mesh-sdk/go/process/gate.go

package process

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

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/approval"
	"git.novox.be/novox/mesh-sdk/go/busretry"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
)

// Gating is what a holder's gate reaches beyond this connection (design 47 G §677; lead answers A-219, A-224, A-228): the
// operator channel (mesh-contracts' operatorchannelclient.GateChannel over this connection, reading the router's record
// under the module's own name, §700) and the holder's held finding writer.
type Gating struct {
	Channel  approval.Channel
	Findings approval.FindingWriter
	// Build is the holder's own fingerprint; empty reads it from the process's environment (OwnBuild).
	Build string
	Logf  func(string, ...any)
}

// Gate is the holder's gate over this connection (§677): its state `approvals`, its calls of the seat `mesh-hand-acts`'s
// `warranted` to record a run, and its seat's `gated-ended`, each on the process's own account. AnswerSeat holds the
// seat's marked verbs through it (Gated); the process hears its decided with HearDecided and reads it at its start with
// the gate's StartHolding.
func (c *Conn) Gate(g Gating) *approval.Gate {
	build := g.Build
	if build == "" {
		build = OwnBuild().Fingerprint
	}
	state := c.State(approval.StateName)
	return approval.New(approval.Deps{Node: c.account.Node, Runner: c.account.Module, Holding: true, Build: build,
		State: state, Bounds: heldBounds{state}, Channel: g.Channel, Findings: g.Findings, Logf: g.Logf,
		Warranted: func(asker, ask string, open func() (bool, error)) error {
			_, err := c.CallKeptWhileOpen(busretry.Open{Key: ask, Open: func(string) (bool, error) { return open() }, Logf: g.Logf},
				"seat:mesh-hand-acts.warranted", map[string]string{"asker": asker, "ask": ask})
			return err
		},
		Say: func(subject string, body any, id string) error {
			return c.emitOn(subject, body, []stdio.EmitOptions{{MessageID: id}})
		}})
}

// heldBounds is the holder's bound keys in its own state approvals (lead answer A-255): a key with its own time to live,
// whose end the bus says as a marker that a watch hears, so a waiting call's 60 minutes need no clock in the process.
type heldBounds struct{ s State }

func (b heldBounds) CreateTTL(key string, value any, ttl time.Duration) error {
	_, err := b.s.CreateTTL(key, value, ttl)
	return err
}

func (b heldBounds) Delete(key string) error { return b.s.Delete(key) }

// Watch hands expired each key under pattern whose own time to live ended, heard from now on: the bus's marker of a time
// to live is a purge, which this state's own writes never make (a delete is not one).
func (b heldBounds) Watch(pattern string, expired func(key string)) (func(), error) {
	ctx, cancel := context.WithCancel(context.Background())
	bind, stop := context.WithTimeout(ctx, stateTimeout)
	kv, err := b.s.store(bind)
	stop()
	if err != nil {
		cancel()
		return nil, err
	}
	w, err := kv.Watch(ctx, watchPattern(pattern), jetstream.UpdatesOnly())
	if err != nil {
		cancel()
		return nil, fmt.Errorf("%s's state %s, watching %q: %w", b.s.Module, b.s.Name, pattern, err)
	}
	go func() {
		defer cancel()
		defer func() { _ = w.Stop() }()
		for e := range w.Updates() {
			if e != nil && e.Operation() == jetstream.KeyValuePurge {
				expired(e.Key())
			}
		}
	}()
	return cancel, nil
}

// HearDecided hands the gate each `decided` of the holder's own asks that hub hears (§677): only the ask's id is read
// from it, the gate reading the router's own record.
func HearDecided(g *approval.Gate, ch approval.Channel, module string, hub *EventHub) error {
	return hub.Handle(ch.Decided(module), func(m jetstream.Msg) {
		var w struct {
			Ask string `json:"ask"`
		}
		if err := json.Unmarshal(m.Data(), &w); err != nil || w.Ask == "" {
			_ = m.Term()
			return
		}
		g.Decided(w.Ask)
		_ = m.Ack()
	})
}

// HeldWithin bounds a caller's wait on a held call's end: the ask's 60 minutes and the 30 a run may come after its choice.
var HeldWithin = approval.AskLasts + approval.RunWithin

// awaitGated waits on the held call's own end on its seat's gated-ended, from the last said on its subject (so an end said
// before the wait began is read), and is its answer: `{answer, approval}` when it ran (lead answer A-158), its words as
// the call's error otherwise. Every failure is an answer, never asked again: the call reached its holder (ADR 0668 §2).
func (c *Conn) awaitGated(subject string, held approval.Held) (json.RawMessage, error) {
	seat, ok := seatOfSubject(subject)
	if !ok {
		return nil, answered(fmt.Sprintf("%s answered held, and is no seat's subject", subject))
	}
	js, err := jetstream.New(c.nc)
	if err != nil {
		return nil, answered(err.Error())
	}
	on := approval.GatedEndedSubject(seat, c.account.User, held.Call)
	stream := seatdef.KeptStream(seat, seatdef.Message{Name: approval.GatedEndedEvent})
	ctx, cancel := context.WithTimeout(context.Background(), HeldWithin)
	defer cancel()
	cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{on},
		DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
	if err != nil {
		return nil, answered(fmt.Sprintf("the call %s waits on ask %s, and its end on %s cannot be heard: %v; %s.approvals reads it",
			held.Call, held.Asked, stream, err, seat))
	}
	for {
		m, err := cons.Next(jetstream.FetchMaxWait(time.Minute))
		if err == nil {
			_ = m.Ack()
			return endOf(m.Data())
		}
		if ctx.Err() != nil || !errors.Is(err, jetstream.ErrNoMessages) && !errors.Is(err, context.DeadlineExceeded) {
			return nil, answered(fmt.Sprintf("the call %s waits on ask %s, and no end of it was heard (%v); %s.approvals reads it",
				held.Call, held.Asked, err, seat))
		}
	}
}

// endOf is a gated-ended's body as the call's answer.
func endOf(data []byte) (json.RawMessage, error) {
	var e approval.GatedEnded
	if err := json.Unmarshal(data, &e); err != nil {
		return nil, answered("the end of the call is not read: " + err.Error())
	}
	if e.Outcome != approval.OutcomeRan || e.Error != "" {
		words := e.Error
		if words == "" {
			words = "the call ended " + e.Outcome
		}
		return nil, answered(fmt.Sprintf("Not run: %s (ask %s, call %s)", words, e.Ask, e.Call))
	}
	return json.Marshal(approval.Answered{Answer: e.Answer, Approval: e.Approval})
}

// seatOfSubject is the seat a call subject `mesh.seat.<seat>.…` names.
func seatOfSubject(subject string) (string, bool) {
	rest, ok := strings.CutPrefix(subject, "mesh.seat.")
	if !ok {
		return "", false
	}
	seat, _, ok := strings.Cut(rest, ".")
	return seat, ok && seat != ""
}