资源不会消失，只会泄露

CREATE FUNCTION create_job_and_charge_credits(
  p_user_id UUID,
  p_credit_cost INTEGER
) RETURNS TABLE(job_id UUID, success BOOLEAN) AS $
DECLARE
  v_current_credits INTEGER;
BEGIN
  -- 🔒 行级锁定：FOR UPDATE
  SELECT video_credits INTO v_current_credits
  FROM profiles
  WHERE id = p_user_id
  FOR UPDATE;

  -- 检查积分
  IF v_current_credits < p_credit_cost THEN
    RETURN QUERY SELECT NULL::UUID, FALSE;
    RETURN;
  END IF;

  -- 扣除积分 + 创建任务（原子操作）
  UPDATE profiles
  SET video_credits = video_credits - p_credit_cost
  WHERE id = p_user_id;

  INSERT INTO ai_generations (user_id, credits_charged, status)
  VALUES (p_user_id, p_credit_cost, 'pending')
  RETURNING id INTO v_job_id;

  RETURN QUERY SELECT v_job_id, TRUE;
END;
$ LANGUAGE plpgsql;

// Cron: 每30分钟执行
async function cleanZombieTasks() {
  const THRESHOLD = 30; // 分钟
  const cutoffTime = new Date(Date.now() - THRESHOLD * 60 * 1000);

  // 查询僵尸任务
  const zombieTasks = await database
    .select('*')
    .from('ai_generations')
    .whereIn('status', ['submitted_to_vendor', 'processing'])
    .where('updated_at', '<', cutoffTime);

  for (const task of zombieTasks) {
    // 主动查询厂商状态
    const vendorStatus = await queryVendorStatus(task.vendor_task_id);

    if (vendorStatus.found) {
      // 厂商有记录，更新状态
      await updateTaskStatus(task.id, vendorStatus.status);
    } else {
      // 厂商也找不到，退款
      await markTaskAsFailed(task.id, 'vendor_lost_task');
      await refundCredits(task.user_id, task.credits_charged);
    }
  }
}