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

package approval

import (
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"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"
)

// StateName is the runner's own state its asks and claims are kept in, per machine (§13).
const StateName = "approvals"

// Store is the runner's state `approvals` (§13): a key's value, a create that refuses a key that exists and an update by
// compare-and-set, each failing with stdio.ErrChanged when another write came first; a put, and its keys.
type Store interface {
	Get(key string) (*stdio.StateEntry, error)
	Create(key string, value any) (uint64, error)
	Update(key string, value any, revision uint64) (uint64, error)
	Put(key string, value any) (uint64, error)
	Keys() ([]string, error)
}

// The bounds of an ask (§10, §12, §13, §14, §17).
const (
	AskLasts      = 60 * time.Minute
	RecordWithin  = 5 * time.Second
	RunWithin     = 30 * time.Minute
	DeclinedFor   = 10 * time.Minute
	MostKeptValue = 256 * 1024
	OnExpiry      = "nothing runs; call again to ask anew"
)

// The options every ask offers (§10), both at level approve, both binding the act's digest.
const (
	OptionApprove = "approve"
	OptionDecline = "decline"
)

// client is operator-channel's client asking under the marked module's name, its asks kept in the runner's state under
// this machine's keys, its publish and the router's record on the runner's own account (§7, §10).
func (g *Gate) client(module string) *operatorchannelclient.Client {
	return &operatorchannelclient.Client{Module: module, State: g.d.State, KeyPrefix: g.d.Node + ".", Now: g.d.Now,
		Publish: func(subject string, body any, id string) error { return g.d.Publish(module, subject, body, id) },
		Record:  func(id string) (*stdio.StateEntry, error) { return g.d.Record(module, id) }}
}

// actKey is the key naming the newest ask of an act (§13).
func (g *Gate) actKey(digest string) string { return g.d.Node + ".act." + digest }

// askKey is the key the client keeps an ask under (§13).
func (g *Gate) askKey(id string) string { return g.d.Node + ".ask." + id }

// newID is an ask's id: 128 random bits.
func newID() string {
	var b [16]byte
	_, _ = rand.Read(b[:])
	return hex.EncodeToString(b[:])
}

// askFor is the ask a marked call raises (§10, §11).
func (g *Gate) askFor(c Call, digest, id string) seat.Ask {
	args, whole := shown(c.Args)
	name := c.Name
	if c.Seat != "" {
		name = c.Seat + "." + c.Name
	}
	headline := headlineOf("Run", name, g.d.Node, whole)
	first, _, _ := strings.Cut(strings.TrimSpace(c.Description), "\n")
	if first == "" {
		first = name
	}
	lines := []string{first, "On " + g.d.Node + ", called through the mesh's tools, which cannot tell who called."}
	if args != "" {
		lines = append(lines, args)
	}
	if !whole {
		lines = append(lines, "The whole is read with mesh_approvals on "+g.d.Node+", whose fingerprint must be the one on the phone.")
	}
	lines = append(lines, "Fingerprint "+Fingerprint(digest))
	return seat.Ask{ID: id, About: digest, Who: operatorchannelclient.Operator, Headline: headline,
		Explanation: strings.Join(lines, "\n"), Expires: operatorchannelclient.FormatTime(g.d.Now().Add(AskLasts)), OnExpiry: OnExpiry,
		Options: []seat.Option{
			{ID: OptionApprove, Label: "Approve", Does: "the same call, made again within 30 minutes, runs once",
				Level: operatorchannelclient.Approve, Binds: digest},
			{ID: OptionDecline, Label: "Decline", Does: "nothing runs; the same call is not asked again for 10 minutes",
				Level: operatorchannelclient.Approve, Binds: digest}}}
}

// headlineOf is "<verb> <names> on <node>?", "<verb> <names>?" when that is longer than the router allows, and the names cut
// with "…" when that too is longer; ending "Part shown" whenever an argument is not shown whole (§10, Q-HL). verb is "Run"
// for a call and "Wait on" for a wait (design 34 every-event §35).
func headlineOf(verb, name, node string, whole bool) string {
	suffix := ""
	if !whole {
		suffix = " Part shown"
	}
	max := operatorchannelclient.HeadlineLength
	fits := func(s string) bool { return len([]rune(s)) <= max }
	if h := verb + " " + name + " on " + node + "?" + suffix; fits(h) {
		return h
	}
	if h := verb + " " + name + "?" + suffix; fits(h) {
		return h
	}
	room := max - len([]rune(verb+" …?"+suffix))
	r := []rune(name)
	if room < 1 {
		room = 1
	}
	if len(r) > room {
		r = r[:room]
	}
	return verb + " " + string(r) + "…?" + suffix
}

// firstCall is a marked call whose act has no ask open: refused at once when its arguments may not be asked (§9), else
// asked (§10, §12).
func (g *Gate) firstCall(c Call) (any, error) {
	if why := refusedUnasked(c.Args); why != "" {
		return nil, g.refused(c, "", why)
	}
	act, err := ActOf(c, g.d.Node)
	if err != nil {
		return nil, g.refused(c, "", err.Error())
	}
	digest, err := operatorchannelclient.ActDigest(act)
	if err != nil {
		return nil, g.refused(c, "", err.Error())
	}
	return nil, g.raise(c, act, digest, "")
}

// raise raises the ask of a marked call whose act has none open (§10): kept in the runner's state with the act and the whole
// arguments, published under the module's name, its act's key naming it; then the 5 seconds for the router's record and
// the first answer (§12), always an error, never a result.
func (g *Gate) raise(c Call, act operatorchannelclient.Act, digest, markedSince string) error {
	id := newID()
	if err := g.keepAsk(c, act, digest, g.askFor(c, digest, id)); err != nil {
		return err
	}
	return g.firstAnswer(c, id, digest, markedSince)
}

// keepAsk raises an ask and keeps it in the runner's state with the act and the whole arguments, its act's key naming it
// (§10, §13); refused, as an error, when it cannot be.
func (g *Gate) keepAsk(c Call, act operatorchannelclient.Act, digest string, a seat.Ask) error {
	cl := g.client(c.Module)
	if size := keptSize(a, act, c.Args); size > MostKeptValue {
		return g.refused(c, digest, fmt.Sprintf("its ask would keep %d bytes, more than the %d a value of the state %s holds", size, MostKeptValue, StateName))
	}
	if err := cl.Ask(a); err != nil {
		return g.refused(c, digest, "the ask could not be raised: "+err.Error())
	}
	if err := g.keepCall(a.ID, act, c.Args); err != nil {
		g.stateFailed(err)
		return g.refused(c, digest, "the ask was raised and its call could not be kept: "+err.Error())
	}
	if _, err := g.d.State.Put(g.actKey(digest), a.ID); err != nil {
		g.stateFailed(err)
		return g.refused(c, digest, "the ask was raised and its act could not be kept: "+err.Error())
	}
	return nil
}

// keptSize is the bytes the kept ask would take.
func keptSize(a seat.Ask, act operatorchannelclient.Act, args json.RawMessage) int {
	raw, _ := json.Marshal(operatorchannelclient.Kept{Ask: a, Act: act, Arguments: args})
	return len(raw)
}

// keepCall writes the act and the whole arguments beside the ask the client kept (§13).
func (g *Gate) keepCall(id string, act operatorchannelclient.Act, args json.RawMessage) error {
	e, err := g.d.State.Get(g.askKey(id))
	if err != nil || e == nil {
		return fmt.Errorf("the kept ask %s cannot be read: %v", id, err)
	}
	var k operatorchannelclient.Kept
	if err := json.Unmarshal(e.Value, &k); err != nil {
		return err
	}
	k.Act, k.Arguments = act, args
	_, err = g.d.State.Update(g.askKey(id), k, e.Revision)
	return err
}

// firstAnswer waits up to RecordWithin for the router's record of the ask and says the call was not run (§12).
func (g *Gate) firstAnswer(c Call, id, digest, markedSince string) error {
	rec, _ := g.awaitRecord(c.Module, id)
	since := ""
	if markedSince != "" {
		since = "; marked since revision " + markedSince
	}
	if rec == nil {
		return fmt.Errorf("Not run, asked; not yet taken by the router (ask %s, fingerprint %s); make the same call again later%s",
			id, Fingerprint(digest), since)
	}
	return fmt.Errorf("Not run: this needs the operator's approval. Asked on %s at %s (ask %s, fingerprint %s); make the same call "+
		"again within 30 minutes of the approval; the ask expires at %s%s", orNone(strings.Join(rec.OfferedOn, ", ")), rec.Opened,
		id, Fingerprint(digest), rec.Expires, since)
}

// record is the router's record of an ask under the module's name, nil when it holds none.
func (g *Gate) record(module, id string) (*seat.AsksByAskerIDValue, error) {
	e, err := g.d.Record(module, id)
	if err != nil || e == nil {
		return nil, err
	}
	var rec seat.AsksByAskerIDValue
	if err := json.Unmarshal(e.Value, &rec); err != nil {
		return nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err)
	}
	return &rec, nil
}

// awaitRecord reads the router's record of a new ask until it is there, within RecordWithin: the seat answers a key's
// value and says nothing when the router takes an ask, so it is read again after a growing pause (§12).
func (g *Gate) awaitRecord(module, id string) (*seat.AsksByAskerIDValue, error) {
	until := g.d.Now().Add(RecordWithin)
	pause := 50 * time.Millisecond
	for {
		rec, err := g.record(module, id)
		if rec != nil || g.d.Now().Add(pause).After(until) {
			return rec, err
		}
		<-g.d.After(pause)
		pause *= 2
	}
}

// refused is a marked call refused, run nothing, written in the runner's log with the ask's fingerprint and never its
// arguments (§19).
func (g *Gate) refused(c Call, digest, why string) error {
	g.d.Logf("[mesh-tools] %s on %s is refused (fingerprint %s): %s", c.Address, g.d.Node, orNone(Fingerprint(digest)), why)
	return fmt.Errorf("Not run: %s needs the operator's approval and %s", c.Address, why)
}