把持久化工作流搬进数据库：微软开源 pg_durable

应用服务 → 调用 Temporal/Step Functions → 写状态到外部存储 → 读/写业务数据到 PostgreSQL

-- 从源码编译安装（需要 pg_config 在 PATH 中）
-- git clone https://github.com/microsoft/pg_durable.git
-- cd pg_durable && make && make install

-- 在目标数据库中启用扩展
CREATE EXTENSION pg_durable;

-- 注册一个名为 "order_process" 的工作流
SELECT pg_durable.register_workflow(
  'order_process',
  '处理电商订单的持久化工作流'
);

-- 添加步骤：验证库存
SELECT pg_durable.add_step(
  'order_process',
  'check_inventory',
  -- step_handler 是一个自定义函数名，pg_durable 会调用它
  'public.check_inventory_handler',
  -- 重试配置：最多 3 次，间隔 5 秒
  max_retries := 3,
  retry_interval := INTERVAL '5 seconds'
);

-- 添加步骤：扣减库存并创建订单记录
SELECT pg_durable.add_step(
  'order_process',
  'create_order',
  'public.create_order_handler',
  max_retries := 2,
  retry_interval := INTERVAL '3 seconds'
);

-- 添加步骤：发送通知
SELECT pg_durable.add_step(
  'order_process',
  'send_notification',
  'public.send_notification_handler',
  max_retries := 5,
  retry_interval := INTERVAL '10 seconds'
);

-- 步骤1：检查库存是否充足
CREATE OR REPLACE FUNCTION public.check_inventory_handler(
  workflow_id UUID,
  step_input JSONB
) RETURNS JSONB LANGUAGE plpgsql AS $$
DECLARE
  product_id INT;
  qty INT;
  available INT;
BEGIN
  product_id := (step_input ->> 'product_id')::INT;
  qty := (step_input ->> 'quantity')::INT;

  SELECT stock INTO available
    FROM inventory
    WHERE product_id = product_id;

  IF available < qty THEN
    -- 返回失败结果，pg_durable 会按重试策略处理
    RETURN jsonb_build_object(
      'status', 'failed',
      'reason', 'insufficient_stock',
      'available', available
    );
  END IF;

  RETURN jsonb_build_object(
    'status', 'completed',
    'product_id', product_id,
    'quantity', qty
  );
END;
$$;

-- 步骤2：创建订单（在同一个数据库事务中完成）
CREATE OR REPLACE FUNCTION public.create_order_handler(
  workflow_id UUID,
  step_input JSONB
) RETURNS JSONB LANGUAGE plpgsql AS $$
DECLARE
  product_id INT;
  qty INT;
  order_id INT;
BEGIN
  product_id := (step_input ->> 'product_id')::INT;
  qty := (step_input ->> 'quantity')::INT;

  -- 扣减库存
  UPDATE inventory SET stock = stock - qty
    WHERE product_id = product_id AND stock >= qty;

  -- 创建订单记录
  INSERT INTO orders (product_id, quantity, status)
    VALUES (product_id, qty, 'created')
    RETURNING id INTO order_id;

  RETURN jsonb_build_object(
    'status', 'completed',
    'order_id', order_id
  );
END;
$$;

-- 步骤3：发送通知（可能调用外部服务）
CREATE OR REPLACE FUNCTION public.send_notification_handler(
  workflow_id UUID,
  step_input JSONB
) RETURNS JSONB LANGUAGE plpgsql AS $$
DECLARE
  order_id INT;
  notify_result JSONB;
BEGIN
  order_id := (step_input ->> 'order_id')::INT;

  -- 这里可以调用外部 HTTP 服务（通过 pg_net 或类似扩展）
  -- 以下为示意，实际需配合异步通知扩展
  notify_result := jsonb_build_object(
    'status', 'completed',
    'order_id', order_id,
    'channel', 'email'
  );

  RETURN notify_result;
END;
$$;

-- 发起一个订单处理工作流
SELECT pg_durable.start_workflow(
  'order_process',
  jsonb_build_object(
    'product_id', 42,
    'quantity', 3
  )
);

-- 查询当前运行中的工作流
SELECT workflow_id, workflow_name, current_step, status, created_at
  FROM pg_durable.workflows
  WHERE status = 'running';

-- 查询某个工作流的步骤执行历史
SELECT step_name, status, attempt_count, started_at, completed_at
  FROM pg_durable.step_history
  WHERE workflow_id = '0192f8c4-7a3e-7d00-8000-000000000001'
  ORDER BY started_at;