feat(workspaces): fork + push/pull

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -1,3 +1,4 @@
+import { db } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
 import { type NextRequest, NextResponse } from 'next/server'
@@ -7,6 +8,7 @@ import { processOutboxEvents } from '@/lib/core/outbox/service'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { workflowDeploymentOutboxHandlers } from '@/lib/workflows/deployment-outbox'
+import { reapStaleBackgroundWork } from '@/lib/workspaces/fork/background-work/store'
 
 const logger = createLogger('OutboxProcessorAPI')
 
@@ -33,12 +35,23 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       minRemainingMs: 95_000,
     })
 
-    logger.info('Outbox processing completed', { requestId, ...result })
+    // Reap fork background-work rows stuck `processing` past their TTL (worker crash /
+    // restart has no in-task hook). Independent of the outbox; a failure here must not
+    // fail the outbox run, so it's guarded separately.
+    let reapedBackgroundWork = 0
+    try {
+      reapedBackgroundWork = await reapStaleBackgroundWork(db)
+    } catch (error) {
+      logger.error('Background-work reap failed', { requestId, error: toError(error).message })
+    }
+
+    logger.info('Outbox processing completed', { requestId, ...result, reapedBackgroundWork })
 
     return NextResponse.json({
       success: true,
       requestId,
       result,
+      reapedBackgroundWork,
     })
   } catch (error) {
     logger.error('Outbox processing failed', { requestId, error: toError(error).message })

apps/sim/app/api/workspaces/[id]/background-work/route.ts

@@ -0,0 +1,45 @@
+import { db } from '@sim/db'
+import { type NextRequest, NextResponse } from 'next/server'
+import { getWorkspaceBackgroundWorkContract } from '@/lib/api/contracts/workspace-fork'
+import { parseRequest } from '@/lib/api/server'
+import { getSession } from '@/lib/auth'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { listSurfacedBackgroundWork } from '@/lib/workspaces/fork/background-work/store'
+import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils'
+
+export const GET = withRouteHandler(
+  async (req: NextRequest, context: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const parsed = await parseRequest(getWorkspaceBackgroundWorkContract, req, context)
+    if (!parsed.success) return parsed.response
+    const { id } = parsed.data.params
+
+    const access = await checkWorkspaceAccess(id, session.user.id)
+    if (!access.exists) {
+      return NextResponse.json({ error: 'Workspace not found' }, { status: 404 })
+    }
+    if (!access.canAdmin) {
+      return NextResponse.json({ error: 'Forbidden' }, { status: 403 })
+    }
+
+    const rows = await listSurfacedBackgroundWork(db, id)
+    return NextResponse.json({
+      items: rows.map((row) => ({
+        id: row.id,
+        workspaceId: row.workspaceId,
+        workflowId: row.workflowId,
+        kind: row.kind,
+        status: row.status,
+        message: row.message,
+        error: row.error,
+        metadata: row.metadata ?? null,
+        startedAt: row.startedAt.toISOString(),
+        completedAt: row.completedAt ? row.completedAt.toISOString() : null,
+      })),
+    })
+  }
+)

apps/sim/app/api/workspaces/[id]/fork/diff/route.ts

@@ -0,0 +1,160 @@
+import { db } from '@sim/db'
+import { type NextRequest, NextResponse } from 'next/server'
+import { getForkDiffContract } from '@/lib/api/contracts/workspace-fork'
+import { parseRequest } from '@/lib/api/server'
+import { getSession } from '@/lib/auth'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { loadTargetDraftSubBlocks } from '@/lib/workspaces/fork/copy/copy-workflows'
+import { loadSourceDeployedStates } from '@/lib/workspaces/fork/copy/deploy-bridge'
+import { assertCanPromote } from '@/lib/workspaces/fork/lineage/authz'
+import { loadForkBlockMap } from '@/lib/workspaces/fork/mapping/block-map-store'
+import {
+  collectForkDependentReconfigs,
+  collectForkResourceUsages,
+} from '@/lib/workspaces/fork/mapping/dependent-reconfigs'
+import {
+  forkDependentValueKey,
+  loadForkDependentValues,
+} from '@/lib/workspaces/fork/mapping/dependent-value-store'
+import { computeForkPromotePlan } from '@/lib/workspaces/fork/promote/promote-plan'
+import { buildForkBlockIdResolver } from '@/lib/workspaces/fork/remap/block-identity'
+import { readTargetDraftDependentValue } from '@/lib/workspaces/fork/remap/remap-references'
+
+export const GET = withRouteHandler(
+  async (req: NextRequest, context: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const parsed = await parseRequest(getForkDiffContract, req, context)
+    if (!parsed.success) return parsed.response
+    const { id } = parsed.data.params
+    const { otherWorkspaceId, direction } = parsed.data.query
+
+    const auth = await assertCanPromote(id, otherWorkspaceId, direction, session.user.id)
+
+    const { deployedWorkflows, sourceStates } = await loadSourceDeployedStates(
+      auth.sourceWorkspaceId
+    )
+    const plan = await computeForkPromotePlan({
+      executor: db,
+      edge: auth.edge,
+      sourceWorkspaceId: auth.sourceWorkspaceId,
+      targetWorkspaceId: auth.targetWorkspaceId,
+      direction,
+      deployedSourceWorkflows: deployedWorkflows,
+      sourceStates,
+    })
+
+    // Resolve dependent-reconfig target block ids through the SAME persisted block map the
+    // sync will use, so a re-pick the modal keys by target block id lands on the block the
+    // promote actually writes (on push that's the parent's original id, not a derived one).
+    const sourceIsParent = auth.sourceWorkspaceId === auth.edge.parentWorkspaceId
+    const blockMap = await loadForkBlockMap(db, auth.edge.childWorkspaceId)
+    const resolveBlockId = buildForkBlockIdResolver(sourceIsParent, blockMap)
+
+    // Stored dependent values are the source of truth for what each selector is set to. Overlay
+    // them as each field's currentValue so the modal pre-fills what the user actually saved. For
+    // an edge that predates the store the fallback is the TARGET's own configured value (loaded
+    // from its draft) - never the source's, which would overwrite the target's selection on the
+    // first sync. Both the stored read and the draft read are scoped to the plan's replace
+    // targets, the only workflows with dependents to reconfigure.
+    const replaceTargetIds = plan.items
+      .filter((item) => item.mode === 'replace')
+      .map((item) => item.targetWorkflowId)
+    const [storedValues, targetDraftByWorkflow] = await Promise.all([
+      loadForkDependentValues(db, auth.edge.childWorkspaceId, replaceTargetIds),
+      loadTargetDraftSubBlocks(db, replaceTargetIds),
+    ])
+    const storedByKey = new Map(
+      storedValues.map((entry) => [
+        forkDependentValueKey(entry.targetWorkflowId, entry.targetBlockId, entry.subBlockKey),
+        entry.value,
+      ])
+    )
+
+    // Source block subBlocks keyed by their resolved target identity, so the first-sync draft
+    // fallback can identity-check a nested tool against the SOURCE dependent tool it came from -
+    // an index alone may point at a different tool in the target draft, whose value isn't the
+    // dependent's. Read structurally (only each subblock's `value`), so the in-memory state's
+    // blocks pass without a cast.
+    const sourceBlocksByTarget = new Map<string, Map<string, Record<string, { value?: unknown }>>>()
+    for (const item of plan.items) {
+      if (item.mode !== 'replace') continue
+      const state = sourceStates.get(item.sourceWorkflowId)
+      if (!state) continue
+      const byBlock = new Map<string, Record<string, { value?: unknown }>>()
+      for (const [sourceBlockId, block] of Object.entries(state.blocks)) {
+        byBlock.set(resolveBlockId(item.targetWorkflowId, sourceBlockId), block.subBlocks ?? {})
+      }
+      sourceBlocksByTarget.set(item.targetWorkflowId, byBlock)
+    }
+
+    const dependentReconfigs = collectForkDependentReconfigs(
+      plan.items,
+      sourceStates,
+      resolveBlockId
+    ).map((field) => ({
+      ...field,
+      currentValue:
+        storedByKey.get(
+          forkDependentValueKey(field.targetWorkflowId, field.targetBlockId, field.subBlockKey)
+        ) ??
+        readTargetDraftDependentValue(
+          targetDraftByWorkflow.get(field.targetWorkflowId)?.get(field.targetBlockId),
+          sourceBlocksByTarget.get(field.targetWorkflowId)?.get(field.targetBlockId),
+          field.subBlockKey
+        ),
+    }))
+
+    const toRef = (reference: (typeof plan.unmappedRequired)[number]) => ({
+      kind: reference.kind,
+      sourceId: reference.sourceId,
+      required: reference.required,
+      blockName: reference.blockName,
+    })
+
+    // Orient the mapping around the workspace the modal is open in (`id`): show the
+    // caller's workflow name first, the sync partner's second, so renames are legible.
+    const currentIsSource = auth.sourceWorkspaceId === id
+    const workflows = [
+      ...plan.items.map((item) => {
+        if (item.mode === 'create') {
+          // The target inherits the source's name, so both sides read the same.
+          return {
+            action: 'create' as const,
+            currentName: item.sourceMeta.name,
+            otherName: item.sourceMeta.name,
+          }
+        }
+        const targetName = item.targetName ?? item.sourceMeta.name
+        return {
+          action: 'update' as const,
+          currentName: currentIsSource ? item.sourceMeta.name : targetName,
+          otherName: currentIsSource ? targetName : item.sourceMeta.name,
+        }
+      }),
+      ...plan.archivedTargets.map((target) => ({
+        action: 'archive' as const,
+        currentName: target.name,
+        otherName: target.name,
+      })),
+    ]
+
+    return NextResponse.json({
+      sourceWorkspaceId: auth.sourceWorkspaceId,
+      targetWorkspaceId: auth.targetWorkspaceId,
+      willUpdate: plan.willUpdate,
+      willCreate: plan.willCreate,
+      willArchive: plan.willArchive,
+      workflows,
+      unmappedRequired: plan.unmappedRequired.map(toRef),
+      unmappedOptional: plan.unmappedOptional.map(toRef),
+      mcpReauthServerIds: plan.mcpReauthServerIds,
+      inlineSecretSources: plan.inlineSecretSources,
+      dependentReconfigs,
+      resourceUsages: collectForkResourceUsages(plan.items, sourceStates),
+    })
+  }
+)

apps/sim/app/api/workspaces/[id]/fork/lineage/route.ts

@@ -0,0 +1,56 @@
+import { db } from '@sim/db'
+import { workspace } from '@sim/db/schema'
+import { eq } from 'drizzle-orm'
+import type { NextRequest } from 'next/server'
+import { NextResponse } from 'next/server'
+import { getForkLineageContract } from '@/lib/api/contracts/workspace-fork'
+import { parseRequest } from '@/lib/api/server'
+import { getSession } from '@/lib/auth'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { assertWorkspaceAdminAccess } from '@/lib/workspaces/fork/lineage/authz'
+import { getForkParent } from '@/lib/workspaces/fork/lineage/lineage'
+import { getUndoableRunForTarget } from '@/lib/workspaces/fork/promote/promote-run-store'
+
+export const GET = withRouteHandler(
+  async (req: NextRequest, context: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const parsed = await parseRequest(getForkLineageContract, req, context)
+    if (!parsed.success) return parsed.response
+    const { id: workspaceId } = parsed.data.params
+
+    await assertWorkspaceAdminAccess(workspaceId, session.user.id)
+
+    const [parent, run] = await Promise.all([
+      getForkParent(workspaceId),
+      getUndoableRunForTarget(db, workspaceId),
+    ])
+
+    let undoableRun: {
+      otherWorkspaceId: string
+      otherName: string
+      direction: 'push' | 'pull'
+    } | null = null
+    if (run) {
+      const [other] = await db
+        .select({ name: workspace.name })
+        .from(workspace)
+        .where(eq(workspace.id, run.sourceWorkspaceId))
+        .limit(1)
+      undoableRun = {
+        otherWorkspaceId: run.sourceWorkspaceId,
+        otherName: other?.name ?? 'workspace',
+        direction: run.direction,
+      }
+    }
+
+    return NextResponse.json({
+      workspaceId,
+      parent,
+      undoableRun,
+    })
+  }
+)

apps/sim/app/api/workspaces/[id]/fork/mapping/route.ts

@@ -0,0 +1,75 @@
+import { db } from '@sim/db'
+import { type NextRequest, NextResponse } from 'next/server'
+import {
+  getForkMappingContract,
+  updateForkMappingContract,
+} from '@/lib/api/contracts/workspace-fork'
+import { parseRequest } from '@/lib/api/server'
+import { getSession } from '@/lib/auth'
+import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { assertCanPromote } from '@/lib/workspaces/fork/lineage/authz'
+import { acquireForkEdgeLock, setForkLockTimeout } from '@/lib/workspaces/fork/lineage/lineage'
+import {
+  applyForkMappingEntries,
+  getForkMappingView,
+  validateForkMappingTargets,
+} from '@/lib/workspaces/fork/mapping/mapping-service'
+
+export const GET = withRouteHandler(
+  async (req: NextRequest, context: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const parsed = await parseRequest(getForkMappingContract, req, context)
+    if (!parsed.success) return parsed.response
+    const { id } = parsed.data.params
+    const { otherWorkspaceId, direction } = parsed.data.query
+
+    const auth = await assertCanPromote(id, otherWorkspaceId, direction, session.user.id)
+
+    const { entries } = await getForkMappingView({
+      edge: auth.edge,
+      sourceWorkspaceId: auth.sourceWorkspaceId,
+      targetWorkspaceId: auth.targetWorkspaceId,
+    })
+
+    return NextResponse.json({
+      childWorkspaceId: auth.edge.childWorkspaceId,
+      parentWorkspaceId: auth.edge.parentWorkspaceId,
+      sourceWorkspaceId: auth.sourceWorkspaceId,
+      targetWorkspaceId: auth.targetWorkspaceId,
+      entries,
+    })
+  }
+)
+
+export const PUT = withRouteHandler(
+  async (req: NextRequest, context: { params: Promise<{ id: string }> }) => {
+    const session = await getSession()
+    if (!session?.user?.id) {
+      return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
+    }
+
+    const parsed = await parseRequest(updateForkMappingContract, req, context)
+    if (!parsed.success) return parsed.response
+    const { id } = parsed.data.params
+    const { otherWorkspaceId, direction, entries } = parsed.data.body
+
+    const auth = await assertCanPromote(id, otherWorkspaceId, direction, session.user.id)
+
+    await validateForkMappingTargets(auth.sourceWorkspaceId, auth.targetWorkspaceId, entries)
+
+    // Serialize concurrent mapping saves on this edge so a push (keyed child-side, deleted
+    // then re-upserted parent-side) can't leave duplicate rows for the same source. Same
+    // edge lock promote/rollback use, with a bounded wait.
+    const updated = await db.transaction(async (tx) => {
+      await setForkLockTimeout(tx)
+      await acquireForkEdgeLock(tx, auth.edge.childWorkspaceId)
+      return applyForkMappingEntries(tx, auth.edge, session.user.id, direction, entries)
+    })
+
+    return NextResponse.json({ success: true as const, updated })
+  }
+)