sxchnge/backend/shared/messaging/internal/natscap/admin.go

package natscap

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

	merrors "git.sendico.io/tech/sxchnge/shared/errors"
	"git.sendico.io/tech/sxchnge/shared/messaging/capability"
	"git.sendico.io/tech/sxchnge/shared/messaging/reliable"
	"github.com/nats-io/nats.go"
	"go.uber.org/zap"
)

func ensureDurableStream(ctx context.Context, logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings) error {
	if js == nil {
		return capability.Unavailable("durable admin")
	}
	expected := &nats.StreamConfig{
		Name:      settings.Stream.Name,
		Subjects:  append([]string(nil), settings.Stream.Subjects...),
		Retention: nats.LimitsPolicy,
		MaxAge:    settings.StreamMaxAge(),
		Replicas:  settings.Stream.Replicas,
	}
	if strings.EqualFold(strings.TrimSpace(settings.Stream.Storage), "memory") {
		expected.Storage = nats.MemoryStorage
	} else {
		expected.Storage = nats.FileStorage
	}

	info, err := js.StreamInfo(expected.Name, nats.Context(ctx))
	if err != nil {
		if errors.Is(err, nats.ErrStreamNotFound) {
			if !settings.AllowsAutoProvision() {
				if logger != nil {
					logger.Warn("Durable stream is missing and auto provisioning is disabled", zap.String("stream", expected.Name))
				}
				return err
			}
			_, err = js.AddStream(expected, nats.Context(ctx))
			return err
		}
		return err
	}
	current := info.Config
	if err := validateStreamConfig(current, *expected, settings.IsProductionValidation() || !settings.AllowsAutoProvision()); err != nil {
		if settings.AllowsAutoProvision() {
			reconciled, reconcileErr := reconcileDurableStream(ctx, logger, js, current, *expected, settings)
			if reconciled {
				return reconcileErr
			}
			if reconcileErr != nil {
				return reconcileErr
			}
		}
		return err
	}
	if settings.AllowsAutoProvision() && current.MaxAge != expected.MaxAge {
		current.MaxAge = expected.MaxAge
		_, err = js.UpdateStream(&current, nats.Context(ctx))
		return err
	}
	return nil
}

func reconcileDurableStream(ctx context.Context, logger *zap.Logger, js durableAdminContext, current nats.StreamConfig, expected nats.StreamConfig, settings reliable.DurableSettings) (bool, error) {
	if js == nil {
		return false, capability.Unavailable("durable admin")
	}
	if strings.TrimSpace(current.Name) != strings.TrimSpace(expected.Name) {
		return false, merrors.DataConflictf("durable stream name drift: got=%s want=%s", current.Name, expected.Name)
	}
	if current.Storage != expected.Storage {
		return false, merrors.DataConflictf("durable stream storage drift: got=%s want=%s", current.Storage.String(), expected.Storage.String())
	}
	if current.Retention != expected.Retention {
		return false, merrors.DataConflictf("durable stream retention drift: got=%s want=%s", current.Retention.String(), expected.Retention.String())
	}
	if current.Replicas != expected.Replicas {
		return false, merrors.DataConflictf("durable stream replicas drift: got=%d want=%d", current.Replicas, expected.Replicas)
	}
	missing, extra := stringSetDiff(expected.Subjects, current.Subjects), stringSetDiff(current.Subjects, expected.Subjects)
	if len(extra) > 0 {
		return false, merrors.DataConflict("durable stream has unexpected subjects")
	}

	if len(missing) == 0 {
		return false, nil
	}
	current.Subjects = append([]string(nil), expected.Subjects...)
	current.MaxAge = expected.MaxAge
	_, err := js.UpdateStream(&current, nats.Context(ctx))
	if err != nil {
		return true, err
	}
	return true, nil
}

func ensureDurableConsumerConfig(ctx context.Context, logger *zap.Logger, js durableAdminContext, settings reliable.DurableSettings, consumer reliable.DurableConsumerSettings) error {
	if js == nil {
		return capability.Unavailable("durable admin")
	}
	deliverPolicy, err := natsDeliverPolicy(consumer.DeliverPolicy)
	if err != nil {
		return err
	}
	expected := &nats.ConsumerConfig{
		Durable:       consumer.Durable,
		DeliverPolicy: deliverPolicy,
		AckPolicy:     nats.AckExplicitPolicy,
		AckWait:       time.Duration(consumer.AckWaitSeconds) * time.Second,
		MaxDeliver:    consumer.MaxDeliver,
		FilterSubject: consumer.Subject,
		ReplayPolicy:  nats.ReplayInstantPolicy,
	}
	if consumer.InactiveThresholdSeconds > 0 {
		expected.InactiveThreshold = time.Duration(consumer.InactiveThresholdSeconds) * time.Second
	}
	info, err := js.ConsumerInfo(settings.Stream.Name, consumer.Durable, nats.Context(ctx))
	if err != nil {
		if errors.Is(err, nats.ErrConsumerNotFound) {
			if !settings.AllowsAutoProvision() {
				if logger != nil {
					logger.Warn("Durable consumer is missing and auto provisioning is disabled", zap.String("stream", settings.Stream.Name), zap.String("subject", consumer.Subject), zap.String("durable", consumer.Durable))
				}
				return err
			}
			_, err = js.AddConsumer(settings.Stream.Name, expected, nats.Context(ctx))
			return err
		}
		return err
	}
	current := info.Config
	if current.DeliverPolicy != expected.DeliverPolicy && logger != nil {
		logger.Warn("Durable consumer deliver policy drift requires delete and recreate",
			zap.String("stream", settings.Stream.Name),
			zap.String("subject", consumer.Subject),
			zap.String("durable", consumer.Durable),
			zap.String("current_deliver_policy", natsDeliverPolicyName(current.DeliverPolicy)),
			zap.String("expected_deliver_policy", natsDeliverPolicyName(expected.DeliverPolicy)),
			zap.Bool("recreate_required", true),
		)
	}
	if err := validateConsumerConfig(current, *expected, settings.IsProductionValidation() || !settings.AllowsAutoProvision()); err != nil {
		return err
	}
	if settings.AllowsAutoProvision() && (current.AckWait != expected.AckWait || current.MaxDeliver != expected.MaxDeliver || current.InactiveThreshold != expected.InactiveThreshold) {
		current.AckWait = expected.AckWait
		current.MaxDeliver = expected.MaxDeliver
		current.InactiveThreshold = expected.InactiveThreshold
		_, err = js.UpdateConsumer(settings.Stream.Name, &current, nats.Context(ctx))
		return err
	}
	return nil
}

func natsDeliverPolicy(deliveryMode reliable.DurableDeliverPolicy) (nats.DeliverPolicy, error) {
	switch deliveryMode {
	case "", reliable.DurableDeliverAll:
		return nats.DeliverAllPolicy, nil
	case reliable.DurableDeliverNew:
		return nats.DeliverNewPolicy, nil
	default:
		return nats.DeliverAllPolicy, merrors.InvalidArgumentf("unsupported durable consumer deliver policy: %q", deliveryMode)
	}
}

func natsDeliverPolicyName(deliveryMode nats.DeliverPolicy) string {
	switch deliveryMode {
	case nats.DeliverAllPolicy:
		return "all"
	case nats.DeliverLastPolicy:
		return "last"
	case nats.DeliverNewPolicy:
		return "new"
	case nats.DeliverByStartSequencePolicy:
		return "by_start_sequence"
	case nats.DeliverByStartTimePolicy:
		return "by_start_time"
	case nats.DeliverLastPerSubjectPolicy:
		return "last_per_subject"
	default:
		return "unknown"
	}
}

func validateStreamConfig(current nats.StreamConfig, expected nats.StreamConfig, strict bool) error {
	if strings.TrimSpace(current.Name) != strings.TrimSpace(expected.Name) {
		return merrors.DataConflictf("durable stream name drift: got=%s want=%s", current.Name, expected.Name)
	}
	if !sameStrings(current.Subjects, expected.Subjects) {
		return merrors.DataConflictf("durable stream subjects drift: current=%v expected=%v", current.Subjects, expected.Subjects)
	}
	if current.Storage != expected.Storage {
		return merrors.DataConflict("durable stream storage drift")
	}
	if current.Retention != expected.Retention {
		return merrors.DataConflict("durable stream retention drift")
	}
	if current.Replicas != expected.Replicas {
		return merrors.DataConflict("durable stream replicas drift")
	}
	if strict && current.MaxAge != expected.MaxAge {
		return merrors.DataConflict("durable stream retention age drift")
	}
	return nil
}

func validateConsumerConfig(current nats.ConsumerConfig, expected nats.ConsumerConfig, strict bool) error {
	if strings.TrimSpace(current.Durable) != strings.TrimSpace(expected.Durable) {
		return merrors.DataConflict("durable consumer name drift")
	}
	if strings.TrimSpace(current.FilterSubject) != strings.TrimSpace(expected.FilterSubject) {
		return merrors.DataConflict("durable consumer filter subject drift")
	}
	if current.DeliverPolicy != expected.DeliverPolicy {
		return merrors.DataConflict("durable consumer deliver policy drift; delete and recreate consumer to apply deliver policy")
	}
	if current.AckPolicy != expected.AckPolicy {
		return merrors.DataConflict("durable consumer ack policy drift")
	}
	if strict && current.AckWait != expected.AckWait {
		return merrors.DataConflict("durable consumer ack wait drift")
	}
	if strict && current.MaxDeliver != expected.MaxDeliver {
		return merrors.DataConflict("durable consumer max deliver drift")
	}
	if strict && current.InactiveThreshold != expected.InactiveThreshold {
		return merrors.DataConflict("durable consumer inactive threshold drift")
	}
	return nil
}

func sameStrings(left, right []string) bool {
	if len(left) != len(right) {
		return false
	}
	seen := map[string]int{}
	for _, value := range left {
		seen[strings.TrimSpace(value)]++
	}
	for _, value := range right {
		key := strings.TrimSpace(value)
		if seen[key] == 0 {
			return false
		}
		seen[key]--
	}
	for _, count := range seen {
		if count != 0 {
			return false
		}
	}
	return true
}

func stringSetDiff(left, right []string) []string {
	if len(left) == 0 {
		return nil
	}
	seen := map[string]int{}
	for _, value := range right {
		seen[strings.TrimSpace(value)]++
	}
	result := make([]string, 0)
	for _, value := range left {
		key := strings.TrimSpace(value)
		if key == "" {
			continue
		}
		if seen[key] > 0 {
			seen[key]--
			continue
		}
		result = append(result, key)
	}
	if len(result) == 0 {
		return nil
	}
	return result
}