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

package keeper

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

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

// 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)
}

// 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,
			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
}