mesh-delivery/modules/mesh-delivery-view/internal/view/history.go

package view

// Two read tools for tracing a module's value back to the message that wrote it (issue 1126 note 76: mesh-catalog's merge
// 9cdac09a was written back over the later 69056948, and no tool read a modules key's history with times, nor what the
// indexer heard of the forge stream FORGE).

import (
	"encoding/json"
	"fmt"
	"slices"
	"strings"
	"time"

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

const (
	// ModuleHistoryTool reads modules.<module>'s kept values with their times.
	ModuleHistoryTool = "indexer_module_history"
	// HeardTool reads the FORGE messages the indexer handled (its state heard).
	HeardTool = "indexer_heard"
	// heardKey is the indexer's one key of its heard state (mesh-delivery-indexer internal/index heard.go HeardKey).
	heardKey = "forge"
)

// Revision is one value a state still keeps for a key (mesh-sdk stdio StateRevision): its revision, when it was written,
// put or del, and the value (none for a delete).
type Revision struct {
	Revision uint64          `json:"revision"`
	At       time.Time       `json:"at"`
	Op       string          `json:"op"`
	Value    json.RawMessage `json:"value,omitempty"`
}

// ModuleRevision is one kept value of modules.<module>, read: where it was computed and what it computed.
type ModuleRevision struct {
	Revision   uint64    `json:"revision"`
	At         time.Time `json:"at"`
	Op         string    `json:"op"`
	Repository string    `json:"repository,omitempty"`
	Commit     string    `json:"commit,omitempty"`
	Src        uint64    `json:"src,omitempty"`
	FP2        string    `json:"fp2,omitempty"`
	None2      string    `json:"none2,omitempty"`
	Manifest   string    `json:"manifest,omitempty"`
	Path       string    `json:"path,omitempty"`
	Unreadable string    `json:"unreadable,omitempty"`
}

// ModuleHistory is a module's kept values, newest first.
type ModuleHistory struct {
	Module string           `json:"module"`
	Values []ModuleRevision `json:"values"`
	Note   string           `json:"note"`
}

// ModuleHistory reads every value the modules state keeps for module, newest first, each with its revision, time, commit,
// source revision and fingerprint.
func (r Reader) ModuleHistory(module string) (ModuleHistory, error) {
	module = strings.TrimSpace(module)
	out := ModuleHistory{Module: module, Note: "newest first; the modules state keeps 20 values a key (its history), so an older one is gone"}
	key, err := delivery.ModulesKey(module)
	if err != nil {
		return out, err
	}
	if r.State.History == nil {
		return out, fmt.Errorf("the modules state's history cannot be read here")
	}
	revs, err := r.State.History(key)
	if err != nil {
		return out, err
	}
	if len(revs) == 0 {
		return out, fmt.Errorf("no history of %s in the modules state", module)
	}
	for _, rev := range newestFirst(revs) {
		row := ModuleRevision{Revision: rev.Revision, At: rev.At, Op: rev.Op}
		if len(rev.Value) > 0 && string(rev.Value) != "null" {
			// Read field by field, whatever the value's version, so an older or newer writer's value still shows.
			var m delivery.Module
			if err := json.Unmarshal(rev.Value, &m); err != nil {
				row.Unreadable = err.Error()
			} else {
				row.Repository, row.Commit, row.Src, row.FP2, row.None2, row.Manifest, row.Path = m.Repository, m.Commit, m.Src, m.FP2, m.None2, m.Manifest, m.Path
			}
		}
		out.Values = append(out.Values, row)
	}
	return out, nil
}

// HeardRow is one FORGE message as the indexer handled it, with the revision and time of its value.
type HeardRow struct {
	Revision uint64          `json:"revision"`
	At       time.Time       `json:"at"`
	Heard    json.RawMessage `json:"heard"`
}

// HeardAnswer is the FORGE messages the indexer handled, newest first.
type HeardAnswer struct {
	Heard []HeardRow `json:"heard"`
	Note  string     `json:"note"`
}

// Heard reads the indexer's heard state: the last FORGE messages it handled (at most limit, 0 for every one kept), newest
// first.
func (r Reader) Heard(limit int) (HeardAnswer, error) {
	out := HeardAnswer{Heard: []HeardRow{}, Note: "newest first; the indexer keeps its last 64 FORGE messages. A message the stream holds and the indexer never handled is not here; " +
		"delivered above 1 is a message the server delivered again"}
	if r.HeardState.History == nil {
		return out, fmt.Errorf("the indexer's heard state cannot be read here")
	}
	revs, err := r.HeardState.History(heardKey)
	if err != nil {
		return out, err
	}
	for _, rev := range newestFirst(revs) {
		if limit > 0 && len(out.Heard) == limit {
			break
		}
		if len(rev.Value) == 0 || string(rev.Value) == "null" {
			continue
		}
		out.Heard = append(out.Heard, HeardRow{Revision: rev.Revision, At: rev.At, Heard: rev.Value})
	}
	return out, nil
}

func newestFirst(revs []Revision) []Revision {
	revs = slices.Clone(revs)
	slices.SortFunc(revs, func(a, b Revision) int {
		switch {
		case a.Revision > b.Revision:
			return -1
		case a.Revision < b.Revision:
			return 1
		}
		return 0
	})
	return revs
}