// 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 (
"fmt"
"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"
)
// 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
// listen port is disabled — phase 1 is one process; the external-NATS
// URL is the scale-out flag (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.
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
}
js, err := nc.JetStream()
if err != nil {
nc.Close()
srv.Shutdown()
return nil, err
}
if _, err := js.AddStream(&nats.StreamConfig{
Name: StreamName,
Subjects: []string{"zit.evt.>"},
}); err != nil && err != nats.ErrStreamNameAlreadyInUse {
nc.Close()
srv.Shutdown()
return nil, fmt.Errorf("events: stream: %w", err)
}
if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: CursorBucket}); err != nil {
if _, kerr := js.KeyValue(CursorBucket); kerr != nil {
nc.Close()
srv.Shutdown()
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 }
// 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. 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)
sub, err := s.nc.Subscribe("zit.evt."+code+".>", func(*nats.Msg) {
select {
case ch <- struct{}{}:
default:
}
})
if err != nil {
close(ch)
return ch, func() {}
}
return ch, func() { sub.Unsubscribe() }
}
func (s *Server) Close() {
s.nc.Close()
s.srv.Shutdown()
s.srv.WaitForShutdown()
}