mesh-tools/node-tools/internal/ledger/ledger.go

// Package ledger is the tool runner's side of a ledger (novox/hq ADR 0597, design 25 § 12.4; issue 1966): a storage type
// whose state is derived from its history. `apply` checks an operation against its type, appends it to the instance's
// history with the last sequence expected, derives the state by the operation's effect and writes it as an update of the
// revision read, and writes the instance's snapshot. What the history, the state and the snapshots are kept in is given
// (History, State, Snapshots): the bus in the runtime, fakes in a test.
package ledger

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"time"

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

// The refusals the stores answer, told by errors.Is.
var (
	// ErrRaced is an append or a snapshot written with an expected sequence that is no longer the last: another writer
	// wrote first.
	ErrRaced = errors.New("another writer wrote the instance's history first")
	// ErrChanged is a state written at a revision that is no longer the key's.
	ErrChanged = errors.New("the state was written by another writer first")
	// ErrFull is a state bucket that refuses a write because it is full.
	ErrFull = errors.New("the state bucket is full")
)

// CodeChanged is the error code a bundle is answered a lost race with, as a lost state update (mesh-sdk stdio.CodeChanged).
const CodeChanged = -32010

// raced is an apply refused because another writer appended to the instance first (ADR 0597 §5).
type raced struct{ instance string }

func (e raced) Error() string {
	return "the operation on " + e.instance + " is not taken: another writer appended to its history first; read it and apply again (novox/hq ADR 0597 §5)"
}
func (e raced) Is(target error) bool { return target == ErrRaced }
func (e raced) ErrorCode() int       { return CodeChanged }

// Kept is one message a history keeps: its sequence and its bytes.
type Kept struct {
	Sequence uint64
	Data     []byte
}

// History is a ledger's history: a stream with a subject per instance, each instance named by its key.
type History interface {
	// Last is the sequence of the instance's last operation, 0 when it has none.
	Last(key string) (uint64, error)
	// Append writes one operation, expecting the instance's last sequence to be expect (0: it has none); ErrRaced when
	// it is not.
	Append(key string, op []byte, expect uint64) (uint64, error)
	// Read is the operations of the instances keys names (a key, with `*` for one name and `>` for the rest) from a
	// sequence on, oldest first, at most limit of them (every one when limit is 0).
	Read(keys string, from uint64, limit int) ([]Kept, error)
}

// State is a ledger's state bucket: each instance's record under its key.
type State interface {
	// Get is the key's value and revision; nil and 0 when it has none.
	Get(key string) ([]byte, uint64, error)
	// Write writes the key's value at the revision read (0: only when it has none); ErrChanged when the key is no longer
	// at it, ErrFull when the bucket is full.
	Write(key string, value []byte, revision uint64) (uint64, error)
	// Keys is every key the bucket holds.
	Keys() ([]string, error)
}

// Snapshots are a ledger's snapshots: the latest record of each instance, on a subject of its own, never expiring.
type Snapshots interface {
	// Get is the key's latest snapshot and the sequence it was written at; nil and 0 when it has none.
	Get(key string) ([]byte, uint64, error)
	// Put writes the key's snapshot, expecting its last to be at expect (0: none); ErrRaced when it is not.
	Put(key string, value []byte, expect uint64) error
}

// Ledger is one ledger of one module: its types and where it is kept.
type Ledger struct {
	Module string
	Name   string
	// Account is the owning module's authenticated account, which every operation it applies is stamped with.
	Account   string
	Types     []catalogue.LedgerType
	History   History
	State     State
	Snapshots Snapshots
	// Indexes are its declared indexes' buckets, by index name.
	Indexes map[string]Index
	// Now is the clock an operation is stamped by; time.Now when nil.
	Now func() time.Time
	// Logf says what an apply could not do after its operation was taken; nothing is said when nil.
	Logf func(format string, args ...any)
}

func (l *Ledger) say(format string, args ...any) {
	if l.Logf != nil {
		l.Logf(format, args...)
	}
}

// Record is what the state and a snapshot keep of an instance: its state and the sequence of the last operation in it.
type Record struct {
	Sequence uint64          `json:"sequence"`
	State    json.RawMessage `json:"state"`
}

// segment is one part of an instance's path.
var segment = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)

// KeyOf is the key an instance is kept under, in the state and as its history's and snapshot's subject token: its path's
// parts joined by dots (`decision/0290` is `decision.0290`).
func KeyOf(instance string) (string, error) {
	parts := strings.Split(instance, "/")
	if len(parts) < 2 {
		return "", fmt.Errorf("%q is not an instance: an instance is named by a path whose first part is its type's name, "+
			"such as decision/0290 (novox/hq ADR 0597 §1)", instance)
	}
	for _, p := range parts {
		if !segment.MatchString(p) {
			return "", fmt.Errorf("%q is not an instance: each part of its path holds letters, digits, - and _ alone", instance)
		}
	}
	return strings.Join(parts, "."), nil
}

// InstanceOf is the instance a key names.
func InstanceOf(key string) string { return strings.ReplaceAll(key, ".", "/") }

func (l *Ledger) now() time.Time {
	if l.Now != nil {
		return l.Now().UTC()
	}
	return time.Now().UTC()
}

// typeOf is the instance's type, by its path's first part.
func (l *Ledger) typeOf(instance string) (catalogue.LedgerType, error) {
	name, _, _ := strings.Cut(instance, "/")
	for _, t := range l.Types {
		if t.Name == name {
			return t, nil
		}
	}
	return catalogue.LedgerType{}, fmt.Errorf("%s's ledger %s has no type %q, which the instance %s names", l.Module, l.Name, name, instance)
}

// operationOf is the operation a type allows by its name.
func operationOf(t catalogue.LedgerType, name string) (catalogue.LedgerOperation, error) {
	for _, op := range t.Operations {
		if op.Name == name {
			return op, nil
		}
	}
	return catalogue.LedgerOperation{}, fmt.Errorf("the type %s allows no operation %q", t.Name, name)
}

// Digest is the digest a type pairs a text field with (ADR 0597 §2): sha256 of the text, as the knowledge base writes it.
func Digest(text string) string {
	sum := sha256.Sum256([]byte(text))
	return "sha256:" + hex.EncodeToString(sum[:])
}

// kinds is the JSON kind each schema type name is held to; a name not here is the judge's to know.
var kinds = map[string]func(any) bool{
	"string":  func(v any) bool { _, ok := v.(string); return ok },
	"number":  func(v any) bool { _, ok := v.(float64); return ok },
	"integer": func(v any) bool { f, ok := v.(float64); return ok && f == float64(int64(f)) },
	"boolean": func(v any) bool { _, ok := v.(bool); return ok },
	"object":  func(v any) bool { _, ok := v.(map[string]any); return ok },
	"array":   func(v any) bool { _, ok := v.([]any); return ok },
}

// checked is an operation as apply checked it: its instance's key, its operation and its body.
type checked struct {
	key  string
	op   catalogue.LedgerOperation
	body map[string]any
}

// check holds an operation to its type (ADR 0597 §2): a type the instance names, an operation it allows, a body that is an
// object whose fields fit the schema, the key its effect names, and every digest matching its text.
func (l *Ledger) check(r buscalls.ApplyRequest) (checked, error) {
	key, err := KeyOf(r.Instance)
	if err != nil {
		return checked{}, err
	}
	t, err := l.typeOf(r.Instance)
	if err != nil {
		return checked{}, err
	}
	op, err := operationOf(t, r.Operation)
	if err != nil {
		return checked{}, err
	}
	body := map[string]any{}
	if len(r.Body) > 0 && string(r.Body) != "null" {
		if err := json.Unmarshal(r.Body, &body); err != nil {
			return checked{}, fmt.Errorf("the operation %s's body is a JSON object: %w", r.Operation, err)
		}
	}
	for name, field := range t.Schema {
		v, has := body[name]
		if !has {
			continue
		}
		if fits, known := kinds[field.Type]; known && !fits(v) {
			return checked{}, fmt.Errorf("the operation %s's field %s is a %s in the type %s", r.Operation, name, field.Type, t.Name)
		}
	}
	for name, field := range t.Schema {
		if field.DigestOf == "" {
			continue
		}
		digest, hasDigest := body[name]
		text, hasText := body[field.DigestOf]
		if !hasDigest && !hasText {
			continue
		}
		s, isText := text.(string)
		if !hasDigest || !hasText || !isText || digest != Digest(s) {
			return checked{}, fmt.Errorf("the operation %s's %s is not the digest of its %s, so it is refused (novox/hq ADR 0597 §2)",
				r.Operation, name, field.DigestOf)
		}
	}
	switch op.Effect {
	case catalogue.EffectSetFields, catalogue.EffectAppend:
	case catalogue.EffectSetEntry, catalogue.EffectRemove:
		if op.Effect == catalogue.EffectRemove && op.Into == "" {
			break
		}
		if _, ok := body[op.Key].(string); !ok {
			return checked{}, fmt.Errorf("the operation %s names its entry by its field %s, a text, which its body lacks", r.Operation, op.Key)
		}
	default:
		return checked{}, fmt.Errorf("the operation %s's effect %q is not one the mesh offers (novox/hq ADR 0597 §2)", r.Operation, op.Effect)
	}
	return checked{key: key, op: op, body: body}, nil
}

// effect is an operation's effect on a state (ADR 0597 §2): it copies only the fields the operation names, and the
// operation's sequence and proven author where it says so; null is a removed instance.
func effect(state json.RawMessage, op catalogue.LedgerOperation, body map[string]any, seq uint64, author buscalls.Author) (json.RawMessage, error) {
	s := map[string]any{}
	if len(state) > 0 && string(state) != "null" {
		if err := json.Unmarshal(state, &s); err != nil {
			return nil, fmt.Errorf("the state held is not an object: %w", err)
		}
	}
	copied := map[string]any{}
	for _, f := range op.Fields {
		if v, ok := body[f]; ok {
			copied[f] = v
		}
	}
	if op.Sequence {
		copied["sequence"] = seq
	}
	// Written last, so a body's own `author` never stands for the proven one (ADR 0597 §7).
	if op.Author {
		copied["author"] = author
	}
	switch op.Effect {
	case catalogue.EffectSetFields:
		for k, v := range copied {
			s[k] = v
		}
	case catalogue.EffectSetEntry:
		m, _ := s[op.Into].(map[string]any)
		if m == nil {
			m = map[string]any{}
		}
		m[body[op.Key].(string)] = copied
		s[op.Into] = m
	case catalogue.EffectAppend:
		list, _ := s[op.Into].([]any)
		s[op.Into] = append(list, copied)
	case catalogue.EffectRemove:
		if op.Into == "" {
			return json.RawMessage("null"), nil
		}
		if m, ok := s[op.Into].(map[string]any); ok {
			delete(m, body[op.Key].(string))
		}
	}
	return json.Marshal(s)
}

// fold applies one kept operation to a record, as apply derived it.
func (l *Ledger) fold(rec Record, k Kept) (Record, error) {
	var o buscalls.Operation
	if err := json.Unmarshal(k.Data, &o); err != nil {
		return rec, fmt.Errorf("the history's operation %d is not one: %w", k.Sequence, err)
	}
	t, err := l.typeOf(o.Instance)
	if err != nil {
		return rec, err
	}
	op, err := operationOf(t, o.Operation)
	if err != nil {
		return rec, err
	}
	body := map[string]any{}
	if len(o.Body) > 0 && string(o.Body) != "null" {
		if err := json.Unmarshal(o.Body, &body); err != nil {
			return rec, err
		}
	}
	if (op.Effect == catalogue.EffectSetEntry || (op.Effect == catalogue.EffectRemove && op.Into != "")) && !isText(body[op.Key]) {
		return rec, fmt.Errorf("the history's operation %d names no entry", k.Sequence)
	}
	state, err := effect(rec.State, op, body, k.Sequence, o.Author)
	if err != nil {
		return rec, err
	}
	return Record{Sequence: k.Sequence, State: state}, nil
}

func isText(v any) bool { _, ok := v.(string); return ok }

// replay folds the instance's operations after rec's sequence, up to upto (every one when 0), onto rec.
func (l *Ledger) replay(key string, rec Record, upto uint64) (Record, error) {
	kept, err := l.History.Read(key, rec.Sequence+1, 0)
	if err != nil {
		return rec, err
	}
	for _, k := range kept {
		if upto != 0 && k.Sequence > upto {
			break
		}
		if k.Sequence <= rec.Sequence {
			continue
		}
		if rec, err = l.fold(rec, k); err != nil {
			return rec, err
		}
	}
	return rec, nil
}

// readState is the instance's record in the state and its revision; an empty record when it has none.
func (l *Ledger) readState(key string) (Record, uint64, error) {
	raw, rev, err := l.State.Get(key)
	if err != nil || raw == nil {
		return Record{}, rev, err
	}
	var rec Record
	if err := json.Unmarshal(raw, &rec); err != nil {
		return Record{}, rev, fmt.Errorf("the state of %s is not a record: %w", key, err)
	}
	return rec, rev, nil
}

// readSnapshot is the instance's snapshot and the sequence it was written at; an empty record when it has none.
func (l *Ledger) readSnapshot(key string) (Record, uint64, error) {
	raw, at, err := l.Snapshots.Get(key)
	if err != nil || raw == nil {
		return Record{}, at, err
	}
	var rec Record
	if err := json.Unmarshal(raw, &rec); err != nil {
		return Record{}, at, fmt.Errorf("the snapshot of %s is not a record: %w", key, err)
	}
	return rec, at, nil
}

// derive is the instance's record caught up with its history up to last: from the state, or from the snapshot when it
// is further on (an instance whose state was lost, or whose operations expired, ADR 0597 §8).
func (l *Ledger) derive(key string, base Record, last uint64) (Record, error) {
	if base.Sequence >= last {
		return base, nil
	}
	snap, _, err := l.readSnapshot(key)
	if err != nil {
		return base, err
	}
	if snap.Sequence > base.Sequence {
		base = snap
	}
	return l.replay(key, base, last)
}

// settle writes want into the state at the revision read (ADR 0597 §5): a lost update re-reads, and catches the state
// read up to want's sequence from the history, never appending again; a state already further on is left as it is. It
// answers what the state holds after it, and whether it is behind because the bucket is full (ADR 0597 §6).
func (l *Ledger) settle(key string, want Record, rev uint64) (Record, bool, error) {
	for tries := 0; tries < 16; tries++ {
		raw, err := json.Marshal(want)
		if err != nil {
			return want, false, err
		}
		_, err = l.State.Write(key, raw, rev)
		switch {
		case err == nil:
			return want, false, nil
		case errors.Is(err, ErrFull):
			return want, true, nil
		case !errors.Is(err, ErrChanged):
			return want, false, err
		}
		cur, curRev, err := l.readState(key)
		if err != nil {
			return want, false, err
		}
		if cur.Sequence >= want.Sequence {
			return cur, false, nil
		}
		if want, err = l.replay(key, cur, want.Sequence); err != nil {
			return want, false, err
		}
		rev = curRev
	}
	return want, false, fmt.Errorf("the state of %s kept being written by other writers; the operation is in its history and the next apply catches it up", key)
}

// writeSnapshot writes rec as the instance's snapshot unless the snapshot is already as far on (ADR 0597 §8).
func (l *Ledger) writeSnapshot(key string, rec Record) error {
	for tries := 0; tries < 16; tries++ {
		snap, at, err := l.readSnapshot(key)
		if err != nil {
			return err
		}
		if at != 0 && snap.Sequence >= rec.Sequence {
			return nil
		}
		raw, err := json.Marshal(rec)
		if err != nil {
			return err
		}
		if err = l.Snapshots.Put(key, raw, at); err == nil || !errors.Is(err, ErrRaced) {
			return err
		}
	}
	return fmt.Errorf("the snapshot of %s kept being written by other writers", key)
}

// Apply applies one operation to one instance, stamped with author (ADR 0597 §3, §5, §6, §8; design 25 § 12.4 steps 1-5):
// checked against its type, appended with the instance's last sequence expected, the state derived by its effect and
// written at the revision read, and the snapshot written. A racing writer's append is refused; a full state bucket
// answers taken with the state behind.
func (l *Ledger) Apply(r buscalls.ApplyRequest, author buscalls.Author) (buscalls.Applied, error) {
	c, err := l.check(r)
	if err != nil {
		return buscalls.Applied{}, err
	}
	last, err := l.History.Last(c.key)
	if err != nil {
		return buscalls.Applied{}, err
	}
	held, rev, err := l.readState(c.key)
	if err != nil {
		return buscalls.Applied{}, err
	}
	base, err := l.derive(c.key, held, last)
	if err != nil {
		return buscalls.Applied{}, err
	}
	body := r.Body
	if len(body) == 0 {
		body = json.RawMessage("{}")
	}
	op, err := json.Marshal(buscalls.Operation{Instance: r.Instance, Operation: r.Operation, Body: body, Author: author, At: l.now()})
	if err != nil {
		return buscalls.Applied{}, err
	}
	seq, err := l.History.Append(c.key, op, last)
	if errors.Is(err, ErrRaced) {
		return buscalls.Applied{}, raced{r.Instance}
	}
	if err != nil {
		return buscalls.Applied{}, err
	}
	state, err := effect(base.State, c.op, c.body, seq, author)
	if err != nil {
		return buscalls.Applied{}, err
	}
	// From here the operation is taken: it is in the history, and a fault below is the state's to catch up, said here
	// and answered as the state behind, never as a refusal of what later takes effect (ADR 0597 §6).
	now, behind, err := l.settle(c.key, Record{Sequence: seq, State: state}, rev)
	if err != nil {
		l.say("[mesh-tools] %s's ledger %s took operation %d on %s and its state could not follow: %v; the next apply "+
			"or a rebuild catches it up", l.Module, l.Name, seq, r.Instance, err)
		behind, now = true, Record{Sequence: seq, State: state}
	}
	if behind {
		now = Record{Sequence: seq, State: state}
	}
	if err := l.writeSnapshot(c.key, now); err != nil {
		l.say("[mesh-tools] %s's ledger %s took operation %d on %s and its snapshot could not be written: %v; the next "+
			"apply or a snapshot repair writes it", l.Module, l.Name, seq, r.Instance, err)
	}
	return buscalls.Applied{Taken: true, Sequence: seq, StateBehind: behind}, nil
}