diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 5ad6a83e6ebf..db3cbe5a7939 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -31,7 +31,7 @@ ENV LC_ALL en_US.UTF-8 COPY default.cf /usr/local/etc/isolate -RUN npm i -g bun@1.3.1 cross-env@7.0.3 esbuild@0.25.0 @anthropic-ai/claude-code +RUN npm i -g bun@1.4.0 cross-env@7.0.3 esbuild@0.25.0 @anthropic-ai/claude-code # install isolated-vm in a parent directory to avoid linking the package in every sandbox diff --git a/Dockerfile b/Dockerfile index c363f43dff2c..c3953f9b0270 100644 --- a/Dockerfile +++ b/Dockerfile @@ -34,9 +34,9 @@ RUN apt-get update && \ # Download, extract, and clean up bun in a single layer so the zip never ships RUN export ARCH=$(uname -m) && \ if [ "$ARCH" = "x86_64" ]; then \ - curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-x64-baseline.zip -o bun.zip; \ + curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-x64-baseline.zip -o bun.zip; \ elif [ "$ARCH" = "aarch64" ]; then \ - curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-aarch64.zip -o bun.zip; \ + curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-aarch64.zip -o bun.zip; \ fi && \ unzip bun.zip && \ mv bun-*/bun /usr/local/bin/bun && \ diff --git a/Dockerfile.worker b/Dockerfile.worker index 09d4a70b314b..430d373c7ebd 100644 --- a/Dockerfile.worker +++ b/Dockerfile.worker @@ -13,9 +13,9 @@ RUN apt-get update && \ RUN export ARCH=$(uname -m) && \ if [ "$ARCH" = "x86_64" ]; then \ - curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-x64-baseline.zip -o bun.zip; \ + curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-x64-baseline.zip -o bun.zip; \ elif [ "$ARCH" = "aarch64" ]; then \ - curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.3.1/bun-linux-aarch64.zip -o bun.zip; \ + curl -fSL --retry 5 --retry-delay 2 https://github.com/oven-sh/bun/releases/download/bun-v1.4.0/bun-linux-aarch64.zip -o bun.zip; \ fi && \ unzip bun.zip && mv bun-*/bun /usr/local/bin/bun && chmod +x /usr/local/bin/bun && rm -rf bun.zip bun-* diff --git a/docker-compose.yml b/docker-compose.yml index 116df299f7c9..83797579e55c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,6 +1,6 @@ services: app: - image: ghcr.io/activepieces/activepieces:0.88.3 + image: ghcr.io/activepieces/activepieces:0.89.0 container_name: activepieces-app restart: unless-stopped ports: @@ -16,7 +16,7 @@ services: networks: - activepieces worker: - image: ghcr.io/activepieces/activepieces:0.88.3 + image: ghcr.io/activepieces/activepieces:0.89.0 restart: unless-stopped depends_on: - app diff --git a/package.json b/package.json index 674dea7f9136..a52c64cdd3a8 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "activepieces", - "version": "0.88.3", - "packageManager": "bun@1.3.3", + "version": "0.89.0", + "packageManager": "bun@1.4.0", "trustedDependencies": [ "sqlite3", "isolated-vm", diff --git a/packages/core/execution/src/lib/flows/dto/create-flow-request.ts b/packages/core/execution/src/lib/flows/dto/create-flow-request.ts index 5ab97d9c5706..02df1d4c1646 100755 --- a/packages/core/execution/src/lib/flows/dto/create-flow-request.ts +++ b/packages/core/execution/src/lib/flows/dto/create-flow-request.ts @@ -1,5 +1,5 @@ import { z } from 'zod' -import { Metadata } from '@activepieces/core-utils' +import { formErrors, Metadata, SAFE_EXTERNAL_ID_PATTERN } from '@activepieces/core-utils' export const CreateFlowRequest = z.object({ displayName: z.string(), @@ -9,6 +9,7 @@ export const CreateFlowRequest = z.object({ projectId: z.string(), templateId: z.string().optional(), metadata: z.optional(Metadata), + externalId: z.string().regex(SAFE_EXTERNAL_ID_PATTERN, formErrors.invalidExternalId).optional(), }) export type CreateFlowRequest = z.infer diff --git a/packages/core/shared/package.json b/packages/core/shared/package.json index 3926a64a2316..100ceb051f08 100644 --- a/packages/core/shared/package.json +++ b/packages/core/shared/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/shared", - "version": "0.151.0", + "version": "0.152.0", "type": "commonjs", "sideEffects": false, "main": "./dist/src/index.js", diff --git a/packages/core/shared/src/lib/automation/tables/dto/tables.dto.ts b/packages/core/shared/src/lib/automation/tables/dto/tables.dto.ts index e203cfa3a80f..9d19e2ceb2bb 100644 --- a/packages/core/shared/src/lib/automation/tables/dto/tables.dto.ts +++ b/packages/core/shared/src/lib/automation/tables/dto/tables.dto.ts @@ -1,12 +1,10 @@ -import { Nullable, OptionalArrayFromQuery } from '@activepieces/core-utils' +import { Nullable, OptionalArrayFromQuery, SAFE_EXTERNAL_ID_PATTERN } from '@activepieces/core-utils' import { z } from 'zod' import { formErrors } from '../../../form-errors' import { FieldState } from '../../project-release/project-state' import { TableAutomationStatus, TableAutomationTrigger } from '../table' import { TableWebhookEventType } from '../table-webhook' -const SAFE_EXTERNAL_ID_PATTERN = /^(?!\.{1,2}$)[A-Za-z0-9._-]{1,128}$/ - export const CreateTableRequest = z.object({ projectId: z.string(), name: z.string(), diff --git a/packages/core/utils/src/lib/activepieces-error.ts b/packages/core/utils/src/lib/activepieces-error.ts index 6ebdcf67b2ee..56813bf56f2b 100644 --- a/packages/core/utils/src/lib/activepieces-error.ts +++ b/packages/core/utils/src/lib/activepieces-error.ts @@ -68,6 +68,7 @@ export type ApErrorParams = | InvalidGitCredentialsParams | InvalidReleaseTypeParams | ProjectExternalIdAlreadyExistsParams + | FlowExternalIdAlreadyExistsParams | SandboxMemoryIssueParams | SandboxExecutionTimeoutParams | SandboxInternalErrorParams @@ -426,6 +427,10 @@ export type ProjectExternalIdAlreadyExistsParams = BaseErrorParams +export type FlowExternalIdAlreadyExistsParams = BaseErrorParams + export type McpPieceRequiresConnectionParams = BaseErrorParams @@ -523,6 +528,7 @@ export enum ErrorCode { EXISTING_USER = 'EXISTING_USER', EXISTING_ALERT_CHANNEL = 'EXISTING_ALERT_CHANNEL', PROJECT_EXTERNAL_ID_ALREADY_EXISTS = 'PROJECT_EXTERNAL_ID_ALREADY_EXISTS', + FLOW_EXTERNAL_ID_ALREADY_EXISTS = 'FLOW_EXTERNAL_ID_ALREADY_EXISTS', FLOW_OPERATION_INVALID = 'FLOW_OPERATION_INVALID', FLOW_OPERATION_IN_PROGRESS = 'FLOW_OPERATION_IN_PROGRESS', FLOW_RUN_RETRY_OUTSIDE_RETENTION = 'FLOW_RUN_RETRY_OUTSIDE_RETENTION', diff --git a/packages/core/utils/src/lib/form-errors.ts b/packages/core/utils/src/lib/form-errors.ts index b4069f5bb449..28e3739e92a4 100644 --- a/packages/core/utils/src/lib/form-errors.ts +++ b/packages/core/utils/src/lib/form-errors.ts @@ -7,3 +7,5 @@ export const formErrors = { invalidFileName: 'invalidFileName', messageRequiresContentOrFiles: 'messageRequiresContentOrFiles', } as const + +export const SAFE_EXTERNAL_ID_PATTERN = /^(?!\.{1,2}$)[A-Za-z0-9._-]{1,128}$/ diff --git a/packages/server/api/src/app/database/migration/postgres/1839000000000-AddFlowProjectIdExternalIdUniqueIndex.ts b/packages/server/api/src/app/database/migration/postgres/1839000000000-AddFlowProjectIdExternalIdUniqueIndex.ts new file mode 100644 index 000000000000..979ee0c44ecd --- /dev/null +++ b/packages/server/api/src/app/database/migration/postgres/1839000000000-AddFlowProjectIdExternalIdUniqueIndex.ts @@ -0,0 +1,44 @@ +import { QueryRunner } from 'typeorm' +import { system } from '../../../helper/system/system' +import { AppSystemProp } from '../../../helper/system/system-props' +import { DatabaseType } from '../../database-type' +import { Migration } from '../../migration' + +export class AddFlowProjectIdExternalIdUniqueIndex1839000000000 implements Migration { + name = 'AddFlowProjectIdExternalIdUniqueIndex1839000000000' + breaking = false + release = '0.88.4' + transaction = false + + public async up(queryRunner: QueryRunner): Promise { + const concurrently = isPGlite() ? '' : 'CONCURRENTLY' + + await queryRunner.query(` + UPDATE "flow" + SET "externalId" = "id" + WHERE "id" IN ( + SELECT "id" FROM ( + SELECT "id", + ROW_NUMBER() OVER ( + PARTITION BY "projectId", "externalId" + ORDER BY "updated" DESC, "created" DESC, "id" DESC + ) AS rn + FROM "flow" + ) ranked + WHERE ranked.rn > 1 + ) + `) + + await queryRunner.query(` + CREATE UNIQUE INDEX ${concurrently} IF NOT EXISTS "idx_flow_project_id_external_id" + ON "flow" ("projectId", "externalId") + `) + } + + public async down(queryRunner: QueryRunner): Promise { + const concurrently = isPGlite() ? '' : 'CONCURRENTLY' + await queryRunner.query(`DROP INDEX ${concurrently} IF EXISTS "idx_flow_project_id_external_id"`) + } +} + +const isPGlite = (): boolean => system.get(AppSystemProp.DB_TYPE) === DatabaseType.PGLITE diff --git a/packages/server/api/src/app/database/postgres-connection.ts b/packages/server/api/src/app/database/postgres-connection.ts index f837132f1cb1..5fe030910a3b 100644 --- a/packages/server/api/src/app/database/postgres-connection.ts +++ b/packages/server/api/src/app/database/postgres-connection.ts @@ -431,6 +431,7 @@ import { WidenMcpOAuthState1835000000000 } from './migration/postgres/1835000000 import { DropTeamsBotInstallation1836000000000 } from './migration/postgres/1836000000000-DropTeamsBotInstallation' import { AddAiProviderStatus1837000000000 } from './migration/postgres/1837000000000-AddAiProviderStatus' import { AddMcpOAuthTokenLastUsedAndClientKey1838000000000 } from './migration/postgres/1838000000000-AddMcpOAuthTokenLastUsedAndClientKey' +import { AddFlowProjectIdExternalIdUniqueIndex1839000000000 } from './migration/postgres/1839000000000-AddFlowProjectIdExternalIdUniqueIndex' const getSslConfig = (): boolean | TlsOptions => { const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL) @@ -877,6 +878,7 @@ export const getMigrations = (): (new () => Migration)[] => { DropTeamsBotInstallation1836000000000, AddAiProviderStatus1837000000000, AddMcpOAuthTokenLastUsedAndClientKey1838000000000, + AddFlowProjectIdExternalIdUniqueIndex1839000000000, ] return migrations } diff --git a/packages/server/api/src/app/file/file.service.ts b/packages/server/api/src/app/file/file.service.ts index f3367b8b8440..2d0c322b4d19 100644 --- a/packages/server/api/src/app/file/file.service.ts +++ b/packages/server/api/src/app/file/file.service.ts @@ -69,6 +69,9 @@ export const fileService = (log: FastifyBaseLogger) => ({ return await fileRepo().save({ ...baseFile, location: FileLocation.S3, s3Key }) } catch (error) { + if (isNil(params.data)) { + throw error + } exceptionHandler.handle(error, log) return saveFileToDb(baseFile, params.data) } diff --git a/packages/server/api/src/app/flows/flow/flow.controller.ts b/packages/server/api/src/app/flows/flow/flow.controller.ts index 52f36b62129a..a704103373a4 100644 --- a/packages/server/api/src/app/flows/flow/flow.controller.ts +++ b/packages/server/api/src/app/flows/flow/flow.controller.ts @@ -28,6 +28,7 @@ export const flowController: FastifyPluginAsyncZod = async (app) => { ownerId: actorUserId(request), templateId: request.body.templateId, ip: networkUtils.clientIp(request), + externalId: request.body.externalId, }) return reply.status(StatusCodes.CREATED).send(newFlow) diff --git a/packages/server/api/src/app/flows/flow/flow.entity.ts b/packages/server/api/src/app/flows/flow/flow.entity.ts index 7ddcc4b53bbc..475f3ae8d689 100644 --- a/packages/server/api/src/app/flows/flow/flow.entity.ts +++ b/packages/server/api/src/app/flows/flow/flow.entity.ts @@ -101,6 +101,11 @@ export const FlowEntity = new EntitySchema({ columns: ['projectId', 'status'], unique: false, }, + { + name: 'idx_flow_project_id_external_id', + columns: ['projectId', 'externalId'], + unique: true, + }, ], relations: { runs: { diff --git a/packages/server/api/src/app/flows/flow/flow.service.ts b/packages/server/api/src/app/flows/flow/flow.service.ts index a5649da2fae6..8adfb78ae930 100644 --- a/packages/server/api/src/app/flows/flow/flow.service.ts +++ b/packages/server/api/src/app/flows/flow/flow.service.ts @@ -30,6 +30,7 @@ import { flowRepo } from './flow.repo' export const flowService = (log: FastifyBaseLogger) => ({ async create({ projectId, request, externalId, ownerId, templateId, createdBy, ip, emitEvents = true }: CreateParams): Promise { + await assertExternalIdIsUnique({ projectId, externalId }) const folderId = await getFolderIdFromRequest({ projectId, folderId: request.folderId, folderName: request.folderName, log }) const newFlow: NewFlow = { id: apId(), @@ -802,6 +803,19 @@ const assertFlowIsNotNull: ( } } +async function assertExternalIdIsUnique({ projectId, externalId }: { projectId: ProjectId, externalId: string | undefined }): Promise { + if (isNil(externalId)) { + return + } + const exists = await flowRepo().existsBy({ projectId, externalId }) + if (exists) { + throw new ActivepiecesError({ + code: ErrorCode.FLOW_EXTERNAL_ID_ALREADY_EXISTS, + params: { externalId }, + }) + } +} + type CreateParams = EventEmissionParams & { projectId: ProjectId request: CreateFlowRequest diff --git a/packages/server/api/src/app/helper/error-handler.ts b/packages/server/api/src/app/helper/error-handler.ts index 236c50a70ea6..269ee6781f83 100644 --- a/packages/server/api/src/app/helper/error-handler.ts +++ b/packages/server/api/src/app/helper/error-handler.ts @@ -91,6 +91,7 @@ const statusCodeMap: Partial> = { [ErrorCode.AUTHORIZATION]: StatusCodes.FORBIDDEN, [ErrorCode.SIGN_UP_DISABLED]: StatusCodes.FORBIDDEN, [ErrorCode.PROJECT_EXTERNAL_ID_ALREADY_EXISTS]: StatusCodes.CONFLICT, + [ErrorCode.FLOW_EXTERNAL_ID_ALREADY_EXISTS]: StatusCodes.CONFLICT, [ErrorCode.INVALID_CREDENTIALS]: StatusCodes.UNAUTHORIZED, [ErrorCode.SESSION_EXPIRED]: StatusCodes.FORBIDDEN, [ErrorCode.EMAIL_IS_NOT_VERIFIED]: StatusCodes.FORBIDDEN, diff --git a/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts b/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts index 19024dcc7668..ae8f91cc51da 100644 --- a/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts +++ b/packages/server/api/src/app/workers/rpc/worker-rpc-service.ts @@ -3,7 +3,7 @@ import { apVersionUtil, onCallService, UNKNOWN_VERSION } from '@activepieces/ser import { ExecutionType, FileCompression, FileLocation, FileType, FlowOperationType, FlowStatus, WebsocketClientEvent, WorkerGroupScope, WorkerToApiContract } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { websocketService } from '../../core/websockets.service' -import { redisConnections } from '../../database/redis-connections' +import { distributedStore, redisConnections } from '../../database/redis-connections' import { agentRpcHandlers } from '../../ee/agent/agent-rpc-handlers' import { chatPersonalizationService } from '../../ee/agent/personalization/chat-personalization-service' import { fileService, getLocationForFile } from '../../file/file.service' @@ -29,6 +29,10 @@ import { getPlatformGroupQueueName, getProjectGroupQueueName, QueueName, WorkerG import { jobBroker } from '../job-queue/job-broker' import { machineService } from '../machine/machine-service' +const FLOW_BUNDLE_PUBLISH_CLAIM_TTL_SECONDS = 60 + +const getFlowBundlePublishClaimKey = (flowVersionId: string): string => `flow_bundle_publish_claim:${flowVersionId}` + const getPollQueueName = (assignment: WorkerGroupAssignment | null): string => { if (isNil(assignment)) { return QueueName.WORKER_JOBS @@ -237,6 +241,16 @@ export function createHandlers(log: FastifyBaseLogger, assignment: WorkerGroupAs if (getLocationForFile(FileType.FLOW_BUNDLE) !== FileLocation.S3) { return { kind: 'skip' } } + // The bundle for a flowVersionId is immutable, and a burst of workers can + // finish provisioning the same version at once. Only the first claimer + // publishes; losing the claim is a normal outcome, not an error. Without + // this, concurrent callers race the deterministic file PK (duplicate key + // on the file table) and hammer the same S3 key (503 Slow Down). The + // claim expires so a publish that died mid-upload gets retried. + const claimed = await distributedStore.putIfAbsent(getFlowBundlePublishClaimKey(input.flowVersionId), 1, FLOW_BUNDLE_PUBLISH_CLAIM_TTL_SECONDS) + if (!claimed) { + return { kind: 'skip' } + } // S3 without signed URLs: the worker streams the bytes back via uploadFlowBundle. if (!signedFileTransport.shouldRedirectForType(FileType.FLOW_BUNDLE)) { return { kind: 'inline' } diff --git a/packages/web/public/locales/en/translation.json b/packages/web/public/locales/en/translation.json index 16e6de2d426f..9fffa65448aa 100644 --- a/packages/web/public/locales/en/translation.json +++ b/packages/web/public/locales/en/translation.json @@ -2304,6 +2304,7 @@ "Highlights": "Highlights", "External ID can only contain letters, numbers and underscores": "External ID can only contain letters, numbers and underscores", "required": "Required", + "invalidExternalId": "External ID may only contain letters, numbers, dot, dash and underscore (up to 128 characters)", "Please connect your account first": "Please connect your account first", "Recommended": "Recommended", "Others": "Others",