Read a ledger: an instance's state, its history oldest first with sequence and author, a watch of its states, a declared index (issue 1966 T3)

node-tools/internal/bus/ledger.go

@@ -5,9 +5,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
+	"time"
 
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
 	"git.novox.be/novox/mesh-sdk/go/membership"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -227,3 +230,114 @@ func (s *busSnapshots) Put(key string, value []byte, expect uint64) error {
 	_, err := publishExpecting(s.write, s.prefix+"."+key, value, expect)
 	return err
 }
+
+// Watch hands deliver every instance's record under a key pattern as it is, then each change; a removed key is handed as
+// nil.
+func (s *busState) Watch(pattern string, deliver func(key string, value []byte, current bool) error) (func(), error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	kv, err := s.kv(ctx)
+	cancel()
+	if err != nil {
+		return nil, err
+	}
+	w, err := kv.Watch(context.Background(), pattern)
+	if err != nil {
+		return nil, err
+	}
+	hand := func(e jetstream.KeyValueEntry, current bool) {
+		var value []byte
+		if e.Operation() == jetstream.KeyValuePut {
+			value = e.Value()
+		}
+		if err := deliver(e.Key(), value, current); err != nil {
+			s.c.Logf("[mesh-tools] a ledger watch of %s was not taken: %v", s.bucket, err)
+		}
+	}
+	timer := time.NewTimer(StateTimeout)
+	defer timer.Stop()
+	for current := true; current; {
+		select {
+		case e, ok := <-w.Updates():
+			if !ok {
+				return nil, errors.New("the ledger's state watch ended before its current states were handed over")
+			}
+			if e == nil {
+				current = false
+				continue
+			}
+			hand(e, true)
+		case <-timer.C:
+			_ = w.Stop()
+			return nil, fmt.Errorf("the ledger's state %s did not hand its current states in %s", s.bucket, StateTimeout)
+		}
+	}
+	go func() {
+		for e := range w.Updates() {
+			if e != nil {
+				hand(e, false)
+			}
+		}
+	}()
+	return func() { _ = w.Stop() }, nil
+}
+
+// LedgerIndexes are the buckets of a ledger binding's declared indexes, by index name, on the runner's own connection.
+func (c *Conn) LedgerIndexes(b membership.BusBinding) map[string]ledger.Index {
+	out := map[string]ledger.Index{}
+	for name, bucket := range b.Indexes {
+		out[name] = &busState{c: c, bucket: bucket}
+	}
+	return out
+}
+
+func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) {
+	if pattern == "" {
+		pattern = ">"
+	}
+	keys, err := s.Keys()
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(keys)
+	var out []buscalls.IndexEntry
+	for _, k := range keys {
+		if limit > 0 && len(out) >= limit {
+			break
+		}
+		if !SubjectMatches(pattern, k) {
+			continue
+		}
+		v, _, err := s.Get(k)
+		if err != nil {
+			return out, err
+		}
+		if v != nil {
+			out = append(out, buscalls.IndexEntry{Key: k, Value: valueOf(v)})
+		}
+	}
+	return out, nil
+}
+
+func (s *busState) Put(key string, value []byte) error {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := s.kv(ctx)
+	if err != nil {
+		return err
+	}
+	_, err = kv.Put(ctx, key, value)
+	return err
+}
+
+func (s *busState) Delete(key string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := s.kv(ctx)
+	if err != nil {
+		return err
+	}
+	if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+		return err
+	}
+	return nil
+}

node-tools/internal/ledger/ledger.go

@@ -87,6 +87,8 @@ type Ledger struct {
 	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.

node-tools/internal/ledger/memory.go

@@ -1,9 +1,12 @@
 package ledger
 
 import (
+	"encoding/json"
 	"sort"
 	"strings"
 	"sync"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
 )
 
 // The memory stores keep what the bus keeps, as the bus answers it, for a test of a ledger without a bus: a stream whose
@@ -156,3 +159,49 @@ func (s *MemorySnapshots) Put(key string, value []byte, expect uint64) error {
 	s.kv[key] = memoryValue{value, s.at}
 	return nil
 }
+
+// MemoryIndex is an index's bucket in memory.
+type MemoryIndex struct {
+	mu sync.Mutex
+	kv map[string][]byte
+}
+
+func (x *MemoryIndex) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) {
+	x.mu.Lock()
+	defer x.mu.Unlock()
+	if pattern == "" {
+		pattern = ">"
+	}
+	var keys []string
+	for k := range x.kv {
+		if Matches(pattern, k) {
+			keys = append(keys, k)
+		}
+	}
+	sort.Strings(keys)
+	var out []buscalls.IndexEntry
+	for _, k := range keys {
+		if limit > 0 && len(out) >= limit {
+			break
+		}
+		out = append(out, buscalls.IndexEntry{Key: k, Value: json.RawMessage(x.kv[k])})
+	}
+	return out, nil
+}
+
+func (x *MemoryIndex) Put(key string, value []byte) error {
+	x.mu.Lock()
+	defer x.mu.Unlock()
+	if x.kv == nil {
+		x.kv = map[string][]byte{}
+	}
+	x.kv[key] = value
+	return nil
+}
+
+func (x *MemoryIndex) Delete(key string) error {
+	x.mu.Lock()
+	defer x.mu.Unlock()
+	delete(x.kv, key)
+	return nil
+}

node-tools/internal/ledger/read.go

@@ -0,0 +1,155 @@
+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).
+	Entries(pattern string, limit int) ([]buscalls.IndexEntry, error)
+	// Put writes one entry; Delete removes one.
+	Put(key string, value []byte) error
+	Delete(key string) error
+}
+
+// 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
+	}
+	entries, err := ix.Entries(r.Key, r.Limit)
+	if err != nil {
+		return buscalls.QueryAnswer{}, err
+	}
+	if entries == nil {
+		entries = []buscalls.IndexEntry{}
+	}
+	return buscalls.QueryAnswer{Entries: entries}, nil
+}

node-tools/internal/ledger/read_test.go

@@ -0,0 +1,67 @@
+package ledger
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// 1966/AC15 and the other reads of a ledger: get, query (ADR 0597 §10; design 25 § 12.5).
+func TestReadsOfALedger(t *testing.T) {
+	cases := []struct {
+		name string
+		run  func(t *testing.T, l *Ledger, f fakes)
+	}{
+		{"1966/AC15 history answers an instance's operations oldest first with sequence and author", func(t *testing.T, l *Ledger, f fakes) {
+			op := buscalls.Author{Account: "ledger.anchor.m", Caller: "op@shanks"}
+			applied(t, l, "decision/1", "note", `{"says":"one"}`, owner)
+			applied(t, l, "decision/2", "note", `{"says":"elsewhere"}`, owner)
+			applied(t, l, "decision/1", "title", `{"title":"Two"}`, op)
+			applied(t, l, "decision/1", "note", `{"says":"three"}`, owner)
+			page, err := l.ReadHistory(buscalls.HistoryRequest{Instance: "decision/1", Limit: 2})
+			if err != nil {
+				t.Fatal(err)
+			}
+			var got []string
+			for _, o := range page.Operations {
+				got = append(got, fmt.Sprintf("%s@%d %s %s", o.Operation, o.Sequence, o.Author.Account, o.Author.Caller))
+			}
+			if strings.Join(got, "; ") != "note@1 ledger.anchor.m ; title@3 ledger.anchor.m op@shanks" || page.Next != 4 || page.Last != 4 {
+				t.Fatalf("the history is %q, next %d, last %d", got, page.Next, page.Last)
+			}
+			rest, _ := l.ReadHistory(buscalls.HistoryRequest{Instance: "decision/1", From: page.Next})
+			if len(rest.Operations) != 1 || rest.Operations[0].Sequence != 4 || rest.Next != 0 {
+				t.Fatalf("read on: %+v", rest)
+			}
+		}},
+		{"get answers the state and its sequence, null for an instance with none", func(t *testing.T, l *Ledger, f fakes) {
+			applied(t, l, "decision/1", "title", `{"title":"One"}`, owner)
+			got, err := l.Get(buscalls.InstanceRequest{Instance: "decision/1"})
+			if err != nil || got.Sequence != 1 || !strings.Contains(string(got.State), `"title":"One"`) {
+				t.Fatalf("got %+v, %v", got, err)
+			}
+			if none, err := l.Get(buscalls.InstanceRequest{Instance: "decision/9"}); err != nil || string(none.State) != "null" {
+				t.Fatalf("none is %+v, %v", none, err)
+			}
+		}},
+		{"query reads a declared index by key, and an undeclared one is refused", func(t *testing.T, l *Ledger, f fakes) {
+			_ = f.index.Put("One.decision.1", []byte(`{"instance":"decision/1"}`))
+			_ = f.index.Put("Two.decision.2", []byte(`{"instance":"decision/2"}`))
+			got, err := l.Query(buscalls.QueryRequest{Index: "by-title", Key: "One.>"})
+			if err != nil || len(got.Entries) != 1 || string(got.Entries[0].Value) != `{"instance":"decision/1"}` {
+				t.Fatalf("queried %+v, %v", got, err)
+			}
+			if _, err := l.Query(buscalls.QueryRequest{Index: "by-date"}); err == nil || !strings.Contains(err.Error(), `no index "by-date"`) {
+				t.Fatalf("an undeclared index: %v", err)
+			}
+		}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			l, f := newLedger()
+			c.run(t, l, f)
+		})
+	}
+}

node-tools/internal/ledger/types_test.go

@@ -29,12 +29,13 @@ type fakes struct {
 	history   *MemoryHistory
 	state     *MemoryState
 	snapshots *MemorySnapshots
+	index     *MemoryIndex
 }
 
 func newLedger() (*Ledger, fakes) {
-	f := fakes{&MemoryHistory{}, &MemoryState{}, &MemorySnapshots{}}
+	f := fakes{&MemoryHistory{}, &MemoryState{}, &MemorySnapshots{}, &MemoryIndex{}}
 	return &Ledger{Module: "m", Name: "records", Types: []catalogue.LedgerType{records}, History: f.history, State: f.state,
-		Snapshots: f.snapshots}, f
+		Snapshots: f.snapshots, Indexes: map[string]Index{"by-title": f.index}}, f
 }
 
 // applied applies an operation and fails the test on a refusal.