Automatically clean up soft-deleted records after X days. (#14862)
Closes #14726 ### Added - `trashRetentionDays` field to workspace entity (default: 14 days) - Automated trash cleanup using BullMQ jobs - Daily cron (00:10 UTC) that enqueues cleanup jobs for all active workspaces - Per-workspace limit: 100k records deleted per day - Calendar-based retention: records deleted on day X are cleaned up X+14 days later (at midnight UTC boundaries) ### Architecture - **Cron (WorkspaceTrashCleanupCronJob):** Runs daily, enqueues jobs in parallel for all workspaces - **Job (WorkspaceTrashCleanupJob):** Processes individual workspace cleanup - **Service (WorkspaceTrashCleanupService):** Discovers tables with `deletedAt`, deletes old records with quota enforcement - **Command:** `npx nx run twenty-server:command cron:workspace:cleanup-trash` to register the cron ### Testing - Unit tests for service with 100% coverage of public API - Tested quota enforcement, error handling, and edge cases --------- Co-authored-by: Félix Malfait <felix.malfait@gmail.com>
This commit is contained in:
Abdullah.
+208
@@ -0,0 +1,208 @@
|
||||
import { Test, type TestingModule } from '@nestjs/testing';
|
||||
|
||||
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
|
||||
import { TrashCleanupService } from 'src/engine/trash-cleanup/services/trash-cleanup.service';
|
||||
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
|
||||
|
||||
describe('TrashCleanupService', () => {
|
||||
let service: TrashCleanupService;
|
||||
let mockFlatEntityMapsCacheService: any;
|
||||
let mockTwentyORMGlobalManager: any;
|
||||
|
||||
beforeEach(async () => {
|
||||
mockFlatEntityMapsCacheService = {
|
||||
getOrRecomputeManyOrAllFlatEntityMaps: jest.fn(),
|
||||
};
|
||||
|
||||
mockTwentyORMGlobalManager = {
|
||||
getRepositoryForWorkspace: jest.fn(),
|
||||
};
|
||||
|
||||
const module: TestingModule = await Test.createTestingModule({
|
||||
providers: [
|
||||
TrashCleanupService,
|
||||
{
|
||||
provide: WorkspaceManyOrAllFlatEntityMapsCacheService,
|
||||
useValue: mockFlatEntityMapsCacheService,
|
||||
},
|
||||
{
|
||||
provide: TwentyORMGlobalManager,
|
||||
useValue: mockTwentyORMGlobalManager,
|
||||
},
|
||||
],
|
||||
}).compile();
|
||||
|
||||
service = module.get<TrashCleanupService>(TrashCleanupService);
|
||||
|
||||
// Suppress logger output in tests
|
||||
jest.spyOn(service['logger'], 'log').mockImplementation();
|
||||
jest.spyOn(service['logger'], 'error').mockImplementation();
|
||||
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it('should be defined', () => {
|
||||
expect(service).toBeDefined();
|
||||
});
|
||||
|
||||
describe('cleanupWorkspaceTrash', () => {
|
||||
const createRepositoryMock = (name: string, initialCount: number) => {
|
||||
let remaining = initialCount;
|
||||
let counter = 0;
|
||||
|
||||
return {
|
||||
find: jest.fn().mockImplementation(({ take }) => {
|
||||
const amount = Math.min(take ?? remaining, remaining);
|
||||
const records = Array.from({ length: amount }, () => ({
|
||||
id: `${name}-${counter++}`,
|
||||
}));
|
||||
|
||||
remaining -= amount;
|
||||
|
||||
return Promise.resolve(records);
|
||||
}),
|
||||
delete: jest.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
};
|
||||
|
||||
const setObjectMetadataCache = (
|
||||
entries: Array<{ id: string; nameSingular: string }>,
|
||||
) => {
|
||||
const byId = entries.reduce<Record<string, any>>(
|
||||
(acc, { id, nameSingular }) => {
|
||||
acc[id] = {
|
||||
id,
|
||||
nameSingular,
|
||||
};
|
||||
|
||||
return acc;
|
||||
},
|
||||
{},
|
||||
);
|
||||
|
||||
mockFlatEntityMapsCacheService.getOrRecomputeManyOrAllFlatEntityMaps.mockResolvedValue(
|
||||
{
|
||||
flatObjectMetadataMaps: {
|
||||
byId,
|
||||
idByUniversalIdentifier: {},
|
||||
},
|
||||
},
|
||||
);
|
||||
};
|
||||
|
||||
it('should return deleted count when cleanup succeeds', async () => {
|
||||
setObjectMetadataCache([
|
||||
{ id: 'obj-company', nameSingular: 'company' },
|
||||
{ id: 'obj-person', nameSingular: 'person' },
|
||||
]);
|
||||
|
||||
const companyRepository = createRepositoryMock('company', 2);
|
||||
const personRepository = createRepositoryMock('person', 1);
|
||||
|
||||
mockTwentyORMGlobalManager.getRepositoryForWorkspace
|
||||
.mockResolvedValueOnce(companyRepository)
|
||||
.mockResolvedValueOnce(personRepository);
|
||||
|
||||
const result = await service.cleanupWorkspaceTrash({
|
||||
workspaceId: 'workspace-id',
|
||||
trashRetentionDays: 14,
|
||||
});
|
||||
|
||||
expect(result).toEqual(3);
|
||||
expect(companyRepository.find).toHaveBeenCalled();
|
||||
expect(personRepository.find).toHaveBeenCalled();
|
||||
expect(companyRepository.delete).toHaveBeenCalledTimes(1);
|
||||
expect(personRepository.delete).toHaveBeenCalledTimes(1);
|
||||
|
||||
const findArgs = companyRepository.find.mock.calls[0][0];
|
||||
|
||||
expect(findArgs.withDeleted).toBe(true);
|
||||
expect(findArgs.order).toEqual({ deletedAt: 'ASC' });
|
||||
});
|
||||
|
||||
it('should return zero when no objects are found', async () => {
|
||||
setObjectMetadataCache([]);
|
||||
|
||||
const result = await service.cleanupWorkspaceTrash({
|
||||
workspaceId: 'workspace-id',
|
||||
trashRetentionDays: 14,
|
||||
});
|
||||
|
||||
expect(result).toEqual(0);
|
||||
expect(
|
||||
mockTwentyORMGlobalManager.getRepositoryForWorkspace,
|
||||
).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should respect max records limit across objects', async () => {
|
||||
(service as any).maxRecordsPerWorkspace = 3;
|
||||
(service as any).batchSize = 3;
|
||||
setObjectMetadataCache([
|
||||
{ id: 'obj-company', nameSingular: 'company' },
|
||||
{ id: 'obj-person', nameSingular: 'person' },
|
||||
]);
|
||||
|
||||
const companyRepository = createRepositoryMock('company', 2);
|
||||
const personRepository = createRepositoryMock('person', 5);
|
||||
|
||||
mockTwentyORMGlobalManager.getRepositoryForWorkspace
|
||||
.mockResolvedValueOnce(companyRepository)
|
||||
.mockResolvedValueOnce(personRepository);
|
||||
|
||||
const result = await service.cleanupWorkspaceTrash({
|
||||
workspaceId: 'workspace-id',
|
||||
trashRetentionDays: 14,
|
||||
});
|
||||
|
||||
expect(result).toEqual(3);
|
||||
expect(companyRepository.delete).toHaveBeenCalledTimes(1);
|
||||
expect(personRepository.delete).toHaveBeenCalledTimes(1);
|
||||
const personDeleteArgs = personRepository.delete.mock.calls[0][0];
|
||||
const deletedIds =
|
||||
personDeleteArgs.id._value ?? personDeleteArgs.id.value;
|
||||
|
||||
expect(deletedIds).toHaveLength(1);
|
||||
expect(personRepository.find).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('should ignore objects without soft deleted records', async () => {
|
||||
setObjectMetadataCache([{ id: 'obj-company', nameSingular: 'company' }]);
|
||||
|
||||
const companyRepository = createRepositoryMock('company', 0);
|
||||
|
||||
mockTwentyORMGlobalManager.getRepositoryForWorkspace.mockResolvedValueOnce(
|
||||
companyRepository,
|
||||
);
|
||||
|
||||
const result = await service.cleanupWorkspaceTrash({
|
||||
workspaceId: 'workspace-id',
|
||||
trashRetentionDays: 14,
|
||||
});
|
||||
|
||||
expect(result).toEqual(0);
|
||||
expect(companyRepository.delete).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should delete records across multiple batches', async () => {
|
||||
setObjectMetadataCache([{ id: 'obj-company', nameSingular: 'company' }]);
|
||||
|
||||
const companyRepository = createRepositoryMock('company', 5);
|
||||
|
||||
mockTwentyORMGlobalManager.getRepositoryForWorkspace.mockResolvedValueOnce(
|
||||
companyRepository,
|
||||
);
|
||||
|
||||
(service as any).batchSize = 2;
|
||||
(service as any).maxRecordsPerWorkspace = 10;
|
||||
|
||||
const result = await service.cleanupWorkspaceTrash({
|
||||
workspaceId: 'workspace-id',
|
||||
trashRetentionDays: 14,
|
||||
});
|
||||
|
||||
expect(result).toEqual(5);
|
||||
expect(companyRepository.find).toHaveBeenCalledTimes(4);
|
||||
expect(companyRepository.delete).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,146 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
|
||||
import { isDefined } from 'twenty-shared/utils';
|
||||
import { In, LessThan } from 'typeorm';
|
||||
|
||||
import { WorkspaceManyOrAllFlatEntityMapsCacheService } from 'src/engine/metadata-modules/flat-entity/services/workspace-many-or-all-flat-entity-maps-cache.service';
|
||||
import {
|
||||
TRASH_CLEANUP_BATCH_SIZE,
|
||||
TRASH_CLEANUP_MAX_RECORDS_PER_WORKSPACE,
|
||||
} from 'src/engine/trash-cleanup/constants/trash-cleanup.constants';
|
||||
import { TwentyORMGlobalManager } from 'src/engine/twenty-orm/twenty-orm-global.manager';
|
||||
|
||||
export type TrashCleanupInput = {
|
||||
workspaceId: string;
|
||||
trashRetentionDays: number;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class TrashCleanupService {
|
||||
private readonly logger = new Logger(TrashCleanupService.name);
|
||||
private readonly maxRecordsPerWorkspace =
|
||||
TRASH_CLEANUP_MAX_RECORDS_PER_WORKSPACE;
|
||||
private readonly batchSize = TRASH_CLEANUP_BATCH_SIZE;
|
||||
|
||||
constructor(
|
||||
private readonly flatEntityMapsCacheService: WorkspaceManyOrAllFlatEntityMapsCacheService,
|
||||
private readonly twentyORMGlobalManager: TwentyORMGlobalManager,
|
||||
) {}
|
||||
|
||||
async cleanupWorkspaceTrash(input: TrashCleanupInput): Promise<number> {
|
||||
const { workspaceId, trashRetentionDays } = input;
|
||||
|
||||
const { flatObjectMetadataMaps } =
|
||||
await this.flatEntityMapsCacheService.getOrRecomputeManyOrAllFlatEntityMaps(
|
||||
{
|
||||
workspaceId,
|
||||
flatMapsKeys: ['flatObjectMetadataMaps'],
|
||||
},
|
||||
);
|
||||
|
||||
const objectNames = Object.values(flatObjectMetadataMaps.byId ?? {})
|
||||
.map((metadata) => metadata?.nameSingular)
|
||||
.filter(isDefined);
|
||||
|
||||
if (objectNames.length === 0) {
|
||||
this.logger.log(`No objects found in workspace ${workspaceId}`);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
const cutoffDate = this.calculateCutoffDate(trashRetentionDays);
|
||||
let deletedCount = 0;
|
||||
|
||||
for (const objectName of objectNames) {
|
||||
if (deletedCount >= this.maxRecordsPerWorkspace) {
|
||||
this.logger.log(
|
||||
`Reached deletion limit (${this.maxRecordsPerWorkspace}) for workspace ${workspaceId}`,
|
||||
);
|
||||
break;
|
||||
}
|
||||
|
||||
const remainingQuota = this.maxRecordsPerWorkspace - deletedCount;
|
||||
const deletedForObject = await this.deleteSoftDeletedRecords({
|
||||
workspaceId,
|
||||
objectName,
|
||||
cutoffDate,
|
||||
remainingQuota,
|
||||
});
|
||||
|
||||
if (deletedForObject > 0) {
|
||||
this.logger.log(
|
||||
`Deleted ${deletedForObject} record(s) from ${objectName} in workspace ${workspaceId}`,
|
||||
);
|
||||
}
|
||||
|
||||
deletedCount += deletedForObject;
|
||||
}
|
||||
|
||||
this.logger.log(
|
||||
`Deleted ${deletedCount} record(s) from workspace ${workspaceId}`,
|
||||
);
|
||||
|
||||
return deletedCount;
|
||||
}
|
||||
|
||||
private async deleteSoftDeletedRecords({
|
||||
workspaceId,
|
||||
objectName,
|
||||
cutoffDate,
|
||||
remainingQuota,
|
||||
}: {
|
||||
workspaceId: string;
|
||||
objectName: string;
|
||||
cutoffDate: Date;
|
||||
remainingQuota: number;
|
||||
}): Promise<number> {
|
||||
if (remainingQuota <= 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const repository =
|
||||
await this.twentyORMGlobalManager.getRepositoryForWorkspace(
|
||||
workspaceId,
|
||||
objectName,
|
||||
{ shouldBypassPermissionChecks: true },
|
||||
);
|
||||
|
||||
let deleted = 0;
|
||||
|
||||
while (deleted < remainingQuota) {
|
||||
const take = Math.min(this.batchSize, remainingQuota - deleted);
|
||||
|
||||
const recordsToDelete = await repository.find({
|
||||
withDeleted: true,
|
||||
select: ['id'],
|
||||
where: {
|
||||
deletedAt: LessThan(cutoffDate),
|
||||
},
|
||||
order: { deletedAt: 'ASC' },
|
||||
take,
|
||||
loadEagerRelations: false,
|
||||
});
|
||||
|
||||
if (recordsToDelete.length === 0) {
|
||||
break;
|
||||
}
|
||||
|
||||
await repository.delete({
|
||||
id: In(recordsToDelete.map((record) => record.id)),
|
||||
});
|
||||
|
||||
deleted += recordsToDelete.length;
|
||||
}
|
||||
|
||||
return deleted;
|
||||
}
|
||||
|
||||
private calculateCutoffDate(trashRetentionDays: number): Date {
|
||||
const cutoffDate = new Date();
|
||||
|
||||
cutoffDate.setUTCHours(0, 0, 0, 0);
|
||||
cutoffDate.setDate(cutoffDate.getDate() - trashRetentionDays + 1);
|
||||
|
||||
return cutoffDate;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user