mesh-sdk/go/buscompose/state.go

package buscompose

import (
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/announce"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"sort"
	"strings"
	"time"
)

// A module's state on the bus (novox/hq ADR 0201, design 32 §4, design 25 §3).
//
// A module names the state it keeps (`state`) and the state of others it reads (`reads`), and each
// is a key-value bucket: the server's own last-per-subject stream with direct reads, delete markers
// and watches, which is the state relationship the mesh already uses for declarations, opened to
// modules. The controller creates every bucket from the catalogue — from registration, like a
// seat's stream, so a reader may watch before the owner runs anywhere — and no module can.
//
// Pure, like everything else in this package that decides what the bus holds; jetstream.go is the
// part that asks a server.

// The mesh's caps on a bucket, the same for every module: a value is a piece of state, not a file,
// and a bucket that grew without bound would be one module filling the bus's disk for everyone.
const (
	StateMaxValueBytes = 256 * 1024
	StateMaxBytes      = 64 * 1024 * 1024
)

// A Bucket is one module's declared state as the bus holds it.
type Bucket struct {
	Module string
	Name   string
	// History is how many values a key keeps; zero is one.
	History int
	// TTLSeconds is how long a value lives; zero is until replaced or deleted.
	TTLSeconds int
	// PerMachine says each machine's instance writes only the key named for its machine (novox/hq ADR
	// 0260), and is granted that key and no other.
	PerMachine bool
	// KeyTTL says the bucket allows a time to live per key (novox/hq ADR 0433 §1): its owner writes each key with its
	// own, and the server writes a delete marker, kept KeyTTLMarkerTTL, when it ends.
	KeyTTL bool
}

// KeyTTLMarkerTTL is how long the marker an expired key leaves is kept: long enough for a watcher to hear it.
const KeyTTLMarkerTTL = time.Minute

// BucketName is the bucket a module's state lives in: the module and the local name joined by an
// underscore, which neither may contain, so two modules can never derive one bucket.
func BucketName(module, name string) string { return module + "_" + name }

// Bucket is this bucket's name on the bus.
func (b Bucket) Bucket() string { return BucketName(b.Module, b.Name) }

// Why is carried into the server's description of the bucket, so somebody reading the server's
// own state finds whose it is and why it is kept.
func (b Bucket) Why() string {
	return fmt.Sprintf("%s's state %q (novox/hq ADR 0201): its current value per key, written by %s, "+
		"read by whatever declares it reads it; kept when %s is unassigned, because it is data",
		b.Module, b.Name, b.Module, b.Module)
}

// bucketOfRead is the bucket a read names, `<module>.<name>`, or false when it names none.
func bucketOfRead(read string) (string, bool) {
	at := strings.LastIndex(read, ".")
	if at <= 0 || at == len(read)-1 {
		return "", false
	}
	module, name := read[:at], read[at+1:]
	if !safeSubject.MatchString(module) || !safeSubject.MatchString(name) {
		return "", false
	}
	return BucketName(module, name), true
}

// readGrants is what reaching a bucket's keys needs: binding to it, a direct get, an ordered consumer for listing and
// watching and the deletion of that consumer and its flow control. key is one key's subject tail, a pattern such as
// `one.>` for a prefix of keys, or empty for every key. The bucket is made with AllowDirect, so a get is the direct get
// and never the stream's message get.
func readGrants(bucket, key string) []string {
	stream := "KV_" + bucket
	out := []string{"$JS.API.STREAM.INFO." + stream}
	if key == "" {
		out = append(out,
			"$JS.API.DIRECT.GET."+stream+".>",
			"$JS.API.CONSUMER.CREATE."+stream+".>")
	} else {
		out = append(out,
			"$JS.API.DIRECT.GET."+stream+".$KV."+bucket+"."+key,
			"$JS.API.CONSUMER.CREATE."+stream+".*.$KV."+bucket+"."+key)
	}
	return append(out,
		"$JS.API.CONSUMER.DELETE."+stream+".>",
		"$JS.FC."+stream+".>")
}

// NodeKeysGrants is what a node's node-engine is granted of a bucket it writes by node, `<node>.>` and no other key
// (novox/hq ADR 0429 §1: each key one writer, enforced by credentials): the writes of those keys and what a get,
// create, update and watch of them need. The node name is checked as every subject token is.
func NodeKeysGrants(bucket, node string) ([]string, error) {
	if !safeSubject.MatchString(node) {
		return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node)
	}
	return append([]string{"$KV." + bucket + "." + node + ".>"}, readGrants(bucket, node+".>")...), nil
}

// stateGrants is what a principal publishes to reach the state its modules keep and read: for every
// bucket, binding to it, reading a key directly, and an ordered consumer for listing and watching,
// created and deleted on the bucket's own stream, with its flow control answered; for a bucket an
// owner keeps, writing under the bucket's own subjects too.
//
// **Measured against a running server, 2026-10-04** (novox/hq research 024), and each one is there
// because leaving it out failed: without STREAM.INFO nothing binds; without DIRECT.GET nothing is
// read; without CONSUMER.CREATE no key is listed and nothing is watched; without CONSUMER.DELETE a
// watch cannot be stopped and lingers on the server. A write outside these is refused by the server
// — and reaches the writer as a timeout, not a refusal, which is why the runtime refuses first.
//
// **One key, where the state is per machine** (novox/hq ADR 0260). The owner's instance on a machine
// writes and reads the key named for that machine, and a holder granted a read for its machine reads
// that key: a direct get of the key's own subject, and a consumer whose filter is that subject — the
// server's create-with-filter form, which a client uses for a watch of one key and which the server
// holds to the filter it names. What cannot be narrowed by key: binding (STREAM.INFO), deleting a
// consumer (named at random by the client) and flow control; none of them reads or writes a value.
func stateGrants(a stateAccess) []string {
	var out []string
	read := func(bucket, key string) { out = append(out, readGrants(bucket, key)...) }
	machineKey := ""
	if safeSubject.MatchString(a.Node) {
		machineKey = a.Node
	}
	perMachine := map[string]bool{}
	for _, n := range a.PerMachine {
		perMachine[n] = true
	}
	for _, name := range a.Keeps {
		if !safeSubject.MatchString(name) {
			continue
		}
		bucket := BucketName(a.Module, name)
		if perMachine[name] {
			if machineKey == "" {
				continue // a machine with no usable name reaches none of it
			}
			read(bucket, machineKey)
			out = append(out, "$KV."+bucket+"."+machineKey)
			continue
		}
		read(bucket, "")
		out = append(out, "$KV."+bucket+".>")
	}
	// **And it says what it keeps** (novox/hq ADR 0495 clause 2, issue 1790): a module keeping state announces it on its
	// own event subject, from its code; refused, its readers are read as before.
	if len(a.Keeps) > 0 && safeSubject.MatchString(a.Module) {
		out = append(out, announce.Subject(a.Module))
	}
	for _, r := range a.Reads {
		if bucket, ok := bucketOfRead(r); ok {
			if NodeKeyedReads[r] {
				if machineKey != "" {
					read(bucket, machineKey+".>")
				}
				continue
			}
			read(bucket, "")
		}
	}
	for _, k := range a.KeyedReads {
		if bucket, ok := bucketOfRead(k.Read); ok && safeSubject.MatchString(k.Key) {
			read(bucket, k.Key)
		}
	}
	return out
}

// NodeKeyedReads are the states keyed `<node>.>` whose reader reads its own machine's keys alone, whatever its `reads`
// names: the node-engine's data items, each node's disk-load reading its own node's (novox/hq ADR 0577 clause 1). A
// machine with no usable name reaches none of them.
var NodeKeyedReads = map[string]bool{delivery.EngineDataRead: true}

// StreamCreatedAdvisory is the subject the bus says a stream was created on (the server's own advisory), here for a
// bucket's stream KV_<bucket>.
func StreamCreatedAdvisory(bucket string) string {
	return "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" + bucket
}

// stateCreatedHeard is what a principal subscribes to so that a bucket it may bind and that is not on the bus yet is
// bound when it appears (novox/hq issue 1790, 2026-10-26: since the grants part creates a manifest's buckets, ADR 0547,
// a module's delivery can start it before its new state exists): the bus's advisory that the bucket's stream was
// created, for each bucket stateGrants lets it bind, and no other. Read-only: the server publishes it in the mesh's own
// account, and hearing it changes nothing.
func stateCreatedHeard(a stateAccess) []string {
	var out []string
	seen := map[string]bool{}
	add := func(bucket string) {
		if !seen[bucket] {
			seen[bucket] = true
			out = append(out, StreamCreatedAdvisory(bucket))
		}
	}
	perMachine := map[string]bool{}
	for _, n := range a.PerMachine {
		perMachine[n] = true
	}
	for _, name := range a.Keeps {
		if !safeSubject.MatchString(name) || (perMachine[name] && !safeSubject.MatchString(a.Node)) {
			continue
		}
		add(BucketName(a.Module, name))
	}
	for _, r := range a.Reads {
		if bucket, ok := bucketOfRead(r); ok {
			add(bucket)
		}
	}
	for _, k := range a.KeyedReads {
		if bucket, ok := bucketOfRead(k.Read); ok && safeSubject.MatchString(k.Key) {
			add(bucket)
		}
	}
	return out
}

// KeyedRead is one key of another module's state a principal reads, and no other key of it.
type KeyedRead struct {
	Read string `json:"read"`
	Key  string `json:"key"`
}

// stateAccess is what one principal may do with state: the module's own buckets (some per machine),
// what it reads whole, and what it reads for its machine's key alone.
type stateAccess struct {
	Module, Node      string
	Keeps, PerMachine []string
	Reads             []string
	KeyedReads        []KeyedRead
}

// perMachineNames is the local names of a module's buckets keyed by machine.
func perMachineNames(buckets []Bucket) []string {
	var out []string
	for _, b := range buckets {
		if b.PerMachine {
			out = append(out, b.Name)
		}
	}
	return out
}

// stateIssuedFor is every bucket a module's code may reach on a machine, as its membership lists them.
func stateIssuedFor(d Declared, node string) []StateIssued {
	var out []StateIssued
	for _, b := range d.State {
		issued := StateIssued{Name: b.Name, Bucket: BucketName(d.Module, b.Name), Writes: true}
		if b.PerMachine {
			issued.Key = node
		}
		out = append(out, issued)
	}
	for _, r := range d.Reads {
		if bucket, ok := bucketOfRead(r); ok {
			out = append(out, StateIssued{Name: r, Bucket: bucket})
		}
	}
	for _, k := range d.KeyedReads {
		if bucket, ok := bucketOfRead(k.Read); ok {
			out = append(out, StateIssued{Name: k.Read, Bucket: bucket, Key: k.Key})
		}
	}
	return out
}

// stateNames is the local names of a module's own buckets.
func stateNames(buckets []Bucket) []string {
	out := make([]string, 0, len(buckets))
	for _, b := range buckets {
		out = append(out, b.Name)
	}
	return out
}

// A BucketAsserter is the part of a JetStream connection bucket assertion needs.
type BucketAsserter interface {
	// EnsureBucket creates the bucket if absent and brings its options to match if present, never
	// discarding what it holds.
	EnsureBucket(b Bucket) error
	// BucketNames is every key-value bucket on the server.
	BucketNames() ([]string, error)
}

// RaiseBuckets asserts every declared bucket and answers the buckets on the server that nothing
// declares any more.
//
// **Those are reported, never removed** (novox/hq ADR 0201, ADR 0030): what a module stored is
// data, and a manifest edited, a module renamed or a catalogue entry dropped is an ordinary day's
// work that must not take data with it. Removing one is a person's act.
func RaiseBuckets(a BucketAsserter, buckets []Bucket) (undeclared []string, err error) {
	sorted := append([]Bucket(nil), buckets...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Bucket() < sorted[j].Bucket() })
	declared := map[string]bool{}
	for _, b := range sorted {
		if err := a.EnsureBucket(b); err != nil {
			return nil, fmt.Errorf("asserting %s's state %q: %w", b.Module, b.Name, err)
		}
		declared[b.Bucket()] = true
	}
	names, err := a.BucketNames()
	if err != nil {
		return nil, fmt.Errorf("listing the bus's state: %w", err)
	}
	for _, n := range names {
		// A seat's cancelled set is the mesh's own (novox/hq ADR 0219), not a module's state. The ended controller's
		// buckets are undeclared like any other until they are removed (ADR 0590 clause 6).
		if !declared[n] && !IsCancelledSet(n) {
			undeclared = append(undeclared, n)
		}
	}
	sort.Strings(undeclared)
	return undeclared, nil
}

// ResolvedBucket is mesh-config's state of resolved values, keyed `<node>.<module>.<key>`.
const ResolvedBucket = "mesh-config_resolved"

// NodeComposeReads is what a node's agent reads to compose its own modules (novox/hq ADR 0547 §2, ADR 0567 §3, §9,
// issue 1126): every key of the seat mesh-placement's placement and candidates (both the assigner's states), of the
// indexer's modules and manifests and of the builder's builds, and of mesh-config's resolved values only its own node's,
// `<node>.>`. Every candidate key, not only `*.<node>`: a node answering its candidate composes with the whole
// proposal laid over (the seed places every node at once, and a binding crosses nodes), and nothing in a placement is
// secret. Reads only; the node is checked as every subject token is by the caller's principal.
func NodeComposeReads(node string) []string {
	var out []string
	for _, bucket := range []string{delivery.PlacementBucket, delivery.PlacementCandidatesBucket, delivery.ModulesBucket,
		delivery.ManifestsBucket, delivery.BuilderBuildsBucket} {
		out = append(out, readGrants(bucket, "")...)
	}
	return append(out, readGrants(ResolvedBucket, node+".>")...)
}

// NodeComposeAdvisories are the stream-created advisories of the seat mesh-placement's placement and candidates
// (ADR 0567 §9): what a node agent hears to bind each once it is made. Those two and no wider.
func NodeComposeAdvisories() []string {
	return []string{StreamCreatedAdvisory(delivery.PlacementBucket), StreamCreatedAdvisory(delivery.PlacementCandidatesBucket)}
}

// ServedReadGrants is the read of each key of mesh-config's resolved values named, `<node>.<module>.<key>`
// (Principal.ServedReads, issue 1126): a get and a watch of that key alone. Each part is checked as every subject token
// is, so a key cannot widen the grant.
func ServedReadGrants(keys []string) ([]string, error) {
	var out []string
	seen := map[string]bool{}
	for _, k := range keys {
		parts := strings.Split(k, ".")
		if len(parts) != 3 {
			return nil, fmt.Errorf("%q is not <node>.<module>.<key>", k)
		}
		for _, part := range parts {
			if !safeSubject.MatchString(part) {
				return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", part)
			}
		}
		for _, g := range readGrants(ResolvedBucket, k) {
			if !seen[g] {
				seen[g] = true
				out = append(out, g)
			}
		}
	}
	return out, nil
}