synapbus/internal/api/agent_events_test.go

package api

import (
	"bufio"
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/go-chi/chi/v5"
	_ "modernc.org/sqlite"

	"github.com/synapbus/synapbus/internal/agents"
	"github.com/synapbus/synapbus/internal/channels"
	"github.com/synapbus/synapbus/internal/messaging"
)

// agentEventsEnv wires real services over a temporary in-memory SQLite
// database behind the real agent API-key middleware.
type agentEventsEnv struct {
	t        *testing.T
	srv      *httptest.Server
	hub      *SSEHub
	msgSvc   *messaging.MessagingService
	chanSvc  *channels.Service
	keys     map[string]string // agent name -> API key
	channels map[string]int64  // channel name -> id
}

func newAgentEventsEnv(t *testing.T, agentNames ...string) *agentEventsEnv {
	t.Helper()
	db := newTestDBFull(t)

	agentSvc := agents.NewAgentService(agents.NewSQLiteAgentStore(db), nil)
	msgSvc := messaging.NewMessagingService(messaging.NewSQLiteMessageStore(db), nil)
	chanSvc := channels.NewService(channels.NewSQLiteChannelStore(db), msgSvc, nil)

	hub := NewSSEHub()
	hub.SetAgentEventBacklog(msgSvc)
	b := NewSSEBroadcaster(hub, agentSvc, chanSvc)
	b.SetMessageService(msgSvc)
	msgSvc.AddMessageListener(b)

	env := &agentEventsEnv{
		t: t, hub: hub, msgSvc: msgSvc, chanSvc: chanSvc,
		keys: map[string]string{}, channels: map[string]int64{},
	}
	ctx := context.Background()
	for _, name := range agentNames {
		_, key, err := agentSvc.Register(ctx, name, name, "ai", nil, 1)
		if err != nil {
			t.Fatalf("register %s: %v", name, err)
		}
		env.keys[name] = key
	}

	r := chi.NewRouter()
	r.Group(func(r chi.Router) {
		r.Use(agents.RequiredAuthMiddlewareWithOAuth(agentSvc, nil, nil))
		r.Get("/api/agent-events", hub.HandleAgentEvents)
	})
	env.srv = httptest.NewServer(r)
	t.Cleanup(env.srv.Close)
	t.Cleanup(hub.Close) // runs before srv.Close so open streams end
	return env
}

func (e *agentEventsEnv) makeChannel(name string, members ...string) {
	e.t.Helper()
	ctx := context.Background()
	ch, err := e.chanSvc.CreateChannel(ctx, channels.CreateChannelRequest{Name: name, CreatedBy: members[0]})
	if err != nil {
		e.t.Fatalf("create channel: %v", err)
	}
	e.channels[name] = ch.ID
	for _, m := range members {
		if err := e.chanSvc.JoinChannel(ctx, ch.ID, m); err != nil {
			e.t.Fatalf("join %s: %v", m, err)
		}
	}
}

func (e *agentEventsEnv) dm(from, to, body, subject string) int64 {
	e.t.Helper()
	m, err := e.msgSvc.SendMessage(context.Background(), from, to, body, messaging.SendOptions{Subject: subject})
	if err != nil {
		e.t.Fatalf("send dm: %v", err)
	}
	return m.ID
}

func (e *agentEventsEnv) post(from, channel, body string) int64 {
	e.t.Helper()
	id := e.channels[channel]
	m, err := e.msgSvc.SendMessage(context.Background(), from, "", body, messaging.SendOptions{ChannelID: &id})
	if err != nil {
		e.t.Fatalf("post: %v", err)
	}
	return m.ID
}

// sseFrame is one parsed SSE frame.
type sseFrame struct {
	ID    string
	Event string
	Data  string
}

type sseClient struct {
	t      *testing.T
	frames chan sseFrame
	cancel context.CancelFunc
	status int
	body   string
}

// connect opens the stream. For non-200 statuses it records status/body only.
func (e *agentEventsEnv) connect(key, lastEventID string) *sseClient {
	e.t.Helper()
	ctx, cancel := context.WithCancel(context.Background())
	req, _ := http.NewRequestWithContext(ctx, "GET", e.srv.URL+"/api/agent-events", nil)
	if key != "" {
		req.Header.Set("Authorization", "Bearer "+key)
	}
	if lastEventID != "" {
		req.Header.Set("Last-Event-ID", lastEventID)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		cancel()
		e.t.Fatalf("connect: %v", err)
	}
	c := &sseClient{t: e.t, frames: make(chan sseFrame, 512), cancel: cancel, status: resp.StatusCode}
	e.t.Cleanup(cancel)
	if resp.StatusCode != http.StatusOK {
		var sb strings.Builder
		sc := bufio.NewScanner(resp.Body)
		for sc.Scan() {
			sb.WriteString(sc.Text())
		}
		resp.Body.Close()
		c.body = sb.String()
		return c
	}
	go func() {
		defer resp.Body.Close()
		defer close(c.frames)
		sc := bufio.NewScanner(resp.Body)
		var f sseFrame
		for sc.Scan() {
			line := sc.Text()
			switch {
			case line == "":
				if f.Event != "" {
					c.frames <- f
				}
				f = sseFrame{}
			case strings.HasPrefix(line, "id: "):
				f.ID = line[4:]
			case strings.HasPrefix(line, "event: "):
				f.Event = line[7:]
			case strings.HasPrefix(line, "data: "):
				f.Data = line[6:]
			}
		}
	}()
	return c
}

func (c *sseClient) next() sseFrame {
	c.t.Helper()
	select {
	case f, ok := <-c.frames:
		if !ok {
			c.t.Fatal("stream closed while waiting for event")
		}
		return f
	case <-time.After(5 * time.Second):
		c.t.Fatal("timed out waiting for SSE event")
	}
	return sseFrame{}
}

// nextMessage skips nothing: it expects the next frame to be new_message.
func (c *sseClient) nextMessage() (sseFrame, map[string]any) {
	c.t.Helper()
	f := c.next()
	if f.Event != "new_message" {
		c.t.Fatalf("event = %q (data %s), want new_message", f.Event, f.Data)
	}
	var m map[string]any
	if err := json.Unmarshal([]byte(f.Data), &m); err != nil {
		c.t.Fatalf("bad json %q: %v", f.Data, err)
	}
	return f, m
}

func (c *sseClient) expectConnected(agent string) {
	c.t.Helper()
	f := c.next()
	if f.Event != "connected" || !strings.Contains(f.Data, `"agent":"`+agent+`"`) {
		c.t.Fatalf("first frame = %+v, want connected for %s", f, agent)
	}
}

func TestAgentEvents_Auth(t *testing.T) {
	env := newAgentEventsEnv(t, "alice")
	tests := []struct {
		name string
		key  string
	}{
		{"missing key", ""},
		{"invalid key", "not-a-real-key"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := env.connect(tt.key, "")
			if c.status != http.StatusUnauthorized {
				t.Fatalf("status = %d, want 401", c.status)
			}
		})
	}

	t.Run("valid key gets connected event", func(t *testing.T) {
		c := env.connect(env.keys["alice"], "")
		if c.status != http.StatusOK {
			t.Fatalf("status = %d, want 200", c.status)
		}
		c.expectConnected("alice")
	})
}

func TestAgentEvents_HandlerWithoutAgentContext(t *testing.T) {
	hub := NewSSEHub()
	rr := httptest.NewRecorder()
	hub.HandleAgentEvents(rr, httptest.NewRequest("GET", "/api/agent-events", nil))
	if rr.Code != http.StatusUnauthorized {
		t.Fatalf("status = %d, want 401", rr.Code)
	}
}

func TestAgentEvents_Delivery(t *testing.T) {
	const secret = "TOP-SECRET-BODY-TEXT"

	tests := []struct {
		name string
		// send performs the action; returns the message id to expect for "bob".
		send    func(env *agentEventsEnv) int64
		want    map[string]any // expected fields (besides message_id)
		absent  []string       // fields that must not be present
		viewers []string
	}{
		{
			name: "dm to bob",
			send: func(env *agentEventsEnv) int64 { return env.dm("alice", "bob", secret, "hello subject") },
			want: map[string]any{"from_agent": "alice", "to_agent": "bob", "subject": "hello subject"},
			absent: []string{"channel", "body"},
		},
		{
			name: "channel message to member",
			send: func(env *agentEventsEnv) int64 {
				env.makeChannel("room", "alice", "bob")
				return env.post("alice", "room", secret)
			},
			want:   map[string]any{"channel": "room"},
			absent: []string{"to_agent", "from_agent", "body"},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			env := newAgentEventsEnv(t, "alice", "bob")
			bob := env.connect(env.keys["bob"], "")
			bob.expectConnected("bob")
			alice := env.connect(env.keys["alice"], "")
			alice.expectConnected("alice")

			id := tt.send(env)

			f, m := bob.nextMessage()
			if f.ID != strconv.FormatInt(id, 10) {
				t.Errorf("id line = %q, want %d", f.ID, id)
			}
			if int64(m["message_id"].(float64)) != id {
				t.Errorf("message_id = %v, want %d", m["message_id"], id)
			}
			for k, v := range tt.want {
				if m[k] != v {
					t.Errorf("%s = %v, want %v", k, m[k], v)
				}
			}
			for _, k := range tt.absent {
				if _, ok := m[k]; ok {
					t.Errorf("field %q must not be present: %s", k, f.Data)
				}
			}
			if strings.Contains(f.Data, secret) {
				t.Errorf("event leaks body: %s", f.Data)
			}

			// Sender must not get its own message: send a canary to alice
			// and make sure it is the next thing she sees.
			canary := env.dm("bob", "alice", "canary", "")
			f2, _ := alice.nextMessage()
			if f2.ID != strconv.FormatInt(canary, 10) {
				t.Errorf("alice's next event id = %s, want canary %d (own message leaked?)", f2.ID, canary)
			}
		})
	}
}

func TestAgentEvents_Isolation(t *testing.T) {
	env := newAgentEventsEnv(t, "alice", "bob", "carol")
	env.makeChannel("private-ish", "bob", "carol") // alice is NOT a member
	env.makeChannel("shared", "alice", "bob")

	alice := env.connect(env.keys["alice"], "")
	alice.expectConnected("alice")

	// None of these may reach alice.
	env.dm("bob", "carol", "b->c", "")           // DM between others
	env.post("bob", "private-ish", "not for you") // channel alice has not joined
	env.dm("alice", "bob", "alice->bob", "")     // her own message

	// Canary she is allowed to see; it must be the very next event.
	canary := env.post("bob", "shared", "visible")
	f, m := alice.nextMessage()
	if f.ID != strconv.FormatInt(canary, 10) || m["channel"] != "shared" {
		t.Fatalf("alice got %+v, want only the canary in 'shared'", f)
	}
}

func TestAgentEvents_Resume(t *testing.T) {
	t.Run("replays visible newer messages once, in order", func(t *testing.T) {
		env := newAgentEventsEnv(t, "alice", "bob", "carol")
		env.makeChannel("room", "alice", "bob")
		env.makeChannel("other", "bob", "carol")

		first := env.dm("bob", "alice", "old", "") // client already has this
		m2 := env.dm("bob", "alice", "missed dm", "s")
		env.dm("bob", "carol", "not alice's", "")
		env.post("bob", "other", "not a member")
		m5 := env.post("bob", "room", "missed channel")
		env.dm("alice", "bob", "own", "") // own message excluded

		c := env.connect(env.keys["alice"], strconv.FormatInt(first, 10))
		c.expectConnected("alice")

		f, m := c.nextMessage()
		if f.ID != strconv.FormatInt(m2, 10) || m["subject"] != "s" {
			t.Fatalf("first replay = %+v", f)
		}
		f, m = c.nextMessage()
		if f.ID != strconv.FormatInt(m5, 10) || m["channel"] != "room" {
			t.Fatalf("second replay = %+v", f)
		}

		// Live message after replay arrives exactly once.
		live := env.dm("bob", "alice", "live", "")
		f, _ = c.nextMessage()
		if f.ID != strconv.FormatInt(live, 10) {
			t.Fatalf("live event id = %s, want %d (duplicate or missing?)", f.ID, live)
		}
	})

	t.Run("exactly 200 are replayed", func(t *testing.T) {
		env := newAgentEventsEnv(t, "alice", "bob")
		var ids []int64
		for i := 0; i < maxAgentBacklog; i++ {
			ids = append(ids, env.dm("bob", "alice", "x", ""))
		}
		c := env.connect(env.keys["alice"], "0")
		c.expectConnected("alice")
		for i, id := range ids {
			f, _ := c.nextMessage()
			if f.ID != strconv.FormatInt(id, 10) {
				t.Fatalf("replay #%d id = %s, want %d", i, f.ID, id)
			}
		}
	})

	t.Run("more than 200 sends resync_required only", func(t *testing.T) {
		env := newAgentEventsEnv(t, "alice", "bob")
		for i := 0; i < maxAgentBacklog+1; i++ {
			env.dm("bob", "alice", "x", "")
		}
		c := env.connect(env.keys["alice"], "0")
		c.expectConnected("alice")
		f := c.next()
		if f.Event != "resync_required" {
			t.Fatalf("event = %q, want resync_required", f.Event)
		}
		// No replay; the next event is the live one.
		live := env.dm("bob", "alice", "live", "")
		lf, _ := c.nextMessage()
		if lf.ID != strconv.FormatInt(live, 10) {
			t.Fatalf("after resync got id %s, want live %d", lf.ID, live)
		}
	})

	t.Run("invalid Last-Event-ID is rejected", func(t *testing.T) {
		env := newAgentEventsEnv(t, "alice")
		c := env.connect(env.keys["alice"], "abc")
		if c.status != http.StatusBadRequest {
			t.Fatalf("status = %d, want 400", c.status)
		}
	})
}

func TestAgentEvents_ConnectionLimit(t *testing.T) {
	env := newAgentEventsEnv(t, "alice", "bob")

	var conns []*sseClient
	for i := 0; i < maxAgentConnections; i++ {
		c := env.connect(env.keys["alice"], "")
		if c.status != http.StatusOK {
			t.Fatalf("connection %d status = %d, want 200", i+1, c.status)
		}
		c.expectConnected("alice")
		conns = append(conns, c)
	}

	over := env.connect(env.keys["alice"], "")
	if over.status != http.StatusTooManyRequests {
		t.Fatalf("6th connection status = %d, want 429", over.status)
	}
	if !strings.Contains(over.body, "too_many_connections") {
		t.Errorf("429 body = %q, want explicit error", over.body)
	}

	// The limit is per agent.
	other := env.connect(env.keys["bob"], "")
	if other.status != http.StatusOK {
		t.Fatalf("other agent status = %d, want 200", other.status)
	}

	// Closing a connection frees a slot.
	conns[0].cancel()
	deadline := time.Now().Add(5 * time.Second)
	for {
		c := env.connect(env.keys["alice"], "")
		if c.status == http.StatusOK {
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("slot was not released after disconnect")
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func TestAgentEvents_SlowClientDroppedWithoutBlocking(t *testing.T) {
	hub := NewSSEHub()
	sub := hub.addAgentSub("alice")
	if sub == nil {
		t.Fatal("addAgentSub returned nil")
	}
	done := make(chan struct{})
	go func() {
		defer close(done)
		for i := 1; i <= agentSubBuffer+10; i++ { // nobody reads sub.ch
			hub.BroadcastAgentMessage([]string{"alice"}, AgentMessageEvent{MessageID: int64(i)})
		}
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("broadcast blocked on a slow client")
	}
	hub.mu.RLock()
	n := len(hub.agentSubs["alice"])
	hub.mu.RUnlock()
	if n != 0 {
		t.Fatalf("slow subscription still registered (%d)", n)
	}
}

func TestAgentEvents_Heartbeat(t *testing.T) {
	env := newAgentEventsEnv(t, "alice")
	env.hub.heartbeat = 50 * time.Millisecond
	c := env.connect(env.keys["alice"], "")
	c.expectConnected("alice")
	if f := c.next(); f.Event != "heartbeat" {
		t.Fatalf("event = %q, want heartbeat", f.Event)
	}
}