mesh-delivery/modules/mesh-grants/internal/grants/running.go

package grants

import (
	"bytes"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
)

// **What each node runs, as its node-engine says it** (novox/hq design 47, ADR 0660 §2): the seat node-engine's event
// `running`, kept last per subject on its own stream, is heard on the part's own durable and kept in the state `running`,
// one key `<node>.<module>` holding the event's body. holdersAfterPass reads the manifest of the digest each says, not the
// build the assigner wants.

// StateRunning is the part's state of what each node-engine said it runs.
const StateRunning = "running"

// RunningKey is a key of StateRunning.
func RunningKey(node, module string) string { return node + "." + module }

// HearRunning keeps one running event in the state running when it differs from the key's value (the time it was seen
// aside), and says whether the part must compose again. An event that is not a running body, or names a subject other
// than its own node and module, is refused with the reason; the caller drops it.
func (p *Part) HearRunning(subject string, data []byte) (changed bool, err error) {
	var r nodeengine.RunningEvent
	if err := json.Unmarshal(data, &r); err != nil {
		return false, fmt.Errorf("%s is not a running body: %w", subject, err)
	}
	// A `running` with stopped carries no fingerprint and no manifest (design 47 G §404, §365): the have no longer holds the module.
	if r.Node == "" || r.Module == "" || (r.Manifest == "" && !r.Stopped) {
		return false, fmt.Errorf("%s names no node, module or manifest digest", subject)
	}
	if !strings.HasSuffix(subject, "."+RunningKey(r.Node, r.Module)) {
		return false, fmt.Errorf("%s does not carry the node and module its body names (%s)", subject, RunningKey(r.Node, r.Module))
	}
	key := RunningKey(r.Node, r.Module)
	was, _, found, err := p.Store.Get(StateRunning, key)
	if err != nil {
		return false, fmt.Errorf("its own %s %s cannot be read: %w", StateRunning, key, err)
	}
	if found {
		var w nodeengine.RunningEvent
		// An event older by `at` than what the part holds changes nothing: a node's module's newer one stands (design 47 G §556).
		if json.Unmarshal(was, &w) == nil && r.At.Before(w.At) {
			return false, nil
		}
		if json.Unmarshal(was, &w) == nil && w.Manifest == r.Manifest && w.Fingerprint == r.Fingerprint && w.Stopped == r.Stopped {
			return false, nil
		}
	}
	if _, err := p.Store.Put(StateRunning, key, bytes.Clone(data)); err != nil {
		return false, fmt.Errorf("its own %s %s cannot be written: %w", StateRunning, key, err)
	}
	return true, nil
}

// StreamRead is the part's whole read of a stream kept last per subject at its start (design 47 G §556): through an ordered
// consumer made for that one read, every subject's last message, before its durable consumer is followed. Until it ends the
// part acts on none of the stream's events, and `grants_first_views` names the stream `waiting`.
type StreamRead struct {
	Stream string `json:"stream"`
	// State is `waiting` until the read ends, then `read`.
	State string    `json:"state"`
	Since time.Time `json:"since"`
}

// ReadingRunning begins the whole read of the node-engine's stream running at the part's start; the part acts on none of
// `running` until RunningRead ends it.
func (p *Part) ReadingRunning(stream string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	p.runningRead = &StreamRead{Stream: stream, State: "waiting", Since: p.now()}
}

// RunningRead ends the whole read: the part acts on `running` from the next pass.
func (p *Part) RunningRead() {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.runningRead != nil {
		p.runningRead.State = "read"
	}
}

// StreamReads are the whole reads the part makes at its start, with their state, for `grants_first_views`.
func (p *Part) StreamReads() []StreamRead {
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.runningRead == nil {
		return nil
	}
	r := *p.runningRead
	r.Since = r.Since.Truncate(time.Second)
	return []StreamRead{r}
}

// readRan reads each key of StateRunning and the manifest its digest names (the indexer's, by digest), into in.Ran. A key
// whose manifest cannot be read is a gap, and that node's module is read as it was before (the trunk's).
func (p *Part) readRan(in *Inputs) {
	if reads := p.StreamReads(); len(reads) > 0 && reads[0].State == "waiting" {
		// The whole read of `running` has not ended: the part acts on none of it, and keeps the list it has (design 47 G §556).
		in.Gaps = append(in.Gaps, Gap{Input: StateRunning, Failed: true,
			Why: "the stream " + reads[0].Stream + " is waiting: its whole read at the part's start has not ended"})
		return
	}
	in.Ran = map[string]catalogue.Manifest{}
	in.Heard = map[string]catalogue.Heard{}
	in.RanByFingerprint = map[string]catalogue.Manifest{}
	keys, err := p.Store.Keys(StateRunning)
	if err != nil {
		in.Gaps = append(in.Gaps, Gap{Input: StateRunning, Why: "its own state cannot be read, so no seat is read from what a node runs: " + err.Error()})
		return
	}
	sort.Strings(keys)
	for _, key := range keys {
		raw, _, found, err := p.Store.Get(StateRunning, key)
		if err != nil || !found {
			continue
		}
		var r nodeengine.RunningEvent
		if json.Unmarshal(raw, &r) != nil {
			continue
		}
		if r.Stopped {
			// A stopped module is heard as stopped and claims nothing on that node (design 47 G §404, §459).
			in.Heard[key] = catalogue.Heard{Stopped: true}
			continue
		}
		if r.Manifest == "" {
			continue
		}
		m, err := p.cache.get(p.Read, r.Manifest)
		if err != nil {
			in.Gaps = append(in.Gaps, Gap{Input: "running manifest of " + key, Why: "the manifest the node-engine says it runs cannot be read, so the trunk's is used: " + err.Error()})
			continue
		}
		in.Ran[key] = m
		in.Heard[key] = catalogue.Heard{Fingerprint: r.Fingerprint}
		in.RanByFingerprint[r.Fingerprint] = m
	}
}

// RunningSaid is what the tool grants_running answers: each node and module the part has heard, the build and manifest
// digest, and when it was seen.
type RunningSaid struct {
	Running []nodeengine.RunningEvent `json:"running"`
	Note    string                    `json:"note,omitempty"`
}

// RunningOf reads the state running, one key (`<node>.<module>`) or all.
func (p *Part) RunningOf(key string) (any, error) {
	keys := []string{key}
	if key == "" {
		var err error
		if keys, err = p.Store.Keys(StateRunning); err != nil {
			return nil, err
		}
		sort.Strings(keys)
	}
	out := RunningSaid{Running: []nodeengine.RunningEvent{}}
	for _, k := range keys {
		raw, _, found, err := p.Store.Get(StateRunning, k)
		if err != nil {
			return nil, err
		}
		var r nodeengine.RunningEvent
		if found && json.Unmarshal(raw, &r) == nil {
			out.Running = append(out.Running, r)
		}
	}
	if len(out.Running) == 0 {
		out.Note = "no node-engine has said it runs anything that this part heard: seats are read from the trunk's manifests"
	}
	return out, nil
}