natsnet/dotnet/src/ZB.MOM.NatsNet.Server/Internal/DataStructures/StreamDeletionMeta.cs

// Copyright 2025 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/sdm.go in the NATS server Go source.

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

/// <summary>
/// Per-sequence data tracked by <see cref="StreamDeletionMeta"/>.
/// Mirrors <c>SDMBySeq</c> in server/sdm.go.
/// </summary>
public readonly struct SdmBySeq
{
    /// <summary>Whether this sequence was the last message for its subject.</summary>
    public bool Last { get; init; }

    /// <summary>Timestamp (nanoseconds UTC) when the removal/SDM was last proposed.</summary>
    public long Ts { get; init; }
}

/// <summary>
/// Tracks pending subject delete markers (SDMs) and message removals for a stream.
/// Used by JetStream cluster consensus to avoid redundant proposals.
/// Mirrors <c>SDMMeta</c> in server/sdm.go.
/// </summary>
public sealed class StreamDeletionMeta
{
    // Per-subject pending-count totals.
    private readonly Dictionary<string, ulong> _totals = new(1);

    // Per-sequence data keyed by sequence number.
    private readonly Dictionary<ulong, SdmBySeq> _pending = new(1);

    // -------------------------------------------------------------------------
    // Header constants (forward-declared; populated in session 19 — JetStream).
    // isSubjectDeleteMarker checks these header keys.
    // -------------------------------------------------------------------------

    // Mirrors JSMarkerReason header key (defined in jetstream.go).
    internal const string HeaderJsMarkerReason = "Nats-Marker-Reason";

    // Mirrors KVOperation header key (defined in jetstream.go).
    internal const string HeaderKvOperation = "KV-Operation";

    // Mirrors KVOperationValuePurge (defined in jetstream.go).
    internal const string KvOperationValuePurge = "PURGE";

    /// <summary>
    /// Returns true when the given header block contains a subject delete marker
    /// (either a JetStream marker or a KV purge operation).
    /// Mirrors <c>isSubjectDeleteMarker</c> in server/sdm.go.
    /// </summary>
    public static bool IsSubjectDeleteMarker(ReadOnlySpan<byte> hdr)
    {
        // Simplified header scan: checks whether JSMarkerReason key is present
        // or whether KV-Operation equals "PURGE".
        // Full implementation depends on SliceHeader from session 08 (client.go).
        // Until then this provides the correct contract.
        var text = System.Text.Encoding.UTF8.GetString(hdr);
        if (text.Contains(HeaderJsMarkerReason))
            return true;
        if (text.Contains($"{HeaderKvOperation}: {KvOperationValuePurge}"))
            return true;
        return false;
    }

    /// <summary>
    /// Tries to get the pending entry for <paramref name="seq"/>.
    /// </summary>
    public bool TryGetPending(ulong seq, out SdmBySeq entry) => _pending.TryGetValue(seq, out entry);

    /// <summary>
    /// Sets the pending entry for <paramref name="seq"/>.
    /// </summary>
    public void SetPending(ulong seq, SdmBySeq entry) => _pending[seq] = entry;

    /// <summary>
    /// Returns the pending count for <paramref name="subj"/>, or 0 if not tracked.
    /// </summary>
    public ulong GetSubjectTotal(string subj) => _totals.TryGetValue(subj, out var cnt) ? cnt : 0;

    /// <summary>
    /// Clears all tracked data.
    /// Mirrors <c>SDMMeta.empty</c>.
    /// </summary>
    public void Empty()
    {
        _totals.Clear();
        _pending.Clear();
    }

    /// <summary>
    /// Tracks <paramref name="seq"/> as pending and returns whether it was
    /// the last message for its subject.  If the sequence is already tracked
    /// the existing <c>Last</c> value is returned without modification.
    /// Mirrors <c>SDMMeta.trackPending</c>.
    /// </summary>
    public bool TrackPending(ulong seq, string subj, bool last)
    {
        if (_pending.TryGetValue(seq, out var p))
            return p.Last;

        _pending[seq] = new SdmBySeq { Last = last, Ts = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L };
        _totals[subj] = _totals.TryGetValue(subj, out var cnt) ? cnt + 1 : 1;
        return last;
    }

    /// <summary>
    /// Removes <paramref name="seq"/> and decrements the pending count for
    /// <paramref name="subj"/>, deleting the subject entry when it reaches zero.
    /// Mirrors <c>SDMMeta.removeSeqAndSubject</c>.
    /// </summary>
    public void RemoveSeqAndSubject(ulong seq, string subj)
    {
        if (!_pending.Remove(seq))
            return;

        if (_totals.TryGetValue(subj, out var msgs))
        {
            if (msgs <= 1)
                _totals.Remove(subj);
            else
                _totals[subj] = msgs - 1;
        }
    }
}