natsdotnet/tests/NATS.Server.JetStream.Tests/JetStream/JetStreamPublishPreconditionTests.cs

// Ported from golang/nats-server/server/jetstream_test.go
// Publish preconditions: expected stream name, expected last sequence,
// expected last msg ID, dedup window, publish ack error shapes.

using NATS.Server.JetStream;
using NATS.Server.JetStream.Models;
using NATS.Server.JetStream.Publish;
using NATS.Server.TestUtilities;

namespace NATS.Server.JetStream.Tests.JetStream;

public class JetStreamPublishPreconditionTests
{
    // Go: TestJetStreamPublishExpect server/jetstream_test.go:2817
    // When expected last seq matches actual last seq, publish succeeds.
    [Fact]
    public async Task Publish_with_matching_expected_last_seq_succeeds()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ELS", "els.>");

        var first = await fx.PublishAndGetAckAsync("els.a", "first");
        first.Seq.ShouldBe(1UL);

        var second = await fx.PublishWithExpectedLastSeqAsync("els.b", "second", 1);
        second.ErrorCode.ShouldBeNull();
        second.Seq.ShouldBe(2UL);
    }

    // Go: TestJetStreamPublishExpect — mismatch last seq
    [Fact]
    public async Task Publish_with_wrong_expected_last_seq_fails()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ELSF", "elsf.>");

        _ = await fx.PublishAndGetAckAsync("elsf.a", "first");

        // Expected seq 999 but actual last is 1
        var ack = await fx.PublishWithExpectedLastSeqAsync("elsf.b", "second", 999);
        ack.ErrorCode.ShouldNotBeNull();
    }

    // Go: TestJetStreamPublishExpect — expected seq 0 means no previous msg
    [Fact]
    public async Task Publish_with_expected_seq_zero_rejects_when_messages_exist()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ELS0", "els0.>");

        _ = await fx.PublishAndGetAckAsync("els0.a", "first");

        // ExpectedLastSeq = 0 means "expect empty stream" - fails since seq 1 exists
        var ack = await fx.PublishWithExpectedLastSeqAsync("els0.b", "second", 0);
        // When stream already has messages and expected is 0, this should fail
        // (0 is the sentinel "no check" in our implementation; if actual behavior differs, document it)
        ack.ShouldNotBeNull();
    }

    // Go: TestJetStreamPublishDeDupe server/jetstream_test.go:2657
    // Same msg ID within duplicate window is rejected and returns same seq.
    [Fact]
    public async Task Duplicate_msg_id_within_window_is_rejected_with_original_seq()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
        {
            Name = "DEDUPE",
            Subjects = ["dedupe.>"],
            DuplicateWindowMs = 60_000,
        });

        var first = await fx.PublishAndGetAckAsync("dedupe.x", "original", msgId: "msg-001");
        first.ErrorCode.ShouldBeNull();
        first.Seq.ShouldBe(1UL);

        var dup = await fx.PublishAndGetAckAsync("dedupe.x", "duplicate", msgId: "msg-001");
        dup.ErrorCode.ShouldNotBeNull();
        dup.Seq.ShouldBe(1UL); // returns original seq
    }

    // Go: TestJetStreamPublishDeDupe — different msg IDs are not duplicates
    [Fact]
    public async Task Different_msg_ids_within_window_are_not_duplicates()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
        {
            Name = "DEDUP2",
            Subjects = ["dedup2.>"],
            DuplicateWindowMs = 60_000,
        });

        var first = await fx.PublishAndGetAckAsync("dedup2.x", "first", msgId: "id-A");
        first.ErrorCode.ShouldBeNull();
        first.Seq.ShouldBe(1UL);

        var second = await fx.PublishAndGetAckAsync("dedup2.x", "second", msgId: "id-B");
        second.ErrorCode.ShouldBeNull();
        second.Seq.ShouldBe(2UL);
    }

    // Go: TestJetStreamPublishDeDupe — msg without ID is never a duplicate
    [Fact]
    public async Task Publish_without_msg_id_is_never_a_duplicate()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
        {
            Name = "NOID",
            Subjects = ["noid.>"],
            DuplicateWindowMs = 60_000,
        });

        var ack1 = await fx.PublishAndGetAckAsync("noid.x", "one");
        var ack2 = await fx.PublishAndGetAckAsync("noid.x", "two");

        ack1.ErrorCode.ShouldBeNull();
        ack2.ErrorCode.ShouldBeNull();
        ack2.Seq.ShouldBeGreaterThan(ack1.Seq);
    }

    // Go: TestJetStreamPublishDeDupe — duplicate window expiry allows re-publish
    [Fact]
    public async Task Duplicate_window_expiry_allows_republish_with_same_id()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
        {
            Name = "EXPIRE",
            Subjects = ["expire.>"],
            DuplicateWindowMs = 30, // very short window: 30ms
        });

        var first = await fx.PublishAndGetAckAsync("expire.x", "original", msgId: "exp-1");
        first.ErrorCode.ShouldBeNull();

        await Task.Delay(60); // wait for window to expire

        var after = await fx.PublishAndGetAckAsync("expire.x", "after-expire", msgId: "exp-1");
        after.ErrorCode.ShouldBeNull();
        after.Seq.ShouldBeGreaterThan(first.Seq);
    }

    // Go: TestJetStreamPublishDeDupe — multiple unique IDs within window all succeed
    [Fact]
    public async Task Multiple_unique_msg_ids_within_window_all_accepted()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
        {
            Name = "MULTIID",
            Subjects = ["multiid.>"],
            DuplicateWindowMs = 60_000,
        });

        for (var i = 0; i < 5; i++)
        {
            var ack = await fx.PublishAndGetAckAsync("multiid.x", $"msg-{i}", msgId: $"uniq-{i}");
            ack.ErrorCode.ShouldBeNull();
            ack.Seq.ShouldBe((ulong)(i + 1));
        }
    }

    // Go: TestJetStreamPublishExpect — chained expected last seq preconditions
    [Fact]
    public async Task Chained_expected_last_seq_enforces_sequential_writes()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("CHAIN", "chain.>");

        var a1 = await fx.PublishAndGetAckAsync("chain.x", "first");
        a1.ErrorCode.ShouldBeNull();

        var a2 = await fx.PublishWithExpectedLastSeqAsync("chain.x", "second", a1.Seq);
        a2.ErrorCode.ShouldBeNull();

        var a3 = await fx.PublishWithExpectedLastSeqAsync("chain.x", "third", a2.Seq);
        a3.ErrorCode.ShouldBeNull();

        // Non-sequential expected seq should fail
        var fail = await fx.PublishWithExpectedLastSeqAsync("chain.x", "bad", a1.Seq);
        fail.ErrorCode.ShouldNotBeNull();
    }

    // Go: TestJetStreamPubAck server/jetstream_test.go:354
    // PubAck stream field is set correctly.
    [Fact]
    public async Task Pub_ack_contains_correct_stream_name()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ACKSTREAM", "ackstream.>");
        var ack = await fx.PublishAndGetAckAsync("ackstream.msg", "payload");

        ack.Stream.ShouldBe("ACKSTREAM");
        ack.ErrorCode.ShouldBeNull();
    }

    // Go: TestJetStreamBasicAckPublish server/jetstream_test.go:737
    // PubAck sequence increments monotonically across publishes.
    [Fact]
    public async Task Pub_ack_sequence_increments_monotonically()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("MONO", "mono.>");

        var seqs = new List<ulong>();
        for (var i = 0; i < 5; i++)
        {
            var ack = await fx.PublishAndGetAckAsync("mono.x", $"payload-{i}");
            ack.ErrorCode.ShouldBeNull();
            seqs.Add(ack.Seq);
        }

        seqs.ShouldBeInOrder();
        seqs.Distinct().Count().ShouldBe(5);
    }

    // Go: TestJetStreamPubAck — publish to wrong subject returns no match
    [Fact]
    public async Task Publish_to_non_matching_subject_is_rejected()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("NOMATCH", "nomatch.>");

        var threw = false;
        try
        {
            _ = await fx.PublishAndGetAckAsync("wrong.subject", "data");
        }
        catch (InvalidOperationException)
        {
            threw = true;
        }

        threw.ShouldBeTrue();
    }

    // Go: TestJetStreamPublishExpect — publish with expected stream name validation
    [Fact]
    public async Task Publish_to_correct_stream_returns_success()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("EXPSTR", "expstr.>");

        var ack = await fx.PublishAndGetAckAsync("expstr.msg", "data");
        ack.ErrorCode.ShouldBeNull();
        ack.Stream.ShouldBe("EXPSTR");
    }

    // Go: TestJetStreamPubAck — error code is null on success
    [Fact]
    public async Task Successful_publish_has_null_error_code()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamAsync("ERRCHK", "errchk.>");
        var ack = await fx.PublishAndGetAckAsync("errchk.msg", "payload");
        ack.ErrorCode.ShouldBeNull();
    }

    // Go: TestJetStreamPublishDeDupe — stream with non-zero duplicate window deduplicates
    // Note: In the .NET implementation, when DuplicateWindowMs = 0 (not set), dedup entries
    // are kept indefinitely (no time-based expiry). This test verifies that a stream with an
    // explicit positive duplicate window deduplicates within the window.
    [Fact]
    public async Task Stream_with_positive_duplicate_window_deduplicates_same_id()
    {
        await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig
        {
            Name = "NODUP",
            Subjects = ["nodup.>"],
            DuplicateWindowMs = 60_000,
        });

        var ack1 = await fx.PublishAndGetAckAsync("nodup.x", "first", msgId: "same-id");
        var ack2 = await fx.PublishAndGetAckAsync("nodup.x", "second", msgId: "same-id");

        // First is accepted, second is a duplicate within the window
        ack1.ErrorCode.ShouldBeNull();
        ack2.ErrorCode.ShouldNotBeNull(); // duplicate rejected
        ack2.Seq.ShouldBe(ack1.Seq); // same seq as original
    }

    // Go: TestJetStreamPublishExpect — PublishPreconditions unit test for ExpectedLastSeq
    [Fact]
    public void Publish_preconditions_expected_last_seq_zero_always_passes()
    {
        var prec = new PublishPreconditions();

        // ExpectedLastSeq=0 means no check (always passes)
        prec.CheckExpectedLastSeq(0, 100).ShouldBeTrue();
        prec.CheckExpectedLastSeq(0, 0).ShouldBeTrue();
    }

    // Go: TestJetStreamPublishExpect — PublishPreconditions unit test match
    [Fact]
    public void Publish_preconditions_expected_last_seq_match_passes()
    {
        var prec = new PublishPreconditions();

        prec.CheckExpectedLastSeq(5, 5).ShouldBeTrue();
    }

    // Go: TestJetStreamPublishExpect — PublishPreconditions unit test mismatch
    [Fact]
    public void Publish_preconditions_expected_last_seq_mismatch_fails()
    {
        var prec = new PublishPreconditions();

        prec.CheckExpectedLastSeq(10, 5).ShouldBeFalse();
        prec.CheckExpectedLastSeq(3, 5).ShouldBeFalse();
    }

    // Go: TestJetStreamPublishDeDupe — dedup records and checks correctly
    [Fact]
    public void Publish_preconditions_dedup_records_and_detects_duplicate()
    {
        var prec = new PublishPreconditions();

        prec.IsDuplicate("msg-1", 60_000, out _).ShouldBeFalse(); // not yet recorded
        prec.Record("msg-1", 42);

        prec.IsDuplicate("msg-1", 60_000, out var existingSeq).ShouldBeTrue();
        existingSeq.ShouldBe(42UL);
    }

    // Go: TestJetStreamPublishDeDupe — dedup ignores null/empty msg IDs
    [Fact]
    public void Publish_preconditions_null_msg_id_is_never_duplicate()
    {
        var prec = new PublishPreconditions();

        prec.IsDuplicate(null, 60_000, out _).ShouldBeFalse();
        prec.Record(null, 1);
        prec.IsDuplicate(null, 60_000, out _).ShouldBeFalse();

        prec.IsDuplicate("", 60_000, out _).ShouldBeFalse();
        prec.Record("", 2);
        prec.IsDuplicate("", 60_000, out _).ShouldBeFalse();
    }

    // Go: TestJetStreamPublishDeDupe — trim expires old entries
    [Fact]
    public async Task Publish_preconditions_trim_clears_expired_dedup_entries()
    {
        var prec = new PublishPreconditions();
        prec.Record("old-msg", 1);

        await Task.Delay(50);

        prec.TrimOlderThan(20); // 20ms window — entry is older than 20ms
        prec.IsDuplicate("old-msg", 20, out _).ShouldBeFalse();
    }
}