284 lines
9.5 KiB
Go
284 lines
9.5 KiB
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(¤t, 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(¤t, 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, ¤t, 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
|
|
}
|