// inbox.go is RFC 0011 §6 at the ops seam: an inbox a person can page
// and CLEAR (`inbox.list`, `inbox.mark`), the channel preferences that
// decide what reaches them (`me.prefs`, `me.prefs.set`), and the three
// calls a browser needs to receive a push with no tab open
// (`me.push.key`, `me.push.subscribe`, `me.push.unsubscribe`).
//
// The division of labour is RFC 0011's: notify owns notifications,
// quarry owns the attention stream and the forge-local switches nobody
// else can know about (a muted repository, a toast preference, which
// browsers a person turned push on in). This file merges the two into
// the one list a person reads, and splits a mark back out to both.
//
// Like every "me" op, these carry no Caps: the router resolves a
// capability set against the repository named in the path, and these
// name none. Auth Token is the real gate.
package api
import (
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"log"
"net/url"
"sort"
"strconv"
"strings"
"time"
"craftdesign.group/quarry/internal/ops"
)
// ErrRemoteUnsupported is the ops layer's word for "the notification
// companion did not answer that": a preferences call it replied 404 or
// 405 to. It is not a bug in the caller, and the ops below degrade
// around it — reporting which half of a write landed — rather than
// failing an operation on a half quarry cannot reach.
//
// The composition root maps notify's own sentinel onto this one, so
// internal/api keeps naming what it needs without importing a client.
var ErrRemoteUnsupported = errors.New("the notification companion does not serve this yet")
// InboxMessage is one notify notification in this package's own terms
// — stated here rather than imported, for AttentionEvent's reason: the
// ops layer names what it needs and the composition root owns the
// transport.
type InboxMessage struct {
ID string
// Workflow is what sent it, and the only thing on a message that
// says WHY it arrived. `inbox.list` turns it into a reason.
Workflow string
Subject string
Body string
Link string
At string
Read bool
Seen bool
}
// PushDevice is one browser a person turned push on in. quarry holds no
// endpoint (see oltp.PushDevice) — this is the list a preferences page
// shows so somebody can turn one of them off.
type PushDevice struct {
EndpointHash string `json:"endpoint_hash"`
UA string `json:"ua,omitempty"`
At string `json:"at,omitempty"`
// PendingDelete is a device somebody asked to remove that the
// notification companion has not confirmed dropping yet. Shown so a
// page can say "removal pending" rather than list it as live.
PendingDelete bool `json:"pending_remote_delete,omitempty"`
}
// InboxPlane is the notify companion as the person-addressed ops need
// it. nil is a forge with no notify: `inbox.count` falls back to the
// local attention tail, `inbox.list` shows only local events, and the
// push and preference ops answer that this forge has no such thing
// rather than pretending a setting took.
type InboxPlane interface {
// Unread counts the account's unread in-app notifications.
Unread(ctx context.Context, account string) (int, error)
// List is one page, newest first: the messages, and the cursor for
// the next page ("" when this was the last).
List(ctx context.Context, account, before string, limit int) ([]InboxMessage, string, error)
// Mark moves messages to "read" or "archived" — either the named
// ids or, with all, the account's whole inbox.
Mark(ctx context.Context, account string, ids []string, all bool, as string) error
// Preferences answers workflow -> channel -> enabled, keyed by
// trigger identifier. An absent workflow or channel is ENABLED,
// which is notify's own rule.
//
// resolved reports whether the companion answered at all. When it
// is false, a family with no row is UNKNOWN rather than on: nothing
// was read, and rendering "on" would tell somebody a channel is
// enabled on the strength of a lookup that never landed.
Preferences(ctx context.Context, account string) (prefs map[string]map[string]bool, resolved bool, err error)
// SetPreferences writes the whole matrix.
SetPreferences(ctx context.Context, account string, prefs map[string]map[string]bool) error
// PushKey is the VAPID public key a browser subscribes with.
PushKey(ctx context.Context) (string, error)
// PushSubscribe stores one browser's subscription.
PushSubscribe(ctx context.Context, account, endpoint, p256dh, auth string) error
// PushUnsubscribe drops one endpoint's subscription.
PushUnsubscribe(ctx context.Context, account, endpoint string) error
}
// AccountPrefStore is the forge-local half of a person's preferences:
// the switches notify has no opinion about (toasts, desktop
// notifications) and the per-repository mute quarry implements by not
// triggering at all. nil means this forge keeps no preferences.
type AccountPrefStore interface {
AccountPref(account, key string) string
SetAccountPref(account, key, value string) error
}
// PushDeviceStore remembers which browsers an account enabled push in.
// nil means the preferences page lists no devices — the subscriptions
// still work, since they live in notify.
type PushDeviceStore interface {
AddPushDevice(account, endpoint, ua string) error
DeletePushDevice(account, endpoint string) error
PushDevices(account string) ([]PushDevice, error)
// MarkPushDevicePendingDelete keeps a device row that notify has
// NOT confirmed dropping, remembering the endpoint so the delete
// can be retried. See me.push.unsubscribe for why forgetting the
// row instead would leave a phone buzzing with nothing to stop it.
MarkPushDevicePendingDelete(account, endpoint string) error
// PendingPushDeletes lists those endpoints, newest first.
PendingPushDeletes(account string) ([]string, error)
}
// prefMute is the account_pref key holding the muted repositories, one
// JSON array rather than a row per repository: the whole set is read on
// every notify trigger, and one key lookup beats a scan.
const prefMute = "mute"
// prefToasts and prefDesktop are the two forge-local switches. Both
// default OFF when unset, which is what the attention stream itself
// does (internal/web: it emits a banner or a desktop notification only
// for an account whose preference reads "on") and what RFC 0011 §5
// requires of the browser prompt — a permission dialog belongs to a
// control somebody clicked. A default of ON here would make
// /me/notifications draw two switches the stream ignores.
const (
prefToasts = "toasts"
prefDesktop = "notify_desktop"
)
// prefFamilies maps RFC 0011 §6's four preference FAMILIES to the
// workflows that carry them. A family is what a person recognizes ("a
// reply") — the workflow name is quarry's own plumbing, and a
// preferences page must not make somebody learn it.
var prefFamilies = []struct{ family, workflow string }{
{"mention", "quarry-mention"},
{"reply", "quarry-forum-reply"},
{"landing", "quarry-landing"},
{"activity", "quarry-activity"},
}
// prefChannels maps quarry's channel names to notify's. Two spellings
// exist because "in_app" is notify's wire token and an HTML form field
// named with an underscore reads as somebody else's vocabulary leaking
// into ours.
var prefChannels = map[string]string{"inapp": "in_app", "push": "push", "email": "email"}
// prefChannelOrder is the matrix's column order — a map would render
// the page differently on every load.
var prefChannelOrder = []string{"inapp", "push", "email"}
// workflowReason turns a workflow's TRIGGER IDENTIFIER — quarry's own
// slug, the thing it names a workflow by and triggers it with — into
// the one-word reason an inbox row shows.
//
// It is emphatically NOT keyed by what a notify message carries:
// Message.WorkflowID is the workflow ROW id (a ULID). The composition
// root reads the identifier notify puts on the message, falling back to
// notify's workflow list, before a row reaches here.
var workflowReason = map[string]string{
"quarry-mention": "mention",
"quarry-forum-reply": "reply",
"quarry-landing": "landing",
"quarry-activity": "activity",
"quarry-error": "error",
"quarry-slow": "slow",
}
// reasonOf reads that map, with two fallbacks. A quarry workflow added
// later still reads sensibly (its slug minus the prefix). Anything else
// — an id that could not be resolved because notify was unreachable —
// gets a GENERIC word rather than a ULID: an inbox row that says
// "01J0AA…" is worse than one that says "notification", and the row is
// still listed, which is the point of never blocking a page on the map.
func reasonOf(workflow string) string {
if r, ok := workflowReason[workflow]; ok {
return r
}
if rest, ok := strings.CutPrefix(workflow, "quarry-"); ok && rest != "" {
return rest
}
return "notification"
}
// localIDPrefix marks an inbox row that came from quarry's own
// attention stream rather than notify. `inbox.mark` splits its ids back
// apart on it, so one list and one mark form serve both sources without
// the page having to keep them in separate buckets.
const localIDPrefix = "local:"
func localID(seq uint64) string { return localIDPrefix + strconv.FormatUint(seq, 10) }
// splitInboxIDs separates notify message ids from local attention
// sequences. An id that claims to be local but is not a number is
// dropped rather than forwarded to notify: it is quarry's own spelling
// gone wrong, and notify would 404 on it.
func splitInboxIDs(ids []string) (notify []string, seqs []uint64) {
for _, id := range ids {
if rest, ok := strings.CutPrefix(id, localIDPrefix); ok {
if n, err := strconv.ParseUint(rest, 10, 64); err == nil {
seqs = append(seqs, n)
}
continue
}
if id != "" {
notify = append(notify, id)
}
}
return notify, seqs
}
// inboxCursor is `inbox.list`'s page position, and it names BOTH
// sources because the page is a merge of both. A cursor that carried
// only notify's position lost mail: page one asked notify for `limit`
// messages, merged local events in, threw the overflow away, and handed
// back notify's cursor for the WHOLE page — so every notify message the
// merge truncated was skipped and never shown again.
//
// - Notify/Skip address notify's half: its own opaque page cursor,
// plus how many rows of that page this reader has already been
// given. Skip is needed because notify's cursor addresses a PAGE,
// not a message, and the merge routinely consumes part of one.
// - Local/LocalSet address quarry's half by attention sequence: the
// next page takes events with Seq strictly below Local. Unset on a
// first page, which means "no bound yet".
// - The two Done flags say a source is exhausted, which "" cannot:
// an empty notify cursor is its FIRST page.
//
// Opaque on the wire (base64url) so a page treats it as a bookmark and
// nobody builds one by hand out of somebody else's sequence numbers.
type inboxCursor struct {
Notify string
Skip int
Local uint64
LocalSet bool
NotifyDone bool
LocalDone bool
}
func (c inboxCursor) encode() string {
if c == (inboxCursor{}) {
return ""
}
// The notify half is escaped: it is somebody else's opaque token and
// may hold the separators this encoding uses.
fields := []string{
"n:" + url.QueryEscape(c.Notify),
"o:" + strconv.Itoa(c.Skip),
}
if c.LocalSet {
fields = append(fields, "l:"+strconv.FormatUint(c.Local, 10))
}
if c.NotifyDone {
fields = append(fields, "nd:1")
}
if c.LocalDone {
fields = append(fields, "ld:1")
}
return base64.RawURLEncoding.EncodeToString([]byte(strings.Join(fields, "|")))
}
// decodeInboxCursor reads what encode wrote. Anything it cannot read is
// a FIRST page rather than an error: a stale or hand-edited bookmark
// should show somebody their inbox, not a failure they cannot act on.
func decodeInboxCursor(s string) inboxCursor {
var c inboxCursor
if s == "" {
return c
}
raw, err := base64.RawURLEncoding.DecodeString(s)
if err != nil {
return inboxCursor{}
}
for _, field := range strings.Split(string(raw), "|") {
key, val, ok := strings.Cut(field, ":")
if !ok {
return inboxCursor{}
}
switch key {
case "n":
v, err := url.QueryUnescape(val)
if err != nil {
return inboxCursor{}
}
c.Notify = v
case "o":
n, err := strconv.Atoi(val)
if err != nil || n < 0 || n > inboxMaxSkip {
// Out of range is a cursor this code never issued.
// Treated as unreadable, which is a FIRST page — the
// same answer a stale bookmark gets, and never a number
// that reaches notify's ?limit=.
return inboxCursor{}
}
c.Skip = n
case "l":
n, err := strconv.ParseUint(val, 10, 64)
if err != nil {
return inboxCursor{}
}
c.Local, c.LocalSet = n, true
case "nd":
c.NotifyDone = val == "1"
case "ld":
c.LocalDone = val == "1"
default:
return inboxCursor{}
}
}
return c
}
// inboxMaxSkip is the hard ceiling on a cursor's in-page offset. Skip
// is arithmetic on a number that arrives from OUTSIDE — a cursor is a
// URL parameter anybody can write — and it is added to the page limit
// before that limit is handed to notify. Unclamped, `o:2000000000` asks
// the companion for two billion rows. Rebasing (see notifyWindow) keeps
// the honest value below one page, so anything near this ceiling is
// already a cursor nobody issued.
const inboxMaxSkip = 1000
// inboxLocalScan bounds how deep into the attention window one page
// reaches for local events. The window is bounded already (that is what
// `inbox.recent` serves whole), and the alternative — a store query per
// page — would put a cursor into a seam that has none.
const inboxLocalScan = 500
// inboxMaxNotifyPages bounds the walk that skips past rows already
// served. Bounded rather than unbounded because a notify that answers
// every page with an empty body and a next cursor would otherwise spin.
const inboxMaxNotifyPages = 8
// encodeMuteList and decodeMuteList are the muted set's storage form: a
// JSON array, so a repository name is a whole element. The previous
// space-separated spelling could not hold a name with a space in it and
// would have silently split one into two mutes that mute nothing.
// decodeMuteList still reads it, because an existing forge has rows in
// that shape and an upgrade must not un-mute anybody.
func encodeMuteList(repos []string) string {
if len(repos) == 0 {
return "[]"
}
out := append([]string(nil), repos...)
sort.Strings(out)
b, err := json.Marshal(out)
if err != nil {
return "[]"
}
return string(b)
}
func decodeMuteList(stored string) []string {
trimmed := strings.TrimSpace(stored)
if trimmed == "" {
return nil
}
var out []string
if trimmed[0] == '[' {
if err := json.Unmarshal([]byte(trimmed), &out); err != nil {
return nil
}
} else {
out = strings.Fields(trimmed)
}
out = slicesCompactNonEmpty(out)
sort.Strings(out)
return out
}
// slicesCompactNonEmpty drops blanks and duplicates, so a form that
// posts the same repository twice mutes it once.
func slicesCompactNonEmpty(in []string) []string {
seen := make(map[string]bool, len(in))
out := make([]string, 0, len(in))
for _, v := range in {
v = strings.TrimSpace(v)
if v == "" || seen[v] {
continue
}
seen[v] = true
out = append(out, v)
}
return out
}
// mutedRepos reads an account's muted set.
func (d Deps) mutedRepos(account string) map[string]bool {
fields := d.mutedList(account)
if len(fields) == 0 {
return nil
}
out := make(map[string]bool, len(fields))
for _, f := range fields {
out[f] = true
}
return out
}
// mutedList is the same set in the order a page shows it.
func (d Deps) mutedList(account string) []string {
if d.Prefs == nil || account == "" {
return nil
}
return decodeMuteList(d.Prefs.AccountPref(account, prefMute))
}
// muted reports whether account has silenced notify for repo. The mute
// is deliberately NOT applied to the attention stream: a person who
// muted a repository still finds its events in their own inbox when
// they go looking — what they turned off is being TOLD (RFC 0011 §6,
// "a per-repo mute that quarry implements by not triggering").
func (d Deps) muted(account, repo string) bool {
if account == "" || repo == "" {
return false
}
return d.mutedRepos(account)[repo]
}
// prefBool reads a forge-local switch that defaults OFF. Only "on"
// means on, which is the reading the attention stream already uses.
func (d Deps) prefBool(account, key string) bool {
if d.Prefs == nil {
return false
}
return d.Prefs.AccountPref(account, key) == "on"
}
type inboxRow struct {
// ID addresses the row in `inbox.mark`: a notify message id, or
// "local:<seq>" for one of quarry's own attention events.
ID string `json:"id"`
// Source is "notify" or "local" — which half of RFC 0011 the row
// came from, so a page can style a row it can only mark locally.
Source string `json:"source"`
// Reason is why it arrived: mention, reply, landing, activity.
Reason string `json:"reason,omitempty"`
Repo string `json:"repo,omitempty"`
From string `json:"from,omitempty"`
Summary string `json:"summary,omitempty"`
Link string `json:"link,omitempty"`
At string `json:"at"`
Read bool `json:"read"`
// Seq is the attention sequence for a local row, 0 for a notify one.
Seq uint64 `json:"seq,omitempty"`
}
type inboxListIn struct {
// Before is the opaque `next` cursor from a previous page. It
// addresses BOTH sources at once and embeds the companion's own
// cursor, whose length is not quarry's to bound tightly — hence a
// ceiling generous enough for one of those plus this encoding's
// fields, rather than one sized for a bare notify token.
Before string `json:"before" validate:"max=1024"`
// Limit bounds the page; 0 takes the op's default.
Limit int `json:"limit" validate:"min=0,max=200"`
}
type inboxListOut struct {
Rows []inboxRow `json:"rows"`
// Next is the cursor for the following page, "" at the end.
Next string `json:"next,omitempty"`
// Read is the local attention read pointer: every local row above
// it is unread, which is the same number the badge is measured from.
Read uint64 `json:"read"`
}
type inboxMarkIn struct {
// IDs are the rows to mark; repeat the key (or send a JSON array).
IDs []string `json:"ids" validate:"max=500"`
// All marks the whole inbox instead, ignoring ids.
All bool `json:"all"`
// As is the state to move to: read, or done (archived — off the
// list entirely).
As string `json:"as" validate:"oneof=read|done,max=8"`
}
type inboxMarkOut struct {
OK bool `json:"ok"`
// Seq is the local attention pointer after the mark.
Seq uint64 `json:"seq,omitempty"`
}
type mePrefsIn struct{}
type mePrefsOut struct {
// Families is family -> channel -> enabled, over RFC 0011 §6's
// {mention, reply, landing, activity} x {inapp, push, email}.
Families map[string]map[string]bool `json:"families"`
// Channels is the column order the matrix renders in.
Channels []string `json:"channels"`
// Toasts and Desktop are the two forge-local switches, both on
// unless somebody turned them off.
Toasts bool `json:"toasts"`
Desktop bool `json:"notify_desktop"`
// Muted names the repositories this account silenced.
Muted []string `json:"muted"`
// Unknown names the families whose stored preferences could not be
// read, because the companion's workflow identities were
// unreachable. Their row in Families is quarry's DEFAULT, not what
// notify holds — a page must say "unknown" for these rather than
// show them as on.
Unknown []string `json:"unknown,omitempty"`
// Devices are the browsers push was enabled in.
Devices []PushDevice `json:"devices"`
// Push reports whether push can be configured at all here — false
// on a forge with no notify companion, which is what lets a page
// say so rather than offer a control that cannot work.
Push bool `json:"push"`
// Applied reports the two halves a preference write has to reach.
// On `me.prefs` it is which halves this forge HAS; on
// `me.prefs.set` it is which halves the write actually LANDED in —
// and those differ, because a companion that did not answer the
// preferences route leaves the notify half unwritten. A page that
// cannot see the difference tells
// somebody a channel is off when it is still on.
Applied prefApplied `json:"applied"`
}
// prefApplied is that two-halves report: quarry's own forge-local
// switches, and the channel matrix notify holds.
type prefApplied struct {
Local bool `json:"local"`
Notify bool `json:"notify"`
}
type mePrefsSetIn struct {
// Each family carries the channels that stay ON. The page posts the
// WHOLE matrix (an unticked box is an absent value), so what arrives
// is the intended end state and not a delta.
Mention []string `json:"mention" validate:"max=8"`
Reply []string `json:"reply" validate:"max=8"`
Landing []string `json:"landing" validate:"max=8"`
Activity []string `json:"activity" validate:"max=8"`
// Toasts and Desktop are explicit: "on" or "off". An ABSENT value
// changes nothing. Reading absence as "off" is how a caller that
// only meant to set one switch silently turned the other one off.
Toasts string `json:"toasts" validate:"oneof=on|off,max=8"`
Desktop string `json:"notify_desktop" validate:"oneof=on|off,max=8"`
// Mute and Unmute are list OPERATIONS, not a replacement set: a
// caller that names neither changes no mute at all. A replacement
// set cannot be told apart from a form that forgot to post the
// field, and the failure mode there is un-muting every repository
// somebody had silenced.
Mute []string `json:"mute" validate:"max=200"`
Unmute []string `json:"unmute" validate:"max=200"`
}
type mePushKeyIn struct{}
type mePushKeyOut struct {
// PublicKey is notify's VAPID application-server key, base64url.
// Empty when no notify is configured — a page reads that as "push
// is not available here".
PublicKey string `json:"publicKey"`
}
// pushKeysIn is the browser's PushSubscription.keys, nested exactly as
// PushSubscription.toJSON() spells it so a page can post the
// subscription object verbatim without reshaping it first.
type pushKeysIn struct {
P256dh string `json:"p256dh" validate:"max=256"`
Auth string `json:"auth" validate:"max=256"`
}
type mePushIn struct {
Endpoint string `json:"endpoint" validate:"required,max=2048"`
Keys pushKeysIn `json:"keys"`
// UA labels the device on the preferences page. Optional: an
// unlabelled device is still one somebody can turn off.
UA string `json:"ua" validate:"max=256"`
}
type mePushOut struct {
OK bool `json:"ok"`
// Devices is the account's device list after the change, so the
// page that just subscribed can render it without a second call.
Devices []PushDevice `json:"devices"`
// Pending is true when notify could not confirm dropping the
// endpoint, so quarry kept the row and will retry. A page reads it
// as "asked, not yet done" rather than showing a device that
// quietly came back.
Pending bool `json:"pending_remote_delete,omitempty"`
}
type mePushDeleteIn struct {
Endpoint string `json:"endpoint" validate:"required,max=2048"`
}
// mergeInbox builds one time-sorted page out of notify's messages and
// quarry's own attention events. Newest first, which is the order an
// inbox is read in.
//
// Both halves are shown because they are not the same thing: notify
// holds what was SENT (and survives a forge restart), the attention
// stream holds what quarry itself noticed, and a forge running no
// notify has only the second. A row present in both is not
// de-duplicated — they carry different ids and different read state,
// and guessing that two rows are one is how an inbox starts hiding mail.
func mergeInbox(msgs []InboxMessage, events []AttentionEvent, readSeq uint64, limit int) []inboxRow {
rows := make([]inboxRow, 0, len(msgs)+len(events))
for _, m := range msgs {
rows = append(rows, inboxRow{
ID: m.ID, Source: "notify", Reason: reasonOf(m.Workflow),
Summary: m.Subject, Link: m.Link, At: m.At, Read: m.Read,
})
}
for _, e := range events {
if e.Kind == AttentionGap {
continue
}
rows = append(rows, inboxRow{
ID: localID(e.Seq), Source: "local", Reason: e.Kind, Repo: e.Repo,
From: e.From, Summary: e.Summary, Link: e.Link, At: e.At,
Read: e.Seq <= readSeq, Seq: e.Seq,
})
}
sort.SliceStable(rows, func(i, j int) bool {
// Two LOCAL rows are ordered by sequence, not by their stamp.
// Seq is the attention plane's own order AND the key the page
// cursor resumes from, so ordering by anything else could put a
// higher sequence after a lower one and let the next page's
// "everything below the mark" step over it.
//
// The two orders agree in practice — an attention event's At is
// stamped by its publisher immediately before the plane assigns
// its Seq — but only Seq is the resume key, so only Seq decides.
if rows[i].Source == "local" && rows[j].Source == "local" {
return rows[i].Seq > rows[j].Seq
}
return inboxAfter(rows[i].At, rows[j].At)
})
if limit > 0 && len(rows) > limit {
rows = rows[:limit]
}
return rows
}
// inboxAfter reports whether a is newer than b. Parsed rather than
// string-compared because the two sources are two clocks with two
// formats: notify stamps its own CreatedAt and quarry stamps
// RFC3339Nano, and a lexical compare would sort "2026-08-16T00:00:00Z"
// against "2026-08-16T00:00:00.000000001Z" by string length. An
// unparseable stamp sorts last rather than crashing the page.
func inboxAfter(a, b string) bool {
ta, aok := parseInboxTime(a)
tb, bok := parseInboxTime(b)
if aok && bok {
return ta.After(tb)
}
if aok != bok {
return aok
}
return a > b
}
func parseInboxTime(s string) (time.Time, bool) {
if s == "" {
return time.Time{}, false
}
for _, layout := range []string{time.RFC3339Nano, time.RFC3339, "2006-01-02 15:04:05"} {
if t, err := time.Parse(layout, s); err == nil {
return t, true
}
}
return time.Time{}, false
}
// registerInbox wires RFC 0011 §6's ops. Registered unconditionally,
// for registerAttention's reason: docs/api.md must be the same document
// on every node, and a forge with no notify answers honestly (an empty
// page, no push key) rather than 404ing a documented surface.
func registerInbox(r *ops.Registry, d Deps) error {
regs := []func() error{
func() error { return registerInboxList(r, d) },
func() error { return registerInboxMark(r, d) },
func() error { return registerMePrefs(r, d) },
func() error { return registerMePush(r, d) },
}
for _, reg := range regs {
if err := reg(); err != nil {
return err
}
}
return nil
}
// notifyWindow is notify's half of one page: up to limit rows starting
// at the cursor's position, and — the part that makes paging correct —
// where notify's half STANDS after each one of them.
//
// Carrying a position per row is what removes the arithmetic. The merge
// decides how many notify rows survive the limit only after both
// sources are sorted together, so the resume point cannot be computed
// ahead of time; and computing it afterwards by adding to Skip is what
// made Skip grow without bound (every page re-fetched from the same
// notify cursor and asked for limit+Skip rows, so a long inbox read got
// quadratically more expensive and the page-boundary paths never ran).
// Here the boundary is crossed while the window is being built, so
// after[i] already names the next page's cursor when rows[i] is the
// last of a notify page.
type notifyWindow struct {
rows []InboxMessage
// after[i] is where notify's half stands once rows[0..i] have been
// served. Parallel to rows.
after []inboxCursor
// none is where it stands when the page serves none of them —
// which is not always the cursor we came in with, because the walk
// past already-served rows may itself have crossed a page.
none inboxCursor
}
// buildNotifyWindow walks to the cursor's position and collects at most
// limit rows from there, following notify's own `next` across as many
// of its pages as that takes.
func buildNotifyWindow(ctx context.Context, d Deps, who string, cur inboxCursor, limit int) (notifyWindow, error) {
win := notifyWindow{none: cur}
at := cur
if d.Inbox == nil || cur.NotifyDone {
return win, nil
}
for page := 0; page < inboxMaxNotifyPages && len(win.rows) < limit; page++ {
// limit+Skip because Skip rows of this page are already served.
// Both are bounded (inboxMaxSkip, and the op's own max=200), so
// what reaches notify is bounded too.
rows, next, err := d.Inbox.List(ctx, who, at.Notify, limit+at.Skip)
if err != nil {
return notifyWindow{}, err
}
if at.Skip >= len(rows) {
// The whole page is behind us. A companion that ignores
// ?limit= and serves a fixed page size lands here.
if next == "" {
at.NotifyDone, at.Skip = true, 0
break
}
at.Skip -= len(rows)
at.Notify = next
win.none = at
continue
}
for i := at.Skip; i < len(rows) && len(win.rows) < limit; i++ {
pos := at
pos.Skip = i + 1
if pos.Skip >= len(rows) {
// Serving this row finishes the page: REBASE onto
// notify's own next cursor rather than growing Skip.
pos.Skip = 0
if next == "" {
pos.NotifyDone = true
} else {
pos.Notify = next
}
}
win.rows = append(win.rows, rows[i])
win.after = append(win.after, pos)
}
if len(win.rows) >= limit {
break
}
if next == "" {
at.NotifyDone, at.Skip = true, 0
break
}
at.Notify, at.Skip = next, 0
}
if len(win.rows) == 0 {
win.none = at
}
return win, nil
}
func registerInboxList(r *ops.Registry, d Deps) error {
return ops.Register(r, ops.Op[inboxListIn, inboxListOut]{
Name: "inbox.list",
Doc: "One page of everything addressed to the signed-in account, newest first, merging the " +
"notify inbox with quarry's own attention stream. Each row carries a `reason` (mention, " +
"reply, landing, activity) derived from the workflow that sent it, and an `id` that " +
"`inbox.mark` takes back — a notify message id, or `local:<seq>` for an attention event. " +
"`before` is the opaque `next` cursor from a previous page; it addresses BOTH sources at " +
"once, so a row the page limit cut off is served by the next page rather than skipped.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/me/inbox",
Run: func(ctx context.Context, in inboxListIn) (inboxListOut, error) {
who := ops.User(ctx)
if who == "" {
return inboxListOut{}, nil
}
limit := in.Limit
if limit == 0 {
limit = 50
}
cur := decodeInboxCursor(in.Before)
out := inboxListOut{}
if d.AttnRead != nil {
out.Read = d.AttnRead.AttentionReadSeq(who)
}
win, err := buildNotifyWindow(ctx, d, who, cur, limit)
if err != nil {
// An outage is not an empty inbox, for inbox.count's
// reason: a page that hides outages teaches people to
// stop trusting it.
return inboxListOut{}, err
}
// The local half is bounded by SEQUENCE: everything below
// the cursor's mark is still to come. The scan is capped
// rather than paged because the attention window is itself
// bounded — past inboxLocalScan the local half simply ends,
// and `inbox.recent` is the op that serves the window whole.
var events []AttentionEvent
if d.Attention != nil && !cur.LocalDone {
all, err := d.Attention.RecentAttention(who, inboxLocalScan)
if err != nil {
return inboxListOut{}, err
}
for _, e := range all {
if cur.LocalSet && e.Seq >= cur.Local {
continue
}
events = append(events, e)
}
}
out.Rows = mergeInbox(win.rows, events, out.Read, limit)
out.Next = advanceInbox(win, len(events), out.Rows)
return out, nil
},
})
}
// advanceInbox reads the emitted page back to say where each source
// stands. Derived from what was EMITTED rather than from what was
// fetched, which is the whole correction: the old cursor was notify's
// position after the fetch, so every fetched-but-truncated message was
// stepped over.
//
// notify's side needs no arithmetic — the window already carries the
// position after each of its rows, so the k-th one emitted names it.
//
// The local side resumes by SEQUENCE, and the lowest sequence emitted
// is that mark. It is exact because the emitted local rows are a
// sequence-descending prefix of the fetched ones: mergeInbox orders the
// local source by Seq (see there), and truncation drops a suffix.
func advanceInbox(win notifyWindow, fetchedLocal int, rows []inboxRow) string {
emittedNotify, lowestLocal := 0, uint64(0)
for _, row := range rows {
if row.Source == "notify" {
emittedNotify++
continue
}
if lowestLocal == 0 || row.Seq < lowestLocal {
lowestLocal = row.Seq
}
}
next := win.none
if emittedNotify > 0 {
next = win.after[emittedNotify-1]
}
if lowestLocal > 0 {
next.Local, next.LocalSet = lowestLocal, true
}
if len(rows) == 0 || fetchedLocal == 0 {
next.LocalDone = true
}
if next.NotifyDone && next.LocalDone {
return ""
}
return next.encode()
}
func registerInboxMark(r *ops.Registry, d Deps) error {
return ops.Register(r, ops.Op[inboxMarkIn, inboxMarkOut]{
Name: "inbox.mark",
Doc: "Mark inbox rows read, or done (archived — off the list). Takes the ids `inbox.list` " +
"gave, or `all` for the whole inbox. Ids reach notify; `local:<seq>` ids advance the " +
"account's attention read pointer, the same forward-only pointer `attention.read` moves, " +
"so clearing the inbox clears the badge whichever half a row came from.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/me/inbox/mark",
Run: func(ctx context.Context, in inboxMarkIn) (inboxMarkOut, error) {
who := ops.User(ctx)
if who == "" {
return inboxMarkOut{}, fmt.Errorf("marking the inbox needs a signed-in account: %w", ops.ErrInvalid)
}
as := in.As
if as == "" {
as = "read"
}
if !in.All && len(in.IDs) == 0 {
return inboxMarkOut{}, fmt.Errorf("name some ids or ask for all: %w", ops.ErrInvalid)
}
ids, seqs := splitInboxIDs(in.IDs)
if d.Inbox != nil && (in.All || len(ids) > 0) {
// "done" is quarry's word; notify's state is archived.
state := "read"
if as == "done" {
state = "archived"
}
if err := d.Inbox.Mark(ctx, who, ids, in.All, state); err != nil {
return inboxMarkOut{}, err
}
}
out := inboxMarkOut{OK: true}
if d.AttnRead == nil {
return out, nil
}
// The local pointer moves to the highest sequence the mark
// named — or, on an all-mark, to the account's own tip. It
// only ever goes forward (MarkAttentionRead's rule), so a
// stale page cannot resurrect dismissed mail.
var to uint64
if in.All {
if d.Attention != nil {
tip, err := d.Attention.AttentionTip(who)
if err != nil {
return inboxMarkOut{}, err
}
to = tip
}
} else {
for _, s := range seqs {
if s > to {
to = s
}
}
}
if to > 0 {
if err := d.AttnRead.MarkAttentionRead(who, to); err != nil {
return inboxMarkOut{}, err
}
}
out.Seq = d.AttnRead.AttentionReadSeq(who)
return out, nil
},
})
}
// currentPrefs renders the account's matrix: notify's stored rows over
// the default (everything on).
func (d Deps) currentPrefs(ctx context.Context, who string) (mePrefsOut, error) {
out := mePrefsOut{
Families: map[string]map[string]bool{},
Channels: prefChannelOrder,
Toasts: d.prefBool(who, prefToasts),
Desktop: d.prefBool(who, prefDesktop),
Push: d.Inbox != nil,
Applied: prefApplied{Local: d.Prefs != nil, Notify: d.Inbox != nil},
}
var stored map[string]map[string]bool
resolved := true
if d.Inbox != nil {
var err error
stored, resolved, err = d.Inbox.Preferences(ctx, who)
if err != nil {
return mePrefsOut{}, err
}
}
for _, f := range prefFamilies {
row := map[string]bool{}
s, ok := stored[f.workflow]
for _, ch := range prefChannelOrder {
// Absent means ENABLED, which is notify's own resolution
// rule (internal/preferences: a channel is on unless a
// level explicitly sets it false). Defaulting to off here
// would silence a forge that has simply never saved a
// preference.
on := true
if ok {
if v, ok := s[prefChannels[ch]]; ok {
on = v
}
}
row[ch] = on
}
out.Families[f.family] = row
if !ok && !resolved {
// No row AND no way to know what to look under. The values
// above are a default being shown, not a setting read.
out.Unknown = append(out.Unknown, f.family)
}
}
out.Muted = d.mutedList(who)
if d.PushDevices != nil {
devs, err := d.PushDevices.PushDevices(who)
if err != nil {
return mePrefsOut{}, err
}
out.Devices = devs
}
return out, nil
}
func registerMePrefs(r *ops.Registry, d Deps) error {
if err := ops.Register(r, ops.Op[mePrefsIn, mePrefsOut]{
Name: "me.prefs",
Doc: "The signed-in account's notification preferences: RFC 0011 §6's family x channel " +
"matrix ({mention, reply, landing, activity} x {inapp, push, email}) as notify holds it, " +
"plus the forge-local switches quarry owns — in-page toasts, desktop notifications, the " +
"repositories this account muted, and the browsers push was enabled in. A channel with no " +
"stored preference is ON, which is notify's own rule.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/me/prefs",
Run: func(ctx context.Context, in mePrefsIn) (mePrefsOut, error) {
who := ops.User(ctx)
if who == "" {
return mePrefsOut{}, nil
}
return d.currentPrefs(ctx, who)
},
}); err != nil {
return err
}
return ops.Register(r, ops.Op[mePrefsSetIn, mePrefsOut]{
Name: "me.prefs.set",
Doc: "Update the signed-in account's preferences. Each family carries the channels that stay " +
"ON (repeat the key), so the whole matrix is posted and an absent channel is off. The two " +
"forge-local switches are EXPLICIT — `toasts` and `notify_desktop` take \"on\" or " +
"\"off\", and leaving one out changes it not at all. `mute` and `unmute` are list " +
"operations for the same reason: naming neither changes no mute. A muted repository still " +
"reaches the account's own inbox, it just stops being sent. The answer's `applied` says " +
"which halves landed: the forge-local one always does, the notify one does not on a " +
"companion that did not answer its preferences route.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/me/prefs",
Run: func(ctx context.Context, in mePrefsSetIn) (mePrefsOut, error) {
who := ops.User(ctx)
if who == "" {
return mePrefsOut{}, fmt.Errorf("preferences need a signed-in account: %w", ops.ErrInvalid)
}
var applied prefApplied
// The forge-local half goes FIRST and on its own. It is the
// half quarry owns outright, and a notify route that has
// not shipped must not cost somebody the mute they came
// here to set.
if d.Prefs != nil {
if err := d.setLocalPrefs(who, in); err != nil {
return mePrefsOut{}, err
}
applied.Local = true
}
if d.Inbox != nil && d.matrixWritable(ctx, who) {
prefs := map[string]map[string]bool{}
wanted := map[string][]string{
"mention": in.Mention, "reply": in.Reply,
"landing": in.Landing, "activity": in.Activity,
}
for _, f := range prefFamilies {
on := map[string]bool{}
for _, ch := range wanted[f.family] {
if notifyCh, ok := prefChannels[ch]; ok {
on[notifyCh] = true
}
}
row := map[string]bool{}
for _, ch := range prefChannelOrder {
row[prefChannels[ch]] = on[prefChannels[ch]]
}
prefs[f.workflow] = row
}
switch err := d.Inbox.SetPreferences(ctx, who, prefs); {
case err == nil:
applied.Notify = true
case errors.Is(err, ErrRemoteUnsupported):
// The companion did not answer. Logged, not
// returned — the local half took, and `applied`
// says so honestly instead of the op failing.
log.Printf("me.prefs.set: %s: notify did not answer its preferences route: %v", who, err)
default:
return mePrefsOut{}, err
}
}
out, err := d.currentPrefs(ctx, who)
if err != nil {
return mePrefsOut{}, err
}
out.Applied = applied
return out, nil
},
})
}
// matrixWritable reports whether the channel matrix may be written at
// all. It may not when the companion did not answer its preferences
// route: `me.prefs` then reports those families UNKNOWN and shows
// quarry's DEFAULT rather than what notify holds, so writing the matrix
// back would store that default — turning every channel on for somebody
// who only came here to unmute a repository. `applied.notify` reports
// false and the local half still takes.
func (d Deps) matrixWritable(ctx context.Context, who string) bool {
_, resolved, err := d.Inbox.Preferences(ctx, who)
if err != nil {
log.Printf("me.prefs.set: %s: preferences unreadable, matrix left alone: %v", who, err)
return false
}
if !resolved {
log.Printf("me.prefs.set: %s: notify did not answer, matrix left alone", who)
}
return resolved
}
// setLocalPrefs writes the half quarry owns: the two switches, and the
// muted set as a list operation. Every write here is conditional on the
// caller having actually named the thing — absence is not a value.
func (d Deps) setLocalPrefs(who string, in mePrefsSetIn) error {
if v := switchValue(in.Toasts); v != "" {
if err := d.Prefs.SetAccountPref(who, prefToasts, v); err != nil {
return err
}
}
if v := switchValue(in.Desktop); v != "" {
if err := d.Prefs.SetAccountPref(who, prefDesktop, v); err != nil {
return err
}
}
if len(in.Mute) == 0 && len(in.Unmute) == 0 {
return nil
}
set := map[string]bool{}
for _, repo := range d.mutedList(who) {
set[repo] = true
}
for _, repo := range slicesCompactNonEmpty(in.Mute) {
set[repo] = true
}
for _, repo := range slicesCompactNonEmpty(in.Unmute) {
delete(set, repo)
}
out := make([]string, 0, len(set))
for repo := range set {
out = append(out, repo)
}
return d.Prefs.SetAccountPref(who, prefMute, encodeMuteList(out))
}
// switchValue normalizes one forge-local switch to what account_pref
// stores, and — the point of it — keeps ABSENT distinct from OFF.
//
// The reading it replaces treated an unposted switch as off, so a form
// that posted `toasts` and not `notify_desktop` silently turned desktop
// notifications off for the person who submitted it. A switch nobody
// mentioned is a switch nobody changed; "" is that, and the caller
// writes nothing.
//
// "off" is stored explicitly rather than as the empty deletion because
// the DEFAULT is on: a deleted row reads as "never set", which is on,
// and the switch would refuse to stay off.
func switchValue(v string) string {
// Only "", "on" and "off" can arrive: the field's `oneof=on|off`
// rule refuses everything else before the op runs, so the lenient
// spellings this used to accept were unreachable branches.
switch strings.ToLower(strings.TrimSpace(v)) {
case "on":
return "on"
case "off":
return "off"
default:
return ""
}
}
func registerMePush(r *ops.Registry, d Deps) error {
if err := ops.Register(r, ops.Op[mePushKeyIn, mePushKeyOut]{
Name: "me.push.key",
Doc: "notify's VAPID application-server public key, base64url — what a browser passes to " +
"pushManager.subscribe. Empty when this forge has no notify companion, which a page reads " +
"as \"push is not available here\". quarry never holds the private half.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/me/push/key",
Run: func(ctx context.Context, in mePushKeyIn) (mePushKeyOut, error) {
if d.Inbox == nil {
return mePushKeyOut{}, nil
}
key, err := d.Inbox.PushKey(ctx)
if err != nil {
return mePushKeyOut{}, err
}
return mePushKeyOut{PublicKey: key}, nil
},
}); err != nil {
return err
}
if err := ops.Register(r, ops.Op[mePushIn, mePushOut]{
Name: "me.push.subscribe",
Doc: "Register a browser for push. The body is the browser's own PushSubscription JSON " +
"({endpoint, keys:{p256dh, auth}}), forwarded to notify as subscriber credentials for the " +
"webpush provider. quarry stores no endpoint and no key — only a hash, a user-agent label " +
"and a timestamp, so the preferences page can list the devices somebody may want to turn " +
"off. Also sweeps this account's unconfirmed unsubscribes: a device notify would not drop " +
"earlier is retried here, which is the one moment quarry knows the companion is reachable.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/me/push",
Run: func(ctx context.Context, in mePushIn) (mePushOut, error) {
who := ops.User(ctx)
if who == "" {
return mePushOut{}, fmt.Errorf("a push subscription needs a signed-in account: %w", ops.ErrInvalid)
}
if d.Inbox == nil {
return mePushOut{}, fmt.Errorf("this forge has no notification companion: %w", ops.ErrUnavailable)
}
if err := d.Inbox.PushSubscribe(ctx, who, in.Endpoint, in.Keys.P256dh, in.Keys.Auth); err != nil {
return mePushOut{}, err
}
d.sweepPendingPushDeletes(ctx, who)
out := mePushOut{OK: true}
if d.PushDevices != nil {
if err := d.PushDevices.AddPushDevice(who, in.Endpoint, in.UA); err != nil {
return mePushOut{}, err
}
devs, err := d.PushDevices.PushDevices(who)
if err != nil {
return mePushOut{}, err
}
out.Devices = devs
}
return out, nil
},
}); err != nil {
return err
}
return ops.Register(r, ops.Op[mePushDeleteIn, mePushOut]{
Name: "me.push.unsubscribe",
Doc: "Stop pushing to one browser: drops the endpoint's credential in notify and forgets the " +
"device row. A credential notify does not hold (404) and one the push service already " +
"discarded (410 Gone) are both done, since that is the end state asked for. A notify " +
"that FAILED the delete has not dropped anything, so quarry keeps the device row marked " +
"`pending_remote_delete`, holding the endpoint only until the delete can be confirmed, " +
"and retries on the next `me.push.subscribe`. Forgetting the row on an unconfirmed " +
"delete is how a phone keeps buzzing with nothing left in the interface to stop it.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/me/push/delete",
Run: func(ctx context.Context, in mePushDeleteIn) (mePushOut, error) {
who := ops.User(ctx)
if who == "" {
return mePushOut{}, fmt.Errorf("removing a push subscription needs a signed-in account: %w", ops.ErrInvalid)
}
out := mePushOut{OK: true}
if d.Inbox != nil {
// An unconfirmed delete is PENDING, not a failed op: the
// person asked for this browser to stop, and answering
// them an error while the row stays live tells them
// nothing about which it is.
if err := d.Inbox.PushUnsubscribe(ctx, who, in.Endpoint); err != nil {
log.Printf("me.push.unsubscribe: %s: notify did not confirm the delete: %v", who, err)
out.Pending = true
}
}
if d.PushDevices != nil {
if out.Pending {
if err := d.PushDevices.MarkPushDevicePendingDelete(who, in.Endpoint); err != nil {
return mePushOut{}, err
}
} else if err := d.PushDevices.DeletePushDevice(who, in.Endpoint); err != nil {
return mePushOut{}, err
}
devs, err := d.PushDevices.PushDevices(who)
if err != nil {
return mePushOut{}, err
}
out.Devices = devs
}
return out, nil
},
})
}
// sweepPendingPushDeletes retries the unsubscribes notify could not
// confirm. Best-effort and silent on failure: it runs off somebody
// ENABLING push, and a device that still cannot be dropped stays
// pending for the next attempt rather than failing the subscribe.
func (d Deps) sweepPendingPushDeletes(ctx context.Context, who string) {
if d.PushDevices == nil || d.Inbox == nil {
return
}
pending, err := d.PushDevices.PendingPushDeletes(who)
if err != nil {
log.Printf("me.push: %s: pending deletes: %v", who, err)
return
}
for _, endpoint := range pending {
if err := d.Inbox.PushUnsubscribe(ctx, who, endpoint); err != nil {
log.Printf("me.push: %s: retry unsubscribe: %v", who, err)
return // still not there; every other row would fail the same way
}
if err := d.PushDevices.DeletePushDevice(who, endpoint); err != nil {
log.Printf("me.push: %s: forget device: %v", who, err)
}
}
}