// 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 (
"encoding/json"
"fmt"
"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"
)
// 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"`
}
// Room is the chat plane over one NATS JetStream context.
type Room struct {
nc *nats.Conn
js nats.JetStreamContext
}
// 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,
}
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 }
// 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
}
ack, err := r.js.Publish(subject(code), raw)
if err != nil {
return 0, 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.
//
// Implementation: an ephemeral ordered consumer walks the channel from
// its start (or from a computed StartSequence when before narrows the
// window enough to make that worthwhile) rather than fetching the whole
// history on every page. Fetching stops as soon as limit+1 messages with
// seq < before have been read from the stream in order, so a request
// for the newest page of a 10k-message window reads only the tail of
// it, not the whole thing.
func (r *Room) History(code string, before uint64, limit int) ([]Message, uint64, error) {
if limit <= 0 {
limit = 100
}
info, err := r.js.StreamInfo(StreamName, &nats.StreamInfoRequest{SubjectsFilter: subject(code)})
if err != nil {
return nil, 0, fmt.Errorf("chat: history: stream info: %w", err)
}
var firstSeq, lastSeq uint64
if ss, ok := info.State.Subjects[subject(code)]; ok && ss > 0 {
// Subjects map only carries counts, not first/last; fall back to
// the stream-wide bounds, which are a safe (if looser) start for
// a single-subject-per-stream design like this one is today.
firstSeq, lastSeq = info.State.FirstSeq, info.State.LastSeq
} else {
return nil, 0, nil // nothing ever posted on this channel
}
if before == 0 || before > lastSeq+1 {
before = lastSeq + 1
}
if before <= firstSeq {
return nil, 0, nil // everything requested is already outside the window
}
// Start close to where the page should begin: at most limit
// messages back from `before`, never before the window's first
// sequence. This is an estimate (some sequences may belong to other
// subjects sharing the stream in a future multi-subject layout); the
// per-message subject/seq filter below still applies, so a wrong
// estimate costs an extra page of fetching, never wrong data — and
// it is why a page never needs to scan the whole window.
start := firstSeq
if before > uint64(limit)+firstSeq {
start = before - uint64(limit)
}
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, 0, fmt.Errorf("chat: history: subscribe: %w", err)
}
defer sub.Unsubscribe()
var all []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, 0, fmt.Errorf("chat: history: fetch: %w", ferr)
}
for _, raw := range msgs {
meta, merr := raw.Metadata()
if merr != nil {
continue
}
seq := meta.Sequence.Stream
if seq >= before {
continue // read past the page's upper bound
}
var m Message
if json.Unmarshal(raw.Data, &m) == nil {
m.Seq = seq
all = append(all, m)
}
}
if len(msgs) < 256 {
break
}
}
var next uint64
if len(all) > limit {
// The oldest overflow entries become the next page's cursor.
next = all[len(all)-limit].Seq + 1
all = all[len(all)-limit:]
} else if start > firstSeq {
// Fewer than limit were fetched, but the estimate did not reach
// all the way to the window's start, so there may be more.
next = start
}
return all, next, 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.
func (r *Room) Tail(code string, from uint64) (<-chan Message, func()) {
out := make(chan Message) // unbuffered: never drops, backpressure blocks the callback below
done := make(chan struct{})
// An ordered consumer is push-mode only (the library rejects
// PullSubscribe + OrderedConsumer outright), so delivery comes
// through a callback rather than Fetch. Blocking that callback on
// out<-m is what gives Tail its never-drops property: JetStream's
// ordered consumer applies flow control, so a slow reader stalls
// the server's delivery to this consumer rather than losing
// anything, exactly the behaviour Subscribe's core-NATS 32-slot
// buffer does not have.
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 out <- m:
case <-done:
}
}, opts...)
if err != nil {
close(out)
return out, func() {}
}
var once sync.Once
cancel := func() {
once.Do(func() {
close(done)
sub.Unsubscribe()
})
}
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.unread) reads instead
// of paging History for it.
func (r *Room) Bounds(code string) (first, last uint64, err error) {
info, err := r.js.StreamInfo(StreamName, &nats.StreamInfoRequest{SubjectsFilter: subject(code)})
if err != nil {
return 0, 0, fmt.Errorf("chat: bounds: %w", err)
}
if _, ok := info.State.Subjects[subject(code)]; !ok {
return 0, 0, nil
}
return info.State.FirstSeq, info.State.LastSeq, 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) {
sub, err := r.js.PullSubscribe(subject(code), "",
nats.StartSequence(from), nats.AckNone(), nats.BindStream(StreamName),
nats.InactiveThreshold(30*time.Second))
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 {
byRef := make(map[uint64]Message, len(msgs))
for _, m := range msgs {
if m.Kind == KindEdit || m.Kind == KindDel {
byRef[m.Ref] = m
}
}
out := make([]Message, 0, len(msgs))
for _, m := range msgs {
if m.Kind == KindEdit || m.Kind == KindDel {
continue // tombstones never appear themselves
}
if t, ok := byRef[m.Seq]; ok {
switch t.Kind {
case KindEdit:
m.Text = t.Text
m.Edited = true
case KindDel:
m.Text = ""
m.Deleted = true
}
}
out = append(out, m)
}
return out
}