mesh-controller/cmd/mesh-controller/assignment_accounts.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"slices"
	"strings"
	"sync"
	"time"

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

	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/chain"
	"github.com/novox/mesh-controller/internal/inventory"
)

// An assignment that stands issues its node's bus accounts and the module's grant (novox/hq ADR 0543, issue 1945; a
// step of research 072's grants: the memberships and credentials heard from placement, not from a send or a build).
//
// ADR 0539 clause 7 issues a module's credential and grant when its build is done. A module already built and assigned
// to one more node never has a build done again, so nothing issued that node's account: on 2026-10-25 at 17:19
// mesh-build-node and mesh-build-go were assigned to ace, composed for it, and the bus's user list never named
// ace.mesh-build-node nor ace.mesh-build-go; one runtime found no credential, the other was refused by the bus, both
// restarting every 5 s. The assign act composes the assigned node alone (composedAfterAct): never the node holding the
// bus, never the grant.
//
// So the write of an assignments key is the event. When a key `<node>.<module>` of the assignments bucket comes to stand
// (a new key, or one that was ended or pending), the lease holder issues the module's accounts on that node that are not
// minted (idempotent: a minted account is never issued again, a rotation is on purpose), composes the node holding the
// bus so its user list carries them, and writes the grants. A rewrite of an assignment that already stood (the
// provisioner's plan kept in its value) does nothing. A starting controller looks at every standing assignment once,
// from one reading of the catalogue: it issues what is missing, then composes the bus's node and writes the grants
// once. The assign act issues too; both mint only an account no row names, in one statement, so exactly one of them
// mints and seals it (mintBusUser, the review of issue 1945).

// assignmentAccountSteps are what an assignment that comes to stand does, each a step a test may stand in for.
type assignmentAccountSteps struct {
	// holds says this controller holds the lease: only its holder issues.
	holds func(ctx context.Context) bool
	// forgetStale forgets every module's accounts on a node where the module no longer resolves, and answers a line for
	// each (issue 1945: rows of an unassignment before this release, marked sealed by migration 0108).
	forgetStale func(ctx context.Context) ([]string, error)
	// catalogue reads the catalogue once, for every assignment a look issues for.
	catalogue func(ctx context.Context) (map[string]catalogue.Manifest, error)
	// issue issues the module's accounts on the node that are not minted, from the catalogue read; issued says it
	// minted one, line what it did.
	issue func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (issued bool, line string, err error)
	// busHolder is the node holding the bus, "" for none.
	busHolder func(ctx context.Context) (string, error)
	// compose writes the nodes' compositions; grants writes the grants issued.
	compose func(ctx context.Context, nodes []string) error
	grants  func(ctx context.Context) error
}

// settle issues what the standing assignment of module on node lacks. fresh says the assignment just came to stand: the
// bus's node is composed and the grants written even when every account was minted before (a module assigned again
// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for
// nothing.
func (s assignmentAccountSteps) settle(ctx context.Context, node, module string) (string, error) {
	if module == controllerModule {
		// The controller's own module is issued at the terminal alone: its `broker` secret is the credential it is
		// using (rollout.go), and it is assigned and unassigned there only (refusedMovingTheController).
		return "", nil
	}
	if !s.holds(ctx) {
		return "", errNotHolding
	}
	shelf, err := s.catalogue(ctx)
	if err != nil {
		return "", fmt.Errorf("the catalogue cannot be read: %w", err)
	}
	_, line, err := s.issue(ctx, shelf, node, module)
	if err != nil {
		return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err)
	}
	said, err := s.cover(ctx)
	prefix := module + " on " + node + ": "
	if line != "" {
		prefix += line + "; "
	}
	return prefix + said, err
}

// look is the start's look: the accounts every standing assignment lacks, issued one after another from one reading of
// the catalogue, then the bus's node composed and the grants written once, when it looked at any assignment, whatever
// it issued (an assignment that came to stand while no controller heard it may have its accounts and still lack its
// place in the bus's user list or its grant). It answers what it did.
func (s assignmentAccountSteps) look(ctx context.Context, standing []chain.Assignment) (string, error) {
	var forgot []string
	if len(standing) > 0 && s.forgetStale != nil {
		// Only from a reading of the assignments that holds some: an empty or failed one would forget every account.
		if !s.holds(ctx) {
			return "", errNotHolding
		}
		var err error
		if forgot, err = s.forgetStale(ctx); err != nil {
			return "", fmt.Errorf("the accounts of modules no longer on their nodes are not forgotten: %w", err)
		}
	}
	var looked []chain.Assignment
	for _, a := range standing {
		if a.Module != controllerModule {
			looked = append(looked, a)
		}
	}
	if len(looked) == 0 {
		return strings.Join(forgot, "\n"), nil
	}
	if !s.holds(ctx) {
		return "", errNotHolding
	}
	shelf, err := s.catalogue(ctx)
	if err != nil {
		return "", fmt.Errorf("the catalogue cannot be read: %w", err)
	}
	var lines []string
	for _, a := range looked {
		if ctx.Err() != nil {
			return "", ctx.Err()
		}
		if _, line, err := s.issue(ctx, shelf, a.Node, a.Module); err != nil {
			return "", fmt.Errorf("%s's bus accounts on %s are not issued: %w", a.Module, a.Node, err)
		} else if line != "" {
			lines = append(lines, a.Module+" on "+a.Node+": "+line)
		}
	}
	said, err := s.cover(ctx)
	lines = append(append(forgot, lines...), fmt.Sprintf("%d standing assignment(s) looked at; %s", len(looked), said))
	return strings.Join(lines, "\n"), err
}

// errNotHolding is a controller that does not hold the lease: it issues nothing, and tries again until it holds it.
var errNotHolding = errors.New("this controller does not hold the lease")

// cover composes the node holding the bus and writes the grants, and says so.
func (s assignmentAccountSteps) cover(ctx context.Context) (string, error) {
	holder, err := s.busHolder(ctx)
	if err != nil {
		return "", fmt.Errorf("the node holding the bus cannot be read: %w", err)
	}
	if holder != "" {
		if err := s.compose(ctx, []string{holder}); err != nil {
			return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
		}
	}
	if err := s.grants(ctx); err != nil {
		return "", err
	}
	if holder != "" {
		return "the bus's user list composed on " + holder + ", the grants written", nil
	}
	return "the grants written", nil
}

// assignmentStands says a value of the assignments bucket holds an assignment that stands: made (a module only pulled in
// by another names no time assigned), not ended and in no state (chain.Assignment.Stands).
func assignmentStands(a chain.Assignment) bool {
	return !a.Assigned.IsZero() && a.Stands() && a.Node != "" && a.Module != ""
}

// assignmentTurns keeps which keys of the assignments bucket stood when last heard, to tell an assignment that comes to
// stand from a rewrite of one that stood.
type assignmentTurns struct {
	mu    sync.Mutex
	stood map[string]bool
}

// heard records a value of key (nil for a delete or a purge) and answers whether it holds an assignment that just came
// to stand.
func (t *assignmentTurns) heard(key string, value []byte) (chain.Assignment, bool) {
	var a chain.Assignment
	stands := false
	if value != nil && json.Unmarshal(value, &a) == nil {
		stands = assignmentStands(a)
	}
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.stood == nil {
		t.stood = map[string]bool{}
	}
	was := t.stood[key]
	t.stood[key] = stands
	return a, stands && !was
}

// ownAccountOf is the username of the account a module speaks on the bus with on node, its own and no process's: the
// module's when it reads a `broker` own secret, else its provisioner client's when it runs one; false when it has none
// (a module whose accounts are its processes' alone, or one that does not speak on the bus).
func ownAccountOf(m catalogue.Manifest, node string) (string, bool) {
	if _, reads := m.OwnSecrets["broker"]; reads {
		return broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username(), true
	}
	if runsTheProvisionerClient(m) {
		return broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username(), true
	}
	return "", false
}

// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its
// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an
// account (mayIssue), has nothing to issue; an account minted before is kept.
func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
	node, module string) (bool, string, error) {
	// Mint only what no row names: the assign act hears its own write and may issue at once (the review of issue 1945).
	ctx = onlyIfAbsent(ctx)
	m, known := shelf[module]
	if !known || mayIssue(m) != nil {
		return false, "", nil
	}
	if _, own := ownAccountOf(m, node); own {
		minted, err := inv.BusMinted(ctx) // sealed accounts alone: a row with no sealed credential is not issued
		if err != nil {
			return false, "", fmt.Errorf("the issued accounts cannot be read: %w", err)
		}
		if missingAccounts(m, node, minted) {
			busAddress, err := broker.BusAddress()
			if err != nil {
				return false, "", err
			}
			// Its processes' and provisioner client's accounts too: issueOnTheNewBus issues each that is not issued.
			ctx, got := withIssued(ctx)
			if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil {
				return false, "", err
			}
			if issued := got.list(); len(issued) > 0 {
				return true, "issued and sealed to " + node + ": " + strings.Join(issued, ", "), nil
			}
			return false, "", nil
		}
		return false, "its bus accounts on " + node + " stood issued and sealed", nil
	}
	line, err := issueMissingProcessAccounts(ctx, inv, m, node)
	if err != nil {
		return false, "", err
	}
	return line != "", line, nil
}

// missingAccounts says some account of m on node is not issued: its own, a process's, or its provisioner client's.
func missingAccounts(m catalogue.Manifest, node string, minted []string) bool {
	if user, own := ownAccountOf(m, node); own && !slices.Contains(minted, user) {
		return true
	}
	if runsTheProvisionerClient(m) &&
		!slices.Contains(minted, broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()) {
		return true
	}
	return len(processAccountsMissing(m, node, minted)) > 0
}

// liveAssignmentAccountSteps are the steps on the controller's own stores and the bus: the composer's passes, as the
// done keys' take-ins use them, so a node asked for by both is composed once a pass.
func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
	inv := open.inventory
	return assignmentAccountSteps{
		holds: func(ctx context.Context) bool {
			// The lease's gate, as every act's: a serving controller the bus does not grant the lease yet acts unleased.
			_, err := theLease.epoch(ctx)
			return err == nil
		},
		forgetStale: func(ctx context.Context) ([]string, error) { return forgetStaleAccounts(ctx, open) },
		catalogue:   func(ctx context.Context) (map[string]catalogue.Manifest, error) { return inv.Catalogue(ctx) },
		issue: func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (bool, string, error) {
			return issueAssignmentAccounts(ctx, inv, shelf, node, module)
		},
		busHolder: func(ctx context.Context) (string, error) {
			holders, err := seatHolders(ctx, inv)
			if err != nil {
				return "", err
			}
			return holders[catalogue.BrokerSeat].Node, nil
		},
		compose: func(ctx context.Context, nodes []string) error {
			return composerFor(open).compose(ctx, nodes, false)
		},
		grants: func(ctx context.Context) error {
			return composerFor(open).compose(ctx, nil, true)
		},
	}
}

// assignmentAccountsRetry is how long a settle that failed waits before it tries again.
var assignmentAccountsRetry = 30 * time.Second

// untilSettled runs settle until it works or ctx ends: a failure, or a controller not holding the lease, is said and
// tried again every assignmentAccountsRetry, never dropped, since no later write of the key may come.
func untilSettled(ctx context.Context, what string, settle func(context.Context) (string, error)) {
	for {
		said, err := settle(ctx)
		if err == nil {
			if said != "" {
				for _, line := range strings.Split(said, "\n") {
					fmt.Printf("assignments: %s\n", line)
				}
			}
			return
		}
		if ctx.Err() != nil {
			return
		}
		fmt.Printf("assignments: %s: %v; tried again in %s\n", what, err, assignmentAccountsRetry)
		select {
		case <-ctx.Done():
			return
		case <-time.After(assignmentAccountsRetry):
		}
	}
}

// watchAssignmentAccounts settles every assignment of the assignments bucket that comes to stand, for as long as ctx: the
// bucket's current values first (the start's look, settled as not fresh), then each write. A watch that ends, or a
// bucket not there yet, is watched again a minute later.
func watchAssignmentAccounts(ctx context.Context, open *stores) {
	steps := liveAssignmentAccountSteps(open)
	goServing(ctx, func() {
		runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
			inBackground(ctx, func() {
				untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
					return steps.settle(ctx, a.Node, a.Module)
				})
			})
		}, func(look []chain.Assignment) {
			// The start's look, once in the process: one reading of the catalogue, one assignment after another,
			// then one composition of the bus's node and the grants (issues 1945 and 1277).
			inBackground(ctx, func() {
				untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
					return steps.look(ctx, look)
				})
			})
		}, func() bool {
			select {
			case <-ctx.Done():
				return false
			case <-time.After(time.Minute):
				return true
			}
		})
	})
}

// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each
// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its
// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the
// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of
// fresh settles replaces the look (the review of issue 1945).
func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
	fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) {
	turns := &assignmentTurns{}
	first := true
	for {
		var atStart []chain.Assignment
		if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) {
			if !isFresh {
				atStart = append(atStart, a)
				return
			}
			fresh(a)
		}, func() {
			first = false
			look(atStart)
			atStart = nil
		}); err != nil && ctx.Err() == nil {
			fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
		}
		if first {
			turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew
		}
		if ctx.Err() != nil || !pause() {
			return
		}
	}
}

// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
// to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once
// they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one
// ended, is fresh: an assignment that came to stand while no watch heard it is settled whole, never as the start's look
// (the review of issue 1945).
func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
	turns *assignmentTurns, first bool, settle func(a chain.Assignment, fresh bool), read func()) error {
	ctx, stop := context.WithCancel(ctx)
	defer stop()
	entries, err := watch(ctx)
	if err != nil {
		return err
	}
	current := true
	for {
		select {
		case <-ctx.Done():
			return nil
		case e, ok := <-entries:
			if !ok {
				if ctx.Err() != nil {
					return nil
				}
				return fmt.Errorf("the watch ended")
			}
			if e.Caught {
				if current {
					current = false // the current values are read: what follows is written from now on
					if first {
						read()
					}
				}
				continue
			}
			var value []byte
			if !e.Deleted {
				value = e.Value
			}
			if a, turned := turns.heard(e.Key, value); turned {
				settle(a, !current || !first)
			}
		}
	}
}

// staleAccounts is, by node, each module some account of which stands on a node where the module no longer resolves:
// the accounts of a module's own, its processes' and its provisioner client's. A node whose resolution is unknown is
// left alone.
func staleAccounts(users map[string]inventory.BusUser, resolved map[string]map[string]bool) map[string][]string {
	seen := map[string]map[string]bool{}
	for _, u := range users {
		switch u.Kind {
		case inventory.BusModule, inventory.BusNodeTools, inventory.BusProcess, inventory.BusProvisioner:
		default:
			continue
		}
		on, known := resolved[u.Node]
		if u.Node == "" || u.Module == "" || !known || on[u.Module] {
			continue
		}
		if seen[u.Node] == nil {
			seen[u.Node] = map[string]bool{}
		}
		seen[u.Node][u.Module] = true
	}
	out := map[string][]string{}
	for node, modules := range seen {
		out[node] = slices.Sorted(maps.Keys(modules))
	}
	return out
}

// forgetStaleAccounts forgets the accounts of every module on a node where it no longer resolves (assigned or
// required), from each node's own resolution; a node that does not resolve is left alone.
func forgetStaleAccounts(ctx context.Context, open *stores) ([]string, error) {
	inv := open.inventory
	users, err := inv.BusUsers(ctx)
	if err != nil {
		return nil, err
	}
	ctx = withInputCache(ctx) // one reading of the mesh for every node resolved
	resolved := map[string]map[string]bool{}
	for _, u := range users {
		if _, done := resolved[u.Node]; done || u.Node == "" {
			continue
		}
		plan, _, err := resolveNode(ctx, open, u.Node)
		if err != nil {
			continue // unknown: nothing of this node is forgotten
		}
		on := map[string]bool{}
		for _, m := range plan.Modules {
			on[m.Module] = true
		}
		resolved[u.Node] = on
	}
	var lines []string
	stale := staleAccounts(users, resolved)
	for _, node := range slices.Sorted(maps.Keys(stale)) {
		for _, module := range stale[node] {
			forgot, err := inv.ForgetBusAccountsOf(ctx, node, module, true)
			if err != nil {
				return lines, err
			}
			if len(forgot) > 0 {
				lines = append(lines, fmt.Sprintf("%s no longer runs on %s: its bus accounts forgotten: %s", module, node,
					strings.Join(forgot, ", ")))
			}
		}
	}
	return lines, nil
}