Compare commits

...

2 Commits

Author SHA1 Message Date
Vikhyath Mondreti
6e78df7d90 fix edit workflow 2026-01-16 19:43:29 -08:00
Vikhyath Mondreti
df8dd052c8 fix(sockets): dying on deployed webhooks 2026-01-16 19:41:42 -08:00
3 changed files with 3 additions and 86 deletions

View File

@@ -2499,7 +2499,9 @@ export const editWorkflowServerTool: BaseServerTool<EditWorkflowParams, any> = {
async execute(params: EditWorkflowParams, context?: { userId: string }): Promise<any> {
const logger = createLogger('EditWorkflowServerTool')
const { operations, workflowId, currentUserWorkflow } = params
if (!operations || operations.length === 0) throw new Error('operations are required')
if (!Array.isArray(operations) || operations.length === 0) {
throw new Error('operations are required and must be an array')
}
if (!workflowId) throw new Error('workflowId is required')
logger.info('Executing edit_workflow', {

View File

@@ -1,7 +1,6 @@
import crypto from 'crypto'
import {
db,
webhook,
workflow,
workflowBlocks,
workflowDeploymentVersion,
@@ -22,7 +21,6 @@ import { generateLoopBlocks, generateParallelBlocks } from '@/stores/workflows/w
const logger = createLogger('WorkflowDBHelpers')
export type WorkflowDeploymentVersion = InferSelectModel<typeof workflowDeploymentVersion>
type WebhookRecord = InferSelectModel<typeof webhook>
type SubflowInsert = InferInsertModel<typeof workflowSubflows>
export interface WorkflowDeploymentVersionResponse {
@@ -337,18 +335,6 @@ export async function saveWorkflowToNormalizedTables(
// Start a transaction
await db.transaction(async (tx) => {
// Snapshot existing webhooks before deletion to preserve them through the cycle
let existingWebhooks: WebhookRecord[] = []
try {
existingWebhooks = await tx.select().from(webhook).where(eq(webhook.workflowId, workflowId))
} catch (webhookError) {
// Webhook table might not be available in test environments
logger.debug('Could not load webhooks before save, skipping preservation', {
error: webhookError instanceof Error ? webhookError.message : String(webhookError),
})
}
// Clear existing data for this workflow
await Promise.all([
tx.delete(workflowBlocks).where(eq(workflowBlocks.workflowId, workflowId)),
tx.delete(workflowEdges).where(eq(workflowEdges.workflowId, workflowId)),
@@ -419,42 +405,6 @@ export async function saveWorkflowToNormalizedTables(
if (subflowInserts.length > 0) {
await tx.insert(workflowSubflows).values(subflowInserts)
}
// Re-insert preserved webhooks if any exist and their blocks still exist
if (existingWebhooks.length > 0) {
try {
const webhookInserts = existingWebhooks
.filter((wh) => !!state.blocks?.[wh.blockId ?? ''])
.map((wh) => ({
id: wh.id,
workflowId: wh.workflowId,
blockId: wh.blockId,
path: wh.path,
provider: wh.provider,
providerConfig: wh.providerConfig,
credentialSetId: wh.credentialSetId,
isActive: wh.isActive,
createdAt: wh.createdAt,
updatedAt: new Date(),
}))
if (webhookInserts.length > 0) {
await tx.insert(webhook).values(webhookInserts)
logger.debug(`Preserved ${webhookInserts.length} webhook(s) through workflow save`, {
workflowId,
})
}
} catch (webhookInsertError) {
// Webhook preservation is optional - don't fail the entire save if it errors
logger.warn('Could not preserve webhooks during save', {
error:
webhookInsertError instanceof Error
? webhookInsertError.message
: String(webhookInsertError),
workflowId,
})
}
}
})
return { success: true }

View File

@@ -1,7 +1,6 @@
import * as schema from '@sim/db'
import { webhook, workflow, workflowBlocks, workflowEdges, workflowSubflows } from '@sim/db'
import { createLogger } from '@sim/logger'
import type { InferSelectModel } from 'drizzle-orm'
import { and, eq, inArray, or, sql } from 'drizzle-orm'
import { drizzle } from 'drizzle-orm/postgres-js'
import postgres from 'postgres'
@@ -1175,14 +1174,6 @@ async function handleWorkflowOperationTx(
parallelCount: Object.keys(parallels || {}).length,
})
// Snapshot existing webhooks before deletion to preserve them through the cycle
// (workflowBlocks has CASCADE DELETE to webhook table)
const existingWebhooks = await tx
.select()
.from(webhook)
.where(eq(webhook.workflowId, workflowId))
// Delete all existing blocks (this will cascade delete edges and webhooks via ON DELETE CASCADE)
await tx.delete(workflowBlocks).where(eq(workflowBlocks.workflowId, workflowId))
// Delete all existing subflows
@@ -1248,32 +1239,6 @@ async function handleWorkflowOperationTx(
await tx.insert(workflowSubflows).values(parallelValues)
}
// Re-insert preserved webhooks if any exist and their blocks still exist
type WebhookRecord = InferSelectModel<typeof webhook>
if (existingWebhooks.length > 0) {
const webhookInserts = existingWebhooks
.filter((wh: WebhookRecord) => !!blocks?.[wh.blockId ?? ''])
.map((wh: WebhookRecord) => ({
id: wh.id,
workflowId: wh.workflowId,
blockId: wh.blockId,
path: wh.path,
provider: wh.provider,
providerConfig: wh.providerConfig,
credentialSetId: wh.credentialSetId,
isActive: wh.isActive,
createdAt: wh.createdAt,
updatedAt: new Date(),
}))
if (webhookInserts.length > 0) {
await tx.insert(webhook).values(webhookInserts)
logger.debug(`Preserved ${webhookInserts.length} webhook(s) through state replacement`, {
workflowId,
})
}
}
logger.info(`Successfully replaced workflow state for ${workflowId}`)
break
}