DodoSSH/src/DodoSSH.Client.Storage/OutboxStore.cs

using DodoSSH.Contracts;
using Microsoft.EntityFrameworkCore;

namespace DodoSSH.Client.Storage;

/// <summary>A local change to queue.</summary>
/// <param name="VaultId">Owning vault.</param>
/// <param name="EntityType">Kind of item.</param>
/// <param name="EntityId">The item. Client-generated UUIDv7, so items can be made offline.</param>
/// <param name="Operation">Upsert or delete.</param>
/// <param name="ExpectedVersion">The version the client believes the server holds; null to create.</param>
/// <param name="Payload">Ciphertext. Required for an upsert.</param>
/// <param name="Fields">Plaintext columns the server needs.</param>
/// <param name="Ancestor">
/// The version this edit branched from, so a conflict can be merged rather than arbitrated. Null when
/// creating, where there is nothing to have branched from.
/// </param>
public sealed record QueuedChange(
    Guid VaultId,
    SyncEntityType EntityType,
    Guid EntityId,
    SyncOperation Operation,
    int? ExpectedVersion,
    EncryptedPayload? Payload,
    SyncPlaintextFields? Fields,
    StoredAncestor? Ancestor);

/// <summary>
/// Changes made here that the server has not yet accepted.
/// </summary>
/// <remarks>
/// <para>
/// One row per item, and that is a database constraint rather than a convention. Two queued edits to
/// one item would have to be pushed in order, and the second's <c>expectedVersion</c> is the version
/// the first will produce — which is not known when it is queued. Coalescing sidesteps that instead of
/// managing it, and the row holds a desired end state rather than a delta, so coalescing loses nothing.
/// </para>
/// <para>
/// <b>Why a coalesced row gets a new operation id.</b> The id is the server's exactly-once key. If a
/// push has already gone out and the user edits again, keeping the id would let the server answer
/// <c>Duplicate</c> for an operation whose contents have since changed — silently discarding the newer
/// edit. A fresh id means the newer state is offered on its own terms: if the earlier push did land,
/// the version has moved on, the push comes back <c>Conflict</c>, and the merge resolves it against an
/// ancestor that is this client's own earlier edit. That merge finds no disagreement, so it converges
/// on the newest state with nothing for the user to arbitrate.
/// </para>
/// <para>
/// ◆ <b>Which is only true if an answer is applied to the operation it answers.</b> A push reads a row,
/// sends it, and acts on the reply some time later; a user who saves in between coalesces a new state into
/// that same row. So <see cref="CompleteAsync"/>, <see cref="ReviseAsync"/> and <see cref="ParkAsync"/>
/// take the operation that was sent and do nothing when the row has since become a different one. Keyed by
/// sequence alone, an acknowledgement of the earlier state deleted the row holding the later one, and the
/// edit was gone with nothing to say so.
/// </para>
/// </remarks>
public sealed class OutboxStore(
    IDbContextFactory<ClientCacheContext> contexts,
    LocalCacheProtector protector,
    TimeProvider clock)
{
    /// <summary>
    /// Queues a change the user just made, coalescing into any row already pending for the item.
    /// </summary>
    /// <remarks>
    /// A coalesced row keeps the ancestor and <c>expectedVersion</c> of the row it replaces, because
    /// the new state is still a descendant of that same base. Taking the caller's values instead would
    /// throw away the common ancestor after the first edit, and with it the ability to merge.
    /// </remarks>
    public async Task<PendingOperation> QueueAsync(
        QueuedChange change,
        CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(change);
        Validate(change);

        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var row = await FindRowAsync(context, change.VaultId, change.EntityType, change.EntityId, cancellationToken)
            .ConfigureAwait(false);

        if (row is null)
        {
            row = new OutboxRow
            {
                VaultId = change.VaultId,
                EntityType = change.EntityType,
                EntityId = change.EntityId,
                ExpectedVersion = change.ExpectedVersion,
            };

            SetAncestor(row, change.EntityType, change.EntityId, change.Ancestor);
            context.Add(row);
        }

        row.OperationId = Guid.CreateVersion7();
        row.Operation = change.Operation;
        row.QueuedAtUtc = clock.GetUtcNow();
        row.Attempts = 0;
        row.LastError = null;
        row.IsParked = false;

        SetPayload(row, change.EntityType, change.EntityId, change.Payload, change.Fields);

        await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);

        return ToPending(row);
    }

    /// <summary>
    /// Replaces a pending operation with the outcome of a merge.
    /// </summary>
    /// <remarks>
    /// <para>
    /// Distinct from <see cref="QueueAsync"/> because the intent is opposite: this <em>does</em> move
    /// the ancestor forward, to the server version the merge was performed against. Without that the
    /// re-push would conflict against the same base for ever.
    /// </para>
    /// <para>
    /// It also <b>keeps the attempt count</b>, where queueing resets it. That difference is what makes
    /// the retry bound real: a row that has conflicted five times needs a person to look at it whether
    /// or not each attempt carried a freshly merged payload, whereas a user making a new edit has
    /// genuinely started over.
    /// </para>
    /// </remarks>
    public async Task<PendingOperation?> ReviseAsync(
        PendingOperation revising,
        SyncOperation operation,
        int? expectedVersion,
        EncryptedPayload? payload,
        SyncPlaintextFields? fields,
        StoredAncestor? ancestor,
        CancellationToken cancellationToken)
    {
        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        ArgumentNullException.ThrowIfNull(revising);

        var row = await context.Set<OutboxRow>()
            .SingleOrDefaultAsync(
                r => r.Sequence == revising.Sequence && r.OperationId == revising.OperationId,
                cancellationToken)
            .ConfigureAwait(false);

        if (row is null)
        {
            // Completed, or replaced by a newer edit since it was read. Either way the merge was of a state
            // that is no longer the one queued, and the next push will meet the server with the current one.
            return null;
        }

        row.OperationId = Guid.CreateVersion7();
        row.Operation = operation;
        row.ExpectedVersion = expectedVersion;
        row.LastError = null;
        row.IsParked = false;

        SetPayload(row, row.EntityType, row.EntityId, payload, fields);
        SetAncestor(row, row.EntityType, row.EntityId, ancestor);

        await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);

        return ToPending(row);
    }

    /// <summary>Reads the next operations to push, oldest first, skipping parked ones.</summary>
    public async Task<IReadOnlyList<PendingOperation>> TakeAsync(
        Guid vaultId,
        int limit,
        CancellationToken cancellationToken)
    {
        ArgumentOutOfRangeException.ThrowIfLessThan(limit, 1);

        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var rows = await context.Set<OutboxRow>()
            .AsNoTracking()
            .Where(r => r.VaultId == vaultId && !r.IsParked)
            .OrderBy(r => r.Sequence)
            .Take(limit)
            .ToListAsync(cancellationToken)
            .ConfigureAwait(false);

        return [.. rows.Select(ToPending)];
    }

    /// <summary>Reads the operation pending for one item, if any.</summary>
    public async Task<PendingOperation?> FindAsync(
        Guid vaultId,
        SyncEntityType entityType,
        Guid entityId,
        CancellationToken cancellationToken)
    {
        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var row = await FindRowAsync(context, vaultId, entityType, entityId, cancellationToken)
            .ConfigureAwait(false);

        return row is null ? null : ToPending(row);
    }

    /// <summary>
    /// Reads every pending operation for a vault, parked ones included.
    /// </summary>
    /// <remarks>
    /// What the interface needs, as opposed to what the pusher needs. A parked change is still the
    /// user's current intent for that item and must be what they see; hiding it because the server
    /// refused it would show them the old values and look like their edit was lost.
    /// </remarks>
    public async Task<IReadOnlyList<PendingOperation>> ListAllAsync(
        Guid vaultId,
        CancellationToken cancellationToken)
    {
        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var rows = await context.Set<OutboxRow>()
            .AsNoTracking()
            .Where(r => r.VaultId == vaultId)
            .OrderBy(r => r.Sequence)
            .ToListAsync(cancellationToken)
            .ConfigureAwait(false);

        return [.. rows.Select(ToPending)];
    }

    /// <summary>Reads operations the server refused, which need a person.</summary>
    public async Task<IReadOnlyList<PendingOperation>> ListParkedAsync(
        Guid vaultId,
        CancellationToken cancellationToken)
    {
        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var rows = await context.Set<OutboxRow>()
            .AsNoTracking()
            .Where(r => r.VaultId == vaultId && r.IsParked)
            .OrderBy(r => r.Sequence)
            .ToListAsync(cancellationToken)
            .ConfigureAwait(false);

        return [.. rows.Select(ToPending)];
    }

    /// <summary>Records that an operation has been sent, so a repeated failure can be noticed.</summary>
    public Task MarkDispatchedAsync(long sequence, CancellationToken cancellationToken) =>
        UpdateAsync(sequence, operationId: null, row => row.Attempts++, cancellationToken);

    /// <summary>Removes an operation the server accepted, unless a newer edit has replaced it since.</summary>
    /// <returns>Whether a row was removed.</returns>
    public async Task<bool> CompleteAsync(PendingOperation operation, CancellationToken cancellationToken)
    {
        ArgumentNullException.ThrowIfNull(operation);

        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var removed = await context.Set<OutboxRow>()
            .Where(r => r.Sequence == operation.Sequence && r.OperationId == operation.OperationId)
            .ExecuteDeleteAsync(cancellationToken)
            .ConfigureAwait(false);

        return removed > 0;
    }

    /// <summary>
    /// Stops retrying an operation and records why.
    /// </summary>
    /// <remarks>
    /// For the answers that will not change on retry — the server calling an operation structurally
    /// invalid, or the caller no longer having permission. Retrying either would spin forever and, far
    /// worse, would block every change queued behind it in a vault the user can still write to.
    /// </remarks>
    public Task ParkAsync(PendingOperation operation, string reason, CancellationToken cancellationToken) =>
        UpdateAsync(
            operation.Sequence,
            operation.OperationId,
            row =>
            {
                row.IsParked = true;
                row.LastError = reason;
            },
            cancellationToken);

    /// <summary>Records a transient failure without parking the operation.</summary>
    public Task RecordFailureAsync(long sequence, string reason, CancellationToken cancellationToken) =>
        UpdateAsync(sequence, operationId: null, row => row.LastError = reason, cancellationToken);

    private static Task<OutboxRow?> FindRowAsync(
        ClientCacheContext context,
        Guid vaultId,
        SyncEntityType entityType,
        Guid entityId,
        CancellationToken cancellationToken) =>
        context.Set<OutboxRow>()
            .SingleOrDefaultAsync(
                r => r.VaultId == vaultId && r.EntityType == entityType && r.EntityId == entityId,
                cancellationToken);

    private static void Validate(QueuedChange change)
    {
        if (change.Operation == SyncOperation.Upsert && change.Payload is null)
        {
            throw new ArgumentException("An upsert requires a payload.", nameof(change));
        }

        if (change.Operation == SyncOperation.Unspecified)
        {
            throw new ArgumentException("An operation is required.", nameof(change));
        }

        if (change.EntityId == Guid.Empty)
        {
            throw new ArgumentException("An entity id is required.", nameof(change));
        }
    }

    private async Task UpdateAsync(
        long sequence,
        Guid? operationId,
        Action<OutboxRow> mutate,
        CancellationToken cancellationToken)
    {
        var context = contexts.CreateDbContext();
        await using var scope = context.ConfigureAwait(false);

        var row = await context.Set<OutboxRow>()
            .SingleOrDefaultAsync(
                r => r.Sequence == sequence && (operationId == null || r.OperationId == operationId),
                cancellationToken)
            .ConfigureAwait(false);

        if (row is null)
        {
            return;
        }

        mutate(row);
        await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
    }

    private void SetPayload(
        OutboxRow row,
        SyncEntityType entityType,
        Guid entityId,
        EncryptedPayload? payload,
        SyncPlaintextFields? fields)
    {
        row.Payload = payload?.Envelope;
        row.WrappedDataKey = payload?.WrappedDataKey;
        row.DataKeyId = payload?.DataKeyId;
        row.KeyGeneration = payload?.KeyGeneration ?? 0;
        row.AadVersion = payload?.AadVersion ?? 0;

        row.ProtectedFields = Seal(entityType, entityId, fields);
    }

    private void SetAncestor(
        OutboxRow row,
        SyncEntityType entityType,
        Guid entityId,
        StoredAncestor? ancestor)
    {
        row.AncestorVersion = ancestor?.Version;
        row.AncestorPayload = ancestor?.Payload.Envelope;
        row.AncestorWrappedDataKey = ancestor?.Payload.WrappedDataKey;
        row.AncestorDataKeyId = ancestor?.Payload.DataKeyId;
        row.AncestorKeyGeneration = ancestor?.Payload.KeyGeneration;
        row.AncestorAadVersion = ancestor?.Payload.AadVersion;

        row.AncestorProtectedFields = Seal(entityType, entityId, ancestor?.Fields);
    }

    private byte[]? Seal(SyncEntityType entityType, Guid entityId, SyncPlaintextFields? fields) =>
        fields is null
            ? null
            : protector.Protect(
                AadResourceTypes.For(entityType), entityId, PlaintextFieldsCodec.Encode(fields));

    private SyncPlaintextFields? Open(SyncEntityType entityType, Guid entityId, byte[]? sealedFields)
    {
        if (sealedFields is null)
        {
            return null;
        }

        var plaintext = protector.TryUnprotect(
            AadResourceTypes.For(entityType), entityId, sealedFields);

        return plaintext is null ? null : PlaintextFieldsCodec.TryDecode(plaintext);
    }

    private PendingOperation ToPending(OutboxRow row)
    {
        var ancestorPayload = CacheMapping.ToAncestorPayload(row);

        var ancestor = ancestorPayload is null || row.AncestorVersion is null
            ? null
            : new StoredAncestor(
                row.AncestorVersion.Value,
                ancestorPayload,
                Open(row.EntityType, row.EntityId, row.AncestorProtectedFields));

        return new PendingOperation(
            row.Sequence,
            row.OperationId,
            row.VaultId,
            row.EntityType,
            row.EntityId,
            row.Operation,
            row.ExpectedVersion,
            CacheMapping.ToPayload(
                row.Payload, row.WrappedDataKey, row.DataKeyId, row.KeyGeneration, row.AadVersion),
            Open(row.EntityType, row.EntityId, row.ProtectedFields),
            ancestor,
            row.QueuedAtUtc,
            row.Attempts,
            row.LastError,
            row.IsParked);
    }
}