mesh-catalog/modules/distribution/cmd/distribution/windows.go

package main

// The windows' driver (design 51 *The move*): at the program's start, on each `delivered` of the status view and on each
// node-engine's `manifest-reads-changed` (ADR 0618 §4) it reads every node through node_haves and the node-engine
// seat's manifest-reads, and steps the old-name and the direct-write windows in distribution's state windows. The
// program does it, not the tools bundle: it holds distribution's own account, its durable consumer of the events
// stream and its own stream for the closing events.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/health"
	"git.novox.be/novox/mesh-sdk/go/process"

	"distribution/internal/holder"
)

// The addresses the driver asks (manifest invokes).
const (
	havesAsk = "mesh-delivery-view.node_haves"
	readsAsk = "seat:node-engine.manifest-reads"
	// callersAsk is distribution's own tool answering its config value callers as resolved.
	callersAsk = holder.Self + "." + holder.CallersTool
)

// caller is a process's call of a tool or a seat verb.
type caller interface {
	Call(key string, body any) (json.RawMessage, error)
	CallOn(node, key string, body any) (json.RawMessage, error)
}

// busCallers reads the caller list from distribution's own tools bundle (holder.CallersTool, manifest invokes): a
// program has no runtime that resolves its module's config values, the tools bundle has (stdio.Config), so the program
// asks it on each wake rather than read an environment variable nothing renders.
type busCallers struct{ c caller }

func (b busCallers) Read() (holder.Callers, error) {
	raw, err := b.c.Call(callersAsk, map[string]any{})
	if err != nil {
		return nil, fmt.Errorf("%s: %w", callersAsk, err)
	}
	var a struct {
		Callers json.RawMessage `json:"callers"`
	}
	if err := json.Unmarshal(unwrap(raw), &a); err != nil || len(a.Callers) == 0 {
		return nil, fmt.Errorf("%s answered what is not a caller list: %s", callersAsk, unwrap(raw))
	}
	return holder.ParseCallers(string(a.Callers))
}

// busDelivery answers holder.Delivery through the bus.
type busDelivery struct{ c caller }

func (d busDelivery) Haves() ([]holder.NodeHave, error) {
	raw, err := d.c.Call(havesAsk, map[string]any{})
	if err != nil {
		return nil, err
	}
	var a struct {
		Keys map[string]struct {
			FP string `json:"fp"`
		} `json:"keys"`
		Cut bool `json:"cut"`
	}
	if err := json.Unmarshal(unwrap(raw), &a); err != nil {
		return nil, fmt.Errorf("node_haves answered what is not its listing: %w", err)
	}
	if a.Cut {
		return nil, errors.New("node_haves cut its answer, so not every node is read")
	}
	out := make([]holder.NodeHave, 0, len(a.Keys))
	for key, h := range a.Keys {
		node, module, ok := strings.Cut(key, ".")
		if ok {
			out = append(out, holder.NodeHave{Node: node, Module: module, FP: h.FP})
		}
	}
	return out, nil
}

func (d busDelivery) Reads(node string) (bool, []string, error) {
	raw, err := d.c.CallOn(node, readsAsk, map[string]any{})
	if err != nil {
		if notServed(err) {
			// A node-engine not serving manifest-reads states nothing (design 51); what the call answered (an away node,
			// an older node-engine, a timeout) goes with it, so the window's tool says why the node holds it.
			return false, nil, err
		}
		// Any other failure is a read that did not happen: served true with its error, so the wake judges nothing.
		return true, nil, err
	}
	var a struct {
		Reads []string `json:"manifest-reads"`
	}
	if err := json.Unmarshal(unwrap(raw), &a); err != nil {
		return true, nil, fmt.Errorf("%s's manifest-reads answered what is not its statement: %w", node, err)
	}
	return true, a.Reads, nil
}

// notServed is whether a call's error says nothing serves the address (an away node, an older node-engine).
func notServed(err error) bool {
	if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	m := strings.ToLower(err.Error())
	for _, w := range []string{"no responders", "serves no verb", "not served", "unknown tool", "timeout"} {
		if strings.Contains(m, w) {
			return true
		}
	}
	return false
}

// unwrap takes a tool's answer out of the MCP content form the tool runner may answer in.
func unwrap(raw json.RawMessage) json.RawMessage {
	var s string
	if json.Unmarshal(raw, &s) == nil {
		return unwrap(json.RawMessage(s))
	}
	var c struct {
		Content []struct {
			Text string `json:"text"`
		} `json:"content"`
	}
	if json.Unmarshal(raw, &c) == nil && len(c.Content) > 0 {
		return json.RawMessage(c.Content[0].Text)
	}
	return raw
}

// kvWindows answers holder.WindowState over the state windows.
type kvWindows struct{ kv jetstream.KeyValue }

func (k kvWindows) get(key string, v any) error {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	e, err := k.kv.Get(ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil
	}
	if err != nil {
		return err
	}
	return json.Unmarshal(e.Value(), v)
}

func (k kvWindows) Window(name string) (holder.Window, error) {
	var w holder.Window
	return w, k.get(name, &w)
}

func (k kvWindows) PutWindow(w holder.Window) error {
	b, _ := json.Marshal(w)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_, err := k.kv.Put(ctx, w.Name, b)
	return err
}

func (k kvWindows) Publishing() (holder.Publishing, error) {
	var p holder.Publishing
	return p, k.get(holder.PublishingKey, &p)
}

// findingsKV writes the driver's findings into distribution's state findings, as the tools bundle writes its own.
func findingsKV(kv jetstream.KeyValue, logf func(string, ...any)) func(kind, subject, words string, raised bool) {
	return func(kind, subject, words string, raised bool) {
		if kv == nil {
			return
		}
		key := health.FindingKey(health.ScopeSeat, holder.Self, kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject))
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if !raised {
			_ = kv.Delete(ctx, key)
			return
		}
		f := health.Finding{V: health.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:],
			Subject: health.Subject{Scope: health.ScopeSeat, ID: holder.Self}, Severity: health.Warning,
			Summary: words, Headline: words, Explanation: words + "; registry_old_name or registry_direct_writes drops a retired node by hand",
			Resolved: "cleared when the window closes", Source: kind, Observer: holder.Self, At: time.Now().UTC()}
		b, _ := json.Marshal(f)
		if _, err := kv.Put(ctx, key, b); err != nil {
			logf("the finding %s could not be written: %v", key, err)
		}
	}
}

// sayClosed says a window's closing on distribution's own family distribution.events, once (its id is the window's).
func sayClosed(js jetstream.JetStream, cred process.Account) func(holder.Closed) error {
	return func(c holder.Closed) error {
		event := holder.OldNameClosedEvent
		if c.Window == holder.DirectWritesWindow {
			event = holder.DirectWritesClosedEvent
		}
		b, _ := json.Marshal(c)
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_, err := js.Publish(ctx, cred.StreamSubject(event), b, jetstream.WithMsgID(holder.Self+"."+event))
		return err
	}
}

// wakes says which events wake the driver, by subject: the status view's delivered, a node-engine's
// manifest-reads-changed (its node the subject's last token).
func wakes(subject string) (string, bool) {
	switch {
	case strings.HasSuffix(subject, ".mesh-delivery-view.event.delivered"):
		return "delivered", true
	case strings.Contains(subject, ".node-engine.event.manifest-reads-changed."):
		return "manifest-reads-changed on " + subject[strings.LastIndex(subject, ".")+1:], true
	}
	return "", false
}

// drive wakes the driver at start and on each waking event of the program's durable consumer of the events stream.
// A wake that could not judge is said and its event asked again a minute later.
func drive(ctx context.Context, js jetstream.JetStream, cred process.Account, d *holder.Driver, logger *log.Logger) {
	wake := func(why string) error {
		said, err := d.Wake(why)
		for _, s := range said {
			logger.Printf("%s", s)
		}
		if err != nil {
			logger.Printf("the windows were not judged (%s): %v", why, err)
		}
		return err
	}
	_ = wake("distribution started")
	cons, err := cred.EventConsumer(ctx, js, process.EventsStream)
	if err != nil {
		logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err)
		return
	}
	cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
		why, ok := wakes(m.Subject())
		if !ok {
			_ = m.Ack()
			return
		}
		if wake(why) != nil {
			_ = m.NakWithDelay(time.Minute)
			return
		}
		_ = m.Ack()
	}, logger.Printf)
	if err != nil {
		logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err)
		return
	}
	<-ctx.Done()
	cc.Stop()
}