mesh-delivery/modules/mesh-grants/internal/buscompose/busstate.go

package buscompose

import (
	"slices"
	"sort"
	"strings"

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

// **The bus module's process has a bus account of its own** (novox/hq ADR 0595 §4, §5; design 25 § 12.2, § 12.3;
// DLV-G2): composed for the process of the module holding the seat mesh-broker (a seat, never a module's name), it
// creates, updates, reads, snapshots and deletes every module's bus state under that module's names only, each stream
// by its name (a key-value bucket's and an object store's by the stream the bus keeps it in); it hears what its module
// consumes (the indexer's manifest-accepted) on a durable consumer of its own, says its module's own stream's events,
// and raises its conditions through the operator channel. As every process account it serves nothing (ADR 0436 §3): the
// seat's four reading verbs are served by its module's tools, and the four acting verbs by nobody on the bus, a
// person's ask reaching the process through its own state. No account holds the acting verbs' subjects, the tool
// runner's included: each is denied to every principal whose publish would reach it.
//
// The names are one rule (BusStateStreams), read by the bus module where it creates them: until mesh-sdk carries it
// as one function (asked of the mesh-sdk member of the milestone), this is the grants part's side of it.

// The verbs of the seat mesh-broker (design 25 § 12.3), until its definition is generated in mesh-contracts: the four
// reading verbs any caller its role grants may ask, and the four acting verbs the bus module's process alone serves.
var (
	BrokerReadingVerbs = []string{"list", "show", "check", "orphans"}
	BrokerActingVerbs  = []string{"create", "apply", "recreate", "drop"}
)

// BrokerVerbs are the eight.
func BrokerVerbs() []string {
	return append(append([]string(nil), BrokerReadingVerbs...), BrokerActingVerbs...)
}

// brokerSeat is the seat mesh-broker as its verbs' subjects name it: held once, for the mesh.
var brokerSeat = Seat{Name: catalogue.BrokerSeat}

// BrokerVerbSubjects are where the seat's verbs are asked: each verb's tool subject, and its caller-named form (issue
// 365).
func BrokerVerbSubjects(verbs []string) []string {
	var out []string
	for _, v := range verbs {
		out = append(out, seatToolSubject(brokerSeat, v, ""), seatSubject(brokerSeat, CallKind, v)+".>")
	}
	sort.Strings(out)
	return out
}

// BrokerActingOnly are the subjects of the seat's acting verbs, which no principal publishes on, however wide its
// grant: the bus module's process serves them, and a person's ask reaches it on its own tool subject instead
// (design 25 § 12.3).
func BrokerActingOnly() []string { return BrokerVerbSubjects(BrokerActingVerbs) }

// BusStateStreams are the streams one bus-state entry of module is kept in on the bus, by the mesh's names
// `<module>_<local>` (design 25 § 12.1): a stream's own; a key-value bucket's KV_ stream; an object store's OBJ_
// stream; a ledger's history stream, its snapshot stream, and its state and index buckets' KV_ streams. A name that is
// not one plain subject token names nothing.
func BusStateStreams(module string, e catalogue.BusStateEntry) []string {
	base := module + "_" + e.Name
	if !safeSubject.MatchString(base) || strings.Contains(e.Name, "_") || strings.Contains(module, "_") {
		return nil
	}
	var out []string
	switch e.Storage {
	case catalogue.StorageKeyValue:
		out = []string{"KV_" + base}
	case catalogue.StorageObjectStore:
		out = []string{"OBJ_" + base}
	case catalogue.StorageLedger:
		out = []string{base, base + "_snapshot", "KV_" + base + "_state"}
		for _, t := range e.Types {
			for _, ix := range t.Indexes {
				if safeSubject.MatchString(ix.Name) {
					out = append(out, "KV_"+base+"_index-"+ix.Name)
				}
			}
		}
	default:
		out = []string{base}
	}
	sort.Strings(out)
	return slices.Compact(out)
}

// keeperConsumer is the durable consumer the bus module's process hears its module's consumes on, on EVENTS:
// `<node>_<module>_<process>`, as a process's own consumer is named (QueueConsumer), filtered on the consumed events.
func keeperConsumer(p Principal) (Consumer, bool) {
	var filters []string
	for _, c := range p.Consumes {
		if s, err := consumedSubject(c); err == nil && !strings.HasPrefix(s, moduleStreamPrefix) {
			filters = append(filters, s)
		}
	}
	// And the events of the seats its module uses that are said to it by name, the operator channel's `decided` for its own
	// asks among them (design 25 § 12.3: the process wakes on the warrant, heard on its durable consumer, never on a timer).
	for _, s := range SeatTrafficOf(p.Module, nil, p.Uses, nil).Subscribe {
		if strings.Contains(s, ".event.") {
			filters = append(filters, s)
		}
	}
	if len(filters) == 0 {
		return Consumer{}, false
	}
	sort.Strings(filters)
	return Consumer{Name: p.Node + "_" + p.Module + "_" + p.Process, Stream: "EVENTS", Filters: slices.Compact(filters),
		AckWaitSeconds: 30, MaxDeliver: 5, FromNow: true,
		Why: "what the bus module's process " + p.Process + " hears of its module's consumes (novox/hq ADR 0595 §4): the " +
			"indexer's accepted verdicts, from now, its first view at start holding what came before"}, true
}

// keeperConsumerGrants are the subjects the process reads and acknowledges its consumer through.
func keeperConsumerGrants(c Consumer) []string {
	return []string{"$JS.API.CONSUMER.INFO." + c.Stream + "." + c.Name, "$JS.API.CONSUMER.MSG.NEXT." + c.Stream + "." + c.Name,
		"$JS.ACK." + c.Stream + "." + c.Name + ".>"}
}

// ownStreamReadGrants are what reading a module's own streams back needs (the bus module's tool history): each stream's
// information and an ordered consumer of its own, as a log's reader has (readLogGrants).
func ownStreamReadGrants(module string, streams []ModuleStream) []string {
	var out []string
	for _, s := range streams {
		if s.Module != module || !safeSubject.MatchString(s.Name) {
			continue
		}
		out = append(out, "$JS.API.STREAM.INFO."+s.Name, "$JS.API.CONSUMER.CREATE."+s.Name+".>",
			"$JS.API.CONSUMER.MSG.NEXT."+s.Name+".*", "$JS.API.CONSUMER.INFO."+s.Name+".*", "$JS.API.CONSUMER.DELETE."+s.Name+".*")
	}
	return out
}

// BusStateNames are every stream the bus module's process may define: those of every entry a manifest asks for now
// (asked), and those the bus module said it created and has not dropped (bound, by module: names of streams), so an
// orphan whose manifest left the trunk can still be dropped. Sorted, once each.
func BusStateNames(asked map[string][]catalogue.BusStateEntry, bound map[string][]string) []string {
	var out []string
	for module, entries := range asked {
		for _, e := range entries {
			out = append(out, BusStateStreams(module, e)...)
		}
	}
	for _, names := range bound {
		for _, n := range names {
			if safeSubject.MatchString(n) {
				out = append(out, n)
			}
		}
	}
	sort.Strings(out)
	return slices.Compact(out)
}

// busStateKeeperGrants are what the bus module's process publishes to keep the bus state named: each stream's create,
// update, information, deletion and snapshot with the acknowledgement of the snapshot's chunks (ADR 0626 §1; the server
// waits on `$JS.SNAPSHOT.ACK.<stream>.…` past its window), and the stream list it finds orphans in. Never a consumer of
// one, nor any stream's own subjects: a snapshot carries an entry's data to the keeper's inbox alone, and the keeper
// hears no message of it (ADR 0626 §1, restricted streams included).
func busStateKeeperGrants(names []string) []string {
	out := []string{"$JS.API.STREAM.NAMES", "$JS.API.STREAM.LIST"}
	for _, n := range names {
		for _, op := range []string{"CREATE", "UPDATE", "INFO", "DELETE", "SNAPSHOT"} {
			out = append(out, "$JS.API.STREAM."+op+"."+n)
		}
		out = append(out, "$JS.SNAPSHOT.ACK."+n+".>")
	}
	return out
}

// brokerSeatEvents are the subjects of the seat mesh-broker's events, from the held seat's definition: what the bus
// module's keeper says them on, like every seat's holder (the lead's ruling, registry milestone 13:0x).
func brokerSeatEvents(holds []Seat) []string {
	var out []string
	for _, s := range holds {
		if s.Name != catalogue.BrokerSeat {
			continue
		}
		for _, e := range s.Emits {
			if safeSubject.MatchString(e) {
				out = append(out, seatSubject(s, "event", e))
			}
		}
	}
	return out
}

// keepsBusState is the bus module's process defining one of the bus-state streams it keeps: the writers table's
// exemption from stream definitions being the grants part's (ADR 0595 §4).
func keepsBusState(p Principal, pattern string) bool {
	if p.Kind != KindProcess || len(p.KeepsBusState) == 0 {
		return false
	}
	for _, op := range []string{"CREATE", "UPDATE", "DELETE"} {
		if name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM."+op+"."); ok {
			return slices.Contains(p.KeepsBusState, name)
		}
	}
	return false
}

// withoutActing are a holder's served verbs of mesh-broker without the acting four.
func withoutActing(served []string) []string {
	var out []string
	for _, v := range served {
		if !slices.Contains(BrokerActingVerbs, v) {
			out = append(out, v)
		}
	}
	return out
}