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

package grants

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/membership"
)

// The part's own states, as its manifest declares them, and the keys of the two that hold one value.
const (
	StateUserList    = "user-list"
	StateMemberships = "memberships"
	StateAssertions  = "assertions"
	StateTakeover    = "takeover"
	// StateGrants is the part's grants by module and fingerprint, the key layout of the controller's (grants.go).
	StateGrants = delivery.PartGrantsState
	KeyList     = "list"
	KeySwitch   = "switch"
	// KeyPass is the last pass's result, and KeyAsked a tool's ask to compose again now, both in StateUserList.
	KeyPass  = "pass"
	KeyAsked = "asked"
)

// States is every state the part keeps.
var States = []string{StateAccounts, StateAssertions, StateGrants, StateLease, StateMemberships, StateTakeover, StateUserList, StateWithheld}

// The events the part says.
const (
	EventUserList   = "user-list-changed"
	EventMembership = "membership-changed"
	// EventGrantWritten is said when the part wrote a grant of its state grants (ADR 0547 §5, ADR 0561 §2).
	EventGrantWritten = "grant-written"
	// EventAsserted is said when the part created a bucket, log or stream a manifest declares (issue 1790).
	EventAsserted = "asserted"
)

// Emits is every event the part says.
var Emits = []string{EventAsserted, EventGrantWritten, EventHashesAccepted, EventMembership, EventUserList}

// Store is the part's own state, by local name: the only thing the part writes.
type Store interface {
	Get(state, key string) ([]byte, uint64, bool, error)
	Put(state, key string, value []byte) (uint64, error)
	// Keys and Delete are for the wanted accounts (accounts.go) and the grants (grants.go): every other state is only put.
	Keys(state string) ([]string, error)
	Delete(state, key string) error
}

// Emitter says one of the part's events.
type Emitter interface {
	Emit(event string, body any) error
}

// UserListState is the value of `user-list` in shadow: the list the grants part composed (users and grants, no hashes),
// what the bus loaded, and how they differ. Inputs is the digest of all it says but At: a value is written again only
// when it changes.
type UserListState struct {
	V      int    `json:"v"`
	Mode   string `json:"mode"`
	Inputs string `json:"inputs"`
	Digest string `json:"digest"`
	Users  int    `json:"users"`
	// LeftOut are the assignments the shadow list leaves out, each with why.
	LeftOut []Gap                     `json:"left_out"`
	Compare Comparison                `json:"compare"`
	Loaded  *delivery.UserListWritten `json:"loaded,omitempty"`
	// Credentials says per user whether the vault holds its hash and how its line would differ (ADR 0547 clause 9).
	Credentials Credentials `json:"credentials"`
	BusNode     string      `json:"bus_node,omitempty"`
	Gaps        []Gap       `json:"gaps"`
	At          time.Time   `json:"at"`
}

func (s UserListState) inputs() string {
	s.At, s.Inputs = time.Time{}, ""
	// When the bus node said it loaded a list is no input: the same list said again, at a later time and the key's later
	// revision, composes nothing new (issue 1126: written again every 3 s, 2026-10-28).
	if s.Loaded != nil {
		l := *s.Loaded
		l.Revision, l.At = 0, time.Time{}
		s.Loaded = &l
	}
	raw, _ := json.Marshal(s)
	sum := sha256.Sum256(raw)
	return "sha256:" + hex.EncodeToString(sum[:])
}

// Result is what one composition did.
type Result struct {
	UserList           UserListState `json:"user_list"`
	Revision           uint64        `json:"revision"`
	Written            bool          `json:"written"`
	Memberships        int           `json:"memberships"`
	MembershipsWritten int           `json:"memberships_written"`
	Published          []string      `json:"published,omitempty"`
	NotPublished       []string      `json:"not_published,omitempty"`
	// Retired is, by `<node>.<module>`, the seat verbs its last published membership served that mesh-sdk's seat
	// catalogue no longer has (Membership.Retired): the pass keeps it, since the state memberships holds the bare
	// membership alone (issue 774).
	Retired           map[string][]string `json:"retired,omitempty"`
	Assertions        int                 `json:"assertions"`
	AssertionsWritten int                 `json:"assertions_written"`
	// Asserted is each module whose declared streams this pass created, `<module> at <digest>: created …`, and
	// NotAsserted each creation refused, `<module> <stream>: <why>` (issue 1790).
	Asserted    []string `json:"asserted,omitempty"`
	NotAsserted []string `json:"not_asserted,omitempty"`
	// Refused is every manifest the indexer holds that this pass could not read, `manifest of <module>: <why>`: its
	// module counts as placed nowhere (issue 1945).
	Refused []string `json:"refused,omitempty"`
	// RootNotFree is each machine the composition needed a root-free word for and did not get one, `<machine>: not
	// free, <why>`: a verified sender is composed as from no root-free machine there (issue 1945).
	RootNotFree []string `json:"root_not_free,omitempty"`
	// Sealed is the list sealed to the bus node as the key `user-list.sealed` names it (ADR 0547 clause 7).
	Sealed *SealedState `json:"sealed,omitempty"`
	// Consumers is what the pass found and created of the durable consumers an account needs (ADR 0561 §6).
	Consumers *ConsumersState `json:"consumers,omitempty"`
	// Accounts is what the pass wrote of the wanted accounts the vault mints from (ADR 0561 §2).
	Accounts *AccountsWritten `json:"accounts,omitempty"`
	// Grants is what the pass wrote of its grants (grants.go).
	Grants *GrantsWritten `json:"grants,omitempty"`
	// Withheld is what the pass wrote of the hearings it withholds (withheld.go).
	Withheld *WithheldWritten `json:"withheld,omitempty"`
	// PrincipalsRefused is each principal whose composed grants the composer refused, `<user>: <why>`: it keeps its grants
	// of the list in force, or stays out where it holds none (refused.go); the condition mesh-grants.principal-refused says it.
	PrincipalsRefused   []string `json:"principals_refused,omitempty"`
	Errors              []string `json:"errors,omitempty"`
	PublishedOrAsserted string   `json:"published_or_asserted"`
}

// NothingTakesEffect is what every composition says it did on the bus beyond its own state and the memberships.
const NothingTakesEffect = "memberships (in its own state memberships, each said on its subject, unless one would leave a seat a placed module claims with no holder, or drop a capability its holder still asks for, which stays a shadow difference), and each bucket, log or stream a manifest declares that the bus lacks, created once with the manifest's limits (the operator, 2026-10-26 08:54); the user list, the grants and any change to a stream the bus holds stay in shadow until take-over"

// Part is the grants part: it reads, calls and writes through these, and nothing else.
type Part struct {
	Read  Reader
	Call  Caller
	Store Store
	Emit  Emitter
	// Publish publishes memberships: the process's connection; nil in the tools, which ask the process instead.
	Publish Publisher
	// Create creates a declared stream the bus lacks: the process's connection; nil in the tools, which ask the process.
	Create Creator
	// Asker asks the operator (take-over, hand-back): the process's client of the router; nil in the tools, which write
	// the request and let the process ask it.
	Asker Asker
	// Logf logs what no state or condition says (an error of acting on an answer): the process's; nil in the tools.
	Logf func(string, ...any)
	// Objects is the part's object bucket of sealed lists: the process's connection; nil in the tools.
	Objects Objects
	// Consumers creates an account's durable consumer from the lease write (consumers.go): the process's connection;
	// nil in the tools, which compare only.
	Consumers ConsumerCreator
	// Bus is where the nodes reach the bus, which each wanted account carries (accounts.go): the process's own credential's;
	// empty in the tools, which write no account.
	Bus BusAddress
	// Say says a condition to the operator: the process's; nil in the tools.
	Say func(asks.Condition) error
	// Forget takes away the record the asker kept of an ask it failed to publish; Nonce is a test's id suffix.
	Forget func(id string) error
	Nonce  func() string
	// Lease is the process's hand on the controller's lease (leasewrite.go); nil in the tools, which ask the process.
	Lease LeaseBus
	// SaySeat says the seat mesh-grants' events (lease-handed, lease-altered, removal-failed): the process's; nil in the
	// tools.
	SaySeat SeatSayer
	// deciding is set while a take-over is decided, or a lease act composes, so the composition it makes asks and
	// reconciles nothing.
	deciding atomic.Bool
	// removalHeard is set when the marker of removal-due was heard, and taken by the next pass (leasewrite.go).
	removalHeard atomic.Bool
	Now          func() time.Time

	cache Manifests
	mu    sync.Mutex
	last  Result
	made  creations
	// askSeen is the revision of the last ask to compose (KeyAsked) this part acted on.
	askSeen uint64
	// heldSaid is the reason of the held-list condition last said (empty: none raised); heldKnown is whether one was said
	// since the start, so the first pass after a start always says its state (a condition raised before a restart is cleared
	// by it). holdMu guards both.
	holdMu    sync.Mutex
	heldSaid  string
	heldKnown bool
	// differSaid is the account consumers' differences last said as a condition, once differKnown (consumers.go).
	differSaid  string
	differKnown bool
	// refusedSaid is the refused principals last said as a condition, once refusedKnown (refused.go).
	refusedSaid  string
	refusedKnown bool
	// missingSaid is the modules without a grant last said as a condition, once missingKnown (grants.go).
	missingSaid  string
	missingKnown bool
	// hashHold are the users whose hash changes held the pass's list, or nil (sealed.go).
	hashHold []string
}

// Ask is the value of KeyAsked: a tool's ask to compose now, and, from grants_assert, to create one module's missing
// streams again even where they were created at this manifest before.
type Ask struct {
	By     string    `json:"by"`
	At     time.Time `json:"at"`
	Assert string    `json:"assert,omitempty"`
	// Consumer is, from grants_consumer_match, one account consumer `<stream>.<name>` to bring to the composed
	// configuration by hand (ADR 0561 §6).
	Consumer string `json:"consumer,omitempty"`
}

// asked is a new ask, read once per ask; the zero Ask when none.
func (p *Part) asked() Ask {
	raw, rev, found, err := p.Store.Get(StateUserList, KeyAsked)
	if err != nil || !found || rev == p.askSeen {
		return Ask{}
	}
	p.askSeen = rev
	var a Ask
	_ = json.Unmarshal(raw, &a)
	return a
}

func (p *Part) now() time.Time {
	if p.Now == nil {
		return time.Now().UTC()
	}
	return p.Now().UTC()
}

// Compose composes the user list, the memberships and the assertions from what the bus holds now, compares them, and
// writes each of its own keys whose value changed, saying each change. It never stops at a refusal: a read refused is
// a gap, a write refused is said in Errors, and the next event composes again.
func (p *Part) Compose() Result {
	// No lock is held across a round trip on the bus: the reads, calls, publishes and writes below take none, and the
	// lock is taken only to keep the result, so a tool answers while a pass waits on the bus.
	res := Result{PublishedOrAsserted: NothingTakesEffect}
	defer func() {
		p.mu.Lock()
		p.last = res
		p.mu.Unlock()
	}()
	in := ReadInputs(p.Read, &p.cache)
	for _, g := range in.Refused {
		res.Refused = append(res.Refused, g.Input+": "+g.Why)
	}
	// recs say what runs (memberships, consumers); listRecs what the user list and the grants carry: what runs and what
	// the trunk names, so the list grants a build's subjects before the want moves to it (issue 774).
	recs, _ := Records(in)
	listRecs, left := ListRecords(in)
	// From the lease write's cut, the controller's write on the lease is taken from the list (ADR 0582 clause 3).
	listRecs.LeaseCut = p.leaseCut()
	holdings := HoldingsOf(in)
	res.RootNotFree = WithRootFree(&listRecs, p.Call)
	recs.RootFree = listRecs.RootFree
	st := UserListState{V: 1, Mode: p.SwitchState(), LeftOut: left, Loaded: in.Loaded, BusNode: in.BusNode,
		Gaps: append(append([]Gap(nil), in.Gaps...), MembershipGaps...), At: p.now()}
	if st.LeftOut == nil {
		st.LeftOut = []Gap{}
	}
	users, refused, err := ComposeEach(listRecs)
	if err != nil {
		res.Errors = append(res.Errors, "the shadow list could not be composed: "+err.Error())
	}
	// A refused principal keeps its grants of the list in force, said once per change (refused.go).
	kept, _ := p.keptOf(refused)
	for _, user := range sortedKeys(refused) {
		res.PrincipalsRefused = append(res.PrincipalsRefused, user+": "+refused[user])
	}
	p.sayPrincipalsRefused(refused, kept)
	users = WithHashes(withKept(users, kept), in.Hashes)
	st.Digest, st.Users = Digest(users), len(users)
	loaded, loadErr := LoadedUsers(p.Call)
	st.Compare = Compare(users, loaded, Placed(in)).Excusing(ExcusedBy(MultiHeld(listRecs)))
	if loadErr != nil {
		st.Compare.Unread = loadErr.Error()
	}
	st.Credentials = CredentialsOf(users, loaded, in.Hashes)
	st.Inputs = st.inputs()
	res.UserList = st
	if raw, rev, found, err := p.Store.Get(StateUserList, KeyList); err != nil {
		res.Errors = append(res.Errors, fmt.Sprintf("its own %s cannot be read: %v", StateUserList, err))
	} else {
		var was UserListState
		if found && json.Unmarshal(raw, &was) == nil && was.Inputs == st.Inputs {
			res.Revision = rev
		} else if data, err := json.Marshal(st); err == nil {
			if rev, err := p.Store.Put(StateUserList, KeyList, data); err != nil {
				res.Errors = append(res.Errors, fmt.Sprintf("its own %s cannot be written: %v", StateUserList, err))
			} else {
				res.Revision, res.Written = rev, true
				p.say(&res, EventUserList, map[string]any{"digest": st.Digest, "revision": rev, "users": st.Users,
					"differing": len(st.Compare.Differing()), "mode": Shadow})
			}
		}
	}
	// The accounts the vault mints from, said in the state only it reads (ADR 0561 §2).
	if p.Store != nil {
		p.writeAccounts(&res, in, listRecs)
	}
	// The hearings withheld of what runs, for the self-check's probe D-heard (withheld.go).
	if p.Store != nil {
		p.writeWithheld(&res, in, recs)
	}
	list, n, leftOut, dropped, err := ComposeList(listRecs, in.Hashes, refused, kept)
	// The grants the assigner reads once the lease is handed away, in shadow until then (ADR 0547 §4, §5); after take-over
	// one that adds a permission only once the bus loaded a list that carries it (issue 774).
	if p.Store != nil {
		inForce, unknown := p.loadedContents(in, loaded, loadErr)
		p.writeGrants(&res, in, listRecs, users, inForce, unknown)
	}
	// The list itself, sealed to the bus node and named in `user-list.sealed` (ADR 0547 clause 7).
	if err != nil {
		res.Errors = append(res.Errors, "the user list could not be composed whole: "+err.Error())
	} else {
		// The users the bus loaded: what a first list after take-over must keep where the part has not yet written one.
		var bus, lost []string
		if st.Compare.Unread == "" {
			bus = names(loaded)
			lost = holdings.LastHolders(holdings.ByUsers(loaded), holdings.ByUsers(users))
		}
		p.writeSealed(&res, list, n, leftOut, dropped, in, bus, lost, users)
	}
	ms := Memberships(recs)
	res.Memberships = len(ms)
	members := make([]membership.Membership, 0, len(ms))
	for _, m := range ms {
		members = append(members, m.Membership)
	}
	seated := holdings.ByMemberships(members...)
	// From the lease write the memberships are written to the state alone and no longer said on their subject, which
	// nothing reads any more (novox/hq ADR 0571 clause 10, issue 774).
	notice, _ := leaseHandedAway(p.Read)
	notice = !notice
	for _, key := range sortedKeys(ms) {
		// The last membership is the state's own (ADR 0571 clause 7), never the assignments stream's, which the bus
		// module's tools remove at the lease write (research 072 row 33): a last read there would freeze every membership
		// once it is gone.
		was, _, found, err := p.Store.Get(StateMemberships, key)
		if err != nil {
			res.NotPublished = append(res.NotPublished, key+": not published: the membership last published cannot be read from its own "+StateMemberships+": "+err.Error())
			continue
		}
		m, now := publish(p.Publish, was, found, ms[key], holdings, seated)
		// Published is written to the state, where every runtime reads it (process.FollowMembership); only one
		// published, so a membership held back never reaches a reader. The subject is a notice after it.
		if now {
			body := membership.EncodeGrantsMembership(m.Membership)
			if _, err := p.Store.Put(StateMemberships, key, body); err != nil {
				res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be written: %v", StateMemberships, key, err))
				m.Published, now = "not published (tried again at the next event): its own "+StateMemberships+" cannot be written", false
			} else {
				res.MembershipsWritten++
				if notice {
					if err := p.Publish.PublishMembership(m.Node, m.Module, body); err != nil {
						res.Errors = append(res.Errors, fmt.Sprintf("the membership of %s is written, its notice on its subject not said: %v", key, err))
					}
				}
			}
		}
		// One unchanged is written bare once where the state still holds the wrapper written before 2026-10-30.
		if m.Published == MembershipUnchanged && p.changed(&res, StateMemberships, key, membership.EncodeGrantsMembership(m.Membership)) {
			res.MembershipsWritten++
		}
		switch {
		case now:
			res.Published = append(res.Published, key)
			p.say(&res, EventMembership, map[string]any{"node": m.Node, "module": m.Module, "digest": m.Digest, "published": true})
		case m.Published != MembershipUnchanged:
			res.NotPublished = append(res.NotPublished, key+": "+m.Published)
		}
		if len(m.Retired) > 0 {
			if res.Retired == nil {
				res.Retired = map[string][]string{}
			}
			res.Retired[key] = m.Retired
		}
	}
	held, heldErr := HeldStreams(p.Call)
	var ask Ask
	if p.Create != nil || p.Consumers != nil {
		ask = p.asked()
	}
	forced := ask.Assert
	for _, module := range sortedKeys(in.Manifests) {
		if !assigned(in, module) {
			continue
		}
		a := Assert(module, in.Digests[module], in.Manifests[module], held, heldErr)
		// Created only where the bus was read and lacks it, never changed where it holds it (issue 1790).
		if p.Create != nil && heldErr == nil {
			if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], forced == module); len(made) > 0 {
				res.Asserted = append(res.Asserted, fmt.Sprintf("%s at %s: created %s", module, a.Manifest, strings.Join(made, ", ")))
				p.say(&res, EventAsserted, map[string]any{"module": module, "manifest": a.Manifest, "created": made})
			}
			for _, why := range notCreated(a) {
				res.NotAsserted = append(res.NotAsserted, module+" "+why)
			}
		}
		res.Assertions++
		data, _ := json.Marshal(a)
		if p.changed(&res, StateAssertions, module, data) {
			res.AssertionsWritten++
		}
	}
	p.consumers(&res, recs, ask.Consumer)
	if _, _, found, err := p.Store.Get(StateTakeover, KeySwitch); err == nil && !found {
		data, _ := json.Marshal(Switch{V: 1, State: Shadow, Acts: []Act{}})
		if _, err := p.Store.Put(StateTakeover, KeySwitch, data); err != nil {
			res.Errors = append(res.Errors, fmt.Sprintf("its own %s cannot be written: %v", StateTakeover, err))
		}
	}
	if !p.deciding.Load() {
		if err := p.RaiseRequested(res.UserList); err != nil {
			res.Errors = append(res.Errors, err.Error())
		}
		if err := p.Reconcile(); err != nil {
			res.Errors = append(res.Errors, err.Error())
		}
		p.leaseStep(&res)
	}
	return res
}

// composeQuiet composes as a step of an act: asking and reconciling nothing.
func (p *Part) composeQuiet() Result {
	p.deciding.Store(true)
	defer p.deciding.Store(false)
	return p.Compose()
}

// Last is the result of the last pass this part made, kept under the lock alone.
func (p *Part) Last() Result {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.last
}

func assigned(in Inputs, module string) bool {
	// The node-engine runs on every node though no node is assigned it (novox/hq ADR 0563 §2, §3): its manifest is
	// placed everywhere, so its declared state is asserted and created as a placed module's.
	if module == buscompose.EngineModule {
		return true
	}
	for _, ms := range in.Assigned {
		for _, m := range ms {
			if m == module {
				return true
			}
		}
	}
	return false
}

// changed writes key when its value differs from the one held, and says whether it wrote.
func (p *Part) changed(res *Result, state, key string, data []byte) bool {
	raw, _, found, err := p.Store.Get(state, key)
	if err != nil {
		res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be read: %v", state, key, err))
		return false
	}
	if found && bytes.Equal(raw, data) {
		return false
	}
	if _, err := p.Store.Put(state, key, data); err != nil {
		res.Errors = append(res.Errors, fmt.Sprintf("its own %s %s cannot be written: %v", state, key, err))
		return false
	}
	return true
}

func (p *Part) say(res *Result, event string, body any) {
	if p.Emit == nil {
		return
	}
	if err := p.Emit.Emit(event, body); err != nil {
		res.Errors = append(res.Errors, fmt.Sprintf("the event %s could not be said: %v", event, err))
	}
}

func sortedKeys[V any](m map[string]V) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

func names(us []User) []string {
	out := make([]string, 0, len(us))
	for _, u := range us {
		out = append(out, u.User)
	}
	return out
}