mesh-catalog/modules/nats/cmd/nats-tools/keptools.go

package main

import (
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"

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

	"nats-tools/internal/keeper"
)

// keptWait is how long a call to take or restore a kept data waits for the keeper's end on the watch of its requests: under
// the runtime's own bound on a call, so a long snapshot is answered as "handed, not ended" and read again with its request.
const keptWait = 50 * time.Second

// KeptHeld reads the keeper's state kept: the restore points it took and the data it set aside.
type KeptHeld func() ([]keeper.Point, []keeper.Aside, error)

// KeptInState is the state kept through the runtime that launched this bundle.
func KeptInState() ([]keeper.Point, []keeper.Aside, error) {
	st := stdio.State(keeper.StateKept)
	keys, err := st.Keys()
	if err != nil {
		return nil, nil, err
	}
	sort.Strings(keys)
	var points []keeper.Point
	var aside []keeper.Aside
	for _, key := range keys {
		e, err := st.Get(key)
		if err != nil || e == nil {
			continue
		}
		switch {
		case strings.HasPrefix(key, "point."):
			var p keeper.Point
			if json.Unmarshal(e.Value, &p) == nil {
				points = append(points, p)
			}
		case strings.HasPrefix(key, "aside."):
			var a keeper.Aside
			if json.Unmarshal(e.Value, &a) == nil {
				aside = append(aside, a)
			}
		}
	}
	return points, aside, nil
}

func formOf(a map[string]any) int {
	f, _ := a["form"].(float64)
	return int(f)
}

// keptTools are the seat mesh-broker's verbs for a module's kept data and the bus module's tool that reads what it holds:
// each verb hands its request to the keeper, which acts on its own account, and waits for the end on the watch of its
// requests (never a poll). Asked again with `request`, a verb reads the end of the request it made.
func keptTools(rs requestStore, held KeptHeld) []stdio.Tool {
	entry := func(extra map[string]any) map[string]any {
		in := map[string]any{"module": str("the module whose kept data it is, e.g. mesh-delivery-indexer"), "local-name": str("the kept data's local name, e.g. modules"),
			"form": num("the kept data's form (a manifest field; absent reads 0)"), "fingerprint": str("the fingerprint the restore point was taken before"),
			"why": str("why, in plain words: kept with the act"), "request": str("a request this verb made before: its end is read, nothing new is asked")}
		for k, v := range extra {
			in[k] = v
		}
		return in
	}
	run := func(verb string, build func(a map[string]any, r *keeper.Request) error) func(a map[string]any) (any, error) {
		return func(a map[string]any) (any, error) {
			name := "mesh-broker." + verb
			if id := arg(a, "request"); id != "" {
				r, err := rs.Get(id)
				if err != nil || r == nil {
					return nil, fmt.Errorf("no request %s is held (%v)", id, err)
				}
				if r.Done.IsZero() && rs.Ended != nil {
					if got := rs.Ended(id, keptWait); got != nil {
						return got, nil
					}
					if again, err := rs.Get(id); err == nil && again != nil {
						r = again
					}
				}
				return r, nil
			}
			r := keeper.Request{Verb: verb, Why: strings.TrimSpace(arg(a, "why")), Caller: stdio.CallerOf(a), At: rs.Now()}
			if err := build(a, &r); err != nil {
				return nil, err
			}
			if r.Why == "" {
				return nil, fmt.Errorf("%s says why", name)
			}
			r.ID = requestID(r.At)
			if err := rs.Put(r.ID, r); err != nil {
				return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err)
			}
			if rs.Ended != nil {
				if got := rs.Ended(r.ID, keptWait); got != nil {
					return got, nil
				}
			}
			return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", "read": fmt.Sprintf("%s with request %s", name, r.ID)}, nil
		}
	}
	point := func(a map[string]any, r *keeper.Request) error {
		r.Module, r.Local, r.Form, r.Fingerprint = arg(a, "module"), arg(a, "local-name"), formOf(a), arg(a, "fingerprint")
		if r.Module == "" || r.Local == "" || r.Fingerprint == "" {
			return fmt.Errorf("a restore point is of one kept data: a module, its local name and the fingerprint it is taken before")
		}
		return nil
	}
	readAside := func() (any, error) {
		_, aside, err := held()
		if err != nil {
			return nil, err
		}
		if aside == nil {
			aside = []keeper.Aside{}
		}
		return map[string]any{"aside": aside}, nil
	}
	return []stdio.Tool{
		{Name: "mesh-broker.snapshot-kept",
			Description: "Take the restore point of one module's kept data on the bus (milestone G, design 47): the data whole as the server's own snapshot, the position of each durable consumer the module reads with, and the catch-up tool the newer manifest declares, kept by the bus module's keeper. Answers the directory the restore point is in. Handed to the keeper, recorded as a hand act.",
			Input:       entry(map[string]any{"catch-up": str("the tool a restore is followed by, from the newer manifest, kept in the restore point")}),
			Run: run(keeper.VerbSnapshotKept, func(a map[string]any, r *keeper.Request) error {
				r.CatchUp = arg(a, "catch-up")
				return point(a, r)
			})},
		{Name: "mesh-broker.restore-kept",
			Description: "Restore one module's kept data on the bus from a restore point beside the live data, never over it (design 47): the live data is snapshotted into a set-aside directory under a name of its own and then removed from the bus, the restore point's streams are put in its place and the module's other consumers are set back to their positions. Answers what was set aside and the catch-up tool the restore point carries. Handed to the keeper, recorded as a hand act.",
			Input:       entry(map[string]any{"from": str("the directory the restore point was restored to, an absolute path")}),
			Run: run(keeper.VerbRestoreKept, func(a map[string]any, r *keeper.Request) error {
				r.From = arg(a, "from")
				if r.From == "" {
					return fmt.Errorf("a restore names the directory the restore point is in")
				}
				return point(a, r)
			})},
		{Name: "mesh-broker.discard-aside",
			Description: "Remove the data a restore set aside, by its id (from `aside`), only on a person's call: it says who and why, and the keeper records the hand act before it removes anything.",
			Input:       map[string]any{"id": str("the id of the data set aside"), "why": str("why, in plain words: kept with the act"), "request": str("a request made before: its end is read")},
			Run: run(keeper.VerbDiscardAside, func(a map[string]any, r *keeper.Request) error {
				r.From = arg(a, "id")
				if r.From == "" {
					return fmt.Errorf("discard-aside names the id of the data set aside")
				}
				return nil
			})},
		{Name: "mesh-broker.aside", Description: "The data restores set aside, each with its module, kept data, size, when, who and why: what a person decides to discard.",
			Run: func(map[string]any) (any, error) { return readAside() }},
		{Name: "nats_kept",
			Description: "What the bus module holds of its modules' kept data besides the live data: every restore point it took (module, kept data, form, the fingerprint it was taken before, its consumers' positions, the catch-up tool, its size) and the data a restore set aside. Read from the keeper's state kept.",
			Run: func(map[string]any) (any, error) {
				points, aside, err := held()
				if err != nil {
					return nil, err
				}
				if points == nil {
					points = []keeper.Point{}
				}
				if aside == nil {
					aside = []keeper.Aside{}
				}
				return map[string]any{"points": points, "aside": aside}, nil
			}},
	}
}