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

package index

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

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
	"git.novox.be/novox/mesh-sdk/go/process"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
)

// The gate's source revision of a watched branch (design 47 G §15, §115), read through the seat mesh-delivery's verb source
// and its event source-moved, never the gate's state: it names every repository's merged head, and its revision is the
// mesh's source revision of that branch. The indexer keeps the latest it has seen. A module is stamped
// with it (Module.Src) when the head the source names for its repository is the merge commit the module's fingerprint
// was computed at: from the modules bucket when the merge was heard first, and when the merge is heard after the source.

// sourceNow is the latest source revision seen.
type sourceNow struct {
	rev   uint64
	heads map[string]string
}

// srcOf is the source revision that names this merge commit as the repository's head, or 0 when the latest source seen
// names another head or none has been seen.
func (ix *Indexer) srcOf(repository, merge string) uint64 {
	ix.mu.Lock()
	defer ix.mu.Unlock()
	if ix.source.rev != 0 && merge != "" && ix.source.heads[repository] == merge {
		return ix.source.rev
	}
	return 0
}

// ApplySource takes the source revision the verb source answered in. An older revision than the latest seen is left. For each
// repository the source names, each module built from it whose fingerprint was computed at the head the source names, and
// that carries no Src yet, is written again with Src the revision (once: a later revision leaves it); a module computed at another commit is left (a merge heard later is stamped
// when it is written, applyMerged), and a repository none of the indexer's modules is built from writes nothing. The
// modules of a repository are those the modules bucket holds of it (ADR 0501 clause 6).
func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) error {
	rev := uint64(a.Revision)
	s := struct{ Heads map[string]string }{Heads: map[string]string{}}
	for _, r := range a.Repositories {
		s.Heads[r.Repository] = r.Head
	}
	ix.kickJudge(ctx)
	ix.mu.Lock()
	if rev < ix.source.rev {
		ix.mu.Unlock()
		return nil
	}
	ix.source = sourceNow{rev: rev, heads: maps.Clone(s.Heads)}
	ix.mu.Unlock()
	if ix.Modules == nil {
		return ix.RecomputeCandidates(ctx)
	}
	if err := ix.kickCatchUp(ctx, s.Heads); err != nil {
		return err
	}
	// A registered module the forge could not read is read again at each new source revision (issue 1945), in the
	// background: a forge that does not answer must not hold this revision. One taken in later is stamped when it is written
	// (applyManifests reads srcOf).
	ix.kickRetry(ctx)
	held, err := ix.heldModules(ctx)
	if err != nil {
		return err
	}
	for _, key := range slices.Sorted(maps.Keys(held)) {
		repository := ""
		for r := range s.Heads {
			if trunk.SameRepository(r, held[key].Repository) {
				repository = r
			}
		}
		if repository == "" {
			continue
		}
		head := s.Heads[repository]
		err = write(ctx, ix.Modules, key, func(value []byte, found bool) ([]byte, bool, error) {
			if !found {
				return nil, false, nil
			}
			cur, err := meshcatalogclient.DecodeModule(value)
			if err != nil {
				ix.logf("modules.%s holds what is not a module, left: %v", key, err)
				return nil, false, nil
			}
			// Stamped once, with the first revision naming the commit as the head: a later revision (a merge of another
			// repository) leaves the value as it is, so no revision of the source rewrites a module that did not change.
			if cur.Repository != repository || cur.Commit != head || cur.Src != 0 {
				return nil, false, nil
			}
			cur.Src = int64(rev)
			if cur.FpSrc == 0 {
				cur.FpSrc = int64(rev) // its fingerprint moved at this merge, whose revision is heard only now (ADR 0601 §1)
			}
			data, err := meshcatalogclient.EncodeModule(cur)
			return data, false, err
		})
		if err != nil {
			return fmt.Errorf("modules.%s: %w", key, err)
		}
	}
	// The stamps are kept with the view (KeepView), so a restart reads them back.
	if err := ix.KeepView(ctx); err != nil {
		return err
	}
	// A trunk that moved makes the previews made onto the old one out of date.
	return ix.RecomputeCandidates(ctx)
}

// kickCatchUp takes the trunk in at a first start, when the objects keep no trunk revision yet (the lead's ruling A on
// milestone 47/G: the catch-up path): the head the source names for every repository, by the merged message the forge
// bridge gives for it, each applied as its merge, and revision 1 kept from them all at its end. Until then the revision
// the tools answer waits, never an empty catalog. The catch-up runs off the source handler, in its own goroutine, said
// when it ends (lead-47's ruling R1 on the rebase): a bridge that does not answer holds the kept call (ADR 0668) to its
// bound, and must not hold this source revision's stamps and the pulls' candidates with it. A repository whose message is
// not read keeps no revision, said: the next source revision runs the catch-up again (no timer). Objects not on the bus
// yet run none, said, until then.
func (ix *Indexer) kickCatchUp(ctx context.Context, heads map[string]string) error {
	if ix.Revisions == nil || ix.Message == nil || len(heads) == 0 {
		return nil
	}
	if _, any, err := newestTrunk(ctx, ix.Revisions); errors.Is(err, process.ErrNotYet) {
		ix.logf("catalog graph: the trunk is not taken in yet: its objects %s are not on the bus", CatalogRevisionsObjects)
		return nil
	} else if err != nil || any {
		return err
	}
	if !ix.catchingUp.CompareAndSwap(false, true) {
		return nil
	}
	heads = maps.Clone(heads)
	ix.catchUpWG.Add(1)
	go func() {
		defer ix.catchUpWG.Done()
		unread := ix.takeInHeads(ctx, heads)
		ix.catchingUp.Store(false)
		if len(unread) > 0 {
			ix.logf("catalog graph: the first start's catch-up ended and no first revision is kept: %d repositories were not taken in (%s); taken in again at the next source revision",
				len(unread), strings.Join(unread, ", "))
			return
		}
		if err := ix.KeepView(ctx); err != nil {
			ix.logf("catalog graph: the first start's catch-up took in the trunk of %d repositories, but its revision could not be kept: %v; taken in again at the next source revision", len(heads), err)
			return
		}
		ix.logf("catalog graph: the first start's catch-up ended: the trunk of %d repositories taken in", len(heads))
	}()
	return nil
}

// takeInHeads applies the forge bridge's merged message of each repository's head, as its merge, and answers the
// repositories whose message was not read or not applied.
func (ix *Indexer) takeInHeads(ctx context.Context, heads map[string]string) []string {
	var unread []string
	for _, repository := range slices.Sorted(maps.Keys(heads)) {
		head := heads[repository]
		if head == "" {
			continue
		}
		f, err := ix.mergedAt(ctx, repository, head)
		if err == nil {
			_, err = ix.ApplyMessage(ctx, f)
		}
		if err != nil {
			unread = append(unread, repository)
			ix.logf("catalog graph: the trunk of %s at %.12s is not taken in: %v", repository, head, err)
		}
	}
	return unread
}

// CatchUp is the indexer's catch-up tool's work (indexer_take_in with no module and no repository, the catch-up its
// manifest names beside the catalog graph revisions): after a restore point of them is restored, the node agent calls it
// with no arguments, and the indexer takes in the head the gate's source names for every repository, as the first
// start's catch-up does, whatever revision the objects keep, then keeps the view as the next revision when an entry
// moved. Its answer names the repositories taken in and those whose merged message could not be read.
func (ix *Indexer) CatchUp(ctx context.Context) TakenIn {
	out := TakenIn{Module: "catch-up"}
	ix.mu.Lock()
	heads := maps.Clone(ix.source.heads)
	ix.mu.Unlock()
	switch {
	case len(heads) == 0:
		out.Answer = "the gate's source names no head yet, so nothing is taken in; the next source revision is taken in as it comes"
		return out
	case ix.Modules == nil || ix.Message == nil:
		out.Answer = "the indexer holds no view of the trunk or reads no merged message, so nothing is taken in"
		return out
	case !ix.catchingUp.CompareAndSwap(false, true):
		out.Answer = "a catch-up runs already; its end keeps the revision"
		return out
	}
	unread := ix.takeInHeads(ctx, heads)
	ix.catchingUp.Store(false)
	if err := ix.KeepView(ctx); err != nil {
		out.Answer = "the repositories were taken in but the revision could not be kept: " + err.Error()
		return out
	}
	out.Stored = len(unread) == 0
	out.Unread = unread
	out.Answer = fmt.Sprintf("the heads the gate's source names for %d repositories taken in", len(heads)-len(unread))
	if len(unread) > 0 {
		out.Answer += fmt.Sprintf("; %d not (%s): their merged message was not read, and the call may be made again", len(unread), strings.Join(unread, ", "))
	}
	return out
}

// mergedAt is the forge bridge's merged message of a repository's head.
func (ix *Indexer) mergedAt(ctx context.Context, repository, head string) (forgemsg.Forge, error) {
	data, err := ix.Message(forTakeIn(ctx, repository, head), repository, head, 0)
	if err != nil {
		return forgemsg.Forge{}, fmt.Errorf("the forge bridge could not answer: %w", err)
	}
	f, err := forgemsg.DecodeForge(data)
	if err != nil {
		return f, fmt.Errorf("the forge bridge's answer is not a forge message: %w", err)
	}
	if f.Kind != forgemsg.ForgeMerged || f.Merge != head || !trunk.SameRepository(f.Repository, repository) {
		return f, fmt.Errorf("the forge bridge's answer is not the merge of %s at %s", repository, head)
	}
	return f, nil
}