// Package grants is the grants part (novox/hq ADR 0547 clauses 5 to 8, issue 1945): it composes the bus's user list, each
// module's membership and each manifest's buckets, logs and streams from what the bus holds, with the bus composer
// (mesh-sdk buscompose), and compares them with what the bus loaded and holds. It publishes memberships (2026-10-25
// 22:31), creates a bucket, log or stream a manifest declares that the bus lacks (2026-10-26 08:54, issue 1790), writes
// the grants the assigner reads and the user list the bus node follows alone, and makes the account consumers, since the
// controller ended by its unassignment (ADR 0590, issue 1126). It changes no stream the bus holds.
package grants

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

	"mesh-grants/internal/buscompose"

	"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-contracts/identity-and-access/go/clients/meshvaultclient"
	"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshnodeenroller"
	"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshvault"
	"git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
	"git.novox.be/novox/mesh-contracts/placement/go/clients/meshwantsclient"
	"git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
	"git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
)

// The states of other parts the grants part reads, as its manifest's reads name them (`<module>.<state>`).
const (
	ReadModules   = "mesh-delivery-indexer.modules"
	ReadManifests = "mesh-delivery-indexer.manifests"
	// ReadVerdicts is the indexer's verdicts, one key per manifest digest: a manifest it has not judged yet has none, and is
	// held like a refused one (ADR 0614 §8).
	ReadVerdicts = "mesh-delivery-indexer.verdicts"
	ReadHave     = "node-engine.have"
	ReadResolved = "mesh-config.resolved"
	ReadWants    = "mesh-delivery-assigner.wants"
	ReadBuilds   = "mesh-delivery-builder.builds"
	// ReadAccounts is the vault's bcrypt hashes of the bus accounts it holds, by bus user (ADR 0547 clause 9).
	ReadAccounts = "mesh-vault.bus-accounts"
	// ReadPlacement is the seat mesh-placement's placement (ADR 0567 §1, §9): each node's modules and the seats
	// they hold.
	ReadPlacement = buscompose.PlacementRead
	// ReadEnrolments is the seat mesh-enrolment's open enrolments, read as a named reader (ADR 0587 §2, §4): a key at
	// live or redeeming composes the user enrol.<node>.
	ReadEnrolments = buscompose.EnrolmentsRead
	// ReadNodeLogins is the vault's node-logins, read as its named reader (ADR 0583 §1 insight): each node's login in use
	// and the one rotating, both composed as users of the node's permissions.
	ReadNodeLogins = buscompose.NodeLoginsRead
)

// Reads is every state the part reads, in the manifest's order. Not the controller's assignments: the seat's placement
// alone places modules (novox/hq ADR 0549 §5, ADR 0571 §10; issue 774).
var Reads = []string{ReadResolved, ReadHave, ReadBuilds, ReadAccounts, ReadEnrolments, ReadNodeLogins}

// ViewReads are the owners' states the part reads by their first views and views families, never their buckets
// (firstviews.go, ADR 0615): each read in Reads' place.
var ViewReads = []string{ReadModules, ReadManifests, ReadVerdicts, ReadWants, ReadPlacement}

// The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
const (
	CallUsers   = "nats.nats_users"
	CallUserCan = "nats.nats_user_can"
	CallStreams = "nats.nats_streams"
)

// Invokes is every tool the part calls.
var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct, CallSeatDefinitions}, firstViewTools()...)

// CallSeatDefinitions reads the seat list (mesh-sdk seatlist): the verb of the seat mesh-catalog.
const CallSeatDefinitions = "seat:mesh-catalog.seat-definitions"

// CallHandAct records a person's act (grants_first_view_again).
const CallHandAct = "seat:mesh-hand-acts.hand-act"

func firstViewTools() []string {
	var out []string
	for _, fv := range FirstViews {
		out = append(out, fv.Tool)
	}
	return out
}

// Reader reads another part's state, named `<module>.<state>`. A refusal (a read not granted yet, a bucket not there) is
// an error the part names as a gap and composes past: it never stops for one.
type Reader interface {
	Keys(state string) ([]string, error)
	Get(state, key string) ([]byte, bool, error)
}

// Caller calls another module's tool, `<module>.<tool>`, and answers its result.
type Caller interface {
	Call(key string, body any) (json.RawMessage, error)
}

// Gap is an input the shadow cannot read from the bus, named with why: never guessed.
type Gap struct {
	Input string `json:"input"`
	Why   string `json:"why"`
	// Failed says the gap is a read that failed this pass, not one that stands (ADR 0547 clause 7: a
	// list composed past a failed read is not sealed).
	Failed bool `json:"failed,omitempty"`
}

// Standing gaps: what the controller composes from its own store, which no bus value carries in this milestone.
var standingGaps = []Gap{
	{Input: "people", Why: "the people and the tools they may invoke are in the controller's store alone: no person user is composed"},
	{Input: "seat holdings", Why: "who holds each seat held once for the mesh is in the controller's store alone: a mesh seat claimed on two machines is composed as held on both"},
	{Input: "announced states", Why: "the controller does not announce its states-announced readable, so every read is composed from `reads`, as when announcements cannot be read"},
}

// resolved is a key of mesh-config's resolved values, `<node>.<module>.<key>`.
type resolved struct {
	Node   string          `json:"node"`
	Module string          `json:"module"`
	Key    string          `json:"key"`
	Value  json.RawMessage `json:"value,omitempty"`
}

// Inputs is what the shadow composes from, read from the bus.
type Inputs struct {
	// Nodes is every machine seen: in a node agent's have or the placement.
	Nodes []string
	// Assigned is what the seat's placement puts on each machine, sorted (issue 774: never the controller's assignments).
	Assigned map[string][]string
	// Placed is each node's placement as the seat mesh-placement holds it (ADR 0567 §1); empty before any is read.
	// Where it is, a seat held once for the mesh is held where placement says, not wherever a module claiming it runs.
	Placed map[string]meshmoduleassigner.Placement
	// Manifests is each module's manifest at the indexer's trunk, by module.
	Manifests map[string]catalogue.Manifest
	// Bound is, by module, the streams of each bus-state entry the bus module said it created and has not dropped, from
	// the part's own state memberships (bindings.go): set by the pass, never read from another part.
	Bound map[string][]string
	// Digests is each module's manifest digest, by module.
	Digests map[string]string
	// Fingerprints is the fingerprint of the new form the indexer's trunk value names for each module, by module: what a
	// grant is written for (grants.go).
	Fingerprints map[string]string
	// NoFingerprint is why the indexer's value of a module names no fingerprint of the new form, by module (its none2, or
	// that it names none): a module here gets no grant, and that is said (grants.go MissingGrants).
	NoFingerprint map[string]string
	// Running is the manifest a machine is wanted to run, by `<node>.<module>`: the one the builder built at the
	// fingerprint the assigner's want names. An assignment with none is composed from the trunk's manifest.
	Running map[string]catalogue.Manifest
	// Values is each resolved config value, by `<node>/<module>/<key>`.
	Values map[string]any
	// BusNode is the machine running the bus's module, and Loaded what its node agent says it loaded.
	BusNode string
	Loaded  *nodeengine.UserListWritten
	// Hashes are the vault's hashes of the bus accounts, by bus user (ReadAccounts); kept in memory, never written.
	Hashes map[string]HeldHash
	// Enrolling is every node whose enrolment key is at live or redeeming (ReadEnrolments, ADR 0587 §4, §5), sorted: one
	// user enrol.<node> each. A key at asked has no token yet; an ended or expired key is gone from the bucket.
	Enrolling []string
	// NodeLogins is each node's key of the vault's node-logins (ReadNodeLogins, ADR 0583 §1); a node absent logs in as
	// node.<n> alone.
	NodeLogins map[string]meshvault.NodeLoginsByNodeValue
	// Refused is every manifest the indexer holds that the part cannot read or mesh-sdk's reader refuses, assigned or
	// not, each with why: its module counts as placed nowhere, so it is said loudly, never passed over (issue 1945).
	Refused []Gap
	// Held is, by module, the manifest a held module's running build was judged on (its have's fingerprint, the builder's
	// build of it): what the seats it defines, the streams others hear from it and its reads are composed from while its
	// trunk manifest is refused or waits, so no other principal loses a grant to it (ADR 0601 clause 3).
	Held map[string]catalogue.Manifest
	Gaps []Gap
}

// HeldHash is one key of the vault's bus-accounts: the hash, and what it is of. The hash is never said by the part, and
// written only into the list sealed to the bus node.
type HeldHash struct {
	Hash       string    `json:"hash"`
	Generation int       `json:"generation,omitempty"`
	At         time.Time `json:"at"`
	Adopted    bool      `json:"adopted,omitempty"`
	// Fingerprint is the vault's fingerprint of the value the hash is of: with the generation, what says the password
	// changed, never the hash itself (a hash of the same password differs by its salt).
	Fingerprint string `json:"fingerprint,omitempty"`
	// Rotating is the vault's rotation of the account under way: a change of its hash it announces (ADR 0561 §4, §7).
	Rotating *struct {
		Login string `json:"login"`
	} `json:"rotating,omitempty"`
}

// credentialOf is what says which password a held hash is of, never the hash: adopted, or the vault's generation and the
// fingerprint of its value.
func credentialOf(h HeldHash) string {
	if h.Adopted {
		return "adopted"
	}
	return fmt.Sprintf("generation %d %s", h.Generation, h.Fingerprint)
}

// readHashes reads the vault's bus-accounts; a refusal is a gap, and every user then has no hash held.
func readHashes(r Reader, gaps *[]Gap) map[string]HeldHash {
	out := map[string]HeldHash{}
	for user, raw := range each(r, ReadAccounts, gaps) {
		var h HeldHash
		if json.Unmarshal(raw, &h) != nil || h.Hash == "" {
			*gaps = append(*gaps, Gap{Input: ReadAccounts + " " + user, Why: "holds no hash"})
			continue
		}
		out[user] = h
	}
	return out
}

// BusModule is the bus's module: it holds mesh-broker.
const BusModule = "nats"

// Manifests caches manifests by digest: a digest names one manifest for ever.
type Manifests struct {
	mu     sync.Mutex
	parsed map[string]catalogue.Manifest
	// gate says whether a manifest naming some seats may be composed yet (the seat list reader, ADR 0636 §1); nil gates none.
	gate SeatGate
}

// SeatGate is what the part asks of the seat list reader before it composes a module new to it: whether a module that
// claims or invokes the given seats may be acted on (seatlist.Reader.MayAct). Until the seat list has been read only a module
// whose seats are all bootstrapped ones may; the rest is held as a manifest waiting for a judge is, its running build's
// manifest composing what runs, so nothing that runs is stopped (held.go).
type SeatGate interface {
	MayAct(seats ...string) bool
}

// readsTheList says whether a module is what the part needs to hear the seat and read the list: the part's own module (its
// consumer on EVENTS, its grant of the seat's verb and events) and the holder of the seat the list is read through. The unread
// list never holds these, whatever other seats their manifests name: held, the seat's event could not reach the part and
// the list would never be read (ADR 0636 §1).
func readsTheList(module string, m catalogue.Manifest) bool {
	return module == Module || slices.ContainsFunc(m.Claims, func(c catalogue.Claim) bool { return c.Name == meshcatalog.Seat })
}

// SeatsOf are the seats a manifest claims or invokes, which the seat list must name before the part composes the module.
func SeatsOf(m catalogue.Manifest) []string {
	var seats []string
	for _, c := range m.Claims {
		seats = append(seats, c.Name)
	}
	for _, in := range m.Invokes {
		if seat, _, ok := catalogue.SeatInvoke(in); ok {
			seats = append(seats, seat)
		}
	}
	return seats
}

func (c *Manifests) get(r Reader, digest string) (catalogue.Manifest, error) {
	// The lock guards the map alone, never a read on the bus: a tool never waits on a pass's round trip.
	c.mu.Lock()
	m, ok := c.parsed[digest]
	c.mu.Unlock()
	if ok {
		return m, nil
	}
	raw, found, err := r.Get(ReadManifests, digest)
	if err != nil {
		return catalogue.Manifest{}, err
	}
	if !found {
		return catalogue.Manifest{}, fmt.Errorf("the indexer holds no manifest %s", digest)
	}
	v, err := meshcatalogclient.DecodeManifest(raw)
	if err != nil {
		return catalogue.Manifest{}, err
	}
	// Decoded, never judged (novox/hq ADR 0578 §3): the indexer's judge judged it, and its verdict is taken where the
	// module's value is read (Inputs); a field this part does not know is set aside.
	m, err = catalogue.Decode([]byte(v.Manifest))
	if err != nil {
		return catalogue.Manifest{}, fmt.Errorf("the manifest %s does not decode: %w", digest, err)
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.parsed == nil {
		c.parsed = map[string]catalogue.Manifest{}
	}
	c.parsed[digest] = m
	return m, nil
}

// each reads every current value of a state; a refusal is a gap and an empty map.
func each(r Reader, state string, gaps *[]Gap) map[string][]byte {
	keys, err := r.Keys(state)
	if err != nil {
		*gaps = append(*gaps, Gap{Input: state, Why: "refused or absent: " + err.Error(), Failed: true})
		return map[string][]byte{}
	}
	out := map[string][]byte{}
	for _, k := range keys {
		v, found, err := r.Get(state, k)
		if err != nil {
			*gaps = append(*gaps, Gap{Input: state + " " + k, Why: err.Error(), Failed: true})
			continue
		}
		if found {
			out[k] = v
		}
	}
	return out
}

// ReadInputs reads what the shadow composes from. It never fails: what it cannot read is a gap.
func ReadInputs(r Reader, cache *Manifests) Inputs {
	cache.mu.Lock()
	gate := cache.gate
	cache.mu.Unlock()
	in := Inputs{Assigned: map[string][]string{}, Manifests: map[string]catalogue.Manifest{}, Digests: map[string]string{}, Fingerprints: map[string]string{},
		NoFingerprint: map[string]string{}, Running: map[string]catalogue.Manifest{},
		Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)}
	nodes := map[string]bool{}
	// **The seat's placement alone places modules** (novox/hq ADR 0549 §5, ADR 0567 §9; issue 774): the controller's
	// assignments are read no more, and the seats held once for the mesh are held where it says. Not read is a failed
	// read: nothing is placed, so the list composed past it is not sealed.
	if keys, err := r.Keys(ReadPlacement); err != nil {
		in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement, Why: "refused or absent: nothing is placed (" + err.Error() + ")", Failed: true})
	} else {
		// Each node's keys in either form, assembled by the seat's one reader (meshmoduleassignerclient.PlacementsOf; research 072
		// row 37): its node key and a key per module, or its legacy key `placement.<node>`.
		values := map[string][]byte{}
		for _, k := range keys {
			raw, found, err := r.Get(ReadPlacement, k)
			if err != nil || !found {
				continue
			}
			if _, _, _, err := meshmoduleassignerclient.ParseKey(k); err != nil {
				in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement + " " + k, Why: "not a placement key, so it places nothing"})
				continue
			}
			values[k] = raw
		}
		placed, err := meshmoduleassignerclient.PlacementsOf(values)
		if err != nil {
			in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement, Why: "a key does not read, so its node is composed from what does: " + err.Error()})
		}
		for node, p := range placed {
			if in.Placed == nil {
				in.Placed = map[string]meshmoduleassigner.Placement{}
			}
			in.Placed[node] = p
		}
	}
	for node, p := range in.Placed {
		nodes[node] = true
		for _, m := range meshmoduleassignerclient.Assigned(p) {
			if !slices.Contains(in.Assigned[node], m) {
				in.Assigned[node] = append(in.Assigned[node], m)
			}
			if m == BusModule && in.BusNode == "" {
				in.BusNode = node
			}
		}
	}
	if len(in.Placed) > 0 {
		in.Gaps = slices.DeleteFunc(in.Gaps, func(g Gap) bool { return g.Input == "seat holdings" })
	}
	haveKeys, err := r.Keys(ReadHave)
	if err != nil {
		in.Gaps = append(in.Gaps, Gap{Input: ReadHave, Why: "refused or absent: " + err.Error(), Failed: true})
	}
	for _, k := range haveKeys {
		if node, _, isNode, err := nodeengineclient.ParseHaveKey(k); err == nil && isNode {
			nodes[node] = true
		}
	}
	in.Gaps = append(in.Gaps, Gap{Input: "machines", Why: "read from the node agents' haves and the placement: a machine with neither is not seen"})
	for n := range nodes {
		in.Nodes = append(in.Nodes, n)
	}
	sort.Strings(in.Nodes)
	if in.BusNode != "" {
		if raw, found, err := r.Get(ReadHave, in.BusNode+"."+BusModule); err != nil {
			in.Gaps = append(in.Gaps, Gap{Input: "the loaded list's have", Why: err.Error(), Failed: true})
		} else if found {
			if h, err := nodeengineclient.DecodeHave(raw); err == nil {
				in.Loaded = h.UserList
			}
		}
	}
	wanted := map[string]bool{}
	for node := range in.Assigned {
		sort.Strings(in.Assigned[node])
		for _, m := range in.Assigned[node] {
			wanted[m] = true
		}
	}
	modules := make([]string, 0, len(wanted))
	for m := range wanted {
		modules = append(modules, m)
	}
	sort.Strings(modules)
	// The seats every module declares, across the whole catalogue: read every module the indexer holds, not only the
	// assigned ones, since a seat is declared by one module and held by another.
	indexed, err := r.Keys(ReadModules)
	if err != nil {
		in.Gaps = append(in.Gaps, Gap{Input: ReadModules, Why: "refused or absent: " + err.Error(), Failed: true})
	}
	for _, module := range indexed {
		raw, found, err := r.Get(ReadModules, module)
		if err != nil {
			// One module's value that does not read holds that module alone, as a refused manifest does (ADR 0614 §8).
			g := Gap{Input: "manifest of " + module, Why: "its value in " + ReadModules + " does not read: " + err.Error()}
			in.Refused = append(in.Refused, g)
			in.Gaps = append(in.Gaps, g)
			continue
		}
		if !found {
			continue
		}
		v, err := meshcatalogclient.DecodeModule(raw)
		if err == nil && v.Manifest == "" {
			err = fmt.Errorf("the indexer's value names no manifest digest")
		}
		if err == nil && strings.HasPrefix(v.Fp2, meshcatalogclient.SourcePrefix2) {
			in.Fingerprints[module] = v.Fp2
		} else if err == nil {
			why := v.None2
			if why == "" {
				why = "the indexer's value names no fingerprint of the new form"
			}
			in.NoFingerprint[module] = why
		}
		var m catalogue.Manifest
		if err == nil {
			// The indexer's refusal, or its wait for a judge that knows a field, carried in its value (ADR 0578 §3): held and named as a refused manifest.
			if verdict := meshcatalogclient.ModuleVerdict(module, v); verdict != nil && meshcatalogclient.VerdictHeld(*verdict) {
				err = errors.New(meshcatalogclient.VerdictWhy(*verdict))
			} else if verdict == nil {
				// The indexer writes a changed manifest's value before it judges it, and a new digest has no verdict: until
				// the verdict is written the manifest is held (ADR 0614 §8).
				err = noVerdictYet(r, v.Manifest)
			}
		}
		if err == nil {
			m, err = cache.get(r, v.Manifest)
		}
		if err == nil && gate != nil && !readsTheList(module, m) && !gate.MayAct(SeatsOf(m)...) {
			err = errors.New("it waits for the seat list: the list has not been read yet and the manifest names seats the bootstrapped list does not (novox/hq ADR 0636 §1)")
		}
		if err != nil {
			// Every manifest, assigned or not: one refused declares no seat, so its holders compose as placed nowhere.
			g := Gap{Input: "manifest of " + module, Why: err.Error()}
			in.Refused = append(in.Refused, g)
			in.Gaps = append(in.Gaps, g)
			continue
		}
		in.Manifests[module] = m
		in.Digests[module] = v.Manifest
	}
	// A module the mesh provides itself has no repository the indexer reads: its manifest comes with mesh-sdk, and the
	// controller's Provide overwrites any row of it at every start, so the provided one always wins here too (issue 1945,
	// the private network on every machine). An indexer manifest of the same name is said as a gap, never used.
	for _, name := range overlay.ProvidedModules() {
		if !slices.Contains(modules, name) {
			continue
		}
		m, raw, err := overlay.ProvidedManifest(name)
		in.Gaps = slices.DeleteFunc(in.Gaps, func(g Gap) bool { return g.Input == "manifest of "+name })
		in.Refused = slices.DeleteFunc(in.Refused, func(g Gap) bool { return g.Input == "manifest of "+name })
		if _, held := in.Digests[name]; held {
			in.Gaps = append(in.Gaps, Gap{Input: "manifest of " + name, Why: "the indexer also holds one, which is not used: the mesh provides this module's manifest itself, and the controller overwrites any row of it"})
		}
		if err != nil {
			g := Gap{Input: "manifest of " + name, Why: err.Error()}
			in.Refused = append(in.Refused, g)
			in.Gaps = append(in.Gaps, g)
			delete(in.Manifests, name)
			delete(in.Digests, name)
		} else {
			in.Manifests[name] = m
			in.Digests[name] = fmt.Sprintf("provided:sha256:%x", sha256.Sum256(raw))
		}
	}
	for _, module := range modules {
		if _, ok := in.Manifests[module]; !ok && !gapNamed(in.Gaps, "manifest of "+module) {
			in.Gaps = append(in.Gaps, Gap{Input: "manifest of " + module,
				Why: "assigned, and the indexer holds no manifest of it at the trunk (a module registered from outside the indexed repositories): composed with no authority"})
		}
	}
	// What each machine is wanted to run: the want's fingerprint, the builder's build of it, the manifest it built.
	var fromTrunk []string
	for key, raw := range each(r, ReadWants, &in.Gaps) {
		node, module, err := meshwantsclient.ParseWantKey(key)
		if err != nil || !slices.Contains(in.Assigned[node], module) {
			continue
		}
		w, err := meshwantsclient.DecodeWant(raw)
		if err != nil {
			fromTrunk = append(fromTrunk, node+"."+module)
			continue
		}
		if w.Removed || w.Fp == "" {
			continue
		}
		bk, err := builds.BuilderBuildKey(module, w.Fp)
		if err != nil {
			continue
		}
		braw, found, err := r.Get(ReadBuilds, bk)
		if err != nil || !found {
			fromTrunk = append(fromTrunk, node+"."+module)
			continue
		}
		b, err := builds.DecodeBuilderBuild(braw)
		if err != nil || b.Manifest == "" {
			fromTrunk = append(fromTrunk, node+"."+module)
			continue
		}
		if m, err := cache.get(r, b.Manifest); err == nil {
			in.Running[node+"."+module] = m
		} else {
			fromTrunk = append(fromTrunk, node+"."+module)
		}
	}
	if len(fromTrunk) > 0 {
		sort.Strings(fromTrunk)
		in.Gaps = append(in.Gaps, Gap{Input: "wanted manifests", Why: "no build record names the manifest at the wanted fingerprint, so the trunk's is used: " + strings.Join(fromTrunk, ", ")})
	}
	in.Held = heldManifests(r, in, cache)
	// A held module placed on a node whose running manifest is not found declares no seat, stream or object bucket to
	// anyone: composing past it would drop the grants of those who use them, so the users of that module alone stay as the
	// list in force holds them (held.go, named unreadable; ADR 0614 §8, ADR 0601 clause 3). The gap says which module and why,
	// and holds nothing else: it is not a failed read.
	for _, g := range in.Refused {
		module, ok := strings.CutPrefix(g.Input, "manifest of ")
		if _, found := in.Held[module]; !ok || found {
			continue
		}
		for _, assigned := range in.Assigned {
			if slices.Contains(assigned, module) {
				in.Gaps = append(in.Gaps, Gap{Input: "running manifest of " + module,
					Why: "its manifest is unreadable (held, and no running build's manifest is found), so the seats and streams it declares cannot be composed: its users stay as the list in force holds them"})
				break
			}
		}
	}
	in.Hashes = readHashes(r, &in.Gaps)
	// The seat mesh-enrolment's keys are every token since the controller's end (issue 1126).
	in.Enrolling, _ = readEnrolling(r, &in.Gaps)
	in.NodeLogins = readNodeLogins(r, &in.Gaps)
	for _, raw := range each(r, ReadResolved, &in.Gaps) {
		var v resolved
		if json.Unmarshal(raw, &v) != nil || v.Node == "" || v.Module == "" || v.Key == "" || len(v.Value) == 0 {
			continue
		}
		var value any
		if json.Unmarshal(v.Value, &value) == nil {
			in.Values[v.Node+"/"+v.Module+"/"+v.Key] = value
		}
	}
	sortGaps(in.Gaps)
	return in
}

// readEnrolling is every node with an enrolment key at live or redeeming (ADR 0587 §4, §5), and whether the seat's state
// was read. Not on the bus yet is a standing gap, never a failed read: the list is still composed and sealed, with no
// enrolment user. A key that does not read is named and composes no user.
func readEnrolling(r Reader, gaps *[]Gap) ([]string, bool) {
	keys, err := r.Keys(ReadEnrolments)
	if err != nil {
		*gaps = append(*gaps, Gap{Input: ReadEnrolments, Why: "not read yet: no enrolment user is composed (" + err.Error() + ")"})
		return nil, false
	}
	var out []string
	for _, k := range keys {
		node, isNode := meshnodeenroller.ParseEnrolmentsByNodeNodeKey(k)
		if !isNode {
			continue
		}
		raw, found, err := r.Get(ReadEnrolments, k)
		if err != nil {
			*gaps = append(*gaps, Gap{Input: ReadEnrolments + " " + k, Why: err.Error(), Failed: true})
			continue
		}
		if !found {
			continue
		}
		e, err := meshnodeenroller.DecodeEnrolmentsByNodeNodeValue(raw)
		if err != nil || e.Node != node {
			*gaps = append(*gaps, Gap{Input: ReadEnrolments + " " + k, Why: "does not read as the enrolment of " + node +
				", so it composes no user"})
			continue
		}
		if e.Stage == buscompose.EnrolmentLive || e.Stage == buscompose.EnrolmentRedeeming {
			out = append(out, node)
		}
	}
	sort.Strings(out)
	return out, true
}

// readNodeLogins is every node's key of the vault's node-logins (ADR 0583 §1). Not on the bus yet is a standing gap,
// never a failed read: every node then logs in as node.<n> alone, as before any rotation. A key that does not read is
// named and leaves its node at node.<n>; a read that fails is a failed read, since a rotating node's new login would be
// left out.
func readNodeLogins(r Reader, gaps *[]Gap) map[string]meshvault.NodeLoginsByNodeValue {
	keys, err := r.Keys(ReadNodeLogins)
	if err != nil {
		*gaps = append(*gaps, Gap{Input: ReadNodeLogins, Why: "not read yet: every node logs in as node.<node> (" + err.Error() + ")"})
		return nil
	}
	out := map[string]meshvault.NodeLoginsByNodeValue{}
	for _, node := range keys {
		raw, found, err := r.Get(ReadNodeLogins, node)
		if err != nil {
			*gaps = append(*gaps, Gap{Input: ReadNodeLogins + " " + node, Why: err.Error(), Failed: true})
			continue
		}
		if !found {
			continue
		}
		l, err := meshvaultclient.DecodeNodeLogin(raw)
		if err != nil || l.Node != node {
			*gaps = append(*gaps, Gap{Input: ReadNodeLogins + " " + node, Why: "does not read as the logins of " + node +
				", so it logs in as node." + node})
			continue
		}
		out[node] = l
	}
	return out
}

// Rotating are the logins the vault's node-logins names in a rotation under way, as a set: a hash of one changes by the
// vault's rotation (ADR 0583 §2), never counted as a writer changing several hashes at once.
func (in Inputs) Rotating() map[string]bool {
	out := map[string]bool{}
	for _, l := range in.NodeLogins {
		if meshvaultclient.Rotates(l) {
			out[l.Login], out[l.Rotating] = true, true
		}
	}
	return out
}

func gapNamed(gaps []Gap, input string) bool {
	for _, g := range gaps {
		if g.Input == input {
			return true
		}
	}
	return false
}

func sortGaps(g []Gap) {
	sort.SliceStable(g, func(i, j int) bool { return strings.Compare(g[i].Input, g[j].Input) < 0 })
}

// ReadFailed names every read this pass could not make: what a list composed now may lack. A manifest the indexer holds
// (refused, or waiting for a judge that knows a field of it) is not one: it holds its module's users in the list as the
// list in force has them (held.go), and never holds the list, so one manifest never freezes the bus's users (issue 774).
func (in Inputs) ReadFailed() []string {
	var out []string
	for _, g := range in.Gaps {
		if g.Failed {
			out = append(out, g.Input+": "+g.Why)
		}
	}
	return out
}

// ReadFailures is ReadFailed and every manifest refused or waiting: what the part retires nothing on (the accounts it
// keeps, the hearings it withholds).
func (in Inputs) ReadFailures() []string {
	out := in.ReadFailed()
	for _, g := range in.Refused {
		out = append(out, g.Input+": "+g.Why)
	}
	return out
}

// heldManifests are the manifests the running builds of the held modules (in.Refused) were judged on, by module: the have
// of the module on a node names the fingerprint that runs, and the builder's build of it the manifest. A module that runs
// nowhere, or whose build or manifest cannot be read, has none, and that is said as a gap.
func heldManifests(r Reader, in Inputs, cache *Manifests) map[string]catalogue.Manifest {
	out := map[string]catalogue.Manifest{}
	for _, g := range in.Refused {
		module, ok := strings.CutPrefix(g.Input, "manifest of ")
		if !ok {
			continue
		}
		for _, node := range in.Nodes {
			raw, found, err := r.Get(ReadHave, node+"."+module)
			if err != nil || !found {
				continue
			}
			h, err := nodeengineclient.DecodeHave(raw)
			if err != nil || h.Fp == "" {
				continue
			}
			bk, err := builds.BuilderBuildKey(module, h.Fp)
			if err != nil {
				continue
			}
			braw, found, err := r.Get(ReadBuilds, bk)
			if err != nil || !found {
				continue
			}
			b, err := builds.DecodeBuilderBuild(braw)
			if err != nil || b.Manifest == "" {
				continue
			}
			if m, err := cache.get(r, b.Manifest); err == nil {
				out[module] = m
				break
			}
		}
	}
	return out
}

// declared are the manifests the seats, hearings and reads are composed from: the trunk's, and for a held module the
// manifest its running build was judged on (Held).
func (in Inputs) declared() map[string]catalogue.Manifest {
	if len(in.Held) == 0 {
		return in.Manifests
	}
	out := make(map[string]catalogue.Manifest, len(in.Manifests)+len(in.Held))
	for m, man := range in.Held {
		out[m] = man
	}
	for m, man := range in.Manifests {
		out[m] = man
	}
	return out
}

// noVerdictYet is nil when the indexer's verdicts state holds an accepted verdict on the manifest, and the reason it is held
// when it holds none yet, cannot be read, or refuses it.
func noVerdictYet(r Reader, digest string) error {
	key, err := meshcatalogclient.VerdictsKey(digest)
	if err != nil {
		return nil
	}
	raw, found, err := r.Get(ReadVerdicts, key)
	if err != nil {
		return fmt.Errorf("its verdict cannot be read: %w", err)
	}
	if !found {
		return errors.New("the indexer has written no verdict on this manifest yet")
	}
	v, err := meshcatalogclient.DecodeVerdict(raw)
	if err != nil {
		return fmt.Errorf("its verdict does not read: %w", err)
	}
	if meshcatalogclient.VerdictHeld(v) {
		return errors.New(meshcatalogclient.VerdictWhy(v))
	}
	return nil
}