mesh-host/internal/link/recovery.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// The node agent's wait for the recovery module (design 47 G §554, §560, A-77, A-92): the ending events of one module's kept data
// are heard through an ordered consumer that begins at the last message of the subject and follows the rest, and a wait's bound
// is a key of the state bounds with its own time to live, ended by the bus's marker.

// BoundsBucket is the node-engine's state `bounds`, named by its holder as every state is.
const BoundsBucket = "mesh-host_bounds"

// RecoveryBus is a link that hears the recovery module's events and keeps a bound.
type RecoveryBus interface {
	// HearEnding calls heard with the last message of the subject on the stream, then each after it, until stop is called.
	HearEnding(stream, subject string, heard func([]byte)) (stop func(), err error)
	// BoundCreate writes the bound key with its time to live; one that stands is left as it is. BoundWatch calls ended once the
	// bus's marker says the key's time ended; BoundDelete ends it.
	BoundCreate(ctx context.Context, key string, value []byte, ttl time.Duration) error
	BoundWatch(ctx context.Context, key string, ended func()) (stop func(), err error)
	BoundDelete(ctx context.Context, key string) error
}

func hearEnding(js nats.JetStreamContext, stream, subject string, heard func([]byte)) (func(), error) {
	sub, err := js.Subscribe(subject, func(m *nats.Msg) { heard(m.Data) }, nats.OrderedConsumer(), nats.DeliverLast(), nats.BindStream(stream))
	if err != nil {
		return nil, fmt.Errorf("the stream %s cannot be followed: %w", stream, err)
	}
	return func() { _ = sub.Unsubscribe() }, nil
}

func boundKV(ctx context.Context, conn *nats.Conn) (jetstream.KeyValue, error) {
	js, err := jetstream.New(conn)
	if err != nil {
		return nil, err
	}
	kv, err := js.KeyValue(ctx, BoundsBucket)
	if err != nil {
		return nil, fmt.Errorf("the state %s cannot be bound: %w", BoundsBucket, err)
	}
	return kv, nil
}

func boundCreate(ctx context.Context, conn *nats.Conn, key string, value []byte, ttl time.Duration) error {
	kv, err := boundKV(ctx, conn)
	if err != nil {
		return err
	}
	if _, err := kv.Create(ctx, key, value, jetstream.KeyTTL(ttl)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
		return err
	}
	return nil
}

func boundWatch(ctx context.Context, conn *nats.Conn, key string, ended func()) (func(), error) {
	kv, err := boundKV(ctx, conn)
	if err != nil {
		return nil, err
	}
	w, err := kv.Watch(ctx, key, jetstream.UpdatesOnly())
	if err != nil {
		return nil, err
	}
	go func() {
		for e := range w.Updates() {
			if e != nil && e.Operation() == jetstream.KeyValuePurge {
				ended()
			}
		}
	}()
	return func() { _ = w.Stop() }, nil
}

func boundDelete(ctx context.Context, conn *nats.Conn, key string) error {
	kv, err := boundKV(ctx, conn)
	if err != nil {
		return err
	}
	if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
		return err
	}
	return nil
}

func (l *natsLink) HearEnding(stream, subject string, heard func([]byte)) (func(), error) {
	return hearEnding(l.js, stream, subject, heard)
}
func (l *natsLink) BoundCreate(ctx context.Context, key string, value []byte, ttl time.Duration) error {
	return boundCreate(ctx, l.conn, key, value, ttl)
}
func (l *natsLink) BoundWatch(ctx context.Context, key string, ended func()) (func(), error) {
	return boundWatch(ctx, l.conn, key, ended)
}
func (l *natsLink) BoundDelete(ctx context.Context, key string) error {
	return boundDelete(ctx, l.conn, key)
}

func (b OverNATS) HearEnding(stream, subject string, heard func([]byte)) (func(), error) {
	return hearEnding(b.JS, stream, subject, heard)
}
func (b OverNATS) BoundCreate(ctx context.Context, key string, value []byte, ttl time.Duration) error {
	return boundCreate(ctx, b.Conn, key, value, ttl)
}
func (b OverNATS) BoundWatch(ctx context.Context, key string, ended func()) (func(), error) {
	return boundWatch(ctx, b.Conn, key, ended)
}
func (b OverNATS) BoundDelete(ctx context.Context, key string) error {
	return boundDelete(ctx, b.Conn, key)
}

var (
	_ RecoveryBus = (*natsLink)(nil)
	_ RecoveryBus = OverNATS{}
)

func (q *Session) recoveryBus() (RecoveryBus, error) {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	h, ok := bus.(RecoveryBus)
	if bus == nil || !ok {
		return nil, ErrNoLink
	}
	return h, nil
}

// RecoveryEnding is the ending event of an act as the node agent reads it, the fields every ending carries.
type RecoveryEnding struct {
	Act      string `json:"act"`
	Outcome  string `json:"outcome"`
	Error    string `json:"error,omitempty"`
	Snapshot string `json:"snapshot,omitempty"`
	// Why is the hand act's words for a cancel (design 47 G A-117, §560).
	Why string `json:"why,omitempty"`
}

// HearRecoveryEnding follows the events `ended` (kind "ended") or `restored` (kind "restored") of one module's kept data: the last
// said first, then each after it, until stop is called.
func (q *Session) HearRecoveryEnding(kind, module, kept string) (<-chan RecoveryEnding, func(), error) {
	bus, err := q.recoveryBus()
	if err != nil {
		return nil, nil, err
	}
	stream, subject := meshrecovery.StreamEnded, meshrecovery.SubjectEnded(module, kept)
	if kind == "restored" {
		stream, subject = meshrecovery.StreamRestored, meshrecovery.SubjectRestored(module, kept)
	}
	out := make(chan RecoveryEnding, 64)
	stop, err := bus.HearEnding(stream, subject, func(raw []byte) {
		var ev RecoveryEnding
		if json.Unmarshal(raw, &ev) == nil && ev.Act != "" {
			select {
			case out <- ev:
			default:
			}
		}
	})
	if err != nil {
		return nil, nil, err
	}
	return out, stop, nil
}

// KeepBound writes the bound key `<what>.<id>` of a wait with its time to live and watches it: expired is closed when the bus's
// marker says its time ended, and end deletes it.
func (q *Session) KeepBound(ctx context.Context, what, id, module string, since time.Time, ttl time.Duration) (<-chan struct{}, func(), error) {
	bus, err := q.recoveryBus()
	if err != nil {
		return nil, nil, err
	}
	key := what + "." + id
	value, err := json.Marshal(struct {
		Module string `json:"module,omitempty"`
		Since  string `json:"since"`
		V      int64  `json:"v"`
	}{module, since.UTC().Format(time.RFC3339), 1})
	if err != nil {
		return nil, nil, err
	}
	expired := make(chan struct{})
	stopWatch, err := bus.BoundWatch(ctx, key, func() {
		select {
		case <-expired:
		default:
			close(expired)
		}
	})
	if err != nil {
		return nil, nil, err
	}
	if err := bus.BoundCreate(ctx, key, value, ttl); err != nil {
		stopWatch()
		return nil, nil, err
	}
	return expired, func() {
		stopWatch()
		_ = bus.BoundDelete(context.WithoutCancel(ctx), key)
	}, nil
}

// CallTool calls a module's tool on this machine's own node tools with its arguments and answers its refusal as an error (design 47 G
// A-108: the catch-up of a restore point is called so, as the node-engine is granted to).
func (q *Session) CallTool(ctx context.Context, module, tool string, args map[string]any) error {
	q.init()
	q.mu.Lock()
	bus := q.bus
	q.mu.Unlock()
	asker, ok := bus.(ToolBus)
	if bus == nil || !ok {
		return ErrNoLink
	}
	if args == nil {
		args = map[string]any{}
	}
	body, err := json.Marshal(args)
	if err != nil {
		return err
	}
	reply, err := asker.Ask(ctx, ToolSubject(module, tool, q.Membership.Node), body)
	if err != nil {
		return err
	}
	var env struct {
		Error string `json:"error"`
	}
	if err := json.Unmarshal(reply, &env); err != nil {
		return fmt.Errorf("its answer is not a tool's answer: %w", err)
	}
	if env.Error != "" {
		return errors.New(env.Error)
	}
	return nil
}