feat: schedule daily log cleanup

Author: geclos

apps/workers/src/workers/index.ts

@@ -82,4 +82,11 @@ export async function setupSchedules() {
     { pattern: '0 0 4 * * *' },
     { opts: { attempts: 1 } },
   )
+
+  // Every day at 1 AM - Schedule cleanup jobs for free plan workspaces
+  await maintenanceQueue.upsertJobScheduler(
+    'scheduleWorkspaceCleanupJobs',
+    { pattern: '0 0 1 * * *' },
+    { opts: { attempts: 1 } },
+  )
 }

apps/workers/src/workers/worker-definitions/maintenanceWorker.ts

@@ -14,6 +14,8 @@ const jobMappings = {
   refreshProjectStatsCacheJob: jobs.refreshProjectStatsCacheJob,
   refreshDocumentsStatsCacheJob: jobs.refreshDocumentsStatsCacheJob,
   refreshDocumentStatsCacheJob: jobs.refreshDocumentStatsCacheJob,
+  scheduleWorkspaceCleanupJobs: jobs.scheduleWorkspaceCleanupJobs,
+  cleanupWorkspaceOldLogsJob: jobs.cleanupWorkspaceOldLogsJob,
 }
 
 export function startMaintenanceWorker() {

packages/core/src/jobs/job-definitions/maintenance/cleanupWorkspaceOldLogsJob.ts

@@ -0,0 +1,111 @@
+import { Job } from 'bullmq'
+import { and, eq, exists, lt, sql } from 'drizzle-orm'
+import { commits, documentLogs, projects, providerLogs } from '../../../schema'
+import Transaction from '../../../lib/Transaction'
+import { Result } from '../../../lib/Result'
+
+export type CleanupWorkspaceOldLogsJobData = {
+  workspaceId: number
+}
+
+/**
+ * Job that deletes document logs and associated provider logs older than 30 days
+ * for a specific workspace.
+ *
+ * This job uses optimized SQL operations to avoid loading data into memory:
+ * 1. Calculates the cutoff date (30 days ago)
+ * 2. Deletes provider logs using a correlated subquery
+ * 3. Deletes document logs using a correlated subquery
+ * 4. Uses batch processing if needed for very large datasets
+ */
+export const cleanupWorkspaceOldLogsJob = async (
+  job: Job<CleanupWorkspaceOldLogsJobData>,
+) => {
+  const { workspaceId } = job.data
+
+  // Calculate cutoff date (30 days ago)
+  const cutoffDate = new Date()
+  cutoffDate.setDate(cutoffDate.getDate() - 30)
+
+  const transaction = new Transaction()
+  return await transaction.call(async (tx) => {
+    let totalDeletedProviderLogs = 0
+    let totalDeletedDocumentLogs = 0
+
+    // Delete provider logs in batches to avoid memory issues and long-running transactions
+    const batchSize = 1000
+    let deletedProviderLogsBatch: number
+
+    do {
+      // Delete provider logs associated with old document logs from this workspace
+      // Use EXISTS with a correlated subquery to avoid loading UUIDs into memory
+      const deletedProviderLogsResult = await tx
+        .delete(providerLogs)
+        .where(
+          and(
+            eq(providerLogs.workspaceId, workspaceId),
+            exists(
+              tx
+                .select({ dummy: sql`1` })
+                .from(documentLogs)
+                .innerJoin(commits, eq(commits.id, documentLogs.commitId))
+                .innerJoin(projects, eq(projects.id, commits.projectId))
+                .where(
+                  and(
+                    eq(projects.workspaceId, workspaceId),
+                    lt(documentLogs.createdAt, cutoffDate),
+                    eq(documentLogs.uuid, providerLogs.documentLogUuid),
+                  ),
+                )
+                .limit(1),
+            ),
+          ),
+        )
+        .returning({ id: providerLogs.id })
+
+      deletedProviderLogsBatch = deletedProviderLogsResult.length
+      totalDeletedProviderLogs += deletedProviderLogsBatch
+
+      // Add a small delay between batches to prevent overwhelming the database
+      if (deletedProviderLogsBatch === batchSize) {
+        await new Promise((resolve) => setTimeout(resolve, 100))
+      }
+    } while (deletedProviderLogsBatch === batchSize)
+
+    // Delete document logs in batches
+    let deletedDocumentLogsBatch: number
+
+    do {
+      // Delete old document logs from this workspace
+      // Use JOIN directly in DELETE for optimal performance
+      const deletedDocumentLogsResult = await tx.execute(sql`
+        DELETE FROM ${documentLogs}
+        WHERE ${documentLogs.id} IN (
+          SELECT dl.id 
+          FROM ${documentLogs} dl
+          INNER JOIN ${commits} c ON c.id = dl.commit_id
+          INNER JOIN ${projects} p ON p.id = c.project_id
+          WHERE p.workspace_id = ${workspaceId}
+            AND dl.created_at < ${cutoffDate}
+          LIMIT ${batchSize}
+        )
+      `)
+
+      deletedDocumentLogsBatch = deletedDocumentLogsResult.rowCount || 0
+      totalDeletedDocumentLogs += deletedDocumentLogsBatch
+
+      // Add a small delay between batches to prevent overwhelming the database
+      if (deletedDocumentLogsBatch === batchSize) {
+        await new Promise((resolve) => setTimeout(resolve, 100))
+      }
+    } while (deletedDocumentLogsBatch === batchSize)
+
+    return Result.ok({
+      success: true,
+      workspaceId,
+      deletedDocumentLogs: totalDeletedDocumentLogs,
+      deletedProviderLogs: totalDeletedProviderLogs,
+      cutoffDate: cutoffDate.toISOString(),
+    })
+  })
+}

packages/core/src/jobs/job-definitions/maintenance/index.ts

@@ -2,3 +2,5 @@ export * from './refreshDocumentsStatsCacheJob'
 export * from './refreshDocumentStatsCacheJob'
 export * from './refreshProjectsStatsCacheJob'
 export * from './refreshProjectStatsCacheJob'
+export * from './scheduleWorkspaceCleanupJobs'
+export * from './cleanupWorkspaceOldLogsJob'

packages/core/src/jobs/job-definitions/maintenance/scheduleWorkspaceCleanupJobs.ts

@@ -0,0 +1,51 @@
+import { Job } from 'bullmq'
+import { eq, inArray } from 'drizzle-orm'
+import { FREE_PLANS } from '../../../plans'
+import { database } from '../../../client'
+import { subscriptions, workspaces } from '../../../schema'
+import { maintenanceQueue } from '../../queues'
+
+export type ScheduleWorkspaceCleanupJobsData = Record<string, never>
+
+/**
+ * Job that runs nightly to schedule cleanup jobs for workspaces on free plans.
+ *
+ * This job:
+ * 1. Finds all workspaces with free plans (hobby_v1, hobby_v2)
+ * 2. Enqueues individual cleanup jobs for each workspace
+ * 3. Each cleanup job will delete document logs and provider logs older than 30 days
+ */
+export const scheduleWorkspaceCleanupJobs = async (
+  _: Job<ScheduleWorkspaceCleanupJobsData>,
+) => {
+  // Find all workspaces with free plans
+  const freeWorkspaces = await database
+    .select({
+      id: workspaces.id,
+    })
+    .from(workspaces)
+    .innerJoin(
+      subscriptions,
+      eq(subscriptions.id, workspaces.currentSubscriptionId),
+    )
+    .where(inArray(subscriptions.plan, FREE_PLANS))
+    .then((r) => r)
+
+  let enqueuedJobs = 0
+
+  // Enqueue individual cleanup job for each free workspace
+  for (const workspace of freeWorkspaces) {
+    await maintenanceQueue.add(
+      'cleanupWorkspaceOldLogsJob',
+      { workspaceId: workspace.id },
+      { attempts: 3 },
+    )
+    enqueuedJobs++
+  }
+
+  return {
+    success: true,
+    freeWorkspacesCount: freeWorkspaces.length,
+    enqueuedJobs,
+  }
+}