diff --git a/node-tools/internal/bus/ledger.go b/node-tools/internal/bus/ledger.go index 5591379..417c64d 100644 --- a/node-tools/internal/bus/ledger.go +++ b/node-tools/internal/bus/ledger.go @@ -10,7 +10,6 @@ import ( "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" @@ -290,7 +289,7 @@ func (c *Conn) LedgerIndexes(b membership.BusBinding) map[string]ledger.Index { return out } -func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) { +func (s *busState) Entries(pattern string, limit int) ([]ledger.Indexed, error) { if pattern == "" { pattern = ">" } @@ -299,7 +298,7 @@ func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, er return nil, err } sort.Strings(keys) - var out []buscalls.IndexEntry + var out []ledger.Indexed for _, k := range keys { if limit > 0 && len(out) >= limit { break @@ -312,7 +311,7 @@ func (s *busState) Entries(pattern string, limit int) ([]buscalls.IndexEntry, er return out, err } if v != nil { - out = append(out, buscalls.IndexEntry{Key: k, Value: valueOf(v)}) + out = append(out, ledger.Indexed{Key: k, Value: v}) } } return out, nil diff --git a/node-tools/internal/ledger/memory.go b/node-tools/internal/ledger/memory.go index 240a225..0b14670 100644 --- a/node-tools/internal/ledger/memory.go +++ b/node-tools/internal/ledger/memory.go @@ -1,12 +1,9 @@ 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 @@ -166,7 +163,7 @@ type MemoryIndex struct { kv map[string][]byte } -func (x *MemoryIndex) Entries(pattern string, limit int) ([]buscalls.IndexEntry, error) { +func (x *MemoryIndex) Entries(pattern string, limit int) ([]Indexed, error) { x.mu.Lock() defer x.mu.Unlock() if pattern == "" { @@ -179,12 +176,12 @@ func (x *MemoryIndex) Entries(pattern string, limit int) ([]buscalls.IndexEntry, } } sort.Strings(keys) - var out []buscalls.IndexEntry + var out []Indexed for _, k := range keys { if limit > 0 && len(out) >= limit { break } - out = append(out, buscalls.IndexEntry{Key: k, Value: json.RawMessage(x.kv[k])}) + out = append(out, Indexed{Key: k, Value: x.kv[k]}) } return out, nil } diff --git a/node-tools/internal/ledger/read.go b/node-tools/internal/ledger/read.go index 08c2e1c..87f4cea 100644 --- a/node-tools/internal/ledger/read.go +++ b/node-tools/internal/ledger/read.go @@ -16,13 +16,19 @@ import ( // 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) + // (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) @@ -144,12 +150,17 @@ func (l *Ledger) Query(r buscalls.QueryRequest) (buscalls.QueryAnswer, error) { if err != nil { return buscalls.QueryAnswer{}, err } - entries, err := ix.Entries(r.Key, r.Limit) + held, err := ix.Entries(r.Key, r.Limit) if err != nil { return buscalls.QueryAnswer{}, err } - if entries == nil { - entries = []buscalls.IndexEntry{} + 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 } diff --git a/node-tools/internal/ledger/read_test.go b/node-tools/internal/ledger/read_test.go index 2539c63..755ca70 100644 --- a/node-tools/internal/ledger/read_test.go +++ b/node-tools/internal/ledger/read_test.go @@ -47,10 +47,10 @@ func TestReadsAndRepairsOfALedger(t *testing.T) { } }}, {"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"}`)) + _ = f.index.Put("One.decision.1", []byte(`{"instance":"decision/1","sequence":1}`)) + _ = f.index.Put("Two.decision.2", []byte(`{"instance":"decision/2","sequence":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"}` { + if err != nil || len(got.Entries) != 1 || got.Entries[0] != (buscalls.IndexEntry{Instance: "decision/1", Sequence: 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"`) { @@ -82,7 +82,7 @@ func TestReadsAndRepairsOfALedger(t *testing.T) { } }}, {"reindex builds a declared index from the states, and lag says an index behind until it does", func(t *testing.T, l *Ledger, f fakes) { - applied(t, l, "decision/1", "title", `{"title":"One ledger"}`, owner) + applied(t, l, "decision/1", "title", `{"title":"One"}`, owner) applied(t, l, "decision/2", "note", `{"says":"untitled"}`, owner) _ = f.index.Put("stale.decision.9", []byte(`{}`)) if lag, _ := l.Lag(buscalls.LedgerRequest{}); len(lag.Indexes) != 1 || lag.Indexes[0].Behind != 1 { @@ -92,7 +92,7 @@ func TestReadsAndRepairsOfALedger(t *testing.T) { t.Fatalf("reindexed %+v, %v", got, err) } entries, _ := f.index.Entries("", 0) - if len(entries) != 1 || entries[0].Key != "One_ledger.decision.1" || string(entries[0].Value) != `{"instance":"decision/1","sequence":1}` { + if len(entries) != 1 || entries[0].Key != buscalls.IndexKey([]string{"One"}, "decision/1") || string(entries[0].Value) != `{"instance":"decision/1","sequence":1}` { t.Fatalf("the index holds %+v", entries) } if lag, _ := l.Lag(buscalls.LedgerRequest{}); len(lag.Indexes) != 0 || len(lag.Instances) != 0 { diff --git a/node-tools/internal/ledger/repair.go b/node-tools/internal/ledger/repair.go index e0d2803..7221d79 100644 --- a/node-tools/internal/ledger/repair.go +++ b/node-tools/internal/ledger/repair.go @@ -3,7 +3,6 @@ package ledger import ( "encoding/json" "fmt" - "regexp" "sort" "strings" @@ -130,41 +129,33 @@ func (l *Ledger) Snapshot(r buscalls.InstanceRequest) (buscalls.Repaired, error) return out, nil } -// unsafe is what a key part cannot hold. -var unsafe = regexp.MustCompile(`[^-_A-Za-z0-9]`) - -// IndexEntryOf is an instance's entry in a declared index, as reindex writes it: its key the values of the fields the -// index is on, in their order, then the instance's key, each part with what a key cannot hold written as `_`; its value -// the instance and the sequence of the state it was taken from. False when the state holds none of those fields. -func IndexEntryOf(on []string, key string, rec Record) (string, []byte, bool) { +// indexEntryOf is an instance's entry in a declared index, as reindex writes it and the owner's watcher keeps it +// (mesh-sdk buscalls.IndexKey, buscalls.IndexEntry): its key the values of the fields the index is on, in their order, +// then the instance; its value the instance and the sequence of the state it was taken from. False when the state holds +// none of those fields. A value that is not text is written as its JSON. +func indexEntryOf(on []string, key string, rec Record) (string, []byte, bool) { var state map[string]any if json.Unmarshal(rec.State, &state) != nil || state == nil { return "", nil, false } - parts := make([]string, 0, len(on)+1) + values := make([]string, 0, len(on)) found := false for _, f := range on { v, has := state[f] - if !has { - parts = append(parts, "_") - continue - } - found = true + found = found || has s, ok := v.(string) if !ok { raw, _ := json.Marshal(v) s = string(raw) } - if s = unsafe.ReplaceAllString(s, "_"); s == "" { - s = "_" - } - parts = append(parts, s) + values = append(values, s) } if !found { return "", nil, false } - value, _ := json.Marshal(map[string]any{"instance": InstanceOf(key), "sequence": rec.Sequence}) - return strings.Join(append(parts, key), "."), value, true + instance := InstanceOf(key) + value, _ := json.Marshal(buscalls.IndexEntry{Instance: instance, Sequence: rec.Sequence}) + return buscalls.IndexKey(values, instance), value, true } // declared is the fields an index is on and the types declaring it. @@ -205,7 +196,7 @@ func (l *Ledger) wanted(index string) (map[string][]byte, map[string]uint64, err if err != nil { return nil, nil, err } - if k, v, ok := IndexEntryOf(on, key, held); ok { + if k, v, ok := indexEntryOf(on, key, held); ok { want[k], seqs[k] = v, held.Sequence } } @@ -295,9 +286,7 @@ func (l *Ledger) Lag(buscalls.LedgerRequest) (buscalls.Lagged, error) { } at := map[string]uint64{} for _, e := range held { - var v struct { - Sequence uint64 `json:"sequence"` - } + var v buscalls.IndexEntry _ = json.Unmarshal(e.Value, &v) at[e.Key] = v.Sequence }