// Package server tests exercise the relay coordinator's pure logic: the
// subscription registry and its caps, envelope peeking, the IP/CORS/NIP-11
// decisions, the worker-pool bookkeeping (driven over constructed channels,
// with real spawn domains only for startup/respawn), the async reaper, and the
// database reply dispatch. Anything that needs a live socket is out of scope.
package server

import (
	"bytes"
	"testing"
	"time"

	"git.smesh.lol/nostr/pkg/envelope"
	"git.smesh.lol/nostr/pkg/event"
	"git.smesh.lol/nostr/pkg/kind"
	"git.smesh.lol/nostr/pkg/signer/p8k"
	"git.smesh.lol/nostr/pkg/tag"
	"git.smesh.lol/morly/pkg/pool"
	"git.smesh.lol/morly/pkg/relay/config"
	"git.smesh.lol/morly/pkg/relay/ratelimit"
	"git.smesh.lol/morly/pkg/relay/tree"
	"git.smesh.lol/morly/pkg/relay/wire"
	"git.smesh.lol/morly/pkg/transport"
)

// --- helpers ---

// stCfg is a minimal config: no relay URL (ungated), no auth, no CORS origins.
func stCfg() (c *config.C) {
	c = &config.C{}
	c.AppName = "musiquay-test"
	c.ACLMode = "none"
	c.NIP70Enforce = false
	c.NegentropyEnabled = false
	c.MaxSubscriptions = 10000
	c.QueryResultLimit = 256
	c.MaxConnPerIP = 100
	return
}

// stServer builds a Server with no spawned domains and an empty transport.
// transport.New stores the handler; nil never gets called.
func stServer(c *config.C) (s *Server) {
	s = &Server{
		cfg:          c,
		conns:        map[int32]*cstate{},
		dbPending:    map[uint32]int32{},
		pendingReq:   map[uint32]int32{},
		asyncPending: map[uint32]asyncHTTPEntry{},
		t:            transport.New(nil, c.MaxConnPerIP),
	}
	return
}

// stFakeDB is a DB handle whose channels are buffered and whose done channel
// stays open, so Valid() is true and dbSend never blocks.
func stFakeDB() (d *DB) {
	d = &DB{
		in:    chan tree.Request{8},
		out:   chan tree.Response{8},
		ready: chan struct{}{8},
		done:  chan struct{}{},
	}
	return
}

func stSignedEvent(t *testing.T, k uint16, tags *tag.S, content string) (ev *event.E) {
	t.Helper()
	sg := p8k.MustNew()
	if gerr := sg.Generate(); gerr != nil {
		t.Fatal(gerr)
		return
	}
	ev = event.New()
	ev.CreatedAt = time.Now().Unix()
	ev.Kind = k
	ev.Tags = tags
	ev.Content = []byte(content)
	if serr := ev.Sign(sg); serr != nil {
		t.Fatal(serr)
		return
	}
	return
}

func stEventMsg(ev *event.E) (msg []byte) {
	es := &envelope.EventSubmission{E: ev}
	return es.Marshal(nil)
}

// stHas is a local substring scan. bytes.Contains/Index can miss a needle in a
// long haystack in this compiler, so the coverage tests do not trust them.
func stHas(hay, needle []byte) (found bool) {
	if len(needle) == 0 {
		return true
	}
	if len(needle) > len(hay) {
		return false
	}
	var i int32
	var j int32
	var match bool
	for i = 0; i+len(needle) <= len(hay); i++ {
		match = true
		for j = 0; j < len(needle); j++ {
			if hay[i+j] != needle[j] {
				match = false
				break
			}
		}
		if match {
			return true
		}
	}
	return false
}

func stTryProxyReq(ch chan wire.ProxyRequest) (r wire.ProxyRequest, ok bool) {
	select {
	case v := <-ch:
		r = v
		ok = true
	default:
	}
	return
}

func stTryIngestReq(ch chan wire.IngestRequest) (r wire.IngestRequest, ok bool) {
	select {
	case v := <-ch:
		r = v
		ok = true
	default:
	}
	return
}

func stTryDBReq(ch chan tree.Request) (r tree.Request, ok bool) {
	select {
	case v := <-ch:
		r = v
		ok = true
	default:
	}
	return
}

// stIngestSrv builds an ingest pool with constructed channels. The done
// channels stay open so workerAlive reports every worker alive.
func stIngestSrv(n int32) (s *Server) {
	s = stServer(stCfg())
	// push, not []chan T{:n}: the inline slice-of-channel literal is
	// miscompiled to an empty slice, so the pool is grown element by element.
	s.workers = pool.NewPool(n)
	for i := int32(0); i < n; i++ {
		s.workerIn = push(s.workerIn, chan wire.IngestRequest{4})
		s.workerOut = push(s.workerOut, chan wire.IngestResponse{4})
		s.workerDone = push(s.workerDone, chan struct{}{})
		s.workerReady = push(s.workerReady, chan struct{}{4})
	}
	s.db = stFakeDB()
	return
}

// stProxySrv builds a proxy pool with constructed channels.
func stProxySrv(n int32) (s *Server) {
	s = stServer(stCfg())
	s.proxyPool = pool.NewPool(n)
	s.proxyBusyTime = []int64{:n}
	for i := int32(0); i < n; i++ {
		s.proxyIn = push(s.proxyIn, chan wire.ProxyRequest{4})
		s.proxyOut = push(s.proxyOut, chan wire.ProxyResponse{4})
		s.proxyDone = push(s.proxyDone, chan struct{}{})
	}
	return
}

// stBlossomSrv builds a blossom pool with constructed channels.
func stBlossomSrv(n int32) (s *Server) {
	s = stServer(stCfg())
	s.blossomPool = pool.NewPool(n)
	for i := int32(0); i < n; i++ {
		s.blossomIn = push(s.blossomIn, chan wire.BlossomRequest{4})
		s.blossomOut = push(s.blossomOut, chan wire.BlossomResponse{4})
		s.blossomDone = push(s.blossomDone, chan struct{}{})
	}
	return
}

// --- envelope peeking ---

func TestPeekEventKind(t *testing.T) {
	ev := stSignedEvent(t, 1, nil, "hello")
	if k := peekEventKind(stEventMsg(ev)); k != 1 {
		t.Fatalf("peekEventKind = %d, want 1", k)
		return
	}
	eph := stSignedEvent(t, 20000, nil, "eph")
	if k := peekEventKind(stEventMsg(eph)); k != 20000 {
		t.Fatalf("peekEventKind ephemeral = %d, want 20000", k)
		return
	}
	replace := stSignedEvent(t, 30023, nil, "article")
	if k := peekEventKind(stEventMsg(replace)); k != 30023 {
		t.Fatalf("peekEventKind param replaceable = %d", k)
		return
	}
}

// --- worker lifecycle ---

func TestWorkerAlive(t *testing.T) {
	d := chan struct{}{}
	if !workerAlive(d) {
		t.Fatal("open channel must be alive")
		return
	}
	close(d)
	if workerAlive(d) {
		t.Fatal("closed channel must be dead")
		return
	}
}

// --- URL matching ---

func TestStripScheme(t *testing.T) {
	if string(stripScheme([]byte("wss://relay.example.com"))) != "relay.example.com" {
		t.Fatalf("wss strip = %s", stripScheme([]byte("wss://relay.example.com")))
		return
	}
	if string(stripScheme([]byte("ws://relay.example.com"))) != "relay.example.com" {
		t.Fatalf("ws strip = %s", stripScheme([]byte("ws://relay.example.com")))
		return
	}
	if string(stripScheme([]byte("https://relay.example.com"))) != "https://relay.example.com" {
		t.Fatalf("https must be untouched, got %s", stripScheme([]byte("https://relay.example.com")))
		return
	}
	if string(stripScheme([]byte("relay.example.com"))) != "relay.example.com" {
		t.Fatal("bare host must be untouched")
		return
	}
}

func TestRelayURLMatch(t *testing.T) {
	if !relayURLMatch([]byte("wss://relay.example.com"), []byte("wss://relay.example.com")) {
		t.Fatal("identical URLs must match")
		return
	}
	if !relayURLMatch([]byte("wss://relay.example.com"), []byte("relay.example.com")) {
		t.Fatal("scheme difference must match")
		return
	}
	if !relayURLMatch([]byte("wss://relay.example.com/"), []byte("relay.example.com")) {
		t.Fatal("trailing slash must match")
		return
	}
	if !relayURLMatch([]byte("WSS://RELAY.EXAMPLE.COM"), []byte("relay.example.com")) {
		t.Fatal("case difference must match")
		return
	}
	if !relayURLMatch([]byte("ws://relay.example.com"), []byte("wss://relay.example.com")) {
		t.Fatal("ws vs wss must match after scheme strip")
		return
	}
	if relayURLMatch([]byte("wss://a.example.com"), []byte("b.example.com")) {
		t.Fatal("different hosts must not match")
		return
	}
	if relayURLMatch([]byte("https://relay.example.com"), []byte("relay.example.com")) {
		t.Fatal("unrecognised scheme must not be stripped")
		return
	}
}

func TestMakeCopyAndToLower(t *testing.T) {
	orig := []byte("AbC")
	cp := makeCopy(orig)
	if !bytes.Equal(cp, []byte("AbC")) {
		t.Fatalf("makeCopy = %s", cp)
		return
	}
	cp[0] = 'x'
	if orig[0] != 'A' {
		t.Fatal("makeCopy must not alias the source")
		return
	}
	up := toLower([]byte("AbC-9"))
	if string(up) != "abc-9" {
		t.Fatalf("toLower = %s", up)
		return
	}
}

// --- IP access control ---

func TestIPBlacklistAndWhitelist(t *testing.T) {
	c := stCfg()
	c.IPBlacklist = []string{"10.0.0.", "1.2.3.4"}
	c.IPWhitelist = []string{"192.168."}
	s := stServer(c)
	if !s.ipBlacklisted("10.0.0.7") {
		t.Fatal("10.0.0.7 must be blacklisted")
		return
	}
	if !s.ipBlacklisted("1.2.3.4") {
		t.Fatal("exact blacklist entry must match")
		return
	}
	if s.ipBlacklisted("11.0.0.7") {
		t.Fatal("11.0.0.7 must not be blacklisted")
		return
	}
	if !s.ipWhitelisted("192.168.1.1") {
		t.Fatal("192.168.1.1 must be whitelisted")
		return
	}
	if s.ipWhitelisted("10.0.0.1") {
		t.Fatal("10.0.0.1 must not be whitelisted")
		return
	}

	empty := stServer(stCfg())
	if empty.ipWhitelisted("192.168.1.1") {
		t.Fatal("empty whitelist must never match")
		return
	}
	if empty.ipBlacklisted("10.0.0.1") {
		t.Fatal("empty blacklist must never match")
		return
	}
}

func TestOnAccept(t *testing.T) {
	c := stCfg()
	c.MaxGlobalConns = 5
	c.IPBlacklist = []string{"9.9."}
	s := stServer(c)
	if !s.OnAccept(1, "1.1.1.1") {
		t.Fatal("a normal IP under the global cap must be accepted")
		return
	}
	if s.OnAccept(2, "9.9.9.9") {
		t.Fatal("a blacklisted IP must be refused")
		return
	}
	unlimited := stCfg()
	unlimited.MaxGlobalConns = 0
	s2 := stServer(unlimited)
	if !s2.OnAccept(1, "1.1.1.1") {
		t.Fatal("MaxGlobalConns=0 must mean unlimited")
		return
	}
}

func TestOnWSUpgrade(t *testing.T) {
	c := stCfg()
	c.MaxConnPerIP = 3
	c.IPWhitelist = []string{"10."}
	s := stServer(c)

	wl, allow := s.OnWSUpgrade(1, "1.1.1.1", 1)
	if wl || !allow {
		t.Fatalf("under limit: whitelisted=%v allow=%v", wl, allow)
		return
	}
	wl2, allow2 := s.OnWSUpgrade(1, "1.1.1.1", 3)
	if wl2 || allow2 {
		t.Fatalf("at limit: whitelisted=%v allow=%v", wl2, allow2)
		return
	}
	wl3, allow3 := s.OnWSUpgrade(1, "10.1.1.1", 99)
	if !wl3 || !allow3 {
		t.Fatalf("whitelisted over limit must pass: whitelisted=%v allow=%v", wl3, allow3)
		return
	}

	unlimited := stCfg()
	unlimited.MaxConnPerIP = 0
	s2 := stServer(unlimited)
	wl4, allow4 := s2.OnWSUpgrade(1, "1.1.1.1", 1000)
	if wl4 || !allow4 {
		t.Fatal("MaxConnPerIP=0 must mean unlimited")
		return
	}
}

func TestConnHelpers(t *testing.T) {
	s := stServer(stCfg())
	if s.t.ConnCount() != 0 {
		t.Fatalf("ConnCount = %d, want 0", s.t.ConnCount())
		return
	}
	if s.t.ConnIP(7) != "" {
		t.Fatalf("unknown fd ConnIP = %q, want empty", s.t.ConnIP(7))
		return
	}
	if s.t.ConnIsWhitelisted(7) {
		t.Fatal("unknown fd must not be whitelisted")
		return
	}
}

// --- CORS ---

func TestCorsHeaders(t *testing.T) {
	c := stCfg()
	s := stServer(c)

	open := s.corsHeaders(map[string]string{})
	if open["Access-Control-Allow-Origin"] != "*" {
		t.Fatalf("default origin = %q", open["Access-Control-Allow-Origin"])
		return
	}
	echo := s.corsHeaders(map[string]string{"origin": "https://a.example"})
	if echo["Access-Control-Allow-Origin"] != "https://a.example" {
		t.Fatalf("echoed origin = %q", echo["Access-Control-Allow-Origin"])
		return
	}

	c.CORSOrigins = []string{"https://a.example"}
	allowed := s.corsHeaders(map[string]string{"origin": "https://a.example"})
	if allowed["Access-Control-Allow-Origin"] != "https://a.example" {
		t.Fatal("listed origin must be allowed")
		return
	}
	denied := s.corsHeaders(map[string]string{"origin": "https://evil.example"})
	if len(denied) != 0 {
		t.Fatalf("unlisted origin must produce no headers, got %d", len(denied))
		return
	}

	c.CORSOrigins = []string{"*"}
	wild := s.corsHeaders(map[string]string{"origin": "https://evil.example"})
	if wild["Access-Control-Allow-Origin"] != "https://evil.example" {
		t.Fatal("wildcard entry must allow any origin")
		return
	}
}

// --- NIP-11 ---

func TestNip11JSON(t *testing.T) {
	c := stCfg()
	c.MaxSubscriptions = 0 // exercises the 10000 default
	s := stServer(c)
	s.Version = "1.2.3"

	buf := s.nip11JSON()
	if !stHas(buf, []byte("\"name\":\"musiquay-test\"")) {
		t.Fatalf("missing name in %s", buf)
		return
	}
	if !stHas(buf, []byte("\"version\":\"1.2.3\"")) {
		t.Fatal("missing version")
		return
	}
	if !stHas(buf, []byte("\"supported_nips\":[1,9,11,40,42,45,50]")) {
		t.Fatalf("unexpected nips list in %s", buf)
		return
	}
	if !stHas(buf, []byte("\"max_subscriptions\":10000")) {
		t.Fatal("max_subscriptions default must be 10000")
		return
	}
	if !stHas(buf, []byte("\"auth_required\":false")) {
		t.Fatal("auth_required must be false")
		return
	}
	if !stHas(buf, []byte("\"restricted_writes\":false")) {
		t.Fatal("restricted_writes must be false")
		return
	}
	if stHas(buf, []byte("\"pubkey\"")) {
		t.Fatal("pubkey must be omitted with no admins")
		return
	}

	c2 := stCfg()
	c2.NIP70Enforce = true
	c2.NegentropyEnabled = true
	c2.Admins = []string{"deadbeef"}
	c2.RelayURL = "wss://relay.example.com"
	c2.AuthRequired = true
	c2.ACLMode = "managed"
	c2.MaxSubscriptions = 5
	c2.QueryResultLimit = 7
	s2 := stServer(c2)
	s2.Version = "9"
	buf2 := s2.nip11JSON()
	if !stHas(buf2, []byte(",70")) {
		t.Fatal("NIP-70 must be advertised")
		return
	}
	if !stHas(buf2, []byte(",77")) {
		t.Fatal("NIP-77 must be advertised")
		return
	}
	if !stHas(buf2, []byte("\"pubkey\":\"deadbeef\"")) {
		t.Fatal("admin pubkey must be advertised")
		return
	}
	if !stHas(buf2, []byte("\"contact\":\"wss://relay.example.com\"")) {
		t.Fatal("relay URL must be the contact")
		return
	}
	if !stHas(buf2, []byte("\"max_subscriptions\":5")) {
		t.Fatal("max_subscriptions must be the configured value")
		return
	}
	if !stHas(buf2, []byte("\"max_limit\":7")) {
		t.Fatal("max_limit must be the query result limit")
		return
	}
	if !stHas(buf2, []byte("\"auth_required\":true")) {
		t.Fatal("auth_required must be true")
		return
	}
	if !stHas(buf2, []byte("\"restricted_writes\":true")) {
		t.Fatal("restricted_writes must be true")
		return
	}
}

// --- async reaper ---

func TestAsyncReapStuck(t *testing.T) {
	s := stServer(stCfg())
	now := time.Now().UnixNano()
	s.asyncPending[1] = asyncHTTPEntry{connFD: 10, createdAt: now - 120_000_000_000}
	s.asyncPending[2] = asyncHTTPEntry{connFD: 11, createdAt: now}
	s.asyncPending[3] = asyncHTTPEntry{connFD: 12, createdAt: 0}
	s.asyncReapStuck()

	if _, ok := s.asyncPending[1]; ok {
		t.Fatal("an entry older than 60s must be reaped")
		return
	}
	if _, ok := s.asyncPending[2]; !ok {
		t.Fatal("a fresh entry must survive")
		return
	}
	if _, ok := s.asyncPending[3]; !ok {
		t.Fatal("createdAt=0 must not be treated as stale")
		return
	}
	if len(s.asyncPending) != 2 {
		t.Fatalf("asyncPending size = %d, want 2", len(s.asyncPending))
		return
	}
}

// --- proxy queue bookkeeping ---

func TestDrainProxyQueue(t *testing.T) {
	s := stProxySrv(2)
	s.proxyQueue = push(s.proxyQueue, pendingProxy{
		connFD: 7, connClose: true, req: wire.ProxyRequest{ReqID: 1, URL: []byte("https://a")},
	})
	s.proxyQueue = push(s.proxyQueue, pendingProxy{
		connFD: 8, connClose: false, req: wire.ProxyRequest{ReqID: 2, URL: []byte("https://b")},
	})
	s.drainProxyQueue()

	if len(s.proxyQueue) != 0 {
		t.Fatalf("queue must drain, %d left", len(s.proxyQueue))
		return
	}
	if len(s.asyncPending) != 2 {
		t.Fatalf("asyncPending = %d, want 2", len(s.asyncPending))
		return
	}
	if !s.proxyPool.Busy[0] || !s.proxyPool.Busy[1] {
		t.Fatal("both workers must be marked busy")
		return
	}
	if s.proxyBusyTime[0] == 0 || s.proxyBusyTime[1] == 0 {
		t.Fatal("busy time must be stamped")
		return
	}
	r0, ok0 := stTryProxyReq(s.proxyIn[0])
	if !ok0 || r0.ReqID != 1 {
		t.Fatalf("worker 0 request = %v ok=%v", r0.ReqID, ok0)
		return
	}
	r1, ok1 := stTryProxyReq(s.proxyIn[1])
	if !ok1 || r1.ReqID != 2 {
		t.Fatalf("worker 1 request = %v ok=%v", r1.ReqID, ok1)
		return
	}
	e, eok := s.asyncPending[1]
	if !eok || e.connFD != 7 || !e.connClose {
		t.Fatal("asyncPending entry 1 must carry the pending connection")
		return
	}
}

func TestDrainProxyQueueStopsWhenBusy(t *testing.T) {
	s := stProxySrv(1)
	s.proxyQueue = push(s.proxyQueue, pendingProxy{connFD: 1, req: wire.ProxyRequest{ReqID: 1}})
	s.proxyQueue = push(s.proxyQueue, pendingProxy{connFD: 2, req: wire.ProxyRequest{ReqID: 2}})
	s.drainProxyQueue()

	if len(s.proxyQueue) != 1 {
		t.Fatalf("one entry must stay queued, %d left", len(s.proxyQueue))
		return
	}
	if !s.proxyPool.Busy[0] {
		t.Fatal("the single worker must be busy")
		return
	}
	if s.proxyQueue[0].req.ReqID != 2 {
		t.Fatalf("the queued entry must be req 2, got %d", s.proxyQueue[0].req.ReqID)
		return
	}
}

func TestDoDispatchProxyBranches(t *testing.T) {
	s := stProxySrv(1)
	s.selfHost = "self.example/"

	if !s.doDispatchProxy(1, "/proxy/", map[string]string{}) {
		t.Fatal("empty target must be handled synchronously")
		return
	}
	if len(s.asyncPending) != 0 {
		t.Fatal("empty target must not register async state")
		return
	}

	if !s.doDispatchProxy(2, "/proxy/self.example/foo", map[string]string{}) {
		t.Fatal("self-host target must be handled synchronously")
		return
	}
	if len(s.asyncPending) != 0 {
		t.Fatal("self-host redirect must not register async state")
		return
	}

	// Queue-full path: the only worker is busy and 64 entries are already
	// waiting, so the request must be refused with 503.
	s.proxyPool.Busy[0] = true
	for i := int32(0); i < 64; i++ {
		s.proxyQueue = push(s.proxyQueue, pendingProxy{req: wire.ProxyRequest{ReqID: uint32(i)}})
	}
	if !s.doDispatchProxy(3, "/proxy/https://other.example/x", map[string]string{}) {
		t.Fatal("a refused proxy request still reports handled")
		return
	}
	if len(s.proxyQueue) != 64 {
		t.Fatalf("a full queue must not grow, size = %d", len(s.proxyQueue))
		return
	}
}

// --- ingest pool ---

func TestDispatchToWorker(t *testing.T) {
	s := stIngestSrv(1)
	if !s.dispatchToWorker(5, []byte("[\"EVENT\",{}]")) {
		t.Fatal("an idle worker must accept the request")
		return
	}
	if !s.workers.Busy[0] {
		t.Fatal("the worker must be marked busy")
		return
	}
	if len(s.pendingReq) != 1 {
		t.Fatalf("pendingReq = %d, want 1", len(s.pendingReq))
		return
	}
	req, ok := stTryIngestReq(s.workerIn[0])
	if !ok || req.ReqID != 1 {
		t.Fatalf("queued ingest request = %v ok=%v", req.ReqID, ok)
		return
	}
	if s.dispatchToWorker(5, []byte("x")) {
		t.Fatal("a busy pool must refuse dispatch")
		return
	}
}

func TestCompleteIngestResponseReject(t *testing.T) {
	s := stIngestSrv(1)
	fd := int32(7)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	s.pendingReq[3] = fd

	var r wire.IngestResponse
	r.ReqID = 3
	r.Verdict = wire.VerdictReject
	r.Reason = []byte("invalid: nope")
	s.completeIngestResponse(r)

	if _, ok := s.pendingReq[3]; ok {
		t.Fatal("a completed request must be dropped from pendingReq")
		return
	}
	if len(s.dbPending) != 0 {
		t.Fatal("a rejected event must not reach the database")
		return
	}
}

func TestCompleteIngestResponseAccept(t *testing.T) {
	s := stIngestSrv(1)
	fd := int32(7)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	s.pendingReq[3] = fd

	var r wire.IngestResponse
	r.ReqID = 3
	r.Verdict = wire.VerdictAccept
	r.Bytes = []byte("[\"EVENT\",{}]")
	s.completeIngestResponse(r)

	if _, ok := s.pendingReq[3]; ok {
		t.Fatal("an accepted request must be dropped from pendingReq")
		return
	}
	if len(s.dbPending) != 1 {
		t.Fatalf("dbPending = %d, want 1", len(s.dbPending))
		return
	}
	req, got := stTryDBReq(s.db.in)
	if !got {
		t.Fatal("the accepted event must be forwarded to the database")
		return
	}
	if req.Op != tree.OpPersist || req.ConnID != fd || !req.Verified {
		t.Fatalf("db request op=%d conn=%d verified=%v", req.Op, req.ConnID, req.Verified)
		return
	}
}

func TestCompleteIngestResponseUnknownAndGone(t *testing.T) {
	s := stIngestSrv(1)
	s.conns[7] = &cstate{subs: map[string]*sub{}}

	var unknown wire.IngestResponse
	unknown.ReqID = 99
	unknown.Verdict = wire.VerdictAccept
	s.completeIngestResponse(unknown)
	if len(s.dbPending) != 0 {
		t.Fatal("an unknown ReqID must be ignored")
		return
	}

	// Known request but the connection is gone: drop it without forwarding.
	s.pendingReq[4] = 42
	var gone wire.IngestResponse
	gone.ReqID = 4
	gone.Verdict = wire.VerdictAccept
	s.completeIngestResponse(gone)
	if _, ok := s.pendingReq[4]; ok {
		t.Fatal("a request for a closed connection must be dropped")
		return
	}
	if len(s.dbPending) != 0 {
		t.Fatal("a closed connection must not reach the database")
		return
	}
}

func TestPollIngestWorkersDrainsReady(t *testing.T) {
	s := stIngestSrv(2)
	fd := int32(5)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	s.pendingReq[9] = fd
	s.workers.Busy[0] = true

	s.workerOut[0] <- wire.IngestResponse{ReqID: 9, Verdict: wire.VerdictReject, Reason: []byte("invalid: nope")}
	s.workerReady[0] <- struct{}{}
	s.pollIngestWorkers()

	if s.workers.Busy[0] {
		t.Fatal("a drained worker must be idle again")
		return
	}
	if _, ok := s.pendingReq[9]; ok {
		t.Fatal("the drained response must clear pendingReq")
		return
	}
	if s.workers.Busy[1] {
		t.Fatal("the untouched worker must stay idle")
		return
	}
}

// --- proxy responses ---

func TestCompleteProxyResponseStatusBranches(t *testing.T) {
	s := stProxySrv(1)

	s.asyncPending[1] = asyncHTTPEntry{connFD: 9}
	var errResp wire.ProxyResponse
	errResp.ReqID = 1
	errResp.Status = -1
	errResp.Err = []byte("boom")
	s.completeProxyResponse(errResp)
	if len(s.asyncPending) != 0 {
		t.Fatal("502 branch must clear asyncPending")
		return
	}

	s.asyncPending[2] = asyncHTTPEntry{connFD: 9}
	var ctResp wire.ProxyResponse
	ctResp.ReqID = 2
	ctResp.Status = 415
	s.completeProxyResponse(ctResp)
	if len(s.asyncPending) != 0 {
		t.Fatal("415 branch must clear asyncPending")
		return
	}

	s.asyncPending[3] = asyncHTTPEntry{connFD: 9}
	var okResp wire.ProxyResponse
	okResp.ReqID = 3
	okResp.Status = 200
	okResp.ContentType = []byte("image/png")
	okResp.Body = []byte("BODY")
	s.completeProxyResponse(okResp)
	if len(s.asyncPending) != 0 {
		t.Fatal("2xx branch must clear asyncPending")
		return
	}

	s.asyncPending[4] = asyncHTTPEntry{connFD: 9}
	var upResp wire.ProxyResponse
	upResp.ReqID = 4
	upResp.Status = 404
	s.completeProxyResponse(upResp)
	if len(s.asyncPending) != 0 {
		t.Fatal("default branch must clear asyncPending")
		return
	}

	// An unknown ReqID is a no-op.
	var stale wire.ProxyResponse
	stale.ReqID = 77
	stale.Status = 200
	s.completeProxyResponse(stale)
	if len(s.asyncPending) != 0 {
		t.Fatal("an unknown proxy response must not create state")
		return
	}
}

func TestPollProxyWorkersDrains(t *testing.T) {
	s := stProxySrv(1)
	s.asyncPending[4] = asyncHTTPEntry{connFD: 9}
	s.proxyPool.Busy[0] = true
	s.proxyBusyTime[0] = time.Now().UnixNano()

	s.proxyOut[0] <- wire.ProxyResponse{ReqID: 4, Status: 200, Body: []byte("b")}
	s.pollProxyWorkers()

	if s.proxyPool.Busy[0] {
		t.Fatal("a drained proxy worker must be idle")
		return
	}
	if s.proxyBusyTime[0] != 0 {
		t.Fatal("a drained proxy worker must clear its busy timestamp")
		return
	}
	if len(s.asyncPending) != 0 {
		t.Fatal("the response must clear asyncPending")
		return
	}
}

// --- blossom responses ---

func TestCompleteBlossomResponseBranches(t *testing.T) {
	s := stBlossomSrv(1)

	s.asyncPending[1] = asyncHTTPEntry{connFD: 9}
	var ctResp wire.BlossomResponse
	ctResp.ReqID = 1
	ctResp.Status = 200
	ctResp.CT = []byte("image/png")
	ctResp.Body = []byte("blob")
	s.completeBlossomResponse(ctResp)
	if len(s.asyncPending) != 0 {
		t.Fatal("content-type branch must clear asyncPending")
		return
	}

	s.asyncPending[2] = asyncHTTPEntry{connFD: 9}
	var sizeResp wire.BlossomResponse
	sizeResp.ReqID = 2
	sizeResp.Status = 200
	sizeResp.Size = 12345
	sizeResp.Body = []byte("blob")
	s.completeBlossomResponse(sizeResp)
	if len(s.asyncPending) != 0 {
		t.Fatal("size branch must clear asyncPending")
		return
	}

	var stale wire.BlossomResponse
	stale.ReqID = 77
	stale.Status = 200
	s.completeBlossomResponse(stale)
	if len(s.asyncPending) != 0 {
		t.Fatal("an unknown blossom response must be a no-op")
		return
	}
}

func TestPollBlossomWorkersDrains(t *testing.T) {
	s := stBlossomSrv(1)
	s.asyncPending[4] = asyncHTTPEntry{connFD: 9}
	s.blossomPool.Busy[0] = true

	s.blossomOut[0] <- wire.BlossomResponse{ReqID: 4, Status: 200, Body: []byte("b")}
	s.pollBlossomWorkers()

	if s.blossomPool.Busy[0] {
		t.Fatal("a drained blossom worker must be idle")
		return
	}
	if len(s.asyncPending) != 0 {
		t.Fatal("the response must clear asyncPending")
		return
	}
}

// --- database dispatch ---

func TestDbSend(t *testing.T) {
	s := stServer(stCfg())
	if s.dbSend(tree.Request{Op: tree.OpPersist, ConnID: 1}) {
		t.Fatal("dbSend without a database must fail")
		return
	}

	s.db = stFakeDB()
	if !s.dbSend(tree.Request{Op: tree.OpPersist, ConnID: 5}) {
		t.Fatal("dbSend with a live database must succeed")
		return
	}
	if len(s.dbPending) != 1 || s.nextDBReq != 1 {
		t.Fatalf("dbPending=%d nextDBReq=%d", len(s.dbPending), s.nextDBReq)
		return
	}
	req, ok := stTryDBReq(s.db.in)
	if !ok || req.ReqID != 1 || req.ConnID != 5 {
		t.Fatalf("queued db request id=%d conn=%d ok=%v", req.ReqID, req.ConnID, ok)
		return
	}

	closed := stFakeDB()
	close(closed.done)
	s.db = closed
	if s.dbSend(tree.Request{}) {
		t.Fatal("dbSend to a stopped database must fail")
		return
	}
}

func TestPollDBDrains(t *testing.T) {
	s := stServer(stCfg())
	s.db = stFakeDB()
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	s.dbPending[1] = fd

	var resp tree.Response
	resp.Op = tree.OpPersist
	resp.ReqID = 1
	resp.OK = true
	resp.EventID = []byte{:32}
	resp.Reason = []byte("saved")
	s.db.out <- resp
	s.db.ready <- struct{}{}
	s.pollDB()

	if _, ok := s.dbPending[1]; ok {
		t.Fatal("the drained reply must clear dbPending")
		return
	}
}

func TestCompleteDBHistoryAndCount(t *testing.T) {
	s := stServer(stCfg())
	fd := int32(4)
	cs := &cstate{subs: map[string]*sub{}}
	cs.subs["s1"] = &sub{id: "s1"}
	s.conns[fd] = cs

	var frames [][]byte
	frames = push(frames, []byte("[\"EVENT\",{}]"))
	frames = push(frames, []byte("[\"EVENT\",{}]"))

	var h tree.Response
	h.Op = tree.OpHistory
	h.ConnID = fd
	h.SubID = []byte("s1")
	h.Events = frames
	s.completeDB(h)

	var unknownSub tree.Response
	unknownSub.Op = tree.OpHistory
	unknownSub.ConnID = fd
	unknownSub.SubID = []byte("gone")
	unknownSub.Events = frames
	s.completeDB(unknownSub)

	var goneConn tree.Response
	goneConn.Op = tree.OpHistory
	goneConn.ConnID = 99
	goneConn.SubID = []byte("s1")
	s.completeDB(goneConn)

	var count tree.Response
	count.Op = tree.OpCount
	count.ConnID = fd
	count.SubID = []byte("s1")
	count.Count = 5
	s.completeDB(count)

	var countGone tree.Response
	countGone.Op = tree.OpCount
	countGone.ConnID = 99
	countGone.SubID = []byte("s1")
	s.completeDB(countGone)

	var noop tree.Response
	noop.Op = 200
	s.completeDB(noop)

	if _, still := s.conns[fd].subs["s1"]; !still {
		t.Fatal("a history reply must not remove a live subscription")
		return
	}
}

func TestCompletePersistMissingAndGone(t *testing.T) {
	s := stServer(stCfg())
	s.conns[4] = &cstate{subs: map[string]*sub{}}

	var missing tree.Response
	missing.Op = tree.OpPersist
	missing.ReqID = 88
	s.completeDB(missing)

	s.dbPending[5] = 99
	var gone tree.Response
	gone.Op = tree.OpPersist
	gone.ReqID = 5
	gone.OK = true
	s.completeDB(gone)
	if _, ok := s.dbPending[5]; ok {
		t.Fatal("a persist reply for a closed connection must be dropped")
		return
	}
}

// --- HTTP routing ---

func stFallback(method, path string, headers map[string]string, body []byte) (int32, map[string]string, []byte) {
	return 201, map[string]string{"X-Fallback": "1"}, []byte("fb")
}

func TestRouteHTTP(t *testing.T) {
	s := stServer(stCfg())
	s.Version = "9"

	status, h, body := s.routeHTTP(1, "GET", "/health", map[string]string{}, nil)
	if status != 200 || string(body) != "ok" {
		t.Fatalf("health = %d %s", status, body)
		return
	}
	if h["Content-Type"] != "text/plain" {
		t.Fatalf("health content-type = %s", h["Content-Type"])
		return
	}

	ms, _, _ := s.routeHTTP(1, "GET", "/metrics", map[string]string{}, nil)
	if ms != 200 {
		t.Fatalf("metrics status = %d", ms)
		return
	}

	rs, _, rb := s.routeHTTP(1, "GET", "/metrics/reset", map[string]string{}, nil)
	if rs != 200 || string(rb) != "reset\n" {
		t.Fatalf("metrics reset = %d %s", rs, rb)
		return
	}

	ns, nh, nb := s.routeHTTP(1, "GET", "/anything", map[string]string{"accept": "application/nostr+json"}, nil)
	if ns != 200 || nh["Content-Type"] != "application/nostr+json" {
		t.Fatalf("nip11 = %d %s", ns, nh["Content-Type"])
		return
	}
	if !stHas(nb, []byte("\"software\"")) {
		t.Fatal("nip11 body must carry the software field")
		return
	}

	fs, fh, fb := s.routeHTTP(1, "GET", "/missing", map[string]string{}, nil)
	if fs != 404 || string(fb) != "404 page not found\n" {
		t.Fatalf("fallback-less 404 = %d %s", fs, fb)
		return
	}
	_ = fh

	s.Fallback = stFallback
	bs, bh, bb := s.routeHTTP(1, "GET", "/missing", map[string]string{}, nil)
	if bs != 201 || string(bb) != "fb" || bh["X-Fallback"] != "1" {
		t.Fatalf("fallback = %d %s %s", bs, bb, bh["X-Fallback"])
		return
	}

	// CORS is merged into a fallback response when enabled.
	s.cfg.CORSEnabled = true
	bs2, bh2, _ := s.routeHTTP(1, "GET", "/missing", map[string]string{"origin": "https://a"}, nil)
	if bs2 != 201 {
		t.Fatalf("fallback with CORS = %d", bs2)
		return
	}
	if bh2["Access-Control-Allow-Origin"] != "https://a" {
		t.Fatal("fallback CORS headers must be merged")
		return
	}
}

func TestOnHTTP(t *testing.T) {
	c := stCfg()
	c.CORSEnabled = true
	s := stServer(c)
	s.Version = "9"

	status, _, body, connClose := s.OnHTTP(1, "GET", "/health", map[string]string{}, nil)
	if status != 200 || string(body) != "ok" || connClose {
		t.Fatalf("plain health = %d %s close=%v", status, body, connClose)
		return
	}

	os_, oh, _, _ := s.OnHTTP(1, "OPTIONS", "/x", map[string]string{"origin": "https://a"}, nil)
	if os_ != 204 {
		t.Fatalf("OPTIONS = %d, want 204", os_)
		return
	}
	if oh["Access-Control-Allow-Origin"] != "https://a" {
		t.Fatal("OPTIONS must answer with CORS headers")
		return
	}

	_, _, _, cc := s.OnHTTP(1, "GET", "/health", map[string]string{"connection": "close"}, nil)
	if !cc {
		t.Fatal("connection: close must be reported")
		return
	}

	// Blacklist re-check. The transport has no connection for fd 1, so ConnIP
	// returns "", and an empty prefix matches it. That exercises the 403 branch.
	c.IPBlacklist = []string{""}
	bs, _, _, _ := s.OnHTTP(1, "GET", "/health", map[string]string{}, nil)
	if bs != 403 {
		t.Fatalf("blacklisted request = %d, want 403", bs)
		return
	}
}

func TestOnHTTPProxyAndBlossomDispatch(t *testing.T) {
	ps := stProxySrv(1)
	ps.selfHost = "self.example/"
	status, _, _, _ := ps.OnHTTP(1, "GET", "/proxy/https://other.example/x", map[string]string{}, nil)
	if status != transport.HTTPDeferred {
		t.Fatalf("proxy dispatch status = %d, want deferred", status)
		return
	}
	if len(ps.asyncPending) != 1 || !ps.proxyPool.Busy[0] {
		t.Fatal("proxy dispatch must register async state")
		return
	}

	bs := stBlossomSrv(1)
	bstatus, _, _, _ := bs.OnHTTP(1, "GET", "/blossom/abc", map[string]string{}, nil)
	if bstatus != transport.HTTPDeferred {
		t.Fatalf("blossom dispatch status = %d, want deferred", bstatus)
		return
	}
	if len(bs.asyncPending) != 1 || !bs.blossomPool.Busy[0] {
		t.Fatal("blossom dispatch must register async state")
		return
	}
}

func TestDoDispatchBlossomBusy(t *testing.T) {
	s := stBlossomSrv(1)
	s.blossomPool.Busy[0] = true
	if !s.doDispatchBlossom(1, "GET", "/blossom/abc", map[string]string{}, nil) {
		t.Fatal("a busy blossom pool still reports handled")
		return
	}
	if len(s.asyncPending) != 0 {
		t.Fatal("a busy blossom pool must not register async state")
		return
	}
}

// --- event ingestion decisions ---

func TestHandleEventRateLimited(t *testing.T) {
	c := stCfg()
	c.RelayURL = "wss://relay.example"
	s := stServer(c)
	s.db = stFakeDB()
	s.writeLimiter = ratelimit.New(1.0, 1)
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}

	ev := stSignedEvent(t, 1, nil, "x")
	msg := stEventMsg(ev)
	s.handleEvent(fd, msg)
	if len(s.dbPending) != 1 {
		t.Fatalf("the first unauthed write must pass, dbPending = %d", len(s.dbPending))
		return
	}
	s.handleEvent(fd, msg)
	if len(s.dbPending) != 1 {
		t.Fatal("the rate-limited write must not reach the database")
		return
	}
}

func TestHandleEventAuthToWrite(t *testing.T) {
	c := stCfg()
	c.RelayURL = "wss://relay.example"
	c.AuthToWrite = true
	c.FreeWriteLimit = 0 // no limiter; AuthToWrite is the only gate
	s := stServer(c)
	s.db = stFakeDB()
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}

	ev := stSignedEvent(t, 1, nil, "x")
	s.handleEvent(fd, stEventMsg(ev))
	if len(s.dbPending) != 0 {
		t.Fatal("auth-required must reject an unauthed write")
		return
	}
}

func TestHandleEventWriteExemptMarmot(t *testing.T) {
	// access.IsMLS reads kind package pointer globals, which stay nil until
	// kind.Ensure runs. handleEvent never calls Ensure, and kind state is
	// per-domain, so a gated relay with MarmotOpen (or NIP46BypassAuth) true
	// segfaults the parent domain on the first unauthed write. Reported;
	// Ensure here so the exemption decision itself is what gets tested.
	kind.Ensure()
	c := stCfg()
	c.RelayURL = "wss://relay.example"
	c.AuthToWrite = true
	c.MarmotOpen = true
	s := stServer(c)
	s.db = stFakeDB()
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}

	ev := stSignedEvent(t, 443, nil, "mls")
	s.handleEvent(fd, stEventMsg(ev))
	if len(s.dbPending) != 1 {
		t.Fatalf("an exempt MLS kind must bypass auth, dbPending = %d", len(s.dbPending))
		return
	}
}

func TestHandleEventWriteExemptNIP46(t *testing.T) {
	kind.Ensure()
	c := stCfg()
	c.RelayURL = "wss://relay.example"
	c.AuthToWrite = true
	c.NIP46BypassAuth = true
	s := stServer(c)
	s.db = stFakeDB()
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}

	ev := stSignedEvent(t, 24133, nil, "connect")
	s.handleEvent(fd, stEventMsg(ev))
	if len(s.dbPending) != 1 {
		t.Fatalf("an exempt NIP-46 connect must bypass auth, dbPending = %d", len(s.dbPending))
		return
	}
}

func TestHandleEventDispatchesToWorker(t *testing.T) {
	s := stIngestSrv(1)
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	ev := stSignedEvent(t, 1, nil, "x")
	s.handleEvent(fd, stEventMsg(ev))
	if len(s.pendingReq) != 1 || !s.workers.Busy[0] {
		t.Fatal("a live ingest pool must take the event")
		return
	}
	if len(s.dbPending) != 0 {
		t.Fatal("the worker path must not talk to the database directly")
		return
	}
}

func TestHandleEventMalformedAndMissingConn(t *testing.T) {
	s := stServer(stCfg())
	s.db = stFakeDB()
	s.handleEvent(99, []byte("[\"EVENT\",{}]"))
	if len(s.dbPending) != 0 {
		t.Fatal("a missing connection must be a no-op")
		return
	}
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	s.handleEvent(fd, []byte("[\"EVENT\",]"))
	if len(s.dbPending) != 0 {
		t.Fatal("a malformed EVENT must not reach the database")
		return
	}
}

// --- dispatch routing ---

func TestDispatchEventPersists(t *testing.T) {
	s := stServer(stCfg())
	s.db = stFakeDB()
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	ev := stSignedEvent(t, 1, nil, "hello")
	s.dispatch(fd, stEventMsg(ev))
	if len(s.dbPending) != 1 {
		t.Fatalf("dispatch must persist an EVENT, dbPending = %d", len(s.dbPending))
		return
	}
	req, ok := stTryDBReq(s.db.in)
	if !ok || req.Op != tree.OpPersist || req.ConnID != fd {
		t.Fatalf("persist request op=%d conn=%d ok=%v", req.Op, req.ConnID, ok)
		return
	}
}

func TestDispatchJailedAndUnknown(t *testing.T) {
	s := stServer(stCfg())
	s.db = stFakeDB()
	fd := int32(4)
	cs := &cstate{subs: map[string]*sub{}, jailed: true}
	s.conns[fd] = cs
	ev := stSignedEvent(t, 1, nil, "hello")
	s.dispatch(fd, stEventMsg(ev))
	if len(s.dbPending) != 0 {
		t.Fatal("a jailed connection must be silent")
		return
	}
	s.dispatch(99, stEventMsg(ev))
	if len(s.dbPending) != 0 {
		t.Fatal("an unknown connection must be a no-op")
		return
	}
	// An unrecognised label is dropped by dispatch.
	s.conns[fd].jailed = false
	s.dispatch(fd, []byte("[\"NOTICE\",\"hi\"]"))
	if len(s.dbPending) != 0 {
		t.Fatal("an unknown label must be dropped")
		return
	}
}

func TestHandleCountAndTaint(t *testing.T) {
	s := stServer(stCfg())
	s.db = stFakeDB()
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	s.handleCount(fd, []byte("[\"COUNT\",\"c1\",{}]"))
	if len(s.dbPending) != 1 {
		t.Fatalf("handleCount dbPending = %d, want 1", len(s.dbPending))
		return
	}
	req, ok := stTryDBReq(s.db.in)
	if !ok || req.Op != tree.OpCount || req.ConnID != fd {
		t.Fatalf("count request op=%d conn=%d ok=%v", req.Op, req.ConnID, ok)
		return
	}

	// A malformed COUNT is dropped.
	s.handleCount(fd, []byte("[\"COUNT\",\"c1\"]"))

	// An undersized hex id taints the filter and jails the connection.
	s.handleCount(fd, []byte("[\"COUNT\",\"c2\",{\"ids\":[\"abcd\"]}]"))
	if !s.conns[fd].jailed {
		t.Fatal("an undersized hex id must jail the connection")
		return
	}

	// Missing connection.
	s.handleCount(99, []byte("[\"COUNT\",\"c1\",{}]"))

	// Auth required and unauthed.
	c2 := stCfg()
	c2.AuthRequired = true
	s2 := stServer(c2)
	s2.db = stFakeDB()
	s2.conns[4] = &cstate{subs: map[string]*sub{}}
	s2.handleCount(4, []byte("[\"COUNT\",\"c1\",{}]"))
	if len(s2.dbPending) != 0 {
		t.Fatal("auth-required COUNT must be silent when unauthed")
		return
	}
}

func TestHandleEventRejectReason(t *testing.T) {
	s := stServer(stCfg())
	fd := int32(4)
	s.conns[fd] = &cstate{subs: map[string]*sub{}}
	ev := stSignedEvent(t, 1, nil, "x")
	// Directly drive eventReject; with no transport connection the send is a
	// no-op, so this only proves it parses and does not panic.
	s.eventReject(fd, stEventMsg(ev), "rate-limited: too many events")
}

func TestOnTickReaps(t *testing.T) {
	s := stServer(stCfg())
	now := time.Now().UnixNano()
	s.asyncPending[1] = asyncHTTPEntry{connFD: 3, createdAt: now - 120_000_000_000}
	s.OnTick()
	if _, ok := s.asyncPending[1]; ok {
		t.Fatal("OnTick must run the async reaper")
		return
	}
}