mesh-sdk/go/seat.go

package stdio

import (
	"encoding/json"
	"strings"
	"time"
)

// Seats that name their caller or their kind (novox/hq ADR 0259 §3): a module submits an accept or says an
// event under its own name or kind, takes a held seat's work, asks a proof and answers one, and reads a
// record a seat keeps for it. The runtime does each as the module, only as far as its membership lists,
// and refuses the rest with the reason.

// Work is one piece of a seat's work this bundle takes: the subject it was submitted on, its body, and the
// headers the submitter's runtime set (x-source is the submitting module, x-event-id its id).
type Work struct {
	Worker  string            `json:"worker,omitempty"`
	Subject string            `json:"subject"`
	Body    json.RawMessage   `json:"body"`
	Headers map[string]string `json:"headers,omitempty"`
}

const seatTimeout = 30 * time.Second

// SeatPublish submits an accept, or says an event, on a seat, de-duplicated by id: the same id twice is
// one message. An empty id is the runtime's to make.
func SeatPublish(subject string, body any, id string) error {
	c, err := live()
	if err != nil {
		return err
	}
	_, err = c.ask("mesh/seat.publish", map[string]any{"subject": subject, "body": body, "id": id}, seatTimeout)
	return err
}

// SeatProve asks a proof — core request and reply, never kept — and returns its answer.
func SeatProve(subject string, body any) (json.RawMessage, error) {
	c, err := live()
	if err != nil {
		return nil, err
	}
	return c.ask("mesh/seat.prove", map[string]any{"subject": subject, "body": body}, seatTimeout)
}

// SeatAnswer answers every proof on a subject (a pattern, `*` one token) with what handle returns.
func SeatAnswer(subject string, handle func(subject string, body json.RawMessage) (any, error)) error {
	c, err := live()
	if err != nil {
		return err
	}
	c.mu.Lock()
	c.provers[subject] = handle
	c.mu.Unlock()
	_, err = c.ask("mesh/seat.answer", map[string]any{"subject": subject}, seatTimeout)
	return err
}

// SeatTake takes a held seat's work from a worker, handing each piece to handle: taken when it returns
// nil, offered again after an error. Make the work safe to do twice, or skip an x-event-id already done.
func SeatTake(worker string, handle func(Work) error) error {
	c, err := live()
	if err != nil {
		return err
	}
	c.mu.Lock()
	c.takers[worker] = handle
	c.mu.Unlock()
	_, err = c.ask("mesh/seat.take", map[string]any{"worker": worker}, seatTimeout)
	return err
}

// SeatRecord reads one key of a record a seat keeps for this module, under its own name, or nil.
func SeatRecord(bucket, key string) (*StateEntry, error) {
	c, err := live()
	if err != nil {
		return nil, err
	}
	raw, err := c.ask("mesh/seat.record", map[string]any{"bucket": bucket, "key": key}, seatTimeout)
	if err != nil || len(raw) == 0 || string(raw) == "null" {
		return nil, err
	}
	var e StateEntry
	return &e, json.Unmarshal(raw, &e)
}

// KindHeld is one kind of a kinded bench and who holds it, with what it promises (novox/hq ADR 0259 §5).
type KindHeld struct {
	Seat         string   `json:"seat"`
	Kind         string   `json:"kind"`
	Module       string   `json:"module"`
	Node         string   `json:"node"`
	Capabilities []string `json:"capabilities,omitempty"`
}

// Promises says whether this kind's claim declares a capability.
func (k KindHeld) Promises(capability string) bool {
	for _, c := range k.Capabilities {
		if c == capability {
			return true
		}
	}
	return false
}

// SeatKinds is every kind of the kinded benches this module uses or watches, as the controller issued
// them in its membership: the one account of which channel is which, never a channel's own word.
func SeatKinds() ([]KindHeld, error) {
	c, err := live()
	if err != nil {
		return nil, err
	}
	raw, err := c.ask("mesh/seat.kinds", map[string]any{}, seatTimeout)
	if err != nil {
		return nil, err
	}
	var out []KindHeld
	return out, json.Unmarshal(raw, &out)
}

// Create writes a key only when it has none, and returns the revision; a key that has one is refused.
func (s KeptState) Create(key string, value any) (uint64, error) {
	return s.revisionOf(s.ask("create", map[string]any{"key": key, "value": value}))
}

// Update writes a key only when it is still at revision, and returns the new one; a key changed
// meanwhile is refused, so of two writers that read one value only the first writes.
func (s KeptState) Update(key string, value any, revision uint64) (uint64, error) {
	return s.revisionOf(s.ask("update", map[string]any{"key": key, "value": value, "revision": revision}))
}

func (s KeptState) revisionOf(raw json.RawMessage, err error) (uint64, error) {
	if err != nil {
		return 0, err
	}
	var put struct {
		Revision uint64 `json:"revision"`
	}
	return put.Revision, json.Unmarshal(raw, &put)
}

// subjectMatches is a subject against a pattern in the bus's wildcards: `*` one token, `>` the rest.
func subjectMatches(pattern, subject string) bool {
	p, s := strings.Split(pattern, "."), strings.Split(subject, ".")
	for i, part := range p {
		if part == ">" {
			return len(s) > i
		}
		if i >= len(s) || (part != "*" && part != s[i]) {
			return false
		}
	}
	return len(p) == len(s)
}