mesh-tools/node-tools/internal/runtime/busstate.go

package runtime

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"sort"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/buscalls"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
)

// The capabilities the tool runner states of the calls it answers (novox/hq ADR 0595 §9, ADR 0597; design 25 § 12.4): each is
// a family of the methods a bundle asks it, named so a manifest's `capabilities` can ask for it, and stated in its announcement
// (announce.Capabilities) for its machine's node-engine to take into the node's record. A module whose manifest asks for one is
// held off a machine whose runner does not state it.
const (
	CapBusStateCalls = "bus-state-calls"
	CapLedgerCalls   = "ledger-calls"
	// CapPutSessions is the put and get sessions' word (design 47 G §518, §622), mesh-sdk's.
	CapPutSessions = stdio.CapPutSessions
	// CapApprovalMark says the runner honours the approval mark: its gate sits on every handler it serves (design 46
	// 7-approval-mark §6, §20), so the grants part composes no deny of its modules' marked subjects.
	CapApprovalMark = "approval-mark/1"
)

// callFamilies are the methods each capability stands for, the ones BusState and BusStateWatch answer.
var callFamilies = map[string][]string{
	CapBusStateCalls: {buscalls.ObjectsPut, buscalls.ObjectsGet, buscalls.ObjectsList, buscalls.ObjectsDelete, buscalls.ObjectsInfo,
		buscalls.ObjectsWatch, buscalls.ObjectsSentGet, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead},
	CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery,
		buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag},
	CapPutSessions: {buscalls.ObjectsPutChunk, buscalls.ObjectsGetChunk},
}

// CallCapabilities is every capability of the calls this runner answers, sorted. The delayed settle of a pulled seat
// message names a fate, not a method, so it stands beside the method families (novox/hq ADR 0664 §5).
func CallCapabilities() []string {
	out := []string{stdio.CapSeatSettleDelay, CapApprovalMark}
	for c := range callFamilies {
		out = append(out, c)
	}
	sort.Strings(out)
	return out
}

// BusState answers a bundle's calls on its module's bus state in the new form (mesh-sdk buscalls; novox/hq ADR 0595
// §9, design 25 § 12.5; issue 1966): each as the module whose bundle asks, never a module the request names, so a
// bundle reaches its own module's bindings alone.
func (b *moduleBus) BusState(method string, params json.RawMessage) (json.RawMessage, error) {
	conn := b.all.conn
	var answer any
	var err error
	switch method {
	case buscalls.ObjectsPut:
		var r buscalls.ObjectsPutRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.ObjectsPut(b.module, r)
		}
	case buscalls.ObjectsGet:
		var r buscalls.ObjectsNameRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = b.all.objectsGet(b.module, r)
			if err == nil && answer == nil {
				return json.RawMessage("null"), nil
			}
		}
	case buscalls.ObjectsDelete:
		var r buscalls.ObjectsNameRequest
		if err = decode(method, params, &r); err == nil {
			err = conn.ObjectsDelete(b.module, r)
			answer = buscalls.Done{}
		}
	case buscalls.ObjectsSentGet:
		var r buscalls.ObjectsSentGetRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = b.all.sentGet(b.module, r)
		}
	case buscalls.ObjectsPutChunk:
		answer, err = b.all.putChunk(b.module, params)
	case buscalls.ObjectsGetChunk:
		answer, err = b.all.getChunk(b.module, params)
	case buscalls.ObjectsList:
		var r buscalls.ObjectsRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.ObjectsList(b.module, r)
		}
	case buscalls.ObjectsInfo:
		var r buscalls.ObjectsRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.ObjectsInfo(b.module, r)
		}
	case buscalls.StreamRead:
		var r buscalls.StreamReadRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.StreamRead(b.module, r)
		}
	case buscalls.StreamFill:
		var r buscalls.StreamRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.StreamFill(b.module, r)
		}
	case buscalls.WorkQueueSize:
		var r buscalls.WorkQueueRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.WorkQueueSize(b.module, r)
		}
	case buscalls.WorkQueueDead:
		var r buscalls.WorkQueueDeadRequest
		if err = decode(method, params, &r); err == nil {
			answer, err = conn.WorkQueueDead(b.module, r)
		}
	case buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerQuery, buscalls.LedgerRebuild,
		buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag:
		answer, err = b.ledgerCall(method, params)
	default:
		return nil, fmt.Errorf("the runtime answers no %s", method)
	}
	if err != nil {
		return nil, err
	}
	return json.Marshal(answer)
}

// BusStateWatch hands a bundle each change of its objects or its ledger, the current ones first.
func (b *moduleBus) BusStateWatch(method string, params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
	switch method {
	case buscalls.ObjectsWatch:
		var r buscalls.ObjectsWatchRequest
		if err := decode(method, params, &r); err != nil {
			return nil, err
		}
		return b.all.conn.ObjectsWatch(b.module, r, func(change buscalls.ObjectsChange) error {
			raw, err := json.Marshal(change)
			if err != nil {
				return err
			}
			return deliver(raw)
		})
	case buscalls.LedgerWatch:
		return b.ledgerWatch(params, deliver)
	}
	return nil, fmt.Errorf("the runtime answers no %s", method)
}

// decode reads a bus-state call's request.
func decode(method string, params json.RawMessage, into any) error {
	if err := json.Unmarshal(params, into); err != nil {
		return fmt.Errorf("%s's request is not of its shape (mesh-sdk buscalls): %w", method, err)
	}
	return nil
}

// objectsGet answers a get: the object whole when it is at most one chunk, else the session that reads it chunk by chunk
// (design 47 G §490, §518); nil when there is none.
func (c *consumers) objectsGet(module string, r buscalls.ObjectsNameRequest) (any, error) {
	st := c.objects()
	if st == nil {
		return nil, fmt.Errorf("the runner has no bus to read %s of %s from", r.Name, module)
	}
	ctx, cancel := context.WithTimeout(context.Background(), bus.StateTimeout)
	defer cancel()
	info, rd, found, err := st.ObjectsOpen(ctx, module, r)
	if err != nil || !found {
		return nil, err
	}
	if info.Size > buscalls.ChunkBytes {
		return c.openGet(module, info, rd, sessionDigest(info.Digest), false)
	}
	defer rd.Close()
	data, err := io.ReadAll(rd)
	if err != nil {
		return nil, err
	}
	return &buscalls.ObjectData{Info: info, Data: data}, nil
}

// sentGet answers a get of a sent object by its reference: whole when the reference's size is at most one chunk, else a
// session (design 47 G §620).
func (c *consumers) sentGet(module string, r buscalls.ObjectsSentGetRequest) (any, error) {
	if r.Size <= buscalls.ChunkBytes {
		got, err := c.conn.SentGet(r)
		if err != nil {
			return nil, err
		}
		return got, nil
	}
	st := c.objects()
	if st == nil {
		return nil, fmt.Errorf("the runner has no bus to read %s from", r.Object)
	}
	ctx, cancel := context.WithTimeout(context.Background(), bus.StateTimeout)
	defer cancel()
	info, rd, err := st.SentOpen(ctx, r)
	if err != nil {
		return nil, err
	}
	return c.openGet(module, info, rd, sessionDigest(r.Digest), true)
}