// Package chat is the 0001 §5 per-repository live channel: messages fan
// out over NATS to subscribers on any node, with a JetStream recent
// window (replay on join). Durability-in-the-DAG — folding batches into
// forum-thread artifacts so history clones with the repository — is the
// archiver's job (chat.promote, RFC 0011 §1) and the window is the
// interim durability. A chat post mutates no store, so it needs no
// write lease: any node accepts posts and publishes them to the shared
// channel.
//
// RFC 0011 §2: the message identity is the JetStream stream sequence on
// chat.<code> — nothing new is minted. Edit and delete are new messages
// that reference an earlier Seq (append-only; a reader Folds them), so
// history stays honest and replay stays trivial.
package chat
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log"
"sync"
"time"
"github.com/nats-io/nats.go"
)
const (
// StreamName holds every repo's channel; subjects are chat.<code>.
StreamName = "QUARRYCHAT"
// window bounds the replay-on-join history per channel.
windowAge = 7 * 24 * time.Hour
windowMsgs = 10_000
)
// Kind values for Message. "robot" is reserved for RFC 0011 §10 (Phase
// 2/4's activity robot); this package does not emit it yet.
const (
KindMsg = "msg"
KindEdit = "edit"
KindDel = "del"
KindRobot = "robot"
// KindGap is not a posted message: Tail synthesises it when the
// resume point a caller asked for is older than anything the
// channel 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.
KindGap = "gap"
)
// Message is one chat line.
type Message struct {
Seq uint64 `json:"seq"`
Kind string `json:"kind"` // "msg" | "edit" | "del" | "robot"
Ref uint64 `json:"ref,omitempty"` // edit/del: seq replaced; msg: reply-to
Repo string `json:"repo"`
User string `json:"user"` // "anonymous" when unauthenticated
Text string `json:"text"`
// Origin distinguishes a bridged or robot-authored line from an
// ordinary post: "" | "fossil" | "robot" (RFC 0011 §9, §10).
Origin string `json:"origin,omitempty"`
At string `json:"at"` // RFC3339Nano UTC, stamped by the receiving node
// Edited and Deleted are set by Fold, never by Post: they describe
// what a tombstone did to this message, not a field anyone posts.
Edited bool `json:"edited,omitempty"`
Deleted bool `json:"deleted,omitempty"`
}
// RobotLog remembers which of a channel's sequences are the activity
// robot's own lines, so Unread can subtract them from the count the
// broker already holds instead of reading the window to tell them
// apart (RFC 0011 §10).
//
// A robot line never raises anybody's badge, so nothing ever moves the
// read pointer past one: on a channel the robot talks in and nobody
// answers, "count what is pending and scan it" is an unbounded scan
// that grows for as long as the pointer stays where it is.
//
// nil is a forge with no OLTP, which also runs no robot: with nothing
// to subtract, the broker's count is already the whole answer.
type RobotLog interface {
// RecordRobotLine notes one robot line, idempotently on (code, seq).
RecordRobotLine(code string, seq uint64, at string) error
// RobotLines counts this channel's robot lines with seq > since.
RobotLines(code string, since uint64) (int, error)
}
// Room is the chat plane over one NATS JetStream context.
type Room struct {
nc *nats.Conn
js nats.JetStreamContext
// Robots is the robot-line counter Unread subtracts. Set by the
// composition root; nil leaves the broker's count unadjusted.
Robots RobotLog
}
// New ensures the chat stream exists (limits retention, per-subject
// window) and returns the room. It shares the caller's NATS connection.
func New(nc *nats.Conn, js nats.JetStreamContext) (*Room, error) {
cfg := &nats.StreamConfig{
Name: StreamName,
Subjects: []string{"chat.>"},
Retention: nats.LimitsPolicy,
Discard: nats.DiscardOld,
MaxAge: windowAge,
MaxMsgsPerSubject: windowMsgs,
// Duplicate suppression, done by the broker rather than by a
// map in one process: a second identical post inside this
// window is ACKED with the ORIGINAL message's sequence and
// never stored. That is what makes a double-tapped Send, a
// re-submitted form and a retried request all land on the same
// message instead of three copies of it. See postID.
Duplicates: dupWindow,
}
if _, err := js.AddStream(cfg); err != nil && err != nats.ErrStreamNameAlreadyInUse {
// An older config (different retention/subjects) is updated in place.
if _, uerr := js.UpdateStream(cfg); uerr != nil {
return nil, fmt.Errorf("chat: stream: %w", err)
}
}
return &Room{nc: nc, js: js}, nil
}
func subject(code string) string { return "chat." + code }
// dupWindow is how long an identical post is treated as the SAME post.
//
// Ten seconds is chosen against the two behaviours it has to separate.
// An accident — a double tap, a form re-submitted by the back button, a
// client retry after a timeout — arrives within a second or two. A
// person deliberately saying "ok" twice takes longer than that to type
// it again, and is entitled to. A long window would quietly swallow
// real messages; a short one would let the accidents through.
const dupWindow = 10 * time.Second
// postID is the identity of a post's CONTENT: same channel, same
// author, same kind, same text, same reply target. Two publishes that
// agree on all five inside dupWindow are one post that was sent twice.
//
// The text is hashed rather than carried: a Nats-Msg-Id is a header, and
// a 4KB message body does not belong in one.
func postID(code string, m Message) string {
h := sha256.New()
for _, part := range []string{code, m.User, m.Kind, m.Text} {
h.Write([]byte(part))
h.Write([]byte{0})
}
fmt.Fprintf(h, "%d", m.Ref)
return hex.EncodeToString(h.Sum(nil)[:16])
}
// Post stamps the message with the current UTC time, publishes it to
// the repo's channel, and returns the JetStream stream sequence the
// broker assigned it — the message's permanent identity (RFC 0011 §2).
func (r *Room) Post(code string, m Message) (uint64, error) {
if m.User == "" {
m.User = "anonymous"
}
if m.Kind == "" {
m.Kind = KindMsg
}
m.At = time.Now().UTC().Format(time.RFC3339Nano)
raw, err := json.Marshal(m)
if err != nil {
return 0, err
}
// The timestamp is deliberately NOT part of the identity below —
// two sends a second apart differ only by it, and that is exactly
// the pair this is here to collapse.
ack, err := r.js.Publish(subject(code), raw, nats.MsgId(postID(code, m)))
if err != nil {
return 0, err
}
// A duplicate was not stored, so there is no new robot line to
// record and nothing new happened: the caller gets the sequence of
// the message that IS there, which is the answer it wanted.
if ack.Duplicate {
return ack.Sequence, nil
}
// The robot's own lines are remembered as they are posted: the
// alternative is reading them back out of the window every time a
// badge is drawn, which is what Unread refuses to do.
if m.Kind == KindRobot && r.Robots != nil {
if err := r.Robots.RecordRobotLine(code, ack.Sequence, m.At); err != nil {
log.Printf("chat: robot line %s#%d: %v", code, ack.Sequence, err)
}
}
return ack.Sequence, nil
}
// History returns at most limit messages with seq < before (before == 0
// means "the latest window"), oldest-first, plus next — the seq to pass
// as before for the older page, or 0 when there is nothing older.
//
// before is EXCLUSIVE, so next is the oldest returned message's own seq:
// the next page ends where this one begins, with no message counted
// twice.
//
// Cost: the stream holds every repository's channel, so a channel's
// sequences are sparse — "limit messages back" is not "limit sequences
// back". The scan therefore starts at before-limit and, if that window
// held fewer than limit of THIS channel's messages, doubles the reach
// and scans again, until the page fills or the channel's first sequence
// is reached. A dense channel costs one scan of about limit messages; a
// channel that is one in N of the stream costs at most ceil(log2 N)
// scans, the last of which reads about 2*limit messages. Every scan is
// subject-filtered by the broker, so the cost is in round-trips, never
// in messages belonging to other channels.
func (r *Room) History(code string, before uint64, limit int) ([]Message, uint64, error) {
if limit <= 0 {
limit = 100
}
first, last, err := r.Bounds(code)
if err != nil {
return nil, 0, err
}
if last == 0 {
return nil, 0, nil // nothing ever posted on this channel
}
if before == 0 || before > last+1 {
before = last + 1
}
if before <= first {
return nil, 0, nil // everything requested is already outside the window
}
reach := uint64(limit)
for {
start := first
if before-first > reach {
start = before - reach
}
all, serr := r.scan(code, start, before)
if serr != nil {
return nil, 0, serr
}
if len(all) < limit && start > first {
reach *= 2
continue
}
var next uint64
if len(all) > limit {
all = all[len(all)-limit:]
}
if len(all) > 0 && !(start <= first && all[0].Seq == first) {
// The page does not begin at the channel's own first
// message, so an older page exists; before is exclusive, so
// the cursor is this page's oldest seq itself.
next = all[0].Seq
}
return all, next, nil
}
}
// Forward is History's mirror: at most limit messages with seq > after,
// oldest-first, plus next — the seq to pass as after for the page below
// this one, or 0 when this page reaches the channel's newest message.
//
// It exists because a permalink needs BOTH directions. A reader who
// followed "#8" from a ticket needs the messages that answered it as
// much as the ones that provoked it, and "the next five messages" is
// not "the next five sequences": the stream is shared by every
// repository's channel, so a quiet channel's own messages are sparse
// among its neighbours'. The reach therefore doubles until the page
// fills, exactly as History does going backwards.
//
// Whether more exists is read from the SCAN — one message beyond the
// page is fetched and dropped — never from the channel's tip. A cursor
// derived from a bound is a cursor that is wrong whenever the bound is.
func (r *Room) Forward(code string, after uint64, limit int) ([]Message, uint64, error) {
if limit <= 0 {
limit = 100
}
first, last, err := r.Bounds(code)
if err != nil {
return nil, 0, err
}
if last == 0 || after >= last {
return nil, 0, nil
}
if first > 0 && after < first-1 {
after = first - 1
}
reach := uint64(limit)
for {
end := after + 1 + reach // exclusive
if end > last+1 {
end = last + 1
}
all, serr := r.scan(code, after+1, end)
if serr != nil {
return nil, 0, serr
}
// One past the page is what says "there is more", so the loop
// wants limit+1 before it stops widening.
if len(all) <= limit && end < last+1 {
reach *= 2
continue
}
var next uint64
if len(all) > limit {
all = all[:limit]
next = all[len(all)-1].Seq
}
return all, next, nil
}
}
// scan reads this channel's messages with start <= seq < before, in seq
// order. The subject filter is the broker's: a pull subscription bound
// to one subject never delivers another channel's messages, which is
// what keeps every per-channel number honest on a shared stream.
func (r *Room) scan(code string, start, before uint64) ([]Message, error) {
sub, err := r.js.PullSubscribe(subject(code), "",
nats.StartSequence(start), nats.AckNone(), nats.BindStream(StreamName),
nats.InactiveThreshold(30*time.Second))
if err != nil {
return nil, fmt.Errorf("chat: history: subscribe: %w", err)
}
defer sub.Unsubscribe()
var out []Message
for {
msgs, ferr := sub.Fetch(256, nats.MaxWait(200*time.Millisecond))
if ferr == nats.ErrTimeout && len(msgs) == 0 {
break
}
if ferr != nil && ferr != nats.ErrTimeout {
return nil, fmt.Errorf("chat: history: fetch: %w", ferr)
}
done := false
for _, raw := range msgs {
meta, merr := raw.Metadata()
if merr != nil {
continue
}
seq := meta.Sequence.Stream
if seq >= before {
done = true // read past the page's upper bound
break
}
var m Message
if json.Unmarshal(raw.Data, &m) == nil {
m.Seq = seq
out = append(out, m)
}
}
if done || len(msgs) < 256 {
break
}
}
return out, nil
}
// Tail delivers messages from seq from+1 onward (from == 0 means "new
// only, starting now") until cancel, as an ordered consumer: unlike
// Subscribe, it never drops under backpressure — a slow reader blocks
// the consumer's own delivery instead of losing a message, because a
// reconnecting client's whole point is "everything after N" (RFC 0011
// §2). Subscribe remains for tick-style "something changed" uses, where
// dropping is the correct behaviour.
//
// cancel closes the returned channel once the consumer is stopped, so a
// caller bridging or ranging over it learns the stream ended instead of
// blocking on a channel nothing will ever write to again.
//
// When from names a sequence older than anything the channel still
// holds, JetStream silently clamps the start to the window's first
// message — a resume that looks successful and is not. Tail says so:
// the first delivery is a KindGap signal (Ref = the from asked for, Seq
// = the window's first) and the replay follows it, so a renderer can
// re-render rather than append onto a history it no longer shares.
func (r *Room) Tail(code string, from uint64) (<-chan Message, func()) {
out := make(chan Message) // unbuffered: never drops, backpressure blocks the pump below
done := make(chan struct{})
var gap *Message
if from > 0 {
if first, _, err := r.Bounds(code); err == nil && first > 0 && from+1 < first {
gap = &Message{Kind: KindGap, Repo: code, Ref: from, Seq: first}
}
}
// An ordered consumer is push-mode only (the library rejects
// PullSubscribe + OrderedConsumer outright), so delivery comes
// through a callback rather than Fetch. The callback hands off to a
// pump goroutine that alone owns out: blocking the handoff is what
// gives Tail its never-drops property (JetStream's ordered consumer
// applies flow control, so a slow reader stalls the server's
// delivery rather than losing anything), and having one owner is
// what makes closing out on cancel safe.
raw := make(chan Message)
opts := []nats.SubOpt{nats.OrderedConsumer()}
if from > 0 {
opts = append(opts, nats.StartSequence(from+1))
} else {
opts = append(opts, nats.DeliverNew())
}
sub, err := r.js.Subscribe(subject(code), func(msg *nats.Msg) {
var m Message
if json.Unmarshal(msg.Data, &m) != nil {
return
}
if meta, merr := msg.Metadata(); merr == nil {
m.Seq = meta.Sequence.Stream
}
select {
case raw <- m:
case <-done:
}
}, opts...)
if err != 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 m := <-raw:
select {
case out <- m:
case <-done:
return
}
}
}
}()
var once sync.Once
cancel := func() {
once.Do(func() {
sub.Unsubscribe()
close(done)
})
}
return out, cancel
}
// Bounds answers the window's first and last sequence for one channel
// (0, 0 when nothing has ever been posted) — what chat.get needs to
// tell "never existed" from "aged out of the window" apart, and what a
// caller needing the channel's current tip (chat.read) reads instead of
// paging History for it.
//
// Both numbers come from subject-filtered views. The stream is shared by
// every repository's channel, so its own FirstSeq/LastSeq describe the
// FLEET, not this channel: reading them here is how a quiet repo ends up
// claiming a busy neighbour's messages.
func (r *Room) Bounds(code string) (first, last uint64, err error) {
msg, gerr := r.js.GetLastMsg(StreamName, subject(code))
if gerr != nil {
if errors.Is(gerr, nats.ErrMsgNotFound) {
return 0, 0, nil
}
return 0, 0, fmt.Errorf("chat: bounds: last: %w", gerr)
}
last = msg.Sequence
first, err = r.firstSeq(code)
if err != nil {
return 0, 0, err
}
return first, last, nil
}
// firstSeq is the oldest sequence this channel still holds, 0 when it
// holds none: a subject-filtered consumer delivering from the start,
// read for its first message's sequence alone.
func (r *Room) firstSeq(code string) (uint64, error) {
sub, err := r.js.PullSubscribe(subject(code), "",
nats.AckNone(), nats.BindStream(StreamName),
nats.InactiveThreshold(30*time.Second))
if err != nil {
return 0, fmt.Errorf("chat: bounds: first: %w", err)
}
defer sub.Unsubscribe()
// Ask how much there is before asking for it: a Fetch on an empty
// channel would wait out its whole timeout to return nothing.
if ci, cerr := sub.ConsumerInfo(); cerr == nil && ci.NumPending == 0 {
return 0, nil
}
msgs, ferr := sub.Fetch(1, nats.MaxWait(2*time.Second))
if ferr != nil && ferr != nats.ErrTimeout {
return 0, fmt.Errorf("chat: bounds: first: %w", ferr)
}
if len(msgs) == 0 {
return 0, nil
}
meta, merr := msgs[0].Metadata()
if merr != nil {
return 0, fmt.Errorf("chat: bounds: first: %w", merr)
}
return meta.Sequence.Stream, nil
}
// Unread counts this channel's messages after since, excluding the
// activity robot's (RFC 0011 §10: robot lines never count as unread —
// they are the channel talking about the repository, and a badge that
// lights up for them is a badge people learn to ignore).
//
// The broker's NumPending is the whole count, and the robot-line store
// says how many of them are the robot's: two numbers, both O(1), and
// no message is fetched. That matters because nothing advances the
// read pointer past a robot line — a channel whose robot is chatty and
// whose reader is away has an unread run that only grows, and reading
// it back per badge render is a scan whose size is the reader's
// absence.
//
// It is emphatically not last-since: sequences are the STREAM's, shared
// by every channel, so that subtraction counts other repositories'
// traffic as this reader's unread mail.
func (r *Room) Unread(code string, since uint64) (int, error) {
opts := []nats.SubOpt{nats.AckNone(), nats.BindStream(StreamName),
nats.InactiveThreshold(30 * time.Second)}
if since > 0 {
opts = append(opts, nats.StartSequence(since+1))
}
sub, err := r.js.PullSubscribe(subject(code), "", opts...)
if err != nil {
return 0, fmt.Errorf("chat: unread: %w", err)
}
defer sub.Unsubscribe()
ci, cerr := sub.ConsumerInfo()
if cerr != nil {
return 0, fmt.Errorf("chat: unread: %w", cerr)
}
n := int(ci.NumPending)
if n == 0 || r.Robots == nil {
return n, nil
}
// Bounded by the WINDOW, not by the reader's pointer alone. Robot
// rows outlive the messages they describe (they are swept by age,
// the channel by age and size), so a channel whose robot was busy
// last week would otherwise subtract lines the broker is no longer
// counting — and clamp a real unread run to zero.
from := since
if first, _, ferr := r.Bounds(code); ferr == nil && first > 0 && first-1 > from {
from = first - 1
}
bots, berr := r.Robots.RobotLines(code, from)
if berr != nil {
// A store that cannot answer leaves the honest ceiling rather
// than no number at all: an over-count is a badge that is too
// eager, a failure is a badge that lies about being empty.
log.Printf("chat: unread: robot lines for %s: %v", code, berr)
return n, nil
}
if n -= bots; n < 0 {
n = 0
}
return n, nil
}
// Range returns messages with seq >= from (and, when to > 0, seq <= to)
// in seq order — the bounded forward scan chat.get and chat.edit/delete
// use to fetch one message (Range(code, seq, seq)) or its direct
// replies (Range(code, seq+1, 0)) without a second index.
func (r *Room) Range(code string, from, to uint64) ([]Message, error) {
// from == 0 means "from the beginning", and in JetStream that is
// DeliverAll — NOT StartSequence(0). Asking for start-sequence zero
// sets the policy to "deliver by start sequence" while leaving the
// sequence unset, which the server rejects outright:
//
// consumer delivery policy is deliver by start sequence, but
// optional start sequence is not set
//
// Bounds answers (0, 0) for a channel nobody has ever posted in, so
// every empty channel took this path and failed — 1,308 times in a
// day, once per repository per backfill tick.
opts := []nats.SubOpt{nats.AckNone(), nats.BindStream(StreamName),
nats.InactiveThreshold(30 * time.Second)}
if from > 0 {
opts = append(opts, nats.StartSequence(from))
} else {
opts = append(opts, nats.DeliverAll())
}
sub, err := r.js.PullSubscribe(subject(code), "", opts...)
if err != nil {
return nil, fmt.Errorf("chat: range: subscribe: %w", err)
}
defer sub.Unsubscribe()
var out []Message
for {
msgs, ferr := sub.Fetch(256, nats.MaxWait(200*time.Millisecond))
if ferr == nats.ErrTimeout && len(msgs) == 0 {
break
}
if ferr != nil && ferr != nats.ErrTimeout {
return nil, fmt.Errorf("chat: range: fetch: %w", ferr)
}
for _, raw := range msgs {
meta, merr := raw.Metadata()
if merr != nil {
continue
}
seq := meta.Sequence.Stream
if to > 0 && seq > to {
continue
}
var m Message
if json.Unmarshal(raw.Data, &m) == nil {
m.Seq = seq
out = append(out, m)
}
}
if len(msgs) < 256 {
break
}
}
return out, nil
}
// Subscribe delivers live messages on one channel until cancel. The
// channel buffers a little; a slow reader drops the oldest (chat is
// live, not a log — the window covers catch-up, and Tail is the
// never-drops alternative for a reconnecting client).
func (r *Room) Subscribe(code string) (<-chan Message, func()) {
ch := make(chan Message, 32)
sub, err := r.nc.Subscribe(subject(code), func(msg *nats.Msg) {
var m Message
if json.Unmarshal(msg.Data, &m) != nil {
return
}
select {
case ch <- m:
default: // drop under backpressure
}
})
if err != nil {
close(ch)
return ch, func() {}
}
r.nc.Flush() // register the SUB before returning (external NATS races)
return ch, func() { sub.Unsubscribe() }
}
// Fold applies edit/delete tombstones to the messages they reference and
// drops the tombstones themselves from the result, so a renderer never
// has to know the append-only shape underneath. msgs must be in seq
// order (History and Tail both deliver that way); the result preserves
// the order of what remains.
func Fold(msgs []Message) []Message {
// A delete is terminal: an edit naming a message that also carries a
// del tombstone changes nothing, whichever order the two arrived in.
// Anything else would let an edit resurrect text someone deleted
// just by being published second — or first.
deleted := make(map[uint64]bool)
edited := make(map[uint64]Message)
for _, m := range msgs {
switch m.Kind {
case KindDel:
deleted[m.Ref] = true
case KindEdit:
edited[m.Ref] = m
}
}
out := make([]Message, 0, len(msgs))
for _, m := range msgs {
switch m.Kind {
case KindEdit, KindDel:
continue // tombstones never appear themselves
case KindGap:
continue // a signal about the stream, not a line in it
}
switch {
case deleted[m.Seq]:
m.Text = ""
m.Deleted = true
default:
if t, ok := edited[m.Seq]; ok {
m.Text = t.Text
m.Edited = true
}
}
out = append(out, m)
}
return out
}