mesh-sdk/go/views/writer_test.go

package views_test

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"runtime"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/views"
)

// fakeState is an observer's state `findings` in memory.
type fakeState struct {
	mu sync.Mutex
	m  map[string][]byte
}

func (s *fakeState) Get(key string) (*stdio.StateEntry, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if v, ok := s.m[key]; ok {
		return &stdio.StateEntry{Key: key, Value: v}, nil
	}
	return nil, nil
}
func (s *fakeState) Put(key string, value any) (uint64, error) {
	raw, _ := json.Marshal(value)
	s.mu.Lock()
	defer s.mu.Unlock()
	s.m[key] = raw
	return 1, nil
}
func (s *fakeState) Delete(key string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.m, key)
	return nil
}
func (s *fakeState) Keys() ([]string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var k []string
	for key := range s.m {
		k = append(k, key)
	}
	return k, nil
}

type said struct {
	subject string
	data    []byte
}

type rig struct {
	t       *testing.T
	state   *fakeState
	now     time.Time
	mu      sync.Mutex
	sent    []said
	fail    int // publishes refused before one is taken
	waits   []time.Duration
	logs    []string
	observe func(context.Context) (views.Observed, error)
	handle  func([]byte)
	listens int
	stopped int
	w       *views.Writer
	ch      chan struct{}
}

var subjects = views.WriterSubjects{
	Written: "mesh.stream.m.event.finding-written.n1", Cleared: "mesh.stream.m.event.finding-cleared.n1",
	Stated: "mesh.stream.m.event.findings-stated.n1", Asked: "mesh.seat.s.event.findings-asked",
}

func newRig(t *testing.T) *rig { return newRigKinds(t) }

func newRigKinds(t *testing.T, kinds ...string) *rig {
	r := &rig{t: t, state: &fakeState{m: map[string][]byte{}}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), ch: make(chan struct{}, 100)}
	r.observe = func(context.Context) (views.Observed, error) { return views.Observed{}, nil }
	w, err := views.NewWriter(views.WriterConfig{
		Module: "m", State: r.state, Subjects: subjects, Kinds: kinds,
		Observe: func(ctx context.Context) (views.Observed, error) { return r.observe(ctx) },
		Logf: func(f string, a ...any) {
			r.mu.Lock()
			r.logs = append(r.logs, fmt.Sprintf(f, a...))
			r.mu.Unlock()
		},
		Now: func() time.Time { r.mu.Lock(); defer r.mu.Unlock(); return r.now },
		Wait: func(ctx context.Context, d time.Duration) bool {
			r.mu.Lock()
			defer r.mu.Unlock()
			r.waits = append(r.waits, d)
			r.now = r.now.Add(d)
			return ctx.Err() == nil
		},
		Publish: func(_ context.Context, subject string, data []byte, _ string) error {
			r.mu.Lock()
			defer r.mu.Unlock()
			if r.fail > 0 {
				r.fail--
				return errors.New("no ack")
			}
			r.sent = append(r.sent, said{subject, data})
			r.ch <- struct{}{}
			return nil
		},
		Listen: func(_ context.Context, subject string, handle func([]byte)) (func(), error) {
			r.mu.Lock()
			defer r.mu.Unlock()
			if subject != subjects.Asked {
				t.Errorf("listening on %s", subject)
			}
			r.listens++
			r.handle = handle
			return func() { r.mu.Lock(); r.stopped++; r.mu.Unlock() }, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	r.w = w
	return r
}

func (r *rig) events() []said {
	r.mu.Lock()
	defer r.mu.Unlock()
	return slices.Clone(r.sent)
}

func finding(id, summary string) views.FindingBody {
	return views.FindingBody{Kind: "k", Scope: "module", ID: id, Severity: "warning", Summary: summary, Source: "probe",
		Subject: views.FindingSubject{Scope: "module", ID: id}}
}

// A finding written says its event with the layout whole after the key, keeps the key's since, and says again only when
// what it says changed; a finding cleared says its event, also of a key that does not exist (design 47 G §383, §485).
func TestWriterWritesAndClearsFindingsAsEvents(t *testing.T) {
	r := newRig(t)
	ctx := context.Background()
	first := r.now
	if err := r.w.Write(ctx, finding("a", "one")); err != nil {
		t.Fatal(err)
	}
	r.now = r.now.Add(time.Hour)
	f := finding("a", "one")
	f.Run = "other run"
	if err := r.w.Write(ctx, f); err != nil { // nothing said changed
		t.Fatal(err)
	}
	if err := r.w.Write(ctx, finding("a", "two")); err != nil {
		t.Fatal(err)
	}
	ev := r.events()
	if len(ev) != 2 || ev[0].subject != subjects.Written || ev[1].subject != subjects.Written {
		t.Fatalf("events: %+v", ev)
	}
	var got views.FindingBody
	if err := json.Unmarshal(ev[1].data, &got); err != nil {
		t.Fatal(err)
	}
	if got.Summary != "two" || !got.Since.Equal(first) || got.Observer != "m" || got.Key() != "module.a.k" || !got.At.Equal(r.now) {
		t.Fatalf("body: %+v", got)
	}
	if e, _ := r.state.Get("module.a.k"); e == nil {
		t.Fatal("the key was not written")
	}
	if err := r.w.Clear(ctx, "module", "a", "k", "gone"); err != nil {
		t.Fatal(err)
	}
	if err := r.w.Clear(ctx, "module", "never", "k", "held by the keeper"); err != nil {
		t.Fatal(err)
	}
	ev = r.events()
	var cl views.FindingCleared
	if len(ev) != 4 || ev[3].subject != subjects.Cleared || json.Unmarshal(ev[3].data, &cl) != nil || cl.ID != "never" || cl.Why != "held by the keeper" {
		t.Fatalf("clear events: %+v", ev)
	}
	if e, _ := r.state.Get("module.a.k"); e != nil {
		t.Fatal("the key was not deleted")
	}
}

// An event the bus does not acknowledge is sent again for one minute, then written in the journal as an error naming the
// finding (design 47 G §383).
func TestWriterSendsAgainForAMinuteThenLogs(t *testing.T) {
	r := newRig(t)
	r.fail = 2
	if err := r.w.Write(context.Background(), finding("a", "one")); err != nil || len(r.events()) != 1 {
		t.Fatalf("resent: %v %+v", err, r.events())
	}
	r.fail = 1000
	start := r.now
	if err := r.w.Write(context.Background(), finding("b", "one")); err != nil {
		t.Fatal(err)
	}
	if r.now.Sub(start) < views.PublishRetryFor || r.now.Sub(start) > views.PublishRetryFor+10*time.Second {
		t.Fatalf("resent for %s", r.now.Sub(start))
	}
	if len(r.logs) != 1 || !strings.Contains(r.logs[0], "module.b.k") || !strings.Contains(r.logs[0], "error") {
		t.Fatalf("journal: %v", r.logs)
	}
}

// A start states the findings observed now, writes their keys, deletes the keys neither stated nor pending, and pends the
// rest; an ask naming another module is not answered, one naming none or this one is, with its id; an observe that fails
// says no set and is called again once a minute until a set is said; a connection made again makes the consumer again and
// states again (design 47 G §384, §461, §485).
func TestWriterStatesAtStartAndOnAsksAndRecovers(t *testing.T) {
	r := newRig(t)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	r.state.Put("module.old.k", finding("old", "x"))
	r.state.Put("module.wait.k", finding("wait", "x"))
	first := r.now
	r.state.Put("module.a.k", func() views.FindingBody { f := finding("a", "before"); f.Since = first.Add(-time.Hour); return f }())
	calls := 0
	r.observe = func(context.Context) (views.Observed, error) {
		calls++
		if calls <= 2 {
			return views.Observed{}, errors.New("not yet")
		}
		return views.Observed{Findings: []views.FindingBody{finding("a", "now")},
			Pending: []views.PendingKey{{Key: "module.wait.k", Since: first, Bound: "5m"}}}, nil
	}
	if err := r.w.Start(ctx); err != nil {
		t.Fatal(err)
	}
	defer r.w.Close()
	stated := waitEvent(t, r, 1)
	if calls != 3 || len(r.waits) != 2 || r.waits[0] != views.ObserveRetryEvery {
		t.Fatalf("observe called %d times, waits %v", calls, r.waits)
	}
	var st views.FindingsStated
	if err := json.Unmarshal(stated.data, &st); err != nil {
		t.Fatal(err)
	}
	if stated.subject != subjects.Stated || st.Asked != "" || len(st.Findings) != 1 || st.Findings[0].Summary != "now" ||
		!st.Findings[0].Since.Equal(first.Add(-time.Hour)) || len(st.Pending) != 1 || st.Pending[0].Key != "module.wait.k" {
		t.Fatalf("statement: %s", stated.data)
	}
	keys, _ := r.state.Keys()
	slices.Sort(keys)
	if !slices.Equal(keys, []string{"module.a.k", "module.wait.k"}) {
		t.Fatalf("keys: %v", keys)
	}
	r.handle([]byte(`{"ask":"a1","at":"2026-11-11T12:00:00Z","by":"keeper","module":"other"}`))
	r.handle([]byte(`{"ask":"a2","at":"2026-11-11T12:00:00Z","by":"keeper","module":"m"}`))
	st = FindingsStatedOf(t, waitEvent(t, r, 2))
	if st.Asked != "a2" || len(r.events()) != 2 {
		t.Fatalf("ask answered: %+v %d", st, len(r.events()))
	}
	r.w.Reconnected()
	waitEvent(t, r, 3)
	if r.listens != 2 || r.stopped != 1 {
		t.Fatalf("consumers made %d, stopped %d", r.listens, r.stopped)
	}
}

func FindingsStatedOf(t *testing.T, e said) views.FindingsStated {
	var st views.FindingsStated
	if err := json.Unmarshal(e.data, &st); err != nil {
		t.Fatal(err)
	}
	return st
}

// waitEvent is the n-th event said, woken by each event said (nothing sleeps for time).
func waitEvent(t *testing.T, r *rig, n int) said {
	t.Helper()
	for len(r.events()) < n {
		select {
		case <-r.ch:
		case <-time.After(10 * time.Second):
			t.Fatalf("event %d never said", n)
		}
	}
	return r.events()[n-1]
}

// A write asked during a statement waits for it, so the statement never deletes the key the write made (design 47 G §485).
func TestWriterRunsOneActAtATime(t *testing.T) {
	r := newRig(t)
	inside, release := make(chan struct{}), make(chan struct{})
	r.observe = func(context.Context) (views.Observed, error) {
		close(inside)
		<-release
		return views.Observed{}, nil
	}
	done := make(chan error, 1)
	go func() { done <- r.w.State(context.Background(), "") }()
	<-inside
	wrote, started := make(chan error, 1), make(chan struct{})
	go func() { close(started); wrote <- r.w.Write(context.Background(), finding("a", "one")) }()
	<-started
	for range 1000 {
		runtime.Gosched()
	}
	close(release)
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	if err := <-wrote; err != nil {
		t.Fatal(err)
	}
	if e, _ := r.state.Get("module.a.k"); e == nil {
		t.Fatal("the statement deleted the key the write made")
	}
}

// A writer owns its kinds (design 47 G §580), states them, deletes no key of another kind, and covers a views reader's
// findings written through it beside observe's (§639); an owner or waiting-on of no form is refused, naming the field (§678).
func TestWriterKindsAndReaderFindings(t *testing.T) {
	ctx := context.Background()
	r := newRigKinds(t, "k")
	other := finding("x", "another bundle's")
	other.Kind = "other"
	r.state.m[other.Key()] = []byte(`{}`)
	if err := r.w.Write(ctx, other); err == nil || !strings.Contains(err.Error(), `"other"`) {
		t.Errorf("a write of another kind: %v", err)
	}
	if err := r.w.Clear(ctx, "module", "x", "other", "why"); err == nil || !strings.Contains(err.Error(), `"other"`) {
		t.Errorf("a clear of another kind: %v", err)
	}
	for field, f := range map[string]views.FindingBody{"owner": {Owner: "Not A Form"}, "waiting-on": {WaitingOn: "x y"}} {
		b := finding("o", "s")
		b.Owner, b.WaitingOn = f.Owner, f.WaitingOn
		if err := r.w.Write(ctx, b); err == nil || !strings.Contains(err.Error(), field) {
			t.Errorf("%s of no form: %v", field, err)
		}
	}
	for _, v := range []string{"me@node-1", "mesh-catalog", "forge:jochen"} {
		b := finding("ok", "s")
		b.Owner = v
		if err := r.w.Write(ctx, b); err != nil {
			t.Errorf("owner %q: %v", v, err)
		}
	}
	// a views reader writes through the part's writer, and the statement keeps its open findings
	rf := views.WriterFindings{Writer: r.w, Body: func(f views.Finding) views.FindingBody {
		return views.FindingBody{Kind: f.Kind, Scope: "module", ID: f.ID, Severity: "warning", Summary: f.Summary, Source: "views"}
	}}
	if err := rf.Put(views.Finding{Kind: views.ViewWaitsKind, ID: "m.o.s", Summary: "waits"}); err != nil {
		t.Fatal(err)
	}
	r.observe = func(context.Context) (views.Observed, error) {
		return views.Observed{Findings: []views.FindingBody{finding("a", "observed")}}, nil
	}
	if err := r.w.State(ctx, ""); err != nil {
		t.Fatal(err)
	}
	if _, ok := r.state.m["module.m.o.s.view-waits"]; !ok {
		t.Errorf("the statement deleted the reader's finding: %v", r.state.m)
	}
	if _, ok := r.state.m[other.Key()]; !ok {
		t.Errorf("the statement deleted another bundle's key")
	}
	if _, ok := r.state.m["module.ok.k"]; ok {
		t.Errorf("a key of its own kind neither stated nor pending stays")
	}
	var stated views.FindingsStated
	for _, e := range r.events() {
		if e.subject == subjects.Stated {
			_ = json.Unmarshal(e.data, &stated)
		}
	}
	if !slices.Contains(stated.Kinds, "k") || !slices.Contains(stated.Kinds, views.ViewWaitsKind) || len(stated.Findings) != 2 {
		t.Errorf("the statement: kinds %v, %d findings", stated.Kinds, len(stated.Findings))
	}
	if err := rf.Delete(views.Finding{Kind: views.ViewWaitsKind, ID: "m.o.s"}); err != nil {
		t.Fatal(err)
	}
	if _, ok := r.state.m["module.m.o.s.view-waits"]; ok {
		t.Errorf("the reader's finding was not cleared")
	}
}