Function syncActiveSideEffects

apps/sim/lib/workflows/deployment-outbox.ts:95–217

(rawPayload: unknown)

}

const syncActiveSideEffects = async (rawPayload: unknown): Promise<void> => {
  const payload = parseSyncActiveSideEffectsPayload(rawPayload)
  const requestId = payload.requestId ?? generateRequestId()
  const [workflowRecord] = await db
    .select()
    .from(workflowTable)
    .where(eq(workflowTable.id, payload.workflowId))
    .limit(1)

  if (!workflowRecord) {
    logger.warn(`[${requestId}] Workflow missing during deployment side-effect sync`, {
      workflowId: payload.workflowId,
    })
    return
  }

  const [versionRow] = await db
    .select({
      id: workflowDeploymentVersion.id,
      state: workflowDeploymentVersion.state,
      isActive: workflowDeploymentVersion.isActive,
    })
    .from(workflowDeploymentVersion)
    .where(
      and(
        eq(workflowDeploymentVersion.workflowId, payload.workflowId),
        eq(workflowDeploymentVersion.id, payload.deploymentVersionId)
      )
    )
    .limit(1)

  if (!versionRow?.isActive) {
    logger.info(`[${requestId}] Skipping stale deployment side-effect sync`, {
      workflowId: payload.workflowId,
      deploymentVersionId: payload.deploymentVersionId,
    })
    if (versionRow) {
      await cleanupDeploymentVersionIfInactive({
        workflowId: payload.workflowId,
        deploymentVersionId: payload.deploymentVersionId,
        workflow: workflowRecord as Record<string, unknown>,
        userId: payload.userId,
        requestId,
      })
    }
    return
  }

  const state = versionRow.state as { blocks?: Record<string, BlockState> }
  const blocks = state.blocks ?? {}
  const workflowData = workflowRecord as Record<string, unknown>

  if (!(await cleanupStaleDeploymentIfNeeded({ payload, workflow: workflowData, requestId }))) {
    return
  }

  const request = new NextRequest(new URL('/api/webhooks', getBaseUrl()))
  const triggerSaveResult = await saveTriggerWebhooksForDeploy({