package chat
import (
"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()
}