[PM-33527] Implement Organization Delete Tasks with Events

src/Admin/AdminConsole/Controllers/OrganizationsController.cs

@@ -29,6 +29,7 @@
 using Bit.Core.Billing.Pricing;
 using Bit.Core.Billing.Providers.Services;
 using Bit.Core.Billing.Services;
+using Bit.Core.Dirt.Enums;
 using Bit.Core.Enums;
 using Bit.Core.Models.OrganizationConnectionConfigs;
 using Bit.Core.OrganizationFeatures.OrganizationSponsorships.FamiliesForEnterprise.Interfaces;
@@ -587,7 +588,10 @@ await _providerBillingService.ScaleSeats(
             }
         }
 
-        await _organizationRepository.DeleteAsync(organization);
+        // Enqueue the event-log cleanup in the same transaction as the delete. This is an
+        // established organization, so its events must be purged from storage for GDPR.
+        await _organizationRepository.DeleteAndCreateDeleteTasksAsync(
+            organization, [OrganizationDeleteTaskType.EventsCleanup]);
         await _organizationAbilityCacheService.DeleteOrganizationAbilityAsync(organization.Id);
 
         return RedirectToAction("Index");

src/Admin/Jobs/JobsHostedService.cs

@@ -1,8 +1,11 @@
 using System.Runtime.InteropServices;
 using Bit.Admin.Auth.Jobs;
 using Bit.Admin.Tools.Jobs;
+using Bit.Core.Dirt.Services;
+using Bit.Core.Dirt.Services.Implementations;
 using Bit.Core.Jobs;
 using Bit.Core.Settings;
+using Microsoft.Extensions.DependencyInjection.Extensions;
 using Quartz;
 
 namespace Bit.Admin.Jobs;
@@ -71,6 +74,15 @@ public override async Task StartAsync(CancellationToken cancellationToken)
             .StartNow()
             .WithCronSchedule("0 0 2 ? * * *")
             .Build();
+        // Quartz keys triggers by identity, so a trigger instance cannot be shared between jobs:
+        // scheduling the second one throws because the key already exists. Every job below has its
+        // own instance, hence a dedicated five-minute trigger here rather than reusing the one
+        // DeleteSendsJob holds.
+        var organizationDeleteTasksTrigger = TriggerBuilder.Create()
+            .WithIdentity("OrganizationDeleteTasksTrigger")
+            .StartNow()
+            .WithCronSchedule("0 */5 * * * ?")
+            .Build();
 
         var jobs = new List<Tuple<Type, ITrigger>>
         {
@@ -80,6 +92,7 @@ public override async Task StartAsync(CancellationToken cancellationToken)
             new Tuple<Type, ITrigger>(typeof(DatabaseExpiredSponsorshipsJob), everyMondayAtMidnightTrigger),
             new Tuple<Type, ITrigger>(typeof(DeleteAuthRequestsJob), everyFifteenMinutesTrigger),
             new Tuple<Type, ITrigger>(typeof(DeleteUnverifiedOrganizationDomainsJob), everyDayAtTwoAmUtcTrigger),
+            new Tuple<Type, ITrigger>(typeof(OrganizationDeleteTasksJob), organizationDeleteTasksTrigger),
         };
 
         if (!(_globalSettings.SqlServer?.DisableDatabaseMaintenanceJobs ?? false))
@@ -103,6 +116,9 @@ public static void AddJobsServices(IServiceCollection services, bool selfHosted)
         {
             services.AddTransient<AliveJob>();
         }
+        services.AddTransient<OrganizationDeleteTasksJob>();
+        services.TryAddEnumerable(
+            ServiceDescriptor.Transient<IOrganizationDeleteTaskHandler, EventsCleanupOrganizationDeleteTaskHandler>());
         services.AddTransient<DatabaseUpdateStatisticsJob>();
         services.AddTransient<DatabaseRebuildlIndexesJob>();
         services.AddTransient<DatabaseExpiredGrantsJob>();

src/Admin/Jobs/OrganizationDeleteTasksJob.cs

@@ -0,0 +1,145 @@
+#nullable enable
+
+using Azure;
+using Bit.Core;
+using Bit.Core.Dirt.Entities;
+using Bit.Core.Dirt.Enums;
+using Bit.Core.Dirt.Repositories;
+using Bit.Core.Dirt.Services;
+using Bit.Core.Jobs;
+using Bitwarden.Server.Sdk.Features;
+using Quartz;
+
+namespace Bit.Admin.Jobs;
+
+/// <summary>
+/// Drains the <c>OrganizationDeleteTask</c> queue: claims the next pending task and dispatches it to
+/// the <see cref="IOrganizationDeleteTaskHandler"/> registered for its type, deleting in bounded
+/// batches within a run budget so a large cleanup resumes across runs. All lease, progress, and
+/// error bookkeeping lives here; handlers only implement the per-type batch delete.
+/// </summary>
+public class OrganizationDeleteTasksJob : BaseJob
+{
+    private static readonly TimeSpan _runBudget = TimeSpan.FromMinutes(4);
+
+    // A missing handler is expected transiently while a rolling deploy is in flight, but a
+    // genuinely orphaned type (enqueued with no handler that will ever be deployed) stays
+    // unhandled long after any deploy completes. Below this age we log the miss at Warning;
+    // beyond it we escalate to Error so a stuck type can be alerted on rather than lost in
+    // steady Warning volume.
+    private static readonly TimeSpan _orphanedTaskEscalationThreshold = TimeSpan.FromHours(1);
+
+    private readonly IOrganizationDeleteTaskRepository _cleanupRepository;
+    private readonly IReadOnlyDictionary<OrganizationDeleteTaskType, IOrganizationDeleteTaskHandler> _handlers;
+    private readonly IFeatureService _featureService;
+
+    public OrganizationDeleteTasksJob(
+        IOrganizationDeleteTaskRepository cleanupRepository,
+        IEnumerable<IOrganizationDeleteTaskHandler> handlers,
+        IFeatureService featureService,
+        ILogger<OrganizationDeleteTasksJob> logger)
+        : base(logger)
+    {
+        _cleanupRepository = cleanupRepository;
+        // Throws at construction if two handlers claim the same type, failing fast on misconfiguration.
+        _handlers = handlers.ToDictionary(handler => handler.TaskType);
+        _featureService = featureService;
+    }
+
+    protected override async Task ExecuteJobAsync(IJobExecutionContext context)
+    {
+        if (!_featureService.IsEnabled(FeatureFlagKeys.OrganizationEventCleanup))
+        {
+            return;
+        }
+
+        var pending = await _cleanupRepository.ClaimNextPendingAsync();
+        if (pending is null)
+        {
+            return;
+        }
+
+        if (!_handlers.TryGetValue(pending.TaskType, out var handler))
+        {
+            // No handler is registered for this type. This is expected transiently when the server
+            // enqueuing tasks is ahead of this worker during a rolling deploy. We deliberately do NOT
+            // record a failure: doing so would burn the retry budget and could permanently abandon a
+            // task that only needs a newer worker. The claim lease expires and the task is reclaimed
+            // on a later run. Escalate to Error once it has been unhandled long enough that deploy
+            // skew is no longer a plausible explanation.
+            var unhandledFor = DateTime.UtcNow - pending.CreationDate;
+            var logLevel = unhandledFor >= _orphanedTaskEscalationThreshold ? LogLevel.Error : LogLevel.Warning;
+            _logger.Log(logLevel, Constants.BypassFiltersEventId,
+                "No handler registered for organization delete task type {TaskType} (task {TaskId}); leaving for retry. Unhandled for {UnhandledMinutes:N0} minutes.",
+                pending.TaskType, pending.Id, unhandledFor.TotalMinutes);
+            return;
+        }
+
+        _logger.LogInformation(Constants.BypassFiltersEventId,
+            "Starting {TaskType} cleanup for organization {OrganizationId} (task {TaskId})",
+            pending.TaskType, pending.OrganizationId, pending.Id);
+
+        var deadline = DateTime.UtcNow.Add(_runBudget);
+        var drained = false;
+        var totalDeleted = 0L;
+
+        try
+        {
+            while (DateTime.UtcNow < deadline && !context.CancellationToken.IsCancellationRequested)
+            {
+                var deleted = await handler.DeleteBatchAsync(pending, context.CancellationToken);
+                if (deleted == 0)
+                {
+                    // An empty batch is the only proof there is nothing left to purge.
+                    drained = true;
+                    break;
+                }
+
+                await _cleanupRepository.UpdateProgressAsync(pending.Id, deleted);
+                totalDeleted += deleted;
+            }
+
+            // Completion has to be driven by an empty batch, not by the absence of deletions: if
+            // cancellation was already signalled or the budget already spent, the loop never runs
+            // and marking the task complete would strand the organization's events forever.
+            if (drained)
+            {
+                await _cleanupRepository.UpdateCompletedAsync(pending.Id);
+                _logger.LogInformation(Constants.BypassFiltersEventId,
+                    "Completed {TaskType} cleanup for organization {OrganizationId}; deleted {Deleted} items this run",
+                    pending.TaskType, pending.OrganizationId, totalDeleted);
+            }
+            else
+            {
+                _logger.LogInformation(Constants.BypassFiltersEventId,
+                    "Paused {TaskType} cleanup for organization {OrganizationId}; deleted {Deleted} items this run, will resume",
+                    pending.TaskType, pending.OrganizationId, totalDeleted);
+            }
+        }
+        catch (Exception ex)
+        {
+            // Store a sanitized error, never ex.Message: Azure SDK messages can embed
+            // row-key identifiers (e.g. UserId=..., CipherId=...) that must not be persisted.
+            var failureCount = await _cleanupRepository.UpdateErrorAsync(pending.Id, BuildSanitizedError(ex));
+
+            // Individual failures surface through the rethrow below, but the transition to
+            // "abandoned" would not: once the cap is reached the task stops being claimed and goes
+            // quiet. Deleting these logs is a GDPR obligation, so a cleanup that has stopped
+            // retrying for good needs to be alertable rather than inferred from counting retries.
+            if (failureCount >= OrganizationDeleteTask.MaxFailureCount)
+            {
+                _logger.LogError(Constants.BypassFiltersEventId,
+                    "Abandoning {TaskType} cleanup for organization {OrganizationId} (task {TaskId}) after {FailureCount} failures; it will not be retried.",
+                    pending.TaskType, pending.OrganizationId, pending.Id, failureCount);
+            }
+
+            throw;
+        }
+    }
+
+    private static string BuildSanitizedError(Exception ex) => ex switch
+    {
+        RequestFailedException rfe => $"{nameof(RequestFailedException)} (Status: {rfe.Status}, ErrorCode: {rfe.ErrorCode})",
+        _ => ex.GetType().FullName ?? ex.GetType().Name,
+    };
+}

src/Core/AdminConsole/OrganizationFeatures/Organizations/OrganizationDeleteCommand.cs

@@ -6,6 +6,7 @@
 using Bit.Core.Auth.Repositories;
 using Bit.Core.Billing;
 using Bit.Core.Billing.Services;
+using Bit.Core.Dirt.Enums;
 using Bit.Core.Exceptions;
 using Bit.Core.Repositories;
 using Bit.Core.Tools.Services;
@@ -64,7 +65,7 @@ public async Task DeleteAsync(Organization organization)
 
         await _sendFileStorageService.DeleteFilesForOrganizationAsync(organization.Id);
         await _cipherService.DeleteAttachmentsForOrganizationAsync(organization.Id);
-        await _organizationRepository.DeleteAsync(organization);
+        await _organizationRepository.DeleteAndCreateDeleteTasksAsync(organization, [OrganizationDeleteTaskType.EventsCleanup]);
         await _organizationAbilityCacheService.DeleteOrganizationAbilityAsync(organization.Id);
     }
 

src/Core/AdminConsole/Repositories/IOrganizationRepository.cs

@@ -2,6 +2,7 @@
 using Bit.Core.AdminConsole.Entities;
 using Bit.Core.AdminConsole.Enums.Provider;
 using Bit.Core.Billing.Organizations.Models;
+using Bit.Core.Dirt.Enums;
 using Bit.Core.Models.Data.Organizations;
 using Bit.Core.Models.Data.Organizations.OrganizationUsers;
 
@@ -84,4 +85,16 @@ public interface IOrganizationRepository : IRepository<Organization, Guid>
     /// <param name="confirmOwnerAction">Action to confirm the organization owner, obtained from
     /// <see cref="IOrganizationUserRepository.BuildConfirmOwnerAction"/></param>
     Task InitializeOrganizationAsync(Organization organization, Func<DbConnection, DbTransaction, Task> confirmOwnerAction);
+
+    /// <summary>
+    /// Deletes the organization and, within the same database transaction, enqueues one
+    /// <c>OrganizationDeleteTask</c> per supplied task type. This guarantees the deletion and the
+    /// cleanup-task records commit atomically, so durable downstream cleanup (e.g. purging
+    /// Table Storage event logs for GDPR) is never lost if the deletion succeeds. Any team can
+    /// enqueue its own cleanup type by adding it to <paramref name="taskTypes"/> without changing
+    /// this signature. An empty collection deletes the organization without enqueuing any task.
+    /// </summary>
+    /// <param name="organization">The organization to delete.</param>
+    /// <param name="taskTypes">The cleanup task types to enqueue, one row created per type.</param>
+    Task DeleteAndCreateDeleteTasksAsync(Organization organization, IEnumerable<OrganizationDeleteTaskType> taskTypes);
 }

src/Core/Constants.cs

@@ -301,6 +301,7 @@ public static partial class FeatureFlagKeys
     public const string AccessIntelligenceAdoptionUxImprovements = "pm-34723-access-intelligence-adoption-ux-improvements";
     public const string EventManagementForGenericHec = "event-management-for-generic-hec";
     public const string BrowserExtensionHealthReport = "pm-35928-premium-user-health-reports";
+    public const string OrganizationEventCleanup = "pm-33527-organization-event-cleanup";
 
     /* UIF Team */
     public const string RouterFocusManagement = "router-focus-management";

src/Core/Dirt/Entities/OrganizationDeleteTask.cs

@@ -0,0 +1,32 @@
+using Bit.Core.Dirt.Enums;
+using Bit.Core.Entities;
+using Bit.Core.Utilities;
+
+namespace Bit.Core.Dirt.Entities;
+
+public class OrganizationDeleteTask : ITableObject<Guid>
+{
+    /// <summary>
+    /// How long a claimed task stays leased before another worker may reclaim it. Every repository
+    /// implementation has to agree on this, so it lives with the entity rather than in one of them.
+    /// </summary>
+    public const int LeaseDurationMinutes = 10;
+
+    /// <summary>
+    /// A task is abandoned once it has failed this many times, so a permanently failing cleanup
+    /// cannot be reclaimed forever.
+    /// </summary>
+    public const int MaxFailureCount = 5;
+
+    public Guid Id { get; set; }
+    public Guid OrganizationId { get; set; }
+    public OrganizationDeleteTaskType TaskType { get; set; }
+    public DateTime CreationDate { get; set; } = DateTime.UtcNow;
+    public DateTime RevisionDate { get; set; } = DateTime.UtcNow;
+    public DateTime? StartDate { get; set; }
+    public DateTime? CompletedDate { get; set; }
+    public long ItemsDeletedCount { get; set; }
+    public int FailureCount { get; set; }
+    public string? LastError { get; set; }
+    public void SetNewId() => Id = CombGuid.Generate();
+}

src/Core/Dirt/Enums/OrganizationDeleteTaskType.cs

@@ -0,0 +1,6 @@
+namespace Bit.Core.Dirt.Enums;
+
+public enum OrganizationDeleteTaskType : byte
+{
+    EventsCleanup = 0,
+}

src/Core/Dirt/Repositories/IEventRepository.cs

@@ -34,4 +34,11 @@ Task<PagedResult<IEvent>> GetManyBySendAsync(Guid organizationId, Guid sendId, D
     Task CreateManyAsync(IEnumerable<IEvent> e);
     Task<PagedResult<IEvent>> GetManyByOrganizationServiceAccountAsync(Guid organizationId, Guid serviceAccountId,
         DateTime startDate, DateTime endDate, PageOptions pageOptions);
+
+    /// <summary>
+    /// Deletes a bounded batch of events for the given organization and returns the number deleted;
+    /// 0 means nothing is left. Callers invoke repeatedly (persisting progress between calls) until
+    /// 0 is returned. Used to purge orphaned event logs when an organization is deleted (GDPR).
+    /// </summary>
+    Task<int> DeleteManyByOrganizationIdAsync(Guid organizationId, CancellationToken cancellationToken = default);
 }

src/Core/Dirt/Repositories/IOrganizationDeleteTaskRepository.cs

@@ -0,0 +1,17 @@
+using Bit.Core.Dirt.Entities;
+
+namespace Bit.Core.Dirt.Repositories;
+
+public interface IOrganizationDeleteTaskRepository
+{
+    Task CreateAsync(OrganizationDeleteTask task);
+    Task<OrganizationDeleteTask?> ClaimNextPendingAsync();
+    Task UpdateProgressAsync(Guid id, long delta);
+    /// <summary>
+    /// Records a failure against the task and returns its new failure count, so the caller can tell
+    /// when a task has reached <see cref="OrganizationDeleteTask.MaxFailureCount"/> and will no
+    /// longer be claimed.
+    /// </summary>
+    Task<int> UpdateErrorAsync(Guid id, string message);
+    Task UpdateCompletedAsync(Guid id);
+}