Artifact [97ea372ca7]

Artifact 97ea372ca7abcef580454e1a8b1bd1641025028b89dea3f08da413650d6d4692:

• File internal/events/events.go — part of check-in [b0dc69c9a7] at 2026-08-15 11:14:05 on branch trunk — events: embedded NATS leg of the ingest pipeline (0001 §4)

  Watcher publishes parsed artifacts to zit.evt..artifact; durable pull consumer folds; cursors in KV; reset event on store rewrite. (user: dmestas, size: 2432) [annotate] [blame] [check-ins using]

// 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,
	})
	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)
}

func (s *Server) Close() {
	s.nc.Close()
	s.srv.Shutdown()
	s.srv.WaitForShutdown()
}