package api
import (
"context"
"errors"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/evt"
"craftdesign.group/quarry/internal/ops"
)
// fakeAttention is the attention plane as the ops see it: a sequence
// per account, and a tail nobody in these tests reads.
type fakeAttention struct {
mu sync.Mutex
seq uint64
sent map[string][]AttentionEvent
}
func newFakeAttention() *fakeAttention {
return &fakeAttention{sent: map[string][]AttentionEvent{}}
}
func (f *fakeAttention) Attend(account string, e AttentionEvent) (uint64, error) {
f.mu.Lock()
defer f.mu.Unlock()
f.seq++
e.Seq = f.seq
f.sent[account] = append(f.sent[account], e)
return f.seq, nil
}
func (f *fakeAttention) TailAttention(string, uint64) (<-chan AttentionEvent, func()) {
ch := make(chan AttentionEvent)
return ch, func() { close(ch) }
}
func (f *fakeAttention) AttentionPending(account string, since uint64) (int, error) {
f.mu.Lock()
defer f.mu.Unlock()
n := 0
for _, e := range f.sent[account] {
if e.Seq > since {
n++
}
}
return n, nil
}
func (f *fakeAttention) AttentionTip(account string) (uint64, error) {
f.mu.Lock()
defer f.mu.Unlock()
evs := f.sent[account]
if len(evs) == 0 {
return 0, nil
}
return evs[len(evs)-1].Seq, nil
}
// waitFor blocks until account holds at least n events or the wait
// runs out. fireNotify publishes off the op's goroutine now, so a test
// that asserts on the stream has to wait for it.
func (f *fakeAttention) waitFor(t *testing.T, account string, n int) []AttentionEvent {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for {
got := f.to(account)
if len(got) >= n {
return got
}
if time.Now().After(deadline) {
return got
}
time.Sleep(5 * time.Millisecond)
}
}
func (f *fakeAttention) RecentAttention(account string, limit int) ([]AttentionEvent, error) {
evs := f.to(account)
if limit > 0 && len(evs) > limit {
evs = evs[len(evs)-limit:]
}
return evs, nil
}
func (f *fakeAttention) to(account string) []AttentionEvent {
f.mu.Lock()
defer f.mu.Unlock()
return append([]AttentionEvent(nil), f.sent[account]...)
}
type memAttnRead struct {
mu sync.Mutex
seqs map[string]uint64
}
func newMemAttnRead() *memAttnRead { return &memAttnRead{seqs: map[string]uint64{}} }
func (m *memAttnRead) MarkAttentionRead(account string, seq uint64) error {
m.mu.Lock()
defer m.mu.Unlock()
if seq > m.seqs[account] {
m.seqs[account] = seq
}
return nil
}
func (m *memAttnRead) AttentionReadSeq(account string) uint64 {
m.mu.Lock()
defer m.mu.Unlock()
return m.seqs[account]
}
// The stream is written whether or not notify exists — a forge with no
// notify configured still tells people what was addressed to them.
func TestFireNotifyAttendsWithoutNotify(t *testing.T) {
fa := newFakeAttention()
d := Deps{Attention: fa}
d.fireNotify(NotifyEvent{
Kind: evt.ForumPosted, Workflow: "quarry-mention", To: "alice",
Payload: map[string]any{"repo": "alpha", "from": "bob",
"text": "hey @alice look", "link": "/r/alpha/forum/abc"},
TxnID: "mention:forum:abc:alice",
})
got := fa.waitFor(t, "alice", 1)
if len(got) != 1 {
t.Fatalf("alice got %d attention events, want 1", len(got))
}
if got[0].Kind != string(evt.ForumPosted) {
t.Errorf("kind is %q, want %q", got[0].Kind, evt.ForumPosted)
}
if got[0].Repo != "alpha" || got[0].From != "bob" ||
got[0].Link != "/r/alpha/forum/abc" || got[0].Summary == "" {
t.Errorf("event carries %+v", got[0])
}
}
// And it is written when notify IS configured, independently: the two
// paths are not alternatives.
func TestFireNotifyAttendsAlongsideNotify(t *testing.T) {
fa := newFakeAttention()
d := Deps{Attention: fa, Notify: &fakeNotify{}}
d.fireNotify(NotifyEvent{
Workflow: "quarry-forum-reply", To: "alice",
Payload: map[string]any{"repo": "alpha", "from": "bob",
"title": "a thread", "link": "/r/alpha/forum/abc"},
})
got := fa.waitFor(t, "alice", 1)
if len(got) != 1 || got[0].Kind != "reply" || got[0].Summary != "a thread" {
t.Fatalf("alice got %+v", got)
}
}
// Topic traffic is the repository talking, not attention: it must not
// land on anybody's personal subject.
func TestFireNotifyDoesNotAttendTopics(t *testing.T) {
fa := newFakeAttention()
d := Deps{Attention: fa}
for _, e := range []NotifyEvent{
{Kind: evt.TicketOpened, Workflow: "quarry-activity", To: "repo:alpha",
Payload: map[string]any{"repo": "alpha"}},
{Workflow: "quarry-error", To: "admins", Payload: map[string]any{}},
{Workflow: "quarry-mention", To: "repo:alpha", Payload: map[string]any{}},
} {
d.fireNotify(e)
}
// Nothing to wait FOR here: the assertion is an absence, so give
// the goroutines a moment to be wrong in.
time.Sleep(50 * time.Millisecond)
fa.mu.Lock()
defer fa.mu.Unlock()
if len(fa.sent) != 0 {
t.Fatalf("topic events reached %v", fa.sent)
}
}
func attentionRegistry(t *testing.T, d Deps) *ops.Registry {
t.Helper()
r := ops.NewRegistry()
if err := registerAttention(r, d); err != nil {
t.Fatal(err)
}
return r
}
func TestAttentionReadIsForwardOnly(t *testing.T) {
fa, store := newFakeAttention(), newMemAttnRead()
d := Deps{Attention: fa, AttnRead: store}
r := attentionRegistry(t, d)
ctx := ops.WithUser(context.Background(), "alice")
for i := 0; i < 3; i++ {
if _, err := fa.Attend("alice", AttentionEvent{Kind: "mention"}); err != nil {
t.Fatal(err)
}
}
out, err := r.Invoke(ctx, "attention.read", attentionReadIn{Seq: 3})
if err != nil {
t.Fatal(err)
}
if got := out.(attentionReadOut).Seq; got != 3 {
t.Fatalf("pointer is %d, want 3", got)
}
// A stale client replays an older cursor: the pointer must not move
// backwards and resurrect mail already dismissed.
out, err = r.Invoke(ctx, "attention.read", attentionReadIn{Seq: 1})
if err != nil {
t.Fatal(err)
}
if got := out.(attentionReadOut).Seq; got != 3 {
t.Fatalf("pointer went to %d on a backwards mark, want 3", got)
}
}
func TestInboxCountFallsBackToTheLocalTail(t *testing.T) {
fa, store := newFakeAttention(), newMemAttnRead()
r := attentionRegistry(t, Deps{Attention: fa, AttnRead: store})
ctx := ops.WithUser(context.Background(), "alice")
for i := 0; i < 4; i++ {
if _, err := fa.Attend("alice", AttentionEvent{Kind: "mention"}); err != nil {
t.Fatal(err)
}
}
if _, err := fa.Attend("bob", AttentionEvent{Kind: "mention"}); err != nil {
t.Fatal(err)
}
out, err := r.Invoke(ctx, "inbox.count", inboxCountIn{})
if err != nil {
t.Fatal(err)
}
if got := out.(inboxCountOut).Unread; got != 4 {
t.Fatalf("unread is %d, want 4 (bob's is not alice's)", got)
}
// Marking everything read with no seq clears it.
if _, err := r.Invoke(ctx, "attention.read", attentionReadIn{}); err != nil {
t.Fatal(err)
}
out, err = r.Invoke(ctx, "inbox.count", inboxCountIn{})
if err != nil {
t.Fatal(err)
}
if got := out.(inboxCountOut).Unread; got != 0 {
t.Fatalf("unread is %d after marking read, want 0", got)
}
}
type memRobots struct{ events map[string]string }
func (m *memRobots) SetChatRobot(repo, events string) error {
if events == "" {
delete(m.events, repo)
return nil
}
m.events[repo] = events
return nil
}
func (m *memRobots) ChatRobot(repo string) string { return m.events[repo] }
func TestChatRobotValidatesPatterns(t *testing.T) {
store := &memRobots{events: map[string]string{}}
d := Deps{ChatRobots: store, Repos: NewRepos(map[string]*Repo{"alpha": {Code: "alpha"}})}
r := ops.NewRegistry()
if err := registerChatRobot(r, d); err != nil {
t.Fatal(err)
}
ctx := ops.WithUser(context.Background(), "alice")
if _, err := r.Invoke(ctx, "chat.robot",
chatRobotIn{Repo: "alpha", Events: "checkin landing.accepted"}); err != nil {
t.Fatal(err)
}
if got := store.ChatRobot("alpha"); got != "checkin landing.accepted" {
t.Fatalf("stored %q", got)
}
_, err := r.Invoke(ctx, "chat.robot", chatRobotIn{Repo: "alpha", Events: "nonsense.event"})
if err == nil {
t.Fatal("chat.robot accepted a pattern no event can match")
}
if !errors.Is(err, ops.ErrInvalid) {
t.Fatalf("refusal is %v, want an invalid-request error", err)
}
// The refused write left the stored patterns alone.
if got := store.ChatRobot("alpha"); got != "checkin landing.accepted" {
t.Fatalf("a refused write changed the stored patterns to %q", got)
}
// Empty is off.
if _, err := r.Invoke(ctx, "chat.robot", chatRobotIn{Repo: "alpha"}); err != nil {
t.Fatal(err)
}
out, err := r.Invoke(ctx, "chat.robot.get", chatRobotGetIn{Repo: "alpha"})
if err != nil {
t.Fatal(err)
}
if got := out.(chatRobotOut); got.On || got.Events != "" {
t.Fatalf("robot reads back %+v after being turned off", got)
}
}
// blockingAttention holds every publish open until it is released —
// the broker that is slow, or gone, while a PubAckWait runs out.
type blockingAttention struct {
fakeAttention
entered chan string
release chan struct{}
}
func newBlockingAttention() *blockingAttention {
return &blockingAttention{
fakeAttention: fakeAttention{sent: map[string][]AttentionEvent{}},
entered: make(chan string, 8),
release: make(chan struct{}),
}
}
func (b *blockingAttention) Attend(account string, e AttentionEvent) (uint64, error) {
b.entered <- account
<-b.release
return b.fakeAttention.Attend(account, e)
}
// RFC 0011 §5's order is "attention first", not "attention on the write
// path": a broker that has stopped acknowledging must cost the op
// nothing at all.
func TestFireNotifyDoesNotBlockOnAttention(t *testing.T) {
ba := newBlockingAttention()
defer close(ba.release)
d := Deps{Attention: ba}
done := make(chan struct{})
go func() {
defer close(done)
d.fireNotify(NotifyEvent{
Workflow: "quarry-mention", To: "alice",
Payload: map[string]any{"repo": "alpha", "from": "bob", "text": "hi @alice"},
})
}()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("fireNotify blocked on the attention publish")
}
select {
case <-ba.entered:
case <-time.After(2 * time.Second):
t.Fatal("the attention publish never started")
}
}
// recordingNotify remembers when Trigger ran, so the order attention
// and notify observe can be asserted rather than assumed.
type orderNotify struct{ fired chan struct{} }
func (o *orderNotify) Trigger(context.Context, NotifyEvent) error {
close(o.fired)
return nil
}
func (o *orderNotify) Provision(context.Context) error { return nil }
// Attention is still written BEFORE notify is triggered for the same
// event — the whole reason an open tab hears from quarry rather than
// from a round trip through the companion.
func TestFireNotifyAttendsBeforeNotify(t *testing.T) {
ba := newBlockingAttention()
on := &orderNotify{fired: make(chan struct{})}
d := Deps{Attention: ba, Notify: on}
d.fireNotify(NotifyEvent{
Workflow: "quarry-mention", To: "alice",
Payload: map[string]any{"repo": "alpha", "from": "bob", "text": "hi @alice"},
})
<-ba.entered
select {
case <-on.fired:
t.Fatal("notify fired while the attention publish was still open")
case <-time.After(100 * time.Millisecond):
}
close(ba.release)
select {
case <-on.fired:
case <-time.After(2 * time.Second):
t.Fatal("notify never fired after the attention publish finished")
}
}
// A personal delivery goes only to somebody who can read the repository
// it is about. Thread subscriptions and replied-to authors are stored,
// and a site admin auto-subscribed while the flag reached every
// repository kept being sent a private repository's post bodies by
// inbox, email and push (#318). Checked at delivery, so a stored
// subscription can never outlive the access it was made with.
func TestAPersonalDeliveryNeedsTheRepository(t *testing.T) {
fa := newFakeAttention()
fn := &fakeNotify{}
d := Deps{Attention: fa, Notify: fn,
CanRead: func(ctx context.Context, repo string) bool {
return repo != "vault" || ops.User(ctx) == "ada"
}}
for _, e := range []NotifyEvent{
{Workflow: "quarry-forum-reply", To: "root",
Payload: map[string]any{"repo": "vault", "title": "private", "text": "a private body"}},
{Workflow: "quarry-mention", To: "root",
Payload: map[string]any{"repo": "vault", "text": "@root a private body"}},
{Workflow: "quarry-forum-reply", To: "ada",
Payload: map[string]any{"repo": "vault", "title": "private", "text": "a private body"}},
{Workflow: "quarry-forum-reply", To: "root",
Payload: map[string]any{"repo": "alpha", "title": "public", "text": "a public body"}},
} {
d.fireNotify(e)
}
fa.waitFor(t, "ada", 1)
got := fa.waitFor(t, "root", 1)
for _, e := range got {
if e.Repo == "vault" {
t.Errorf("FAIL: a reader of nothing in vault was told about it: %+v", e)
}
}
deadline := time.Now().Add(2 * time.Second)
for len(fn.snapshot()) < 2 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
for _, e := range fn.snapshot() {
if e.To == "root" && e.Payload["repo"] == "vault" {
t.Errorf("FAIL: vault's post went to root by notify: %+v", e)
}
}
if n := len(fn.snapshot()); n != 2 {
t.Errorf("FAIL: notify was asked %d times, want ada's vault reply and root's alpha reply", n)
}
}