diff --git a/apps/docs/content/docs/tables/index.mdx b/apps/docs/content/docs/tables/index.mdx index 41b821d995e..e44ca483464 100644 --- a/apps/docs/content/docs/tables/index.mdx +++ b/apps/docs/content/docs/tables/index.mdx @@ -24,7 +24,6 @@ Every column has a type, which decides how its values are stored and validated. | **Currency** | An amount in a currency you pick per column | `$1,234.56` | | **Boolean** | `true` or `false` | `true` | | **Date** | A date | `2026-03-16` | -| **Expiration** | An absolute row expiration time, stored as Unix epoch seconds (seconds since January 1, 1970 UTC) | `1773671400` | | **JSON** | An object or array | `{ "tier": "pro" }` | | **Select** | One of a fixed set of options, or several | `Pro` | @@ -32,8 +31,6 @@ Types are enforced as you enter values, so a Number column only takes numbers. A Currency column stores a plain number and renders it in the currency you choose for that column, so filters, sorts, and exports all see the amount itself. Changing a column's currency relabels it — it does not convert the amounts. -A table can have one Expiration column. Adding it enables row expiration; rows with a non-empty expiration value become eligible for deletion after that time passes. Cleanup runs periodically, so actual row removal may happen after the expiration timestamp rather than exactly at it. Deleting the Expiration column disables expiration for the table. Expiration cells use the date editor, while APIs and workflows read and write integer Unix epoch seconds. - ## Editing a table Open the **Tables** section in the sidebar and click **New table** to create one. Add columns from the column header, type into a cell to edit it, and paste rows from a spreadsheet to bulk-load. Filter and sort from the toolbar without changing the underlying data. The editor has full keyboard support; see [keyboard shortcuts](/keyboard-shortcuts). diff --git a/apps/docs/content/docs/workflows/blocks/agent.mdx b/apps/docs/content/docs/workflows/blocks/agent.mdx index 960bd7bab02..6bb78780dd7 100644 --- a/apps/docs/content/docs/workflows/blocks/agent.mdx +++ b/apps/docs/content/docs/workflows/blocks/agent.mdx @@ -112,7 +112,7 @@ Live tool-call chips stream for **OpenAI, Anthropic, Azure Anthropic, Google, Ve | Anthropic | Summaries only — These generations omit full thinking; Sim requests summarized thinking on streaming runs. | `claude-fable-5-1`, `claude-fable-5`, `claude-sonnet-5`, `claude-opus-5`, `claude-opus-4-8`, `claude-opus-4-7`, `claude-opus-4-6`, `claude-sonnet-4-6`, `claude-opus-4-5`, `claude-opus-4-1`, `claude-sonnet-4-5`, `claude-haiku-4-5` | | Azure OpenAI | Summaries only — Requires OpenAI organization verification; falls back to no summaries. | `azure/gpt-5.4`, `azure/gpt-5.4-mini`, `azure/gpt-5.4-nano`, `azure/gpt-5.2`, `azure/gpt-5.1`, `azure/gpt-5.1-codex`, `azure/gpt-5`, `azure/gpt-5-mini`, `azure/gpt-5-nano`, `azure/o3`, `azure/o4-mini` | | Azure Anthropic | Summaries only — These generations omit full thinking; Sim requests summarized thinking on streaming runs. | `azure-anthropic/claude-opus-4-6`, `azure-anthropic/claude-opus-4-5`, `azure-anthropic/claude-sonnet-4-5`, `azure-anthropic/claude-opus-4-1`, `azure-anthropic/claude-haiku-4-5` | -| Google | Summaries only | `gemini-3.6-flash`, `gemini-3.5-flash-lite`, `gemini-3.5-flash`, `gemini-3.1-pro-preview`, `gemini-3.1-flash-lite`, `gemini-3-flash-preview`, `gemini-2.5-pro`, `gemini-2.5-flash`, `gemini-2.5-flash-lite` | +| Google | Summaries only | `gemini-3.8-flash`, `gemini-3.6-flash`, `gemini-3.5-flash-lite`, `gemini-3.5-flash`, `gemini-3.1-pro-preview`, `gemini-3.1-flash-lite`, `gemini-3-flash-preview`, `gemini-2.5-pro`, `gemini-2.5-flash`, `gemini-2.5-flash-lite` | | Vertex AI | Summaries only | `vertex/gemini-3.5-flash`, `vertex/gemini-3.1-pro-preview`, `vertex/gemini-3.1-flash-lite`, `vertex/gemini-3-flash-preview`, `vertex/gemini-2.5-pro`, `vertex/gemini-2.5-flash`, `vertex/gemini-2.5-flash-lite` | | DeepSeek | Full thinking deltas | `deepseek-v4-pro`, `deepseek-v4-flash`, `deepseek-reasoner` | | xAI | Full thinking deltas | `grok-4.6`, `grok-4.5`, `grok-4.3`, `grok-4.20-multi-agent-0309` | diff --git a/apps/docs/openapi-v2-tables.json b/apps/docs/openapi-v2-tables.json index d91748c097d..fb9688e5f6c 100644 --- a/apps/docs/openapi-v2-tables.json +++ b/apps/docs/openapi-v2-tables.json @@ -4979,16 +4979,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { @@ -5266,16 +5257,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Column data type." }, "required": { @@ -5454,16 +5436,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { @@ -5563,16 +5536,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Column data type." }, "required": { @@ -5669,7 +5633,7 @@ "type": { "description": "Replacement column data type.", "type": "string", - "enum": ["string", "number", "currency", "boolean", "date", "ttl", "json", "select"] + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"] }, "required": { "description": "Whether inserts must supply a value for this column.", @@ -7433,16 +7397,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { @@ -7642,16 +7597,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Output column data type." }, "required": { @@ -7792,16 +7738,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Output column data type." }, "required": { @@ -7919,16 +7856,7 @@ }, "type": { "type": "string", - "enum": [ - "string", - "number", - "currency", - "boolean", - "date", - "ttl", - "json", - "select" - ], + "enum": ["string", "number", "currency", "boolean", "date", "json", "select"], "description": "Data type of values stored in the column." }, "required": { diff --git a/apps/sim/app/(landing)/integrations/data/landing-content.ts b/apps/sim/app/(landing)/integrations/data/landing-content.ts index eec845c9475..32064764949 100644 --- a/apps/sim/app/(landing)/integrations/data/landing-content.ts +++ b/apps/sim/app/(landing)/integrations/data/landing-content.ts @@ -38,6 +38,6 @@ export const INTEGRATION_LANDING_CONTENT: Record ({ ownedAttachableWorkspacesWhere: vi.fn(() => undefined), })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), deferOutboxHandler: (reason: string, _minimum?: number, consumeAttempt = true) => ({ outcome: 'deferred', reason, diff --git a/apps/sim/lib/admin/member-operation.ts b/apps/sim/lib/admin/member-operation.ts index 71eb2981bc8..e68090a08e4 100644 --- a/apps/sim/lib/admin/member-operation.ts +++ b/apps/sim/lib/admin/member-operation.ts @@ -16,7 +16,7 @@ import { import { reconcileOrganizationSeats } from '@/lib/billing/organizations/seats' import { ENTITLED_SUBSCRIPTION_STATUSES } from '@/lib/billing/subscriptions/utils' import { - deferOutboxHandler, + continueOutboxHandler, enqueueOutboxEvent, type OutboxHandler, outboxEventHasSourceOperationId, @@ -688,7 +688,7 @@ export const processAdminMemberOperation: OutboxHandler = async (rawPay } if (nextWorkspaceIndex < payload.request.workspaceIds.length) { - return deferOutboxHandler('Continuing bounded member workspace moves', undefined, false) + return continueOutboxHandler('Continuing bounded member workspace moves') } } diff --git a/apps/sim/lib/api/contracts/tables.ts b/apps/sim/lib/api/contracts/tables.ts index 9d3a95e206e..30dd03797e3 100644 --- a/apps/sim/lib/api/contracts/tables.ts +++ b/apps/sim/lib/api/contracts/tables.ts @@ -53,7 +53,9 @@ export const domainObjectSchema = () => z.custom(isRecordLike) * Column types are a fixed enum derived from `COLUMN_TYPES` so callers cannot * send arbitrary strings the server would reject downstream. */ -export const columnTypeSchema = z.enum(COLUMN_TYPES) +export const columnTypeSchema = z + .enum(COLUMN_TYPES) + .meta({ omitEnumValuesFromOpenApi: ['ttl'] as const }) /** One choice in a `select` column. `id` is the stable cell key. */ export const selectOptionSchema = z.object({ diff --git a/apps/sim/lib/billing/enterprise-provisioning.test.ts b/apps/sim/lib/billing/enterprise-provisioning.test.ts index a71b36f8bb5..58afc58ab15 100644 --- a/apps/sim/lib/billing/enterprise-provisioning.test.ts +++ b/apps/sim/lib/billing/enterprise-provisioning.test.ts @@ -61,6 +61,11 @@ vi.mock('@/lib/billing/webhooks/enterprise-reconciliation-lease', () => ({ ), })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), deferOutboxHandler: (reason: string, minimumBackoffMs?: number, consumeAttempt = true) => ({ outcome: 'deferred', reason, diff --git a/apps/sim/lib/billing/enterprise-provisioning.ts b/apps/sim/lib/billing/enterprise-provisioning.ts index 739ca721211..6a5f1152037 100644 --- a/apps/sim/lib/billing/enterprise-provisioning.ts +++ b/apps/sim/lib/billing/enterprise-provisioning.ts @@ -79,6 +79,7 @@ import { withEnterpriseReconciliationLease } from '@/lib/billing/webhooks/enterp import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-handlers' import { env } from '@/lib/core/config/env' import { + continueOutboxHandler, deferOutboxHandler, enqueueOutboxEvent, type OutboxEventContext, @@ -3108,7 +3109,7 @@ export const reconcileEnterpriseMembers: OutboxHandler = async (rawPayl if (!nextCursor) return await context.checkpointPayload({ afterUserId: nextCursor }) - return deferOutboxHandler('Continuing bounded Enterprise member reconciliation', undefined, false) + return continueOutboxHandler('Continuing bounded Enterprise member reconciliation') } export const enterpriseIssuanceOutboxHandlers = { diff --git a/apps/sim/lib/core/outbox/service.test.ts b/apps/sim/lib/core/outbox/service.test.ts index dc4fd2f2504..9817363630c 100644 --- a/apps/sim/lib/core/outbox/service.test.ts +++ b/apps/sim/lib/core/outbox/service.test.ts @@ -25,6 +25,7 @@ vi.mock('@sim/utils/id', () => ({ })) import { + continueOutboxHandler, deferOutboxHandler, enqueueOrReschedulePendingOutboxEvent, enqueueOutboxEvent, @@ -387,6 +388,20 @@ describe('processOutboxEvents — handler success and retry', () => { expect(deferredUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null }) }) + it('re-runs a continued handler without consuming its attempt budget', async () => { + const handler = vi.fn(async () => continueOutboxHandler('continuing bounded cleanup')) + queueTableRows(outboxEvent, [makePendingRow({ attempts: 4, maxAttempts: 5 })]) + holdLease() + + const result = await processOutboxEvents({ 'test.event': handler }) + + expect(result.retried).toBe(1) + const continuedUpdate = updateSets().find( + (set) => set.status === 'pending' && 'attempts' in set + ) + expect(continuedUpdate).toMatchObject({ attempts: 4, lastError: null, lockedAt: null }) + }) + it('dead-letters on failure when attempts reaches maxAttempts', async () => { const handler = vi.fn(async () => { throw new Error('permanent failure') diff --git a/apps/sim/lib/core/outbox/service.ts b/apps/sim/lib/core/outbox/service.ts index a36600aaa43..a4ede767836 100644 --- a/apps/sim/lib/core/outbox/service.ts +++ b/apps/sim/lib/core/outbox/service.ts @@ -81,8 +81,9 @@ export interface DeferredOutboxHandlerResult { minimumBackoffMs?: number /** * Defaults to true for an external acknowledgement with a finite retry - * budget. Set false only for an internal dependency whose own outbox row - * independently reaches completed or dead-letter. + * budget. False is reserved for waits on an internal dependency whose own + * outbox row independently reaches completed or dead-letter, and for + * bounded continuation after durable progress (`continueOutboxHandler`). */ consumeAttempt?: boolean } @@ -100,6 +101,19 @@ export function deferOutboxHandler( } } +/** + * Yields after durable progress so the worker re-runs the event without + * spending an attempt. For bounded batches whose remaining work shrinks on + * every run; a run that made no progress must throw or `deferOutboxHandler` + * instead, or the event never reaches a terminal state. + */ +export function continueOutboxHandler( + reason: string, + minimumBackoffMs?: number +): DeferredOutboxHandlerResult { + return deferOutboxHandler(reason, minimumBackoffMs, false) +} + export type OutboxHandler = ( payload: T, context: OutboxEventContext diff --git a/apps/sim/lib/webhooks/deploy.test.ts b/apps/sim/lib/webhooks/deploy.test.ts index 9732e9d0308..77ce7cc1642 100644 --- a/apps/sim/lib/webhooks/deploy.test.ts +++ b/apps/sim/lib/webhooks/deploy.test.ts @@ -1,9 +1,15 @@ /** * @vitest-environment node */ -import { account, credential } from '@sim/db/schema' -import { queueTableRows, resetDbChainMock, resetEnvFlagsMock, setEnvFlags } from '@sim/testing' -import { eq } from 'drizzle-orm' +import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db/schema' +import { + dbChainMockFns, + queueTableRows, + resetDbChainMock, + resetEnvFlagsMock, + setEnvFlags, +} from '@sim/testing' +import { eq, ne } from 'drizzle-orm' import { afterAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest' import type { SubBlockConfig } from '@/blocks/types' import type { BlockState } from '@/stores/workflows/workflow/types' @@ -29,6 +35,14 @@ vi.mock('@/lib/webhooks/utils.server', () => ({ vi.mock('@/lib/webhooks/pending-verification', () => ({ PendingWebhookVerificationTracker: vi.fn(), })) +const { mockIsDeploymentVersionActive, mockIsDeploymentVersionProtected } = vi.hoisted(() => ({ + mockIsDeploymentVersionActive: vi.fn(), + mockIsDeploymentVersionProtected: vi.fn(), +})) +vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ + isDeploymentVersionActive: mockIsDeploymentVersionActive, + isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtected, +})) const { mockGetSlackBotCredential, @@ -52,9 +66,11 @@ vi.mock('@/lib/webhooks/providers/slack', () => ({ import { buildProviderConfig, + cleanupInactiveDeploymentWebhooks, resolveTriggerCredentialId, resolveWebhookConfigForBlock, } from '@/lib/webhooks/deploy' +import { cleanupExternalWebhook } from '@/lib/webhooks/provider-subscriptions' import { getBlock } from '@/blocks' import { getTrigger } from '@/triggers' @@ -639,3 +655,107 @@ describe('resolveWebhookConfigForBlock — TikTok routing', () => { expect(result?.error.message).toContain('Reconnect') }) }) + +describe('cleanupInactiveDeploymentWebhooks', () => { + const workflow = { id: 'workflow-1', userId: 'user-1', workspaceId: 'workspace-1' } + const input = { + workflowId: 'workflow-1', + workflow, + requestId: 'request-1', + protectedDeploymentVersionId: null, + limit: 5, + } + + function staleWebhookRow(id: string) { + return { + id, + workflowId: 'workflow-1', + deploymentVersionId: 'version-1', + provider: 'github', + providerConfig: {}, + archivedAt: null, + createdAt: new Date('2026-07-14T08:00:00.000Z'), + } + } + + beforeEach(() => { + mockIsDeploymentVersionActive.mockResolvedValue(false) + mockIsDeploymentVersionProtected.mockResolvedValue(false) + }) + + it('retires one bounded batch of stale rows and reports the remainder', async () => { + queueTableRows(webhook, [ + staleWebhookRow('wh-1'), + staleWebhookRow('wh-2'), + staleWebhookRow('wh-3'), + ]) + queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }]) + queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }]) + + await expect(cleanupInactiveDeploymentWebhooks({ ...input, limit: 2 })).resolves.toEqual({ + hasMore: true, + }) + + expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(2) + expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledWith( + expect.objectContaining({ id: 'wh-1' }), + workflow, + 'request-1', + { throwOnError: true } + ) + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(2) + }) + + it('reports completion once the batch drains every stale row', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + queueTableRows(workflowDeploymentVersion, [{ id: 'version-1' }]) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: false }) + + expect(vi.mocked(cleanupExternalWebhook)).toHaveBeenCalledTimes(1) + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + }) + + it('excludes the version the current operation is preparing from the batch', async () => { + queueTableRows(webhook, []) + + await expect( + cleanupInactiveDeploymentWebhooks({ ...input, protectedDeploymentVersionId: 'version-3' }) + ).resolves.toEqual({ hasMore: false }) + + expect(ne).toHaveBeenCalledWith(webhook.deploymentVersionId, 'version-3') + }) + + it('stops before any provider call once the fence reports a change', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + + await expect( + cleanupInactiveDeploymentWebhooks({ ...input, shouldContinue: async () => false }) + ).resolves.toEqual({ hasMore: true }) + + expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) + + it('leaves a row alone when its version was re-activated after the batch was selected', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + mockIsDeploymentVersionActive.mockResolvedValue(true) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true }) + + expect(mockIsDeploymentVersionActive).toHaveBeenCalledWith('workflow-1', 'version-1') + expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) + + it('leaves a row alone when its version became the current candidate mid-batch', async () => { + queueTableRows(webhook, [staleWebhookRow('wh-1')]) + mockIsDeploymentVersionProtected.mockResolvedValue(true) + + await expect(cleanupInactiveDeploymentWebhooks(input)).resolves.toEqual({ hasMore: true }) + + expect(mockIsDeploymentVersionProtected).toHaveBeenCalledWith('workflow-1', 'version-1') + expect(vi.mocked(cleanupExternalWebhook)).not.toHaveBeenCalled() + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/webhooks/deploy.ts b/apps/sim/lib/webhooks/deploy.ts index 136720edcc1..7543e0a0595 100644 --- a/apps/sim/lib/webhooks/deploy.ts +++ b/apps/sim/lib/webhooks/deploy.ts @@ -3,7 +3,7 @@ import { account, credential, webhook, workflowDeploymentVersion } from '@sim/db import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateShortId } from '@sim/utils/id' -import { and, eq, inArray, isNull, or } from 'drizzle-orm' +import { and, asc, eq, inArray, isNull, ne, or } from 'drizzle-orm' import type { NextRequest } from 'next/server' import { isSlackExtendedScopesEnabled } from '@/lib/core/config/env-flags' import { getProviderIdFromServiceId } from '@/lib/oauth' @@ -33,6 +33,10 @@ import { replaceSlackStreamAuthoringConfig, } from '@/lib/webhooks/slack-stream-config' import { findConflictingWebhookPathOwner } from '@/lib/webhooks/utils.server' +import { + isDeploymentVersionActive, + isDeploymentVersionProtectedByCurrentOperation, +} from '@/lib/workflows/persistence/deployment-operations' import { buildCanonicalIndex, buildSubBlockValues, @@ -1276,39 +1280,15 @@ export async function cleanupWebhooksForWorkflow( if (!skipExternalCleanup) { for (const wh of existingWebhooks) { - if (shouldDeleteWebhook && !(await shouldDeleteWebhook())) { - logger.info(`[${requestId}] Stopping webhook cleanup because deployment became active`, { - workflowId, - deploymentVersionId, - webhookId: wh.id, - }) - return - } - - try { - await cleanupExternalWebhook(wh, workflow, requestId, { - throwOnError: strictExternalCleanup, - }) - } catch (cleanupError) { - logger.warn(`[${requestId}] Failed to cleanup external webhook ${wh.id}`, cleanupError) - if (strictExternalCleanup) throw cleanupError - // Continue with other webhooks even if one fails - } - - const deleted = await deleteWebhookRecordAfterCleanup({ - workflowId, + const deleted = await cleanupWebhookRow({ + webhook: wh, + workflow, + requestId, deploymentVersionId, - webhookId: wh.id, + strictExternalCleanup, shouldDeleteWebhook, }) - if (!deleted) { - logger.info(`[${requestId}] Stopping webhook DB cleanup because deployment became active`, { - workflowId, - deploymentVersionId, - webhookId: wh.id, - }) - return - } + if (!deleted) return } } else { for (const wh of existingWebhooks) { @@ -1336,6 +1316,141 @@ export async function cleanupWebhooksForWorkflow( ) } +type WebhookRow = typeof webhook.$inferSelect + +/** + * Tears down one webhook's provider subscription and then deletes its row. + * Returns false when `shouldDeleteWebhook` reports the deployment became + * active again, in which case the caller must stop touching its rows. + */ +async function cleanupWebhookRow(params: { + webhook: WebhookRow + workflow: Record + requestId: string + deploymentVersionId?: string | null + strictExternalCleanup: boolean + shouldDeleteWebhook?: () => Promise +}): Promise { + const { webhook: wh, workflow, requestId, deploymentVersionId, strictExternalCleanup } = params + const workflowId = wh.workflowId + if (params.shouldDeleteWebhook && !(await params.shouldDeleteWebhook())) { + logger.info(`[${requestId}] Stopping webhook cleanup because deployment became active`, { + workflowId, + deploymentVersionId, + webhookId: wh.id, + }) + return false + } + + try { + await cleanupExternalWebhook(wh, workflow, requestId, { throwOnError: strictExternalCleanup }) + } catch (cleanupError) { + logger.warn(`[${requestId}] Failed to cleanup external webhook ${wh.id}`, cleanupError) + if (strictExternalCleanup) throw cleanupError + } + + const deleted = await deleteWebhookRecordAfterCleanup({ + workflowId, + deploymentVersionId, + webhookId: wh.id, + shouldDeleteWebhook: params.shouldDeleteWebhook, + }) + if (!deleted) { + logger.info(`[${requestId}] Stopping webhook DB cleanup because deployment became active`, { + workflowId, + deploymentVersionId, + webhookId: wh.id, + }) + } + return deleted +} + +export interface InactiveDeploymentWebhookCleanupResult { + /** True when rows remain beyond this batch and the caller should run again. */ + hasMore: boolean +} + +/** + * Tears down webhooks still owned by inactive deployment versions of a + * workflow, at most `limit` rows per call. Provider teardown costs one call + * per row, so the work is bounded here and `hasMore` asks the caller to come + * back; every finished row leaves the remaining set smaller, so repeated calls + * converge. `protectedDeploymentVersionId` is the version an in-flight + * operation is preparing, inactive until cutover but live preparation state. + * Each row is re-checked right before its provider call: the version must + * still be inactive and must not have become the current operation's + * candidate, since either can change while the batch runs and the fenced row + * delete that follows cannot undo provider teardown. + */ +export async function cleanupInactiveDeploymentWebhooks(params: { + workflowId: string + workflow: Record + requestId: string + protectedDeploymentVersionId: string | null + limit: number + shouldContinue?: () => Promise +}): Promise { + const { workflowId, workflow, requestId, shouldContinue } = params + const inactiveVersionIds = db + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, workflowId), + eq(workflowDeploymentVersion.isActive, false) + ) + ) + const staleWebhooks = await db + .select() + .from(webhook) + .where( + and( + eq(webhook.workflowId, workflowId), + isNull(webhook.archivedAt), + inArray(webhook.deploymentVersionId, inactiveVersionIds), + params.protectedDeploymentVersionId + ? ne(webhook.deploymentVersionId, params.protectedDeploymentVersionId) + : undefined + ) + ) + .orderBy(asc(webhook.createdAt)) + .limit(params.limit + 1) + + const batch = staleWebhooks.slice(0, params.limit) + if (batch.length === 0) return { hasMore: false } + + logger.info( + `[${requestId}] Cleaning up ${batch.length} webhook(s) owned by inactive deployments`, + { + workflowId, + webhookIds: batch.map((wh) => wh.id), + } + ) + + for (const wh of batch) { + const deploymentVersionId = wh.deploymentVersionId + const deleted = await cleanupWebhookRow({ + webhook: wh, + workflow, + requestId, + deploymentVersionId, + strictExternalCleanup: true, + shouldDeleteWebhook: async () => { + if (shouldContinue && !(await shouldContinue())) return false + if (!deploymentVersionId) return true + if (await isDeploymentVersionActive(workflowId, deploymentVersionId)) return false + return !(await isDeploymentVersionProtectedByCurrentOperation( + workflowId, + deploymentVersionId + )) + }, + }) + if (!deleted) return { hasMore: true } + } + + return { hasMore: staleWebhooks.length > params.limit } +} + /** * Deletes a webhook record unless the deployment became active again. * diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index 4483575af30..9cfa87dbce9 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/apps/sim/lib/workflows/deployment-outbox.test.ts @@ -30,6 +30,10 @@ const { mockRecordAudit, mockEmitWorkflowDeployedEvent, mockCaptureServerEvent, + mockCleanupInactiveDeploymentWebhooks, + mockDeleteInactiveDeploymentSchedules, + mockGetProtectedDeploymentVersionId, + mockIsDeploymentVersionActive, mockTx, } = vi.hoisted(() => ({ mockPrepareWebhooks: vi.fn(), @@ -51,6 +55,10 @@ const { mockRecordAudit: vi.fn(), mockEmitWorkflowDeployedEvent: vi.fn(), mockCaptureServerEvent: vi.fn(), + mockCleanupInactiveDeploymentWebhooks: vi.fn(), + mockDeleteInactiveDeploymentSchedules: vi.fn(), + mockGetProtectedDeploymentVersionId: vi.fn(), + mockIsDeploymentVersionActive: vi.fn(), mockTx: { select: vi.fn(), update: vi.fn(), execute: vi.fn() }, })) @@ -66,6 +74,11 @@ vi.mock('@sim/audit', () => ({ vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock })) vi.mock('@/lib/core/outbox/service', () => ({ + continueOutboxHandler: (reason: string) => ({ + outcome: 'deferred', + reason, + consumeAttempt: false, + }), enqueueOutboxEvent: vi.fn(), processOutboxEventById: vi.fn(), })) @@ -85,6 +98,7 @@ vi.mock('@/lib/mcp/workflow-mcp-sync', () => ({ })) vi.mock('@/lib/webhooks/deploy', () => ({ + cleanupInactiveDeploymentWebhooks: mockCleanupInactiveDeploymentWebhooks, cleanupWebhooksForWorkflow: mockCleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy: vi.fn(), saveTriggerWebhooksForDeploy: vi.fn(), @@ -102,7 +116,9 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ activateDeploymentOperation: mockActivateDeploymentOperation, beginDeploymentOperationActivation: mockBeginDeploymentOperationActivation, getDeploymentOperation: mockGetDeploymentOperation, + getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + isDeploymentVersionActive: mockIsDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation: mockIsDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness: mockMarkDeploymentComponentReadiness, @@ -113,6 +129,7 @@ vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ vi.mock('@/lib/workflows/schedules', () => ({ createSchedulesForDeploy: mockCreateSchedulesForDeploy, + deleteInactiveDeploymentSchedules: mockDeleteInactiveDeploymentSchedules, deleteSchedulesForWorkflow: vi.fn(), })) @@ -219,6 +236,10 @@ describe('versioned deployment preparation outbox', () => { }) mockIsDeploymentOperationCurrent.mockResolvedValue(false) mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(false) + mockIsDeploymentVersionActive.mockResolvedValue(false) + mockGetProtectedDeploymentVersionId.mockResolvedValue(null) + mockDeleteInactiveDeploymentSchedules.mockResolvedValue({ status: 'deleted', count: 0 }) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValue({ hasMore: false }) }) it('activates only after every preparation component is ready', async () => { @@ -558,6 +579,8 @@ describe('versioned deployment preparation outbox', () => { await expect(handler()(payload(), context(new AbortController(), 3))).resolves.toBeUndefined() expect(mockCleanupRetiredWebhookRegistrations).not.toHaveBeenCalled() + expect(mockDeleteInactiveDeploymentSchedules).not.toHaveBeenCalled() + expect(mockCleanupInactiveDeploymentWebhooks).not.toHaveBeenCalled() expect(mockMarkDeploymentOperationFailed).not.toHaveBeenCalled() expect(mockRecordDeploymentOperationRetry).not.toHaveBeenCalled() }) @@ -589,11 +612,35 @@ describe('versioned deployment preparation outbox', () => { { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, ]) - await handler()(payload(), context()) + const outboxContext = context() + + await expect(handler()(payload(), outboxContext)).resolves.toBeUndefined() expect(mockCleanupRetiredWebhookRegistrations).toHaveBeenCalledTimes(1) expect(mockRecordAudit).toHaveBeenCalledTimes(1) expect(mockEmitWorkflowDeployedEvent).toHaveBeenCalledTimes(1) + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + operationFence: { + workflowId: 'workflow-1', + operationId: 'operation-1', + generation: 2, + deploymentVersionId: 'version-2', + statuses: ['active'], + }, + }) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledWith( + expect.objectContaining({ + workflowId: 'workflow-1', + protectedDeploymentVersionId: null, + limit: 20, + }) + ) + expect(outboxContext.checkpointPayload).toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) }) /** @@ -619,29 +666,140 @@ describe('versioned deployment preparation outbox', () => { ) }) - it('keeps v1 cleanup from deleting a candidate owned by the current v2 operation', async () => { + it('continues through the outbox while stale webhooks remain, then checkpoints the cleanup', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW })) + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValueOnce({ hasMore: true }) + const outboxContext = context() + + await expect(handler()(payload(), outboxContext)).resolves.toEqual({ + outcome: 'deferred', + reason: expect.any(String), + consumeAttempt: false, + }) + expect(outboxContext.checkpointPayload).not.toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) + + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + await expect(handler()(payload(), outboxContext)).resolves.toBeUndefined() + + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledTimes(2) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledTimes(2) + expect(outboxContext.checkpointPayload).toHaveBeenCalledWith( + expect.objectContaining({ + checkpoints: expect.objectContaining({ inactiveCleanupCompleted: true }), + }) + ) + }) + + it('stops legacy inactive cleanup as soon as its lease is aborted', async () => { + const controller = new AbortController() + controller.abort() + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_INACTIVE_SIDE_EFFECTS + ] + + await expect( + cleanupHandler( + { workflowId: 'workflow-1', activeDeploymentVersionId: 'version-2', userId: 'user-1' }, + context(controller) + ) + ).rejects.toMatchObject({ name: 'AbortError' }) + + expect(mockDeleteInactiveDeploymentSchedules).not.toHaveBeenCalled() + expect(mockCleanupInactiveDeploymentWebhooks).not.toHaveBeenCalled() + }) + + it('retires undeployed side effects by row and shields the candidate owned by the current v2 operation', async () => { queueTableRows(schemaMock.workflow, [ { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, ]) - queueTableRows(schemaMock.workflowDeploymentVersion, [{ isActive: false }]) queueTableRows(schemaMock.workflow, [{ isDeployed: true }]) - mockIsDeploymentVersionProtectedByCurrentOperation.mockResolvedValue(true) + mockGetProtectedDeploymentVersionId.mockResolvedValue('version-2') const cleanupHandler = createWorkflowDeploymentOutboxHandlers()[ WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS ] - await cleanupHandler( - { + await expect( + cleanupHandler( + { + workflowId: 'workflow-1', + deploymentVersionIds: ['version-2'], + userId: 'user-1', + requestId: 'request-1', + }, + context() + ) + ).resolves.toBeUndefined() + + expect(mockDeleteInactiveDeploymentSchedules).toHaveBeenCalledWith({ + workflowId: 'workflow-1', + operationFence: undefined, + }) + expect(mockCleanupInactiveDeploymentWebhooks).toHaveBeenCalledWith( + expect.objectContaining({ workflowId: 'workflow-1', - deploymentVersionIds: ['version-2'], - userId: 'user-1', - requestId: 'request-1', - }, - context() + protectedDeploymentVersionId: 'version-2', + limit: 20, + }) ) - expect(mockCleanupWebhooksForWorkflow).not.toHaveBeenCalled() expect(mockCreateSchedulesForDeploy).not.toHaveBeenCalled() }) + + it('continues undeploy cleanup through the outbox before touching MCP tools', async () => { + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + mockCleanupInactiveDeploymentWebhooks.mockResolvedValueOnce({ hasMore: true }) + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS + ] + + await expect( + cleanupHandler({ workflowId: 'workflow-1', userId: 'user-1' }, context()) + ).resolves.toEqual({ + outcome: 'deferred', + reason: expect.any(String), + consumeAttempt: false, + }) + + expect(mockNotifyMcpToolServers).not.toHaveBeenCalled() + }) + + it('lets a timed-out undeploy stop between null-version webhooks', async () => { + queueTableRows(schemaMock.workflow, [ + { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' }, + ]) + queueTableRows(schemaMock.workflow, [{ isDeployed: false }]) + const controller = new AbortController() + const cleanupHandler = + createWorkflowDeploymentOutboxHandlers()[ + WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS.CLEANUP_UNDEPLOYED_SIDE_EFFECTS + ] + + await expect( + cleanupHandler({ workflowId: 'workflow-1', userId: 'user-1' }, context(controller)) + ).resolves.toBeUndefined() + + expect(mockCleanupWebhooksForWorkflow).toHaveBeenCalledTimes(1) + const shouldDeleteWebhook = mockCleanupWebhooksForWorkflow.mock + .calls[0][6] as () => Promise + queueTableRows(schemaMock.workflow, [{ isDeployed: false }]) + await expect(shouldDeleteWebhook()).resolves.toBe(true) + + controller.abort() + await expect(shouldDeleteWebhook()).rejects.toMatchObject({ name: 'AbortError' }) + }) }) diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 62875063496..f3462161392 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -3,10 +3,12 @@ import type { PrincipalActor } from '@sim/auth/principal' import { db, workflowDeploymentVersion, workflow as workflowTable } from '@sim/db' import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { and, eq, ne } from 'drizzle-orm' +import { and, eq } from 'drizzle-orm' import { NextRequest } from 'next/server' import { env } from '@/lib/core/config/env' import { + continueOutboxHandler, + type DeferredOutboxHandlerResult, enqueueOutboxEvent, type OutboxEventContext, type OutboxHandler, @@ -24,6 +26,7 @@ import { } from '@/lib/mcp/workflow-mcp-sync' import { captureServerEvent } from '@/lib/posthog/server' import { + cleanupInactiveDeploymentWebhooks, cleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy, saveTriggerWebhooksForDeploy, @@ -44,7 +47,9 @@ import { beginDeploymentOperationActivation, type DeploymentOperationGeneration, getDeploymentOperation, + getProtectedDeploymentVersionId, isDeploymentOperationCurrent, + isDeploymentVersionActive, isDeploymentVersionProtectedByCurrentOperation, markDeploymentComponentReadiness, markDeploymentOperationFailed, @@ -52,7 +57,11 @@ import { setDeploymentTxTimeouts, type WorkflowDeploymentOperation, } from '@/lib/workflows/persistence/deployment-operations' -import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from '@/lib/workflows/schedules' +import { + createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, + deleteSchedulesForWorkflow, +} from '@/lib/workflows/schedules' import { emitWorkflowDeployedEvent } from '@/lib/workspace-events/emitter' import type { BlockState } from '@/stores/workflows/workflow/types' @@ -77,6 +86,16 @@ export const DEPLOYMENT_READINESS_COMPONENTS = ['webhooks', 'schedules', 'mcp'] */ const DEPLOYMENT_PREPARATION_MAX_ATTEMPTS = 4 +/** + * Webhooks retired per outbox attempt when cleaning up inactive deployment + * versions. Each costs a provider call, so the batch keeps one attempt well + * inside the handler timeout; the handler continues through the outbox while + * rows remain. + */ +const INACTIVE_WEBHOOK_CLEANUP_BATCH_SIZE = 20 + +const INACTIVE_CLEANUP_CONTINUATION_REASON = 'Continuing inactive deployment side-effect cleanup' + interface DeploymentPreparationCheckpoints { webhooksPrepared?: boolean schedulesPrepared?: boolean @@ -132,7 +151,12 @@ interface SyncActiveSideEffectsPayload { interface CleanupUndeployedSideEffectsPayload { workflowId: string - deploymentVersionIds: string[] + /** + * Versions the undeploy retired. Cleanup finds stale rows from the versions' + * current state instead; kept for one release so events written by earlier + * pods still parse, and events written here still parse on them. + */ + deploymentVersionIds?: string[] userId: string requestId?: string } @@ -249,7 +273,7 @@ function createPrepareDeploymentHandler( return async (rawPayload, context) => { const payload = parsePrepareDeploymentV2Payload(rawPayload) try { - await prepareDeploymentOperation(payload, context, prepareWebhooks) + return await prepareDeploymentOperation(payload, context, prepareWebhooks) } catch (error) { const isFinalAttempt = context.attempts + 1 >= context.maxAttempts if (isNonRetryableDeploymentError(error) || isFinalAttempt) { @@ -305,7 +329,7 @@ async function prepareDeploymentOperation( payload: PrepareDeploymentV2Payload, context: OutboxEventContext, prepareWebhooks: PrepareDeploymentWebhooksHook -): Promise { +): Promise { context.signal.throwIfAborted() let operation = await getDeploymentOperation(payload) context.signal.throwIfAborted() @@ -338,7 +362,7 @@ async function prepareDeploymentOperation( * whatever else has started since, while only the fenced cleanup is * skipped. */ - await runPostActivationWork({ + return runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -346,7 +370,6 @@ async function prepareDeploymentOperation( checkpoint, context, }) - return } if (operation.status !== 'preparing' && operation.status !== 'activating') return @@ -488,7 +511,7 @@ async function prepareDeploymentOperation( notifyMcpToolServers(affectedMcpServers) context.signal.throwIfAborted() - await runPostActivationWork({ + return runPostActivationWork({ payload, operation, workflow: workflowRecord as Record, @@ -517,6 +540,10 @@ async function prepareDeploymentOperation( * them on the same predicate would drop them for good in the window where a * newer generation exists but has not activated — this activation is still * the live one there, and nothing else will emit them. + * + * Inactive-version cleanup is bounded per attempt. While rows remain, the + * handler yields a continuation so the outbox re-runs it without spending an + * attempt; the notifications above are checkpointed and never repeat. */ async function runPostActivationWork(params: { payload: PrepareDeploymentV2Payload @@ -525,20 +552,21 @@ async function runPostActivationWork(params: { checkpoints: DeploymentPreparationCheckpoints checkpoint: (patch: Partial) => Promise context: OutboxEventContext -}): Promise { +}): Promise { await emitPostActivationSideEffects(params) await cleanupRetiredWebhooksForOperation({ payload: params.payload, workflow: params.workflow, context: params.context, }) - await cleanupInactiveDeploymentsForOperation({ + const cleanupComplete = await cleanupInactiveDeploymentsForOperation({ payload: params.payload, workflow: params.workflow, checkpoints: params.checkpoints, checkpoint: params.checkpoint, context: params.context, }) + return cleanupComplete ? undefined : continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) } async function prepareReadinessComponent(params: { @@ -622,14 +650,19 @@ async function cleanupRetiredWebhooksForOperation(params: { }) } +/** + * Returns false while inactive-version cleanup still has rows to retire, so + * the caller yields a continuation instead of completing the event. A + * superseded attempt returns true: the newer generation owns the cleanup now. + */ async function cleanupInactiveDeploymentsForOperation(params: { payload: PrepareDeploymentV2Payload workflow: Record checkpoints: DeploymentPreparationCheckpoints checkpoint: (patch: Partial) => Promise context: OutboxEventContext -}): Promise { - if (params.checkpoints.inactiveCleanupCompleted) return +}): Promise { + if (params.checkpoints.inactiveCleanupCompleted) return true const operationFence = { workflowId: params.payload.workflowId, operationId: params.payload.operationId, @@ -644,18 +677,18 @@ async function cleanupInactiveDeploymentsForOperation(params: { return isCurrent } - if (!(await shouldContinue())) return - await cleanupInactiveDeploymentVersions({ + if (!(await shouldContinue())) return true + const { complete } = await cleanupInactiveDeploymentSideEffects({ workflowId: params.payload.workflowId, - activeDeploymentVersionId: params.payload.deploymentVersionId, workflow: params.workflow, - userId: params.payload.userId, requestId: params.payload.requestId, shouldContinue, operationFence, }) - if (!(await shouldContinue())) return + if (!(await shouldContinue())) return true + if (!complete) return false await params.checkpoint({ inactiveCleanupCompleted: true }) + return true } async function emitPostActivationSideEffects(params: { @@ -898,9 +931,10 @@ const syncActiveSideEffects = async (rawPayload: unknown): Promise => { }) } -const cleanupInactiveSideEffects = async (rawPayload: unknown): Promise => { +const cleanupInactiveSideEffects: OutboxHandler = async (rawPayload, context) => { const payload = parseCleanupInactiveSideEffectsPayload(rawPayload) const requestId = payload.requestId ?? generateRequestId() + context.signal.throwIfAborted() const [workflowRecord] = await db .select() .from(workflowTable) @@ -909,18 +943,19 @@ const cleanupInactiveSideEffects = async (rawPayload: unknown): Promise => if (!workflowRecord) return - await cleanupInactiveDeploymentVersions({ + const { complete } = await cleanupInactiveDeploymentSideEffects({ workflowId: payload.workflowId, - activeDeploymentVersionId: payload.activeDeploymentVersionId, workflow: workflowRecord as Record, - userId: payload.userId, requestId, + shouldContinue: unlessAborted(context.signal), }) + if (!complete) return continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) } -const cleanupUndeployedSideEffects = async (rawPayload: unknown): Promise => { +const cleanupUndeployedSideEffects: OutboxHandler = async (rawPayload, context) => { const payload = parseCleanupUndeployedSideEffectsPayload(rawPayload) const requestId = payload.requestId ?? generateRequestId() + context.signal.throwIfAborted() const [workflowRecord] = await db .select() .from(workflowTable) @@ -930,47 +965,45 @@ const cleanupUndeployedSideEffects = async (rawPayload: unknown): Promise if (!workflowRecord) return const workflowData = workflowRecord as Record - for (const deploymentVersionId of payload.deploymentVersionIds) { - const [versionRow] = await db - .select({ isActive: workflowDeploymentVersion.isActive }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, payload.workflowId), - eq(workflowDeploymentVersion.id, deploymentVersionId) - ) - ) - .limit(1) - - if (!versionRow || versionRow.isActive) continue - await cleanupDeploymentVersionIfInactive({ - workflowId: payload.workflowId, - workflow: workflowData, - userId: payload.userId, - requestId, - deploymentVersionId, - }) - } + const { complete } = await cleanupInactiveDeploymentSideEffects({ + workflowId: payload.workflowId, + workflow: workflowData, + requestId, + shouldContinue: unlessAborted(context.signal), + }) + if (!complete) return continueOutboxHandler(INACTIVE_CLEANUP_CONTINUATION_REASON) + context.signal.throwIfAborted() await cleanupNullVersionWebhooksIfStillUndeployed({ workflowId: payload.workflowId, workflow: workflowData, requestId, + signal: context.signal, }) + context.signal.throwIfAborted() await removeMcpToolsIfStillUndeployed(payload.workflowId, requestId) } +/** Continuation gate for handlers without an operation fence: stops only when the lease aborts. */ +function unlessAborted(signal: AbortSignal): () => Promise { + return async () => { + signal.throwIfAborted() + return true + } +} + /** * Run inactive-version cleanup synchronously as part of the active-version sync, right * after the active version's webhooks/schedules are registered. * - * {@link cleanupInactiveDeploymentVersions} re-checks that each version is still inactive - * before tearing anything down, so it can never touch the now-active version. Running it - * inline — rather than only enqueueing it — closes the window where a lost + * {@link cleanupInactiveDeploymentSideEffects} only selects rows whose version is inactive and + * re-checks each webhook right before its delete, so it can never touch the now-active version. + * Running it inline — rather than only enqueueing it — closes the window where a lost * `CLEANUP_INACTIVE` outbox event leaves superseded webhooks behind as live-but-never-polled - * `is_active` orphans. The deferred event is kept as a fallback so cleanup still retries if - * the inline pass throws, without failing the already-succeeded registration. + * `is_active` orphans. The deferred event is kept as a fallback so cleanup still continues if + * the inline pass throws or has more rows than one bounded pass retires, without failing the + * already-succeeded registration. */ async function syncInactiveDeploymentCleanup(params: { workflowId: string @@ -980,57 +1013,64 @@ async function syncInactiveDeploymentCleanup(params: { requestId: string }): Promise { try { - await cleanupInactiveDeploymentVersions(params) + const { complete } = await cleanupInactiveDeploymentSideEffects({ + workflowId: params.workflowId, + workflow: params.workflow, + requestId: params.requestId, + }) + if (complete) return + logger.info( + `[${params.requestId}] Inline inactive-deployment cleanup has more rows; continuing through the outbox` + ) } catch (cleanupError) { logger.warn( `[${params.requestId}] Inline inactive-deployment cleanup failed; deferring to outbox retry`, cleanupError ) - await enqueueWorkflowInactiveDeploymentCleanup(db, { - workflowId: params.workflowId, - activeDeploymentVersionId: params.activeDeploymentVersionId, - userId: params.userId, - requestId: params.requestId, - }) } + await enqueueWorkflowInactiveDeploymentCleanup(db, { + workflowId: params.workflowId, + activeDeploymentVersionId: params.activeDeploymentVersionId, + userId: params.userId, + requestId: params.requestId, + }) } -async function cleanupInactiveDeploymentVersions(params: { +/** + * Retires schedules and webhooks still owned by inactive deployment versions + * of the workflow. Work is keyed by side-effect rows, never by versions, so a + * workflow deployed hundreds of times costs no more than one deployed twice. + * Schedules go in one fenced statement; webhooks need a provider call each + * and drain in bounded batches, with `complete: false` asking the caller to + * run again. `shouldContinue` gates every step and throws once the outbox + * lease is aborted. + */ +async function cleanupInactiveDeploymentSideEffects(params: { workflowId: string - activeDeploymentVersionId: string workflow: Record - userId: string requestId: string shouldContinue?: () => Promise operationFence?: DeploymentCleanupOperationFence -}): Promise { - if (params.shouldContinue && !(await params.shouldContinue())) return - const inactiveVersions = await db - .select({ id: workflowDeploymentVersion.id }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, params.workflowId), - ne(workflowDeploymentVersion.id, params.activeDeploymentVersionId), - eq(workflowDeploymentVersion.isActive, false) - ) - ) +}): Promise<{ complete: boolean }> { + if (params.shouldContinue && !(await params.shouldContinue())) return { complete: false } - for (const version of inactiveVersions) { - if (params.shouldContinue && !(await params.shouldContinue())) return - if (await isDeploymentVersionProtectedByCurrentOperation(params.workflowId, version.id)) { - continue - } - await cleanupDeploymentVersionIfInactive({ - workflowId: params.workflowId, - workflow: params.workflow, - userId: params.userId, - requestId: params.requestId, - deploymentVersionId: version.id, - shouldContinue: params.shouldContinue, - operationFence: params.operationFence, - }) - } + const schedules = await deleteInactiveDeploymentSchedules({ + workflowId: params.workflowId, + operationFence: params.operationFence, + }) + if (schedules.status === 'superseded') return { complete: false } + + if (params.shouldContinue && !(await params.shouldContinue())) return { complete: false } + const protectedDeploymentVersionId = await getProtectedDeploymentVersionId(params.workflowId) + const { hasMore } = await cleanupInactiveDeploymentWebhooks({ + workflowId: params.workflowId, + workflow: params.workflow, + requestId: params.requestId, + protectedDeploymentVersionId, + limit: INACTIVE_WEBHOOK_CLEANUP_BATCH_SIZE, + shouldContinue: params.shouldContinue, + }) + return { complete: !hasMore } } async function cleanupDeploymentVersionIfInactive(params: { @@ -1155,25 +1195,6 @@ async function cleanupStaleDeploymentIfNeeded(params: { return false } -async function isDeploymentVersionActive( - workflowId: string, - deploymentVersionId: string -): Promise { - const [versionRow] = await db - .select({ id: workflowDeploymentVersion.id }) - .from(workflowDeploymentVersion) - .where( - and( - eq(workflowDeploymentVersion.workflowId, workflowId), - eq(workflowDeploymentVersion.id, deploymentVersionId), - eq(workflowDeploymentVersion.isActive, true) - ) - ) - .limit(1) - - return Boolean(versionRow) -} - async function removeMcpToolsIfStillUndeployed( workflowId: string, requestId: string @@ -1194,12 +1215,18 @@ async function removeMcpToolsIfStillUndeployed( notifyMcpToolServers(tools) } +/** + * The per-row gate also throws once the outbox lease aborts, so a timed-out + * undeploy stops between webhooks instead of overlapping its reaped retry. + */ async function cleanupNullVersionWebhooksIfStillUndeployed(params: { workflowId: string workflow: Record requestId: string + signal: AbortSignal }): Promise { const isStillUndeployed = async () => { + params.signal.throwIfAborted() const [workflowRecord] = await db .select({ isDeployed: workflowTable.isDeployed }) .from(workflowTable) @@ -1458,7 +1485,7 @@ function parseCleanupUndeployedSideEffectsPayload( const record = parsePayloadRecord(payload) const workflowId = parseRequiredString(record.workflowId, 'workflowId') const userId = parseRequiredString(record.userId, 'userId') - const deploymentVersionIds = parseRequiredStringArray( + const deploymentVersionIds = parseOptionalStringArray( record.deploymentVersionIds, 'deploymentVersionIds' ) @@ -1467,7 +1494,12 @@ function parseCleanupUndeployedSideEffectsPayload( ? record.requestId : undefined - return { workflowId, deploymentVersionIds, userId, requestId } + return { + workflowId, + ...(deploymentVersionIds ? { deploymentVersionIds } : {}), + userId, + requestId, + } } function parseCleanupInactiveSideEffectsPayload( @@ -1509,12 +1541,13 @@ function parseRequiredPositiveInteger(value: unknown, fieldName: string): number return value } -function parseRequiredStringArray(value: unknown, fieldName: string): string[] { +function parseOptionalStringArray(value: unknown, fieldName: string): string[] | undefined { + if (value === undefined) return undefined if ( !Array.isArray(value) || value.some((item) => typeof item !== 'string' || item.length === 0) ) { - throw new Error(`Deployment outbox payload is missing ${fieldName}`) + throw new Error(`Deployment outbox payload has an invalid ${fieldName}`) } return value } diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts index 757e54eb2e8..1a0ba58f210 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.test.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.test.ts @@ -21,6 +21,7 @@ vi.mock('@sim/utils/id', () => ({ import { activateDeploymentOperation, + getProtectedDeploymentVersionId, markDeploymentComponentReadiness, markDeploymentOperationFailed, prepareWorkflowDeployment, @@ -489,3 +490,30 @@ describe('deployment operation persistence', () => { expect(dbChainMockFns.update).not.toHaveBeenCalledWith(schemaMock.workflow) }) }) + +describe('getProtectedDeploymentVersionId', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('returns the version the in-flight current operation is preparing', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { deploymentVersionId: 'version-3', protocolVersion: 2, status: 'preparing' }, + ]) + + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBe('version-3') + }) + + it('protects nothing once the latest operation is terminal', async () => { + dbChainMockFns.limit.mockResolvedValueOnce([ + { deploymentVersionId: 'version-3', protocolVersion: 2, status: 'active' }, + ]) + + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBeNull() + }) + + it('protects nothing for a workflow without operations', async () => { + await expect(getProtectedDeploymentVersionId(WORKFLOW_ID)).resolves.toBeNull() + }) +}) diff --git a/apps/sim/lib/workflows/persistence/deployment-operations.ts b/apps/sim/lib/workflows/persistence/deployment-operations.ts index 9146f35b601..3d5a95517a6 100644 --- a/apps/sim/lib/workflows/persistence/deployment-operations.ts +++ b/apps/sim/lib/workflows/persistence/deployment-operations.ts @@ -70,6 +70,15 @@ export interface DeploymentOperationGeneration { generation: number } +/** + * Identifies the operation a fenced step belongs to, optionally narrowed to + * the version it targets and the statuses it may currently hold. + */ +export type DeploymentOperationFence = DeploymentOperationGeneration & { + deploymentVersionId?: string + statuses?: readonly DeploymentOperationStatus[] +} + export interface WorkflowDeploymentStatus { activeDeployment: { deploymentVersionId: string @@ -282,10 +291,7 @@ export async function getDeploymentOperation( * Confirms an operation still owns the workflow's latest generation. */ export async function isDeploymentOperationCurrent( - params: DeploymentOperationGeneration & { - deploymentVersionId?: string - statuses?: readonly DeploymentOperationStatus[] - }, + params: DeploymentOperationFence, executor: Pick = db ): Promise { const [latestOperation] = await executor @@ -322,6 +328,19 @@ export async function isDeploymentVersionProtectedByCurrentOperation( deploymentVersionId: string, executor: Pick = db ): Promise { + return (await getProtectedDeploymentVersionId(workflowId, executor)) === deploymentVersionId +} + +/** + * The deployment version the current operation is still preparing, or null + * once the latest operation is terminal. Cleanup must leave this version + * alone: it is inactive until cutover, yet its schedules and webhook + * candidates are live preparation state. + */ +export async function getProtectedDeploymentVersionId( + workflowId: string, + executor: Pick = db +): Promise { const [latestOperation] = await executor .select({ deploymentVersionId: workflowDeploymentOperation.deploymentVersionId, @@ -333,12 +352,42 @@ export async function isDeploymentVersionProtectedByCurrentOperation( .orderBy(desc(workflowDeploymentOperation.generation)) .limit(1) - return ( - latestOperation?.deploymentVersionId === deploymentVersionId && - latestOperation.protocolVersion === DEPLOYMENT_OPERATION_PROTOCOL_VERSION && - isDeploymentOperationStatus(latestOperation.status) && - IN_FLIGHT_STATUSES.includes(latestOperation.status) - ) + if ( + !latestOperation || + latestOperation.protocolVersion !== DEPLOYMENT_OPERATION_PROTOCOL_VERSION || + !isDeploymentOperationStatus(latestOperation.status) || + !IN_FLIGHT_STATUSES.includes(latestOperation.status) + ) { + return null + } + return latestOperation.deploymentVersionId +} + +/** + * True when the given deployment version is the workflow's active one. + * Cleanup re-checks this immediately before any provider teardown because a + * version can be re-activated between a batch being selected and its rows + * being processed, and the fenced row delete that follows cannot undo a + * provider call. + */ +export async function isDeploymentVersionActive( + workflowId: string, + deploymentVersionId: string, + executor: Pick = db +): Promise { + const [versionRow] = await executor + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, workflowId), + eq(workflowDeploymentVersion.id, deploymentVersionId), + eq(workflowDeploymentVersion.isActive, true) + ) + ) + .limit(1) + + return Boolean(versionRow) } /** diff --git a/apps/sim/lib/workflows/schedules/deploy.test.ts b/apps/sim/lib/workflows/schedules/deploy.test.ts index afb5dc2abdc..62421cc8fdf 100644 --- a/apps/sim/lib/workflows/schedules/deploy.test.ts +++ b/apps/sim/lib/workflows/schedules/deploy.test.ts @@ -3,12 +3,21 @@ * * @vitest-environment node */ -import { dbChainMock, dbChainMockFns, resetDbChainMock, schemaMock } from '@sim/testing' +import { + dbChainMock, + dbChainMockFns, + flattenMockConditions, + resetDbChainMock, + schemaMock, +} from '@sim/testing' import { afterAll, afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockRandomUUID } = vi.hoisted(() => ({ - mockRandomUUID: vi.fn(), -})) +const { mockRandomUUID, mockGetProtectedDeploymentVersionId, mockIsDeploymentOperationCurrent } = + vi.hoisted(() => ({ + mockRandomUUID: vi.fn(), + mockGetProtectedDeploymentVersionId: vi.fn(), + mockIsDeploymentOperationCurrent: vi.fn(), + })) vi.mock('@sim/db', () => ({ ...dbChainMock, ...schemaMock })) @@ -16,7 +25,17 @@ vi.mock('@/lib/webhooks/deploy', () => ({ cleanupWebhooksForWorkflow: vi.fn().mockResolvedValue(undefined), })) -import { createSchedulesForDeploy, deleteSchedulesForWorkflow } from './deploy' +vi.mock('@/lib/workflows/persistence/deployment-operations', () => ({ + getProtectedDeploymentVersionId: mockGetProtectedDeploymentVersionId, + isDeploymentOperationCurrent: mockIsDeploymentOperationCurrent, + setDeploymentTxTimeouts: vi.fn(), +})) + +import { + createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, + deleteSchedulesForWorkflow, +} from './deploy' import type { BlockState } from './utils' import * as scheduleUtils from './utils' import { findScheduleBlocks, validateScheduleBlock, validateWorkflowSchedules } from './validation' @@ -795,3 +814,66 @@ describe('Schedule Deploy Utilities', () => { }) }) }) + +describe('deleteInactiveDeploymentSchedules', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mockIsDeploymentOperationCurrent.mockResolvedValue(true) + mockGetProtectedDeploymentVersionId.mockResolvedValue(null) + }) + + it('deletes every schedule owned by an inactive version in one statement', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'schedule-1' }, { id: 'schedule-2' }]) + + await expect(deleteInactiveDeploymentSchedules({ workflowId: 'workflow-1' })).resolves.toEqual({ + status: 'deleted', + count: 2, + }) + + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + const conditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) + expect(conditions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'inArray', + column: schemaMock.workflowSchedule.deploymentVersionId, + }), + expect.objectContaining({ type: 'isNull', column: schemaMock.workflowSchedule.archivedAt }), + ]) + ) + expect(conditions).not.toEqual( + expect.arrayContaining([expect.objectContaining({ type: 'ne' })]) + ) + }) + + it('shields the version an in-flight operation is preparing', async () => { + mockGetProtectedDeploymentVersionId.mockResolvedValue('version-3') + + await deleteInactiveDeploymentSchedules({ workflowId: 'workflow-1' }) + + const conditions = flattenMockConditions(dbChainMockFns.where.mock.calls.at(-1)?.[0]) + expect(conditions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: 'ne', + left: schemaMock.workflowSchedule.deploymentVersionId, + right: 'version-3', + }), + ]) + ) + }) + + it('deletes nothing once a newer operation owns the workflow', async () => { + mockIsDeploymentOperationCurrent.mockResolvedValue(false) + + await expect( + deleteInactiveDeploymentSchedules({ + workflowId: 'workflow-1', + operationFence: { workflowId: 'workflow-1', operationId: 'operation-1', generation: 2 }, + }) + ).resolves.toEqual({ status: 'superseded' }) + + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/workflows/schedules/deploy.ts b/apps/sim/lib/workflows/schedules/deploy.ts index 1f2a1bfd662..475212ffd23 100644 --- a/apps/sim/lib/workflows/schedules/deploy.ts +++ b/apps/sim/lib/workflows/schedules/deploy.ts @@ -1,9 +1,15 @@ -import { db, workflowSchedule } from '@sim/db' +import { db, workflow, workflowDeploymentVersion, workflowSchedule } from '@sim/db' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' -import { and, eq, inArray, isNull } from 'drizzle-orm' +import { and, eq, inArray, isNull, ne } from 'drizzle-orm' import type { DbOrTx } from '@/lib/db/types' +import { + type DeploymentOperationFence, + getProtectedDeploymentVersionId, + isDeploymentOperationCurrent, + setDeploymentTxTimeouts, +} from '@/lib/workflows/persistence/deployment-operations' import type { BlockState } from '@/lib/workflows/schedules/utils' import { findScheduleBlocks, validateScheduleBlock } from '@/lib/workflows/schedules/validation' @@ -203,3 +209,68 @@ export async function deleteSchedulesForWorkflow( : `Deleted all schedules for workflow ${workflowId}` ) } + +export type InactiveDeploymentScheduleCleanupResult = + | { status: 'deleted'; count: number } + | { status: 'superseded' } + +/** + * Deletes every schedule still owned by an inactive deployment version of the + * workflow in one fenced statement. Keyed by schedule rows rather than by + * versions, so the cost follows what is stale instead of how many times the + * workflow has been deployed. The version an in-flight operation is preparing + * is left alone: it is inactive until cutover, but its schedules are live + * preparation state. The workflow row lock serializes this with activation so + * `isActive` cannot flip underneath the delete. + */ +export async function deleteInactiveDeploymentSchedules(params: { + workflowId: string + /** When set, nothing is deleted once a newer operation has taken over the workflow. */ + operationFence?: DeploymentOperationFence +}): Promise { + return db.transaction(async (tx) => { + await setDeploymentTxTimeouts(tx) + await tx + .select({ id: workflow.id }) + .from(workflow) + .where(eq(workflow.id, params.workflowId)) + .for('update') + if (params.operationFence && !(await isDeploymentOperationCurrent(params.operationFence, tx))) { + return { status: 'superseded' } + } + + const protectedDeploymentVersionId = await getProtectedDeploymentVersionId( + params.workflowId, + tx + ) + const inactiveVersionIds = tx + .select({ id: workflowDeploymentVersion.id }) + .from(workflowDeploymentVersion) + .where( + and( + eq(workflowDeploymentVersion.workflowId, params.workflowId), + eq(workflowDeploymentVersion.isActive, false) + ) + ) + const deleted = await tx + .delete(workflowSchedule) + .where( + and( + eq(workflowSchedule.workflowId, params.workflowId), + isNull(workflowSchedule.archivedAt), + inArray(workflowSchedule.deploymentVersionId, inactiveVersionIds), + protectedDeploymentVersionId + ? ne(workflowSchedule.deploymentVersionId, protectedDeploymentVersionId) + : undefined + ) + ) + .returning({ id: workflowSchedule.id }) + + if (deleted.length > 0) { + logger.info( + `Deleted ${deleted.length} schedule(s) owned by inactive deployments of workflow ${params.workflowId}` + ) + } + return { status: 'deleted', count: deleted.length } + }) +} diff --git a/apps/sim/lib/workflows/schedules/index.ts b/apps/sim/lib/workflows/schedules/index.ts index 67dd11fd2c8..026d35ecd8d 100644 --- a/apps/sim/lib/workflows/schedules/index.ts +++ b/apps/sim/lib/workflows/schedules/index.ts @@ -1,5 +1,7 @@ export { createSchedulesForDeploy, + deleteInactiveDeploymentSchedules, deleteSchedulesForWorkflow, + type InactiveDeploymentScheduleCleanupResult, } from './deploy' export { validateWorkflowSchedules } from './validation' diff --git a/apps/sim/providers/models.ts b/apps/sim/providers/models.ts index 83767ac68ef..1b09f47db67 100644 --- a/apps/sim/providers/models.ts +++ b/apps/sim/providers/models.ts @@ -1594,6 +1594,26 @@ export const PROVIDER_DEFINITIONS: Record = { icon: GeminiIcon, color: '#4285F4', models: [ + { + id: 'gemini-3.8-flash', + pricing: { + input: 0.75, + cachedInput: 0.075, + output: 3.75, + updatedAt: '2026-09-02', + }, + capabilities: { + temperature: { min: 0, max: 2 }, + thinking: { + levels: ['low', 'medium', 'high'], + default: 'medium', + }, + maxOutputTokens: 65536, + }, + contextWindow: 1048576, + releaseDate: '2026-09-02', + recommended: true, + }, { id: 'gemini-3.6-flash', pricing: { @@ -1612,7 +1632,6 @@ export const PROVIDER_DEFINITIONS: Record = { }, contextWindow: 1048576, releaseDate: '2026-07-21', - recommended: true, }, { id: 'gemini-3.5-flash-lite', diff --git a/apps/sim/public/library/what-is-human-in-the-loop-in-ai-agents/cover.jpg b/apps/sim/public/library/what-is-human-in-the-loop-in-ai-agents/cover.jpg new file mode 100644 index 00000000000..44fe6f1148a Binary files /dev/null and b/apps/sim/public/library/what-is-human-in-the-loop-in-ai-agents/cover.jpg differ diff --git a/packages/deployment-config/src/integrations.json b/packages/deployment-config/src/integrations.json index 1ce3e5cc2ee..0c2c80ea82b 100644 --- a/packages/deployment-config/src/integrations.json +++ b/packages/deployment-config/src/integrations.json @@ -1,5 +1,5 @@ { - "updatedAt": "2026-09-01", + "updatedAt": "2026-09-02", "integrations": [ { "type": "onepassword", @@ -21569,7 +21569,7 @@ "body": "Sim requests only the Slack permissions its actions and triggers need, and never shows private channel names or messages to people who are not members of those channels in Slack.", "href": "/privacy" }, - "aiDisclaimer": "Sim agents use AI models to generate messages and responses sent to Slack. AI-generated content can be inaccurate or incomplete, so review automated outputs before relying on them, especially for important communications." + "aiDisclaimer": "A paid Slack plan is required to use Sim's AI agent in Slack's app container. Other Sim features continue to work on free Slack plans. Sim agents use AI models to generate messages and responses sent to Slack. AI-generated content can be inaccurate or incomplete, so review automated outputs before relying on them, especially for important communications." } }, { diff --git a/scripts/openapi/documents.test.ts b/scripts/openapi/documents.test.ts index d2c31bd694b..c7c3b510584 100644 --- a/scripts/openapi/documents.test.ts +++ b/scripts/openapi/documents.test.ts @@ -279,6 +279,10 @@ describe('generated OpenAPI documents', () => { expect(tableProperties.ownerEmail).toMatchObject({ type: 'string', format: 'email' }) }) + it('omits feature-flagged table column types', () => { + expect(JSON.stringify(generatedDocument(tablesOpenApiDocument))).not.toContain('"ttl"') + }) + it('keeps billing as its own API reference group', () => { const spec = generatedDocument(billingOpenApiDocument) expect((spec.tags as JsonObject[]).map((tag) => tag.name)).toEqual(['Billing']) diff --git a/scripts/openapi/generator.test.ts b/scripts/openapi/generator.test.ts index b6b9b453fb9..14fbd811396 100644 --- a/scripts/openapi/generator.test.ts +++ b/scripts/openapi/generator.test.ts @@ -194,6 +194,35 @@ describe('OpenAPI generator', () => { }) }) + it('omits feature-flagged enum values from generated schemas', () => { + const columnType = z.enum(['string', 'ttl']).meta({ omitEnumValuesFromOpenApi: ['ttl'] }) + const body = z + .object({ type: columnType.describe('Column data type.') }) + .meta({ id: 'HiddenEnumRequest', title: 'Hidden enum request', description: 'Request body.' }) + const response = z + .object({ ok: z.boolean().describe('Whether the request succeeded.') }) + .meta({ id: 'HiddenEnumResponse', title: 'Hidden enum response', description: 'Response.' }) + const contract = defineRouteContract({ + method: 'POST', + path: '/hidden-enum', + body, + response: { mode: 'json', schema: response }, + }) + const route = defineOpenApiRoute( + contract, + operation('hiddenEnum', { description: 'Response.' }), + { body, response } + ) + const spec = generateOpenApiDocument(document([route])) + const schemas = (spec.components as JsonObject).schemas as JsonObject + const requestProperties = (schemas.HiddenEnumRequest as JsonObject).properties as JsonObject + const documentedColumnType = requestProperties.type as JsonObject + + expect(columnType.safeParse('ttl').success).toBe(true) + expect(documentedColumnType.enum).toEqual(['string']) + expect(documentedColumnType).not.toHaveProperty('omitEnumValuesFromOpenApi') + }) + it('handles every route response mode and media type', () => { const emptyContract = defineRouteContract({ method: 'DELETE', diff --git a/scripts/openapi/generator.ts b/scripts/openapi/generator.ts index 550113a1ce8..93fb3bb762b 100644 --- a/scripts/openapi/generator.ts +++ b/scripts/openapi/generator.ts @@ -105,6 +105,31 @@ function stripLegacySchemaIds(value: unknown): unknown { ) } +function omitEnumValuesFromOpenApi( + metadata: z.core.GlobalMeta | undefined, + schema: JsonObject, + label: string +): void { + const omittedValues = metadata?.omitEnumValuesFromOpenApi + if (omittedValues === undefined) return + + invariant( + Array.isArray(omittedValues) && omittedValues.length > 0, + `${label} omitEnumValuesFromOpenApi must be a non-empty array` + ) + invariant( + Array.isArray(schema.enum), + `${label} omitEnumValuesFromOpenApi requires an enum schema` + ) + for (const value of omittedValues) { + invariant(schema.enum.includes(value), `${label} omits an enum value that does not exist`) + } + + schema.enum = schema.enum.filter((value) => !omittedValues.includes(value)) + invariant(schema.enum.length > 0, `${label} cannot omit every enum value`) + Reflect.deleteProperty(schema, 'omitEnumValuesFromOpenApi') +} + function comparableSchema(schema: ApiSchema, io: SchemaIo): unknown { const cached = comparableSchemaCache.get(schema)?.get(io) if (cached) return cached @@ -206,14 +231,12 @@ function generateSchema( unrepresentable: 'any', cycles: 'ref', reused: 'inline', - override: ({ zodSchema, path }) => { + override: ({ zodSchema, jsonSchema, path }) => { const current = zodSchema as ApiSchema - validateExamples( - current, - z.globalRegistry.get(current)?.examples, - io, - `${label} at ${path.join('.') || ''}` - ) + const metadata = z.globalRegistry.get(current) + const schemaLabel = `${label} at ${path.join('.') || ''}` + validateExamples(current, metadata?.examples, io, schemaLabel) + omitEnumValuesFromOpenApi(metadata, jsonSchema as JsonObject, schemaLabel) }, }) as JsonObject const byIo = generatedSchemaCache.get(schema) ?? new Map()