// Package events is the event plane (0001 §1): embedded NATS with
// JetStream, the dagnats/notify operational shape. Everything here lives
// under --data and is never truth — the stream is rederivable from the
// stores, which is what lets the disposability harness wipe it freely.
package events
import (
"errors"
"fmt"
"log"
"time"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
)
const (
// StreamName holds every store event; subjects follow the 0001 §4
// shape zit.evt.<project_code>.<kind>.
StreamName = "ZITEVT"
// CursorBucket is the KV bucket for watcher cursors (advisory —
// a lost cursor re-reads from zero).
CursorBucket = "cursors"
)
// The event stream's window: a live tail, not the stores' history.
// ingest's window-gap rule (window.go) asks whether a store's own events
// aged off the front of it unfolded, so these limits bound how long a
// node may be away or behind, not how long a store may sit quiet.
// Measured on production, 2026-09-27: events average ~50 KB (the largest
// 136 KB; parsed artifacts ride whole) and a busy hour carries ~85, so
// age is the limit that binds; the byte cap is the disk ceiling.
// JetStream reserves WindowBytes against free disk when the stream is
// made. WindowMsgs is unlimited (-1).
const (
WindowAge = time.Hour
WindowBytes = 1 << 30
WindowMsgs = -1
)
// Server is an embedded NATS server plus an in-process client.
type Server struct {
srv *server.Server
nc *nats.Conn
js nats.JetStreamContext
}
// Start runs an embedded server with JetStream storage under dir — the
// single-node default. The listen port is disabled; scale-out points
// the same binary at an external NATS with Connect instead (0003 §2).
func Start(dir string) (*Server, error) {
srv, err := server.NewServer(&server.Options{
DontListen: true,
JetStream: true,
StoreDir: dir,
// Parsed artifacts ride the stream whole; large stores (sqlite's
// baseline manifests, big clusters) blow the 1MB default. An
// external NATS must set max_payload to match (see Connect).
MaxPayload: 8 << 20,
})
if err != nil {
return nil, err
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
return nil, fmt.Errorf("events: embedded NATS not ready")
}
nc, err := nats.Connect("", nats.InProcessServer(srv))
if err != nil {
srv.Shutdown()
return nil, err
}
s, err := finishConnect(srv, nc)
if err != nil {
srv.Shutdown()
return nil, err
}
return s, nil
}
// Connect joins an external NATS cluster (the scale-out flag): N quarry
// nodes share one event plane, so each folds off the same stream and one
// holds the write lease. The external server MUST allow a max_payload of
// at least 8 MiB — parsed artifacts ride the stream whole and a 1 MiB
// default rejects large-store manifests. reconnect keeps the node
// serving reads across a NATS blip; writes refuse while disconnected
// (the lease coordinator is unreachable, 0001 §3).
func Connect(url string) (*Server, error) {
nc, err := nats.Connect(url,
nats.MaxReconnects(-1),
nats.ReconnectWait(time.Second),
)
if err != nil {
return nil, fmt.Errorf("events: dial %s: %w", url, err)
}
s, err := finishConnect(nil, nc)
if err != nil {
nc.Close()
return nil, err
}
return s, nil
}
// finishConnect derives the JetStream context and bootstraps the stream
// and KV buckets — the setup shared by embedded and external nodes.
func finishConnect(srv *server.Server, nc *nats.Conn) (*Server, error) {
js, err := nc.JetStream()
if err != nil {
nc.Close()
return nil, err
}
// Limits retention with a disk cap: every node's per-node durable
// consumer reads the FULL stream independently (work-queue retention
// would load-balance one copy across nodes, leaving each index
// partial — the store's truth demands every node fold every event).
// The Window limits bound the footprint so the stream stays a live
// tail, not the store's whole history (the 5.6 GB-beside-218 MB
// problem). Retention can't change on a live stream, so an
// older-config stream is dropped with its cursors, forcing a census
// republish that idempotent folds absorb. Two nodes may race the
// migration on a shared NATS, so a not-found delete is fine.
desired := &nats.StreamConfig{
Name: StreamName,
Subjects: []string{"zit.evt.>"},
Retention: nats.LimitsPolicy,
Discard: nats.DiscardOld,
MaxAge: WindowAge,
MaxBytes: WindowBytes,
MaxMsgs: WindowMsgs,
}
if info, ierr := js.StreamInfo(StreamName); ierr == nil && info.Config.Retention != nats.LimitsPolicy {
if derr := js.DeleteStream(StreamName); derr != nil && !errors.Is(derr, nats.ErrStreamNotFound) {
nc.Close()
return nil, fmt.Errorf("events: migrate stream: %w", derr)
}
_ = js.DeleteKeyValue(CursorBucket)
}
// Adding a stream that exists changes nothing, so a stream an older
// binary made would keep that binary's limits; the update makes the
// limits above the live ones. Raising limits keeps every message;
// lowering them drops the oldest past the new bound, as discard-old
// would have anyway. A refused update is not fatal: the forge ran on
// the stream's limits before, and the boot log and ops page show the
// live ones.
if _, err := js.AddStream(desired); err != nil {
if !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
nc.Close()
return nil, fmt.Errorf("events: stream: %w", err)
}
if _, uerr := js.UpdateStream(desired); uerr != nil {
log.Printf("events: stream limits not applied, running on the stream's own: %v", uerr)
}
}
if _, err := js.AddStream(vocabConfig); err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
nc.Close()
return nil, fmt.Errorf("events: vocabulary stream: %w", err)
}
if _, err := js.AddStream(attnConfig); err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
nc.Close()
return nil, fmt.Errorf("events: attention stream: %w", err)
}
if err := addCounterStream(nc); err != nil {
nc.Close()
return nil, err
}
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: CursorBucket}); err != nil {
if _, kerr := js.KeyValue(CursorBucket); kerr != nil {
nc.Close()
return nil, fmt.Errorf("events: kv: %w", err)
}
}
return &Server{srv: srv, nc: nc, js: js}, nil
}
// JS exposes the JetStream context for publishers and consumers.
func (s *Server) JS() nats.JetStreamContext { return s.js }
// Window is the event stream's span as it stands: a store whose own
// unfolded events sit below FirstSeq has lost them and is rebuilt.
type Window struct {
FirstSeq uint64
LastSeq uint64
Msgs uint64
Bytes uint64
// Oldest is when the first message landed; zero on an empty stream.
Oldest time.Time
// MaxAge is the stream's configured age limit, as it stands on the
// server rather than as this binary declares it.
MaxAge time.Duration
}
// Window reads the event stream's span.
func (s *Server) Window() (Window, error) {
info, err := s.js.StreamInfo(StreamName)
if err != nil {
return Window{}, err
}
st := info.State
w := Window{FirstSeq: st.FirstSeq, LastSeq: st.LastSeq, Msgs: st.Msgs, Bytes: st.Bytes, MaxAge: info.Config.MaxAge}
if st.Msgs > 0 {
w.Oldest = st.FirstTime
}
return w, nil
}
// Conn exposes the raw NATS connection for planes that need core
// pub/sub (chat live delivery) alongside JetStream.
func (s *Server) Conn() *nats.Conn { return s.nc }
// Cursors returns the advisory cursor KV bucket.
func (s *Server) Cursors() (nats.KeyValue, error) {
return s.js.KeyValue(CursorBucket)
}
// Watch returns a channel that ticks when anything happens on one
// repo's subjects — the change signal live pages coalesce on. code "*"
// is the NATS wildcard: one subscription covering every repo. The
// channel never blocks the publisher (capacity 1, drop-if-full: a tick
// means "something changed", not "N things changed"). cancel stops the
// subscription.
func (s *Server) Watch(code string) (<-chan struct{}, func()) {
ch := make(chan struct{}, 1)
// One repository's kinds exactly: names may contain dots, and a tick
// for a.b's push on a's page tells a's readers when private a.b moved.
subj := "zit.evt." + code + ".*"
if code == "*" {
subj = "zit.evt.>"
}
sub, err := s.nc.Subscribe(subj, func(*nats.Msg) {
select {
case ch <- struct{}{}:
default:
}
})
// Ensure the SUB has reached the server before returning: over an
// external NATS a publish that races an unflushed subscription would
// be missed (in-process embedded delivery is synchronous, so this is
// a no-op there).
if err == nil {
s.nc.Flush()
}
if err != nil {
close(ch)
return ch, func() {}
}
return ch, func() { sub.Unsubscribe() }
}
// WatchNoun signals when a VOCABULARY event lands for one noun — the
// quarry.evt subjects Raise publishes on, as against Watch's zit.evt
// artifact stream. Same coalescing contract: a tick means "something
// changed", never "N things changed".
//
// This is what a live page should wait on. A page that instead ticks on
// a timer asks the engine the same question over and over whether or
// not the answer moved, and every viewer multiplies that; one poller
// feeding many subscribers is the shape (RFC 0005).
func (s *Server) WatchNoun(code, noun string) (<-chan struct{}, func()) {
ch := make(chan struct{}, 1)
sub, err := s.nc.Subscribe("quarry.evt."+code+"."+noun+".*", func(*nats.Msg) {
select {
case ch <- struct{}{}:
default:
}
})
if err == nil {
s.nc.Flush()
}
if err != nil {
close(ch)
return ch, func() {}
}
return ch, func() { sub.Unsubscribe() }
}
func (s *Server) Close() {
s.nc.Close()
// External NATS: no embedded server to stop.
if s.srv != nil {
s.srv.Shutdown()
s.srv.WaitForShutdown()
}
}