mesh-host/internal/link/bus.go

package link

import (
	"context"
	"errors"
	"fmt"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go"
)

// Bus is the link open now as the session holds it. What it can do is asked of it by what it implements (HaveBus,
// HealthBus, ToolBus, FilesBus): the report it once published into the controller's CONTROL stream went with the
// controller (issue 1126), and what the node applied is said in its haves.
type Bus interface{}

// --- The bus the mesh runs on today -----------------------------------------------------------

// OverNATS is the bus as a connection.
type OverNATS struct {
	Conn *nats.Conn
	JS   nats.JetStreamContext
	// Node is this node's name, which names its own object bucket (FileObject). Empty where nothing reads a file.
	Node string
}

// HealthSubject is where this node says its long-running resources' health between reports (novox/hq
// ADR 0240): its own, inside the grant every host already has, and on core NATS like the heartbeat — a
// statement lost is said again within a minute while anything is not healthy.
func HealthSubject(node string) string { return "mesh.control." + node + ".health" }

// HealthBus is a link that can say a health statement. Separate from Bus so a link that cannot is still
// a link: the statement then waits for the next report, which carries it too.
type HealthBus interface {
	Health(ctx context.Context, node string, body []byte) error
}

// Health says a statement, core and flushed, as a heartbeat is.
func (b OverNATS) Health(ctx context.Context, node string, body []byte) error {
	if err := b.Conn.Publish(HealthSubject(node), body); err != nil {
		return err
	}
	flush, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	return b.Conn.FlushWithContext(flush)
}

// CLISubject is where this node's engine asks the controller a line `mesh-cli` was given on this machine
// (novox/hq ADR 0272 §3): its own node's, inside the grant every engine already has (`mesh.control.<node>.>`), so the
// bus server is the authority on which node asked. Core NATS, never the CONTROL stream, which names its own subjects.
func CLISubject(node string) string { return "mesh.control." + node + ".cli" }

// ToolSubject is a module's tool on one machine's node tools — the subject a node-engine asks a declared
// tool check on (novox/hq ADR 0240, to-be 48 §3), granted it for its own machine's instance and no other.
func ToolSubject(module, tool, node string) string {
	return "mesh.mod." + module + ".tool." + tool + "." + node
}

// ToolBus is a link that can ask a module's tool and wait for its answer.
type ToolBus interface {
	Ask(ctx context.Context, subject string, body []byte) ([]byte, error)
}

// Ask asks on core NATS and waits for the answer, within ctx.
func (b OverNATS) Ask(ctx context.Context, subject string, body []byte) ([]byte, error) {
	msg, err := b.Conn.RequestWithContext(ctx, subject, body)
	if err != nil {
		return nil, err
	}
	return msg.Data, nil
}

// HaveBind binds the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132), once for a pass of writes.
// **Only binds**: a node-engine's account reaches no part of the JetStream API beyond what it is granted, and the bucket
// is the controller's to create and to configure, so a bucket that is not there is an error said as what it is. Every
// call on the bucket waits no longer than ctx has left: a refused publish is answered by nothing, and the client's own
// wait is five seconds.
func (b OverNATS) HaveBind(ctx context.Context) (HaveBucket, error) {
	js := b.JS
	wait := 5 * time.Second
	if at, ok := ctx.Deadline(); ok {
		wait = time.Until(at)
		if wait <= 0 {
			return nil, context.DeadlineExceeded
		}
	}
	if b.Conn != nil {
		bounded, err := b.Conn.JetStream(nats.MaxWait(wait))
		if err != nil {
			return nil, err
		}
		js = bounded
	}
	kv, err := js.KeyValue(delivery.HaveBucket)
	if err != nil {
		return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err)
	}
	return natsHaveBucket{kv}, nil
}

type natsHaveBucket struct{ kv nats.KeyValue }

func (n natsHaveBucket) Get(_ context.Context, key string) ([]byte, uint64, error) {
	entry, err := n.kv.Get(key)
	if errors.Is(err, nats.ErrKeyNotFound) {
		return nil, 0, ErrHaveAbsent
	}
	if err != nil {
		return nil, 0, err
	}
	return entry.Value(), entry.Revision(), nil
}

func (n natsHaveBucket) Set(_ context.Context, key string, value []byte, revision uint64) error {
	var err error
	if revision == 0 {
		_, err = n.kv.Create(key, value)
		return err
	}
	_, err = n.kv.Update(key, value, revision)
	return err
}

var _ HaveBus = OverNATS{}