mesh-catalog/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go

package main

// The records kept in the ledger (novox/hq ADR 0597 §11, to-be 55 §2.1): the Store the rules of the knowledge base are
// written against, on the tool runner's ledger calls. A record is one instance, `decision/0290` or `research/081`; every
// change to it is one operation appended to its history; its head, the entry, is the state apply derives from the
// operations and is never written by the module; a section's text is in the `write-section` operation that wrote it and
// nowhere else, the head naming it by its title, its digest and that operation's sequence.

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/buscalls"
)

// MostSectionBytes is the most a section holds (to-be 55 §4), within a history entry's 256 KiB.
const MostSectionBytes = 40 << 10

// sectionRef is how a head names a section: its title, the digest of its text and the sequence of the write-section that
// wrote it.
type sectionRef struct {
	Name     string `json:"name"`
	Digest   string `json:"digest"`
	Sequence uint64 `json:"sequence"`
}

// storedRecord is a record as the head and a version keep it: the record whole but for its sections' text.
type storedRecord struct {
	Record
	Sections []sectionRef `json:"sections,omitempty"`
}

// entryHead is the head of an instance: the version served, the one proposed over it, and how many versions there are.
type entryHead struct {
	Current        *storedRecord `json:"current"`
	CurrentVersion int           `json:"current-version"`
	Pending        *storedRecord `json:"pending,omitempty"`
	PendingVersion int           `json:"pending-version,omitempty"`
	Versions       int           `json:"versions"`
}

// versionHead is a version as its operation keeps it.
type versionHead struct {
	N  int    `json:"n"`
	By string `json:"by"`
	// ImportedBy is who wrote the version in the database it was copied from: the ledger's author is the module's own
	// account, and the old author is kept apart, marked imported (ADR 0597 §7).
	ImportedBy string       `json:"imported-by,omitempty"`
	At         time.Time    `json:"at"`
	Why        string       `json:"why"`
	Insight    bool         `json:"insight,omitempty"`
	Record     storedRecord `json:"record"`
}

// headState is the state of an instance, as apply derives it from the operations.
type headState struct {
	Kind     string     `json:"kind"`
	Number   int        `json:"number"`
	Entry    *entryHead `json:"entry"`
	Sequence uint64     `json:"sequence"`
}

// SectionDigest is the digest a section's text is kept under; the ledger refuses a write-section whose digest is another.
func SectionDigest(text string) string {
	sum := sha256.Sum256([]byte(text))
	return "sha256:" + hex.EncodeToString(sum[:])
}

// InstanceOf is the instance a record is: `decision/0290`, `research/081`.
func InstanceOf(k Key) string {
	if k.Kind == KindResearch {
		return fmt.Sprintf("research/%03d", k.Number)
	}
	return fmt.Sprintf("decision/%04d", k.Number)
}

// ledgerStore is the Store on the ledger.
type ledgerStore struct {
	pipe   Pipe
	states States

	mu sync.Mutex
	// heads are the instances read, by key, each with the sequence it was read at; texts are the sections' text by digest.
	heads   map[Key]*headState
	texts   map[string]string
	scanned bool
	// drafted are the sections written for a draft and named by no version yet: a version that carries one reuses its
	// write-section and writes none again.
	drafted map[Key][]sectionRef
	// index is the search index of the module's memory, nil until built.
	index      *searchIndex
	indexAsync bool
	logf       func(format string, a ...any)
}

// newLedgerStore is the store over a pipe and the module's states.
func newLedgerStore(p Pipe, s States) *ledgerStore {
	return &ledgerStore{pipe: p, states: s, heads: map[Key]*headState{}, texts: map[string]string{}, drafted: map[Key][]sectionRef{}}
}

// stripped is r as a stored record naming its sections by ref: the refs of the sections already written are kept, the
// new ones are filled by the caller.
func stripped(r *Record, refs []sectionRef) storedRecord {
	c := r.Clone()
	c.Sections = nil
	return storedRecord{Record: *c, Sections: refs}
}

// read is one instance's head from the ledger, or nil when it has none.
func (s *ledgerStore) read(k Key) (*headState, error) {
	var got buscalls.InstanceState
	if err := s.pipe.Call(buscalls.LedgerGet, buscalls.InstanceRequest{Ledger: LedgerName, Instance: InstanceOf(k)}, &got); err != nil {
		return nil, err
	}
	if len(got.State) == 0 || string(got.State) == "null" {
		return nil, nil
	}
	var h headState
	if err := json.Unmarshal(got.State, &h); err != nil {
		return nil, fmt.Errorf("the head of %s is not a record's: %w", k, err)
	}
	h.Sequence = got.Sequence
	if h.Entry == nil {
		// Sections were written for it and no version: a draft of a number not held yet is no record.
		return nil, nil
	}
	return &h, nil
}

// fresh is the head, read again unless this process wrote it or read it and nothing else writes: one instance of the
// module holds the seat, so what it holds is what the ledger has.
func (s *ledgerStore) fresh(k Key) (*headState, error) {
	if h := s.heads[k]; h != nil {
		return h, nil
	}
	h, err := s.read(k)
	if err != nil || h == nil {
		return nil, err
	}
	s.heads[k] = h
	return h, nil
}

// heldPrefix starts the key of the state `findings` that says a record is held: the ledger answers an instance by its path and
// lists none, so the store keeps one small key for each record it creates, and a restart reads them to know what to ask for.
const heldPrefix = "held."

func heldKey(k Key) string { return fmt.Sprintf("%s%s.%04d", heldPrefix, k.Kind, k.Number) }

func (s *ledgerStore) hold(k Key) {
	if _, err := s.states.State(findingsState).Put(heldKey(k), 1); err != nil && s.logf != nil {
		s.logf("[mesh-knowledge] that %s is held could not be kept: %v", k, err)
	}
}

// scan reads every instance the ledger holds, once: the keys that say a record is held name them; a ledger written before
// those keys existed is probed number by number, a long run of none ending it, and the keys are then written.
func (s *ledgerStore) scan() error {
	if s.scanned {
		return nil
	}
	keys, err := s.states.State(findingsState).Keys()
	if err != nil {
		return err
	}
	found := 0
	for _, key := range keys {
		rest, ok := strings.CutPrefix(key, heldPrefix)
		if !ok {
			continue
		}
		var k Key
		var n int
		kind, num, ok := strings.Cut(rest, ".")
		if _, err := fmt.Sscanf(num, "%d", &n); !ok || err != nil {
			continue
		}
		k = Key{kind, n}
		if _, err := s.fresh(k); err != nil {
			return err
		}
		found++
	}
	if found == 0 {
		const gap = 1000
		for _, kind := range Kinds {
			misses := 0
			for n := 1; misses < gap; n++ {
				k := Key{kind, n}
				h, err := s.fresh(k)
				if err != nil {
					return err
				}
				if h == nil {
					misses++
					continue
				}
				misses = 0
				s.hold(k)
			}
		}
	}
	s.scanned = true
	return nil
}

// text is the text of a section by its ref, read from the write-section that wrote it when it is not held, and held to its
// digest.
func (s *ledgerStore) text(ref sectionRef, k Key) (string, error) {
	if t, ok := s.texts[ref.Digest]; ok {
		return t, nil
	}
	var page buscalls.HistoryPage
	if err := s.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(k), From: ref.Sequence, Limit: 1}, &page); err != nil {
		return "", err
	}
	if len(page.Operations) == 0 || page.Operations[0].Sequence != ref.Sequence || page.Operations[0].Operation != OpWriteSection {
		return "", fmt.Errorf("the section %q of %s names operation %d, which is no write-section of it", ref.Name, k, ref.Sequence)
	}
	var body struct {
		Text string `json:"text"`
	}
	if err := json.Unmarshal(page.Operations[0].Body, &body); err != nil {
		return "", err
	}
	if SectionDigest(body.Text) != ref.Digest {
		return "", fmt.Errorf("the section %q of %s has a text whose digest is not %s: the history is damaged", ref.Name, k, ref.Digest)
	}
	s.texts[ref.Digest] = body.Text
	return body.Text, nil
}

// whole is a stored record with its sections' text.
func (s *ledgerStore) whole(sr *storedRecord, k Key) (*Record, error) {
	if sr == nil {
		return nil, nil
	}
	r := sr.Record.Clone()
	r.Sections = nil
	for _, ref := range sr.Sections {
		t, err := s.text(ref, k)
		if err != nil {
			return nil, err
		}
		r.Sections = append(r.Sections, Section{Name: ref.Name, Text: t})
	}
	return r, nil
}

// entryOf is the entry a head is.
func (s *ledgerStore) entryOf(k Key, h *headState) (*Entry, error) {
	cur, err := s.whole(h.Entry.Current, k)
	if err != nil {
		return nil, err
	}
	pend, err := s.whole(h.Entry.Pending, k)
	if err != nil {
		return nil, err
	}
	return &Entry{Current: cur, CurrentVersion: h.Entry.CurrentVersion, Pending: pend, PendingVersion: h.Entry.PendingVersion,
		Versions: h.Entry.Versions, Revision: int64(h.Sequence)}, nil
}

func (s *ledgerStore) Get(_ context.Context, k Key) (*Entry, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	h, err := s.fresh(k)
	if err != nil || h == nil {
		return nil, err
	}
	return s.entryOf(k, h)
}

func (s *ledgerStore) All(context.Context) ([]*Entry, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.scan(); err != nil {
		return nil, err
	}
	keys := make([]Key, 0, len(s.heads))
	for k := range s.heads {
		keys = append(keys, k)
	}
	sortKeys(keys)
	out := make([]*Entry, 0, len(keys))
	for _, k := range keys {
		e, err := s.entryOf(k, s.heads[k])
		if err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, nil
}

// write appends a write-section for each section of r that is new, and answers the refs of all its sections. known are
// the refs already written, by name and digest.
func (s *ledgerStore) write(ctx context.Context, k Key, r *Record, known []sectionRef) ([]sectionRef, error) {
	var refs []sectionRef
	for _, sec := range r.Sections {
		d := SectionDigest(sec.Text)
		found := false
		for _, kr := range known {
			if kr.Name == sec.Name && kr.Digest == d {
				refs, found = append(refs, kr), true
				break
			}
		}
		if found {
			continue
		}
		if len(sec.Text) > MostSectionBytes {
			return nil, fmt.Errorf("the section %q of %s is %d bytes; a section holds at most %d: split it", sec.Name, k, len(sec.Text), MostSectionBytes)
		}
		seq, err := s.apply(ctx, k, OpWriteSection, map[string]any{"title": sec.Name, "text": sec.Text, "digest": d})
		if err != nil {
			return nil, err
		}
		s.texts[d] = sec.Text
		refs = append(refs, sectionRef{Name: sec.Name, Digest: d, Sequence: seq})
	}
	return refs, nil
}

// apply hands one operation to the ledger and answers its sequence.
func (s *ledgerStore) apply(ctx context.Context, k Key, op string, body map[string]any) (uint64, error) {
	raw, err := json.Marshal(body)
	if err != nil {
		return 0, err
	}
	var got buscalls.Applied
	if err := s.pipe.Call(buscalls.LedgerApply, buscalls.ApplyRequest{Ledger: LedgerName, Instance: InstanceOf(k), Operation: op, Body: raw, Call: callOf(ctx)}, &got); err != nil {
		return 0, err
	}
	if !got.Taken {
		return 0, fmt.Errorf("the ledger did not take %s on %s", op, k)
	}
	return got.Sequence, nil
}

// operationFor is the operation a change of an entry is, named by what changed, so the history says it.
func operationFor(old *Entry, v Version, current bool) string {
	if old == nil {
		switch {
		case v.Record.Status == StatusReserved:
			return OpReserve
		case v.Record.Source != "" && v.Record.Author == "" && len(v.Record.Writers) == 0:
			return OpImport
		}
		return OpPropose
	}
	prev, next := old.Working(), v.Record
	switch {
	case prev.Status == StatusReserved && next.Status == StatusProposed:
		return OpPropose
	case next.Status == StatusWithdrawn && prev.Status != StatusWithdrawn:
		return OpWithdraw
	case next.Accepted() && !prev.Accepted(), next.Status == StatusSuperseded && prev.Status != StatusSuperseded:
		return OpAccept
	case (next.Retracting != nil) != (prev.Retracting != nil), len(next.Retractions) != len(prev.Retractions), next.Status == StatusRetracted && prev.Status != StatusRetracted:
		return OpRetract
	case next.Ask != nil && (prev.Ask == nil || prev.Ask.Said != next.Ask.Said) && next.Ask.Warrant != "":
		return OpWarrant
	case next.Ask != nil && (prev.Ask == nil || prev.Ask.ID != next.Ask.ID):
		return OpAsk
	case len(next.Reviews) != len(prev.Reviews):
		return OpReview
	case (next.Drafting != nil) != (prev.Drafting != nil) || (next.Drafting != nil && prev.Drafting != nil && *next.Drafting != *prev.Drafting):
		return OpDraft
	case next.Author != prev.Author:
		return OpTakeOver
	}
	return OpSubmit
}

func (s *ledgerStore) Create(ctx context.Context, v Version) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := v.Record.Key()
	h, err := s.fresh(k)
	if err != nil {
		return err
	}
	if h != nil {
		return ErrExists
	}
	v.N = 1
	refs, err := s.write(ctx, k, v.Record, s.drafted[k])
	if err != nil {
		return err
	}
	return s.commit(ctx, k, operationFor(nil, v, true), &entryHead{CurrentVersion: 1, Versions: 1, Current: ptr(stripped(v.Record, refs))}, v, refs, true)
}

func ptr[T any](v T) *T { return &v }

// commit appends the operation that carries the new head and the version, and holds the head it derived.
func (s *ledgerStore) commit(ctx context.Context, k Key, op string, e *entryHead, v Version, refs []sectionRef, current bool) error {
	body := map[string]any{"kind": k.Kind, "number": k.Number, "entry": e,
		"version":          versionHead{N: v.N, By: v.By, At: v.At, Why: v.Why, Insight: v.Insight, Record: stripped(v.Record, refs)},
		"pending-sections": map[string]any{}}
	seq, err := s.apply(ctx, k, op, body)
	if err != nil {
		return err
	}
	s.heads[k] = &headState{Kind: k.Kind, Number: k.Number, Entry: e, Sequence: seq}
	delete(s.drafted, k)
	if e.Versions == 1 {
		s.hold(k)
	}
	s.indexed(v.Record, current)
	return nil
}

// WriteSection appends a write-section of the record (SectionWriter): a draft's text is in the history at once.
func (s *ledgerStore) WriteSection(ctx context.Context, k Key, title, text string) (sectionRef, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(text) > MostSectionBytes {
		return sectionRef{}, fmt.Errorf("the section %q is %d bytes; a section holds at most %d: split it", title, len(text), MostSectionBytes)
	}
	d := SectionDigest(text)
	seq, err := s.apply(ctx, k, OpWriteSection, map[string]any{"title": title, "text": text, "digest": d})
	if err != nil {
		return sectionRef{}, err
	}
	s.texts[d] = text
	ref := sectionRef{Name: title, Digest: d, Sequence: seq}
	s.drafted[k] = append(s.drafted[k], ref)
	return ref, nil
}

// SectionText is the text a ref names (SectionWriter).
func (s *ledgerStore) SectionText(_ context.Context, k Key, ref sectionRef) (string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.text(ref, k)
}

// Working is the refs of the version a writer works on (SectionWriter).
func (s *ledgerStore) Working(_ context.Context, k Key) ([]sectionRef, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	h, err := s.fresh(k)
	if err != nil || h == nil {
		return nil, err
	}
	if h.Entry.Pending != nil {
		return append([]sectionRef(nil), h.Entry.Pending.Sections...), nil
	}
	return append([]sectionRef(nil), h.Entry.Current.Sections...), nil
}

// Known tells the store sections already written for a draft, after a restart (SectionWriter).
func (s *ledgerStore) Known(k Key, refs []sectionRef) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.drafted[k] = append([]sectionRef(nil), refs...)
}

func (s *ledgerStore) Save(ctx context.Context, v Version, current, clearPending bool, revision int64) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	k := v.Record.Key()
	h, err := s.fresh(k)
	if err != nil {
		return err
	}
	if h == nil {
		return fmt.Errorf("%s is not held", k)
	}
	if int64(h.Sequence) != revision {
		return ErrChanged
	}
	old, err := s.entryOf(k, h)
	if err != nil {
		return err
	}
	known := append([]sectionRef(nil), s.drafted[k]...)
	for _, sr := range []*storedRecord{h.Entry.Current, h.Entry.Pending} {
		if sr != nil {
			known = append(known, sr.Sections...)
		}
	}
	refs, err := s.write(ctx, k, v.Record, known)
	if err != nil {
		return err
	}
	e := *h.Entry
	v.N = e.Versions + 1
	e.Versions = v.N
	sr := ptr(stripped(v.Record, refs))
	if current {
		e.Current, e.CurrentVersion = sr, v.N
	} else {
		e.Pending, e.PendingVersion = sr, v.N
	}
	if clearPending {
		e.Pending, e.PendingVersion = nil, 0
	}
	return s.commit(ctx, k, operationFor(old, v, current), &e, v, refs, current)
}

func (s *ledgerStore) Versions(_ context.Context, k Key) ([]Version, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	h, err := s.fresh(k)
	if err != nil || h == nil {
		return nil, err
	}
	var out []Version
	for from := uint64(0); ; {
		var page buscalls.HistoryPage
		if err := s.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(k), From: from}, &page); err != nil {
			return nil, err
		}
		for _, op := range page.Operations {
			var body struct {
				Version *versionHead `json:"version"`
			}
			if op.Operation == OpWriteSection || json.Unmarshal(op.Body, &body) != nil || body.Version == nil {
				continue
			}
			r, err := s.whole(&body.Version.Record, k)
			if err != nil {
				return nil, err
			}
			out = append(out, Version{N: body.Version.N, By: body.Version.By, At: body.Version.At, Why: body.Version.Why, Record: r, Insight: body.Version.Insight})
		}
		if page.Next == 0 {
			break
		}
		from = page.Next
	}
	sort.SliceStable(out, func(i, j int) bool { return out[i].N < out[j].N })
	return out, nil
}

// Search ranks the served versions from the index of the module's memory; it is built on the first question when no start
// built it, and refuses with "index not built" while a start builds it.
func (s *ledgerStore) Search(ctx context.Context, question string, limit int) ([]Hit, error) {
	s.mu.Lock()
	idx, async := s.index, s.indexAsync
	s.mu.Unlock()
	if idx == nil {
		if async {
			return nil, ErrIndexNotBuilt
		}
		if err := s.BuildIndex(ctx); err != nil {
			return nil, err
		}
		s.mu.Lock()
		idx = s.index
		s.mu.Unlock()
	}
	return idx.search(question, limit)
}

func (s *ledgerStore) Inbound(_ context.Context, k Key) ([]Inbound, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if err := s.scan(); err != nil {
		return nil, err
	}
	// The links are in the heads, which hold no text: nothing is read of a section.
	var served []*Record
	for _, h := range s.heads {
		served = append(served, &h.Entry.Current.Record)
	}
	return inboundOf(served, k), nil
}

// The whole-set findings and what the operator was told are keys of the state `findings`.
const (
	findingsState = "findings"
	findingsKey   = "set"
	toldPrefix    = "told."
)

func (s *ledgerStore) SetFindings(_ context.Context, f []Finding) error {
	_, err := s.states.State(findingsState).Put(findingsKey, f)
	return err
}

func (s *ledgerStore) Findings(context.Context) ([]Finding, error) {
	e, err := s.states.State(findingsState).Get(findingsKey)
	if err != nil || e == nil {
		return nil, err
	}
	var out []Finding
	return out, json.Unmarshal(e.Value, &out)
}

func (s *ledgerStore) Told(_ context.Context, topic string) (bool, error) {
	_, err := s.states.State(findingsState).Create(toldPrefix+topic, time.Now().UTC())
	switch {
	case err == nil:
		return true, nil
	case isChanged(err):
		return false, nil
	}
	return false, err
}

func (s *ledgerStore) ToldOf(_ context.Context, prefix string) ([]string, error) {
	keys, err := s.states.State(findingsState).Keys()
	if err != nil {
		return nil, err
	}
	var out []string
	for _, k := range keys {
		if t, ok := strings.CutPrefix(k, toldPrefix); ok && strings.HasPrefix(t, prefix) {
			out = append(out, t)
		}
	}
	sort.Strings(out)
	return out, nil
}

func (s *ledgerStore) Untold(_ context.Context, key string) error {
	return s.states.State(findingsState).Delete(toldPrefix + key)
}

func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) }

// Operations are the operations of a record that made a version (OperationReader).
func (s *ledgerStore) Operations(_ context.Context, k Key) ([]OperationInfo, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var out []OperationInfo
	for from := uint64(0); ; {
		var page buscalls.HistoryPage
		if err := s.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(k), From: from}, &page); err != nil {
			return nil, err
		}
		for _, op := range page.Operations {
			var body struct {
				Version *versionHead `json:"version"`
			}
			if op.Operation == OpWriteSection || json.Unmarshal(op.Body, &body) != nil || body.Version == nil {
				continue
			}
			out = append(out, OperationInfo{Sequence: op.Sequence, Operation: op.Operation, Version: body.Version.N, By: body.Version.By, At: body.Version.At, Why: body.Version.Why})
		}
		if page.Next == 0 {
			return out, nil
		}
		from = page.Next
	}
}