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

package ledger

import (
	"encoding/json"
	"errors"
	"fmt"

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

// Reading a ledger (ADR 0597 §10, design 25 § 12.5; issue 1966 T3): an instance's state, its history, a watch of its
// states, and a declared index. Only the owning module reads its ledger: the runtime opens a ledger for the module whose
// bundle asks, never one the request names.

// Index is one declared index's bucket, kept by a watcher in the owning module's process and rebuilt by reindex (ADR
// 0597 §4).
type Index interface {
	// Entries is the entries whose key a pattern names (`*` one name, `>` the rest; empty is every key), at most limit
	// (every one when 0), each as the bucket holds it.
	Entries(pattern string, limit int) ([]Indexed, error)
	// Put writes one entry; Delete removes one.
	Put(key string, value []byte) error
	Delete(key string) error
}

// Indexed is one key of an index's bucket and its value, a buscalls.IndexEntry as JSON.
type Indexed struct {
	Key   string
	Value []byte
}

// Watcher is a state bucket that hands each instance's record as it is and as it changes, under a key pattern.
type Watcher interface {
	Watch(pattern string, deliver func(key string, value []byte, current bool) error) (func(), error)
}

// HistoryMost is the most operations one history read answers.
const HistoryMost = 1000

// Get is an instance's state as the state holds it, and the sequence of its last operation in it; null and 0 for an
// instance with none.
func (l *Ledger) Get(r buscalls.InstanceRequest) (buscalls.InstanceState, error) {
	key, err := KeyOf(r.Instance)
	if err != nil {
		return buscalls.InstanceState{}, err
	}
	rec, _, err := l.readState(key)
	if err != nil {
		return buscalls.InstanceState{}, err
	}
	state := rec.State
	if len(state) == 0 {
		state = json.RawMessage("null")
	}
	return buscalls.InstanceState{Instance: r.Instance, State: state, Sequence: rec.Sequence}, nil
}

// operationsOf reads kept operations as the history answers them, each with its sequence.
func operationsOf(kept []Kept) ([]buscalls.Operation, error) {
	out := make([]buscalls.Operation, 0, len(kept))
	for _, k := range kept {
		var o buscalls.Operation
		if err := json.Unmarshal(k.Data, &o); err != nil {
			return nil, fmt.Errorf("the history's operation %d is not one: %w", k.Sequence, err)
		}
		o.Sequence = k.Sequence
		out = append(out, o)
	}
	return out, nil
}

// ReadHistory is an instance's operations from a sequence on, oldest first, each with its sequence and its proven author.
func (l *Ledger) ReadHistory(r buscalls.HistoryRequest) (buscalls.HistoryPage, error) {
	key, err := KeyOf(r.Instance)
	if err != nil {
		return buscalls.HistoryPage{}, err
	}
	limit := r.Limit
	if limit <= 0 || limit > HistoryMost {
		limit = HistoryMost
	}
	last, err := l.History.Last(key)
	if err != nil {
		return buscalls.HistoryPage{}, err
	}
	kept, err := l.History.Read(key, r.From, limit+1)
	if err != nil {
		return buscalls.HistoryPage{}, err
	}
	page := buscalls.HistoryPage{Last: last}
	if len(kept) > limit {
		page.Next = kept[limit].Sequence
		kept = kept[:limit]
	}
	if page.Operations, err = operationsOf(kept); err != nil {
		return buscalls.HistoryPage{}, err
	}
	return page, nil
}

// Watch hands deliver the state of every instance Instances names (`*` one name, `>` the rest, in an instance's path),
// the current ones first, then each change.
func (l *Ledger) Watch(r buscalls.LedgerWatchRequest, deliver func(buscalls.InstanceChange) error) (func(), error) {
	w, ok := l.State.(Watcher)
	if !ok {
		return nil, errors.New("this ledger's state cannot be watched")
	}
	pattern := ">"
	if r.Instances != "" {
		pattern = keyPattern(r.Instances)
	}
	return w.Watch(pattern, func(key string, value []byte, current bool) error {
		var rec Record
		if value != nil {
			if err := json.Unmarshal(value, &rec); err != nil {
				return fmt.Errorf("the state of %s is not a record: %w", key, err)
			}
		}
		state := rec.State
		if len(state) == 0 {
			state = json.RawMessage("null")
		}
		return deliver(buscalls.InstanceChange{Watch: r.Watch, Instance: InstanceOf(key), State: state, Sequence: rec.Sequence,
			Current: current})
	})
}

// keyPattern is an instance pattern as a key pattern: its path's parts joined by dots.
func keyPattern(instances string) string {
	out := []byte(instances)
	for i, c := range out {
		if c == '/' {
			out[i] = '.'
		}
	}
	return string(out)
}

// index is a declared index by its name, or the reason there is none.
func (l *Ledger) index(name string) (Index, error) {
	if ix, ok := l.Indexes[name]; ok {
		return ix, nil
	}
	return nil, fmt.Errorf("%s's ledger %s declares no index %q", l.Module, l.Name, name)
}

// Query is a declared index's entries whose key Key names.
func (l *Ledger) Query(r buscalls.QueryRequest) (buscalls.QueryAnswer, error) {
	ix, err := l.index(r.Index)
	if err != nil {
		return buscalls.QueryAnswer{}, err
	}
	held, err := ix.Entries(r.Key, r.Limit)
	if err != nil {
		return buscalls.QueryAnswer{}, err
	}
	entries := make([]buscalls.IndexEntry, 0, len(held))
	for _, h := range held {
		var e buscalls.IndexEntry
		if err := json.Unmarshal(h.Value, &e); err != nil {
			return buscalls.QueryAnswer{}, fmt.Errorf("the index %s's entry %s is not one: %w", r.Index, h.Key, err)
		}
		entries = append(entries, e)
	}
	return buscalls.QueryAnswer{Entries: entries}, nil
}