mesh-catalog/modules/restic/cmd/restic-backups/itembus.go

package main

// The live bus of the item events (items.go): the module's own broker account, as the recovery module's process holds its own
// (design 47 G A-150: the process `night` has a bus account of its own, says its items and reads the recovery module's restores).

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/keys"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/seatlist"
	"git.novox.be/novox/mesh-sdk/go/views"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// The streams and subjects of the seats' events the holder reads and says.
const (
	backupSeat           = "node-backup"
	recoverySeat         = "mesh-recovery"
	streamItemStarted    = "SEAT_node-backup_item-started"
	streamItemEnded      = "SEAT_node-backup_item-ended"
	streamRestoreStarted = "SEAT_mesh-recovery_restore-started"
	streamRestored       = "SEAT_mesh-recovery_restored"
	readWait             = 5 * time.Second
)

func recoveryEventSubject(event, module, kept string) string {
	return "mesh.seat." + recoverySeat + ".event." + event + "." + module + "." + kept
}

func itemEventSubject(event, node, module, item string) string {
	return "mesh.seat." + backupSeat + ".event." + event + "." + node + "." + module + "." + item
}

// liveBus is ItemBus over the module's own account.
type liveBus struct {
	conn    *process.Conn
	nc      *nats.Conn
	js      jetstream.JetStream
	node    string
	account process.Account
}

// connectItems opens the bus as the module's own account when the process was given its credential (MESH_BROKER_FILE); nil, nil
// when it was not.
func connectItems(getenv func(string) string) (*liveBus, error) {
	if strings.TrimSpace(getenv(process.BrokerFileEnv)) == "" {
		return nil, nil
	}
	account, err := process.ReadBrokerAccount(getenv)
	if err != nil {
		return nil, err
	}
	conn, err := process.Connect(account)
	if err != nil {
		return nil, err
	}
	nc, js, err := account.Dial()
	if err != nil {
		conn.Close()
		return nil, err
	}
	return &liveBus{conn: conn, nc: nc, js: js, node: account.Node, account: account}, nil
}

// Broker answers how to find the holder of the seat mesh-broker.
func (l *liveBus) Broker() (string, error) { return seatlist.HolderOf(l.conn.Call, BrokerSeat) }

// Close ends both connections.
func (l *liveBus) Close() { l.nc.Close(); l.conn.Close() }

// PutOffs is the module's own state `put-off`, its keys with their own time to live (A-155).
func (l *liveBus) PutOffs() PutOffState { return stateOfPutOffs{l.conn.State(PutOffStateName)} }

// Findings starts the module's held finding writer over its state `findings` (§561), as the recovery module does (§461): started
// before the night's first backup, so what it writes is held and written. It answers how to close it.
func (l *liveBus) Findings(ctx context.Context, logf func(string, ...any)) (Findings, func(), error) {
	// The module's finding stream is its `finding-events` entry, asked of the seat mesh-broker's show once (A-274, §529); the
	// night's process lives one night, so it hears no recreated of it. Until it answers, the writer holds its events.
	binding := process.NewBinding(BrokerSeat, l.account.Module, l.conn.CallKept, logf, views.FindingEventsLocal)
	binding.Ask(views.FindingEventsLocal)
	w, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{
		Asked: meshconditions.SubjectFindingsAsked, Module: l.account.Module, Node: l.account.Node, State: l.conn.State("findings"),
		Kinds: []string{FindingBackupWaits}, JS: l.js, Conn: l.nc, Bindings: binding, Logf: logf})
	if err != nil {
		return nil, nil, err
	}
	if err := w.Start(ctx); err != nil {
		return nil, nil, err
	}
	return w, w.Close, nil
}

func (l *liveBus) Say(event string, ref ItemRef, body map[string]any) error {
	return l.conn.EmitSeat(backupSeat, event+"."+l.node+"."+keys.EscapeName(ref.Module)+"."+keys.EscapeName(ref.Item), body)
}

// last reads every subject's last message of a stream through an ordered consumer made for this one read (§556).
func (l *liveBus) last(ctx context.Context, stream, filter string, take func(subject string, data []byte)) error {
	cons, err := l.js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{
		FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
	if err != nil {
		return fmt.Errorf("the read of %s: %w", stream, err)
	}
	info, err := cons.Info(ctx)
	if err != nil {
		return fmt.Errorf("the read of %s: %w", stream, err)
	}
	for left := info.NumPending; left > 0; {
		m, err := cons.Next(jetstream.FetchMaxWait(readWait))
		if err != nil {
			return fmt.Errorf("the read of %s: %w", stream, err)
		}
		take(m.Subject(), m.Data())
		meta, err := m.Metadata()
		if err != nil {
			return err
		}
		left = meta.NumPending
	}
	return nil
}

func atOf(data []byte) time.Time {
	var b struct {
		At time.Time `json:"at"`
	}
	_ = json.Unmarshal(data, &b)
	return b.At
}

func (l *liveBus) Unended() ([]ItemRef, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 2*readWait)
	defer cancel()
	started, ended := map[ItemRef]time.Time{}, map[ItemRef]time.Time{}
	for _, c := range []struct {
		stream, event string
		into          map[ItemRef]time.Time
	}{{streamItemStarted, EventItemStarted, started}, {streamItemEnded, EventItemEnded, ended}} {
		prefix := itemEventSubject(c.event, l.node, "", "")
		prefix = strings.TrimSuffix(prefix, "..")
		c := c
		err := l.last(ctx, c.stream, itemEventSubject(c.event, l.node, "*", "*"), func(subject string, data []byte) {
			module, item, ok := strings.Cut(strings.TrimPrefix(subject, prefix+"."), ".")
			if !ok {
				return
			}
			m, e1 := keys.UnescapeName(module)
			i, e2 := keys.UnescapeName(item)
			if e1 == nil && e2 == nil {
				c.into[ItemRef{m, i}] = atOf(data)
			}
		})
		if err != nil {
			return nil, err
		}
	}
	var out []ItemRef
	for ref, at := range started {
		if end, ok := ended[ref]; !ok || end.Before(at) {
			out = append(out, ref)
		}
	}
	sortRefs(out)
	return out, nil
}

func (l *liveBus) Restoring(module string) (string, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 2*readWait)
	defer cancel()
	type act struct {
		id string
		at time.Time
	}
	started, restored := map[string]act{}, map[string]act{}
	for _, c := range []struct {
		stream, event string
		into          map[string]act
	}{{streamRestoreStarted, "restore-started", started}, {streamRestored, "restored", restored}} {
		prefix := recoveryEventSubject(c.event, keys.EscapeName(module), "")
		c := c
		err := l.last(ctx, c.stream, recoveryEventSubject(c.event, keys.EscapeName(module), "*"), func(subject string, data []byte) {
			var b struct {
				Act string    `json:"act"`
				At  time.Time `json:"at"`
			}
			if json.Unmarshal(data, &b) == nil {
				c.into[strings.TrimPrefix(subject, prefix)] = act{b.Act, b.At}
			}
		})
		if err != nil {
			return "", err
		}
	}
	var running []string
	for kept, s := range started {
		if r, ok := restored[kept]; !ok || r.id != s.id {
			running = append(running, s.id)
		}
	}
	if len(running) == 0 {
		return "", nil
	}
	if len(running) > 1 {
		sortStrings(running)
	}
	return running[0], nil
}

func (l *liveBus) WaitRestored(ctx context.Context, want string) error {
	cons, err := l.js.OrderedConsumer(ctx, streamRestored, jetstream.OrderedConsumerConfig{
		FilterSubjects: []string{"mesh.seat." + recoverySeat + ".event.restored.*.*"}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
	if err != nil {
		return err
	}
	done := make(chan struct{})
	var once sync.Once
	cc, err := cons.Consume(func(m jetstream.Msg) {
		var b struct {
			Act string `json:"act"`
		}
		if json.Unmarshal(m.Data(), &b) == nil && b.Act == want {
			once.Do(func() { close(done) })
		}
	})
	if err != nil {
		return err
	}
	defer cc.Stop()
	// The wait ends on the restored or when ctx ends (the bound's expiry cancels it), never on a fetch's own time.
	select {
	case <-done:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func sortStrings(s []string) {
	for i := 1; i < len(s); i++ {
		for j := i; j > 0 && s[j] < s[j-1]; j-- {
			s[j], s[j-1] = s[j-1], s[j]
		}
	}
}

// PutOffStateName is the module's own state of the items put off for a restore, whose keys have a time to live (key-ttl).
const PutOffStateName = "put-off"

// putOffBucket is the state's calls this module makes on its bucket.
type putOffBucket interface {
	CreateTTL(key string, value any, ttl time.Duration) (uint64, error)
	Watch(key string, handle func(stdio.StateChange) error) error
	Delete(key string) error
	Keys() ([]string, error)
	Get(key string) (*stdio.StateEntry, error)
}

// errBoundEnded ends a bound key's watch once the key is gone.
var errBoundEnded = errors.New("the bound key is removed")

// stateOfPutOffs times the wait for a restore by a bound key of the state put-off (A-77, A-155): the key is watched before it is
// written, so its expiry is never missed, and the watch ends with the key's removal.
type stateOfPutOffs struct{ s putOffBucket }

func (p stateOfPutOffs) Hold(key string, rec PutOff, ttl time.Duration, expired func()) (func(), error) {
	var ended atomic.Bool
	if err := p.s.Watch(key, func(c stdio.StateChange) error {
		if c.Op == "put" || c.Current {
			return nil
		}
		if !ended.Swap(true) {
			expired()
		}
		return errBoundEnded
	}); err != nil {
		return nil, err
	}
	if _, err := p.s.CreateTTL(key, rec, ttl); err != nil {
		ended.Store(true)
		_ = p.s.Delete(key)
		return nil, err
	}
	return func() {
		if !ended.Swap(true) {
			_ = p.s.Delete(key)
		}
	}, nil
}

func (p stateOfPutOffs) All() (map[string]PutOff, error) {
	names, err := p.s.Keys()
	if err != nil {
		return nil, err
	}
	out := map[string]PutOff{}
	for _, k := range names {
		e, err := p.s.Get(k)
		if err != nil || e == nil {
			continue
		}
		var rec PutOff
		if json.Unmarshal(e.Value, &rec) == nil {
			out[k] = rec
		}
	}
	return out, nil
}