DodoSSH/src/DodoSSH.Api/Features/Sync/SyncService.cs

using DodoSSH.Api.Setup;
using DodoSSH.Contracts;
using DodoSSH.Domain;
using DodoSSH.Domain.Sync;
using DodoSSH.Infrastructure;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Options;
using Npgsql;

namespace DodoSSH.Api.Features.Sync;

/// <summary>Applies and reads a vault's change log.</summary>
internal sealed class SyncService(
    DodoDbContext database,
    IOptions<SyncOptions> syncOptions,
    ICursorKeyProvider cursorKeys,
    TimeProvider clock,
    ILogger<SyncService> logger)
{
    private readonly SyncOptions options = syncOptions.Value;

    /// <summary>Reads changes after a cursor.</summary>
    internal async Task<SyncPullResponse> PullAsync(
        Vault vault,
        SyncPullRequest request,
        CancellationToken cancellationToken)
    {
        var afterSequence = 0L;
        var issuedUnder = 0L;
        if (!string.IsNullOrEmpty(request.Cursor)
            && !SyncCursor.TryDecode(cursorKeys.Key, request.Cursor, vault.Id, out afterSequence, out issuedUnder))
        {
            throw new InvalidCursorException();
        }

        // Read before the changes, for the cursor this read hands out; see the check after it.
        var horizonBefore = await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false);

        var (changes, hasMore) = await ReadPageAsync(vault.Id, afterSequence, request, cancellationToken)
            .ConfigureAwait(false);

        // ◆ After the read, not before it. The collector deletes a tombstone's entries and raises the horizon
        // in one transaction, so a read that already missed an entry is followed by a horizon that says so;
        // checked first, the two could straddle a collection and the deletion would go unnoticed for good.
        //
        // A cursor behind the horizon is refused only if it was issued before it. One issued since belongs to
        // a read from the beginning that started after the collection, and its early pages are behind the
        // horizon because old, untouched items are; refusing those would send that read back to the start
        // for ever. See SyncCursor.
        var horizon = await HorizonAsync(vault.Id, cancellationToken).ConfigureAwait(false);

        if (afterSequence > 0 && afterSequence < horizon && issuedUnder < horizon)
        {
            throw new InvalidCursorException();
        }

        var hydrated = await HydrateAsync(vault, changes, cancellationToken).ConfigureAwait(false);

        // When nothing came back the cursor must not move, or a concurrent write landing between
        // this read and the next would be skipped forever.
        var nextSequence = changes.Count > 0 ? changes[^1].Sequence : afterSequence;

        return new SyncPullResponse(
            Changes: hydrated,
            // The horizon from before the read, never the one after: a collection landing between the two
            // could have taken an entry this page would otherwise have carried, and a cursor claiming the
            // later horizon would vouch for a client that never heard of it. The earlier one costs at most
            // a restart.
            NextCursor: SyncCursor.Encode(cursorKeys.Key, vault.Id, nextSequence, horizonBefore),
            HasMore: hasMore,
            ServerTime: clock.GetUtcNow(),
            CurrentKeyGeneration: (uint)vault.KeyGeneration);
    }

    /// <summary>The changes after a position, one page of them, and whether there are more.</summary>
    private async Task<(List<VaultChange> Changes, bool HasMore)> ReadPageAsync(
        Guid vaultId,
        long afterSequence,
        SyncPullRequest request,
        CancellationToken cancellationToken)
    {
        var limit = Math.Clamp(
            request.Limit ?? options.DefaultPullLimit,
            1,
            options.MaxPullLimit);

        var types = request.EntityTypes is { Count: > 0 }
            ? request.EntityTypes.Select(ToDomain).ToArray()
            : null;

        // One extra row, so "is there more" needs no second query.
        var query = database.VaultChanges
            .Where(c => c.VaultId == vaultId && c.Sequence > afterSequence);

        if (types is not null)
        {
            query = query.Where(c => types.Contains(c.EntityType));
        }

        var changes = await query
            .OrderBy(c => c.Sequence)
            .Take(limit + 1)
            .ToListAsync(cancellationToken)
            .ConfigureAwait(false);

        var hasMore = changes.Count > limit;
        if (hasMore)
        {
            changes.RemoveAt(changes.Count - 1);
        }

        return (changes, hasMore);
    }

    /// <summary>
    /// Applies a batch of operations.
    /// </summary>
    /// <remarks>
    /// <para>
    /// The whole batch runs in one transaction whose <em>first</em> statement takes a per-vault
    /// advisory lock. That is load-bearing: <c>bigserial</c> hands out sequence values before
    /// commit, so without serialising writers per vault, transaction A can take sequence 5 while B
    /// takes 6 and commits first. A reader advancing its cursor to 6 then misses 5 permanently.
    /// See ADR 0003.
    /// </para>
    /// <para>
    /// Individual operations that conflict are skipped rather than aborting the batch, so one stale
    /// item cannot block everything else a client queued while offline.
    /// </para>
    /// </remarks>
    internal async Task<SyncPushResponse> PushAsync(
        Vault vault,
        Guid actorUserId,
        SyncPushRequest request,
        CancellationToken cancellationToken)
    {
        ValidateBatchLimits(request);

        return await database.ExecuteFreshAsync([vault], async () =>
        {
            var transaction = await database.Database
                .BeginTransactionAsync(cancellationToken)
                .ConfigureAwait(false);
            await using var _ = transaction.ConfigureAwait(false);

            await AcquireVaultLockAsync(vault.Id, cancellationToken).ConfigureAwait(false);

            var results = new List<SyncPushResult>(request.Operations.Count);
            var highestSequence = 0L;

            foreach (var operation in request.Operations)
            {
                var result = await ApplyAsync(vault, actorUserId, operation, cancellationToken)
                    .ConfigureAwait(false);

                results.Add(result);

                if (result.ChangeSequence is { } sequence && sequence > highestSequence)
                {
                    highestSequence = sequence;
                }
            }

            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);

            // If nothing was applied, report the log head so the client does not rewind.
            if (highestSequence == 0)
            {
                highestSequence = await CurrentHeadAsync(vault.Id, cancellationToken)
                    .ConfigureAwait(false);
            }

            return new SyncPushResponse(
                Results: results,
                Cursor: SyncCursor.Encode(cursorKeys.Key, vault.Id, highestSequence));
        }).ConfigureAwait(false);
    }

    private void ValidateBatchLimits(SyncPushRequest request)
    {
        if (request.Operations.Count == 0)
        {
            throw new PushBatchInvalidException("A push must contain at least one operation.");
        }

        if (request.Operations.Count > options.MaxOperationsPerPush)
        {
            throw new PushBatchTooLargeException(
                $"A push may contain at most {options.MaxOperationsPerPush} operations; "
                + $"{request.Operations.Count} were supplied.");
        }

        long total = 0;
        foreach (var operation in request.Operations)
        {
            var length = operation.Payload?.Envelope.Length ?? 0;

            if (length > options.MaxItemPayloadBytes)
            {
                throw new PushBatchTooLargeException(
                    $"Item {operation.EntityId} payload is {length} bytes; the limit is "
                    + $"{options.MaxItemPayloadBytes}.");
            }

            total += length;
        }

        if (total > options.MaxPayloadBytes)
        {
            throw new PushBatchTooLargeException(
                $"Total push payload is {total} bytes; the limit is {options.MaxPayloadBytes}.");
        }
    }

    /// <summary>
    /// Serialises writers to one vault for the life of the transaction.
    /// </summary>
    /// <remarks>
    /// <c>hashtextextended</c> over the vault id gives a stable 64-bit lock key. Contention is
    /// per-vault, and a push is a single transaction anyway, so the cost is negligible next to the
    /// silent sync corruption it prevents. Requires <c>Multiplexing=false</c> on the connection,
    /// which is the Npgsql default — enabling it would break the lock's session affinity.
    /// </remarks>
    private Task<int> AcquireVaultLockAsync(Guid vaultId, CancellationToken cancellationToken) =>
        database.Database.ExecuteSqlAsync(
            $"SELECT pg_advisory_xact_lock(hashtextextended({vaultId.ToString()}, 0))",
            cancellationToken);

    private async Task<SyncPushResult> ApplyAsync(
        Vault vault,
        Guid actorUserId,
        SyncPushOperation operation,
        CancellationToken cancellationToken)
    {
        if (ItemKinds.For(operation.EntityType) is not { } kind)
        {
            // Reserved in the contract but not synced here yet. A newer client gets a precise
            // per-operation answer rather than a whole-batch failure.
            return Invalid(operation, $"Entity type {operation.EntityType} is not yet supported.");
        }

        if (operation.Operation is not (SyncOperation.Upsert or SyncOperation.Delete))
        {
            return Invalid(operation, "Operation must be Upsert or Delete.");
        }

        // Exactly-once at operation granularity. A batch-level idempotency key alone would
        // double-apply the operations that did land when a client retries a partially-overlapping
        // batch after a timeout.
        var receipt = await database.SyncOperationReceipts
            .SingleOrDefaultAsync(r => r.OperationId == operation.OperationId, cancellationToken)
            .ConfigureAwait(false);

        if (receipt is not null)
        {
            return new SyncPushResult(
                operation.OperationId,
                SyncOperationStatus.Duplicate,
                receipt.ResultVersion,
                receipt.AppliedChangeSequence,
                ServerEntity: null,
                Detail: null);
        }

        var item = await kind.FindAsync(database, operation.EntityId, cancellationToken)
            .ConfigureAwait(false);

        // An id that exists in another vault must not be addressable from this one, and must not
        // reveal that it exists elsewhere.
        if (item is not null && item.VaultId != vault.Id)
        {
            return new SyncPushResult(
                operation.OperationId,
                SyncOperationStatus.Forbidden,
                Version: null,
                ChangeSequence: null,
                ServerEntity: null,
                Detail: null);
        }

        return operation.Operation == SyncOperation.Delete
            ? await ApplyDeleteAsync(vault, kind, actorUserId, operation, item, cancellationToken)
                .ConfigureAwait(false)
            : await ApplyUpsertAsync(vault, kind, actorUserId, operation, item, cancellationToken)
                .ConfigureAwait(false);
    }

    private async Task<SyncPushResult> ApplyUpsertAsync(
        Vault vault,
        IItemKind kind,
        Guid actorUserId,
        SyncPushOperation operation,
        IVaultItem? existing,
        CancellationToken cancellationToken)
    {
        if (operation.Payload is null)
        {
            return Invalid(operation, "An upsert requires a payload.");
        }

        if (!ValidatePayload(operation.Payload, out var payloadError))
        {
            return Invalid(operation, payloadError);
        }

        var fields = operation.PlaintextFields ?? new SyncPlaintextFields();

        if (!kind.ValidateFields(fields, out var fieldError))
        {
            return Invalid(operation, fieldError);
        }

        var now = clock.GetUtcNow();

        if (existing is null || existing.DeletedAtUtc is not null)
        {
            return await CreateAsync(
                vault, kind, actorUserId, operation, existing, fields, now, cancellationToken)
                .ConfigureAwait(false);
        }

        if (operation.ExpectedVersion != existing.Version)
        {
            return await ConflictAsync(vault, kind, operation, existing, cancellationToken)
                .ConfigureAwait(false);
        }

        existing.Version++;
        ApplyFields(kind, existing, operation.Payload, fields, actorUserId, now);

        return await RecordAsync(
            vault, kind, actorUserId, operation, existing, ChangeOperation.Upsert, cancellationToken)
            .ConfigureAwait(false);
    }

    /// <summary>Creates a new item, or rejects an upsert that cannot become one.</summary>
    private async Task<SyncPushResult> CreateAsync(
        Vault vault,
        IItemKind kind,
        Guid actorUserId,
        SyncPushOperation operation,
        IVaultItem? existing,
        SyncPlaintextFields fields,
        DateTimeOffset now,
        CancellationToken cancellationToken)
    {
        // A tombstone wins over a late upsert. The client is told, so it can resurrect the item
        // deliberately under a new id rather than silently undoing someone else's delete.
        if (existing?.DeletedAtUtc is not null)
        {
            return await ConflictAsync(vault, kind, operation, existing, cancellationToken)
                .ConfigureAwait(false);
        }

        if (operation.ExpectedVersion is not null)
        {
            // The client believes it is updating something that does not exist here.
            return await ConflictAsync(vault, kind, operation, existing: null, cancellationToken)
                .ConfigureAwait(false);
        }

        var created = kind.Add(database, operation.EntityId, vault.Id);

        created.Version = 1;
        created.CreatedAtUtc = now;
        created.CreatedByUserId = actorUserId;

        ApplyFields(kind, created, operation.Payload!, fields, actorUserId, now);

        return await RecordAsync(
            vault, kind, actorUserId, operation, created, ChangeOperation.Upsert, cancellationToken)
            .ConfigureAwait(false);
    }

    private async Task<SyncPushResult> ApplyDeleteAsync(
        Vault vault,
        IItemKind kind,
        Guid actorUserId,
        SyncPushOperation operation,
        IVaultItem? existing,
        CancellationToken cancellationToken)
    {
        if (existing is null)
        {
            return Invalid(operation, "Cannot delete an item that does not exist.");
        }

        if (existing.DeletedAtUtc is not null)
        {
            // Already a tombstone. Idempotent rather than an error: a client retrying a delete it
            // is unsure about should not have to distinguish these.
            return new SyncPushResult(
                operation.OperationId,
                SyncOperationStatus.Applied,
                existing.Version,
                existing.ChangeSequence,
                ServerEntity: null,
                Detail: null);
        }

        if (operation.ExpectedVersion is not null && operation.ExpectedVersion != existing.Version)
        {
            return await ConflictAsync(vault, kind, operation, existing, cancellationToken)
                .ConfigureAwait(false);
        }

        var now = clock.GetUtcNow();
        existing.Version++;
        existing.DeletedAtUtc = now;
        existing.UpdatedAtUtc = now;
        existing.UpdatedByUserId = actorUserId;

        kind.ClearFieldsOnDelete(existing);

        return await RecordAsync(
            vault, kind, actorUserId, operation, existing, ChangeOperation.Delete, cancellationToken)
            .ConfigureAwait(false);
    }

    /// <remarks>
    /// The ciphertext and the bookkeeping are identical for every item type; only the plaintext columns
    /// differ, and those are the type's own business. Splitting it this way is what stops a new item type
    /// from having to restate — and possibly misstate — how a payload is stored.
    /// </remarks>
    private static void ApplyFields(
        IItemKind kind,
        IVaultItem item,
        EncryptedPayload payload,
        SyncPlaintextFields fields,
        Guid actorUserId,
        DateTimeOffset now)
    {
        item.Payload = payload.Envelope;
        item.DataKeyWrap = payload.WrappedDataKey;
        item.ContentKeyId = payload.DataKeyId;
        item.KeyGeneration = (int)payload.KeyGeneration;
        item.PayloadAadVersion = payload.AadVersion;
        item.DeletedAtUtc = null;
        item.UpdatedAtUtc = now;
        item.UpdatedByUserId = actorUserId;

        kind.ApplyFields(item, fields);
    }

    /// <summary>
    /// Rejects a payload missing its data key.
    /// </summary>
    /// <remarks>
    /// The server cannot read any of these bytes, so this is a structural check and nothing more.
    /// It is still worth making: docs/crypto.md §3 requires a per-item data key, the payload's AAD
    /// binds <see cref="EncryptedPayload.DataKeyId"/>, and a row stored without a wrap is a row no
    /// client will ever be able to open. Better to refuse it here — where the client is told which
    /// operation was wrong — than to store an item that silently reads as corrupt forever.
    /// </remarks>
    private static bool ValidatePayload(EncryptedPayload payload, out string error)
    {
        error = string.Empty;

        if (payload.Envelope.Length == 0)
        {
            error = "A payload envelope cannot be empty.";
            return false;
        }

        if (payload.WrappedDataKey.Length == 0)
        {
            error = "A payload requires its data key, wrapped under the vault key.";
            return false;
        }

        if (payload.DataKeyId == Guid.Empty)
        {
            error = "A payload requires a data key identifier; it is part of the payload's AAD.";
            return false;
        }

        return true;
    }

    private async Task<SyncPushResult> RecordAsync(
        Vault vault,
        IItemKind kind,
        Guid actorUserId,
        SyncPushOperation operation,
        IVaultItem item,
        ChangeOperation changeOperation,
        CancellationToken cancellationToken)
    {
        var change = new VaultChange
        {
            VaultId = vault.Id,
            EntityType = kind.ChangeType,
            EntityId = item.Id,
            Operation = changeOperation,
            Revision = item.Version,
            ActorUserId = actorUserId,
            OccurredAtUtc = clock.GetUtcNow(),
        };

        database.VaultChanges.Add(change);

        // Saved before the denormalised pointer is set, because the sequence is assigned by the
        // database on insert.
        await database.SaveChangesAsync(cancellationToken).ConfigureAwait(false);

        item.ChangeSequence = change.Sequence;

        database.SyncOperationReceipts.Add(new SyncOperationReceipt
        {
            OperationId = operation.OperationId,
            VaultId = vault.Id,
            AppliedChangeSequence = change.Sequence,
            ResultVersion = item.Version,
            CreatedAtUtc = clock.GetUtcNow(),
        });

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

        return new SyncPushResult(
            operation.OperationId,
            SyncOperationStatus.Applied,
            item.Version,
            change.Sequence,
            ServerEntity: null,
            Detail: null);
    }

    private async Task<SyncPushResult> ConflictAsync(
        Vault vault,
        IItemKind kind,
        SyncPushOperation operation,
        IVaultItem? existing,
        CancellationToken cancellationToken)
    {
        // The server cannot merge ciphertext, so it hands back its current state and the client
        // performs a three-way merge against its retained common ancestor. Never last-writer-wins.
        Contracts.SyncChange? serverEntity = null;

        if (existing is not null)
        {
            var hydrated = await HydrateAsync(
                vault,
                [new VaultChange
                {
                    Sequence = existing.ChangeSequence,
                    VaultId = existing.VaultId,
                    EntityType = kind.ChangeType,
                    EntityId = existing.Id,
                    Operation = existing.DeletedAtUtc is null
                        ? ChangeOperation.Upsert
                        : ChangeOperation.Delete,
                    Revision = existing.Version,
                    OccurredAtUtc = existing.UpdatedAtUtc,
                }],
                cancellationToken).ConfigureAwait(false);

            serverEntity = hydrated.Count > 0 ? hydrated[0] : null;
        }

        SyncLog.PushConflict(logger, operation.EntityId, operation.ExpectedVersion, existing?.Version);

        return new SyncPushResult(
            operation.OperationId,
            SyncOperationStatus.Conflict,
            existing?.Version,
            existing?.ChangeSequence,
            serverEntity,
            Detail: null);
    }

    /// <summary>Attaches current row state to change-log entries.</summary>
    /// <remarks>
    /// <para>
    /// A page of changes can mix item types, so rows are loaded per type — one query each, not one per
    /// change — and then matched back in the log's order. Ordering is the log's, never the load's: a pull
    /// that returned changes grouped by type would hand the client a sequence its cursor cannot resume from.
    /// </para>
    /// <para>
    /// ◆ <b>An entry an item has since moved past is left out.</b> The log keeps every revision, but the
    /// table keeps only the latest ciphertext — so hydrating an old entry paired that entry's revision number
    /// with the newest payload, and the payload's AAD binds the revision it was sealed at. A client opened it
    /// at the old number, failed, and parked its own unsent edit to that item as undecryptable. The later
    /// entry is in this page or a later one, since the item's <c>ChangeSequence</c> is that entry's sequence;
    /// it carries the same payload under the right number, so nothing is lost by sending it once. Leaving
    /// the old entries out also stops a fresh device downloading an item once per edit it ever had. The
    /// cursor is computed from the unfiltered page by the caller, so it still moves past them.
    /// </para>
    /// </remarks>
    private async Task<List<Contracts.SyncChange>> HydrateAsync(
        Vault vault,
        List<VaultChange> changes,
        CancellationToken cancellationToken)
    {
        var rows = new Dictionary<(ChangeEntityType Type, Guid Id), IVaultItem>();
        var kinds = new Dictionary<ChangeEntityType, IItemKind>();

        foreach (var group in changes.GroupBy(change => change.EntityType))
        {
            if (KindForChange(group.Key) is not { } kind)
            {
                // A change this server wrote under a type it no longer synchronises. Nothing sensible to
                // hydrate, and dropping the entry would leave a gap in the client's cursor — so it is
                // emitted below with no payload, which a client already handles as a tombstone.
                continue;
            }

            kinds[group.Key] = kind;

            var ids = group.Select(change => change.EntityId).Distinct().ToArray();

            var loaded = await kind.LoadAsync(database, vault.Id, ids, cancellationToken)
                .ConfigureAwait(false);

            foreach (var (id, item) in loaded)
            {
                rows[(group.Key, id)] = item;
            }
        }

        return [.. changes
            .Where(change => rows.GetValueOrDefault((change.EntityType, change.EntityId)) is not { } row
                || row.ChangeSequence <= change.Sequence)
            .Select(change => Describe(
                change,
                kinds.GetValueOrDefault(change.EntityType),
                rows.GetValueOrDefault((change.EntityType, change.EntityId))))];
    }

    /// <summary>Turns one change-log entry and its row into what a client receives.</summary>
    private static Contracts.SyncChange Describe(VaultChange change, IItemKind? kind, IVaultItem? item)
    {
        // A delete carries no payload: there is nothing left to decrypt, and shipping the pre-delete
        // ciphertext would undermine the point of the tombstone.
        var isDelete = change.Operation == ChangeOperation.Delete || item?.DeletedAtUtc is not null;
        var opaque = isDelete || item is null;

        return new Contracts.SyncChange(
            EntityType: kind?.WireType ?? ToWire(change.EntityType),
            EntityId: change.EntityId,
            Operation: isDelete ? SyncOperation.Delete : SyncOperation.Upsert,
            Version: change.Revision,
            ChangeSequence: change.Sequence,
            Payload: opaque
                ? null
                : new EncryptedPayload(
                    item!.Payload,
                    // Non-null for every row a push can create: ValidatePayload refuses an operation
                    // without them. The columns stay nullable because they are also the seam for M5's
                    // per-item grants.
                    item.DataKeyWrap ?? [],
                    item.ContentKeyId ?? Guid.Empty,
                    (uint)item.KeyGeneration,
                    (byte)item.PayloadAadVersion),
            PlaintextFields: opaque || kind is null ? null : kind.Hydrate(item!),
            UpdatedAt: change.OccurredAtUtc);
    }

    /// <summary>The kind behind a change-log entry, or null for a type this server does not synchronise.</summary>
    private static IItemKind? KindForChange(ChangeEntityType type) => ItemKinds.For(ToWire(type));

    /// <remarks>
    /// Never below the vault's horizon: collecting a tombstone can take the newest entries with it, and a
    /// head reported lower than a cursor the client already holds would rewind it under the horizon and
    /// cost a full resync for nothing.
    /// </remarks>
    private async Task<long> CurrentHeadAsync(Guid vaultId, CancellationToken cancellationToken)
    {
        var head = await database.VaultChanges
            .Where(c => c.VaultId == vaultId)
            .OrderByDescending(c => c.Sequence)
            .Select(c => c.Sequence)
            .FirstOrDefaultAsync(cancellationToken)
            .ConfigureAwait(false);

        return Math.Max(head, await HorizonAsync(vaultId, cancellationToken).ConfigureAwait(false));
    }

    /// <summary>The vault's horizon as the database holds it now, not as the request loaded it.</summary>
    private Task<long> HorizonAsync(Guid vaultId, CancellationToken cancellationToken) =>
        database.Vaults
            .Where(v => v.Id == vaultId)
            .Select(v => v.SyncHorizon)
            .SingleAsync(cancellationToken);

    private static SyncPushResult Invalid(SyncPushOperation operation, string detail) =>
        new(
            operation.OperationId,
            SyncOperationStatus.Invalid,
            Version: null,
            ChangeSequence: null,
            ServerEntity: null,
            Detail: detail);

    /// <remarks>
    /// The two enums are numerically aligned by hand — and they do not even use the same member names for
    /// the same value, <c>Host</c> against <c>SshHost</c> — so nothing in the type system defends this cast.
    /// <c>EntityTypeAlignmentTests</c> is what defends it; if it fails, changes would start being filed
    /// under another item type's name and a pull would hand clients the wrong ciphertext.
    /// </remarks>
    private static ChangeEntityType ToDomain(SyncEntityType type) => (ChangeEntityType)(int)type;

    /// <inheritdoc cref="ToDomain" />
    private static SyncEntityType ToWire(ChangeEntityType type) => (SyncEntityType)(int)type;
}