mesh-sdk/go/views/writer.go

package views

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"os"
	"regexp"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

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

// The finding writer (design 47 G §461, §485, §503): the one writer every observer uses to keep its findings in its own
// state `findings` and to say them as events (§383, §384). It names no seat and no module: the observer hands in the
// subjects of `finding-written`, `finding-cleared`, `findings-stated` and `findings-asked`, which the generated
// contract of the seat that defines them makes from its module name and node (§461, §439).

// The time constants of the writer (design 47 G §383, §485).
const (
	// PublishRetryFor is how long an event the bus did not acknowledge is sent again before it is logged as an error.
	PublishRetryFor = time.Minute
	// ObserveRetryEvery is how often a failing observe is called again until a set is said.
	ObserveRetryEvery = time.Minute
)

// A body is handled in its wire form (design 47 G A-3): the writer is generic over the contract's body types, which stay plain
// structs, and marshals a body, reads and sets `kind`, `scope`, `id`, `subject`, `at`, `since` and `observer` by their
// contract names. wire is that form.
type wire = map[string]any

// wireOf is v in its wire form.
func wireOf(v any) (wire, error) {
	raw, err := json.Marshal(v)
	if err != nil {
		return nil, err
	}
	var m wire
	if err := json.Unmarshal(raw, &m); err != nil {
		return nil, fmt.Errorf("a body is not an object: %w", err)
	}
	return m, nil
}

func str(m wire, field string) string { s, _ := m[field].(string); return s }

// FindingKey is the key of the finding of scope, id and kind: `<scope>.<id>.<kind>` (novox/hq ADR 0564 §2).
func FindingKey(scope, id, kind string) string { return scope + "." + id + "." + kind }

func keyOf(m wire) string { return FindingKey(str(m, "scope"), str(m, "id"), str(m, "kind")) }

// requireFields refuses a contract body whose wire form lacks a field the writer reads or sets (design 47 G A-3).
func requireFields[T any](what string, fields ...string) error {
	var zero T
	m, err := wireOf(&zero)
	if err != nil {
		return fmt.Errorf("the %s body of the contract cannot be read: %w", what, err)
	}
	for _, f := range fields {
		if _, ok := m[f]; !ok {
			return fmt.Errorf("the %s body of the contract has no field %q on its wire form", what, f)
		}
	}
	return nil
}

// ViewKinds are the finding kinds a views reader writes through the part's writer (design 47 G §639).
var ViewKinds = []string{ViewWaitsKind, UnansweredKind, UnsaidKind, TooLargeKind, EntryNotFoundKind}

var (
	sessionForm = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*@[a-z0-9][a-z0-9-]*$`)
	seatForm    = regexp.MustCompile(`^[a-z][a-z0-9-]*$`)
	authorForm  = regexp.MustCompile(`^forge:[A-Za-z0-9._-]+$`)
)

// formOfWho refuses a finding's `owner` or `waiting-on` of none of the three forms (design 47 G §678): a session
// (who@node), a seat's name, or a commit's author (forge:<account>); empty is none.
func formOfWho(field, v string) error {
	if v == "" || sessionForm.MatchString(v) || seatForm.MatchString(v) || authorForm.MatchString(v) {
		return nil
	}
	return fmt.Errorf("the finding's %s %q is none of a session (who@node), a seat's name or a commit's author (forge:<account>)", field, v)
}

// PendingKey is a key the observer cannot judge yet since its start: its `since` and its bound (a duration such as 5m).
type PendingKey struct {
	Key   string    `json:"key"`
	Since time.Time `json:"since"`
	Bound string    `json:"bound"`
}

// Observed is what an observer's own truth says now: the findings it observes and the keys it cannot judge yet.
type Observed[F any] struct {
	Findings []F
	Pending  []PendingKey
}

// FindingState is the observer's own state `findings` (stdio.State and process.State both are one).
type FindingState interface {
	Get(key string) (*stdio.StateEntry, error)
	Put(key string, value any) (uint64, error)
	Delete(key string) error
	Keys() ([]string, error)
}

// WriterSubjects are the subjects of the three events an observer's module says on its own stream.
type WriterSubjects struct {
	Written, Cleared, Stated string
}

// FindingSubjects are the subjects of the module's three own events on node, in §383's form
// `mesh.stream.<module>.event.<event>.<node>` (design 47 G A-13, §461, §439, §383). The writer forms them itself; a module
// passes none. The fourth, the subject of the ask it hears, is the seat's and is given (WriterConfig.Asked, A-17).
func FindingSubjects(module, node string) WriterSubjects {
	of := func(event string) string { return "mesh.stream." + module + ".event." + event + "." + node }
	return WriterSubjects{Written: of("finding-written"), Cleared: of("finding-cleared"), Stated: of("findings-stated")}
}

// FindingEventsLocal is the local name of the `events` template a part requires for its finding stream (design 47 G §529, A-30).
const FindingEventsLocal = "finding-events"

// FindingBindings is the part's bus bindings as the writer reads them (process.Binding is one): Asks says whether the part
// requires the entry of that local name, Names gives its names on the bus.
type FindingBindings interface {
	Asks(local string) bool
	Names(local string) (membership.BusBinding, error)
}

// subject is where the writer says event: under the binding `finding-events` when the process was handed one, whose subject
// is `mesh.bus.<module>.finding-events` and each message `<subject>.<event>.<node>`; with none, on the old stream
// (FindingSubjects). A binding not answered yet is an error, so the event is sent again until it is (design 47 G A-274, §529).
func (w *Writer[F, C, S, A]) subject(event string) (string, error) {
	if w.cfg.Bindings != nil && w.cfg.Bindings.Asks(FindingEventsLocal) {
		b, err := w.cfg.Bindings.Names(FindingEventsLocal)
		if err != nil {
			return "", err
		}
		if b.Subject == "" {
			return "", fmt.Errorf("the bus entry %q of %s names no subject", FindingEventsLocal, w.cfg.Module)
		}
		return b.Subject + "." + event + "." + w.cfg.Node, nil
	}
	switch event {
	case "finding-written":
		return w.subj.Written, nil
	case "finding-cleared":
		return w.subj.Cleared, nil
	}
	return w.subj.Stated, nil
}

// WriterConfig makes a Writer.
type WriterConfig[F, C, S, A any] struct {
	// Module is the observer's module name: it fills the observer of a finding and tells an ask for another module.
	Module string
	// State is the observer's own state `findings`.
	State FindingState
	// Kinds are the finding kinds this writer owns (design 47 G §580): it refuses a Write or a Clear of another kind,
	// naming it, its statement carries them as `kinds` and deletes only keys of them. None given, it owns every kind.
	// The kinds of a views reader (ViewKinds) are the reader's own and are always the writer's besides.
	Kinds []string
	// JS is the bus connection's JetStream; Conn, when given, lets the writer hear the connection made again.
	JS   jetstream.JetStream
	Conn *nats.Conn
	// Bindings are the part's bus bindings: when it asks the entry `finding-events` (an `events` template its manifest
	// requires) the writer says its events under that binding's subjects, else on the old stream (design 47 G A-274); a
	// module passes no subjects. nil is the old stream.
	Bindings FindingBindings
	// Asked is the subject the writer hears `findings-asked` on: the seat's event, which the module fills from the generated
	// contract of the seat that defines it (design 47 G A-17); mesh-sdk names no seat.
	Asked string
	// Node is the machine the module runs on: it forms the subjects of the four events (FindingSubjects) and a views
	// reader's findings name it in their subject (design 47 G §687, A-13); none given, MESH_NODE.
	Node string
	// Observe answers the findings the observer observes now and its pending keys, or an error. It is called under the
	// writer's one lock, so it must not call Write or Clear.
	Observe func(ctx context.Context) (Observed[F], error)
	// Send puts a set larger than one bus message in the observer's sent store and answers what to publish instead
	// (stateobjects.Sender.Send); nil sends the body as it is.
	Send func(label string, body []byte) ([]byte, error)
	// Logf is the observer's journal.
	Logf func(string, ...any)
	// Now is the clock, Wait pauses for d and answers false when ctx ended; nil are the real ones.
	Now  func() time.Time
	Wait func(ctx context.Context, d time.Duration) bool
	// Publish sends one message and answers when the bus acknowledged it, and Listen makes a consumer that delivers only
	// the asks published after it is made; nil are JS's.
	Publish func(ctx context.Context, subject string, data []byte, id string) error
	Listen  func(ctx context.Context, subject string, handle func(data []byte)) (stop func(), err error)
}

// Writer keeps an observer's findings and says them (design 47 G §461, §485). Its acts, Write, Clear and a statement,
// run one at a time in the order they are asked.
type Writer[F, C, S, A any] struct {
	cfg  WriterConfig[F, C, S, A]
	subj WriterSubjects // the four subjects, formed by FindingSubjects
	mu   sync.Mutex     // the one lock of the acts
	seq  uint64
	ctl  sync.Mutex // guards the consumer, the asks and the loop's context
	stop func()
	asks []string
	want bool
	wake chan struct{}
	run  context.Context
	end  context.CancelFunc
}

// NewWriter is the writer for cfg.
func NewWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A]) (*Writer[F, C, S, A], error) {
	switch {
	case cfg.Module == "":
		return nil, errors.New("a finding writer needs its module's name")
	case cfg.State == nil:
		return nil, errors.New("a finding writer needs the state `findings`")
	case cfg.Observe == nil:
		return nil, errors.New("a finding writer needs the observe function")
	case cfg.Asked == "":
		return nil, errors.New("a finding writer needs the subject of findings-asked, from the generated contract of its seat")
	case cfg.JS == nil && (cfg.Publish == nil || cfg.Listen == nil):
		return nil, errors.New("a finding writer needs the bus connection's JetStream")
	}
	if err := requireFields[F]("finding", "kind", "scope", "id", "subject", "at", "since", "observer"); err != nil {
		return nil, err
	}
	if err := requireFields[C]("finding-cleared", "kind", "scope", "id", "at", "why"); err != nil {
		return nil, err
	}
	if err := requireFields[S]("findings-stated", "at", "findings", "pending"); err != nil {
		return nil, err
	}
	if err := requireFields[A]("findings-asked", "ask"); err != nil {
		return nil, err
	}
	if cfg.Node == "" {
		cfg.Node = os.Getenv("MESH_NODE")
	}
	if cfg.Node == "" {
		return nil, errors.New("a finding writer needs its node, which forms the subjects of its events")
	}
	if cfg.Now == nil {
		cfg.Now = func() time.Time { return time.Now().UTC() }
	}
	if cfg.Wait == nil {
		cfg.Wait = func(ctx context.Context, d time.Duration) bool {
			t := time.NewTimer(d)
			defer t.Stop()
			select {
			case <-ctx.Done():
				return false
			case <-t.C:
				return true
			}
		}
	}
	if cfg.Logf == nil {
		cfg.Logf = func(string, ...any) {}
	}
	if cfg.Publish == nil {
		cfg.Publish = jetStreamPublish(cfg.JS)
	}
	if cfg.Listen == nil {
		cfg.Listen = jetStreamListen(cfg.JS)
	}
	return &Writer[F, C, S, A]{cfg: cfg, subj: FindingSubjects(cfg.Module, cfg.Node), wake: make(chan struct{}, 1)}, nil
}

func jetStreamPublish(js jetstream.JetStream) func(context.Context, string, []byte, string) error {
	return func(ctx context.Context, subject string, data []byte, id string) error {
		pub, cancel := context.WithTimeout(ctx, 10*time.Second)
		defer cancel()
		_, err := js.PublishMsg(pub, &nats.Msg{Subject: subject, Data: data}, jetstream.WithMsgID(id))
		return err
	}
}

// jetStreamListen makes an ephemeral consumer of its own on the stream that holds subject, from the next message on.
func jetStreamListen(js jetstream.JetStream) func(context.Context, string, func([]byte)) (func(), error) {
	return func(ctx context.Context, subject string, handle func([]byte)) (func(), error) {
		stream, err := js.StreamNameBySubject(ctx, subject)
		if err != nil {
			return nil, fmt.Errorf("the stream of %s: %w", subject, err)
		}
		cons, err := js.CreateConsumer(ctx, stream, jetstream.ConsumerConfig{
			FilterSubject: subject, DeliverPolicy: jetstream.DeliverNewPolicy, AckPolicy: jetstream.AckNonePolicy,
			InactiveThreshold: 10 * time.Minute,
		})
		if err != nil {
			return nil, fmt.Errorf("a consumer of %s: %w", subject, err)
		}
		cc, err := cons.Consume(func(m jetstream.Msg) { handle(m.Data()) })
		if err != nil {
			return nil, fmt.Errorf("consuming %s: %w", subject, err)
		}
		return cc.Stop, nil
	}
}

// Start makes the consumer of `findings-asked` and then states the findings now, and again on each ask and each time the
// connection is made again (design 47 G §384, §485). It returns once the consumer exists; the statements run in the
// writer's own goroutine until ctx ends or Close.
func (w *Writer[F, C, S, A]) Start(ctx context.Context) error {
	w.ctl.Lock()
	defer w.ctl.Unlock()
	if w.run != nil {
		return errors.New("the finding writer is started")
	}
	w.run, w.end = context.WithCancel(ctx)
	if err := w.listen(); err != nil {
		w.end()
		w.run = nil
		return err
	}
	w.asks, w.want = append(w.asks, ""), true
	if w.cfg.Conn != nil {
		prev := w.cfg.Conn.Opts.ReconnectedCB
		w.cfg.Conn.SetReconnectHandler(func(nc *nats.Conn) {
			if prev != nil {
				prev(nc)
			}
			w.Reconnected()
		})
	}
	go w.loop(w.run)
	w.signal()
	return nil
}

// Close ends the writer's consumer and its statements.
func (w *Writer[F, C, S, A]) Close() {
	w.ctl.Lock()
	defer w.ctl.Unlock()
	if w.stop != nil {
		w.stop()
		w.stop = nil
	}
	if w.end != nil {
		w.end()
	}
}

func (w *Writer[F, C, S, A]) listen() error {
	stop, err := w.cfg.Listen(w.run, w.cfg.Asked, w.heard)
	if err != nil {
		return err
	}
	if w.stop != nil {
		w.stop()
	}
	w.stop = stop
	return nil
}

// Reconnected is told that the bus connection was made again: the writer makes its consumer again, from the newest ask,
// and states its findings again (design 47 G §485).
func (w *Writer[F, C, S, A]) Reconnected() {
	w.ctl.Lock()
	defer w.ctl.Unlock()
	if w.run == nil || w.run.Err() != nil {
		return
	}
	if err := w.listen(); err != nil {
		w.cfg.Logf("the finding writer could not make its consumer of %s again: %v", w.cfg.Asked, err)
	}
	w.asks, w.want = append(w.asks, ""), true
	w.signal()
}

func (w *Writer[F, C, S, A]) signal() {
	select {
	case w.wake <- struct{}{}:
	default:
	}
}

// heard is an ask: answered when it names no module or this one.
func (w *Writer[F, C, S, A]) heard(data []byte) {
	var body A
	if err := json.Unmarshal(data, &body); err != nil {
		w.cfg.Logf("an ask for findings could not be read: %v", err)
		return
	}
	a, err := wireOf(&body)
	if err != nil {
		w.cfg.Logf("an ask for findings could not be read: %v", err)
		return
	}
	if str(a, "module") != "" && str(a, "module") != w.cfg.Module {
		return
	}
	w.ctl.Lock()
	w.asks, w.want = append(w.asks, str(a, "ask")), true
	w.ctl.Unlock()
	w.signal()
}

func (w *Writer[F, C, S, A]) loop(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return
		case <-w.wake:
		}
		w.ctl.Lock()
		asks, want := w.asks, w.want
		w.asks, w.want = nil, false
		w.ctl.Unlock()
		if !want {
			continue
		}
		asked := ""
		if len(asks) > 0 {
			asked = asks[len(asks)-1]
		}
		for {
			err := w.State(ctx, asked)
			if err == nil {
				break
			}
			w.cfg.Logf("the findings could not be stated, no set said, trying again in %s: %v", ObserveRetryEvery, err)
			if !w.cfg.Wait(ctx, ObserveRetryEvery) {
				return
			}
		}
	}
}

// owns says whether the writer's statement covers kind: every kind when none were given (design 47 G §580), else its
// own and the views reader's.
func (w *Writer[F, C, S, A]) owns(kind string) bool {
	return len(w.cfg.Kinds) == 0 || slices.Contains(w.cfg.Kinds, kind) || slices.Contains(ViewKinds, kind)
}

// statedKinds are the kinds a statement names: none when the writer owns every kind.
func (w *Writer[F, C, S, A]) statedKinds() []string {
	if len(w.cfg.Kinds) == 0 {
		return nil
	}
	k := append(slices.Clone(w.cfg.Kinds), ViewKinds...)
	slices.Sort(k)
	return slices.Compact(k)
}

// ownsKey says whether key is one of a kind the writer owns: its last part is the kind (`<scope>.<id>.<kind>`).
func (w *Writer[F, C, S, A]) ownsKey(key string) bool {
	if len(w.cfg.Kinds) == 0 {
		return true
	}
	i := strings.LastIndex(key, ".")
	return i >= 0 && w.owns(key[i+1:])
}

func (w *Writer[F, C, S, A]) now() time.Time { return w.cfg.Now().UTC() }

// unsaid is a finding as compared to say whether what it says changed: not the time it was last seen, nor what saw it.
func unsaid(f wire) string {
	m := maps.Clone(f)
	delete(m, "at")
	delete(m, "run")
	raw, _ := json.Marshal(m)
	return string(raw)
}

func (w *Writer[F, C, S, A]) read(key string) (wire, error) {
	e, err := w.cfg.State.Get(key)
	if err != nil {
		return nil, err
	}
	if e == nil {
		return nil, nil
	}
	var old wire
	if err := json.Unmarshal(e.Value, &old); err != nil {
		return nil, fmt.Errorf("the finding %s in the state findings: %w", key, err)
	}
	return old, nil
}

// CheckFinding refuses a finding the keeper could not make a condition of, naming the field of the first rule it breaks:
// the key's three parts (`scope`, `id` and `kind`, each given, the kind with no dot, since the kind is the key's last
// part), and the forms of `owner` and `waiting-on` (design 47 G §678, §689). It judges the finding's wire form (design 47
// G A-3): any value that marshals as a finding, the contract's generated type among them. The writer refuses such a
// finding before it writes it, and the keeper judges the findings others write with it.
func CheckFinding(finding any) error {
	m, ok := finding.(wire)
	if !ok {
		var err error
		if m, err = wireOf(finding); err != nil {
			return fmt.Errorf("the finding could not be read: %w", err)
		}
	}
	switch {
	case strings.TrimSpace(str(m, "scope")) == "":
		return errors.New("the finding's scope is not given")
	case strings.TrimSpace(str(m, "id")) == "":
		return errors.New("the finding's id is not given")
	case strings.TrimSpace(str(m, "kind")) == "":
		return errors.New("the finding's kind is not given")
	case strings.Contains(str(m, "kind"), "."):
		return fmt.Errorf("the finding's kind %q holds a dot, and the kind is the last part of its key", str(m, "kind"))
	}
	if err := formOfWho("owner", str(m, "owner")); err != nil {
		return err
	}
	return formOfWho("waiting-on", str(m, "waiting-on"))
}

// complete fills what the observer leaves to the writer and keeps the key's `since` where the key exists (design 47 G §384).
func (w *Writer[F, C, S, A]) complete(finding F) (wire, wire, error) {
	f, err := wireOf(&finding)
	if err != nil {
		return nil, nil, fmt.Errorf("the finding could not be read: %w", err)
	}
	if err := CheckFinding(f); err != nil {
		return f, nil, err
	}
	if !w.owns(str(f, "kind")) {
		return f, nil, fmt.Errorf("the finding kind %q is not one this writer owns (%v)", str(f, "kind"), w.cfg.Kinds)
	}
	old, err := w.read(keyOf(f))
	if err != nil {
		return f, nil, err
	}
	if str(f, "observer") == "" {
		f["observer"] = w.cfg.Module
	}
	at, _ := time.Parse(time.RFC3339Nano, str(f, "at"))
	if at.IsZero() {
		at = w.now()
	}
	f["at"] = at.UTC()
	since, _ := time.Parse(time.RFC3339Nano, str(f, "since"))
	if old != nil {
		if os, _ := time.Parse(time.RFC3339Nano, str(old, "since")); !os.IsZero() {
			since = os
		}
	}
	if since.IsZero() {
		since = at
	}
	f["since"] = since.UTC()
	return f, old, nil
}

// WriteWire is Write of a finding given in its wire form (a views reader's findings, design 47 G §687).
func (w *Writer[F, C, S, A]) WriteWire(ctx context.Context, finding map[string]any) error {
	var f F
	if err := fillFrom(&f, finding); err != nil {
		return err
	}
	return w.Write(ctx, f)
}

// ModuleNode is the module and the node the writer's findings are about.
func (w *Writer[F, C, S, A]) ModuleNode() (string, string) { return w.cfg.Module, w.cfg.Node }

// Write keeps the finding in its key and says `finding-written` only when what the finding says changed (design 47 G
// §383, §461). The key's `since` stays where the key exists.
func (w *Writer[F, C, S, A]) Write(ctx context.Context, finding F) error {
	w.mu.Lock()
	defer w.mu.Unlock()
	f, old, err := w.complete(finding)
	if err != nil {
		return err
	}
	if _, err := w.cfg.State.Put(keyOf(f), f); err != nil {
		return fmt.Errorf("writing the finding %s: %w", keyOf(f), err)
	}
	if old != nil && unsaid(old) == unsaid(f) {
		return nil
	}
	raw, _ := json.Marshal(f)
	w.say(ctx, "finding-written", keyOf(f), raw)
	return nil
}

// Clear deletes the finding's key and says `finding-cleared`; a key that does not exist is deleted of nothing and is said
// all the same, so a keeper holding the finding clears it (design 47 G §485).
func (w *Writer[F, C, S, A]) Clear(ctx context.Context, scope, id, kind, why string) error {
	w.mu.Lock()
	defer w.mu.Unlock()
	key := FindingKey(scope, id, kind)
	if !w.owns(kind) {
		return fmt.Errorf("the finding kind %q is not one this writer owns (%v)", kind, w.cfg.Kinds)
	}
	if err := w.cfg.State.Delete(key); err != nil {
		return fmt.Errorf("clearing the finding %s: %w", key, err)
	}
	var cleared C
	if err := fillFrom(&cleared, wire{"kind": kind, "scope": scope, "id": id, "at": w.now(), "why": why}); err != nil {
		return err
	}
	raw, _ := json.Marshal(&cleared)
	w.say(ctx, "finding-cleared", key, raw)
	return nil
}

// State judges every finding anew with observe, writes each stated finding's key, deletes each key neither stated nor
// pending, and says `findings-stated` (design 47 G §384, §461); asked is the ask it answers, none at a start. An error of
// observe says no set.
func (w *Writer[F, C, S, A]) State(ctx context.Context, asked string) error {
	w.mu.Lock()
	defer w.mu.Unlock()
	obs, err := w.cfg.Observe(ctx)
	if err != nil {
		return fmt.Errorf("observe: %w", err)
	}
	keep := map[string]bool{}
	findings, pending := []wire{}, []wire{}
	for _, finding := range obs.Findings {
		f, _, err := w.complete(finding)
		if err != nil {
			return err
		}
		if _, err := w.cfg.State.Put(keyOf(f), f); err != nil {
			return fmt.Errorf("writing the finding %s: %w", keyOf(f), err)
		}
		keep[keyOf(f)] = true
		findings = append(findings, f)
	}
	for _, p := range obs.Pending {
		keep[p.Key] = true
		pending = append(pending, wire{"key": p.Key, "since": p.Since, "bound": p.Bound})
	}
	keys, err := w.cfg.State.Keys()
	if err != nil {
		return fmt.Errorf("the keys of the state findings: %w", err)
	}
	slices.Sort(keys)
	for _, k := range keys {
		// a views reader's open findings are stated beside observe's (design 47 G §639)
		if i := strings.LastIndex(k, "."); !keep[k] && i >= 0 && slices.Contains(ViewKinds, k[i+1:]) {
			if old, err := w.read(k); err == nil && old != nil {
				keep[k] = true
				findings = append(findings, old)
			}
		}
		if !keep[k] && w.ownsKey(k) {
			if err := w.cfg.State.Delete(k); err != nil {
				return fmt.Errorf("deleting the finding %s: %w", k, err)
			}
		}
	}
	stated := wire{"at": w.now(), "findings": findings, "pending": pending}
	if asked != "" {
		stated["asked"] = asked
	}
	if k := w.statedKinds(); k != nil {
		stated["kinds"] = k
	}
	var body S
	if err := fillFrom(&body, stated); err != nil {
		return err
	}
	raw, _ := json.Marshal(&body)
	if w.cfg.Send != nil {
		if raw, err = w.cfg.Send("findings-stated."+w.cfg.Module, raw); err != nil {
			w.cfg.Logf("the findings-stated of %s could not be sent, the next statement covers it: %v", w.cfg.Module, err)
			return nil
		}
	}
	w.say(ctx, "findings-stated", w.cfg.Module, raw)
	return nil
}

// say publishes one event awaiting the bus's acknowledgement, sent again for PublishRetryFor when unacknowledged, then
// written in the journal as an error naming what it said (design 47 G §383); the next statement covers it.
func (w *Writer[F, C, S, A]) say(ctx context.Context, event, what string, data []byte) {
	w.seq++
	id := fmt.Sprintf("%s-%s-%d-%d", w.cfg.Module, event, w.now().UnixNano(), w.seq)
	start := w.now()
	for delay := time.Second; ; delay = min(delay*2, 10*time.Second) {
		subject, err := w.subject(event)
		if err == nil {
			err = w.cfg.Publish(ctx, subject, data, id)
		}
		if err == nil {
			return
		}
		if w.now().Sub(start) >= PublishRetryFor || ctx.Err() != nil || !w.cfg.Wait(ctx, delay) {
			w.cfg.Logf("error: %s for %s was not acknowledged by the bus, the next statement covers it: %v", event, what, err)
			return
		}
	}
}

// fillFrom sets the contract body v from its wire form.
func fillFrom(v any, m wire) error {
	raw, err := json.Marshal(m)
	if err != nil {
		return err
	}
	return json.Unmarshal(raw, v)
}