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 <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 <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 <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 <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 <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 <hariombalhara@gmail.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 <hariom@cal.com> Co-authored-by: Hariom <hariombalhara@gmail.com>
This commit is contained in:
8 files changed
+819
No files matched your search
@@ -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);
|
||||
@@ -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": "* * * * *"
|
||||
|
||||
@@ -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 }
|
||||
);
|
||||
}
|
||||
}
|
||||
+432
@@ -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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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<typeof createQueuedResponse>[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);
|
||||
});
|
||||
});
|
||||
});
|
||||
+53
@@ -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,
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
+30
@@ -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<QueuedFormResponseRepositoryFindManyResult[]>;
|
||||
deleteByIds(ids: string[]): Promise<{ count: number }>;
|
||||
}
|
||||
+152
@@ -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<QueuedFormResponseRepositoryInterface>;
|
||||
let mockLogger: ReturnType<typeof createMockLogger>;
|
||||
|
||||
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<QueuedFormResponseRepositoryInterface>;
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
});
|
||||
+96
@@ -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,
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user