natsdotnet/tests/NATS.Server.JetStream.Tests/JetStream/Storage/FileStorePurgeTests.cs

// Reference: golang/nats-server/server/filestore_test.go
// Tests ported from: TestFileStorePurge, TestFileStoreCompact,
//                    TestFileStoreCompactLastPlusOne, TestFileStoreCompactMsgCountBug,
//                    TestFileStorePurgeExWithSubject, TestFileStorePurgeExKeepOneBug,
//                    TestFileStorePurgeExNoTombsOnBlockRemoval,
//                    TestFileStoreStreamTruncate

using System.Text;
using NATS.Server.JetStream.Storage;

namespace NATS.Server.JetStream.Tests.JetStream.Storage;

public sealed class FileStorePurgeTests : IDisposable
{
    private readonly string _dir;

    public FileStorePurgeTests()
    {
        _dir = Path.Combine(Path.GetTempPath(), $"nats-js-fs-purge-{Guid.NewGuid():N}");
        Directory.CreateDirectory(_dir);
    }

    public void Dispose()
    {
        if (Directory.Exists(_dir))
            Directory.Delete(_dir, recursive: true);
    }

    private FileStore CreateStore(string subdirectory, FileStoreOptions? options = null)
    {
        var dir = Path.Combine(_dir, subdirectory);
        var opts = options ?? new FileStoreOptions();
        opts.Directory = dir;
        return new FileStore(opts);
    }

    // Go: TestFileStorePurge server/filestore_test.go:709
    [Fact]
    public async Task Purge_removes_all_messages()
    {
        await using var store = CreateStore("purge-all");

        for (var i = 0; i < 100; i++)
            await store.AppendAsync("foo", new byte[128], default);

        (await store.GetStateAsync(default)).Messages.ShouldBe((ulong)100);

        await store.PurgeAsync(default);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)0);
        state.Bytes.ShouldBe((ulong)0);
    }

    // Go: TestFileStorePurge server/filestore_test.go:740
    [Fact]
    public async Task Purge_recovers_same_state_after_restart()
    {
        var subDir = "purge-restart";

        await using (var store = CreateStore(subDir))
        {
            for (var i = 0; i < 50; i++)
                await store.AppendAsync("foo", "Hello"u8.ToArray(), default);

            await store.PurgeAsync(default);
        }

        await using (var store = CreateStore(subDir))
        {
            var state = await store.GetStateAsync(default);
            state.Messages.ShouldBe((ulong)0);
            state.Bytes.ShouldBe((ulong)0);
        }
    }

    // Go: TestFileStorePurge server/filestore_test.go:776
    [Fact]
    public async Task Store_after_purge_works()
    {
        await using var store = CreateStore("purge-then-store");

        for (var i = 0; i < 20; i++)
            await store.AppendAsync("foo", "Hello"u8.ToArray(), default);

        await store.PurgeAsync(default);

        // New messages after purge.
        for (var i = 0; i < 10; i++)
        {
            var seq = await store.AppendAsync("foo", "After purge"u8.ToArray(), default);
            seq.ShouldBeGreaterThan((ulong)0);
        }

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)10);
    }

    // Go: TestFileStoreCompact server/filestore_test.go:822
    [Fact]
    public async Task Compact_removes_messages_below_sequence()
    {
        await using var store = CreateStore("compact-below-seq");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        // Compact removes all messages with seq < 5, leaving seqs 5-10 (6 messages).
        var removed = store.Compact(5);
        removed.ShouldBe((ulong)4);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)6);
        state.FirstSeq.ShouldBe((ulong)5);
        state.LastSeq.ShouldBe((ulong)10);

        // Seqs 1-4 must be gone.
        for (ulong seq = 1; seq <= 4; seq++)
            (await store.LoadAsync(seq, default)).ShouldBeNull();

        // Seqs 5-10 must still be present.
        for (ulong seq = 5; seq <= 10; seq++)
            (await store.LoadAsync(seq, default)).ShouldNotBeNull();
    }

    // Go: TestFileStoreCompact server/filestore_test.go:851
    [Fact]
    public async Task Compact_beyond_last_seq_resets_first()
    {
        await using var store = CreateStore("compact-beyond-last");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        // Compact at seq 100 (beyond last seq 10) removes all messages.
        var removed = store.Compact(100);
        removed.ShouldBe((ulong)10);

        var apiState = await store.GetStateAsync(default);
        apiState.Messages.ShouldBe((ulong)0);

        // FastState / State() should report _first watermark = 100.
        var state = store.State();
        state.Msgs.ShouldBe((ulong)0);
        state.FirstSeq.ShouldBe((ulong)100);
    }

    // Go: TestFileStoreCompact server/filestore_test.go:862
    [Fact]
    public async Task Compact_recovers_after_restart()
    {
        var subDir = "compact-restart";

        await using (var store = CreateStore(subDir))
        {
            for (var i = 0; i < 10; i++)
                await store.AppendAsync("foo", "data"u8.ToArray(), default);

            store.Compact(5);
        }

        // Reopen the same directory and verify state is preserved.
        await using (var store = CreateStore(subDir))
        {
            var state = await store.GetStateAsync(default);
            state.Messages.ShouldBe((ulong)6);
            state.FirstSeq.ShouldBe((ulong)5);
            state.LastSeq.ShouldBe((ulong)10);
        }
    }

    // Go: TestFileStoreCompactLastPlusOne server/filestore_test.go:875
    [Fact]
    public async Task Compact_last_plus_one_clears_all()
    {
        await using var store = CreateStore("compact-last-plus-one");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        var lastSeq = (await store.GetStateAsync(default)).LastSeq;
        lastSeq.ShouldBe((ulong)10);

        // Compact at lastSeq+1 removes all messages.
        var removed = store.Compact(lastSeq + 1);
        removed.ShouldBe((ulong)10);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)0);
    }

    // Go: TestFileStoreCompactMsgCountBug server/filestore_test.go:916
    [Fact]
    public async Task Compact_with_prior_deletes_counts_correctly()
    {
        await using var store = CreateStore("compact-prior-deletes");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        // Remove seq 3 and 7 before compacting.
        await store.RemoveAsync(3, default);
        await store.RemoveAsync(7, default);

        // Compact at seq 5: removes seqs < 5 that still exist (1, 2, 4 — seq 3 already gone).
        store.Compact(5);

        // Remaining: seqs 5, 6, 8, 9, 10 (seq 7 was already deleted).
        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)5);
        state.FirstSeq.ShouldBe((ulong)5);
        state.LastSeq.ShouldBe((ulong)10);

        // Confirm seq 5, 6, 8, 9, 10 are loadable; 3, 7 are gone.
        (await store.LoadAsync(5, default)).ShouldNotBeNull();
        (await store.LoadAsync(6, default)).ShouldNotBeNull();
        (await store.LoadAsync(8, default)).ShouldNotBeNull();
        (await store.LoadAsync(9, default)).ShouldNotBeNull();
        (await store.LoadAsync(10, default)).ShouldNotBeNull();
        (await store.LoadAsync(3, default)).ShouldBeNull();
        (await store.LoadAsync(7, default)).ShouldBeNull();
    }

    // Go: TestFileStoreStreamTruncate server/filestore_test.go:991
    [Fact]
    public async Task Truncate_removes_messages_after_sequence()
    {
        await using var store = CreateStore("truncate-after-seq");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        // Truncate at seq 5: removes seqs > 5, leaving seqs 1-5.
        store.Truncate(5);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)5);
        state.FirstSeq.ShouldBe((ulong)1);
        state.LastSeq.ShouldBe((ulong)5);

        // Seqs 6-10 must be gone.
        for (ulong seq = 6; seq <= 10; seq++)
            (await store.LoadAsync(seq, default)).ShouldBeNull();

        // Seqs 1-5 must still be present.
        for (ulong seq = 1; seq <= 5; seq++)
            (await store.LoadAsync(seq, default)).ShouldNotBeNull();
    }

    // Go: TestFileStoreStreamTruncate server/filestore_test.go:1025
    [Fact]
    public async Task Truncate_with_interior_deletes()
    {
        await using var store = CreateStore("truncate-interior-deletes");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        // Remove seq 3 and 7 before truncating.
        await store.RemoveAsync(3, default);
        await store.RemoveAsync(7, default);

        // Truncate at seq 5: removes seqs > 5 that still exist (6, 8, 9, 10 — seq 7 already gone).
        store.Truncate(5);

        // Remaining: seqs 1, 2, 4, 5 (seq 3 was already deleted).
        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)4);
        state.LastSeq.ShouldBe((ulong)5);

        (await store.LoadAsync(1, default)).ShouldNotBeNull();
        (await store.LoadAsync(2, default)).ShouldNotBeNull();
        (await store.LoadAsync(3, default)).ShouldBeNull();
        (await store.LoadAsync(4, default)).ShouldNotBeNull();
        (await store.LoadAsync(5, default)).ShouldNotBeNull();
        (await store.LoadAsync(6, default)).ShouldBeNull();
        (await store.LoadAsync(7, default)).ShouldBeNull();
    }

    // Go: TestFileStorePurgeExWithSubject server/filestore_test.go:3743
    [Fact]
    public async Task PurgeEx_with_subject_removes_matching()
    {
        await using var store = CreateStore("purgeex-subject");

        // Interleave "foo" and "bar" messages.
        for (var i = 0; i < 5; i++)
        {
            await store.AppendAsync("foo", "foo-data"u8.ToArray(), default);
            await store.AppendAsync("bar", "bar-data"u8.ToArray(), default);
        }

        var before = await store.GetStateAsync(default);
        before.Messages.ShouldBe((ulong)10);

        // PurgeEx with subject="foo", seq=0, keep=0: removes all "foo" messages.
        var removed = store.PurgeEx("foo", 0, 0);
        removed.ShouldBe((ulong)5);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)5);

        // All remaining messages should be on "bar".
        var messages = await store.ListAsync(default);
        messages.Count.ShouldBe(5);
        foreach (var msg in messages)
            msg.Subject.ShouldBe("bar");
    }

    // Go: TestFileStorePurgeExKeepOneBug server/filestore_test.go:3382
    [Fact]
    public async Task PurgeEx_keep_one_preserves_last()
    {
        await using var store = CreateStore("purgeex-keep-one");

        ulong lastSeq = 0;
        for (var i = 0; i < 5; i++)
            lastSeq = await store.AppendAsync("foo", "data"u8.ToArray(), default);

        lastSeq.ShouldBe((ulong)5);

        // PurgeEx with keep=1: should remove 4 messages, keeping only the last one.
        var removed = store.PurgeEx("foo", 0, 1);
        removed.ShouldBe((ulong)4);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)1);

        // The remaining message must be the one with the highest sequence.
        var remaining = await store.ListAsync(default);
        remaining.Count.ShouldBe(1);
        remaining[0].Sequence.ShouldBe(lastSeq);
    }

    // Go: TestFileStorePurgeExNoTombsOnBlockRemoval server/filestore_test.go:3823
    [Fact]
    public async Task PurgeEx_no_tombstones_on_block_removal()
    {
        await using var store = CreateStore("purgeex-no-tombs");

        // Store messages on "foo" and "bar".
        for (var i = 0; i < 5; i++)
            await store.AppendAsync("foo", "foo-data"u8.ToArray(), default);

        var barSeqs = new List<ulong>();
        for (var i = 0; i < 5; i++)
            barSeqs.Add(await store.AppendAsync("bar", "bar-data"u8.ToArray(), default));

        // PurgeEx removes all "foo" messages.
        store.PurgeEx("foo", 0, 0);

        // "bar" messages must still be loadable and state must be consistent.
        foreach (var seq in barSeqs)
        {
            var msg = await store.LoadAsync(seq, default);
            msg.ShouldNotBeNull();
            msg!.Subject.ShouldBe("bar");
        }

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)5);
    }

    // Go: TestFileStorePurge server/filestore_test.go:709
    [Fact]
    public async Task Purge_then_list_returns_empty()
    {
        await using var store = CreateStore("purge-list");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        await store.PurgeAsync(default);

        var messages = await store.ListAsync(default);
        messages.Count.ShouldBe(0);
    }

    // Go: TestFileStorePurge server/filestore_test.go:709
    [Fact]
    public async Task Multiple_purges_are_safe()
    {
        await using var store = CreateStore("multi-purge");

        for (var i = 0; i < 5; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        await store.PurgeAsync(default);
        await store.PurgeAsync(default); // Double purge should not error.

        (await store.GetStateAsync(default)).Messages.ShouldBe((ulong)0);
    }

    // Go: TestFileStorePurge server/filestore_test.go:709
    [Fact]
    public async Task Purge_empty_store_is_safe()
    {
        await using var store = CreateStore("purge-empty");

        await store.PurgeAsync(default);

        (await store.GetStateAsync(default)).Messages.ShouldBe((ulong)0);
    }

    // Go: TestFileStorePurge server/filestore_test.go:709
    [Fact]
    public async Task Purge_with_prior_removes()
    {
        await using var store = CreateStore("purge-prior-rm");

        for (var i = 0; i < 10; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        // Remove some messages first.
        await store.RemoveAsync(2, default);
        await store.RemoveAsync(4, default);
        await store.RemoveAsync(6, default);

        (await store.GetStateAsync(default)).Messages.ShouldBe((ulong)7);

        await store.PurgeAsync(default);

        var state = await store.GetStateAsync(default);
        state.Messages.ShouldBe((ulong)0);
        state.Bytes.ShouldBe((ulong)0);
    }

    // Go: TestFileStorePurge server/filestore_test.go:776
    [Fact]
    public async Task Purge_then_store_then_purge_again()
    {
        await using var store = CreateStore("purge-cycle");

        for (var i = 0; i < 5; i++)
            await store.AppendAsync("foo", "data"u8.ToArray(), default);

        await store.PurgeAsync(default);

        for (var i = 0; i < 3; i++)
            await store.AppendAsync("foo", "new data"u8.ToArray(), default);

        (await store.GetStateAsync(default)).Messages.ShouldBe((ulong)3);

        await store.PurgeAsync(default);
        (await store.GetStateAsync(default)).Messages.ShouldBe((ulong)0);
    }

    // Go: TestFileStorePurge server/filestore_test.go:709
    [Fact]
    public async Task Purge_data_file_is_deleted()
    {
        var subDir = "purge-file";
        var dir = Path.Combine(_dir, subDir);

        await using (var store = CreateStore(subDir))
        {
            for (var i = 0; i < 10; i++)
                await store.AppendAsync("foo", "data"u8.ToArray(), default);

            await store.PurgeAsync(default);
        }

        // The data file should be cleaned up or empty after purge.
        var dataFile = Path.Combine(dir, "messages.jsonl");
        if (File.Exists(dataFile))
        {
            var content = File.ReadAllText(dataFile);
            content.Trim().ShouldBeEmpty();
        }
    }
}