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

package buscompose

import (
	"fmt"
	"slices"
	"sort"
	"strings"
)

// The bus account of one process of a module that declares bus (novox/hq ADR 0436 §1, §2, §3, §5).
//
// **Never the module's own account's grants, never the tool runner's.** A process's authority is read
// straight from its bus declaration — its emits, its queue's consumer, its state, whether it reads the
// records of its module's own asks, and the tools and seat verbs it calls — and nothing else: it holds no
// seat, serves no tool and hears no module stream; it answers a seat's verb only where its resource declares `serves` beside its bus (design 47 G A-109). Modelled on
// provisionerGrants, the other kind of process account the mesh mints (ADR 0349 clauses 6 and 11).

// answeredSubjects are the subscribes of the seat verbs an account answers, each seat:<seat>.<verb>: the call and tool subjects
// of each (design 47 G A-107, A-109, A-115).
func answeredSubjects(verbs []string) ([]string, error) {
	var sub []string
	for _, v := range verbs {
		seat, verb, ok := strings.Cut(strings.TrimPrefix(v, "seat:"), ".")
		if !strings.HasPrefix(v, "seat:") || !ok || seat == "" || verb == "" {
			return nil, fmt.Errorf("%q does not name a seat's verb: one serves seat:<seat>.<verb>", v)
		}
		sub = append(sub, "mesh.seat."+seat+".call."+verb+".*", "mesh.seat."+seat+".tool."+verb)
	}
	return sub, nil
}

// processGrants is one process's whole authority: to publish, and to subscribe.
func processGrants(p Principal) (pub, sub []string, err error) {
	// 1. Its own events, under its module's own namespace — the same subjects the module's own principal
	// would publish them on, had it declared them there instead (novox/hq ADR 0436 §1).
	own := "mesh.mod." + p.Module
	for _, e := range p.Emits {
		pub = append(pub, own+".event."+e)
	}

	// 1b. The events of its module's events streams asked for in `requires` (design 42 zfs-owns-zfs, lead answer A-38; ADR
	// 0595 §1, §2): publish on each stream's subjects, the subject boundOf gives the module's own account, and nothing of
	// its reading. A name that cannot be one token is granted nothing.
	for _, local := range p.BusStreams {
		if safeSubject.MatchString(p.Module) && safeSubject.MatchString(local) && !strings.Contains(local, "_") && !strings.Contains(p.Module, "_") {
			pub = append(pub, "mesh.bus."+p.Module+"."+local+".>")
		}
	}

	// 2. The one durable consumer of each work-queue stream its bus names (novox/hq ADR 0436 §1, §2, §6):
	// asked about, pulled from and acknowledged, by the same name ConsumersOf composes the consumer under
	// (QueueConsumer) — and no other stream's.
	for _, q := range p.Queues {
		consumer := p.Node + "_" + p.Module + "_" + q.Process
		pub = append(pub,
			"$JS.API.CONSUMER.INFO."+q.Stream+"."+consumer,
			"$JS.API.CONSUMER.MSG.NEXT."+q.Stream+"."+consumer,
			"$JS.ACK."+q.Stream+"."+consumer+".>")
	}

	// 3. Its module's own state, by the local names its bus names alone — never the whole of what the
	// module keeps (novox/hq ADR 0436 §1), read and written as stateGrants gives any module's own bucket.
	kept := stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State}
	pub = append(pub, stateGrants(kept)...)
	// And it hears each of them created, so one not on the bus at its start is bound when it appears (novox/hq issue
	// 1790; process.AwaitBucket).
	sub = append(sub, stateCreatedHeard(kept)...)

	// 4. The router's records of its module's own asks (novox/hq ADR 0436 §1, ADR 0259 §3): the same
	// direct-get subjects a module's own principal reads for the seats its Uses names that keep them —
	// read here from the one list, so the two cannot disagree.
	if p.Asks {
		pub = append(pub, SeatTrafficOf(p.Module, nil, p.Uses, nil).Records...)
	}

	// 5. The tools and seat verbs it calls (novox/hq ADR 0152, ADR 0436 §1): the same derivation a
	// module's own principal's invokes takes, so `callerNamed` (applied to every non-controller principal)
	// adds the caller-named `call` variants for it exactly as it would for a module.
	invoked, err := invokedSubjects(p.Invokes)
	if err != nil {
		return nil, nil, err
	}
	pub = append(pub, invoked...)

	// 5b. The seat verbs it answers (design 47 G A-107, A-109): its subscribe of each verb's call and tool subjects (and
	// the chunk subjects, chunks.go).
	answered, err := answeredSubjects(p.AnswersSeat)
	if err != nil {
		return nil, nil, err
	}
	sub = append(sub, answered...)

	// 5c. The ends of its calls of a marked verb, heard on the seat's gated-ended, its own alone; and those of the marked verbs
	// it answers, said to any caller (design 47 G §677, A-264).
	gp, gs := gatedEndedGrants(p.Username(), p.MarkedVerbs, p.Invokes, p.AnswersSeat)
	pub, sub = append(pub, gp...), append(sub, gs...)

	// 5d. What it hears of its module's consumes, as its bus names them (design 47 G A-340, as 1b gives the publish): each
	// event's subject, and its own durable consumer on EVENTS and on each kept seat event's stream, asked about, pulled from
	// and acknowledged. The bus module's keeper hears its module's consumes in 6 instead.
	if !p.KeepsBus {
		for _, c := range p.Consumes {
			subject, err := consumedSubject(c)
			if err != nil {
				return nil, nil, err
			}
			sub = append(sub, subject)
		}
		sub = append(sub, watchedEvents(p.Watches)...)
		for _, h := range p.Hears {
			sub = append(sub, h.Filters...)
		}
		pub = append(pub, heardGrants(p, p.Hears)...)
		if c, ok := processConsumer(p); ok {
			pub = append(pub, keeperConsumerGrants(c)...)
		}
	}

	// 6. The bus module's process (busstate.go, ADR 0595 §4, §5): every module's bus state it keeps, by name; its own
	// consumer of what its module consumes; its module's own stream's events; and its conditions through the seats its
	// module uses (the operator channel's accept.condition in its module's name).
	if p.KeepsBus {
		pub = append(pub, busStateKeeperGrants(p.KeepsBusState)...)
		// And its standing restore of the streams ADR 0436 §3 restricts, used on a `snapshot-restore` call alone (design 47 G §681, A-60).
		pub = append(pub, restoreGrants(p.KeepsRestricted)...)
		if c, ok := keeperConsumer(p); ok {
			pub = append(pub, keeperConsumerGrants(c)...)
		}
		pub = append(pub, ownStreamGrants(p.Module, p.Streams)...)
		pub = append(pub, p.SaysSeatEvents...)
		tp, ts := seatTrafficGrants(SeatTrafficOf(p.Module, nil, p.Uses, nil))
		pub, sub = append(pub, tp...), append(sub, ts...)
	}

	return pub, sub, nil
}

// processHears splits what a process's bus consumes (design 47 G A-340) as its module's consumes are split: the events of a
// seat kept in a stream of its own (SplitKept), the seats' events on the shared stream (the module's Watches, narrowed to
// what the process names), and the modules' events. Only what its module consumes is kept; the judge refuses the rest.
func processHears(kept []KeptEvent, d Declared, consumes []string) (fromModules []string, watches []Seat, hears []Heard) {
	rest, hears := SplitKept(kept, consumes)
	for _, c := range rest {
		if slices.Contains(d.Consumes, c) {
			fromModules = append(fromModules, c)
		}
		for _, w := range d.Watches {
			for _, e := range w.Emits {
				if c == w.Name+"."+e || w.Kinded && c == w.Name+"."+e+".*" {
					watches = append(watches, Seat{Name: w.Name, Scope: w.Scope, Emits: []string{e}, Kinded: w.Kinded,
						Proofs: w.Proofs, DeclaredBy: w.DeclaredBy, HeldByTheEngine: w.HeldByTheEngine})
				}
			}
		}
	}
	return fromModules, watches, hears
}

// processConsumer is the durable consumer on EVENTS a process hears its bus's consumes on, `<node>_<module>_<process>`,
// filtered to exactly those events, from now (design 47 G A-340). None when it hears nothing there.
func processConsumer(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)
		}
	}
	filters = append(filters, watchedEvents(p.Watches)...)
	if len(filters) == 0 {
		return Consumer{}, false
	}
	sort.Strings(filters)
	return Consumer{Name: consumerDurable(p), Stream: "EVENTS", Filters: slices.Compact(filters), AckWaitSeconds: 30, MaxDeliver: 5,
		FromNow: true, Why: "what " + p.Module + "'s process " + p.Process + " hears of its module's consumes (design 47 G A-340)"}, true
}