diff --git a/packages/twenty-server/src/database/commands/cron-register-all.command.ts b/packages/twenty-server/src/database/commands/cron-register-all.command.ts index e85424e796..8588fab285 100644 --- a/packages/twenty-server/src/database/commands/cron-register-all.command.ts +++ b/packages/twenty-server/src/database/commands/cron-register-all.command.ts @@ -9,6 +9,7 @@ import { ApplicationVersionCheckCronCommand } from 'src/engine/core-modules/appl import { BillingReminderCronCommand } from 'src/engine/core-modules/billing/reminders/crons/commands/billing-reminder.cron.command'; import { EnterpriseKeyValidationCronCommand } from 'src/engine/core-modules/enterprise/cron/command/enterprise-key-validation.cron.command'; import { EventLogCleanupCronCommand } from 'src/engine/core-modules/event-logs/cleanup/commands/event-log-cleanup.cron.command'; +import { PendingFileCleanupCronCommand } from 'src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command'; import { RotateSigningKeysCronCommand } from 'src/engine/core-modules/jwt/crons/commands/rotate-signing-keys.cron.command'; import { CronTriggerCronCommand } from 'src/engine/core-modules/logic-function/logic-function-trigger/triggers/cron/cron-trigger.cron.command'; import { CheckPublicDomainsValidRecordsCronCommand } from 'src/engine/core-modules/public-domain/crons/commands/check-public-domains-valid-records.cron.command'; @@ -65,6 +66,7 @@ export class CronRegisterAllCommand extends CommandRunner { private readonly marketplaceCatalogSyncCronCommand: MarketplaceCatalogSyncCronCommand, private readonly applicationVersionCheckCronCommand: ApplicationVersionCheckCronCommand, private readonly staleRegistrationCleanupCronCommand: StaleRegistrationCleanupCronCommand, + private readonly pendingFileCleanupCronCommand: PendingFileCleanupCronCommand, private readonly billingReminderCronCommand: BillingReminderCronCommand, private readonly twentyConfigService: TwentyConfigService, ) { @@ -183,6 +185,10 @@ export class CronRegisterAllCommand extends CommandRunner { name: 'StaleRegistrationCleanup', command: this.staleRegistrationCleanupCronCommand, }, + { + name: 'PendingFileCleanup', + command: this.pendingFileCleanupCronCommand, + }, { name: 'BillingReminder', command: this.billingReminderCronCommand, diff --git a/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command.ts b/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command.ts new file mode 100644 index 0000000000..8d734e2421 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command.ts @@ -0,0 +1,33 @@ +import { Command, CommandRunner } from 'nest-commander'; + +import { PENDING_FILE_CLEANUP_CRON_PATTERN } from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants'; +import { PendingFileCleanupCronJob } from 'src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job'; +import { InjectMessageQueue } from 'src/engine/core-modules/message-queue/decorators/message-queue.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; +import { MessageQueueService } from 'src/engine/core-modules/message-queue/services/message-queue.service'; + +@Command({ + name: 'cron:pending-file-cleanup', + description: + 'Starts a cron job to clean up stale pending direct file uploads', +}) +export class PendingFileCleanupCronCommand extends CommandRunner { + constructor( + @InjectMessageQueue(MessageQueue.cronQueue) + private readonly messageQueueService: MessageQueueService, + ) { + super(); + } + + async run(): Promise { + await this.messageQueueService.addCron({ + jobName: PendingFileCleanupCronJob.name, + data: undefined, + options: { + repeat: { + pattern: PENDING_FILE_CLEANUP_CRON_PATTERN, + }, + }, + }); + } +} diff --git a/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants.ts b/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants.ts new file mode 100644 index 0000000000..0201882e66 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants.ts @@ -0,0 +1,7 @@ +export const PENDING_FILE_CLEANUP_CRON_PATTERN = '0 * * * *'; + +// A pending file only becomes reapable long after its upload URL has expired, +// so the cleanup can never race a legitimate in-flight upload. +export const PENDING_FILE_MAX_AGE_MS = 24 * 60 * 60 * 1000; + +export const PENDING_FILE_CLEANUP_BATCH_SIZE = 200; diff --git a/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job.ts b/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job.ts new file mode 100644 index 0000000000..d601439a6f --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job.ts @@ -0,0 +1,41 @@ +import { Injectable, Logger } from '@nestjs/common'; + +import { SentryCronMonitor } from 'src/engine/core-modules/cron/sentry-cron-monitor.decorator'; +import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service'; +import { PENDING_FILE_CLEANUP_CRON_PATTERN } from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants'; +import { PendingFileCleanupService } from 'src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service'; +import { Process } from 'src/engine/core-modules/message-queue/decorators/process.decorator'; +import { Processor } from 'src/engine/core-modules/message-queue/decorators/processor.decorator'; +import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants'; + +@Injectable() +@Processor(MessageQueue.cronQueue) +export class PendingFileCleanupCronJob { + private readonly logger = new Logger(PendingFileCleanupCronJob.name); + + constructor( + private readonly pendingFileCleanupService: PendingFileCleanupService, + private readonly exceptionHandlerService: ExceptionHandlerService, + ) {} + + @Process(PendingFileCleanupCronJob.name) + @SentryCronMonitor( + PendingFileCleanupCronJob.name, + PENDING_FILE_CLEANUP_CRON_PATTERN, + ) + async handle(): Promise { + try { + const deletedCount = + await this.pendingFileCleanupService.cleanupStalePendingFiles(); + + if (deletedCount > 0) { + this.logger.log( + `Pending file cleanup completed: ${deletedCount} stale file(s) deleted`, + ); + } + } catch (error) { + this.exceptionHandlerService.captureExceptions([error]); + throw error; + } + } +} diff --git a/packages/twenty-server/src/engine/core-modules/file/file-upload/file-upload.module.ts b/packages/twenty-server/src/engine/core-modules/file/file-upload/file-upload.module.ts index 96fae862e7..29ffab2c1a 100644 --- a/packages/twenty-server/src/engine/core-modules/file/file-upload/file-upload.module.ts +++ b/packages/twenty-server/src/engine/core-modules/file/file-upload/file-upload.module.ts @@ -6,9 +6,12 @@ import { ApplicationModule } from 'src/engine/core-modules/application/applicati import { FileStorageModule } from 'src/engine/core-modules/file-storage/file-storage.module'; import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity'; import { FileUploadController } from 'src/engine/core-modules/file/file-upload/controllers/file-upload.controller'; +import { PendingFileCleanupCronCommand } from 'src/engine/core-modules/file/file-upload/crons/commands/pending-file-cleanup.cron.command'; +import { PendingFileCleanupCronJob } from 'src/engine/core-modules/file/file-upload/crons/jobs/pending-file-cleanup.cron.job'; import { FileUploadTokenGuard } from 'src/engine/core-modules/file/file-upload/guards/file-upload-token.guard'; import { FileUploadResolver } from 'src/engine/core-modules/file/file-upload/resolvers/file-upload.resolver'; import { FileUploadService } from 'src/engine/core-modules/file/file-upload/services/file-upload.service'; +import { PendingFileCleanupService } from 'src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service'; import { FileUrlModule } from 'src/engine/core-modules/file/file-url/file-url.module'; import { JwtModule } from 'src/engine/core-modules/jwt/jwt.module'; import { FieldMetadataEntity } from 'src/engine/metadata-modules/field-metadata/field-metadata.entity'; @@ -32,9 +35,12 @@ import { provideWorkspaceScopedRepository } from 'src/engine/twenty-orm/workspac FileUploadService, FileUploadResolver, FileUploadTokenGuard, + PendingFileCleanupService, + PendingFileCleanupCronJob, + PendingFileCleanupCronCommand, provideWorkspaceScopedRepository(FileEntity), ], - exports: [FileUploadService], + exports: [FileUploadService, PendingFileCleanupCronCommand], controllers: [FileUploadController], }) export class FileUploadModule {} diff --git a/packages/twenty-server/src/engine/core-modules/file/file-upload/services/__tests__/pending-file-cleanup.service.spec.ts b/packages/twenty-server/src/engine/core-modules/file/file-upload/services/__tests__/pending-file-cleanup.service.spec.ts new file mode 100644 index 0000000000..d5275857e3 --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/file/file-upload/services/__tests__/pending-file-cleanup.service.spec.ts @@ -0,0 +1,179 @@ +import { Test, type TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; + +import { FileFolder } from 'twenty-shared/types'; + +import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; +import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service'; +import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity'; +import { + PENDING_FILE_CLEANUP_BATCH_SIZE, + PENDING_FILE_MAX_AGE_MS, +} from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants'; +import { PendingFileCleanupService } from 'src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service'; +import { FILE_STATUS } from 'src/engine/core-modules/file/types/file-status.types'; + +describe('PendingFileCleanupService', () => { + let service: PendingFileCleanupService; + + const fileRepository = { + find: jest.fn(), + delete: jest.fn(), + }; + + const applicationRepository = { + findOne: jest.fn(), + }; + + const fileStorageService = { + deleteFile: jest.fn(), + }; + + beforeEach(async () => { + jest.clearAllMocks(); + + applicationRepository.findOne.mockResolvedValue({ + id: 'application-id', + universalIdentifier: 'application-uid', + }); + fileRepository.delete.mockResolvedValue({ affected: 1 }); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + PendingFileCleanupService, + { + provide: getRepositoryToken(FileEntity), + useValue: fileRepository, + }, + { + provide: getRepositoryToken(ApplicationEntity), + useValue: applicationRepository, + }, + { + provide: FileStorageService, + useValue: fileStorageService, + }, + ], + }).compile(); + + service = module.get(PendingFileCleanupService); + }); + + it('should query for stale pending files with the batch size cap', async () => { + fileRepository.find.mockResolvedValue([]); + + const before = Date.now() - PENDING_FILE_MAX_AGE_MS; + + await service.cleanupStalePendingFiles(); + + const after = Date.now() - PENDING_FILE_MAX_AGE_MS; + + expect(fileRepository.find).toHaveBeenCalledTimes(1); + + const where = fileRepository.find.mock.calls[0][0].where; + + expect(where.status).toBe(FILE_STATUS.PENDING); + // LessThan wraps the threshold in _value + const threshold = where.createdAt._value.getTime(); + + expect(threshold).toBeGreaterThanOrEqual(before); + expect(threshold).toBeLessThanOrEqual(after); + expect(fileRepository.find.mock.calls[0][0].take).toBe( + PENDING_FILE_CLEANUP_BATCH_SIZE, + ); + }); + + it('should delete each stale file guarded by status and return the deleted count', async () => { + fileRepository.find.mockResolvedValue([ + { + id: 'file-1', + workspaceId: 'workspace-1', + applicationId: 'application-id', + path: `${FileFolder.FilesField}/some/file-1.png`, + }, + { + id: 'file-2', + workspaceId: 'workspace-2', + applicationId: 'application-id', + path: `${FileFolder.Workflow}/file-2.pdf`, + }, + ]); + + const deletedCount = await service.cleanupStalePendingFiles(); + + expect(deletedCount).toBe(2); + expect(fileRepository.delete).toHaveBeenNthCalledWith(1, { + id: 'file-1', + status: FILE_STATUS.PENDING, + }); + expect(fileRepository.delete).toHaveBeenNthCalledWith(2, { + id: 'file-2', + status: FILE_STATUS.PENDING, + }); + expect(fileStorageService.deleteFile).toHaveBeenNthCalledWith(1, { + workspaceId: 'workspace-1', + applicationUniversalIdentifier: 'application-uid', + fileFolder: FileFolder.FilesField, + resourcePath: 'some/file-1.png', + }); + expect(fileStorageService.deleteFile).toHaveBeenNthCalledWith(2, { + workspaceId: 'workspace-2', + applicationUniversalIdentifier: 'application-uid', + fileFolder: FileFolder.Workflow, + resourcePath: 'file-2.pdf', + }); + }); + + it('should skip storage deletion when the file was confirmed concurrently', async () => { + fileRepository.find.mockResolvedValue([ + { + id: 'file-1', + workspaceId: 'workspace-1', + applicationId: 'application-id', + path: `${FileFolder.FilesField}/file-1.png`, + }, + ]); + // A racing completeFileUpload promoted the row, so the guarded delete is a no-op. + fileRepository.delete.mockResolvedValueOnce({ affected: 0 }); + + const deletedCount = await service.cleanupStalePendingFiles(); + + expect(deletedCount).toBe(0); + expect(fileStorageService.deleteFile).not.toHaveBeenCalled(); + }); + + it('should keep cleaning remaining files when one deletion fails', async () => { + fileRepository.find.mockResolvedValue([ + { + id: 'file-1', + workspaceId: 'workspace-1', + applicationId: 'application-id', + path: `${FileFolder.FilesField}/file-1.png`, + }, + { + id: 'file-2', + workspaceId: 'workspace-2', + applicationId: 'application-id', + path: `${FileFolder.FilesField}/file-2.png`, + }, + ]); + fileStorageService.deleteFile + .mockRejectedValueOnce(new Error('storage exploded')) + .mockResolvedValueOnce(undefined); + + const deletedCount = await service.cleanupStalePendingFiles(); + + expect(deletedCount).toBe(1); + expect(fileStorageService.deleteFile).toHaveBeenCalledTimes(2); + }); + + it('should not delete anything when there are no stale files', async () => { + fileRepository.find.mockResolvedValue([]); + + const deletedCount = await service.cleanupStalePendingFiles(); + + expect(deletedCount).toBe(0); + expect(fileRepository.delete).not.toHaveBeenCalled(); + expect(fileStorageService.deleteFile).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/twenty-server/src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service.ts b/packages/twenty-server/src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service.ts new file mode 100644 index 0000000000..91d9eef6ed --- /dev/null +++ b/packages/twenty-server/src/engine/core-modules/file/file-upload/services/pending-file-cleanup.service.ts @@ -0,0 +1,98 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; + +import { FileFolder } from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { LessThan, Repository } from 'typeorm'; + +import { ApplicationEntity } from 'src/engine/core-modules/application/application.entity'; +import { FileStorageService } from 'src/engine/core-modules/file-storage/file-storage.service'; +import { FileEntity } from 'src/engine/core-modules/file/entities/file.entity'; +import { + PENDING_FILE_CLEANUP_BATCH_SIZE, + PENDING_FILE_MAX_AGE_MS, +} from 'src/engine/core-modules/file/file-upload/crons/constants/pending-file-cleanup.constants'; +import { FILE_STATUS } from 'src/engine/core-modules/file/types/file-status.types'; +import { removeFileFolderFromFileEntityPath } from 'src/engine/core-modules/file/utils/remove-file-folder-from-file-entity-path.utils'; + +@Injectable() +export class PendingFileCleanupService { + private readonly logger = new Logger(PendingFileCleanupService.name); + + constructor( + // eslint-disable-next-line twenty/prefer-workspace-scoped-repository -- the reaper runs in a cron with no workspace context and must sweep stale PENDING files across every workspace + @InjectRepository(FileEntity) + private readonly fileRepository: Repository, + // eslint-disable-next-line twenty/prefer-workspace-scoped-repository -- resolves the application universalIdentifier of a cross-workspace file while reaping outside any workspace context + @InjectRepository(ApplicationEntity) + private readonly applicationRepository: Repository, + private readonly fileStorageService: FileStorageService, + ) {} + + // Deletes file records stuck in PENDING (direct uploads that were initiated + // but never completed) together with any partially uploaded object. Never + // promotes to UPLOADED: a file that was never confirmed is referenced by + // nothing, the client recovery path is re-uploading under a fresh fileId. + async cleanupStalePendingFiles(): Promise { + const staleThreshold = new Date(Date.now() - PENDING_FILE_MAX_AGE_MS); + + const staleFiles = await this.fileRepository.find({ + where: { + status: FILE_STATUS.PENDING, + createdAt: LessThan(staleThreshold), + }, + take: PENDING_FILE_CLEANUP_BATCH_SIZE, + }); + + let deletedCount = 0; + + for (const file of staleFiles) { + try { + // Claim the row atomically: delete it only while it is still PENDING. + // If completeFileUpload promoted it to UPLOADED between the fetch above + // and here, the delete affects no rows and the now-live file (and its + // object) are left untouched. + const { affected } = await this.fileRepository.delete({ + id: file.id, + status: FILE_STATUS.PENDING, + }); + + if (!isDefined(affected) || affected === 0) { + continue; + } + + await this.deleteStorageObject(file); + + deletedCount++; + } catch (error) { + this.logger.warn( + `Failed to clean up stale pending file ${file.id} in workspace ${file.workspaceId}: ${error.message}`, + ); + } + } + + return deletedCount; + } + + // The row has already been removed, so this only tidies the (possibly + // partial, possibly absent) storage object. A failure here leaks bytes but + // never data, so it is logged rather than retried. + private async deleteStorageObject(file: FileEntity): Promise { + const [fileFolder] = file.path.split('/'); + + const application = await this.applicationRepository.findOne({ + where: { id: file.applicationId, workspaceId: file.workspaceId }, + }); + + if (!isDefined(application)) { + return; + } + + await this.fileStorageService.deleteFile({ + workspaceId: file.workspaceId, + applicationUniversalIdentifier: application.universalIdentifier, + fileFolder: fileFolder as FileFolder, + resourcePath: removeFileFolderFromFileEntityPath(file.path), + }); + } +}