mesh-catalog/modules/zfs/cmd/zfs-tools/bus.go

package main

import (
	"encoding/json"
	"os"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/buscalls"
	"git.novox.be/novox/mesh-sdk/go/process"
)

// busStates is the module's states as this process reaches them (its bus account's `state`).
type busStates struct{ c *process.Conn }

func (b busStates) Put(state, key string, v any) error {
	_, err := b.c.State(state).Put(key, v)
	return err
}
func (b busStates) Delete(state, key string) error      { return b.c.State(state).Delete(key) }
func (b busStates) Keys(state string) ([]string, error) { return b.c.State(state).Keys() }
func (b busStates) Get(state, key string) (json.RawMessage, bool, error) {
	e, err := b.c.State(state).Get(key)
	if err != nil {
		return nil, false, err
	}
	if e == nil {
		return nil, false, nil
	}
	return e.Value, true, nil
}

// JoinBus gives the machine this process's events and states, as its own bus account says. Without the
// account or the bus the machine keeps none, and what it changes is said as a warning (§18: with the bus down
// the apply converges from the config file it last got). It answers how to leave the bus.
func (m *Machine) JoinBus() func() {
	acct, err := process.ReadBrokerAccount(os.Getenv)
	if err != nil {
		m.Warnings = append(m.Warnings, err.Error())
		return func() {}
	}
	c, err := process.Connect(acct)
	if err != nil {
		m.Warnings = append(m.Warnings, err.Error())
		return func() {}
	}
	m.Emit = func(kind string, body any) error { return c.Emit(kind, body, stdio.EmitOptions{}) }
	m.States = busStates{c}
	m.Findings = meshconditionsclient.FindingWriter{State: c.State(StateFindings)}
	m.Call = c.Call
	return c.Close
}

// toolStates is the module's states as the tools bundle reaches them, through the tool runner.
type toolStates struct{}

func (toolStates) Put(state, key string, v any) error {
	_, err := stdio.State(state).Put(key, v)
	return err
}
func (toolStates) Delete(state, key string) error      { return stdio.State(state).Delete(key) }
func (toolStates) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
func (toolStates) Get(state, key string) (json.RawMessage, bool, error) {
	e, err := stdio.State(state).Get(key)
	if err != nil || e == nil {
		return nil, false, err
	}
	return e.Value, true, nil
}

// eventHub hands the module's events, heard once through the tool runner, to the hand acts waiting for one.
type eventHub struct {
	mu      sync.Mutex
	started bool
	waiters map[int]hubWaiter
	next    int
}

type hubWaiter struct {
	kinds []string
	ch    chan map[string]any
}

func (h *eventHub) await(kinds []string, bound time.Duration) (map[string]any, bool) {
	h.mu.Lock()
	if h.waiters == nil {
		h.waiters = map[int]hubWaiter{}
	}
	h.next++
	id, w := h.next, hubWaiter{kinds, make(chan map[string]any, 1)}
	h.waiters[id] = w
	first := !h.started
	h.started = true
	h.mu.Unlock()
	defer func() { h.mu.Lock(); delete(h.waiters, id); h.mu.Unlock() }()
	if first {
		_ = stdio.Subscribe("#", func(e stdio.Envelope) error {
			var body map[string]any
			_ = json.Unmarshal(e.Body, &body)
			h.mu.Lock()
			defer h.mu.Unlock()
			for _, w := range h.waiters {
				for _, k := range w.kinds {
					if e.Key == k || strings.HasSuffix(e.Key, "."+k) {
						select {
						case w.ch <- body:
						default:
						}
					}
				}
			}
			return nil
		})
	}
	select {
	case ev := <-w.ch:
		return ev, true
	case <-time.After(bound):
		return nil, false
	}
}

// busLedger is the ledger `snapshots` through the tool runner.
type busLedger struct{}

func (busLedger) Apply(instance, operation string, body any, call string) error {
	raw, err := json.Marshal(body)
	if err != nil {
		return err
	}
	_, err = stdio.BusState(buscalls.LedgerApply, buscalls.ApplyRequest{Ledger: LedgerName, Instance: instance,
		Operation: operation, Body: raw, Call: call})
	return err
}

func (busLedger) Entries() ([]LedgerEntry, error) {
	raw, err := stdio.BusState(buscalls.LedgerQuery, buscalls.QueryRequest{Ledger: LedgerName, Index: "by-dataset-kind"})
	if err != nil {
		return nil, err
	}
	var q buscalls.QueryAnswer
	if err := json.Unmarshal(raw, &q); err != nil {
		return nil, err
	}
	var out []LedgerEntry
	for _, ie := range q.Entries {
		raw, err := stdio.BusState(buscalls.LedgerGet, buscalls.InstanceRequest{Ledger: LedgerName, Instance: ie.Instance})
		if err != nil {
			return nil, err
		}
		var st buscalls.InstanceState
		if err := json.Unmarshal(raw, &st); err != nil {
			return nil, err
		}
		var e LedgerEntry
		if err := json.Unmarshal(st.State, &e); err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, nil
}

// EventsStream is the module's events stream's local name in its `requires` (§25).
const EventsStream = "events"

// busStream is the module's events stream through the tool runner.
type busStream struct{}

// Last reads the stream from its first sequence on, page by page, and keeps the last n events of the kind.
func (busStream) Last(n int, kind string) ([]LoggedEvent, error) {
	raw, err := stdio.BusState(buscalls.StreamFill, buscalls.StreamRequest{Stream: EventsStream})
	if err != nil {
		return nil, err
	}
	var fill buscalls.StreamFilled
	if err := json.Unmarshal(raw, &fill); err != nil {
		return nil, err
	}
	var out []LoggedEvent
	from := fill.First
	for from != 0 && from <= fill.Last {
		raw, err := stdio.BusState(buscalls.StreamRead, buscalls.StreamReadRequest{Stream: EventsStream, From: from, Limit: 500})
		if err != nil {
			return nil, err
		}
		var page buscalls.StreamPage
		if err := json.Unmarshal(raw, &page); err != nil {
			return nil, err
		}
		for _, msg := range page.Messages {
			k := KindOfSubject(msg.Subject)
			if kind != "" && k != kind {
				continue
			}
			out = append(out, LoggedEvent{Sequence: msg.Sequence, Kind: k, At: msg.At, Body: msg.Body})
			if len(out) > n {
				out = out[1:]
			}
		}
		from = page.Next
	}
	return out, nil
}