package chat
import (
"runtime"
"testing"
"time"
natsserver "github.com/nats-io/nats-server/v2/server"
"craftdesign.group/quarry/internal/events"
)
func room(t *testing.T, ev *events.Server) *Room {
t.Helper()
r, err := New(ev.Conn(), ev.JS())
if err != nil {
t.Fatal(err)
}
return r
}
// Post returns the JetStream sequence, monotonically increasing per
// channel, and History replays the window oldest-first carrying that
// same seq back; a second channel is isolated from the first.
func TestPostAndHistory(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
var seqs []uint64
for _, txt := range []string{"first", "second", "third"} {
seq, err := r.Post("repoA", Message{User: "dan", Text: txt})
if err != nil {
t.Fatal(err)
}
seqs = append(seqs, seq)
}
if _, err := r.Post("repoB", Message{User: "eve", Text: "other channel"}); err != nil {
t.Fatal(err)
}
if seqs[0] == 0 || seqs[1] <= seqs[0] || seqs[2] <= seqs[1] {
t.Fatalf("seq not monotonic: %v", seqs)
}
hist, next, err := r.History("repoA", 0, 50)
if err != nil {
t.Fatal(err)
}
if next != 0 {
t.Fatalf("next = %d, want 0 (nothing older than a 3-message window)", next)
}
if len(hist) != 3 || hist[0].Text != "first" || hist[2].Text != "third" {
t.Fatalf("history = %+v", hist)
}
for i, m := range hist {
if m.Seq != seqs[i] {
t.Fatalf("history[%d].Seq = %d, want %d", i, m.Seq, seqs[i])
}
}
if hist[0].At == "" || hist[0].User != "dan" || hist[0].Kind != KindMsg {
t.Fatalf("message not stamped: %+v", hist[0])
}
other, _, _ := r.History("repoB", 0, 50)
if len(other) != 1 || other[0].Text != "other channel" {
t.Fatalf("channel isolation broken: %+v", other)
}
// Empty author defaults to anonymous.
r.Post("repoA", Message{Text: "no name"})
hist, _, _ = r.History("repoA", 0, 50)
if hist[len(hist)-1].User != "anonymous" {
t.Fatalf("anon default missing: %+v", hist[len(hist)-1])
}
}
// History(before) pages backward exactly: a limit-1 page returns the
// single newest message and Next names the seq to ask for the one
// before it; walking Next repeatedly reconstructs the whole window in
// reverse with no gaps and no repeats.
func TestHistoryPagesExactly(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
var seqs []uint64
for i := 0; i < 5; i++ {
seq, err := r.Post("r", Message{User: "dan", Text: string(rune('a' + i))})
if err != nil {
t.Fatal(err)
}
seqs = append(seqs, seq)
}
var walked []uint64
before := uint64(0)
for i := 0; i < 10; i++ { // bounded so a Next bug cannot infinite-loop the test
page, next, err := r.History("r", before, 2)
if err != nil {
t.Fatal(err)
}
if len(page) == 0 {
break
}
pageSeqs := make([]uint64, len(page))
for i, m := range page {
pageSeqs[i] = m.Seq
}
// Pages arrive newest-to-oldest; each page itself is oldest-first,
// so prepending the whole page (not element-by-element) restores
// forward order across the full walk.
walked = append(pageSeqs, walked...)
if next == 0 {
break
}
before = next
}
if len(walked) != len(seqs) {
t.Fatalf("walked %v, want all of %v", walked, seqs)
}
for i, seq := range seqs {
if walked[i] != seq {
t.Fatalf("walked[%d] = %d, want %d (full walk: %v)", i, walked[i], seq, walked)
}
}
}
// Tail(from) delivers exactly the messages after from, then live
// messages posted afterward, in order.
func TestTailResumesFromSeq(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
var seqs []uint64
for i := 0; i < 3; i++ {
seq, err := r.Post("r", Message{User: "dan", Text: string(rune('a' + i))})
if err != nil {
t.Fatal(err)
}
seqs = append(seqs, seq)
}
ch, cancel := r.Tail("r", seqs[0])
defer cancel()
want := []string{"b", "c"}
for _, w := range want {
select {
case m := <-ch:
if m.Text != w {
t.Fatalf("got %q, want %q", m.Text, w)
}
case <-time.After(5 * time.Second):
t.Fatalf("timed out waiting for replay of %q", w)
}
}
if _, err := r.Post("r", Message{User: "dan", Text: "live"}); err != nil {
t.Fatal(err)
}
select {
case m := <-ch:
if m.Text != "live" {
t.Fatalf("got %q, want %q", m.Text, "live")
}
case <-time.After(5 * time.Second):
t.Fatal("live message not delivered after replay")
}
}
// Tail(0) delivers new messages only — nothing posted before the tail
// opened replays.
func TestTailZeroIsNewOnly(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
if _, err := r.Post("r", Message{User: "dan", Text: "before"}); err != nil {
t.Fatal(err)
}
ch, cancel := r.Tail("r", 0)
defer cancel()
if _, err := r.Post("r", Message{User: "dan", Text: "after"}); err != nil {
t.Fatal(err)
}
select {
case m := <-ch:
if m.Text != "after" {
t.Fatalf("got %q, want %q (no pre-existing message)", m.Text, "after")
}
case <-time.After(5 * time.Second):
t.Fatal("live message not delivered")
}
}
// A slow Tail reader loses nothing: posts pile up while nobody reads,
// then all of them arrive once the reader catches up — the property
// Subscribe deliberately does NOT have (it drops under backpressure).
func TestTailNeverDropsUnderBackpressure(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
ch, cancel := r.Tail("r", 0)
defer cancel()
const n = 50
for i := 0; i < n; i++ {
if _, err := r.Post("r", Message{User: "dan", Text: string(rune('a' + i%26))}); err != nil {
t.Fatal(err)
}
}
// Nobody has read from ch yet; a dropping implementation (like
// Subscribe's 32-slot buffer) would have lost most of these by now.
got := 0
timeout := time.After(10 * time.Second)
for got < n {
select {
case <-ch:
got++
case <-timeout:
t.Fatalf("only %d/%d messages delivered — Tail dropped under backpressure", got, n)
}
}
}
// Subscribe, by contrast, is allowed to drop: a slow reader with a full
// 32-slot buffer loses messages rather than blocking the publisher,
// which is the documented, correct behaviour for the tick-style uses
// it remains for.
func TestSubscribeDropsUnderBackpressure(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
ch, cancel := r.Subscribe("r")
defer cancel()
const n = 200 // comfortably over the 32-slot buffer
for i := 0; i < n; i++ {
if _, err := r.Post("r", Message{User: "dan", Text: "x"}); err != nil {
t.Fatal(err)
}
}
time.Sleep(200 * time.Millisecond)
got := len(ch)
if got >= n {
t.Fatalf("Subscribe delivered all %d messages with no reader draining; expected drops", n)
}
}
// Fold applies an edit in place and marks it, applies a delete by
// blanking the text and marking it, and drops both tombstones from the
// result — a reader never sees the append-only shape underneath.
func TestFold(t *testing.T) {
msgs := []Message{
{Seq: 1, Kind: KindMsg, Text: "hello"},
{Seq: 2, Kind: KindMsg, Text: "world"},
{Seq: 3, Kind: KindEdit, Ref: 1, Text: "hello, edited"},
{Seq: 4, Kind: KindMsg, Text: "untouched"},
{Seq: 5, Kind: KindDel, Ref: 2},
}
got := Fold(msgs)
if len(got) != 3 {
t.Fatalf("Fold length = %d, want 3 (tombstones dropped): %+v", len(got), got)
}
if got[0].Seq != 1 || got[0].Text != "hello, edited" || !got[0].Edited {
t.Fatalf("edited message = %+v", got[0])
}
if got[1].Seq != 2 || got[1].Text != "" || !got[1].Deleted {
t.Fatalf("deleted message = %+v", got[1])
}
if got[2].Seq != 4 || got[2].Text != "untouched" || got[2].Edited || got[2].Deleted {
t.Fatalf("untouched message = %+v", got[2])
}
}
// A live subscriber receives posts as they arrive.
func TestLiveSubscribe(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
ch, cancel := r.Subscribe("repo")
defer cancel()
if _, err := r.Post("repo", Message{User: "dan", Text: "live message"}); err != nil {
t.Fatal(err)
}
select {
case m := <-ch:
if m.Text != "live message" || m.User != "dan" {
t.Fatalf("got %+v", m)
}
case <-time.After(5 * time.Second):
t.Fatal("live message not delivered")
}
}
// Chat fans out across nodes on a shared external NATS — a post on one
// node reaches a subscriber on another (the multi-node property).
func TestCrossNodeChat(t *testing.T) {
url := externalURL(t)
evA, err := events.Connect(url)
if err != nil {
t.Fatal(err)
}
defer evA.Close()
evB, err := events.Connect(url)
if err != nil {
t.Fatal(err)
}
defer evB.Close()
a, b := room(t, evA), room(t, evB)
ch, cancel := b.Subscribe("shared")
defer cancel()
if _, err := a.Post("shared", Message{User: "dan", Text: "cross-node hello"}); err != nil {
t.Fatal(err)
}
select {
case m := <-ch:
if m.Text != "cross-node hello" {
t.Fatalf("got %+v", m)
}
case <-time.After(5 * time.Second):
t.Fatal("cross-node chat not delivered")
}
}
func externalURL(t *testing.T) string {
t.Helper()
srv, err := natsserver.NewServer(&natsserver.Options{
Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(),
})
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("external NATS not ready")
}
t.Cleanup(srv.Shutdown)
return srv.ClientURL()
}
// cancel() closes the delivery channel, so every consumer downstream of
// a Tail (main.go's bridge goroutine, the SSE handler) learns the stream
// is over by the range/receive ending rather than blocking forever.
func TestTailCancelClosesChannel(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
ch, cancel := r.Tail("r", 0)
cancel()
select {
case _, open := <-ch:
if open {
t.Fatal("cancel() delivered a message instead of closing the channel")
}
case <-time.After(5 * time.Second):
t.Fatal("cancel() did not close the Tail channel")
}
cancel() // idempotent: a second cancel must not panic
}
// A cancelled Tail leaves no goroutine behind — not the pump that owns
// the delivery channel, and not the JetStream subscription's own.
func TestTailLeavesNoGoroutines(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
// Warm up: the first Tail creates library-side machinery that is not
// per-call, so the baseline is taken after one full cycle.
_, warm := r.Tail("r", 0)
warm()
time.Sleep(200 * time.Millisecond)
before := runtime.NumGoroutine()
for i := 0; i < 10; i++ {
ch, cancel := r.Tail("r", 0)
if _, err := r.Post("r", Message{User: "dan", Text: "x"}); err != nil {
t.Fatal(err)
}
select {
case <-ch:
case <-time.After(5 * time.Second):
t.Fatal("no delivery")
}
cancel()
}
// Goroutines unwind asynchronously; give them a bounded chance.
var after int
for i := 0; i < 50; i++ {
time.Sleep(100 * time.Millisecond)
after = runtime.NumGoroutine()
if after <= before+2 {
return
}
}
t.Fatalf("goroutines leaked: %d before, %d after 10 cancelled Tails", before, after)
}
// Two channels interleaved on the one stream: every per-channel number
// (bounds, unread, history) must come from a subject-filtered view, so
// A's answers never count B's messages. Stream-wide FirstSeq/LastSeq
// arithmetic gets all three of these wrong.
func TestChannelsAreCountedSeparately(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
var a, b []uint64
for i := 0; i < 5; i++ {
sa, err := r.Post("A", Message{User: "dan", Text: "a"})
if err != nil {
t.Fatal(err)
}
sb, err := r.Post("B", Message{User: "eve", Text: "b"})
if err != nil {
t.Fatal(err)
}
a, b = append(a, sa), append(b, sb)
}
first, last, err := r.Bounds("A")
if err != nil {
t.Fatal(err)
}
if first != a[0] || last != a[len(a)-1] {
t.Fatalf("Bounds(A) = %d,%d, want %d,%d (B's sequences leaked in)", first, last, a[0], a[len(a)-1])
}
first, last, err = r.Bounds("B")
if err != nil {
t.Fatal(err)
}
if first != b[0] || last != b[len(b)-1] {
t.Fatalf("Bounds(B) = %d,%d, want %d,%d", first, last, b[0], b[len(b)-1])
}
if f, l, err := r.Bounds("never"); err != nil || f != 0 || l != 0 {
t.Fatalf("Bounds(never) = %d,%d,%v, want 0,0,nil", f, l, err)
}
// Unread counts THIS channel's messages after since, never the
// stream-wide difference last-since (which here would be ~2x).
if n, err := r.Unread("A", 0); err != nil || n != 5 {
t.Fatalf("Unread(A, 0) = %d,%v, want 5", n, err)
}
if n, err := r.Unread("A", a[2]); err != nil || n != 2 {
t.Fatalf("Unread(A, %d) = %d,%v, want 2", a[2], n, err)
}
if n, err := r.Unread("A", a[4]); err != nil || n != 0 {
t.Fatalf("Unread(A, latest) = %d,%v, want 0", n, err)
}
if n, err := r.Unread("never", 0); err != nil || n != 0 {
t.Fatalf("Unread(never, 0) = %d,%v, want 0", n, err)
}
// History pages A only, and its Next is exclusive: walking it must
// reconstruct A's five sequences with no repeats and no B rows.
var walked []uint64
before := uint64(0)
for i := 0; i < 10; i++ {
page, next, err := r.History("A", before, 2)
if err != nil {
t.Fatal(err)
}
if len(page) == 0 {
break
}
seqs := make([]uint64, len(page))
for j, m := range page {
if m.Text != "a" {
t.Fatalf("History(A) returned a B message: %+v", m)
}
seqs[j] = m.Seq
}
walked = append(seqs, walked...)
if next == 0 {
break
}
before = next
}
if len(walked) != len(a) {
t.Fatalf("walked %v, want exactly %v", walked, a)
}
for i := range a {
if walked[i] != a[i] {
t.Fatalf("walked[%d] = %d, want %d (full walk %v)", i, walked[i], a[i], walked)
}
}
}
// Tail from a sequence the window has already dropped cannot silently
// restart at the window's first message — the reader would believe it
// resumed. A gap signal precedes the replay so the consumer can
// re-render instead of appending.
func TestTailReportsAGapWhenFromIsBelowTheWindow(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
// Another channel's messages take the low sequences, so channel "r"
// genuinely starts above 1 and a lower "from" names a sequence this
// channel never had.
for i := 0; i < 3; i++ {
if _, err := r.Post("other", Message{User: "x", Text: "pad"}); err != nil {
t.Fatal(err)
}
}
if _, err := r.Post("r", Message{User: "dan", Text: "first here"}); err != nil {
t.Fatal(err)
}
first, _, err := r.Bounds("r")
if err != nil {
t.Fatal(err)
}
if first < 3 {
t.Fatalf("channel first = %d, want it above the padding", first)
}
out, cancel := r.Tail("r", first-2)
defer cancel()
select {
case m := <-out:
if m.Kind != KindGap {
t.Fatalf("first delivery = %+v, want a %q signal", m, KindGap)
}
if m.Ref != first-2 || m.Seq != first {
t.Fatalf("gap = %+v, want Ref %d (requested from) and Seq %d (window first)", m, first-2, first)
}
case <-time.After(5 * time.Second):
t.Fatal("no gap signal for a from below the window")
}
// The replay itself still follows the signal.
select {
case m := <-out:
if m.Kind != KindMsg || m.Text != "first here" {
t.Fatalf("after the gap = %+v, want the replayed message", m)
}
case <-time.After(5 * time.Second):
t.Fatal("no replay after the gap signal")
}
}
// A resume that the window can honour reports no gap.
func TestTailReportsNoGapWhenFromIsInsideTheWindow(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
one, err := r.Post("r", Message{User: "dan", Text: "one"})
if err != nil {
t.Fatal(err)
}
if _, err := r.Post("r", Message{User: "dan", Text: "two"}); err != nil {
t.Fatal(err)
}
out, cancel := r.Tail("r", one)
defer cancel()
select {
case m := <-out:
if m.Kind == KindGap {
t.Fatalf("gap reported for an in-window resume: %+v", m)
}
if m.Text != "two" {
t.Fatalf("got %+v, want the message after from", m)
}
case <-time.After(5 * time.Second):
t.Fatal("no delivery")
}
}
// A del tombstone is terminal: an edit that arrives after it, or before
// it, never resurrects the text.
func TestFoldDeleteIsTerminal(t *testing.T) {
deleteThenEdit := []Message{
{Seq: 1, Kind: KindMsg, Text: "hello"},
{Seq: 2, Kind: KindDel, Ref: 1},
{Seq: 3, Kind: KindEdit, Ref: 1, Text: "resurrected"},
}
editThenDelete := []Message{
{Seq: 1, Kind: KindMsg, Text: "hello"},
{Seq: 2, Kind: KindEdit, Ref: 1, Text: "edited"},
{Seq: 3, Kind: KindDel, Ref: 1},
}
for name, msgs := range map[string][]Message{
"delete then edit": deleteThenEdit,
"edit then delete": editThenDelete,
} {
got := Fold(msgs)
if len(got) != 1 {
t.Fatalf("%s: Fold = %+v, want one message", name, got)
}
if !got[0].Deleted || got[0].Text != "" {
t.Fatalf("%s: deleted message resurrected: %+v", name, got[0])
}
}
}
// A gap signal is not a message: Fold never renders it.
func TestFoldDropsGapSignals(t *testing.T) {
got := Fold([]Message{
{Seq: 5, Kind: KindGap, Ref: 2},
{Seq: 6, Kind: KindMsg, Text: "hi"},
})
if len(got) != 1 || got[0].Text != "hi" {
t.Fatalf("Fold = %+v, want the message alone", got)
}
}