mesh-delivery/modules/mesh-grants/internal/grants/assertions.go

package grants

import (
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"

	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
)

// Item is one bucket, log or stream a manifest declares, as composed, beside what the bus holds of it.
type Item struct {
	Kind   string `json:"kind"`
	Stream string `json:"stream"`
	// Composed is what an assertion would ask for; Held what the bus holds, absent when it holds no such stream.
	Composed Limits  `json:"composed"`
	Held     *Limits `json:"held,omitempty"`
	// Differences says how they differ, in words; empty when they agree in what can be read from here.
	Differences []string `json:"differences,omitempty"`
	// Created says what became of creating it when the bus lacked it: "created", or why not (issue 1790). Empty when
	// the bus holds it or no creation was asked here.
	Created string `json:"created,omitempty"`
}

// Limits are the facts of a stream an assertion sets and the bus module's nats_streams reads back. Its size cap is not
// read back by that tool, so it is composed and not compared.
type Limits struct {
	Subjects      []string `json:"subjects"`
	Retention     string   `json:"retention"`
	MaxAgeHours   int      `json:"max_age_hours,omitempty"`
	MaxPerSubject int      `json:"max_per_subject,omitempty"`
	MaxBytes      int64    `json:"max_bytes,omitempty"`
}

// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted says this pass created
// one the bus lacked (the operator, 2026-10-26 08:54); one the bus holds is never changed.
type Assertion struct {
	Module   string `json:"module"`
	Manifest string `json:"manifest"`
	Items    []Item `json:"items"`
	Asserted bool   `json:"asserted"`
	Says     string `json:"says"`
}

// ShadowAsserts is what an assertion says where nothing may be created: in the tools, which ask the part's process.
const ShadowAsserts = "in shadow: composed and compared, nothing asserted (ADR 0547 clause 5 asserts from take-over)"

// CreatesMissing is what an assertion of the part's process says (the operator, 2026-10-26 08:54; issue 1790).
const CreatesMissing = "a bucket, log or stream the manifest declares and the bus lacks is created with the manifest's limits, once per " +
	"manifest; one the bus holds is never changed or deleted, a difference is only said (ADR 0547 clause 5 asserts the rest from take-over)"

// Creator creates one stream the bus does not hold, with the configuration given, and never updates or deletes one: the
// part's process alone, whose account is granted the create of each declared stream by name (the operator, 2026-10-26
// 08:54). A stream the bus holds already answers ErrHeld.
type Creator interface {
	CreateDeclared(cfg jetstream.StreamConfig) error
}

// ErrHeld is a create refused because the bus holds the stream already (made meanwhile): left as it is.
var ErrHeld = errors.New("the bus holds it already, so it is left as it is")

// Created words of an item.
const (
	ItemCreated    = "created"
	ItemNotCreated = "not created: "
)

// Configs are the streams a manifest declares, as the controller asserts them at its start (mesh-sdk buscompose), by
// stream name. Pure.
func Configs(m catalogue.Manifest) map[string]jetstream.StreamConfig {
	out := map[string]jetstream.StreamConfig{}
	for _, b := range bucketsOf(m) {
		c := buscompose.BucketStreamConfig(b)
		out[c.Name] = c
	}
	for _, l := range logsOf(m) {
		c := l.Config()
		out[c.Name] = c
	}
	for _, s := range streamsOf(m) {
		c := buscompose.ModuleStreamConfig(s)
		out[c.Name] = c
	}
	return out
}

// creations remembers each stream created, by the manifest digest it was created at: created once per manifest, so a
// stream deleted since is not made again until a new manifest or a hand assert (grants_assert).
type creations struct {
	mu   sync.Mutex
	made map[string]string
}

func (c *creations) madeAt(stream string) string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.made[stream]
}

func (c *creations) keep(stream, digest string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.made == nil {
		c.made = map[string]string{}
	}
	c.made[stream] = digest
}

// createMissing creates each item of a the bus lacks, with the manifest's limits, and says what became of it. It never
// touches an item the bus holds, and does nothing when what the bus holds was not read. force asks again a stream
// created at this manifest and gone since (a hand assert). It answers the streams created.
func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, force bool) []string {
	cfgs := Configs(m)
	var made []string
	a.Says = CreatesMissing
	for i := range a.Items {
		it := &a.Items[i]
		if it.Held != nil || !slices.Contains(it.Differences, "the bus holds no stream "+it.Stream) {
			continue
		}
		if !force && done.madeAt(it.Stream) == a.Manifest {
			it.Created = ItemNotCreated + "created at this manifest and gone since; grants_assert creates it again"
			continue
		}
		cfg, ok := cfgs[it.Stream]
		if !ok {
			it.Created = ItemNotCreated + "no configuration composed for it"
			continue
		}
		switch err := c.CreateDeclared(cfg); {
		case err == nil:
			it.Created = ItemCreated
			done.keep(it.Stream, a.Manifest)
			made = append(made, it.Kind+" "+it.Stream)
		case errors.Is(err, ErrHeld):
			it.Created = ItemNotCreated + ErrHeld.Error()
		default:
			it.Created = ItemNotCreated + err.Error() + " (tried again at the next event)"
		}
	}
	if len(made) > 0 {
		a.Asserted = true
	}
	return made
}

// notCreated is each item of a creation refused, with why, as `<stream>: <why>`.
func notCreated(a Assertion) []string {
	var out []string
	for _, it := range a.Items {
		if why, ok := strings.CutPrefix(it.Created, ItemNotCreated); ok && it.Created != "" {
			out = append(out, it.Stream+": "+why)
		}
	}
	return out
}

// Composed is a module's buckets, logs and streams as an assertion would ask for them. Pure.
func Composed(m catalogue.Manifest) []Item {
	var out []Item
	for _, b := range bucketsOf(m) {
		per := b.History
		if per < 1 {
			per = 1
		}
		out = append(out, Item{Kind: "bucket", Stream: "KV_" + b.Bucket(), Composed: Limits{
			Subjects: []string{"$KV." + b.Bucket() + ".>"}, Retention: string(buscompose.RetentionLimits),
			MaxAgeHours: b.TTLSeconds / 3600, MaxPerSubject: per}})
	}
	for _, l := range logsOf(m) {
		c := l.Config()
		out = append(out, Item{Kind: "log", Stream: l.Stream(), Composed: Limits{Subjects: set(c.Subjects),
			Retention: string(buscompose.RetentionLimits), MaxAgeHours: int(c.MaxAge.Hours()), MaxBytes: c.MaxBytes}})
	}
	for _, s := range streamsOf(m) {
		st := s.Stream()
		out = append(out, Item{Kind: "stream", Stream: st.Name, Composed: Limits{Subjects: set(st.Subjects),
			Retention: string(st.Retention), MaxAgeHours: st.MaxAge / 3600, MaxBytes: st.MaxBytes}})
	}
	return out
}

// HeldStreams reads every stream the bus holds through its module's nats_streams, by name.
func HeldStreams(c Caller) (map[string]Limits, error) {
	raw, err := c.Call(CallStreams, map[string]any{})
	if err != nil {
		return nil, fmt.Errorf("%s: %w", CallStreams, err)
	}
	var got struct {
		Detail []struct {
			Name          string   `json:"name"`
			Subjects      []string `json:"subjects"`
			Retention     string   `json:"retention"`
			MaxAgeHours   int      `json:"max_age_hours"`
			MaxPerSubject int      `json:"max_per_subject"`
		} `json:"detail"`
	}
	if err := json.Unmarshal(unwrap(raw), &got); err != nil {
		return nil, fmt.Errorf("%s answered no streams: %w", CallStreams, err)
	}
	out := map[string]Limits{}
	for _, s := range got.Detail {
		out[s.Name] = Limits{Subjects: set(s.Subjects), Retention: s.Retention, MaxAgeHours: s.MaxAgeHours, MaxPerSubject: s.MaxPerSubject}
	}
	return out, nil
}

// Assert composes a module's items and compares each with what the bus holds; held nil is not read. It asserts nothing.
func Assert(module, digest string, m catalogue.Manifest, held map[string]Limits, heldErr error) Assertion {
	a := Assertion{Module: module, Manifest: digest, Items: Composed(m), Says: ShadowAsserts}
	for i := range a.Items {
		it := &a.Items[i]
		if heldErr != nil {
			it.Differences = []string{"what the bus holds was not read: " + heldErr.Error()}
			continue
		}
		h, ok := held[it.Stream]
		if !ok {
			it.Differences = []string{"the bus holds no stream " + it.Stream}
			continue
		}
		it.Held = &h
		if !slices.Equal(h.Subjects, it.Composed.Subjects) {
			it.Differences = append(it.Differences, fmt.Sprintf("subjects %v held, %v composed", h.Subjects, it.Composed.Subjects))
		}
		if h.Retention != it.Composed.Retention {
			it.Differences = append(it.Differences, fmt.Sprintf("retention %s held, %s composed", h.Retention, it.Composed.Retention))
		}
		if h.MaxAgeHours != it.Composed.MaxAgeHours {
			it.Differences = append(it.Differences, fmt.Sprintf("age %d h held, %d h composed", h.MaxAgeHours, it.Composed.MaxAgeHours))
		}
		if it.Kind == "bucket" && max(h.MaxPerSubject, 1) != it.Composed.MaxPerSubject {
			it.Differences = append(it.Differences, fmt.Sprintf("history %d held, %d composed", h.MaxPerSubject, it.Composed.MaxPerSubject))
		}
	}
	sort.SliceStable(a.Items, func(i, j int) bool { return a.Items[i].Stream < a.Items[j].Stream })
	return a
}