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

package keeper

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

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/bussnap"
)

// JetStream is the bus as the keeper's own account reaches it (Bus): each object a stream, a key-value bucket or an
// object store of JetStream, read, created, grown and deleted by its stream.
type JetStream struct {
	JS jetstream.JetStream
	// Snap takes a snapshot of streams and says where it is kept (bussnap through the keeper's own connection).
	Snap func(ctx context.Context, streams []string) (string, error)
	// NC and Dir are the connection and the scratch directory the keeper's restore in place of a snapshot uses (design 47 G §681).
	NC  *nats.Conn
	Dir string
	Log func(string, ...any)
	// Server reads the server's own storage limit in bytes (0 none), from its monitoring endpoint.
	Server func(ctx context.Context) (int64, error)
}

// Limits reads the account's storage limit (-1 is none) and the server's (design 47 G §547).
func (b JetStream) Limits(ctx context.Context) (Limits, error) {
	info, err := b.JS.AccountInfo(ctx)
	if err != nil {
		return Limits{}, err
	}
	var l Limits
	if info.Limits.MaxStore > 0 {
		l.AccountBytes = info.Limits.MaxStore
	}
	if b.Server != nil {
		// The server's limit is only shown, never held against (§547): one that cannot be read is shown as none.
		if n, err := b.Server(ctx); err == nil {
			l.ServerBytes = n
		}
	}
	return l, nil
}

// Held reads the object's stream.
func (b JetStream) Held(ctx context.Context, o Object) (Settings, Fill, bool, error) {
	s, err := b.JS.Stream(ctx, o.Stream())
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return Settings{}, Fill{}, false, nil
	}
	if err != nil {
		return Settings{}, Fill{}, false, retryable(err)
	}
	info, err := s.Info(ctx)
	if err != nil {
		return Settings{}, Fill{}, false, retryable(err)
	}
	return SettingsOf(o.Kind, info.Config), Fill{Bytes: info.State.Bytes, Messages: info.State.Msgs}, true, nil
}

// SettingsOf reads a stream's configuration as the settings an object of kind is asked with.
func SettingsOf(kind string, c jetstream.StreamConfig) Settings {
	s := Settings{MaxBytes: c.MaxBytes, MaxAgeSeconds: int64(c.MaxAge / time.Second), Discard: DiscardOld,
		WorkQueue: c.Retention == jetstream.WorkQueuePolicy, DuplicatesSeconds: int64(c.Duplicates / time.Second)}
	if c.Discard == jetstream.DiscardNew {
		s.Discard = DiscardNew
	}
	if c.MaxMsgSize > 0 {
		s.MaxMsgSize = int64(c.MaxMsgSize)
	}
	switch kind {
	case KindBucket:
		s.History = c.MaxMsgsPerSubject
		s.KeyTTL = c.AllowMsgTTL
		s.DuplicatesSeconds = 0
	case KindObjects:
		s.DuplicatesSeconds = 0
	default:
		if c.MaxMsgsPerSubject > 0 {
			s.History = c.MaxMsgsPerSubject
		}
		if !s.WorkQueue {
			s.DuplicatesSeconds = 0
		}
	}
	return s
}

// Create creates the object; a refusal of the account's grant is ErrRetry: the grants part composes the keeper's grant
// to create a module's names from the same accepted manifests, and may not have yet.
func (b JetStream) Create(ctx context.Context, o Object) error {
	var err error
	switch o.Kind {
	case KindBucket:
		_, err = b.JS.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: o.Name, History: uint8(o.Settings.History),
			TTL: time.Duration(o.Settings.MaxAgeSeconds) * time.Second, MaxBytes: o.Settings.MaxBytes,
			MaxValueSize: int32(o.Settings.MaxMsgSize), Storage: jetstream.FileStorage, Replicas: 1,
			LimitMarkerTTL: limitMarker(o.Settings.KeyTTL)})
	case KindObjects:
		_, err = b.JS.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: o.Name, MaxBytes: o.Settings.MaxBytes,
			TTL: time.Duration(o.Settings.MaxAgeSeconds) * time.Second, Storage: jetstream.FileStorage, Replicas: 1})
	default:
		_, err = b.JS.CreateStream(ctx, streamConfig(o))
	}
	return retryable(err)
}

func limitMarker(on bool) time.Duration {
	if on {
		return time.Second
	}
	return 0
}

func streamConfig(o Object) jetstream.StreamConfig {
	c := jetstream.StreamConfig{Name: o.Name, Subjects: o.Subjects, Retention: jetstream.LimitsPolicy,
		MaxBytes: o.Settings.MaxBytes, MaxAge: time.Duration(o.Settings.MaxAgeSeconds) * time.Second,
		Discard: jetstream.DiscardOld, Storage: jetstream.FileStorage, Replicas: 1, MaxMsgs: -1, MaxMsgsPerSubject: -1}
	if o.Settings.Discard == DiscardNew {
		c.Discard = jetstream.DiscardNew
	}
	if o.Settings.WorkQueue {
		c.Retention = jetstream.WorkQueuePolicy
		c.Duplicates = time.Duration(o.Settings.DuplicatesSeconds) * time.Second
	}
	if o.Settings.MaxMsgSize > 0 {
		c.MaxMsgSize = int32(o.Settings.MaxMsgSize)
	}
	if o.Settings.History > 0 {
		c.MaxMsgsPerSubject = o.Settings.History
	}
	return c
}

// Update brings the held stream to the object's settings, keeping every other part of its configuration.
func (b JetStream) Update(ctx context.Context, o Object) error {
	s, err := b.JS.Stream(ctx, o.Stream())
	if err != nil {
		return retryable(err)
	}
	info, err := s.Info(ctx)
	if err != nil {
		return retryable(err)
	}
	c := info.Config
	c.MaxBytes = o.Settings.MaxBytes
	c.MaxAge = time.Duration(o.Settings.MaxAgeSeconds) * time.Second
	if o.Settings.History > 0 {
		c.MaxMsgsPerSubject = o.Settings.History
	}
	_, err = b.JS.UpdateStream(ctx, c)
	return retryable(err)
}

// Delete deletes the object's stream.
func (b JetStream) Delete(ctx context.Context, o Object) error {
	err := b.JS.DeleteStream(ctx, o.Stream())
	if errors.Is(err, jetstream.ErrStreamNotFound) {
		return nil
	}
	return retryable(err)
}

// Snapshot takes the streams' snapshot.
func (b JetStream) Snapshot(ctx context.Context, streams []string) (string, error) {
	if b.Snap == nil {
		return "", fmt.Errorf("no snapshot is taken here, so nothing is removed")
	}
	return b.Snap(ctx, streams)
}

// retryable marks a refusal of the account's permissions, or a bus that answers nothing, as ErrRetry.
func retryable(err error) error {
	if err == nil {
		return nil
	}
	words := strings.ToLower(err.Error())
	if strings.Contains(words, "permission") || strings.Contains(words, "authorization") || errors.Is(err, context.DeadlineExceeded) ||
		strings.Contains(words, "no responders") || strings.Contains(words, "timeout") {
		return fmt.Errorf("%w: %v", ErrRetry, err)
	}
	return err
}

// RestrictedMetadata is the key a restricted stream's metadata says it under (ADR 0363 §4): the grants part writes it.
const RestrictedMetadata = "mesh-restricted"

// Restricted reads the streams' metadata: true when any says it is restricted.
func (b JetStream) Restricted(ctx context.Context, streams []string) (bool, error) {
	for _, name := range streams {
		s, err := b.JS.Stream(ctx, name)
		if err != nil {
			return false, retryable(err)
		}
		info, err := s.Info(ctx)
		if err != nil {
			return false, retryable(err)
		}
		if _, ok := info.Config.Metadata[RestrictedMetadata]; ok {
			return true, nil
		}
	}
	return false, nil
}

// SnapshotStreams reads the streams a snapshot file holds from its manifest.
func (b JetStream) SnapshotStreams(path string) ([]string, error) {
	m, err := bussnap.ManifestOf(path)
	if err != nil {
		return nil, err
	}
	var out []string
	for _, s := range m.Streams {
		out = append(out, s.Name)
	}
	return out, nil
}

// Live says whether a stream of that name exists.
func (b JetStream) Live(ctx context.Context, name string) (bool, error) {
	return bussnap.Live(ctx, b.NC, name)
}

// Restore puts the snapshot's streams back under their own names.
func (b JetStream) Restore(ctx context.Context, path string) error {
	log := b.Log
	if log == nil {
		log = func(string, ...any) {}
	}
	return bussnap.RestoreFile(ctx, b.NC, path, b.Dir, log)
}

// DeleteStream removes the stream by its name.
func (b JetStream) DeleteStream(ctx context.Context, name string) error {
	return bussnap.DeleteStream(ctx, b.NC, name)
}