mesh-sdk/go/views/viewstest/owner.go

// Package viewstest 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 viewstest

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/catalogue"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/views"
)

// 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 answering on nc: tool is the owner's read tool as a reader's invokes names it
// (`<owner>.<tool>`), stream its views stream as its manifest's `streams` names it, both given as the reader's
// ProcessReader.Follow takes them. The tool answers a key's history too (key with history), from what the bucket keeps:
// a test that reads history makes the bucket first with the history it needs. It returns the bucket the test writes.
func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state, tool, stream 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 = catalogue.StreamBusName(stream)
	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(tool, owner+".")
	sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) {
		var in struct {
			After   string `json:"after"`
			Key     string `json:"key"`
			History bool   `json:"history"`
		}
		_ = json.Unmarshal(m.Data, &in)
		var a any
		var err error
		if in.History {
			a, err = views.ReadHistory(owner, state, Source{ctx, kv}, in.Key)
		} else {
			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, tool, stream, durable string) []string {
	tool = strings.TrimPrefix(tool, owner+".")
	stream = catalogue.StreamBusName(stream)
	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, revision and the time the bus wrote it.
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(), Written: e.Created()}, true, nil
}

// History is every value the bucket keeps for key, oldest first, as the tool runner's mesh/state.history says it
// (a put, or a delete with a null value); none kept is an empty history.
func (s Source) History(key string) ([]views.Revision, error) {
	kept, err := s.KV.History(s.Ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return []views.Revision{}, nil
	}
	if err != nil {
		return nil, err
	}
	out := make([]views.Revision, 0, len(kept))
	for _, e := range kept {
		r := views.Revision{Revision: e.Revision(), At: e.Created().UTC(), Op: "delete", Value: json.RawMessage("null")}
		if e.Operation() == jetstream.KeyValuePut {
			r.Op, r.Value = "put", json.RawMessage(e.Value())
		}
		out = append(out, r)
	}
	return out, nil
}