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

package main

// The three tools of this module that change the bus (novox/hq ADR 0564 clause 5, issue 1952 notes 6, 7 and 9):
// remove one named consumer, remove one named stream, and remove one node's keys under a prefix of a key-value bucket. Everything else
// here reads.
//
// **Narrow by what the tool needs, and narrower by what the bus grants.** The tools reach the bus on the
// credential the module's manifest names (NATS_TOOLS_CREDENTIAL); the bus refuses whatever that credential is not
// granted, so a wrong stream, consumer or bucket is refused by the server and said here, not trusted to the
// arguments. Each tool refuses without `why`, reads what it would remove first (and says it, so a person can ask
// with dry_run and then for real), removes, reads that it is gone, and says what it did as an event of the module
// (the audit trail: who asked is in the call the tool runner audits, why and what are in the event).

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

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

// Bus opens JetStream on the bus as the module's credential for one call; close ends the connection.
type Bus func(ctx context.Context) (js jetstream.JetStream, close func(), err error)

// Said keeps the audit line: an event of this module.
type Said func(event string, body any) error

const (
	consumerRemoved   = "consumer-removed"
	bucketKeysRemoved = "bucket-keys-removed"
	streamRemoved     = "stream-removed"
	// maxKeysShown bounds the keys an answer lists; the count is always whole.
	maxKeysShown = 200
)

var (
	// A stream, consumer or bucket is one subject token: no dot, no wildcard, no space.
	name = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
	// A key prefix is subject tokens and their dots; no wildcard, so it names keys and not a pattern.
	keyPrefix = regexp.MustCompile(`^[A-Za-z0-9_./=-]+$`)
)

func why(a map[string]any) (string, error) {
	w := strings.TrimSpace(arg(a, "why"))
	if w == "" {
		return "", errors.New("why is required: say why this is removed, for the audit trail (novox/hq ADR 0564 clause 5); nothing was removed")
	}
	return w, nil
}

// answerWithin is how long the bus is given to answer one request. **The bus does not refuse a request its
// credential is not granted: it drops it** (a permissions violation is told to the connection's error handler,
// never answered), so an unanswered request is how a refusal shows here.
var answerWithin = 10 * time.Second

// unanswered says what a request with no answer most likely is.
func unanswered(err error, what string) error {
	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
		return fmt.Errorf("%s: the bus did not answer within %s; it drops, without an answer, what the module's credential is not granted, "+
			"so this is most likely outside its grant (or the bus is not reachable): %w", what, answerWithin, err)
	}
	return fmt.Errorf("%s: %w", what, err)
}

func step(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(ctx, answerWithin)
}

// ConsumerRemoved is what nats_consumer_remove answers and says.
type ConsumerRemoved struct {
	Stream   string         `json:"stream"`
	Consumer string         `json:"consumer"`
	Why      string         `json:"why"`
	Was      map[string]any `json:"was"`
	Removed  bool           `json:"removed"`
	At       time.Time      `json:"at"`
}

// RemoveConsumer removes one named consumer of one stream, after reading it, and says so.
func RemoveConsumer(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
	w, err := why(a)
	if err != nil {
		return nil, err
	}
	stream, consumer := arg(a, "stream"), arg(a, "consumer")
	if !name.MatchString(stream) || !name.MatchString(consumer) {
		return nil, errors.New("stream and consumer are required, each one name (letters, digits, - and _): nothing was removed")
	}
	js, closeBus, err := bus(ctx)
	if err != nil {
		return nil, err
	}
	defer closeBus()
	sctx, cancel := step(ctx)
	c, err := js.Consumer(sctx, stream, consumer)
	cancel()
	if err != nil {
		if errors.Is(err, jetstream.ErrConsumerNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) {
			return nil, fmt.Errorf("%s has no consumer %s (%v): nothing was removed", stream, consumer, err)
		}
		return nil, unanswered(err, fmt.Sprintf("the consumer %s of %s could not be read, so it was not removed", consumer, stream))
	}
	info := c.CachedInfo()
	was := map[string]any{
		"durable": info.Config.Durable, "filter_subject": info.Config.FilterSubject, "filter_subjects": info.Config.FilterSubjects,
		"deliver_subject": info.Config.DeliverSubject, "pending": info.NumPending, "ack_pending": info.NumAckPending,
		"redelivered": info.NumRedelivered, "created": info.Created,
	}
	sctx, cancel = step(ctx)
	err = js.DeleteConsumer(sctx, stream, consumer)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("consumer %s of %s was not removed", consumer, stream))
	}
	sctx, cancel = step(ctx)
	_, err = js.Consumer(sctx, stream, consumer)
	cancel()
	// Only the bus saying it is not there is a removal; a read that failed otherwise confirms nothing.
	if err == nil {
		return nil, fmt.Errorf("consumer %s of %s is still there after the removal was accepted", consumer, stream)
	}
	if !errors.Is(err, jetstream.ErrConsumerNotFound) && !errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil, unanswered(err, fmt.Sprintf("the removal of consumer %s of %s was accepted but could not be confirmed", consumer, stream))
	}
	out := ConsumerRemoved{Stream: stream, Consumer: consumer, Why: w, Was: was, Removed: true, At: time.Now().UTC()}
	if err := said(consumerRemoved, out); err != nil {
		return nil, fmt.Errorf("consumer %s of %s WAS removed (%s) but the audit event could not be said: %w", consumer, stream, w, err)
	}
	return out, nil
}

// KeysRemoved is what nats_bucket_remove answers and says.
type KeysRemoved struct {
	Bucket  string    `json:"bucket"`
	Prefix  string    `json:"prefix"`
	Why     string    `json:"why"`
	DryRun  bool      `json:"dry_run"`
	Count   int       `json:"count"`
	Keys    []string  `json:"keys"`
	Cut     bool      `json:"keys_cut,omitempty"`
	Removed int       `json:"removed"`
	At      time.Time `json:"at"`
}

// RemoveKeys removes the keys of a key-value bucket under a prefix: counted first, and only those.
func RemoveKeys(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
	w, err := why(a)
	if err != nil {
		return nil, err
	}
	bucket, node, prefix := arg(a, "bucket"), arg(a, "node"), arg(a, "prefix")
	if !name.MatchString(bucket) {
		return nil, errors.New("bucket is required, a bucket's name as nats_buckets lists it: nothing was removed")
	}
	if (node == "") == (prefix == "") {
		return nil, errors.New("give exactly one of node (that node's keys: those starting `<node>.`) or prefix (keys starting so): " +
			"a removal of a whole bucket is not this tool's; nothing was removed")
	}
	if node != "" {
		if !name.MatchString(node) {
			return nil, errors.New("node is one machine's name: nothing was removed")
		}
		prefix = node + "."
	}
	if !keyPrefix.MatchString(prefix) {
		return nil, errors.New("prefix is key text without wildcards or spaces: nothing was removed")
	}
	dry, _ := a["dry_run"].(bool)
	js, closeBus, err := bus(ctx)
	if err != nil {
		return nil, err
	}
	defer closeBus()
	sctx, cancel := step(ctx)
	kv, err := js.KeyValue(sctx, bucket)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("bucket %s could not be opened, nothing was removed", bucket))
	}
	sctx, cancel = step(ctx)
	stream, err := js.Stream(sctx, "KV_"+bucket)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("bucket %s has no stream to read its keys from", bucket))
	}
	// The server narrows to the prefix's tokens when the prefix ends on a token; any other prefix is cut here.
	filter := "$KV." + bucket + ".>"
	if strings.HasSuffix(prefix, ".") {
		filter = "$KV." + bucket + "." + prefix + ">"
	}
	sctx, cancel = step(ctx)
	info, err := stream.Info(sctx, jetstream.WithSubjectFilter(filter))
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("the keys of %s could not be read, so none were removed", bucket))
	}
	var keys []string
	for subject := range info.State.Subjects {
		if key := strings.TrimPrefix(subject, "$KV."+bucket+"."); strings.HasPrefix(key, prefix) {
			keys = append(keys, key)
		}
	}
	sort.Strings(keys)
	out := KeysRemoved{Bucket: bucket, Prefix: prefix, Why: w, DryRun: dry, Count: len(keys), Keys: keys, At: time.Now().UTC()}
	if len(keys) > maxKeysShown {
		out.Keys, out.Cut = keys[:maxKeysShown], true
	}
	if want, ok := a["expect"].(float64); ok && int(want) != len(keys) {
		return nil, fmt.Errorf("%d keys of %s start with %q, not the %d expected: nothing was removed", len(keys), bucket, prefix, int(want))
	}
	if dry || len(keys) == 0 {
		return out, nil
	}
	for _, k := range keys {
		sctx, cancel := step(ctx)
		err := kv.Purge(sctx, k)
		cancel()
		if err != nil {
			return nil, unanswered(err, fmt.Sprintf("removing %q from %s failed after %d of %d keys, the rest were left", k, bucket, out.Removed, len(keys)))
		}
		out.Removed++
	}
	if err := said(bucketKeysRemoved, out); err != nil {
		return nil, fmt.Errorf("%d keys of %s under %q WERE removed (%s) but the audit event could not be said: %w", out.Removed, bucket, prefix, w, err)
	}
	return out, nil
}

// StreamRemoved is what nats_stream_remove answers and says.
type StreamRemoved struct {
	Stream    string    `json:"stream"`
	Why       string    `json:"why"`
	DryRun    bool      `json:"dry_run"`
	Present   bool      `json:"present"`
	Messages  uint64    `json:"messages"`
	Bytes     uint64    `json:"bytes"`
	Consumers int       `json:"consumers"`
	Removed   bool      `json:"removed"`
	Said      string    `json:"said"`
	At        time.Time `json:"at"`
}

// RemoveStream removes one named stream whole, after reading it, and says so (the controller's leftovers: its streams
// CONTROL and ASSIGNMENTS, its buckets' and object stores' streams; novox/hq ADR 0590 §6). **A stream already gone is no error**: the removal is the same act
// however often it is asked, and says it found nothing. A stream with a consumer is refused, naming how many: a reader
// is still attached, and that is for a person to look at before the stream goes.
func RemoveStream(ctx context.Context, bus Bus, said Said, a map[string]any) (any, error) {
	w, err := why(a)
	if err != nil {
		return nil, err
	}
	stream := arg(a, "stream")
	if !name.MatchString(stream) {
		return nil, errors.New("stream is required, one stream's name (letters, digits, - and _): nothing was removed")
	}
	dry, _ := a["dry_run"].(bool)
	js, closeBus, err := bus(ctx)
	if err != nil {
		return nil, err
	}
	defer closeBus()
	out := StreamRemoved{Stream: stream, Why: w, DryRun: dry, At: time.Now().UTC()}
	sctx, cancel := step(ctx)
	s, err := js.Stream(sctx, stream)
	cancel()
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		out.Said = stream + " is not there: nothing to remove"
		return out, nil
	}
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("the stream %s could not be read, so it was not removed", stream))
	}
	sctx, cancel = step(ctx)
	info, err := s.Info(sctx)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("the stream %s could not be read, so it was not removed", stream))
	}
	out.Present, out.Messages, out.Bytes, out.Consumers = true, info.State.Msgs, info.State.Bytes, info.State.Consumers
	if info.State.Consumers > 0 {
		return nil, fmt.Errorf("%s has %d consumers: a reader is attached, so it was not removed", stream, info.State.Consumers)
	}
	if dry {
		out.Said = fmt.Sprintf("%s holds %d messages (%d bytes) and no consumer; it would be removed", stream, out.Messages, out.Bytes)
		return out, nil
	}
	sctx, cancel = step(ctx)
	err = js.DeleteStream(sctx, stream)
	cancel()
	if err != nil {
		return nil, unanswered(err, fmt.Sprintf("stream %s was not removed", stream))
	}
	sctx, cancel = step(ctx)
	_, err = js.Stream(sctx, stream)
	cancel()
	// Only the bus saying it is not there is a removal; a read that failed otherwise confirms nothing.
	if err == nil {
		return nil, fmt.Errorf("stream %s is still there after the removal was accepted", stream)
	}
	if !errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil, unanswered(err, fmt.Sprintf("the removal of stream %s was accepted but could not be confirmed", stream))
	}
	out.Removed = true
	out.Said = fmt.Sprintf("%s removed: %d messages (%d bytes) and no consumer", stream, out.Messages, out.Bytes)
	if err := said(streamRemoved, out); err != nil {
		return nil, fmt.Errorf("stream %s WAS removed (%s) but the audit event could not be said: %w", stream, w, err)
	}
	return out, nil
}

// removalTools are the module's three tools that change the bus; every other tool here reads.
func removalTools(bus Bus, said Said) []stdio.Tool {
	run := func(f func(context.Context, Bus, Said, map[string]any) (any, error)) func(map[string]any) (any, error) {
		return func(a map[string]any) (any, error) { return f(context.Background(), bus, said, a) }
	}
	return []stdio.Tool{
		{Name: "nats_consumer_remove",
			Description: "Removes (deletes) one named JetStream consumer of one stream, e.g. a consumer the controller left, such as `controller` of DEAD_LETTER_NOTICES or `controller-merges` of EVENTS (novox/hq ADR 0590 §6, recorded as a hand act under issue 1126). Needs `why`, kept in the audit trail; refuses an unknown stream or consumer; reads the consumer first and answers what it was (filters, pending, unacknowledged); the stream and its other consumers are not touched. The bus refuses any consumer the module's credential is not granted.",
			Input:       map[string]any{"stream": str("the stream, e.g. DEAD_LETTER_NOTICES"), "consumer": str("the consumer's name, e.g. controller"), "why": str("why it is removed, for the audit trail (required)")},
			Run:         run(RemoveConsumer)},
		{Name: "nats_stream_remove",
			Description: "Removes (deletes) one named JetStream stream whole, e.g. a stream the controller left: CONTROL, ASSIGNMENTS, one of its buckets as KV_<bucket> or a node's declaration files as OBJ_declaration-files-<node> (novox/hq ADR 0590 §6; a bucket's whole removal is this tool's, never nats_bucket_remove's). Needs `why`, kept in the audit trail; reads the stream first and answers its messages, bytes and consumers; refuses a stream that still has a consumer; a stream already gone is answered as such and is no error, so asking again is safe. `dry_run` only reads. The bus refuses any stream the module's credential is not granted.",
			Input:       map[string]any{"stream": str("the stream, e.g. ASSIGNMENTS"), "why": str("why it is removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "read the stream, remove nothing"}},
			Run:         run(RemoveStream)},
		{Name: "nats_bucket_remove",
			Description: "Removes (purges) keys from a key-value bucket: one node's keys (`node`, the keys starting `<node>.`) or the keys under a `prefix`, e.g. a full acts bucket of the node-engine. Needs `why`, kept in the audit trail; never the whole bucket. Counts the keys first and lists them (`dry_run` only counts; `expect` refuses when the count differs); removes only those keys, and their history. See the buckets and their keys with nats_buckets. 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-host_acts"), "node": str("a node whose keys (`<node>.`) go"), "prefix": str("key prefix instead of node"),
				"why": str("why they are removed, for the audit trail (required)"), "dry_run": map[string]any{"type": "boolean", "description": "count and list the keys, remove none"}, "expect": num("the count of keys expected; refuses when it differs")},
			Run: run(RemoveKeys)},
	}
}