// Package composer is the node-engine's composition as one function over a plain request (novox/hq ADR 0482 clause 4,
// ADR 0593 §7): what the command mesh-compose runs, so a group that changes the node-engine's composition is composed by
// the composer built at that group's head of the node-engine's repository, in a build module's sandbox.
//
// **It reads its request and nothing else.** No bus, store, network, file or environment: the manifests come inside the
// request (raw, as their repository holds them), the rest of the mesh is the request's input, and the answer is a value.
// The request's and the verdict's shapes are mesh-sdk's composer package; their logic, which composes, is here.
package composer

import (
	"encoding/json"
	"fmt"
	"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
	"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay"
	"maps"
	"slices"
	"sort"
	"strings"

	sdk "git.novox.be/novox/mesh-delivery/sdk/composer"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/keys"
	"git.novox.be/novox/mesh-sdk/go/sdkversion"

	"github.com/novox/mesh-host/internal/compose"
	"github.com/novox/mesh-host/internal/compose/frombus"
)

// The shapes of mesh-sdk's composer package, by the names this package's code was written with.
type (
	Request      = sdk.Request
	Case         = sdk.Case
	Manifest     = sdk.Manifest
	Verdict      = sdk.Verdict
	CaseVerdict  = sdk.CaseVerdict
	ManifestRead = sdk.ManifestRead
	Machine      = sdk.Machine
	Path         = sdk.Path
)

const (
	Version      = sdk.Version
	DepthResolve = sdk.DepthResolve
	DepthCompose = sdk.DepthCompose
)

// ErrUnfilled is a manifest named by a digest that nobody replaced by its bytes (sdk.ErrUnfilled).
var ErrUnfilled = sdk.ErrUnfilled

// inputOf is the request's input as this composer's composition input: the request carries it as JSON (mesh-sdk holds no
// composition), the same bytes either way.
func inputOf(r Request) (compose.Input, error) {
	var in compose.Input
	raw, err := json.Marshal(r.Input)
	if err != nil {
		return in, fmt.Errorf("composer: the request's input cannot be read: %w", err)
	}
	if len(raw) == 0 || string(raw) == "null" {
		return in, nil
	}
	if err := json.Unmarshal(raw, &in); err != nil {
		return in, fmt.Errorf("composer: the request's input is not a composition input: %w", err)
	}
	return in, nil
}

// Run composes every case of r and answers the verdict, its SDK this binary's mesh-sdk. An error is a request it
// cannot run (an unknown depth, a case named twice, a manifest left as a digest); a machine that does not compose is a
// verdict.
func Run(r Request) (Verdict, error) {
	depth := r.Depth
	if depth == "" {
		depth = DepthCompose
	}
	if depth != DepthResolve && depth != DepthCompose {
		return Verdict{}, fmt.Errorf("composer: the depth %q is not resolve or compose", r.Depth)
	}
	v := Verdict{V: Version, SDK: sdkversion.Version()}
	seen := map[string]bool{}
	for _, c := range r.Cases {
		if seen[c.Name] {
			return Verdict{}, fmt.Errorf("composer: the case %q is named twice", c.Name)
		}
		seen[c.Name] = true
		cv, err := runCase(r, c, depth)
		if err != nil {
			return Verdict{}, err
		}
		v.Cases = append(v.Cases, cv)
	}
	return v, nil
}

// parse decodes every manifest leniently and takes its verdict (novox/hq ADR 0578 §3, §5): the composer judges none. A
// manifest refused by its verdict is held, composed as its stub without its items (as the bus reader holds one); bytes
// that do not decode are left out. Both are named in reads.
func parse(m map[string]Manifest, reads map[string]ManifestRead, verdicts map[string]json.RawMessage) (map[string]catalogue.Manifest, error) {
	out := map[string]catalogue.Manifest{}
	for name, man := range m {
		if len(man.Raw) == 0 {
			return nil, fmt.Errorf("%w: %s (digest %s)", ErrUnfilled, name, man.Digest)
		}
		decoded, err := catalogue.Decode(man.Raw)
		read := ManifestRead{Accepted: err == nil, SetAside: decoded.SetAside()}
		if err != nil {
			read.Refused = fmt.Sprintf("the manifest of %s does not decode: %v", name, err)
		} else if raw, ok := verdicts[name]; ok {
			v, verr := caseVerdict(raw)
			if verr != nil {
				return nil, fmt.Errorf("composer: the verdict of %s is not one: %w", name, verr)
			}
			if refused, _ := meshcatalogclient.TakeVerdict(&v); refused {
				if v.Module == "" {
					v.Module = name
				}
				read = ManifestRead{Refused: meshcatalogclient.VerdictWhy(v)}
				decoded.Items = nil
				decoded = compose.Stub(decoded)
			}
		}
		if reads != nil {
			reads[name] = read
		}
		if err == nil {
			out[name] = decoded
		}
	}
	return out, nil
}

// caseVerdict reads a verdict a case carries. It is not the indexer's kept value but the checker's account of it, made
// from the indexer's modules value: a checker of the trunk writes it with version 0 and no judge, one of the registry
// milestone with no version and no judge, and the composer of a group's head is asked by the checker that runs. So it is
// read in the kept value's form at version 0 (or none) and at keys.Version, its judge optional; a later version, a value
// that does not decode and a state no reader knows are refused, never taken as accepted.
func caseVerdict(raw json.RawMessage) (meshcatalog.VerdictsByDigestValue, error) {
	var v meshcatalog.VerdictsByDigestValue
	if err := json.Unmarshal(raw, &v); err != nil {
		return v, fmt.Errorf("the value does not fit a verdict's shape: %w", err)
	}
	if v.V != 0 && v.V != keys.Version {
		return v, fmt.Errorf("%w: the value is version %d, this reader reads 0 and %d", keys.ErrUnknownVersion, v.V, keys.Version)
	}
	switch v.State {
	case meshcatalogclient.VerdictAccepted, meshcatalogclient.VerdictRefused, meshcatalogclient.VerdictManifestWaits:
	default:
		return v, fmt.Errorf("the state %q is none of %s, %s and %s", v.State, meshcatalogclient.VerdictAccepted,
			meshcatalogclient.VerdictRefused, meshcatalogclient.VerdictManifestWaits)
	}
	return v, nil
}

func runCase(r Request, c Case, depth string) (CaseVerdict, error) {
	cv := CaseVerdict{Name: c.Name, Manifests: map[string]ManifestRead{}}
	shelf, err := parse(c.Manifests, cv.Manifests, c.Verdicts)
	if err != nil {
		return CaseVerdict{}, err
	}
	base, err := inputOf(r)
	if err != nil {
		return CaseVerdict{}, err
	}
	held := base
	base, awaited := overrule(base, c, cv.Manifests, shelf)
	assignments := map[string]*meshmoduleassigner.Assignments{}
	for node, raw := range c.Assignments {
		if string(raw) == "null" {
			assignments[node] = nil
			continue
		}
		p, perr := meshmoduleassignerclient.DecodeAssignments(raw)
		if perr != nil {
			return CaseVerdict{}, fmt.Errorf("composer: the case %q: the assignment of %s is not one: %w", c.Name, node, perr)
		}
		assignments[node] = &p
	}
	in, err := frombus.Placed(base, assignments)
	if err != nil {
		return CaseVerdict{}, fmt.Errorf("composer: the case %q: %w", c.Name, err)
	}
	// The module the mesh provides itself (the private network) has no repository the indexer reads, so no case carries
	// it: its manifest comes with mesh-sdk and always wins, as the bus reader has it (frombus readBase). Without it a node
	// it is assigned to composed nothing (issue 1126, check of mesh-host #200: "nothing provides mesh-wireguard").
	for _, name := range overlay.ProvidedModules() {
		m, _, err := overlay.ProvidedManifest(name)
		if err != nil {
			return CaseVerdict{}, fmt.Errorf("composer: %w", err)
		}
		shelf[name] = m
	}
	// A module the input carries as its node-engine took it in from its build (SentInput) is composed so: the bus
	// reader composes it from its build, never from the indexer's manifest a case gives. One the case overrules the
	// node's hold on is composed at the build the node would compose it at (Input.Holds), else from the case's manifest.
	for name, m := range base.Catalogue {
		if _, over := awaited[name]; !over {
			shelf[name] = m
		}
	}
	for name, hold := range base.Holds {
		if _, over := awaited[name]; over && hold.Built != nil {
			shelf[name] = *hold.Built
		}
	}
	// A module the input's node-engine holds as refused (compose.Input.Refused: its verdict, or bytes that do not decode) is
	// held here as there, and its read says why, whatever the case's own verdicts: the check then names the cause the node
	// holds it for, never a pass the node would not compose, nor a rule about its stub (issue 774, group check 278aa5f0ccbe).
	for name, why := range base.Refused {
		if read, carried := cv.Manifests[name]; carried && read.Accepted {
			cv.Manifests[name] = ManifestRead{Refused: "its node-engine holds it: " + why}
		}
	}
	in.Catalogue = shelf
	in.Nodes = slices.Clone(in.Nodes)
	in.Holdings = slices.Clone(in.Holdings)
	for i := range in.Nodes {
		if len(c.Kept[in.Nodes[i].Name]) == 0 {
			continue // what the input keeps of this node, as its engine took it in, stands
		}
		in.Nodes[i].Kept = nil
		if kept := c.Kept[in.Nodes[i].Name]; len(kept) > 0 {
			k, err := parse(kept, nil, nil)
			if err != nil {
				return CaseVerdict{}, err
			}
			in.Nodes[i].Kept = k
		}
	}
	if r.HoldingsFromClaims {
		in.Holdings = append(in.Holdings, HoldingsOf(shelf, in.Nodes)...)
	}
	if len(r.Machines) == 0 {
		return cv, nil
	}
	cv.Machines = map[string]Machine{}
	for _, node := range r.Machines {
		m := machine(in, node, depth, r.Declarations)
		if words := awaitedOn(in, node, awaited); words != "" {
			// Composed without the build it runs at: named per node and module, never a silent pass (issue 1921). The
			// machine is then not judged, as for a node whose running state its engine does not send (ADR 0593 §9). Where
			// it does not compose without that build (a tool runner's bundle), it is composed with the module held as its
			// node-engine holds it, and named so; a node that refuses then too is the change's refusal.
			how := "composed from the change's manifest without its build: "
			if !m.Composes && m.NotJudged == "" {
				if again := machine(heldAgain(in, held, awaited), node, depth, r.Declarations); again.Composes {
					m, how = again, "composed with it held as its node-engine holds it, since it does not compose without its build: "
				}
			}
			words = how + words
			if m.Composes {
				m = Machine{NotJudged: words, Declared: m.Declared}
			} else if m.NotJudged == "" {
				m.Reason += "; " + words
			}
		}
		cv.Machines[node] = m
	}
	return cv, nil
}

// overrule lifts the hold of the input's node-engine on each module whose manifest the case carries and judges itself
// (novox/hq ADR 0578 §2): a module the input holds (Refused: the running indexer's verdict, or the running node-engine's
// decoder), whose bytes this composer reads whole, and that the case gives an accepted verdict or none (a reader takes a
// manifest with no verdict as accepted, ADR 0578 §3; the running checker drops the verdicts of the change's manifests,
// issue 1921). It is then no longer Refused nor held in how the node composes. A case's own refusal and bytes this
// composer does not decode stay held. awaited is each module lifted with no build the input carries: the words for the verdict, or
// "" where it is composed at its build.
func overrule(base compose.Input, c Case, reads map[string]ManifestRead, shelf map[string]catalogue.Manifest) (compose.Input, map[string]string) {
	awaited := map[string]string{}
	for name, why := range base.Refused {
		hold := base.Holds[name]
		if _, carried := c.Manifests[name]; !carried || !reads[name].Accepted || shelf[name].ReadNoFurther() {
			continue
		}
		held := "held by the running node-engine"
		switch hold.By {
		case compose.HeldByVerdict:
			held = "refused only by the running indexer"
		case compose.HeldByDecoder:
			held = "not read by the running node-engine's decoder"
		}
		judged := "accepted by the check's own validator"
		if _, given := c.Verdicts[name]; !given {
			judged = "given no verdict by the check, so taken as accepted (novox/hq ADR 0578 §3)"
		}
		awaited[name] = ""
		if hold.Built == nil && compose.NamesBuild(shelf[name]) {
			build := "its build"
			if hold.Build != "" {
				build += " " + hold.Build
			}
			awaited[name] = fmt.Sprintf("%s is %s (%s) and %s, and %s is awaited", name, held, why, judged, build)
		}
	}
	if len(awaited) == 0 {
		return base, awaited
	}
	base.Refused = maps.Clone(base.Refused)
	for name := range awaited {
		delete(base.Refused, name)
	}
	if how := base.ComposedHow; how != nil {
		copied := *how
		copied.Held = maps.Clone(how.Held)
		for name := range awaited {
			delete(copied.Held, name)
		}
		base.ComposedHow = &copied
	}
	return base, awaited
}

// heldAgain is in with each module overrule lifted and gave no build held again as held, the input before overrule,
// holds it: its stub, Refused and how the node composes.
func heldAgain(in, held compose.Input, awaited map[string]string) compose.Input {
	in.Catalogue = maps.Clone(in.Catalogue)
	in.Refused = maps.Clone(in.Refused)
	if in.Refused == nil {
		in.Refused = map[string]string{}
	}
	if how := in.ComposedHow; how != nil {
		copied := *how
		copied.Held = maps.Clone(how.Held)
		in.ComposedHow = &copied
	}
	for name, words := range awaited {
		if words == "" {
			continue
		}
		if stub, ok := held.Catalogue[name]; ok {
			in.Catalogue[name] = stub
		} else {
			in.Catalogue[name] = compose.Stub(in.Catalogue[name])
		}
		in.Refused[name] = held.Refused[name]
		if in.ComposedHow != nil && held.ComposedHow != nil {
			if why, was := held.ComposedHow.Held[name]; was {
				if in.ComposedHow.Held == nil {
					in.ComposedHow.Held = map[string]string{}
				}
				in.ComposedHow.Held[name] = why
			}
		}
	}
	return in
}

// awaitedOn is the words of each module placed on node that is composed without its build (overrule), joined, sorted.
func awaitedOn(in compose.Input, node string, awaited map[string]string) string {
	var out []string
	for _, n := range in.Nodes {
		if n.Name != node {
			continue
		}
		for _, m := range n.Assigned {
			if words := awaited[m]; words != "" {
				out = append(out, words)
			}
		}
	}
	sort.Strings(out)
	return strings.Join(out, "; ")
}

// HoldingsOf is every seat an assigned module claims, held on the node it is assigned to, in node order: the holdings
// of an input that has none of record (the assigner's, mesh-delivery-assigner internal/assign input).
func HoldingsOf(shelf map[string]catalogue.Manifest, nodes []compose.Node) []catalogue.Held {
	var out []catalogue.Held
	sorted := slices.Clone(nodes)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name })
	for _, n := range sorted {
		for _, mod := range n.Assigned {
			for _, cl := range shelf[mod].Claims {
				scope := cl.Scope
				if seat, ok := catalogue.SeatNamed(cl.Name); ok && scope == "" {
					scope = seat.Scope
				}
				out = append(out, catalogue.Held{Claim: cl.Name, Scope: scope, Node: n.Name, Module: mod})
			}
		}
	}
	return out
}

func machine(in compose.Input, node, depth string, declarations bool) Machine {
	if depth == DepthResolve {
		if _, err := compose.Resolve(in, node); err != nil {
			return Machine{Reason: err.Error()}
		}
		return Machine{Composes: true}
	}
	// Composed as the node-engine composes it, when its input says how (compose.Input.ComposedHow, issue 774): the same
	// path, at the same builds, a module folded in its own composition kept as it runs.
	if !declarations {
		if _, how, err := frombus.ComposeSent(in, node); how {
			if err != nil {
				return Machine{Reason: err.Error()}
			}
			return Machine{Composes: true}
		}
	}
	d, problems, err := compose.Compose(in, node)
	if err != nil {
		// An input from a node-engine before ComposedHow says nothing of what runs of the modules folded in the node's
		// own composition: the sandbox cannot compose them as the node does, so it does not judge the node, and names
		// it (ADR 0593 §9), never a failure of the change and never a silent pass.
		if folded := foldedUnsaid(in, node); len(folded) > 0 && in.ComposedHow == nil {
			return Machine{NotJudged: NotJudged + strings.Join(folded, ", ") +
				" (its own composition carries them as they run, and the sandbox cannot): " + err.Error()}
		}
		return Machine{Reason: err.Error()}
	}
	m := Machine{Composes: len(problems) == 0, Problems: problems, Refusals: d.Refusals()}
	if len(problems) > 0 {
		m.Reason = "the node-engine's validator refuses the declaration"
	}
	if declarations {
		m.Declared = pathsOf(d)
	}
	return m
}

// NotJudged opens the words of a machine the sandbox does not judge (Machine.NotJudged, the contract's field, ADR 0593 §9):
// its node-engine sent an input that does not say how it composes (compose.Input.ComposedHow), and the node runs modules
// folded in its own composition. The machine is then neither composed nor refused; a reader decides by the field.
const NotJudged = "not judged: its node-engine sends no running state for "

// foldedUnsaid is each module node runs folded in its own composition (Node.InOwn) whose manifest names a build: what
// the node composes as it runs, and an input without ComposedHow cannot.
func foldedUnsaid(in compose.Input, node string) []string {
	var out []string
	for _, n := range in.Nodes {
		if n.Name != node {
			continue
		}
		for m, ok := range n.InOwn {
			man, known := n.Kept[m]
			if !known {
				man, known = in.Catalogue[m]
			}
			if ok && known && compose.NamesBuild(man) {
				out = append(out, m)
			}
		}
	}
	sort.Strings(out)
	return out
}

// pathsOf is every resource of a declaration that names a path, with its owner, group and mode as declared, sorted by
// path.
func pathsOf(d compose.Declared) []Path {
	var out []Path
	str := func(r map[string]any, key string) string {
		if v, ok := r[key]; ok && v != nil {
			return fmt.Sprint(v)
		}
		return ""
	}
	for _, r := range d.Resources {
		p := str(r, "path")
		if p == "" {
			continue
		}
		out = append(out, Path{Path: p, Kind: str(r, "type"), Owner: str(r, "owner"), Group: str(r, "group"),
			Mode: str(r, "mode"), Module: d.ModuleOf[str(r, "id")]})
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Path != out[j].Path {
			return out[i].Path < out[j].Path
		}
		return out[i].Kind < out[j].Kind
	})
	return out
}

// Package is the import path of the command, which a build module builds at the node-engine's repository's head of a
// group that changes the node-engine's composition (novox/hq ADR 0593 §7): mesh-sdk's composer.Package names the same.
const Package = "github.com/novox/mesh-host/cmd/mesh-compose"

// Decode reads a request (sdk.Decode); it refuses one that is not JSON or of another version.
func Decode(data []byte) (Request, error) { return sdk.Decode(data) }

// SentInput is the composition input a node-engine answers for the sandbox (nodeengine.ComposeCandidateInput.Input), from the
// input it composed node from: the one shape both read (issue 1126, checks of mesh-host #200). Every other node's modules
// are left to the cases (the indexer's manifests, as the bus reader reads them too); node's own placed modules stay as the
// bus reader took them in from their builds (their artifacts, the texts of their templates archive, what is handed to the
// provisioner), with what node keeps at another build, which no case can give: the sandbox has no build or archive to
// read. Without them the sandbox composed another node than the engine (restic's templates and artifacts unread, postgres's
// secret reader handed and the module not).
func SentInput(in compose.Input, node string) compose.Input {
	own := map[string]bool{}
	for _, n := range in.Nodes {
		if n.Name == node {
			for _, m := range n.Assigned {
				own[m] = true
			}
		}
	}
	taken := in.Catalogue
	in.Catalogue = nil
	for name, m := range taken {
		if own[name] {
			if in.Catalogue == nil {
				in.Catalogue = map[string]catalogue.Manifest{}
			}
			in.Catalogue[name] = m
		}
	}
	in.Nodes = slices.Clone(in.Nodes)
	for i := range in.Nodes {
		if in.Nodes[i].Name != node {
			in.Nodes[i].Kept = nil
		}
	}
	return in
}