// 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 `_` (made when absent, history // 1), the owner's marks, its views stream with a durable consumer per reader (`_`), 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 }