wire_test.go

package catbird

import (
	"bufio"
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/google/uuid"
)

var testSecret = []byte("01234567890123456789012345678901") // 32 bytes

func startTestWire(t *testing.T, wire *Wire) {
	t.Helper()

	ctx, cancel := context.WithCancel(t.Context())
	errCh := make(chan error, 1)

	go func() {
		errCh <- wire.Start(ctx)
	}()

	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("wire.Start() failed: %v", err)
		}
	case <-time.After(500 * time.Millisecond):
	}

	t.Cleanup(func() {
		cancel()
		select {
		case err := <-errCh:
			if err != nil && !errors.Is(err, context.Canceled) {
				t.Logf("wire shutdown error: %v", err)
			}
		case <-time.After(10 * time.Second):
			t.Log("wire did not shutdown within timeout")
		}
	})
}

func TestWireTokenMintVerify(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)

	token := wire.Token([]string{"work.01JABC", "user.01JXYZ"}, TokenOpts{
		Identity: "user-123",
		ValidFor: time.Hour,
	})

	payload, err := wire.verifyToken(token)
	if err != nil {
		t.Fatalf("verifyToken: %v", err)
	}
	if len(payload.Topics) != 2 || payload.Topics[0] != "work.01JABC" || payload.Topics[1] != "user.01JXYZ" {
		t.Errorf("topics = %v", payload.Topics)
	}
	if payload.Identity != "user-123" {
		t.Errorf("identity = %q", payload.Identity)
	}
}

func TestWireTokenNoOpts(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)

	token := wire.Token([]string{"topic-a"})
	payload, err := wire.verifyToken(token)
	if err != nil {
		t.Fatalf("verifyToken: %v", err)
	}
	if payload.Identity != "" {
		t.Errorf("expected empty identity, got %q", payload.Identity)
	}
	if payload.Expiry != 0 {
		t.Errorf("expected no expiry, got %d", payload.Expiry)
	}
}

func TestWireTokenExpired(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)

	// Build an expired token directly (Token() ignores non-positive ValidFor)
	p := tokenPayload{Topics: []string{"topic"}, Expiry: time.Now().Add(-time.Minute).Unix()}
	b, _ := json.Marshal(p)
	ct, _ := wireEncrypt(wire.secret, b)
	token := base64.RawURLEncoding.EncodeToString(ct)

	_, err := wire.verifyToken(token)
	if err == nil {
		t.Fatal("expected error for expired token")
	}
}

func TestWireTokenWrongSecret(t *testing.T) {
	getTestClient(t)
	wire1 := NewWire(testPool, testSecret)
	wire2 := NewWire(testPool, []byte("different-secret-key-32-bytesXXX"))

	token := wire1.Token([]string{"topic"})
	_, err := wire2.verifyToken(token)
	if err == nil {
		t.Fatal("expected error for wrong secret")
	}
}

func TestWireTokenInvalid(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)

	_, err := wire.verifyToken("not-a-valid-token")
	if err == nil {
		t.Fatal("expected error for garbage token")
	}
}

func TestWireServeSSEUnauthorized(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	rr := httptest.NewRecorder()
	req := httptest.NewRequest("GET", "/events/bad-token", nil)
	wire.ServeSSE(rr, req, "bad-token")

	if rr.Code != http.StatusUnauthorized {
		t.Errorf("expected 401, got %d", rr.Code)
	}
}

func TestWireNotifySSE(t *testing.T) {
	cb := getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	topic := "test." + uuid.NewString()[:8]
	token := wire.Token([]string{topic})

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		wire.ServeSSE(w, r, r.URL.Query().Get("token"))
	}))
	defer srv.Close()

	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(srv.URL + "?token=" + token)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		t.Fatalf("expected 200, got %d", resp.StatusCode)
	}

	time.Sleep(100 * time.Millisecond)

	// Notify via Client (goes through pg_notify, cross-node path)
	if err := cb.Notify(t.Context(), topic, ""); err != nil {
		t.Fatal(err)
	}

	// Read SSE event — event name is the topic
	events := readSSEEvents(t, resp, 1, 2*time.Second)
	if len(events) == 0 {
		t.Fatal("no SSE events received")
	}
	if events[0].event != topic {
		t.Errorf("event = %q, want %q", events[0].event, topic)
	}
}

func TestWireNotifyWithData(t *testing.T) {
	cb := getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	topic := "test." + uuid.NewString()[:8]
	token := wire.Token([]string{topic})

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		wire.ServeSSE(w, r, r.URL.Query().Get("token"))
	}))
	defer srv.Close()

	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(srv.URL + "?token=" + token)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()

	time.Sleep(100 * time.Millisecond)

	if err := cb.Notify(t.Context(), topic, `<div>hello</div>`); err != nil {
		t.Fatal(err)
	}

	events := readSSEEvents(t, resp, 1, 2*time.Second)
	if len(events) == 0 {
		t.Fatal("no SSE events received")
	}
	if events[0].event != topic {
		t.Errorf("event = %q, want %q", events[0].event, topic)
	}
	if events[0].data != "<div>hello</div>" {
		t.Errorf("data = %q, want %q", events[0].data, "<div>hello</div>")
	}
}

func TestWireLocalNotify(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	topic := "test." + uuid.NewString()[:8]
	token := wire.Token([]string{topic})

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		wire.ServeSSE(w, r, r.URL.Query().Get("token"))
	}))
	defer srv.Close()

	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(srv.URL + "?token=" + token)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()

	time.Sleep(100 * time.Millisecond)

	// Notify via Wire (local + pg_notify)
	if err := wire.Notify(t.Context(), topic, ""); err != nil {
		t.Fatal(err)
	}

	events := readSSEEvents(t, resp, 1, 2*time.Second)
	if len(events) == 0 {
		t.Fatal("no SSE events received")
	}
	if events[0].event != topic {
		t.Errorf("event = %q, want %q", events[0].event, topic)
	}
}

func TestWirePresence(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	topic := "presence." + uuid.NewString()[:8]
	token := wire.Token([]string{topic}, TokenOpts{Identity: "alice"})

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		wire.ServeSSE(w, r, r.URL.Query().Get("token"))
	}))
	defer srv.Close()

	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(srv.URL + "?token=" + token)
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(200 * time.Millisecond)

	identities, err := wire.Presence(t.Context(), topic)
	if err != nil {
		t.Fatal(err)
	}
	if len(identities) != 1 || identities[0] != "alice" {
		t.Errorf("presence = %v, want [alice]", identities)
	}

	// Disconnect
	resp.Body.Close()
	time.Sleep(200 * time.Millisecond)

	identities, err = wire.Presence(t.Context(), topic)
	if err != nil {
		t.Fatal(err)
	}
	if len(identities) != 0 {
		t.Errorf("presence after disconnect = %v, want []", identities)
	}
}

func TestWirePresenceMultipleConnections(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	topic := "presence." + uuid.NewString()[:8]

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		wire.ServeSSE(w, r, r.URL.Query().Get("token"))
	}))
	defer srv.Close()

	client := &http.Client{Timeout: 5 * time.Second}

	// Two connections from same identity
	token1 := wire.Token([]string{topic}, TokenOpts{Identity: "bob"})
	resp1, err := client.Get(srv.URL + "?token=" + token1)
	if err != nil {
		t.Fatal(err)
	}
	defer resp1.Body.Close()

	token2 := wire.Token([]string{topic}, TokenOpts{Identity: "bob"})
	resp2, err := client.Get(srv.URL + "?token=" + token2)
	if err != nil {
		t.Fatal(err)
	}
	defer resp2.Body.Close()

	time.Sleep(200 * time.Millisecond)

	// Should still show one identity (DISTINCT)
	identities, err := wire.Presence(t.Context(), topic)
	if err != nil {
		t.Fatal(err)
	}
	if len(identities) != 1 || identities[0] != "bob" {
		t.Errorf("presence = %v, want [bob]", identities)
	}

	// Close first connection
	resp1.Body.Close()
	time.Sleep(200 * time.Millisecond)

	// Still present via second connection
	identities, err = wire.Presence(t.Context(), topic)
	if err != nil {
		t.Fatal(err)
	}
	if len(identities) != 1 || identities[0] != "bob" {
		t.Errorf("presence after first disconnect = %v, want [bob]", identities)
	}
}

func TestWireDeliverLocal(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	sub := &wireSubscriber{
		id:           uuid.NewString(),
		ch:           make(chan wireEvent, wireChannelSize),
		topics:       []string{"topic-a"},
		cancel:       cancel,
		lastDelivery: time.Now(),
	}

	wire.addSubscriber(sub)
	defer wire.removeSubscriber(sub)

	wire.deliverLocal("topic-a", wireEvent{topic: "topic-a", message: "pong"})

	select {
	case ev := <-sub.ch:
		if ev.topic != "topic-a" {
			t.Errorf("topic = %q, want %q", ev.topic, "topic-a")
		}
		if ev.message != "pong" {
			t.Errorf("message = %q, want %q", ev.message, "pong")
		}
	case <-time.After(time.Second):
		t.Fatal("timeout")
	}

	// Deliver to non-existent topic — should not panic
	wire.deliverLocal("no-such-topic", wireEvent{topic: "no-such-topic", message: ""})

	_ = ctx
}

func TestWireDeliverLocalWildcard(t *testing.T) {
	getTestClient(t)
	wire := NewWire(testPool, testSecret)

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	sub := &wireSubscriber{
		id:           uuid.NewString(),
		ch:           make(chan wireEvent, wireChannelSize),
		topics:       []string{"user.123.#"},
		cancel:       cancel,
		lastDelivery: time.Now(),
	}

	wire.addSubscriber(sub)
	defer wire.removeSubscriber(sub)

	// Publish to a subtopic — should match the # wildcard
	wire.deliverLocal("user.123.batch_edit.done", wireEvent{topic: "user.123.batch_edit.done", message: "ok"})

	select {
	case ev := <-sub.ch:
		if ev.topic != "user.123.batch_edit.done" {
			t.Errorf("topic = %q, want %q", ev.topic, "user.123.batch_edit.done")
		}
		if ev.message != "ok" {
			t.Errorf("message = %q, want %q", ev.message, "ok")
		}
	case <-time.After(time.Second):
		t.Fatal("timeout — wildcard subscriber did not receive event")
	}

	// Non-matching topic should not deliver
	wire.deliverLocal("user.456.notification", wireEvent{topic: "user.456.notification"})

	select {
	case ev := <-sub.ch:
		t.Fatalf("unexpected event: %+v", ev)
	case <-time.After(50 * time.Millisecond):
	}

	_ = ctx
}

func TestWireNotifySSEWildcard(t *testing.T) {
	cb := getTestClient(t)
	wire := NewWire(testPool, testSecret)
	startTestWire(t, wire)

	base := "user." + uuid.NewString()[:8]
	token := wire.Token([]string{base + ".#"})

	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		wire.ServeSSE(w, r, r.URL.Query().Get("token"))
	}))
	defer srv.Close()

	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(srv.URL + "?token=" + token)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()

	time.Sleep(100 * time.Millisecond)

	subtopic := base + ".batch_edit.done"
	if err := cb.Notify(t.Context(), subtopic, "finished"); err != nil {
		t.Fatal(err)
	}

	events := readSSEEvents(t, resp, 1, 2*time.Second)
	if len(events) == 0 {
		t.Fatal("no SSE events received for wildcard subscription")
	}
	if events[0].event != subtopic {
		t.Errorf("event = %q, want %q", events[0].event, subtopic)
	}
	if events[0].data != "finished" {
		t.Errorf("data = %q, want %q", events[0].data, "finished")
	}
}

// --- SSE test helpers ---

type sseEvent struct {
	event string
	data  string
}

// --- Listen tests (migrated from listener_test.go) ---

func TestWireListenExactTopic(t *testing.T) {
	cb := getTestClient(t)

	topic := "test.listen." + uuid.NewString()[:8]

	var mu sync.Mutex
	var received []string

	wire := NewWire(testPool, testSecret)
	wire.Listen(topic, func(ctx context.Context, topic, message string) {
		mu.Lock()
		received = append(received, topic)
		mu.Unlock()
	})
	startTestWire(t, wire)

	if err := cb.Notify(t.Context(), topic, ""); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if len(received) != 1 || received[0] != topic {
		t.Errorf("received = %v, want [%s]", received, topic)
	}
}

func TestWireListenWildcardStar(t *testing.T) {
	cb := getTestClient(t)

	prefix := "test.lwc." + uuid.NewString()[:8]

	var mu sync.Mutex
	var received []string

	wire := NewWire(testPool, testSecret)
	wire.Listen(prefix+".*", func(ctx context.Context, topic, message string) {
		mu.Lock()
		received = append(received, topic)
		mu.Unlock()
	})
	startTestWire(t, wire)

	// Should match
	if err := cb.Notify(t.Context(), prefix+".created", ""); err != nil {
		t.Fatal(err)
	}
	if err := cb.Notify(t.Context(), prefix+".updated", ""); err != nil {
		t.Fatal(err)
	}
	// Should not match (two tokens after prefix)
	if err := cb.Notify(t.Context(), prefix+".sub.created", ""); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if len(received) != 2 {
		t.Errorf("received = %v, want 2 events", received)
	}
}

func TestWireListenWildcardHash(t *testing.T) {
	cb := getTestClient(t)

	prefix := "test.lhash." + uuid.NewString()[:8]

	var mu sync.Mutex
	var received []string

	wire := NewWire(testPool, testSecret)
	wire.Listen(prefix+".#", func(ctx context.Context, topic, message string) {
		mu.Lock()
		received = append(received, topic)
		mu.Unlock()
	})
	startTestWire(t, wire)

	// All should match
	if err := cb.Notify(t.Context(), prefix, ""); err != nil {
		t.Fatal(err)
	}
	if err := cb.Notify(t.Context(), prefix+".one", ""); err != nil {
		t.Fatal(err)
	}
	if err := cb.Notify(t.Context(), prefix+".one.two", ""); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if len(received) != 3 {
		t.Errorf("received %d events, want 3: %v", len(received), received)
	}
}

func TestWireListenWithData(t *testing.T) {
	cb := getTestClient(t)

	topic := "test.ldata." + uuid.NewString()[:8]

	var mu sync.Mutex
	var gotData string

	wire := NewWire(testPool, testSecret)
	wire.Listen(topic, func(ctx context.Context, topic, message string) {
		mu.Lock()
		gotData = message
		mu.Unlock()
	})
	startTestWire(t, wire)

	if err := cb.Notify(t.Context(), topic, `<div>hello</div>`); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if gotData != "<div>hello</div>" {
		t.Errorf("data = %q, want %q", gotData, "<div>hello</div>")
	}
}

func TestWireListenMultipleHandlers(t *testing.T) {
	cb := getTestClient(t)

	topic := "test.lmulti." + uuid.NewString()[:8]

	var mu sync.Mutex
	var count int

	wire := NewWire(testPool, testSecret)
	wire.Listen(topic, func(ctx context.Context, topic, message string) {
		mu.Lock()
		count++
		mu.Unlock()
	})
	wire.Listen(topic, func(ctx context.Context, topic, message string) {
		mu.Lock()
		count++
		mu.Unlock()
	})
	startTestWire(t, wire)

	if err := cb.Notify(t.Context(), topic, ""); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if count != 2 {
		t.Errorf("count = %d, want 2", count)
	}
}

func TestWireListenNoMatch(t *testing.T) {
	cb := getTestClient(t)

	prefix := "test.lnomatch." + uuid.NewString()[:8]

	var mu sync.Mutex
	var received bool

	wire := NewWire(testPool, testSecret)
	wire.Listen(prefix+".specific", func(ctx context.Context, topic, message string) {
		mu.Lock()
		received = true
		mu.Unlock()
	})
	startTestWire(t, wire)

	if err := cb.Notify(t.Context(), prefix+".other", ""); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if received {
		t.Error("handler was called for non-matching topic")
	}
}

func TestWireListenSkipSelf(t *testing.T) {
	cb := getTestClient(t)

	topic := "test.lskipself." + uuid.NewString()[:8]

	var mu sync.Mutex
	var count int

	wire := NewWire(testPool, testSecret)
	wire.Listen(topic, func(ctx context.Context, topic, message string) {
		mu.Lock()
		count++
		mu.Unlock()
	})
	startTestWire(t, wire)

	// Notify with SentBy: wire.ID() — should be skipped
	if err := cb.Notify(t.Context(), topic, "", NotifyOpts{SentBy: wire.ID()}); err != nil {
		t.Fatal(err)
	}

	// Notify without SentBy — should be delivered
	if err := cb.Notify(t.Context(), topic, ""); err != nil {
		t.Fatal(err)
	}

	time.Sleep(500 * time.Millisecond)

	mu.Lock()
	defer mu.Unlock()
	if count != 1 {
		t.Errorf("count = %d, want 1 (skip-self should have prevented one delivery)", count)
	}
}

func readSSEEvents(t *testing.T, resp *http.Response, count int, timeout time.Duration) []sseEvent {
	t.Helper()

	ch := make(chan sseEvent, count)
	go func() {
		scanner := bufio.NewScanner(resp.Body)
		var currentEvent, currentData string
		for scanner.Scan() {
			line := scanner.Text()
			if v, ok := strings.CutPrefix(line, "event: "); ok {
				currentEvent = v
			} else if v, ok := strings.CutPrefix(line, "data: "); ok {
				currentData = v
			} else if line == "data:" {
				currentData = ""
			} else if line == "" && currentEvent != "" {
				ch <- sseEvent{event: currentEvent, data: currentData}
				currentEvent = ""
				currentData = ""
			}
		}
	}()

	var events []sseEvent
	deadline := time.After(timeout)
	for range count {
		select {
		case ev := <-ch:
			events = append(events, ev)
		case <-deadline:
			return events
		}
	}
	return events
}