mesh-contracts/go/clients/operatorchannelclient/client.go

package operatorchannelclient

// Client is a module's asking of a person (novox/hq ADR 0259 §10): it publishes an ask under the module's own name,
// cancels it, reads its state from the router's record, and hands the module each warrant once — only one for an ask
// it made, for an option it offered, at that option's level, before it expired. The module then acts with its own
// grants.
//
// It keeps what it asked, and which warrants it acted on, in a state of the module's own (declared in its manifest,
// by default `asked`): the ask is written before it is published, and a warrant is marked acted on by compare-and-set
// before the module's handler runs, so a redelivery, or a second instance, acts once.

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

	seat "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
	stdio "git.novox.be/novox/mesh-sdk/go"
)

// Client asks as one module.
type Client struct {
	// Module is the asking module's name: the one the bus grants its asks under.
	Module string
	// State is the module's own state the asks are kept in.
	State Store
	// Now is the clock; nil is time.Now.
	Now func() time.Time
	// Publish, Record and Subscribe reach the bus; nil is the stdio SDK's, through the module's runtime.
	Publish   func(subject string, body any, id string) error
	Record    func(key string) (*stdio.StateEntry, error)
	Subscribe func(pattern string, handle func(stdio.Envelope) error) error
}

// Store is where the asks are kept: stdio.State(name) is one.
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)
}

// New is a client for the module the runtime serves, keeping its asks in the state named.
func New(state string) *Client {
	return &Client{Module: os.Getenv("MESH_MODULE"), State: stdio.State(state)}
}

func (c *Client) now() time.Time {
	if c.Now != nil {
		return c.Now()
	}
	return time.Now()
}

// Kept is one ask as the module keeps it.
type Kept struct {
	Ask    seat.Ask      `json:"ask"`
	Asked  time.Time     `json:"asked"`
	Acted  bool          `json:"acted,omitempty"`
	Ended  string        `json:"ended,omitempty"`
	Answer *seat.Warrant `json:"answer,omitempty"`
}

func key(id string) string { return "ask." + id }

// Ask checks an ask, keeps it, and publishes it under the module's own name. An ask with an id already kept
// is refused: a new question is a new id.
func (c *Client) Ask(a seat.Ask) error {
	if err := Check(a, c.now()); err != nil {
		return err
	}
	if _, err := c.State.Create(key(a.ID), Kept{Ask: a, Asked: c.now()}); err != nil {
		if errors.Is(err, stdio.ErrChanged) {
			return fmt.Errorf("an ask %s was already made: a new question is a new id", a.ID)
		}
		return err
	}
	return c.publish(seat.AcceptSubjectAsk(c.Module), a, a.ID)
}

// Cancel takes an ask back: the router ends it and edits every copy.
func (c *Client) Cancel(id string) error {
	return c.publish(seat.AcceptSubjectCancel(c.Module), map[string]string{"id": id}, "cancel."+id)
}

func (c *Client) publish(subject string, body any, id string) error {
	if c.Publish != nil {
		return c.Publish(subject, body, id)
	}
	return stdio.SeatPublish(subject, body, id)
}

// Where is an ask's state as the router keeps it: its state and its warrant, if any — read under the
// module's own name and no other's.
func (c *Client) Where(id string) (state string, w *seat.Warrant, err error) {
	read := c.Record
	if read == nil {
		read = func(k string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", k) }
	}
	e, err := read(id)
	if err != nil || e == nil {
		return "", nil, err
	}
	var rec struct {
		State   string        `json:"state"`
		Warrant *seat.Warrant `json:"warrant"`
	}
	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.State, rec.Warrant, nil
}

// Handled is what came of a warrant: acted on, refused (and why), or already done.
type Handled string

const (
	Acted    Handled = "acted"
	Refused  Handled = "refused"
	Done     Handled = "already acted on"
	Recorded Handled = "ended without a choice"
)

// Take hands act a warrant once, after holding it to the ask it answers. act performs the option with the
// module's own grants; its error is said and the warrant stays acted on (a person's decision is not
// performed twice; the module says the failure in its own record).
func (c *Client) Take(w seat.Warrant, act func(seat.Option, seat.Warrant) error) (Handled, error) {
	if w.Asker != c.Module {
		return Refused, fmt.Errorf("a warrant for %s reached %s", w.Asker, c.Module)
	}
	e, err := c.State.Get(key(w.Ask))
	if err != nil {
		return "", err
	}
	if e == nil {
		return Refused, fmt.Errorf("%s holds no ask %s", c.Module, w.Ask)
	}
	var k Kept
	if err := json.Unmarshal(e.Value, &k); err != nil {
		return "", err
	}
	if k.Acted || k.Ended != "" {
		return Done, nil
	}
	if w.Outcome != OutcomeChosen {
		k.Ended = w.Outcome
		k.Answer = &w
		if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil && !errors.Is(err, stdio.ErrChanged) {
			return "", err
		}
		return Recorded, nil
	}
	o, err := For(w, c.Module, k.Ask)
	if err != nil {
		return Refused, err
	}
	// Marked before acting: of two deliveries, or two instances, only the one whose write stands acts.
	k.Acted, k.Ended, k.Answer = true, OutcomeChosen, &w
	if _, err := c.State.Update(key(w.Ask), k, e.Revision); err != nil {
		if errors.Is(err, stdio.ErrChanged) {
			return Done, nil
		}
		return "", err
	}
	return Acted, act(o, w)
}

// Listen hands every warrant for this module to Take, with act. It asks the runtime for the module's
// events matching the router's `decided` subject for this module.
func (c *Client) Listen(act func(seat.Option, seat.Warrant) error, said func(Handled, seat.Warrant, error)) error {
	subscribe := c.Subscribe
	if subscribe == nil {
		subscribe = stdio.Subscribe
	}
	return subscribe(seat.Seat+"."+seat.EventDecided+"."+c.Module, func(e stdio.Envelope) error {
		if !strings.HasPrefix(e.Key, seat.Seat+"."+seat.EventDecided+".") {
			return nil
		}
		var w seat.Warrant
		if err := json.Unmarshal(e.Body, &w); err != nil {
			if said != nil {
				said(Refused, w, fmt.Errorf("a warrant could not be read: %w", err))
			}
			return nil
		}
		h, err := c.Take(w, act)
		if h == "" && err != nil {
			return err // the state could not be reached: offered again
		}
		if said != nil {
			said(h, w, err)
		}
		return nil
	})
}