// inbox_test.go proves RFC 0011 Phase 3a end to end: quarry's inbox is
// notify's inbox and quarry's own attention stream shown as ONE list,
// a mark reaches both halves, a push subscription reaches notify's
// credentials route, a muted repository stops being SENT without
// stopping being noticed, and a preferences write lands as notify
// preferences.
//
// The fake notify here implements the routes notify shipped: notify#300
// (mark), notify#301 (count and pagination), notify#302 (web push) and
// notify#305 (the subscriber-preferences pair keyed by trigger
// identifier, the per-device credential DELETE, and the trigger
// identifier carried on every message row).
package harness
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"net/url"
"path/filepath"
"sort"
"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/notify"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/web"
"craftdesign.group/quarry/internal/zitcli"
)
// phase3Req is one call the Phase 3 fake saw, with its query — which
// matters here in a way it did not for Phase 0: unsubscribe names its
// endpoint in the query string and nowhere else.
type phase3Req struct {
method, path, query string
body map[string]any
}
// phase3Notify answers the Phase 3 routes and remembers every call.
// Messages are stateful so a mark actually changes what a later list
// returns — a fake that accepts a mark and keeps serving the same
// unread page would let a broken "clear the inbox" pass.
type phase3Notify struct {
mu sync.Mutex
reqs []phase3Req
msgs []notify.Message
// prefsGlobal and prefsWorkflows are the two sections of notify's
// preferences DTO, held apart because that is how notify holds
// them: a PUT MERGES onto what is stored, and a fake that replaced
// the lot would let a client that clobbers the global section pass.
prefsGlobal map[string]bool
prefsWorkflows map[string]map[string]bool
// sawGlobalWrite records a PUT that named the global section at
// all. quarry's matrix has no cell for it and must never write it.
sawGlobalWrite bool
// fixedPage, when set, makes the message route serve exactly that
// many rows and IGNORE ?limit= — which is what notify does today
// (internal/inboxapi serves the whole list) and what forces quarry
// to cross the companion's own page boundaries.
fixedPage int
// noWorkflows makes GET /v1/workflows fail, which is a companion
// quarry cannot resolve workflow identities against.
noWorkflows bool
// credentialDeleteFails makes the webpush credential DELETE answer
// 500 — a companion that did NOT drop the credential, which is the
// only answer that leaves quarry owing a retry.
credentialDeleteFails bool
// noPreferences makes the preferences pair answer 404: quarry then
// has no idea what notify holds and must say so.
noPreferences bool
// legacyRows serves message rows with no WorkflowIdentifier, which
// is what a notify older than notify#305 answers.
legacyRows bool
// owner scopes a seeded message to one subscriber, keyed by message
// id. A fake that served every subscriber the same inbox could not
// catch quarry addressing the wrong one.
owner map[string]string
}
// phase3Workflows is what notify's GET /v1/workflows answers: the pair
// nothing else in its API relates — the workflow ULID a Message carries
// and the trigger identifier quarry named it by. The ULIDs here are
// deliberately NOT quarry's spelling, so a reason derived from a
// message can only be right by going through this map.
var phase3Workflows = map[string]string{
"01JQMENTION0000000000000": "quarry-mention",
"01JQREPLY000000000000000": "quarry-forum-reply",
"01JQLANDING00000000000000": "quarry-landing",
"01JQACTIVITY0000000000000": "quarry-activity",
}
func (f *phase3Notify) record(r *http.Request) map[string]any {
var body map[string]any
if b, _ := io.ReadAll(r.Body); len(b) > 0 {
json.Unmarshal(b, &body)
}
f.reqs = append(f.reqs, phase3Req{r.Method, r.URL.Path, r.URL.RawQuery, body})
return body
}
func (f *phase3Notify) ServeHTTP(w http.ResponseWriter, r *http.Request) {
f.mu.Lock()
defer f.mu.Unlock()
body := f.record(r)
who := r.URL.Query().Get("subscriberId")
switch {
case r.Method == "GET" && r.URL.Path == "/v1/workflows" && f.noWorkflows:
w.WriteHeader(http.StatusInternalServerError)
case r.Method == "GET" && r.URL.Path == "/v1/workflows":
out := make([]map[string]any, 0, len(phase3Workflows))
for id, ident := range phase3Workflows {
out = append(out, map[string]any{"id": id, "name": ident, "triggerIdentifier": ident})
}
json.NewEncoder(w).Encode(out)
case r.Method == "GET" && r.URL.Path == "/v1/messages":
var live []notify.Message
for _, m := range f.msgs {
if m.Archived || !f.mine(m.ID, who) {
continue
}
live = append(live, m)
}
sort.SliceStable(live, func(a, b int) bool { return live[a].CreatedAt > live[b].CreatedAt })
unread := 0
for _, m := range live {
if !m.Read {
unread++
}
}
// notify#301's pagination, with an opaque cursor of its own —
// quarry must never be able to read one, which is the point of
// spelling it differently here than quarry spells its own.
from := 0
if c := r.URL.Query().Get("before"); c != "" {
n, err := strconv.Atoi(strings.TrimPrefix(c, "off-"))
if err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
from = n
}
if from > len(live) {
from = len(live)
}
page := live[from:]
size := 0
if f.fixedPage > 0 {
size = f.fixedPage
} else if l, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && l > 0 {
size = l
}
if size > 0 && size < len(page) {
page = page[:size]
}
next := ""
if from+len(page) < len(live) {
next = "off-" + strconv.Itoa(from+len(page))
}
if page == nil {
page = []notify.Message{}
}
if f.legacyRows {
// An older notify: the row names its workflow by ULID and
// nothing else, so the reason has to come from the map.
stripped := make([]notify.Message, 0, len(page))
for _, m := range page {
m.WorkflowIdentifier = ""
stripped = append(stripped, m)
}
page = stripped
}
json.NewEncoder(w).Encode(map[string]any{
"data": page, "next": next, "unreadCount": unread, "unseenCount": unread,
})
case r.Method == "GET" && r.URL.Path == "/v1/messages/count":
n := 0
for _, m := range f.msgs {
if !m.Read && !m.Archived && f.mine(m.ID, who) {
n++
}
}
json.NewEncoder(w).Encode(map[string]any{"count": n})
case r.Method == "POST" && r.URL.Path == "/v1/messages/mark":
as, _ := body["as"].(string)
all, _ := body["all"].(bool)
who, _ = body["subscriberId"].(string)
named := map[string]bool{}
if ids, ok := body["ids"].([]any); ok {
for _, id := range ids {
named[id.(string)] = true
}
}
for i := range f.msgs {
if !f.mine(f.msgs[i].ID, who) {
continue
}
if !all && !named[f.msgs[i].ID] {
continue
}
switch as {
case "seen":
f.msgs[i].Seen = true
case "read":
f.msgs[i].Read, f.msgs[i].Seen = true, true
case "archived":
f.msgs[i].Archived, f.msgs[i].Read = true, true
}
}
w.WriteHeader(http.StatusOK)
case r.Method == "GET" && r.URL.Path == "/v1/integrations/webpush/key":
json.NewEncoder(w).Encode(map[string]any{"publicKey": "BFakeVapidPublicKey"})
case strings.HasSuffix(r.URL.Path, "/preferences") && f.noPreferences:
w.WriteHeader(http.StatusNotFound)
case r.Method == "PUT" && strings.HasSuffix(r.URL.Path, "/preferences"):
if _, ok := body["global"]; ok {
f.sawGlobalWrite = true
}
// Read-merge-write, like notify's own handler: a channel the
// request leaves out keeps whatever was stored.
if workflows, ok := body["workflows"].(map[string]any); ok {
for wf, raw := range workflows {
channels, _ := raw.(map[string]any)
if f.prefsWorkflows[wf] == nil {
f.prefsWorkflows[wf] = map[string]bool{}
}
for ch, on := range channels {
f.prefsWorkflows[wf][ch], _ = on.(bool)
}
}
}
fallthrough
case r.Method == "GET" && strings.HasSuffix(r.URL.Path, "/preferences"):
json.NewEncoder(w).Encode(map[string]any{
"global": f.prefsGlobal, "workflows": f.prefsWorkflows})
case r.Method == "POST" && r.URL.Path == "/v1/events/trigger":
// A trigger does not merely get recorded: notify DELIVERS it,
// and an in_app workflow's delivery is a message in the
// subscriber's inbox carrying the trigger's transaction id.
// Without that, the fake cannot show the shape a live forge has
// — the same event standing in both halves of RFC 0011 §6.
f.deliver(body)
w.WriteHeader(http.StatusOK)
case r.Method == "POST" && (r.URL.Path == "/v1/subscribers" ||
r.URL.Path == "/v1/topics" || r.URL.Path == "/v1/workflows"),
r.Method == "POST" && strings.HasSuffix(r.URL.Path, "/subscribers"),
r.Method == "DELETE" && strings.Contains(r.URL.Path, "/v1/topics/"),
r.Method == "PUT" && strings.HasSuffix(r.URL.Path, "/credentials"):
// Triggers, subscribers, topics, workflow creation and the
// credential PUT only need recording and a success.
w.WriteHeader(http.StatusOK)
case r.Method == "DELETE" && strings.HasSuffix(r.URL.Path, "/credentials/webpush"):
if f.credentialDeleteFails {
w.WriteHeader(http.StatusInternalServerError)
return
}
w.WriteHeader(http.StatusNoContent)
default:
// Everything else is a route notify does not have. Answering
// 200 to an unmatched path is how a fake lets a typo'd URL —
// or a route quarry invented — pass as working.
w.WriteHeader(http.StatusNotFound)
}
}
// deliver turns one recorded trigger into the in-app message notify
// would have written for it. Topic triggers deliver nothing here: a
// topic fans out to a membership this fake does not keep, and the
// person-addressed half is what the inbox merge is about.
//
// The caller holds f.mu.
func (f *phase3Notify) deliver(body map[string]any) {
to, _ := body["to"].(map[string]any)
who, _ := to["subscriberId"].(string)
name, _ := body["name"].(string)
txn, _ := body["transactionId"].(string)
if who == "" || name == "" {
return
}
var workflowID string
for id, ident := range phase3Workflows {
if ident == name {
workflowID = id
}
}
if workflowID == "" {
return
}
payload, _ := body["payload"].(map[string]any)
subject, _ := payload["text"].(string)
if s, ok := payload["title"].(string); ok && s != "" {
subject = s
}
if f.owner == nil {
f.owner = map[string]string{}
}
id := "delivered-" + strconv.Itoa(len(f.msgs)+1)
f.owner[id] = who
f.msgs = append(f.msgs, named([]notify.Message{{
ID: id, WorkflowID: workflowID, Subject: subject, TransactionID: txn,
CreatedAt: time.Now().UTC().Format(time.RFC3339Nano),
}})...)
}
// mine reports whether a seeded message belongs to who. An unassigned
// message belongs to everybody, which keeps the older tests short.
func (f *phase3Notify) mine(id, who string) bool {
if f.owner == nil || f.owner[id] == "" {
return true
}
return f.owner[id] == who
}
// seedFor seeds messages owned by one subscriber.
func (f *phase3Notify) seedFor(who string, msgs ...notify.Message) {
f.mu.Lock()
defer f.mu.Unlock()
if f.owner == nil {
f.owner = map[string]string{}
}
for _, m := range msgs {
f.owner[m.ID] = who
}
f.msgs = append(f.msgs, named(msgs)...)
}
// named fills in each row's trigger identifier the way notify does,
// resolving it at read time from the workflow the row names. Seeds say
// only the ULID, so this is what makes every seeded row a CURRENT
// notify's row rather than an older one's.
func named(msgs []notify.Message) []notify.Message {
out := make([]notify.Message, 0, len(msgs))
for _, m := range msgs {
if m.WorkflowIdentifier == "" {
m.WorkflowIdentifier = phase3Workflows[m.WorkflowID]
}
out = append(out, m)
}
return out
}
func (f *phase3Notify) find(method, path string) []phase3Req {
f.mu.Lock()
defer f.mu.Unlock()
var out []phase3Req
for _, r := range f.reqs {
if r.method == method && r.path == path {
out = append(out, r)
}
}
return out
}
func (f *phase3Notify) seed(msgs ...notify.Message) {
f.mu.Lock()
defer f.mu.Unlock()
f.msgs = append(f.msgs, named(msgs)...)
}
// waitForPhase3 polls until (method, path) has been called want times,
// since every notify call quarry makes off an op is best-effort async.
func waitForPhase3(t *testing.T, f *phase3Notify, method, path string, want int) {
t.Helper()
for i := 0; i < 100; i++ {
if len(f.find(method, path)) >= want {
return
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("never saw %d %s %s", want, method, path)
}
// harnessInbox is cmd/quarry's inboxCounter, restated: the ops name
// api.InboxPlane and the client speaks notify's wire.
type harnessInbox struct {
client *notify.Client
names *notify.WorkflowNames
}
// harnessRemote is cmd/quarry's `remote`, restated: notify's "I did not
// answer that" becomes the ops layer's own word for it.
func harnessRemote(err error) error {
if err != nil && errors.Is(err, notify.ErrRouteMissing) {
return fmt.Errorf("%w: %w", api.ErrRemoteUnsupported, err)
}
return err
}
func (i harnessInbox) Unread(ctx context.Context, account string) (int, error) {
return i.client.UnreadCount(ctx, account)
}
func (i harnessInbox) List(ctx context.Context, account, before string, limit int) ([]api.InboxMessage, string, error) {
msgs, next, _, _, err := i.client.Messages(ctx, account, before, limit)
if err != nil {
return nil, "", err
}
reasons := i.names.Identify(ctx, msgs)
out := make([]api.InboxMessage, 0, len(msgs))
for _, m := range msgs {
out = append(out, api.InboxMessage{ID: m.ID, Workflow: reasons[m.ID], Subject: m.Subject,
Body: m.Content, Link: m.Link(), At: m.CreatedAt, Read: m.Read, Seen: m.Seen,
TxnID: m.TransactionID})
}
return out, next, nil
}
func (i harnessInbox) Mark(ctx context.Context, account string, ids []string, all bool, as string) error {
return i.client.MarkMessages(ctx, account, ids, all, as)
}
func (i harnessInbox) Preferences(ctx context.Context, account string) (map[string]map[string]bool, bool, error) {
prefs, err := i.client.Preferences(ctx, account)
if err != nil {
return nil, false, nil
}
// Only the per-workflow section is surfaced: quarry's matrix has a
// row per family and no cell for notify's global layer.
return prefs.Workflows, true, nil
}
func (i harnessInbox) DigestWindows(ctx context.Context, account string) (map[string]api.DigestWindow, error) {
return nil, nil
}
func (i harnessInbox) SetPreferences(ctx context.Context, account string, prefs map[string]map[string]bool,
_ map[string]*api.DigestWindow) error {
return harnessRemote(i.client.SetPreferences(ctx, account, prefs, nil))
}
func (i harnessInbox) PushKey(ctx context.Context) (string, error) {
return i.client.WebPushKey(ctx)
}
func (i harnessInbox) PushSubscribe(ctx context.Context, account, endpoint, p256dh, auth string) error {
return i.client.PutWebPush(ctx, account, notify.PushSubscription{
Endpoint: endpoint, Keys: notify.PushKeys{P256dh: p256dh, Auth: auth}})
}
func (i harnessInbox) PushUnsubscribe(ctx context.Context, account, endpoint string) error {
return harnessRemote(i.client.DeleteWebPush(ctx, account, endpoint))
}
// harnessPushDevices adapts forge OLTP to api.PushDeviceStore.
type harnessPushDevices struct{ db *oltp.DB }
func (p harnessPushDevices) AddPushDevice(account, endpoint, ua string) error {
return p.db.AddPushDevice(account, endpoint, ua)
}
func (p harnessPushDevices) DeletePushDevice(account, endpoint string) error {
return p.db.DeletePushDevice(account, endpoint)
}
func (p harnessPushDevices) MarkPushDevicePendingDelete(account, endpoint string) error {
return p.db.MarkPushDevicePendingDelete(account, endpoint)
}
func (p harnessPushDevices) PendingPushDeletes(account string) ([]string, error) {
return p.db.PendingPushDeletes(account)
}
func (p harnessPushDevices) PushDevices(account string) ([]api.PushDevice, error) {
rows, err := p.db.PushDevices(account)
if err != nil {
return nil, err
}
out := make([]api.PushDevice, 0, len(rows))
for _, r := range rows {
out = append(out, api.PushDevice{EndpointHash: r.EndpointHash, UA: r.UA, At: r.At,
PendingDelete: r.PendingDelete})
}
return out, nil
}
type inboxForge struct {
srv *httptest.Server
// web is the same registry served through the HTML plane: the
// preferences page, the inbox page and the push routes a browser
// and a service worker reach with a cookie and no Basic credential.
web *httptest.Server
odb *oltp.DB
ev *events.Server
fake *phase3Notify
}
// harnessWebSessions is the browser half of the same accounts the API
// half authenticates with Basic. The token is the account name, which
// is all a test needs and nothing a deployment would do.
type harnessWebSessions struct{ odb *oltp.DB }
func (h harnessWebSessions) Login(user, password string) (string, time.Time, bool) {
if password != "pw" {
return "", time.Time{}, false
}
return user, time.Now().Add(time.Hour), true
}
func (h harnessWebSessions) Resolve(token string) (string, bool) { return token, token != "" }
func (h harnessWebSessions) Logout(string) {}
func (h harnessWebSessions) Allow(user, repo string, need ops.Auth) bool {
return user != ""
}
// harnessWebPrefs is the forge-local preference store as the web layer
// reads it — the same table the ops write through, because the stream
// and the preferences page must not disagree about a switch.
type harnessWebPrefs struct{ odb *oltp.DB }
func (h harnessWebPrefs) Pref(account, key string) string {
return h.odb.AccountPref(account, key)
}
func (h harnessWebPrefs) SetPref(account, key, value string) error {
return h.odb.SetAccountPref(account, key, value)
}
// buildInboxForge is a real forge with BOTH halves wired — the notify
// companion and the local attention stream — which is the only
// configuration in which the merge, the two-sided mark and the mute can
// be observed at all.
func buildInboxForge(t *testing.T) inboxForge {
t.Helper()
fake := &phase3Notify{
// notify reports every global channel explicitly, an absent one
// meaning enabled.
prefsGlobal: map[string]bool{"in_app": true, "push": true,
"email": true, "sms": true, "chat": true},
prefsWorkflows: map[string]map[string]bool{},
}
nsrv := httptest.NewServer(fake)
t.Cleanup(nsrv.Close)
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)
}
nclient := notify.New(nsrv.URL, "k")
deps := api.Deps{
Version: "0.0.0-harness",
Repos: api.NewRepos(nil),
Chat: room,
Notify: &harnessNotify{client: nclient},
Inbox: harnessInbox{client: nclient, names: notify.NewWorkflowNames(nclient)},
Attention: harnessAttention{ev: ev},
AttnRead: odb,
Prefs: odb,
PushDevices: harnessPushDevices{db: odb},
Accounts: harnessAccounts{db: odb},
Forum: odb,
}
deps.CapsFor = func(account, repo string) api.CapSet { return odb.EffectiveCaps(account, repo) }
repoPath := buildStore(t, "inboxseed")
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"})
if _, err := pipe.Sync(); err != nil {
t.Fatal(err)
}
for _, name := range []string{"dan", "eve"} {
if err := odb.AddAccount(name, "pw", false); err != nil {
t.Fatal(err)
}
if err := odb.SetCaps(name, "alpha", "C3h"); 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)
websrv := httptest.NewServer(web.HandlerWith(r, nil, nil, []web.Option{
web.WithSessions(harnessWebSessions{odb: odb}),
web.WithPrefs(harnessWebPrefs{odb: odb}),
}, web.Verify(verify)))
t.Cleanup(websrv.Close)
return inboxForge{srv: srv, web: websrv, odb: odb, ev: ev, fake: fake}
}
// inboxCall performs one authenticated API call, form-encoded.
func inboxCall(t *testing.T, f inboxForge, user, method, path string, form url.Values) (int, []byte) {
t.Helper()
var body io.Reader
if form != nil {
body = strings.NewReader(form.Encode())
}
req, err := http.NewRequest(method, f.srv.URL+path, body)
if err != nil {
t.Fatal(err)
}
if form != nil {
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
}
req.SetBasicAuth(user, "pw")
resp, err := f.srv.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
b, _ := io.ReadAll(resp.Body)
return resp.StatusCode, b
}
type inboxListBody struct {
Rows []struct {
ID string `json:"id"`
Source string `json:"source"`
Reason string `json:"reason"`
Repo string `json:"repo"`
Summary string `json:"summary"`
Link string `json:"link"`
At string `json:"at"`
Read bool `json:"read"`
Seq uint64 `json:"seq"`
} `json:"rows"`
Next string `json:"next"`
Read uint64 `json:"read"`
}
func listInbox(t *testing.T, f inboxForge, user string) inboxListBody {
t.Helper()
code, body := inboxCall(t, f, user, "GET", "/api/v1/me/inbox", nil)
if code != http.StatusOK {
t.Fatalf("inbox.list: %d %s", code, body)
}
var out inboxListBody
if err := json.Unmarshal(body, &out); err != nil {
t.Fatalf("inbox.list body: %v (%s)", err, body)
}
return out
}
// One list, both sources, newest first, each row carrying the reason it
// arrived — which is the whole claim of RFC 0011 §1's `inbox.list`.
//
// The notify messages carry workflow ULIDs, because that is what
// notify's Message.WorkflowID is. A reason that came out right can only
// have come through GET /v1/workflows: quarry's own spelling
// ("quarry-mention") appears nowhere on the message.
func TestInboxListMergesNotifyAndAttention(t *testing.T) {
f := buildInboxForge(t)
f.fake.seed(
notify.Message{ID: "m-old", WorkflowID: "01JQREPLY000000000000000", Subject: "eve replied",
CreatedAt: "2026-08-01T00:00:00Z"},
notify.Message{ID: "m-new", WorkflowID: "01JQMENTION0000000000000", Subject: "eve mentioned you",
CreatedAt: "2026-08-03T00:00:00Z",
Extras: json.RawMessage(`{"redirect":{"url":"/r/alpha/chat"}}`)},
)
// One local attention event, stamped between the two notify ones,
// so a merge that simply concatenated would be visible as an order.
if _, err := f.ev.Attend("dan", events.AttentionEvent{
Kind: "landing", Repo: "alpha", From: "eve", Summary: "landing accepted",
Link: "/r/alpha/landings/1", At: "2026-08-02T00:00:00Z",
}); err != nil {
t.Fatal(err)
}
got := listInbox(t, f, "dan")
if len(got.Rows) != 3 {
t.Fatalf("rows = %+v", got.Rows)
}
wantIDs := []string{"m-new", "", "m-old"} // "" = the local row, id checked below
wantSources := []string{"notify", "local", "notify"}
wantReasons := []string{"mention", "landing", "reply"}
for i, row := range got.Rows {
if row.Source != wantSources[i] || row.Reason != wantReasons[i] {
t.Fatalf("row %d = %+v, want source %q reason %q", i, row, wantSources[i], wantReasons[i])
}
if wantIDs[i] != "" && row.ID != wantIDs[i] {
t.Fatalf("row %d id = %q, want %q", i, row.ID, wantIDs[i])
}
}
local := got.Rows[1]
if !strings.HasPrefix(local.ID, "local:") || local.Seq == 0 {
t.Fatalf("the local row must be addressable by sequence: %+v", local)
}
if local.Repo != "alpha" || local.Link != "/r/alpha/landings/1" {
t.Fatalf("the local row lost its repo or link: %+v", local)
}
// The notify row's link comes out of its extras redirect.
if got.Rows[0].Link != "/r/alpha/chat" {
t.Fatalf("notify row link = %q", got.Rows[0].Link)
}
}
// "Clear everything" must reach BOTH halves: notify's bulk route for
// the messages, and the forward-only attention pointer for the local
// events. Clearing one and not the other leaves a badge that will not
// go out.
func TestInboxMarkAllReachesNotifyAndThePointer(t *testing.T) {
f := buildInboxForge(t)
f.fake.seed(notify.Message{ID: "m1", WorkflowID: "01JQMENTION0000000000000", Subject: "hi",
CreatedAt: "2026-08-01T00:00:00Z"})
seq, err := f.ev.Attend("dan", events.AttentionEvent{
Kind: "mention", Repo: "alpha", Summary: "hey", At: "2026-08-02T00:00:00Z"})
if err != nil {
t.Fatal(err)
}
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/inbox/mark",
url.Values{"all": {"true"}, "as": {"read"}})
if code != http.StatusOK {
t.Fatalf("inbox.mark: %d %s", code, body)
}
var out struct {
OK bool `json:"ok"`
Seq uint64 `json:"seq"`
}
json.Unmarshal(body, &out)
if !out.OK || out.Seq != seq {
t.Fatalf("mark out = %+v, want the local pointer at %d", out, seq)
}
if got := f.odb.AttentionReadSeq("dan"); got != seq {
t.Fatalf("attention pointer = %d, want %d", got, seq)
}
marks := f.fake.find("POST", "/v1/messages/mark")
if len(marks) != 1 {
t.Fatalf("notify saw %d bulk marks, want 1", len(marks))
}
if marks[0].body["all"] != true || marks[0].body["subscriberId"] != "dan" ||
marks[0].body["as"] != "read" {
t.Fatalf("mark body = %+v", marks[0].body)
}
// And the count the badge reads is now zero from notify's side.
code, body = inboxCall(t, f, "dan", "GET", "/api/v1/me/inbox/count", nil)
if code != http.StatusOK {
t.Fatalf("inbox.count: %d %s", code, body)
}
var count struct {
Unread int `json:"unread"`
}
json.Unmarshal(body, &count)
if count.Unread != 0 {
t.Fatalf("unread after mark-all = %d", count.Unread)
}
}
// A "done" mark on a local id moves only the pointer, and a notify id
// goes only to notify: the split is by id spelling, not by a second
// field the page has to remember to set.
func TestInboxMarkSplitsIDsByOrigin(t *testing.T) {
f := buildInboxForge(t)
f.fake.seed(notify.Message{ID: "m1", WorkflowID: "01JQMENTION0000000000000", CreatedAt: "2026-08-01T00:00:00Z"})
seq, _ := f.ev.Attend("dan", events.AttentionEvent{Kind: "mention", At: "2026-08-02T00:00:00Z"})
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/inbox/mark",
url.Values{"ids": {"m1", "local:" + strconv.FormatUint(seq, 10)}, "as": {"done"}})
if code != http.StatusOK {
t.Fatalf("inbox.mark: %d %s", code, body)
}
marks := f.fake.find("POST", "/v1/messages/mark")
if len(marks) != 1 {
t.Fatalf("notify marks = %d", len(marks))
}
// "done" is quarry's word; notify's state is archived.
if marks[0].body["as"] != "archived" {
t.Fatalf("as = %v, want archived", marks[0].body["as"])
}
ids, _ := marks[0].body["ids"].([]any)
if len(ids) != 1 || ids[0] != "m1" {
t.Fatalf("a local id must never be forwarded to notify: %+v", ids)
}
if got := f.odb.AttentionReadSeq("dan"); got != seq {
t.Fatalf("pointer = %d, want %d", got, seq)
}
// An archived message leaves the list entirely.
if rows := listInbox(t, f, "dan").Rows; len(rows) != 1 || rows[0].Source != "local" {
t.Fatalf("rows after archive = %+v", rows)
}
}
// The browser's own PushSubscription JSON, posted verbatim, must reach
// notify's credentials route with the subscriber id — and leave quarry
// holding a device row but never the endpoint.
func TestPushSubscribeForwardsAndRecordsADevice(t *testing.T) {
f := buildInboxForge(t)
const endpoint = "https://push.example.test/abc123"
req, err := http.NewRequest("POST", f.srv.URL+"/api/v1/me/push", strings.NewReader(
`{"endpoint":"`+endpoint+`","expirationTime":null,"keys":{"p256dh":"PPP","auth":"AAA"},"ua":"Firefox"}`))
if err != nil {
t.Fatal(err)
}
req.Header.Set("Content-Type", "application/json")
req.SetBasicAuth("dan", "pw")
resp, err := f.srv.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode != http.StatusOK {
t.Fatalf("me.push.subscribe: %d %s", resp.StatusCode, body)
}
puts := f.fake.find("PUT", "/v1/subscribers/dan/credentials")
if len(puts) != 1 {
t.Fatalf("credentials PUTs = %d", len(puts))
}
if puts[0].body["providerId"] != "webpush" {
t.Fatalf("providerId = %v", puts[0].body["providerId"])
}
creds, _ := puts[0].body["credentials"].(map[string]any)
if creds["endpoint"] != endpoint {
t.Fatalf("credentials = %+v", creds)
}
keys, _ := creds["keys"].(map[string]any)
if keys["p256dh"] != "PPP" || keys["auth"] != "AAA" {
t.Fatalf("keys = %+v", keys)
}
devs, err := f.odb.PushDevices("dan")
if err != nil {
t.Fatal(err)
}
if len(devs) != 1 || devs[0].UA != "Firefox" {
t.Fatalf("devices = %+v", devs)
}
if devs[0].EndpointHash != oltp.PushEndpointHash(endpoint) {
t.Fatalf("the device row must key on a hash, not the endpoint: %+v", devs[0])
}
// Unsubscribing names the endpoint in the query and forgets the row.
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/push/delete",
url.Values{"endpoint": {endpoint}})
if code != http.StatusOK {
t.Fatalf("me.push.unsubscribe: %d %s", code, body)
}
dels := f.fake.find("DELETE", "/v1/subscribers/dan/credentials/webpush")
if len(dels) != 1 || !strings.Contains(dels[0].query, url.QueryEscape(endpoint)) {
t.Fatalf("delete = %+v", dels)
}
if devs, _ := f.odb.PushDevices("dan"); len(devs) != 0 {
t.Fatalf("devices after unsubscribe = %+v", devs)
}
}
// The push key is relayed, never held: quarry has no VAPID private
// half and the public one comes from notify on demand.
// A person who has never saved a preference has everything on: notify
// reads an absent channel as enabled, and quarry must not disagree
// with it just because it has no row of its own.
func TestPrefsDefaultToEverythingOn(t *testing.T) {
f := buildInboxForge(t)
code, body := inboxCall(t, f, "eve", "GET", "/api/v1/me/prefs", nil)
if code != http.StatusOK {
t.Fatalf("me.prefs: %d %s", code, body)
}
var out struct {
Families map[string]map[string]bool `json:"families"`
Channels []string `json:"channels"`
Toasts bool `json:"toasts"`
Desktop bool `json:"notify_desktop"`
Muted []string `json:"muted"`
}
json.Unmarshal(body, &out)
if len(out.Families) != 4 || len(out.Channels) != 3 {
t.Fatalf("matrix = %+v / %+v", out.Families, out.Channels)
}
for family, row := range out.Families {
for _, ch := range out.Channels {
if !row[ch] {
t.Fatalf("%s.%s defaulted off", family, ch)
}
}
}
// The two forge-local switches default OFF, which is what the
// attention stream itself does: a banner over the page somebody is
// reading, and a browser permission prompt, are both opt-in.
if out.Toasts || out.Desktop || len(out.Muted) != 0 {
t.Fatalf("forge-local defaults = %+v", out)
}
}
func TestPushKeyIsRelayed(t *testing.T) {
f := buildInboxForge(t)
code, body := inboxCall(t, f, "dan", "GET", "/api/v1/me/push/key", nil)
if code != http.StatusOK {
t.Fatalf("me.push.key: %d %s", code, body)
}
var out struct {
PublicKey string `json:"publicKey"`
}
json.Unmarshal(body, &out)
if out.PublicKey != "BFakeVapidPublicKey" {
t.Fatalf("publicKey = %q", out.PublicKey)
}
}
// A muted repository stops being SENT and does not stop being noticed:
// the attention event is still published (the person finds it in their
// own inbox when they look), and no trigger reaches notify.
func TestMutedRepoSkipsNotifyButNotAttention(t *testing.T) {
f := buildInboxForge(t)
// eve mutes alpha through the ops surface, not the store, so the
// preferences write is what the mute is proved from.
code, body := inboxCall(t, f, "eve", "POST", "/api/v1/me/prefs",
url.Values{"mute": {"alpha"}, "toasts": {"on"}})
if code != http.StatusOK {
t.Fatalf("me.prefs.set: %d %s", code, body)
}
before := len(f.fake.find("POST", "/v1/events/trigger"))
if code, body := inboxCall(t, f, "dan", "POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hey @eve"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d %s", code, body)
}
// The attention event still lands...
deadline := time.Now().Add(3 * time.Second)
var rows []struct {
Source string
Reason string
}
for time.Now().Before(deadline) {
got := listInbox(t, f, "eve")
rows = rows[:0]
for _, r := range got.Rows {
if r.Source == "local" {
rows = append(rows, struct{ Source, Reason string }{r.Source, r.Reason})
}
}
if len(rows) > 0 {
break
}
time.Sleep(50 * time.Millisecond)
}
if len(rows) != 1 || rows[0].Reason != "mention" {
t.Fatalf("a mute must not silence the attention stream: %+v", rows)
}
// ...and notify heard nothing at all about it.
time.Sleep(300 * time.Millisecond) // fireNotify is async; give a skipped trigger time to appear
for _, r := range f.fake.find("POST", "/v1/events/trigger")[before:] {
to, _ := r.body["to"].(map[string]any)
if to != nil && to["subscriberId"] == "eve" {
t.Fatalf("a muted repository still triggered notify: %+v", r.body)
}
}
}
// me.prefs.set writes the whole matrix to notify as per-workflow
// preference rows — an unticked channel is stored false, not simply
// left out, because notify reads an absent channel as ENABLED.
func TestPrefsSetWritesNotifyPreferences(t *testing.T) {
f := buildInboxForge(t)
// Desktop notifications on first, so the write below — which names
// `toasts` and not `notify_desktop` — has something to leave alone.
if code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/prefs",
url.Values{"notify_desktop": {"on"}}); code != http.StatusOK {
t.Fatalf("me.prefs.set: %d %s", code, body)
}
// `families` names the four being written. It is load-bearing for
// activity, which is being cleared: in form encoding an empty list
// emits nothing at all, so "activity: {}" arrives looking exactly
// like a caller who never mentioned activity. Naming it is how a
// caller says "this one, to nothing" rather than "not mine to
// touch" — see mePrefsSetIn.Families.
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/prefs", url.Values{
"families": {"mention", "reply", "landing", "activity"},
"mention": {"inapp", "push"},
"reply": {"inapp"},
"landing": {"inapp", "push", "email"},
"activity": {},
"toasts": {"off"},
})
if code != http.StatusOK {
t.Fatalf("me.prefs.set: %d %s", code, body)
}
puts := f.fake.find("PUT", "/v1/subscribers/dan/preferences")
if len(puts) == 0 {
t.Fatal("the matrix never reached notify")
}
// The LAST write is the one under test: the setup call above wrote
// one too. That one named no families, so it wrote none — which is
// why this assertion needs the write that names all four.
put := puts[len(puts)-1].body
if _, ok := put["global"]; ok {
t.Fatalf("quarry wrote notify's global section: %+v", put)
}
byWorkflow := map[string]map[string]any{}
workflows, _ := put["workflows"].(map[string]any)
for wf, raw := range workflows {
ch, _ := raw.(map[string]any)
byWorkflow[wf] = ch
}
want := map[string]map[string]bool{
"quarry-mention": {"in_app": true, "push": true, "email": false},
"quarry-forum-reply": {"in_app": true, "push": false, "email": false},
"quarry-landing": {"in_app": true, "push": true, "email": true},
"quarry-activity": {"in_app": false, "push": false, "email": false},
}
for workflow, channels := range want {
got, ok := byWorkflow[workflow]
if !ok {
t.Fatalf("no preference row for %s: %+v", workflow, byWorkflow)
}
for ch, on := range channels {
if got[ch] != on {
t.Fatalf("%s.%s = %v, want %v (row %+v)", workflow, ch, got[ch], on, got)
}
}
// The route is keyed by TRIGGER IDENTIFIER. A ULID-keyed entry
// would be a workflow notify has never heard of, and it fails
// the whole PUT closed with 422.
for id := range phase3Workflows {
if _, ok := byWorkflow[id]; ok {
t.Fatalf("the payload is keyed by ULID %s: %+v", id, byWorkflow)
}
}
}
// And the store agrees: an unticked cell is an EXPLICIT false
// override under the trigger identifier, not an absence notify
// would read back as enabled.
f.fake.mu.Lock()
reply := f.fake.prefsWorkflows["quarry-forum-reply"]
mention := f.fake.prefsWorkflows["quarry-mention"]
sawGlobal := f.fake.sawGlobalWrite
f.fake.mu.Unlock()
if on, ok := reply["push"]; !ok || on {
t.Fatalf("reply push = %v (present %v), want an explicit false: %+v", on, ok, reply)
}
if on, ok := mention["push"]; !ok || !on {
t.Fatalf("mention push = %v (present %v), want an explicit true: %+v", on, ok, mention)
}
if sawGlobal {
t.Fatal("quarry must never write notify's global preferences")
}
// And the round trip: me.prefs reads back what was just written,
// plus the forge-local switch nobody else knows about.
code, body = inboxCall(t, f, "dan", "GET", "/api/v1/me/prefs", nil)
if code != http.StatusOK {
t.Fatalf("me.prefs: %d %s", code, body)
}
var out struct {
Families map[string]map[string]bool `json:"families"`
Toasts bool `json:"toasts"`
Desktop bool `json:"notify_desktop"`
Push bool `json:"push"`
}
json.Unmarshal(body, &out)
if !out.Families["mention"]["push"] || out.Families["mention"]["email"] {
t.Fatalf("mention row = %+v", out.Families["mention"])
}
if out.Families["activity"]["inapp"] {
t.Fatalf("activity was turned off and read back on: %+v", out.Families["activity"])
}
if out.Toasts {
t.Fatal("toasts were turned off and read back on")
}
// An UNPOSTED switch is UNCHANGED. The write above named `toasts`
// and not `notify_desktop`, and desktop was turned on before it —
// so it must still be on. The reading this replaces — absence means
// off — meant a form that set one switch silently turned the other
// one off for whoever submitted it.
if !out.Desktop {
t.Fatal("a switch nobody mentioned must not be turned off")
}
if !out.Push {
t.Fatal("push must be configurable on a forge with notify wired")
}
}
// Turning push off for mentions and nothing else: the override lands
// under the TRIGGER IDENTIFIER as an explicit false, and reads back
// identically. An override notify stored under a key quarry cannot
// name, or as an absence notify resolves to ENABLED, is a box somebody
// unticked that keeps pushing.
func TestPrefsPushOffForMentionsIsAnExplicitOverride(t *testing.T) {
f := buildInboxForge(t)
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/prefs", url.Values{
"mention": {"inapp", "email"},
"reply": {"inapp", "push", "email"},
"landing": {"inapp", "push", "email"},
"activity": {"inapp", "push", "email"},
})
if code != http.StatusOK {
t.Fatalf("me.prefs.set: %d %s", code, body)
}
f.fake.mu.Lock()
stored := map[string]bool{}
for ch, on := range f.fake.prefsWorkflows["quarry-mention"] {
stored[ch] = on
}
f.fake.mu.Unlock()
on, ok := stored["push"]
if !ok {
t.Fatalf("no push override under quarry-mention: %+v", stored)
}
if on {
t.Fatalf("push came back on: %+v", stored)
}
code, body = inboxCall(t, f, "dan", "GET", "/api/v1/me/prefs", nil)
if code != http.StatusOK {
t.Fatalf("me.prefs: %d %s", code, body)
}
var out struct {
Families map[string]map[string]bool `json:"families"`
Unknown []string `json:"unknown"`
}
json.Unmarshal(body, &out)
if len(out.Unknown) != 0 {
t.Fatalf("a companion that answers leaves nothing unknown: %+v", out.Unknown)
}
if out.Families["mention"]["push"] {
t.Fatalf("mention push read back on: %+v", out.Families["mention"])
}
if !out.Families["mention"]["inapp"] || !out.Families["reply"]["push"] {
t.Fatalf("a cell nobody unticked came back off: %+v", out.Families)
}
}
// A page of rows that carry their own trigger identifier costs NO
// workflow lookup: the reason is on the message, and asking notify what
// it just told us is a round trip per page of inbox.
func TestInboxReasonsComeFromTheMessageNotTheWorkflowList(t *testing.T) {
f := buildInboxForge(t)
f.fake.seedFor("dan", notify.Message{ID: "m1", WorkflowID: "01JQMENTION0000000000000",
Subject: "eve mentioned you", CreatedAt: "2026-08-01T00:00:00Z"})
got := listInbox(t, f, "dan")
var reason string
for _, row := range got.Rows {
if row.ID == "m1" {
reason = row.Reason
}
}
if reason != "mention" {
t.Fatalf("reason = %q, want it read off the message", reason)
}
if calls := f.fake.find("GET", "/v1/workflows"); len(calls) != 0 {
t.Fatalf("the workflow list was asked %d times for a page that names its own workflows", len(calls))
}
}
// The same inbox against an OLDER notify, whose rows name their
// workflow by ULID alone: the map is the fallback that keeps the reason
// quarry's own word rather than an opaque id.
func TestInboxReasonsFallBackToTheWorkflowListForOlderRows(t *testing.T) {
f := buildInboxForge(t)
f.fake.mu.Lock()
f.fake.legacyRows = true
f.fake.mu.Unlock()
f.fake.seedFor("dan", notify.Message{ID: "m1", WorkflowID: "01JQMENTION0000000000000",
Subject: "eve mentioned you", CreatedAt: "2026-08-01T00:00:00Z"})
got := listInbox(t, f, "dan")
for _, row := range got.Rows {
if row.ID == "m1" && row.Reason != "mention" {
t.Fatalf("reason = %q, want it resolved through the workflow list", row.Reason)
}
}
if calls := f.fake.find("GET", "/v1/workflows"); len(calls) == 0 {
t.Fatal("a row with no identifier must still be resolved through the workflow list")
}
}
// The two switches and the muted set are EXPLICIT operations: naming
// one changes one, and naming neither changes neither.
func TestPrefsSetLeavesUnmentionedSettingsAlone(t *testing.T) {
f := buildInboxForge(t)
readPrefs := func() (bool, bool, []string) {
t.Helper()
code, body := inboxCall(t, f, "dan", "GET", "/api/v1/me/prefs", nil)
if code != http.StatusOK {
t.Fatalf("me.prefs: %d %s", code, body)
}
var out struct {
Toasts bool `json:"toasts"`
Desktop bool `json:"notify_desktop"`
Muted []string `json:"muted"`
}
json.Unmarshal(body, &out)
return out.Toasts, out.Desktop, out.Muted
}
set := func(form url.Values) {
t.Helper()
if code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/prefs", form); code != http.StatusOK {
t.Fatalf("me.prefs.set: %d %s", code, body)
}
}
set(url.Values{"toasts": {"on"}, "notify_desktop": {"off"}, "mute": {"alpha", "beta"}})
toasts, desktop, muted := readPrefs()
if !toasts || desktop {
t.Fatalf("toasts/desktop = %v/%v, want on/off", toasts, desktop)
}
if len(muted) != 2 || muted[0] != "alpha" || muted[1] != "beta" {
t.Fatalf("muted = %+v", muted)
}
// A write that mentions neither switch nor either list op leaves
// all three exactly as they were.
set(url.Values{"mention": {"inapp"}})
toasts, desktop, muted = readPrefs()
if !toasts || desktop || len(muted) != 2 {
t.Fatalf("an unrelated write changed things: %v/%v/%+v", toasts, desktop, muted)
}
// unmute is a list op too: it removes what it names and nothing else.
set(url.Values{"unmute": {"alpha"}})
if _, _, muted = readPrefs(); len(muted) != 1 || muted[0] != "beta" {
t.Fatalf("muted after unmute = %+v", muted)
}
// And the answer says which halves landed.
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/prefs", url.Values{"toasts": {"on"}})
if code != http.StatusOK {
t.Fatalf("me.prefs.set: %d %s", code, body)
}
var applied struct {
Applied struct {
Local bool `json:"local"`
Notify bool `json:"notify"`
} `json:"applied"`
}
json.Unmarshal(body, &applied)
if !applied.Applied.Local || !applied.Applied.Notify {
t.Fatalf("applied = %+v, want both halves on a forge that has both", applied.Applied)
}
}
// Paging the merged inbox must lose nothing and repeat nothing.
//
// The bug this pins: page one asked notify for `limit` messages, merged
// the local events in, threw away everything past the limit, and then
// handed back NOTIFY'S cursor for the whole page — so every notify
// message the merge truncated was stepped over and never shown again.
// Seven rows from two sources, interleaved in time, read three at a
// time, must come back as seven rows in one strict order.
func TestInboxPagesBothSourcesWithoutGapsOrDuplicates(t *testing.T) {
f := buildInboxForge(t)
at := func(day int) string { return "2026-08-0" + strconv.Itoa(day) + "T00:00:00Z" }
// Newest first once sorted: n1(10) l1(9) n2(8) l2(7) n3(6) l3(5) n4(4).
f.fake.seedFor("dan",
notify.Message{ID: "n1", WorkflowID: "01JQMENTION0000000000000", Subject: "n1", CreatedAt: "2026-08-10T00:00:00Z"},
notify.Message{ID: "n2", WorkflowID: "01JQMENTION0000000000000", Subject: "n2", CreatedAt: at(8)},
notify.Message{ID: "n3", WorkflowID: "01JQREPLY000000000000000", Subject: "n3", CreatedAt: at(6)},
notify.Message{ID: "n4", WorkflowID: "01JQREPLY000000000000000", Subject: "n4", CreatedAt: at(4)},
)
// Published oldest first, so the attention sequence rises with time
// — which is the assumption the local half of the cursor rests on.
for _, e := range []struct {
summary string
day int
}{
{"l3", 5}, {"l2", 7}, {"l1", 9},
} {
if _, err := f.ev.Attend("dan", events.AttentionEvent{
Kind: "landing", Repo: "alpha", Summary: e.summary, At: at(e.day),
}); err != nil {
t.Fatal(err)
}
}
var seen []string
before := ""
for page := 0; page < 6; page++ {
code, body := inboxCall(t, f, "dan", "GET",
"/api/v1/me/inbox?limit=3&before="+url.QueryEscape(before), nil)
if code != http.StatusOK {
t.Fatalf("page %d: %d %s", page, code, body)
}
var out inboxListBody
if err := json.Unmarshal(body, &out); err != nil {
t.Fatalf("page %d: %v (%s)", page, err, body)
}
if len(out.Rows) > 3 {
t.Fatalf("page %d served %d rows for a limit of 3", page, len(out.Rows))
}
for _, row := range out.Rows {
seen = append(seen, row.Summary)
}
if out.Next == "" {
break
}
if out.Next == before {
t.Fatalf("page %d handed back its own cursor", page)
}
before = out.Next
}
want := []string{"n1", "l1", "n2", "l2", "n3", "l3", "n4"}
if len(seen) != len(want) {
t.Fatalf("paged rows = %+v, want %+v", seen, want)
}
for i := range want {
if seen[i] != want[i] {
t.Fatalf("paged rows = %+v, want %+v", seen, want)
}
}
}
// A notify that FAILED to drop a credential must not leave quarry
// pretending it did. The device row stays, marked, and the next
// successful subscribe retries it. Forgetting it instead is how a
// browser keeps receiving pushes with nothing left in the interface to
// stop them.
func TestUnsubscribeKeepsTheDeviceUntilNotifyConfirms(t *testing.T) {
f := buildInboxForge(t)
f.fake.mu.Lock()
f.fake.credentialDeleteFails = true
f.fake.mu.Unlock()
const endpoint = "https://push.example.test/pending"
subscribe := func() {
t.Helper()
req, err := http.NewRequest("POST", f.srv.URL+"/api/v1/me/push", strings.NewReader(
`{"endpoint":"`+endpoint+`","keys":{"p256dh":"PPP","auth":"AAA"},"ua":"Firefox"}`))
if err != nil {
t.Fatal(err)
}
req.Header.Set("Content-Type", "application/json")
req.SetBasicAuth("dan", "pw")
resp, err := f.srv.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
t.Fatalf("me.push.subscribe: %d %s", resp.StatusCode, b)
}
}
subscribe()
code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/push/delete",
url.Values{"endpoint": {endpoint}})
if code != http.StatusOK {
t.Fatalf("me.push.unsubscribe: %d %s", code, body)
}
var out struct {
OK bool `json:"ok"`
Pending bool `json:"pending_remote_delete"`
Devices []struct {
PendingDelete bool `json:"pending_remote_delete"`
} `json:"devices"`
}
json.Unmarshal(body, &out)
if !out.OK || !out.Pending {
t.Fatalf("an unconfirmed delete must say so: %s", body)
}
if len(out.Devices) != 1 || !out.Devices[0].PendingDelete {
t.Fatalf("the device row must stay, marked: %s", body)
}
if pending, _ := f.odb.PendingPushDeletes("dan"); len(pending) != 1 || pending[0] != endpoint {
t.Fatalf("the retry needs the endpoint: %+v", pending)
}
// notify recovers; the next subscribe sweeps what it owed.
f.fake.mu.Lock()
f.fake.credentialDeleteFails = false
f.fake.mu.Unlock()
subscribe()
if pending, _ := f.odb.PendingPushDeletes("dan"); len(pending) != 0 {
t.Fatalf("the sweep left %+v owed", pending)
}
}
// The same claim as TestInboxPagesBothSourcesWithoutGapsOrDuplicates,
// but against a companion that serves a FIXED page size and ignores the
// limit quarry asks for — which is what notify actually does today. The
// merged page therefore crosses notify's own page boundaries, and the
// cursor has to rebase onto its `next` rather than count an offset
// against a page it has already left.
func TestInboxPagesAcrossNotifyPageBoundaries(t *testing.T) {
f := buildInboxForge(t)
f.fake.mu.Lock()
f.fake.fixedPage = 2
f.fake.mu.Unlock()
at := func(day int) string { return "2026-08-" + strconv.Itoa(day) + "T00:00:00Z" }
f.fake.seedFor("dan",
notify.Message{ID: "n1", WorkflowID: "01JQMENTION0000000000000", Subject: "n1", CreatedAt: at(20)},
notify.Message{ID: "n2", WorkflowID: "01JQMENTION0000000000000", Subject: "n2", CreatedAt: at(18)},
notify.Message{ID: "n3", WorkflowID: "01JQREPLY000000000000000", Subject: "n3", CreatedAt: at(16)},
notify.Message{ID: "n4", WorkflowID: "01JQREPLY000000000000000", Subject: "n4", CreatedAt: at(14)},
notify.Message{ID: "n5", WorkflowID: "01JQREPLY000000000000000", Subject: "n5", CreatedAt: at(12)},
notify.Message{ID: "n6", WorkflowID: "01JQREPLY000000000000000", Subject: "n6", CreatedAt: at(10)},
)
// Oldest published first, so the attention sequence rises with the
// stamp — the invariant the local half of the cursor rests on.
for _, e := range []struct {
summary string
day int
}{{"l3", 15}, {"l2", 17}, {"l1", 19}} {
if _, err := f.ev.Attend("dan", events.AttentionEvent{
Kind: "landing", Repo: "alpha", Summary: e.summary, At: at(e.day),
}); err != nil {
t.Fatal(err)
}
}
var seen []string
before := ""
for page := 0; page < 8; page++ {
code, body := inboxCall(t, f, "dan", "GET",
"/api/v1/me/inbox?limit=3&before="+url.QueryEscape(before), nil)
if code != http.StatusOK {
t.Fatalf("page %d: %d %s", page, code, body)
}
var out inboxListBody
if err := json.Unmarshal(body, &out); err != nil {
t.Fatalf("page %d: %v (%s)", page, err, body)
}
for _, row := range out.Rows {
seen = append(seen, row.Summary)
}
if out.Next == "" {
break
}
if out.Next == before {
t.Fatalf("page %d handed back its own cursor", page)
}
before = out.Next
}
want := []string{"n1", "l1", "n2", "l2", "n3", "l3", "n4", "n5", "n6"}
if len(seen) != len(want) {
t.Fatalf("paged rows = %+v, want %+v", seen, want)
}
for i := range want {
if seen[i] != want[i] {
t.Fatalf("paged rows = %+v, want %+v", seen, want)
}
}
}
// The badge counts BOTH halves. inbox.list shows notify's messages and
// quarry's own attention events as one list, and they are disjoint
// sources with disjoint ids and disjoint read state — so a badge that
// counts either one alone leaves rows nothing will ever clear it for.
// Both halves are non-zero here, and differ, so a revert to either-or
// cannot pass by coincidence.
func TestInboxCountAddsNotifyAndLocalUnread(t *testing.T) {
f := buildInboxForge(t)
f.fake.seedFor("dan",
notify.Message{ID: "c1", WorkflowID: "01JQMENTION0000000000000", CreatedAt: "2026-08-01T00:00:00Z"},
notify.Message{ID: "c2", WorkflowID: "01JQMENTION0000000000000", CreatedAt: "2026-08-02T00:00:00Z"},
)
for i := 0; i < 3; i++ {
if _, err := f.ev.Attend("dan", events.AttentionEvent{
Kind: "mention", Repo: "alpha", At: "2026-08-03T00:00:00Z"}); err != nil {
t.Fatal(err)
}
}
count := func() int {
t.Helper()
code, body := inboxCall(t, f, "dan", "GET", "/api/v1/me/inbox/count", nil)
if code != http.StatusOK {
t.Fatalf("inbox.count: %d %s", code, body)
}
var out struct {
Unread int `json:"unread"`
}
json.Unmarshal(body, &out)
return out.Unread
}
// 2 from notify + 3 local: neither half alone reaches five.
if got := count(); got != 5 {
t.Fatalf("unread = %d, want 5 (2 notify + 3 local)", got)
}
// Clearing one half leaves the other's count standing.
if code, body := inboxCall(t, f, "dan", "POST", "/api/v1/me/inbox/mark",
url.Values{"ids": {"c1", "c2"}, "as": {"read"}}); code != http.StatusOK {
t.Fatalf("inbox.mark: %d %s", code, body)
}
if got := count(); got != 3 {
t.Fatalf("unread after clearing notify = %d, want the 3 local ones", got)
}
}
// A companion that does not answer its preferences route leaves quarry
// with no idea what it holds. The matrix must then say UNKNOWN rather
// than render its default as if it were a setting somebody had chosen.
func TestPrefsReportUnknownWhenNotifyDoesNotAnswer(t *testing.T) {
f := buildInboxForge(t)
f.fake.mu.Lock()
f.fake.noPreferences = true
f.fake.mu.Unlock()
code, body := inboxCall(t, f, "eve", "GET", "/api/v1/me/prefs", nil)
if code != http.StatusOK {
t.Fatalf("me.prefs: %d %s", code, body)
}
var out struct {
Families map[string]map[string]bool `json:"families"`
Unknown []string `json:"unknown"`
}
json.Unmarshal(body, &out)
if len(out.Unknown) != len(out.Families) || len(out.Unknown) == 0 {
t.Fatalf("unknown = %+v, want every family (%+v)", out.Unknown, out.Families)
}
// And with the identities readable it is a real answer again.
// Decoded into a FRESH value: an absent `unknown` leaves the field
// of a reused one exactly as the previous body set it.
g := buildInboxForge(t)
code, body = inboxCall(t, g, "eve", "GET", "/api/v1/me/prefs", nil)
if code != http.StatusOK {
t.Fatalf("me.prefs: %d %s", code, body)
}
var healthy struct {
Families map[string]map[string]bool `json:"families"`
Unknown []string `json:"unknown"`
}
json.Unmarshal(body, &healthy)
if len(healthy.Unknown) != 0 {
t.Fatalf("unknown = %+v on a companion that answers", healthy.Unknown)
}
if len(healthy.Families) != 4 {
t.Fatalf("families = %+v", healthy.Families)
}
}
// The defect this file's fake could not previously show: with notify
// configured, ONE mention is written to both halves of RFC 0011 §6 —
// quarry publishes the attention event and triggers notify with the
// same idempotency key — so the inbox listed it twice and the badge
// counted it twice. Paired on that key, it is one row and one count.
//
// End to end through the real chat op, the real fireNotify, the real
// attention stream and a notify that delivers what it is triggered
// with, because every one of those is where the key has to survive.
func TestOneMentionRendersAndCountsOnceAcrossBothHalves(t *testing.T) {
f := buildInboxForge(t)
before := inboxCount(t, f, "eve")
if code, body := inboxCall(t, f, "dan", "POST", "/api/v1/r/alpha/chat",
url.Values{"text": {"hey @eve about the landing"}}); code != http.StatusOK {
t.Fatalf("chat.post: %d %s", code, body)
}
waitForPhase3(t, f.fake, "POST", "/v1/events/trigger", 1)
// Both halves have to have landed before the merge is worth
// asserting on: the notify row arrives through the trigger above,
// the attention event through its own publish.
var rows []inboxListBody
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
got := listInbox(t, f, "eve")
if len(got.Rows) > 0 {
rows = append(rows[:0], got)
if got.Rows[0].Seq > 0 {
break
}
}
time.Sleep(50 * time.Millisecond)
}
if len(rows) == 0 {
t.Fatal("the mention reached neither half of eve's inbox")
}
got := rows[0]
if len(got.Rows) != 1 {
t.Fatalf("eve's inbox has %d rows for one mention, want 1: %+v", len(got.Rows), got.Rows)
}
row := got.Rows[0]
if row.Source != "notify" {
t.Errorf("the surviving row came from %q, want notify's (it holds the read state)", row.Source)
}
if row.Seq == 0 {
t.Errorf("the row lost the attention sequence, so a mark cannot move the pointer: %+v", row)
}
if row.Reason != "mention" {
t.Errorf("reason = %q, want mention", row.Reason)
}
// One mention, one count — the +2 the deployed forge showed is the
// bug, and +1 is the whole assertion.
if after := inboxCount(t, f, "eve"); after != before+1 {
t.Fatalf("inbox.count went %d -> %d for one mention, want %d", before, after, before+1)
}
// And marking that one row clears both halves: notify is told its
// own id, and the local pointer moves to the twin's sequence.
code, body := inboxCall(t, f, "eve", "POST", "/api/v1/me/inbox/mark",
url.Values{"ids": {row.ID}, "as": {"read"}})
if code != http.StatusOK {
t.Fatalf("inbox.mark: %d %s", code, body)
}
if got := f.odb.AttentionReadSeq("eve"); got != row.Seq {
t.Fatalf("attention pointer = %d after marking the paired row, want %d", got, row.Seq)
}
marks := f.fake.find("POST", "/v1/messages/mark")
if len(marks) != 1 {
t.Fatalf("notify saw %d marks, want 1", len(marks))
}
ids, _ := marks[0].body["ids"].([]any)
if len(ids) != 1 || strings.Contains(ids[0].(string), "local:") {
t.Fatalf("notify was handed %v, want its own id alone", marks[0].body["ids"])
}
if after := inboxCount(t, f, "eve"); after != before {
t.Fatalf("inbox.count = %d after the mark, want it back at %d", after, before)
}
}
// inboxCount reads the badge's number through the op.
func inboxCount(t *testing.T, f inboxForge, user string) int {
t.Helper()
code, body := inboxCall(t, f, user, "GET", "/api/v1/me/inbox/count", nil)
if code != http.StatusOK {
t.Fatalf("inbox.count: %d %s", code, body)
}
var out struct {
Unread int `json:"unread"`
}
if err := json.Unmarshal(body, &out); err != nil {
t.Fatalf("inbox.count body: %v (%s)", err, body)
}
return out.Unread
}