mesh-sdk/go/buscalls/buscalls.go

// Package buscalls is the contract of the calls a bundle makes on its own bus state through the tool runner, over the
// bundle's pipe (novox/hq ADR 0595 §9, ADR 0597 §10, design 25 § 12.5): the object store's, a stream's read-back and
// fill, a work queue's size and dead letters, and the ledger's. Each call names an entry by the local name its module
// asks for it by in `requires`; the tool runner serves it from the module's membership's bindings, its own entries
// only. Only the shapes and the method names are here: what the tool runner refuses, and why, is its own.
//
// It is a package of its own, imported only by the modules that use these calls and by the tool runner, so a change
// here rebuilds those and not the module library's importers (ADR 0595 §9). It imports nothing of mesh-sdk.
package buscalls

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

// MetaCall is where in a `tools/call`'s `_meta` the tool runner hands a bundle the id of the tool call in flight: the id
// ApplyRequest.Call names, so the runner stamps the caller it proved for that call as the operation's author (novox/hq
// ADR 0597 §7). The runner writes `_meta`; a caller writes only `arguments`, so no caller can name another call.
const MetaCall = "mesh/call"

// The methods a bundle asks the tool runner, one per call.
const (
	ObjectsPut    = "mesh/objects.put"
	ObjectsGet    = "mesh/objects.get"
	ObjectsList   = "mesh/objects.list"
	ObjectsDelete = "mesh/objects.delete"
	ObjectsInfo   = "mesh/objects.info"
	ObjectsWatch  = "mesh/objects.watch"

	StreamRead = "mesh/stream.read"
	StreamFill = "mesh/stream.fill"

	WorkQueueSize = "mesh/workqueue.size"
	WorkQueueDead = "mesh/workqueue.dead"

	LedgerApply   = "mesh/ledger.apply"
	LedgerGet     = "mesh/ledger.get"
	LedgerHistory = "mesh/ledger.history"
	LedgerWatch   = "mesh/ledger.watch"
	LedgerQuery   = "mesh/ledger.query"
	// The ledger's repairs, which the tool runner also serves as tools of every module owning a ledger, its own
	// ledgers only (design 25 § 12.5).
	LedgerRebuild  = "mesh/ledger.rebuild"
	LedgerReindex  = "mesh/ledger.reindex"
	LedgerSnapshot = "mesh/ledger.snapshot"
	LedgerLag      = "mesh/ledger.lag"
)

// The methods the tool runner sends the bundle for a watch it asked, each naming the watch by its id, as `mesh/log`
// does for a log.
const (
	ObjectsWatched = "mesh/objects"
	LedgerWatched  = "mesh/ledger"
)

// Fill is how much an entry holds against its cap: its messages (or objects), its bytes and its cap in bytes.
type Fill struct {
	Count    uint64 `json:"count"`
	Bytes    uint64 `json:"bytes"`
	MaxBytes uint64 `json:"max_bytes"`
}

// Done is the answer of a call that answers nothing but that it was done.
type Done struct{}

// --- the object store

// ObjectsPutRequest stores one object whole under its name, replacing one of that name.
type ObjectsPutRequest struct {
	Objects     string `json:"objects"`
	Name        string `json:"name"`
	Data        []byte `json:"data"`
	Description string `json:"description,omitempty"`
}

// ObjectsNameRequest names one object of an object store: get, delete.
type ObjectsNameRequest struct {
	Objects string `json:"objects"`
	Name    string `json:"name"`
}

// ObjectsRequest names an object store: list, info.
type ObjectsRequest struct {
	Objects string `json:"objects"`
}

// ObjectsWatchRequest asks the tool runner to send every object stored or deleted, as ObjectsChange under the watch's id.
type ObjectsWatchRequest struct {
	Objects string `json:"objects"`
	Watch   string `json:"watch"`
}

// ObjectInfo is one object as its store holds it: put's answer, and each of list's.
type ObjectInfo struct {
	Name        string    `json:"name"`
	Size        uint64    `json:"size"`
	Chunks      uint32    `json:"chunks"`
	Digest      string    `json:"digest"`
	Description string    `json:"description,omitempty"`
	Modified    time.Time `json:"modified"`
	Deleted     bool      `json:"deleted,omitempty"`
}

// ObjectData is get's answer: the object's info and its bytes.
type ObjectData struct {
	Info ObjectInfo `json:"info"`
	Data []byte     `json:"data"`
}

// ObjectsListed is list's answer: every object the store holds.
type ObjectsListed struct {
	Objects []ObjectInfo `json:"objects"`
}

// ObjectsChange is one object stored or deleted, sent under ObjectsWatched; Current says it was there when the watch
// began.
type ObjectsChange struct {
	Watch   string     `json:"watch"`
	Object  ObjectInfo `json:"object"`
	Current bool       `json:"current"`
}

// --- a stream

// StreamReadRequest reads a module's own stream from a sequence on, at most Limit messages, Subjects narrowing it.
type StreamReadRequest struct {
	Stream   string `json:"stream"`
	From     uint64 `json:"from"`
	Subjects string `json:"subjects,omitempty"`
	Limit    int    `json:"limit,omitempty"`
}

// StreamRequest names a stream: fill.
type StreamRequest struct {
	Stream string `json:"stream"`
}

// StreamMessage is one message of a stream.
type StreamMessage struct {
	Sequence uint64          `json:"sequence"`
	Subject  string          `json:"subject"`
	Body     json.RawMessage `json:"body"`
	At       time.Time       `json:"at"`
}

// StreamPage is read's answer: the messages in order, the sequence to read on from (0 when nothing follows) and the
// stream's last sequence when it was read.
type StreamPage struct {
	Messages []StreamMessage `json:"messages"`
	Next     uint64          `json:"next"`
	Last     uint64          `json:"last"`
}

// StreamFilled is fill's answer: the stream's fill and its first and last sequence.
type StreamFilled struct {
	Fill
	First uint64 `json:"first"`
	Last  uint64 `json:"last"`
}

// --- a work queue

// WorkQueueRequest names a work queue: size.
type WorkQueueRequest struct {
	Queue string `json:"queue"`
}

// WorkQueueDeadRequest asks a work queue's dead letters, at most Limit of them.
type WorkQueueDeadRequest struct {
	Queue string `json:"queue"`
	Limit int    `json:"limit,omitempty"`
}

// WorkQueueSized is size's answer: the messages waiting to be taken, against the queue's cap.
type WorkQueueSized struct {
	Fill
}

// DeadLetter is one message delivered as often as its queue allows and never settled.
type DeadLetter struct {
	Sequence   uint64          `json:"sequence"`
	Subject    string          `json:"subject"`
	Deliveries int             `json:"deliveries"`
	Body       json.RawMessage `json:"body,omitempty"`
	At         time.Time       `json:"at"`
}

// WorkQueueDeadLetters is dead's answer.
type WorkQueueDeadLetters struct {
	Dead []DeadLetter `json:"dead"`
}

// --- the ledger (ADR 0597)

// ApplyRequest applies one operation to one instance of a ledger. Call is the id of the tool call the operation
// serves, as the tool runner handed it under MetaCall, whose caller the tool runner proved and stamps as the author
// (ADR 0597 §7); empty, the operation carries the owning module's account alone. An author the bundle writes into Body
// is never taken as the author.
type ApplyRequest struct {
	Ledger    string          `json:"ledger"`
	Instance  string          `json:"instance"`
	Operation string          `json:"operation"`
	Body      json.RawMessage `json:"body,omitempty"`
	Call      string          `json:"call,omitempty"`
}

// Applied is apply's answer: the operation was taken, with its sequence in the instance's history; StateBehind says
// the state could not follow because its bucket is full, and catches up at the next apply after room is made (ADR
// 0597 §6).
type Applied struct {
	Taken       bool   `json:"taken"`
	Sequence    uint64 `json:"sequence"`
	StateBehind bool   `json:"state_behind,omitempty"`
}

// Author is who an operation is by, as `apply` stamped it (ADR 0597 §7): the owning module's account, and the caller
// of the tool call it served, as the tool runner proved it.
type Author struct {
	Account string `json:"account"`
	Caller  string `json:"caller,omitempty"`
}

// InstanceRequest names one instance of a ledger: get, snapshot.
type InstanceRequest struct {
	Ledger   string `json:"ledger"`
	Instance string `json:"instance"`
}

// InstanceState is get's answer: the instance's state and the sequence of the last operation it holds.
type InstanceState struct {
	Instance string          `json:"instance"`
	State    json.RawMessage `json:"state"`
	Sequence uint64          `json:"sequence"`
}

// HistoryRequest reads an instance's operations from a sequence on, at most Limit of them.
type HistoryRequest struct {
	Ledger   string `json:"ledger"`
	Instance string `json:"instance"`
	From     uint64 `json:"from,omitempty"`
	Limit    int    `json:"limit,omitempty"`
}

// Operation is one operation as the history keeps it.
type Operation struct {
	Sequence  uint64          `json:"sequence"`
	Instance  string          `json:"instance"`
	Operation string          `json:"operation"`
	Body      json.RawMessage `json:"body,omitempty"`
	Author    Author          `json:"author"`
	At        time.Time       `json:"at"`
}

// HistoryPage is history's answer: the operations oldest first, the sequence to read on from (0 when nothing
// follows) and the instance's last sequence when it was read.
type HistoryPage struct {
	Operations []Operation `json:"operations"`
	Next       uint64      `json:"next"`
	Last       uint64      `json:"last"`
}

// LedgerWatchRequest asks the tool runner to send every change of the instances Instances names (an instance, with
// `*` for one name and `>` for the rest) as InstanceChange under the watch's id: the current states first.
type LedgerWatchRequest struct {
	Ledger    string `json:"ledger"`
	Instances string `json:"instances"`
	Watch     string `json:"watch"`
}

// InstanceChange is one instance's new state, sent under LedgerWatched; Current says it was the state when the watch
// began.
type InstanceChange struct {
	Watch    string          `json:"watch"`
	Instance string          `json:"instance"`
	State    json.RawMessage `json:"state"`
	Sequence uint64          `json:"sequence"`
	Current  bool            `json:"current"`
}

// QueryRequest reads a declared index of a ledger: the entries whose key Key names (a key, with `*` for one name and
// `>` for the rest; empty is every key), at most Limit of them.
type QueryRequest struct {
	Ledger string `json:"ledger"`
	Index  string `json:"index"`
	Key    string `json:"key,omitempty"`
	Limit  int    `json:"limit,omitempty"`
}

// IndexEntry is the value of one key of a declared index (IndexKey): the instance it points at, and the sequence of
// the operation it was written from. The owner's watcher and the tool runner's reindex both write it so.
type IndexEntry struct {
	Instance string `json:"instance"`
	Sequence uint64 `json:"sequence"`
}

// IndexKey is the key of an instance in a declared index: the values of the index's fields, in its order, then the
// parts of the instance's path, all joined by ".". Each value and each path part is encoded (keyToken): every byte
// outside [A-Za-z0-9_-] is written "=" and two upper-case hex digits ("." is "=2E", " " is "=20", "=" is "=3D"), so two
// different keys never meet and every part is a token the bus takes. `17950.listing.a17` is the key of price 17950
// and instance `listing/a17`. The owner's watcher and the tool runner's reindex both write it, so a reindex leaves the
// index as the watcher keeps it.
func IndexKey(fields []string, instance string) string {
	parts := make([]string, 0, len(fields)+2)
	for _, f := range fields {
		parts = append(parts, keyToken(f))
	}
	for _, p := range strings.Split(instance, "/") {
		parts = append(parts, keyToken(p))
	}
	return strings.Join(parts, ".")
}

// keyToken is v with every byte outside [A-Za-z0-9_-] written "=" and two upper-case hex digits (IndexKey).
func keyToken(v string) string {
	var b strings.Builder
	for i := 0; i < len(v); i++ {
		c := v[i]
		if c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-' {
			b.WriteByte(c)
			continue
		}
		fmt.Fprintf(&b, "=%02X", c)
	}
	return b.String()
}

// QueryAnswer is query's answer.
type QueryAnswer struct {
	Entries []IndexEntry `json:"entries"`
}

// RebuildFrom is what a rebuild derives the state from: the history, or the snapshot and the operations after it.
const (
	RebuildFromHistory  = "history"
	RebuildFromSnapshot = "snapshot"
)

// RebuildRequest derives an instance's state again (every instance's when Instance is empty), From the history or the
// snapshot.
type RebuildRequest struct {
	Ledger   string `json:"ledger"`
	Instance string `json:"instance,omitempty"`
	From     string `json:"from,omitempty"`
}

// ReindexRequest builds a declared index again from the history.
type ReindexRequest struct {
	Ledger string `json:"ledger"`
	Index  string `json:"index"`
}

// LedgerRequest names a ledger: lag.
type LedgerRequest struct {
	Ledger string `json:"ledger"`
}

// Repaired is the answer of rebuild, reindex and snapshot: how many instances (or index entries) it wrote.
type Repaired struct {
	Written int `json:"written"`
}

// InstanceLag is one instance whose state or snapshot is behind its history: the sequences each holds.
type InstanceLag struct {
	Instance string `json:"instance"`
	History  uint64 `json:"history"`
	State    uint64 `json:"state"`
	Snapshot uint64 `json:"snapshot"`
}

// IndexLag is how far one declared index is behind the history.
type IndexLag struct {
	Index  string `json:"index"`
	Behind uint64 `json:"behind"`
}

// Lagged is lag's answer: every instance and index behind its history; none when nothing is behind.
type Lagged struct {
	Instances []InstanceLag `json:"instances,omitempty"`
	Indexes   []IndexLag    `json:"indexes,omitempty"`
}