// attention.go is RFC 0011 §5's person-addressed plane at the ops seam:
// the stream an open tab tails (`me.attention`), the count a badge reads
// (`inbox.count`), the pointer that clears it (`attention.read`), and
// §10's per-repository activity robot switch (`chat.robot`).
//
// Everything here is about a PERSON, not a repository, which is why the
// "me" ops name no repo and carry no repository capability.
package api
import (
"context"
"fmt"
"log"
"strings"
"time"
"craftdesign.group/quarry/internal/evt"
"craftdesign.group/quarry/internal/ops"
)
// AttentionEvent is one thing addressed to one person, in this
// package's own terms. Stated here rather than imported from the event
// plane for the same reason NotifyEvent is: the ops layer names what it
// needs, and the composition root owns which transport carries it.
type AttentionEvent struct {
Seq uint64 `json:"seq"`
Kind string `json:"kind"`
Repo string `json:"repo,omitempty"`
From string `json:"from,omitempty"`
Summary string `json:"summary,omitempty"`
Link string `json:"link,omitempty"`
At string `json:"at"`
// Ref is set only on a gap signal: the sequence that was asked for.
Ref uint64 `json:"ref,omitempty"`
}
// AttentionGap is the Kind a tail carries when the resume point asked
// for is older than the window — not an event, a statement about the
// stream. Mirrors chat's gap frame exactly.
const AttentionGap = "gap"
// AttentionPlane is RFC 0011 §5's stream. A nil Deps.Attention disables
// it, which is how a forge with no event plane simply tells nobody
// anything live — every call site here is nil-safe.
type AttentionPlane interface {
// Attend publishes one event for account and returns its sequence.
Attend(account string, e AttentionEvent) (uint64, error)
// TailAttention delivers account's events from from+1 onward (0 =
// new only) until cancel, never dropping under backpressure.
TailAttention(account string, from uint64) (<-chan AttentionEvent, func())
// AttentionPending counts account's events after since.
AttentionPending(account string, since uint64) (int, error)
// AttentionTip is the highest sequence account currently holds, 0
// when the window holds none of theirs. The account's own tip, not
// the stream's — the stream's belongs to everybody at once.
AttentionTip(account string) (uint64, error)
// RecentAttention is at most limit of account's most recent
// events, oldest-first: the load-time state an inbox renders with
// no stream open at all (Principle 3).
RecentAttention(account string, limit int) ([]AttentionEvent, error)
}
// AttentionStore holds the read pointer the local unread count is
// measured from. nil means nothing is ever marked read, so the count is
// the whole window — honest, and what a forge with no OLTP gets.
type AttentionStore interface {
MarkAttentionRead(account string, seq uint64) error
AttentionReadSeq(account string) uint64
}
// ChatRobotStore is §10's per-repository switch: which events the robot
// speaks aloud in the channel, in webhook pattern syntax. nil means the
// robot cannot be configured, which is a forge with no OLTP.
type ChatRobotStore interface {
SetChatRobot(repo, events string) error
ChatRobot(repo string) string
}
// The "me" ops below carry no Caps, and RFC 0011 §8's `p` is not
// spelled here, for the reason chat.unread already records: the router
// resolves a capability set against the repository named in the path or
// form, and these routes name none. A letter would be checked against
// the EMPTY repository — the fleet default's stock strings — and refuse
// every ordinary account. Auth Token is the real gate; these surfaces
// are about the signed-in person and show nothing a repository owns.
// attend publishes one attention event best-effort: never blocks the
// op, never fails it, and never lets a broker problem become a write
// failure. Errors are logged, exactly as fireNotify's are.
func (d Deps) attend(account string, e AttentionEvent) {
if d.Attention == nil || account == "" {
return
}
if _, err := d.Attention.Attend(account, e); err != nil {
log.Printf("attention: %s: %v", account, err)
}
}
// personAddressed names the workflows that address a PERSON rather than
// a repository's topic (RFC 0011 §5). Topic activity is not attention:
// it is the repository talking, and it reaches the same people through
// the channel and the timeline. The admin-only error and slow-page
// workflows are not either — "admins" is a topic, not somebody.
var personAddressed = map[string]string{
"quarry-mention": "mention",
"quarry-forum-reply": "reply",
"quarry-landing": "landing",
}
// attentionOf renders a NotifyEvent as the attention event for it, and
// reports whether it is addressed to a person at all.
func attentionOf(e NotifyEvent) (string, AttentionEvent, bool) {
kind, ok := personAddressed[e.Workflow]
if !ok {
return "", AttentionEvent{}, false
}
// A topic is not a person. notify's topics are spelled with a
// prefix ("repo:<name>") or are the reserved "admins" group;
// publishing either to attn.<who> would address a subject nobody
// tails and, worse, would look like mail to an account with that
// name if one ever existed.
if e.To == "" || e.To == "admins" || strings.Contains(e.To, ":") {
return "", AttentionEvent{}, false
}
str := func(k string) string {
s, _ := e.Payload[k].(string)
return s
}
if e.Kind.Valid() {
kind = string(e.Kind)
}
summary := str("summary")
if summary == "" {
summary = str("title")
}
if summary == "" {
summary = attentionSummary(str("text"))
}
from := str("from")
if from == "" {
from = str("actor")
}
return e.To, AttentionEvent{
Kind: kind, Repo: str("repo"), From: from,
Summary: summary, Link: str("link"),
At: time.Now().UTC().Format(time.RFC3339Nano),
}, true
}
// attentionSummary is one line of somebody's prose, bounded: a toast
// and an inbox row each have one line to spend, and a 4 KiB chat
// message would otherwise ride the stream whole to fill it.
func attentionSummary(s string) string {
s = firstLine(s)
const limit = 160
if len(s) <= limit {
return s
}
// Cut on a rune boundary: a byte slice through UTF-8 renders as a
// replacement character in the middle of somebody's sentence.
cut := limit
for cut > 0 && s[cut]&0xC0 == 0x80 {
cut--
}
return strings.TrimSpace(s[:cut]) + "…"
}
type meAttentionIn struct {
// From resumes after this seq (Last-Event-ID, overlaid by the
// router); 0 streams new events only.
From uint64 `json:"from"`
}
type meAttentionOut struct {
Seq uint64 `json:"seq"`
Kind string `json:"kind"`
Repo string `json:"repo,omitempty"`
From string `json:"from,omitempty"`
Summary string `json:"summary,omitempty"`
Link string `json:"link,omitempty"`
At string `json:"at"`
}
type inboxRecentIn struct {
// Limit bounds the page; 0 takes the op's own default.
Limit int `json:"limit" validate:"min=0,max=200"`
}
type inboxRecentOut struct {
// Events are newest-first, which is the order an inbox reads in.
Events []meAttentionOut `json:"events"`
// Read is the account's read pointer: everything above it is
// unread, and the page needs the number rather than a per-row flag
// so it can say the same thing the badge does.
Read uint64 `json:"read"`
}
type inboxCountIn struct{}
type inboxCountOut struct {
Unread int `json:"unread"`
}
type attentionReadIn struct {
// Seq is the pointer to move to; 0 means "everything I have been
// shown", which the op resolves to the account's current tip.
Seq uint64 `json:"seq"`
}
type attentionReadOut struct {
OK bool `json:"ok"`
Seq uint64 `json:"seq"`
}
type chatRobotIn struct {
Repo string `json:"repo" validate:"required,max=64"`
// Events is webhook pattern syntax (`*`, a noun, or an exact kind),
// space-separated. Empty turns the robot off.
Events string `json:"events" validate:"max=512"`
}
type chatRobotOut struct {
Repo string `json:"repo"`
Events string `json:"events"`
On bool `json:"on"`
}
type chatRobotGetIn struct {
Repo string `json:"repo" validate:"required,max=64"`
}
// registerAttention wires the person-addressed ops. Registered
// unconditionally: a forge with no attention plane still documents the
// surface and answers it honestly (an empty stream, a zero count),
// which is what keeps docs/api.md the same document on every node.
func registerAttention(r *ops.Registry, d Deps) error {
if err := ops.Register(r, ops.Op[meAttentionIn, meAttentionOut]{
Name: "me.attention",
Doc: "Everything addressed to the signed-in account, as Server-Sent Events. The body is " +
"text/event-stream, not JSON: each event's `id:` line is its stream sequence, and a client " +
"that reconnects with `Last-Event-ID: <seq>` resumes after it. A resume older than the " +
"window replays from the window's first event, preceded by an `event: gap` frame carrying " +
"{\"from\":<asked>,\"first\":<available>}.",
Kind: ops.Stream,
Auth: ops.Token,
Route: "GET /api/v1/me/attention",
StreamRun: func(ctx context.Context, in meAttentionIn, s ops.Streamer) error {
who := ops.User(ctx)
if who == "" {
return fmt.Errorf("the attention stream needs a signed-in account: %w", ops.ErrInvalid)
}
if d.Attention == nil {
// No plane: an empty stream, not an error. The page's
// load-time state is still true (Principle 3).
<-ctx.Done()
return nil
}
events, cancel := d.Attention.TailAttention(who, in.From)
defer cancel()
heartbeat := time.NewTicker(20 * time.Second)
defer heartbeat.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-heartbeat.C:
s.Ping()
case e, open := <-events:
if !open {
return nil
}
if e.Kind == AttentionGap {
if err := s.Event("gap", map[string]uint64{
"from": e.Ref, "first": e.Seq,
}); err != nil {
return err
}
continue
}
if err := s.Emit(e.Seq, meAttentionOut{
Seq: e.Seq, Kind: e.Kind, Repo: e.Repo, From: e.From,
Summary: e.Summary, Link: e.Link, At: e.At,
}); err != nil {
return err
}
}
}
},
}); err != nil {
return err
}
if err := ops.Register(r, ops.Op[inboxCountIn, inboxCountOut]{
Name: "inbox.count",
Doc: "How much is waiting for the signed-in account. Counted from the notify inbox when one " +
"is configured — that is the inbox they can open and clear — and from the local attention " +
"stream after their read pointer when it is not.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/me/inbox/count",
Run: func(ctx context.Context, in inboxCountIn) (inboxCountOut, error) {
who := ops.User(ctx)
if who == "" {
return inboxCountOut{}, nil
}
if d.Inbox != nil {
n, err := d.Inbox.Unread(ctx, who)
if err != nil {
// An outage is not "you have nothing": a badge that
// hides outages teaches people to ignore it.
return inboxCountOut{}, err
}
return inboxCountOut{Unread: n}, nil
}
if d.Attention == nil {
return inboxCountOut{}, nil
}
var since uint64
if d.AttnRead != nil {
since = d.AttnRead.AttentionReadSeq(who)
}
n, err := d.Attention.AttentionPending(who, since)
if err != nil {
return inboxCountOut{}, err
}
return inboxCountOut{Unread: n}, nil
},
}); err != nil {
return err
}
if err := ops.Register(r, ops.Op[inboxRecentIn, inboxRecentOut]{
Name: "inbox.recent",
Doc: "What was recently addressed to the signed-in account, newest first, with their read " +
"pointer. This is the local attention window — the same events me.attention streams — so " +
"an inbox is a page a forge with no notify companion still renders correctly.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/me/inbox/recent",
Run: func(ctx context.Context, in inboxRecentIn) (inboxRecentOut, error) {
who := ops.User(ctx)
if who == "" {
return inboxRecentOut{}, nil
}
out := inboxRecentOut{}
if d.AttnRead != nil {
out.Read = d.AttnRead.AttentionReadSeq(who)
}
if d.Attention == nil {
return out, nil
}
limit := in.Limit
if limit == 0 {
limit = 50
}
evs, err := d.Attention.RecentAttention(who, limit)
if err != nil {
return inboxRecentOut{}, err
}
// Newest first: RecentAttention answers in stream order,
// which is the order they happened, not the order an inbox
// is read in.
for i := len(evs) - 1; i >= 0; i-- {
e := evs[i]
if e.Kind == AttentionGap {
continue
}
out.Events = append(out.Events, meAttentionOut{
Seq: e.Seq, Kind: e.Kind, Repo: e.Repo, From: e.From,
Summary: e.Summary, Link: e.Link, At: e.At,
})
}
return out, nil
},
}); err != nil {
return err
}
return ops.Register(r, ops.Op[attentionReadIn, attentionReadOut]{
Name: "attention.read",
Doc: "Move the signed-in account's attention read pointer to seq (default: everything so far). " +
"Forward-only — a stale client replaying an old cursor cannot resurrect mail already dismissed.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/me/attention/read",
Run: func(ctx context.Context, in attentionReadIn) (attentionReadOut, error) {
who := ops.User(ctx)
if who == "" {
return attentionReadOut{}, fmt.Errorf("marking read needs a signed-in account: %w", ops.ErrInvalid)
}
if d.AttnRead == nil {
return attentionReadOut{}, fmt.Errorf("this forge keeps no read state: %w", ops.ErrUnavailable)
}
seq := in.Seq
if seq == 0 {
// "Everything so far" is the pointer plus what is still
// pending, which is the account's own tip — never the
// stream's, which belongs to every account at once.
if d.Attention != nil {
tip, err := d.Attention.AttentionTip(who)
if err != nil {
return attentionReadOut{}, err
}
seq = tip
}
}
if seq == 0 {
return attentionReadOut{OK: true}, nil
}
if err := d.AttnRead.MarkAttentionRead(who, seq); err != nil {
return attentionReadOut{}, err
}
return attentionReadOut{OK: true, Seq: d.AttnRead.AttentionReadSeq(who)}, nil
},
})
}
// registerChatRobot wires RFC 0011 §10's switch. It lives with the chat
// ops (it configures a channel) but reads and writes forge-local state,
// so a forge with no OLTP registers the surface and refuses the write
// rather than pretending the setting took.
func registerChatRobot(r *ops.Registry, d Deps) error {
if err := ops.Register(r, ops.Op[chatRobotIn, chatRobotOut]{
Name: "chat.robot",
Caps: "a",
Doc: "Set which events the activity robot announces in a repository's chat channel, in the " +
"webhook pattern syntax (`*`, a noun like `checkin`, or an exact kind like `landing.accepted`), " +
"space-separated. Empty turns the robot off.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/r/{repo}/chat/robot",
Run: func(ctx context.Context, in chatRobotIn) (chatRobotOut, error) {
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return chatRobotOut{}, err
}
if d.ChatRobots == nil {
return chatRobotOut{}, fmt.Errorf("this forge keeps no robot settings: %w", ops.ErrUnavailable)
}
events := strings.Join(strings.Fields(in.Events), " ")
// Refused at the moment it is set, by the same rule the
// webhook surface uses: a robot that silently announces
// nothing reads as a broken forge.
if err := evt.CheckPatterns(events); err != nil {
return chatRobotOut{}, fmt.Errorf("%s: %w", err, ops.ErrInvalid)
}
if err := d.ChatRobots.SetChatRobot(in.Repo, events); err != nil {
return chatRobotOut{}, err
}
return chatRobotOut{Repo: in.Repo, Events: events, On: events != ""}, nil
},
}); err != nil {
return err
}
return ops.Register(r, ops.Op[chatRobotGetIn, chatRobotOut]{
Name: "chat.robot.get",
Caps: "a",
Doc: "The repository's activity-robot event patterns; empty when the robot is off. Its own op " +
"rather than a field on repo.get, because it is chat configuration and only an administrator may see it.",
Kind: ops.Query,
Auth: ops.Token,
Route: "GET /api/v1/r/{repo}/chat/robot",
Run: func(ctx context.Context, in chatRobotGetIn) (chatRobotOut, error) {
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return chatRobotOut{}, err
}
if d.ChatRobots == nil {
return chatRobotOut{Repo: in.Repo}, nil
}
events := d.ChatRobots.ChatRobot(in.Repo)
return chatRobotOut{Repo: in.Repo, Events: events, On: events != ""}, nil
},
})
}