mesh-delivery/modules/mesh-delivery-indexer/internal/index/toolchains.go

package index

import (
	"context"
	"fmt"
	"maps"
	"reflect"
	"slices"
	"strings"

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

	"mesh-delivery-indexer/judge"
)

// The new-form fingerprint of a module is written, and written again with no merge, when what it is of changes
// (novox/hq ADR 0465 clause 2; issue 1444): a module's tree and lock hashes come with a merge and are kept in
// modules.<m>.Inputs2; its bases' and its toolchain's come from here. The indexer holds no graph, so each recompute reads
// the manifest of every module from manifests.<digest> and takes the modules in the order their bases come first.
//
// **No module's fingerprint takes the builder as an input** (novox/hq ADR 0530 §1, issue 1887 I1): the builder orders and
// records builds and makes no artifact's bytes (ADR 0498 §1), so its Builder input is empty for every module, the builder's
// own included; a change of how an artifact is built is a change of its build module's recipe (ADR 0530 §2).

// ToolchainModules are the build module of each language, whose state toolchains (delivery.ToolchainsBucket) the
// indexer reads: the language's agreed toolchain is the one every node's entry in it agrees on.
var ToolchainModules = map[string]string{"go": "mesh-build-go", "typescript": "mesh-build-node"}

// ToolchainStreams are the stream each build module carries its toolchain.reported event on, by module, as the bus names it
// (a stream's local name in capitals, catalogue.StreamBusName): `toolchains-go` and `toolchains-node`, declared by the
// build modules (streams, with the event toolchain.reported) and heard by this module (consumes).
var ToolchainStreams = map[string]string{"mesh-build-go": "TOOLCHAINS_GO", "mesh-build-node": "TOOLCHAINS_NODE"}

// PutToolchain takes one change of a build module's toolchains bucket in, without writing any module: a put of a node's
// toolchain of a language (key <node>.<language>, delivery.ToolchainKey), or its delete. It answers whether what the
// language's toolchain is of (version and recipe) changed. A key, or a value, that is not a toolchain of the module's
// language is said in the log and left out.
func (ix *Indexer) PutToolchain(module, key string, value []byte, deleted bool) bool {
	language := ""
	for l, m := range ToolchainModules {
		if m == module {
			language = l
		}
	}
	if language == "" {
		ix.logf("toolchains of %q are not read: it is no build module the indexer knows", module)
		return false
	}
	node, keyLanguage, err := delivery.ParseToolchainKey(key)
	if err != nil || keyLanguage != language {
		ix.logf("%s_toolchains.%s is not a %s toolchain's key, left out: %v", module, key, language, err)
		return false
	}
	ix.mu.Lock()
	defer ix.mu.Unlock()
	if deleted {
		if _, had := ix.toolchains[language][node]; !had {
			return false
		}
		delete(ix.toolchains[language], node)
		return true
	}
	tc, err := delivery.DecodeToolchain(value)
	if err != nil || tc.Language != language {
		ix.logf("%s_toolchains.%s is not a %s toolchain, left out: %v", module, key, language, err)
		return false
	}
	if ix.toolchains == nil {
		ix.toolchains = map[string]map[string]delivery.Toolchain{}
	}
	if ix.toolchains[language] == nil {
		ix.toolchains[language] = map[string]delivery.Toolchain{}
	}
	cur, had := ix.toolchains[language][node]
	ix.toolchains[language][node] = tc
	return !had || cur.Version != tc.Version || cur.Recipe != tc.Recipe
}

// ApplyToolchain takes one change of a build module's toolchains in (PutToolchain) and, when the language's toolchain
// changed, writes the modules again that it moves (Recompute). The same version and recipe seen again writes nothing.
func (ix *Indexer) ApplyToolchain(ctx context.Context, module, key string, value []byte, deleted bool) error {
	if !ix.PutToolchain(module, key, value, deleted) {
		return nil
	}
	return ix.Recompute(ctx)
}

// agreed is the toolchain of each language every reporting node agrees on, or why there is none.
func (ix *Indexer) agreed() map[string]toolchainOrNone {
	ix.mu.Lock()
	defer ix.mu.Unlock()
	out := map[string]toolchainOrNone{}
	for language := range ToolchainModules {
		var entries []delivery.NodeToolchain
		for _, node := range slices.Sorted(maps.Keys(ix.toolchains[language])) {
			entries = append(entries, delivery.NodeToolchain{Node: node, Toolchain: ix.toolchains[language][node]})
		}
		tc, none := delivery.AgreedToolchain(entries)
		out[language] = toolchainOrNone{tc, none}
	}
	return out
}

type toolchainOrNone struct {
	tc   delivery.Toolchain2
	none string
}

// known is a module the indexer holds the inputs of: its record as read, the manifest's shape and the key it is written at.
type known struct {
	key       string
	cur       delivery.Module
	manifest  catalogue.Manifest
	languages map[string]string
	bases     []string
	contexts  bool
	// held is the fingerprint the bucket holds for the module and the toolchains it was computed with (for a merge, the
	// value before it): what stillComputed keeps while the builders agree on no toolchain.
	held delivery.Module
}

// Recompute writes again every module whose new-form fingerprint, or reason for none, is no longer what it holds: from
// the inputs kept at its merge (Inputs2), its manifest (manifests.<digest>), the fingerprints of its bases and of the
// builder, and the toolchains the build modules report. A module is computed after its bases and the builder, whatever
// the order the bucket lists them in; modules that stand on each other have a cycle as their reason. A module is written
// only when its fingerprint or its reason changes, and only if its manifest is still the one computed from. A module with
// no kept inputs (a message that carried no manifest) is left as it is.
func (ix *Indexer) Recompute(ctx context.Context) error { return ix.recomputeWith(ctx, nil, nil) }

// recomputeWith is Recompute for a merge as well: the records the merge gives (overlay, by module key) stand in place
// of what the bucket holds, are computed with the others, and are written once, with their final fingerprint or reason
// (issue 1444: a merge writes each module once). A record of the overlay that has no kept inputs is written as it is. w is
// the merge's write (order.go), nil for none; it is told each module written.
func (ix *Indexer) recomputeWith(ctx context.Context, overlay map[string]delivery.Module, w *mergeWrite) error {
	if err := ix.fillPaths(ctx); err != nil {
		return err
	}
	if err := ix.recomputeModules(ctx, overlay, w); err != nil {
		return err
	}
	// What the modules moved to reaches the candidates' fingerprints too (a base, the builder, a toolchain).
	return ix.RecomputeCandidates(ctx)
}

func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]delivery.Module, w *mergeWrite) error {
	if ix.Modules == nil {
		return nil
	}
	if ix.Manifests == nil {
		return ix.writeOverlay(ctx, overlay, nil, w)
	}
	ix.recompute.Lock()
	defer ix.recompute.Unlock()
	return ix.recomputeLocked(ctx, overlay, w)
}

// recomputeLocked is recomputeModules with ix.recompute held by the caller (a hold that moves, holds.go).
func (ix *Indexer) recomputeLocked(ctx context.Context, overlay map[string]delivery.Module, w *mergeWrite) error {
	if ix.Manifests == nil {
		return ix.writeOverlay(ctx, overlay, nil, w)
	}
	keys, err := ix.Modules.Keys(ctx)
	if err != nil {
		return fmt.Errorf("listing the modules: %w", err)
	}
	for key := range overlay {
		if !slices.Contains(keys, key) {
			keys = append(keys, key)
		}
	}
	slices.Sort(keys)
	held := map[string]delivery.Module{}
	stale := map[string]bool{} // values an SDK that could not read their manifest wrote, judged again (issue 1923)
	all := map[string]*known{}
	for _, key := range keys {
		cur, merged := overlay[key]
		var before delivery.Module
		value, _, found, err := ix.Modules.Get(ctx, key)
		if err != nil {
			return fmt.Errorf("modules.%s: %w", key, err)
		}
		if found {
			if before, err = delivery.DecodeModule(value); err != nil && !merged {
				ix.logf("modules.%s holds what is not a module, left: %v", key, err)
				continue
			}
		}
		if !merged {
			if !found {
				continue
			}
			cur = before
		}
		held[key] = cur
		if cur.Manifest == "" || cur.Inputs2 == nil {
			continue
		}
		raw, m, err := ix.manifestRaw(ctx, cur.Manifest)
		if err != nil {
			ix.logf("modules.%s: its manifest %s cannot be read, left: %v", key, cur.Manifest, err)
			continue
		}
		if again, ok := ix.rejudged(ctx, m, cur, raw); ok {
			cur, stale[key] = again, true
			held[key] = cur
		}
		all[key] = &known{key: key, cur: cur, manifest: m, languages: languagesOf(m), bases: basesOf(m), contexts: hasContext(m), held: before}
	}
	tools := ix.agreed()
	published := ix.publishedAll(ctx)
	series := ix.SeriesAll(ctx) // each package's series, the newest a user accepting it takes (DLV-I6)
	type result struct {
		fp, none string
		tcs      map[string]delivery.ToolchainUsed // by language, when there is a fingerprint
		pkgs     delivery.PackageVersions          // the package versions, when there is a fingerprint
	}
	done := map[string]result{}
	visiting := map[string]bool{}
	var of func(module string, chain []string) result
	of = func(module string, chain []string) result {
		if r, ok := done[module]; ok {
			return r
		}
		k, ours := all[module]
		if !ours { // a module the indexer holds no inputs of: its record, as written
			cur, has := held[module]
			switch {
			case !has:
				return result{fp: "", none: "the indexer holds no record of it"}
			case cur.FP2 != "":
				return result{fp: cur.FP2, none: ""}
			case cur.None2 != "":
				return result{fp: "", none: cur.None2}
			}
			return result{fp: "", none: "its record carries no fingerprint of the new form"}
		}
		if visiting[module] {
			return result{fp: "", none: "a cycle of bases: " + strings.Join(append(chain, module), " -> ")}
		}
		visiting[module] = true
		defer delete(visiting, module)
		chain = append(chain, module)
		kept := *k.cur.Inputs2
		kept.Bases, kept.Builder = nil, ""
		var missing []string
		for _, b := range k.bases {
			if r := of(b, chain); r.fp != "" {
				kept.Bases = append(kept.Bases, r.fp)
			} else {
				missing = append(missing, fmt.Sprintf("the base %s has no fingerprint (%s)", b, r.none))
			}
		}
		slices.Sort(kept.Bases)
		slices.Sort(missing)
		toolchains := map[string]delivery.Toolchain2{}
		var noToolchain string
		for _, artifact := range slices.Sorted(maps.Keys(k.languages)) {
			t := tools[k.languages[artifact]]
			if t.none != "" {
				if noToolchain == "" {
					noToolchain = t.none
				}
				continue
			}
			toolchains[artifact] = t.tc
		}
		var r result
		pub, pins := acceptedSeries(series, k.manifest, k.cur.Repository, published)
		switch {
		case k.contexts:
			r.none = None2Context
		case len(missing) > 0:
			r.none = strings.Join(missing, "; ")
		case noToolchain != "":
			r.none = noToolchain
			if fp, pkgs, ok := stillComputed(module, kept, k, toolchains, held, pub, pins); ok {
				r = result{fp: fp, tcs: k.held.Toolchains, pkgs: pkgs}
			}
		default:
			f := formOf(module, k.cur.Repository, &kept, toolchains, k.held, held, pub, pins, ix.logf)
			r.fp, r.none = f.fp, f.none
			if r.fp != "" {
				r.tcs, r.pkgs = toolchainsUsed(toolchains), f.versions
			}
		}
		done[module] = r
		k.cur.Inputs2 = &kept
		return r
	}
	finals := map[string]delivery.Module{}
	edges := map[string]graphEdges{} // the module graph's nodes of this recompute (graph.go)
	for _, key := range slices.Sorted(maps.Keys(all)) {
		k := all[key]
		r := of(key, nil)
		edges[key] = graphEdges{module: key, cur: k.cur, manifest: k.manifest, fp: r.fp, none: r.none, tcs: r.tcs, pkgs: r.pkgs}
		if _, merged := overlay[key]; merged {
			final := overlay[key]
			final.FP2, final.None2, final.Inputs2, final.Toolchains, final.Packages = r.fp, r.none, k.cur.Inputs2, r.tcs, r.pkgs
			// What the bases gave is kept as it was while the value it gave does not change, so a merge
			// that moves nothing for the module writes nothing (as Recompute does).
			if value, _, found, err := ix.Modules.Get(ctx, key); err == nil && found {
				if old, err := delivery.DecodeModule(value); err == nil && old.Inputs2 != nil && old.FP2 == r.fp && old.None2 == r.none {
					in := *final.Inputs2
					in.Bases, in.Builder = old.Inputs2.Bases, old.Inputs2.Builder
					final.Inputs2 = &in
				}
			}
			finals[key] = final
			continue
		}
		if !stale[key] && k.cur.FP2 == r.fp && k.cur.None2 == r.none && reflect.DeepEqual(k.cur.Toolchains, r.tcs) && reflect.DeepEqual(k.cur.Packages, r.pkgs) {
			continue
		}
		digest, inputs, paths := k.cur.Manifest, k.cur.Inputs2, k.cur.ArtifactPaths
		wrote := false
		err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
			if !found {
				return nil, false, nil
			}
			cur, err := delivery.DecodeModule(value)
			if err != nil || cur.Manifest != digest || cur.Inputs2 == nil {
				return nil, false, nil // a merge changed it meanwhile; its own recompute follows
			}
			if cur.FP2 == r.fp && cur.None2 == r.none && reflect.DeepEqual(cur.Toolchains, r.tcs) && reflect.DeepEqual(cur.Packages, r.pkgs) {
				return nil, false, nil
			}
			if cur.FP2 != r.fp {
				// Moved with no merge (a toolchain, a package, a base): no release carries this build, and the source revision
				// moves only at a merge, so it is not known (0) and the move is asked of a person; never the revision of now,
				// which may be the very release that last merged the module (ADR 0601 §1). The source stamp leaves it, since
				// Src is set.
				cur.FPSrc = 0
			}
			cur.FP2, cur.None2, cur.Toolchains, cur.Packages = r.fp, r.none, r.tcs, r.pkgs
			cur.Inputs2, cur.ArtifactPaths = inputs, paths
			data, err := delivery.EncodeModule(cur)
			wrote = err == nil
			return data, false, err
		})
		if err != nil {
			return fmt.Errorf("modules.%s: %w", key, err)
		}
		if wrote {
			w.wrote(key)
		}
	}
	if err := ix.writeOverlay(ctx, overlay, finals, w); err != nil {
		return err
	}
	return ix.writeGraph(ctx, edges)
}

// writeOverlay writes the records of a merge, each once and only if it differs from what the bucket holds; finals has
// the ones that were computed, the others are written as the merge gave them.
//
// **A module the merge did not change is not written** (the operator, 2026-10-25: "only if a module has got a new version in
// commit x, should a new module version get created linked to commit x"). A value names the commit its fingerprint first
// appeared at: when the merge gives a module the value it holds but for the commit (and the source revision stamped on it),
// the held value stands, its commit unchanged, so a catalogue merge touching one module writes one key, not one per module of
// the repository, and the controller's take-in hears only that one. The repository's trunk head is the gate's source.main.
//
// A merged message older than the repository's taken merge never reaches here (order.go): it is asked before any write.
func (ix *Indexer) writeOverlay(ctx context.Context, overlay, finals map[string]delivery.Module, w *mergeWrite) error {
	for _, key := range slices.Sorted(maps.Keys(overlay)) {
		next := overlay[key]
		if f, ok := finals[key]; ok {
			next = f
		}
		wrote := false
		err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
			wrote = false
			next := next
			cur, err := delivery.DecodeModule(value)
			held := found && err == nil
			if held && cur.Manifest == next.Manifest {
				// The judge's verdict stands while the digest does (ADR 0578 §3, ADR 0579 §1): a recompute never writes a
				// refused or outgrown module as accepted; only the judge changes it. So do the reader capabilities it named.
				next.Verdict, next.Problems, next.Unknown, next.Warnings = cur.Verdict, cur.Problems, cur.Unknown, cur.Warnings
				next.Needs = cur.Needs
			}
			// The source revision at which FP2 last changed (ADR 0601 §1): this merge's when it moved it (0 until the
			// source revision naming the merge is heard, which stamps it), else the one held.
			if held && cur.FP2 == next.FP2 {
				next.FPSrc = cur.FPSrc
			} else {
				next.FPSrc = next.Src
			}
			if held && sameButWhere(cur, next) {
				return nil, false, nil
			}
			if next.Manifest != "" && (!held || cur.Manifest != next.Manifest) {
				w.taken(next.Manifest) // a new module, or a changed manifest: fresh, judged in full (ADR 0579 §1)
			}
			data, err := delivery.EncodeModule(next)
			wrote = err == nil
			return data, false, err
		})
		if err != nil {
			return fmt.Errorf("modules.%s: %w", key, err)
		}
		if wrote {
			w.wrote(key)
		}
	}
	return nil
}

// sameButWhere says the held value cur is next but for where it was computed (Commit and Src) and the judge's verdict,
// which only the judge writes: the module is unchanged.
func sameButWhere(cur, next delivery.Module) bool {
	next = withV(next)
	next.Commit, next.Src, next.FPSrc = cur.Commit, cur.Src, cur.FPSrc
	next.Verdict, next.Problems, next.Unknown, next.Warnings = cur.Verdict, cur.Problems, cur.Unknown, cur.Warnings
	next.Needs = cur.Needs
	return reflect.DeepEqual(cur, next)
}

// manifestRaw is manifests.<digest>: the manifest as stored and parsed by the running SDK.
func (ix *Indexer) manifestRaw(ctx context.Context, digest string) (string, catalogue.Manifest, error) {
	key, err := delivery.ManifestsKey(digest)
	if err != nil {
		return "", catalogue.Manifest{}, err
	}
	value, _, found, err := ix.Manifests.Get(ctx, key)
	if err != nil {
		return "", catalogue.Manifest{}, err
	}
	if !found {
		return "", catalogue.Manifest{}, fmt.Errorf("manifests.%s is not there", key)
	}
	kept, err := delivery.DecodeManifest(value)
	if err != nil {
		return "", catalogue.Manifest{}, err
	}
	m, err := judge.ForInputs([]byte(kept.JSON))
	return kept.JSON, m, err
}

// rejudged is a value an SDK that could not read its manifest wrote, as the running SDK judges it: its kept inputs (the
// tree, the lock hashes by file name, the named paths' objects) taken through the manifest as a merge takes them, and the
// reason that gives. It answers false for any other value, and for one whose manifest the running SDK still cannot make
// inputs of (issue 1923).
func (ix *Indexer) rejudged(ctx context.Context, m catalogue.Manifest, cur delivery.Module, raw string) (delivery.Module, bool) {
	if cur.FP2 != "" || cur.Inputs2 == nil || !strings.HasPrefix(cur.None2, None2Unreadable) {
		return cur, false
	}
	in := inputsOfNamed(m.Module, raw, cur.Inputs2.Tree, cur.Inputs2.Locks, cur.Inputs2.Paths, cur.Inputs2.Packages, ix.baseFP(ctx))
	if in.Kept == nil {
		return cur, false
	}
	in.Kept.Imports, in.Kept.ImportForm = cur.Inputs2.Imports, cur.Inputs2.ImportForm
	cur.Inputs2, cur.None2, cur.ArtifactPaths = in.Kept, in.None2, in.ArtifactPaths
	return cur, true
}

// hasContext is whether an artifact of the manifest reads another repository's tree.
func hasContext(m catalogue.Manifest) bool {
	if m.Build == nil {
		return false
	}
	for _, a := range m.Build.Artifacts {
		if a.Context != nil {
			return true
		}
	}
	return false
}

// stillComputed is the fingerprint a module already holds, kept while a language it is built with has no agreed toolchain
// (issue 1940, 2026-10-25 20:56: a build node assigned again brought a stale report into the agreement, every module's
// fingerprint was blanked to "builders disagree", and the controller's take-in refused modules whose fingerprints were
// computed and built). Only computing a new fingerprint needs the agreement: the held one is kept when its kept inputs,
// with the toolchains it was computed with in place of the languages agreed on none, give it again, in its form. Any
// other input that moved (a merge, a base, an agreed toolchain, an imported package) gives another fingerprint, and the
// module has none until they agree.
func stillComputed(module string, kept delivery.KeptInputs2, k *known, agreed map[string]delivery.Toolchain2, held map[string]delivery.Module, published map[string]delivery.PublishedPackages, pins delivery.PackageVersions) (string, delivery.PackageVersions, bool) {
	if k.held.FP2 == "" {
		return "", nil, false
	}
	toolchains := map[string]delivery.Toolchain2{}
	for artifact, language := range k.languages {
		if tc, ok := agreed[artifact]; ok {
			toolchains[artifact] = tc
			continue
		}
		used, ok := k.held.Toolchains[language]
		if !ok {
			return "", nil, false
		}
		toolchains[artifact] = delivery.Toolchain2{Language: language, Version: used.Version, Recipe: used.Recipe}
	}
	f := formOf(module, k.held.Repository, &kept, toolchains, k.held, held, published, pins, nil)
	return f.fp, f.versions, f.none == "" && f.fp == k.held.FP2
}

// toolchainsUsed are the toolchains a fingerprint was computed with, by language (an artifact's toolchain is its language's).
func toolchainsUsed(by map[string]delivery.Toolchain2) map[string]delivery.ToolchainUsed {
	if len(by) == 0 {
		return nil
	}
	out := map[string]delivery.ToolchainUsed{}
	for _, t := range by {
		out[t.Language] = delivery.ToolchainUsed{Version: t.Version, Recipe: t.Recipe}
	}
	return out
}

// RecomputeAtStart is Recompute once the indexer has started. While the indexer has read no toolchain at all (it has just started and the toolchains buckets are not settled), every
// Go module would be written as having no toolchain reported and then back: it computes nothing then, and the settling of
// the toolchains computes them.
func (ix *Indexer) RecomputeAtStart(ctx context.Context) error {
	ix.mu.Lock()
	none := len(ix.toolchains) == 0
	ix.mu.Unlock()
	if none {
		ix.logf("no toolchain is read yet: the new-form fingerprints are computed when one is")
		return nil
	}
	return ix.Recompute(ctx)
}

// fillPaths gives every module whose key names no directory (a key written before the indexer learned paths) the one its
// newest build record names, once, by compare-and-set, changing nothing else (issue 1614): the builder needs it, and a
// rarely merged repository would leave it without one for weeks. A key that names a directory is never written here.
// It reads the controller's build records for nothing else: since R4g no fingerprint takes a build record (ADR 0530 §1).
func (ix *Indexer) fillPaths(ctx context.Context) error {
	if ix.Modules == nil || ix.Builds == nil {
		return nil
	}
	held, err := ix.heldModules(ctx)
	if err != nil {
		return err
	}
	for _, key := range slices.Sorted(maps.Keys(held)) {
		if held[key].Path != "" {
			continue
		}
		dir, ok := ix.Builds.PathOf(key, held[key].Repository)
		if !ok {
			continue
		}
		err := write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
			if !found {
				return nil, false, nil
			}
			cur, err := delivery.DecodeModule(value)
			if err != nil || cur.Path != "" {
				return nil, false, nil
			}
			cur.Path = dir
			data, err := delivery.EncodeModule(cur)
			return data, false, err
		})
		if err != nil {
			return fmt.Errorf("modules.%s: %w", key, err)
		}
		ix.logf("modules.%s names no directory: %q filled from its newest build record (issue 1614)", key, dir)
	}
	return nil
}