mesh-catalog/modules/nats/cmd/nats-tools/expiry.go

package main

// The tool of this module that sets and restores a key-value bucket's expiry (novox/hq ADR 0582, "How it is
// checked": the bus module's expiry tool sets and restores a bucket's expiry and records the act; issue 774). The
// grants part calls it at the lease write, to take the 15 seconds off the lease bucket, and at the hand-back, to put
// them back. Like the removal tools it refuses without `why`, reads what it changes first, reads it back after, and
// says what it did as an event of the module; it also records the act through the seat mesh-hand-acts (ADR 0565
// clause 11), because it changes what the mesh keeps.

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/handacts"
	"github.com/nats-io/nats.go/jetstream"
)

const bucketExpirySet = "bucket-expiry-set"

// kvPrefix is the name a bucket's stream carries.
const kvPrefix = "KV_"

// Recorded records one hand act through the seat mesh-hand-acts.
type Recorded func(handacts.Request) error

// RecordedBySeat records an act through the runtime, to the seat mesh-hand-acts alone.
func RecordedBySeat(q handacts.Request) error {
	_, err := handacts.Recorder{Ask: stdio.Ask}.Record(q)
	return err
}

// BucketExpirySet is what nats_bucket_expiry answers and says.
type BucketExpirySet struct {
	Bucket  string    `json:"bucket"`
	Was     string    `json:"was"`
	Now     string    `json:"now"`
	Why     string    `json:"why"`
	Warrant string    `json:"warrant,omitempty"`
	DryRun  bool      `json:"dry_run,omitempty"`
	Changed bool      `json:"changed"`
	Said    string    `json:"said"`
	At      time.Time `json:"at"`
}

// expiryText names an expiry: a duration, or "none".
func expiryText(d time.Duration) string {
	if d == 0 {
		return "none"
	}
	return d.String()
}

// parseExpiry reads `max_age`: "none" removes the expiry, otherwise a positive duration such as 15s.
func parseExpiry(s string) (time.Duration, error) {
	s = strings.TrimSpace(s)
	if s == "none" {
		return 0, nil
	}
	d, err := time.ParseDuration(s)
	if err != nil || d <= 0 {
		return 0, fmt.Errorf("max_age is required: a positive duration such as 15s, or none to remove the expiry (got %q): nothing was changed", s)
	}
	return d, nil
}

// SetBucketExpiry sets one key-value bucket's expiry (MaxAge of its stream) after reading it, reads it back, records
// the act and says so. A bucket already at that expiry is answered as such and records nothing: asking again is safe.
// A name that is no key-value bucket is refused and nothing changes.
func SetBucketExpiry(ctx context.Context, bus Bus, said Said, record Recorded, a map[string]any) (any, error) {
	w, err := why(a)
	if err != nil {
		return nil, err
	}
	bucket := arg(a, "bucket")
	if !name.MatchString(bucket) {
		return nil, errors.New("bucket is required, one bucket's name as nats_buckets lists it (letters, digits, - and _): nothing was changed")
	}
	maxAge, err := parseExpiry(arg(a, "max_age"))
	if err != nil {
		return nil, err
	}
	dry, _ := a["dry_run"].(bool)
	js, closeBus, err := bus(ctx)
	if err != nil {
		return nil, err
	}
	defer closeBus()
	stream := kvPrefix + bucket
	out := BucketExpirySet{Bucket: bucket, Why: w, Warrant: strings.TrimSpace(arg(a, "warrant")), DryRun: dry, At: time.Now().UTC()}
	sctx, cancel := step(ctx)
	s, err := js.Stream(sctx, stream)
	cancel()
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil, fmt.Errorf("%s is no key-value bucket (there is no stream %s): nothing was changed", bucket, stream)
	}
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("the bucket %s could not be read, so its expiry was not changed", bucket))
	}
	cfg := s.CachedInfo().Config
	out.Was, out.Now = expiryText(cfg.MaxAge), expiryText(maxAge)
	if cfg.MaxAge == maxAge {
		out.Said = fmt.Sprintf("%s already expires after %s: nothing to change", bucket, out.Now)
		return out, nil
	}
	if dry {
		out.Said = fmt.Sprintf("%s expires after %s; it would be set to %s", bucket, out.Was, out.Now)
		return out, nil
	}
	cfg.MaxAge = maxAge
	// A stream refuses a window of duplicates longer than its expiry; a bucket made with 15 s carries 15 s.
	if maxAge > 0 && (cfg.Duplicates == 0 || cfg.Duplicates > maxAge) {
		cfg.Duplicates = maxAge
	}
	sctx, cancel = step(ctx)
	_, err = js.UpdateStream(sctx, cfg)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("the expiry of %s was not changed", bucket))
	}
	sctx, cancel = step(ctx)
	s, err = js.Stream(sctx, stream)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("the new expiry of %s was accepted but could not be read back", bucket))
	}
	if got := s.CachedInfo().Config.MaxAge; got != maxAge {
		return nil, fmt.Errorf("%s reads back with an expiry of %s after the change to %s was accepted", bucket, expiryText(got), out.Now)
	}
	out.Changed = true
	out.Said = fmt.Sprintf("%s expiry set from %s to %s", bucket, out.Was, out.Now)
	what := fmt.Sprintf("set the expiry of the bucket %s from %s to %s", bucket, out.Was, out.Now)
	if out.Warrant != "" {
		what += " (warrant " + out.Warrant + ")"
	}
	if err := record(handacts.Request{What: what, Why: w, Cause: "bucket-expiry"}); err != nil {
		return nil, fmt.Errorf("the expiry of %s WAS changed (%s) but the hand act could not be recorded: %w", bucket, what, err)
	}
	if err := said(bucketExpirySet, out); err != nil {
		return nil, fmt.Errorf("the expiry of %s WAS changed (%s) but the audit event could not be said: %w", bucket, w, err)
	}
	return out, nil
}

// expiryTools is the module's tool that sets a bucket's expiry.
func expiryTools(bus Bus, said Said, record Recorded) []stdio.Tool {
	return []stdio.Tool{{Name: "nats_bucket_expiry",
		Description: "Sets or restores the expiry of one key-value bucket: `max_age` is a duration such as 15s, or `none` to keep keys with no expiry (e.g. the lease bucket of mesh-controller at the lease write, back to 15s at the hand-back; novox/hq ADR 0582). Needs `why`, kept in the audit trail and recorded as a hand act through mesh-hand-acts (`warrant` names the operator's ask when there is one); refuses a name that is no key-value bucket; reads the expiry first and back after, and a bucket already at that expiry is answered as such. `dry_run` only reads. The bus refuses any bucket the module's credential is not granted.",
		Input: map[string]any{"bucket": str("the bucket as nats_buckets lists it, e.g. mesh-controller_lease"), "max_age": str("a duration such as 15s, or none to remove the expiry"),
			"why": str("why it is changed, for the audit trail (required)"), "warrant": str("the operator's ask this follows, when there is one"),
			"dry_run": map[string]any{"type": "boolean", "description": "read the expiry, change nothing"}},
		Run: func(a map[string]any) (any, error) {
			return SetBucketExpiry(context.Background(), bus, said, record, a)
		}}}
}