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

package main

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/handacts"

	"nats-tools/internal/keeper"
)

// A person's acting tools (design 25 § 12.3; CAT-N8): nats_create and nats_apply (acts), nats_recreate and nats_drop
// (dangerous, granted only by name). None holds a grant of the seat's acting verbs: each hands its request to the
// keeper through the keeper's own state requests, stamped with the caller the tool runner proved, and reads its end
// back. recreate and drop first raise the operator's ask, binding exactly that act; the keeper acts only on its warrant,
// after a snapshot.

// requestStore is the keeper's state requests, as this bundle reaches it.
type requestStore struct {
	Put    func(key string, r keeper.Request) error
	Get    func(key string) (*keeper.Request, error)
	Keys   func() ([]string, error)
	Delete func(key string) error
	// Spent reads the keeper's spend records, its state spent, which this bundle reads and never writes.
	Spent func() ([]keeper.Acted, error)
	Ask   func(a oc.Ask) error
	Now   func() time.Time
	// Ended answers the request's end once the keeper wrote it, or nil once within passed: heard on the state's watch,
	// never read again on a timer.
	Ended func(id string, within time.Duration) *keeper.Request
	// Record records a hand act through the seat mesh-hand-acts.
	Record Recorded
}

// RequestsInState is the store through the runtime that launched this bundle: one watch of the state requests for the
// bundle's life (the runtime's watches last as long as the process), handing each end to the call waiting for it.
func RequestsInState() requestStore {
	s := stdio.State(keeper.StateRequests)
	ends := newEnds(func(handle func(key string, value []byte)) error {
		return s.Watch("", func(c stdio.StateChange) error {
			handle(c.Key, c.Value)
			return nil
		})
	})
	return requestStore{
		Put: func(key string, r keeper.Request) error { _, err := s.Put(key, r); return err },
		Get: func(key string) (*keeper.Request, error) {
			e, err := s.Get(key)
			if err != nil || e == nil {
				return nil, err
			}
			var r keeper.Request
			return &r, json.Unmarshal(e.Value, &r)
		},
		Keys:   s.Keys,
		Delete: s.Delete,
		Spent: func() ([]keeper.Acted, error) {
			spent := stdio.State(keeper.StateSpent)
			keys, err := spent.Keys()
			if err != nil {
				return nil, err
			}
			sort.Strings(keys)
			out := []keeper.Acted{}
			for _, key := range keys {
				var a keeper.Acted
				if e, err := spent.Get(key); err == nil && e != nil && json.Unmarshal(e.Value, &a) == nil {
					out = append(out, a)
				}
			}
			return out, nil
		},
		Ask:    func(a oc.Ask) error { return stdio.SeatPublish(oc.AcceptSubjectAsk(keeper.Module), a, a.ID) },
		Now:    func() time.Time { return time.Now().UTC() },
		Ended:  ends.wait,
		Record: RecordedBySeat,
	}
}

// ends hands a request's end to the call waiting for it, from one watch of the state started at the first wait.
type ends struct {
	watch   func(handle func(key string, value []byte)) error
	mu      sync.Mutex
	started bool
	waiting map[string][]chan *keeper.Request
}

func newEnds(watch func(handle func(key string, value []byte)) error) *ends {
	return &ends{watch: watch, waiting: map[string][]chan *keeper.Request{}}
}

func (e *ends) heard(key string, value []byte) {
	var r keeper.Request
	if json.Unmarshal(value, &r) != nil || r.Done.IsZero() {
		return
	}
	e.mu.Lock()
	chans := e.waiting[key]
	delete(e.waiting, key)
	e.mu.Unlock()
	for _, c := range chans {
		c <- &r
	}
}

// wait waits for id's end at most within: nil when it did not come, or the watch could not be taken.
func (e *ends) wait(id string, within time.Duration) *keeper.Request {
	c := make(chan *keeper.Request, 1)
	e.mu.Lock()
	e.waiting[id] = append(e.waiting[id], c)
	start := !e.started
	e.started = true
	e.mu.Unlock()
	if start {
		if err := e.watch(e.heard); err != nil {
			e.mu.Lock()
			e.started = false
			e.mu.Unlock()
			return nil
		}
	}
	select {
	case r := <-c:
		return r
	case <-time.After(within):
		e.mu.Lock()
		e.waiting[id] = slices.DeleteFunc(e.waiting[id], func(w chan *keeper.Request) bool { return w == c })
		e.mu.Unlock()
		return nil
	}
}

func requestID(now time.Time) string {
	var b [4]byte
	_, _ = rand.Read(b[:])
	return now.Format("20060102t150405") + "-" + hex.EncodeToString(b[:])
}

// askFor is the operator's ask for a recreate or a drop of one entry, its approve option binding exactly that act.
func askFor(r keeper.Request, now time.Time) (oc.Ask, error) {
	binds, err := occ.ActDigest(r.Act())
	if err != nil {
		return oc.Ask{}, err
	}
	what := map[string]string{keeper.VerbDrop: "remove", keeper.VerbRecreate: "rebuild"}[r.Verb]
	name := keeper.BusName(r.Module, r.Local)
	return oc.Ask{ID: "bus-" + r.ID, Who: occ.Operator, Expires: occ.FormatTime(now.Add(keeper.AskLasts)),
		OnExpiry: "nothing is removed and the entry stays as it is",
		About:    keeper.ConditionKey(map[string]string{keeper.VerbDrop: "orphaned", keeper.VerbRecreate: "drifted"}[r.Verb], r.Module, r.Local),
		Headline: fmt.Sprintf("%s's kept data %s: %s it?", r.Module, r.Local, what),
		Explanation: fmt.Sprintf("%s asks to %s %s on the bus, %s. A snapshot of it is taken first; its data is lost from the bus.",
			r.Caller, what, name, r.Why),
		Options: []oc.Option{
			{ID: "go", Label: strings.ToUpper(what[:1]) + what[1:], Does: fmt.Sprintf("%s %s after a snapshot", what, name), Level: occ.Destroy, Binds: binds},
			{ID: "keep", Label: "Keep it", Does: "nothing is removed", Level: occ.Acknowledge},
		}}, nil
}

func requestTools(rs requestStore) []stdio.Tool {
	entry := map[string]any{"module": str("the module, e.g. mesh-knowledge"), "local": str("its local name, e.g. records"),
		"why": str("why, in plain words: kept with the act"), "request": str("a request this tool made before: its end is read, nothing new is asked")}
	tool := func(verb, description string) stdio.Tool {
		return stdio.Tool{Name: "nats_" + verb, Description: description, Input: entry, Run: func(a map[string]any) (any, error) {
			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)
				}
				return r, nil
			}
			r := keeper.Request{Verb: verb, Module: arg(a, "module"), Local: arg(a, "local"), Why: strings.TrimSpace(arg(a, "why")),
				// The caller the tool runner proved, never a name the arguments give.
				Caller: stdio.CallerOf(a), At: rs.Now()}
			if r.Module == "" || r.Local == "" || r.Why == "" {
				return nil, fmt.Errorf("nats_%s names one entry, a module and its local name, and says why", verb)
			}
			r.ID = requestID(r.At)
			if verb == keeper.VerbRecreate || verb == keeper.VerbDrop {
				ask, err := askFor(r, r.At)
				if err != nil {
					return nil, err
				}
				r.Ask = ask.ID
				// The ask first: a request is never handed for an ask that was not raised (an ask with no request acts on
				// nothing, since the keeper acts only on a request).
				if err := rs.Ask(ask); err != nil {
					return nil, fmt.Errorf("the operator's ask cannot be raised, so nothing will be removed: %w", err)
				}
				if err := rs.Put(r.ID, r); err != nil {
					return nil, fmt.Errorf("the request cannot be handed to the keeper, so nothing will be removed whatever the operator answers: %w", err)
				}
				return map[string]any{"request": r.ID, "ask": ask.ID, "state": "the operator is asked; the keeper acts once they approve, after a snapshot",
					"read": fmt.Sprintf("nats_%s with request %s", verb, r.ID)}, nil
			}
			if err := rs.Put(r.ID, r); err != nil {
				return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err)
			}
			return awaitEnd(rs, r, "nats_"+verb), nil
		}}
	}
	firstView := stdio.Tool{Name: "nats_first_view_again",
		Description: "Ask the bus module's keeper to ask the indexer's indexer_bus_state, its first view of every accepted manifest's bus state, at once (design 25 § 12.2): by hand, beside its own backoff, recorded as a hand act; answers what the first view answered and its revision.",
		Input:       map[string]any{"why": str("why, in plain words: kept with the act"), "request": str("a request made before: its end is read")},
		Run: func(a map[string]any) (any, error) {
			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)
				}
				return r, nil
			}
			r := keeper.Request{Verb: keeper.VerbFirstViewAgain, Why: strings.TrimSpace(arg(a, "why")), Caller: stdio.CallerOf(a), At: rs.Now()}
			if r.Why == "" {
				return nil, fmt.Errorf("nats_first_view_again says why")
			}
			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)
			}
			return awaitEnd(rs, r, "nats_first_view_again"), nil
		}}
	return []stdio.Tool{
		requestsTool(rs),
		firstView,
		tool(keeper.VerbCreate, "Create one entry of a module's kept data on the bus now, as its accepted manifest asks for it (design 25 § 12.3): handed to the bus module's keeper, which alone may create, and recorded as a hand act."),
		tool(keeper.VerbApply, "Apply what a module's manifest asks of one entry where the bus holds less (a larger cap, a longer age): only a difference that grows; handed to the keeper and recorded as a hand act."),
		tool(keeper.VerbRecreate, "Rebuild one entry whose settings drifted in a way that does not grow: the operator is asked first, and the keeper acts only on their warrant, after a snapshot of the entry. Its data is lost from the bus (the snapshot keeps it). Dangerous."),
		tool(keeper.VerbDrop, "Remove one orphan, an entry no accepted manifest asks for: the operator is asked first, and the keeper acts only on their warrant, after a snapshot of the entry. Dangerous."),
	}
}

// awaitEnd waits for a request's end, which the keeper writes as soon as its watch of requests hands it the request;
// past half a minute it answers how to read the end later.
func awaitEnd(rs requestStore, r keeper.Request, tool string) any {
	if rs.Ended != nil {
		if got := rs.Ended(r.ID, 30*time.Second); got != nil {
			return got
		}
	}
	if got, err := rs.Get(r.ID); err == nil && got != nil && !got.Done.IsZero() {
		return got
	}
	return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", "read": fmt.Sprintf("%s with request %s", tool, r.ID)}
}

// requestsTool reads the keeper's state requests (design 25 § 12.3; review round 1, tools F3): every request with what
// waits, what ended and why, and each warrant spent; and drops one request not ended yet by hand, recorded as a hand act,
// so a request nobody will answer leaves by a person's act and never only by its key ageing out.
func requestsTool(rs requestStore) stdio.Tool {
	return stdio.Tool{Name: "nats_requests",
		Description: "The persons' requests the bus module's tools handed its keeper (nats_create, nats_apply, nats_recreate, nats_drop, nats_first_view_again): each one waiting (and for what: the operator's warrant on its ask, or the keeper's first view) or ended, with its outcome or why not, and each warrant spent (the keeper's state spent, which only it writes: a warrant acts once). With drop and why, removes one request not ended yet, recorded as a hand act; the operator's ask it raised then acts on nothing.",
		Input:       map[string]any{"drop": str("a request not ended yet, to remove by hand"), "why": str("why, in plain words: kept with the act")},
		Run: func(a map[string]any) (any, error) {
			if id := strings.TrimSpace(arg(a, "drop")); id != "" {
				why := strings.TrimSpace(arg(a, "why"))
				if why == "" {
					return nil, fmt.Errorf("nats_requests drop says why")
				}
				r, err := rs.Get(id)
				if err != nil || r == nil || strings.HasPrefix(id, keeper.ActedPrefix) {
					return nil, fmt.Errorf("no request %s is held (%v)", id, err)
				}
				if !r.Done.IsZero() {
					return nil, fmt.Errorf("the request %s ended at %s (%s): only a request that waits is dropped", id, r.Done.Format(time.RFC3339), r.Outcome)
				}
				if err := rs.Record(handacts.Request{What: fmt.Sprintf("dropped the waiting request %s (%s of %s's %s, ask %s)", id, r.Verb, r.Module, r.Local, r.Ask),
					Why: why, Cause: "nats_requests"}); err != nil {
					return nil, fmt.Errorf("the hand act cannot be recorded, so the request stays: %w", err)
				}
				if err := rs.Delete(id); err != nil {
					return nil, err
				}
				return map[string]any{"dropped": id, "ask": r.Ask, "state": "the keeper acts on it no more; the operator's ask it raised acts on nothing"}, nil
			}
			keys, err := rs.Keys()
			if err != nil {
				return nil, err
			}
			sort.Strings(keys)
			var waiting, ended []keeper.Request
			spent, err := rs.Spent()
			if err != nil {
				return nil, fmt.Errorf("the warrants spent cannot be read: %w", err)
			}
			for _, key := range keys {
				if strings.HasPrefix(key, keeper.ActedPrefix) {
					continue
				}
				r, err := rs.Get(key)
				if err != nil || r == nil {
					continue
				}
				if r.Done.IsZero() {
					waiting = append(waiting, *r)
				} else {
					ended = append(ended, *r)
				}
			}
			return map[string]any{"waiting": orEmpty(waiting), "ended": orEmpty(ended), "warrants-spent": spent,
				"why-waiting": "a recreate or a drop waits for the operator's warrant on its ask, acted on once they answer, and ends unanswered when its ask ends; any request waits while the keeper holds no first view"}, nil
		}}
}

func orEmpty(rs []keeper.Request) []keeper.Request {
	if rs == nil {
		return []keeper.Request{}
	}
	return rs
}