package api
import (
"context"
"errors"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/ops"
)
// fakeTail hands out a fixed list of events, then holds the stream
// open the way a live consumer does.
type fakeTail struct {
events []ForgeEvent
repo string
from uint64
asked bool
canceled chan struct{}
}
func (f *fakeTail) TailEvents(repo string, from uint64) (<-chan ForgeEvent, func()) {
f.repo, f.from, f.asked = repo, from, true
ch := make(chan ForgeEvent)
f.canceled = make(chan struct{})
go func() {
for _, e := range f.events {
select {
case ch <- e:
case <-f.canceled:
return
}
}
}()
var once sync.Once
return ch, func() { once.Do(func() { close(f.canceled) }) }
}
// frame is one thing a stream wrote: an id'd event or a named frame.
type frame struct {
name string
id uint64
data any
}
type recStreamer struct {
mu sync.Mutex
frames []frame
}
func (r *recStreamer) Emit(id uint64, data any) error {
r.mu.Lock()
defer r.mu.Unlock()
r.frames = append(r.frames, frame{id: id, data: data})
return nil
}
func (r *recStreamer) Event(name string, data any) error {
r.mu.Lock()
defer r.mu.Unlock()
r.frames = append(r.frames, frame{name: name, data: data})
return nil
}
func (r *recStreamer) Ping() {}
func (r *recStreamer) wait(t *testing.T, n int) []frame {
t.Helper()
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) {
r.mu.Lock()
if len(r.frames) >= n {
out := append([]frame(nil), r.frames...)
r.mu.Unlock()
return out
}
r.mu.Unlock()
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("FAIL: the stream wrote %d frames, want %d", len(r.frames), n)
return nil
}
// run streams until n frames are written, then ends the stream.
func run(t *testing.T, ctx context.Context, d Deps, repo string, from uint64, kinds []string,
visible func(string) bool, admin bool, n int) []frame {
t.Helper()
ctx, cancel := context.WithCancel(ctx)
s := &recStreamer{}
done := make(chan error, 1)
go func() { done <- d.streamEvents(ctx, s, repo, from, kinds, visible, admin) }()
s.wait(t, n)
time.Sleep(50 * time.Millisecond) // anything past n would land here
cancel()
if err := <-done; err != nil {
t.Fatalf("FAIL: stream ended with %v", err)
}
s.mu.Lock()
defer s.mu.Unlock()
return s.frames
}
var at = time.Date(2026, 11, 11, 13, 55, 2, 0, time.UTC)
// A listener hears "live" once the tail is attached, then each event it
// asked for by kind, in quarry's own words, without the token a change
// came in on; a stale resume point is said out loud first.
func TestTheEventStreamSendsWhatWasAskedFor(t *testing.T) {
ft := &fakeTail{events: []ForgeEvent{
{Gap: true, From: 3, Seq: 9},
{Seq: 9, Kind: "landing.accepted", Repo: "quarry", At: at,
Payload: map[string]any{"landing": "da0c7b1b5356aa", "actor": "dmestas", "via": "TOKEN1"}},
{Seq: 10, Kind: "wiki.edited", Repo: "quarry", At: at},
{Seq: 11, Kind: "ci.completed", Repo: "quarry", At: at,
Payload: map[string]any{"runId": "r1", "status": "fail", "local": true}},
}}
d := Deps{EventTail: ft}
frames := run(t, context.Background(), d, "quarry", 3, []string{"landing.*", "ci.completed"},
func(string) bool { return true }, false, 4)
if !ft.asked || ft.repo != "quarry" || ft.from != 3 {
t.Fatalf("FAIL: tail asked for repo %q from %d", ft.repo, ft.from)
}
if len(frames) != 4 {
t.Fatalf("FAIL: %d frames, want live, gap and two events: %+v", len(frames), frames)
}
if frames[0].name != "live" {
t.Fatalf("FAIL: first frame %+v, want live", frames[0])
}
if g, ok := frames[1].data.(map[string]uint64); frames[1].name != "gap" || !ok || g["from"] != 3 || g["first"] != 9 {
t.Fatalf("FAIL: second frame %+v, want a gap from 3 to 9", frames[1])
}
landing := frames[2].data.(eventOut)
if frames[2].id != 9 || landing.Kind != "landing.accepted" || landing.Actor != "dmestas" ||
!strings.HasPrefix(landing.Link, "/r/quarry/landing/da0c7b1b5356") || landing.At != "2026-11-11T13:55:02Z" {
t.Fatalf("FAIL: landing event %+v (id %d)", landing, frames[2].id)
}
if _, leaked := landing.Payload["via"]; leaked {
t.Fatal("FAIL: the token a change came in on was sent to a listener")
}
if ci := frames[3].data.(eventOut); frames[3].id != 11 || ci.Payload["status"] != "fail" || ci.Payload["local"] != "true" {
t.Fatalf("FAIL: ci event %+v, want its status carried", ci)
}
t.Log("ok")
}
// The forge-wide stream carries only repositories the caller can read,
// and who holds which access only to administrators.
func TestTheEventStreamWithholdsWhatTheCallerMayNotSee(t *testing.T) {
ix := wikiIndex(t)
repos := NewRepos(map[string]*Repo{"open": {Index: ix}, "vault": {Index: ix}})
ft := &fakeTail{events: []ForgeEvent{
{Seq: 1, Kind: "landing.accepted", Repo: "vault", At: at},
{Seq: 2, Kind: "account.caps", Repo: "open", At: at},
{Seq: 3, Kind: "landing.accepted", Repo: "open", At: at},
}}
d := Deps{EventTail: ft, Repos: repos,
CanRead: func(ctx context.Context, repo string) bool { return repo != "vault" }}
ctx := ops.WithUser(context.Background(), "ada")
frames := run(t, ctx, d, "", 0, nil, d.eventVisibility(ctx), d.siteAdmin(ctx), 2)
if len(frames) != 2 || frames[1].id != 3 {
t.Fatalf("FAIL: a reader got %+v, want live and open's landing only", frames)
}
// A site administrator: s at the forge, resolved the router's way.
d.CapsFor = func(user, repo string) CapSet { return capsOf("s") }
frames = run(t, ctx, d, "", 0, nil, d.eventVisibility(ctx), d.siteAdmin(ctx), 3)
if len(frames) != 3 || frames[1].id != 2 || frames[2].id != 3 {
t.Fatalf("FAIL: an administrator got %+v, want the access change and open's landing", frames)
}
t.Log("ok")
}
// A kind that is not a pattern is refused before anything is streamed.
func TestTheEventStreamRefusesABadKindPattern(t *testing.T) {
err := Deps{EventTail: &fakeTail{}}.streamEvents(context.Background(), &recStreamer{}, "", 0,
[]string{"landing.["}, func(string) bool { return true }, false)
if !errors.Is(err, ops.ErrInvalid) {
t.Fatalf("FAIL: a bad pattern answered %v, want invalid", err)
}
t.Log("ok")
}