// attention_test.go proves RFC 0011 Phase 2 end to end with **notify
// deliberately absent**. That is the whole point of the local attention
// subject: a forge running no notify companion at all still tells
// people the things addressed to them, and it tells them first.
package harness
import (
"bufio"
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"net/url"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/chat"
"craftdesign.group/quarry/internal/events"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/zitcli"
)
// harnessAttention is the composition root's adapter, restated here:
// the ops name api.AttentionEvent and the event plane speaks its own,
// so something has to carry one across.
type harnessAttention struct{ ev *events.Server }
func (a harnessAttention) Attend(account string, e api.AttentionEvent) (uint64, error) {
return a.ev.Attend(account, events.AttentionEvent{
Kind: e.Kind, Repo: e.Repo, From: e.From,
Summary: e.Summary, Link: e.Link, At: e.At, TxnID: e.TxnID,
})
}
func (a harnessAttention) RecentAttention(account string, limit int) ([]api.AttentionEvent, error) {
evs, err := a.ev.RecentAttention(account, limit)
if err != nil {
return nil, err
}
out := make([]api.AttentionEvent, 0, len(evs))
for _, e := range evs {
out = append(out, api.AttentionEvent{Seq: e.Seq, Kind: e.Kind, Repo: e.Repo,
From: e.From, Summary: e.Summary, Link: e.Link, At: e.At,
TxnID: e.TxnID, Ref: e.Ref})
}
return out, nil
}
func (a harnessAttention) TailAttention(account string, from uint64) (<-chan api.AttentionEvent, func()) {
src, stop := a.ev.TailAttention(account, from)
out := make(chan api.AttentionEvent)
done := make(chan struct{})
go func() {
defer close(out)
for e := range src {
select {
case out <- api.AttentionEvent{Seq: e.Seq, Kind: e.Kind, Repo: e.Repo,
From: e.From, Summary: e.Summary, Link: e.Link, At: e.At,
TxnID: e.TxnID, Ref: e.Ref}:
case <-done:
return
}
}
}()
var once sync.Once
return out, func() { once.Do(func() { stop(); close(done) }) }
}
func (a harnessAttention) AttentionPending(account string, since uint64) (int, error) {
return a.ev.AttentionPending(account, since)
}
func (a harnessAttention) AttentionTip(account string) (uint64, error) {
return a.ev.AttentionTip(account)
}
type attnForge struct {
srv *httptest.Server
odb *oltp.DB
room *chat.Room
ev *events.Server
sync func() (int, error)
path string
}
// buildAttentionForge stands up a real forge with a real store, a real
// chat plane, the attention stream and the activity robot — and NO
// notify plane at all.
func buildAttentionForge(t *testing.T, robotPatterns string) attnForge {
t.Helper()
dataDir := t.TempDir()
ev, err := events.Start(filepath.Join(dataDir, "nats"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ev.Close)
odb, err := oltp.Open(filepath.Join(dataDir, "oltp.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { odb.Close() })
room, err := chat.New(ev.Conn(), ev.JS())
if err != nil {
t.Fatal(err)
}
// As serve wires it: the robot's own lines are remembered, which is
// what keeps them out of the unread count.
room.Robots = odb
if robotPatterns != "" {
if err := odb.SetChatRobot("alpha", robotPatterns); err != nil {
t.Fatal(err)
}
}
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
if err := chat.NewRobot(room, ev.JS(),
func(repo string) string { return odb.ChatRobot(repo) },
func(repo string) (string, bool) { return repo, true }).Start(ctx); err != nil {
t.Fatal(err)
}
deps := api.Deps{
Version: "0.0.0-harness",
Repos: api.NewRepos(nil),
Chat: room,
// Notify is deliberately nil: the attention plane must not need it.
Attention: harnessAttention{ev: ev},
AttnRead: odb,
ChatRobots: odb,
Accounts: harnessAccounts{db: odb},
Forum: odb,
ChatRefs: odb,
}
deps.CapsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
repoPath := buildStore(t, "attnseed")
ix, err := index.Open(filepath.Join(dataDir, "alpha.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
cli := zitcli.New(repoPath)
pipe := ingest.NewPipeline(cli, ev, ix, "alpha", "harness")
deps.Repos.Put("alpha", &api.Repo{Index: ix, Zit: cli, Sync: pipe.Sync, Code: "alpha"})
// Twice: the first round rebuilds from the store's census and the
// second settles the once-per-process tip check, and neither is
// news — a rebuild that raised events would announce the whole
// history. Only a round after these takes the bus, which is the
// path the robot is a consumer of.
for i := 0; i < 2; i++ {
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
}
for _, acct := range []struct{ name, caps string }{
{"dan", "C3h"}, {"eve", "C3h"},
} {
if err := odb.AddAccount(acct.name, "pw", false); err != nil {
t.Fatal(err)
}
if err := odb.SetCaps(acct.name, "alpha", acct.caps); err != nil {
t.Fatal(err)
}
}
verify := func(repo, user, pass string, need ops.Auth) api.Decision {
if user == "" || pass != "pw" {
return api.Denied
}
return api.Allowed
}
r := ops.NewRegistry()
if err := api.RegisterAll(r, deps); err != nil {
t.Fatal(err)
}
mux, err := api.RouterWithCaps(r, "0.0.0-harness", deps.CapsFor, deps.CanRead, verify)
if err != nil {
t.Fatal(err)
}
srv := httptest.NewServer(mux)
t.Cleanup(srv.Close)
return attnForge{srv: srv, odb: odb, room: room, ev: ev, sync: pipe.Sync, path: repoPath}
}
type attnCaller struct {
t *testing.T
srv *httptest.Server
user string
}
func (c attnCaller) do(method, path string, form url.Values) (int, []byte) {
c.t.Helper()
var body io.Reader
if form != nil {
body = strings.NewReader(form.Encode())
}
req, err := http.NewRequest(method, c.srv.URL+path, body)
if err != nil {
c.t.Fatal(err)
}
if form != nil {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
}
req.SetBasicAuth(c.user, "pw")
resp, err := c.srv.Client().Do(req)
if err != nil {
c.t.Fatal(err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
return resp.StatusCode, b
}
func (c attnCaller) unread() int {
c.t.Helper()
code, body := c.do("GET", "/api/v1/me/inbox/count", nil)
if code != http.StatusOK {
c.t.Fatalf("inbox.count: %d %s", code, body)
}
var out struct {
Unread int `json:"unread"`
}
json.Unmarshal(body, &out)
return out.Unread
}
// attnStream is one open attention stream. A single goroutine owns the
// body for the stream's whole life and hands each event to events, so a
// wait that gives up never leaves a second reader behind on the same
// connection.
type attnStream struct {
events <-chan map[string]any
}
// openAttention opens the account's attention stream once the op has
// sent `event: live`, which it sends only after its tail is attached (so
// nothing the test causes next can be missed). The router's own ": live"
// comment is NOT that signal: it is written before the op runs, and a
// mention posted on it could land before the consumer existed and never
// be delivered.
func openAttention(t *testing.T, srv *httptest.Server, user string) (*attnStream, func()) {
t.Helper()
client := &http.Client{Transport: &http.Transport{DisableKeepAlives: true}}
req, _ := http.NewRequest("GET", srv.URL+"/api/v1/me/attention", nil)
req.SetBasicAuth(user, "pw")
resp, err := client.Do(req)
if err != nil {
t.Fatal(err)
}
rd := bufio.NewReader(resp.Body)
for {
line, err := rd.ReadString('\n')
if err != nil {
resp.Body.Close()
t.Fatalf("%s's attention stream ended before its tail attached: %v", user, err)
}
if line == "event: live\n" {
break
}
}
events := make(chan map[string]any)
done := make(chan struct{})
go func() {
defer close(events)
for {
line, err := rd.ReadString('\n')
if err != nil {
return
}
if !strings.HasPrefix(line, "data: ") {
continue
}
var m map[string]any
// An event is an object; the stream's control frames
// ("live", the heartbeat) carry none and are not one.
if json.Unmarshal([]byte(strings.TrimSpace(strings.TrimPrefix(line, "data: "))), &m) != nil || m == nil {
continue
}
select {
case events <- m:
case <-done:
return
}
}
}()
var once sync.Once
return &attnStream{events: events}, func() {
once.Do(func() {
close(done)
resp.Body.Close()
})
}
}
// untilDeadline is how long a wait for something that IS coming may
// take: the rest of the test binary's time, less a margin to fail with
// a message rather than a timeout panic. Late in a long package run the
// margin would leave nothing, so it is dropped under ten seconds and the
// wait never falls below one second.
func untilDeadline(t *testing.T) time.Duration {
dl, ok := t.Deadline()
if !ok {
return time.Duration(1<<63 - 1)
}
left := time.Until(dl)
if left > 10*time.Second {
left -= 5 * time.Second
}
return max(left, time.Second)
}
// awaitEvent reads the next event. It waits for the event, not a
// window: the tail is attached before the producer runs, so an event
// that is coming arrives, and only the test binary's own deadline ends
// the wait for one that never does.
func awaitEvent(t *testing.T, s *attnStream, what string) map[string]any {
t.Helper()
got, ok := nextEvent(s, untilDeadline(t))
if !ok {
t.Fatalf("%s: the stream ended or the test ran out of time with no event", what)
}
return got
}
// nextEvent takes one event, or reports that none arrived within d.
func nextEvent(s *attnStream, d time.Duration) (map[string]any, bool) {
select {
case m, open := <-s.events:
return m, open
case <-time.After(d):
return nil, false
}
}
// A mention reaches the mentioned account and nobody else — with no
// notify configured anywhere in the process.
func TestMentionReachesTheAttentionStreamWithoutNotify(t *testing.T) {
f := buildAttentionForge(t, "")
eveStream, closeEve := openAttention(t, f.srv, "eve")
defer closeEve()
danStream, closeDan := openAttention(t, f.srv, "dan")
defer closeDan()
dan := attnCaller{t: t, srv: f.srv, user: "dan"}
if code, body := dan.do("POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"@eve can you look at this"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d %s", code, body)
}
got := awaitEvent(t, eveStream, "eve's attention stream")
if got["from"] != "dan" || got["repo"] != "alpha" {
t.Fatalf("eve's event is %v", got)
}
if _, ok := got["link"]; !ok {
t.Errorf("the event carries no link: %v", got)
}
// The author hears nothing about their own mention.
if e, ok := nextEvent(danStream, 2*time.Second); ok {
t.Fatalf("dan heard about his own mention: %v", e)
}
}
// A reply reaches the thread's subscriber — the author, auto-subscribed
// by their own post — and not the replier.
func TestForumReplyReachesTheSubscriberOnly(t *testing.T) {
f := buildAttentionForge(t, "")
dan := attnCaller{t: t, srv: f.srv, user: "dan"}
eve := attnCaller{t: t, srv: f.srv, user: "eve"}
code, body := dan.do("POST", "/api/v1/r/alpha/forum",
url.Values{"title": {"a question"}, "body": {"how does this work"}})
if code != http.StatusOK {
t.Fatalf("forum.post: %d %s", code, body)
}
var posted struct {
Hash string `json:"hash"`
}
json.Unmarshal(body, &posted)
if posted.Hash == "" {
t.Fatalf("forum.post returned no hash: %s", body)
}
danStream, closeDan := openAttention(t, f.srv, "dan")
defer closeDan()
eveStream, closeEve := openAttention(t, f.srv, "eve")
defer closeEve()
code, body = eve.do("POST", "/api/v1/r/alpha/forum/"+posted.Hash,
url.Values{"body": {"like this"}})
if code != http.StatusOK {
t.Fatalf("forum.reply: %d %s", code, body)
}
got := awaitEvent(t, danStream, "the thread author's attention stream")
if got["from"] != "eve" {
t.Fatalf("dan's event is %v", got)
}
if e, ok := nextEvent(eveStream, 2*time.Second); ok {
t.Fatalf("the replier heard about her own reply: %v", e)
}
}
// The count is the local tail after the read pointer, and marking read
// clears it.
func TestAttentionReadClearsTheInboxCount(t *testing.T) {
f := buildAttentionForge(t, "")
dan := attnCaller{t: t, srv: f.srv, user: "dan"}
eve := attnCaller{t: t, srv: f.srv, user: "eve"}
eveStream, closeEve := openAttention(t, f.srv, "eve")
defer closeEve()
for i := 0; i < 2; i++ {
if code, body := dan.do("POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"@eve ping " + strconv.Itoa(i)}}); code != http.StatusOK {
t.Fatalf("chat.post: %d %s", code, body)
}
}
// The publish is best-effort and off the op's critical path, so the
// count is read once both events have been delivered, not assumed
// immediate: a delivered event is one the broker already stores.
for i := 0; i < 2; i++ {
awaitEvent(t, eveStream, "eve's attention stream")
}
if n := eve.unread(); n != 2 {
t.Fatalf("eve's unread is %d, want 2", n)
}
if n := dan.unread(); n != 0 {
t.Fatalf("dan's unread is %d, want 0 — it is eve who was named", n)
}
if code, body := eve.do("POST", "/api/v1/me/attention/read", url.Values{}); code != http.StatusOK {
t.Fatalf("attention.read: %d %s", code, body)
}
if n := eve.unread(); n != 0 {
t.Fatalf("eve's unread is %d after marking read, want 0", n)
}
}
// The robot announces a matching event in the channel, and the line it
// writes never lights the unread badge.
func TestRobotAnnouncesCheckinsWithoutRaisingUnread(t *testing.T) {
f := buildAttentionForge(t, "checkin")
dan := attnCaller{t: t, srv: f.srv, user: "dan"}
// A person's own line first, so "unread" has something true to
// count and the robot's absence from it is a real distinction.
code, body := dan.do("POST", "/api/v1/r/alpha/chat", url.Values{"text": {"hello"}})
if code != http.StatusOK {
t.Fatalf("chat.post: %d %s", code, body)
}
var hello struct {
Seq uint64 `json:"seq"`
}
if err := json.Unmarshal(body, &hello); err != nil || hello.Seq == 0 {
t.Fatalf("chat.post returned no seq: %s", body)
}
before, err := f.room.Unread("alpha", 0)
if err != nil {
t.Fatal(err)
}
if before != 1 {
t.Fatalf("unread before the robot spoke is %d, want 1", before)
}
// Everything after the person's line, by sequence: a tail that
// starts from a known seq cannot miss what comes after it.
lines, stop := f.room.Tail("alpha", hello.Seq)
defer stop()
// A check-in on the store, folded by ONE explicit round of the real
// pipeline, which is what raises checkin.created on the vocabulary
// stream.
commitInStore(t, f, "robot.txt", "a change the robot should announce")
if n, err := f.sync(); err != nil || n == 0 {
t.Fatalf("the round after the commit folded %d artifacts, err=%v", n, err)
}
timeout := time.After(untilDeadline(t))
var robot chat.Message
for robot.Seq == 0 {
select {
case m, open := <-lines:
if !open {
t.Fatal("the channel's tail ended before the robot spoke")
}
if m.Kind == chat.KindRobot {
robot = m
}
case <-timeout:
t.Fatal("the robot announced nothing for a check-in it was asked about")
}
}
if robot.Origin != "robot" {
t.Errorf("the robot's line has origin %q", robot.Origin)
}
if !strings.Contains(robot.Text, "check-in") {
t.Errorf("the robot's line reads %q", robot.Text)
}
after, err := f.room.Unread("alpha", 0)
if err != nil {
t.Fatal(err)
}
if after != before {
t.Fatalf("unread went from %d to %d: the robot's line counted", before, after)
}
}
// commitInStore writes one check-in through zit itself — the robot must
// react to what the FOLD raises, not to something a test published on
// the vocabulary subject by hand.
func commitInStore(t *testing.T, f attnForge, name, body string) {
t.Helper()
if err := os.WriteFile(filepath.Join(f.path, name), []byte(body+"\n"), 0o644); err != nil {
t.Fatal(err)
}
zrun(t, f.path, "add", name)
zrun(t, f.path, "commit", "-m", "robot: "+name)
}