// attn.go is RFC 0011 §5's attention stream: one small JetStream stream
// carrying what is addressed to a PERSON, on the subject `attn.<account>`.
//
// It exists so an open tab is told by quarry rather than by a round trip
// through the notify companion, and so a forge with no notify configured
// at all still tells people the things that were addressed to them.
// Nothing here is truth — an attention event is a notification about a
// fact the stores already hold (RFC 0001 §11), which is why it lives on
// a 30-day window and a wiped stream costs nobody anything.
package events
import (
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go"
)
const (
// AttnStream carries every account's attention; subjects are
// attn.<account>.
AttnStream = "QUARRYATTN"
// attnWindow bounds one person's backlog. Thirty days is the
// inbox's reach; a thousand per subject is the cap RFC 0011 §8
// states, and both are per SUBJECT, so a busy account never
// evicts a quiet one's mail.
attnWindowAge = 30 * 24 * time.Hour
attnWindowMsgs = 1_000
)
// KindAttnGap is not a published event: TailAttention synthesises it
// when the resume point a caller asked for is older than anything the
// subject still holds. Seq carries the window's first sequence and Ref
// the sequence that was asked for, so a renderer can re-render rather
// than append onto a history it no longer shares. Same signal, same
// shape and same reason as chat's KindGap.
const KindAttnGap = "gap"
// AttentionEvent is one thing addressed to one person.
type AttentionEvent struct {
// Seq is the JetStream stream sequence — the event's identity and
// the `id:` line an SSE client resumes from. Stamped on delivery,
// never by the publisher.
Seq uint64 `json:"seq"`
// Kind is the vocabulary kind that raised it (RFC 0005), or
// "mention" / "reply" for the person-addressed events the
// vocabulary has no noun for.
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"`
}
// attnConfig is Limits retention with a per-subject window, like chat's:
// every reader tails the same copy, and the cap is per person.
var attnConfig = &nats.StreamConfig{
Name: AttnStream,
Subjects: []string{"attn.>"},
Retention: nats.LimitsPolicy,
Discard: nats.DiscardOld,
MaxAge: attnWindowAge,
MaxMsgsPerSubject: attnWindowMsgs,
}
// attnSubject is the account's subject, refusing any name that would
// not address exactly one person. A dot would split the subject into
// tokens, and `*` or `>` are NATS wildcards: either would let one
// account's publish or tail reach another's mail, which is the whole
// property this subject is here to provide.
func attnSubject(account string) (string, error) {
if account == "" {
return "", errors.New("events: attention needs an account")
}
if strings.ContainsAny(account, ".*> \t\r\n") {
return "", fmt.Errorf("events: %q cannot address one account", account)
}
return "attn." + account, nil
}
// Attend publishes one attention event for account and returns the
// sequence the broker assigned it — the event's permanent identity.
//
// Best-effort from the caller's point of view, exactly as Raise is: a
// dropped attention event costs a missed toast, never a fact.
func (s *Server) Attend(account string, e AttentionEvent) (uint64, error) {
if s == nil {
return 0, nil
}
subj, err := attnSubject(account)
if err != nil {
return 0, err
}
if e.At == "" {
e.At = time.Now().UTC().Format(time.RFC3339Nano)
}
e.Seq = 0 // the broker's, never the publisher's
raw, err := json.Marshal(e)
if err != nil {
return 0, fmt.Errorf("events: marshal attention: %w", err)
}
ack, err := s.js.Publish(subj, raw)
if err != nil {
return 0, fmt.Errorf("events: publish attention: %w", err)
}
return ack.Sequence, nil
}
// TailAttention delivers one account's events from seq from+1 onward
// (from == 0 means "new only, starting now") until cancel, as an
// ordered consumer: it never drops under backpressure, because a
// reconnecting client's whole point is "everything after N".
//
// cancel closes the returned channel once the consumer is stopped, so a
// caller ranging over it learns the stream ended instead of blocking on
// a channel nothing will write to again.
//
// When from names a sequence older than anything the subject still
// holds, JetStream silently clamps the start to the window's first
// message. TailAttention says so: the first delivery is a KindAttnGap
// signal and the replay follows it.
func (s *Server) TailAttention(account string, from uint64) (<-chan AttentionEvent, func()) {
out := make(chan AttentionEvent) // unbuffered: never drops, backpressure blocks the pump
subj, err := attnSubject(account)
if err != nil {
close(out)
return out, func() {}
}
done := make(chan struct{})
var gap *AttentionEvent
if from > 0 {
if first, _, berr := s.attnBounds(subj); berr == nil && first > 0 && from+1 < first {
gap = &AttentionEvent{Kind: KindAttnGap, Ref: from, Seq: first}
}
}
// An ordered consumer is push-mode only, so delivery comes through
// a callback that hands off to a pump goroutine which alone owns
// out: blocking the handoff is what gives this its never-drops
// property, and one owner is what makes closing out on cancel safe.
raw := make(chan AttentionEvent)
opts := []nats.SubOpt{nats.OrderedConsumer()}
if from > 0 {
opts = append(opts, nats.StartSequence(from+1))
} else {
opts = append(opts, nats.DeliverNew())
}
sub, serr := s.js.Subscribe(subj, func(msg *nats.Msg) {
var e AttentionEvent
if json.Unmarshal(msg.Data, &e) != nil {
return
}
if meta, merr := msg.Metadata(); merr == nil {
e.Seq = meta.Sequence.Stream
}
select {
case raw <- e:
case <-done:
}
}, opts...)
if serr != nil {
close(out)
return out, func() {}
}
go func() {
defer close(out)
if gap != nil {
select {
case out <- *gap:
case <-done:
return
}
}
for {
select {
case <-done:
return
case e := <-raw:
select {
case out <- e:
case <-done:
return
}
}
}
}()
var once sync.Once
return out, func() {
once.Do(func() {
sub.Unsubscribe()
close(done)
})
}
}
// AttentionPending counts one account's events after since — the number
// the broker already keeps, read off a subject-filtered consumer's
// NumPending without fetching a single message.
//
// Subject-filtered for the same reason chat.Unread is: sequences belong
// to the STREAM every account shares, so last-minus-since would count
// other people's mail as this person's unread.
func (s *Server) AttentionPending(account string, since uint64) (int, error) {
subj, err := attnSubject(account)
if err != nil {
return 0, err
}
opts := []nats.SubOpt{nats.AckNone(), nats.BindStream(AttnStream),
nats.InactiveThreshold(30 * time.Second)}
if since > 0 {
opts = append(opts, nats.StartSequence(since+1))
}
sub, err := s.js.PullSubscribe(subj, "", opts...)
if err != nil {
return 0, fmt.Errorf("events: attention pending: %w", err)
}
defer sub.Unsubscribe()
ci, cerr := sub.ConsumerInfo()
if cerr != nil {
return 0, fmt.Errorf("events: attention pending: %w", cerr)
}
return int(ci.NumPending), nil
}
// RecentAttention returns at most limit of an account's most recent
// events, oldest-first — the load-time state the inbox renders without
// any stream open (Principle 3).
func (s *Server) RecentAttention(account string, limit int) ([]AttentionEvent, error) {
if limit <= 0 {
limit = 50
}
subj, err := attnSubject(account)
if err != nil {
return nil, err
}
first, last, err := s.attnBounds(subj)
if err != nil || last == 0 {
return nil, err
}
start := first
// The stream holds every account, so "limit back" in SEQUENCES is
// not "limit events": start no earlier than the subject's own
// first, and trim to limit after the subject-filtered scan.
if last > uint64(limit) && last-uint64(limit) > first {
start = last - uint64(limit)
}
all, err := s.attnScan(subj, start)
if err != nil {
return nil, err
}
if len(all) > limit {
all = all[len(all)-limit:]
}
return all, nil
}
// attnScan reads one subject's events with seq >= start, in seq order.
func (s *Server) attnScan(subj string, start uint64) ([]AttentionEvent, error) {
sub, err := s.js.PullSubscribe(subj, "",
nats.StartSequence(start), nats.AckNone(), nats.BindStream(AttnStream),
nats.InactiveThreshold(30*time.Second))
if err != nil {
return nil, fmt.Errorf("events: attention scan: %w", err)
}
defer sub.Unsubscribe()
var out []AttentionEvent
for {
msgs, ferr := sub.Fetch(256, nats.MaxWait(200*time.Millisecond))
if ferr != nil && !errors.Is(ferr, nats.ErrTimeout) {
return nil, fmt.Errorf("events: attention scan: %w", ferr)
}
for _, m := range msgs {
var e AttentionEvent
if json.Unmarshal(m.Data, &e) != nil {
continue
}
if meta, merr := m.Metadata(); merr == nil {
e.Seq = meta.Sequence.Stream
}
out = append(out, e)
}
if len(msgs) < 256 {
break
}
}
return out, nil
}
// attnBounds answers one subject's first and last sequence (0, 0 when
// it holds nothing). Both come from subject-filtered views: the
// stream's own bounds describe every account at once.
func (s *Server) attnBounds(subj string) (first, last uint64, err error) {
msg, gerr := s.js.GetLastMsg(AttnStream, subj)
if gerr != nil {
if errors.Is(gerr, nats.ErrMsgNotFound) {
return 0, 0, nil
}
return 0, 0, fmt.Errorf("events: attention bounds: %w", gerr)
}
last = msg.Sequence
sub, serr := s.js.PullSubscribe(subj, "",
nats.AckNone(), nats.BindStream(AttnStream),
nats.InactiveThreshold(30*time.Second))
if serr != nil {
return 0, 0, fmt.Errorf("events: attention bounds: %w", serr)
}
defer sub.Unsubscribe()
msgs, ferr := sub.Fetch(1, nats.MaxWait(2*time.Second))
if ferr != nil && !errors.Is(ferr, nats.ErrTimeout) {
return 0, 0, fmt.Errorf("events: attention bounds: %w", ferr)
}
if len(msgs) == 0 {
return 0, last, nil
}
meta, merr := msgs[0].Metadata()
if merr != nil {
return 0, 0, fmt.Errorf("events: attention bounds: %w", merr)
}
return meta.Sequence.Stream, last, nil
}
// AttentionTip is the highest sequence one account currently holds, 0
// when the window holds none of theirs — what "mark everything read"
// resolves to. The account's own tip, from a subject-filtered view: the
// stream's last sequence belongs to every account at once.
func (s *Server) AttentionTip(account string) (uint64, error) {
subj, err := attnSubject(account)
if err != nil {
return 0, err
}
_, last, err := s.attnBounds(subj)
return last, err
}