mesh-catalog/modules/nats/internal/keeper/plan.go

// Package keeper is the bus module's keeper (novox/hq design 25 § 12.2-12.3, ADR 0595 §4-§6): it creates the bus state
// every accepted manifest asks for in `requires`, compares what the bus holds with what was asked, finds orphans and
// reads the fill of what refuses when full, and says each on the bus module's own events. It reads nothing of another
// part's state: what to create comes on the indexer's manifest-accepted, and its first view from the indexer's tool
// indexer_bus_state. Everything it touches it reaches through the interfaces in keeper.go; this file is pure.
package keeper

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

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

// The kinds of bus object an entry is made of: a stream, a key-value bucket (stream KV_<name>) and an object store
// (stream OBJ_<name>).
const (
	KindStream  = "stream"
	KindBucket  = "key-value"
	KindObjects = "object-store"
)

// Separator joins a module's name and a local name into a bus name (design 25 § 12.1: `<module>_<local>`); a local
// name holding it is refused, so no two modules' names meet.
const Separator = "_"

// Fixed settings of the templates (design 25 § 12.1's table): never options.
const (
	MiB           = int64(1) << 20
	StateCap      = 64 * MiB
	ValueMost     = 256 << 10
	EntryMost     = 256 << 10
	DefaultCapMiB = 1024
	EventsDays    = 7
)

// Settings are one bus object's settings, as asked and as held: what check compares.
type Settings struct {
	// MaxBytes is its cap in bytes; MaxAgeSeconds its age, 0 for none.
	MaxBytes      int64 `json:"max-bytes"`
	MaxAgeSeconds int64 `json:"max-age-seconds,omitempty"`
	// History is how many values a key keeps (a bucket's), or how many messages a subject keeps (a snapshot's); 0 none.
	History int64 `json:"history,omitempty"`
	// Discard is what it does when full: "new" (refuses) or "old" (drops the oldest).
	Discard string `json:"discard"`
	// WorkQueue says a message is removed once taken.
	WorkQueue bool `json:"work-queue,omitempty"`
	// MaxMsgSize is the largest message (a value, a log entry), 0 none.
	MaxMsgSize int64 `json:"max-msg-size,omitempty"`
	// DuplicatesSeconds is a work queue's duplicate window.
	DuplicatesSeconds int64 `json:"duplicates-seconds,omitempty"`
	// KeyTTL says a bucket's key may be given a lifetime of its own.
	KeyTTL bool `json:"key-ttl,omitempty"`
}

// The words of Discard.
const (
	DiscardNew = "new"
	DiscardOld = "old"
)

// Refuses says the object refuses a new message when full: the ones the fill pass reads (ADR 0595 §6).
func (s Settings) Refuses() bool { return s.Discard == DiscardNew }

// Object is one bus object an entry is made of: its kind, its name as the kind names it (a bucket by its bucket name,
// a stream by its stream name), its subjects for a stream, and its settings.
type Object struct {
	Kind     string   `json:"kind"`
	Name     string   `json:"name"`
	Subjects []string `json:"subjects,omitempty"`
	Settings Settings `json:"settings"`
}

// Stream is the stream the bus keeps the object on: KV_<bucket>, OBJ_<store>, or the stream itself.
func (o Object) Stream() string {
	switch o.Kind {
	case KindBucket:
		return "KV_" + o.Name
	case KindObjects:
		return "OBJ_" + o.Name
	}
	return o.Name
}

// Plan is what one entry is on the bus: its binding (the names the module receives) and the objects it is made of.
type Plan struct {
	Module   string                  `json:"module"`
	Local    string                  `json:"local"`
	Storage  string                  `json:"storage"`
	Template string                  `json:"template,omitempty"`
	Binding  membership.BusBinding   `json:"binding"`
	Objects  []Object                `json:"objects"`
	Entry    catalogue.BusStateEntry `json:"entry"`
}

// BusName is `<module>_<local>`, the name the mesh gives an entry on the bus (design 25 § 12.1).
func BusName(module, local string) string { return module + Separator + local }

// StreamSubject is the subject a stream entry's messages are under (`mesh.bus.<module>.<local>`, messages on its `.>`).
func StreamSubject(module, local string) string { return "mesh.bus." + module + "." + local }

// PlanOf is the plan of one entry of module, or why it cannot be made: a local name holding the separator, a storage
// type and template that do not go together, an option of the wrong shape. The judge refuses each of these first
// (ADR 0595 §8); this is the keeper's own guard, so a manifest judged by an older judge is refused, never half-made.
func PlanOf(module string, e catalogue.BusStateEntry) (Plan, error) {
	if e.Name == "" || strings.Contains(e.Name, Separator) || strings.ContainsAny(e.Name, ".*> ") {
		return Plan{}, fmt.Errorf("the local name %q is empty or holds %q, a dot, a wildcard or a space: the mesh names it %s on the bus, and two modules' names must never meet", e.Name, Separator, BusName(module, "<local>"))
	}
	name := BusName(module, e.Name)
	p := Plan{Module: module, Local: e.Name, Storage: e.Storage, Template: e.Template, Entry: e,
		Binding: membership.BusBinding{Local: e.Name, Storage: e.Storage, Template: e.Template}}
	opts := options{raw: e.Options}
	conf := options{raw: e.Configuration}
	streamOf := func(s Settings) Object {
		subject := StreamSubject(module, e.Name)
		p.Binding.Stream, p.Binding.Subject = name, subject
		return Object{Kind: KindStream, Name: name, Subjects: []string{subject + ".>"}, Settings: s}
	}
	switch {
	case e.Storage == catalogue.StorageKeyValue && e.Template == catalogue.TemplateState:
		s, err := stateSettings(opts)
		if err != nil {
			return Plan{}, err
		}
		p.Binding.Bucket = name
		p.Objects = []Object{{Kind: KindBucket, Name: name, Settings: s}}
	case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateEvents:
		age, err := opts.int(catalogue.OptionAgeDays, EventsDays)
		if err != nil {
			return Plan{}, err
		}
		capMiB, err := opts.int(catalogue.OptionCapMiB, DefaultCapMiB)
		if err != nil {
			return Plan{}, err
		}
		p.Objects = []Object{streamOf(Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: DiscardOld})}
	case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateWorkQueue:
		capMiB, err := opts.int(catalogue.OptionCapMiB, DefaultCapMiB)
		if err != nil {
			return Plan{}, err
		}
		age, err := opts.int(catalogue.OptionAgeDays, 0)
		if err != nil {
			return Plan{}, err
		}
		dup, err := opts.int(catalogue.OptionDuplicates, 0)
		if err != nil {
			return Plan{}, err
		}
		p.Objects = []Object{streamOf(Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: DiscardNew,
			WorkQueue: true, DuplicatesSeconds: dup})}
	case e.Storage == catalogue.StorageStream && e.Template == catalogue.TemplateLog:
		s, err := logSettings(opts)
		if err != nil {
			return Plan{}, err
		}
		p.Objects = []Object{streamOf(s)}
	case e.Storage == catalogue.StorageObjectStore && e.Template == catalogue.TemplateObjects:
		capMiB, err := opts.int(catalogue.OptionCapMiB, 0)
		if err != nil {
			return Plan{}, err
		}
		if capMiB <= 0 {
			return Plan{}, fmt.Errorf("objects %q gives no %s, which the template objects requires", e.Name, catalogue.OptionCapMiB)
		}
		p.Binding.ObjectStore = name
		p.Objects = []Object{{Kind: KindObjects, Name: name, Settings: Settings{MaxBytes: capMiB * MiB, Discard: DiscardNew}}}
	case e.Storage == catalogue.StorageLedger && e.Template == catalogue.TemplateLedger:
		history, err := logSettings(opts)
		if err != nil {
			return Plan{}, err
		}
		// Keeping days only in a ledger's own configuration (design 25 § 12.1).
		age, err := conf.int(catalogue.OptionAgeDays, 0)
		if err != nil {
			return Plan{}, err
		}
		history.MaxAgeSeconds = days(age)
		state, err := stateSettings(options{})
		if err != nil {
			return Plan{}, err
		}
		subject := buscompose.LedgerHistorySubject(module, e.Name)
		snapshot := buscompose.LedgerSnapshotSubject(module, e.Name)
		p.Binding.History, p.Binding.Subject = name, subject
		p.Binding.State = name + "_state"
		p.Binding.Snapshot = snapshot
		p.Binding.Types = e.Types
		p.Objects = []Object{
			{Kind: KindStream, Name: name, Subjects: []string{subject + ".>"}, Settings: history},
			{Kind: KindBucket, Name: p.Binding.State, Settings: state},
		}
		var indexes []string
		for _, t := range e.Types {
			for _, ix := range t.Indexes {
				indexes = append(indexes, ix.Name)
			}
		}
		sort.Strings(indexes)
		for _, ix := range indexes {
			if p.Binding.Indexes == nil {
				p.Binding.Indexes = map[string]string{}
			}
			if _, twice := p.Binding.Indexes[ix]; twice {
				continue
			}
			p.Binding.Indexes[ix] = name + "_index-" + ix
			p.Objects = append(p.Objects, Object{Kind: KindBucket, Name: p.Binding.Indexes[ix], Settings: state})
		}
		// The snapshot: the latest state per instance, never expiring (ADR 0597 §8), one message per subject.
		p.Objects = append(p.Objects, Object{Kind: KindStream, Name: name + "_snapshot", Subjects: []string{snapshot + ".>"},
			Settings: Settings{MaxBytes: history.MaxBytes, History: 1, Discard: DiscardOld}})
	case e.Storage == catalogue.StorageStream && e.Template == "":
		// A bare stream: its own configuration, which must give when full, a cap and an age (design 25 § 12.1).
		var full string
		if raw, ok := conf.raw[catalogue.ConfigWhenFull]; ok {
			_ = json.Unmarshal(raw, &full)
		}
		capMiB, err := conf.int(catalogue.OptionCapMiB, 0)
		if err != nil {
			return Plan{}, err
		}
		age, err := conf.int(catalogue.OptionAgeDays, -1)
		if err != nil {
			return Plan{}, err
		}
		discard := map[string]string{catalogue.WhenFullRefuse: DiscardNew, catalogue.WhenFullDropOldest: DiscardOld}[full]
		if discard == "" || capMiB <= 0 || age < 0 {
			return Plan{}, fmt.Errorf("the bare stream %q must give %s (refuse or drop-oldest), %s and %s in its configuration", e.Name,
				catalogue.ConfigWhenFull, catalogue.OptionCapMiB, catalogue.OptionAgeDays)
		}
		p.Objects = []Object{streamOf(Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: discard})}
	default:
		return Plan{}, fmt.Errorf("the storage type %q with the template %q is none the bus module makes (design 25 § 12.1)", e.Storage, e.Template)
	}
	return p, nil
}

func stateSettings(o options) (Settings, error) {
	history, err := o.int(catalogue.OptionPastValues, 1)
	if err != nil {
		return Settings{}, err
	}
	ttl, err := o.int(catalogue.OptionTimeToLive, 0)
	if err != nil {
		return Settings{}, err
	}
	keyTTL, err := o.bool(catalogue.OptionLifetimePerKey)
	if err != nil {
		return Settings{}, err
	}
	if perMachine, err := o.bool(catalogue.OptionPerMachine); err != nil || perMachine {
		if err != nil {
			return Settings{}, err
		}
		return Settings{}, fmt.Errorf("the option %s (one bucket per machine) is not made by the bus module yet: the module keeps one bucket", catalogue.OptionPerMachine)
	}
	return Settings{MaxBytes: StateCap, MaxAgeSeconds: ttl, History: history, Discard: DiscardNew, MaxMsgSize: ValueMost, KeyTTL: keyTTL}, nil
}

func logSettings(o options) (Settings, error) {
	capMiB, err := o.int(catalogue.OptionCapMiB, DefaultCapMiB)
	if err != nil {
		return Settings{}, err
	}
	age, err := o.int(catalogue.OptionAgeDays, 0)
	if err != nil {
		return Settings{}, err
	}
	return Settings{MaxBytes: capMiB * MiB, MaxAgeSeconds: days(age), Discard: DiscardNew, MaxMsgSize: EntryMost}, nil
}

func days(n int64) int64 {
	if n <= 0 {
		return 0
	}
	return int64((time.Duration(n) * 24 * time.Hour) / time.Second)
}

// options reads a template's options as the manifest wrote them.
type options struct{ raw map[string]json.RawMessage }

func (o options) int(name string, fallback int64) (int64, error) {
	raw, ok := o.raw[name]
	if !ok {
		return fallback, nil
	}
	n, err := strconv.ParseInt(strings.TrimSpace(string(raw)), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("the option %s is %s, not a whole number", name, raw)
	}
	return n, nil
}

func (o options) bool(name string) (bool, error) {
	raw, ok := o.raw[name]
	if !ok {
		return false, nil
	}
	var b bool
	if err := json.Unmarshal(raw, &b); err != nil {
		return false, fmt.Errorf("the option %s is %s, not true or false", name, raw)
	}
	return b, nil
}

// Grows is whether held reaches asked by growing alone (a larger cap, a longer age, more past values), so the keeper
// applies it without a person (ADR 0595 §6); false for any other difference, which is a drift a person decides. Equal
// settings grow by nothing and are no difference (Differs).
func Grows(asked, held Settings) bool {
	if asked.Discard != held.Discard || asked.WorkQueue != held.WorkQueue || asked.MaxMsgSize != held.MaxMsgSize ||
		asked.DuplicatesSeconds != held.DuplicatesSeconds || asked.KeyTTL != held.KeyTTL {
		return false
	}
	return asked.MaxBytes >= held.MaxBytes && longer(asked.MaxAgeSeconds, held.MaxAgeSeconds) && asked.History >= held.History
}

// longer is whether age a keeps at least as long as b, 0 being for ever.
func longer(a, b int64) bool {
	if a == 0 {
		return true
	}
	return b != 0 && a >= b
}

// Differs says how held differs from asked, field by field, or "" when it does not.
func Differs(asked, held Settings) string {
	var out []string
	add := func(name string, a, h any) {
		if fmt.Sprint(a) != fmt.Sprint(h) {
			out = append(out, fmt.Sprintf("%s asked %v, held %v", name, a, h))
		}
	}
	add("max-bytes", asked.MaxBytes, held.MaxBytes)
	add("max-age-seconds", asked.MaxAgeSeconds, held.MaxAgeSeconds)
	add("history", asked.History, held.History)
	add("discard", asked.Discard, held.Discard)
	add("work-queue", asked.WorkQueue, held.WorkQueue)
	add("max-msg-size", asked.MaxMsgSize, held.MaxMsgSize)
	add("duplicates-seconds", asked.DuplicatesSeconds, held.DuplicatesSeconds)
	add("key-ttl", asked.KeyTTL, held.KeyTTL)
	return strings.Join(out, "; ")
}