package jsonrpc_test

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"strings"
	"testing"
	"time"

	"codeberg.org/turbo-editors/turbo-core/jsonrpc"
)

// newlineFraming marks messages off with a newline, which is what the Agent
// Client Protocol does. It is the simplest framing there is, so the tests are
// about the JSON-RPC layer rather than about parsing headers.
type newlineFraming struct{}

func (newlineFraming) ReadMessage(r *bufio.Reader) ([]byte, error) {
	line, err := r.ReadBytes('\n')
	if err != nil {
		return nil, err
	}
	return line, nil
}

func (newlineFraming) WriteMessage(w io.Writer, body []byte) error {
	_, err := fmt.Fprintf(w, "%s\n", body)
	return err
}

// frame is one message the peer read, or the reason it could not.
type frame struct {
	msg map[string]any
	err error
}

// peer is the far end of a connection, driven from a test.
//
// It is deliberately not built on Conn: a peer sharing the code under test
// could not catch that code writing a malformed frame.
//
// It reads continuously into a buffered channel rather than on demand, because
// net.Pipe is synchronous — a write blocks until somebody reads. With on-demand
// reads, a Reply made from the test goroutine deadlocks against the read that
// was going to collect it. Reading continuously is also what a real client
// does.
type peer struct {
	t      *testing.T
	stream net.Conn
	frames chan frame
}

// newPair returns a Conn and the raw peer at the other end of it.
func newPair(t *testing.T, onNotify jsonrpc.NotificationFunc, onRequest jsonrpc.RequestFunc) (*jsonrpc.Conn, *peer) {
	t.Helper()

	ours, theirs := net.Pipe()
	conn := jsonrpc.NewConn(ours, newlineFraming{}, onNotify, onRequest)
	go func() { _ = conn.Run() }()

	p := &peer{t: t, stream: theirs, frames: make(chan frame, 32)}
	go p.readLoop()

	t.Cleanup(func() { _ = conn.Close(); _ = theirs.Close() })
	return conn, p
}

// readLoop decodes every line the connection sends until it ends.
//
// It never calls t.Fatalf: that is not allowed outside the test goroutine, and
// doing it here hangs the run instead of failing it. Failures travel back as
// values instead.
func (p *peer) readLoop() {
	reader := bufio.NewReader(p.stream)
	for {
		line, err := reader.ReadBytes('\n')
		if err != nil {
			close(p.frames)
			return
		}

		var msg map[string]any
		if err := json.Unmarshal(line, &msg); err != nil {
			p.frames <- frame{err: fmt.Errorf("the connection wrote %q, which is not JSON: %w", line, err)}
			continue
		}
		p.frames <- frame{msg: msg}
	}
}

// send writes one raw line to the connection under test.
func (p *peer) send(line string) {
	p.t.Helper()

	_ = p.stream.SetWriteDeadline(time.Now().Add(2 * time.Second))
	if _, err := fmt.Fprintf(p.stream, "%s\n", line); err != nil {
		p.t.Fatalf("writing to the connection: %v", err)
	}
}

// read returns the next message the connection sent, failing the test if none
// arrives.
func (p *peer) read() map[string]any {
	p.t.Helper()

	select {
	case got, open := <-p.frames:
		if !open {
			p.t.Fatal("the connection ended without sending anything more")
		}
		if got.err != nil {
			p.t.Fatal(got.err)
		}
		return got.msg
	case <-time.After(2 * time.Second):
		p.t.Fatal("the connection sent nothing")
		return nil
	}
}

// readsNothing fails the test if the connection sends anything within a short
// window. It is how "this must not be answered" is stated.
func (p *peer) readsNothing(what string) {
	p.t.Helper()

	select {
	case got := <-p.frames:
		p.t.Fatalf("%s: the connection sent %v", what, got.msg)
	case <-time.After(150 * time.Millisecond):
	}
}

func TestACallGetsItsAnswer(t *testing.T) {
	conn, peer := newPair(t, nil, nil)

	answered := make(chan error, 1)
	var result struct{ Name string }
	go func() { answered <- conn.Call(t.Context(), "greet", map[string]string{"who": "world"}, &result) }()

	request := peer.read()
	if request["method"] != "greet" {
		t.Fatalf("the connection sent %v, want a greet request", request)
	}
	if request["jsonrpc"] != jsonrpc.Version {
		t.Errorf("the request says jsonrpc %v, want %q", request["jsonrpc"], jsonrpc.Version)
	}
	peer.send(fmt.Sprintf(`{"jsonrpc":"2.0","id":%v,"result":{"Name":"hello"}}`, request["id"]))

	if err := <-answered; err != nil {
		t.Fatalf("Call() error = %v", err)
	}
	if result.Name != "hello" {
		t.Errorf("the result is %q, want %q", result.Name, "hello")
	}
}

func TestAPeersRequestIdDoesNotStealAnAnswerMeantForOurOwnCall(t *testing.T) {
	// Both sides number their requests from one, and an agent really does send
	// `id: 1` while a call of ours with the same id is in flight — that is what
	// the Agent Client Protocol's agents do. The two directions are separate
	// id spaces, and a connection that shared one map would hand the peer's
	// *request* to the caller waiting for an answer, or answer the request
	// with the caller's result.
	asked := make(chan string, 1)
	onRequest := func(req *jsonrpc.Request) {
		asked <- req.Method
		req.Reply(map[string]string{"outcome": "allowed"}, nil)
	}
	conn, peer := newPair(t, nil, onRequest)

	answered := make(chan error, 1)
	var result struct{ Name string }
	go func() { answered <- conn.Call(t.Context(), "ours", nil, &result) }()

	ourRequest := peer.read()
	if fmt.Sprint(ourRequest["id"]) != "1" {
		t.Fatalf("our first call took id %v, want 1 — the rest of this test assumes it", ourRequest["id"])
	}

	// The peer now asks a question carrying the same id as our call in flight.
	peer.send(`{"jsonrpc":"2.0","id":1,"method":"theirs","params":{}}`)

	select {
	case method := <-asked:
		if method != "theirs" {
			t.Errorf("the handler was given %q, want %q", method, "theirs")
		}
	case <-time.After(2 * time.Second):
		t.Fatal("the peer's request never reached the handler")
	}

	reply := peer.read()
	if fmt.Sprint(reply["id"]) != "1" {
		t.Errorf("the answer carries id %v, want the peer's own 1", reply["id"])
	}
	if reply["result"] == nil {
		t.Errorf("the peer's request was answered with %v, want a result", reply)
	}

	// Our own call must still be waiting, and must still get its own answer.
	select {
	case err := <-answered:
		t.Fatalf("our call finished early with %v; the peer's request stole its answer", err)
	case <-time.After(50 * time.Millisecond):
	}

	peer.send(`{"jsonrpc":"2.0","id":1,"result":{"Name":"ours"}}`)
	if err := <-answered; err != nil {
		t.Fatalf("Call() error = %v", err)
	}
	if result.Name != "ours" {
		t.Errorf("our call got %q, want %q", result.Name, "ours")
	}
}

func TestARequestMayBeAnsweredLongAfterItArrived(t *testing.T) {
	// This is the whole reason this package exists apart from lsp: a permission
	// dialog is answered by a person, on another goroutine, several turns of an
	// event loop later. A handler that had to return its answer could not wait.
	held := make(chan *jsonrpc.Request, 1)
	_, peer := newPair(t, nil, func(req *jsonrpc.Request) { held <- req })

	peer.send(`{"jsonrpc":"2.0","id":7,"method":"session/request_permission","params":{}}`)

	request := <-held
	if request.Method != "session/request_permission" {
		t.Fatalf("the handler was given %q", request.Method)
	}

	// Nothing may be written until the answer is given.
	peer.readsNothing("answered before the handler did")

	// From another goroutine, as the event loop's dialog really does.
	go request.Reply(map[string]any{"outcome": map[string]string{"outcome": "selected"}}, nil)

	reply := peer.read()
	if fmt.Sprint(reply["id"]) != "7" {
		t.Errorf("the answer carries id %v, want 7", reply["id"])
	}
	if reply["result"] == nil {
		t.Errorf("the answer is %v, want a result", reply)
	}
}

func TestOnlyTheFirstReplyToARequestIsSent(t *testing.T) {
	// A window closing tears down a pending permission that may already have
	// been answered. Two responses carrying one id would desynchronise the
	// peer, which matches answers to requests by exactly that id.
	held := make(chan *jsonrpc.Request, 1)
	_, peer := newPair(t, nil, func(req *jsonrpc.Request) { held <- req })

	peer.send(`{"jsonrpc":"2.0","id":3,"method":"ask","params":{}}`)
	request := <-held

	go request.Reply("first", nil)

	first := peer.read()

	// Only now, with the first answer known to be on the wire, can the later
	// ones be tried: Reply holds the once until its write has finished.
	request.Reply("second", nil)
	request.Reply(nil, errors.New("third"))
	if first["result"] != "first" {
		t.Errorf("the answer is %v, want the first one", first)
	}

	// A second frame must never arrive.
	peer.readsNothing("a second answer was sent")
}

func TestAnErrorFromAHandlerReachesThePeerAsAnErrorObject(t *testing.T) {
	_, peer := newPair(t, nil, func(req *jsonrpc.Request) {
		req.Reply(nil, &jsonrpc.ResponseError{Code: jsonrpc.CodeInvalidParams, Message: "no uri"})
	})

	peer.send(`{"jsonrpc":"2.0","id":2,"method":"fs/read_text_file","params":{}}`)

	reply := peer.read()
	failure, ok := reply["error"].(map[string]any)
	if !ok {
		t.Fatalf("the answer is %v, want an error object", reply)
	}
	if failure["code"] != float64(jsonrpc.CodeInvalidParams) {
		t.Errorf("the code is %v, want %d", failure["code"], jsonrpc.CodeInvalidParams)
	}
	if failure["message"] != "no uri" {
		t.Errorf("the message is %v, want %q", failure["message"], "no uri")
	}
}

func TestAPlainErrorFromAHandlerBecomesAnInternalError(t *testing.T) {
	_, peer := newPair(t, nil, func(req *jsonrpc.Request) {
		req.Reply(nil, errors.New("the file is a directory"))
	})

	peer.send(`{"jsonrpc":"2.0","id":2,"method":"fs/read_text_file","params":{}}`)

	failure, ok := peer.read()["error"].(map[string]any)
	if !ok {
		t.Fatal("want an error object")
	}
	if failure["code"] != float64(jsonrpc.CodeInternalError) {
		t.Errorf("the code is %v, want %d", failure["code"], jsonrpc.CodeInternalError)
	}
	if !strings.Contains(fmt.Sprint(failure["message"]), "the file is a directory") {
		t.Errorf("the message is %v, want the handler's own words in it", failure["message"])
	}
}

func TestARequestWithNoHandlerIsAnsweredMethodNotFound(t *testing.T) {
	// Silence would leave the peer waiting for ever. A peer that asks for
	// something this client does not do has to be told so.
	_, peer := newPair(t, nil, nil)

	peer.send(`{"jsonrpc":"2.0","id":4,"method":"terminal/create","params":{}}`)

	failure, ok := peer.read()["error"].(map[string]any)
	if !ok {
		t.Fatal("want an error object")
	}
	if failure["code"] != float64(jsonrpc.CodeMethodNotFound) {
		t.Errorf("the code is %v, want %d", failure["code"], jsonrpc.CodeMethodNotFound)
	}
	if failure["message"] != "terminal/create" {
		t.Errorf("the message is %v, want the method's name", failure["message"])
	}
}

func TestANotificationReachesTheHandlerAndIsNeverAnswered(t *testing.T) {
	// A notification is exactly a message with no id, and answering one would
	// send a response the peer has nothing to match it to.
	arrived := make(chan string, 1)
	_, peer := newPair(t, func(method string, _ json.RawMessage) { arrived <- method }, nil)

	peer.send(`{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"s"}}`)

	select {
	case method := <-arrived:
		if method != "session/update" {
			t.Errorf("the handler was given %q", method)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("the notification never arrived")
	}

	peer.readsNothing("the notification was answered")
}

func TestNotifySendsNoId(t *testing.T) {
	conn, peer := newPair(t, nil, nil)

	go func() { _ = conn.Notify("session/cancel", map[string]string{"sessionId": "s"}) }()

	sent := peer.read()
	if _, carries := sent["id"]; carries {
		t.Errorf("the notification carries an id: %v", sent)
	}
	if sent["method"] != "session/cancel" {
		t.Errorf("the notification is %v", sent)
	}
}

func TestACallFailsWhenTheConnectionEnds(t *testing.T) {
	// A peer that dies mid-request must not leave the caller blocked for ever.
	conn, peer := newPair(t, nil, nil)

	answered := make(chan error, 1)
	go func() { answered <- conn.Call(context.Background(), "greet", nil, nil) }()

	peer.read() // wait until the request is really on the wire
	_ = peer.stream.Close()

	select {
	case err := <-answered:
		if !errors.Is(err, jsonrpc.ErrClosed) {
			t.Errorf("Call() error = %v, want ErrClosed", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("the call never came back")
	}
}

func TestACallEndsWithItsContext(t *testing.T) {
	conn, peer := newPair(t, nil, nil)

	ctx, cancel := context.WithCancel(context.Background())
	answered := make(chan error, 1)
	go func() { answered <- conn.Call(ctx, "slow", nil, nil) }()

	peer.read()
	cancel()

	select {
	case err := <-answered:
		if !errors.Is(err, context.Canceled) {
			t.Errorf("Call() error = %v, want context.Canceled", err)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("the call ignored its context")
	}
}

func TestAResponseToAnIdWeNeverSentIsIgnored(t *testing.T) {
	// The connection must not panic or die on one; an agent replaying a
	// recorded session can easily send one.
	conn, peer := newPair(t, nil, nil)

	peer.send(`{"jsonrpc":"2.0","id":999,"result":{}}`)

	// Still usable afterwards, which is the whole point.
	answered := make(chan error, 1)
	go func() { answered <- conn.Call(t.Context(), "greet", nil, nil) }()
	request := peer.read()
	peer.send(fmt.Sprintf(`{"jsonrpc":"2.0","id":%v,"result":null}`, request["id"]))

	if err := <-answered; err != nil {
		t.Errorf("the connection stopped working after a stray response: %v", err)
	}
}

func TestCallingOnAClosedConnectionFails(t *testing.T) {
	conn, _ := newPair(t, nil, nil)

	if err := conn.Close(); err != nil {
		t.Fatalf("Close() error = %v", err)
	}
	if err := conn.Call(t.Context(), "greet", nil, nil); !errors.Is(err, jsonrpc.ErrClosed) {
		t.Errorf("Call() error = %v, want ErrClosed", err)
	}
}