Skip to content
quarry Artifact [3a43d3cc6f]
Login

Artifact [3a43d3cc6f]

Check-ins Using Download Hex

Artifact 3a43d3cc6f95760aeae939b537f3753048a44461e233635a17d29477c9446a85:

// 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"
	"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
// 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).
	// MaxBytes/MaxAge bound the footprint so the stream stays a live
	// tail, not the store's whole history (the 5.6 GB-beside-218 MB
	// problem); a node behind the window rebuilds from the store via
	// reindex. 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,
		MaxBytes:  1 << 30,   // 1 GiB ceiling across all repos
		MaxAge:    time.Hour, // live window; older needs a reindex
	}
	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)
	}
	if _, err := js.AddStream(desired); err != nil && !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
		nc.Close()
		return nil, fmt.Errorf("events: stream: %w", err)
	}
	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.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 }

// 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)
	sub, err := s.nc.Subscribe("zit.evt."+code+".>", 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() }
}

func (s *Server) Close() {
	s.nc.Close()
	// External NATS: no embedded server to stop.
	if s.srv != nil {
		s.srv.Shutdown()
		s.srv.WaitForShutdown()
	}
}