diff --git a/deploy/charts/mail-server/templates/deployment.yaml b/deploy/charts/mail-server/templates/deployment.yaml index fcc0e59..1f84c0a 100644 --- a/deploy/charts/mail-server/templates/deployment.yaml +++ b/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: diff --git a/deploy/charts/mail-server/values.yaml b/deploy/charts/mail-server/values.yaml index 806f0c8..77fb153 100644 --- a/deploy/charts/mail-server/values.yaml +++ b/deploy/charts/mail-server/values.yaml @@ -43,6 +43,8 @@ mta: webhook: username: REPLACE_ME +executeCronjobs: false + secretName: mail-server-secrets probes: diff --git a/package-lock.json b/package-lock.json index fe44bc5..b831ddd 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,6 +17,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", @@ -1286,6 +1287,19 @@ "@nestjs/core": "^11.0.0" } }, + "node_modules/@nestjs/schedule": { + "version": "6.1.3", + "resolved": "https://registry.npmjs.org/@nestjs/schedule/-/schedule-6.1.3.tgz", + "integrity": "sha512-RflMFOpR16Dwd1jAUbeB4mfGTCh65fvEdL4mSjQPJChpkRGRjIXjb+6YQcK2faQrVT60c9DmLmoVR7/ONCtuYQ==", + "license": "MIT", + "dependencies": { + "cron": "4.4.0" + }, + "peerDependencies": { + "@nestjs/common": "^10.0.0 || ^11.0.0", + "@nestjs/core": "^10.0.0 || ^11.0.0" + } + }, "node_modules/@nestjs/schematics": { "version": "11.1.0", "resolved": "https://registry.npmjs.org/@nestjs/schematics/-/schematics-11.1.0.tgz", @@ -2353,6 +2367,12 @@ "@types/node": "*" } }, + "node_modules/@types/luxon": { + "version": "3.7.4", + "resolved": "https://registry.npmjs.org/@types/luxon/-/luxon-3.7.4.tgz", + "integrity": "sha512-V536ZAd6ZJztrrBlLcDFaaZrXNAL2E5uGmssWf/dpSiLkmkLScXUYhUBnWPmtW+cIqnNHzf6//TCMpIc9SCRRQ==", + "license": "MIT" + }, "node_modules/@types/ms": { "version": "2.1.0", "resolved": "https://registry.npmjs.org/@types/ms/-/ms-2.1.0.tgz", @@ -3978,6 +3998,23 @@ } } }, + "node_modules/cron": { + "version": "4.4.0", + "resolved": "https://registry.npmjs.org/cron/-/cron-4.4.0.tgz", + "integrity": "sha512-fkdfq+b+AHI4cKdhZlppHveI/mgz2qpiYxcm+t5E5TsxX7QrLS1VE0+7GENEk9z0EeGPcpSciGv6ez24duWhwQ==", + "license": "MIT", + "dependencies": { + "@types/luxon": "~3.7.0", + "luxon": "~3.7.0" + }, + "engines": { + "node": ">=18.x" + }, + "funding": { + "type": "ko-fi", + "url": "https://ko-fi.com/intcreator" + } + }, "node_modules/cross-env": { "version": "10.1.0", "resolved": "https://registry.npmjs.org/cross-env/-/cross-env-10.1.0.tgz", @@ -6658,6 +6695,15 @@ "node": "20 || >=22" } }, + "node_modules/luxon": { + "version": "3.7.2", + "resolved": "https://registry.npmjs.org/luxon/-/luxon-3.7.2.tgz", + "integrity": "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew==", + "license": "MIT", + "engines": { + "node": ">=12" + } + }, "node_modules/magic-string": { "version": "0.30.17", "resolved": "https://registry.npmjs.org/magic-string/-/magic-string-0.30.17.tgz", diff --git a/package.json b/package.json index 2bf1a7d..db057d3 100644 --- a/package.json +++ b/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", diff --git a/src/app.module.ts b/src/app.module.ts index d425e65..a9bbb72 100644 --- a/src/app.module.ts +++ b/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, diff --git a/src/config/configuration.ts b/src/config/configuration.ts index a897e3f..af05284 100644 --- a/src/config/configuration.ts +++ b/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: { diff --git a/src/modules/account/account-purge.service.spec.ts b/src/modules/account/account-purge.service.spec.ts new file mode 100644 index 0000000..e254d9f --- /dev/null +++ b/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; + let accountService: DeepMocked; + let config: DeepMocked; + + beforeEach(async () => { + vi.useFakeTimers(); + vi.setSystemTime(NOW); + + const module: TestingModule = await Test.createTestingModule({ + providers: [AccountPurgeService], + }) + .useMocker(() => createMock()) + .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 = { + '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 }); + }); +}); diff --git a/src/modules/account/account-purge.service.ts b/src/modules/account/account-purge.service.ts new file mode 100644 index 0000000..f8b809a --- /dev/null +++ b/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 { + const batchSize = + options.batchSize ?? this.config.get('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 { + if (batchSize <= 0) return []; + + const stalled = await this.accounts.claimStalledDeletions({ + updatedBefore: dayjs() + .subtract( + this.config.get('accounts.purgeStalledAfterMinutes')!, + 'minute', + ) + .toDate(), + limit: batchSize, + }); + + const expired = await this.accounts.claimExpiredSuspended({ + suspendedBefore: dayjs() + .subtract( + this.config.get('accounts.suspendedRetentionDays')!, + 'day', + ) + .toDate(), + limit: batchSize - stalled.length, + }); + + return [...stalled, ...expired]; + } +} diff --git a/src/modules/account/account.module.ts b/src/modules/account/account.module.ts index 77152db..03192cd 100644 --- a/src/modules/account/account.module.ts +++ b/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 {} diff --git a/src/modules/account/account.service.spec.ts b/src/modules/account/account.service.spec.ts index 36557ff..78529d8 100644 --- a/src/modules/account/account.service.spec.ts +++ b/src/modules/account/account.service.spec.ts @@ -17,7 +17,10 @@ import { AccountRepository } from './repositories/account.repository.js'; import { AddressRepository } from './repositories/address.repository.js'; import { DomainRepository } from './repositories/domain.repository.js'; import { MailAddressKeysRepository } from './repositories/mail-address-keys.repository.js'; -import { BridgeClient } from '../infrastructure/bridge/bridge.service.js'; +import { + BridgeApiError, + BridgeClient, +} from '../infrastructure/bridge/bridge.service.js'; import { PaymentsService } from '../infrastructure/payments/payments.service.js'; import type { Tier } from '../infrastructure/payments/payments.types.js'; import { @@ -69,6 +72,11 @@ describe('AccountService', () => { bridge = module.get(BridgeClient); payments = module.get(PaymentsService); config = module.get(ConfigService); + + bridge.deleteMailBucket.mockResolvedValue({ + maxSpaceBytes: 1000, + totalUsedSpaceBytes: 0, + }); }); describe('getAccount', () => { @@ -599,7 +607,7 @@ describe('AccountService', () => { expect(accounts.delete).toHaveBeenCalledWith(createdAccount.id, { force: true, }); - expect(accounts.setNetworkBucketId).not.toHaveBeenCalled(); + expect(addresses.setNetworkBucketId).not.toHaveBeenCalled(); }); it('when the provider delete fails during rollback, then still hard-deletes the account', async () => { @@ -670,7 +678,7 @@ describe('AccountService', () => { }); describe('deleteAccount', () => { - it('when account has addresses, then deletes all principals and account', async () => { + it('when account has addresses, then destroys every principal', async () => { const addr1 = newMailAddressAttributes({ isDefault: true }); const addr2 = newMailAddressAttributes({ isDefault: false }); const account = MailAccount.build( @@ -686,14 +694,28 @@ describe('AccountService', () => { expect(provider.deleteAccount).toHaveBeenCalledWith( addr2.providerExternalId, ); - expect(addresses.deleteProviderLink).toHaveBeenCalledWith(addr1.id); - expect(addresses.deleteProviderLink).toHaveBeenCalledWith(addr2.id); - expect(accounts.delete).toHaveBeenCalledWith(account.id); }); - it('when account has a network bucket, then deletes it via the bridge', async () => { + it('when the rows are dropped, then they are hard deleted so the cascades fire', async () => { + const account = MailAccount.build(newMailAccountAttributes()); + accounts.findByUserId.mockResolvedValue(account); + + await service.deleteAccount(account.userId); + + expect(accounts.delete).toHaveBeenCalledWith(account.id, { force: true }); + }); + + it('when addresses hold network buckets, then releases each of them', async () => { + const addr1 = newMailAddressAttributes({ + isDefault: true, + networkBucketId: 'bucket-1', + }); + const addr2 = newMailAddressAttributes({ + isDefault: false, + networkBucketId: 'bucket-2', + }); const account = MailAccount.build( - newMailAccountAttributes({ networkBucketId: 'bucket-1' }), + newMailAccountAttributes({ addresses: [addr1, addr2] }), ); accounts.findByUserId.mockResolvedValue(account); @@ -703,12 +725,17 @@ describe('AccountService', () => { account.userId, 'bucket-1', ); - expect(accounts.delete).toHaveBeenCalledWith(account.id); + expect(bridge.deleteMailBucket).toHaveBeenCalledWith( + account.userId, + 'bucket-2', + ); }); - it('when account has no network bucket, then does not call the bridge', async () => { + it('when there is no network bucket, then does not call the bridge', async () => { const account = MailAccount.build( - newMailAccountAttributes({ networkBucketId: null }), + newMailAccountAttributes({ + addresses: [newMailAddressAttributes({ networkBucketId: null })], + }), ); accounts.findByUserId.mockResolvedValue(account); @@ -717,16 +744,76 @@ describe('AccountService', () => { expect(bridge.deleteMailBucket).not.toHaveBeenCalled(); }); - it('when bridge bucket deletion fails, then logs a warning and still deletes the account', async () => { + it('when the bucket is already gone, then treats the 404 as released', async () => { const account = MailAccount.build( - newMailAccountAttributes({ networkBucketId: 'bucket-1' }), + newMailAccountAttributes({ + addresses: [ + newMailAddressAttributes({ networkBucketId: 'bucket-1' }), + ], + }), ); accounts.findByUserId.mockResolvedValue(account); - bridge.deleteMailBucket.mockRejectedValue(new Error('Bridge down')); + bridge.deleteMailBucket.mockRejectedValue( + new BridgeApiError('gone', 404, 'not found'), + ); await service.deleteAccount(account.userId); - expect(accounts.delete).toHaveBeenCalledWith(account.id); + expect(accounts.delete).toHaveBeenCalledWith(account.id, { force: true }); + }); + + it('when the bridge refuses to release a bucket, then keeps the rows and rethrows', async () => { + const account = MailAccount.build( + newMailAccountAttributes({ + addresses: [ + newMailAddressAttributes({ networkBucketId: 'bucket-1' }), + ], + }), + ); + accounts.findByUserId.mockResolvedValue(account); + bridge.deleteMailBucket.mockRejectedValue( + new BridgeApiError('refused', 409, 'shard-backed'), + ); + + await expect(service.deleteAccount(account.userId)).rejects.toThrow( + BridgeApiError, + ); + expect(accounts.delete).not.toHaveBeenCalled(); + }); + + it('when the bridge is down, then keeps the rows and rethrows', async () => { + const account = MailAccount.build( + newMailAccountAttributes({ + addresses: [ + newMailAddressAttributes({ networkBucketId: 'bucket-1' }), + ], + }), + ); + accounts.findByUserId.mockResolvedValue(account); + bridge.deleteMailBucket.mockRejectedValue(new Error('Bridge down')); + + await expect(service.deleteAccount(account.userId)).rejects.toThrow( + 'Bridge down', + ); + expect(accounts.delete).not.toHaveBeenCalled(); + }); + + it('when a principal cannot be destroyed, then keeps the rows and the buckets', async () => { + const account = MailAccount.build( + newMailAccountAttributes({ + addresses: [ + newMailAddressAttributes({ networkBucketId: 'bucket-1' }), + ], + }), + ); + accounts.findByUserId.mockResolvedValue(account); + provider.deleteAccount.mockRejectedValue(new Error('Stalwart down')); + + await expect(service.deleteAccount(account.userId)).rejects.toThrow( + 'Stalwart down', + ); + expect(bridge.deleteMailBucket).not.toHaveBeenCalled(); + expect(accounts.delete).not.toHaveBeenCalled(); }); it('when account does not exist, then throws NotFoundException', async () => { @@ -761,6 +848,22 @@ describe('AccountService', () => { expect(accounts.suspend).toHaveBeenCalledWith(account.id); }); + it('when account is being deleted, then throws ConflictException', async () => { + const account = MailAccount.build( + newMailAccountAttributes({ + status: MailAccountState.Deleting, + suspendedAt: new Date(), + }), + ); + accounts.findByUserId.mockResolvedValue(account); + + await expect(service.suspendAccount(account.userId)).rejects.toThrow( + ConflictException, + ); + expect(provider.suspendAccount).not.toHaveBeenCalled(); + expect(accounts.suspend).not.toHaveBeenCalled(); + }); + it('when account is already suspended, then is a no-op', async () => { const account = MailAccount.build( newMailAccountAttributes({ @@ -809,6 +912,22 @@ describe('AccountService', () => { expect(accounts.reactivate).toHaveBeenCalledWith(account.id); }); + it('when account is being deleted, then throws ConflictException', async () => { + const account = MailAccount.build( + newMailAccountAttributes({ + status: MailAccountState.Deleting, + suspendedAt: new Date(), + }), + ); + accounts.findByUserId.mockResolvedValue(account); + + await expect(service.reactivateAccount(account.userId)).rejects.toThrow( + ConflictException, + ); + expect(provider.reactivateAccount).not.toHaveBeenCalled(); + expect(accounts.reactivate).not.toHaveBeenCalled(); + }); + it('when account is already active, then is a no-op', async () => { const account = MailAccount.build( newMailAccountAttributes({ status: MailAccountState.Active }), diff --git a/src/modules/account/account.service.ts b/src/modules/account/account.service.ts index be455c7..fa27cb3 100644 --- a/src/modules/account/account.service.ts +++ b/src/modules/account/account.service.ts @@ -9,7 +9,10 @@ import { } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import dayjs from 'dayjs'; -import { BridgeClient } from '../infrastructure/bridge/bridge.service.js'; +import { + BridgeClient, + isBridgeNotFound, +} from '../infrastructure/bridge/bridge.service.js'; import { PaymentsService } from '../infrastructure/payments/payments.service.js'; import { MailNotSetupException } from '../provisioning/mail-not-setup.exception.js'; import { AccountProvider } from './account-provider.port.js'; @@ -275,27 +278,18 @@ export class AccountService { const account = await this.getAccountOrFail(driveUserUuid); await Promise.all( - account.addresses.map(async (a) => { - await this.provider.deleteAccount(a.providerExternalId); - await this.addresses.deleteProviderLink(a.id); - await this.deleteNetworkBucket(driveUserUuid, a); - }), + account.addresses.map((a) => + this.provider.deleteAccount(a.providerExternalId), + ), ); - if (account.networkBucketId) { - try { - await this.bridge.deleteMailBucket( - driveUserUuid, - account.networkBucketId, - ); - } catch (error) { - this.logger.warn( - `Failed to delete network bucket '${account.networkBucketId}' for '${driveUserUuid}': ${(error as Error).message}`, - ); - } - } + await Promise.all( + account.addresses.map((a) => + this.releaseNetworkBucket(driveUserUuid, a.networkBucketId), + ), + ); + await this.accounts.delete(account.id, { force: true }); - await this.accounts.delete(account.id); this.logger.log(`Deleted account for user '${driveUserUuid}'`); } @@ -496,14 +490,45 @@ export class AccountService { await this.addresses.setNetworkBucketId(addressId, bucket.id); } + /** + * Deletes a network bucket and the quota it holds. + * + * A 404 means the bucket, or the user behind it, is already gone — nothing + * left to release, so that counts as done. Anything else is left to the + * caller: giving up here would drop the rows carrying the bucket id and + * strand the charge with no way to find it again. + */ + private async releaseNetworkBucket( + userUuid: string, + networkBucketId: string | null, + ): Promise { + if (!networkBucketId) return; + + try { + const { totalUsedSpaceBytes } = await this.bridge.deleteMailBucket( + userUuid, + networkBucketId, + ); + this.logger.log( + `Deleted network bucket '${networkBucketId}' for '${userUuid}', user now at ${totalUsedSpaceBytes} bytes`, + ); + } catch (error) { + if (isBridgeNotFound(error)) { + this.logger.log( + `Network bucket '${networkBucketId}' for '${userUuid}' was already gone`, + ); + return; + } + throw error; + } + } + private async deleteNetworkBucket( userUuid: string, address: MailAddress, ): Promise { - if (!address.networkBucketId) return; - try { - await this.bridge.deleteMailBucket(userUuid, address.networkBucketId); + await this.releaseNetworkBucket(userUuid, address.networkBucketId); } catch (error) { this.logger.warn( `Failed to delete network bucket '${address.networkBucketId}' for '${userUuid}': ${(error as Error).message}`, @@ -511,6 +536,19 @@ export class AccountService { } } + /** + * Once an account has been claimed for deletion its mailboxes are already + * being torn down, so there is nothing coherent left to suspend or bring + * back. The caller has to provision a new account instead. + */ + private assertNotBeingDeleted(account: MailAccount): void { + if (account.isBeingDeleted) { + throw new ConflictException( + `Account for user '${account.userId}' is being deleted`, + ); + } + } + private async getAccountOrFail(userId: string): Promise { const account = await this.accounts.findByUserId(userId); if (!account) { @@ -521,6 +559,7 @@ export class AccountService { async suspendAccount(userId: string): Promise { const account = await this.getAccountOrFail(userId); + this.assertNotBeingDeleted(account); if (account.isSuspended) { this.logger.log(`Account for user '${userId}' is already suspended`); return; @@ -539,6 +578,7 @@ export class AccountService { async reactivateAccount(userId: string): Promise { const account = await this.getAccountOrFail(userId); + this.assertNotBeingDeleted(account); if (!account.isSuspended) { this.logger.log(`Account for user '${userId}' is already active`); return; diff --git a/src/modules/account/domain/mail-account.domain.ts b/src/modules/account/domain/mail-account.domain.ts index 38071d1..79d788f 100644 --- a/src/modules/account/domain/mail-account.domain.ts +++ b/src/modules/account/domain/mail-account.domain.ts @@ -6,6 +6,7 @@ import { export enum MailAccountState { Active = 'active', Suspended = 'suspended', + Deleting = 'deleting', } export interface MailAccountAttributes { @@ -45,4 +46,8 @@ export class MailAccount { get isSuspended(): boolean { return this.status === MailAccountState.Suspended; } + + get isBeingDeleted(): boolean { + return this.status === MailAccountState.Deleting; + } } diff --git a/src/modules/account/repositories/account.repository.spec.ts b/src/modules/account/repositories/account.repository.spec.ts index f75411a..3041ac0 100644 --- a/src/modules/account/repositories/account.repository.spec.ts +++ b/src/modules/account/repositories/account.repository.spec.ts @@ -2,6 +2,8 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { Test, type TestingModule } from '@nestjs/testing'; import { getModelToken } from '@nestjs/sequelize'; import { createMock, type DeepMocked } from '@golevelup/ts-vitest'; +import { QueryTypes } from 'sequelize'; +import { Sequelize } from 'sequelize-typescript'; import { AccountRepository } from './account.repository.js'; import { MailAccountModel } from '../models/mail-account.model.js'; import { MailAddressModel } from '../models/mail-address.model.js'; @@ -11,6 +13,7 @@ import { MailAccountState } from '../domain/mail-account.domain.js'; describe('AccountRepository', () => { let repository: AccountRepository; let accountModel: DeepMocked; + let sequelize: DeepMocked; beforeEach(async () => { const module: TestingModule = await Test.createTestingModule({ @@ -26,6 +29,7 @@ describe('AccountRepository', () => { repository = module.get(AccountRepository); accountModel = module.get(getModelToken(MailAccountModel)); + sequelize = module.get(Sequelize); }); const buildModel = (overrides: Partial = {}) => @@ -167,4 +171,103 @@ describe('AccountRepository', () => { ); }); }); + + describe('claimExpiredSuspended', () => { + const threshold = new Date('2026-08-01T00:00:00.000Z'); + + it('when accounts are past retention, then claims them and returns them', async () => { + sequelize.query.mockResolvedValue([ + { id: 'acc-1', userId: 'user-1' }, + ] as never); + + const claimed = await repository.claimExpiredSuspended({ + suspendedBefore: threshold, + limit: 10, + }); + + expect(claimed).toEqual([{ id: 'acc-1', userId: 'user-1' }]); + }); + + it('when claiming, then moves suspended rows into deleting in one statement', async () => { + sequelize.query.mockResolvedValue([] as never); + + await repository.claimExpiredSuspended({ + suspendedBefore: threshold, + limit: 10, + }); + + expect(sequelize.query).toHaveBeenCalledWith( + expect.stringContaining('UPDATE mail_accounts'), + { + replacements: { + deleting: MailAccountState.Deleting, + suspended: MailAccountState.Suspended, + threshold, + limit: 10, + }, + type: QueryTypes.SELECT, + }, + ); + }); + + it('when claiming, then the statement leases only rows nobody else holds', async () => { + sequelize.query.mockResolvedValue([] as never); + + await repository.claimExpiredSuspended({ + suspendedBefore: threshold, + limit: 10, + }); + + const clauses = [ + 'deleted_at IS NULL', + 'status = :suspended', + 'suspended_at < :threshold', + 'FOR UPDATE SKIP LOCKED', + 'RETURNING id, user_id AS "userId"', + ]; + for (const clause of clauses) { + expect(sequelize.query).toHaveBeenCalledWith( + expect.stringContaining(clause), + expect.anything(), + ); + } + }); + + it('when the batch has no room left, then does not touch the database', async () => { + const claimed = await repository.claimExpiredSuspended({ + suspendedBefore: threshold, + limit: 0, + }); + + expect(claimed).toEqual([]); + expect(sequelize.query).not.toHaveBeenCalled(); + }); + }); + + describe('claimStalledDeletions', () => { + it('when a claim has gone stale, then takes it back for another run', async () => { + const threshold = new Date('2026-08-21T12:00:00.000Z'); + sequelize.query.mockResolvedValue([ + { id: 'acc-1', userId: 'user-1' }, + ] as never); + + const claimed = await repository.claimStalledDeletions({ + updatedBefore: threshold, + limit: 5, + }); + + expect(sequelize.query).toHaveBeenCalledWith( + expect.stringContaining('updated_at < :threshold'), + { + replacements: { + deleting: MailAccountState.Deleting, + threshold, + limit: 5, + }, + type: QueryTypes.SELECT, + }, + ); + expect(claimed).toEqual([{ id: 'acc-1', userId: 'user-1' }]); + }); + }); }); diff --git a/src/modules/account/repositories/account.repository.ts b/src/modules/account/repositories/account.repository.ts index 762b14a..b8bab3a 100644 --- a/src/modules/account/repositories/account.repository.ts +++ b/src/modules/account/repositories/account.repository.ts @@ -1,5 +1,7 @@ import { Injectable } from '@nestjs/common'; import { InjectModel } from '@nestjs/sequelize'; +import { QueryTypes } from 'sequelize'; +import { Sequelize } from 'sequelize-typescript'; import { MailAccount, MailAccountState, @@ -9,11 +11,17 @@ import { MailAddressModel } from '../models/mail-address.model.js'; import { MailProviderAccountModel } from '../models/mail-provider-account.model.js'; import { toAddressAttributes } from './address.repository.js'; +export interface ClaimedAccount { + id: string; + userId: string; +} + @Injectable() export class AccountRepository { constructor( @InjectModel(MailAccountModel) private readonly accountModel: typeof MailAccountModel, + private readonly sequelize: Sequelize, ) {} async findByUserId(userId: string): Promise { @@ -64,6 +72,77 @@ export class AccountRepository { ); } + async claimExpiredSuspended(params: { + suspendedBefore: Date; + limit: number; + }): Promise { + return this.runClaim( + `UPDATE mail_accounts + SET status = :deleting, updated_at = NOW() + WHERE id IN ( + SELECT id + FROM mail_accounts + WHERE deleted_at IS NULL + AND status = :suspended + AND suspended_at IS NOT NULL + AND suspended_at < :threshold + ORDER BY suspended_at + LIMIT :limit + FOR UPDATE SKIP LOCKED + ) + RETURNING id, user_id AS "userId"`, + params.limit, + { + suspended: MailAccountState.Suspended, + threshold: params.suspendedBefore, + }, + ); + } + + /** + * Re-claims accounts left mid-purge by a run that died, so the next one + * finishes them instead of leaving them stuck in 'deleting' forever. + */ + async claimStalledDeletions(params: { + updatedBefore: Date; + limit: number; + }): Promise { + return this.runClaim( + `UPDATE mail_accounts + SET status = :deleting, updated_at = NOW() + WHERE id IN ( + SELECT id + FROM mail_accounts + WHERE deleted_at IS NULL + AND status = :deleting + AND updated_at < :threshold + ORDER BY updated_at + LIMIT :limit + FOR UPDATE SKIP LOCKED + ) + RETURNING id, user_id AS "userId"`, + params.limit, + { threshold: params.updatedBefore }, + ); + } + + private async runClaim( + sql: string, + limit: number, + replacements: Record, + ): Promise { + if (limit <= 0) return []; + + return this.sequelize.query(sql, { + replacements: { + deleting: MailAccountState.Deleting, + limit, + ...replacements, + }, + type: QueryTypes.SELECT, + }); + } + private toDomain(model: MailAccountModel): MailAccount { return MailAccount.build({ id: model.id, diff --git a/src/modules/infrastructure/bridge/bridge.service.spec.ts b/src/modules/infrastructure/bridge/bridge.service.spec.ts index 368ba5a..22a9a23 100644 --- a/src/modules/infrastructure/bridge/bridge.service.spec.ts +++ b/src/modules/infrastructure/bridge/bridge.service.spec.ts @@ -139,14 +139,17 @@ describe('BridgeClient', () => { }); describe('deleteMailBucket', () => { - it('when Bridge returns 204, then signs a gateway token and DELETEs the bucket', async () => { + it('when Bridge returns 200, then signs a gateway token, DELETEs the bucket and returns the snapshot', async () => { + const snapshot = { maxSpaceBytes: 1000, totalUsedSpaceBytes: 240 }; jwtService.sign.mockReturnValue('signed-jwt'); httpRequest.mockResolvedValue({ - statusCode: 204, - body: { text: () => Promise.resolve('') }, + statusCode: 200, + body: { text: () => Promise.resolve(JSON.stringify(snapshot)) }, }); - await service.deleteMailBucket('user-1', 'bucket-1'); + const result = await service.deleteMailBucket('user-1', 'bucket-1'); + + expect(result).toEqual(snapshot); expect(jwtService.sign).toHaveBeenCalledWith( { payload: { uuid: 'user-1' } }, @@ -168,7 +171,7 @@ describe('BridgeClient', () => { ); }); - it('when Bridge returns a non-204 status, then throws BridgeApiError with statusCode and details', async () => { + it('when Bridge returns a non-200 status, then throws BridgeApiError with statusCode and details', async () => { jwtService.sign.mockReturnValue('signed-jwt'); httpRequest.mockResolvedValue({ statusCode: 404, diff --git a/src/modules/infrastructure/bridge/bridge.service.ts b/src/modules/infrastructure/bridge/bridge.service.ts index 391a185..2c3bbd7 100644 --- a/src/modules/infrastructure/bridge/bridge.service.ts +++ b/src/modules/infrastructure/bridge/bridge.service.ts @@ -79,7 +79,10 @@ export class BridgeClient implements OnModuleInit, OnModuleDestroy { return JSON.parse(text) as MailBucket; } - async deleteMailBucket(userUuid: string, bucketId: string): Promise { + async deleteMailBucket( + userUuid: string, + bucketId: string, + ): Promise { const token = this.signGatewayToken(userUuid); const { statusCode, body } = await this.httpClient.request({ @@ -93,13 +96,15 @@ export class BridgeClient implements OnModuleInit, OnModuleDestroy { const text = await body.text(); - if (statusCode !== 204) { + if (statusCode !== 200) { throw new BridgeApiError( `Failed to delete mail bucket '${bucketId}' for user '${userUuid}': HTTP ${statusCode}`, statusCode, text, ); } + + return JSON.parse(text) as UserSpaceSnapshot; } async createBucketEntry( @@ -210,3 +215,7 @@ export class BridgeApiError extends Error { this.name = 'BridgeApiError'; } } + +export function isBridgeNotFound(error: unknown): boolean { + return error instanceof BridgeApiError && error.statusCode === 404; +} diff --git a/src/modules/infrastructure/stalwart/stalwart-account.provider.ts b/src/modules/infrastructure/stalwart/stalwart-account.provider.ts index 58ac622..d9316ca 100644 --- a/src/modules/infrastructure/stalwart/stalwart-account.provider.ts +++ b/src/modules/infrastructure/stalwart/stalwart-account.provider.ts @@ -52,8 +52,13 @@ export class StalwartAccountProvider extends AccountProvider { } async deleteAccount(email: string): Promise { - await this.stalwart.deleteAccountByEmail(email); - this.logger.log(`Deleted account '${email}'`); + const deleted = await this.stalwart.deleteAccountByEmail(email); + + this.logger.log( + deleted + ? `Deleted account '${email}'` + : `Account '${email}' was already gone`, + ); } async suspendAccount(email: string): Promise { diff --git a/src/modules/infrastructure/stalwart/stalwart.service.spec.ts b/src/modules/infrastructure/stalwart/stalwart.service.spec.ts index 10f6ce8..08e954d 100644 --- a/src/modules/infrastructure/stalwart/stalwart.service.spec.ts +++ b/src/modules/infrastructure/stalwart/stalwart.service.spec.ts @@ -106,6 +106,11 @@ const DOMAIN_BATCH_HIT = jmapResponse([ getResp('x:Domain/get', [{ id: 'dom1', name: 'test.com' }]), ]); +const DOMAIN_BATCH_MISS = jmapResponse([ + queryResp('x:Domain/query', []), + getResp('x:Domain/get', []), +]); + describe('StalwartService', () => { let service: StalwartService; @@ -288,7 +293,7 @@ describe('StalwartService', () => { await expect( service.deleteAccountByEmail('alice@test.com'), - ).resolves.toBeUndefined(); + ).resolves.toBe(true); expect(mockRequest).toHaveBeenCalledTimes(2); const setCall = bodyOf(1).methodCalls[1]!; @@ -301,7 +306,7 @@ describe('StalwartService', () => { }); }); - it('when account not found, then throws StalwartApiError', async () => { + it('when account not found, then reports nothing was destroyed', async () => { mockRequest .mockResolvedValueOnce(DOMAIN_BATCH_HIT) .mockResolvedValueOnce( @@ -313,6 +318,14 @@ describe('StalwartService', () => { await expect( service.deleteAccountByEmail('ghost@test.com'), + ).resolves.toBe(false); + }); + + it('when the domain is unknown, then throws StalwartApiError', async () => { + mockRequest.mockResolvedValueOnce(DOMAIN_BATCH_MISS); + + await expect( + service.deleteAccountByEmail('ghost@unknown.com'), ).rejects.toThrow(StalwartApiError); }); diff --git a/src/modules/infrastructure/stalwart/stalwart.service.ts b/src/modules/infrastructure/stalwart/stalwart.service.ts index 68d7e7b..6299e73 100644 --- a/src/modules/infrastructure/stalwart/stalwart.service.ts +++ b/src/modules/infrastructure/stalwart/stalwart.service.ts @@ -151,11 +151,14 @@ export class StalwartService implements OnModuleInit, OnModuleDestroy { return get.list[0] ?? null; } - async deleteAccountByEmail(email: string): Promise { + async deleteAccountByEmail(email: string): Promise { const { local, domain } = splitEmail(email); const domainId = await this.resolveDomainId(domain); if (!domainId) { - throw new StalwartApiError(`Account '${email}' not found`, null); + throw new StalwartApiError( + `Cannot delete account '${email}': domain '${domain}' is not configured in Stalwart`, + { domain }, + ); } const response = await this.jmapCall< @@ -176,7 +179,7 @@ export class StalwartService implements OnModuleInit, OnModuleDestroy { ]); const query = response.methodResponses[0]![1] as JmapQueryResponse; if (query.ids.length === 0) { - throw new StalwartApiError(`Account '${email}' not found`, null); + return false; } const set = response @@ -189,6 +192,8 @@ export class StalwartService implements OnModuleInit, OnModuleDestroy { failed, ); } + + return true; } async suspendAccountByEmail(email: string): Promise { diff --git a/src/modules/jobs/constants.ts b/src/modules/jobs/constants.ts new file mode 100644 index 0000000..13a834e --- /dev/null +++ b/src/modules/jobs/constants.ts @@ -0,0 +1,3 @@ +export enum JobName { + ACCOUNT_PURGE = 'account-purge', +} diff --git a/src/modules/jobs/jobs.module.ts b/src/modules/jobs/jobs.module.ts new file mode 100644 index 0000000..3ded75a --- /dev/null +++ b/src/modules/jobs/jobs.module.ts @@ -0,0 +1,10 @@ +import { Module } from '@nestjs/common'; +import { ScheduleModule } from '@nestjs/schedule'; +import { AccountModule } from '../account/account.module.js'; +import { AccountPurgeScheduler } from './tasks/account-purge/account-purge.scheduler.js'; + +@Module({ + imports: [ScheduleModule.forRoot(), AccountModule], + providers: [AccountPurgeScheduler], +}) +export class JobsModule {} diff --git a/src/modules/jobs/tasks/account-purge/account-purge.scheduler.spec.ts b/src/modules/jobs/tasks/account-purge/account-purge.scheduler.spec.ts new file mode 100644 index 0000000..6700b6d --- /dev/null +++ b/src/modules/jobs/tasks/account-purge/account-purge.scheduler.spec.ts @@ -0,0 +1,86 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import { Test, type TestingModule } from '@nestjs/testing'; +import { createMock, type DeepMocked } from '@golevelup/ts-vitest'; +import { ConfigService } from '@nestjs/config'; +import { AccountPurgeScheduler } from './account-purge.scheduler.js'; +import { AccountPurgeService } from '../../../account/account-purge.service.js'; + +describe('AccountPurgeScheduler', () => { + let scheduler: AccountPurgeScheduler; + let purge: DeepMocked; + let config: DeepMocked; + + const enable = (enabled: boolean) => config.get.mockReturnValue(enabled); + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [AccountPurgeScheduler], + }) + .useMocker(() => createMock()) + .compile(); + + scheduler = module.get(AccountPurgeScheduler); + purge = module.get(AccountPurgeService); + config = module.get(ConfigService); + + purge.purgeExpiredAccounts.mockResolvedValue({ + claimed: 0, + purged: 0, + failed: 0, + }); + }); + + it('when the tick fires on a cronjob instance, then runs a purge', async () => { + enable(true); + + await scheduler.handleCron(); + + expect(config.get).toHaveBeenCalledWith('executeCronjobs'); + expect(purge.purgeExpiredAccounts).toHaveBeenCalled(); + }); + + it('when this is not a cronjob instance, then the tick does nothing', async () => { + enable(false); + + await scheduler.handleCron(); + + expect(purge.purgeExpiredAccounts).not.toHaveBeenCalled(); + }); + + it('when the flag is unset, then purging stays off', async () => { + config.get.mockReturnValue(undefined); + + await scheduler.handleCron(); + + expect(purge.purgeExpiredAccounts).not.toHaveBeenCalled(); + }); + + it('when a run is still going, then the next tick is skipped', async () => { + enable(true); + let release!: () => void; + purge.purgeExpiredAccounts.mockReturnValue( + new Promise((resolve) => { + release = () => resolve({ claimed: 0, purged: 0, failed: 0 }); + }), + ); + + const first = scheduler.handleCron(); + await scheduler.handleCron(); + + expect(purge.purgeExpiredAccounts).toHaveBeenCalledTimes(1); + + release(); + await first; + }); + + it('when a run fails, then the tick swallows it and lets the next one through', async () => { + enable(true); + purge.purgeExpiredAccounts.mockRejectedValueOnce(new Error('DB down')); + + await expect(scheduler.handleCron()).resolves.toBeUndefined(); + + await scheduler.handleCron(); + + expect(purge.purgeExpiredAccounts).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/modules/jobs/tasks/account-purge/account-purge.scheduler.ts b/src/modules/jobs/tasks/account-purge/account-purge.scheduler.ts new file mode 100644 index 0000000..d048119 --- /dev/null +++ b/src/modules/jobs/tasks/account-purge/account-purge.scheduler.ts @@ -0,0 +1,41 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { Cron, CronExpression } from '@nestjs/schedule'; +import { AccountPurgeService } from '../../../account/account-purge.service.js'; +import { JobName } from '../../constants.js'; + +@Injectable() +export class AccountPurgeScheduler { + private readonly logger = new Logger(AccountPurgeScheduler.name); + private running = false; + + constructor( + private readonly purge: AccountPurgeService, + private readonly config: ConfigService, + ) {} + + @Cron(CronExpression.EVERY_DAY_AT_4AM, { + name: JobName.ACCOUNT_PURGE, + timeZone: 'UTC', + }) + async handleCron(): Promise { + if (!this.config.get('executeCronjobs')) return; + + if (this.running) { + this.logger.warn('Previous purge run is still going; skipping this tick'); + return; + } + + this.running = true; + try { + await this.purge.purgeExpiredAccounts(); + } catch (error) { + this.logger.error( + `Purge run failed: ${(error as Error).message}`, + (error as Error).stack, + ); + } finally { + this.running = false; + } + } +} diff --git a/src/modules/usage/mail-usage.service.ts b/src/modules/usage/mail-usage.service.ts index 7d5af97..74c68c7 100644 --- a/src/modules/usage/mail-usage.service.ts +++ b/src/modules/usage/mail-usage.service.ts @@ -1,7 +1,7 @@ import { Injectable, Logger } from '@nestjs/common'; import { - BridgeApiError, BridgeClient, + isBridgeNotFound, } from '../infrastructure/bridge/bridge.service.js'; import { DuplicateEntryKeyError, @@ -103,7 +103,7 @@ export class MailUsageService { existing.bridgeEntryId, ); } catch (error) { - if (!(error instanceof BridgeApiError && error.statusCode === 404)) { + if (!isBridgeNotFound(error)) { throw error; } this.logger.debug(