DodoSSH/src/DodoSSH.Client.Sync/SyncEngine.cs

using System.Globalization;
using System.Runtime.InteropServices;
using DodoSSH.Client.Api;
using DodoSSH.Client.Storage;
using DodoSSH.Contracts;

namespace DodoSSH.Client.Sync;

/// <summary>
/// One vault's synchronisation pass: pull, reconcile, push, pull again.
/// </summary>
/// <remarks>
/// <para>
/// Pull first, so a local change is merged against the newest server state before it is offered — which
/// turns most would-be conflicts into ordinary merges and keeps the push round count down. Push second.
/// Pull once more at the end only if something was pushed, so the mirror reflects the versions the
/// server actually assigned.
/// </para>
/// <para>
/// <b>Pulling does not decrypt.</b> A change with no local work pending is copied into the mirror as
/// ciphertext and nothing more. Decryption happens when a merge needs it, or when the interface reads an
/// item. For a five-thousand-item first sync that is the difference between plumbing bytes and running
/// ten thousand AEAD operations for nothing.
/// </para>
/// </remarks>
public sealed class SyncEngine
{
    private readonly ISyncApi api;
    private readonly ItemStore items;
    private readonly OutboxStore outbox;
    private readonly SyncStateStore syncState;
    private readonly ConflictStore conflicts;
    private readonly VaultKeyring keyring;
    private readonly TimeProvider clock;
    private readonly SyncOptions options;

    /// <remarks>
    /// One per synchronised item type, built once. The keys are also the pull filter — see
    /// <see cref="ItemKinds"/> — so a type this engine cannot reconcile is never requested, and a type it
    /// can reconcile cannot be left out of the request.
    /// </remarks>
    private readonly Dictionary<SyncEntityType, IItemReconciler> reconcilers;

    /// <summary>Creates the engine.</summary>
    public SyncEngine(
        ISyncApi api,
        ItemStore items,
        OutboxStore outbox,
        SyncStateStore syncState,
        ConflictStore conflicts,
        VaultKeyring keyring,
        TimeProvider clock,
        SyncOptions? options = null)
    {
        ArgumentNullException.ThrowIfNull(api);
        ArgumentNullException.ThrowIfNull(items);
        ArgumentNullException.ThrowIfNull(outbox);
        ArgumentNullException.ThrowIfNull(syncState);
        ArgumentNullException.ThrowIfNull(conflicts);
        ArgumentNullException.ThrowIfNull(keyring);
        ArgumentNullException.ThrowIfNull(clock);

        this.api = api;
        this.items = items;
        this.outbox = outbox;
        this.syncState = syncState;
        this.conflicts = conflicts;
        this.keyring = keyring;
        this.clock = clock;
        this.options = options ?? SyncOptions.Default;

        reconcilers = ItemKinds.Reconcilers(outbox, conflicts, keyring);
    }

    /// <summary>Runs a full pass over one vault.</summary>
    public async Task<SyncReport> SyncAsync(Guid vaultId, CancellationToken cancellationToken)
    {
        var report = new SyncReportBuilder(vaultId);

        await PullAsync(vaultId, report, cancellationToken).ConfigureAwait(false);

        var pushedAnything = false;

        for (var round = 1; ; round++)
        {
            var outcome = await DrainAsync(vaultId, report, cancellationToken).ConfigureAwait(false);

            if (outcome.Sent == 0)
            {
                break;
            }

            pushedAnything = true;

            if (!outcome.NeedsAnotherRound)
            {
                break;
            }

            if (round >= options.MaxPushRounds)
            {
                report.RoundsExhausted = true;
                break;
            }
        }

        if (pushedAnything)
        {
            await PullAsync(vaultId, report, cancellationToken).ConfigureAwait(false);
        }

        return report.Build();
    }

    /// <summary>
    /// Reads every change available and applies it.
    /// </summary>
    /// <remarks>
    /// The cursor is saved after each page's changes are applied, which makes applying at-least-once
    /// rather than exactly-once: a process that dies between the two re-reads that page next time. That
    /// is deliberate and safe, because applying a change is a blind overwrite of a mirror row and a
    /// resurrection takes a deterministic id. The other ordering — save the cursor first — would lose
    /// changes outright, which no amount of idempotence can repair.
    /// </remarks>
    private async Task PullAsync(
        Guid vaultId,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);

        // At most one restart per pull. A server that rejects the cursor it has just issued is not
        // telling this client anything it can act on, and replaying the whole log against it would turn
        // one broken deployment into an unbounded amount of work.
        var alreadyStartedOver = false;

        for (var page = 0; page < options.MaxPullPages; page++)
        {
            SyncPullResponse response;

            try
            {
                response = await api.SyncPullAsync(
                    vaultId,
                    new SyncPullRequest(state.Cursor, options.PullPageSize, ItemKinds.SyncedTypes),
                    cancellationToken).ConfigureAwait(false);
            }
            catch (DodoSshApiException exception)
                when (!alreadyStartedOver && WasTheCursorRefused(exception, state.Cursor))
            {
                alreadyStartedOver = true;
                state = await StartOverAsync(state, report, cancellationToken).ConfigureAwait(false);

                continue;
            }

            foreach (var change in response.Changes)
            {
                await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false);
                report.Pulled++;

                // Counted apart for the same reason the pushed ones are: a machine reads back the log
                // entries it just wrote, so a pass that "pulled five changes" may have carried nothing
                // anybody did. See SyncReport.PulledItems.
                if (IsLog(change.EntityType))
                {
                    report.PulledLogEntries++;
                }
            }

            var advanced = !string.Equals(state.Cursor, response.NextCursor, StringComparison.Ordinal);

            state = Record(vaultId, state, response, report);
            await syncState.SaveAsync(state, cancellationToken).ConfigureAwait(false);

            if (!response.HasMore)
            {
                break;
            }

            // A server that claims more but neither returns a change nor moves the cursor would spin
            // this loop for ever. Stopping is the only safe reading of that answer.
            if (!advanced && response.Changes.Count == 0)
            {
                break;
            }
        }
    }

    /// <summary>
    /// Whether the server refused the cursor this client sent, rather than failing for some other reason.
    /// </summary>
    /// <remarks>
    /// Read from the problem code and not from the prose, which is free to change, and never claimed for a
    /// request that carried no cursor: "from the beginning" is the one position a client is allowed to ask
    /// for, so a rejection of <em>that</em> is a server this code cannot reason about and has to surface.
    /// It is also what keeps the retry from looping — the restarted request sends no cursor.
    /// </remarks>
    /// <summary>Whether a change is one the machine wrote about itself rather than one somebody made.</summary>
    /// <remarks>
    /// Used only for reporting. Log entries sync exactly like every other item — they are pulled, pushed,
    /// merged and stored by the same code — and this distinguishes them nowhere except in the two numbers a
    /// person reads.
    /// </remarks>
    private static bool IsLog(SyncEntityType type) =>
        type is SyncEntityType.ConnectionLogEntry or SyncEntityType.ActivityLogEntry;

    private static bool WasTheCursorRefused(DodoSshApiException exception, string? cursor) =>
        !string.IsNullOrEmpty(cursor)
        && string.Equals(exception.Code, ProblemCodes.InvalidCursor, StringComparison.Ordinal);

    /// <summary>
    /// Forgets this vault's position and reads the log again from the beginning.
    /// </summary>
    /// <remarks>
    /// <para>
    /// A cursor the server will not accept is not a transient failure. Every later pass reads the same
    /// stored cursor and is refused the same way, so a vault that met one stayed there for good — pulling
    /// nothing, and pushing nothing either, because the pass threw before it reached the outbox. The user
    /// saw a 400 saying to resync from the beginning and had no way to do it. This is that way.
    /// </para>
    /// <para>
    /// The causes are all on the far side: a rotated cursor signing key, a vault served from a restored
    /// database whose sequences no longer reach that far, a cache copied between machines. None of them is
    /// something the person at the keyboard did, and none of them is something they could act on if asked.
    /// </para>
    /// <para>
    /// <b>The mirror is deliberately kept.</b> Replaying from the beginning rewrites every row the server
    /// still has — applying a change is a blind overwrite — so the re-pull repairs the mirror as it goes.
    /// Clearing it first would claim more than the evidence supports: the position was refused, not the
    /// contents, and a machine that loses its connection halfway through the replay would be left with
    /// less than it started with. <c>SyncStateStore.ResetAsync</c> is the heavier remedy, for when the
    /// cache itself is the thing in doubt.
    /// </para>
    /// <para>
    /// That leaves one gap, and it is worth naming rather than leaving to be discovered: once the server
    /// starts collecting tombstones — <c>DodoOptions.TombstoneRetentionDays</c>, not implemented yet — a
    /// replay no longer carries a deletion older than the retention window. A machine that missed such a
    /// delete and then had its cursor refused would keep the row. Nothing here can tell that from a row
    /// the server still has, so the answer when it matters will be to clear the mirror as well, not to
    /// guess.
    /// </para>
    /// <para>
    /// Written down before the replay begins, so a process that dies mid-replay starts the next one from
    /// the beginning as well, rather than meeting the same refusal again.
    /// </para>
    /// </remarks>
    private async Task<StoredSyncState> StartOverAsync(
        StoredSyncState state,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        var restarted = state with { Cursor = null };

        await syncState.SaveAsync(restarted, cancellationToken).ConfigureAwait(false);

        report.ResyncedFromStart = true;

        return restarted;
    }

    private StoredSyncState Record(
        Guid vaultId,
        StoredSyncState state,
        SyncPullResponse response,
        SyncReportBuilder report)
    {
        var now = clock.GetUtcNow();
        var skew = (long)(response.ServerTime - now).TotalMilliseconds;

        report.ServerKeyGeneration = response.CurrentKeyGeneration;
        report.ServerTimeSkewMs = skew;

        // A generation ahead of the key this client holds means the vault was rekeyed and this client's
        // grant has not been re-wrapped. Items pulled meanwhile are stored but cannot be read.
        keyring.TryGet(vaultId, out _, out var held);
        report.RekeyRequired = response.CurrentKeyGeneration > held;

        return state with
        {
            Cursor = response.NextCursor,
            KeyGeneration = response.CurrentKeyGeneration,
            LastPulledAt = now,
            ServerTimeSkewMs = skew,
        };
    }

    private async Task ApplyAsync(
        Guid vaultId,
        SyncChange change,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        if (!reconcilers.TryGetValue(change.EntityType, out var reconciler))
        {
            // Reserved in the contract but not yet syncable here. The pull filter already asks for only
            // the types this build handles, so reaching this means a newer server sent something extra —
            // and ignoring it keeps that from breaking an older client's pull. Not mirrored either: a row
            // this build can never read is cache with no reader.
            return;
        }

        await ItemMirror.WriteAsync(items, vaultId, change, cancellationToken).ConfigureAwait(false);

        var pending = await outbox
            .FindAsync(vaultId, change.EntityType, change.EntityId, cancellationToken)
            .ConfigureAwait(false);

        if (pending is null || pending.IsParked)
        {
            return;
        }

        await reconciler.ReconcileAsync(vaultId, change, pending, report, cancellationToken)
            .ConfigureAwait(false);
    }

    /// <summary>What one push round achieved.</summary>
    [StructLayout(LayoutKind.Auto)]
    private readonly record struct DrainOutcome(int Sent, int Conflicts, bool BatchWasFull)
    {
        internal bool NeedsAnotherRound => Conflicts > 0 || BatchWasFull;
    }

    /// <summary>
    /// Sends one batch and acts on each per-operation answer.
    /// </summary>
    /// <remarks>
    /// <b>The cursor in the push response is deliberately ignored.</b> It sits after this push's own
    /// changes, so adopting it would skip any change another client committed at a lower sequence
    /// between this client's last pull and this push — permanently. Continuing from the cursor this
    /// client already holds re-reads its own writes, which costs one redundant page and is idempotent.
    /// </remarks>
    private async Task<DrainOutcome> DrainAsync(
        Guid vaultId,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        var pending = await outbox
            .TakeAsync(vaultId, options.MaxOperationsPerPush, cancellationToken)
            .ConfigureAwait(false);

        if (pending.Count == 0)
        {
            return default;
        }

        var operations = new List<SyncPushOperation>(pending.Count);
        var byOperationId = new Dictionary<Guid, PendingOperation>(pending.Count);

        foreach (var operation in pending)
        {
            var payload = await CurrentAsync(vaultId, operation, cancellationToken)
                .ConfigureAwait(false);

            operations.Add(new SyncPushOperation(
                operation.OperationId,
                operation.EntityType,
                operation.EntityId,
                operation.Operation,
                operation.ExpectedVersion,
                payload,
                operation.Fields));

            byOperationId[operation.OperationId] = operation;

            await outbox.MarkDispatchedAsync(operation.Sequence, cancellationToken)
                .ConfigureAwait(false);
        }

        var response = await api
            .SyncPushAsync(vaultId, new SyncPushRequest(operations), cancellationToken)
            .ConfigureAwait(false);

        var conflicted = 0;

        foreach (var result in response.Results)
        {
            if (!byOperationId.TryGetValue(result.OperationId, out var operation))
            {
                // An id this client did not send. Nothing sane to do with it.
                continue;
            }

            // The attempt count was incremented above, so the bound is read from the fresh value.
            var attempts = operation.Attempts + 1;

            if (await HandleAsync(vaultId, operation with { Attempts = attempts }, result, report, cancellationToken)
                .ConfigureAwait(false))
            {
                conflicted++;
            }
        }

        return new DrainOutcome(
            pending.Count, conflicted, pending.Count == options.MaxOperationsPerPush);
    }

    /// <summary>
    /// The payload to send, re-sealed under the current key if it was queued before a rotation.
    /// </summary>
    /// <remarks>
    /// <para>
    /// <b>Nothing may reach the server under a superseded generation, and this is where that is
    /// enforced.</b> A change queued offline is sealed under whatever key was current when it was made,
    /// and a rotation can land in between — so sending it as it stands would store a brand-new item
    /// under the key the person who was just removed still holds. The vault's own re-sealing pass cannot
    /// help: it runs over what the server holds, and this has not been sent yet.
    /// </para>
    /// <para>
    /// The revised payload is written back to the outbox before it goes, so a push whose answer is lost
    /// is retried as the same bytes. Re-sealing on each attempt instead would produce a different
    /// envelope every time, which is harmless on the wire and would leave the queued row disagreeing
    /// with what the server may already have accepted.
    /// </para>
    /// </remarks>
    private async Task<EncryptedPayload?> CurrentAsync(
        Guid vaultId,
        PendingOperation operation,
        CancellationToken cancellationToken)
    {
        if (operation.Payload is not { } payload
            || !keyring.TryGet(vaultId, out _, out var generation)
            || payload.KeyGeneration >= generation)
        {
            return operation.Payload;
        }

        var version = SyncVersions.NextVersion(operation.ExpectedVersion);

        var resealed = PayloadReseal.TryReseal(
            keyring,
            vaultId,
            operation.EntityType,
            operation.EntityId,
            payload,
            openAtVersion: version,
            sealAtVersion: version);

        if (resealed is null)
        {
            // The generation this change was queued under is one this session no longer holds. It goes
            // as it stands: the server takes it either way, and a queued edit that cannot be re-sealed
            // is still the user's work.
            return payload;
        }

        await outbox.ReviseAsync(
                operation.Sequence,
                operation.Operation,
                operation.ExpectedVersion,
                resealed,
                operation.Fields,
                operation.Ancestor,
                cancellationToken)
            .ConfigureAwait(false);

        return resealed;
    }

    /// <returns>Whether this answer warrants another push round.</returns>
    private async Task<bool> HandleAsync(
        Guid vaultId,
        PendingOperation operation,
        SyncPushResult result,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        switch (result.Status)
        {
            case SyncOperationStatus.Applied:
            case SyncOperationStatus.Duplicate:
                // Duplicate means an earlier push of this exact operation id already landed, so the
                // stored state is what this operation intended. Treated as success on purpose: that is
                // what makes a retry after a timeout exactly-once rather than merely at-least-once.
                await AcceptAsync(vaultId, operation, result, cancellationToken).ConfigureAwait(false);
                report.Pushed++;

                // Counted separately so the interface can stay quiet about a pass that carried nothing but
                // log entries. Every user action now queues one, so without this the background pass would
                // have something to announce after literally every save — and would overwrite the message
                // the save itself had just put on the status line.
                if (IsLog(operation.EntityType))
                {
                    report.PushedLogEntries++;
                }

                return false;

            case SyncOperationStatus.Conflict:
                return await ResolveAsync(vaultId, operation, result, report, cancellationToken)
                    .ConfigureAwait(false);

            case SyncOperationStatus.Forbidden:
                await RejectAsync(
                    vaultId,
                    operation,
                    "You no longer have permission to change this item.",
                    report,
                    cancellationToken).ConfigureAwait(false);
                return false;

            case SyncOperationStatus.Invalid:
                await RejectAsync(
                    vaultId,
                    operation,
                    result.Detail ?? "The server rejected this change as invalid.",
                    report,
                    cancellationToken).ConfigureAwait(false);
                return false;

            default:
                await outbox.RecordFailureAsync(
                    operation.Sequence,
                    $"Unexpected push status {result.Status}.",
                    cancellationToken).ConfigureAwait(false);
                return false;
        }
    }

    /// <summary>Records an accepted operation and clears it from the outbox.</summary>
    private async Task AcceptAsync(
        Guid vaultId,
        PendingOperation operation,
        SyncPushResult result,
        CancellationToken cancellationToken)
    {
        var expected = SyncVersions.NextVersion(operation.ExpectedVersion);
        var version = result.Version ?? expected;
        var isDelete = operation.Operation == SyncOperation.Delete;

        // The payload was sealed at the version this client predicted, and the AAD binds that version.
        // If the server assigned a different one — which its own version check should make impossible —
        // storing the payload would leave a mirror row that never decrypts. Skip the write; the pull at
        // the end of the pass brings the authoritative row.
        var canMirror = isDelete || version == expected;

        if (canMirror)
        {
            await items.SaveAsync(
                new StoredItem(
                    vaultId,
                    operation.EntityType,
                    operation.EntityId,
                    version,
                    result.ChangeSequence ?? 0,
                    isDelete ? null : operation.Payload,
                    isDelete ? null : operation.Fields,
                    isDelete,
                    clock.GetUtcNow()),
                cancellationToken).ConfigureAwait(false);
        }

        await outbox.CompleteAsync(operation.Sequence, cancellationToken).ConfigureAwait(false);
    }

    private async Task<bool> ResolveAsync(
        Guid vaultId,
        PendingOperation operation,
        SyncPushResult result,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        if (operation.Attempts >= options.MaxAttemptsBeforeParking)
        {
            await RejectAsync(
                vaultId,
                operation,
                string.Create(
                    CultureInfo.InvariantCulture,
                    $"Could not be reconciled after {operation.Attempts} attempts."),
                report,
                cancellationToken).ConfigureAwait(false);

            return false;
        }

        if (!reconcilers.TryGetValue(operation.EntityType, out var reconciler))
        {
            // Only reachable if something queued a type this build does not synchronise, which the
            // repositories cannot do. Parked rather than dropped, so the change is visible to a user
            // instead of retried for ever against a path that cannot handle it.
            await RejectAsync(
                vaultId,
                operation,
                $"This version of DodoSSH cannot reconcile items of type {operation.EntityType}.",
                report,
                cancellationToken).ConfigureAwait(false);

            return false;
        }

        if (result.ServerEntity is null)
        {
            // The version check failed but the server has no such row. Re-offer it as a create.
            return await RetryAsCreateAsync(vaultId, reconciler, operation, report, cancellationToken)
                .ConfigureAwait(false);
        }

        await ItemMirror.WriteAsync(items, vaultId, result.ServerEntity, cancellationToken)
            .ConfigureAwait(false);

        await reconciler
            .ReconcileAsync(vaultId, result.ServerEntity, operation, report, cancellationToken)
            .ConfigureAwait(false);

        return true;
    }

    private async Task<bool> RetryAsCreateAsync(
        Guid vaultId,
        IItemReconciler reconciler,
        PendingOperation operation,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        if (operation.Operation == SyncOperation.Delete)
        {
            // Nothing there to delete, so the intent is already satisfied.
            await outbox.CompleteAsync(operation.Sequence, cancellationToken).ConfigureAwait(false);
            return false;
        }

        // Re-sealing needs the item's own cipher, so the reconciler does it. A reason back means the
        // change can never be sent, not that it should be retried.
        var failure = await reconciler
            .ReofferAsCreateAsync(vaultId, operation, cancellationToken)
            .ConfigureAwait(false);

        if (failure is null)
        {
            return true;
        }

        await RejectAsync(vaultId, operation, failure, report, cancellationToken).ConfigureAwait(false);

        return false;
    }

    /// <summary>Parks an operation the server will never accept, and says why.</summary>
    private async Task RejectAsync(
        Guid vaultId,
        PendingOperation operation,
        string reason,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        await outbox.ParkAsync(operation.Sequence, reason, cancellationToken).ConfigureAwait(false);

        await conflicts.RecordAsync(
            vaultId,
            operation.EntityType,
            operation.EntityId,
            ConflictKind.Rejected,
            ConflictDetails.Encode(reason),
            cancellationToken).ConfigureAwait(false);

        report.Parked++;
    }
}