mesh-sdk/go/views/status.go

package views

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

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

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

// StatusState is a reader's own state of its first views (design 25 § 12.2): at StatusKey each first view's Status, as
// its process last said it, and at AgainKey a hand ask of its re-read tool, which the process takes and deletes. A
// process serves no tool (ADR 0436), so its read and re-read tools, in its tools bundle, meet it here.
const StatusState = "first-views"

// StatusKey is where a reader keeps its first view of owner's state.
func StatusKey(owner, state string) string { return owner + "." + state }

// AgainKey is where a re-read tool asks a reader's process to ask owner's first view of state again.
func AgainKey(owner, state string) string { return "again." + owner + "." + state }

// HandAsk is one ask of a re-read tool: who asked (the runtime's proof), why and when.
type HandAsk struct {
	V     int       `json:"v"`
	Owner string    `json:"owner"`
	State string    `json:"state"`
	Why   string    `json:"why"`
	By    string    `json:"by"`
	At    time.Time `json:"at"`
}

// KeepStatus is a Follower's Said that writes its status at StatusKey of kv (the reader's StatusState).
func KeepStatus(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Status) {
	return func(s Status) {
		data, err := json.Marshal(s)
		if err == nil {
			_, err = kv.Put(ctx, StatusKey(s.Owner, s.State), data)
		}
		if err != nil && logf != nil {
			logf("views: the status of the first view %s is not kept: %v", s.Tool, err)
		}
	}
}

// HearAgain takes each hand ask written at an AgainKey of kv, those standing at the start included, asks that
// follower's first view again with the ask's words and who asked, and deletes the ask. It returns once the watch is set.
func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower, logf func(string, ...any)) error {
	w, err := kv.Watch(ctx, "again.>")
	if err != nil {
		return err
	}
	go func() {
		defer func() { _ = w.Stop() }()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				if e == nil || e.Operation() != jetstream.KeyValuePut {
					continue
				}
				var a HandAsk
				if err := json.Unmarshal(e.Value(), &a); err != nil {
					logf("views: the hand ask %s is not read: %v", e.Key(), err)
					_ = kv.Delete(ctx, e.Key())
					continue
				}
				i := slices.IndexFunc(followers, func(f *Follower) bool { return f.Owner == a.Owner && f.State == a.State })
				if i < 0 {
					logf("views: the hand ask %s names no first view this part holds", e.Key())
					_ = kv.Delete(ctx, e.Key())
					continue
				}
				logf("views: %s asks the first view %s again by hand: %s", a.By, followers[i].Tool, a.Why)
				_ = kv.Delete(ctx, e.Key(), jetstream.LastRevision(e.Revision()))
				f := followers[i]
				go func() { _ = f.Again(ctx, a.Why+" ("+a.By+")") }()
			}
		}
	}()
	return nil
}

// StatusTools are a reader's read tool of its first views (read) and its re-read tool (again), in its tools bundle,
// over its StatusState: read lists what the process last said of each; again writes a hand ask the process takes,
// naming the caller the runtime proved, and is recorded in the status's by-hand count and why. again is a hand act:
// acts records it through the seat mesh-hand-acts before the ask is written, and one the seat does not take is said and
// nothing is asked (the zero Recorder records nothing, so again refuses).
func StatusTools(reader, read, again string, st stdio.KeptState, acts handacts.Recorder, now func() time.Time) []stdio.Tool {
	return []stdio.Tool{
		{Name: read,
			Description: fmt.Sprintf("%s's first views of the indexer's and the assigner's states (novox/hq ADR 0615): per owner and state, "+
				"the tool it asks, the revision its picture held when last answered, when it asked and was answered, how many asks "+
				"(by hand among them) and whether its condition (unanswered past two minutes) is raised.", reader),
			Input: map[string]any{"type": "object", "properties": map[string]any{}},
			Run: func(map[string]any) (any, error) {
				keys, err := st.Keys()
				if err != nil {
					return nil, fmt.Errorf("the state %s cannot be listed: %w", StatusState, err)
				}
				out := []Status{}
				for _, k := range keys {
					if strings.HasPrefix(k, "again.") {
						continue
					}
					e, err := st.Get(k)
					if err != nil || e == nil {
						continue
					}
					var s Status
					if json.Unmarshal(e.Value, &s) == nil {
						out = append(out, s)
					}
				}
				return map[string]any{"first-views": out}, nil
			}},
		{Name: again,
			Description: fmt.Sprintf("Ask %s's first view of an owner's state again now, by hand (a hand act, its caller and why kept in "+
				"the status %s reads): for a picture a person doubts. The part asks once and follows the owner's events after the answer.", reader, read),
			Input: map[string]any{"type": "object", "required": []string{"owner", "state", "why"}, "properties": map[string]any{
				"owner": map[string]any{"type": "string", "description": "the owner: mesh-delivery-indexer or mesh-delivery-assigner"},
				"state": map[string]any{"type": "string", "description": "the owner's state, as the read tool names it"},
				"why":   map[string]any{"type": "string", "description": "why it is asked again, in plain words"}}},
			Run: func(args map[string]any) (any, error) {
				owner, _ := args["owner"].(string)
				state, _ := args["state"].(string)
				why, _ := args["why"].(string)
				if owner == "" || state == "" || strings.TrimSpace(why) == "" {
					return nil, errors.New("again names the owner, the state and why")
				}
				a := HandAsk{V: Version, Owner: owner, State: state, Why: strings.TrimSpace(why), By: stdio.CallerOf(args), At: now().UTC()}
				if _, err := acts.Record(handacts.Request{What: fmt.Sprintf("asked %s's first view of %s's %s again by hand", reader, owner, state),
					Why: a.Why, Cause: "first-view-again", Verb: again}); err != nil {
					return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was asked again: %w", err)
				}
				if _, err := st.Put(AgainKey(owner, state), a); err != nil {
					return nil, fmt.Errorf("the hand ask is not written: %w", err)
				}
				return map[string]any{"asked": a, "read": read}, nil
			}},
	}
}