mesh-controller/internal/inventory/busrecords.go

package inventory

import (
	"context"
	"fmt"
	"strings"

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

// What the bus's user list is derived from, read out of the mesh's records.
//
// The deriving itself is pure and lives in the broker package; this is the reading, and it is kept
// apart for the reason that package keeps its own types: a permission must be a function of what a
// module declared, and a query that decided anything would be a second place authority came from.

// BusRecords is every fact the composer needs about who may reach the bus.
//
// **A module's authority comes from the manifest, not from the assignment.** The assignment says
// *where* it runs; what it may say is in what it declared, so the two are read together and the
// manifest is the one that decides.
func (i *Inventory) BusRecords(ctx context.Context) (broker.Records, error) {
	nodes, err := i.Nodes(ctx)
	if err != nil {
		return broker.Records{}, fmt.Errorf("cannot read the mesh's machines: %w", err)
	}
	declared, err := i.Catalogue(ctx)
	if err != nil {
		return broker.Records{}, fmt.Errorf("cannot read the catalogue: %w", err)
	}

	// Every seat any module declares, by name, so a module's claim can be resolved to the protocol
	// that seat promises. **Across the whole catalogue, not one manifest**: a seat is declared by
	// one module and held by another, which is the whole reason a seat exists (ADR 0118).
	seats := map[string]catalogue.SeatDeclaration{}
	for _, m := range declared {
		for _, s := range m.DefinesSeats {
			seats[s.Name] = s
		}
	}
	// And the mesh's own, which carry protocol too (novox/hq ADR 0121). Added after the modules'
	// rather than before, because a `mesh-*` name is the mesh's and registration refuses a module
	// declaring one — so this cannot be shadowed, and if it ever were, the mesh's own would win.
	for _, own := range catalogue.SeatsWithAProtocol() {
		seats[own.Name] = catalogue.SeatDeclaration{
			Name: own.Name, Scope: own.Scope,
			Accepts: own.Accepts, Emits: own.Emits, Serves: own.Serves,
		}
	}

	// Who holds each seat held once for the mesh, where the mesh recorded it (novox/hq issue 218).
	holdings, err := i.Holdings(ctx)
	if err != nil {
		return broker.Records{}, fmt.Errorf("cannot read who holds the mesh's seats: %w", err)
	}

	out := broker.Records{Assigned: map[string][]broker.Declared{}, People: map[string][]string{},
		Interchangeable: map[string]bool{}}
	for _, n := range nodes {
		out.Nodes = append(out.Nodes, n.Name)
		modules, err := i.Assigned(ctx, n.Name)
		if err != nil {
			return broker.Records{}, fmt.Errorf("cannot read what %s runs: %w", n.Name, err)
		}
		for _, module := range modules {
			m, known := declared[module]
			if !known {
				// Assigned and not in the catalogue. Said rather than composed with no authority:
				// a user with an empty permission list is a module that starts, connects, and is
				// refused by the server on its first publish — an authorisation error that says
				// nothing about a missing manifest.
				//
				// **The catalogue refuses to forget an assigned module, so this is the second line
				// and not the first.** It earns its place there anyway: relying on another
				// package's invariant is how a rule ends up enforced by nothing.
				return broker.Records{}, fmt.Errorf(
					"%s is assigned to %s and is not in the catalogue, so what it may say cannot "+
						"be derived", module, n.Name)
			}
			d := declaredFor(m, seats)
			d.Holds = heldHere(d.Holds, holdings, n.Name, module)
			out.Assigned[n.Name] = append(out.Assigned[n.Name], d)
			if m.Instances == catalogue.InstancesInterchangeable {
				out.Interchangeable[m.Module] = true
			}
		}
	}

	enrolling, err := i.NodesWithALiveToken(ctx)
	if err != nil {
		return broker.Records{}, err
	}
	out.Enrolling = enrolling

	people, err := i.People(ctx)
	if err != nil {
		return broker.Records{}, err
	}
	for _, p := range people {
		out.People[p.Name] = p.Invokes
	}
	return out, nil
}

// declaredFor is one module's manifest as the composer needs it: what it says about itself, and the
// protocol of every seat it holds or uses.
func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration) broker.Declared {
	// A consumed name is a module's event unless it names a seat, and only somebody holding the seat
	// set can tell (novox/hq ADR 0121). Split here, because the composer cannot look at a name and
	// know — and a role's event read as a module's is a subscription to a namespace nobody owns.
	var fromModules []string
	var watches []broker.Seat
	for _, c := range m.Consumes {
		emitter, event, named := strings.Cut(c, ".")
		if named {
			// A seat's event when the seat says it; else the event of the module of that name — a seat and
			// the module holding it may share a name (mesh-delivery, novox/hq ADR 0239).
			if s, isASeat := seats[emitter]; isASeat && catalogue.SeatSays(s.Emits, event) {
				watches = append(watches, broker.Seat{Name: s.Name, Emits: []string{event}})
				continue
			}
		}
		fromModules = append(fromModules, c)
	}

	d := broker.Declared{
		Module:   m.Module,
		Emits:    m.EmitsAll(),
		Consumes: fromModules,
		Watches:  watches,
		// The tools it answers, which is `tools` and not `serves`: the manifest's `serves` is the
		// facts a consumer needs to reach a provision, a different meaning under a similar word.
		Serves: m.Tools,
		// And what it calls (novox/hq ADR 0152) — the console's `*`, nothing else's.
		Invokes: m.Invokes,
		// And the state it keeps and reads (novox/hq ADR 0201).
		State: bucketsOf(m),
		Reads: m.Reads,
		// And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them.
		Checks: catalogue.HealthChecks(m),
	}
	// Whether it can be given an account at all: delivered as its own secret named broker, so one
	// that declares none has nowhere to read it (novox/hq issue 195).
	if _, reads := m.OwnSecrets["broker"]; !reads {
		d.NoAccount = true
	}
	// The bus's own module — the one holding mesh-broker — copies the bus's streams for the night's
	// backup, and its account is granted that and nothing else (novox/hq ADR 0235).
	d.SnapshotsTheBus = m.ClaimsSeat(catalogue.BrokerSeat)
	for _, c := range m.Claims {
		// Every seat with a protocol, the mesh's own included. One that says only who does a job is
		// not here and grants nothing, which is most of them.
		if s, hasAProtocol := seats[c.Name]; hasAProtocol {
			d.Holds = append(d.Holds, asSeat(s))
		}
	}
	for _, name := range m.Uses {
		if s, declaredSomewhere := seats[name]; declaredSomewhere {
			d.Uses = append(d.Uses, asSeat(s))
		}
	}
	return d
}

// bucketsOf is the state a module keeps, as the bus holds it.
func bucketsOf(m catalogue.Manifest) []broker.Bucket {
	var out []broker.Bucket
	for _, s := range m.State {
		out = append(out, broker.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds})
	}
	return out
}

// DeclaredBuckets is every bucket the catalogue declares, registered modules assigned or not: a
// bucket exists from registration, like a seat's stream, so a module reading it may watch before its
// owner runs anywhere (novox/hq ADR 0201).
func (i *Inventory) DeclaredBuckets(ctx context.Context) ([]broker.Bucket, error) {
	declared, err := i.Catalogue(ctx)
	if err != nil {
		return nil, fmt.Errorf("cannot read the catalogue: %w", err)
	}
	var out []broker.Bucket
	for _, m := range declared {
		out = append(out, bucketsOf(m)...)
	}
	return out, nil
}

func asSeat(s catalogue.SeatDeclaration) broker.Seat {
	return broker.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits,
		Serves: catalogue.VerbNames(s.Serves)}
}

// MeshSeats are the mesh's own seats that carry a protocol, as the bus needs them: what to make a work
// queue for, and whose holder gets a worker on it (novox/hq ADR 0121).
func MeshSeats() []broker.DeclaredSeat {
	var out []broker.DeclaredSeat
	for _, s := range catalogue.SeatsWithAProtocol() {
		out = append(out, broker.DeclaredSeat{
			Name: s.Name, Accepts: s.Accepts, Emits: s.Emits, Serves: catalogue.VerbNames(s.Serves),
		})
	}
	return out
}

// NodesWithALiveToken is every machine holding a token that could still be presented — issued, not
// expired, not redeemed.
//
// **One enrolment user per such token** (design 25 §6): the inbox an answer goes to is scoped to the
// token, because an answer carries that machine's credentials sealed to it and a shared inbox is one
// machine able to read another's.
func (i *Inventory) NodesWithALiveToken(ctx context.Context) ([]string, error) {
	rows, err := i.store.Pool().Query(ctx,
		`select distinct n.name
		   from enrolment_token t join node n on n.id = t.node
		  where t.redeemed is null and t.expires > now()
		  order by n.name`)
	if err != nil {
		return nil, fmt.Errorf("cannot read which machines hold a live token: %w", err)
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		out = append(out, name)
	}
	return out, rows.Err()
}

// heldHere keeps of what a module claims only the seats it holds on this machine (novox/hq issue 218).
// A seat held once per machine is held by every assignment that claims it. A seat held once for the
// mesh is held by one assignment: where the mesh recorded who holds it, a claim on any other machine
// grants nothing and issues nothing — or the module would serve the role's verbs from a machine that
// is not the role's, and a question to the mesh's store would be answered from the wrong database. A
// mesh seat with no holder on record is left as it was derived.
func heldHere(claimed []broker.Seat, holdings []catalogue.Held, node, module string) []broker.Seat {
	recorded := map[string][]catalogue.Held{}
	for _, h := range holdings {
		if h.Scope == catalogue.ScopeMesh {
			recorded[h.Claim] = append(recorded[h.Claim], h)
		}
	}
	var out []broker.Seat
	for _, s := range claimed {
		holders, onRecord := recorded[s.Name]
		if s.Scope != catalogue.ScopeMesh || !onRecord {
			out = append(out, s)
			continue
		}
		for _, h := range holders {
			if h.Node == node && h.Module == module {
				out = append(out, s)
				break
			}
		}
	}
	return out
}