// 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, 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
}
// openAttention opens the account's attention stream and returns a
// reader over it, after the server has flushed its opening comment (so
// the consumer is live before the test causes anything).
func openAttention(t *testing.T, srv *httptest.Server, user string) (*bufio.Reader, 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)
if _, err := rd.ReadString('\n'); err != nil { // ": live"
t.Fatal(err)
}
return rd, func() { resp.Body.Close() }
}
// nextEvent reads one data: frame, or reports that none arrived.
func nextEvent(rd *bufio.Reader, d time.Duration) (map[string]any, bool) {
type res struct {
m map[string]any
ok bool
}
ch := make(chan res, 1)
go func() {
for {
line, err := rd.ReadString('\n')
if err != nil {
ch <- res{}
return
}
if strings.HasPrefix(line, "data: ") {
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 {
ch <- res{m, true}
return
}
}
}
}()
select {
case r := <-ch:
return r.m, r.ok
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, ok := nextEvent(eveStream, 10*time.Second)
if !ok {
t.Fatal("eve's attention stream delivered nothing")
}
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, ok := nextEvent(danStream, 10*time.Second)
if !ok {
t.Fatal("the thread's author heard nothing about the reply")
}
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"}
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 polled rather than assumed to be immediate.
deadline := time.Now().Add(10 * time.Second)
for eve.unread() < 2 && time.Now().Before(deadline) {
time.Sleep(100 * time.Millisecond)
}
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.
if code, body := dan.do("POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hello"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d %s", code, 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)
}
// A check-in on the store, folded by the real pipeline, which is
// what raises checkin.created on the vocabulary stream.
commitInStore(t, f, "robot.txt", "a change the robot should announce")
var robot chat.Message
deadline := time.Now().Add(20 * time.Second)
for time.Now().Before(deadline) {
if _, err := f.sync(); err != nil {
t.Fatal(err)
}
msgs, _, herr := f.room.History("alpha", 0, 50)
if herr != nil {
t.Fatal(herr)
}
for _, m := range msgs {
if m.Kind == chat.KindRobot {
robot = m
}
}
if robot.Seq != 0 {
break
}
time.Sleep(250 * time.Millisecond)
}
if robot.Seq == 0 {
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)
}