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 `: `. 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 }