package callstest

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"io"
	"net"
	"net/http"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"
)

// sseReader is a browser's EventSource, near enough: one open GET
// against the harness's real server, its lines parsed into events on a
// channel while the test goroutine reads them one at a time.
type sseReader struct {
	t       *testing.T
	cancel  context.CancelFunc
	body    io.ReadCloser
	events  chan sseEvent
	hb      chan struct{}
	done    chan struct{}
	closing chan struct{}
	once    sync.Once
}

type sseEvent struct {
	name string
	data []byte
}

// stream opens an SSE stream and fails the test unless the server
// accepted it. A refusal is read with c.get instead, which needs no
// socket: the status and the JSON body are the whole answer (§4.5).
func (c *client) stream(target string) *sseReader {
	c.t.Helper()
	srv := c.x.server()
	ctx, cancel := context.WithCancel(context.Background())
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+target, nil)
	if err != nil {
		cancel()
		c.t.Fatal(err)
	}
	for _, ck := range c.cookies {
		req.AddCookie(ck)
	}
	req.Header.Set("Accept", "text/event-stream")
	resp, err := srv.Client().Do(req)
	if err != nil {
		cancel()
		c.t.Fatalf("GET %s: %v", target, err)
	}
	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		cancel()
		c.t.Fatalf("GET %s: %d\n%s", target, resp.StatusCode, b)
	}
	if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") {
		resp.Body.Close()
		cancel()
		c.t.Fatalf("GET %s: Content-Type %q", target, ct)
	}
	s := &sseReader{
		t: c.t, cancel: cancel, body: resp.Body,
		events: make(chan sseEvent, 64), hb: make(chan struct{}, 64),
		done: make(chan struct{}), closing: make(chan struct{}),
	}
	c.t.Cleanup(s.close)
	go s.read()
	return s
}

// read parses the wire format the way an EventSource does: a field is
// "name:" with one optional space after the colon, several data lines
// join with a newline, a line starting with a colon is a comment, and a
// blank line dispatches what has been collected.
func (s *sseReader) read() {
	defer close(s.done)
	sc := bufio.NewScanner(s.body)
	sc.Buffer(make([]byte, 0, 64*1024), 1<<20)
	var ev sseEvent
	var hasData bool
	for sc.Scan() {
		line := sc.Text()
		switch {
		case line == "":
			if ev.name != "" || hasData {
				s.dispatch(ev)
			}
			ev, hasData = sseEvent{}, false
		case strings.HasPrefix(line, ":"):
			select {
			case s.hb <- struct{}{}:
			default:
			}
		case strings.HasPrefix(line, "event:"):
			ev.name = field(line, "event:")
		case strings.HasPrefix(line, "data:"):
			if hasData {
				ev.data = append(ev.data, '\n')
			}
			ev.data = append(ev.data, field(line, "data:")...)
			hasData = true
		}
	}
}

func field(line, name string) string {
	return strings.TrimPrefix(strings.TrimPrefix(line, name), " ")
}

// dispatch hands one event to the test goroutine. It blocks rather than
// dropping: an overflowing buffer means a test is not reading what it
// asked for, and that should fail out loud rather than quietly lose the
// event that would have failed the assertion.
func (s *sseReader) dispatch(ev sseEvent) {
	select {
	case s.events <- ev:
	case <-s.closing:
	case <-time.After(5 * time.Second):
		s.t.Errorf("sse: nothing read %q for five seconds; the event buffer overflowed", ev.name)
	}
}

// next waits for one event.
func (s *sseReader) next(timeout time.Duration) (string, []byte, bool) {
	select {
	case ev := <-s.events:
		return ev.name, ev.data, true
	case <-time.After(timeout):
		return "", nil, false
	}
}

// hello reads the first event, which is always hello (spec §4.1), and
// returns the peer id and the boot id it carries.
func (s *sseReader) hello() (peer, boot string) {
	s.t.Helper()
	name, data, ok := s.next(2 * time.Second)
	if !ok || name != "hello" {
		s.t.Fatalf("first event is %q (ok=%v), want hello", name, ok)
	}
	var body struct{ Peer, Boot string }
	if err := json.Unmarshal(data, &body); err != nil {
		s.t.Fatalf("hello %s: %v", data, err)
	}
	return body.Peer, body.Boot
}

// nextOf skips events until the named one arrives, or the timeout
// passes — for a test that expects some event (a roster, a peer) ahead
// of the one it actually wants to assert on.
func (s *sseReader) nextOf(event string, timeout time.Duration) (string, []byte, bool) {
	s.t.Helper()
	deadline := time.After(timeout)
	for {
		select {
		case ev := <-s.events:
			if ev.name == event {
				return ev.name, ev.data, true
			}
		case <-deadline:
			return "", nil, false
		}
	}
}

// waitHeartbeat waits for one SSE comment line, which is the only thing
// the server sends on an idle stream.
func (s *sseReader) waitHeartbeat(timeout time.Duration) bool {
	select {
	case <-s.hb:
		return true
	case <-time.After(timeout):
		return false
	}
}

// waitClosed waits for the server to end the stream.
func (s *sseReader) waitClosed(timeout time.Duration) bool {
	select {
	case <-s.done:
		return true
	case <-time.After(timeout):
		return false
	}
}

// close is what a browser navigating away does. It waits for the read
// goroutine to finish so the test never races the server's own exit.
func (s *sseReader) close() {
	s.once.Do(func() {
		close(s.closing)
		s.cancel()
		s.body.Close()
		select {
		case <-s.done:
		case <-time.After(2 * time.Second):
			s.t.Error("the stream reader did not stop")
		}
	})
}

// rawStream is the stream with nobody draining it in the background —
// the one thing sseReader cannot be, because its read goroutine is
// always reading. A test that wants the server's own 64-event
// subscriber buffer to overflow (spec §4.4) has to stop reading the
// socket: that is what backs the handler's write up, and only a handler
// that is not draining lets the buffer behind it fill.
//
// Nothing here parses more of the wire format than the tests need: an
// event name, its data, and heartbeat comments skipped.
type rawStream struct {
	t      *testing.T
	cancel context.CancelFunc
	body   io.ReadCloser
	r      *bufio.Reader
	once   sync.Once
}

// tinyWindowClient dials with a deliberately small receive buffer. It is
// what makes "the client stopped reading" mean something: with the
// kernel's own defaults, loopback will happily absorb megabytes on the
// reader's behalf, and the server's write never blocks, so the hub's
// buffer never fills and the test proves nothing. SO_RCVBUF also turns
// receive-window autotuning off, which is the point — the window must
// stay small rather than growing to meet the flood.
func tinyWindowClient() *http.Client {
	d := &net.Dialer{
		Control: func(network, address string, c syscall.RawConn) error {
			var setErr error
			if err := c.Control(func(fd uintptr) {
				setErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_RCVBUF, 4096)
			}); err != nil {
				return err
			}
			return setErr
		},
	}
	return &http.Client{Transport: &http.Transport{DialContext: d.DialContext}}
}

func (c *client) rawStream(target string) *rawStream {
	c.t.Helper()
	srv := c.x.server()
	ctx, cancel := context.WithCancel(context.Background())
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+target, nil)
	if err != nil {
		cancel()
		c.t.Fatal(err)
	}
	for _, ck := range c.cookies {
		req.AddCookie(ck)
	}
	req.Header.Set("Accept", "text/event-stream")
	resp, err := tinyWindowClient().Do(req)
	if err != nil {
		cancel()
		c.t.Fatalf("GET %s: %v", target, err)
	}
	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		cancel()
		c.t.Fatalf("GET %s: %d\n%s", target, resp.StatusCode, b)
	}
	// A small reader on purpose: the socket must stop being drained the
	// moment the test stops calling next, not a buffer's worth later.
	s := &rawStream{t: c.t, cancel: cancel, body: resp.Body, r: bufio.NewReaderSize(resp.Body, 512)}
	c.t.Cleanup(s.close)
	return s
}

// next reads one event, blocking until the server sends one or the
// timeout cancels the request out from under the read.
func (s *rawStream) next(timeout time.Duration) (string, []byte, bool) {
	s.t.Helper()
	stop := time.AfterFunc(timeout, s.cancel)
	defer stop.Stop()
	var name string
	var data []byte
	for {
		raw, err := s.r.ReadString('\n')
		if err != nil {
			return "", nil, false
		}
		line := strings.TrimRight(raw, "\r\n")
		switch {
		case line == "":
			if name != "" || data != nil {
				return name, data, true
			}
		case strings.HasPrefix(line, ":"): // heartbeat
		case strings.HasPrefix(line, "event:"):
			name = field(line, "event:")
		case strings.HasPrefix(line, "data:"):
			data = append(data, field(line, "data:")...)
		}
	}
}

// hello is next's first event, which is always hello (§4.1).
func (s *rawStream) hello() string {
	s.t.Helper()
	name, data, ok := s.next(2 * time.Second)
	if !ok || name != "hello" {
		s.t.Fatalf("first event is %q (ok=%v), want hello", name, ok)
	}
	var body struct{ Peer string }
	if err := json.Unmarshal(data, &body); err != nil {
		s.t.Fatalf("hello %s: %v", data, err)
	}
	return body.Peer
}

func (s *rawStream) close() {
	s.once.Do(func() {
		s.cancel()
		s.body.Close()
	})
}

// signal is POST /m/{sqid}/signal, the upstream half of §4.1: a JSON
// body carrying the peer and the message, under CSRF. body is decoded
// into a map and given "peer", rather than spliced in as a string — a
// caller's body of "{}" or "" is not a special case this way, and every
// call sends one well-formed JSON document. The endpoint itself arrives
// with the signal handler; this is the client for it.
func (c *client) signal(sqid, peer, body string) (*http.Response, []byte) {
	c.t.Helper()
	msg := map[string]any{}
	if strings.TrimSpace(body) != "" {
		if err := json.Unmarshal([]byte(body), &msg); err != nil {
			c.t.Fatalf("signal body %q: %v", body, err)
		}
	}
	msg["peer"] = peer
	payload, err := json.Marshal(msg)
	if err != nil {
		c.t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.x.server().URL+"/m/"+sqid+"/signal", bytes.NewReader(payload))
	if err != nil {
		c.t.Fatal(err)
	}
	for _, ck := range c.cookies {
		req.AddCookie(ck)
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Origin", testOrigin)
	resp, err := c.x.server().Client().Do(req)
	if err != nil {
		c.t.Fatalf("POST signal: %v", err)
	}
	b, err := io.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		c.t.Fatalf("POST signal: reading body: %v", err)
	}
	return resp, b
}

// signalRaw posts body verbatim, unlike signal, which always injects a
// well-formed "peer" — for the shapes signal cannot express at all: a
// malformed document, or one with no peer field.
func (c *client) signalRaw(sqid, body string) (*http.Response, []byte) {
	c.t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.x.server().URL+"/m/"+sqid+"/signal", strings.NewReader(body))
	if err != nil {
		c.t.Fatal(err)
	}
	for _, ck := range c.cookies {
		req.AddCookie(ck)
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Origin", testOrigin)
	resp, err := c.x.server().Client().Do(req)
	if err != nil {
		c.t.Fatalf("POST signal: %v", err)
	}
	b, err := io.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		c.t.Fatalf("POST signal: reading body: %v", err)
	}
	return resp, b
}