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

using System.Globalization;
using System.Runtime.InteropServices;
using System.Security.Cryptography;
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 ReadStateAsync(vaultId, report, 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;
            }

            await ApplyPageAsync(vaultId, response.Changes, report, cancellationToken).ConfigureAwait(false);
            CountPulled(response.Changes, report);

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

            state = Record(vaultId, state, response, report);
            state = await CarrySweepAsync(vaultId, state, response, report, cancellationToken).ConfigureAwait(false);
            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;
            }
        }
    }

    private static void CountPulled(IReadOnlyList<SyncChange> changes, SyncReportBuilder report)
    {
        foreach (var change in changes)
        {
            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++;
            }
        }
    }

    /// <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;

    /// <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>
    private static bool WasTheCursorRefused(DodoSshApiException exception, string? cursor) =>
        !string.IsNullOrEmpty(cursor)
        && string.Equals(exception.Code, ProblemCodes.InvalidCursor, StringComparison.Ordinal);

    /// <summary>Where the pull should carry on from, which is the start when the types it reads have changed.</summary>
    /// <remarks>
    /// A cursor read under a different list of types is past changes of the types it did not ask for, so an
    /// upgrade that adds one — ObjectStore, say — would never receive the items of that type made
    /// before it. Starting over once is the only way to see them; applying is idempotent, which is what makes
    /// the restart for a refused cursor safe as well. A state from before the list was kept has none and
    /// starts over once for the same reason: it may already have missed some.
    /// </remarks>
    private async Task<StoredSyncState> ReadStateAsync(
        Guid vaultId,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        var state = await syncState.ReadAsync(vaultId, cancellationToken).ConfigureAwait(false);

        if (state.Cursor is not null && !string.Equals(state.PulledTypes, PulledTypes, StringComparison.Ordinal))
        {
            return await StartOverAsync(state, report, cancellationToken).ConfigureAwait(false);
        }

        // A pull from nothing that is not a restart — the vault's first, or a cache reset — sweeps as well.
        // There is usually nothing to remove; the one time there is, it is the same case.
        return state is { Cursor: null, SweepMark: null }
            ? await SweepingAsync(state, cancellationToken).ConfigureAwait(false)
            : state;
    }

    /// <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>
    /// A replay does not carry a deletion the server has collected — <c>DodoOptions.TombstoneRetentionDays</c>
    /// — and that is the commonest refusal of all: a machine away longer than the retention. The rows it no
    /// longer hears about are the ones the replay does not mark, and are swept when it ends; see
    /// <see cref="CarrySweepAsync"/>.
    /// </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 = await SweepingAsync(state with { Cursor = null }, cancellationToken).ConfigureAwait(false);

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

        report.ResyncedFromStart = true;

        return restarted;
    }

    /// <summary>The state with a sweep begun: a new mark, and the bound set from what the cache holds now.</summary>
    /// <remarks>
    /// The mark is random rather than the time, because a sweep must never share one with the last: rows that
    /// one marked would count as seen by this one. A clock can stand still, or go back.
    /// </remarks>
    private async Task<StoredSyncState> SweepingAsync(StoredSyncState state, CancellationToken cancellationToken) =>
        state with
        {
            SweepMark = BitConverter.ToInt64(RandomNumberGenerator.GetBytes(sizeof(long))),
            SweepSeenUpTo = await items.HighestSequenceAsync(state.VaultId, cancellationToken).ConfigureAwait(false),
        };

    /// <summary>
    /// Marks what a pull from nothing has read, and at its end removes what it never did.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The server forgets a deletion once it is past its retention. A machine that was away longer is
    /// refused its cursor and reads everything again — and everything again is the items that still exist,
    /// with nothing said about the ones that went. Without this their rows stayed here for good, and the
    /// next edit of one put it back on the server for everybody.
    /// </para>
    /// <para>
    /// A pull from nothing can take more than one pass, so the mark is in the stored state, not in memory.
    /// Every pull from nothing sweeps — the first one ever as well, where there is nothing to remove. What
    /// is kept, and why, is <see cref="ItemStore.SweepAsync"/>'s to say.
    /// </para>
    /// </remarks>
    private async Task<StoredSyncState> CarrySweepAsync(
        Guid vaultId,
        StoredSyncState state,
        SyncPullResponse response,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        if (state.SweepMark is not { } mark)
        {
            return state;
        }

        var read = response.Changes.Where(change => reconcilers.ContainsKey(change.EntityType)).ToList();

        await items.MarkSeenAsync(
            vaultId,
            mark,
            read.Select(change => (change.EntityType, change.EntityId)).ToList(),
            cancellationToken).ConfigureAwait(false);

        var seenUpTo = read.Aggregate(state.SweepSeenUpTo, (highest, change) => Math.Max(highest, change.ChangeSequence));

        if (response.HasMore)
        {
            return state with { SweepSeenUpTo = seenUpTo };
        }

        report.Pruned += await items
            .SweepAsync(vaultId, ItemKinds.SyncedTypes, mark, seenUpTo, cancellationToken)
            .ConfigureAwait(false);

        return state with { SweepMark = null, SweepSeenUpTo = 0 };
    }

    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,
            PulledTypes = PulledTypes,
        };
    }

    /// <summary>The types this build pulls, in a form a stored state can be compared against.</summary>
    private static string PulledTypes { get; } =
        string.Join(',', ItemKinds.SyncedTypes.Select(type => (int)type).Order());

    /// <summary>Applies one page of pulled changes, in order, with as few writes as that allows.</summary>
    /// <remarks>
    /// <para>
    /// Each change used to cost a write to mirror it and a read to look for a local edit of the same item —
    /// two or more SQLite transactions apiece, which on a first sync of a large keychain was most of the time
    /// the sync took. The outbox is read once per page instead, and every run of changes with no local edit
    /// waiting is mirrored in a single write.
    /// </para>
    /// <para>
    /// ◆ A change that does have an edit waiting goes through <see cref="ApplyAsync"/> exactly as before, with
    /// the run in front of it written first: the reconciler reads the mirror and the outbox as they stand at
    /// that change, and both are what the one-by-one loop would have left them as. The one-read snapshot only
    /// decides which path a change takes; the path that matters re-reads the outbox for itself.
    /// </para>
    /// </remarks>
    private async Task ApplyPageAsync(
        Guid vaultId,
        IReadOnlyList<SyncChange> changes,
        SyncReportBuilder report,
        CancellationToken cancellationToken)
    {
        if (changes.Count == 0)
        {
            return;
        }

        var waiting = (await outbox.ListAllAsync(vaultId, cancellationToken).ConfigureAwait(false))
            .Select(pending => (pending.EntityType, pending.EntityId))
            .ToHashSet();

        var run = new List<SyncChange>();

        foreach (var change in changes)
        {
            if (!reconcilers.ContainsKey(change.EntityType))
            {
                // See ApplyAsync: a type this build cannot sync is neither mirrored nor reconciled.
                continue;
            }

            if (!waiting.Contains((change.EntityType, change.EntityId)))
            {
                run.Add(change);
                continue;
            }

            await ItemMirror.WriteManyAsync(items, vaultId, run, cancellationToken).ConfigureAwait(false);
            run.Clear();

            await ApplyAsync(vaultId, change, report, cancellationToken).ConfigureAwait(false);
        }

        await ItemMirror.WriteManyAsync(items, vaultId, run, cancellationToken).ConfigureAwait(false);
    }

    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,
                operation.Operation,
                operation.ExpectedVersion,
                resealed,
                operation.Fields,
                operation.Ancestor,
                cancellationToken)
            .ConfigureAwait(false);

        return resealed;
    }

    /// <summary>Whether a refused operation is a delete of an item the server never had.</summary>
    /// <remarks>
    /// A delete queued over a create that never reached the server carries no expected version, and the
    /// server's only reason to refuse one of those is that it has no such item — which is the state the delete
    /// asked for, so it is done. It used to be parked as a rejection: a "conflict" about an item nobody else
    /// had ever seen, which the user could neither resolve nor dismiss. A create that did land is no exception
    /// to this — the server has the item then, and applies the delete.
    /// </remarks>
    private static bool DeletesSomethingNeverCreated(PendingOperation operation) =>
        operation is { Operation: SyncOperation.Delete, ExpectedVersion: null };

    /// <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 when DeletesSomethingNeverCreated(operation):
                await outbox.CompleteAsync(operation, 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);
        }

        var completed = await outbox.CompleteAsync(operation, cancellationToken).ConfigureAwait(false);

        if (!completed && canMirror && !isDelete)
        {
            await RebaseOntoAcceptedAsync(vaultId, operation, version, cancellationToken).ConfigureAwait(false);
        }
    }

    /// <summary>
    /// Moves an edit made while <paramref name="accepted"/> was in flight onto the version that push produced.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The user saved again between the push being read and its answer arriving, so the row now holds a
    /// later state branched from the same base as the one the server just took. Left alone it is still the
    /// user's edit and still goes — but it goes expecting the old version, comes back <c>Conflict</c>, and is
    /// merged against an ancestor that predates the save the server already has, which makes every field
    /// changed twice look like a disagreement with somebody else. Its true parent is the state just accepted,
    /// so that becomes its ancestor and expected version, and its payload is re-sealed at the version it
    /// will now produce, because the AAD binds that.
    /// </para>
    /// <para>
    /// Only for a row still branched from the same base. One that is not has been revised by something else
    /// since, and the ordinary conflict path is the one that knows what to do with it; so is a row whose
    /// payload this session cannot re-seal.
    /// </para>
    /// </remarks>
    private async Task RebaseOntoAcceptedAsync(
        Guid vaultId,
        PendingOperation accepted,
        int version,
        CancellationToken cancellationToken)
    {
        var later = await outbox
            .FindAsync(vaultId, accepted.EntityType, accepted.EntityId, cancellationToken)
            .ConfigureAwait(false);

        if (later is null || later.ExpectedVersion != accepted.ExpectedVersion || accepted.Payload is null)
        {
            return;
        }

        EncryptedPayload? payload = null;

        if (later.Payload is { } queued)
        {
            payload = PayloadReseal.TryReseal(
                keyring,
                vaultId,
                later.EntityType,
                later.EntityId,
                queued,
                openAtVersion: SyncVersions.NextVersion(later.ExpectedVersion),
                sealAtVersion: SyncVersions.NextVersion(version));

            if (payload is null)
            {
                return;
            }
        }

        await outbox.ReviseAsync(
                later,
                later.Operation,
                version,
                payload,
                later.Fields,
                new StoredAncestor(version, accepted.Payload, accepted.Fields),
                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, 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)
        {
            // An edit of a version this machine once saw means the server had the item and has since let go
            // of it: deleted elsewhere, and the deletion collected past its retention before this machine
            // heard of it. The edit is the user's work and goes back, but not silently — somebody deleted
            // this on purpose. A create has no expected version and is the ordinary case.
            if (operation.ExpectedVersion is not null)
            {
                await conflicts.RecordAsync(
                    vaultId,
                    operation.EntityType,
                    operation.EntityId,
                    ConflictKind.RecreatedAfterCollection,
                    ConflictDetails.Encode(
                        "This was deleted elsewhere while this machine had unsaved changes to it, long enough ago "
                        + "that the server no longer remembered the deletion. The changes were kept and the item "
                        + "was put back."),
                    cancellationToken).ConfigureAwait(false);

                report.Recreated++;
            }

            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, reason, cancellationToken).ConfigureAwait(false);

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

        report.Parked++;
    }
}