jschoubben
28c3b27fec
ADR 0615 §1: sdk/firstview is a process's picture of an owner's state, read as the bucket it replaces (Get, Keys, Watch; every write refused), asked through <owner>.<short>_read_<state> with values and kept from the owner's views family; firstviewtest serves an owner in a whole-module test. The teller declares invokes assigner_read_asks, consumes views.asks, states first-views and findings, and the tools teller_first_views and teller_first_view_again. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
110 lines
4.0 KiB
Go
110 lines
4.0 KiB
Go
// Package firstviewtest is an owner of a state as a reader's whole-module test needs it (novox/hq ADR 0615): the
|
|
// owner's read tool answering the first view from its bucket at its mark, and its views family said by the real
|
|
// views emitter onto its views stream, which the reader's durable consumer reads.
|
|
package firstviewtest
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"git.novox.be/novox/mesh-sdk/go/views"
|
|
|
|
"git.novox.be/novox/mesh-delivery/sdk/firstview"
|
|
)
|
|
|
|
// Serve is owner's state on the bus with the owner's side of it: the bucket `<owner>_<state>` (made when absent, history
|
|
// 1), the owner's marks, its views stream with a durable consumer per reader (`<node>_<module>`), the emitter saying
|
|
// each change, and the read tool (firstview.ReadTool) answering on nc. It returns the bucket the test writes.
|
|
func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state string, readers ...string) jetstream.KeyValue {
|
|
t.Helper()
|
|
bucket := owner + "_" + state
|
|
kv, err := js.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
if kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
marks, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: owner + "_" + views.MarksState, History: 1})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stream := firstview.ViewsStream(owner)
|
|
if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{process.ModuleStreamSubject(owner, views.EventPrefix+">")}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, r := range readers {
|
|
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: r, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
em := views.Emitter{Owner: owner, Subject: func(e string) string { return process.ModuleStreamSubject(owner, e) },
|
|
Publish: views.JetStreamPublish(js), Marks: marks}
|
|
if err := em.Start(ctx, state, kv, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
tool := strings.TrimPrefix(firstview.ReadTool(owner, state), owner+".")
|
|
sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) {
|
|
var in struct {
|
|
After string `json:"after"`
|
|
}
|
|
_ = json.Unmarshal(m.Data, &in)
|
|
a, err := views.FirstView(owner, state, Source{ctx, kv}, Source{ctx, marks}, in.After, time.Now())
|
|
out := map[string]any{"result": a, "node": "n0"}
|
|
if err != nil {
|
|
out = map[string]any{"error": err.Error()}
|
|
}
|
|
raw, _ := json.Marshal(out)
|
|
_ = m.Respond(raw)
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
|
return kv
|
|
}
|
|
|
|
// Grants are the subjects a reader's account publishes to follow owner's state (its `invokes` and `consumes`): the
|
|
// read tool's call subjects, its durable consumer's info, next and acknowledgements on the owner's views stream.
|
|
func Grants(owner, state, durable string) []string {
|
|
tool := strings.TrimPrefix(firstview.ReadTool(owner, state), owner+".")
|
|
stream := firstview.ViewsStream(owner)
|
|
return []string{"mesh.mod." + owner + ".call." + tool + ".>", "mesh.mod." + owner + ".call." + tool,
|
|
"$JS.API.CONSUMER.INFO." + stream + "." + durable, "$JS.API.CONSUMER.MSG.NEXT." + stream + "." + durable,
|
|
"$JS.ACK." + stream + "." + durable + ".>"}
|
|
}
|
|
|
|
// Source is a bucket as a views.Source.
|
|
type Source struct {
|
|
Ctx context.Context
|
|
KV jetstream.KeyValue
|
|
}
|
|
|
|
// Keys is every key; none is no error.
|
|
func (s Source) Keys() ([]string, error) {
|
|
keys, err := s.KV.Keys(s.Ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return nil, nil
|
|
}
|
|
return keys, err
|
|
}
|
|
|
|
// Get is one key's value and revision.
|
|
func (s Source) Get(key string) (*views.Entry, bool, error) {
|
|
e, err := s.KV.Get(s.Ctx, key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, false, nil
|
|
}
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
return &views.Entry{Value: e.Value(), Revision: e.Revision()}, true, nil
|
|
}
|