mesh-tools/node-tools/internal/approval/record.go

package approval

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

	"git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	stdio "git.novox.be/novox/mesh-sdk/go"
)

// The bounds of a run's recording (§13, §16; lead answers A-157, A-158).
const (
	UnansweredAfter = 5 * time.Minute
	KeptFor         = 48 * time.Hour
	SaidBefore      = time.Hour
)

// Bounds is the runner's key-ttl state `bound` (lead answers A-157, A-158): a key created with its own time to live, which
// the bus removes when it passes, every watch hearing the removal as an expiry, so the gate's waits need no timer.
type Bounds interface {
	CreateTTL(key string, value any, ttl time.Duration) error
	Delete(key string) error
	// Watch hands expired every key under the pattern whose time to live passed, until the stop is called.
	Watch(pattern string, expired func(key string)) (stop func(), err error)
}

// Answered is a call run on the operator's approval: the tool's own answer and the line naming the approval (§16), in the
// form a process holder's `gated-ended` says it (design 47 G §677; lead answer A-158).
type Answered struct {
	Answer   any    `json:"answer"`
	Approval string `json:"approval"`
}

// The bound keys of a run (lead answer A-158; §13).
func (g *Gate) unansweredKey(id string) string { return g.d.Node + ".unanswered." + id }
func (g *Gate) unrecordedKey(id string) string { return g.d.Node + ".unrecorded." + id }

// ran is a call run on the operator's approval: answered with the tool's answer and the approval's line, and its warrant
// recorded in the hand-act log (§16).
func (g *Gate) ran(c Call, id string, rec *seat.AsksByAskerIDValue, answer any, err error) (any, error) {
	channel := ""
	at := ""
	if w := rec.Warrant; w != nil {
		channel, at = w.Channel, w.At
	}
	if channel == "" {
		channel = strings.Join(rec.OfferedOn, ", ")
	}
	line := fmt.Sprintf("run on the operator's approval via %s at %s, ask %s", orNone(channel), at, id)
	g.d.Logf("[mesh-tools] %s on %s %s", c.Address, g.d.Node, line)
	g.bindRecording(c.Module, id)
	g.d.Go(func() { g.recordRun(c.Module, id) })
	if err != nil {
		return nil, fmt.Errorf("%w (%s)", err, line)
	}
	return Answered{Answer: answer, Approval: line}, nil
}

// bindRecording keeps the run's two bound keys: one that raises `call-unanswered` when `warranted` has not answered within
// five minutes, one that says an unrecorded run an hour before its key's 48 hours end (§13, §16).
func (g *Gate) bindRecording(module, id string) {
	if g.d.Bounds == nil {
		return
	}
	for key, ttl := range map[string]time.Duration{g.unansweredKey(id): UnansweredAfter, g.unrecordedKey(id): KeptFor - SaidBefore} {
		if err := g.d.Bounds.CreateTTL(key, map[string]string{"asker": module, "ask": id}, ttl); err != nil {
			g.d.Logf("[mesh-tools] the bound %s of the run of ask %s is not kept: %v", key, id, err)
		}
	}
}

// recordRun calls `mesh-hand-acts.warranted` for one acted ask, each try a kept call open while the ask lacks `recorded`
// (§16, A-157), and on its answer writes `recorded` and clears what waited on it. One recording per ask at a time.
func (g *Gate) recordRun(module, id string) {
	if !g.recording(id, true) {
		return
	}
	defer g.recording(id, false)
	if g.d.Warranted == nil {
		g.d.Logf("[mesh-tools] the run of ask %s is not recorded: the tool runner reaches no hand-act log", id)
		return
	}
	if err := g.d.Warranted(module, id, func() (bool, error) { return g.unrecorded(id) }); err != nil {
		g.d.Logf("[mesh-tools] mesh-hand-acts.warranted has not answered for ask %s (%v): asked again when the hand-act seat's "+
			"holder is heard again and at the next start", id, err)
		return
	}
	if err := g.markRecorded(id); err != nil {
		g.d.Logf("[mesh-tools] ask %s is recorded in the hand-act log and its mark recorded is not kept: %v", id, err)
		return
	}
	if g.d.Bounds != nil {
		_ = g.d.Bounds.Delete(g.unansweredKey(id))
		_ = g.d.Bounds.Delete(g.unrecordedKey(id))
	}
	g.finding(id, KindCallUnanswered, false, "mesh-hand-acts.warranted answered for ask "+id, "")
}

// recording claims or releases an ask's recording: false when one runs already.
func (g *Gate) recording(id string, start bool) bool {
	g.mu.Lock()
	defer g.mu.Unlock()
	if g.inFlight == nil {
		g.inFlight = map[string]bool{}
	}
	if start {
		if g.inFlight[id] {
			return false
		}
		g.inFlight[id] = true
		return true
	}
	delete(g.inFlight, id)
	return true
}

// kept is the ask the client keeps under its id, with its revision; nil when there is none.
func (g *Gate) kept(id string) (*operatorchannelclient.Kept, uint64, error) {
	e, err := g.d.State.Get(g.askKey(id))
	if err != nil || e == nil {
		return nil, 0, err
	}
	var k operatorchannelclient.Kept
	if err := json.Unmarshal(e.Value, &k); err != nil {
		return nil, 0, err
	}
	return &k, e.Revision, nil
}

// unrecorded says the ask is acted on and its run not yet recorded: the key that keeps `warranted` open (§16).
func (g *Gate) unrecorded(id string) (bool, error) {
	k, _, err := g.kept(id)
	if err != nil || k == nil {
		return false, err
	}
	return k.Acted && !k.Recorded, nil
}

// markRecorded writes `recorded` on the kept ask by compare-and-set, again when another write came first.
func (g *Gate) markRecorded(id string) error {
	for {
		k, rev, err := g.kept(id)
		if err != nil || k == nil {
			return fmt.Errorf("the kept ask cannot be read: %v", err)
		}
		if k.Recorded {
			return nil
		}
		k.Recorded = true
		if _, err := g.d.State.Update(g.askKey(id), k, rev); err == nil {
			return nil
		} else if !isChanged(err) {
			return err
		}
	}
}

// Recover calls `warranted` again for every acted, unrecorded ask this machine's keys hold within their 48 hours (§16): at
// each start, and when the hand-act seat's holder is heard again (A-157).
func (g *Gate) Recover() {
	if g.d.State == nil {
		return
	}
	keys, err := g.d.State.Keys()
	if err != nil {
		g.d.Logf("[mesh-tools] the acted, unrecorded asks cannot be read (%v): read again at the next start or holder heard", err)
		return
	}
	prefix := g.d.Node + ".ask."
	for _, key := range keys {
		if !strings.HasPrefix(key, prefix) {
			continue
		}
		id := strings.TrimPrefix(key, prefix)
		k, _, err := g.kept(id)
		if err != nil || k == nil || !k.Acted || k.Recorded || g.d.Now().Sub(k.Asked) > KeptFor {
			continue
		}
		module := moduleOf(k)
		g.d.Go(func() { g.recordRun(module, id) })
	}
}

// moduleOf is the module an ask was asked under: the warrant's asker.
func moduleOf(k *operatorchannelclient.Kept) string {
	if k.Answer != nil {
		return k.Answer.Asker
	}
	return ""
}

// WatchBounds hears the run's bound keys expire (A-157): `call-unanswered` raised when `warranted` has not answered within
// five minutes; an unrecorded run said in the log an hour before its key's 48 hours end (§13).
func (g *Gate) WatchBounds() (func(), error) {
	if g.d.Bounds == nil {
		return func() {}, nil
	}
	return g.d.Bounds.Watch(g.d.Node+".**", func(key string) {
		if key == g.unreadableKey() {
			g.unreadableExpired()
			return
		}
		rest := strings.TrimPrefix(key, g.d.Node+".")
		kind, id, _ := strings.Cut(rest, ".")
		if open, err := g.unrecorded(id); err != nil || !open {
			return
		}
		switch kind {
		case "unanswered":
			g.finding(id, KindCallUnanswered, true, fmt.Sprintf("mesh-hand-acts.warranted has not answered for ask %s, run on the "+
				"operator's approval on %s, within %s: the run is not yet in the hand-act log", id, g.d.Node, UnansweredAfter),
				"a run's hand-act is unrecorded on "+g.d.Node)
		case "unrecorded":
			g.d.Logf("[mesh-tools] ERROR the run of ask %s on %s is still unrecorded an hour before its key's 48 hours end", id, g.d.Node)
		}
	})
}

// isChanged says a compare-and-set lost to another write.
func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) }