fix(retention): switch data retention to be org-level (#4270)

* fix(retention): switch data retention to be org-level

* fix lint

* cleanup mothership ran logs

* fix cleanup dispatcher

* fix ui flash for data retention settings

* fix lint

* remove raw sql string interprolation

apps/docs/content/docs/en/enterprise/data-retention.mdx

@@ -6,7 +6,7 @@ description: Control how long execution logs, deleted resources, and copilot dat
 import { FAQ } from '@/components/ui/faq'
 import { Image } from '@/components/ui/image'
 
-Data Retention lets workspace admins on Enterprise plans configure how long three categories of data are kept before they are permanently deleted. Each workspace in your organization can have its own independent configuration.
+Data Retention lets organization owners and admins on Enterprise plans configure how long three categories of data are kept before they are permanently deleted. The configuration applies to every workspace in the organization.
 
 ---
 
@@ -58,9 +58,9 @@ Each setting is independent. You can configure a short log retention period alon
 
 ---
 
-## Per-workspace configuration
+## Organization-wide configuration
 
-Retention is configured at the **workspace level**, not organization-wide. Each workspace in your organization can have a different configuration. Changes to one workspace's settings do not affect other workspaces.
+Retention is configured at the **organization level**. A single configuration applies to every workspace in the organization — there are no per-workspace overrides.
 
 ---
 
@@ -73,7 +73,7 @@ By default, all three settings are unconfigured — no data is automatically del
 <FAQ items={[
   {
     question: "Who can configure data retention settings?",
-    answer: "Only workspace admins can configure data retention settings. On Sim Cloud, the workspace must be on an Enterprise plan."
+    answer: "Only organization owners and admins can configure data retention settings. On Sim Cloud, the organization must be on an Enterprise plan."
   },
   {
     question: "Is deletion immediate once the retention period expires?",
@@ -85,7 +85,7 @@ By default, all three settings are unconfigured — no data is automatically del
   },
   {
     question: "Does the retention period apply to all workspaces in my organization?",
-    answer: "No. Retention is configured per workspace. Each workspace in your organization can have a different configuration."
+    answer: "Yes. Retention is configured once per organization and applies to every workspace in the organization."
   },
   {
     question: "What happens if I shorten the retention period?",

apps/sim/app/api/organizations/[id]/data-retention/route.ts

@@ -0,0 +1,214 @@
+import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
+import { db } from '@sim/db'
+import { member, organization } from '@sim/db/schema'
+import { createLogger } from '@sim/logger'
+import { and, eq } from 'drizzle-orm'
+import { type NextRequest, NextResponse } from 'next/server'
+import { z } from 'zod'
+import { getSession } from '@/lib/auth'
+import {
+  CLEANUP_CONFIG,
+  type OrganizationRetentionSettings,
+} from '@/lib/billing/cleanup-dispatcher'
+import { isOrganizationOnEnterprisePlan } from '@/lib/billing/core/subscription'
+import { isBillingEnabled } from '@/lib/core/config/feature-flags'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+
+const logger = createLogger('DataRetentionAPI')
+
+const MIN_HOURS = 24
+const MAX_HOURS = 43800
+
+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(),
+})
+
+function enterpriseDefaults(): OrganizationRetentionSettings {
+  return {
+    logRetentionHours: CLEANUP_CONFIG['cleanup-logs'].defaults.enterprise,
+    softDeleteRetentionHours: CLEANUP_CONFIG['cleanup-soft-deletes'].defaults.enterprise,
+    taskCleanupHours: CLEANUP_CONFIG['cleanup-tasks'].defaults.enterprise,
+  }
+}
+
+function normalizeConfigured(
+  settings: Partial<OrganizationRetentionSettings> | null | undefined
+): OrganizationRetentionSettings {
+  return {
+    logRetentionHours: settings?.logRetentionHours ?? null,
+    softDeleteRetentionHours: settings?.softDeleteRetentionHours ?? null,
+    taskCleanupHours: settings?.taskCleanupHours ?? null,
+  }
+}
+
+/**
+ * GET /api/organizations/[id]/data-retention
+ * Returns the organization's data retention settings.
+ * Accessible by any member of the organization.
+ */
+export const GET = withRouteHandler(
+  async (_request: NextRequest, { params }: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const { id: organizationId } = await params
+
+    const [memberEntry] = await db
+      .select({ id: member.id })
+      .from(member)
+      .where(and(eq(member.organizationId, organizationId), eq(member.userId, session.user.id)))
+      .limit(1)
+
+    if (!memberEntry) {
+      return NextResponse.json(
+        { error: 'Forbidden - Not a member of this organization' },
+        { status: 403 }
+      )
+    }
+
+    const [org] = await db
+      .select({ dataRetentionSettings: organization.dataRetentionSettings })
+      .from(organization)
+      .where(eq(organization.id, organizationId))
+      .limit(1)
+
+    if (!org) {
+      return NextResponse.json({ error: 'Organization not found' }, { status: 404 })
+    }
+
+    const isEnterprise = !isBillingEnabled || (await isOrganizationOnEnterprisePlan(organizationId))
+    const configured = normalizeConfigured(org.dataRetentionSettings)
+    const defaults = enterpriseDefaults()
+
+    return NextResponse.json({
+      success: true,
+      data: {
+        isEnterprise,
+        defaults,
+        configured,
+        effective: isEnterprise ? configured : defaults,
+      },
+    })
+  }
+)
+
+/**
+ * PUT /api/organizations/[id]/data-retention
+ * Updates the organization's data retention settings.
+ * Requires enterprise plan and owner/admin role.
+ */
+export const PUT = withRouteHandler(
+  async (request: NextRequest, { params }: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const { id: organizationId } = await params
+
+    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 [memberEntry] = await db
+      .select({ role: member.role })
+      .from(member)
+      .where(and(eq(member.organizationId, organizationId), eq(member.userId, session.user.id)))
+      .limit(1)
+
+    if (!memberEntry) {
+      return NextResponse.json(
+        { error: 'Forbidden - Not a member of this organization' },
+        { status: 403 }
+      )
+    }
+
+    if (memberEntry.role !== 'owner' && memberEntry.role !== 'admin') {
+      return NextResponse.json(
+        { error: 'Forbidden - Only organization owners and admins can update data retention' },
+        { status: 403 }
+      )
+    }
+
+    if (isBillingEnabled) {
+      const hasEnterprise = await isOrganizationOnEnterprisePlan(organizationId)
+      if (!hasEnterprise) {
+        return NextResponse.json(
+          { error: 'Data Retention is available on Enterprise plans only' },
+          { status: 403 }
+        )
+      }
+    }
+
+    const [currentOrg] = await db
+      .select({
+        name: organization.name,
+        dataRetentionSettings: organization.dataRetentionSettings,
+      })
+      .from(organization)
+      .where(eq(organization.id, organizationId))
+      .limit(1)
+
+    if (!currentOrg) {
+      return NextResponse.json({ error: 'Organization not found' }, { status: 404 })
+    }
+
+    const current = normalizeConfigured(currentOrg.dataRetentionSettings)
+    const merged: OrganizationRetentionSettings = { ...current }
+    if (parsed.data.logRetentionHours !== undefined) {
+      merged.logRetentionHours = parsed.data.logRetentionHours
+    }
+    if (parsed.data.softDeleteRetentionHours !== undefined) {
+      merged.softDeleteRetentionHours = parsed.data.softDeleteRetentionHours
+    }
+    if (parsed.data.taskCleanupHours !== undefined) {
+      merged.taskCleanupHours = parsed.data.taskCleanupHours
+    }
+
+    const [updated] = await db
+      .update(organization)
+      .set({ dataRetentionSettings: merged, updatedAt: new Date() })
+      .where(eq(organization.id, organizationId))
+      .returning({ dataRetentionSettings: organization.dataRetentionSettings })
+
+    if (!updated) {
+      return NextResponse.json({ error: 'Organization not found' }, { status: 404 })
+    }
+
+    recordAudit({
+      workspaceId: null,
+      actorId: session.user.id,
+      action: AuditAction.ORGANIZATION_UPDATED,
+      resourceType: AuditResourceType.ORGANIZATION,
+      resourceId: organizationId,
+      actorName: session.user.name ?? undefined,
+      actorEmail: session.user.email ?? undefined,
+      resourceName: currentOrg.name,
+      description: 'Updated data retention settings',
+      metadata: { changes: parsed.data },
+      request,
+    })
+
+    const configured = normalizeConfigured(updated.dataRetentionSettings)
+    const defaults = enterpriseDefaults()
+
+    return NextResponse.json({
+      success: true,
+      data: {
+        isEnterprise: true,
+        defaults,
+        configured,
+        effective: configured,
+      },
+    })
+  }
+)

apps/sim/app/api/workspaces/[id]/data-retention/route.ts

@@ -1,226 +0,0 @@
-import { AuditAction, AuditResourceType, recordAudit } from '@sim/audit'
-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 { 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 { isBillingEnabled } from '@/lib/core/config/feature-flags'
-import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
-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 const GET = withRouteHandler(
-  async (_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 = !isBillingEnabled || 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 const PUT = withRouteHandler(
-  async (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 })
-      }
-
-      if (isBillingEnabled) {
-        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 })
-    }
-  }
-)

apps/sim/background/cleanup-logs.ts

@@ -1,5 +1,5 @@
 import { db } from '@sim/db'
-import { workflowExecutionLogs } from '@sim/db/schema'
+import { jobExecutionLogs, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { task } from '@trigger.dev/sdk'
 import { and, inArray, lt } from 'drizzle-orm'
@@ -112,6 +112,63 @@ async function cleanupTier(
   return results
 }
 
+interface JobLogCleanupResults {
+  deleted: number
+  deleteFailed: number
+}
+
+async function cleanupJobExecutionLogsTier(
+  workspaceIds: string[],
+  retentionDate: Date,
+  label: string
+): Promise<JobLogCleanupResults> {
+  const results: JobLogCleanupResults = { deleted: 0, deleteFailed: 0 }
+  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: jobExecutionLogs.id })
+      .from(jobExecutionLogs)
+      .where(
+        and(
+          inArray(jobExecutionLogs.workspaceId, workspaceIds),
+          lt(jobExecutionLogs.startedAt, retentionDate)
+        )
+      )
+      .limit(BATCH_SIZE)
+
+    if (batch.length === 0) {
+      hasMore = false
+      break
+    }
+
+    const logIds = batch.map((log) => log.id)
+    try {
+      const deleted = await db
+        .delete(jobExecutionLogs)
+        .where(inArray(jobExecutionLogs.id, logIds))
+        .returning({ id: jobExecutionLogs.id })
+
+      results.deleted += deleted.length
+    } catch (deleteError) {
+      results.deleteFailed += logIds.length
+      logger.error(`Batch delete failed for ${label} (job_execution_logs):`, { deleteError })
+    }
+
+    batchesProcessed++
+    hasMore = batch.length === BATCH_SIZE
+
+    logger.info(
+      `[${label}] job_execution_logs batch ${batchesProcessed}: ${batch.length} rows processed`
+    )
+  }
+
+  return results
+}
+
 export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void> {
   const startTime = Date.now()
 
@@ -135,7 +192,12 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
 
   const results = await cleanupTier(workspaceIds, retentionDate, label)
   logger.info(
-    `[${label}] Result: ${results.deleted} deleted, ${results.deleteFailed} failed out of ${results.total} candidates`
+    `[${label}] workflow_execution_logs: ${results.deleted} deleted, ${results.deleteFailed} failed out of ${results.total} candidates`
+  )
+
+  const jobLogResults = await cleanupJobExecutionLogsTier(workspaceIds, retentionDate, label)
+  logger.info(
+    `[${label}] job_execution_logs: ${jobLogResults.deleted} deleted, ${jobLogResults.deleteFailed} failed`
   )
 
   // Snapshot cleanup runs only on the free job to avoid running it N times for N enterprise workspaces.

apps/sim/ee/data-retention/components/data-retention-settings.tsx

@@ -3,16 +3,17 @@
 import { useEffect, useState } from 'react'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { useParams } from 'next/navigation'
 import { Button, Combobox, toast } from '@/components/emcn'
+import { useSession } from '@/lib/auth/auth-client'
 import { isBillingEnabled } from '@/lib/core/config/feature-flags'
-import { useUserPermissionsContext } from '@/app/workspace/[workspaceId]/providers/workspace-permissions-provider'
+import { getUserRole } from '@/lib/workspaces/organization/utils'
 import { SettingRow } from '@/ee/components/setting-row'
 import { DataRetentionSkeleton } from '@/ee/data-retention/components/data-retention-skeleton'
 import {
-  useUpdateWorkspaceRetention,
-  useWorkspaceRetention,
+  useOrganizationRetention,
+  useUpdateOrganizationRetention,
 } from '@/ee/data-retention/hooks/data-retention'
+import { useOrganizations } from '@/hooks/queries/organization'
 
 const logger = createLogger('DataRetentionSettings')
 
@@ -68,12 +69,18 @@ function RetentionSelect({ value, onChange }: RetentionSelectProps) {
 }
 
 export function DataRetentionSettings() {
-  const params = useParams<{ workspaceId: string }>()
-  const workspaceId = params.workspaceId
+  const { data: session, isPending: sessionPending } = useSession()
+  const { data: orgsData, isLoading: orgsLoading } = useOrganizations()
 
-  const { data, isLoading } = useWorkspaceRetention(workspaceId)
-  const { canAdmin } = useUserPermissionsContext()
-  const updateMutation = useUpdateWorkspaceRetention()
+  const activeOrganization = orgsData?.activeOrganization
+  const orgId = activeOrganization?.id
+
+  const { data, isLoading: retentionLoading } = useOrganizationRetention(orgId)
+  const updateMutation = useUpdateOrganizationRetention()
+
+  const userEmail = session?.user?.email
+  const userRole = getUserRole(activeOrganization, userEmail)
+  const canManage = userRole === 'owner' || userRole === 'admin'
 
   const [logDays, setLogDays] = useState('')
   const [softDeleteDays, setSoftDeleteDays] = useState('')
@@ -103,9 +110,10 @@ export function DataRetentionSettings() {
     taskCleanupDays !== savedTaskCleanupDays
 
   async function handleSave() {
+    if (!orgId) return
     try {
       await updateMutation.mutateAsync({
-        workspaceId,
+        orgId,
         settings: {
           logRetentionHours: daysToHours(logDays),
           softDeleteRetentionHours: daysToHours(softDeleteDays),
@@ -123,7 +131,17 @@ export function DataRetentionSettings() {
     }
   }
 
-  if (isLoading) return <DataRetentionSkeleton />
+  if (sessionPending || orgsLoading || (orgId && retentionLoading)) {
+    return <DataRetentionSkeleton />
+  }
+
+  if (!orgId) {
+    return (
+      <div className='flex h-full items-center justify-center text-[var(--text-muted)] text-sm'>
+        Data retention is configured per organization. Join or create an organization to continue.
+      </div>
+    )
+  }
 
   if (!data) {
     return (
@@ -141,10 +159,10 @@ export function DataRetentionSettings() {
     )
   }
 
-  if (!canAdmin) {
+  if (!canManage) {
     return (
       <div className='flex h-full items-center justify-center text-[var(--text-muted)] text-sm'>
-        Only workspace admins can configure data retention settings.
+        Only organization owners and admins can configure data retention settings.
       </div>
     )
   }

apps/sim/ee/data-retention/hooks/data-retention.ts

@@ -1,7 +1,6 @@
 'use client'
 
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'
-import type { PlanCategory } from '@/lib/billing/plan-helpers'
 
 export interface RetentionValues {
   logRetentionHours: number | null
@@ -10,7 +9,6 @@ export interface RetentionValues {
 }
 
 export interface DataRetentionResponse {
-  plan: PlanCategory
   isEnterprise: boolean
   defaults: RetentionValues
   configured: RetentionValues
@@ -19,14 +17,14 @@ export interface DataRetentionResponse {
 
 export const dataRetentionKeys = {
   all: ['dataRetention'] as const,
-  settings: (workspaceId: string) => [...dataRetentionKeys.all, 'settings', workspaceId] as const,
+  settings: (orgId: string) => [...dataRetentionKeys.all, 'settings', orgId] as const,
 }
 
 async function fetchDataRetention(
-  workspaceId: string,
+  orgId: string,
   signal?: AbortSignal
 ): Promise<DataRetentionResponse> {
-  const response = await fetch(`/api/workspaces/${workspaceId}/data-retention`, { signal })
+  const response = await fetch(`/api/organizations/${orgId}/data-retention`, { signal })
 
   if (!response.ok) {
     const error = await response.json().catch(() => ({}))
@@ -37,26 +35,26 @@ async function fetchDataRetention(
   return data as DataRetentionResponse
 }
 
-export function useWorkspaceRetention(workspaceId: string | undefined) {
+export function useOrganizationRetention(orgId: string | undefined) {
   return useQuery({
-    queryKey: dataRetentionKeys.settings(workspaceId ?? ''),
-    queryFn: ({ signal }) => fetchDataRetention(workspaceId as string, signal),
-    enabled: Boolean(workspaceId),
+    queryKey: dataRetentionKeys.settings(orgId ?? ''),
+    queryFn: ({ signal }) => fetchDataRetention(orgId as string, signal),
+    enabled: Boolean(orgId),
     staleTime: 60 * 1000,
   })
 }
 
 interface UpdateRetentionVariables {
-  workspaceId: string
+  orgId: string
   settings: Partial<RetentionValues>
 }
 
-export function useUpdateWorkspaceRetention() {
+export function useUpdateOrganizationRetention() {
   const queryClient = useQueryClient()
 
   return useMutation({
-    mutationFn: async ({ workspaceId, settings }: UpdateRetentionVariables) => {
-      const response = await fetch(`/api/workspaces/${workspaceId}/data-retention`, {
+    mutationFn: async ({ orgId, settings }: UpdateRetentionVariables) => {
+      const response = await fetch(`/api/organizations/${orgId}/data-retention`, {
         method: 'PUT',
         headers: { 'Content-Type': 'application/json' },
         body: JSON.stringify(settings),
@@ -70,8 +68,8 @@ export function useUpdateWorkspaceRetention() {
       const { data } = await response.json()
       return data as DataRetentionResponse
     },
-    onSettled: (_data, _error, { workspaceId }) => {
-      queryClient.invalidateQueries({ queryKey: dataRetentionKeys.settings(workspaceId) })
+    onSettled: (_data, _error, { orgId }) => {
+      queryClient.invalidateQueries({ queryKey: dataRetentionKeys.settings(orgId) })
     },
   })
 }

apps/sim/lib/billing/cleanup-dispatcher.ts

@@ -1,9 +1,9 @@
 import { db } from '@sim/db'
-import { subscription, workspace } from '@sim/db/schema'
+import { organization, 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 { and, eq, inArray, isNotNull, isNull, sql } 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'
@@ -16,11 +16,15 @@ const BATCH_TRIGGER_CHUNK_SIZE = 1000
 
 export type CleanupJobType = 'cleanup-logs' | 'cleanup-soft-deletes' | 'cleanup-tasks'
 
-export type WorkspaceRetentionColumn =
+export type OrganizationRetentionKey =
   | 'logRetentionHours'
   | 'softDeleteRetentionHours'
   | 'taskCleanupHours'
 
+export type OrganizationRetentionSettings = {
+  [K in OrganizationRetentionKey]: number | null
+}
+
 export type NonEnterprisePlan = Exclude<PlanCategory, 'enterprise'>
 
 const NON_ENTERPRISE_PLANS = ['free', 'pro', 'team'] as const satisfies readonly NonEnterprisePlan[]
@@ -30,35 +34,36 @@ export type CleanupJobPayload =
   | { plan: 'enterprise'; workspaceId: string }
 
 interface CleanupJobConfig {
-  column: WorkspaceRetentionColumn
+  key: OrganizationRetentionKey
   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.
+ * Single source of truth for cleanup retention: which key each job type reads
+ * from `organization.dataRetentionSettings`, and the default retention (in
+ * hours) per plan. Enterprise is always `null` here — enterprise orgs must
+ * set their own value.
  */
 export const CLEANUP_CONFIG = {
   'cleanup-logs': {
-    column: 'logRetentionHours',
+    key: 'logRetentionHours',
     defaults: { free: 30 * DAY, pro: null, team: null, enterprise: null },
   },
   'cleanup-soft-deletes': {
-    column: 'softDeleteRetentionHours',
+    key: 'softDeleteRetentionHours',
     defaults: { free: 30 * DAY, pro: 90 * DAY, team: 90 * DAY, enterprise: null },
   },
   'cleanup-tasks': {
-    column: 'taskCleanupHours',
+    key: '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.
+ * per-workspace (routed through the owning organization's retention config).
  */
 export async function resolveWorkspaceIdsForPlan(plan: NonEnterprisePlan): Promise<string[]> {
   if (plan === 'free') {
@@ -105,8 +110,8 @@ export interface ResolvedCleanupScope {
 /**
  * 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).
+ * has no retention configured (default is null, or the enterprise org has not
+ * set this key).
  */
 export async function resolveCleanupScope(
   jobType: CleanupJobType,
@@ -121,17 +126,19 @@ export async function resolveCleanupScope(
     return { workspaceIds, retentionHours, label: payload.plan }
   }
 
-  const [ws] = await db
-    .select({ hours: workspace[config.column] })
+  const [row] = await db
+    .select({ settings: organization.dataRetentionSettings })
     .from(workspace)
+    .innerJoin(organization, eq(organization.id, workspace.organizationId))
     .where(eq(workspace.id, payload.workspaceId))
     .limit(1)
 
-  if (ws?.hours == null) return null
+  const hours = row?.settings?.[config.key]
+  if (hours == null) return null
 
   return {
     workspaceIds: [payload.workspaceId],
-    retentionHours: ws.hours,
+    retentionHours: hours,
     label: `enterprise/${payload.workspaceId}`,
   }
 }
@@ -189,7 +196,8 @@ async function runInlineIfNeeded(
  * 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
+ * - One enterprise job per workspace whose owning organization has a non-null
+ *   retention value for this job's key
  *
  * Uses Trigger.dev batchTrigger when available, otherwise parallel enqueue via
  * the JobQueueBackend abstraction. On the database backend (no external worker),
@@ -211,30 +219,34 @@ export async function dispatchCleanupJobs(
     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]
+  // Enterprise: workspaces whose owning org is on an active enterprise sub and
+  // has a non-NULL value for this job's retention key. groupBy dedupes in case
+  // multiple entitled subscription rows exist for the same org.
   const enterpriseRows = await db
     .select({ id: workspace.id })
     .from(workspace)
+    .innerJoin(organization, eq(organization.id, workspace.organizationId))
     .innerJoin(
       subscription,
       and(
-        eq(subscription.referenceId, workspace.billedAccountUserId),
+        eq(subscription.referenceId, organization.id),
         inArray(subscription.status, ENTITLED_SUBSCRIPTION_STATUSES),
         eq(subscription.plan, 'enterprise')
       )
     )
-    .where(and(isNull(workspace.archivedAt), isNotNull(retentionCol)))
+    .where(
+      and(
+        isNull(workspace.archivedAt),
+        isNotNull(sql`${organization.dataRetentionSettings}->>${config.key}`)
+      )
+    )
     .groupBy(workspace.id)
 
   const enterpriseCount = enterpriseRows.length
 
   const planLabels = plansWithDefaults.join('+') || 'none'
   logger.info(
-    `[${jobType}] Dispatching: plans=[${planLabels}] + ${enterpriseCount} enterprise jobs (column: ${config.column})`
+    `[${jobType}] Dispatching: plans=[${planLabels}] + ${enterpriseCount} enterprise jobs (key: ${config.key})`
   )
 
   if (enterpriseCount === 0) {

packages/db/migrations/0196_retention_org_level.sql

@@ -0,0 +1,4 @@
+ALTER TABLE "organization" ADD COLUMN "data_retention_settings" json;--> statement-breakpoint
+ALTER TABLE "workspace" DROP COLUMN "log_retention_hours";--> statement-breakpoint
+ALTER TABLE "workspace" DROP COLUMN "soft_delete_retention_hours";--> statement-breakpoint
+ALTER TABLE "workspace" DROP COLUMN "task_cleanup_hours";

packages/db/migrations/meta/_journal.json

@@ -1366,6 +1366,13 @@
       "when": 1776883116756,
       "tag": "0195_normal_white_queen",
       "breakpoints": true
+    },
+    {
+      "idx": 196,
+      "version": "7",
+      "when": 1776916912442,
+      "tag": "0196_retention_org_level",
+      "breakpoints": true
     }
   ]
 }

packages/db/schema.ts

@@ -980,6 +980,11 @@ export const organization = pgTable('organization', {
     privacyUrl?: string
     hidePoweredBySim?: boolean
   }>(),
+  dataRetentionSettings: json('data_retention_settings').$type<{
+    logRetentionHours?: number | null
+    softDeleteRetentionHours?: number | null
+    taskCleanupHours?: number | null
+  }>(),
   orgUsageLimit: decimal('org_usage_limit'),
   storageUsedBytes: bigint('storage_used_bytes', { mode: 'number' }).notNull().default(0),
   departedMemberUsage: decimal('departed_member_usage').notNull().default('0'),
@@ -1079,9 +1084,6 @@ 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(),