natsnet/dotnet/src/ZB.MOM.NatsNet.Server/JetStream/NatsConsumer.Config.cs

using ZB.MOM.NatsNet.Server.Internal.DataStructures;

namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsConsumer
{
    internal const int DefaultMaxAckPending = 1000;
    internal static readonly TimeSpan DefaultAckWait = TimeSpan.FromSeconds(30);
    internal static readonly TimeSpan DefaultDeleteWait = TimeSpan.FromSeconds(5);
    internal static readonly TimeSpan DefaultPinnedTtl = TimeSpan.FromMinutes(2);

    internal static JsApiError? SetConsumerConfigDefaults(
        ConsumerConfig config,
        StreamConfig streamConfig,
        JetStreamAccountLimits? selectedLimits,
        bool pedantic)
    {
        ArgumentNullException.ThrowIfNull(config);
        ArgumentNullException.ThrowIfNull(streamConfig);
        var streamReplicas = Math.Max(1, streamConfig.Replicas);

        if (config.MaxDeliver is 0 or < -1)
        {
            if (pedantic && config.MaxDeliver < -1)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("max_deliver must be set to -1"));
            config.MaxDeliver = -1;
        }

        if (config.MaxWaiting < 0)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("max_waiting must not be negative"));
            config.MaxWaiting = 0;
        }

        if (config.MaxAckPending < -1)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("max_ack_pending must be set to -1"));
            config.MaxAckPending = -1;
        }

        if (config.MaxRequestBatch < 0)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("max_batch must not be negative"));
            config.MaxRequestBatch = 0;
        }

        if (config.MaxRequestExpires < TimeSpan.Zero)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("max_expires must not be negative"));
            config.MaxRequestExpires = TimeSpan.Zero;
        }

        if (config.MaxRequestMaxBytes < 0)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("max_bytes must not be negative"));
            config.MaxRequestMaxBytes = 0;
        }

        if (config.Heartbeat < TimeSpan.Zero)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("idle_heartbeat must not be negative"));
            config.Heartbeat = TimeSpan.Zero;
        }

        if (config.InactiveThreshold < TimeSpan.Zero)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("inactive_threshold must not be negative"));
            config.InactiveThreshold = TimeSpan.Zero;
        }

        if (config.PinnedTTL < TimeSpan.Zero)
        {
            if (pedantic)
                return JsApiErrors.NewJSPedanticError(new InvalidOperationException("priority_timeout must not be negative"));
            config.PinnedTTL = TimeSpan.Zero;
        }

        if (config.AckWait == TimeSpan.Zero)
            config.AckWait = DefaultAckWait;

        if (config.MaxAckPending == 0 && config.AckPolicy != AckPolicy.AckNone)
        {
            config.MaxAckPending = selectedLimits?.MaxAckPending > 0
                ? selectedLimits.MaxAckPending
                : DefaultMaxAckPending;
        }

        if (config.InactiveThreshold == TimeSpan.Zero && string.IsNullOrWhiteSpace(config.Durable))
            config.InactiveThreshold = DefaultDeleteWait;

        if (config.PinnedTTL == TimeSpan.Zero && config.PriorityPolicy == PriorityPolicy.PriorityPinnedClient)
            config.PinnedTTL = DefaultPinnedTtl;

        if (config.Replicas == 0 || config.Replicas > streamReplicas)
            config.Replicas = streamReplicas;

        if (!string.IsNullOrWhiteSpace(config.Name) && string.IsNullOrWhiteSpace(config.Durable))
            config.Durable = config.Name;

        return null;
    }

    internal static JsApiError? CheckConsumerCfg(
        ConsumerConfig config,
        StreamConfig streamConfig,
        JetStreamAccountLimits? selectedLimits,
        bool isRecovering)
    {
        ArgumentNullException.ThrowIfNull(config);
        ArgumentNullException.ThrowIfNull(streamConfig);
        var streamReplicas = Math.Max(1, streamConfig.Replicas);

        if (!string.IsNullOrWhiteSpace(config.Durable) &&
            !string.IsNullOrWhiteSpace(config.Name) &&
            !string.Equals(config.Durable, config.Name, StringComparison.Ordinal))
        {
            return JsApiErrors.NewJSConsumerCreateDurableAndNameMismatchError();
        }

        if (HasPathSeparators(config.Durable) || HasPathSeparators(config.Name))
            return JsApiErrors.NewJSConsumerNameContainsPathSeparatorsError();

        if (config.Replicas > streamReplicas)
            return JsApiErrors.NewJSConsumerReplicasExceedsStreamError();

        if (!Enum.IsDefined(config.AckPolicy))
            return JsApiErrors.NewJSConsumerAckPolicyInvalidError();

        if (!Enum.IsDefined(config.ReplayPolicy))
            return JsApiErrors.NewJSConsumerReplayPolicyInvalidError();

        if (!Enum.IsDefined(config.DeliverPolicy))
            return JsApiErrors.NewJSConsumerInvalidPolicyError(new InvalidOperationException("deliver policy invalid"));

        if (config.FilterSubjects is { Length: > 0 } && !string.IsNullOrWhiteSpace(config.FilterSubject))
            return JsApiErrors.NewJSConsumerDuplicateFilterSubjectsError();

        var filters = config.FilterSubjects is { Length: > 0 }
            ? SubjectTokens.Subjects(config.FilterSubjects)
            : (string.IsNullOrWhiteSpace(config.FilterSubject) ? [] : [config.FilterSubject]);

        for (var i = 0; i < filters.Length; i++)
        {
            if (string.IsNullOrWhiteSpace(filters[i]))
                return JsApiErrors.NewJSConsumerEmptyFilterError();
            if (!SubscriptionIndex.IsValidSubject(filters[i]))
                return JsApiErrors.NewJSConsumerFilterNotSubsetError();

            for (var j = i + 1; j < filters.Length; j++)
            {
                if (SubscriptionIndex.SubjectsCollide(filters[i], filters[j]))
                    return JsApiErrors.NewJSConsumerOverlappingSubjectFiltersError();
            }
        }

        var isPush = !string.IsNullOrWhiteSpace(config.DeliverSubject);
        if (isPush)
        {
            if (!SubscriptionIndex.IsValidSubject(config.DeliverSubject!))
                return JsApiErrors.NewJSConsumerInvalidDeliverSubjectError();

            if (SubscriptionIndex.SubjectHasWildcard(config.DeliverSubject!))
                return JsApiErrors.NewJSConsumerDeliverToWildcardsError();

            if (config.MaxWaiting > 0)
                return JsApiErrors.NewJSConsumerPushMaxWaitingError();
        }
        else
        {
            if (config.RateLimit > 0)
                return JsApiErrors.NewJSConsumerPullWithRateLimitError();
        }

        if (config.MaxAckPending > 0 && selectedLimits?.MaxAckPending > 0 && config.MaxAckPending > selectedLimits.MaxAckPending)
            return JsApiErrors.NewJSConsumerMaxPendingAckExcessError(selectedLimits.MaxAckPending);

        if (streamConfig.Retention == RetentionPolicy.WorkQueuePolicy && config.AckPolicy != AckPolicy.AckExplicit)
            return JsApiErrors.NewJSConsumerWQRequiresExplicitAckError();

        if (config.Direct)
        {
            if (isPush)
                return JsApiErrors.NewJSConsumerDirectRequiresPushError();
            if (!string.IsNullOrWhiteSpace(config.Durable))
                return JsApiErrors.NewJSConsumerDirectRequiresEphemeralError();
        }

        _ = isRecovering;
        return null;
    }

    internal void UpdateInactiveThreshold(ConsumerConfig config)
    {
        ArgumentNullException.ThrowIfNull(config);

        _mu.EnterWriteLock();
        try
        {
            _deleteThreshold = config.InactiveThreshold > TimeSpan.Zero
                ? config.InactiveThreshold
                : DefaultDeleteWait;

            Config.InactiveThreshold = _deleteThreshold;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void UpdatePauseState(ConsumerConfig config, DateTime? nowUtc = null)
    {
        ArgumentNullException.ThrowIfNull(config);
        var now = nowUtc ?? DateTime.UtcNow;

        _mu.EnterWriteLock();
        try
        {
            Config.PauseUntil = config.PauseUntil;
            _isPaused = config.PauseUntil.HasValue && config.PauseUntil.Value > now;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal ConsumerAssignment? ConsumerAssignment()
    {
        _mu.EnterReadLock();
        try
        {
            return _assignment;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void SetConsumerAssignment(ConsumerAssignment? assignment)
    {
        _mu.EnterWriteLock();
        try
        {
            _assignment = assignment;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    private static bool HasPathSeparators(string? value)
    {
        if (string.IsNullOrWhiteSpace(value))
            return false;

        return value.Contains('/') || value.Contains('\\');
    }
}