package chat
import (
"errors"
"fmt"
"runtime"
"strings"
"sync"
"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++ {
// Distinct text per message: duplicate suppression would fold
// identical posts into one, and this test is about delivery.
if _, err := r.Post("r", Message{User: "dan", Text: fmt.Sprintf("m%d", i)}); 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: fmt.Sprintf("x%d", i)}); 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: fmt.Sprintf("a%d", i)})
if err != nil {
t.Fatal(err)
}
sb, err := r.Post("B", Message{User: "eve", Text: fmt.Sprintf("b%d", i)})
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 !strings.HasPrefix(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: fmt.Sprintf("pad%d", i)}); 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)
}
}
// memRobots is the RobotLog a forge's OLTP provides, in memory: it
// remembers the robot's own sequences and counts the ones after a
// pointer, which is all Unread asks of it.
type memRobots struct {
mu sync.Mutex
seqs []uint64
}
func (m *memRobots) RecordRobotLine(_ string, seq uint64, _ string) error {
m.mu.Lock()
defer m.mu.Unlock()
m.seqs = append(m.seqs, seq)
return nil
}
func (m *memRobots) RobotLines(_ string, since uint64) (int, error) {
m.mu.Lock()
defer m.mu.Unlock()
n := 0
for _, s := range m.seqs {
if s > since {
n++
}
}
return n, nil
}
// countingRobots is a RobotLog that also says how it was asked, so a
// test can prove Unread reached for the counter and not for the window.
type countingRobots struct {
mu sync.Mutex
seqs map[uint64]bool
asked int
// answer overrides the true count, which is how "did you use this
// or did you scan?" becomes an assertion rather than a hope.
answer int
}
func (c *countingRobots) RecordRobotLine(_ string, seq uint64, _ string) error {
c.mu.Lock()
defer c.mu.Unlock()
c.seqs[seq] = true
return nil
}
func (c *countingRobots) RobotLines(_ string, _ uint64) (int, error) {
c.mu.Lock()
defer c.mu.Unlock()
c.asked++
return c.answer, nil
}
// Unread subtracts the robot's lines from the count the broker already
// holds — it never reads the window to tell them apart. The stub
// answers a number the window would contradict, so a Unread that
// scanned would give the window's answer instead of this one.
func TestUnreadSubtractsRobotLinesWithoutScanning(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
bots := &countingRobots{seqs: map[uint64]bool{}, answer: 1}
r.Robots = bots
for i := 0; i < 3; i++ {
if _, err := r.Post("A", Message{User: "dan", Text: fmt.Sprintf("a%d", i)}); err != nil {
t.Fatal(err)
}
}
for i := 0; i < 2; i++ {
if _, err := r.Post("A", Message{Kind: KindRobot, User: "robot", Text: fmt.Sprintf("checkin %d", i)}); err != nil {
t.Fatal(err)
}
}
if len(bots.seqs) != 2 {
t.Fatalf("recorded %d robot lines, want 2", len(bots.seqs))
}
n, err := r.Unread("A", 0)
if err != nil {
t.Fatal(err)
}
// 5 pending, stub says 1 is the robot's: a scan would have found 2.
if n != 4 {
t.Fatalf("Unread(A, 0) = %d, want 4 (NumPending 5 minus the counter's 1)", n)
}
if bots.asked != 1 {
t.Fatalf("the counter was asked %d times, want 1", bots.asked)
}
}
// Nothing advances a read pointer past a robot line, so the unread run
// on a robot-only channel grows without bound. The count stays O(1)
// anyway: a stale pointer with a wide window costs the same two
// questions as a fresh one, and fetches nothing.
func TestUnreadOnStaleWindowFetchesNothing(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
bots := &countingRobots{seqs: map[uint64]bool{}}
r.Robots = bots
const n = 2000
for i := 0; i < n; i++ {
if _, err := r.Post("A", Message{Kind: KindRobot, User: "robot", Text: fmt.Sprintf("checkin %d", i)}); err != nil {
t.Fatal(err)
}
}
bots.answer = n
start := time.Now()
got, err := r.Unread("A", 0)
if err != nil {
t.Fatal(err)
}
if took := time.Since(start); took > 2*time.Second {
t.Fatalf("Unread over a %d-message window took %s — it is reading the window", n, took)
}
if got != 0 {
t.Fatalf("Unread(A, 0) = %d, want 0 (every line is the robot's)", got)
}
}
// Robot rows outlive the messages they describe: they are swept by age
// and the channel by age AND size, so a busy neighbour can push this
// channel's oldest messages out from under rows that still exist. The
// subtraction must be bounded by what the broker is still counting, or
// a badge goes dark while a person's message waits in it.
func TestUnreadIgnoresRobotLinesBelowTheWindow(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
bots := &memRobots{}
r.Robots = bots
// Another channel's traffic takes the first sequences, so channel
// A's window starts above them.
for i := 0; i < 5; i++ {
if _, err := r.Post("B", Message{User: "eve", Text: fmt.Sprintf("elsewhere %d", i)}); err != nil {
t.Fatal(err)
}
}
// Rows for lines A no longer holds — what a sweep has not yet
// reached, and what the old subtraction counted.
for seq := uint64(1); seq <= 5; seq++ {
if err := bots.RecordRobotLine("A", seq, ""); err != nil {
t.Fatal(err)
}
}
if _, err := r.Post("A", Message{User: "dan", Text: "still waiting"}); err != nil {
t.Fatal(err)
}
n, err := r.Unread("A", 0)
if err != nil {
t.Fatal(err)
}
if n != 1 {
t.Fatalf("Unread(A, 0) = %d, want 1 — aged-out robot rows were subtracted", n)
}
}
// Range(code, 0, ...) means "from the beginning". It must not become
// JetStream's StartSequence(0), which sets the delivery policy to "by
// start sequence" and then leaves the sequence unset — the server
// rejects that outright with:
//
// consumer delivery policy is deliver by start sequence, but
// optional start sequence is not set
//
// Bounds answers a zero first sequence for a channel nobody has posted
// in, so the reply backfill handed exactly that to Range for every
// empty channel and failed 1,308 times in one day.
func TestRangeFromZeroReadsFromTheBeginning(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
// An empty channel is the case that was failing: no error, no rows.
got, err := r.Range("empty-channel", 0, 0)
if err != nil {
t.Fatalf("Range over a channel nobody has posted in: %v", err)
}
if len(got) != 0 {
t.Errorf("empty channel returned %d messages", len(got))
}
// And with content, "from zero" means every message, not none.
for _, text := range []string{"first", "second", "third"} {
if _, perr := r.Post("c1", Message{User: "dan", Text: text}); perr != nil {
t.Fatal(perr)
}
}
got, err = r.Range("c1", 0, 0)
if err != nil {
t.Fatalf("Range from zero: %v", err)
}
if len(got) != 3 {
t.Fatalf("Range from zero returned %d messages, want all 3", len(got))
}
if got[0].Text != "first" || got[2].Text != "third" {
t.Errorf("out of order or wrong messages: %+v", got)
}
}
// TestIdenticalPostsWithinTheWindowAreOneMessage pins the guarantee
// behind "you cannot send the same thing twice by accident". A double
// tap, a re-submitted form and a retried request all publish the same
// content; the broker must answer all three with ONE message.
//
// It is a guarantee and not a UI courtesy, which is why it is tested
// here rather than against the composer: the API route reaches Post by
// the same door.
func TestIdenticalPostsWithinTheWindowAreOneMessage(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
first, err := r.Post("A", Message{User: "dan", Text: "same"})
if err != nil {
t.Fatal(err)
}
again, err := r.Post("A", Message{User: "dan", Text: "same"})
if err != nil {
t.Fatal(err)
}
// The second post is ACKED — the sender is not shown an error for
// an accident — but with the FIRST message's sequence.
if again != first {
t.Fatalf("second identical post got seq %d, want the first's %d", again, first)
}
msgs, err := r.Range("A", 0, 0)
if err != nil {
t.Fatal(err)
}
if len(msgs) != 1 {
t.Fatalf("channel holds %d messages, want 1", len(msgs))
}
// The suppression is keyed on content, not on the author being
// noisy: a different author, a different text and a different
// reply target each make a genuinely different post.
for _, m := range []Message{
{User: "eve", Text: "same"},
{User: "dan", Text: "different"},
{User: "dan", Text: "same", Ref: first},
} {
seq, err := r.Post("A", m)
if err != nil {
t.Fatal(err)
}
if seq == first {
t.Fatalf("%+v was folded into the first message", m)
}
}
}
// Forward is the permalink's other half. The properties that matter:
// it pages, it says whether more exists, and it finds this channel's
// messages however sparsely they sit among the stream's.
func TestForwardPagesThisChannelOnly(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
// Interleaved with a busy neighbour, so "five messages on" is far
// from "five sequences on" — the case the doubling reach exists for.
var a []uint64
for i := 0; i < 12; i++ {
for j := 0; j < 3; j++ {
if _, err := r.Post("B", Message{User: "eve", Text: fmt.Sprintf("noise %d.%d", i, j)}); err != nil {
t.Fatal(err)
}
}
seq, err := r.Post("A", Message{User: "dan", Text: fmt.Sprintf("a%d", i)})
if err != nil {
t.Fatal(err)
}
a = append(a, seq)
}
// From the third message, five on: a3..a7, with more to come.
page, next, err := r.Forward("A", a[2], 5)
if err != nil {
t.Fatal(err)
}
if len(page) != 5 {
t.Fatalf("got %d messages, want 5", len(page))
}
for i, m := range page {
if m.Seq != a[3+i] {
t.Fatalf("page[%d] = seq %d, want %d", i, m.Seq, a[3+i])
}
if !strings.HasPrefix(m.Text, "a") {
t.Fatalf("page carries a neighbour's message: %+v", m)
}
}
if next != a[7] {
t.Fatalf("next = %d, want %d (the page's own newest)", next, a[7])
}
// The last page says so by answering 0 — the caller draws no
// "later" control rather than one that leads nowhere.
page, next, err = r.Forward("A", a[8], 5)
if err != nil {
t.Fatal(err)
}
if len(page) != 3 || next != 0 {
t.Fatalf("tail page = %d messages, next %d; want 3 and 0", len(page), next)
}
// Past the end is empty, not an error: a reader who followed a link
// to the newest message asks this every time.
if page, next, err := r.Forward("A", a[len(a)-1], 5); err != nil || len(page) != 0 || next != 0 {
t.Fatalf("Forward past the tip = %d,%d,%v; want 0,0,nil", len(page), next, err)
}
}
// Purge is the only irreversible thing here, and the mistake it must
// never make is erasing a neighbour's message: the stream is shared by
// every repository's channel, so a stray sequence names somebody
// else's row just as readily as your own.
func TestPurgeRefusesAnotherChannelsSequence(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
mine, err := r.Post("A", Message{User: "dan", Text: "mine"})
if err != nil {
t.Fatal(err)
}
theirs, err := r.Post("B", Message{User: "eve", Text: "theirs"})
if err != nil {
t.Fatal(err)
}
if err := r.Purge("A", theirs); !errors.Is(err, ErrNotThisChannel) {
t.Fatalf("purging B's message as A returned %v, want ErrNotThisChannel", err)
}
if msgs, err := r.Range("B", 0, 0); err != nil || len(msgs) != 1 {
t.Fatalf("B lost a message to A's purge: %d, %v", len(msgs), err)
}
// Its own, it erases.
if err := r.Purge("A", mine); err != nil {
t.Fatal(err)
}
if msgs, err := r.Range("A", 0, 0); err != nil || len(msgs) != 0 {
t.Fatalf("A still holds %d messages after purging its only one, %v", len(msgs), err)
}
// Purging what is already gone is the requested end state, not an
// error: a second click must not fail.
if err := r.Purge("A", mine); err != nil {
t.Fatalf("re-purging returned %v, want nil", err)
}
}
// Activity is the whole fleet's pulse: a message in ANY channel ticks
// it, once per burst, so a page can refresh what it says about other
// rooms without tailing them all.
func TestActivityTicksOnAnyChannel(t *testing.T) {
ev, err := events.Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
r := room(t, ev)
tick, stop := r.Activity()
defer stop()
for _, code := range []string{"A", "B"} {
if _, err := r.Post(code, Message{User: "dan", Text: "in " + code}); err != nil {
t.Fatal(err)
}
select {
case <-tick:
case <-time.After(3 * time.Second):
t.Fatalf("no tick for a message in %s", code)
}
}
}