From a71d949a0c6398253523ee4884a53e4b4490397f Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 5 Aug 2025 22:17:50 +0530 Subject: [PATCH] feat: add cron job to cleanup old queued form responses (#22035) * feat: add cron job to cleanup old queued form responses - Add cleanup cron job for App_RoutingForms_QueuedFormResponse table - Delete records with null actualResponseId older than 1 hour - Schedule to run twice daily (every 12 hours) via vercel.json - Follow existing cron job patterns for authentication and structure Co-Authored-By: hariom@cal.com * fix: correct date comparison logic to delete older records - Change from lte to gte in createdAt comparison - Now properly deletes records older than 1 hour Co-Authored-By: hariom@cal.com * refactor: move cleanup logic to routing forms lib and import in API endpoint - Create cleanupExpiredQueuedFormResponses function in routing forms lib - Update API endpoint to import and use the new cleanup function - Follow established patterns for code organization in Cal.com - Maintain same functionality and authentication logic Co-Authored-By: hariom@cal.com * refactor: move entire cron logic to routing-forms cron folder - Create routing-forms/cron/queuedFormResponseCleanup.ts with complete handler - Move authentication, HTTP handling, and database operations to routing-forms - Update API endpoint to simply import and call the cron handler - Remove old lib cleanup function - Follow user's preferred organizational pattern Co-Authored-By: hariom@cal.com * fix: correct date comparison logic to delete records older than 1 hour - Change from gte (greater than or equal) to lt (less than) cutoffTime - Now properly deletes records where createdAt < cutoffTime (older than 1 hour) - Previous logic was deleting newer records instead of older ones Co-Authored-By: hariom@cal.com * fixes * refactor: reorganize queuedFormResponse into domain-driven structure - Move repository and service files to routingForm/queuedFormResponse subdirectory - Clean up unused methods and tests for better maintainability - Fix import paths and type definitions - Ensure all tests pass and maintain backward compatibility * fix: align QueuedFormResponse interface types with Prisma schema - Update actualResponseId type from string to number to match Prisma Int type - Make updatedAt nullable (Date | null) to match Prisma DateTime? type - Fix import path for QueuedFormResponseRepository interface Co-Authored-By: hariom@cal.com * refactor: reorganize queuedFormResponse into domain-driven structure - Move repository and service files to routingForm/queuedFormResponse subdirectory - Clean up unused methods and tests for better maintainability - Fix import paths and type definitions - Ensure all tests pass and maintain backward compatibility * test: add comprehensive where clause validation tests and missing test coverage * chore: update cron schedule to weekly for 7-day retention period * revert: restore 12-hour cron schedule for better batch management * Add integration tests * refactor: apply test data builder pattern to queuedFormResponse integration tests --------- Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: hariom@cal.com Co-authored-by: Hariom --- .../cron/queuedFormResponseCleanup/route.ts | 5 + apps/web/vercel.json | 4 + .../cron/queuedFormResponseCleanup.ts | 47 ++ ...FormResponseRepository.integration-test.ts | 432 ++++++++++++++++++ .../PrismaQueuedFormResponseRepository.ts | 53 +++ .../QueuedFormResponseRepository.interface.ts | 30 ++ .../QueuedFormResponseService.test.ts | 152 ++++++ .../QueuedFormResponseService.ts | 96 ++++ 8 files changed, 819 insertions(+) create mode 100644 apps/web/app/api/cron/queuedFormResponseCleanup/route.ts create mode 100644 packages/app-store/routing-forms/cron/queuedFormResponseCleanup.ts create mode 100644 packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.integration-test.ts create mode 100644 packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.ts create mode 100644 packages/lib/server/repository/routingForm/queuedFormResponse/QueuedFormResponseRepository.interface.ts create mode 100644 packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.test.ts create mode 100644 packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.ts diff --git a/apps/web/app/api/cron/queuedFormResponseCleanup/route.ts b/apps/web/app/api/cron/queuedFormResponseCleanup/route.ts new file mode 100644 index 0000000000..39c593ee66 --- /dev/null +++ b/apps/web/app/api/cron/queuedFormResponseCleanup/route.ts @@ -0,0 +1,5 @@ +import { defaultResponderForAppDir } from "app/api/defaultResponderForAppDir"; + +import { handleQueuedFormResponseCleanup } from "@calcom/app-store/routing-forms/cron/queuedFormResponseCleanup"; + +export const POST = defaultResponderForAppDir(handleQueuedFormResponseCleanup); diff --git a/apps/web/vercel.json b/apps/web/vercel.json index fbe6dae2f6..ef206b78f8 100644 --- a/apps/web/vercel.json +++ b/apps/web/vercel.json @@ -4,6 +4,10 @@ "path": "/api/cron/calendar-cache-cleanup", "schedule": "0 5 * * *" }, + { + "path": "/api/cron/queuedFormResponseCleanup", + "schedule": "0 */12 * * *" + }, { "path": "/api/tasks/cron", "schedule": "* * * * *" diff --git a/packages/app-store/routing-forms/cron/queuedFormResponseCleanup.ts b/packages/app-store/routing-forms/cron/queuedFormResponseCleanup.ts new file mode 100644 index 0000000000..69c92bbb07 --- /dev/null +++ b/packages/app-store/routing-forms/cron/queuedFormResponseCleanup.ts @@ -0,0 +1,47 @@ +import type { NextRequest } from "next/server"; +import { NextResponse } from "next/server"; + +import logger from "@calcom/lib/logger"; +import { PrismaQueuedFormResponseRepository } from "@calcom/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository"; +import { QueuedFormResponseService } from "@calcom/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService"; +import prisma from "@calcom/prisma"; + +function validateRequest(request: NextRequest) { + const apiKey = request.headers.get("authorization") || request.nextUrl.searchParams.get("apiKey"); + + if (![process.env.CRON_API_KEY, `Bearer ${process.env.CRON_SECRET}`].includes(`${apiKey}`)) { + return NextResponse.json({ message: "Not authenticated" }, { status: 401 }); + } + + return null; +} + +export async function handleQueuedFormResponseCleanup(request: NextRequest) { + const authError = validateRequest(request); + if (authError) { + return authError; + } + + try { + const queuedFormResponseRepo = new PrismaQueuedFormResponseRepository(prisma); + const queuedFormResponseService = new QueuedFormResponseService({ + logger, + queuedFormResponseRepo, + }); + + const result = await queuedFormResponseService.cleanupExpiredResponses({ + batchSize: 1000, + }); + + return NextResponse.json({ + status: "success", + data: result, + }); + } catch (error) { + console.error("Error during queued form response cleanup:", error); + return NextResponse.json( + { status: "error", message: "Failed to cleanup queued form responses" }, + { status: 500 } + ); + } +} diff --git a/packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.integration-test.ts b/packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.integration-test.ts new file mode 100644 index 0000000000..795e9b1ac9 --- /dev/null +++ b/packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.integration-test.ts @@ -0,0 +1,432 @@ +import { describe, expect, beforeAll, afterAll, beforeEach, it } from "vitest"; + +import prisma from "@calcom/prisma"; + +import { PrismaQueuedFormResponseRepository } from "./PrismaQueuedFormResponseRepository"; + +const repository = new PrismaQueuedFormResponseRepository(prisma); + +// Test Data Builders +const createTestUser = async (overrides?: { email?: string; username?: string }) => { + const timestamp = Date.now(); + return prisma.user.create({ + data: { + email: overrides?.email || `test-qfr-${timestamp}@example.com`, + username: overrides?.username || `test-qfr-${timestamp}`, + }, + }); +}; + +const createTestForm = async (userId: number, overrides?: { name?: string }) => { + return prisma.app_RoutingForms_Form.create({ + data: { + name: overrides?.name || `Test Form ${Date.now()}`, + userId, + fields: [], + routes: [], + settings: {}, + }, + }); +}; + +const createFormResponse = async ( + formId: string, + overrides?: { + response?: Record; + chosenRouteId?: string; + } +) => { + return prisma.app_RoutingForms_FormResponse.create({ + data: { + formId, + response: overrides?.response || { test: "response" }, + chosenRouteId: overrides?.chosenRouteId || "test-route-id", + }, + }); +}; + +const createQueuedResponse = async ( + formId: string, + overrides?: { + response?: Record; + chosenRouteId?: string; + actualResponseId?: number | null; + createdAt?: Date; + } +) => { + return prisma.app_RoutingForms_QueuedFormResponse.create({ + data: { + formId, + response: overrides?.response || { field1: "value" }, + chosenRouteId: overrides?.chosenRouteId || "test-route-id", + actualResponseId: overrides?.actualResponseId ?? null, + ...(overrides?.createdAt && { createdAt: overrides.createdAt }), + }, + }); +}; + +const createQueuedResponsesBatch = async ( + formId: string, + count: number, + baseOverrides?: Parameters[1] +) => { + const responses = []; + for (let i = 0; i < count; i++) { + const response = await createQueuedResponse(formId, { + ...baseOverrides, + response: { field1: `value${i}` }, + }); + responses.push(response); + } + return responses; +}; + +// Date Helpers +const daysAgo = (days: number) => new Date(Date.now() - days * 24 * 60 * 60 * 1000); + +// Assertion Helpers +const expectQueuedResponseCount = async (formId: string, expectedCount: number) => { + const count = await prisma.app_RoutingForms_QueuedFormResponse.count({ + where: { formId }, + }); + expect(count).toBe(expectedCount); +}; + +const expectQueuedResponseExists = async (responseId: string) => { + const response = await prisma.app_RoutingForms_QueuedFormResponse.findUnique({ + where: { id: responseId }, + }); + expect(response).not.toBeNull(); +}; + +const expectQueuedResponseNotExists = async (responseId: string) => { + const response = await prisma.app_RoutingForms_QueuedFormResponse.findUnique({ + where: { id: responseId }, + }); + expect(response).toBeNull(); +}; + +const expectResponseFields = (response: unknown) => { + const expectedFields = [ + "id", + "formId", + "response", + "chosenRouteId", + "createdAt", + "updatedAt", + "actualResponseId", + ]; + expectedFields.forEach((field) => { + expect(response).toHaveProperty(field); + }); + expect(Object.keys(response as object).sort()).toEqual(expectedFields.sort()); +}; + +// Cleanup Helpers +const cleanupQueuedResponses = async (formId: string) => { + await prisma.app_RoutingForms_QueuedFormResponse.deleteMany({ + where: { formId }, + }); +}; + +describe("PrismaQueuedFormResponseRepository Integration Tests", () => { + let testForm: { id: string }; + let testUser: { id: number }; + let testFormResponse: { id: number }; + let testResponses: Array<{ id: string; actualResponseId: number | null; createdAt: Date }> = []; + + beforeAll(async () => { + // Create test user + testUser = await createTestUser(); + + // Create test form + testForm = await createTestForm(testUser.id); + + // Create test form response for actualResponseId references + testFormResponse = await createFormResponse(testForm.id); + }); + + afterAll(async () => { + // Clean up all test data + await prisma.app_RoutingForms_QueuedFormResponse.deleteMany({ + where: { formId: testForm.id }, + }); + await prisma.app_RoutingForms_FormResponse.deleteMany({ + where: { formId: testForm.id }, + }); + await prisma.app_RoutingForms_Form.delete({ + where: { id: testForm.id }, + }); + await prisma.user.delete({ + where: { id: testUser.id }, + }); + await prisma.$disconnect(); + }); + + beforeEach(async () => { + // Clear queued responses before each test + await cleanupQueuedResponses(testForm.id); + testResponses = []; + }); + + describe("findMany", () => { + it("should find queued responses by actualResponseId", async () => { + // Create additional form responses for testing + const formResponse1 = await createFormResponse(testForm.id, { + response: { field1: "form-response-1" }, + }); + const formResponse2 = await createFormResponse(testForm.id, { + response: { field1: "form-response-2" }, + }); + + // Create test queued responses + const queuedResponse1 = await createQueuedResponse(testForm.id, { + response: { field1: "value1" }, + actualResponseId: formResponse1.id, + }); + + const queuedResponse2 = await createQueuedResponse(testForm.id, { + response: { field1: "value2" }, + actualResponseId: formResponse2.id, + }); + + const queuedResponse3 = await createQueuedResponse(testForm.id, { + response: { field1: "value3" }, + actualResponseId: null, + }); + + testResponses.push(queuedResponse1, queuedResponse2, queuedResponse3); + + // Test finding by specific actualResponseId + const result = await repository.findMany({ + where: { actualResponseId: formResponse1.id }, + }); + + expect(result).toHaveLength(1); + expect(result[0].id).toBe(queuedResponse1.id); + expect(result[0].actualResponseId).toBe(formResponse1.id); + expect(result[0].formId).toBe(testForm.id); + expect(result[0].response).toEqual({ field1: "value1" }); + }); + + it("should find queued responses by actualResponseId null", async () => { + // Create test queued responses + await createQueuedResponse(testForm.id, { + response: { field1: "value1" }, + actualResponseId: testFormResponse.id, + }); + + const queuedResponse2 = await createQueuedResponse(testForm.id, { + response: { field1: "value2" }, + actualResponseId: null, + }); + + const queuedResponse3 = await createQueuedResponse(testForm.id, { + response: { field1: "value3" }, + actualResponseId: null, + }); + + // Test finding by actualResponseId null + const result = await repository.findMany({ + where: { actualResponseId: null }, + }); + + expect(result).toHaveLength(2); + const resultIds = result.map((r) => r.id).sort(); + const expectedIds = [queuedResponse2.id, queuedResponse3.id].sort(); + expect(resultIds).toEqual(expectedIds); + }); + + it("should find queued responses created before a specific date", async () => { + // Create test queued responses with different timestamps + const now = new Date(); + const oneDayAgo = daysAgo(1); + const twoDaysAgo = daysAgo(2); + + // Create responses with specific timestamps + const oldResponse1 = await createQueuedResponse(testForm.id, { + response: { field1: "old1" }, + createdAt: twoDaysAgo, + }); + + const oldResponse2 = await createQueuedResponse(testForm.id, { + response: { field1: "old2" }, + createdAt: twoDaysAgo, + }); + + const recentResponse = await createQueuedResponse(testForm.id, { + response: { field1: "recent" }, + createdAt: now, + }); + + testResponses.push(oldResponse1, oldResponse2, recentResponse); + + // Test finding responses created before oneDayAgo + const result = await repository.findMany({ + where: { createdAt: { lt: oneDayAgo } }, + }); + + expect(result).toHaveLength(2); + const resultIds = result.map((r) => r.id).sort(); + const expectedIds = [oldResponse1.id, oldResponse2.id].sort(); + expect(resultIds).toEqual(expectedIds); + }); + + it("should apply take parameter to limit results", async () => { + // Create multiple test queued responses + const responses = await createQueuedResponsesBatch(testForm.id, 5, { actualResponseId: null }); + testResponses.push(...responses); + + // Test with take parameter + const result = await repository.findMany({ + where: { actualResponseId: null }, + params: { take: 3 }, + }); + + expect(result).toHaveLength(3); + }); + + it("should handle combined where conditions", async () => { + // Create test queued responses + const now = new Date(); + const oneDayAgo = daysAgo(1); + const twoDaysAgo = daysAgo(2); + + const oldWithNull = await createQueuedResponse(testForm.id, { + response: { field1: "old-null" }, + actualResponseId: null, + createdAt: twoDaysAgo, + }); + + await createQueuedResponse(testForm.id, { + response: { field1: "old-with-id" }, + actualResponseId: testFormResponse.id, + createdAt: twoDaysAgo, + }); + + await createQueuedResponse(testForm.id, { + response: { field1: "recent-null" }, + actualResponseId: null, + createdAt: now, + }); + + // Test finding old responses with null actualResponseId + const result = await repository.findMany({ + where: { + actualResponseId: null, + createdAt: { lt: oneDayAgo }, + }, + }); + + expect(result).toHaveLength(1); + expect(result[0].id).toBe(oldWithNull.id); + }); + + it("should throw error when where is empty", async () => { + await expect( + repository.findMany({ + where: {}, + }) + ).rejects.toThrow("where is empty"); + }); + + it("should throw error when where has only undefined values", async () => { + await expect( + repository.findMany({ + where: { + actualResponseId: undefined, + createdAt: undefined, + }, + }) + ).rejects.toThrow("where is empty"); + }); + + it("should return empty array when no matches found", async () => { + const result = await repository.findMany({ + where: { actualResponseId: 999999 }, + }); + + expect(result).toHaveLength(0); + expect(result).toEqual([]); + }); + + it("should return all fields defined in safeSelect", async () => { + const formResponseForTest = await createFormResponse(testForm.id, { + response: { field1: "form-response-test" }, + }); + + const response = await createQueuedResponse(testForm.id, { + response: { field1: "test-value" }, + actualResponseId: formResponseForTest.id, + }); + + const result = await repository.findMany({ + where: { actualResponseId: formResponseForTest.id }, + }); + + expect(result).toHaveLength(1); + const foundResponse = result[0]; + + // Verify all fields from safeSelect are present + expect(foundResponse).toHaveProperty("id", response.id); + expect(foundResponse).toHaveProperty("formId", response.formId); + expect(foundResponse).toHaveProperty("response", response.response); + expect(foundResponse).toHaveProperty("chosenRouteId", response.chosenRouteId); + expect(foundResponse).toHaveProperty("createdAt"); + expect(foundResponse).toHaveProperty("updatedAt"); + expect(foundResponse).toHaveProperty("actualResponseId", response.actualResponseId); + + // Verify no extra fields are included + expectResponseFields(foundResponse); + }); + }); + + describe("deleteByIds", () => { + it("should delete queued responses by their ids", async () => { + // Create test queued responses + const response1 = await createQueuedResponse(testForm.id, { response: { field1: "value1" } }); + const response2 = await createQueuedResponse(testForm.id, { response: { field1: "value2" } }); + const response3 = await createQueuedResponse(testForm.id, { response: { field1: "value3" } }); + + // Delete response1 and response2 + const result = await repository.deleteByIds([response1.id, response2.id]); + + expect(result.count).toBe(2); + + // Verify only response3 remains + await expectQueuedResponseCount(testForm.id, 1); + await expectQueuedResponseExists(response3.id); + }); + + it("should return count 0 when deleting non-existent ids", async () => { + const result = await repository.deleteByIds(["non-existent-id-1", "non-existent-id-2"]); + + expect(result.count).toBe(0); + }); + + it("should handle empty array of ids", async () => { + const result = await repository.deleteByIds([]); + + expect(result.count).toBe(0); + }); + + it("should handle single id deletion", async () => { + const response = await createQueuedResponse(testForm.id, { response: { field1: "single-delete" } }); + + const result = await repository.deleteByIds([response.id]); + + expect(result.count).toBe(1); + + // Verify it's deleted + await expectQueuedResponseNotExists(response.id); + }); + + it("should handle mix of existing and non-existing ids", async () => { + const response = await createQueuedResponse(testForm.id, { response: { field1: "mixed-delete" } }); + + const result = await repository.deleteByIds([response.id, "non-existent-id"]); + + expect(result.count).toBe(1); + }); + }); +}); diff --git a/packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.ts b/packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.ts new file mode 100644 index 0000000000..421172e593 --- /dev/null +++ b/packages/lib/server/repository/routingForm/queuedFormResponse/PrismaQueuedFormResponseRepository.ts @@ -0,0 +1,53 @@ +import type { PrismaClient } from "@calcom/prisma"; +import prisma from "@calcom/prisma"; + +import type { + QueuedFormResponseRepositoryFindManyArgs, + QueuedFormResponseRepositoryFindManyWhere, + QueuedFormResponseRepositoryInterface, +} from "./QueuedFormResponseRepository.interface"; + +const safeSelect = { + id: true, + formId: true, + response: true, + chosenRouteId: true, + createdAt: true, + updatedAt: true, + actualResponseId: true, +}; + +export class PrismaQueuedFormResponseRepository implements QueuedFormResponseRepositoryInterface { + constructor(private readonly prismaClient: PrismaClient = prisma) {} + + private assertWhereIsNotEmpty(where: QueuedFormResponseRepositoryFindManyWhere) { + let found = false; + Object.entries(where).forEach(([_, value]) => { + if (value !== undefined) { + found = true; + } + }); + if (!found) { + throw new Error("where is empty"); + } + } + + async findMany({ where, params }: QueuedFormResponseRepositoryFindManyArgs) { + this.assertWhereIsNotEmpty(where); + return await this.prismaClient.app_RoutingForms_QueuedFormResponse.findMany({ + where, + select: safeSelect, + ...(params?.take && { take: params.take }), + }); + } + + async deleteByIds(ids: string[]) { + return await this.prismaClient.app_RoutingForms_QueuedFormResponse.deleteMany({ + where: { + id: { + in: ids, + }, + }, + }); + } +} diff --git a/packages/lib/server/repository/routingForm/queuedFormResponse/QueuedFormResponseRepository.interface.ts b/packages/lib/server/repository/routingForm/queuedFormResponse/QueuedFormResponseRepository.interface.ts new file mode 100644 index 0000000000..7e0f95ba7c --- /dev/null +++ b/packages/lib/server/repository/routingForm/queuedFormResponse/QueuedFormResponseRepository.interface.ts @@ -0,0 +1,30 @@ +export type QueuedFormResponseRepositoryFindManyWhere = { + actualResponseId?: number | null; + createdAt?: { + lt: Date; + }; +}; + +export type QueuedFormResponseRepositoryFindManyArgs = { + where: QueuedFormResponseRepositoryFindManyWhere; + params?: { + take?: number; + }; +}; + +export type QueuedFormResponseRepositoryFindManyResult = { + id: string; + formId: string; + response: unknown; + chosenRouteId: string | null; + createdAt: Date; + updatedAt: Date | null; + actualResponseId: number | null; +}; + +export interface QueuedFormResponseRepositoryInterface { + findMany( + args: QueuedFormResponseRepositoryFindManyArgs + ): Promise; + deleteByIds(ids: string[]): Promise<{ count: number }>; +} diff --git a/packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.test.ts b/packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.test.ts new file mode 100644 index 0000000000..188cd93ccd --- /dev/null +++ b/packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.test.ts @@ -0,0 +1,152 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +import type logger from "@calcom/lib/logger"; + +import type { QueuedFormResponseRepositoryInterface } from "../../repository/routingForm/queuedFormResponse/QueuedFormResponseRepository.interface"; +import { QueuedFormResponseService } from "./QueuedFormResponseService"; + +describe("QueuedFormResponseService", () => { + let service: QueuedFormResponseService; + let mockRepo: jest.Mocked; + let mockLogger: ReturnType; + + function createMockLogger() { + const subLogger = { + info: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + debug: vi.fn(), + }; + + return { + getSubLogger: vi.fn().mockReturnValue(subLogger), + subLogger, + }; + } + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + + mockRepo = { + findMany: vi.fn(), + deleteByIds: vi.fn(), + } as unknown as jest.Mocked; + + mockLogger = createMockLogger(); + + service = new QueuedFormResponseService({ + logger: mockLogger as unknown as typeof logger, + queuedFormResponseRepo: mockRepo, + }); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + describe("cleanupExpiredResponses", () => { + it("should clean up expired responses in batches", async () => { + const now = Date.now(); + vi.spyOn(Date, "now").mockReturnValue(now); + + // Mock first batch + mockRepo.findMany + .mockResolvedValueOnce([{ id: "1" }, { id: "2" }] as any) + .mockResolvedValueOnce([{ id: "3" }] as any) + .mockResolvedValueOnce([]); + + mockRepo.deleteByIds.mockResolvedValueOnce({ count: 2 }).mockResolvedValueOnce({ count: 1 }); + + const resultPromise = service.cleanupExpiredResponses({ + batchSize: 2, + }); + + // Advance timers to handle delays + await vi.runAllTimersAsync(); + const result = await resultPromise; + + // Verify cutoff time calculation (7 days is hardcoded) + const expectedCutoffTime = new Date(now - 7 * 24 * 60 * 60 * 1000); + expect(mockRepo.findMany).toHaveBeenCalledWith({ + where: { + actualResponseId: null, + createdAt: { + lt: expectedCutoffTime, + }, + }, + params: { + take: 2, + }, + }); + + // Verify deletion calls + expect(mockRepo.deleteByIds).toHaveBeenCalledWith(["1", "2"]); + expect(mockRepo.deleteByIds).toHaveBeenCalledWith(["3"]); + + // Verify result + expect(result).toEqual({ + count: 3, + batches: 2, + }); + }); + + it("should use default values when config is not provided", async () => { + mockRepo.findMany.mockResolvedValueOnce([]); + + const result = await service.cleanupExpiredResponses(); + const expectedCutoffTime = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000); + + // Verify default batch size and older than days + expect(mockRepo.findMany).toHaveBeenCalledWith({ + where: { + actualResponseId: null, + createdAt: { + lt: expectedCutoffTime, + }, + }, + params: { + take: 1000, // default batch size + }, + }); + + expect(result).toEqual({ + count: 0, + batches: 0, + }); + }); + + it("should handle no expired responses", async () => { + mockRepo.findMany.mockResolvedValueOnce([]); + + const result = await service.cleanupExpiredResponses(); + + expect(mockRepo.deleteByIds).not.toHaveBeenCalled(); + expect(result).toEqual({ + count: 0, + batches: 0, + }); + }); + + it("should stop when batch is not full", async () => { + // Return partial batch + mockRepo.findMany.mockResolvedValueOnce([{ id: "1" }, { id: "2" }, { id: "3" }]); + mockRepo.deleteByIds.mockResolvedValueOnce({ count: 3 }); + + const result = await service.cleanupExpiredResponses({ batchSize: 5 }); + + expect(mockRepo.findMany).toHaveBeenCalledTimes(1); + expect(mockRepo.deleteByIds).toHaveBeenCalledTimes(1); + expect(result).toEqual({ + count: 3, + batches: 1, + }); + }); + + it("should handle errors during batch processing", async () => { + mockRepo.findMany.mockRejectedValueOnce(new Error("Database error")); + + await expect(service.cleanupExpiredResponses()).rejects.toThrow("Database error"); + }); + }); +}); diff --git a/packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.ts b/packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.ts new file mode 100644 index 0000000000..69ca09cfa0 --- /dev/null +++ b/packages/lib/server/service/routingForm/queuedFormResponse/QueuedFormResponseService.ts @@ -0,0 +1,96 @@ +import type logger from "@calcom/lib/logger"; + +import type { QueuedFormResponseRepositoryInterface } from "../../../repository/routingForm/queuedFormResponse/QueuedFormResponseRepository.interface"; + +interface Dependencies { + logger: typeof logger; + queuedFormResponseRepo: QueuedFormResponseRepositoryInterface; +} + +interface CleanupConfig { + batchSize?: number; +} + +export class QueuedFormResponseService { + constructor(private readonly deps: Dependencies) {} + + private getExpiredResponses({ batchSize }: { batchSize: number }) { + const cutoffTime = this.getExpiryCutoffTime(); + return this.deps.queuedFormResponseRepo.findMany({ + where: { + actualResponseId: null, + createdAt: { + lt: cutoffTime, + }, + }, + params: { + take: batchSize, + }, + }); + } + /** + * Calculate the cutoff time. All responses older than this time could be considered expired. + */ + private getExpiryCutoffTime(): Date { + const olderThanDays = 7; + return new Date(Date.now() - olderThanDays * 24 * 60 * 60 * 1000); + } + + /** + * Clean up expired queued form responses in batches + */ + async cleanupExpiredResponses(config: CleanupConfig = {}): Promise<{ + count: number; + batches: number; + }> { + const { batchSize = 1000 } = config; + + const log = this.deps.logger.getSubLogger({ + prefix: ["[QueuedFormResponseService]", "cleanupExpiredResponses"], + }); + + let totalDeleted = 0; + let batchCount = 0; + let hasMore = true; + + while (hasMore) { + try { + const expiredResponses = await this.getExpiredResponses({ batchSize }); + + if (expiredResponses.length === 0) { + hasMore = false; + break; + } + + const idsToDelete = expiredResponses.map((r: { id: string }) => r.id); + + const deleteResult = await this.deps.queuedFormResponseRepo.deleteByIds(idsToDelete); + + totalDeleted += deleteResult.count; + batchCount++; + + log.info(`Batch ${batchCount}: Deleted ${deleteResult.count} records. Total: ${totalDeleted}`); + + // If we found fewer records than the batch size, we're done + if (expiredResponses.length < batchSize) { + hasMore = false; + } + + // Add a small delay between batches to reduce database load + if (hasMore) { + await new Promise((resolve) => setTimeout(resolve, 100)); + } + } catch (error) { + log.error(`Error processing batch ${batchCount + 1}:`, error); + throw error; + } + } + + log.info(`Cleanup completed. Total deleted: ${totalDeleted}, Batches: ${batchCount}`); + + return { + count: totalDeleted, + batches: batchCount, + }; + } +}