mesh-host/internal/selfcompose/shadow.go

package selfcompose

import (
	"bytes"
	"context"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"sync"
	"time"

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

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

// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every proposal with a key in
// the placement seat's state `candidates`, it composes this node in shadow from the trunk's placement with the proposal's
// keys laid over it and its own inputs (the sealed values and boxes it holds among them), compares the result with what
// the node runs, and hands the answers to be written into the node's own have key, `<node>.node`, under `candidates`.
// It applies nothing and hands no composition: what runs is still composed from `placement` alone, by the Composer.
//
// **Event-driven, like the Composer.** It watches every candidate key, the trunk's placement, every node's record, the
// indexer's modules, this node's config values, and this node's module haves (what it runs moved); a burst of changes
// answers once. It runs whether or not the node is placed yet: a node that applies the controller's compositions answers
// against what it runs from them. An answer is handed only when it changes, and a proposal's answer is dropped when its
// keys go.
type Shadow struct {
	c      *Composer
	answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)

	mu         sync.Mutex
	candidates map[string]map[string]candidateKey // by proposal, by node
	answers    map[string]delivery.CandidateAnswer
	// wants are the node's wants by module (agent.NodeModule for its own): what a proposal deleting its placement returns
	// it to, the controller's compositions at them.
	wants map[string]delivery.Want
	// ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy
	// from the declaration kept before the switch (issue 1126, probe #532 on shanks).
	ran map[string]string
	// heldNow is, by module, how its have says the node keeps it today as it runs: "held", "failed", or "runs a build of
	// the old form" (a `build` and no `fp`, which no registered build of the trunk names); "" when it is composed as any
	// other. A candidate that cannot compose it, or holds it only behind one of these, keeps it as today
	// (frombus.Scope.HeldToday, issue 1126).
	heldNow map[string]string
	// answered is, by proposal, the candidate revisions and the input changes (moved) each answer was composed at: a proposal
	// is answered again only when either moved (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84
	// every 13-17 s, woken by its own modules' haves whose health words changed at every look).
	answered map[string]answeredAt
	// moved counts, by input (inputWords), the changes of each of the node's own inputs, and retry names the proposals
	// answered again after a read the bus did not answer: what woke each answer is said with it (issue 1126,
	// 2026-10-28, seed-9's probe: a safety check could not tell a loop from inputs moving).
	moved [len(inputWords)]uint64
	retry map[string]bool
	// owed says an answer met a read the bus did not answer: it is no answer to keep, and its proposal is answered again
	// after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not
	// yet ... no responders" in its answer to seed-7).
	owed bool
}

// answeredAt is what one proposal's answer was composed at: its keys' revisions, by node, and the changes of each input.
type answeredAt struct {
	revs  string
	moved [len(inputWords)]uint64
}

// inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it.
var inputWords = [...]string{"the candidate", "placement", "records", "modules", "values", "haves", "wants"}

// wokenBy says what moved between the answer a proposal stands at (had, when done) and the one about to be composed (at):
// its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words.
func wokenBy(had answeredAt, done, retry bool, at answeredAt) string {
	switch {
	case retry:
		return "a retry after a read the bus did not answer"
	case !done:
		return "its first answer since the shadow began"
	}
	var words []string
	if had.revs != at.revs {
		words = append(words, inputWords[0])
	}
	for i := 1; i < len(inputWords); i++ {
		if had.moved[i] != at.moved[i] {
			words = append(words, inputWords[i])
		}
	}
	if len(words) == 0 {
		return "nothing that moved"
	}
	return strings.Join(words, ", ")
}

// candidateKey is one candidate key as last read: its value and revision.
type candidateKey struct {
	c   delivery.PlacementCandidate
	rev uint64
}

// NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key.
func NewShadow(env Env, answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)) *Shadow {
	return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{},
		answers: map[string]delivery.CandidateAnswer{}, wants: map[string]delivery.Want{}, ran: map[string]string{}}
}

// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch
// yet (the candidates' state not made, or the node's grant of it not loaded at a delivery) or a watch that ended is
// tried again rather than leaving the node answering nothing for the rest of the link (review of L2b, round 1, finding
// 2). Each distinct reason is said once.
func (s *Shadow) Keep(ctx context.Context, backoff func(try int) time.Duration) {
	said := ""
	for try := 1; ctx.Err() == nil; try++ {
		began := time.Now()
		err := s.Run(ctx)
		if ctx.Err() != nil {
			return
		}
		if time.Since(began) > time.Minute {
			try = 1 // it ran: a later stop begins the waits again
		}
		why := "it stopped"
		if err != nil {
			why = err.Error()
		}
		if why != said {
			s.c.env.Say("this node answers no placement candidate for now, and tries again: " + why)
			said = why
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(backoff(try)):
		}
	}
}

// Run answers until ctx ends or an input's watch ends.
func (s *Shadow) Run(ctx context.Context) error {
	// Each run watches afresh and starts from nothing read: what changed while it was down (a candidate key deleted)
	// is then not missed, and the watches of a run before are not left open (review of L2b, round 2).
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	s.mu.Lock()
	s.candidates = map[string]map[string]candidateKey{}
	s.wants = map[string]delivery.Want{}
	s.ran = map[string]string{}
	s.answered = map[string]answeredAt{}
	s.moved = [len(inputWords)]uint64{}
	s.retry = map[string]bool{}
	s.c = New(s.c.env)
	s.mu.Unlock()
	c := s.c
	node := c.env.Node
	// In inputWords' order.
	ins := []input{
		{delivery.PlacementCandidatesBucket, ">", nil, nil},
		{delivery.PlacementBucket, ">", (*Composer).takePlacement, nil},
		{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, nil},
		{delivery.ModulesBucket, ">", (*Composer).takeModule, nil},
		{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
		// What this node runs moved: an apply ends in its module's have.
		{delivery.HaveBucket, node + ".>", nil, nil},
		// Its wants: a proposal deleting its placement returns it to the controller's compositions at them.
		{delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil},
	}
	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 fmt.Errorf("the node answers placement candidates 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)
	}
	caught := make([]bool, len(ins))
	ready, dirty := false, false
	// Not ready after a minute is said, naming each input not read yet, in the journal and in an answer to each proposal it
	// knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531).
	waiting := time.NewTimer(notReadyAfter)
	defer waiting.Stop()
	var again <-chan time.Time
	tries := 0
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-again:
			again = nil
			dirty = true
		case <-waiting.C:
			if !ready {
				s.notReady(ctx, ins, caught)
			}
			continue
		case t := <-merged:
			if !t.ok {
				return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket)
			}
			// Only a change of what an answer is composed from answers again: a key written again with the value it held,
			// this node's own key (its answers), or a module's have whose health or words moved but not what it runs, is
			// none (issue 1126, 2026-10-28: each look rewrote shanks' unhealthy modules' haves, and each woke a full answer).
			changed := false
			switch {
			case t.u.Caught:
				caught[t.in] = true
			case t.in == 0:
				changed = s.takeCandidate(t.u)
			case t.in == 2:
				changed = s.takeRecord(t.u)
			case t.in == 5:
				folded := c.takeModuleHave(t.u)
				changed = s.takeRan(t.u) || folded
			case t.in == 6:
				changed = s.takeWant(t.u)
			default:
				changed = s.takeChanged(ins[t.in], t.u)
			}
			if changed {
				if t.in != 0 {
					s.moved[t.in]++ // the candidates' own changes are their revisions
				}
				dirty = true
			}
			if len(merged) > 0 {
				continue
			}
		}
		if !ready {
			if ready = all(caught); !ready {
				continue
			}
			dirty = true
		}
		if dirty {
			dirty = false
			s.answerAll(ctx)
			if s.owed {
				wait := LaterWait(tries)
				tries++
				c.env.Say(fmt.Sprintf("this node answers again in %s the placement candidates whose answer met a bus that did not answer", wait))
				again = time.After(wait)
			} else {
				tries, again = 0, nil
			}
		}
	}
}

// takeChanged takes u through in.take and says whether the entry it names changed: a placement, a module or a config
// value written again with the value it held changes nothing composed.
func (s *Shadow) takeChanged(in input, u agent.Update) bool {
	before, had, known := s.entry(in.bucket, u.Key)
	in.take(s.c, u)
	if !known {
		return true
	}
	after, has, _ := s.entry(in.bucket, u.Key)
	return had != has || !reflect.DeepEqual(before, after)
}

// entry is what the composer holds of one key of bucket, whether it holds it, and whether the bucket is one it knows.
func (s *Shadow) entry(bucket, key string) (v any, has, known bool) {
	c := s.c
	c.mu.Lock()
	defer c.mu.Unlock()
	switch bucket {
	case delivery.PlacementBucket:
		node, err := delivery.ParsePlacementKey(key)
		if err != nil {
			return nil, false, true
		}
		p, ok := c.placements[node]
		return p, ok, true
	case delivery.ModulesBucket:
		name, err := delivery.ParseModulesKey(key)
		if err != nil {
			return nil, false, true
		}
		m, ok := c.modules[name]
		return m, ok, true
	case ResolvedBucket:
		rest, ok := strings.CutPrefix(key, c.env.Node+".")
		if !ok {
			return nil, false, true
		}
		module, k, ok := strings.Cut(rest, ".")
		if !ok {
			return nil, false, true
		}
		val, ok := c.values[module][k]
		return val, ok, true
	}
	return nil, false, false
}

// revsOf is a proposal's keys' revisions, by node, as one word.
func revsOf(keys map[string]candidateKey) string {
	nodes := make([]string, 0, len(keys))
	for n := range keys {
		nodes = append(nodes, n)
	}
	sort.Strings(nodes)
	var b strings.Builder
	for _, n := range nodes {
		fmt.Fprintf(&b, "%s=%d;", n, keys[n].rev)
	}
	return b.String()
}

// takeCandidate takes one candidate key, and says whether it changed anything.
func (s *Shadow) takeCandidate(u agent.Update) bool {
	proposal, node, err := delivery.ParsePlacementCandidateKey(u.Key)
	if err != nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if u.Deleted {
		if _, ok := s.candidates[proposal][node]; !ok {
			return false
		}
		delete(s.candidates[proposal], node)
		if len(s.candidates[proposal]) == 0 {
			delete(s.candidates, proposal)
		}
		return true
	}
	c, err := delivery.DecodePlacementCandidate(u.Value)
	if err != nil || c.Proposal != proposal || (c.Placement != nil && c.Placement.Node != node) {
		// Not one: read as gone.
		if _, ok := s.candidates[proposal][node]; ok {
			delete(s.candidates[proposal], node)
			return true
		}
		return false
	}
	if s.candidates[proposal] == nil {
		s.candidates[proposal] = map[string]candidateKey{}
	}
	s.candidates[proposal][node] = candidateKey{c: c, rev: u.Revision}
	return true
}

// takeRecord takes a node's record, and says whether it changed: the node's own key changes with each answer written,
// which moves nothing composed.
func (s *Shadow) takeRecord(u agent.Update) bool {
	node, _, isNode, err := delivery.ParseHaveKey(u.Key)
	if err != nil || !isNode {
		return false
	}
	c := s.c
	c.mu.Lock()
	before, had := c.records[node]
	c.mu.Unlock()
	c.takeRecord(u)
	c.mu.Lock()
	after, has := c.records[node]
	c.mu.Unlock()
	return had != has || (has && !delivery.SameNodeRecord(before, after))
}

// How long the shadow waits before it says it is not ready, and how long a proposal's answer runs before it is said slow.
var notReadyAfter, slowAfter = time.Minute, time.Minute

// notReady says which inputs the shadow has not read yet, and answers each proposal it knows refused for it.
func (s *Shadow) notReady(ctx context.Context, ins []input, caught []bool) {
	var waits []string
	for i, in := range ins {
		if !caught[i] {
			waits = append(waits, in.bucket+" ("+in.pattern+")")
		}
	}
	why := fmt.Sprintf("the node has not read %s after %s; it answers once it has", strings.Join(waits, ", "), notReadyAfter)
	s.c.env.Say("this node answers no placement candidate yet: " + why)
	s.mu.Lock()
	next := map[string]delivery.CandidateAnswer{}
	for proposal, keys := range s.candidates {
		own, ok := keys[s.c.env.Node]
		if !ok {
			continue // no key of this node's: nothing asks it to answer
		}
		next[proposal] = delivery.CandidateAnswer{Proposal: proposal, Revision: own.rev, Refused: why, Verdict: delivery.CandidateRefused}
	}
	s.answers = next
	s.mu.Unlock()
	if len(next) > 0 && s.answer != nil {
		s.answer(ctx, next)
	}
}

// answerAll answers every proposal, the newest candidate first, and hands the answers each time one changes (issue 1126,
// T3: ace answered a closed pull request's group for 22 minutes before the newer one, all handed at the end). Newest
// is the highest revision of the proposal's keys: a withdrawn proposal's candidate is the older and waits; the seat
// drops a closed one's keys, which ends its answer here (a node does not read the forge). Each answer says when it
// starts, every progressEvery while it runs what it waits on, and when it ends (T2: the node said nothing for 22
// minutes).
func (s *Shadow) answerAll(ctx context.Context) {
	c := s.c
	node := c.env.Node
	s.mu.Lock()
	type listed struct {
		proposal string
		rev      uint64
		at       answeredAt
	}
	var proposals []listed
	for g, keys := range s.candidates {
		l := listed{proposal: g, at: answeredAt{revs: revsOf(keys), moved: s.moved}}
		for _, k := range keys {
			l.rev = max(l.rev, k.rev)
		}
		proposals = append(proposals, l)
	}
	// A proposal whose keys' revisions and the node's inputs are those its answer stands at is not answered again.
	for g := range s.answered {
		if _, ok := s.candidates[g]; !ok {
			delete(s.answered, g)
		}
	}
	// Each proposal's last answer stands until it is answered again; a proposal whose keys went is dropped.
	next := map[string]delivery.CandidateAnswer{}
	for _, l := range proposals {
		if a, ok := s.answers[l.proposal]; ok {
			next[l.proposal] = a
		}
	}
	s.owed = false
	s.mu.Unlock()
	sort.Slice(proposals, func(i, j int) bool {
		if proposals[i].rev != proposals[j].rev {
			return proposals[i].rev > proposals[j].rev
		}
		return proposals[i].proposal < proposals[j].proposal
	})
	hand := func() {
		s.mu.Lock()
		same := len(next) == len(s.answers)
		for g, a := range next {
			if old, ok := s.answers[g]; !ok || !sameAnswer(old, a) {
				same = false
			}
		}
		out := make(map[string]delivery.CandidateAnswer, len(next))
		for g, a := range next {
			out[g] = a
		}
		if !same {
			s.answers = out
		}
		s.mu.Unlock()
		if !same && s.answer != nil {
			given := make(map[string]delivery.CandidateAnswer, len(out))
			for g, a := range out {
				given[g] = a
			}
			s.answer(ctx, given)
			c.env.Say(fmt.Sprintf("%s answered %d placement candidate proposal(s)", node, len(given)))
		}
	}
	for _, l := range proposals {
		if ctx.Err() != nil {
			return
		}
		proposal := l.proposal
		s.mu.Lock()
		at, done := s.answered[proposal]
		retry := s.retry[proposal]
		s.mu.Unlock()
		if done && at == l.at {
			continue
		}
		began := time.Now()
		c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d; woken by %s)", proposal, l.rev, wokenBy(at, done, retry, l.at)))
		stop := make(chan struct{})
		go func() {
			tick := time.NewTicker(progressEvery)
			defer tick.Stop()
			for {
				select {
				case <-stop:
					return
				case <-tick.C:
					c.env.Say(fmt.Sprintf("the answer to %s has run %s and waits on %s", proposal,
						time.Since(began).Round(time.Second), c.waitingOn()))
				}
			}
		}()
		missed := c.missed.Load()
		a, ok := s.answerProposal(ctx, proposal)
		close(stop)
		provisional := c.missed.Load() > missed && ctx.Err() == nil
		took := time.Since(began)
		if took > slowAfter {
			c.env.Say(fmt.Sprintf("the answer to %s took %s", proposal, took.Round(time.Second)))
			if ok {
				// Words that do not move with the time it took, so an answer the same as before is not written again.
				a.Note = strings.TrimPrefix(a.Note+"; ", "; ") + "this answer took over " + slowAfter.String() + " to compose"
			}
		}
		if ok {
			a = delivery.FitAnswer(a, delivery.MaxAnswerBytes)
			next[proposal] = a
			c.env.Say(fmt.Sprintf("answered placement candidate %s in %s: %s", proposal, took.Round(time.Millisecond), a.Verdict))
		} else {
			delete(next, proposal)
			c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", proposal))
		}
		s.mu.Lock()
		if provisional {
			// What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
			s.owed = true
			s.retry[proposal] = true
			delete(s.answered, proposal)
			c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", proposal))
		} else {
			s.answered[proposal] = l.at
			delete(s.retry, proposal)
		}
		s.mu.Unlock()
		hand()
	}
	hand() // a proposal whose keys went, when no proposal is left to answer
}

// How often a running answer says what it waits on.
var progressEvery = time.Minute

// answerProposal is the node's answer to one proposal, and false when the node has nothing to answer: neither the trunk nor
// the proposal places it.
func (s *Shadow) answerProposal(ctx context.Context, proposal string) (delivery.CandidateAnswer, bool) {
	c := s.c
	node := c.env.Node
	defer c.waiting("composing")()
	s.mu.Lock()
	keys := s.candidates[proposal]
	given := make(map[string]*delivery.Placement, len(keys))
	for n, k := range keys {
		if k.c.Reached && k.c.Placement == nil {
			continue // reached, with no key of its own: nothing laid over it
		}
		given[n] = k.c.Placement
	}
	// The revision answered is the node's own candidate key's; a node with none is not asked.
	own, mine := keys[node]
	rev := own.rev
	s.mu.Unlock()

	c.mu.Lock()
	modules := make(map[string]delivery.Module, len(c.modules))
	for k, v := range c.modules {
		modules[k] = v
	}
	c.mu.Unlock()
	if !mine {
		return delivery.CandidateAnswer{}, false
	}

	var running []delivery.Composed
	if c.env.Running != nil {
		running = c.env.Running()
	}
	if p, ok := given[node]; ok && p == nil && !own.c.Reached {
		// The proposal deletes this node's placement key: the node then returns to the controller's compositions, as one the
		// catalogue does not place (endWhenUnplaced). Those at its wants are compared with what runs (issue 1126, the
		// seed's revert, whose check read "would stop running node").
		s.mu.Lock()
		wants := make(map[string]delivery.Want, len(s.wants))
		for m, w := range s.wants {
			wants[m] = w
		}
		s.mu.Unlock()
		return returning(ctx, c.get, node, wants, running, proposal, rev), true
	}

	placement := given[node]
	if placement == nil {
		c.mu.Lock()
		p := c.placements[node]
		c.mu.Unlock()
		placement = &p
	}
	// What it runs inside its own composition, of what the candidate places (the seed's trunk places nothing).
	c.mu.Lock()
	inOwn := c.inOwnOf(*placement)
	c.mu.Unlock()
	fps := shadowBuilds(running, *placement, modules, inOwn)
	kept := map[string]string{}
	for module, fp := range fps {
		if mod, ok := modules[module]; ok && mod.FP2 == fp {
			continue
		}
		kept[module] = fp // composed from that build (frombus.Scope.KeptAt, issue 1126)
	}
	// The boxes it runs, by what each module's have and its own key say after a restart (Held), are reused for the
	// modules it runs: a value sealed again is no difference.
	held := map[string]delivery.Want{agent.NodeModule: {}}
	for module := range fps {
		held[module] = delivery.Want{FP: fps[module]}
	}
	c.mu.Lock()
	c.wants = held
	c.mu.Unlock()
	a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn,
		SameValue: c.sameValue, HeldToday: s.heldToday()}, fps, running, proposal, rev)
	s.mu.Lock()
	ran := make(map[string]string, len(s.ran))
	for m, d := range s.ran {
		ran[m] = d
	}
	s.mu.Unlock()
	return asHavesSay(a, ran), true
}

// heldToday is what the node holds today: each module whose have's phase is held, and what its own composer holds, the
// node's own composition ("node") among them when it composes none of its own.
func (s *Shadow) heldToday() map[string]string {
	out := map[string]string{}
	s.mu.Lock()
	for m, how := range s.heldNow {
		if how != "" {
			out[m] = how
		}
	}
	s.mu.Unlock()
	for _, m := range s.c.heldNow() {
		if _, said := out[m]; !said {
			out[m] = "held"
		}
	}
	return out
}

// takeRan keeps the digest one of this node's module haves says runs, and says whether it changed: a have whose health
// or words alone moved runs the same.
func (s *Shadow) takeRan(u agent.Update) bool {
	node, module, isNode, err := delivery.ParseHaveKey(u.Key)
	if err != nil || isNode || node != s.c.env.Node {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	before, had := s.ran[module]
	h, err := delivery.DecodeHave(u.Value)
	if s.heldNow == nil {
		s.heldNow = map[string]string{}
	}
	wasHeld := s.heldNow[module]
	s.heldNow[module] = ""
	if !u.Deleted && err == nil {
		switch {
		case h.Phase == delivery.PhaseHeld:
			s.heldNow[module] = "held"
		case h.Phase == delivery.PhaseFailed:
			s.heldNow[module] = "failed"
		case h.Phase != delivery.PhaseRemoved && h.FP == "" && h.Build != "":
			// A build of the old form: its have names a `build` and no `fp`, which no registered build of the trunk
			// names, so the controller keeps it as it runs (issue 1126: novox's de-spiegel at build-…); a seed keeps it so.
			s.heldNow[module] = "runs a build of the old form"
		}
	}
	holdMoved := wasHeld != s.heldNow[module] // a hold's start or end changes the answer too
	if u.Deleted || err != nil || h.Composed == "" || h.Phase == delivery.PhaseRemoved {
		delete(s.ran, module)
		return had || holdMoved
	}
	s.ran[module] = h.Composed
	return !had || before != h.Composed || holdMoved
}

// asHavesSay is a with each module that differs only from the copy the agent keeps of the declaration before the switch
// counted the same where the composed digest is the one its have says runs: what runs is the composition at that
// digest, never the kept copy (issue 1126, probe #532: shanks' xterm, zsh, zsh-autosuggestions and
// zsh-syntax-highlighting).
func asHavesSay(a delivery.CandidateAnswer, ran map[string]string) delivery.CandidateAnswer {
	if a.Refused != "" || len(a.Differs) == 0 {
		return a
	}
	kept := a.Differs[:0:0]
	for _, d := range a.Differs {
		if d.Composed != "" && ran[d.Module] == d.Composed {
			a.Same = append(a.Same, d.Module)
			continue
		}
		kept = append(kept, d)
	}
	a.Differs = kept
	sort.Strings(a.Same)
	a.Verdict = delivery.VerdictOf(a)
	return a
}

// shadowBuilds is the build each module placed is composed at in shadow: the one it runs, and for one the candidate
// adds, the trunk's. A module it runs inside its own composition (inOwn, from its haves: mesh-wireguard, one at a build
// of the old form) is given none, so it is folded there again, as the node's own composer folds it, rather than
// composed at the trunk's build as if the candidate added it (issue 1126: nats and photos have a trunk build that no
// want names).
func shadowBuilds(running []delivery.Composed, placement delivery.Placement, modules map[string]delivery.Module, inOwn []string) map[string]string {
	fps := map[string]string{}
	for _, r := range running {
		if r.Module != "" && r.FP != "" {
			fps[r.Module] = r.FP
		}
	}
	folded := map[string]bool{}
	for _, m := range inOwn {
		folded[m] = true
	}
	for _, m := range placement.Modules {
		if _, runs := fps[m.Module]; !runs && !folded[m.Module] {
			if mod, ok := modules[m.Module]; ok && mod.FP2 != "" {
				fps[m.Module] = mod.FP2
			}
		}
	}
	return fps
}

func sameAnswer(a, b delivery.CandidateAnswer) bool {
	x, errA := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}})
	y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}})
	return errA == nil && errB == nil && string(x) == string(y)
}

// takeWant takes one of the node's wants, and says whether it changed.
func (s *Shadow) takeWant(u agent.Update) bool {
	module := ""
	var w delivery.Want
	if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
		module = agent.NodeModule
	} else if node, m, err := delivery.ParseWantKey(u.Key); err == nil && node == s.c.env.Node {
		module = m
		if !u.Deleted {
			if w, err = delivery.DecodeWant(u.Value); err != nil || w.Removed || w.FP == "" {
				u.Deleted = true
			}
		}
	} else {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	old, had := s.wants[module]
	if u.Deleted {
		delete(s.wants, module)
		return had
	}
	s.wants[module] = w
	return !had || old.FP != w.FP
}

// returning is the node's answer to a proposal deleting its placement: it returns to the controller's compositions, each
// read at the fingerprint its want names in the node's composition bucket and compared with what runs
// (delivery.CompareComposed). A want the controller holds no composition for, or one that cannot be read, refuses the
// answer, naming each: returning then would leave that module with nothing to apply.
func returning(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string,
	wants map[string]delivery.Want, running []delivery.Composed, proposal string, rev uint64) delivery.CandidateAnswer {
	a := delivery.CandidateAnswer{Proposal: proposal, Revision: rev}
	bucket, err := delivery.ComposedBucketFor(node)
	if err != nil {
		a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused
		return a
	}
	modules := make([]string, 0, len(wants))
	for m := range wants {
		modules = append(modules, m)
	}
	sort.Strings(modules)
	comps := map[string]delivery.Composed{}
	var missing []string
	for _, m := range modules {
		var key string
		if m == agent.NodeModule {
			key, err = delivery.ComposedNodeKey(node)
		} else {
			key, err = delivery.ComposedKey(node, m, wants[m].FP)
		}
		if err != nil {
			missing = append(missing, m+": "+err.Error())
			continue
		}
		raw, err := get(ctx, bucket, key)
		switch {
		case err != nil:
			missing = append(missing, fmt.Sprintf("%s: the controller's composition %s cannot be read: %v", m, key, err))
			continue
		case raw == nil:
			missing = append(missing, fmt.Sprintf("%s: the controller holds no composition %s, which its want names", m, key))
			continue
		}
		comp, err := delivery.DecodeComposed(raw)
		if err != nil {
			missing = append(missing, fmt.Sprintf("%s: the controller's composition %s: %v", m, key, err))
			continue
		}
		comps[key] = comp
	}
	if len(missing) > 0 {
		a.Refused = "returning to the controller's compositions, " + strings.Join(missing, "; ")
		a.Verdict = delivery.CandidateRefused
		return a
	}
	cmp := delivery.CompareComposed(comps, running)
	a.Returns = delivery.ReturnsToController
	a.Same, a.Differs, a.Added, a.Removed = cmp.Same, cmp.Differs, cmp.Added, cmp.Removed
	a.Verdict = delivery.VerdictOf(a)
	return a
}

// sameValue says whether two sealed values open, with this node's key, to the same value; unknown when either does not
// open (or the node holds no opener). Only the comparison leaves: never a value (issue 1126, the shadow shows its work).
func (c *Composer) sameValue(a, b string) (same, known bool) {
	if c.env.Unseal == nil {
		return false, false
	}
	x, errA := c.env.Unseal(a)
	y, errB := c.env.Unseal(b)
	if errA != nil || errB != nil {
		return false, false
	}
	return bytes.Equal(x, y), true
}