mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-01 14:59:19 +08:00
feat(jobs): Add data retention jobs (#4128)
* feat(jobs): Add data retention jobs Add 3 cron-triggered cleanup jobs dispatched via Trigger.dev (or inline fallback): - cleanup-soft-deletes: hard-deletes soft-deleted workspace resources past retention - cleanup-logs: deletes expired workflow execution logs + S3 files - cleanup-tasks: deletes expired copilot chats, runs, feedback, inbox tasks Enterprise admins can configure per-workspace retention via Settings > Data Retention. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> # Conflicts: # packages/db/migrations/meta/0192_snapshot.json # packages/db/migrations/meta/_journal.json # packages/db/schema.ts * Cleanup orphaned using ids, not timestamp sorting * fix lint
This commit is contained in:
Theodore Li
@@ -0,0 +1,24 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { verifyCronAuth } from '@/lib/auth/internal'
|
||||
import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
const logger = createLogger('SoftDeleteCleanupAPI')
|
||||
|
||||
export async function GET(request: NextRequest) {
|
||||
try {
|
||||
const authError = verifyCronAuth(request, 'soft-delete cleanup')
|
||||
if (authError) return authError
|
||||
|
||||
const result = await dispatchCleanupJobs('cleanup-soft-deletes')
|
||||
|
||||
logger.info('Soft-delete cleanup jobs dispatched', result)
|
||||
|
||||
return NextResponse.json({ triggered: true, ...result })
|
||||
} catch (error) {
|
||||
logger.error('Failed to dispatch soft-delete cleanup jobs:', { error })
|
||||
return NextResponse.json({ error: 'Failed to dispatch soft-delete cleanup' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { verifyCronAuth } from '@/lib/auth/internal'
|
||||
import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
const logger = createLogger('TaskCleanupAPI')
|
||||
|
||||
export async function GET(request: NextRequest) {
|
||||
try {
|
||||
const authError = verifyCronAuth(request, 'task cleanup')
|
||||
if (authError) return authError
|
||||
|
||||
const result = await dispatchCleanupJobs('cleanup-tasks')
|
||||
|
||||
logger.info('Task cleanup jobs dispatched', result)
|
||||
|
||||
return NextResponse.json({ triggered: true, ...result })
|
||||
} catch (error) {
|
||||
logger.error('Failed to dispatch task cleanup jobs:', { error })
|
||||
return NextResponse.json({ error: 'Failed to dispatch task cleanup' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
@@ -1,201 +1,24 @@
|
||||
import { db } from '@sim/db'
|
||||
import { subscription, workflowExecutionLogs, workspace } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, eq, inArray, isNull, lt } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { verifyCronAuth } from '@/lib/auth/internal'
|
||||
import { sqlIsPaid } from '@/lib/billing/plan-helpers'
|
||||
import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
import { snapshotService } from '@/lib/logs/execution/snapshot/service'
|
||||
import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
|
||||
import { dispatchCleanupJobs } from '@/lib/billing/cleanup-dispatcher'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
|
||||
const logger = createLogger('LogsCleanupAPI')
|
||||
|
||||
const BATCH_SIZE = 2000
|
||||
|
||||
export async function GET(request: NextRequest) {
|
||||
try {
|
||||
const authError = verifyCronAuth(request, 'logs cleanup')
|
||||
if (authError) {
|
||||
return authError
|
||||
}
|
||||
if (authError) return authError
|
||||
|
||||
const retentionDate = new Date()
|
||||
retentionDate.setDate(retentionDate.getDate() - Number(env.FREE_PLAN_LOG_RETENTION_DAYS || '7'))
|
||||
const result = await dispatchCleanupJobs('cleanup-logs')
|
||||
|
||||
const freeWorkspacesSubquery = db
|
||||
.select({ id: workspace.id })
|
||||
.from(workspace)
|
||||
.leftJoin(
|
||||
subscription,
|
||||
and(
|
||||
eq(subscription.referenceId, workspace.billedAccountUserId),
|
||||
inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES),
|
||||
sqlIsPaid(subscription.plan)
|
||||
)
|
||||
)
|
||||
.where(isNull(subscription.id))
|
||||
logger.info('Log cleanup jobs dispatched', result)
|
||||
|
||||
const results = {
|
||||
enhancedLogs: {
|
||||
total: 0,
|
||||
archived: 0,
|
||||
archiveFailed: 0,
|
||||
deleted: 0,
|
||||
deleteFailed: 0,
|
||||
},
|
||||
files: {
|
||||
total: 0,
|
||||
deleted: 0,
|
||||
deleteFailed: 0,
|
||||
},
|
||||
snapshots: {
|
||||
cleaned: 0,
|
||||
cleanupFailed: 0,
|
||||
},
|
||||
}
|
||||
|
||||
const startTime = Date.now()
|
||||
const MAX_BATCHES = 10
|
||||
|
||||
let batchesProcessed = 0
|
||||
let hasMoreLogs = true
|
||||
|
||||
logger.info('Starting enhanced logs cleanup for free-plan workspaces')
|
||||
|
||||
while (hasMoreLogs && batchesProcessed < MAX_BATCHES) {
|
||||
const oldEnhancedLogs = await db
|
||||
.select({
|
||||
id: workflowExecutionLogs.id,
|
||||
workflowId: workflowExecutionLogs.workflowId,
|
||||
executionId: workflowExecutionLogs.executionId,
|
||||
stateSnapshotId: workflowExecutionLogs.stateSnapshotId,
|
||||
level: workflowExecutionLogs.level,
|
||||
trigger: workflowExecutionLogs.trigger,
|
||||
startedAt: workflowExecutionLogs.startedAt,
|
||||
endedAt: workflowExecutionLogs.endedAt,
|
||||
totalDurationMs: workflowExecutionLogs.totalDurationMs,
|
||||
executionData: workflowExecutionLogs.executionData,
|
||||
cost: workflowExecutionLogs.cost,
|
||||
files: workflowExecutionLogs.files,
|
||||
createdAt: workflowExecutionLogs.createdAt,
|
||||
})
|
||||
.from(workflowExecutionLogs)
|
||||
.where(
|
||||
and(
|
||||
inArray(workflowExecutionLogs.workspaceId, freeWorkspacesSubquery),
|
||||
lt(workflowExecutionLogs.startedAt, retentionDate)
|
||||
)
|
||||
)
|
||||
.limit(BATCH_SIZE)
|
||||
|
||||
results.enhancedLogs.total += oldEnhancedLogs.length
|
||||
|
||||
for (const log of oldEnhancedLogs) {
|
||||
const today = new Date().toISOString().split('T')[0]
|
||||
|
||||
const enhancedLogKey = `logs/archived/${today}/${log.id}.json`
|
||||
const enhancedLogData = JSON.stringify({
|
||||
...log,
|
||||
archivedAt: new Date().toISOString(),
|
||||
logType: 'enhanced',
|
||||
})
|
||||
|
||||
try {
|
||||
await StorageService.uploadFile({
|
||||
file: Buffer.from(enhancedLogData),
|
||||
fileName: enhancedLogKey,
|
||||
contentType: 'application/json',
|
||||
context: 'logs',
|
||||
preserveKey: true,
|
||||
customKey: enhancedLogKey,
|
||||
metadata: {
|
||||
logId: String(log.id),
|
||||
workflowId: String(log.workflowId ?? ''),
|
||||
executionId: String(log.executionId),
|
||||
logType: 'enhanced',
|
||||
archivedAt: new Date().toISOString(),
|
||||
},
|
||||
})
|
||||
|
||||
results.enhancedLogs.archived++
|
||||
|
||||
if (isUsingCloudStorage() && log.files && Array.isArray(log.files)) {
|
||||
for (const file of log.files) {
|
||||
if (file && typeof file === 'object' && file.key) {
|
||||
results.files.total++
|
||||
try {
|
||||
await StorageService.deleteFile({
|
||||
key: file.key,
|
||||
context: 'execution',
|
||||
})
|
||||
results.files.deleted++
|
||||
|
||||
// Also delete from workspace_files table
|
||||
const { deleteFileMetadata } = await import('@/lib/uploads/server/metadata')
|
||||
await deleteFileMetadata(file.key)
|
||||
|
||||
logger.info(`Deleted execution file: ${file.key}`)
|
||||
} catch (fileError) {
|
||||
results.files.deleteFailed++
|
||||
logger.error(`Failed to delete file ${file.key}:`, { fileError })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const deleteResult = await db
|
||||
.delete(workflowExecutionLogs)
|
||||
.where(eq(workflowExecutionLogs.id, log.id))
|
||||
.returning({ id: workflowExecutionLogs.id })
|
||||
|
||||
if (deleteResult.length > 0) {
|
||||
results.enhancedLogs.deleted++
|
||||
} else {
|
||||
results.enhancedLogs.deleteFailed++
|
||||
logger.warn(`Failed to delete log ${log.id} after archiving: No rows deleted`)
|
||||
}
|
||||
} catch (deleteError) {
|
||||
results.enhancedLogs.deleteFailed++
|
||||
logger.error(`Error deleting log ${log.id} after archiving:`, { deleteError })
|
||||
}
|
||||
} catch (archiveError) {
|
||||
results.enhancedLogs.archiveFailed++
|
||||
logger.error(`Failed to archive log ${log.id}:`, { archiveError })
|
||||
}
|
||||
}
|
||||
|
||||
batchesProcessed++
|
||||
hasMoreLogs = oldEnhancedLogs.length === BATCH_SIZE
|
||||
|
||||
logger.info(`Processed logs batch ${batchesProcessed}: ${oldEnhancedLogs.length} logs`)
|
||||
}
|
||||
|
||||
try {
|
||||
const snapshotRetentionDays = Number(env.FREE_PLAN_LOG_RETENTION_DAYS || '7') + 1 // Keep snapshots 1 day longer
|
||||
const cleanedSnapshots = await snapshotService.cleanupOrphanedSnapshots(snapshotRetentionDays)
|
||||
results.snapshots.cleaned = cleanedSnapshots
|
||||
logger.info(`Cleaned up ${cleanedSnapshots} orphaned snapshots`)
|
||||
} catch (snapshotError) {
|
||||
results.snapshots.cleanupFailed = 1
|
||||
logger.error('Error cleaning up orphaned snapshots:', { snapshotError })
|
||||
}
|
||||
|
||||
const timeElapsed = (Date.now() - startTime) / 1000
|
||||
const reachedLimit = batchesProcessed >= MAX_BATCHES && hasMoreLogs
|
||||
|
||||
return NextResponse.json({
|
||||
message: `Processed ${batchesProcessed} enhanced log batches (${results.enhancedLogs.total} logs, ${results.files.total} files) in ${timeElapsed.toFixed(2)}s${reachedLimit ? ' (batch limit reached)' : ''}`,
|
||||
results,
|
||||
complete: !hasMoreLogs,
|
||||
batchLimitReached: reachedLimit,
|
||||
})
|
||||
return NextResponse.json({ triggered: true, ...result })
|
||||
} catch (error) {
|
||||
logger.error('Error in log cleanup process:', { error })
|
||||
return NextResponse.json({ error: 'Failed to process log cleanup' }, { status: 500 })
|
||||
logger.error('Failed to dispatch log cleanup jobs:', { error })
|
||||
return NextResponse.json({ error: 'Failed to dispatch log cleanup' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
import { db } from '@sim/db'
|
||||
import { workspace } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import { type NextRequest, NextResponse } from 'next/server'
|
||||
import { z } from 'zod'
|
||||
import { AuditAction, AuditResourceType, recordAudit } from '@/lib/audit/log'
|
||||
import { getSession } from '@/lib/auth'
|
||||
import { CLEANUP_CONFIG } from '@/lib/billing/cleanup-dispatcher'
|
||||
import { getHighestPrioritySubscription } from '@/lib/billing/core/plan'
|
||||
import { isEnterprisePlan } from '@/lib/billing/core/subscription'
|
||||
import { getPlanType, type PlanCategory } from '@/lib/billing/plan-helpers'
|
||||
import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils'
|
||||
import { getWorkspaceBilledAccountUserId } from '@/lib/workspaces/utils'
|
||||
|
||||
const logger = createLogger('DataRetentionAPI')
|
||||
|
||||
const MIN_HOURS = 24
|
||||
const MAX_HOURS = 43800 // 5 years
|
||||
|
||||
interface RetentionValues {
|
||||
logRetentionHours: number | null
|
||||
softDeleteRetentionHours: number | null
|
||||
taskCleanupHours: number | null
|
||||
}
|
||||
|
||||
function getPlanDefaults(plan: PlanCategory): RetentionValues {
|
||||
return {
|
||||
logRetentionHours: CLEANUP_CONFIG['cleanup-logs'].defaults[plan],
|
||||
softDeleteRetentionHours: CLEANUP_CONFIG['cleanup-soft-deletes'].defaults[plan],
|
||||
taskCleanupHours: CLEANUP_CONFIG['cleanup-tasks'].defaults[plan],
|
||||
}
|
||||
}
|
||||
|
||||
async function resolveWorkspacePlan(billedAccountUserId: string): Promise<PlanCategory> {
|
||||
const sub = await getHighestPrioritySubscription(billedAccountUserId)
|
||||
return getPlanType(sub?.plan)
|
||||
}
|
||||
|
||||
const updateRetentionSchema = z.object({
|
||||
logRetentionHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
|
||||
softDeleteRetentionHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
|
||||
taskCleanupHours: z.number().int().min(MIN_HOURS).max(MAX_HOURS).nullable().optional(),
|
||||
})
|
||||
|
||||
/**
|
||||
* GET /api/workspaces/[id]/data-retention
|
||||
* Returns the workspace's data retention config including plan defaults and
|
||||
* whether the workspace is on an enterprise plan.
|
||||
*/
|
||||
export async function GET(_request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
const { id: workspaceId } = await params
|
||||
|
||||
const permission = await getUserEntityPermissions(session.user.id, 'workspace', workspaceId)
|
||||
if (!permission) {
|
||||
return NextResponse.json({ error: 'Workspace not found or access denied' }, { status: 404 })
|
||||
}
|
||||
|
||||
const [ws] = await db
|
||||
.select({
|
||||
logRetentionHours: workspace.logRetentionHours,
|
||||
softDeleteRetentionHours: workspace.softDeleteRetentionHours,
|
||||
taskCleanupHours: workspace.taskCleanupHours,
|
||||
billedAccountUserId: workspace.billedAccountUserId,
|
||||
})
|
||||
.from(workspace)
|
||||
.where(eq(workspace.id, workspaceId))
|
||||
.limit(1)
|
||||
|
||||
if (!ws) {
|
||||
return NextResponse.json({ error: 'Workspace not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
const plan = await resolveWorkspacePlan(ws.billedAccountUserId)
|
||||
const defaults = getPlanDefaults(plan)
|
||||
const isEnterpriseWorkspace = plan === 'enterprise'
|
||||
|
||||
return NextResponse.json({
|
||||
success: true,
|
||||
data: {
|
||||
plan,
|
||||
isEnterprise: isEnterpriseWorkspace,
|
||||
defaults,
|
||||
configured: {
|
||||
logRetentionHours: ws.logRetentionHours,
|
||||
softDeleteRetentionHours: ws.softDeleteRetentionHours,
|
||||
taskCleanupHours: ws.taskCleanupHours,
|
||||
},
|
||||
effective: isEnterpriseWorkspace
|
||||
? {
|
||||
logRetentionHours: ws.logRetentionHours,
|
||||
softDeleteRetentionHours: ws.softDeleteRetentionHours,
|
||||
taskCleanupHours: ws.taskCleanupHours,
|
||||
}
|
||||
: {
|
||||
logRetentionHours: defaults.logRetentionHours,
|
||||
softDeleteRetentionHours: defaults.softDeleteRetentionHours,
|
||||
taskCleanupHours: defaults.taskCleanupHours,
|
||||
},
|
||||
},
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error('Failed to get data retention settings', { error })
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* PUT /api/workspaces/[id]/data-retention
|
||||
* Updates the workspace's data retention settings.
|
||||
* Requires admin permission and enterprise plan.
|
||||
*/
|
||||
export async function PUT(request: NextRequest, { params }: { params: Promise<{ id: string }> }) {
|
||||
try {
|
||||
const session = await getSession()
|
||||
if (!session?.user?.id) {
|
||||
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||
}
|
||||
|
||||
const { id: workspaceId } = await params
|
||||
|
||||
const permission = await getUserEntityPermissions(session.user.id, 'workspace', workspaceId)
|
||||
if (permission !== 'admin') {
|
||||
return NextResponse.json({ error: 'Insufficient permissions' }, { status: 403 })
|
||||
}
|
||||
|
||||
const billedAccountUserId = await getWorkspaceBilledAccountUserId(workspaceId)
|
||||
if (!billedAccountUserId) {
|
||||
return NextResponse.json({ error: 'Workspace not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
const hasEnterprise = await isEnterprisePlan(billedAccountUserId)
|
||||
if (!hasEnterprise) {
|
||||
return NextResponse.json(
|
||||
{ error: 'Data Retention configuration is available on Enterprise plans only' },
|
||||
{ status: 403 }
|
||||
)
|
||||
}
|
||||
|
||||
const body = await request.json()
|
||||
const parsed = updateRetentionSchema.safeParse(body)
|
||||
if (!parsed.success) {
|
||||
return NextResponse.json(
|
||||
{ error: parsed.error.errors[0]?.message ?? 'Invalid request body' },
|
||||
{ status: 400 }
|
||||
)
|
||||
}
|
||||
|
||||
const updateData: Record<string, unknown> = { updatedAt: new Date() }
|
||||
|
||||
if (parsed.data.logRetentionHours !== undefined) {
|
||||
updateData.logRetentionHours = parsed.data.logRetentionHours
|
||||
}
|
||||
if (parsed.data.softDeleteRetentionHours !== undefined) {
|
||||
updateData.softDeleteRetentionHours = parsed.data.softDeleteRetentionHours
|
||||
}
|
||||
if (parsed.data.taskCleanupHours !== undefined) {
|
||||
updateData.taskCleanupHours = parsed.data.taskCleanupHours
|
||||
}
|
||||
|
||||
const [updated] = await db
|
||||
.update(workspace)
|
||||
.set(updateData)
|
||||
.where(eq(workspace.id, workspaceId))
|
||||
.returning({
|
||||
logRetentionHours: workspace.logRetentionHours,
|
||||
softDeleteRetentionHours: workspace.softDeleteRetentionHours,
|
||||
taskCleanupHours: workspace.taskCleanupHours,
|
||||
})
|
||||
|
||||
if (!updated) {
|
||||
return NextResponse.json({ error: 'Workspace not found' }, { status: 404 })
|
||||
}
|
||||
|
||||
recordAudit({
|
||||
workspaceId,
|
||||
actorId: session.user.id,
|
||||
action: AuditAction.ORGANIZATION_UPDATED,
|
||||
resourceType: AuditResourceType.WORKSPACE,
|
||||
resourceId: workspaceId,
|
||||
actorName: session.user.name ?? undefined,
|
||||
actorEmail: session.user.email ?? undefined,
|
||||
description: 'Updated data retention settings',
|
||||
metadata: { changes: parsed.data },
|
||||
request,
|
||||
})
|
||||
|
||||
const defaults = getPlanDefaults('enterprise')
|
||||
|
||||
return NextResponse.json({
|
||||
success: true,
|
||||
data: {
|
||||
plan: 'enterprise' as const,
|
||||
isEnterprise: true,
|
||||
defaults,
|
||||
configured: {
|
||||
logRetentionHours: updated.logRetentionHours,
|
||||
softDeleteRetentionHours: updated.softDeleteRetentionHours,
|
||||
taskCleanupHours: updated.taskCleanupHours,
|
||||
},
|
||||
effective: {
|
||||
logRetentionHours: updated.logRetentionHours,
|
||||
softDeleteRetentionHours: updated.softDeleteRetentionHours,
|
||||
taskCleanupHours: updated.taskCleanupHours,
|
||||
},
|
||||
},
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error('Failed to update data retention settings', { error })
|
||||
return NextResponse.json({ error: 'Internal server error' }, { status: 500 })
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,7 @@ const SECTION_TITLES: Record<string, string> = {
|
||||
skills: 'Skills',
|
||||
'workflow-mcp-servers': 'MCP Servers',
|
||||
'credential-sets': 'Email Polling',
|
||||
'data-retention': 'Data Retention',
|
||||
'recently-deleted': 'Recently Deleted',
|
||||
debug: 'Debug',
|
||||
} as const
|
||||
|
||||
@@ -169,6 +169,13 @@ const AuditLogs = dynamic(
|
||||
const SSO = dynamic(() => import('@/ee/sso/components/sso-settings').then((m) => m.SSO), {
|
||||
loading: () => <SettingsSectionSkeleton />,
|
||||
})
|
||||
const DataRetentionSettings = dynamic(
|
||||
() =>
|
||||
import('@/ee/data-retention/components/data-retention-settings').then(
|
||||
(m) => m.DataRetentionSettings
|
||||
),
|
||||
{ loading: () => <SettingsSectionSkeleton /> }
|
||||
)
|
||||
const WhitelabelingSettings = dynamic(
|
||||
() =>
|
||||
import('@/ee/whitelabeling/components/whitelabeling-settings').then(
|
||||
@@ -221,6 +228,7 @@ export function SettingsPage({ section }: SettingsPageProps) {
|
||||
{isBillingEnabled && effectiveSection === 'subscription' && <Subscription />}
|
||||
{isBillingEnabled && effectiveSection === 'organization' && <TeamManagement />}
|
||||
{effectiveSection === 'sso' && <SSO />}
|
||||
{effectiveSection === 'data-retention' && <DataRetentionSettings />}
|
||||
{effectiveSection === 'whitelabeling' && <WhitelabelingSettings />}
|
||||
{effectiveSection === 'byok' && <BYOK />}
|
||||
{effectiveSection === 'copilot' && <Copilot />}
|
||||
|
||||
@@ -2,6 +2,7 @@ import {
|
||||
Card,
|
||||
ClipboardList,
|
||||
Connections,
|
||||
Database,
|
||||
HexSimple,
|
||||
Key,
|
||||
KeySquare,
|
||||
@@ -42,6 +43,7 @@ export type SettingsSection =
|
||||
| 'workflow-mcp-servers'
|
||||
| 'inbox'
|
||||
| 'admin'
|
||||
| 'data-retention'
|
||||
| 'mothership'
|
||||
| 'recently-deleted'
|
||||
|
||||
@@ -178,6 +180,15 @@ export const allNavigationItems: NavigationItem[] = [
|
||||
requiresEnterprise: true,
|
||||
selfHostedOverride: isSSOEnabled,
|
||||
},
|
||||
{
|
||||
id: 'data-retention',
|
||||
label: 'Data Retention',
|
||||
icon: Database,
|
||||
section: 'enterprise',
|
||||
requiresHosted: true,
|
||||
requiresEnterprise: true,
|
||||
showWhenLocked: true,
|
||||
},
|
||||
{
|
||||
id: 'whitelabeling',
|
||||
label: 'Whitelabeling',
|
||||
|
||||
+5
-1
@@ -152,7 +152,11 @@ export function SettingsSidebar({
|
||||
return false
|
||||
}
|
||||
|
||||
if (item.requiresEnterprise && (!hasEnterprisePlan || !isOrgAdminOrOwner)) {
|
||||
if (
|
||||
item.requiresEnterprise &&
|
||||
(!hasEnterprisePlan || !isOrgAdminOrOwner) &&
|
||||
!item.showWhenLocked
|
||||
) {
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
import { db } from '@sim/db'
|
||||
import { workflowExecutionLogs } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { task } from '@trigger.dev/sdk'
|
||||
import { and, inArray, lt } from 'drizzle-orm'
|
||||
import { type CleanupJobPayload, resolveCleanupScope } from '@/lib/billing/cleanup-dispatcher'
|
||||
import { snapshotService } from '@/lib/logs/execution/snapshot/service'
|
||||
import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
|
||||
import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
|
||||
|
||||
const logger = createLogger('CleanupLogs')
|
||||
|
||||
const BATCH_SIZE = 2000
|
||||
const MAX_BATCHES_PER_TIER = 10
|
||||
|
||||
interface TierResults {
|
||||
total: number
|
||||
deleted: number
|
||||
deleteFailed: number
|
||||
filesTotal: number
|
||||
filesDeleted: number
|
||||
filesDeleteFailed: number
|
||||
}
|
||||
|
||||
function emptyTierResults(): TierResults {
|
||||
return {
|
||||
total: 0,
|
||||
deleted: 0,
|
||||
deleteFailed: 0,
|
||||
filesTotal: 0,
|
||||
filesDeleted: 0,
|
||||
filesDeleteFailed: 0,
|
||||
}
|
||||
}
|
||||
|
||||
async function deleteExecutionFiles(files: unknown, results: TierResults): Promise<void> {
|
||||
if (!isUsingCloudStorage() || !files || !Array.isArray(files)) return
|
||||
|
||||
const keys = files.filter((f) => f && typeof f === 'object' && f.key).map((f) => f.key as string)
|
||||
results.filesTotal += keys.length
|
||||
|
||||
await Promise.all(
|
||||
keys.map(async (key) => {
|
||||
try {
|
||||
await StorageService.deleteFile({ key, context: 'execution' })
|
||||
await deleteFileMetadata(key)
|
||||
results.filesDeleted++
|
||||
} catch (fileError) {
|
||||
results.filesDeleteFailed++
|
||||
logger.error(`Failed to delete file ${key}:`, { fileError })
|
||||
}
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
async function cleanupTier(
|
||||
workspaceIds: string[],
|
||||
retentionDate: Date,
|
||||
label: string
|
||||
): Promise<TierResults> {
|
||||
const results = emptyTierResults()
|
||||
if (workspaceIds.length === 0) return results
|
||||
|
||||
let batchesProcessed = 0
|
||||
let hasMore = true
|
||||
|
||||
while (hasMore && batchesProcessed < MAX_BATCHES_PER_TIER) {
|
||||
const batch = await db
|
||||
.select({
|
||||
id: workflowExecutionLogs.id,
|
||||
files: workflowExecutionLogs.files,
|
||||
})
|
||||
.from(workflowExecutionLogs)
|
||||
.where(
|
||||
and(
|
||||
inArray(workflowExecutionLogs.workspaceId, workspaceIds),
|
||||
lt(workflowExecutionLogs.startedAt, retentionDate)
|
||||
)
|
||||
)
|
||||
.limit(BATCH_SIZE)
|
||||
|
||||
results.total += batch.length
|
||||
|
||||
if (batch.length === 0) {
|
||||
hasMore = false
|
||||
break
|
||||
}
|
||||
|
||||
for (const log of batch) {
|
||||
await deleteExecutionFiles(log.files, results)
|
||||
}
|
||||
|
||||
const logIds = batch.map((log) => log.id)
|
||||
try {
|
||||
const deleted = await db
|
||||
.delete(workflowExecutionLogs)
|
||||
.where(inArray(workflowExecutionLogs.id, logIds))
|
||||
.returning({ id: workflowExecutionLogs.id })
|
||||
|
||||
results.deleted += deleted.length
|
||||
} catch (deleteError) {
|
||||
results.deleteFailed += logIds.length
|
||||
logger.error(`Batch delete failed for ${label}:`, { deleteError })
|
||||
}
|
||||
|
||||
batchesProcessed++
|
||||
hasMore = batch.length === BATCH_SIZE
|
||||
|
||||
logger.info(`[${label}] Batch ${batchesProcessed}: ${batch.length} logs processed`)
|
||||
}
|
||||
|
||||
return results
|
||||
}
|
||||
|
||||
export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void> {
|
||||
const startTime = Date.now()
|
||||
|
||||
const scope = await resolveCleanupScope('cleanup-logs', payload)
|
||||
if (!scope) {
|
||||
logger.info(`[${payload.plan}] No retention configured, skipping`)
|
||||
return
|
||||
}
|
||||
|
||||
const { workspaceIds, retentionHours, label } = scope
|
||||
|
||||
if (workspaceIds.length === 0) {
|
||||
logger.info(`[${label}] No workspaces to process`)
|
||||
return
|
||||
}
|
||||
|
||||
const retentionDate = new Date(Date.now() - retentionHours * 60 * 60 * 1000)
|
||||
logger.info(
|
||||
`[${label}] Cleaning ${workspaceIds.length} workspaces, cutoff: ${retentionDate.toISOString()}`
|
||||
)
|
||||
|
||||
const results = await cleanupTier(workspaceIds, retentionDate, label)
|
||||
logger.info(
|
||||
`[${label}] Result: ${results.deleted} deleted, ${results.deleteFailed} failed out of ${results.total} candidates`
|
||||
)
|
||||
|
||||
// Snapshot cleanup runs only on the free job to avoid running it N times for N enterprise workspaces.
|
||||
if (payload.plan === 'free') {
|
||||
try {
|
||||
const retentionDays = Math.floor(retentionHours / 24)
|
||||
const snapshotsCleaned = await snapshotService.cleanupOrphanedSnapshots(retentionDays + 1)
|
||||
logger.info(`Cleaned up ${snapshotsCleaned} orphaned snapshots`)
|
||||
} catch (snapshotError) {
|
||||
logger.error('Error cleaning up orphaned snapshots:', { snapshotError })
|
||||
}
|
||||
}
|
||||
|
||||
const timeElapsed = (Date.now() - startTime) / 1000
|
||||
logger.info(`[${label}] Job completed in ${timeElapsed.toFixed(2)}s`)
|
||||
}
|
||||
|
||||
export const cleanupLogsTask = task({
|
||||
id: 'cleanup-logs',
|
||||
run: runCleanupLogs,
|
||||
})
|
||||
@@ -0,0 +1,273 @@
|
||||
import { db } from '@sim/db'
|
||||
import {
|
||||
a2aAgent,
|
||||
copilotChats,
|
||||
knowledgeBase,
|
||||
mcpServers,
|
||||
memory,
|
||||
userTableDefinitions,
|
||||
workflow,
|
||||
workflowFolder,
|
||||
workflowMcpServer,
|
||||
workspaceFile,
|
||||
workspaceFiles,
|
||||
} from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { task } from '@trigger.dev/sdk'
|
||||
import { and, inArray, isNotNull, lt } from 'drizzle-orm'
|
||||
import { type CleanupJobPayload, resolveCleanupScope } from '@/lib/billing/cleanup-dispatcher'
|
||||
import {
|
||||
batchDeleteByWorkspaceAndTimestamp,
|
||||
DEFAULT_BATCH_SIZE,
|
||||
DEFAULT_MAX_BATCHES_PER_TABLE,
|
||||
deleteRowsById,
|
||||
} from '@/lib/cleanup/batch-delete'
|
||||
import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup'
|
||||
import type { StorageContext } from '@/lib/uploads'
|
||||
import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
|
||||
|
||||
const logger = createLogger('CleanupSoftDeletes')
|
||||
|
||||
interface WorkspaceFileScope {
|
||||
/** Rows from `workspace_file` (singular, legacy workspace-context only). */
|
||||
legacyRows: Array<{ id: string; key: string }>
|
||||
/** Rows from `workspace_files` (plural, multi-context). */
|
||||
multiContextRows: Array<{ id: string; key: string; context: StorageContext }>
|
||||
}
|
||||
|
||||
/**
|
||||
* Select every soft-deleted file row that's eligible for permanent removal.
|
||||
* Returned once and reused for both S3 deletion and DB deletion so the external
|
||||
* cleanup cannot drift from the row-level cleanup.
|
||||
*/
|
||||
async function selectExpiredWorkspaceFiles(
|
||||
workspaceIds: string[],
|
||||
retentionDate: Date
|
||||
): Promise<WorkspaceFileScope> {
|
||||
const limit = DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE
|
||||
|
||||
const [legacyRows, multiContextRows] = await Promise.all([
|
||||
db
|
||||
.select({ id: workspaceFile.id, key: workspaceFile.key })
|
||||
.from(workspaceFile)
|
||||
.where(
|
||||
and(
|
||||
inArray(workspaceFile.workspaceId, workspaceIds),
|
||||
isNotNull(workspaceFile.deletedAt),
|
||||
lt(workspaceFile.deletedAt, retentionDate)
|
||||
)
|
||||
)
|
||||
.limit(limit),
|
||||
db
|
||||
.select({
|
||||
id: workspaceFiles.id,
|
||||
key: workspaceFiles.key,
|
||||
context: workspaceFiles.context,
|
||||
})
|
||||
.from(workspaceFiles)
|
||||
.where(
|
||||
and(
|
||||
inArray(workspaceFiles.workspaceId, workspaceIds),
|
||||
isNotNull(workspaceFiles.deletedAt),
|
||||
lt(workspaceFiles.deletedAt, retentionDate)
|
||||
)
|
||||
)
|
||||
.limit(limit),
|
||||
])
|
||||
|
||||
return {
|
||||
legacyRows,
|
||||
multiContextRows: multiContextRows.map((r) => ({
|
||||
id: r.id,
|
||||
key: r.key,
|
||||
context: r.context as StorageContext,
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
async function cleanupWorkspaceFileStorage(
|
||||
scope: WorkspaceFileScope
|
||||
): Promise<{ filesDeleted: number; filesFailed: number }> {
|
||||
const stats = { filesDeleted: 0, filesFailed: 0 }
|
||||
if (!isUsingCloudStorage()) return stats
|
||||
|
||||
const toDelete: Array<{ key: string; context: StorageContext }> = [
|
||||
...scope.legacyRows.map((r) => ({ key: r.key, context: 'workspace' as StorageContext })),
|
||||
...scope.multiContextRows.map((r) => ({ key: r.key, context: r.context })),
|
||||
]
|
||||
|
||||
await Promise.all(
|
||||
toDelete.map(async ({ key, context }) => {
|
||||
try {
|
||||
await StorageService.deleteFile({ key, context })
|
||||
stats.filesDeleted++
|
||||
} catch (error) {
|
||||
stats.filesFailed++
|
||||
logger.error(`Failed to delete storage file ${key} (context: ${context}):`, { error })
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
/**
|
||||
* Tables cleaned by the generic workspace-scoped batched DELETE. Tables whose
|
||||
* hard-delete triggers external side effects (workflow → copilot chats cascade,
|
||||
* workspace files → S3 storage) are handled explicitly so the SELECT that drives
|
||||
* the external cleanup and the SELECT that drives the DB delete see the same rows.
|
||||
*/
|
||||
const CLEANUP_TARGETS = [
|
||||
{
|
||||
table: workflowFolder,
|
||||
softDeleteCol: workflowFolder.archivedAt,
|
||||
wsCol: workflowFolder.workspaceId,
|
||||
name: 'workflowFolder',
|
||||
},
|
||||
{
|
||||
table: knowledgeBase,
|
||||
softDeleteCol: knowledgeBase.deletedAt,
|
||||
wsCol: knowledgeBase.workspaceId,
|
||||
name: 'knowledgeBase',
|
||||
},
|
||||
{
|
||||
table: userTableDefinitions,
|
||||
softDeleteCol: userTableDefinitions.archivedAt,
|
||||
wsCol: userTableDefinitions.workspaceId,
|
||||
name: 'userTableDefinitions',
|
||||
},
|
||||
{ table: memory, softDeleteCol: memory.deletedAt, wsCol: memory.workspaceId, name: 'memory' },
|
||||
{
|
||||
table: mcpServers,
|
||||
softDeleteCol: mcpServers.deletedAt,
|
||||
wsCol: mcpServers.workspaceId,
|
||||
name: 'mcpServers',
|
||||
},
|
||||
{
|
||||
table: workflowMcpServer,
|
||||
softDeleteCol: workflowMcpServer.deletedAt,
|
||||
wsCol: workflowMcpServer.workspaceId,
|
||||
name: 'workflowMcpServer',
|
||||
},
|
||||
{
|
||||
table: a2aAgent,
|
||||
softDeleteCol: a2aAgent.archivedAt,
|
||||
wsCol: a2aAgent.workspaceId,
|
||||
name: 'a2aAgent',
|
||||
},
|
||||
] as const
|
||||
|
||||
export async function runCleanupSoftDeletes(payload: CleanupJobPayload): Promise<void> {
|
||||
const startTime = Date.now()
|
||||
|
||||
const scope = await resolveCleanupScope('cleanup-soft-deletes', payload)
|
||||
if (!scope) {
|
||||
logger.info(`[${payload.plan}] No retention configured, skipping`)
|
||||
return
|
||||
}
|
||||
|
||||
const { workspaceIds, retentionHours, label } = scope
|
||||
|
||||
if (workspaceIds.length === 0) {
|
||||
logger.info(`[${label}] No workspaces to process`)
|
||||
return
|
||||
}
|
||||
|
||||
const retentionDate = new Date(Date.now() - retentionHours * 60 * 60 * 1000)
|
||||
logger.info(
|
||||
`[${label}] Processing ${workspaceIds.length} workspaces, cutoff: ${retentionDate.toISOString()}`
|
||||
)
|
||||
|
||||
// Select workflows + files once. These sets drive BOTH external cleanup
|
||||
// (chats + S3) AND the DB deletes below — selecting twice could return
|
||||
// different subsets above the LIMIT cap and orphan or prematurely purge data.
|
||||
const [doomedWorkflows, fileScope] = await Promise.all([
|
||||
db
|
||||
.select({ id: workflow.id })
|
||||
.from(workflow)
|
||||
.where(
|
||||
and(
|
||||
inArray(workflow.workspaceId, workspaceIds),
|
||||
isNotNull(workflow.archivedAt),
|
||||
lt(workflow.archivedAt, retentionDate)
|
||||
)
|
||||
)
|
||||
.limit(DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE),
|
||||
selectExpiredWorkspaceFiles(workspaceIds, retentionDate),
|
||||
])
|
||||
|
||||
const doomedWorkflowIds = doomedWorkflows.map((w) => w.id)
|
||||
let chatCleanup: { execute: () => Promise<void> } | null = null
|
||||
|
||||
if (doomedWorkflowIds.length > 0) {
|
||||
const doomedChats = await db
|
||||
.select({ id: copilotChats.id })
|
||||
.from(copilotChats)
|
||||
.where(inArray(copilotChats.workflowId, doomedWorkflowIds))
|
||||
.limit(DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE)
|
||||
|
||||
const doomedChatIds = doomedChats.map((c) => c.id)
|
||||
if (doomedChatIds.length > 0) {
|
||||
chatCleanup = await prepareChatCleanup(doomedChatIds, label)
|
||||
}
|
||||
}
|
||||
|
||||
const fileStats = await cleanupWorkspaceFileStorage(fileScope)
|
||||
|
||||
let totalDeleted = 0
|
||||
|
||||
// Delete the workflow + file rows using the exact IDs we already selected.
|
||||
const workflowResult = await deleteRowsById(
|
||||
workflow,
|
||||
workflow.id,
|
||||
doomedWorkflowIds,
|
||||
`${label}/workflow`
|
||||
)
|
||||
totalDeleted += workflowResult.deleted
|
||||
|
||||
const legacyFileResult = await deleteRowsById(
|
||||
workspaceFile,
|
||||
workspaceFile.id,
|
||||
fileScope.legacyRows.map((r) => r.id),
|
||||
`${label}/workspaceFile`
|
||||
)
|
||||
totalDeleted += legacyFileResult.deleted
|
||||
|
||||
const multiContextFileResult = await deleteRowsById(
|
||||
workspaceFiles,
|
||||
workspaceFiles.id,
|
||||
fileScope.multiContextRows.map((r) => r.id),
|
||||
`${label}/workspaceFiles`
|
||||
)
|
||||
totalDeleted += multiContextFileResult.deleted
|
||||
|
||||
for (const target of CLEANUP_TARGETS) {
|
||||
const result = await batchDeleteByWorkspaceAndTimestamp({
|
||||
tableDef: target.table,
|
||||
workspaceIdCol: target.wsCol,
|
||||
timestampCol: target.softDeleteCol,
|
||||
workspaceIds,
|
||||
retentionDate,
|
||||
tableName: `${label}/${target.name}`,
|
||||
requireTimestampNotNull: true,
|
||||
})
|
||||
totalDeleted += result.deleted
|
||||
}
|
||||
|
||||
logger.info(
|
||||
`[${label}] Complete: ${totalDeleted} rows deleted, ${fileStats.filesDeleted} files cleaned`
|
||||
)
|
||||
|
||||
// Clean up copilot backend + chat storage files after DB rows are gone
|
||||
if (chatCleanup) {
|
||||
await chatCleanup.execute()
|
||||
}
|
||||
|
||||
const timeElapsed = (Date.now() - startTime) / 1000
|
||||
logger.info(`[${label}] Job completed in ${timeElapsed.toFixed(2)}s`)
|
||||
}
|
||||
|
||||
export const cleanupSoftDeletesTask = task({
|
||||
id: 'cleanup-soft-deletes',
|
||||
run: runCleanupSoftDeletes,
|
||||
})
|
||||
@@ -0,0 +1,203 @@
|
||||
import { db } from '@sim/db'
|
||||
import {
|
||||
copilotAsyncToolCalls,
|
||||
copilotChats,
|
||||
copilotFeedback,
|
||||
copilotRunCheckpoints,
|
||||
copilotRuns,
|
||||
mothershipInboxTask,
|
||||
} from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { task } from '@trigger.dev/sdk'
|
||||
import { and, inArray, lt, sql } from 'drizzle-orm'
|
||||
import { type CleanupJobPayload, resolveCleanupScope } from '@/lib/billing/cleanup-dispatcher'
|
||||
import {
|
||||
batchDeleteByWorkspaceAndTimestamp,
|
||||
DEFAULT_BATCH_SIZE,
|
||||
DEFAULT_MAX_BATCHES_PER_TABLE,
|
||||
deleteRowsById,
|
||||
type TableCleanupResult,
|
||||
} from '@/lib/cleanup/batch-delete'
|
||||
import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup'
|
||||
|
||||
const logger = createLogger('CleanupTasks')
|
||||
|
||||
/**
|
||||
* Delete copilot run checkpoints and async tool calls via join through copilotRuns.
|
||||
* These tables don't have a direct workspaceId — we find qualifying run IDs first.
|
||||
*/
|
||||
const RUN_CHILD_TABLES = [
|
||||
{
|
||||
table: copilotRunCheckpoints,
|
||||
runIdCol: copilotRunCheckpoints.runId,
|
||||
name: 'copilotRunCheckpoints',
|
||||
},
|
||||
{
|
||||
table: copilotAsyncToolCalls,
|
||||
runIdCol: copilotAsyncToolCalls.runId,
|
||||
name: 'copilotAsyncToolCalls',
|
||||
},
|
||||
] as const
|
||||
|
||||
async function deleteByRunIds(
|
||||
table: (typeof RUN_CHILD_TABLES)[number]['table'],
|
||||
runIdCol: (typeof RUN_CHILD_TABLES)[number]['runIdCol'],
|
||||
runIds: string[],
|
||||
tableName: string
|
||||
): Promise<TableCleanupResult> {
|
||||
const result: TableCleanupResult = { table: tableName, deleted: 0, failed: 0 }
|
||||
try {
|
||||
const deleted = await db
|
||||
.delete(table)
|
||||
.where(inArray(runIdCol, runIds))
|
||||
.returning({ id: sql`id` })
|
||||
result.deleted = deleted.length
|
||||
logger.info(`[${tableName}] Deleted ${deleted.length} rows`)
|
||||
} catch (error) {
|
||||
result.failed++
|
||||
logger.error(`[${tableName}] Delete failed:`, { error })
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
async function cleanupRunChildren(
|
||||
workspaceIds: string[],
|
||||
retentionDate: Date,
|
||||
label: string
|
||||
): Promise<TableCleanupResult[]> {
|
||||
if (workspaceIds.length === 0) return []
|
||||
|
||||
const runIds = await db
|
||||
.select({ id: copilotRuns.id })
|
||||
.from(copilotRuns)
|
||||
.where(
|
||||
and(inArray(copilotRuns.workspaceId, workspaceIds), lt(copilotRuns.updatedAt, retentionDate))
|
||||
)
|
||||
.limit(DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE)
|
||||
|
||||
if (runIds.length === 0) {
|
||||
return RUN_CHILD_TABLES.map((t) => ({ table: `${label}/${t.name}`, deleted: 0, failed: 0 }))
|
||||
}
|
||||
|
||||
const ids = runIds.map((r) => r.id)
|
||||
|
||||
return Promise.all(
|
||||
RUN_CHILD_TABLES.map((t) => deleteByRunIds(t.table, t.runIdCol, ids, `${label}/${t.name}`))
|
||||
)
|
||||
}
|
||||
|
||||
export async function runCleanupTasks(payload: CleanupJobPayload): Promise<void> {
|
||||
const startTime = Date.now()
|
||||
|
||||
const scope = await resolveCleanupScope('cleanup-tasks', payload)
|
||||
if (!scope) {
|
||||
logger.info(`[${payload.plan}] No retention configured, skipping`)
|
||||
return
|
||||
}
|
||||
|
||||
const { workspaceIds, retentionHours, label } = scope
|
||||
|
||||
if (workspaceIds.length === 0) {
|
||||
logger.info(`[${label}] No workspaces to process`)
|
||||
return
|
||||
}
|
||||
|
||||
const retentionDate = new Date(Date.now() - retentionHours * 60 * 60 * 1000)
|
||||
logger.info(
|
||||
`[${label}] Processing ${workspaceIds.length} workspaces, cutoff: ${retentionDate.toISOString()}`
|
||||
)
|
||||
|
||||
// Collect chat IDs before deleting so we can clean up the copilot backend after
|
||||
const doomedChats = await db
|
||||
.select({ id: copilotChats.id })
|
||||
.from(copilotChats)
|
||||
.where(
|
||||
and(
|
||||
inArray(copilotChats.workspaceId, workspaceIds),
|
||||
lt(copilotChats.updatedAt, retentionDate)
|
||||
)
|
||||
)
|
||||
.limit(DEFAULT_BATCH_SIZE * DEFAULT_MAX_BATCHES_PER_TABLE)
|
||||
|
||||
const doomedChatIds = doomedChats.map((c) => c.id)
|
||||
|
||||
// Prepare chat cleanup (collect file keys + copilot backend call) BEFORE DB deletion
|
||||
const chatCleanup = await prepareChatCleanup(doomedChatIds, label)
|
||||
|
||||
// Delete run children first (checkpoints, tool calls) since they reference runs
|
||||
const runChildResults = await cleanupRunChildren(workspaceIds, retentionDate, label)
|
||||
for (const r of runChildResults) {
|
||||
if (r.deleted > 0) logger.info(`[${r.table}] ${r.deleted} deleted`)
|
||||
}
|
||||
|
||||
// Delete feedback — no direct workspaceId, reuse chat IDs collected above
|
||||
const feedbackResult: TableCleanupResult = {
|
||||
table: `${label}/copilotFeedback`,
|
||||
deleted: 0,
|
||||
failed: 0,
|
||||
}
|
||||
try {
|
||||
if (doomedChatIds.length > 0) {
|
||||
const deleted = await db
|
||||
.delete(copilotFeedback)
|
||||
.where(inArray(copilotFeedback.chatId, doomedChatIds))
|
||||
.returning({ id: copilotFeedback.feedbackId })
|
||||
feedbackResult.deleted = deleted.length
|
||||
logger.info(`[${feedbackResult.table}] Deleted ${deleted.length} rows`)
|
||||
} else {
|
||||
logger.info(`[${feedbackResult.table}] No expired rows found`)
|
||||
}
|
||||
} catch (error) {
|
||||
feedbackResult.failed++
|
||||
logger.error(`[${feedbackResult.table}] Delete failed:`, { error })
|
||||
}
|
||||
|
||||
// Delete copilot runs (has workspaceId directly, cascades checkpoints)
|
||||
const runsResult = await batchDeleteByWorkspaceAndTimestamp({
|
||||
tableDef: copilotRuns,
|
||||
workspaceIdCol: copilotRuns.workspaceId,
|
||||
timestampCol: copilotRuns.updatedAt,
|
||||
workspaceIds,
|
||||
retentionDate,
|
||||
tableName: `${label}/copilotRuns`,
|
||||
})
|
||||
|
||||
// Delete copilot chats using the exact IDs collected above so the chat
|
||||
// cleanup (S3 + copilot backend) and the DB delete can never disagree.
|
||||
const chatsResult = await deleteRowsById(
|
||||
copilotChats,
|
||||
copilotChats.id,
|
||||
doomedChatIds,
|
||||
`${label}/copilotChats`
|
||||
)
|
||||
|
||||
// Delete mothership inbox tasks (has workspaceId directly)
|
||||
const inboxResult = await batchDeleteByWorkspaceAndTimestamp({
|
||||
tableDef: mothershipInboxTask,
|
||||
workspaceIdCol: mothershipInboxTask.workspaceId,
|
||||
timestampCol: mothershipInboxTask.createdAt,
|
||||
workspaceIds,
|
||||
retentionDate,
|
||||
tableName: `${label}/mothershipInboxTask`,
|
||||
})
|
||||
|
||||
const totalDeleted =
|
||||
runChildResults.reduce((s, r) => s + r.deleted, 0) +
|
||||
feedbackResult.deleted +
|
||||
runsResult.deleted +
|
||||
chatsResult.deleted +
|
||||
inboxResult.deleted
|
||||
|
||||
logger.info(`[${label}] Complete: ${totalDeleted} total rows deleted`)
|
||||
|
||||
// Clean up copilot backend + storage files after DB rows are gone
|
||||
await chatCleanup.execute()
|
||||
|
||||
const timeElapsed = (Date.now() - startTime) / 1000
|
||||
logger.info(`Task cleanup completed in ${timeElapsed.toFixed(2)}s`)
|
||||
}
|
||||
|
||||
export const cleanupTasksTask = task({
|
||||
id: 'cleanup-tasks',
|
||||
run: runCleanupTasks,
|
||||
})
|
||||
@@ -0,0 +1,220 @@
|
||||
'use client'
|
||||
|
||||
import { useCallback, useState } from 'react'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { toError } from '@sim/utils/errors'
|
||||
import { Loader2 } from 'lucide-react'
|
||||
import { useParams } from 'next/navigation'
|
||||
import { Button, Label } from '@/components/emcn'
|
||||
import {
|
||||
Select,
|
||||
SelectContent,
|
||||
SelectItem,
|
||||
SelectTrigger,
|
||||
SelectValue,
|
||||
} from '@/components/ui/select'
|
||||
import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/providers/workspace-permissions-provider'
|
||||
import {
|
||||
useUpdateWorkspaceRetention,
|
||||
useWorkspaceRetention,
|
||||
} from '@/ee/data-retention/hooks/data-retention'
|
||||
|
||||
const logger = createLogger('DataRetentionSettings')
|
||||
|
||||
const DAY_OPTIONS = [
|
||||
{ value: '1', label: '1 day' },
|
||||
{ value: '3', label: '3 days' },
|
||||
{ value: '7', label: '7 days' },
|
||||
{ value: '14', label: '14 days' },
|
||||
{ value: '30', label: '30 days' },
|
||||
{ value: '60', label: '60 days' },
|
||||
{ value: '90', label: '90 days' },
|
||||
{ value: '180', label: '180 days' },
|
||||
{ value: '365', label: '1 year' },
|
||||
{ value: '1825', label: '5 years' },
|
||||
{ value: 'never', label: 'Forever' },
|
||||
] as const
|
||||
|
||||
function hoursToDisplayDays(hours: number | null): string {
|
||||
if (hours === null) return 'never'
|
||||
return String(Math.round(hours / 24))
|
||||
}
|
||||
|
||||
function daysToHours(days: string): number | null {
|
||||
if (days === 'never') return null
|
||||
return Number(days) * 24
|
||||
}
|
||||
|
||||
interface SettingRowProps {
|
||||
label: string
|
||||
description?: string
|
||||
children: React.ReactNode
|
||||
}
|
||||
|
||||
function SettingRow({ label, description, children }: SettingRowProps) {
|
||||
return (
|
||||
<div className='flex flex-col gap-1.5'>
|
||||
<Label className='text-[13px] text-[var(--text-primary)]'>{label}</Label>
|
||||
{description && <p className='text-[12px] text-[var(--text-muted)]'>{description}</p>}
|
||||
{children}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
function SectionTitle({ children }: { children: React.ReactNode }) {
|
||||
return <h3 className='mb-4 font-medium text-[15px] text-[var(--text-primary)]'>{children}</h3>
|
||||
}
|
||||
|
||||
interface RetentionSelectProps {
|
||||
value: string
|
||||
onChange: (value: string) => void
|
||||
}
|
||||
|
||||
function RetentionSelect({ value, onChange }: RetentionSelectProps) {
|
||||
const standard = DAY_OPTIONS.find((o) => o.value === value)
|
||||
const options = standard
|
||||
? DAY_OPTIONS
|
||||
: [...DAY_OPTIONS, { value, label: `${value} days (custom)` } as const]
|
||||
|
||||
return (
|
||||
<Select value={value} onValueChange={onChange}>
|
||||
<SelectTrigger className='h-[36px] max-w-[200px] text-[13px]'>
|
||||
<SelectValue />
|
||||
</SelectTrigger>
|
||||
<SelectContent>
|
||||
{options.map((opt) => (
|
||||
<SelectItem key={opt.value} value={opt.value} className='text-[13px]'>
|
||||
{opt.label}
|
||||
</SelectItem>
|
||||
))}
|
||||
</SelectContent>
|
||||
</Select>
|
||||
)
|
||||
}
|
||||
|
||||
export function DataRetentionSettings() {
|
||||
const params = useParams<{ workspaceId: string }>()
|
||||
const workspaceId = params.workspaceId
|
||||
|
||||
const { data, isLoading } = useWorkspaceRetention(workspaceId)
|
||||
const { canAdmin } = useUserPermissionsContext()
|
||||
const updateMutation = useUpdateWorkspaceRetention()
|
||||
|
||||
const [logDays, setLogDays] = useState('')
|
||||
const [softDeleteDays, setSoftDeleteDays] = useState('')
|
||||
const [taskCleanupDays, setTaskCleanupDays] = useState('')
|
||||
const [formInitialized, setFormInitialized] = useState(false)
|
||||
|
||||
const [saveError, setSaveError] = useState<string | null>(null)
|
||||
const [saveSuccess, setSaveSuccess] = useState(false)
|
||||
|
||||
if (data && !formInitialized) {
|
||||
setLogDays(hoursToDisplayDays(data.effective.logRetentionHours))
|
||||
setSoftDeleteDays(hoursToDisplayDays(data.effective.softDeleteRetentionHours))
|
||||
setTaskCleanupDays(hoursToDisplayDays(data.effective.taskCleanupHours))
|
||||
setFormInitialized(true)
|
||||
}
|
||||
|
||||
const handleSave = useCallback(async () => {
|
||||
setSaveError(null)
|
||||
setSaveSuccess(false)
|
||||
|
||||
try {
|
||||
await updateMutation.mutateAsync({
|
||||
workspaceId,
|
||||
settings: {
|
||||
logRetentionHours: daysToHours(logDays),
|
||||
softDeleteRetentionHours: daysToHours(softDeleteDays),
|
||||
taskCleanupHours: daysToHours(taskCleanupDays),
|
||||
},
|
||||
})
|
||||
setSaveSuccess(true)
|
||||
setTimeout(() => setSaveSuccess(false), 3000)
|
||||
} catch (error) {
|
||||
logger.error('Failed to save data retention settings', { error })
|
||||
setSaveError(toError(error).message)
|
||||
}
|
||||
}, [workspaceId, logDays, softDeleteDays, taskCleanupDays])
|
||||
|
||||
if (isLoading) {
|
||||
return (
|
||||
<div className='flex flex-col gap-8'>
|
||||
{[...Array(3)].map((_, i) => (
|
||||
<div key={i} className='flex flex-col gap-3'>
|
||||
<div className='h-4 w-32 animate-pulse rounded bg-[var(--surface-3)]' />
|
||||
<div className='h-9 w-full animate-pulse rounded-lg bg-[var(--surface-3)]' />
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
if (!data) {
|
||||
return (
|
||||
<div className='flex h-full items-center justify-center text-[var(--text-muted)] text-sm'>
|
||||
Failed to load data retention settings.
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
if (!data.isEnterprise) {
|
||||
return (
|
||||
<div className='flex h-full items-center justify-center text-[var(--text-muted)] text-sm'>
|
||||
Data retention is available on Enterprise plans only.
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
if (!canAdmin) {
|
||||
return (
|
||||
<div className='flex h-full items-center justify-center text-[var(--text-muted)] text-sm'>
|
||||
Only workspace admins can configure data retention settings.
|
||||
</div>
|
||||
)
|
||||
}
|
||||
|
||||
return (
|
||||
<div className='flex flex-col gap-8'>
|
||||
<section>
|
||||
<SectionTitle>Retention Periods</SectionTitle>
|
||||
<div className='flex flex-col gap-5'>
|
||||
<SettingRow
|
||||
label='Log retention'
|
||||
description='How long execution logs are kept before they are permanently deleted.'
|
||||
>
|
||||
<RetentionSelect value={logDays} onChange={setLogDays} />
|
||||
</SettingRow>
|
||||
<SettingRow
|
||||
label='Soft deletion cleanup'
|
||||
description='How long deleted resources remain recoverable before they are permanently removed.'
|
||||
>
|
||||
<RetentionSelect value={softDeleteDays} onChange={setSoftDeleteDays} />
|
||||
</SettingRow>
|
||||
<SettingRow
|
||||
label='Task cleanup'
|
||||
description='How long copilot chats, runs, and inbox tasks are kept before they are permanently deleted.'
|
||||
>
|
||||
<RetentionSelect value={taskCleanupDays} onChange={setTaskCleanupDays} />
|
||||
</SettingRow>
|
||||
</div>
|
||||
</section>
|
||||
|
||||
<div className='flex items-center gap-3'>
|
||||
<Button onClick={handleSave} disabled={updateMutation.isPending} className='text-[13px]'>
|
||||
{updateMutation.isPending ? (
|
||||
<>
|
||||
<Loader2 className='mr-2 h-3.5 w-3.5 animate-spin' />
|
||||
Saving…
|
||||
</>
|
||||
) : (
|
||||
'Save changes'
|
||||
)}
|
||||
</Button>
|
||||
{saveSuccess && (
|
||||
<span className='text-[13px] text-green-500'>Settings saved successfully.</span>
|
||||
)}
|
||||
{saveError && <span className='text-[13px] text-red-500'>{saveError}</span>}
|
||||
</div>
|
||||
</div>
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
'use client'
|
||||
|
||||
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
|
||||
import type { PlanCategory } from '@/lib/billing/plan-helpers'
|
||||
|
||||
export interface RetentionValues {
|
||||
logRetentionHours: number | null
|
||||
softDeleteRetentionHours: number | null
|
||||
taskCleanupHours: number | null
|
||||
}
|
||||
|
||||
export interface DataRetentionResponse {
|
||||
plan: PlanCategory
|
||||
isEnterprise: boolean
|
||||
defaults: RetentionValues
|
||||
configured: RetentionValues
|
||||
effective: RetentionValues
|
||||
}
|
||||
|
||||
export const dataRetentionKeys = {
|
||||
all: ['dataRetention'] as const,
|
||||
settings: (workspaceId: string) => [...dataRetentionKeys.all, 'settings', workspaceId] as const,
|
||||
}
|
||||
|
||||
async function fetchDataRetention(
|
||||
workspaceId: string,
|
||||
signal?: AbortSignal
|
||||
): Promise<DataRetentionResponse> {
|
||||
const response = await fetch(`/api/workspaces/${workspaceId}/data-retention`, { signal })
|
||||
|
||||
if (!response.ok) {
|
||||
const error = await response.json().catch(() => ({}))
|
||||
throw new Error(error.error ?? 'Failed to fetch data retention settings')
|
||||
}
|
||||
|
||||
const { data } = await response.json()
|
||||
return data as DataRetentionResponse
|
||||
}
|
||||
|
||||
export function useWorkspaceRetention(workspaceId: string | undefined) {
|
||||
return useQuery({
|
||||
queryKey: dataRetentionKeys.settings(workspaceId ?? ''),
|
||||
queryFn: ({ signal }) => fetchDataRetention(workspaceId as string, signal),
|
||||
enabled: Boolean(workspaceId),
|
||||
staleTime: 60 * 1000,
|
||||
})
|
||||
}
|
||||
|
||||
interface UpdateRetentionVariables {
|
||||
workspaceId: string
|
||||
settings: Partial<RetentionValues>
|
||||
}
|
||||
|
||||
export function useUpdateWorkspaceRetention() {
|
||||
const queryClient = useQueryClient()
|
||||
|
||||
return useMutation({
|
||||
mutationFn: async ({ workspaceId, settings }: UpdateRetentionVariables) => {
|
||||
const response = await fetch(`/api/workspaces/${workspaceId}/data-retention`, {
|
||||
method: 'PUT',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify(settings),
|
||||
})
|
||||
|
||||
if (!response.ok) {
|
||||
const error = await response.json().catch(() => ({}))
|
||||
throw new Error(error.error ?? 'Failed to update data retention settings')
|
||||
}
|
||||
|
||||
const { data } = await response.json()
|
||||
return data as DataRetentionResponse
|
||||
},
|
||||
onSettled: (_data, _error, { workspaceId }) => {
|
||||
queryClient.invalidateQueries({ queryKey: dataRetentionKeys.settings(workspaceId) })
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
import { db } from '@sim/db'
|
||||
import { subscription, workspace } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { toError } from '@sim/utils/errors'
|
||||
import { tasks } from '@trigger.dev/sdk'
|
||||
import { and, eq, inArray, isNotNull, isNull } from 'drizzle-orm'
|
||||
import { type PlanCategory, sqlIsPaid, sqlIsPro, sqlIsTeam } from '@/lib/billing/plan-helpers'
|
||||
import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils'
|
||||
import { getJobQueue } from '@/lib/core/async-jobs'
|
||||
import { shouldExecuteInline } from '@/lib/core/async-jobs/config'
|
||||
import { isTriggerAvailable } from '@/lib/knowledge/documents/service'
|
||||
|
||||
const logger = createLogger('RetentionDispatcher')
|
||||
|
||||
const BATCH_TRIGGER_CHUNK_SIZE = 1000
|
||||
|
||||
export type CleanupJobType = 'cleanup-logs' | 'cleanup-soft-deletes' | 'cleanup-tasks'
|
||||
|
||||
export type WorkspaceRetentionColumn =
|
||||
| 'logRetentionHours'
|
||||
| 'softDeleteRetentionHours'
|
||||
| 'taskCleanupHours'
|
||||
|
||||
export type NonEnterprisePlan = Exclude<PlanCategory, 'enterprise'>
|
||||
|
||||
const NON_ENTERPRISE_PLANS = ['free', 'pro', 'team'] as const satisfies readonly NonEnterprisePlan[]
|
||||
|
||||
export type CleanupJobPayload =
|
||||
| { plan: NonEnterprisePlan }
|
||||
| { plan: 'enterprise'; workspaceId: string }
|
||||
|
||||
interface CleanupJobConfig {
|
||||
column: WorkspaceRetentionColumn
|
||||
defaults: Record<PlanCategory, number | null>
|
||||
}
|
||||
|
||||
const DAY = 24
|
||||
|
||||
/**
|
||||
* Single source of truth for cleanup retention: which workspace column each job
|
||||
* type inspects, and the default retention (in hours) per plan. Enterprise is
|
||||
* always `null` here — enterprise tenants must set their own value per workspace.
|
||||
*/
|
||||
export const CLEANUP_CONFIG = {
|
||||
'cleanup-logs': {
|
||||
column: 'logRetentionHours',
|
||||
defaults: { free: 7 * DAY, pro: null, team: null, enterprise: null },
|
||||
},
|
||||
'cleanup-soft-deletes': {
|
||||
column: 'softDeleteRetentionHours',
|
||||
defaults: { free: 7 * DAY, pro: 30 * DAY, team: 30 * DAY, enterprise: null },
|
||||
},
|
||||
'cleanup-tasks': {
|
||||
column: 'taskCleanupHours',
|
||||
defaults: { free: null, pro: null, team: null, enterprise: null },
|
||||
},
|
||||
} as const satisfies Record<CleanupJobType, CleanupJobConfig>
|
||||
|
||||
/**
|
||||
* Bulk-lookup workspace IDs for a non-enterprise plan category. Enterprise is
|
||||
* per-workspace (has explicit opt-in retention), so it's not handled here.
|
||||
*/
|
||||
export async function resolveWorkspaceIdsForPlan(plan: NonEnterprisePlan): Promise<string[]> {
|
||||
if (plan === 'free') {
|
||||
const rows = await db
|
||||
.select({ id: workspace.id })
|
||||
.from(workspace)
|
||||
.leftJoin(
|
||||
subscription,
|
||||
and(
|
||||
eq(subscription.referenceId, workspace.billedAccountUserId),
|
||||
inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES),
|
||||
sqlIsPaid(subscription.plan)
|
||||
)
|
||||
)
|
||||
.where(and(isNull(subscription.id), isNull(workspace.archivedAt)))
|
||||
|
||||
return rows.map((r) => r.id)
|
||||
}
|
||||
|
||||
const planPredicate = plan === 'pro' ? sqlIsPro(subscription.plan) : sqlIsTeam(subscription.plan)
|
||||
const rows = await db
|
||||
.select({ id: workspace.id })
|
||||
.from(workspace)
|
||||
.innerJoin(
|
||||
subscription,
|
||||
and(
|
||||
eq(subscription.referenceId, workspace.billedAccountUserId),
|
||||
inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES),
|
||||
planPredicate!
|
||||
)
|
||||
)
|
||||
.where(isNull(workspace.archivedAt))
|
||||
.groupBy(workspace.id)
|
||||
|
||||
return rows.map((r) => r.id)
|
||||
}
|
||||
|
||||
export interface ResolvedCleanupScope {
|
||||
workspaceIds: string[]
|
||||
retentionHours: number
|
||||
label: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Translate a queued cleanup payload into a concrete cleanup scope: the set of
|
||||
* workspaces and the retention cutoff to apply. Returns `null` when the plan
|
||||
* has no retention configured (default is null, or the enterprise workspace
|
||||
* has not opted in).
|
||||
*/
|
||||
export async function resolveCleanupScope(
|
||||
jobType: CleanupJobType,
|
||||
payload: CleanupJobPayload
|
||||
): Promise<ResolvedCleanupScope | null> {
|
||||
const config = CLEANUP_CONFIG[jobType]
|
||||
|
||||
if (payload.plan !== 'enterprise') {
|
||||
const retentionHours = config.defaults[payload.plan]
|
||||
if (retentionHours === null) return null
|
||||
const workspaceIds = await resolveWorkspaceIdsForPlan(payload.plan)
|
||||
return { workspaceIds, retentionHours, label: payload.plan }
|
||||
}
|
||||
|
||||
const [ws] = await db
|
||||
.select({ hours: workspace[config.column] })
|
||||
.from(workspace)
|
||||
.where(eq(workspace.id, payload.workspaceId))
|
||||
.limit(1)
|
||||
|
||||
if (ws?.hours == null) return null
|
||||
|
||||
return {
|
||||
workspaceIds: [payload.workspaceId],
|
||||
retentionHours: ws.hours,
|
||||
label: `enterprise/${payload.workspaceId}`,
|
||||
}
|
||||
}
|
||||
|
||||
type RunnerFn = (payload: CleanupJobPayload) => Promise<void>
|
||||
|
||||
async function getInlineRunner(jobType: CleanupJobType): Promise<RunnerFn> {
|
||||
switch (jobType) {
|
||||
case 'cleanup-logs': {
|
||||
const { runCleanupLogs } = await import('@/background/cleanup-logs')
|
||||
return runCleanupLogs
|
||||
}
|
||||
case 'cleanup-soft-deletes': {
|
||||
const { runCleanupSoftDeletes } = await import('@/background/cleanup-soft-deletes')
|
||||
return runCleanupSoftDeletes
|
||||
}
|
||||
case 'cleanup-tasks': {
|
||||
const { runCleanupTasks } = await import('@/background/cleanup-tasks')
|
||||
return runCleanupTasks
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* When the job queue backend is "database" (no Trigger.dev, no BullMQ), the
|
||||
* enqueued rows just sit in async_jobs forever. Run them inline as fire-and-forget
|
||||
* promises, following the same pattern as the workflow execution API route.
|
||||
*/
|
||||
async function runInlineIfNeeded(
|
||||
jobQueue: Awaited<ReturnType<typeof getJobQueue>>,
|
||||
jobType: CleanupJobType,
|
||||
jobId: string,
|
||||
payload: CleanupJobPayload
|
||||
): Promise<void> {
|
||||
if (!shouldExecuteInline()) return
|
||||
const runner = await getInlineRunner(jobType)
|
||||
void (async () => {
|
||||
try {
|
||||
await jobQueue.startJob(jobId)
|
||||
await runner(payload)
|
||||
await jobQueue.completeJob(jobId, null)
|
||||
} catch (error) {
|
||||
const errorMessage = toError(error).message
|
||||
logger.error(`[${jobType}] Inline job ${jobId} failed`, { error: errorMessage })
|
||||
try {
|
||||
await jobQueue.markJobFailed(jobId, errorMessage)
|
||||
} catch (markErr) {
|
||||
logger.error(`[${jobType}] Failed to mark job ${jobId} as failed`, { markErr })
|
||||
}
|
||||
}
|
||||
})()
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispatcher: enqueue cleanup jobs driven by `CLEANUP_CONFIG`.
|
||||
*
|
||||
* - One job per non-enterprise plan with a non-null default
|
||||
* - One enterprise job per workspace with a non-NULL retention value in the column
|
||||
*
|
||||
* Uses Trigger.dev batchTrigger when available, otherwise parallel enqueue via
|
||||
* the JobQueueBackend abstraction. On the database backend (no external worker),
|
||||
* jobs run inline in the same process via fire-and-forget promises.
|
||||
*/
|
||||
export async function dispatchCleanupJobs(
|
||||
jobType: CleanupJobType
|
||||
): Promise<{ jobIds: string[]; jobCount: number; enterpriseCount: number }> {
|
||||
const config = CLEANUP_CONFIG[jobType]
|
||||
const jobQueue = await getJobQueue()
|
||||
const jobIds: string[] = []
|
||||
|
||||
const plansWithDefaults = NON_ENTERPRISE_PLANS.filter((plan) => config.defaults[plan] !== null)
|
||||
|
||||
for (const plan of plansWithDefaults) {
|
||||
const payload: CleanupJobPayload = { plan }
|
||||
const jobId = await jobQueue.enqueue(jobType, payload)
|
||||
jobIds.push(jobId)
|
||||
await runInlineIfNeeded(jobQueue, jobType, jobId, payload)
|
||||
}
|
||||
|
||||
// Enterprise: query workspaces with non-NULL retention column. The JOIN can
|
||||
// match multiple subscription rows per workspace (e.g. active + past_due both
|
||||
// in ENTITLED_SUBSCRIPTION_STATUSES) — groupBy dedupes to one row per workspace
|
||||
// so we don't dispatch the same cleanup job twice.
|
||||
const retentionCol = workspace[config.column]
|
||||
const enterpriseRows = await db
|
||||
.select({ id: workspace.id })
|
||||
.from(workspace)
|
||||
.innerJoin(
|
||||
subscription,
|
||||
and(
|
||||
eq(subscription.referenceId, workspace.billedAccountUserId),
|
||||
inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES),
|
||||
eq(subscription.plan, 'enterprise')
|
||||
)
|
||||
)
|
||||
.where(and(isNull(workspace.archivedAt), isNotNull(retentionCol)))
|
||||
.groupBy(workspace.id)
|
||||
|
||||
const enterpriseCount = enterpriseRows.length
|
||||
|
||||
const planLabels = plansWithDefaults.join('+') || 'none'
|
||||
logger.info(
|
||||
`[${jobType}] Dispatching: plans=[${planLabels}] + ${enterpriseCount} enterprise jobs (column: ${config.column})`
|
||||
)
|
||||
|
||||
if (enterpriseCount === 0) {
|
||||
return { jobIds, jobCount: jobIds.length, enterpriseCount: 0 }
|
||||
}
|
||||
|
||||
if (isTriggerAvailable()) {
|
||||
// Trigger.dev: use batchTrigger, chunked
|
||||
for (let i = 0; i < enterpriseRows.length; i += BATCH_TRIGGER_CHUNK_SIZE) {
|
||||
const chunk = enterpriseRows.slice(i, i + BATCH_TRIGGER_CHUNK_SIZE)
|
||||
const batchResult = await tasks.batchTrigger(
|
||||
jobType,
|
||||
chunk.map((row) => ({
|
||||
payload: { plan: 'enterprise' as const, workspaceId: row.id },
|
||||
options: {
|
||||
tags: [`workspaceId:${row.id}`, `jobType:${jobType}`],
|
||||
},
|
||||
}))
|
||||
)
|
||||
jobIds.push(batchResult.batchId)
|
||||
}
|
||||
} else {
|
||||
// Fallback: parallel enqueue via abstraction
|
||||
const results = await Promise.allSettled(
|
||||
enterpriseRows.map(async (row) => {
|
||||
const payload: CleanupJobPayload = { plan: 'enterprise', workspaceId: row.id }
|
||||
const jobId = await jobQueue.enqueue(jobType, payload)
|
||||
await runInlineIfNeeded(jobQueue, jobType, jobId, payload)
|
||||
return jobId
|
||||
})
|
||||
)
|
||||
|
||||
let succeeded = 0
|
||||
let failed = 0
|
||||
for (const result of results) {
|
||||
if (result.status === 'fulfilled') {
|
||||
jobIds.push(result.value)
|
||||
succeeded++
|
||||
} else {
|
||||
failed++
|
||||
logger.error(`[${jobType}] Failed to enqueue enterprise job:`, { reason: result.reason })
|
||||
}
|
||||
}
|
||||
logger.info(`[${jobType}] Enterprise enqueue: ${succeeded} succeeded, ${failed} failed`)
|
||||
}
|
||||
|
||||
return { jobIds, jobCount: jobIds.length, enterpriseCount }
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
import { db } from '@sim/db'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, inArray, isNotNull, lt, sql } from 'drizzle-orm'
|
||||
import type { PgColumn, PgTable } from 'drizzle-orm/pg-core'
|
||||
|
||||
const logger = createLogger('BatchDelete')
|
||||
|
||||
export const DEFAULT_BATCH_SIZE = 2000
|
||||
export const DEFAULT_MAX_BATCHES_PER_TABLE = 10
|
||||
|
||||
export interface TableCleanupResult {
|
||||
table: string
|
||||
deleted: number
|
||||
failed: number
|
||||
}
|
||||
|
||||
export interface BatchDeleteOptions {
|
||||
tableDef: PgTable
|
||||
workspaceIdCol: PgColumn
|
||||
timestampCol: PgColumn
|
||||
workspaceIds: string[]
|
||||
retentionDate: Date
|
||||
tableName: string
|
||||
/** When true, also requires `timestampCol IS NOT NULL` (soft-delete semantics). */
|
||||
requireTimestampNotNull?: boolean
|
||||
batchSize?: number
|
||||
maxBatches?: number
|
||||
}
|
||||
|
||||
/**
|
||||
* Iteratively delete rows in a table matching a workspace + time-based predicate.
|
||||
*
|
||||
* Uses a SELECT-with-LIMIT → DELETE-by-ID pattern to keep each round bounded in
|
||||
* memory and I/O (PostgreSQL DELETE does not support LIMIT directly).
|
||||
*/
|
||||
export async function batchDeleteByWorkspaceAndTimestamp({
|
||||
tableDef,
|
||||
workspaceIdCol,
|
||||
timestampCol,
|
||||
workspaceIds,
|
||||
retentionDate,
|
||||
tableName,
|
||||
requireTimestampNotNull = false,
|
||||
batchSize = DEFAULT_BATCH_SIZE,
|
||||
maxBatches = DEFAULT_MAX_BATCHES_PER_TABLE,
|
||||
}: BatchDeleteOptions): Promise<TableCleanupResult> {
|
||||
const result: TableCleanupResult = { table: tableName, deleted: 0, failed: 0 }
|
||||
|
||||
if (workspaceIds.length === 0) {
|
||||
logger.info(`[${tableName}] Skipped — no workspaces in scope`)
|
||||
return result
|
||||
}
|
||||
|
||||
const predicates = [inArray(workspaceIdCol, workspaceIds), lt(timestampCol, retentionDate)]
|
||||
if (requireTimestampNotNull) predicates.push(isNotNull(timestampCol))
|
||||
const whereClause = and(...predicates)
|
||||
|
||||
let batchesProcessed = 0
|
||||
let hasMore = true
|
||||
|
||||
while (hasMore && batchesProcessed < maxBatches) {
|
||||
try {
|
||||
const batch = await db
|
||||
.select({ id: sql<string>`id` })
|
||||
.from(tableDef)
|
||||
.where(whereClause)
|
||||
.limit(batchSize)
|
||||
|
||||
if (batch.length === 0) {
|
||||
logger.info(`[${tableName}] No expired rows found`)
|
||||
hasMore = false
|
||||
break
|
||||
}
|
||||
|
||||
const ids = batch.map((r) => r.id)
|
||||
const deleted = await db
|
||||
.delete(tableDef)
|
||||
.where(inArray(sql`id`, ids))
|
||||
.returning({ id: sql`id` })
|
||||
|
||||
result.deleted += deleted.length
|
||||
hasMore = batch.length === batchSize
|
||||
batchesProcessed++
|
||||
|
||||
logger.info(`[${tableName}] Batch ${batchesProcessed}: deleted ${deleted.length} rows`)
|
||||
} catch (error) {
|
||||
result.failed++
|
||||
logger.error(`[${tableName}] Batch delete failed:`, { error })
|
||||
hasMore = false
|
||||
}
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete rows by an explicit list of IDs. Use this when the IDs were selected
|
||||
* upstream (e.g., to drive external cleanup like S3 deletes or a backend API
|
||||
* call) so the DB delete cannot drift from the upstream selection. Paired with
|
||||
* `batchDeleteByWorkspaceAndTimestamp` for tables with no external side effects.
|
||||
*/
|
||||
export async function deleteRowsById(
|
||||
tableDef: PgTable,
|
||||
idCol: PgColumn,
|
||||
ids: string[],
|
||||
tableName: string
|
||||
): Promise<TableCleanupResult> {
|
||||
const result: TableCleanupResult = { table: tableName, deleted: 0, failed: 0 }
|
||||
if (ids.length === 0) return result
|
||||
try {
|
||||
const deleted = await db.delete(tableDef).where(inArray(idCol, ids)).returning({ id: idCol })
|
||||
result.deleted = deleted.length
|
||||
logger.info(`[${tableName}] Deleted ${deleted.length} rows`)
|
||||
} catch (error) {
|
||||
result.failed++
|
||||
logger.error(`[${tableName}] Delete failed:`, { error })
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -0,0 +1,208 @@
|
||||
import { db } from '@sim/db'
|
||||
import { copilotChats, workspaceFiles } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { and, inArray, isNull } from 'drizzle-orm'
|
||||
import { SIM_AGENT_API_URL } from '@/lib/copilot/constants'
|
||||
import { env } from '@/lib/core/config/env'
|
||||
import type { StorageContext } from '@/lib/uploads'
|
||||
import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
|
||||
|
||||
const logger = createLogger('ChatCleanup')
|
||||
|
||||
const COPILOT_CLEANUP_BATCH_SIZE = 1000
|
||||
|
||||
/**
|
||||
* Only storage in these contexts is tied to chat/task lifecycle. Workspace
|
||||
* files, execution logs, knowledge bases, profile pictures, etc. are owned by
|
||||
* other subsystems and must never be touched by chat cleanup — even if a row
|
||||
* somehow ends up with `chatId` set through a future flow.
|
||||
*/
|
||||
const CHAT_SCOPED_CONTEXTS = ['copilot', 'mothership'] as const satisfies readonly StorageContext[]
|
||||
type ChatScopedContext = (typeof CHAT_SCOPED_CONTEXTS)[number]
|
||||
|
||||
interface FileRef {
|
||||
key: string
|
||||
context: ChatScopedContext
|
||||
}
|
||||
|
||||
/**
|
||||
* Collect all file storage keys associated with the given chat IDs.
|
||||
* Two sources:
|
||||
* 1. workspaceFiles rows with chatId FK — filtered to chat-scoped contexts only
|
||||
* 2. fileAttachments[].key inside copilotChats.messages JSONB — all copilot uploads
|
||||
*/
|
||||
export async function collectChatFiles(chatIds: string[]): Promise<FileRef[]> {
|
||||
const files: FileRef[] = []
|
||||
if (chatIds.length === 0) return files
|
||||
|
||||
const seen = new Set<string>()
|
||||
|
||||
const [linkedFiles, chatsWithMessages] = await Promise.all([
|
||||
db
|
||||
.select({ key: workspaceFiles.key, context: workspaceFiles.context })
|
||||
.from(workspaceFiles)
|
||||
.where(
|
||||
and(
|
||||
inArray(workspaceFiles.chatId, chatIds),
|
||||
isNull(workspaceFiles.deletedAt),
|
||||
inArray(workspaceFiles.context, [...CHAT_SCOPED_CONTEXTS])
|
||||
)
|
||||
),
|
||||
db
|
||||
.select({ messages: copilotChats.messages })
|
||||
.from(copilotChats)
|
||||
.where(inArray(copilotChats.id, chatIds)),
|
||||
])
|
||||
|
||||
for (const f of linkedFiles) {
|
||||
if (!seen.has(f.key)) {
|
||||
seen.add(f.key)
|
||||
files.push({ key: f.key, context: f.context as ChatScopedContext })
|
||||
}
|
||||
}
|
||||
|
||||
for (const chat of chatsWithMessages) {
|
||||
const messages = chat.messages as unknown[]
|
||||
if (!Array.isArray(messages)) continue
|
||||
for (const msg of messages) {
|
||||
if (!msg || typeof msg !== 'object') continue
|
||||
const attachments = (msg as Record<string, unknown>).fileAttachments
|
||||
if (!Array.isArray(attachments)) continue
|
||||
for (const attachment of attachments) {
|
||||
if (
|
||||
attachment &&
|
||||
typeof attachment === 'object' &&
|
||||
(attachment as Record<string, unknown>).key
|
||||
) {
|
||||
const key = (attachment as Record<string, unknown>).key as string
|
||||
if (!seen.has(key)) {
|
||||
seen.add(key)
|
||||
files.push({ key, context: 'copilot' })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return files
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete files from cloud storage using the correct context/bucket per file.
|
||||
*/
|
||||
export async function deleteStorageFiles(
|
||||
files: FileRef[],
|
||||
label: string
|
||||
): Promise<{ filesDeleted: number; filesFailed: number }> {
|
||||
const stats = { filesDeleted: 0, filesFailed: 0 }
|
||||
if (files.length === 0 || !isUsingCloudStorage()) return stats
|
||||
|
||||
await Promise.all(
|
||||
files.map(async (file) => {
|
||||
try {
|
||||
await StorageService.deleteFile({ key: file.key, context: file.context })
|
||||
stats.filesDeleted++
|
||||
} catch (error) {
|
||||
stats.filesFailed++
|
||||
logger.error(`[${label}] Failed to delete storage file ${file.key}:`, { error })
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
/**
|
||||
* Call the copilot backend to delete chat data (memory_files, checkpoints, task_chains, etc.)
|
||||
* Chunked at 1000 per request.
|
||||
*/
|
||||
export async function cleanupCopilotBackend(
|
||||
chatIds: string[],
|
||||
label: string
|
||||
): Promise<{ deleted: number; failed: number }> {
|
||||
const stats = { deleted: 0, failed: 0 }
|
||||
|
||||
if (chatIds.length === 0 || !env.COPILOT_API_KEY) {
|
||||
if (!env.COPILOT_API_KEY) {
|
||||
logger.warn(`[${label}] COPILOT_API_KEY not set, skipping copilot backend cleanup`)
|
||||
}
|
||||
return stats
|
||||
}
|
||||
|
||||
for (let i = 0; i < chatIds.length; i += COPILOT_CLEANUP_BATCH_SIZE) {
|
||||
const chunk = chatIds.slice(i, i + COPILOT_CLEANUP_BATCH_SIZE)
|
||||
try {
|
||||
const response = await fetch(`${SIM_AGENT_API_URL}/api/tasks/cleanup`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'x-api-key': env.COPILOT_API_KEY,
|
||||
},
|
||||
body: JSON.stringify({ chatIds: chunk }),
|
||||
})
|
||||
|
||||
if (!response.ok) {
|
||||
const errorBody = await response.text().catch(() => '')
|
||||
logger.error(`[${label}] Copilot backend cleanup failed: ${response.status}`, {
|
||||
errorBody,
|
||||
chatCount: chunk.length,
|
||||
})
|
||||
stats.failed += chunk.length
|
||||
continue
|
||||
}
|
||||
|
||||
const result = await response.json()
|
||||
stats.deleted += result.deleted ?? 0
|
||||
logger.info(
|
||||
`[${label}] Copilot backend cleanup: ${result.deleted} chats deleted (batch ${Math.floor(i / COPILOT_CLEANUP_BATCH_SIZE) + 1})`
|
||||
)
|
||||
} catch (error) {
|
||||
stats.failed += chunk.length
|
||||
logger.error(`[${label}] Copilot backend cleanup request failed:`, { error })
|
||||
}
|
||||
}
|
||||
|
||||
return stats
|
||||
}
|
||||
|
||||
/**
|
||||
* Full chat cleanup: collect file refs, then (after DB deletion by caller)
|
||||
* call copilot backend and delete storage files.
|
||||
*
|
||||
* Usage:
|
||||
* const cleanup = await prepareChatCleanup(chatIds, label)
|
||||
* // ... delete DB rows ...
|
||||
* await cleanup.execute()
|
||||
*/
|
||||
export async function prepareChatCleanup(
|
||||
chatIds: string[],
|
||||
label: string
|
||||
): Promise<{ execute: () => Promise<void> }> {
|
||||
// Collect file refs BEFORE DB deletion (keys + context are lost after cascade)
|
||||
const files = await collectChatFiles(chatIds)
|
||||
if (files.length > 0) {
|
||||
logger.info(`[${label}] Collected ${files.length} files for cleanup`, {
|
||||
files: files.map((f) => ({ key: f.key, context: f.context })),
|
||||
})
|
||||
}
|
||||
|
||||
return {
|
||||
execute: async () => {
|
||||
// Call copilot backend
|
||||
if (chatIds.length > 0) {
|
||||
const copilotResult = await cleanupCopilotBackend(chatIds, label)
|
||||
logger.info(
|
||||
`[${label}] Copilot backend: ${copilotResult.deleted} deleted, ${copilotResult.failed} failed`
|
||||
)
|
||||
}
|
||||
|
||||
// Delete storage files with correct context per file
|
||||
if (files.length > 0) {
|
||||
const fileStats = await deleteStorageFiles(files, label)
|
||||
logger.info(
|
||||
`[${label}] Storage cleanup: ${fileStats.filesDeleted} deleted, ${fileStats.filesFailed} failed`
|
||||
)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -20,6 +20,9 @@ const JOB_TYPE_TO_TASK_ID: Record<JobType, string> = {
|
||||
'schedule-execution': 'schedule-execution',
|
||||
'webhook-execution': 'webhook-execution',
|
||||
'resume-execution': 'resume-execution',
|
||||
'cleanup-logs': 'cleanup-logs',
|
||||
'cleanup-soft-deletes': 'cleanup-soft-deletes',
|
||||
'cleanup-tasks': 'cleanup-tasks',
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -25,6 +25,9 @@ export type JobType =
|
||||
| 'schedule-execution'
|
||||
| 'webhook-execution'
|
||||
| 'resume-execution'
|
||||
| 'cleanup-logs'
|
||||
| 'cleanup-soft-deletes'
|
||||
| 'cleanup-tasks'
|
||||
|
||||
export type AsyncExecutionCorrelationSource = 'workflow' | 'schedule' | 'webhook'
|
||||
|
||||
|
||||
@@ -187,8 +187,6 @@ export const env = createEnv({
|
||||
AZURE_STORAGE_OG_IMAGES_CONTAINER_NAME: z.string().optional(), // Azure container for OpenGraph images
|
||||
AZURE_STORAGE_WORKSPACE_LOGOS_CONTAINER_NAME: z.string().optional(), // Azure container for workspace logos
|
||||
|
||||
// Data Retention
|
||||
FREE_PLAN_LOG_RETENTION_DAYS: z.string().optional(), // Log retention days for free plan users
|
||||
|
||||
// Admission & Burst Protection
|
||||
ADMISSION_GATE_MAX_INFLIGHT: z.string().optional().default('500'), // Max concurrent in-flight execution requests per pod
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
DROP INDEX "chat_archived_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "doc_archived_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "doc_deleted_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "form_archived_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "kc_archived_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "kc_deleted_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "mcp_servers_workspace_deleted_idx";--> statement-breakpoint
|
||||
DROP INDEX "webhook_archived_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "workflow_mcp_tool_archived_at_idx";--> statement-breakpoint
|
||||
DROP INDEX "workflow_schedule_archived_at_idx";--> statement-breakpoint
|
||||
ALTER TABLE "workspace" ADD COLUMN "log_retention_hours" integer;--> statement-breakpoint
|
||||
ALTER TABLE "workspace" ADD COLUMN "soft_delete_retention_hours" integer;--> statement-breakpoint
|
||||
ALTER TABLE "workspace" ADD COLUMN "task_cleanup_hours" integer;--> statement-breakpoint
|
||||
CREATE INDEX "a2a_agent_workspace_archived_partial_idx" ON "a2a_agent" USING btree ("workspace_id","archived_at") WHERE "a2a_agent"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "chat_archived_at_partial_idx" ON "chat" USING btree ("archived_at") WHERE "chat"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "doc_archived_at_partial_idx" ON "document" USING btree ("archived_at") WHERE "document"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "doc_deleted_at_partial_idx" ON "document" USING btree ("deleted_at") WHERE "document"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "form_archived_at_partial_idx" ON "form" USING btree ("archived_at") WHERE "form"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "kb_workspace_deleted_partial_idx" ON "knowledge_base" USING btree ("workspace_id","deleted_at") WHERE "knowledge_base"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "kc_archived_at_partial_idx" ON "knowledge_connector" USING btree ("archived_at") WHERE "knowledge_connector"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "kc_deleted_at_partial_idx" ON "knowledge_connector" USING btree ("deleted_at") WHERE "knowledge_connector"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "mcp_servers_workspace_deleted_partial_idx" ON "mcp_servers" USING btree ("workspace_id","deleted_at") WHERE "mcp_servers"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "memory_workspace_deleted_partial_idx" ON "memory" USING btree ("workspace_id","deleted_at") WHERE "memory"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "usage_log_workspace_created_at_idx" ON "usage_log" USING btree ("workspace_id","created_at");--> statement-breakpoint
|
||||
CREATE INDEX "user_table_def_workspace_archived_partial_idx" ON "user_table_definitions" USING btree ("workspace_id","archived_at") WHERE "user_table_definitions"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "webhook_archived_at_partial_idx" ON "webhook" USING btree ("archived_at") WHERE "webhook"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workflow_workspace_archived_partial_idx" ON "workflow" USING btree ("workspace_id","archived_at") WHERE "workflow"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workflow_folder_workspace_archived_partial_idx" ON "workflow_folder" USING btree ("workspace_id","archived_at") WHERE "workflow_folder"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workflow_mcp_server_workspace_deleted_partial_idx" ON "workflow_mcp_server" USING btree ("workspace_id","deleted_at") WHERE "workflow_mcp_server"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workflow_mcp_tool_archived_at_partial_idx" ON "workflow_mcp_tool" USING btree ("archived_at") WHERE "workflow_mcp_tool"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workflow_schedule_archived_at_partial_idx" ON "workflow_schedule" USING btree ("archived_at") WHERE "workflow_schedule"."archived_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workspace_file_workspace_deleted_partial_idx" ON "workspace_file" USING btree ("workspace_id","deleted_at") WHERE "workspace_file"."deleted_at" IS NOT NULL;--> statement-breakpoint
|
||||
CREATE INDEX "workspace_files_workspace_deleted_partial_idx" ON "workspace_files" USING btree ("workspace_id","deleted_at") WHERE "workspace_files"."deleted_at" IS NOT NULL;
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1345,6 +1345,13 @@
|
||||
"when": 1776538528385,
|
||||
"tag": "0192_invitation_unification",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 193,
|
||||
"version": "7",
|
||||
"when": 1776725749082,
|
||||
"tag": "0193_lying_rocket_racer",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
+65
-14
@@ -140,6 +140,9 @@ export const workflowFolder = pgTable(
|
||||
),
|
||||
parentSortIdx: index('workflow_folder_parent_sort_idx').on(table.parentId, table.sortOrder),
|
||||
archivedAtIdx: index('workflow_folder_archived_at_idx').on(table.archivedAt),
|
||||
workspaceArchivedAtPartialIdx: index('workflow_folder_workspace_archived_partial_idx')
|
||||
.on(table.workspaceId, table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -176,6 +179,9 @@ export const workflow = pgTable(
|
||||
.where(sql`${table.archivedAt} IS NULL`),
|
||||
folderSortIdx: index('workflow_folder_sort_idx').on(table.folderId, table.sortOrder),
|
||||
archivedAtIdx: index('workflow_archived_at_idx').on(table.archivedAt),
|
||||
workspaceArchivedAtPartialIdx: index('workflow_workspace_archived_partial_idx')
|
||||
.on(table.workspaceId, table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -543,7 +549,9 @@ export const workflowSchedule = pgTable(
|
||||
table.workflowId,
|
||||
table.deploymentVersionId
|
||||
),
|
||||
archivedAtIdx: index('workflow_schedule_archived_at_idx').on(table.archivedAt),
|
||||
archivedAtPartialIdx: index('workflow_schedule_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -620,7 +628,9 @@ export const webhook = pgTable(
|
||||
),
|
||||
// Optimize queries for credential set webhooks
|
||||
credentialSetIdIdx: index('webhook_credential_set_id_idx').on(table.credentialSetId),
|
||||
archivedAtIdx: index('webhook_archived_at_idx').on(table.archivedAt),
|
||||
archivedAtPartialIdx: index('webhook_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -899,7 +909,9 @@ export const chat = pgTable(
|
||||
identifierIdx: uniqueIndex('identifier_idx')
|
||||
.on(table.identifier)
|
||||
.where(sql`${table.archivedAt} IS NULL`),
|
||||
archivedAtIdx: index('chat_archived_at_idx').on(table.archivedAt),
|
||||
archivedAtPartialIdx: index('chat_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -941,7 +953,9 @@ export const form = pgTable(
|
||||
.where(sql`${table.archivedAt} IS NULL`),
|
||||
workflowIdIdx: index('form_workflow_id_idx').on(table.workflowId),
|
||||
userIdIdx: index('form_user_id_idx').on(table.userId),
|
||||
archivedAtIdx: index('form_archived_at_idx').on(table.archivedAt),
|
||||
archivedAtPartialIdx: index('form_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -1063,6 +1077,9 @@ export const workspace = pgTable(
|
||||
inboxEnabled: boolean('inbox_enabled').notNull().default(false),
|
||||
inboxAddress: text('inbox_address'),
|
||||
inboxProviderId: text('inbox_provider_id'),
|
||||
logRetentionHours: integer('log_retention_hours'),
|
||||
softDeleteRetentionHours: integer('soft_delete_retention_hours'),
|
||||
taskCleanupHours: integer('task_cleanup_hours'),
|
||||
archivedAt: timestamp('archived_at'),
|
||||
createdAt: timestamp('created_at').notNull().defaultNow(),
|
||||
updatedAt: timestamp('updated_at').notNull().defaultNow(),
|
||||
@@ -1095,6 +1112,9 @@ export const workspaceFile = pgTable(
|
||||
workspaceIdIdx: index('workspace_file_workspace_id_idx').on(table.workspaceId),
|
||||
keyIdx: index('workspace_file_key_idx').on(table.key),
|
||||
deletedAtIdx: index('workspace_file_deleted_at_idx').on(table.deletedAt),
|
||||
workspaceDeletedAtPartialIdx: index('workspace_file_workspace_deleted_partial_idx')
|
||||
.on(table.workspaceId, table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -1131,6 +1151,9 @@ export const workspaceFiles = pgTable(
|
||||
contextIdx: index('workspace_files_context_idx').on(table.context),
|
||||
chatIdIdx: index('workspace_files_chat_id_idx').on(table.chatId),
|
||||
deletedAtIdx: index('workspace_files_deleted_at_idx').on(table.deletedAt),
|
||||
workspaceDeletedAtPartialIdx: index('workspace_files_workspace_deleted_partial_idx')
|
||||
.on(table.workspaceId, table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -1226,6 +1249,9 @@ export const memory = pgTable(
|
||||
table.workspaceId,
|
||||
table.key
|
||||
),
|
||||
workspaceDeletedAtPartialIdx: index('memory_workspace_deleted_partial_idx')
|
||||
.on(table.workspaceId, table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -1268,6 +1294,9 @@ export const knowledgeBase = pgTable(
|
||||
userWorkspaceIdx: index('kb_user_workspace_idx').on(table.userId, table.workspaceId),
|
||||
// Index for soft delete filtering
|
||||
deletedAtIdx: index('kb_deleted_at_idx').on(table.deletedAt),
|
||||
workspaceDeletedAtPartialIdx: index('kb_workspace_deleted_partial_idx')
|
||||
.on(table.workspaceId, table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
/** One active (non-deleted) name per workspace; matches user_table_definitions pattern */
|
||||
workspaceNameActiveUnique: uniqueIndex('kb_workspace_name_active_unique')
|
||||
.on(table.workspaceId, table.name)
|
||||
@@ -1356,8 +1385,12 @@ export const document = pgTable(
|
||||
.where(sql`${table.deletedAt} IS NULL`),
|
||||
// Sync engine: load all active docs for a connector
|
||||
connectorIdIdx: index('doc_connector_id_idx').on(table.connectorId),
|
||||
archivedAtIdx: index('doc_archived_at_idx').on(table.archivedAt),
|
||||
deletedAtIdx: index('doc_deleted_at_idx').on(table.deletedAt),
|
||||
archivedAtPartialIdx: index('doc_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
deletedAtPartialIdx: index('doc_deleted_at_partial_idx')
|
||||
.on(table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
// Text tag indexes
|
||||
tag1Idx: index('doc_tag1_idx').on(table.tag1),
|
||||
tag2Idx: index('doc_tag2_idx').on(table.tag2),
|
||||
@@ -2077,11 +2110,10 @@ export const mcpServers = pgTable(
|
||||
table.enabled
|
||||
),
|
||||
|
||||
// Soft delete pattern - workspace + not deleted
|
||||
workspaceDeletedIdx: index('mcp_servers_workspace_deleted_idx').on(
|
||||
table.workspaceId,
|
||||
table.deletedAt
|
||||
),
|
||||
// Soft delete pattern - workspace + not deleted (partial: only deleted rows)
|
||||
workspaceDeletedIdx: index('mcp_servers_workspace_deleted_partial_idx')
|
||||
.on(table.workspaceId, table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -2136,6 +2168,9 @@ export const workflowMcpServer = pgTable(
|
||||
workspaceIdIdx: index('workflow_mcp_server_workspace_id_idx').on(table.workspaceId),
|
||||
createdByIdx: index('workflow_mcp_server_created_by_idx').on(table.createdBy),
|
||||
deletedAtIdx: index('workflow_mcp_server_deleted_at_idx').on(table.deletedAt),
|
||||
workspaceDeletedAtPartialIdx: index('workflow_mcp_server_workspace_deleted_partial_idx')
|
||||
.on(table.workspaceId, table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -2166,7 +2201,9 @@ export const workflowMcpTool = pgTable(
|
||||
serverWorkflowUnique: uniqueIndex('workflow_mcp_tool_server_workflow_unique')
|
||||
.on(table.serverId, table.workflowId)
|
||||
.where(sql`${table.archivedAt} IS NULL`),
|
||||
archivedAtIdx: index('workflow_mcp_tool_archived_at_idx').on(table.archivedAt),
|
||||
archivedAtPartialIdx: index('workflow_mcp_tool_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -2235,6 +2272,9 @@ export const a2aAgent = pgTable(
|
||||
.on(table.workspaceId, table.workflowId)
|
||||
.where(sql`${table.archivedAt} IS NULL`),
|
||||
archivedAtIdx: index('a2a_agent_archived_at_idx').on(table.archivedAt),
|
||||
workspaceArchivedAtPartialIdx: index('a2a_agent_workspace_archived_partial_idx')
|
||||
.on(table.workspaceId, table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -2386,6 +2426,10 @@ export const usageLog = pgTable(
|
||||
sourceIdx: index('usage_log_source_idx').on(table.source),
|
||||
workspaceIdIdx: index('usage_log_workspace_id_idx').on(table.workspaceId),
|
||||
workflowIdIdx: index('usage_log_workflow_id_idx').on(table.workflowId),
|
||||
workspaceCreatedAtIdx: index('usage_log_workspace_created_at_idx').on(
|
||||
table.workspaceId,
|
||||
table.createdAt
|
||||
),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -2708,8 +2752,12 @@ export const knowledgeConnector = pgTable(
|
||||
(table) => ({
|
||||
knowledgeBaseIdIdx: index('kc_knowledge_base_id_idx').on(table.knowledgeBaseId),
|
||||
statusNextSyncIdx: index('kc_status_next_sync_idx').on(table.status, table.nextSyncAt),
|
||||
archivedAtIdx: index('kc_archived_at_idx').on(table.archivedAt),
|
||||
deletedAtIdx: index('kc_deleted_at_idx').on(table.deletedAt),
|
||||
archivedAtPartialIdx: index('kc_archived_at_partial_idx')
|
||||
.on(table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
deletedAtPartialIdx: index('kc_deleted_at_partial_idx')
|
||||
.on(table.deletedAt)
|
||||
.where(sql`${table.deletedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
@@ -2777,6 +2825,9 @@ export const userTableDefinitions = pgTable(
|
||||
.on(table.workspaceId, table.name)
|
||||
.where(sql`${table.archivedAt} IS NULL`),
|
||||
archivedAtIdx: index('user_table_def_archived_at_idx').on(table.archivedAt),
|
||||
workspaceArchivedAtPartialIdx: index('user_table_def_workspace_archived_partial_idx')
|
||||
.on(table.workspaceId, table.archivedAt)
|
||||
.where(sql`${table.archivedAt} IS NOT NULL`),
|
||||
})
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user