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

package index

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"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-sdk/go/health"
	"git.novox.be/novox/mesh-sdk/go/keys"

	"mesh-delivery-indexer/judge"
)

// manifest-accepted, never lost silently (design 25 § 12.2; DLV-I9, I9b, I9c). The indexer says it for every newly
// accepted manifest, its bus-state entries empty when it asks for none, and an empty one for a module leaving the trunk,
// so the bus module's orphans see it; a manifest the judge set aside to wait gets none. A publish that fails is published
// once more; one still failing writes the indexer's finding naming the module, said again at each judging pass after
// and cleared once said. indexer_say_accepted says it again for one module by hand, from the verdict it holds.
const (
	SayAcceptedTool   = "indexer_say_accepted"
	sayAcceptedPrefix = "say."
	// UnsaidKind is the finding of a manifest-accepted whose publish failed twice.
	UnsaidKind = "accepted-unsaid"
)

// acceptedOf is the event body for the verdict v (its revision in the verdicts state) on the manifest raw of digest, its
// entries an empty list when it asks for none.
func acceptedOf(digest string, v meshcatalog.VerdictsByDigestValue, revision uint64, raw []byte) meshcatalog.ManifestAcceptedEvent {
	m, _ := judge.Manifest(raw, nil)
	a := acceptedEvent(digest, v, m)
	a.Revision = int64(revision)
	return a
}

// leftOf is the empty manifest-accepted for module, which left the trunk, judged by judgeID; revision is the modules
// state's revision of the value that left (0 when none is held), so its event id (acceptedEventID, the
// module and the revision, since its body names no digest) differs for each leave.
func leftOf(module string, revision uint64, judgeID string) (meshcatalog.ManifestAcceptedEvent, string) {
	a := meshcatalog.ManifestAcceptedEvent{V: keys.Version, Module: module, Judge: judgeID, BusState: busStateOf(nil), Revision: int64(revision)}
	return a, acceptedEventID(a)
}

// sayAccepted says a with the event id id, once more when the first publish fails; the module's finding is written when
// both fail and deleted when it is said. Nil SayAccepted says nothing.
func (ix *Indexer) sayAccepted(ctx context.Context, a meshcatalog.ManifestAcceptedEvent, id string) error {
	if ix.SayAccepted == nil {
		return nil
	}
	err := ix.SayAccepted(ctx, a, id)
	if err != nil {
		ix.logf("judge: %s for %s failed, published again: %v", meshcatalog.EventManifestAccepted, a.Module, err)
		err = ix.SayAccepted(ctx, a, id)
	}
	ix.unsaid(ctx, a.Module, err)
	return err
}

// unsaidKey is the key of module's unsaid finding.
func unsaidKey(module string) string {
	return health.FindingKey(health.ScopeModule, Module+".accepted."+module, UnsaidKind)
}

// unsaid writes module's finding for err, or deletes it with none.
func (ix *Indexer) unsaid(ctx context.Context, module string, err error) {
	if ix.Findings == nil {
		return
	}
	key := unsaidKey(module)
	werr := write(ctx, ix.Findings, key, func(value []byte, found bool) ([]byte, bool, error) {
		if err == nil {
			return nil, found, nil
		}
		at := ix.now().UTC()
		summary := fmt.Sprintf("%s of %s was not published: %v", meshcatalog.EventManifestAccepted, module, err)
		data, ferr := health.EncodeFinding(health.Finding{Key: key, Kind: UnsaidKind,
			Subject: health.Subject{Scope: health.ScopeModule, ID: Module + ".accepted." + module}, Severity: health.Warning,
			Summary:  summary,
			Headline: "The indexer could not tell the bus module that " + module + "'s manifest was accepted",
			Explanation: "The bus module creates and removes a module's bus state from this event. The indexer says it again at its " +
				"next judging pass and clears this once it is said; indexer_say_accepted says it now.",
			Resolved: "manifest-accepted of " + module + " was said.",
			Evidence: []health.Evidence{{At: at, Said: summary}}, Source: UnsaidKind, Observer: Module, At: at})
		return data, false, ferr
	})
	if werr != nil {
		ix.logf("judge: the finding of %s for %s is not written: %v", meshcatalog.EventManifestAccepted, module, werr)
	}
}

// sayFor says module's manifest-accepted from what the indexer holds now: from its trunk's accepted verdict, or an empty
// one when the trunk holds the module no more (only when left is allowed: by hand a module with no accepted verdict is
// refused). A held module whose verdict waits or is refused gets none; again gives the event an id of its own, so a hand
// act within the bus's duplicate window is still said.
func (ix *Indexer) sayFor(ctx context.Context, module string, allowLeft, again bool) (string, error) {
	held, err := ix.heldModules(ctx)
	if err != nil {
		return "", fmt.Errorf("the modules state cannot be read: %w", err)
	}
	m, ok := held[module]
	var a meshcatalog.ManifestAcceptedEvent
	var id string
	switch {
	case !ok && !allowLeft:
		return "", fmt.Errorf("the indexer holds no module %s, so it has no accepted verdict to say", module)
	case !ok:
		a, id = leftOf(module, 0, ix.judgeID())
	case m.Manifest == "" || ix.Verdicts == nil || ix.Manifests == nil:
		return "", fmt.Errorf("%s has no accepted verdict the indexer holds", module)
	default:
		key, err := meshcatalogclient.VerdictsKey(m.Manifest)
		if err != nil {
			return "", err
		}
		value, revision, found, err := ix.Verdicts.Get(ctx, key)
		if err != nil {
			return "", fmt.Errorf("the verdict on %s cannot be read: %w", module, err)
		}
		var v meshcatalog.VerdictsByDigestValue
		if found {
			v, err = meshcatalogclient.DecodeVerdict(value)
		}
		if !found || err != nil || meshcatalogclient.VerdictHeld(v) {
			if !allowLeft {
				return "", fmt.Errorf("%s has no accepted verdict on its manifest %s", module, m.Manifest)
			}
			ix.unsaid(ctx, module, nil) // a set-aside or refused manifest gets none
			return "none: its manifest is not accepted", nil
		}
		raw, _, found, err := ix.Manifests.Get(ctx, m.Manifest)
		kept, derr := meshcatalogclient.DecodeManifest(raw)
		if err != nil || !found || derr != nil {
			return "", fmt.Errorf("the manifest %s of %s cannot be read", m.Manifest, module)
		}
		a = acceptedOf(m.Manifest, v, revision, []byte(kept.Manifest))
		id = acceptedEventID(a)
	}
	if again {
		id += "-again-" + ix.now().UTC().Format(time.RFC3339Nano)
	}
	if err := ix.sayAccepted(ctx, a, id); err != nil {
		return "", fmt.Errorf("%s of %s was not published: %w", meshcatalog.EventManifestAccepted, module, err)
	}
	if a.Digest == "" {
		return "said empty: the trunk holds no module " + module, nil
	}
	return fmt.Sprintf("said for %s, manifest %s at revision %d, %d bus-state entries", module, a.Digest, a.Revision, len(a.BusState)), nil
}

// SayUnsaid says again each manifest-accepted whose finding stands (the judging pass's end): cleared once said.
func (ix *Indexer) SayUnsaid(ctx context.Context) {
	if ix.Findings == nil {
		return
	}
	keys, err := ix.Findings.Keys(ctx)
	if err != nil {
		return
	}
	for _, key := range keys {
		if !strings.HasSuffix(key, "."+UnsaidKind) {
			continue
		}
		value, _, found, err := ix.Findings.Get(ctx, key)
		if err != nil || !found {
			continue
		}
		f, err := health.DecodeFinding(value)
		if err != nil || f.Kind != UnsaidKind {
			continue
		}
		module := strings.TrimPrefix(f.Subject.ID, Module+".accepted.")
		if _, err := ix.sayFor(ctx, module, true, false); err != nil {
			ix.logf("judge: %v; its finding stands", err)
		}
	}
}

// SayAcceptedAgain answers indexer_say_accepted: module's manifest-accepted said again now from its accepted verdict.
func (ix *Indexer) SayAcceptedAgain(ctx context.Context, module string) TakenIn {
	out := TakenIn{Module: module}
	answer, err := ix.sayFor(ctx, module, false, true)
	if err != nil {
		out.Answer = err.Error()
		return out
	}
	out.Stored, out.Answer = true, answer
	return out
}

// NewSayAccepted is the value indexer_say_accepted writes for a module: its key in the takeins state and the value.
func NewSayAccepted(module, why, by string, at time.Time) (string, []byte, error) {
	key, data, err := NewTakeIn(module, why, by, at)
	if err != nil {
		return "", nil, err
	}
	var a TakeInAsk
	if err := json.Unmarshal(data, &a); err != nil {
		return "", nil, err
	}
	if a.Why == "" {
		return "", nil, errors.New("say why it is said again (why): it is a hand act")
	}
	a.Say = true
	data, err = json.Marshal(a)
	return sayAcceptedPrefix + key, data, err
}