ha-n95-local-control/internal/xmpp/xmpp_test.go

package xmpp

import (
	"bytes"
	"context"
	"encoding/base64"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"os"
	"strings"
	"sync"
	"testing"
	"time"

	"git.i3omb.com/gronod/ha-n95-local-control/internal/config"
	"git.i3omb.com/gronod/ha-n95-local-control/internal/session"
)

func init() {
	// Keep tests quiet unless they explicitly fail.
	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})))
}

func testConfig() config.Config {
	cfg, err := config.Load([]string{
		"ADVERTISE_IP=192.0.2.10",
		"MQTT_HOST=mqtt.example.invalid",
		"CONTROLLER_JID=n95bridge@ecouser.net/homeassistant",
	})
	if err != nil {
		panic(err)
	}
	return cfg
}

func TestTokenizerSeveralStanzasOneSegment(t *testing.T) {
	tok := NewTokenizer()
	// Feed the stream open first so the tokenizer is in-stream.
	data := `<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>` +
		`<iq from='serial@155.ecorobot.net/atom' to='155.ecorobot.net' id='1234567890' type='get'><ping xmlns='urn:xmpp:ping'/></iq>` +
		`<iq type='set' id='a'><bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'><resource>atom</resource></bind></iq>` +
		`<presence><status>hello world</status></presence>`
	evs := tok.Feed([]byte(data))

	var opens, stanzas int
	for _, ev := range evs {
		switch ev.Kind {
		case StreamOpen:
			opens++
		case Stanza:
			stanzas++
		case TokenError:
			t.Fatalf("tokenizer error: %v, data so far %q", tok.Err(), ev.Data)
		}
	}
	if opens != 1 {
		t.Fatalf("opens = %d, want 1", opens)
	}
	if stanzas != 3 {
		t.Fatalf("stanzas = %d, want 3", stanzas)
	}
	if tok.Err() != nil {
		t.Fatalf("tokenizer error after feed: %v", tok.Err())
	}
	// Remaining buffer should be empty after the three complete stanzas.
	if len(tok.buf) != 0 {
		t.Fatalf("remaining buffer = %q, want empty", tok.buf)
	}
}

func TestTokenizerStanzaSpansSegments(t *testing.T) {
	tok := NewTokenizer()
	tok.Feed([]byte(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`))
	part1 := `<iq from='serial@155.ecorobot.net/atom' to='155.ecorobot.net' id='1000000000' type='get'><ping xmlns='urn:xmpp:ping'/></iq><iq from='serial@155.ecorobot.net/atom' to='155.ecorobot.net' id='1000000001' type='get'`
	part2 := `><ping xmlns='urn:xmpp:ping'/></iq>`
	evs1 := tok.Feed([]byte(part1))
	if len(evs1) != 1 {
		t.Fatalf("first feed events = %d, want 1: %+v", len(evs1), evs1)
	}
	evs2 := tok.Feed([]byte(part2))
	if len(evs2) != 1 {
		t.Fatalf("second feed events = %d, want 1: %+v", len(evs2), evs2)
	}
	if evs2[0].Kind != Stanza {
		t.Fatalf("second event kind = %v, want Stanza", evs2[0].Kind)
	}
}

func TestStreamIDRepeatedAfterSASL(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	registry := session.NewRegistry()
	bus := session.NewBus()
	rec := &recorder{}
	bus.Register(rec)

	cfg := testConfig()
	clock := newFakeClock()
	c := &Conn{
		netConn:  server,
		cfg:      cfg,
		registry: registry,
		bus:      bus,
		clock:    clock,
	}
	go c.serve()

	xc := &xmppClient{t: t, c: client}
	xc.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	open1 := xc.recvUntil("</stream:features>")
	id1 := extractAttr(open1, "id")
	if id1 == "" {
		t.Fatalf("first stream id missing: %q", open1)
	}

	xc.send(saslPlain("serial1", "secret"))
	xc.recvUntil("<success")

	xc.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	open2 := xc.recvUntil("</stream:features>")
	id2 := extractAttr(open2, "id")
	if id1 != id2 {
		t.Fatalf("stream id changed: %q -> %q", id1, id2)
	}

	// A second TCP connection must receive a different id.
	client2, server2 := net.Pipe()
	defer client2.Close()
	defer server2.Close()
	c2 := &Conn{
		netConn:  server2,
		cfg:      cfg,
		registry: session.NewRegistry(),
		bus:      session.NewBus(),
		clock:    newFakeClock(),
	}
	go c2.serve()
	xc2 := &xmppClient{t: t, c: client2}
	xc2.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	open3 := xc2.recvUntil("</stream:features>")
	id3 := extractAttr(open3, "id")
	if id3 == "" || id3 == id1 {
		t.Fatalf("second connection stream id not unique: %q", id3)
	}
}

func TestStartTLSAdvertisedNotWaited(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	c := &Conn{
		netConn:  server,
		cfg:      testConfig(),
		registry: session.NewRegistry(),
		bus:      session.NewBus(),
		clock:    newFakeClock(),
	}
	go c.serve()

	xc := &xmppClient{t: t, c: client}
	xc.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	open1 := xc.recvUntil("</stream:features>")
	if !strings.Contains(open1, `<starttls xmlns="urn:ietf:params:xml:ns:xmpp-tls"><required/></starttls>`) {
		t.Fatalf("missing starttls required: %q", open1)
	}

	// Skip STARTTLS and send SASL directly; server answers success.
	xc.send(saslPlain("serial2", "secret"))
	xc.recvUntil("<success")
}

func TestHandshakeToReady(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	registry := session.NewRegistry()
	bus := session.NewBus()
	rec := &recorder{}
	bus.Register(rec)

	cfg := testConfig()
	c := &Conn{
		netConn:  server,
		cfg:      cfg,
		registry: registry,
		bus:      bus,
		clock:    newFakeClock(),
	}
	go c.serve()

	jid := completeHandshake(t, client, "serial3", "atom")
	wantJID := "serial3@155.ecorobot.net/atom"
	if jid != wantJID {
		t.Fatalf("bind JID = %q, want %q", jid, wantJID)
	}

	mustWaitFor(t, func() bool {
		rec.mu.Lock()
		defer rec.mu.Unlock()
		return len(rec.ready) == 1
	})
	rec.mu.Lock()
	readyCount := len(rec.ready)
	rec.mu.Unlock()
	if readyCount != 1 {
		t.Fatalf("SessionReady count = %d, want 1", readyCount)
	}

	currentGen, ok := registry.Current(wantJID)
	if !ok || currentGen != 1 {
		t.Fatalf("Current(%s) = %d %v, want 1 true", wantJID, currentGen, ok)
	}
}

func TestDummyPresenceSpaces(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	c := &Conn{
		netConn:  server,
		cfg:      testConfig(),
		registry: session.NewRegistry(),
		bus:      session.NewBus(),
		clock:    newFakeClock(),
	}
	go c.serve()

	jid, dummy := completeHandshakeEx(t, client, "serial4", "atom")
	wantTo := fmt.Sprintf(`to="%s"`, jid)
	if !strings.Contains(dummy, `<presence`) || !strings.Contains(dummy, wantTo) || !strings.Contains(dummy, "> dummy </presence>") {
		t.Fatalf("dummy presence = %q, want to=%q with > dummy </presence>", dummy, jid)
	}
}

func TestSASLAcceptedAndNotLogged(t *testing.T) {
	// Capture all log records.
	var logBuf bytes.Buffer
	h := &captureHandler{Level: slog.LevelDebug}
	slog.SetDefault(slog.New(h))
	defer slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})))

	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	c := &Conn{
		netConn:  server,
		cfg:      testConfig(),
		registry: session.NewRegistry(),
		bus:      session.NewBus(),
		clock:    newFakeClock(),
	}
	go c.serve()

	xc := &xmppClient{t: t, c: client}
	xc.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	xc.recvUntil("</stream:features>")

	// Valid PLAIN credential.
	payload := "\x00serial5\x00any-password"
	token := base64.StdEncoding.EncodeToString([]byte(payload))
	xc.send(fmt.Sprintf(`<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>%s</auth>`, token))
	xc.recvUntil("<success")
	_ = logBuf.String()

	h.mu.Lock()
	records := h.records
	h.mu.Unlock()
	joined := strings.Join(records, "\n")
	if strings.Contains(joined, token) {
		t.Fatalf("log contained base64 token")
	}
	if strings.Contains(joined, "any-password") {
		t.Fatalf("log contained password")
	}
	if !strings.Contains(joined, "serial5") {
		t.Fatalf("log did not contain authcid: %q", joined)
	}

	// Malformed credential.
	client2, server2 := net.Pipe()
	defer client2.Close()
	defer server2.Close()
	h2 := &captureHandler{Level: slog.LevelDebug}
	slog.SetDefault(slog.New(h2))
	c2 := &Conn{
		netConn:  server2,
		cfg:      testConfig(),
		registry: session.NewRegistry(),
		bus:      session.NewBus(),
		clock:    newFakeClock(),
	}
	go c2.serve()
	xc2 := &xmppClient{t: t, c: client2}
	xc2.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	xc2.recvUntil("</stream:features>")
	xc2.send(`<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>!!!</auth>`)
	fail := xc2.recvUntil("</failure>")
	if !strings.Contains(fail, "malformed-request") {
		t.Fatalf("expected malformed-request failure: %q", fail)
	}
	h2.mu.Lock()
	records2 := h2.records
	h2.mu.Unlock()
	joined2 := strings.Join(records2, "\n")
	if strings.Contains(joined2, "!!!") {
		t.Fatalf("malformed log contained raw payload")
	}
	if !strings.Contains(joined2, "sasl malformed") {
		t.Fatalf("malformed log did not contain 'sasl malformed': %q", joined2)
	}
}

func TestSessionReplaceAtBind(t *testing.T) {
	registry := session.NewRegistry()
	bus := session.NewBus()
	rec := &recorder{}
	bus.Register(rec)
	cfg := testConfig()
	clock := newFakeClock()

	// Connection A completes handshake through session, not yet READY.
	clientA, serverA := net.Pipe()
	defer clientA.Close()
	cA := &Conn{
		netConn:  serverA,
		cfg:      cfg,
		registry: registry,
		bus:      bus,
		clock:    clock,
	}
	go cA.serve()
	jid := completeHandshakeNoReady(t, clientA, "serial6", "atom")
	// Drain A so the replacement close-with-write does not block on a pipe.
	go func() { _, _ = io.Copy(io.Discard, clientA) }()

	// Connection B starts, finishes SASL, then sends bind for the same JID.
	clientB, serverB := net.Pipe()
	defer clientB.Close()
	cB := &Conn{
		netConn:  serverB,
		cfg:      cfg,
		registry: registry,
		bus:      bus,
		clock:    clock,
	}
	go cB.serve()
	xcB := &xmppClient{t: t, c: clientB}
	xcB.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	xcB.recvUntil("</stream:features>")
	xcB.send(saslPlain("serial6", "pw"))
	xcB.recvUntil("<success")
	xcB.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	xcB.recvUntil("</stream:features>")
	xcB.send(`<iq type='set' id='bindB'><bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'><resource>atom</resource></bind></iq>`)
	bindResult := xcB.recvUntil("</iq>")
	if !strings.Contains(bindResult, jid) {
		t.Fatalf("B bind result missing JID: %q", bindResult)
	}

	// A's later bytes should not be delivered; send a presence on A.
	clientA.Write([]byte(`<presence><status>hello world</status></presence>`))

	mustWaitFor(t, func() bool {
		rec.mu.Lock()
		defer rec.mu.Unlock()
		return len(rec.down) == 1
	})

	rec.mu.Lock()
	replacedReasons := []string{}
	for _, d := range rec.down {
		replacedReasons = append(replacedReasons, d.Reason)
	}
	readyCount := len(rec.ready)
	rec.mu.Unlock()
	if len(replacedReasons) != 1 || replacedReasons[0] != session.ReasonReplaced {
		t.Fatalf("replacement down events = %v, want one replaced", replacedReasons)
	}
	if readyCount != 0 {
		t.Fatalf("SessionReady emitted from replaced A: %d", readyCount)
	}

	// Send on A's generation returns ErrStale.
	if err := cA.registry.Send(jid, 1, []byte("x")); !errors.Is(err, session.ErrStale) {
		t.Fatalf("Send on gen 1 error = %v, want ErrStale", err)
	}
}

func TestBotPingMatchingID(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	c := &Conn{
		netConn:  server,
		cfg:      testConfig(),
		registry: session.NewRegistry(),
		bus:      session.NewBus(),
		clock:    newFakeClock(),
	}
	go c.serve()

	completeHandshake(t, client, "serial7", "atom")
	_ = xcRecvPing(t, client) // consume announce ping

	xc := &xmppClient{t: t, c: client}
	bigID := "12345678901234567890"
	xc.send(fmt.Sprintf(`<iq from='serial7@155.ecorobot.net/atom' to='155.ecorobot.net' id='%s' type='get'><ping xmlns='urn:xmpp:ping'/></iq>`, bigID))
	reply := xc.recvUntil("/>")
	if !strings.Contains(reply, fmt.Sprintf(`id="%s"`, bigID)) {
		t.Fatalf("reply missing id: %q", reply)
	}
	if !strings.Contains(reply, `type="result"`) {
		t.Fatalf("reply not type=result: %q", reply)
	}
	if !strings.Contains(reply, `from="155.ecorobot.net"`) {
		t.Fatalf("reply from wrong: %q", reply)
	}
	if !strings.Contains(reply, fmt.Sprintf(`to="serial7@155.ecorobot.net/atom"`)) {
		t.Fatalf("reply to wrong: %q", reply)
	}
}

func TestAnnouncePingAfterReady(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	registry := session.NewRegistry()
	bus := session.NewBus()
	rec := &recorder{}
	bus.Register(rec)
	cfg := testConfig()
	clock := newFakeClock()

	c := &Conn{
		netConn:  server,
		cfg:      cfg,
		registry: registry,
		bus:      bus,
		clock:    clock,
	}
	go c.serve()

	completeHandshake(t, client, "serial8", "atom")
	// The announce ping is sent immediately after READY.
	ping := xcRecvPing(t, client)
	waitForGoroutines() // let pingLoop reach its select
	if ping == "" {
		t.Fatalf("no announce ping received")
	}
	if !strings.Contains(ping, cfg.ControllerJID) {
		t.Fatalf("ping from not controller JID: %q", ping)
	}

	// Reply to the announce ping.
	id := extractAttr(ping, "id")
	xc := &xmppClient{t: t, c: client}
	xc.send(fmt.Sprintf(`<iq type='result' from='serial8@155.ecorobot.net/atom' to='%s' id='%s'/>`, cfg.ControllerJID, id))

	// Wait for AnnounceOK.
	mustWaitFor(t, func() bool {
		rec.mu.Lock()
		defer rec.mu.Unlock()
		return len(rec.announce) == 1
	})
	rec.mu.Lock()
	gen := rec.announce[0].Generation
	rec.mu.Unlock()
	if gen != 1 {
		t.Fatalf("AnnounceOK generation = %d, want 1", gen)
	}
}

func TestControllerPingDeadlineSignalsDown(t *testing.T) {
	client, server := net.Pipe()
	defer client.Close()
	defer server.Close()

	registry := session.NewRegistry()
	bus := session.NewBus()
	rec := &recorder{}
	bus.Register(rec)
	cfg := testConfig()
	clock := newFakeClock()

	c := &Conn{
		netConn:  server,
		cfg:      cfg,
		registry: registry,
		bus:      bus,
		clock:    clock,
	}
	go c.serve()

	completeHandshake(t, client, "serial9", "atom")
	ping := xcRecvPing(t, client)
	waitForGoroutines() // let pingLoop reach its select
	if ping == "" {
		t.Fatalf("no announce ping received")
	}

	// Advance 12 seconds with no reply.
	clock.Advance(PingResultTimeout)

	mustWaitFor(t, func() bool {
		rec.mu.Lock()
		defer rec.mu.Unlock()
		return len(rec.down) == 1
	})
	rec.mu.Lock()
	reason := rec.down[0].Reason
	rec.mu.Unlock()
	if reason != session.ReasonPingTimeout {
		t.Fatalf("down reason = %q, want ping-timeout", reason)
	}
}

// completeHandshake performs the full client handshake up to READY and returns
// the bound full JID.
func completeHandshake(t *testing.T, client net.Conn, serial, resource string) string {
	jid, _ := completeHandshakeEx(t, client, serial, resource)
	return jid
}

// completeHandshakeEx returns the bound JID and the raw dummy presence response.
func completeHandshakeEx(t *testing.T, client net.Conn, serial, resource string) (jid, dummy string) {
	jid = completeHandshakeNoReady(t, client, serial, resource)
	xc := &xmppClient{t: t, c: client}
	xc.send(`<presence><status>hello world</status></presence>`)
	dummy = xc.recvUntil("</presence>")
	if !strings.Contains(dummy, "> dummy </presence>") {
		t.Fatalf("dummy presence missing: %q", dummy)
	}
	return jid, dummy
}

func completeHandshakeNoReady(t *testing.T, client net.Conn, serial, resource string) string {
	xc := &xmppClient{t: t, c: client}
	xc.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	xc.recvUntil("</stream:features>")
	xc.send(saslPlain(serial, "password"))
	xc.recvUntil("<success")
	xc.send(`<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' xmlns='jabber:client' to='155.ecorobot.net' version='1.0'>`)
	xc.recvUntil("</stream:features>")
	xc.send(fmt.Sprintf(`<iq type='set' id='bind1'><bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'><resource>%s</resource></bind></iq>`, resource))
	bindResult := xc.recvUntil("</iq>")
	jid := extractJID(bindResult)
	if jid == "" {
		t.Fatalf("no JID in bind result: %q", bindResult)
	}
	xc.send(`<iq type='set' id='sess1'><session xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>`)
	xc.recvUntil("/>")
	return jid
}

func saslPlain(serial, password string) string {
	payload := fmt.Sprintf("\x00%s\x00%s", serial, password)
	token := base64.StdEncoding.EncodeToString([]byte(payload))
	return fmt.Sprintf(`<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>%s</auth>`, token)
}

func extractJID(s string) string {
	start := strings.Index(s, "<jid>")
	if start < 0 {
		return ""
	}
	start += len("<jid>")
	end := strings.Index(s[start:], "</jid>")
	if end < 0 {
		return ""
	}
	return s[start : start+end]
}

func extractAttr(s, name string) string {
	prefix := name + "=\""
	start := strings.Index(s, prefix)
	if start < 0 {
		return ""
	}
	start += len(prefix)
	end := strings.Index(s[start:], "\"")
	if end < 0 {
		return ""
	}
	return s[start : start+end]
}

func waitForGoroutines() { time.Sleep(20 * time.Millisecond) }

func xcRecvPing(t *testing.T, client net.Conn) string {
	xc := &xmppClient{t: t, c: client}
	return xc.recvUntil("</iq>")
}

func mustReadAll(c net.Conn) string {
	c.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
	defer c.SetReadDeadline(time.Time{})
	var buf bytes.Buffer
	b := make([]byte, 4096)
	for {
		n, err := c.Read(b)
		if n > 0 {
			buf.Write(b[:n])
		}
		if err != nil {
			break
		}
	}
	return buf.String()
}

func mustWaitFor(t *testing.T, f func() bool) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if f() {
			return
		}
		time.Sleep(10 * time.Millisecond)
	}
	t.Fatalf("condition not satisfied")
}

type xmppClient struct {
	t *testing.T
	c net.Conn
}

func (xc *xmppClient) send(s string) {
	xc.t.Helper()
	_, err := xc.c.Write([]byte(s))
	if err != nil {
		xc.t.Fatalf("client write: %v", err)
	}
}

func (xc *xmppClient) recvUntil(marker string) string {
	xc.t.Helper()
	var buf bytes.Buffer
	b := make([]byte, 1024)
	xc.c.SetReadDeadline(time.Now().Add(2 * time.Second))
	defer xc.c.SetReadDeadline(time.Time{})
	for {
		n, err := xc.c.Read(b)
		if n > 0 {
			buf.Write(b[:n])
		}
		if strings.Contains(buf.String(), marker) {
			return buf.String()
		}
		if err != nil {
			if errors.Is(err, os.ErrDeadlineExceeded) {
				xc.t.Fatalf("timeout waiting for %q, got %q", marker, buf.String())
			}
			xc.t.Fatalf("client read: %v", err)
		}
	}
}

type recorder struct {
	mu       sync.Mutex
	ready    []session.ReadyEvent
	announce []session.ReadyEvent
	down     []session.DownEvent
	stanzas  []session.StanzaEvent
}

func (r *recorder) SessionReady(e session.ReadyEvent) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.ready = append(r.ready, e)
}

func (r *recorder) AnnounceOK(e session.ReadyEvent) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.announce = append(r.announce, e)
}

func (r *recorder) SessionDown(e session.DownEvent) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.down = append(r.down, e)
}

func (r *recorder) Stanza(e session.StanzaEvent) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.stanzas = append(r.stanzas, e)
}

type captureHandler struct {
	Level   slog.Leveler
	mu      sync.Mutex
	records []string
}

func (h *captureHandler) Enabled(_ context.Context, level slog.Level) bool {
	return level >= h.Level.Level()
}

func (h *captureHandler) Handle(_ context.Context, r slog.Record) error {
	var b strings.Builder
	b.WriteString(r.Message)
	r.Attrs(func(a slog.Attr) bool {
		b.WriteString(fmt.Sprintf(" %s=%v", a.Key, a.Value))
		return true
	})
	h.mu.Lock()
	h.records = append(h.records, b.String())
	h.mu.Unlock()
	return nil
}

func (h *captureHandler) WithAttrs(attrs []slog.Attr) slog.Handler { return h }
func (h *captureHandler) WithGroup(name string) slog.Handler       { return h }

// fakeClock is a deterministic clock for tests. It starts at a fixed base time.
type fakeClock struct {
	mu     sync.Mutex
	now    time.Time
	timers []*fakeTimer
}

type fakeTimer struct {
	fire  time.Time
	ch    chan time.Time
	fired bool
}

func newFakeClock() *fakeClock {
	return &fakeClock{now: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)}
}

func (f *fakeClock) Now() time.Time {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.now
}

func (f *fakeClock) After(d time.Duration) <-chan time.Time {
	f.mu.Lock()
	defer f.mu.Unlock()
	ch := make(chan time.Time, 1)
	f.timers = append(f.timers, &fakeTimer{fire: f.now.Add(d), ch: ch})
	f.fireDue()
	return ch
}

func (f *fakeClock) Advance(d time.Duration) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.now = f.now.Add(d)
	f.fireDue()
}

func (f *fakeClock) fireDue() {
	for _, t := range f.timers {
		if !t.fired && !f.now.Before(t.fire) {
			t.fired = true
			select {
			case t.ch <- f.now:
			default:
			}
		}
	}
}