feat: piece branching

docs/developers/piece-reference/piece-branching.mdx

@@ -0,0 +1,57 @@
+---
+title: 'Piece Branching'
+icon: 'hammer'
+description: 'Make a piece branchable'
+---
+Activepieces allows pieces to be directly branched from depending on the action by adding the expected outputs in the `outputs` array of objects when creating an action.
+
+### Waiting for approval action
+
+We will see the branching feature within the context of the approval process, using the waiting for approval action as an example.
+
+**Example:**
+
+```typescript
+import { createAction } from '@activepieces/pieces-framework';
+import { ExecutionType, PauseType, branchedPieceResponse } from '@activepieces/shared';
+
+export const waitForApprovalLink = createAction({
+ name: 'wait_for_approval',
+ displayName: 'Wait for Approval',
+ description: 'Pauses the flow and wait for the approval from the user',
+ props: {},
+ errorHandlingOptions: {
+ continueOnFailure: {
+ hide: true,
+ },
+ retryOnFailure: {
+ hide: true,
+ },
+ },
+ // Here we define the names of the outputs expected to be branched from the piece
+ outputs: [
+ { name: 'approved' }, 
+ { name: 'denied' }
+ ],
+ async run(ctx) {
+ if (ctx.executionType === ExecutionType.BEGIN) {
+ ctx.run.pause({
+ pauseMetadata: {
+ type: PauseType.WEBHOOK,
+ response: {},
+ },
+ });
+
+ // Return an empty Map while execution is paused
+ return branchedPieceResponse()
+ } else {
+ // Return a Map populated with the expected returned data
+ return branchedPieceResponse({
+ version: 'v1', // Optional: adding a version for the expected response. Defaults to 'v1'.
+ approved: ctx.resumePayload.queryParams['action'] === 'approve',
+ denied: ctx.resumePayload.queryParams['action'] !== 'approve',
+ })
+ }
+ },
+});
+```

packages/engine/src/lib/handler/piece-executor.ts

@@ -1,5 +1,5 @@
 import { URL } from 'url'
-import { ActionContext, ConnectionsManager, PauseHook, PauseHookParams, PiecePropertyMap, StaticPropsValue, StopHook, StopHookParams, TagsManager } from '@activepieces/pieces-framework'
+import { Action, ActionContext, ConnectionsManager, PauseHook, PauseHookParams, PiecePropertyMap, RunFunctionReturnType, StaticPropsValue, StopHook, StopHookParams, TagsManager } from '@activepieces/pieces-framework'
 import { ActionType, assertNotNullOrUndefined, AUTHENTICATION_PROPERTY_NAME, ExecutionType, FlowRunStatus, GenericStepOutput, isNil, PauseType, PieceAction, StepOutputStatus } from '@activepieces/shared'
 import { continueIfFailureHandler, handleExecutionError, runWithExponentialBackoff } from '../helper/error-handling'
 import { pieceLoader } from '../helper/piece-loader'
@@ -9,9 +9,19 @@ import { createContextStore } from '../services/storage.service'
 import { ActionHandler, BaseExecutor } from './base-executor'
 import { EngineConstants } from './context/engine-constants'
 import { ExecutionVerdict, FlowExecutorContext } from './context/flow-execution-context'
+import { flowExecutor } from './flow-executor'
 
 type HookResponse = { stopResponse: StopHookParams | undefined, pauseResponse: PauseHookParams | undefined, tags: string[], stopped: boolean, paused: boolean }
 
+type RunBranchablePieceWithVersion = {
+ version: 'v1'
+ pieceOutput: RunFunctionReturnType
+ executionState: FlowExecutorContext
+ action: PieceAction
+ constants: EngineConstants
+ stepOutput: GenericStepOutput<ActionType.PIECE, unknown>
+}
+
 export const pieceExecutor: BaseExecutor<PieceAction> = {
  async handle({
  action,
@@ -128,6 +138,18 @@ const executeAction: ActionHandler<PieceAction> = async ({ action, executionStat
  pauseMetadata: hookResponse.pauseResponse.pauseMetadata,
  })
  }
+ if (action.children && hasBranches(pieceAction)) {
+ const pieceOutput = output as RunFunctionReturnType
+
+ return await runBranchablePieceWithVersion({
+ executionState,
+ pieceOutput,
+ action,
+ constants,
+ stepOutput,
+ version: pieceOutput.version,
+ })
+ }
 
  return newExecutionContext.upsertStep(action.name, stepOutput.setOutput(output)).increaseTask().setVerdict(ExecutionVerdict.RUNNING, undefined)
  }
@@ -144,6 +166,45 @@ const executeAction: ActionHandler<PieceAction> = async ({ action, executionStat
  }
 }
 
+function hasBranches(pieceAction: Action): boolean {
+ return !isNil(pieceAction.outputs) && pieceAction.outputs.length > 1
+}
+
+async function runBranchablePieceWithVersion({ 
+ pieceOutput,
+ executionState, 
+ action, 
+ constants,
+ stepOutput, 
+ version = 'v1', 
+}: RunBranchablePieceWithVersion): Promise<FlowExecutorContext> {
+ const versions = {
+ v1: async (): Promise<FlowExecutorContext> => {
+ let newExecutionContext = executionState
+ let outputValue = undefined
+ for (const [k, v] of pieceOutput.output.entries()) {
+ if (v === false || v === undefined || isNil(v)) {
+ continue
+ }
+
+ if (isNil(action.children)) {
+ continue
+ }
+
+ newExecutionContext = await flowExecutor.execute({
+ action: action.children.filter(child => child.name === k)[0].action,
+ executionState: newExecutionContext,
+ constants,
+ })
+ outputValue = v
+ }
+ return newExecutionContext.upsertStep(action.name, stepOutput.setOutput(outputValue)).increaseTask().setVerdict(ExecutionVerdict.RUNNING, undefined)
+ },
+ }
+
+ return versions[version]()
+}
+
 const createTagsManager = (hookResponse: HookResponse): TagsManager => {
  return {
  add: async (params: {

packages/pieces/community/approval/src/lib/actions/wait-for-approval.ts

@@ -1,35 +1,38 @@
-import { createAction } from '@activepieces/pieces-framework';
-import { ExecutionType, PauseType } from '@activepieces/shared';
-
-export const waitForApprovalLink = createAction({
- name: 'wait_for_approval',
- displayName: 'Wait for Approval',
- description: 'Pauses the flow and wait for the approval from the user',
- props: {},
- errorHandlingOptions: {
- continueOnFailure: {
- hide: true,
- },
- retryOnFailure: {
- hide: true,
- },
- },
- async run(ctx) {
- if (ctx.executionType === ExecutionType.BEGIN) {
- ctx.run.pause({
- pauseMetadata: {
- type: PauseType.WEBHOOK,
- response: {}
- },
- });
-
- return {
- approved: true,
- };
- } else {
- return {
- approved: ctx.resumePayload.queryParams['action'] === 'approve',
- };
- }
- },
-});
+import { createAction } from '@activepieces/pieces-framework';
+import { ExecutionType, PauseType, branchedPieceResponse } from '@activepieces/shared';
+
+export const waitForApprovalLink = createAction({
+ name: 'wait_for_approval',
+ displayName: 'Wait for Approval',
+ description: 'Pauses the flow and wait for the approval from the user',
+ props: {},
+ errorHandlingOptions: {
+ continueOnFailure: {
+ hide: true,
+ },
+ retryOnFailure: {
+ hide: true,
+ },
+ },
+ outputs: [
+ { name: 'approved' }, 
+ { name: 'denied' }
+ ],
+ async run(ctx) {
+ if (ctx.executionType === ExecutionType.BEGIN) {
+ ctx.run.pause({
+ pauseMetadata: {
+ type: PauseType.WEBHOOK,
+ response: {},
+ },
+ });
+
+ return branchedPieceResponse()
+ } else {
+ return branchedPieceResponse({
+ approved: ctx.resumePayload.queryParams['action'] === 'approve',
+ denied: ctx.resumePayload.queryParams['action'] !== 'approve',
+ })
+ }
+ },
+});

packages/pieces/community/framework/src/lib/action/action.ts

@@ -1,78 +1,86 @@
-import { Static, Type } from '@sinclair/typebox';
-import { ActionContext } from '../context';
-import { ActionBase } from '../piece-metadata';
-import { InputPropertyMap } from '../property';
-import { PieceAuthProperty } from '../property/authentication';
-
-export type ActionRunner<PieceAuth extends PieceAuthProperty, ActionProps extends InputPropertyMap> =
- (ctx: ActionContext<PieceAuth, ActionProps>) => Promise<unknown | void>
-
-export const ErrorHandlingOptionsParam = Type.Object({
- retryOnFailure: Type.Object({
- defaultValue: Type.Optional(Type.Boolean()),
- hide: Type.Optional(Type.Boolean()),
- }),
- continueOnFailure: Type.Object({
- defaultValue: Type.Optional(Type.Boolean()),
- hide: Type.Optional(Type.Boolean()),
- }),
-})
-export type ErrorHandlingOptionsParam = Static<typeof ErrorHandlingOptionsParam>
-
-type CreateActionParams<PieceAuth extends PieceAuthProperty, ActionProps extends InputPropertyMap> = {
- /**
- * A dummy parameter used to infer {@code PieceAuth} type
- */
- name: string
- auth?: PieceAuth
- displayName: string
- description: string
- props: ActionProps
- run: ActionRunner<PieceAuth, ActionProps>
- test?: ActionRunner<PieceAuth, ActionProps>
- requireAuth?: boolean
- errorHandlingOptions?: ErrorHandlingOptionsParam
-}
-
-export class IAction<PieceAuth extends PieceAuthProperty, ActionProps extends InputPropertyMap> implements ActionBase {
- constructor(
- public readonly name: string,
- public readonly displayName: string,
- public readonly description: string,
- public readonly props: ActionProps,
- public readonly run: ActionRunner<PieceAuth, ActionProps>,
- public readonly test: ActionRunner<PieceAuth, ActionProps>,
- public readonly requireAuth: boolean,
- public readonly errorHandlingOptions: ErrorHandlingOptionsParam,
- ) { }
-}
-
-export type Action<
- PieceAuth extends PieceAuthProperty = any,
- ActionProps extends InputPropertyMap = any,
-> = IAction<PieceAuth, ActionProps>
-
-export const createAction = <
- PieceAuth extends PieceAuthProperty = PieceAuthProperty,
- ActionProps extends InputPropertyMap = any
->(
- params: CreateActionParams<PieceAuth, ActionProps>,
-) => {
- return new IAction(
- params.name,
- params.displayName,
- params.description,
- params.props,
- params.run,
- params.test ?? params.run,
- params.requireAuth ?? true,
- params.errorHandlingOptions ?? {
- continueOnFailure: {
- defaultValue: false,
- },
- retryOnFailure: {
- defaultValue: false,
- }
- },
- )
-}
+import { Static, Type } from '@sinclair/typebox';
+import { ActionContext } from '../context';
+import { ActionBase, ActionOutput } from '../piece-metadata';
+import { InputPropertyMap } from '../property';
+import { PieceAuthProperty } from '../property/authentication';
+
+export type RunFunctionReturnType = {
+ version: 'v1'
+ output: Map<string, boolean | undefined | null>
+}
+
+export type ActionRunner<PieceAuth extends PieceAuthProperty, ActionProps extends InputPropertyMap> =
+ (ctx: ActionContext<PieceAuth, ActionProps>) => Promise<RunFunctionReturnType | unknown | void>
+
+export const ErrorHandlingOptionsParam = Type.Object({
+ retryOnFailure: Type.Object({
+ defaultValue: Type.Optional(Type.Boolean()),
+ hide: Type.Optional(Type.Boolean()),
+ }),
+ continueOnFailure: Type.Object({
+ defaultValue: Type.Optional(Type.Boolean()),
+ hide: Type.Optional(Type.Boolean()),
+ }),
+})
+export type ErrorHandlingOptionsParam = Static<typeof ErrorHandlingOptionsParam>
+
+type CreateActionParams<PieceAuth extends PieceAuthProperty, ActionProps extends InputPropertyMap> = {
+ /**
+ * A dummy parameter used to infer {@code PieceAuth} type
+ */
+ name: string
+ auth?: PieceAuth
+ displayName: string
+ description: string
+ props: ActionProps
+ run: ActionRunner<PieceAuth, ActionProps>
+ test?: ActionRunner<PieceAuth, ActionProps>
+ requireAuth?: boolean
+ outputs?: ActionOutput[]
+ errorHandlingOptions?: ErrorHandlingOptionsParam
+}
+
+export class IAction<PieceAuth extends PieceAuthProperty, ActionProps extends InputPropertyMap> implements ActionBase {
+ constructor(
+ public readonly name: string,
+ public readonly displayName: string,
+ public readonly description: string,
+ public readonly props: ActionProps,
+ public readonly run: ActionRunner<PieceAuth, ActionProps>,
+ public readonly test: ActionRunner<PieceAuth, ActionProps>,
+ public readonly requireAuth: boolean,
+ public readonly outputs: ActionOutput[],
+ public readonly errorHandlingOptions: ErrorHandlingOptionsParam,
+ ) { }
+}
+
+export type Action<
+ PieceAuth extends PieceAuthProperty = any,
+ ActionProps extends InputPropertyMap = any,
+> = IAction<PieceAuth, ActionProps>
+
+export const createAction = <
+ PieceAuth extends PieceAuthProperty = PieceAuthProperty,
+ ActionProps extends InputPropertyMap = any
+>(
+ params: CreateActionParams<PieceAuth, ActionProps>,
+) => {
+ return new IAction(
+ params.name,
+ params.displayName,
+ params.description,
+ params.props,
+ params.run,
+ params.test ?? params.run,
+ params.requireAuth ?? true,
+ params.outputs ?? [{ name: 'main' }],
+ params.errorHandlingOptions ?? {
+ continueOnFailure: {
+ defaultValue: false,
+ },
+ retryOnFailure: {
+ defaultValue: false,
+ }
+ },
+ )
+}