mesh-host/internal/selfcompose/selfcompose.go

// Package selfcompose is the node agent's own composer (novox/hq ADR 0547 §2, issue 1126): it composes this node's
// modules and its own composition with mesh-sdk's composer, from what is on the bus, and hands the compositions to the
// agent as the agent once read them from the controller's bucket `composed-<node>`.
//
// **It watches, it never polls** (ADR 0547 §2): its inputs are watches of the node's wants, the indexer's placement and
// modules, every node's own have key (where a node's record rides, ADR 0555 §3) and mesh-config's resolved values of this
// node. A change of any of them marks the node to compose again; one loop drains the mark, so a burst of changes composes
// the node once (composing one node takes 0.8 to 4.4 s, measured 2026-10-26). It hands the agent only the compositions that
// differ from the ones it handed before, and takes back one no want names any more. A composition that cannot be made is
// said, once per reason, and the ones handed before stand: a node keeps running what it ran rather than nothing.
package selfcompose

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"sync"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/compose"
	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
	"git.novox.be/novox/mesh-sdk/go/delivery"

	"github.com/novox/mesh-host/internal/agent"
)

// ResolvedBucket is mesh-config's state of resolved values, keyed `<node>.<module>.<key>` (mesh-catalog modules/mesh-config
// resolve.go ResolvedKey); a node reads its own slice alone (ADR 0547 §2).
const ResolvedBucket = "mesh-config_resolved"

// Env is what the composer needs of the node.
type Env struct {
	Node string
	// Watch follows one bucket's keys matching pattern, as the agent's watches do.
	Watch func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error)
	// Get reads one key's latest value; nil and no error when there is none.
	Get  func(ctx context.Context, bucket, key string) ([]byte, error)
	Mesh frombus.Mesh
	Say  func(string)
	// Own asks the vault's `own` in this node's engine's name, with the arguments given: its modules' own secrets and the
	// pair credentials asked, sealed to its key (ADR 0550 §5). Nil asks nothing, and a module with either does not compose.
	Own func(ctx context.Context, args map[string]any) (json.RawMessage, error)
	// Hear subscribes a subject: here the vault's word that a secret changed (VaultSecretChanged), on which the node
	// composes again, so a rotated secret or credential reaches it at once (ADR 0547 §2). Nil hears nothing.
	Hear func(subject string) (<-chan []byte, func(), error)
}

// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
const VaultSecretChanged = "mesh.mod.mesh-vault.event.secret-changed"

// Composer composes the node and is the agent's watcher of its compositions (agent.Env.Composed).
type Composer struct {
	env Env

	mu         sync.Mutex
	wants      map[string]delivery.Want // by module, the node's own under agent.NodeModule
	placements map[string]delivery.Placement
	records    map[string]delivery.NodeRecord
	modules    map[string]delivery.Module
	values     map[string]map[string]any // by module, this node's
	manifests  map[string][]byte         // by digest, read once
	handed     map[string]string         // composed key to the digest handed
	// sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again
	// is other bytes, so the one first had is kept while the generation stands, and the composition stays the same.
	sealedOwn map[string]ownSealed
	said      string
	caught    bool
	rev       uint64
}

// New is a composer for one node.
func New(env Env) *Composer {
	if env.Say == nil {
		env.Say = func(string) {}
	}
	return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{},
		records: map[string]delivery.NodeRecord{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
		manifests: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}}
}

// ownSealed is one own secret's sealed value at a generation and the key it is sealed to.
type ownSealed struct {
	generation int
	key        string
	sealed     string
}

// input is one watched bucket and how its updates are taken.
type input struct {
	bucket, pattern string
	take            func(c *Composer, u agent.Update)
}

func (c *Composer) inputs() []input {
	node := c.env.Node
	return []input{
		{delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant},
		{delivery.PlacementBucket, ">", (*Composer).takePlacement},
		{delivery.HaveBucket, "*." + delivery.HaveNodeWord, (*Composer).takeRecord},
		{delivery.ModulesBucket, ">", (*Composer).takeModule},
		{ResolvedBucket, node + ".>", (*Composer).takeValue},
	}
}

// Watch starts the composer and answers the compositions it makes: each as a key of `composed-<node>` would be, then a
// Caught update once every input was read and the node composed once, then every change. It ends when ctx ends or an
// input's watch ends.
func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, error) {
	ins := c.inputs()
	type tagged struct {
		in int
		u  agent.Update
		ok bool
	}
	merged := make(chan tagged, 64)
	for i, in := range ins {
		ch, err := c.env.Watch(ctx, in.bucket, in.pattern)
		if err != nil {
			return nil, fmt.Errorf("the node composes from %s, which cannot be watched: %w", in.bucket, err)
		}
		go func(i int, ch <-chan agent.Update) {
			for u := range ch {
				select {
				case merged <- tagged{in: i, u: u, ok: true}:
				case <-ctx.Done():
					return
				}
			}
			select {
			case merged <- tagged{in: i}:
			case <-ctx.Done():
			}
		}(i, ch)
	}
	if c.env.Hear != nil {
		heard, stop, err := c.env.Hear(VaultSecretChanged)
		if err != nil {
			return nil, fmt.Errorf("the node composes again when a secret changes, and cannot hear the vault say so: %w", err)
		}
		go func() {
			defer stop()
			for {
				select {
				case <-ctx.Done():
					return
				case _, ok := <-heard:
					if !ok {
						return
					}
					select {
					case merged <- tagged{in: -1, ok: true}:
					case <-ctx.Done():
						return
					}
				}
			}
		}()
	}
	out := make(chan agent.Update, 64)
	go func() {
		defer close(out)
		caught := make([]bool, len(ins))
		ready := false
		dirty := false
		for {
			// Drain whatever came, then compose once for all of it.
			select {
			case <-ctx.Done():
				return
			case t := <-merged:
				if !t.ok {
					c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket))
					return
				}
				switch {
				case t.in < 0:
					dirty = true // a secret changed: its new generation is asked of the vault at the compose
				case t.u.Caught:
					caught[t.in] = true
				default:
					ins[t.in].take(c, t.u)
					dirty = true
				}
				if len(merged) > 0 {
					continue
				}
			}
			if !ready {
				ready = all(caught)
				if !ready {
					continue
				}
				dirty = true
			}
			if !dirty {
				continue
			}
			dirty = false
			for _, u := range c.compose(ctx) {
				select {
				case out <- u:
				case <-ctx.Done():
					return
				}
			}
			if !c.caughtSent() {
				select {
				case out <- agent.Update{Caught: true}:
				case <-ctx.Done():
					return
				}
			}
		}
	}()
	return out, nil
}

func all(b []bool) bool {
	for _, v := range b {
		if !v {
			return false
		}
	}
	return true
}

// caughtSent says the Caught update went out, marking it sent the first time it is asked.
func (c *Composer) caughtSent() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	sent := c.caught
	c.caught = true
	return sent
}

func (c *Composer) takeWant(u agent.Update) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
		if u.Deleted {
			delete(c.wants, agent.NodeModule)
		} else {
			c.wants[agent.NodeModule] = delivery.Want{}
		}
		return
	}
	node, module, err := delivery.ParseWantKey(u.Key)
	if err != nil || node != c.env.Node {
		return
	}
	w, err := delivery.DecodeWant(u.Value)
	if u.Deleted || err != nil || w.Removed || w.FP == "" {
		delete(c.wants, module)
		return
	}
	c.wants[module] = w
}

func (c *Composer) takePlacement(u agent.Update) {
	node, err := delivery.ParsePlacementKey(u.Key)
	if err != nil {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	p, err := delivery.DecodePlacement(u.Value)
	if u.Deleted || err != nil || p.Node != node {
		delete(c.placements, node)
		return
	}
	c.placements[node] = p
}

func (c *Composer) takeRecord(u agent.Update) {
	node, _, isNode, err := delivery.ParseHaveKey(u.Key)
	if err != nil || !isNode {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	h, err := delivery.DecodeHaveNode(u.Value)
	if u.Deleted || err != nil || h.Record == nil {
		delete(c.records, node)
		return
	}
	c.records[node] = *h.Record
}

func (c *Composer) takeModule(u agent.Update) {
	name, err := delivery.ParseModulesKey(u.Key)
	if err != nil {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	m, err := delivery.DecodeModule(u.Value)
	if u.Deleted || err != nil {
		delete(c.modules, name)
		return
	}
	c.modules[name] = m
}

func (c *Composer) takeValue(u agent.Update) {
	rest, ok := strings.CutPrefix(u.Key, c.env.Node+".")
	if !ok {
		return
	}
	module, key, ok := strings.Cut(rest, ".")
	if !ok {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	var r struct {
		Value json.RawMessage `json:"value"`
	}
	var v any
	if u.Deleted || json.Unmarshal(u.Value, &r) != nil || len(r.Value) == 0 || json.Unmarshal(r.Value, &v) != nil {
		delete(c.values[module], key)
		return
	}
	if c.values[module] == nil {
		c.values[module] = map[string]any{}
	}
	c.values[module][key] = v
}

// compose composes the node once and answers the updates to hand: each composition whose digest differs from the one
// handed, and a deletion for each handed one no want names any more.
func (c *Composer) compose(ctx context.Context) []agent.Update {
	c.mu.Lock()
	fps := map[string]string{}
	kept := map[string]string{}
	var behind []string // modules wanted at another build than the trunk's: composed with that build's manifest
	for module, w := range c.wants {
		if module == agent.NodeModule {
			continue
		}
		fps[module] = w.FP
		if m, ok := c.modules[module]; ok && m.FP2 != w.FP {
			behind = append(behind, module)
		}
	}
	_, nodeWanted := c.wants[agent.NodeModule]
	r := reader{c: c}
	c.mu.Unlock()
	sort.Strings(behind)
	for _, module := range behind {
		digest, err := c.builtManifest(ctx, module, fps[module])
		if err != nil {
			c.sayOnce(fmt.Sprintf("%s is wanted at %s, whose manifest cannot be read, so the node does not compose: %v", module, fps[module], err))
			return nil
		}
		kept[module] = digest
	}

	compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, Kept: kept}, fps)
	if err != nil {
		c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
		return nil
	}
	c.sayOnce("")
	c.mu.Lock()
	defer c.mu.Unlock()
	var out []agent.Update
	now := map[string]bool{}
	keys := make([]string, 0, len(compositions))
	for key := range compositions {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		comp := compositions[key]
		if comp.Module == "" && !nodeWanted {
			continue
		}
		now[key] = true
		dig, err := delivery.ComposedDigestOf(comp)
		if err != nil {
			c.env.Say(fmt.Sprintf("the composition %s has no digest, and is not handed: %v", key, err))
			continue
		}
		if c.handed[key] == dig {
			continue
		}
		raw, err := delivery.EncodeComposed(comp)
		if err != nil {
			c.env.Say(fmt.Sprintf("the composition %s cannot be written, and is not handed: %v", key, err))
			continue
		}
		c.rev++
		c.handed[key] = dig
		out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev})
	}
	for key := range c.handed {
		if !now[key] {
			c.rev++
			delete(c.handed, key)
			out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
		}
	}
	return out
}

func (c *Composer) sayOnce(line string) {
	c.mu.Lock()
	if c.said == line {
		c.mu.Unlock()
		return
	}
	c.said = line
	c.mu.Unlock()
	if line != "" {
		c.env.Say(line)
	}
}

// builtManifest is the digest of the manifest a module's build at a fingerprint was made from (the builder's build key):
// what a node composes a module from when its want names another build than the trunk's.
func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string, error) {
	key, err := delivery.BuilderBuildKey(module, fp)
	if err != nil {
		return "", err
	}
	raw, err := c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
	if err != nil {
		return "", err
	}
	if raw == nil {
		return "", fmt.Errorf("the builder holds no build %s", key)
	}
	b, err := delivery.DecodeBuilderBuild(raw)
	if err != nil {
		return "", err
	}
	if b.Manifest == "" {
		return "", fmt.Errorf("the build %s names no manifest", key)
	}
	return b.Manifest, nil
}

// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for
// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read.
func (c *Composer) Policy(module string) string {
	c.mu.Lock()
	m, ok := c.modules[module]
	c.mu.Unlock()
	if !ok || m.Manifest == "" {
		return ""
	}
	raw, err := reader{c: c}.Manifest(context.Background(), m.Manifest)
	if err != nil {
		return ""
	}
	parsed, err := catalogue.ParseManifestForInputs(raw)
	if err != nil {
		return ""
	}
	policy, _, _ := catalogue.DerivedUpgrade(parsed)
	return policy
}

// reader is the composer's view of the bus as frombus reads it.
type reader struct{ c *Composer }

func (r reader) Placements(context.Context) (map[string]delivery.Placement, error) {
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make(map[string]delivery.Placement, len(r.c.placements))
	for k, v := range r.c.placements {
		out[k] = v
	}
	return out, nil
}

func (r reader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make(map[string]delivery.NodeRecord, len(r.c.records))
	for k, v := range r.c.records {
		out[k] = v
	}
	return out, nil
}

func (r reader) Modules(context.Context) (map[string]delivery.Module, error) {
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make(map[string]delivery.Module, len(r.c.modules))
	for k, v := range r.c.modules {
		out[k] = v
	}
	return out, nil
}

func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
	r.c.mu.Lock()
	raw, ok := r.c.manifests[digest]
	r.c.mu.Unlock()
	if ok {
		return raw, nil
	}
	key, err := delivery.ManifestsKey(digest)
	if err != nil {
		return nil, err
	}
	value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key)
	if err != nil {
		return nil, err
	}
	if value == nil {
		return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
	}
	m, err := delivery.DecodeManifest(value)
	if err != nil {
		return nil, err
	}
	r.c.mu.Lock()
	r.c.manifests[digest] = []byte(m.JSON)
	r.c.mu.Unlock()
	return []byte(m.JSON), nil
}

func (r reader) Values(_ context.Context, node, module string) (map[string]any, error) {
	if node != r.c.env.Node {
		return nil, nil // a node reads its own slice alone (ADR 0547 §2)
	}
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	if len(r.c.values[module]) == 0 {
		return nil, nil
	}
	out := make(map[string]any, len(r.c.values[module]))
	for k, v := range r.c.values[module] {
		out[k] = v
	}
	return out, nil
}

func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecretRow, error) {
	if node != r.c.env.Node || r.c.env.Own == nil {
		return nil, nil
	}
	raw, err := r.c.env.Own(ctx, map[string]any{})
	if err != nil {
		return nil, fmt.Errorf("the vault's own: %w", err)
	}
	var a struct {
		Node string `json:"node"`
		Key  string `json:"key"`
		Own  []struct {
			Module     string          `json:"module"`
			Name       string          `json:"name"`
			Origin     string          `json:"origin"`
			Generation int             `json:"generation"`
			Sealed     json.RawMessage `json:"sealed"`
		} `json:"own"`
	}
	if err := json.Unmarshal(raw, &a); err != nil {
		return nil, fmt.Errorf("the vault's own answer is not one: %w", err)
	}
	if a.Node != node {
		return nil, fmt.Errorf("the vault answered %q's own secrets to %s", a.Node, node)
	}
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	out := make([]compose.OwnSecretRow, 0, len(a.Own))
	for _, o := range a.Own {
		id := o.Module + "/" + o.Name
		kept, ok := r.c.sealedOwn[id]
		if !ok || kept.generation != o.Generation || kept.key != a.Key {
			kept = ownSealed{generation: o.Generation, key: a.Key, sealed: string(o.Sealed)}
			r.c.sealedOwn[id] = kept
		}
		out = append(out, compose.OwnSecretRow{Node: node, Module: o.Module, Name: o.Name, Origin: o.Origin,
			SealedTo: a.Key, Sealed: kept.sealed})
	}
	return out, nil
}

func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProvider []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
	if node != r.c.env.Node || r.c.env.Own == nil {
		return nil, nil, nil
	}
	type asked struct {
		Role string `json:"role"`
		compose.CredentialKey
	}
	var ask []asked
	for _, k := range asConsumer {
		ask = append(ask, asked{Role: "consumer", CredentialKey: k})
	}
	for _, k := range asProvider {
		ask = append(ask, asked{Role: "provider", CredentialKey: k})
	}
	raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
	if err != nil {
		return nil, nil, fmt.Errorf("the vault's own: %w", err)
	}
	var a struct {
		Node        string `json:"node"`
		Key         string `json:"key"`
		Credentials []struct {
			Role string `json:"role"`
			compose.CredentialKey
			Name       string          `json:"name"`
			Generation int             `json:"generation"`
			Sealed     json.RawMessage `json:"sealed"`
		} `json:"credentials"`
	}
	if err := json.Unmarshal(raw, &a); err != nil {
		return nil, nil, fmt.Errorf("the vault's own answer is not one: %w", err)
	}
	if a.Node != node {
		return nil, nil, fmt.Errorf("the vault answered %q's credentials to %s", a.Node, node)
	}
	r.c.mu.Lock()
	defer r.c.mu.Unlock()
	var creds []compose.Credential
	var issued []compose.Issued
	for _, c := range a.Credentials {
		// The value first had is kept while its generation stands, so sealing it again moves no composition.
		id := c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local
		kept, ok := r.c.sealedOwn[id]
		if !ok || kept.generation != c.Generation || kept.key != a.Key {
			kept = ownSealed{generation: c.Generation, key: a.Key, sealed: string(c.Sealed)}
			r.c.sealedOwn[id] = kept
		}
		k := c.CredentialKey
		switch c.Role {
		case "consumer":
			creds = append(creds, compose.Credential{Provision: k.Provision, Consumer: k.Consumer, ConsumerModule: k.ConsumerModule,
				Provider: k.Provider, Local: k.Local, State: compose.CredentialHeld, Sealed: kept.sealed})
		case "provider":
			issued = append(issued, compose.Issued{Provision: k.Provision, Consumer: k.Consumer, ConsumerModule: k.ConsumerModule,
				Local: k.Local, Provider: k.Provider, ForProvider: kept.sealed})
		}
	}
	return creds, issued, nil
}