DodoSSH/tests/DodoSSH.Client.Storage.Tests/OutboxStoreTests.cs

using DodoSSH.Contracts;
using static DodoSSH.Client.Storage.Tests.CacheHarness;

namespace DodoSSH.Client.Storage.Tests;

/// <summary>
/// The outbox, whose coalescing rules are where offline work is kept or lost.
/// </summary>
/// <remarks>
/// Two properties carry the weight. The ancestor must survive every coalesce, or a conflict can only
/// be arbitrated rather than merged. And a coalesced row must get a fresh operation id, or the server
/// can answer <c>Duplicate</c> for an operation whose contents have since changed and silently discard
/// the newer edit.
/// </remarks>
public sealed class OutboxStoreTests : IAsyncLifetime
{
    private CacheHarness harness = null!;

    /// <inheritdoc />
    public async ValueTask InitializeAsync() => harness = await CreateAsync();

    /// <inheritdoc />
    public ValueTask DisposeAsync()
    {
        harness.Dispose();
        return ValueTask.CompletedTask;
    }

    [Fact]
    public async Task AQueuedChange_ComesBackWithEverythingItNeedsToBePushed()
    {
        var entityId = Guid.CreateVersion7();
        var ancestor = new StoredAncestor(3, Payload(seed: 40), new SyncPlaintextFields());

        var queued = await harness.Outbox.QueueAsync(
            Change(entityId, expectedVersion: 3, seed: 7, ancestor: ancestor), Token);

        queued.OperationId.ShouldNotBe(Guid.Empty);
        queued.ExpectedVersion.ShouldBe(3);
        queued.Operation.ShouldBe(SyncOperation.Upsert);
        queued.Payload.ShouldNotBeNull();
        queued.Payload.Envelope.ShouldBe(Payload(seed: 7).Envelope);
        queued.Payload.WrappedDataKey.ShouldBe(Payload(seed: 7).WrappedDataKey);
        queued.Payload.DataKeyId.ShouldBe(Payload(seed: 7).DataKeyId);
        queued.Ancestor.ShouldNotBeNull();
        queued.Ancestor.Version.ShouldBe(3);
        queued.Ancestor.Payload.Envelope.ShouldBe(Payload(seed: 40).Envelope);
    }

    [Fact]
    public async Task ASecondEditToTheSameItem_CoalescesIntoOneRow()
    {
        // Two rows would have to be pushed in order, and the second's expectedVersion is the version
        // the first will produce — which is not known when it is queued.
        var entityId = Guid.CreateVersion7();

        await harness.Outbox.QueueAsync(Change(entityId, seed: 1), Token);
        await harness.Outbox.QueueAsync(Change(entityId, seed: 2), Token);

        var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);

        pending.ShouldHaveSingleItem().Payload!.Envelope.ShouldBe(Payload(seed: 2).Envelope);
    }

    [Fact]
    public async Task ACoalescedEdit_KeepsTheOriginalAncestorAndExpectedVersion()
    {
        // The load-bearing rule. The newest state is still a descendant of the base the first edit
        // branched from; adopting the caller's values here would discard the common ancestor after the
        // first edit and leave nothing to merge against.
        var entityId = Guid.CreateVersion7();
        var ancestor = new StoredAncestor(5, Payload(seed: 90), new SyncPlaintextFields());

        await harness.Outbox.QueueAsync(
            Change(entityId, expectedVersion: 5, seed: 1, ancestor: ancestor), Token);

        // A second edit arrives knowing nothing about the base.
        await harness.Outbox.QueueAsync(
            Change(entityId, expectedVersion: null, seed: 2, ancestor: null), Token);

        var pending = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();

        pending.ExpectedVersion.ShouldBe(5);
        pending.Ancestor.ShouldNotBeNull();
        pending.Ancestor.Version.ShouldBe(5);
        pending.Ancestor.Payload.Envelope.ShouldBe(Payload(seed: 90).Envelope);
    }

    [Fact]
    public async Task ACoalescedEdit_GetsAFreshOperationId()
    {
        // Reusing the id would let the server report Duplicate — meaning "already applied" — for an
        // operation whose payload has since changed, and the newer edit would vanish with the push
        // reported as a success.
        var entityId = Guid.CreateVersion7();

        var first = await harness.Outbox.QueueAsync(Change(entityId, seed: 1), Token);
        await harness.Outbox.MarkDispatchedAsync(first.Sequence, Token);

        var second = await harness.Outbox.QueueAsync(Change(entityId, seed: 2), Token);

        second.OperationId.ShouldNotBe(first.OperationId);
        second.Sequence.ShouldBe(first.Sequence);

        // And the retry counter resets, because this is a new operation rather than a further attempt
        // at the old one.
        second.Attempts.ShouldBe(0);
    }

    [Fact]
    public async Task AnUpsertFollowedByADelete_BecomesADelete()
    {
        var entityId = Guid.CreateVersion7();

        await harness.Outbox.QueueAsync(Change(entityId, expectedVersion: 2, seed: 1), Token);
        await harness.Outbox.QueueAsync(
            Change(entityId, SyncOperation.Delete, expectedVersion: 2), Token);

        var pending = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();

        pending.Operation.ShouldBe(SyncOperation.Delete);
        pending.Payload.ShouldBeNull();
    }

    [Fact]
    public async Task ChangesToDifferentItems_DrainInTheOrderTheyWereMade()
    {
        // Order matters for creates that reference each other — a host naming a jump host — so the
        // outbox is a queue, not a set.
        var first = Guid.CreateVersion7();
        var second = Guid.CreateVersion7();
        var third = Guid.CreateVersion7();

        foreach (var id in new[] { first, second, third })
        {
            await harness.Outbox.QueueAsync(Change(id), Token);
        }

        var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);

        pending.Select(p => p.EntityId).ShouldBe([first, second, third]);
    }

    [Fact]
    public async Task CoalescingDoesNotJumpTheQueue()
    {
        // The row keeps its original position. Re-editing the first item should not push it behind
        // items queued after it, because the later ones may depend on it existing.
        var first = Guid.CreateVersion7();
        var second = Guid.CreateVersion7();

        await harness.Outbox.QueueAsync(Change(first), Token);
        await harness.Outbox.QueueAsync(Change(second), Token);
        await harness.Outbox.QueueAsync(Change(first, seed: 9), Token);

        var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);

        pending.Select(p => p.EntityId).ShouldBe([first, second]);
    }

    [Fact]
    public async Task Revise_MovesTheAncestorForwardUnlikeQueue()
    {
        // The opposite intent from a coalesce: a merge has just been performed against a newer server
        // version, so that version becomes the base. Leaving the old ancestor would make the re-push
        // conflict against the same point for ever.
        var entityId = Guid.CreateVersion7();
        var original = new StoredAncestor(1, Payload(seed: 10), new SyncPlaintextFields());

        var queued = await harness.Outbox.QueueAsync(
            Change(entityId, expectedVersion: 1, ancestor: original), Token);

        var merged = new StoredAncestor(4, Payload(seed: 20), new SyncPlaintextFields());

        var revised = await harness.Outbox.ReviseAsync(
            queued.Sequence,
            SyncOperation.Upsert,
            expectedVersion: 4,
            Payload(seed: 30),
            new SyncPlaintextFields(),
            merged,
            Token);

        revised.ShouldNotBeNull();
        revised.ExpectedVersion.ShouldBe(4);
        revised.Ancestor!.Version.ShouldBe(4);
        revised.Ancestor.Payload.Envelope.ShouldBe(Payload(seed: 20).Envelope);
        revised.OperationId.ShouldNotBe(queued.OperationId);
    }

    [Fact]
    public async Task AParkedOperation_IsNotHandedOutForPushing()
    {
        // An operation the server called Invalid will never succeed. Retrying it would spin and, worse,
        // would block every change queued behind it in a vault the user can still write to.
        var parked = Guid.CreateVersion7();
        var healthy = Guid.CreateVersion7();

        var queued = await harness.Outbox.QueueAsync(Change(parked), Token);
        await harness.Outbox.QueueAsync(Change(healthy), Token);

        await harness.Outbox.ParkAsync(queued.Sequence, "Entity type not supported.", Token);

        var pending = await harness.Outbox.TakeAsync(VaultId, 10, Token);
        pending.ShouldHaveSingleItem().EntityId.ShouldBe(healthy);

        var listed = (await harness.Outbox.ListParkedAsync(VaultId, Token)).ShouldHaveSingleItem();
        listed.EntityId.ShouldBe(parked);
        listed.LastError.ShouldBe("Entity type not supported.");
    }

    [Fact]
    public async Task ReEditingAParkedOperation_Unparks()
    {
        // The user's remedy for a rejected change is to change it. That has to actually re-arm it.
        var entityId = Guid.CreateVersion7();

        var queued = await harness.Outbox.QueueAsync(Change(entityId), Token);
        await harness.Outbox.ParkAsync(queued.Sequence, "nope", Token);

        var requeued = await harness.Outbox.QueueAsync(Change(entityId, seed: 5), Token);

        requeued.IsParked.ShouldBeFalse();
        requeued.LastError.ShouldBeNull();
        (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
    }

    [Fact]
    public async Task Complete_RemovesTheOperation()
    {
        var queued = await harness.Outbox.QueueAsync(Change(Guid.CreateVersion7()), Token);

        (await harness.Outbox.CompleteAsync(queued.Sequence, Token)).ShouldBeTrue();
        (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldBeEmpty();

        // Idempotent: a drain that retries after a crash must not fail on an already-cleared row.
        (await harness.Outbox.CompleteAsync(queued.Sequence, Token)).ShouldBeFalse();
    }

    [Fact]
    public async Task MarkDispatched_CountsAttempts()
    {
        var queued = await harness.Outbox.QueueAsync(Change(Guid.CreateVersion7()), Token);

        await harness.Outbox.MarkDispatchedAsync(queued.Sequence, Token);
        await harness.Outbox.MarkDispatchedAsync(queued.Sequence, Token);

        var pending = (await harness.Outbox.TakeAsync(VaultId, 10, Token)).ShouldHaveSingleItem();
        pending.Attempts.ShouldBe(2);
    }

    [Fact]
    public async Task AnUpsertWithoutAPayload_IsRefused()
    {
        // Caught here rather than at the server, where it would come back as one opaque Invalid among
        // a batch of otherwise good operations.
        var change = new QueuedChange(
            VaultId,
            SyncEntityType.Host,
            Guid.CreateVersion7(),
            SyncOperation.Upsert,
            ExpectedVersion: null,
            Payload: null,
            Fields: null,
            Ancestor: null);

        await Should.ThrowAsync<ArgumentException>(
            async () => await harness.Outbox.QueueAsync(change, Token));
    }

    [Fact]
    public async Task ThePendingOperationForAnItem_CanBeLookedUpDirectly()
    {
        // How a pull discovers that an incoming change collides with local work.
        var entityId = Guid.CreateVersion7();
        await harness.Outbox.QueueAsync(Change(entityId), Token);

        (await harness.Outbox.FindAsync(VaultId, SyncEntityType.Host, entityId, Token))
            .ShouldNotBeNull();

        (await harness.Outbox.FindAsync(VaultId, SyncEntityType.Host, Guid.CreateVersion7(), Token))
            .ShouldBeNull();
    }

    private static CancellationToken Token => TestContext.Current.CancellationToken;
}