流程控制

了解如何在 Piece 内部控制流程执行

流程控制让动作可以改变运行的形态——提前停止、发送中间 HTTP 响应，或暂停工作流并在外部信号到达时恢复。所有这些都通过动作的 run 方法中的 ctx 参数暴露。

停止工作流

停止工作流并（可选）向启动它的 Webhook 触发器返回响应。

带响应：

context.run.stop({
  response: {
    status: context.propsValue.status ?? StatusCodes.OK,
    body: context.propsValue.body,
    headers: (context.propsValue.headers as Record<string, string>) ?? {},
  },
});

不带响应：

context.run.stop();

使用等待点暂停

等待点是一个持久的检查点：运行被标记为 PAUSED，其执行状态被持久化，一旦等待点恢复，动作将第二次被调用。等待点可以承受工作节点重启——完整模型请参见持久化执行。

同一个动作会运行两次——第一次创建等待点，第二次读取恢复负载——因此每个暂停动作都需要在 ctx.executionType 上进行分支判断：

import { ExecutionType } from '@activepieces/shared';

async run(ctx) {
  if (ctx.executionType === ExecutionType.BEGIN) {
    // 第一次调用：创建等待点并暂停。
    // 真实的 WEBHOOK 等待点必须向外暴露 waitpoint.buildResumeUrl(...)
    // 请参阅下面的"等待 Webhook 回调"部分。
    const waitpoint = await ctx.run.createWaitpoint({ type: 'WEBHOOK' });
    ctx.run.waitForWaitpoint(waitpoint.id);
    return {};
  }

  // 第二次调用：等待点已恢复。
  return {
    body: ctx.resumePayload.body,
    headers: ctx.resumePayload.headers,
    queryParams: ctx.resumePayload.queryParams,
  };
}

两个钩子完成工作：

• ctx.run.createWaitpoint({ type, ... })——在服务器上注册等待点并返回 { id, resumeUrl, buildResumeUrl }。
• ctx.run.waitForWaitpoint(waitpointId)——告诉引擎该步骤的裁定为 paused；动作返回后运行转换为 PAUSED。

有两种等待点类型。

等待 Webhook 回调

创建一个 WEBHOOK 等待点并暴露其恢复 URL——无论何时调用该 URL，工作流都会恢复。

async run(ctx) {
  if (ctx.executionType === ExecutionType.BEGIN) {
    const waitpoint = await ctx.run.createWaitpoint({ type: 'WEBHOOK' });

    const callbackUrl = waitpoint.buildResumeUrl({
      queryParams: { runId: ctx.run.id },
    });

    // 将 `callbackUrl` 发送到外部世界可以访问的地方
    //（电子邮件、Slack 消息、第三方 API 等）。

    ctx.run.waitForWaitpoint(waitpoint.id);
    return {};
  }

  return {
    approved: ctx.resumePayload.queryParams['action'] === 'approve',
  };
}

buildResumeUrl 接受可选的 sync: true 参数，以向调用者返回工作流剩余部分产生的同步响应，以及一个 queryParams 对象，该对象在恢复时传递到 ctx.resumePayload.queryParams。

立即响应并等待下一个 Webhook

暂停工作流并立即回复 Webhook 触发器——适用于”我们已收到你的提交，稍后将回电”的模式。传入 responseToSend，你的 HTTP 响应将立即发送；工作流随后保持暂停状态，直到返回的 URL 被调用。

async run(ctx) {
  if (ctx.executionType === ExecutionType.BEGIN) {
    const waitpoint = await ctx.run.createWaitpoint({
      type: 'WEBHOOK',
      responseToSend: {
        status: 200,
        headers: {},
        body: { accepted: true },
      },
    });

    const nextWebhookUrl = waitpoint.buildResumeUrl({
      queryParams: { created: new Date().toISOString() },
      sync: true,
    });

    // nextWebhookUrl 是对应方调用以恢复此运行的 URL。

    ctx.run.waitForWaitpoint(waitpoint.id);
    return { nextWebhookUrl };
  }

  return {
    body: ctx.resumePayload.body,
    headers: ctx.resumePayload.headers,
    queryParams: ctx.resumePayload.queryParams,
  };
}

延迟到指定时间戳

创建一个 DELAY 等待点，指定你希望恢复的 UTC 时间戳。服务器会安排一个一次性作业，在 resumeDateTime 触发并自动恢复运行。

async run(ctx) {
  if (ctx.executionType === ExecutionType.RESUME) {
    return { success: true };
  }

  const futureTime = new Date(Date.now() + 60 * 60 * 1000); // 1 小时后

  const waitpoint = await ctx.run.createWaitpoint({
    type: 'DELAY',
    resumeDateTime: futureTime.toUTCString(),
  });
  ctx.run.waitForWaitpoint(waitpoint.id);
  return {};
}

resumeDateTime 受服务器 AP_PAUSED_FLOW_TIMEOUT_DAYS 设置的限制；如果你请求更长的延迟，引擎会抛出 PausedFlowTimeoutError。

读取恢复负载

在 RESUME 分支中，ctx.resumePayload 包含恢复调用携带的数据：

ctx.resumePayload.body        // Webhook 调用者的请求体（DELAY 类型为空）
ctx.resumePayload.headers     // Webhook 调用者的 HTTP 头
ctx.resumePayload.queryParams // 解析后的 ?foo=bar 查询字符串

对于 DELAY 等待点，没有传入的 HTTP 请求，因此负载为空——只需使用 RESUME 分支来产生步骤的最终输出。