From 47b503d35753a27aa628d11191ab85e5dd906e39 Mon Sep 17 00:00:00 2001 From: Kishan Parmar <135701940+kishanprmr@users.noreply.github.com> Date: Tue, 22 Sep 2026 12:02:43 +0530 Subject: [PATCH 1/3] feat(pieces): generate images, manage storage/users, and inspect models across google-gemini, open-router, supabase (#15389) --- bun.lock | 8 +- .../community/google-gemini/package.json | 2 +- .../community/google-gemini/src/index.ts | 10 +- .../src/lib/actions/count-tokens.action.ts | 50 ++++++ .../lib/actions/generate-embeddings.action.ts | 87 ++++++++++ .../src/lib/actions/generate-image.action.ts | 99 ++++++++++++ .../src/lib/actions/list-models.action.ts | 35 ++++ .../google-gemini/src/lib/common/common.ts | 28 +++- .../google-gemini/src/lib/output-schemas.ts | 60 +++++++ .../pieces/community/open-router/package.json | 2 +- .../pieces/community/open-router/src/index.ts | 33 ++-- .../src/lib/actions/get-credits.action.ts | 42 +++++ .../src/lib/actions/get-current-key.action.ts | 56 +++++++ .../actions/list-embedding-models.action.ts | 37 +++++ .../actions/list-model-endpoints.action.ts | 81 ++++++++++ .../src/lib/actions/list-models.action.ts | 64 ++++++++ .../src/lib/actions/list-providers.action.ts | 33 ++++ .../lib/actions/list-user-models.action.ts | 33 ++++ .../open-router/src/lib/common/index.ts | 67 ++++++++ .../open-router/src/lib/output-schemas.ts | 94 ++++++++++- .../pieces/community/supabase/package.json | 2 +- .../pieces/community/supabase/src/index.ts | 56 ++++--- .../supabase/src/lib/actions/create-bucket.ts | 52 ++++++ .../src/lib/actions/create-signed-url.ts | 53 +++++++ .../supabase/src/lib/actions/create-user.ts | 70 ++++++++ .../supabase/src/lib/actions/delete-bucket.ts | 39 +++++ .../supabase/src/lib/actions/delete-file.ts | 47 ++++++ .../supabase/src/lib/actions/delete-user.ts | 39 +++++ .../supabase/src/lib/actions/download-file.ts | 50 ++++++ .../supabase/src/lib/actions/get-user.ts | 47 ++++++ .../supabase/src/lib/actions/invite-user.ts | 54 +++++++ .../supabase/src/lib/actions/list-buckets.ts | 39 +++++ .../supabase/src/lib/actions/list-files.ts | 53 +++++++ .../supabase/src/lib/actions/list-users.ts | 56 +++++++ .../supabase/src/lib/actions/update-user.ts | 64 ++++++++ .../supabase/src/lib/output-schemas.ts | 150 ++++++++++++++++++ 36 files changed, 1736 insertions(+), 56 deletions(-) create mode 100644 packages/pieces/community/google-gemini/src/lib/actions/count-tokens.action.ts create mode 100644 packages/pieces/community/google-gemini/src/lib/actions/generate-embeddings.action.ts create mode 100644 packages/pieces/community/google-gemini/src/lib/actions/generate-image.action.ts create mode 100644 packages/pieces/community/google-gemini/src/lib/actions/list-models.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/get-credits.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/get-current-key.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/list-embedding-models.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/list-model-endpoints.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/list-models.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/list-providers.action.ts create mode 100644 packages/pieces/community/open-router/src/lib/actions/list-user-models.action.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/create-bucket.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/create-signed-url.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/create-user.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/delete-bucket.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/delete-file.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/delete-user.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/download-file.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/get-user.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/invite-user.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/list-buckets.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/list-files.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/list-users.ts create mode 100644 packages/pieces/community/supabase/src/lib/actions/update-user.ts diff --git a/bun.lock b/bun.lock index a60325babcfd..2051d39bfcc5 100644 --- a/bun.lock +++ b/bun.lock @@ -188,7 +188,7 @@ }, "packages/core/utils": { "name": "@activepieces/core-utils", - "version": "0.7.0", + "version": "0.8.0", "dependencies": { "deepmerge-ts": "8.0.2", "ipaddr.js": "2.3.0", @@ -3865,7 +3865,7 @@ }, "packages/pieces/community/google-gemini": { "name": "@activepieces/piece-google-gemini", - "version": "0.2.0", + "version": "0.3.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", @@ -6534,7 +6534,7 @@ }, "packages/pieces/community/open-router": { "name": "@activepieces/piece-open-router", - "version": "0.2.0", + "version": "0.3.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", @@ -8719,7 +8719,7 @@ }, "packages/pieces/community/supabase": { "name": "@activepieces/piece-supabase", - "version": "0.2.0", + "version": "0.3.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", diff --git a/packages/pieces/community/google-gemini/package.json b/packages/pieces/community/google-gemini/package.json index 46181620102f..c262a7425e0e 100644 --- a/packages/pieces/community/google-gemini/package.json +++ b/packages/pieces/community/google-gemini/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-google-gemini", - "version": "0.2.0", + "version": "0.3.0", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "dependencies": { diff --git a/packages/pieces/community/google-gemini/src/index.ts b/packages/pieces/community/google-gemini/src/index.ts index 62216e01a79e..ecb901bca3e5 100644 --- a/packages/pieces/community/google-gemini/src/index.ts +++ b/packages/pieces/community/google-gemini/src/index.ts @@ -2,9 +2,13 @@ import { createCustomApiCallAction } from '@activepieces/pieces-common'; import { createPiece } from '@activepieces/pieces-framework'; import { PieceCategory } from '@activepieces/pieces-framework'; import { chatGemini } from './lib/actions/chat-gemini.action'; +import { countTokensAction } from './lib/actions/count-tokens.action'; import { createVideoAction } from './lib/actions/create-video.action'; import { generateContentFromImageAction } from './lib/actions/generate-content-from-image.action'; import { generateContentAction } from './lib/actions/generate-content.action'; +import { generateEmbeddingsAction } from './lib/actions/generate-embeddings.action'; +import { generateImageAction } from './lib/actions/generate-image.action'; +import { listModelsAction } from './lib/actions/list-models.action'; import { textToSpeechAction } from './lib/actions/text-to-speech.action'; import { generateContentWithFileSearchAction } from './lib/actions/generate-content-with-file-search'; import { googleGeminiAuth } from './lib/auth'; @@ -13,7 +17,7 @@ export const googleGemini = createPiece({ displayName: 'Google Gemini', auth: googleGeminiAuth, description: 'Use the new Gemini models from Google', - minimumSupportedRelease: '0.30.0', + minimumSupportedRelease: '0.87.0', logoUrl: 'https://cdn.activepieces.com/pieces/google-gemini.png', categories: [PieceCategory.ARTIFICIAL_INTELLIGENCE], authors: ["pfernandez98","kishanprmr","MoShizzle","AbdulTheActivePiecer","abuaboud"], @@ -24,6 +28,10 @@ export const googleGemini = createPiece({ chatGemini, textToSpeechAction, createVideoAction, + generateImageAction, + listModelsAction, + generateEmbeddingsAction, + countTokensAction, createCustomApiCallAction({ baseUrl: () => 'https://generativelanguage.googleapis.com/v1beta', auth: googleGeminiAuth, diff --git a/packages/pieces/community/google-gemini/src/lib/actions/count-tokens.action.ts b/packages/pieces/community/google-gemini/src/lib/actions/count-tokens.action.ts new file mode 100644 index 000000000000..3f4bef1c88e1 --- /dev/null +++ b/packages/pieces/community/google-gemini/src/lib/actions/count-tokens.action.ts @@ -0,0 +1,50 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { GoogleGenAI } from '@google/genai'; +import { googleGeminiAuth } from '../auth'; +import { defaultLLM, getGeminiModelOptions } from '../common/common'; +import { countTokensActionOutputSchema } from '../output-schemas'; + +export const countTokensAction = createAction({ + audience: 'both', + name: 'count_tokens', + classification: 'READ', + auth: googleGeminiAuth, + displayName: 'Count Tokens', + description: 'Counts how many tokens a prompt will consume for a given Gemini model.', + aiMetadata: { + description: + 'Counts the tokens a piece of text will consume for a given Gemini model, without generating any content. Use it before generate_content or chat_gemini to check a prompt fits the model\'s input limit or to estimate cost. Idempotent: same text and model always return the same count.', + idempotent: true, + }, + props: { + text: Property.LongText({ + displayName: 'Text', + required: true, + description: 'The text to count tokens for.', + }), + model: Property.Dropdown({ + displayName: 'Model', + required: true, + auth: googleGeminiAuth, + refreshers: [], + defaultValue: defaultLLM, + options: async ({ auth }) => getGeminiModelOptions({ auth }), + }), + }, + outputSchema: countTokensActionOutputSchema, + async run({ auth, propsValue }) { + const { text, model } = propsValue; + + const genAI = new GoogleGenAI({ apiKey: auth.secret_text }); + + const response = await genAI.models.countTokens({ + model, + contents: text, + }); + + return { + totalTokens: response.totalTokens ?? null, + cachedContentTokenCount: response.cachedContentTokenCount ?? null, + }; + }, +}); diff --git a/packages/pieces/community/google-gemini/src/lib/actions/generate-embeddings.action.ts b/packages/pieces/community/google-gemini/src/lib/actions/generate-embeddings.action.ts new file mode 100644 index 000000000000..919a871caec0 --- /dev/null +++ b/packages/pieces/community/google-gemini/src/lib/actions/generate-embeddings.action.ts @@ -0,0 +1,87 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { GoogleGenAI } from '@google/genai'; +import { googleGeminiAuth } from '../auth'; +import { getGeminiEmbeddingModelOptions } from '../common/common'; +import { generateEmbeddingsActionOutputSchema } from '../output-schemas'; + +export const generateEmbeddingsAction = createAction({ + audience: 'both', + name: 'generate_embeddings', + classification: 'READ', + auth: googleGeminiAuth, + displayName: 'Generate Embeddings', + description: 'Converts text into a numerical embedding vector using a Gemini embedding model.', + aiMetadata: { + description: + 'Converts input text into a numerical embedding vector for semantic search, similarity comparison, clustering, or classification. Set Task Type to the intended use (e.g. retrieval_document when indexing content, retrieval_query when searching it) for a better-tuned vector. Idempotent: the same text and settings always produce the same vector.', + idempotent: true, + }, + props: { + text: Property.LongText({ + displayName: 'Text', + required: true, + description: 'The text to generate an embedding for.', + }), + model: Property.Dropdown({ + displayName: 'Model', + required: true, + auth: googleGeminiAuth, + refreshers: [], + defaultValue: 'gemini-embedding-001', + options: async ({ auth }) => getGeminiEmbeddingModelOptions({ auth }), + }), + taskType: Property.StaticDropdown({ + displayName: 'Task Type', + description: 'Tunes the embedding for how it will be used.', + required: false, + options: { + disabled: false, + options: [ + { label: 'Retrieval Query (searching)', value: 'RETRIEVAL_QUERY' }, + { label: 'Retrieval Document (indexing)', value: 'RETRIEVAL_DOCUMENT' }, + { label: 'Semantic Similarity', value: 'SEMANTIC_SIMILARITY' }, + { label: 'Classification', value: 'CLASSIFICATION' }, + { label: 'Clustering', value: 'CLUSTERING' }, + { label: 'Question Answering', value: 'QUESTION_ANSWERING' }, + ], + }, + }), + title: Property.ShortText({ + displayName: 'Title', + description: 'Only used when Task Type is Retrieval Document; improves embedding quality for document search.', + required: false, + }), + outputDimensionality: Property.Number({ + displayName: 'Output Dimensionality', + description: 'Truncate the embedding to this many dimensions. Only supported on newer models.', + required: false, + }), + }, + outputSchema: generateEmbeddingsActionOutputSchema, + async run({ auth, propsValue }) { + const { text, model, taskType, title, outputDimensionality } = propsValue; + + const genAI = new GoogleGenAI({ apiKey: auth.secret_text }); + + const response = await genAI.models.embedContent({ + model, + contents: text, + config: { + taskType, + title, + outputDimensionality, + }, + }); + + const embedding = response.embeddings?.[0]; + + if (!embedding?.values) { + throw new Error('No embedding returned from model response.'); + } + + return { + embedding: embedding.values, + dimensions: embedding.values.length, + }; + }, +}); diff --git a/packages/pieces/community/google-gemini/src/lib/actions/generate-image.action.ts b/packages/pieces/community/google-gemini/src/lib/actions/generate-image.action.ts new file mode 100644 index 000000000000..bb4e2194f7f1 --- /dev/null +++ b/packages/pieces/community/google-gemini/src/lib/actions/generate-image.action.ts @@ -0,0 +1,99 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { GoogleGenAI } from '@google/genai'; +import mime from 'mime-types'; +import { googleGeminiAuth } from '../auth'; +import { getGeminiImageModelOptions } from '../common/common'; +import { generateImageActionOutputSchema } from '../output-schemas'; + +export const generateImageAction = createAction({ + audience: 'both', + name: 'generate_image', + classification: 'READ', + auth: googleGeminiAuth, + displayName: 'Generate Image', + description: 'Generate an image from a text prompt using a Gemini image model.', + aiMetadata: { + description: + 'Generates an image from a text prompt using a Gemini image-generation model (e.g. Nano Banana) and returns it as a downloadable file. Use this for text-to-image; generate_content_from_image goes the other direction (image in, text out), and create_video is for motion output. Not idempotent: each call renders a new image.', + idempotent: false, + }, + props: { + prompt: Property.LongText({ + displayName: 'Prompt', + required: true, + description: 'Describe the image you want to generate.', + }), + model: Property.Dropdown({ + displayName: 'Model', + required: true, + auth: googleGeminiAuth, + refreshers: [], + options: async ({ auth }) => getGeminiImageModelOptions({ auth }), + }), + aspectRatio: Property.StaticDropdown({ + displayName: 'Aspect Ratio', + required: false, + defaultValue: '1:1', + options: { + disabled: false, + options: [ + { label: '1:1 (Square)', value: '1:1' }, + { label: '2:3 (Portrait)', value: '2:3' }, + { label: '3:2 (Landscape)', value: '3:2' }, + { label: '3:4 (Portrait)', value: '3:4' }, + { label: '4:3 (Landscape)', value: '4:3' }, + { label: '9:16 (Portrait)', value: '9:16' }, + { label: '16:9 (Landscape)', value: '16:9' }, + { label: '21:9 (Widescreen)', value: '21:9' }, + ], + }, + }), + imageSize: Property.StaticDropdown({ + displayName: 'Image Size', + required: false, + defaultValue: '1K', + options: { + disabled: false, + options: [ + { label: '1K', value: '1K' }, + { label: '2K', value: '2K' }, + { label: '4K', value: '4K' }, + ], + }, + }), + }, + outputSchema: generateImageActionOutputSchema, + async run({ auth, propsValue, files }) { + const { prompt, model, aspectRatio, imageSize } = propsValue; + + const genAI = new GoogleGenAI({ apiKey: auth.secret_text }); + + const response = await genAI.models.generateContent({ + model, + contents: prompt, + config: { + responseModalities: ['IMAGE'], + imageConfig: { + aspectRatio, + imageSize, + }, + }, + }); + + const imagePart = response.candidates?.[0]?.content?.parts?.find( + (part) => part.inlineData?.data + ); + + if (!imagePart?.inlineData?.data) { + throw new Error('No image data returned from model response.'); + } + + const mimeType = imagePart.inlineData.mimeType ?? 'image/png'; + const extension = mime.extension(mimeType) || 'png'; + + return await files.write({ + data: Buffer.from(imagePart.inlineData.data, 'base64'), + fileName: `image.${extension}`, + }); + }, +}); diff --git a/packages/pieces/community/google-gemini/src/lib/actions/list-models.action.ts b/packages/pieces/community/google-gemini/src/lib/actions/list-models.action.ts new file mode 100644 index 000000000000..c3799f33ad7d --- /dev/null +++ b/packages/pieces/community/google-gemini/src/lib/actions/list-models.action.ts @@ -0,0 +1,35 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { googleGeminiAuth } from '../auth'; +import { listAllModels } from '../common/common'; +import { listModelsActionOutputSchema } from '../output-schemas'; + +export const listModelsAction = createAction({ + audience: 'both', + name: 'list_models', + classification: 'SEARCH', + auth: googleGeminiAuth, + displayName: 'List Models', + description: 'Lists the Gemini and Veo models available to this API key.', + aiMetadata: { + description: + 'Lists every Gemini/Veo model available to this API key, with its token limits and supported generation methods. Use this to discover valid model names before calling generate_content, generate_image, create_video, or generate_embeddings, or to check a model\'s input/output token limits. Safe to retry: read-only.', + idempotent: true, + }, + props: {}, + outputSchema: listModelsActionOutputSchema, + async run({ auth }) { + const models = await listAllModels({ auth }); + + return models.map((model) => ({ + name: model.name.replace('models/', ''), + displayName: model.displayName ?? null, + description: model.description ?? null, + version: model.version ?? null, + inputTokenLimit: model.inputTokenLimit ?? null, + outputTokenLimit: model.outputTokenLimit ?? null, + supportedGenerationMethods: Array.isArray(model.supportedGenerationMethods) + ? model.supportedGenerationMethods.join(', ') + : null, + })); + }, +}); diff --git a/packages/pieces/community/google-gemini/src/lib/common/common.ts b/packages/pieces/community/google-gemini/src/lib/common/common.ts index 8fdd81a67781..ff4833c00be0 100644 --- a/packages/pieces/community/google-gemini/src/lib/common/common.ts +++ b/packages/pieces/community/google-gemini/src/lib/common/common.ts @@ -30,7 +30,15 @@ const isTtsModel = (model: GeminiModel): boolean => const isVeoModel = (model: GeminiModel): boolean => model.name.toLowerCase().includes('veo'); -const listAllModels = async ({ +const isImageModel = (model: GeminiModel): boolean => + model.name.toLowerCase().includes('image') && + !isTtsModel(model) && + !isVeoModel(model); + +const isEmbeddingModel = (model: GeminiModel): boolean => + model.name.toLowerCase().includes('embedding'); + +export const listAllModels = async ({ auth, }: { auth: GeminiAuth; @@ -107,11 +115,27 @@ export const getGeminiVideoModelOptions = async ({ auth?: GeminiAuth; }) => fetchModelOptions({ auth, filter: isVeoModel }); +export const getGeminiImageModelOptions = async ({ + auth, +}: { + auth?: GeminiAuth; +}) => fetchModelOptions({ auth, filter: isImageModel }); + +export const getGeminiEmbeddingModelOptions = async ({ + auth, +}: { + auth?: GeminiAuth; +}) => fetchModelOptions({ auth, filter: isEmbeddingModel }); + type GeminiAuth = AppConnectionValueForAuthProperty; -type GeminiModel = { +export type GeminiModel = { name: string; displayName?: string; + description?: string; + version?: string; + inputTokenLimit?: number; + outputTokenLimit?: number; supportedGenerationMethods?: string[]; }; diff --git a/packages/pieces/community/google-gemini/src/lib/output-schemas.ts b/packages/pieces/community/google-gemini/src/lib/output-schemas.ts index 7c01bfc8c538..b2aa073d4726 100644 --- a/packages/pieces/community/google-gemini/src/lib/output-schemas.ts +++ b/packages/pieces/community/google-gemini/src/lib/output-schemas.ts @@ -140,3 +140,63 @@ export const createVideoActionOutputSchema: OutputSchema = { }, ], }; + +export const generateImageActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'imageFile', + label: 'Image File URL', + value: '', + format: 'url', + }, + ], +}; + +export const listModelsActionOutputSchema: OutputSchema = { + itemLabel: '{name}', + fields: [ + { + key: 'models', + label: 'Models', + value: '', + listItems: [ + { key: 'name', label: 'Model Name' }, + { key: 'displayName', label: 'Display Name' }, + { key: 'description', label: 'Description' }, + { key: 'version', label: 'Version' }, + { key: 'inputTokenLimit', label: 'Input Token Limit', format: 'number' }, + { key: 'outputTokenLimit', label: 'Output Token Limit', format: 'number' }, + { key: 'supportedGenerationMethods', label: 'Supported Methods' }, + ], + }, + ], +}; + +export const generateEmbeddingsActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'embedding', + label: 'Embedding Vector', + }, + { + key: 'dimensions', + label: 'Dimensions', + format: 'number', + }, + ], +}; + +export const countTokensActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'totalTokens', + label: 'Total Tokens', + format: 'number', + }, + { + key: 'cachedContentTokenCount', + label: 'Cached Content Token Count', + format: 'number', + }, + ], +}; diff --git a/packages/pieces/community/open-router/package.json b/packages/pieces/community/open-router/package.json index f547e429b4bf..3b54daa3d7c7 100644 --- a/packages/pieces/community/open-router/package.json +++ b/packages/pieces/community/open-router/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-open-router", - "version": "0.2.0", + "version": "0.3.0", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "scripts": { diff --git a/packages/pieces/community/open-router/src/index.ts b/packages/pieces/community/open-router/src/index.ts index 9cb8617ed9ed..cd0519938dac 100644 --- a/packages/pieces/community/open-router/src/index.ts +++ b/packages/pieces/community/open-router/src/index.ts @@ -1,32 +1,33 @@ -import { - AuthenticationType, - createCustomApiCallAction, - httpClient, - HttpMethod, - HttpRequest, -} from '@activepieces/pieces-common'; -import { createPiece, PieceAuth } from '@activepieces/pieces-framework'; +import { createCustomApiCallAction } from '@activepieces/pieces-common'; +import { createPiece } from '@activepieces/pieces-framework'; import { PieceCategory } from '@activepieces/pieces-framework'; import { askOpenRouterAction } from './lib/actions/ask-open-router'; +import { getCreditsAction } from './lib/actions/get-credits.action'; +import { getCurrentKeyAction } from './lib/actions/get-current-key.action'; +import { listEmbeddingModelsAction } from './lib/actions/list-embedding-models.action'; +import { listModelEndpointsAction } from './lib/actions/list-model-endpoints.action'; +import { listModelsAction } from './lib/actions/list-models.action'; +import { listProvidersAction } from './lib/actions/list-providers.action'; +import { listUserModelsAction } from './lib/actions/list-user-models.action'; import { openRouterAuth } from './lib/auth'; -const markdownDescription = ` -Follow these instructions to get your OpenAI API Key: - -1. Visit the following website: https://openrouter.ai/keys. -2. Once on the website, click on create a key. -3. Once you have created a key, copy it and use it for the Api key field on the site. -`; export const openRouter = createPiece({ displayName: 'OpenRouter', description: 'Use any AI model to generate code, text, or images via OpenRouter.ai.', auth: openRouterAuth, - minimumSupportedRelease: '0.30.0', + minimumSupportedRelease: '0.87.0', logoUrl: 'https://cdn.activepieces.com/pieces/open-router.png', categories: [PieceCategory.ARTIFICIAL_INTELLIGENCE], authors: ["Salem-Alaa","kishanprmr","MoShizzle","abuaboud"], actions: [ askOpenRouterAction, + listModelsAction, + listEmbeddingModelsAction, + listUserModelsAction, + listModelEndpointsAction, + listProvidersAction, + getCreditsAction, + getCurrentKeyAction, createCustomApiCallAction({ baseUrl: () => 'https://openrouter.ai/api/v1', auth: openRouterAuth, diff --git a/packages/pieces/community/open-router/src/lib/actions/get-credits.action.ts b/packages/pieces/community/open-router/src/lib/actions/get-credits.action.ts new file mode 100644 index 000000000000..3774c1046f57 --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/get-credits.action.ts @@ -0,0 +1,42 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { getCreditsActionOutputSchema } from '../output-schemas'; + +interface OpenRouterCreditsResponse { + data: { + total_credits: number; + total_usage: number; + }; +} + +export const getCreditsAction = createAction({ + audience: 'both', + name: 'get_credits', + classification: 'READ', + auth: openRouterAuth, + displayName: 'Get Credits', + description: "Gets this account's current API credit balance and total usage.", + aiMetadata: { + description: + "Returns the total credits purchased and total credits used on this OpenRouter account. Call it before a large or batched job to confirm there is enough balance left; a low or exhausted balance makes every paid model generation fail. Safe to retry: read-only.", + idempotent: true, + }, + props: {}, + outputSchema: getCreditsActionOutputSchema, + async run({ auth }) { + const response = await httpClient.sendRequest({ + url: 'https://openrouter.ai/api/v1/credits', + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + return { + totalCredits: response.body.data.total_credits, + totalUsage: response.body.data.total_usage, + }; + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/actions/get-current-key.action.ts b/packages/pieces/community/open-router/src/lib/actions/get-current-key.action.ts new file mode 100644 index 000000000000..6dee73b83856 --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/get-current-key.action.ts @@ -0,0 +1,56 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { getCurrentKeyActionOutputSchema } from '../output-schemas'; + +interface OpenRouterKeyResponse { + data: { + label: string; + usage: number; + limit: number | null; + is_free_tier: boolean; + rate_limit: { + requests: number; + interval: string; + }; + }; +} + +export const getCurrentKeyAction = createAction({ + audience: 'both', + name: 'get_current_key', + classification: 'READ', + auth: openRouterAuth, + displayName: 'Get API Key Usage', + description: 'Gets usage, spend limit, and rate-limit info for your connected API key.', + aiMetadata: { + description: + "Returns metadata for the connected OpenRouter API key: its label, usage so far, spend limit if one is set, whether it is free-tier, and its rate limit. Use it to check remaining headroom before a batch of generations. Safe to retry: read-only.", + idempotent: true, + }, + props: {}, + outputSchema: getCurrentKeyActionOutputSchema, + async run({ auth }) { + const response = await httpClient.sendRequest({ + url: 'https://openrouter.ai/api/v1/key', + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + const { label, usage, limit, is_free_tier, rate_limit } = response.body.data; + + return { + label, + usage, + limit, + isFreeTier: is_free_tier, + rateLimit: { + requests: rate_limit?.requests ?? null, + interval: rate_limit?.interval ?? null, + }, + }; + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/actions/list-embedding-models.action.ts b/packages/pieces/community/open-router/src/lib/actions/list-embedding-models.action.ts new file mode 100644 index 000000000000..d55992d839f7 --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/list-embedding-models.action.ts @@ -0,0 +1,37 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { OpenRouterModel, mapOpenRouterModelSummary } from '../common'; +import { listModelsActionOutputSchema } from '../output-schemas'; + +export const listEmbeddingModelsAction = createAction({ + audience: 'both', + name: 'list_embedding_models', + classification: 'SEARCH', + auth: openRouterAuth, + displayName: 'List Embedding Models', + description: 'Lists the embedding models available through OpenRouter.', + aiMetadata: { + description: + 'Lists only the models in the OpenRouter catalog whose output is an embedding vector, for use as the model id in a raw call to the embeddings endpoint. Pick List Models instead when the target is a text or chat generation model. Safe to retry: read-only.', + idempotent: true, + }, + props: {}, + outputSchema: listModelsActionOutputSchema, + async run({ auth }) { + const response = await httpClient.sendRequest<{ data: OpenRouterModel[] }>({ + url: 'https://openrouter.ai/api/v1/models', + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + const embeddingModels = response.body.data.filter((model) => + model.architecture?.output_modalities?.includes('embedding') + ); + + return embeddingModels.map(mapOpenRouterModelSummary); + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/actions/list-model-endpoints.action.ts b/packages/pieces/community/open-router/src/lib/actions/list-model-endpoints.action.ts new file mode 100644 index 000000000000..6859649641a2 --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/list-model-endpoints.action.ts @@ -0,0 +1,81 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { OpenRouterEndpoint, OpenRouterModel, mapOpenRouterEndpoint } from '../common'; +import { listModelEndpointsActionOutputSchema } from '../output-schemas'; + +export const listModelEndpointsAction = createAction({ + audience: 'both', + name: 'list_model_endpoints', + classification: 'READ', + auth: openRouterAuth, + displayName: 'List Providers for a Model', + description: 'Lists every provider serving a specific model, with the pricing and uptime each one offers.', + aiMetadata: { + description: + "Lists every provider serving one model on OpenRouter, each with its own pricing, quantization, context length, and uptime. Use it to compare providers for a model discovered with List Models before pinning a request to one. Safe to retry: read-only.", + idempotent: true, + }, + props: { + model: Property.Dropdown({ + displayName: 'Model', + description: 'The model to list providers for.', + required: true, + refreshers: [], + auth: openRouterAuth, + options: async ({ auth }) => { + if (!auth) { + return { + disabled: true, + options: [], + placeholder: 'Please connect your account first', + }; + } + try { + const response = await httpClient.sendRequest<{ data: OpenRouterModel[] }>({ + url: 'https://openrouter.ai/api/v1/models', + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + return { + disabled: false, + options: response.body.data.map((model) => ({ label: model.id, value: model.id })), + }; + } catch (error) { + return { + disabled: true, + options: [], + placeholder: `Couldn't load models:\n${error}`, + }; + } + }, + }), + }, + outputSchema: listModelEndpointsActionOutputSchema, + async run({ auth, propsValue }) { + const [author, ...slugParts] = propsValue.model.split('/'); + const slug = slugParts.join('/'); + + const response = await httpClient.sendRequest<{ + data: { id: string; name: string; endpoints: OpenRouterEndpoint[] }; + }>({ + url: `https://openrouter.ai/api/v1/models/${author}/${slug}/endpoints`, + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + const { id, name, endpoints } = response.body.data; + + return { + id, + name, + endpoints: endpoints.map(mapOpenRouterEndpoint), + }; + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/actions/list-models.action.ts b/packages/pieces/community/open-router/src/lib/actions/list-models.action.ts new file mode 100644 index 000000000000..48c19d89b79c --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/list-models.action.ts @@ -0,0 +1,64 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod, QueryParams } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { OpenRouterModel, mapOpenRouterModelSummary } from '../common'; +import { listModelsActionOutputSchema } from '../output-schemas'; + +const MODEL_CATEGORIES = [ + 'programming', + 'roleplay', + 'marketing', + 'marketing/seo', + 'technology', + 'science', + 'translation', + 'legal', + 'finance', + 'health', + 'trivia', + 'academia', +]; + +export const listModelsAction = createAction({ + audience: 'both', + name: 'list_models', + classification: 'SEARCH', + auth: openRouterAuth, + displayName: 'List Models', + description: 'Lists the models available through OpenRouter, with pricing and context limits.', + aiMetadata: { + description: + 'Lists every model in the OpenRouter catalog, with context length, per-token pricing, and supported parameters. Use this to discover a valid model id before calling Ask LLM or Chat Completion, or to compare pricing and context limits across providers. Optionally filter by category. Safe to retry: read-only.', + idempotent: true, + }, + props: { + category: Property.StaticDropdown({ + displayName: 'Category', + description: 'Only return models tagged for this use case.', + required: false, + options: { + disabled: false, + options: MODEL_CATEGORIES.map((category) => ({ label: category, value: category })), + }, + }), + }, + outputSchema: listModelsActionOutputSchema, + async run({ auth, propsValue }) { + const queryParams: QueryParams = {}; + if (propsValue.category) { + queryParams['category'] = propsValue.category; + } + + const response = await httpClient.sendRequest<{ data: OpenRouterModel[] }>({ + url: 'https://openrouter.ai/api/v1/models', + method: HttpMethod.GET, + queryParams, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + return response.body.data.map(mapOpenRouterModelSummary); + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/actions/list-providers.action.ts b/packages/pieces/community/open-router/src/lib/actions/list-providers.action.ts new file mode 100644 index 000000000000..08933da584cf --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/list-providers.action.ts @@ -0,0 +1,33 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { OpenRouterProvider } from '../common'; +import { listProvidersActionOutputSchema } from '../output-schemas'; + +export const listProvidersAction = createAction({ + audience: 'both', + name: 'list_providers', + classification: 'SEARCH', + auth: openRouterAuth, + displayName: 'List Providers', + description: 'Lists the upstream AI model providers routed through OpenRouter.', + aiMetadata: { + description: + 'Lists every upstream provider OpenRouter can route requests to (e.g. Azure, CoreWeave, Cerebras), with headquarters and policy links. Use it to see which providers exist before restricting a model to specific providers with provider routing options. Safe to retry: read-only.', + idempotent: true, + }, + props: {}, + outputSchema: listProvidersActionOutputSchema, + async run({ auth }) { + const response = await httpClient.sendRequest<{ data: OpenRouterProvider[] }>({ + url: 'https://openrouter.ai/api/v1/providers', + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + return response.body.data; + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/actions/list-user-models.action.ts b/packages/pieces/community/open-router/src/lib/actions/list-user-models.action.ts new file mode 100644 index 000000000000..f3ab7a441d72 --- /dev/null +++ b/packages/pieces/community/open-router/src/lib/actions/list-user-models.action.ts @@ -0,0 +1,33 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { AuthenticationType, httpClient, HttpMethod } from '@activepieces/pieces-common'; +import { openRouterAuth } from '../auth'; +import { OpenRouterModel, mapOpenRouterModelSummary } from '../common'; +import { listModelsActionOutputSchema } from '../output-schemas'; + +export const listUserModelsAction = createAction({ + audience: 'both', + name: 'list_user_models', + classification: 'SEARCH', + auth: openRouterAuth, + displayName: 'List My Available Models', + description: "Lists the models your account can actually use once your provider preferences, privacy settings, and guardrails are applied.", + aiMetadata: { + description: + "Lists the OpenRouter models this account can actually use once its provider preferences, privacy settings, and guardrails are applied - narrower than List Models when the account has restrictions configured. Use it to confirm a model is reachable before calling it. Safe to retry: read-only.", + idempotent: true, + }, + props: {}, + outputSchema: listModelsActionOutputSchema, + async run({ auth }) { + const response = await httpClient.sendRequest<{ data: OpenRouterModel[] }>({ + url: 'https://openrouter.ai/api/v1/models/user', + method: HttpMethod.GET, + authentication: { + type: AuthenticationType.BEARER_TOKEN, + token: auth.secret_text, + }, + }); + + return response.body.data.map(mapOpenRouterModelSummary); + }, +}); diff --git a/packages/pieces/community/open-router/src/lib/common/index.ts b/packages/pieces/community/open-router/src/lib/common/index.ts index 29a39934487d..5edea151b82b 100644 --- a/packages/pieces/community/open-router/src/lib/common/index.ts +++ b/packages/pieces/community/open-router/src/lib/common/index.ts @@ -1,3 +1,27 @@ +export const mapOpenRouterModelSummary = (model: OpenRouterModel) => ({ + id: model.id, + name: model.name, + description: model.description ?? null, + contextLength: model.context_length ?? null, + promptPrice: model.pricing?.prompt ?? null, + completionPrice: model.pricing?.completion ?? null, + supportedParameters: model.supported_parameters?.join(', ') ?? null, +}); + +export const mapOpenRouterEndpoint = (endpoint: OpenRouterEndpoint) => ({ + name: endpoint.name, + modelId: endpoint.model_id ?? null, + modelName: endpoint.model_name ?? null, + providerName: endpoint.provider_name, + contextLength: endpoint.context_length, + promptPrice: endpoint.pricing?.prompt ?? null, + completionPrice: endpoint.pricing?.completion ?? null, + quantization: endpoint.quantization, + maxCompletionTokens: endpoint.max_completion_tokens, + status: endpoint.status, + uptimeLast30m: endpoint.uptime_last_30m, +}); + export interface promptResponse { choices: { text: string; @@ -11,3 +35,46 @@ export interface openRouterModels { id: string; }[]; } + +export interface OpenRouterModel { + id: string; + name: string; + description?: string; + context_length?: number; + architecture?: { + modality?: string; + input_modalities?: string[]; + output_modalities?: string[]; + }; + pricing?: { + prompt?: string; + completion?: string; + }; + supported_parameters?: string[]; +} + +export interface OpenRouterProvider { + name: string; + slug: string; + privacy_policy_url: string | null; + terms_of_service_url: string | null; + status_page_url: string | null; + headquarters: string | null; + datacenters: string[] | null; +} + +export interface OpenRouterEndpoint { + name: string; + model_id?: string; + model_name?: string; + provider_name: string; + context_length: number | null; + pricing?: { + prompt?: string; + completion?: string; + }; + quantization: string | null; + max_completion_tokens: number | null; + status: number; + uptime_last_30m: number | null; +} diff --git a/packages/pieces/community/open-router/src/lib/output-schemas.ts b/packages/pieces/community/open-router/src/lib/output-schemas.ts index 88ffebfdccc0..12fb5f8e3c3d 100644 --- a/packages/pieces/community/open-router/src/lib/output-schemas.ts +++ b/packages/pieces/community/open-router/src/lib/output-schemas.ts @@ -1,4 +1,4 @@ -import { OutputSchema } from '@activepieces/pieces-framework'; +import { OutputSchema, OutputSchemaField } from '@activepieces/pieces-framework'; export const askLmmActionOutputSchema: OutputSchema = { fields: [ @@ -10,3 +10,95 @@ export const askLmmActionOutputSchema: OutputSchema = { }, ], }; + +const modelSummaryListItems: OutputSchemaField[] = [ + { key: 'id', label: 'Model ID' }, + { key: 'name', label: 'Name' }, + { key: 'description', label: 'Description' }, + { key: 'contextLength', label: 'Context Length', format: 'number' }, + { key: 'promptPrice', label: 'Prompt Price (per token)' }, + { key: 'completionPrice', label: 'Completion Price (per token)' }, + { key: 'supportedParameters', label: 'Supported Parameters' }, +]; + +export const listModelsActionOutputSchema: OutputSchema = { + itemLabel: '{id}', + fields: [ + { + key: 'models', + label: 'Models', + value: '', + listItems: modelSummaryListItems, + }, + ], +}; + +export const listProvidersActionOutputSchema: OutputSchema = { + itemLabel: '{name}', + fields: [ + { + key: 'providers', + label: 'Providers', + value: '', + listItems: [ + { key: 'name', label: 'Name' }, + { key: 'slug', label: 'Slug' }, + { key: 'headquarters', label: 'Headquarters' }, + { key: 'privacy_policy_url', label: 'Privacy Policy URL', format: 'url' }, + { key: 'terms_of_service_url', label: 'Terms of Service URL', format: 'url' }, + { key: 'status_page_url', label: 'Status Page URL', format: 'url' }, + ], + }, + ], +}; + +const endpointListItems: OutputSchemaField[] = [ + { key: 'name', label: 'Endpoint Name' }, + { key: 'modelId', label: 'Model ID' }, + { key: 'modelName', label: 'Model Name' }, + { key: 'providerName', label: 'Provider' }, + { key: 'contextLength', label: 'Context Length', format: 'number' }, + { key: 'promptPrice', label: 'Prompt Price (per token)' }, + { key: 'completionPrice', label: 'Completion Price (per token)' }, + { key: 'quantization', label: 'Quantization' }, + { key: 'maxCompletionTokens', label: 'Max Completion Tokens', format: 'number' }, + { key: 'uptimeLast30m', label: 'Uptime (last 30m %)', format: 'number' }, +]; + +export const listModelEndpointsActionOutputSchema: OutputSchema = { + fields: [ + { key: 'id', label: 'Model ID' }, + { key: 'name', label: 'Model Name' }, + { + key: 'endpoints', + label: 'Endpoints', + value: '', + listItems: endpointListItems, + }, + ], +}; + +export const getCreditsActionOutputSchema: OutputSchema = { + fields: [ + { key: 'totalCredits', label: 'Total Credits', format: 'number' }, + { key: 'totalUsage', label: 'Total Usage', format: 'number' }, + ], +}; + +export const getCurrentKeyActionOutputSchema: OutputSchema = { + fields: [ + { key: 'label', label: 'Key Label' }, + { key: 'usage', label: 'Usage', format: 'number' }, + { key: 'limit', label: 'Limit', format: 'number' }, + { key: 'isFreeTier', label: 'Is Free Tier', format: 'boolean' }, + { + key: 'rateLimit', + label: 'Rate Limit', + value: 'rateLimit', + children: [ + { key: 'requests', label: 'Requests', format: 'number' }, + { key: 'interval', label: 'Interval' }, + ], + }, + ], +}; diff --git a/packages/pieces/community/supabase/package.json b/packages/pieces/community/supabase/package.json index e036ff724099..147fa91cee46 100644 --- a/packages/pieces/community/supabase/package.json +++ b/packages/pieces/community/supabase/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-supabase", - "version": "0.2.0", + "version": "0.3.0", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "dependencies": { diff --git a/packages/pieces/community/supabase/src/index.ts b/packages/pieces/community/supabase/src/index.ts index ebf9a6160959..9a41f55db0dd 100644 --- a/packages/pieces/community/supabase/src/index.ts +++ b/packages/pieces/community/supabase/src/index.ts @@ -1,10 +1,7 @@ import { createCustomApiCallAction } from '@activepieces/pieces-common'; import { createPiece, - PieceAuth, - Property, } from '@activepieces/pieces-framework'; -import { createClient } from '@supabase/supabase-js'; import { PieceCategory } from '@activepieces/pieces-framework'; import { uploadFile } from './lib/actions/upload-file'; import { createRow } from './lib/actions/create-row'; @@ -14,39 +11,27 @@ import { upsertRow } from './lib/actions/upsert-row'; import { searchRows } from './lib/actions/search-rows'; import { listTables } from './lib/actions/list-tables'; import { getTableSchema } from './lib/actions/get-table-schema'; +import { createBucket } from './lib/actions/create-bucket'; +import { deleteBucket } from './lib/actions/delete-bucket'; +import { listBuckets } from './lib/actions/list-buckets'; +import { listFiles } from './lib/actions/list-files'; +import { downloadFile } from './lib/actions/download-file'; +import { deleteFile } from './lib/actions/delete-file'; +import { createSignedUrl } from './lib/actions/create-signed-url'; +import { listUsers } from './lib/actions/list-users'; +import { getUser } from './lib/actions/get-user'; +import { createUser } from './lib/actions/create-user'; +import { updateUser } from './lib/actions/update-user'; +import { deleteUser } from './lib/actions/delete-user'; +import { inviteUser } from './lib/actions/invite-user'; import { newRow } from './lib/triggers/new-row'; import { supabaseAuth } from './lib/auth'; -const markdown = ` -## Supabase Connection Setup - -### 1. Get Your Project URL -- Go to your [Supabase Dashboard](https://supabase.com/dashboard) -- Select your project -- Go to **Settings** → **API** -- Copy the **Project URL** (format: \`https://your-project-ref.supabase.co\`) - -### 2. Get Your API Key -Choose the appropriate key based on your use case: - -**For Actions (Database Operations):** -- Use **Service Role Key** (secret) for server-side operations -- Has full access to bypass Row Level Security (RLS) - -**For Triggers (Webhooks):** -- Use **Anonymous Key** (public) if your webhooks don't need elevated permissions -- Use **Service Role Key** for elevated permissions - -**Security Note:** Keep your Service Role Key secret - it bypasses all RLS policies. - -Find your keys in **Settings** → **API** → **Project API keys** -`; - export const supabase = createPiece({ displayName: 'Supabase', description: 'The open-source Firebase alternative', auth: supabaseAuth, - minimumSupportedRelease: '0.30.0', + minimumSupportedRelease: '0.87.0', logoUrl: 'https://cdn.activepieces.com/pieces/supabase.png', categories: [PieceCategory.DEVELOPER_TOOLS], authors: ["kishanprmr","MoShizzle","abuaboud","fortunamide"], @@ -59,6 +44,19 @@ export const supabase = createPiece({ searchRows, listTables, getTableSchema, + createBucket, + deleteBucket, + listBuckets, + listFiles, + downloadFile, + deleteFile, + createSignedUrl, + listUsers, + getUser, + createUser, + updateUser, + deleteUser, + inviteUser, createCustomApiCallAction({ baseUrl: (auth) => auth?.props?.url || '', auth: supabaseAuth, diff --git a/packages/pieces/community/supabase/src/lib/actions/create-bucket.ts b/packages/pieces/community/supabase/src/lib/actions/create-bucket.ts new file mode 100644 index 000000000000..0a4204e6b942 --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/create-bucket.ts @@ -0,0 +1,52 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { createBucketActionOutputSchema } from '../output-schemas'; + +export const createBucket = createAction({ + name: 'create_bucket', + classification: 'WRITE', + displayName: 'Create Storage Bucket', + description: 'Creates a new Storage bucket, public or private.', + audience: 'both', + aiMetadata: { + description: 'Creates a new Storage bucket with a given name, either public (files reachable by direct URL) or private (files require a signed URL or download). Use before uploading files to a bucket that does not exist yet. Not idempotent: creating a bucket with a name that already exists errors.', + idempotent: false, + }, + auth: supabaseAuth, + props: { + name: Property.ShortText({ + displayName: 'Bucket Name', + required: true, + }), + public: Property.Checkbox({ + displayName: 'Public', + description: 'Whether files in this bucket are publicly readable by URL.', + required: false, + defaultValue: false, + }), + fileSizeLimit: Property.Number({ + displayName: 'File Size Limit (bytes)', + required: false, + }), + }, + outputSchema: createBucketActionOutputSchema, + async run(context) { + const { name, public: isPublic, fileSizeLimit } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.storage.createBucket(name, { + public: isPublic ?? false, + fileSizeLimit: fileSizeLimit || undefined, + }); + + if (error) { + throw new Error(`Failed to create bucket: ${error.message}`); + } + + return { + name: data.name, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/create-signed-url.ts b/packages/pieces/community/supabase/src/lib/actions/create-signed-url.ts new file mode 100644 index 000000000000..a84f7cef0002 --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/create-signed-url.ts @@ -0,0 +1,53 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { createSignedUrlActionOutputSchema } from '../output-schemas'; + +export const createSignedUrl = createAction({ + name: 'create_signed_url', + classification: 'READ', + displayName: 'Create Temporary File Link', + description: 'Generates a time-limited URL that grants temporary access to a file in a private Storage bucket.', + audience: 'both', + aiMetadata: { + description: 'Generates a time-limited signed URL for a file in a Storage bucket, so it can be accessed (e.g. shared or fetched) without making the bucket public. Use for files in private buckets, or when a temporary link is preferable to returning the full file content. Read-only and idempotent: repeated calls just mint a new link with the same expiry window.', + idempotent: true, + }, + auth: supabaseAuth, + props: { + bucket: Property.ShortText({ + displayName: 'Bucket', + description: 'The name of the Storage bucket.', + required: true, + }), + path: Property.ShortText({ + displayName: 'File Path', + description: 'The path of the file within the bucket.', + required: true, + }), + expiresIn: Property.Number({ + displayName: 'Expires In (seconds)', + description: 'How long the link stays valid, in seconds.', + required: false, + defaultValue: 3600, + }), + }, + outputSchema: createSignedUrlActionOutputSchema, + async run(context) { + const { bucket, path, expiresIn } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.storage + .from(bucket) + .createSignedUrl(path, expiresIn || 3600); + + if (error) { + throw new Error(`Failed to create signed URL: ${error.message}`); + } + + return { + signedUrl: data.signedUrl, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/create-user.ts b/packages/pieces/community/supabase/src/lib/actions/create-user.ts new file mode 100644 index 000000000000..6546b883827b --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/create-user.ts @@ -0,0 +1,70 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { createUserActionOutputSchema } from '../output-schemas'; + +export const createUser = createAction({ + name: 'create_user', + classification: 'WRITE', + displayName: 'Create User', + description: 'Creates a new user directly in the project\'s Auth system, with an immediately usable password. Requires the Service Role Key.', + audience: 'both', + aiMetadata: { + description: 'Creates a new user in Auth with an email/phone and password that can sign in immediately, skipping the invite/confirmation email flow. Requires the Service Role Key. Use for programmatic account provisioning; use Invite User by Email instead if the user should set their own password via an email link. Not idempotent: calling twice with the same email creates a conflict error, not a second identical user.', + idempotent: false, + }, + auth: supabaseAuth, + props: { + email: Property.ShortText({ + displayName: 'Email', + required: false, + }), + phone: Property.ShortText({ + displayName: 'Phone', + required: false, + }), + password: Property.ShortText({ + displayName: 'Password', + required: false, + }), + emailConfirm: Property.Checkbox({ + displayName: 'Mark Email as Confirmed', + required: false, + defaultValue: false, + }), + userMetadata: Property.Json({ + displayName: 'User Metadata', + description: 'Custom data to attach to the user, as a JSON object.', + required: false, + }), + }, + outputSchema: createUserActionOutputSchema, + async run(context) { + const { email, phone, password, emailConfirm, userMetadata } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + if (!email && !phone) { + throw new Error('Either Email or Phone must be provided.'); + } + + const { data, error } = await supabase.auth.admin.createUser({ + email, + phone, + password, + email_confirm: emailConfirm, + user_metadata: (userMetadata as Record) ?? undefined, + }); + + if (error) { + throw new Error(`Failed to create user: ${error.message}`); + } + + return { + id: data.user.id, + email: data.user.email ?? null, + phone: data.user.phone ?? null, + created_at: data.user.created_at, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/delete-bucket.ts b/packages/pieces/community/supabase/src/lib/actions/delete-bucket.ts new file mode 100644 index 000000000000..1663b5ff87f0 --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/delete-bucket.ts @@ -0,0 +1,39 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { deleteBucketActionOutputSchema } from '../output-schemas'; + +export const deleteBucket = createAction({ + name: 'delete_bucket', + classification: 'DESTRUCTIVE', + displayName: 'Delete Storage Bucket', + description: 'Permanently deletes a Storage bucket. The bucket must be empty first.', + audience: 'both', + aiMetadata: { + description: 'Permanently deletes a Storage bucket by name. The bucket must already be empty (delete its files first) or the call fails. Use only once you have confirmed the bucket name and that it should be removed entirely. Idempotent: deleting an already-deleted bucket errors without further effect.', + idempotent: true, + }, + auth: supabaseAuth, + props: { + name: Property.ShortText({ + displayName: 'Bucket Name', + required: true, + }), + }, + outputSchema: deleteBucketActionOutputSchema, + async run(context) { + const { name } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { error } = await supabase.storage.deleteBucket(name); + + if (error) { + throw new Error(`Failed to delete bucket: ${error.message}`); + } + + return { + success: true, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/delete-file.ts b/packages/pieces/community/supabase/src/lib/actions/delete-file.ts new file mode 100644 index 000000000000..2b1f3ae259da --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/delete-file.ts @@ -0,0 +1,47 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { deleteFileActionOutputSchema } from '../output-schemas'; + +export const deleteFile = createAction({ + name: 'delete_file', + classification: 'DESTRUCTIVE', + displayName: 'Delete File', + description: 'Permanently deletes one or more files from a Storage bucket.', + audience: 'both', + aiMetadata: { + description: 'Permanently deletes one or more files from a Storage bucket, given their full paths. Use only once you have confirmed the paths (List Files in Bucket first). Idempotent: re-running on paths that no longer exist deletes nothing further.', + idempotent: true, + }, + auth: supabaseAuth, + props: { + bucket: Property.ShortText({ + displayName: 'Bucket', + description: 'The name of the Storage bucket.', + required: true, + }), + paths: Property.Array({ + displayName: 'File Paths', + description: 'Paths of the files to delete within the bucket.', + required: true, + }), + }, + outputSchema: deleteFileActionOutputSchema, + async run(context) { + const { bucket, paths } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.storage.from(bucket).remove(paths as string[]); + + if (error) { + throw new Error(`Failed to delete file(s): ${error.message}`); + } + + return { + deleted_files: (data ?? []).map((file) => ({ + name: file.name, + })), + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/delete-user.ts b/packages/pieces/community/supabase/src/lib/actions/delete-user.ts new file mode 100644 index 000000000000..998175c2e6ea --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/delete-user.ts @@ -0,0 +1,39 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { deleteUserActionOutputSchema } from '../output-schemas'; + +export const deleteUser = createAction({ + name: 'delete_user', + classification: 'DESTRUCTIVE', + displayName: 'Delete User', + description: 'Permanently deletes a user from the project\'s Auth system. Requires the Service Role Key.', + audience: 'both', + aiMetadata: { + description: "Permanently deletes a user from Auth by ID, removing their ability to sign in. Requires the Service Role Key. Use only once you have confirmed the user ID (Get User or List Users first) — this cannot be undone. Idempotent: deleting an already-deleted user ID errors without further effect.", + idempotent: true, + }, + auth: supabaseAuth, + props: { + userId: Property.ShortText({ + displayName: 'User ID', + required: true, + }), + }, + outputSchema: deleteUserActionOutputSchema, + async run(context) { + const { userId } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { error } = await supabase.auth.admin.deleteUser(userId); + + if (error) { + throw new Error(`Failed to delete user: ${error.message}`); + } + + return { + success: true, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/download-file.ts b/packages/pieces/community/supabase/src/lib/actions/download-file.ts new file mode 100644 index 000000000000..443fd91143ca --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/download-file.ts @@ -0,0 +1,50 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { downloadFileActionOutputSchema } from '../output-schemas'; + +export const downloadFile = createAction({ + name: 'download_file', + classification: 'READ', + displayName: 'Download File', + description: 'Downloads a file from a Storage bucket and returns its content as base64.', + audience: 'both', + aiMetadata: { + description: 'Downloads a file from a Storage bucket by its path and returns its content as base64, along with its MIME type and size. Use when you need the actual file content rather than a link (for example, to read a small text/JSON file or pass an image on for processing). For large files, prefer Create Temporary File Link instead of loading the full content. Read-only and idempotent.', + idempotent: true, + }, + auth: supabaseAuth, + props: { + bucket: Property.ShortText({ + displayName: 'Bucket', + description: 'The name of the Storage bucket.', + required: true, + }), + path: Property.ShortText({ + displayName: 'File Path', + description: 'The path of the file within the bucket.', + required: true, + }), + }, + outputSchema: downloadFileActionOutputSchema, + async run(context) { + const { bucket, path } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.storage.from(bucket).download(path); + + if (error) { + throw new Error(`Failed to download file: ${error.message}`); + } + + const arrayBuffer = await data.arrayBuffer(); + const base64 = Buffer.from(arrayBuffer).toString('base64'); + + return { + base64, + mimeType: data.type, + size: data.size, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/get-user.ts b/packages/pieces/community/supabase/src/lib/actions/get-user.ts new file mode 100644 index 000000000000..d825d4b9affe --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/get-user.ts @@ -0,0 +1,47 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { getUserActionOutputSchema } from '../output-schemas'; + +export const getUser = createAction({ + name: 'get_user', + classification: 'READ', + displayName: 'Get User', + description: 'Gets a single user\'s Auth record by ID. Requires the Service Role Key.', + audience: 'both', + aiMetadata: { + description: "Returns a single user's Auth record (email, phone, metadata, confirmation and sign-in timestamps) by user ID. Requires the Service Role Key. Use after List Users to inspect one account in full. Read-only and idempotent.", + idempotent: true, + }, + auth: supabaseAuth, + props: { + userId: Property.ShortText({ + displayName: 'User ID', + required: true, + }), + }, + outputSchema: getUserActionOutputSchema, + async run(context) { + const { userId } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.auth.admin.getUserById(userId); + + if (error) { + throw new Error(`Failed to get user: ${error.message}`); + } + + const user = data.user; + return { + id: user.id, + email: user.email ?? null, + phone: user.phone ?? null, + user_metadata: user.user_metadata ?? {}, + app_metadata: user.app_metadata ?? {}, + created_at: user.created_at, + last_sign_in_at: user.last_sign_in_at ?? null, + confirmed_at: user.confirmed_at ?? null, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/invite-user.ts b/packages/pieces/community/supabase/src/lib/actions/invite-user.ts new file mode 100644 index 000000000000..ccfb9d0733bc --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/invite-user.ts @@ -0,0 +1,54 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { inviteUserActionOutputSchema } from '../output-schemas'; + +export const inviteUser = createAction({ + name: 'invite_user', + classification: 'WRITE', + displayName: 'Invite User by Email', + description: 'Sends an invite email to a new user, who sets their own password when accepting. Requires the Service Role Key.', + audience: 'both', + aiMetadata: { + description: 'Creates a new user and sends them an invite email with a link to set their own password. Requires the Service Role Key. Use for user-facing onboarding; use Create User instead when you need to provision a login programmatically without an email step. Not idempotent: calling twice for the same email errors on the second call once the account exists.', + idempotent: false, + }, + auth: supabaseAuth, + props: { + email: Property.ShortText({ + displayName: 'Email', + required: true, + }), + userMetadata: Property.Json({ + displayName: 'User Metadata', + description: 'Custom data to attach to the user, as a JSON object.', + required: false, + }), + redirectTo: Property.ShortText({ + displayName: 'Redirect URL', + description: 'Where to redirect the user after they accept the invite.', + required: false, + }), + }, + outputSchema: inviteUserActionOutputSchema, + async run(context) { + const { email, userMetadata, redirectTo } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.auth.admin.inviteUserByEmail(email, { + data: (userMetadata as Record) ?? undefined, + redirectTo, + }); + + if (error) { + throw new Error(`Failed to invite user: ${error.message}`); + } + + return { + id: data.user.id, + email: data.user.email ?? null, + invited_at: data.user.invited_at ?? null, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/list-buckets.ts b/packages/pieces/community/supabase/src/lib/actions/list-buckets.ts new file mode 100644 index 000000000000..a8b41e70cc4b --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/list-buckets.ts @@ -0,0 +1,39 @@ +import { createAction } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { listBucketsActionOutputSchema } from '../output-schemas'; + +export const listBuckets = createAction({ + name: 'list_buckets', + classification: 'READ', + displayName: 'List Storage Buckets', + description: 'Lists all Storage buckets in the project.', + audience: 'both', + aiMetadata: { + description: 'Lists every Storage bucket in the connected Supabase project, with each bucket\'s visibility (public/private) and file size limit. Use to discover which buckets exist before uploading, listing, or downloading files. Read-only and idempotent.', + idempotent: true, + }, + auth: supabaseAuth, + props: {}, + outputSchema: listBucketsActionOutputSchema, + async run(context) { + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.storage.listBuckets(); + + if (error) { + throw new Error(`Failed to list buckets: ${error.message}`); + } + + return { + buckets: (data ?? []).map((bucket) => ({ + id: bucket.id, + name: bucket.name, + public: bucket.public, + created_at: bucket.created_at, + updated_at: bucket.updated_at, + })), + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/list-files.ts b/packages/pieces/community/supabase/src/lib/actions/list-files.ts new file mode 100644 index 000000000000..3da0f3dd4d0d --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/list-files.ts @@ -0,0 +1,53 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { listFilesActionOutputSchema } from '../output-schemas'; + +export const listFiles = createAction({ + name: 'list_files', + classification: 'SEARCH', + displayName: 'List Files in Bucket', + description: 'Lists files and folders inside a Storage bucket, optionally under a specific folder path.', + audience: 'both', + aiMetadata: { + description: "Lists files and folders inside a Storage bucket, optionally scoped to a folder path. Use to discover what's already stored before downloading, deleting, or generating a link for a file. Read-only and idempotent.", + idempotent: true, + }, + auth: supabaseAuth, + props: { + bucket: Property.ShortText({ + displayName: 'Bucket', + description: 'The name of the Storage bucket.', + required: true, + }), + path: Property.ShortText({ + displayName: 'Folder Path', + description: 'Folder path to list within the bucket. Leave empty to list the bucket root.', + required: false, + }), + }, + outputSchema: listFilesActionOutputSchema, + async run(context) { + const { bucket, path } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.storage.from(bucket).list(path || undefined); + + if (error) { + throw new Error(`Failed to list files: ${error.message}`); + } + + return { + files: (data ?? []).map((file) => ({ + name: file.name, + id: file.id, + updated_at: file.updated_at, + created_at: file.created_at, + last_accessed_at: file.last_accessed_at, + size: file.metadata?.['size'] ?? null, + mimetype: file.metadata?.['mimetype'] ?? null, + })), + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/list-users.ts b/packages/pieces/community/supabase/src/lib/actions/list-users.ts new file mode 100644 index 000000000000..9c2e98ec7925 --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/list-users.ts @@ -0,0 +1,56 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { listUsersActionOutputSchema } from '../output-schemas'; + +export const listUsers = createAction({ + name: 'list_users', + classification: 'SEARCH', + displayName: 'List Users', + description: 'Lists the users registered in the project\'s Auth system. Requires the Service Role Key.', + audience: 'both', + aiMetadata: { + description: "Lists users registered in the connected Supabase project's Auth system, paginated. Requires the Service Role Key. Use to look up or browse accounts before acting on a specific user. Read-only and idempotent.", + idempotent: true, + }, + auth: supabaseAuth, + props: { + page: Property.Number({ + displayName: 'Page', + description: 'Page number for pagination (starts from 1).', + required: false, + defaultValue: 1, + }), + perPage: Property.Number({ + displayName: 'Users Per Page', + required: false, + defaultValue: 50, + }), + }, + outputSchema: listUsersActionOutputSchema, + async run(context) { + const { page, perPage } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.auth.admin.listUsers({ + page: page || 1, + perPage: perPage || 50, + }); + + if (error) { + throw new Error(`Failed to list users: ${error.message}`); + } + + return { + users: data.users.map((user) => ({ + id: user.id, + email: user.email ?? null, + phone: user.phone ?? null, + created_at: user.created_at, + last_sign_in_at: user.last_sign_in_at ?? null, + confirmed_at: user.confirmed_at ?? null, + })), + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/actions/update-user.ts b/packages/pieces/community/supabase/src/lib/actions/update-user.ts new file mode 100644 index 000000000000..78d5539a00a3 --- /dev/null +++ b/packages/pieces/community/supabase/src/lib/actions/update-user.ts @@ -0,0 +1,64 @@ +import { createAction, Property } from '@activepieces/pieces-framework'; +import { createClient } from '@supabase/supabase-js'; +import { supabaseAuth } from '../auth'; +import { updateUserActionOutputSchema } from '../output-schemas'; + +export const updateUser = createAction({ + name: 'update_user', + classification: 'WRITE', + displayName: 'Update User', + description: 'Updates a user\'s email, phone, password, or metadata. Requires the Service Role Key.', + audience: 'both', + aiMetadata: { + description: "Updates an existing user's email, phone, password, or metadata by user ID. Requires the Service Role Key. Only the fields provided are changed; leave the rest empty to keep them as-is. Idempotent: setting the same values again has no further effect.", + idempotent: true, + }, + auth: supabaseAuth, + props: { + userId: Property.ShortText({ + displayName: 'User ID', + required: true, + }), + email: Property.ShortText({ + displayName: 'New Email', + required: false, + }), + phone: Property.ShortText({ + displayName: 'New Phone', + required: false, + }), + password: Property.ShortText({ + displayName: 'New Password', + required: false, + }), + userMetadata: Property.Json({ + displayName: 'User Metadata', + description: 'Replaces the user\'s custom metadata, as a JSON object.', + required: false, + }), + }, + outputSchema: updateUserActionOutputSchema, + async run(context) { + const { userId, email, phone, password, userMetadata } = context.propsValue; + const { url, apiKey } = context.auth.props; + const supabase = createClient(url, apiKey); + + const { data, error } = await supabase.auth.admin.updateUserById(userId, { + email, + phone, + password, + user_metadata: (userMetadata as Record) ?? undefined, + }); + + if (error) { + throw new Error(`Failed to update user: ${error.message}`); + } + + return { + id: data.user.id, + email: data.user.email ?? null, + phone: data.user.phone ?? null, + updated_at: data.user.updated_at ?? null, + }; + }, +}); diff --git a/packages/pieces/community/supabase/src/lib/output-schemas.ts b/packages/pieces/community/supabase/src/lib/output-schemas.ts index fac5b3917dab..68ce60bd2b34 100644 --- a/packages/pieces/community/supabase/src/lib/output-schemas.ts +++ b/packages/pieces/community/supabase/src/lib/output-schemas.ts @@ -414,6 +414,156 @@ export const uploadFileActionOutputSchema: OutputSchema = { ], }; +export const createBucketActionOutputSchema: OutputSchema = { + fields: [ + { key: 'name', label: 'Name', value: 'name' }, + ], +}; + +export const deleteBucketActionOutputSchema: OutputSchema = { + fields: [ + { key: 'success', label: 'Success', value: 'success', format: 'boolean' }, + ], +}; + +export const listBucketsActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'buckets', + label: 'Buckets', + value: '', + labelKey: 'name', + listItems: [ + { key: 'id', label: 'ID' }, + { key: 'name', label: 'Name' }, + { key: 'public', label: 'Public', format: 'boolean' }, + { key: 'created_at', label: 'Created At', format: 'datetime' }, + { key: 'updated_at', label: 'Updated At', format: 'datetime' }, + ], + }, + ], +}; + +export const listFilesActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'files', + label: 'Files', + value: '', + labelKey: 'name', + listItems: [ + { key: 'name', label: 'Name' }, + { key: 'id', label: 'ID' }, + { key: 'updated_at', label: 'Updated At', format: 'datetime' }, + { key: 'created_at', label: 'Created At', format: 'datetime' }, + { key: 'last_accessed_at', label: 'Last Accessed At', format: 'datetime' }, + { key: 'size', label: 'Size (bytes)', format: 'number' }, + { key: 'mimetype', label: 'MIME Type' }, + ], + }, + ], +}; + +export const downloadFileActionOutputSchema: OutputSchema = { + fields: [ + { key: 'base64', label: 'File Content (base64)' }, + { key: 'mimeType', label: 'MIME Type' }, + { key: 'size', label: 'Size (bytes)', format: 'number' }, + ], +}; + +export const deleteFileActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'deleted_files', + label: 'Deleted Files', + value: '', + labelKey: 'name', + listItems: [ + { key: 'name', label: 'Name' }, + ], + }, + ], +}; + +export const createSignedUrlActionOutputSchema: OutputSchema = { + fields: [ + { key: 'signedUrl', label: 'Signed URL', value: 'signedUrl', format: 'url' }, + ], +}; + +export const listUsersActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'users', + label: 'Users', + value: '', + labelKey: 'email', + listItems: [ + { key: 'id', label: 'ID' }, + { key: 'email', label: 'Email' }, + { key: 'phone', label: 'Phone' }, + { key: 'created_at', label: 'Created At', format: 'datetime' }, + { key: 'last_sign_in_at', label: 'Last Sign In At', format: 'datetime' }, + { key: 'confirmed_at', label: 'Confirmed At', format: 'datetime' }, + ], + }, + ], +}; + +export const getUserActionOutputSchema: OutputSchema = { + fields: [ + { key: 'id', label: 'ID' }, + { key: 'email', label: 'Email' }, + { key: 'phone', label: 'Phone' }, + { + key: 'user_metadata', + label: 'User Metadata', + value: 'user_metadata', + }, + { + key: 'app_metadata', + label: 'App Metadata', + value: 'app_metadata', + }, + { key: 'created_at', label: 'Created At', format: 'datetime' }, + { key: 'last_sign_in_at', label: 'Last Sign In At', format: 'datetime' }, + { key: 'confirmed_at', label: 'Confirmed At', format: 'datetime' }, + ], +}; + +export const createUserActionOutputSchema: OutputSchema = { + fields: [ + { key: 'id', label: 'ID' }, + { key: 'email', label: 'Email' }, + { key: 'phone', label: 'Phone' }, + { key: 'created_at', label: 'Created At', format: 'datetime' }, + ], +}; + +export const updateUserActionOutputSchema: OutputSchema = { + fields: [ + { key: 'id', label: 'ID' }, + { key: 'email', label: 'Email' }, + { key: 'phone', label: 'Phone' }, + { key: 'updated_at', label: 'Updated At', format: 'datetime' }, + ], +}; + +export const deleteUserActionOutputSchema: OutputSchema = { + fields: [ + { key: 'success', label: 'Success', value: 'success', format: 'boolean' }, + ], +}; + +export const inviteUserActionOutputSchema: OutputSchema = { + fields: [ + { key: 'id', label: 'ID' }, + { key: 'email', label: 'Email' }, + { key: 'invited_at', label: 'Invited At', format: 'datetime' }, + ], +}; + export const newRowTriggerOutputSchema: OutputSchema = { fields: [ { From da67d472c6d4f1093be60c6e003353a876e12028 Mon Sep 17 00:00:00 2001 From: Amr Elmohamady Date: Tue, 22 Sep 2026 11:23:31 +0300 Subject: [PATCH 2/3] feat(logging): enrich every log with host = os.hostname() (#15665) Co-authored-by: Claude Opus 4.7 --- .agents/skills/debug-failed-run/SKILL.md | 16 ++++ .../evlog-metrics.md | 73 +++++++++++++++++++ .../data-storage-observability/index.md | 5 ++ packages/server/utils/src/evlog-setup.ts | 25 +++++-- .../utils/test/evlog-host-drain.test.ts | 45 ++++++++++++ 5 files changed, 159 insertions(+), 5 deletions(-) create mode 100644 brain/knowledge/data-storage-observability/evlog-metrics.md create mode 100644 packages/server/utils/test/evlog-host-drain.test.ts diff --git a/.agents/skills/debug-failed-run/SKILL.md b/.agents/skills/debug-failed-run/SKILL.md index 67b60ee39e9d..f87f6ca19628 100644 --- a/.agents/skills/debug-failed-run/SKILL.md +++ b/.agents/skills/debug-failed-run/SKILL.md @@ -46,6 +46,22 @@ Use the **`Logs`** source (`id: 6a2a91b1d37162f45ad78233`; key columns `Body`, ` Scope the time window to the job's `processedAt`/`finishedAt` from Step 1 (± a few minutes) to keep queries cheap. +**Host correlation** — every log carries `LogAttributes['host']` (format `-`, e.g. `91.98.135.134-e5cbf708af8d`), stamped by the global evlog host enricher. When a failure looks resource-shaped (OOM, RPC timeout, sandbox crash, silent hang), correlate the failing job's host with its `system.snapshot` metrics at that time: + +```sql +-- Given the failed job's host + finish time, what did that container's memory + event loop look like? +SELECT Timestamp, LogAttributes['memRssMb'] AS rss_mb, + LogAttributes['memHeapUsedMb'] AS heap_mb, + LogAttributes['eventLoopDelayP99Ms'] AS lag_ms +FROM default.otel_logs +WHERE LogAttributes['event'] = 'system.snapshot' + AND LogAttributes['host'] = '' + AND Timestamp BETWEEN - INTERVAL 15 MINUTE AND + INTERVAL 1 MINUTE +ORDER BY Timestamp +``` + +RSS climbing toward the container's cgroup ceiling (~900 MB on workers, ~2 GB on app) or `lag_ms` spiking above ~1 s just before the failure = infra cause, not the code path. The reverse — pull failed jobs on a host whose RSS crossed a threshold — surfaces a leaking container that is dropping a small percentage of runs silently. + > **Those two names exist only in the script's report** (`debug-failed-job.js` formats them from the raw job). If you bypass the script and read jobs straight off the queue with `bullmq` — `q.getJobs(['failed'], …)` — the properties are **`job.processedOn` / `job.finishedOn`**. `job.processedAt` is silently `undefined`, not an error, so a hand-rolled scan reports every failure as having no timestamp and you fall back to `job.timestamp` (enqueue time) without noticing. That shifts failures earlier by however long the queue backlog is (8+ minutes on cloud) and will make a fix look like it did not take effect. Bucket by `finishedOn` when you want "is this still happening". You're looking for the engine/worker log lines that bracket the failure — sandbox crashes, OOM ("no space"/heap), RPC timeouts, connection refresh failures. ## Step 3 — Trace the failure into this repo diff --git a/brain/knowledge/data-storage-observability/evlog-metrics.md b/brain/knowledge/data-storage-observability/evlog-metrics.md new file mode 100644 index 000000000000..4ae4d0580c66 --- /dev/null +++ b/brain/knowledge/data-storage-observability/evlog-metrics.md @@ -0,0 +1,73 @@ +--- +icon: 📈 +--- + +# Logging & Metrics (evlog) + +All structured logging goes through **evlog**. One wide event per unit of work (HTTP request, worker job, flow run) with all context as fields, shipped to a single remote drain. Metrics are derived from log fields on the backend — we do not run a separate metrics pipeline for application signals, with one exception (BullMQ queue depth as native OTLP gauges). + +### Setup + +- `evlogSetup.init` (`packages/server/utils/src/evlog-setup.ts`) wires `initLogger` with service/version/env, level + sampling, a redact list (secrets, tokens, `authorization`, `connection.value`, axios internals), and a drain wrapper (`wrapDrainWithHost`) that stamps `os.hostname()` on every event whose `host` field is undefined. Uses the drain, not `enricherPlugin`, because `enricherPlugin.enrich` hooks only run on the HTTP request-finish path; plain `log.info/warn/error` and `createLogger().emit()` go through `emitWideEvent`, which invokes drains but not enrichers, and workers don't register `evlogFastify` at all — a `host` enricher would be a worker-side no-op. Events that already carry a host (e.g. `system.snapshot`, which sets it at the call site) pass through unchanged. +- Sampling: info sampled per `AP_SAMPLE_RATE_INFO`; warn/error always kept; extra keep rules for `status ≥ 400` and `duration > 2000ms`. +- Field-naming schema is a hard contract — see `packages/server/CLAUDE.md` "Structured Logging Field Schema". One concept = one dotted path (`flowRun.id`, `piece.name`, `job.type`), durations end in `Ms`, bytes in `Bytes`, reserved keys are auto-populated. + +### Drains + +`packages/server/utils/src/evlog-drains.ts` picks one remote destination by env, first match wins: **Axiom → HyperDX → Loki → Better Stack → generic OTLP**. All batch through a shared pipeline (100 / 5s, retry ×3, buffer 5000). In dev, an additional NDJSON file drain writes to `.evlog/logs/` so `/analyze-logs` and jq work locally. + +### Log-based metrics + +The drain destination (HyperDX/ClickStack, Loki, Better Stack) derives metrics from log fields at query time — that is the whole reason the field schema is strict. Typical shapes: + +- `count() where error.code = 'X'` → error rate +- `p95(durationMs) group by path` → latency SLO +- `count() group by piece.name, flowRun.status` → per-piece failure rate + +### System snapshot (process metrics on ClickStack) + +Process-level CPU / memory / event-loop lag reach ClickStack **as evlog events**, not native metrics. Both API (`packages/server/api/src/app/helper/system-snapshot.ts`) and worker (`packages/server/worker/src/lib/utils/system-snapshot.ts`) run a `setInterval(..., 60_000)` that emits one wide event with: + +- `event: 'system.snapshot'`, `host` +- `memRssMb`, `memHeapUsedMb`, `memHeapTotalMb` +- `eventLoopDelayP99Ms` (from `perf_hooks.monitorEventLoopDelay`) +- `queueCounts` (BullMQ, API only) + +Emitted with `_forceKeep: true` so info-sampling cannot drop it. Dashboards and alerts for RSS / heap / event-loop lag are built on this event. + +### BullMQ queue depth (real OTLP metrics) + +The only signal that leaves as a **native metric**, not a log. Same 60s tick calls `otelQueueMetrics.push` (`packages/server/api/src/app/helper/otel-queue-metrics.ts`) which POSTs an OTLP gauge `bullmq.job.count` with `queue` + `state` attributes to `OTEL_EXPORTER_OTLP_ENDPOINT/v1/metrics`. Gated by `AP_OTEL_QUEUE_METRICS_ENABLED`. Queue-depth cardinality (queues × states) would bloat wide events, and gauges are the natural shape. + +### Worker → App healthcheck (in-band, not evlog) + +CPU / RAM / disk / per-executor RSS travel from worker to app **piggyback on every `apiClient.poll(machineInfo)` call** — see `buildMachineInfo` in `packages/server/worker/src/lib/worker.ts` (uses `systemUsage` from `packages/server/utils/src/system-usage.ts`). App caches it in `app-machine-cache.ts` and surfaces the fleet at `GET /v1/health/system`. This powers the Platform health page; it does **not** go to ClickStack. + +### What lives where + +| Signal | Source | Destination | +| --- | --- | --- | +| Latency / error / business KPIs | evlog wide events per request/job | ClickStack via drain — log-based metric | +| Process RSS / heap / event-loop lag | `system.snapshot` every 60s | ClickStack via drain — log-based metric | +| BullMQ queue depth by state | `otelQueueMetrics.push` every 60s | OTLP `/v1/metrics` — native gauge | +| Worker CPU / RAM / disk / executors | `apiClient.poll(machineInfo)` | App cache → `GET /v1/health/system` | +| Managed Postgres / Redis CPU / mem | DigitalOcean managed metrics | Betterstack (via `do-alerts@` group) | +| Hetzner host CPU / mem | host agent, if any | not in ClickStack | + +### Gotchas + +- **Missing `_forceKeep` drops your metric.** A recurring info-level snapshot without `_forceKeep: true` is subject to sampling — the value silently thins out at whatever `sampleRateInfo` is set to, and charts read as gaps or an inflated mean. Every periodic-metric event must set it. +- **Field-name drift breaks every log-based metric.** The same id logged as `runId` in one place and `flowRun.id` in another cannot be joined at query time. The schema in `packages/server/CLAUDE.md` is enforced by convention only — the drain will not warn. +- **`hyperdxToken` wins over `otlpEnabled`.** `resolveRemote` picks the first matching drain (Axiom → HyperDX → Loki → Better Stack → OTLP), so setting an Axiom or HyperDX token silently disables `AP_OTEL_ENABLED`. To send logs to a generic OTLP collector, unset the other provider tokens. +- **The Betterstack drain requires both `betterstackToken` and `betterstackHost`.** Setting only one silently falls through to the next drain. +- **Worker machine info is in-band with polling, not scheduled.** If a worker stops polling (crash, network partition), its numbers on the health page freeze — there is no separate heartbeat. + +### Key files + +- `packages/server/utils/src/evlog-setup.ts` — `evlogSetup.init` / `.flush` +- `packages/server/utils/src/evlog-drains.ts` — drain selection +- `packages/server/utils/src/system-usage.ts` — cgroup CPU / RAM readers +- `packages/server/api/src/app/helper/system-snapshot.ts` — API 60s tick +- `packages/server/worker/src/lib/utils/system-snapshot.ts` — worker 60s tick +- `packages/server/api/src/app/helper/otel-queue-metrics.ts` — OTLP gauge push +- `packages/server/api/src/app/helper/app-machine-cache.ts` — fleet cache for `/v1/health/system` diff --git a/brain/knowledge/data-storage-observability/index.md b/brain/knowledge/data-storage-observability/index.md index 26e0360637a1..a2a84c972058 100644 --- a/brain/knowledge/data-storage-observability/index.md +++ b/brain/knowledge/data-storage-observability/index.md @@ -34,6 +34,10 @@ Security-relevant actions persisted to `audit_event`, queryable by platform admi Platform reporting: daily runs, active flows/users, time-saved estimates. `PlatformAnalyticsReport` cached (5-min TTL) refreshed under a distributed lock; separate daily cron (12:00 UTC) tallies per-piece usage into `pieceMetadata.usage`. minutesSaved = runs × flow.timeSavedPerRun. Powers `/impact` (Summary/Trends/Details). Gated by `analyticsEnabled` — NOT in CE. Frontend queries carry `enabled: platform.plan.analyticsEnabled`. +### Logging & Metrics (evlog) + +All structured logging goes through **evlog** — one wide event per unit of work, one remote drain (Axiom / HyperDX / Loki / Better Stack / OTLP; first match wins). Metrics on ClickStack are **log-based**: dashboards and alerts query numeric fields (`durationMs`, `memRssMb`, `eventLoopDelayP99Ms`) on wide events. Both API and worker emit a 60s `system.snapshot` event with process RSS / heap / event-loop lag so those charts exist at all. BullMQ queue depth is the one signal exported as a **native OTLP gauge** (`bullmq.job.count`), because per-queue-per-state cardinality would bloat wide events. Worker CPU / RAM travels in-band on the poll healthcheck (not to ClickStack) and surfaces on `GET /v1/health/system`. + ### Product Telemetry Anonymous product analytics to PostHog, from both the browser and the app container. Gated per platform by `platform_configuration.isProductTelemetryEnabled`, edited at Platform Admin > Infrastructure > Configurations; `AP_TELEMETRY_ENABLED` survives only as the value a platform's row is *born* with. See [000033](../decisions/000033-platform-configuration-rows-are-authoritative-and-created-on-first-read.md). @@ -81,3 +85,4 @@ Streams platform/project events to webhook URLs in real time — internal AP flo - **Knowledge Base** — documents chunked into vector embeddings for AI search - **Analytics** — usage reporting - **Audit Logs** — the persisted security-action record +- **Logging & Metrics (evlog)** — wide events, drains, log-based metrics, `system.snapshot`, OTLP queue gauge diff --git a/packages/server/utils/src/evlog-setup.ts b/packages/server/utils/src/evlog-setup.ts index fd7b7169e2ee..8fa782852db7 100644 --- a/packages/server/utils/src/evlog-setup.ts +++ b/packages/server/utils/src/evlog-setup.ts @@ -1,8 +1,22 @@ -import { auditEnricher, auditRedactPreset, enricherPlugin, initLogger, RedactConfig } from 'evlog' +import os from 'os' +import { isNil } from '@activepieces/core-utils' +import { auditEnricher, auditRedactPreset, DrainContext, enricherPlugin, initLogger, RedactConfig } from 'evlog' import { apLogger, ApLogger } from './ap-logger' import { evlogDrains, EvlogDrainConfig } from './evlog-drains' -// Module-level flush function; replaced each time init() is called. +const HOSTNAME = os.hostname() + +function wrapDrainWithHost( + inner: (ctx: DrainContext) => void | Promise, +): (ctx: DrainContext) => Promise { + return async (ctx: DrainContext) => { + if (isNil(ctx.event['host'])) { + ctx.event['host'] = HOSTNAME + } + await inner(ctx) + } +} + let activeFlusher: (() => Promise) = async () => undefined // Evlog does not support trace/fatal natively; map to nearest equivalents. @@ -124,6 +138,8 @@ function init({ params }: { params: EvlogSetupParams }): ApLogger { const resolved = evlogDrains.resolve({ config: params.drainConfig }) activeFlusher = resolved.flush + const drainWithHost = resolved.drain ? wrapDrainWithHost(resolved.drain) : undefined + initLogger({ env: { service: params.drainConfig.serviceName, version: params.version, environment: params.environment }, pretty: params.logPretty ?? false, @@ -140,9 +156,7 @@ function init({ params }: { params: EvlogSetupParams }): ApLogger { ], }, redact: REDACT_CONFIG, - drain: resolved.drain, - // Fills audit.context (requestId, traceId, ip, userAgent) on audit-bearing - // events only; all other events pass through untouched. + drain: drainWithHost, plugins: [enricherPlugin('audit-context', auditEnricher())], }) @@ -158,6 +172,7 @@ async function flush(): Promise { export const evlogSetup = { init, flush, + wrapDrainWithHost, } export type EvlogSetupParams = { diff --git a/packages/server/utils/test/evlog-host-drain.test.ts b/packages/server/utils/test/evlog-host-drain.test.ts new file mode 100644 index 000000000000..8a009487d270 --- /dev/null +++ b/packages/server/utils/test/evlog-host-drain.test.ts @@ -0,0 +1,45 @@ +import os from 'os' +import { describe, expect, it, vi } from 'vitest' +import { initLogger, log } from 'evlog' +import type { DrainContext, WideEvent } from 'evlog' +import { evlogSetup } from '../src/evlog-setup' + +function makeCtx(event: Partial): DrainContext { + return { event: event as WideEvent } +} + +describe('evlogSetup.wrapDrainWithHost', () => { + it('preserves a host set explicitly at the call site (system.snapshot pattern)', async () => { + const inner = vi.fn().mockResolvedValue(undefined) + const wrapped = evlogSetup.wrapDrainWithHost(inner) + + const ctx = makeCtx({ event: 'system.snapshot', host: 'explicit-host-abc' } as Partial) + await wrapped(ctx) + + expect(inner.mock.calls[0][0].event.host).toBe('explicit-host-abc') + }) +}) + +describe('evlog log.error path (worker shape)', () => { + it('a plain log.error emitted through evlog gets host stamped by the drain wrapper', async () => { + const captured: DrainContext[] = [] + const captureDrain = async (ctx: DrainContext): Promise => { + captured.push({ event: { ...ctx.event } }) + } + + initLogger({ + env: { service: 'test-worker', version: '0.0.0', environment: 'test' }, + minLevel: 'info', + drain: evlogSetup.wrapDrainWithHost(captureDrain), + }) + + log.error({ event: 'job.failed', job: { id: 'j1', type: 'EXECUTE_FLOW' }, error: 'boom' }, 'job.failed') + + await new Promise(resolve => setImmediate(resolve)) + + expect(captured).toHaveLength(1) + expect(captured[0].event.host).toBe(os.hostname()) + expect(captured[0].event.event).toBe('job.failed') + expect(captured[0].event.level).toBe('error') + }) +}) From 735fc110100da322c98e4b5822e846e86a285b2e Mon Sep 17 00:00:00 2001 From: Hazem Adel Date: Tue, 22 Sep 2026 11:53:22 +0300 Subject: [PATCH 3/3] fix(flows): deleted flows no longer get stuck hidden and still running (#15693) --- CLAUDE.md | 11 +- brain/knowledge/flows-execution/flows.md | 3 +- ...000-FlowExternalIdUniqueIgnoresDeleting.ts | 49 ++++ .../1856000000000-AddFlowTombstoneIndex.ts | 27 ++ .../src/app/database/postgres-connection.ts | 4 + .../server/api/src/app/file/file.module.ts | 2 + .../flows/flow/flow-service-side-effects.ts | 5 +- .../api/src/app/flows/flow/flow.controller.ts | 1 + .../api/src/app/flows/flow/flow.entity.ts | 6 + .../api/src/app/flows/flow/flow.jobs.ts | 18 +- .../api/src/app/flows/flow/flow.service.ts | 69 ++--- .../zombie-polling-interceptor.ts | 11 +- .../flows/flow/flow-delete-recovery.test.ts | 254 ++++++++++++++++++ 13 files changed, 398 insertions(+), 62 deletions(-) create mode 100644 packages/server/api/src/app/database/migration/postgres/1855000000000-FlowExternalIdUniqueIgnoresDeleting.ts create mode 100644 packages/server/api/src/app/database/migration/postgres/1856000000000-AddFlowTombstoneIndex.ts create mode 100644 packages/server/api/test/integration/ce/flows/flow/flow-delete-recovery.test.ts diff --git a/CLAUDE.md b/CLAUDE.md index 3bb2019f4341..3d02a89cf98a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -100,9 +100,14 @@ When running in `--mode=cloud`, do not use OAuth2 connections — the OAuth prov - If the PR includes any contributions to pieces (integrations under `packages/pieces`), also add the appropriate pieces label (in addition to the primary label above): - **`🧩 area/third-party-pieces`** — for third-party integrations (most pieces under `packages/pieces/community/`) - **`🧩 area/core-pieces`** — for core pieces (under `packages/pieces/core/`) -- **Always fill the "Breaking change?" section of the PR template** — tick exactly one box (the `breaking-change-check` CI job fails if it is left unedited). A change is breaking if a self-hoster or API consumer must take action: removed/renamed API fields or endpoints, dropped columns, new required fields, removed/required env vars, or default/limit/behaviour changes. If it is breaking: - - also apply the **`⛓️‍💥 breaking-change`** label (in addition to the primary label above), and - - add an entry to `docs/install/reference/breaking-changes.mdx` describing what changed and the action required. CI enforces that the label and the docs entry travel together. +- **The PR template has TWO required sections and `breaking-change-check` fails if either is left unedited** — `### Breaking change?` *and* `### Security impact?`. Tick exactly one box in each (ticking two in one section also fails). Copy the option text from `.github/pull_request_template.md`; the parser matches `- [x] no` / `- [x] yes` under each heading, so keep the `no`/`yes` as the first word after the checkbox. + - `Breaking change?` has **three** options: `no`, `yes — technical` (removed/renamed API field or endpoint, dropped column, new required field, removed/required env var), `yes — functional` (default/limit/behaviour change, new self-hosted setup step). + - `Security impact?` has **two**: `no`, `yes — security-sensitive` (touches authentication, secrets, permissions/roles, cryptography, SSRF / outbound HTTP, or file handling). This one is a **declaration only** — no label, no docs entry, nothing else is coupled to it. +- **If the change is breaking, the label and the docs entry must travel together** — the check's coupling is bidirectional and it also cross-checks the template answer, so label, docs entry and ticked box must all agree: + - apply the **`⛓️‍💥 breaking-change`** label (in addition to the primary label above), **and** + - add an entry to `docs/install/reference/breaking-changes.mdx`. It must have a `####` title **and** at least one body line — a bare heading, a `---`, or loose prose does not count as an entry. + - Ticking `no` while carrying the label fails, and so does adding a docs entry without the label. +- **Never `gh run rerun` a `pull_request`-triggered check to pick up a PR description edit.** A rerun replays the *original* event payload, so it re-reads the stale body and fails again — and because it is a later attempt on an older run, its failure shadows a newer run that already passed. Edit the body (which fires `edited`, a trigger these workflows listen for) and let the fresh run supersede. To check the answer locally without any run at all: `PR_BODY="$(gh pr view --json body --jq .body)" PR_LABELS="$(gh pr view --json labels --jq '[.labels[].name]')" BASE_REF=main npx tsx tools/scripts/breaking-change-check.ts` - **Non-rollbackable migrations are a separate axis** from customer-facing breaking changes: a migration that runs destructive DDL (`DROP TABLE`/`DROP COLUMN`, `ADD ... NOT NULL` without `DEFAULT`, etc.) must set `breaking = true` on the migration class — this is the rollback-safety flag (used by `rollback-migrations.ts` and the release rollback note), enforced by `check-migration-rollback.ts`. It does **not** by itself require the `⛓️‍💥 breaking-change` label; decide that from the upgrade-impact question above. ## Database Migrations diff --git a/brain/knowledge/flows-execution/flows.md b/brain/knowledge/flows-execution/flows.md index b8700f2ae25d..ea6732d5bafa 100644 --- a/brain/knowledge/flows-execution/flows.md +++ b/brain/knowledge/flows-execution/flows.md @@ -36,7 +36,8 @@ Flows are the core automation primitive: a versioned directed graph of trigger + - **`flowVersionSideEffects.preApplyOperation` writes on the default connection, so it escapes any caller transaction.** `handleSampleDataDeletion` and `handleUpdateTriggerWebhookSimulation` take no `entityManager`; a write they make from inside a `transaction()` survives its rollback. They early-return for `IMPORT_FLOW`/`UPDATE_SAMPLE_DATA_INFO`, so the transactional path above is safe today — but adding an operation type to it silently reintroduces partial commits. `updateLastModified` sits outside that swallow-all catch on purpose: a swallowed statement failure inside a transaction poisons it and resurfaces as a confusing "transaction is aborted" error on the next statement. - **`transaction()` (`core/db/transaction.ts`) is a bare `dataSource.transaction()`** — it acquires a *new* connection, not a savepoint. Nesting it deadlocks, so check every caller before wrapping a service method that others may already call inside a transaction. - Step settings split a piece's props into an always-visible **essential** set and a collapsed **Advanced** section: a prop is Advanced only when it sets `advanced: true` (everything else — incl. `MARKDOWN`, tab/section group members, and checkbox reveal targets — stays essential). `propertyGroups` render as tabs, sectioned cards, or the "Add filter" builder. -- **Flows stuck in `DELETING` keep eating the active-flow limit.** Deletion is a durable BullMQ system job (`delete-flow-`), not synchronous: `delete()` sets `operationStatus=DELETING` and enqueues, and the row plus `status=ENABLED` only go away when the job finishes. That job runs `sampleDataService.deleteForFlow`, whose `DELETE FROM file … metadata->>'flowId'=?` had no index — on the large prod `file` table it seq-scans, blows `statement_timeout`, exhausts its 2 attempts and lands **permanently** in the failed set. The flow is then hidden from the UI list (which filters `!=DELETING`) but still counted by the active-flows quota (`getUsage` counts `status=ENABLED`), so Publish silently shows the "Purchase Extra Active Flows" dialog instead of publishing — this is what breaks the `webhook-should-return-response` e2e monitor. Stuck flows are functionally dead (`preDelete` disables the trigger before the failing delete), so forcing their rows away is safe. Fixes on `fix/flow-delete-sample-data-timeout`: a partial expression index `idx_file_sample_data_flow_id` on `file (type, (metadata->>'flowId'))`, plus `operationStatus != DELETING` in the active-flow counts so the quota stops depending on delete-job success. +- **Flows stuck in `DELETING` keep eating the active-flow limit.** Deletion is a durable BullMQ system job (`delete-flow-`), not synchronous: `delete()` sets `operationStatus=DELETING` and enqueues, and the row plus `status=ENABLED` only go away when the job finishes. That job runs `sampleDataService.deleteForFlow`, whose `DELETE FROM file … metadata->>'flowId'=?` had no index — on the large prod `file` table it seq-scans, blows `statement_timeout`, exhausts its 2 attempts and lands **permanently** in the failed set. The flow is then hidden from the UI list (which filters `!=DELETING`) but still counted by the active-flows quota (`getUsage` counts `status=ENABLED`), so Publish silently shows the "Purchase Extra Active Flows" dialog instead of publishing — this is what breaks the `webhook-should-return-response` e2e monitor. Fixes on `fix/flow-delete-sample-data-timeout`: a partial expression index `idx_file_sample_data_flow_id` on `file (type, (metadata->>'flowId'))`, plus `operationStatus != DELETING` in the active-flow counts so the quota stops depending on delete-job success. **Do not assume a stuck flow is functionally dead** — that was the read at the time, and it is wrong whenever the job died *before* `preDelete` ran: cloud prod held 52 stranded rows, 6 still `ENABLED`, the oldest 176 days, and the polling ones were still firing production runs (GIT-1872 / Pylon #5899). GIT-1872 made the state recoverable instead of terminal: `delete()` writes `status=DISABLED` with `DELETING` and invalidates the execution cache, `delete()` accepts a flow already in `DELETING` so a retry re-enqueues (`upsertJob` already retries a failed job and adds a missing one), and a 15-minute `STRANDED_FLOW_DELETION_SWEEP` re-enqueues rows older than 15 minutes — which is the only mechanism that recovers a row whose redis job is gone, since `removeOnFail: { age: ONE_MONTH }` garbage-collects the failed job and no `attempts` value can outlive that. +- **`FlowStatus` does not gate the polling path — `trigger_source` does, and only the delete job removes it.** `executePollingJob` and `renewWebhookJob` never read `flow.status`; their sole gate is `zombiePollingInterceptor.preDispatch`, which allows the job iff a non-soft-deleted `trigger_source` row exists for the `flowVersionId`. That row is soft-deleted inside `flowSideEffects.preDelete`, which runs *inside* the `DELETE_FLOW` job — so for a flow whose delete job is stranded, writing `status=DISABLED` buys nothing and it keeps creating PRODUCTION runs via `worker-rpc-service.submitPayloads` → `flowRunService.start`, neither of which checks a status either. The webhook family *is* gated on `FlowStatus` (`webhook.service.ts`, `app-event-routing.module.ts`), which is why "disable it and the runs stop" reads as true until you try it on a schedule trigger. GIT-1872 added the `operationStatus = DELETING` check to the interceptor rather than to `flowRunService.start`, to keep the per-run hot path free of an extra query. - **The flows list page (`automations/index.tsx`) fetches a capped window and sorts it client-side, so any new "sort by X" has to be pushed into SQL to be correct.** `use-automations-data.ts` calls `flowsApi.list`/`tablesApi.list` with `limit: 1000` (1500 for folder contents, one budget shared across *all* folders) and passes `cursor: undefined` at every call site; all merging, sorting and page-slicing then happens client-side in `automations/lib/utils.ts` (`mergeAndSortItems`, `buildTreeItems`, `buildFilteredTreeItems`, plus a second folder-children sort inside `buildFilteredTreeItems` that is easy to miss). Because the window is capped, re-sorting it in the browser only reorders a *recency-biased slice* — a rarely-touched flow at row 1001 by `updated` never reaches the client, so it can never appear under A–Z. Ordering therefore has to happen before the `LIMIT`, in Postgres. - **The cursor `Paginator` can be made to order by a joined column without denormalizing anything — `addOrderBy` registered on the query builder *before* `paginate()` leads.** `appendPagingQuery` does `new SelectQueryBuilder(builder)` and then only ever calls `addOrderBy`, so a caller's own criteria comes first and the paginator's `status/updated/id` become the tiebreak chain. That means `queryBuilder.addSelect('LOWER(COALESCE(latest_version."displayName", ''))', alias).addOrderBy(alias, order)` sorts flows by name with **no new column, no migration, no index, no paginator change** — despite `OrderByConfig.field` itself being unable to target a join (`buildOrder` and `buildCursorQuery` both hard-prefix `${this.alias}.`). Three things to respect: (1) the keyset `WHERE` is still built from `status/updated/id`, so this is only sound while no cursor is passed — reject `sortBy` + `cursor` with a 400 and return `createPage(data, null)` so no meaningless `next` is minted; (2) the sort alias must be **all-lowercase**, because TypeORM's DISTINCT path re-emits the order criteria *unquoted* in its second `id IN (...)` query, so a mixed-case alias fails only on that second query; (3) that same DISTINCT path (`SelectQueryBuilder` clears the inner ORDER BY and nulls the inner limit) means **no index on `flow` can serve this ORDER BY** — which is exactly why denormalizing a sort column onto `FlowEntity` buys nothing here. - **Any query that reads `flow_version.trigger` across a whole platform is TOAST-bound, not index-bound.** Trigger blobs are jsonb and TOAST'd on rows past ~2 KB, so a report that walks piece steps across all published flows on a platform pays ~2–5 ms of TOAST fetch + JSON parse per flow no matter how tight the WHERE clause is — a 10k-published-flow platform lands at 30 s–1 min. Endpoints of this shape (`platform/pieces-report/pieces-report.controller.ts` is the current example) must page + stream (`Readable.from(async iterable)`) so memory is bounded even when wall clock isn't; the escape hatch for the biggest platforms is a background job with async delivery, unlocked when a real sync request times out. Postgres jsonpath in-DB is not a shortcut — it still reads the whole TOASTed blob and the branch schema (`nextAction`/`children.*`/`onSuccessAction`/`firstLoopAction`/router children) drifts every time the flow shape changes, which is what `flowStructureUtil.getAllSteps` is authoritative over. diff --git a/packages/server/api/src/app/database/migration/postgres/1855000000000-FlowExternalIdUniqueIgnoresDeleting.ts b/packages/server/api/src/app/database/migration/postgres/1855000000000-FlowExternalIdUniqueIgnoresDeleting.ts new file mode 100644 index 000000000000..8b5120c8ff19 --- /dev/null +++ b/packages/server/api/src/app/database/migration/postgres/1855000000000-FlowExternalIdUniqueIgnoresDeleting.ts @@ -0,0 +1,49 @@ +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 FlowExternalIdUniqueIgnoresDeleting1855000000000 implements Migration { + name = 'FlowExternalIdUniqueIgnoresDeleting1855000000000' + breaking = false + release = '0.92.0' + transaction = false + + public async up(queryRunner: QueryRunner): Promise { + const concurrently = isPGlite() ? '' : 'CONCURRENTLY ' + await queryRunner.query('DROP INDEX IF EXISTS "idx_flow_project_id_external_id_partial"') + await queryRunner.query(` + CREATE UNIQUE INDEX ${concurrently}"idx_flow_project_id_external_id_partial" + ON "flow" ("projectId", "externalId") + WHERE "operationStatus" != 'DELETING' + `) + await queryRunner.query('DROP INDEX IF EXISTS "idx_flow_project_id_external_id"') + await queryRunner.query('ALTER INDEX "idx_flow_project_id_external_id_partial" RENAME TO "idx_flow_project_id_external_id"') + } + + public async down(queryRunner: QueryRunner): Promise { + 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('DROP INDEX IF EXISTS "idx_flow_project_id_external_id"') + await queryRunner.query(` + CREATE UNIQUE INDEX IF NOT EXISTS "idx_flow_project_id_external_id" + ON "flow" ("projectId", "externalId") + `) + } +} + +const isPGlite = (): boolean => system.get(AppSystemProp.DB_TYPE) === DatabaseType.PGLITE diff --git a/packages/server/api/src/app/database/migration/postgres/1856000000000-AddFlowTombstoneIndex.ts b/packages/server/api/src/app/database/migration/postgres/1856000000000-AddFlowTombstoneIndex.ts new file mode 100644 index 000000000000..7643151627a2 --- /dev/null +++ b/packages/server/api/src/app/database/migration/postgres/1856000000000-AddFlowTombstoneIndex.ts @@ -0,0 +1,27 @@ +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 AddFlowTombstoneIndex1856000000000 implements Migration { + name = 'AddFlowTombstoneIndex1856000000000' + breaking = false + release = '0.92.0' + transaction = false + + public async up(queryRunner: QueryRunner): Promise { + const concurrently = isPGlite() ? '' : 'CONCURRENTLY ' + await queryRunner.query(` + CREATE INDEX ${concurrently}IF NOT EXISTS "idx_flow_deleting_updated" + ON "flow" ("updated") + WHERE "operationStatus" = 'DELETING' + `) + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query('DROP INDEX IF EXISTS "idx_flow_deleting_updated"') + } +} + +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 063b86280d79..1a62566fef1d 100644 --- a/packages/server/api/src/app/database/postgres-connection.ts +++ b/packages/server/api/src/app/database/postgres-connection.ts @@ -440,6 +440,8 @@ import { AddMcpActivity1844000000000 } from './migration/postgres/1844000000000- import { NullOrphanFolderReferences1846000000000 } from './migration/postgres/1846000000000-NullOrphanFolderReferences' import { AddAgentConversationAiCredits1850000000000 } from './migration/postgres/1850000000000-AddAgentConversationAiCredits' import { AddMcpOAuthCodeNonce1851000000000 } from './migration/postgres/1851000000000-AddMcpOAuthCodeNonce' +import { FlowExternalIdUniqueIgnoresDeleting1855000000000 } from './migration/postgres/1855000000000-FlowExternalIdUniqueIgnoresDeleting' +import { AddFlowTombstoneIndex1856000000000 } from './migration/postgres/1856000000000-AddFlowTombstoneIndex' const getSslConfig = (): boolean | TlsOptions => { const useSsl = system.get(AppSystemProp.POSTGRES_USE_SSL) @@ -895,6 +897,8 @@ export const getMigrations = (): (new () => Migration)[] => { NullOrphanFolderReferences1846000000000, AddAgentConversationAiCredits1850000000000, AddMcpOAuthCodeNonce1851000000000, + FlowExternalIdUniqueIgnoresDeleting1855000000000, + AddFlowTombstoneIndex1856000000000, ] return migrations } diff --git a/packages/server/api/src/app/file/file.module.ts b/packages/server/api/src/app/file/file.module.ts index a34d0956e357..f0f9aef29dad 100644 --- a/packages/server/api/src/app/file/file.module.ts +++ b/packages/server/api/src/app/file/file.module.ts @@ -2,6 +2,7 @@ import { FileType } from '@activepieces/shared' import { FastifyPluginAsyncZod } from 'fastify-type-provider-zod' import { entitiesMustBeOwnedByCurrentProject } from '../authentication/authorization' import { agentRetention } from '../ee/agent/agent-retention' +import { flowBackgroundJobs } from '../flows/flow/flow.jobs' import { SystemJobName } from '../helper/system-jobs/common' import { systemJobHandlers } from '../helper/system-jobs/job-handlers' import { systemJobsSchedule } from '../helper/system-jobs/system-job' @@ -15,6 +16,7 @@ export const fileModule: FastifyPluginAsyncZod = async (app) => { await fileService(app.log).deleteStaleBulk([FileType.FLOW_RUN_LOG, FileType.FLOW_RUN_LOG_SLICE, FileType.FLOW_STEP_FILE, FileType.TRIGGER_EVENT_FILE, FileType.TRIGGER_PAYLOAD, FileType.WEBHOOK_PAYLOAD, FileType.MCP_CALL_PAYLOAD]) await agentRetention(app.log).deleteStaleFlowStepConversations() await mcpActivityService(app.log).deleteStale() + await flowBackgroundJobs(app.log).reapTombstonedFlows() }) await systemJobsSchedule(app.log).upsertJob({ job: { diff --git a/packages/server/api/src/app/flows/flow/flow-service-side-effects.ts b/packages/server/api/src/app/flows/flow/flow-service-side-effects.ts index 19f368c5d21d..a3d891e2b9e3 100644 --- a/packages/server/api/src/app/flows/flow/flow-service-side-effects.ts +++ b/packages/server/api/src/app/flows/flow/flow-service-side-effects.ts @@ -38,10 +38,7 @@ export const flowSideEffects = (log: FastifyBaseLogger) => ({ }, async preDelete({ flowToDelete }: PreDeleteParams): Promise { - if ( - flowToDelete.status === FlowStatus.DISABLED || - isNil(flowToDelete.publishedVersionId) - ) { + if (isNil(flowToDelete.publishedVersionId)) { return } await triggerSourceService(log).disable({ 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 9edc131f37d1..f4812adf393e 100644 --- a/packages/server/api/src/app/flows/flow/flow.controller.ts +++ b/packages/server/api/src/app/flows/flow/flow.controller.ts @@ -145,6 +145,7 @@ export const flowController: FastifyPluginAsyncZod = async (app) => { const flow = await flowService(request.log).getOnePopulatedOrThrow({ id: request.params.id, projectId: request.projectId, + includeDeleting: true, }) await gitRepoService(request.log).onDeleted({ type: GitPushOperationType.DELETE_FLOW, 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 475f3ae8d689..17cbd372f994 100644 --- a/packages/server/api/src/app/flows/flow/flow.entity.ts +++ b/packages/server/api/src/app/flows/flow/flow.entity.ts @@ -105,6 +105,12 @@ export const FlowEntity = new EntitySchema({ name: 'idx_flow_project_id_external_id', columns: ['projectId', 'externalId'], unique: true, + where: `"operationStatus" != '${FlowOperationStatus.DELETING}'`, + }, + { + name: 'idx_flow_deleting_updated', + columns: ['updated'], + where: `"operationStatus" = '${FlowOperationStatus.DELETING}'`, }, ], relations: { diff --git a/packages/server/api/src/app/flows/flow/flow.jobs.ts b/packages/server/api/src/app/flows/flow/flow.jobs.ts index 987f48768dc1..2fe8dbfb2b9c 100644 --- a/packages/server/api/src/app/flows/flow/flow.jobs.ts +++ b/packages/server/api/src/app/flows/flow/flow.jobs.ts @@ -1,4 +1,5 @@ import { assertNotNullOrUndefined } from '@activepieces/core-utils' +import { FlowOperationStatus } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' import { repoFactory } from '../../core/db/repo-factory' import { SystemJobData, SystemJobName } from '../../helper/system-jobs/common' @@ -9,10 +10,12 @@ import { flowVersionRepo } from '../flow-version/flow-version.service' import { flowExecutionCache } from './flow-execution-cache' import { flowSideEffects } from './flow-service-side-effects' import { flowRepo } from './flow.repo' +import { flowService } from './flow.service' const waitpointRepo = repoFactory(WaitpointEntity) const BATCH_SIZE = 1000 +const TOMBSTONE_REAP_LIMIT = 50 export async function batchDeleteByFlowId(flowId: string): Promise { while (true) { @@ -75,4 +78,17 @@ export const flowBackgroundJobs = (log: FastifyBaseLogger) => ({ await flowExecutionCache(log).invalidate(flow.id) }, -}) \ No newline at end of file + reapTombstonedFlows: async () => { + const tombstoned = await flowRepo().find({ + where: { operationStatus: FlowOperationStatus.DELETING }, + order: { updated: 'ASC' }, + take: TOMBSTONE_REAP_LIMIT, + }) + if (tombstoned.length === 0) { + return + } + log.warn({ flowCount: tombstoned.length }, '[reapTombstonedFlows] Re-queueing deletions that never finished') + await Promise.all(tombstoned.map((flow) => flowService(log).addDeleteFlowJob(flow))) + }, + +}) 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 97a068aa05b2..8a5af5b4700a 100644 --- a/packages/server/api/src/app/flows/flow/flow.service.ts +++ b/packages/server/api/src/app/flows/flow/flow.service.ts @@ -29,6 +29,8 @@ import { flowRepo } from './flow.repo' +const notDeleting = { operationStatus: Not(FlowOperationStatus.DELETING) } + export const flowService = (log: FastifyBaseLogger) => ({ async create({ projectId, request, externalId, ownerId, templateId, createdBy, ip, emitEvents = true }: CreateParams): Promise { await assertExternalIdIsUnique({ projectId, externalId }) @@ -121,7 +123,7 @@ export const flowService = (log: FastifyBaseLogger) => ({ }, }) - const queryBuilder = flowRepo().createQueryBuilder('ff').where({ operationStatus: Not(FlowOperationStatus.DELETING) }) + const queryBuilder = flowRepo().createQueryBuilder('ff').where(notDeleting) if (projectIds) { queryBuilder.andWhere({ projectId: In(projectIds) }) @@ -234,11 +236,13 @@ export const flowService = (log: FastifyBaseLogger) => ({ return flowRepo().existsBy({ id, projectId, + ...notDeleting, }) }, async getOneById(id: string): Promise { const flow = await flowRepo().findOneBy({ id, + ...notDeleting, }) if (isNil(flow)) { return null @@ -251,7 +255,7 @@ export const flowService = (log: FastifyBaseLogger) => ({ } return flow }, - async getOne({ id, projectId, entityManager }: GetOneParams): Promise { + async getOne({ id, projectId, entityManager, includeDeleting = false }: GetOneParams): Promise { const projectExists = await projectService(log).exists({ projectId, }) @@ -261,6 +265,7 @@ export const flowService = (log: FastifyBaseLogger) => ({ return flowRepo(entityManager).findOneBy({ id, projectId, + ...(includeDeleting ? {} : notDeleting), }) }, @@ -277,18 +282,10 @@ export const flowService = (log: FastifyBaseLogger) => ({ removeConnectionsName = false, removeSampleData = false, entityManager, + includeDeleting = false, }: GetOnePopulatedParams): Promise { - const flow = await flowRepo(entityManager).findOne({ - where: { - id, - projectId, - }, - }) - - const projectExists = await projectService(log).exists({ - projectId, - }) - if (isNil(flow) || !projectExists) { + const flow = await this.getOne({ id, projectId, entityManager, includeDeleting }) + if (isNil(flow)) { return null } @@ -316,22 +313,8 @@ export const flowService = (log: FastifyBaseLogger) => ({ } }, - async getOnePopulatedOrThrow({ - id, - projectId, - versionId, - removeConnectionsName = false, - removeSampleData = false, - entityManager, - }: GetOnePopulatedParams): Promise { - const flow = await this.getOnePopulated({ - id, - projectId, - versionId, - removeConnectionsName, - removeSampleData, - entityManager, - }) + async getOnePopulatedOrThrow(params: GetOnePopulatedParams): Promise { + const flow = await this.getOnePopulated(params) assertFlowIsNotNull(flow) return flow @@ -357,14 +340,6 @@ export const flowService = (log: FastifyBaseLogger) => ({ id, projectId, }) - if (flow.operationStatus === FlowOperationStatus.DELETING) { - throw new ActivepiecesError({ - code: ErrorCode.FLOW_OPERATION_IN_PROGRESS, - params: { - message: 'This flow is getting deleted.', - }, - }) - } if (operation.type === FlowOperationType.LOCK_AND_PUBLISH && flow.status === FlowStatus.ENABLED && !isNil(flow.publishedVersionId)) { previouslyPublishedVersion = await flowVersionService(log).getFlowVersionOrThrow({ flowId: id, versionId: flow.publishedVersionId }) } @@ -570,24 +545,19 @@ export const flowService = (log: FastifyBaseLogger) => ({ async delete({ id, projectId, previousFlow, userId, ip, emitEvents = true }: DeleteParams): Promise { const deletedFlow = emitEvents - ? previousFlow ?? await this.getOnePopulatedOrThrow({ id, projectId }) + ? previousFlow ?? await this.getOnePopulatedOrThrow({ id, projectId, includeDeleting: true }) : undefined const flow = await this.getOneOrThrow({ id, projectId, + includeDeleting: true, }) - if (flow.operationStatus !== FlowOperationStatus.NONE) { - throw new ActivepiecesError({ - code: ErrorCode.FLOW_OPERATION_IN_PROGRESS, - params: { - message: `Flow ${id} is already being ${flow.operationStatus}`, - }, - }) - } - await this.addDeleteFlowJob(flow) - await flowRepo().update(id, { + await flowRepo().update({ id, projectId }, { + status: FlowStatus.DISABLED, operationStatus: FlowOperationStatus.DELETING, }) + await flowExecutionCache(log).invalidate(id) + await this.addDeleteFlowJob(flow) log.info({ flow: { id }, project: { id: projectId } }, 'Flow deletion requested') if (!isNil(deletedFlow)) { flowSideEffects(log).onDeleted({ @@ -873,7 +843,7 @@ async function assertExternalIdIsUnique({ projectId, externalId }: { projectId: if (isNil(externalId)) { return } - const exists = await flowRepo().existsBy({ projectId, externalId }) + const exists = await flowRepo().existsBy({ projectId, externalId, ...notDeleting }) if (exists) { throw new ActivepiecesError({ code: ErrorCode.FLOW_EXTERNAL_ID_ALREADY_EXISTS, @@ -916,6 +886,7 @@ type GetOneParams = { id: FlowId projectId: ProjectId entityManager?: EntityManager + includeDeleting?: boolean } type GetOnePopulatedParams = GetOneParams & { diff --git a/packages/server/api/src/app/workers/job-queue/interceptors/zombie-polling-interceptor.ts b/packages/server/api/src/app/workers/job-queue/interceptors/zombie-polling-interceptor.ts index 9b382c23e5f6..a0d17b3c9761 100644 --- a/packages/server/api/src/app/workers/job-queue/interceptors/zombie-polling-interceptor.ts +++ b/packages/server/api/src/app/workers/job-queue/interceptors/zombie-polling-interceptor.ts @@ -1,6 +1,7 @@ import { isNil } from '@activepieces/core-utils' -import { JobData, PollingJobData, RenewWebhookJobData, WorkerJobType } from '@activepieces/shared' +import { FlowOperationStatus, JobData, PollingJobData, RenewWebhookJobData, WorkerJobType } from '@activepieces/shared' import { FastifyBaseLogger } from 'fastify' +import { flowRepo } from '../../../flows/flow/flow.repo' import { triggerSourceRepo } from '../../../trigger/trigger-source/trigger-source-service' import { InterceptorResult, InterceptorVerdict, JobInterceptor } from '../job-interceptor' import { jobQueue } from '../job-queue' @@ -12,14 +13,16 @@ export const zombiePollingInterceptor: JobInterceptor = { if (!ZOMBIE_REPEATING_JOB_TYPES.includes(jobData.jobType)) { return { verdict: InterceptorVerdict.ALLOW } } - const { flowVersionId } = jobData as PollingJobData | RenewWebhookJobData + const { flowId, flowVersionId } = jobData as PollingJobData | RenewWebhookJobData // An active trigger source exists only when the flow is enabled and this exact version is current. // If soft-deleted (disabled or re-published to a new version), findOneBy returns null. const activeTriggerSource = await triggerSourceRepo().findOneBy({ flowVersionId }) - if (!isNil(activeTriggerSource)) { + const flowIsBeingDeleted = await flowRepo().existsBy({ id: flowId, operationStatus: FlowOperationStatus.DELETING }) + const triggerIsStillLive = !isNil(activeTriggerSource) && !flowIsBeingDeleted + if (triggerIsStillLive) { return { verdict: InterceptorVerdict.ALLOW } } - log.warn({ flowVersion: { id: flowVersionId } }, '[zombiePollingInterceptor] No active trigger source — discarding repeat job (flow disabled, re-published, or deleted)') + log.warn({ flow: { id: flowId }, flowVersion: { id: flowVersionId } }, '[zombiePollingInterceptor] No live trigger source — discarding repeat job (flow disabled, re-published, or deleted)') await jobQueue(log).removeRepeatingJob({ flowVersionId }) return { verdict: InterceptorVerdict.DISCARD } }, diff --git a/packages/server/api/test/integration/ce/flows/flow/flow-delete-recovery.test.ts b/packages/server/api/test/integration/ce/flows/flow/flow-delete-recovery.test.ts new file mode 100644 index 000000000000..945a0701e0d4 --- /dev/null +++ b/packages/server/api/test/integration/ce/flows/flow/flow-delete-recovery.test.ts @@ -0,0 +1,254 @@ +import { apId } from '@activepieces/core-utils' +import { TriggerStrategy } from '@activepieces/pieces-framework' +import { Flow, FlowOperationStatus, FlowStatus, FlowTriggerType, LATEST_JOB_DATA_SCHEMA_VERSION, WorkerJobType } from '@activepieces/shared' +import dayjs from 'dayjs' +import { Job } from 'bullmq' +import { FastifyInstance } from 'fastify' +import { StatusCodes } from 'http-status-codes' +import { flowSideEffects } from '../../../../../src/app/flows/flow/flow-service-side-effects' +import { flowBackgroundJobs } from '../../../../../src/app/flows/flow/flow.jobs' +import { InterceptorVerdict } from '../../../../../src/app/workers/job-queue/job-interceptor' +import { zombiePollingInterceptor } from '../../../../../src/app/workers/job-queue/interceptors/zombie-polling-interceptor' +import { db } from '../../../../helpers/db' +import { createMockFlow, createMockFlowVersion } from '../../../../helpers/mocks' +import { createTestContext, TestContext } from '../../../../helpers/test-context' +import { setupTestEnvironment, teardownTestEnvironment } from '../../../../helpers/test-setup' + +let app: FastifyInstance | null = null + +beforeAll(async () => { + app = await setupTestEnvironment() +}) + +afterAll(async () => { + await teardownTestEnvironment() +}) + +async function savePublishedFlow(ctx: TestContext, flow?: Partial): Promise { + const mockFlow = createMockFlow({ + projectId: ctx.project.id, + status: FlowStatus.ENABLED, + ...flow, + }) + await db.save('flow', mockFlow) + const mockVersion = createMockFlowVersion({ flowId: mockFlow.id }) + await db.save('flow_version', mockVersion) + await db.update('flow', mockFlow.id, { publishedVersionId: mockVersion.id }) + return mockFlow +} + +async function waitForRowToDisappear(flowId: string): Promise { + for (let attempt = 0; attempt < 40; attempt++) { + const row = await db.findOneBy('flow', { id: flowId }) + if (row === null) { + return null + } + await new Promise(resolve => setTimeout(resolve, 100)) + } + return db.findOneBy('flow', { id: flowId }) +} + +async function savePollingFlowWithLiveTrigger(ctx: TestContext, flow?: Partial): Promise<{ flowId: string, flowVersionId: string }> { + const mockFlow = createMockFlow({ projectId: ctx.project.id, status: FlowStatus.ENABLED, ...flow }) + await db.save('flow', mockFlow) + const mockVersion = createMockFlowVersion({ flowId: mockFlow.id }) + await db.save('flow_version', mockVersion) + await db.update('flow', mockFlow.id, { publishedVersionId: mockVersion.id }) + await db.save('trigger_source', { + id: apId(), + created: new Date().toISOString(), + updated: new Date().toISOString(), + flowId: mockFlow.id, + flowVersionId: mockVersion.id, + projectId: ctx.project.id, + pieceName: '@activepieces/piece-schedule', + pieceVersion: '0.1.0', + triggerName: 'every_hour', + type: TriggerStrategy.POLLING, + simulate: false, + schedule: null, + deleted: null, + }) + return { flowId: mockFlow.id, flowVersionId: mockVersion.id } +} + +async function dispatchPollingJob(ctx: TestContext, flowId: string, flowVersionId: string): Promise { + const result = await zombiePollingInterceptor.preDispatch({ + jobId: apId(), + jobData: { + jobType: WorkerJobType.EXECUTE_POLLING, + projectId: ctx.project.id, + platformId: ctx.platform.id, + schemaVersion: LATEST_JOB_DATA_SCHEMA_VERSION, + flowId, + flowVersionId, + triggerType: FlowTriggerType.PIECE, + }, + job: {} as Job, + log: app!.log, + }) + return result.verdict +} + +describe('Flow deletion recovery', () => { + it('disables the flow before the delete job runs, so its trigger stops firing', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx) + + const response = await ctx.delete(`/v1/flows/${flow.id}`) + + expect(response?.statusCode).toBe(StatusCodes.NO_CONTENT) + const row = await db.findOneBy('flow', { id: flow.id }) + expect(row?.status ?? FlowStatus.DISABLED).toBe(FlowStatus.DISABLED) + }) + + it('accepts a second delete for a flow already stranded in DELETING', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx, { operationStatus: FlowOperationStatus.DELETING }) + + const response = await ctx.delete(`/v1/flows/${flow.id}`) + + expect(response?.statusCode).toBe(StatusCodes.NO_CONTENT) + expect(await waitForRowToDisappear(flow.id)).toBeNull() + }) + + it('reaps a flow whose delete job was lost, without a user asking again', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx, { operationStatus: FlowOperationStatus.DELETING }) + await db.update('flow', flow.id, { updated: dayjs().subtract(1, 'day').toISOString() }) + + await flowBackgroundJobs(app!.log).reapTombstonedFlows() + + expect(await waitForRowToDisappear(flow.id)).toBeNull() + }) + + it('discards the polling job of a flow being deleted, so it stops firing runs', async () => { + const ctx = await createTestContext(app!) + const { flowId, flowVersionId } = await savePollingFlowWithLiveTrigger(ctx, { operationStatus: FlowOperationStatus.DELETING }) + + expect(await dispatchPollingJob(ctx, flowId, flowVersionId)).toBe(InterceptorVerdict.DISCARD) + }) + + it('still dispatches the polling job of a flow that is not being deleted', async () => { + const ctx = await createTestContext(app!) + const { flowId, flowVersionId } = await savePollingFlowWithLiveTrigger(ctx) + + expect(await dispatchPollingJob(ctx, flowId, flowVersionId)).toBe(InterceptorVerdict.ALLOW) + }) + + it('tears the trigger source down even though delete() already wrote DISABLED', async () => { + const ctx = await createTestContext(app!) + const { flowId } = await savePollingFlowWithLiveTrigger(ctx, { status: FlowStatus.DISABLED }) + const flowToDelete = await db.findOneByOrFail('flow', { id: flowId }) + + await flowSideEffects(app!.log).preDelete({ flowToDelete }) + + const triggerSource = await db.findOneBy<{ deleted: string | null }>('trigger_source', { flowId }) + expect(triggerSource).toBeNull() + }) + + it('keeps reaping the rest when one flow cannot be reclaimed', async () => { + const ctx = await createTestContext(app!) + const doomed = createMockFlow({ + projectId: ctx.project.id, + status: FlowStatus.DISABLED, + operationStatus: FlowOperationStatus.DELETING, + }) + await db.save('flow', doomed) + const doomedVersion = createMockFlowVersion({ flowId: doomed.id }) + await db.save('flow_version', doomedVersion) + await db.update('flow', doomed.id, { publishedVersionId: doomedVersion.id }) + await db.save('trigger_source', { + id: apId(), + created: new Date().toISOString(), + updated: new Date().toISOString(), + flowId: doomed.id, + flowVersionId: apId(), + projectId: ctx.project.id, + pieceName: '@activepieces/piece-schedule', + pieceVersion: '0.1.0', + triggerName: 'every_hour', + type: TriggerStrategy.POLLING, + simulate: false, + schedule: null, + }) + const healthy = await savePublishedFlow(ctx, { operationStatus: FlowOperationStatus.DELETING }) + await db.update('flow', doomed.id, { updated: dayjs().subtract(2, 'hour').toISOString() }) + await db.update('flow', healthy.id, { updated: dayjs().subtract(2, 'hour').toISOString() }) + + await flowBackgroundJobs(app!.log).reapTombstonedFlows() + + expect(await waitForRowToDisappear(healthy.id)).toBeNull() + expect(await db.findOneBy('flow', { id: doomed.id })).not.toBeNull() + }) + + it('is gone to every reader the moment the request returns, not when the job lands', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx, { operationStatus: FlowOperationStatus.DELETING }) + + const opened = await ctx.get(`/v1/flows/${flow.id}`) + const listed = await ctx.get('/v1/flows', { projectId: ctx.project.id }) + + expect(opened?.statusCode).toBe(StatusCodes.NOT_FOUND) + expect(listed?.json().data.map((f: Flow) => f.id)).not.toContain(flow.id) + }) + + it('frees the external id so the flow can be recreated while the tombstone lingers', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx, { operationStatus: FlowOperationStatus.DELETING }) + + const response = await ctx.post('/v1/flows', { + displayName: 'recreated', + projectId: ctx.project.id, + externalId: flow.externalId, + }, { query: { projectId: ctx.project.id } }) + + expect(response?.statusCode).toBe(StatusCodes.CREATED) + expect(response?.json().externalId).toBe(flow.externalId) + }) + + it('survives concurrent deletes and a reaper pass racing the same flow', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx) + + const responses = await Promise.all([ + ctx.delete(`/v1/flows/${flow.id}`), + ctx.delete(`/v1/flows/${flow.id}`), + ctx.delete(`/v1/flows/${flow.id}`), + flowBackgroundJobs(app!.log).reapTombstonedFlows(), + ]) + + for (const response of responses.slice(0, 3)) { + expect(response).not.toBeUndefined() + expect([StatusCodes.NO_CONTENT, StatusCodes.NOT_FOUND]).toContain((response as { statusCode: number }).statusCode) + } + expect(await waitForRowToDisappear(flow.id)).toBeNull() + expect(await db.findBy('flow_version', { flowId: flow.id })).toHaveLength(0) + expect(await db.findBy('trigger_source', { flowId: flow.id })).toHaveLength(0) + }) + + it('is safe to tear the trigger source down twice', async () => { + const ctx = await createTestContext(app!) + const { flowId } = await savePollingFlowWithLiveTrigger(ctx, { status: FlowStatus.DISABLED }) + const flowToDelete = await db.findOneByOrFail('flow', { id: flowId }) + + await flowSideEffects(app!.log).preDelete({ flowToDelete }) + const afterFirst = await db.findBy('trigger_source', { flowId }) + + await flowSideEffects(app!.log).preDelete({ flowToDelete }) + const afterSecond = await db.findBy('trigger_source', { flowId }) + + expect(afterFirst).toHaveLength(0) + expect(afterSecond).toHaveLength(0) + }) + + it('reclaims a just-requested deletion rather than waiting on a clock', async () => { + const ctx = await createTestContext(app!) + const flow = await savePublishedFlow(ctx, { operationStatus: FlowOperationStatus.DELETING }) + + await flowBackgroundJobs(app!.log).reapTombstonedFlows() + await flowBackgroundJobs(app!.log).reapTombstonedFlows() + + expect(await waitForRowToDisappear(flow.id)).toBeNull() + }) +})