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

package grants

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

	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
)

// **The controller's lease, what still depends on the controller, and a dry run of the lease write's check** (issue 774,
// design 47 L3; novox/hq ADR 0547 §11, ADR 0561 §3, ADR 0571 §13). One read-only answer from bus state: the lease key,
// every manifest the indexer holds, and the bus module's streams. It writes nothing, and asks the controller nothing.
//
// It lives on the grants part because the part is where the lease write takes most of its effect and already reads all
// it needs: the lease (buscompose.NamedReaders names it), every manifest at the indexer's trunk, placement, and the bus's
// streams through the bus module's read tools. No record names the home of the lease write's check; ADR 0571 §13 names
// its test in three repositories, and this is the same judgement over the manifests the mesh holds now.

// ToolLeaseCheck is the tool's name.
const ToolLeaseCheck = "grants_lease_check"

// Controller is the module and the seat the check is about.
const Controller = "mesh-controller"

// **The judge is mesh-sdk's** (catalogue.Dependencies, catalogue.Judge; issue 774): the same judgement the three
// repositories' manifest tests run through catalogue.ManifestRefusals (ADR 0571 §13), re-exported here by its names.
const (
	DependsRead    = catalogue.DependsRead
	DependsInvoke  = catalogue.DependsInvoke
	DependsConsume = catalogue.DependsConsume
	DependsUses    = catalogue.DependsUses
	DependsWaits   = catalogue.DependsWaits
	PhaseBefore    = catalogue.PhaseBefore
	PhaseWrite     = catalogue.PhaseWrite
)

type (
	// Dependency, Kept, Verdict and Coverage are mesh-sdk's.
	Dependency = catalogue.Dependency
	Kept       = catalogue.KeptDependency
	Verdict    = catalogue.Verdict
	Coverage   = catalogue.Coverage
)

// KeepList and EndedBy are mesh-sdk's named list and the records that ended what it no longer keeps.
var (
	KeepList = catalogue.KeepList
	EndedBy  = catalogue.EndedBy
)

// Dependencies are every declaration on the controller in the manifests given (catalogue.Dependencies).
func Dependencies(manifests map[string]catalogue.Manifest) []Dependency {
	return catalogue.Dependencies(manifests)
}

// Judge runs the lease write's check in one phase (catalogue.Judge).
func Judge(deps []Dependency, phase string, cov Coverage, handedAway bool, leaseSays string) Verdict {
	return catalogue.Judge(deps, phase, cov, handedAway, leaseSays)
}

// LeaseNow is who holds the lease and what follows from it.
type LeaseNow struct {
	HandedAway bool                    `json:"handed_away"`
	Says       string                  `json:"says"`
	Holder     *buscompose.LeaseHolder `json:"holder,omitempty"`
	Since      *time.Time              `json:"since,omitempty"`
	Does       []string                `json:"does"`
}

// While the controller holds the lease, and once it is handed away: what the holder and the parts watching the key do
// because of it, by the records that say so.
var (
	DoesWhileController = []string{
		"the controller mints every new bus account; the vault mints none and says what it holds back (ADR 0561 §3)",
		"the grants part creates no durable consumer an account needs, and only compares them (ADR 0561 §6)",
		"the grants part publishes memberships on the controller's assignments stream beside its own state (ADR 0571 §10)",
		"mesh-healers writes each act it would take in `would`, and acts on none (ADR 0564 §8)",
		"the artifact sweep lets go of nothing; the controller's take-in sweeps and says `artifacts-gone` (ADR 0571 §4)",
		"the watchman only lists dead letters; the controller keeps DEAD_LETTERS through its consumer of DEAD_LETTER_NOTICES (ADR 0564 §5)",
		"the controller keeps composing for the bus node, whose agent follows its user list, so the hand-backs of config, conditions, placement and the user list still answer (ADR 0547 §8, ADR 0564 §9, ADR 0565 §4, ADR 0567 §11)",
	}
	DoesHandedAway = []string{
		"the vault mints every new bus account (ADR 0561 §1)",
		"the grants part creates the durable consumers an account needs, and stops publishing on the assignments stream (ADR 0561 §6, ADR 0571 §10)",
		"mesh-healers acts (ADR 0564 §8)",
		"the artifact sweep sweeps and records every deletion in the hand-act log (ADR 0571 §4)",
		"the watchman keeps the dead letters (ADR 0564 §5)",
		"every hand-back is answered as gone (ADR 0547 §11, ADR 0564 §9, ADR 0565 §4, ADR 0567 §11)",
	}
)

// Lease reads the lease key and says who holds it, since when, and what follows.
func Lease(r Reader) LeaseNow {
	raw, found, err := r.Get(ReadLease, LeaseKey)
	away, says := buscompose.LeaseHandedAway(raw, found, err)
	out := LeaseNow{HandedAway: away, Says: says, Does: DoesWhileController}
	if away {
		out.Does = DoesHandedAway
	}
	if err == nil && found && len(raw) > 0 {
		var h buscompose.LeaseHolder
		if json.Unmarshal(raw, &h) == nil {
			out.Holder = &h
			if !h.Taken.IsZero() {
				t := h.Taken
				out.Since = &t
			}
		}
	}
	return out
}

// StreamUse is one of the controller's streams (its buckets, the assignments stream, the dead-letter notices) as the bus
// module answers it: who consumes it now, and what it last carried.
type StreamUse struct {
	Stream    string   `json:"stream"`
	Consumers int      `json:"consumers"`
	Durable   []string `json:"durable,omitempty"`
	Messages  int64    `json:"messages"`
	Subjects  int64    `json:"subjects_held,omitempty"`
	Last      string   `json:"last,omitempty"`
}

// The streams of the controller's that are not its buckets: the assignments stream it publishes memberships on, and the
// notices its consumer `controller` turns into dead letters until the lease write (ADR 0564 §5).
const (
	StreamAssignments       = "ASSIGNMENTS"
	StreamDeadLetterNotices = "DEAD_LETTER_NOTICES"
)

// controllerStream says whether a stream is the controller's.
func controllerStream(name string) bool {
	return strings.HasPrefix(name, "KV_"+Controller+"_") || name == StreamAssignments || name == StreamDeadLetterNotices
}

// Duties reads, from the bus module's streams, every stream of the controller's someone consumes, with its durable
// consumers by name (a watcher of a bucket is ephemeral and names nobody). Streams with no consumer are counted apart.
func Duties(c Caller) (used []StreamUse, idle []string, err error) {
	raw, err := c.Call(CallStreams, map[string]any{})
	if err != nil {
		return nil, nil, fmt.Errorf("%s: %w", CallStreams, err)
	}
	type consumer struct {
		Name string `json:"name"`
		Kind string `json:"kind"`
		Max  int    `json:"max_deliver"`
		Ack  string `json:"ack_policy"`
	}
	type stream struct {
		Name      string     `json:"name"`
		Consumers int        `json:"consumers"`
		Messages  int64      `json:"messages"`
		Held      int64      `json:"subjects_held"`
		Last      string     `json:"last"`
		Detail    []consumer `json:"consumer_detail"`
	}
	var got struct {
		Detail []stream `json:"detail"`
	}
	if err := json.Unmarshal(unwrap(raw), &got); err != nil {
		return nil, nil, fmt.Errorf("%s answered no streams: %w", CallStreams, err)
	}
	for _, s := range got.Detail {
		if !controllerStream(s.Name) {
			continue
		}
		if s.Consumers == 0 {
			idle = append(idle, s.Name)
			continue
		}
		u := StreamUse{Stream: s.Name, Consumers: s.Consumers, Messages: s.Messages, Subjects: s.Held, Last: s.Last}
		if one, err := c.Call(CallStreams, map[string]any{"stream": s.Name}); err == nil {
			var d struct {
				Detail []stream `json:"detail"`
			}
			if json.Unmarshal(unwrap(one), &d) == nil && len(d.Detail) > 0 {
				for _, cs := range d.Detail[0].Detail {
					if durable(cs.Name) {
						u.Durable = append(u.Durable, cs.Name)
					}
				}
			}
		}
		used = append(used, u)
	}
	sort.Slice(used, func(i, j int) bool { return used[i].Stream < used[j].Stream })
	sort.Strings(idle)
	return used, idle, nil
}

// durable says whether a consumer's name is one a part gave it: the bus names an ephemeral watcher with eight random
// letters and digits, which no part of the mesh uses.
func durable(name string) bool {
	if len(name) != 8 {
		return true
	}
	for _, r := range name {
		if !(r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9') {
			return true
		}
	}
	return false
}

// LeaseAnswer is the tool's answer.
type LeaseAnswer struct {
	Lease        LeaseNow     `json:"lease"`
	Dependencies []Dependency `json:"dependencies"`
	// Unkept are the dependencies no record keeps, by module: what the check refuses in every phase.
	Unkept    map[string][]string `json:"unkept"`
	Manifests int                 `json:"manifests_read"`
	// Indexed is how many modules the indexer holds; -1 when its keys could not be listed.
	Indexed int   `json:"modules_indexed"`
	Unread  []Gap `json:"manifests_unread,omitempty"`
	// Streams are the controller's streams someone consumes now; Idle those no one does.
	Streams      []StreamUse `json:"streams,omitempty"`
	Idle         []string    `json:"streams_idle,omitempty"`
	StreamsError string      `json:"streams_unread,omitempty"`
	// Check is the dry run at the write (the three alone), Before the same over the whole list before the write's
	// delivery.
	Check  Verdict  `json:"check"`
	Before Verdict  `json:"check_before_delivery"`
	Beside []string `json:"not_judged_here"`
}

// NotJudgedHere are the lease write's other conditions, each with the tool that answers it (design 47 L3, ADR 0547 §11).
var NotJudgedHere = []string{
	"L4 to L7 delivered: the issue tracker's milestones 1952, 1953, 1954 and 1955",
	"every node's record trusts the vault's key: mesh-delivery-view.node_trusts (ADR 0550 §2)",
	"mesh-config's compare empty and its take-over done, and mesh-wants.upgrade compare empty: mesh-config.compare, mesh-wants.upgrade (ADR 0565 §14)",
	"the condition keeper's take-over done: mesh-conditions.compare (ADR 0564 §9)",
}

// LeaseCheck answers the tool: who holds the lease and what follows, every dependency on the controller with what keeps
// it, the controller's streams in use, and the lease write's check run dry. Read-only.
func (p *Part) LeaseCheck() (any, error) {
	in := ReadInputs(p.Read, &p.cache)
	out := LeaseAnswer{Lease: Lease(p.Read), Unkept: map[string][]string{}, Manifests: len(in.Manifests), Unread: in.Refused,
		Beside: NotJudgedHere}
	for _, g := range in.Gaps {
		if g.Failed && strings.HasPrefix(g.Input, ReadModules) {
			out.Unread = append(out.Unread, g)
		}
	}
	out.Dependencies = Dependencies(in.Manifests)
	if out.Dependencies == nil {
		out.Dependencies = []Dependency{}
	}
	for _, d := range out.Dependencies {
		if d.Kept == "" {
			out.Unkept[d.Module] = append(out.Unkept[d.Module], d.Kind+" "+d.What)
		}
	}
	if used, idle, err := Duties(p.Call); err != nil {
		out.StreamsError = err.Error()
	} else {
		out.Streams, out.Idle = used, idle
	}
	// Only the indexer's own modules count as read: a manifest the mesh provides itself must not stand in for one missing.
	cov := Coverage{Indexed: -1}
	for _, g := range out.Unread {
		cov.Unread = append(cov.Unread, catalogue.Unread{Input: g.Input, Why: g.Why})
	}
	if keys, err := p.Read.Keys(ReadModules); err == nil {
		cov.Indexed = len(keys)
		for _, k := range keys {
			if _, ok := in.Manifests[k]; ok {
				cov.Read++
			}
		}
	}
	out.Indexed = cov.Indexed
	out.Check = Judge(out.Dependencies, PhaseWrite, cov, out.Lease.HandedAway, out.Lease.Says)
	out.Before = Judge(out.Dependencies, PhaseBefore, cov, out.Lease.HandedAway, out.Lease.Says)
	return out, nil
}