improvement(outbox): load handler modules only for the event types a run will process (#8731)

* improvement(outbox): load handler modules only for the event types a run will process

* refactor(outbox): take only lazy handler groups; resolve eligible types in one place

* fix(outbox): fail the run after maintenance when a handler module could not load

* improvement(outbox): import every registry event type from a light module; prove failed loads in Postgres

* improvement(outbox): source enterprise and document-processing event types from dependency-free modules

Author: waleedlatif1

apps/sim/app/api/cron/reconcile-billing-seats/route.ts

@@ -3,7 +3,7 @@ import { toError } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { reconcileTeamSeatDrift } from '@/lib/billing/organizations/seat-drift'
-import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers'
+import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import { findDeadLetteredEvents } from '@/lib/core/outbox/service'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'

apps/sim/app/api/v1/admin/outbox/[id]/requeue/route.ts

@@ -6,11 +6,11 @@ import { and, eq, sql } from 'drizzle-orm'
 import { NextResponse } from 'next/server'
 import { adminV1RequeueOutboxEventContract } from '@/lib/api/contracts/v1/admin'
 import { getValidationErrorMessage, parseRequest } from '@/lib/api/server'
+import { enterpriseMetadataSyncPayloadSchema } from '@/lib/billing/enterprise-outbox'
 import {
   ENTERPRISE_METADATA_SYNC_EVENT_TYPE,
   ENTERPRISE_PROVISION_EVENT_TYPE,
-  enterpriseMetadataSyncPayloadSchema,
-} from '@/lib/billing/enterprise-outbox'
+} from '@/lib/billing/enterprise-outbox-events'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'
 

apps/sim/app/api/v1/admin/subscriptions/[id]/route.ts

@@ -33,7 +33,7 @@ import {
 } from '@/lib/api/contracts/v1/admin'
 import { parseRequest } from '@/lib/api/server'
 import { requireStripeClient } from '@/lib/billing/stripe-client'
-import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers'
+import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { withAdminAuthParams } from '@/app/api/v1/admin/middleware'

apps/sim/ee/access-requests/lib/notification-events.ts

@@ -1,2 +1,3 @@
 export const PERMISSION_ACCESS_REQUEST_CREATED_EVENT = 'permission-access-request.created'
 export const PERMISSION_ACCESS_REQUEST_DECIDED_EVENT = 'permission-access-request.decided'
+export const PERMISSION_ACCESS_REQUEST_NOTIFY_ADMIN_EVENT = 'permission-access-request.notify-admin'

apps/sim/ee/access-requests/lib/notifications.ts

@@ -18,12 +18,12 @@ import { getMyAccessRequestHref } from '@/ee/access-requests/lib/navigation'
 import {
   PERMISSION_ACCESS_REQUEST_CREATED_EVENT,
   PERMISSION_ACCESS_REQUEST_DECIDED_EVENT,
+  PERMISSION_ACCESS_REQUEST_NOTIFY_ADMIN_EVENT,
 } from '@/ee/access-requests/lib/notification-events'
 import { isAccessRequestEnabled } from '@/ee/access-requests/lib/settings'
 
 const logger = createLogger('PermissionAccessRequestNotifications')
 const ADMIN_RECIPIENT_PAGE_SIZE = 50
-const ADMIN_NOTIFICATION_EVENT = 'permission-access-request.notify-admin'
 const notificationPayloadSchema = z.object({ requestId: z.string().min(1).max(256) }).strict()
 const createdPayloadSchema = notificationPayloadSchema.extend({
   afterMemberId: z.string().min(1).max(256).optional(),
@@ -176,8 +176,8 @@ export const permissionAccessRequestOutboxHandlers = {
       .insert(outboxEvent)
       .values(
         recipients.map((recipient) => ({
-          id: `${ADMIN_NOTIFICATION_EVENT}:${requestId}:${recipient.userId}`,
-          eventType: ADMIN_NOTIFICATION_EVENT,
+          id: `${PERMISSION_ACCESS_REQUEST_NOTIFY_ADMIN_EVENT}:${requestId}:${recipient.userId}`,
+          eventType: PERMISSION_ACCESS_REQUEST_NOTIFY_ADMIN_EVENT,
           payload: { requestId, recipientUserId: recipient.userId },
         }))
       )
@@ -187,7 +187,7 @@ export const permissionAccessRequestOutboxHandlers = {
       return continueOutboxHandler('Continue access request administrator notifications')
     }
   },
-  [ADMIN_NOTIFICATION_EVENT]: async (rawPayload, context) => {
+  [PERMISSION_ACCESS_REQUEST_NOTIFY_ADMIN_EVENT]: async (rawPayload, context) => {
     const { requestId, recipientUserId } = adminPayloadSchema.parse(rawPayload)
     const request = await loadRequest(requestId)
     if (

apps/sim/ee/workspace-forking/application/admit-sync.ts

@@ -2,6 +2,11 @@ import { generateId } from '@sim/utils/id'
 import { truncate } from '@sim/utils/string'
 import { enqueueOutboxEvent } from '@/lib/core/outbox/service'
 import type { DbOrTx } from '@/lib/db/types'
+import {
+  WORKSPACE_MCP_CHANGED_EVENT,
+  WORKSPACE_OPERATION_OBSERVE_EVENT,
+  WORKSPACE_WORKFLOWS_CHANGED_EVENT,
+} from '@/lib/workspaces/operations/outbox-events'
 import {
   insertWorkspaceOperationReceipt,
   type WorkspaceOperationReport,
@@ -100,7 +105,7 @@ export async function admitForkSync(
   }
   if (params.mcpAttachmentServerIds.length)
     report.effectEventIds!.push(
-      await enqueueOutboxEvent(tx, 'workspace.mcp.changed', {
+      await enqueueOutboxEvent(tx, WORKSPACE_MCP_CHANGED_EVENT, {
         serverIds: params.mcpAttachmentServerIds,
       })
     )
@@ -117,10 +122,10 @@ export async function admitForkSync(
           ? 'completed_with_warnings'
           : 'completed'
   }
-  await enqueueOutboxEvent(tx, 'workspace.workflows.changed', {
+  await enqueueOutboxEvent(tx, WORKSPACE_WORKFLOWS_CHANGED_EVENT, {
     workspaceId: params.targetWorkspaceId,
   })
-  await enqueueOutboxEvent(tx, 'workspace.operation.observe', {
+  await enqueueOutboxEvent(tx, WORKSPACE_OPERATION_OBSERVE_EVENT, {
     workspaceId: report.workspaceId,
     operationId: report.operationId,
   })

apps/sim/ee/workspace-forking/application/content-outbox-event.ts

@@ -0,0 +1 @@
+export const FORK_CONTENT_COPY_EVENT = 'workspace.fork.content.copy'

apps/sim/ee/workspace-forking/application/content-outbox.ts

@@ -11,6 +11,7 @@ import {
 } from '@/lib/core/outbox/service'
 import type { DbOrTx } from '@/lib/db/types'
 import type { WorkspaceOperationReport } from '@/lib/workspaces/operations/receipts'
+import { FORK_CONTENT_COPY_EVENT } from '@/ee/workspace-forking/application/content-outbox-event'
 import {
   type ForkContentCopyPayload,
   runForkContentCopy,
@@ -153,11 +154,11 @@ export async function enqueueDurableForkContent(
   })
   if (Buffer.byteLength(JSON.stringify(payload)) > 8 * 1024 * 1024)
     throw new OrchestrationError('payload_too_large', 'Fork background work exceeds 8 MiB')
-  return enqueueOutboxEvent(tx, 'workspace.fork.content.copy', payload)
+  return enqueueOutboxEvent(tx, FORK_CONTENT_COPY_EVENT, payload)
 }
 
 export const forkContentOutboxHandlers = {
-  'workspace.fork.content.copy': withOutboxHandlerTimeout(async (raw, context) => {
+  [FORK_CONTENT_COPY_EVENT]: withOutboxHandlerTimeout(async (raw, context) => {
     const payload = contentPayloadSchema.parse(raw)
     const [receipt] = await db
       .select({ report: workspaceOperationReceipt.report })

apps/sim/ee/workspace-forking/lib/create-fork.ts

@@ -16,6 +16,7 @@ import {
   collectReferencedFileFolderPaths,
 } from '@/lib/workflows/references/reference-scan'
 import type { ForkRemapKind } from '@/lib/workflows/references/remap-references'
+import { WORKSPACE_OPERATION_OBSERVE_EVENT } from '@/lib/workspaces/operations/outbox-events'
 import {
   findWorkspaceOperationReceipt,
   insertWorkspaceOperationReceipt,
@@ -625,7 +626,7 @@ export async function createFork(params: CreateForkParams): Promise<CreateForkRe
           requestId: admission.requestId,
         })
       }
-      await enqueueOutboxEvent(tx, 'workspace.operation.observe', {
+      await enqueueOutboxEvent(tx, WORKSPACE_OPERATION_OBSERVE_EVENT, {
         workspaceId: report.workspaceId,
         operationId: report.operationId,
       })

apps/sim/lib/admin/dashboard-credit-grant.test.ts

@@ -50,7 +50,6 @@ vi.mock('@/lib/billing/enterprise-provisioning', () => ({
   getLatestEnterpriseProvisionings: vi.fn(async () => new Map()),
 }))
 vi.mock('@/lib/billing/enterprise-outbox', () => ({
-  ENTERPRISE_METADATA_SYNC_EVENT_TYPE: 'stripe.sync-enterprise-metadata',
   resolveEnterpriseMetadataIntent: vi.fn(),
 }))
 vi.mock('@/lib/core/outbox/service', () => outboxServiceMock)