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

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

namespace DodoSSH.Api.Features.Sync;

/// <summary>
/// Removes tombstones past their retention, and change-log entries a later one has superseded.
/// </summary>
/// <remarks>
/// <para>
/// ADR 0003 promised this and nothing did it: every deletion stayed a row for ever, and every edit stayed a
/// change-log entry for ever, so the log grew without bound and a client starting from nothing walked every
/// edit ever made to read the items that exist.
/// </para>
/// <para>
/// <b>Superseded entries go at any time.</b> A pull returns entries after a cursor in order, so a client
/// before both an item's old entry and its new one reads the new one, and a client between them already has
/// the old one. Nothing reads history: a conflict resolves against the newest entry alone.
/// </para>
/// <para>
/// <b>Tombstones go after the retention</b>, and taking them is the one change here a client can notice: a
/// machine that last synced before the deletion would never hear of it. So each vault records the highest
/// sequence collected from it — <see cref="Vault.SyncHorizon"/> — and a pull from a cursor below that is
/// refused. The client answers by starting over, and a pull from nothing removes what the server no longer
/// has; see the client's <c>SyncEngine</c>.
/// </para>
/// <para>
/// Per vault under the same advisory lock a push takes, so a collection never interleaves with a write to
/// the vault — and the items are read again under it, so one restored since it was found is left alone.
/// </para>
/// </remarks>
internal sealed class TombstoneCollector(
    DodoDbContext database,
    IOptions<SyncOptions> syncOptions,
    TimeProvider clock,
    ILogger<TombstoneCollector> logger)
{
    /// <summary>Collects what is due, once.</summary>
    /// <returns>How many tombstones and superseded entries went.</returns>
    internal async Task<int> RunOnceAsync(CancellationToken cancellationToken)
    {
        var before = clock.GetUtcNow() - TimeSpan.FromDays(syncOptions.Value.TombstoneRetentionDays);

        var collected = 0;

        collected += await CollectAsync(db => db.Hosts, ChangeEntityType.SshHost, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.SshKeys, ChangeEntityType.SshKey, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.Credentials, ChangeEntityType.Credential, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.KnownHostKeys, ChangeEntityType.KnownHostKey, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.HostGroups, ChangeEntityType.HostGroup, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.Tags, ChangeEntityType.Tag, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.Snippets, ChangeEntityType.Snippet, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.ConnectionLog, ChangeEntityType.ConnectionLogEntry, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.ActivityLog, ChangeEntityType.ActivityLogEntry, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.ObjectStores, ChangeEntityType.ObjectStore, before, cancellationToken).ConfigureAwait(false);
        collected += await CollectAsync(db => db.SpelunkKeys, ChangeEntityType.SpelunkKey, before, cancellationToken).ConfigureAwait(false);

        collected += await CompactAsync(cancellationToken).ConfigureAwait(false);

        SyncLog.Collected(logger, collected);

        return collected;
    }

    private async Task<int> CollectAsync<T>(
        Func<DodoDbContext, DbSet<T>> table,
        ChangeEntityType type,
        DateTimeOffset before,
        CancellationToken cancellationToken)
        where T : class, IVaultItem
    {
        var due = await table(database)
            .Where(item => item.DeletedAtUtc != null && item.DeletedAtUtc < before)
            .Select(item => new { item.VaultId, item.Id })
            .ToListAsync(cancellationToken)
            .ConfigureAwait(false);

        var collected = 0;

        foreach (var vault in due.GroupBy(item => item.VaultId))
        {
            var ids = vault.Select(item => item.Id).ToArray();

            collected += await database.ExecuteFreshAsync(
                [],
                () => CollectFromVaultAsync(table, type, vault.Key, ids, before, cancellationToken))
                .ConfigureAwait(false);
        }

        return collected;
    }

    private async Task<int> CollectFromVaultAsync<T>(
        Func<DodoDbContext, DbSet<T>> table,
        ChangeEntityType type,
        Guid vaultId,
        Guid[] ids,
        DateTimeOffset before,
        CancellationToken cancellationToken)
        where T : class, IVaultItem
    {
        var transaction = await database.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
        await using var _ = transaction.ConfigureAwait(false);

        await database.Database
            .ExecuteSqlAsync($"SELECT pg_advisory_xact_lock(hashtextextended({vaultId.ToString()}, 0))", cancellationToken)
            .ConfigureAwait(false);

        // Read again under the lock: an item restored since it was found is no longer due.
        var still = await table(database)
            .Where(item => item.VaultId == vaultId
                && ids.Contains(item.Id)
                && item.DeletedAtUtc != null
                && item.DeletedAtUtc < before)
            .Select(item => new { item.Id, item.ChangeSequence })
            .ToListAsync(cancellationToken)
            .ConfigureAwait(false);

        if (still.Count == 0)
        {
            return 0;
        }

        var going = still.Select(item => item.Id).ToArray();
        var horizon = still.Max(item => item.ChangeSequence);

        await table(database)
            .Where(item => item.VaultId == vaultId && going.Contains(item.Id))
            .ExecuteDeleteAsync(cancellationToken)
            .ConfigureAwait(false);

        await database.VaultChanges
            .Where(change => change.VaultId == vaultId && change.EntityType == type && going.Contains(change.EntityId))
            .ExecuteDeleteAsync(cancellationToken)
            .ConfigureAwait(false);

        await database.Vaults
            .Where(v => v.Id == vaultId && v.SyncHorizon < horizon)
            .ExecuteUpdateAsync(set => set.SetProperty(v => v.SyncHorizon, horizon), cancellationToken)
            .ConfigureAwait(false);

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

        return still.Count;
    }

    /// <summary>Removes every change-log entry a later entry for the same item has superseded.</summary>
    private Task<int> CompactAsync(CancellationToken cancellationToken) =>
        database.Database.ExecuteSqlRawAsync(
            "DELETE FROM " + DodoDbContext.SchemaName + ".sync_change AS older "
            + "USING " + DodoDbContext.SchemaName + ".sync_change AS newer "
            + "WHERE newer.vault_id = older.vault_id "
            + "AND newer.entity_type = older.entity_type "
            + "AND newer.entity_id = older.entity_id "
            + "AND newer.sequence > older.sequence",
            cancellationToken);
}

/// <summary>Runs the <see cref="TombstoneCollector"/> on a schedule, in a scope of its own each time.</summary>
internal sealed class TombstoneCollectionService(
    IServiceScopeFactory scopes,
    IOptions<SyncOptions> syncOptions,
    TimeProvider clock,
    ILogger<TombstoneCollectionService> logger) : BackgroundService
{
    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var interval = TimeSpan.FromMinutes(syncOptions.Value.TombstoneCollectionIntervalMinutes);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await Task.Delay(interval, clock, stoppingToken).ConfigureAwait(false);

                var scope = scopes.CreateAsyncScope();
                await using var _ = scope.ConfigureAwait(false);

                await scope.ServiceProvider.GetRequiredService<TombstoneCollector>()
                    .RunOnceAsync(stoppingToken)
                    .ConfigureAwait(false);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                return;
            }
            catch (Exception exception) when (exception is not OutOfMemoryException)
            {
                // The next run tries again; nothing collected late is harmed by waiting.
                SyncLog.CollectionFailed(logger, exception);
            }
        }
    }
}