[PB-5977]: feat(account-purge): implement account purge functionality with scheduling

deploy/charts/mail-server/templates/deployment.yaml

@@ -57,6 +57,8 @@ spec:
               value: {{ .Values.mta.hooksUsername | quote }}
             - name: STALWART_WEBHOOK_USERNAME
               value: {{ .Values.webhook.username | quote }}
+            - name: EXECUTE_JOBS
+              value: {{ .Values.executeCronjobs | quote }}
 
           envFrom:
             - secretRef:

deploy/charts/mail-server/values.yaml

@@ -43,6 +43,8 @@ mta:
 webhook:
   username: REPLACE_ME
 
+executeCronjobs: false
+
 secretName: mail-server-secrets
 
 probes:

package.json

@@ -37,6 +37,7 @@
     "@nestjs/jwt": "^11.0.2",
     "@nestjs/passport": "^11.0.5",
     "@nestjs/platform-express": "^11.1.15",
+    "@nestjs/schedule": "^6.1.3",
     "@nestjs/sequelize": "^11.0.1",
     "@nestjs/swagger": "^11.4.6",
     "class-transformer": "^0.5.1",

src/app.module.ts

@@ -16,6 +16,9 @@ import { HttpGlobalExceptionFilter } from './common/filters/http-global-exceptio
 import { AddressesModule } from './modules/addresses/addresses.module';
 import { StalwartEventsModule } from './modules/stalwart-events/stalwart-events.module';
 import { MtaHooksModule } from './modules/mta-hooks/mta-hooks.module';
+import { JobsModule } from './modules/jobs/jobs.module';
+
+const executeCronjobs = process.env.EXECUTE_JOBS === 'true';
 
 @Module({
   imports: [
@@ -82,6 +85,7 @@ import { MtaHooksModule } from './modules/mta-hooks/mta-hooks.module';
       }),
     }),
     EventEmitterModule.forRoot({ wildcard: true, delimiter: '.' }),
+    ...(executeCronjobs ? [JobsModule] : []),
     HealthModule,
     JmapModule,
     EmailModule,

src/config/configuration.ts

@@ -1,4 +1,5 @@
 export default () => ({
+  executeCronjobs: process.env.EXECUTE_JOBS === 'true',
   port: Number.parseInt(process.env.PORT ?? '3100', 10),
   environment: process.env.NODE_ENV ?? 'development',
   isDevelopment: process.env.NODE_ENV === 'development',
@@ -32,6 +33,14 @@ export default () => ({
       process.env.SUSPENDED_ACCOUNT_RETENTION_DAYS ?? '30',
       10,
     ),
+    purgeBatchSize: Number.parseInt(
+      process.env.ACCOUNT_PURGE_BATCH_SIZE ?? '100',
+      10,
+    ),
+    purgeStalledAfterMinutes: Number.parseInt(
+      process.env.ACCOUNT_PURGE_STALLED_AFTER_MINUTES ?? '60',
+      10,
+    ),
   },
 
   secrets: {

src/modules/account/account-purge.service.spec.ts

@@ -0,0 +1,133 @@
+import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest';
+import { Test, type TestingModule } from '@nestjs/testing';
+import { createMock, type DeepMocked } from '@golevelup/ts-vitest';
+import { ConfigService } from '@nestjs/config';
+import { AccountPurgeService } from './account-purge.service.js';
+import { AccountService } from './account.service.js';
+import { AccountRepository } from './repositories/account.repository.js';
+
+const NOW = new Date('2026-08-21T12:00:00.000Z');
+const RETENTION_DAYS = 30;
+const BATCH_SIZE = 100;
+const STALLED_AFTER_MINUTES = 60;
+
+describe('AccountPurgeService', () => {
+  let service: AccountPurgeService;
+  let accounts: DeepMocked<AccountRepository>;
+  let accountService: DeepMocked<AccountService>;
+  let config: DeepMocked<ConfigService>;
+
+  beforeEach(async () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(NOW);
+
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [AccountPurgeService],
+    })
+      .useMocker(() => createMock<object>())
+      .compile();
+
+    service = module.get(AccountPurgeService);
+    accounts = module.get(AccountRepository);
+    accountService = module.get(AccountService);
+    config = module.get(ConfigService);
+
+    config.get.mockImplementation((key: string) => {
+      const values: Record<string, number> = {
+        'accounts.suspendedRetentionDays': RETENTION_DAYS,
+        'accounts.purgeBatchSize': BATCH_SIZE,
+        'accounts.purgeStalledAfterMinutes': STALLED_AFTER_MINUTES,
+      };
+      return values[key] as never;
+    });
+
+    accounts.claimStalledDeletions.mockResolvedValue([]);
+    accounts.claimExpiredSuspended.mockResolvedValue([]);
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it('when accounts are past retention, then claims and deletes each of them', async () => {
+    accounts.claimExpiredSuspended.mockResolvedValue([
+      { id: 'acc-1', userId: 'user-1' },
+      { id: 'acc-2', userId: 'user-2' },
+    ]);
+
+    const summary = await service.purgeExpiredAccounts();
+
+    expect(accounts.claimExpiredSuspended).toHaveBeenCalledWith({
+      suspendedBefore: new Date('2026-07-22T12:00:00.000Z'),
+      limit: BATCH_SIZE,
+    });
+    expect(accountService.deleteAccount).toHaveBeenCalledWith('user-1');
+    expect(accountService.deleteAccount).toHaveBeenCalledWith('user-2');
+    expect(summary).toEqual({ claimed: 2, purged: 2, failed: 0 });
+  });
+
+  it('when nothing is due, then reports an empty run without deleting anything', async () => {
+    const summary = await service.purgeExpiredAccounts();
+
+    expect(accountService.deleteAccount).not.toHaveBeenCalled();
+    expect(summary).toEqual({ claimed: 0, purged: 0, failed: 0 });
+  });
+
+  it('when a claim has gone stale, then it is retried before newly expired ones', async () => {
+    accounts.claimStalledDeletions.mockResolvedValue([
+      { id: 'acc-stuck', userId: 'user-stuck' },
+    ]);
+
+    await service.purgeExpiredAccounts({ batchSize: 3 });
+
+    expect(accounts.claimStalledDeletions).toHaveBeenCalledWith({
+      updatedBefore: new Date('2026-08-21T11:00:00.000Z'),
+      limit: 3,
+    });
+    expect(accounts.claimExpiredSuspended).toHaveBeenCalledWith({
+      suspendedBefore: new Date('2026-07-22T12:00:00.000Z'),
+      limit: 2,
+    });
+    expect(accountService.deleteAccount).toHaveBeenCalledWith('user-stuck');
+  });
+
+  it('when stalled claims fill the batch, then no new accounts are claimed', async () => {
+    accounts.claimStalledDeletions.mockResolvedValue([
+      { id: 'acc-1', userId: 'user-1' },
+      { id: 'acc-2', userId: 'user-2' },
+    ]);
+
+    await service.purgeExpiredAccounts({ batchSize: 2 });
+
+    expect(accounts.claimExpiredSuspended).toHaveBeenCalledWith({
+      suspendedBefore: new Date('2026-07-22T12:00:00.000Z'),
+      limit: 0,
+    });
+  });
+
+  it('when one account fails, then the rest of the batch still runs', async () => {
+    accounts.claimExpiredSuspended.mockResolvedValue([
+      { id: 'acc-1', userId: 'user-1' },
+      { id: 'acc-2', userId: 'user-2' },
+      { id: 'acc-3', userId: 'user-3' },
+    ]);
+    accountService.deleteAccount.mockImplementation((userId: string) =>
+      userId === 'user-2'
+        ? Promise.reject(new Error('Bridge refused'))
+        : Promise.resolve(),
+    );
+
+    const summary = await service.purgeExpiredAccounts();
+
+    expect(accountService.deleteAccount).toHaveBeenCalledWith('user-3');
+    expect(summary).toEqual({ claimed: 3, purged: 2, failed: 1 });
+  });
+
+  it('when the batch size is zero, then nothing is claimed at all', async () => {
+    const summary = await service.purgeExpiredAccounts({ batchSize: 0 });
+
+    expect(accounts.claimStalledDeletions).not.toHaveBeenCalled();
+    expect(accounts.claimExpiredSuspended).not.toHaveBeenCalled();
+    expect(summary).toEqual({ claimed: 0, purged: 0, failed: 0 });
+  });
+});

src/modules/account/account-purge.service.ts

@@ -0,0 +1,89 @@
+import { Injectable, Logger } from '@nestjs/common';
+import { ConfigService } from '@nestjs/config';
+import dayjs from 'dayjs';
+import { AccountService } from './account.service.js';
+import {
+  AccountRepository,
+  type ClaimedAccount,
+} from './repositories/account.repository.js';
+
+export interface PurgeOptions {
+  batchSize?: number;
+}
+
+export interface PurgeSummary {
+  claimed: number;
+  purged: number;
+  failed: number;
+}
+
+@Injectable()
+export class AccountPurgeService {
+  private readonly logger = new Logger(AccountPurgeService.name);
+
+  constructor(
+    private readonly accounts: AccountRepository,
+    private readonly accountService: AccountService,
+    private readonly config: ConfigService,
+  ) {}
+
+  async purgeExpiredAccounts(
+    options: PurgeOptions = {},
+  ): Promise<PurgeSummary> {
+    const batchSize =
+      options.batchSize ?? this.config.get<number>('accounts.purgeBatchSize')!;
+    const claimed = await this.claimBatch(batchSize);
+
+    if (claimed.length === 0) {
+      return { claimed: 0, purged: 0, failed: 0 };
+    }
+
+    let purged = 0;
+    let failed = 0;
+
+    for (const account of claimed) {
+      try {
+        await this.accountService.deleteAccount(account.userId);
+        purged++;
+      } catch (error) {
+        failed++;
+        this.logger.error(
+          `Failed to purge account '${account.id}' for user '${account.userId}': ${(error as Error).message}`,
+          (error as Error).stack,
+        );
+      }
+    }
+
+    this.logger.log(
+      `Purge run finished: claimed=${claimed.length} purged=${purged} failed=${failed}`,
+    );
+
+    return { claimed: claimed.length, purged, failed };
+  }
+
+  private async claimBatch(batchSize: number): Promise<ClaimedAccount[]> {
+    if (batchSize <= 0) return [];
+
+    const stalled = await this.accounts.claimStalledDeletions({
+      updatedBefore: dayjs()
+        .subtract(
+          this.config.get<number>('accounts.purgeStalledAfterMinutes')!,
+          'minute',
+        )
+        .toDate(),
+      limit: batchSize,
+    });
+
+    const expired = await this.accounts.claimExpiredSuspended({
+      suspendedBefore: dayjs()
+        .subtract(
+          this.config.get<number>('accounts.suspendedRetentionDays')!,
+          'day',
+        )
+        .toDate(),
+      limit: batchSize - stalled.length,
+    });
+
+    return [...stalled, ...expired];
+  }
+}

src/modules/account/account.module.ts

@@ -4,6 +4,7 @@ import { Reflector } from '@nestjs/core';
 import { StalwartModule } from '../infrastructure/stalwart/stalwart.module.js';
 import { PaymentsModule } from '../infrastructure/payments/payments.module.js';
 import { BridgeModule } from '../infrastructure/bridge/bridge.module.js';
+import { AccountPurgeService } from './account-purge.service.js';
 import { AccountService } from './account.service.js';
 import { UserController } from './user.controller.js';
 import { MailAccountGuard } from '../provisioning/provisioning.guard.js';
@@ -39,9 +40,10 @@ import { MailAddressKeysRepository } from './repositories/mail-address-keys.repo
     DomainRepository,
     MailAddressKeysRepository,
     AccountService,
+    AccountPurgeService,
     MailAccountGuard,
     Reflector,
   ],
-  exports: [AccountService],
+  exports: [AccountService, AccountPurgeService],
 })
 export class AccountModule {}