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

package approval

import (
	"context"
	"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"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/outward"
)

// Wait is one mesh_wait call as the gate judges it (design 34 every-event §34 to §36): the event names given, the subjects
// they became (after `from`), the filter and the timeout.
type Wait struct {
	Names    []string
	From     string
	Subjects []string
	Filter   map[string]any
	Timeout  time.Duration
}

// Waited is a wait through the gate (every-event §34 to §36). An unmarked wait is heard at once by hear, given how long it
// may wait. A marked one is heard only on the operator's warrant for that one wait, its timeout running from then; with
// none it subscribes to nothing, its ask raised once under the runner's own module's name, and is answered as an error at
// its timeout, when its caller goes, or as soon as the router's record says the ask was chosen, never in a hot loop. c is
// the call as the console makes it: its address, its tool's name, its caller and its arguments as given.
func (g *Gate) Waited(ctx context.Context, c Call, w Wait, hear func(within time.Duration) (any, error)) (any, error) {
	deadline := g.d.Now().Add(w.Timeout)
	// A wait from a seat is never marked, and a seat's event never is (§32, §34).
	if w.From == "seat" || !outsideSeats(w.Subjects) {
		return hear(w.Timeout)
	}
	if g.d.Catalog == nil {
		return nil, g.holdRefused(ctx, deadline, "this wait is refused: the tool runner reads no catalog half, so the marks are not read")
	}
	if err := g.d.Catalog.Ready(); err != nil {
		// Before the marks are read nothing is asked: the wait waits for the first view within its own timeout and is
		// then judged, and is refused at its timeout when it does not come (§34).
		if g.d.Wait != nil {
			err = g.waitCatalog(ctx, w.Timeout)
		}
		if err != nil {
			return nil, g.holdRefused(ctx, deadline, fmt.Sprintf("this wait is refused at its timeout: the marks are not yet read (%v); "+
				"mesh_first_views says what the tool runner waits for", err))
		}
	}
	if !overlapsAny(w.Subjects, g.waitMarks()) {
		return hear(deadline.Sub(g.d.Now()))
	}
	c.Module, c.Seat, c.Fingerprint = g.d.Runner, "", g.d.Build
	if c.Fingerprint == "" {
		return nil, g.holdRefused(ctx, deadline, "this wait needs the operator's approval, and the tool runner's own fingerprint is "+
			"unknown (its MESH_FINGERPRINT was not set at its start), so no warrant can bind its build")
	}
	if b, ok := g.d.State.(Binding); g.d.State == nil || ok && !b.Bound() {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, "", "the tool runner's membership binds no state "+StateName+" to keep its asks in").Error())
	}
	if why := refusedUnasked(c.Args); why != "" {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, "", why).Error())
	}
	if why := g.held(); why != "" {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, "", why).Error())
	}
	act, err := ActOf(c, g.d.Node)
	if err != nil {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, "", err.Error()).Error())
	}
	digest, err := operatorchannelclient.ActDigest(act)
	if err != nil {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, "", err.Error()).Error())
	}
	named, err := g.d.State.Get(g.actKey(digest))
	if err != nil {
		g.stateFailed(err)
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()).Error())
	}
	g.stateRead()
	anew := func() (any, error) {
		id := newID()
		if err := g.keepAsk(c, act, digest, g.askForWait(c, w, digest, id)); err != nil {
			return nil, g.holdRefused(ctx, deadline, err.Error())
		}
		return nil, g.holdAsk(ctx, deadline, c.Module, id, digest, true)
	}
	if named == nil {
		return anew()
	}
	var id string
	if err := json.Unmarshal(named.Value, &id); err != nil || id == "" {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, fmt.Sprintf("its act's key in %s names no ask", StateName)).Error())
	}
	rec, err := g.record(c.Module, id)
	switch {
	case err != nil:
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router's record of its ask "+id+" cannot be read: "+err.Error()).Error())
	case rec == nil, rec.State == "open", rec.State == "held-back":
		return nil, g.holdAsk(ctx, deadline, c.Module, id, digest, false)
	case rec.State == "refused":
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router refused its ask "+id+": "+orNone(rec.Answer)).Error())
	case rec.State != "chosen":
		return anew() // expired, cancelled, replaced: asked again (§15, §17)
	}
	wt := rec.Warrant
	if wt == nil {
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router's record of its ask "+id+" is chosen and carries no warrant").Error())
	}
	at, _ := operatorchannelclient.Time(wt.At)
	if wt.Option == OptionDecline {
		if g.d.Now().Before(at.Add(DeclinedFor)) {
			return nil, g.holdRefused(ctx, deadline, declinedWords(id, at))
		}
		return anew()
	}
	if g.d.Now().After(at.Add(RunWithin)) {
		return anew()
	}
	var answer any
	var ran error
	h, err := g.client(c.Module).Take(*wt, func(o seat.Option, _ seat.Warrant) error {
		if err := operatorchannelclient.Performs(o, act); err != nil {
			ran = err
			return err
		}
		// On the warrant it subscribes, and the wait's timeout runs from then (§35).
		answer, ran = hear(w.Timeout)
		return nil
	})
	switch {
	case h == operatorchannelclient.Done:
		return anew() // already run: one warrant is one wait (§35)
	case h == operatorchannelclient.Refused:
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its warrant does not hold: "+errText(err)).Error())
	case h == "" && err != nil:
		g.stateFailed(err)
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its state "+StateName+" cannot be written: "+err.Error()).Error())
	case err != nil:
		return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, err.Error()).Error())
	}
	return g.ran(c, id, rec, answer, ran)
}

// outsideSeats says a subject is not a seat's: only such a subject can be marked (§32, §34).
func outsideSeats(subjects []string) bool {
	for _, s := range subjects {
		if !strings.HasPrefix(s, "mesh.seat.") {
			return true
		}
	}
	return false
}

// waitMarks are the marked subjects of every module the catalog half lists (§33): the newest entry of every followed
// branch, and the served entry of each module this runner serves (lead answer A-168), so a mark is never lost.
func (g *Gate) waitMarks() map[string]string {
	m := Marks{Tools: map[string]string{}, Verbs: map[string]string{}, Subjects: map[string]string{}}
	defs := g.d.Catalog.Definitions()
	for _, e := range g.d.Catalog.Entries() {
		m.add(e, defs)
	}
	g.mu.Lock()
	served := make(map[string]string, len(g.served))
	for module, fp := range g.served {
		served[module] = fp
	}
	g.mu.Unlock()
	for _, fp := range served {
		if e, ok, _ := g.d.Catalog.Served(fp); ok {
			m.add(e, defs)
		}
	}
	return m.Subjects
}

// overlapsAny says a subject a wait subscribes overlaps a marked subject (§34).
func overlapsAny(subjects []string, marked map[string]string) bool {
	for _, s := range subjects {
		for mk := range marked {
			if buscompose.Overlaps(s, mk) {
				return true
			}
		}
	}
	return false
}

// waitCatalog waits for the catalog half's first view within the bound, or until the caller goes.
func (g *Gate) waitCatalog(ctx context.Context, within time.Duration) error {
	done := make(chan error, 1)
	go func() { done <- g.d.Wait(within) }()
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

// pause waits until the deadline or the caller's going.
func (g *Gate) pause(ctx context.Context, deadline time.Time) {
	if left := deadline.Sub(g.d.Now()); left > 0 {
		select {
		case <-g.d.After(left):
		case <-ctx.Done():
		}
	}
}

// holdRefused is a refused marked wait, said at its timeout or when its caller goes, never at once, so a watcher's loop
// is never a hot one (§36).
func (g *Gate) holdRefused(ctx context.Context, deadline time.Time, words string) error {
	g.pause(ctx, deadline)
	if !strings.HasPrefix(words, "Not run") {
		words = "Not run: " + words
	}
	return fmt.Errorf("%s", words)
}

// holdAsk holds a marked wait whose ask is open, subscribing to nothing but its ask's `decided` (§36): it answers, as an
// error, at its timeout, when its caller goes, or once the router's record says the ask was chosen, the event only waking it.
func (g *Gate) holdAsk(ctx context.Context, deadline time.Time, module, id, digest string, first bool) error {
	for {
		left := deadline.Sub(g.d.Now())
		heard := false
		if left > 0 && ctx.Err() == nil {
			var err error
			if g.d.Hear != nil {
				heard, err = g.d.Hear(ctx, []string{seat.SubjectDecided(module)}, left, func(body json.RawMessage) bool {
					var w seat.Warrant
					return json.Unmarshal(body, &w) == nil && w.Ask == id
				})
			}
			if g.d.Hear == nil || err != nil {
				g.pause(ctx, deadline)
			}
		}
		rec, _ := g.record(module, id)
		if heard && (rec == nil || rec.State != "chosen") {
			continue // the event only wakes it: the record says nothing chosen yet
		}
		return g.waitWords(rec, id, digest, first)
	}
}

// waitWords are §12's and §17's words for a marked wait not run, "this wait" for "the call" (§36).
func (g *Gate) waitWords(rec *seat.AsksByAskerIDValue, id, digest string, first bool) error {
	switch {
	case rec == nil:
		return fmt.Errorf("Not run, asked; not yet taken by the router (ask %s, fingerprint %s); make this wait again later", id, Fingerprint(digest))
	case rec.State == "chosen" && rec.Warrant != nil && rec.Warrant.Option == OptionDecline:
		at, _ := operatorchannelclient.Time(rec.Warrant.At)
		return fmt.Errorf("%s", declinedWords(id, at))
	case rec.State == "chosen", first:
		return fmt.Errorf("Not run: this wait needs the operator's approval. Asked on %s at %s (ask %s, fingerprint %s); make this wait "+
			"again within 30 minutes of the approval; the ask expires at %s", orNone(strings.Join(rec.OfferedOn, ", ")), rec.Opened,
			id, Fingerprint(digest), rec.Expires)
	case rec.State == "open", rec.State == "held-back":
		return fmt.Errorf("Not run: still waiting for the operator's answer to ask %s (fingerprint %s); it expires at %s",
			id, Fingerprint(digest), rec.Expires)
	}
	return fmt.Errorf("Not run: the ask %s ended %s; make this wait again to ask anew", id, rec.State)
}

func declinedWords(id string, at time.Time) string {
	return fmt.Sprintf("Not run: the operator declined ask %s; this wait is not asked again until %s", id,
		operatorchannelclient.FormatTime(at.Add(DeclinedFor)))
}

// askForWait is a marked wait's ask (§35): "Wait on <names> on <node>?", shortened as §10 says, and an explanation naming
// every event name one per line, the filter, and the bound as the timeout in minutes, then the act's fingerprint.
func (g *Gate) askForWait(c Call, w Wait, digest, id string) seat.Ask {
	a := g.askFor(c, digest, id)
	filter, whole := "none", true
	if len(w.Filter) > 0 {
		raw, _ := json.Marshal(w.Filter)
		filter = string(raw)
		if r, ok := outward.Check(filter); !ok {
			filter, whole = shapeOf(r), false
		}
	}
	a.Headline = headlineOf("Wait on", strings.Join(w.Names, ", "), g.d.Node, whole)
	lines := append([]string{"Wait on " + g.d.Node + ", called through the mesh's tools, which cannot tell who called, for:"}, w.Names...)
	lines = append(lines, "Filter: "+filter, fmt.Sprintf("Timeout: %s minutes", minutes(w.Timeout)), "Fingerprint "+Fingerprint(digest))
	a.Explanation = strings.Join(lines, "\n")
	a.Options[0].Does = "the same wait, made again within 30 minutes, subscribes once, its timeout running from then"
	a.Options[1].Does = "nothing is heard; the same wait is not asked again for 10 minutes"
	return a
}

// minutes is a timeout in minutes, as few digits as say it.
func minutes(d time.Duration) string {
	return strings.TrimSuffix(strings.TrimRight(fmt.Sprintf("%.2f", d.Minutes()), "0"), ".")
}