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

package index

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

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

// A module registered in the catalogue from its own repository (the operator registers it with the repository and the
// path of its manifest) had no manifest here: the indexer learns a repository only from a merged message, and no merge
// followed the registration. The controller's outcome records name each registered module's repository, directory and
// commit (builds.go Registered), so the indexer reads that module's manifest at the repository's head (the gate's source names it) or, with none named, at the
// registered commit, and writes the same keys a merge writes. A later merge to the repository's trunk reaches the module through the forge bridge's merged
// message as for every other repository.

// RepoRead is a repository at a commit as the forge bridge reads it for a merged message: the root tree, the tree of each
// directory holding a module.json (the root spelled ""), their manifests and the hex sha256 of the lock files beside them.
type RepoRead struct {
	Root      string
	Trees     map[string]string
	Manifests map[string]string
	Locks     map[string]map[string]string
}

// IndexRegistered writes the module of a registration as a merge of its repository would, when the modules bucket holds
// none of that name. Its commit is the head the gate's source names for the repository, else the registration's. The
// repository's other modules are neither written nor dropped: they are the merges'. A manifest that is not at the path or
// names another module is said and left. A forge that could not say (a timeout, or a 404 while the forge's account could
// not read the repository) is an error, and the module is kept as unread: it is read again at the next source revision
// and the next merged message (RetryUnread), not only when the indexer starts again (issue 1945).
func (ix *Indexer) IndexRegistered(ctx context.Context, r Registration) error {
	_, err := ix.indexRegistered(ctx, r, false)
	return err
}

// taken is what indexRegistered did with one registration: the module written, held already, or the reason it was not.
type taken struct {
	stored, held            bool
	repository, commit, dir string
	why                     string
}

func (ix *Indexer) indexRegistered(ctx context.Context, r Registration, force bool) (taken, error) {
	if ix.Modules == nil || ix.Read == nil {
		return taken{why: "the indexer reads no forge or writes no modules"}, nil
	}
	key, err := delivery.ModulesKey(r.Module)
	if err != nil {
		ix.logf("%s: the registered module has no usable key, skipped: %v", r.Module, err)
		return taken{why: fmt.Sprintf("the module has no usable key: %v", err)}, nil
	}
	if _, _, found, err := ix.Modules.Get(ctx, key); err != nil {
		return taken{}, fmt.Errorf("modules.%s: %w", key, err)
	} else if found {
		ix.markRead(r.Module)
		return taken{held: true, why: "the modules state holds it already: a merge to its repository writes it from now on"}, nil
	}
	repository := delivery.RepositoryName(r.Repository)
	commit := r.Commit
	dir := strings.Trim(r.Path, "/")
	if dir == delivery.RootPath {
		dir = ""
	}
	// The head the gate's source names for the repository, when it names one, else the registration's commit. A module a
	// merge removed is absent at the head and is said and left.
	ix.mu.Lock()
	for repo, head := range ix.source.heads {
		if delivery.SameRepository(repo, r.Repository) && head != "" {
			commit = head
		}
	}
	ix.mu.Unlock()
	t := taken{repository: repository, commit: commit, dir: dir}
	if repository == "" || commit == "" {
		ix.logf("%s: registered with no repository or commit, so its manifest is not read", r.Module)
		t.why = "the controller's record names no repository or commit"
		return t, nil
	}
	// One try for each repository, commit and directory: a repository with no manifest where it was registered is said once,
	// not at every record of the controller's builds. A person's take-in (force) reads again.
	tried := repository + "@" + commit + ":" + dir
	ix.mu.Lock()
	if ix.tried[tried] && !force {
		ix.mu.Unlock()
		t.why = "tried already at this commit and directory"
		return t, nil
	}
	if ix.tried == nil {
		ix.tried = map[string]bool{}
	}
	ix.tried[tried] = true
	ix.mu.Unlock()
	read, err := ix.Read(ctx, repository, commit, dir)
	if err != nil {
		ix.mu.Lock()
		delete(ix.tried, tried) // a forge that could not say is asked again
		if ix.unread == nil {
			ix.unread = map[string]bool{}
		}
		ix.unread[r.Module] = true // and read again on the next source revision or merged message (RetryUnread)
		ix.mu.Unlock()
		return t, fmt.Errorf("%s: reading %s at %s: %w", r.Module, repository, commit, err)
	}
	ix.markRead(r.Module)
	raw, ok := read.Manifests[dir]
	if !ok || read.Trees[dir] == "" {
		ix.logf("%s: %s at %s holds no module.json at %q, so the registered module is not indexed", r.Module, repository, commit, dir)
		t.why = fmt.Sprintf("%s at %s holds no module.json at %q", repository, commit, dir)
		return t, nil
	}
	if name := manifestModule(raw); name != r.Module {
		ix.logf("%s: the manifest at %q of %s at %s names the module %q, so the registered module is not indexed", r.Module, dir, repository, commit, name)
		t.why = fmt.Sprintf("the manifest at %q of %s at %s names the module %q", dir, repository, commit, name)
		return t, nil
	}
	f := delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: repository, Merge: commit, Root: read.Root,
		Trees:     map[string]string{dir: read.Trees[dir]},
		Manifests: map[string]string{dir: raw}}
	if l := read.Locks[dir]; len(l) > 0 {
		f.Locks = map[string]map[string]string{dir: l}
	}
	if err := ix.applyManifests(ctx, f, false); err != nil {
		return t, err
	}
	if _, _, found, err := ix.Modules.Get(ctx, key); err != nil {
		return t, fmt.Errorf("modules.%s: %w", key, err)
	} else if !found {
		t.why = "its manifest was read but the modules state holds no value of it (the indexer's log says why)"
		return t, nil
	}
	t.stored = true
	ix.logf("%s: the registered module is indexed from %s at %s", r.Module, repository, commit)
	return t, nil
}

// markRead takes a module out of the unread ones: the forge answered for it.
func (ix *Indexer) markRead(module string) {
	ix.mu.Lock()
	delete(ix.unread, module)
	ix.mu.Unlock()
}

// RetryUnread reads again every registered module whose last read the forge could not answer (a timeout, or a 404 while
// the forge's account could not read the repository; issue 1945). It is called on the events that already reach the
// indexer, a new source revision and a merged message, so a repository that became readable is taken in without a hand
// act and without a timer of its own (ADR 0437 §8). What still cannot be read is said and kept for the next.
func (ix *Indexer) RetryUnread(ctx context.Context) {
	ix.mu.Lock()
	names := slices.Sorted(maps.Keys(ix.unread))
	ix.mu.Unlock()
	for _, module := range names {
		if err := ix.IndexRegisteredModule(ctx, module); err != nil {
			ix.logf("the registered module %s is still not indexed: %v", module, err)
		}
	}
}

// kickRetry runs RetryUnread in one background goroutine, so the message or revision that asked it is acknowledged without
// waiting on the forge (each unread module costs up to two forge calls of 30 s each). A kick while one runs makes it run
// once more when it ends; never two at once.
func (ix *Indexer) kickRetry(ctx context.Context) {
	ix.mu.Lock()
	none := len(ix.unread) == 0
	ix.mu.Unlock()
	if none {
		return
	}
	ix.retryAgain.Store(true)
	if !ix.retrying.CompareAndSwap(false, true) {
		return
	}
	ix.retryWG.Add(1)
	go func() {
		defer ix.retryWG.Done()
		for ix.retryAgain.Swap(false) {
			ix.RetryUnread(ctx)
		}
		ix.retrying.Store(false)
		// A kick between the last Swap and the Store found it running and returned: run for it.
		if ix.retryAgain.Load() {
			ix.kickRetry(ctx)
		}
	}()
}

// TakenIn is the answer of a take-in (the tool indexer_take_in): whether the module is now in the modules state, from
// where, and otherwise why not.
type TakenIn struct {
	Module     string `json:"module"`
	Stored     bool   `json:"stored"`
	Held       bool   `json:"held,omitempty"`
	Repository string `json:"repository,omitempty"`
	Commit     string `json:"commit,omitempty"`
	Path       string `json:"path,omitempty"`
	Answer     string `json:"answer"`
}

// TakeIn reads a registered module's manifest now, whatever was tried before, and writes it as a merge would (the tool
// indexer_take_in, issue 1945). A module the modules state holds is answered so and left; one with no registration in the
// controller's records, or whose every registration could not be read or holds no manifest of it, is answered with why.
func (ix *Indexer) TakeIn(ctx context.Context, module string) TakenIn {
	out := TakenIn{Module: module}
	if ix.Builds == nil {
		out.Answer = "the indexer reads no controller's build records, so it knows no registration"
		return out
	}
	regs := ix.Builds.RegistrationsOf(module)
	if len(regs) == 0 {
		if key, err := delivery.ModulesKey(module); err == nil && ix.Modules != nil {
			if _, _, found, err := ix.Modules.Get(ctx, key); err == nil && found {
				out.Held, out.Answer = true, "the modules state holds it already: a merge to its repository writes it from now on"
				return out
			}
		}
		out.Answer = "the controller's build records name no registration of " + module + ": a module from outside the indexed repositories is known only from them, and one inside them from a merge"
		return out
	}
	var whys []string
	for _, r := range regs {
		t, err := ix.indexRegistered(ctx, r, true)
		out.Repository, out.Commit, out.Path = t.repository, t.commit, pathOf(t.dir)
		switch {
		case err != nil:
			whys = append(whys, err.Error())
			continue
		case t.held:
			out.Held, out.Answer = true, t.why
			out.Repository, out.Commit, out.Path = "", "", ""
			return out
		case t.stored:
			out.Stored = true
			out.Answer = fmt.Sprintf("written to the modules state from %s at %s", t.repository, t.commit)
			return out
		}
		whys = append(whys, t.why)
	}
	out.Answer = "not taken in: " + strings.Join(whys, "; ")
	return out
}

// IndexRegisteredModule indexes the registration the controller's records name for a module, when there is one.
func (ix *Indexer) IndexRegisteredModule(ctx context.Context, module string) error {
	if ix.Builds == nil {
		return nil
	}
	var first error
	for _, r := range ix.Builds.RegistrationsOf(module) {
		if err := ix.IndexRegistered(ctx, r); err != nil && first == nil {
			first = err
		}
	}
	return first
}

// IndexRegisteredAll indexes every registered module the modules bucket does not hold. A module that cannot be read is
// said and the others go on; the first error is answered.
func (ix *Indexer) IndexRegisteredAll(ctx context.Context) error {
	if ix.Builds == nil {
		return nil
	}
	var first error
	for _, r := range ix.Builds.Registered() {
		if err := ix.IndexRegistered(ctx, r); err != nil {
			ix.logf("the registered module %s is not indexed: %v", r.Module, err)
			if first == nil {
				first = err
			}
		}
	}
	return first
}