feat(pdd): 引入拼多多原始订单暂存仓 pdd_raw_orders 与两段式解耦流水线

1. 数据库新增 081_pdd_raw_orders.sql 迁移，创建 pdd_raw_orders 原始暂存表及索引； 2. 新增 pdd-raw-order-repo 仓储，支持原汁原味批量暂存、pending 单据消费、状态回填与重置； 3. 重构 pdd-sync-service 为采集暂存（Staging）与业务进单（Ingestion）两段式解耦流水线，实现断点续传； 4. 管理后台路由接入 raw-orders 原始流水查询及 replay 一键历史单重放 API。
2026-09-19 11:58:28 +08:00
parent c0d00d28ce
commit 0b4ee964ed
6 changed files with 780 additions and 117 deletions
@@ -15,6 +15,7 @@ const testStore: PddStoreConfigItem = {

 test('syncStoreOrders performs adaptive pagination when page 1 contains only new orders', async () => {
  const pagesQueried: number[] = []
+  const stagedOrders = new Set<string>()

  // Mock client queryRecentOrders
  const originalQuery = PddClient.prototype.queryRecentOrders
@@ -48,24 +49,31 @@ test('syncStoreOrders performs adaptive pagination when page 1 contains only new
    return { total: 100, items: [] }
  }

+  // Pre-populate page 2 item in staging DB
+  stagedOrders.add('sn_page2_50')
+
  try {
    const result = await syncStoreOrders(testStore, {
      pageSize: 50,
      maxPages: 5,
-      checkExistingOrderIds: async (orderSns: string[]) => {
-        // Page 1 has 0 existing in DB. Page 2 has sn_page2_50 in DB.
-        const existing = new Set<string>()
-        if (orderSns.includes('sn_page2_50')) {
-          existing.add('sn_page2_50')
+      batchUpsertRawFn: async (orders) => {
+        let inserted = 0
+        let updated = 0
+        for (const o of orders) {
+          if (stagedOrders.has(o.order_sn)) {
+            updated++
+          } else {
+            stagedOrders.add(o.order_sn)
+            inserted++
+          }
        }
-        return existing
+        return { insertedCount: inserted, updatedCount: updated }
      },
-      upsertFn: async () => {
+      processPendingFn: async () => {
        return {
-          ignored: false,
-          order: {} as any,
-          orderItems: [],
-          tasks: [],
+          upsertedCount: stagedOrders.size,
+          ignoredCount: 0,
+          failedCount: 0,
        }
      },
    })
@@ -101,18 +109,11 @@ test('syncStoreOrders stops on page 1 if page 1 already contains existing orders
    const result = await syncStoreOrders(testStore, {
      pageSize: 50,
      maxPages: 5,
-      checkExistingOrderIds: async () => {
-        // Page 1 has an existing order in DB!
-        return new Set(['sn_page1_10'])
-      },
-      upsertFn: async () => {
-        return {
-          ignored: false,
-          order: {} as any,
-          orderItems: [],
-          tasks: [],
-        }
+      batchUpsertRawFn: async () => {
+        // Page 1 already all exist in raw DB (insertedCount = 0)
+        return { insertedCount: 0, updatedCount: 50 }
      },
+      processPendingFn: async () => ({ upsertedCount: 0, ignoredCount: 0, failedCount: 0 }),
    })

    assert.equal(result.pagesFetched, 1, 'Should stop at page 1')
@@ -145,8 +146,8 @@ test('syncStoreOrders stops on page 1 if explicit pageNumber > 1 is specified',
    const result = await syncStoreOrders(testStore, {
      pageNumber: 2,
      pageSize: 50,
-      checkExistingOrderIds: async () => new Set(), // even if all are new
-      upsertFn: async () => ({ ignored: false, order: {} as any, orderItems: [], tasks: [] }),
+      batchUpsertRawFn: async () => ({ insertedCount: 50, updatedCount: 0 }),
+      processPendingFn: async () => ({ upsertedCount: 50, ignoredCount: 0, failedCount: 0 }),
    })

    assert.equal(result.pagesFetched, 1, 'Should only fetch the explicit page')
@@ -1,28 +1,31 @@
 /**
 * Pinduoduo Order Synchronization Service
- * 拼多多订单增量同步与进单调度服务
+ * 拼多多订单增量同步与进单调度服务（两段式暂存仓架构）
 *
 * Core Workflow / 核心流程:
- * 1. Pull recent orders from PDD MMS API using anti-content engine.
- *    调用基于纯算 anti-content 的 MMS 客户端拉取近期待发货订单。
- * 2. Adaptive pagination: If all orders on the first page are brand-new, auto-page to pull older orders until hitting existing records.
- *    自适应翻页拉取：当第一页（及后续页）全为新订单时，自适应继续翻页拉取，直至命中数据库已有订单或达到上限。
- * 3. Map PDD payload to platform SourceOrderEvent.
- *    将原始报文转化为 order_site 统一标准进单事件。
- * 4. In-memory ingestion via upsertOrderFromSource().
- *    内存直调平台核心 upsertOrderFromSource()，自动入库并触发打手大厅工单分发。
- * 5. Staggered multi-store scheduling & risk control.
- *    多店铺错峰拉取与优雅异常处理。
+ * Stage 1: Staging (采集与暂存)
+ *   调用基于纯算 anti-content 的 MMS 客户端拉取近期订单，原汁原味批量暂存入 pdd_raw_orders 表。
+ *   支持自适应翻页拉取，直至命中已有记录或达到分页上限。
+ * Stage 2: Processing & Ingestion (业务消费与分发)
+ *   从 pdd_raw_orders 中读取 pending 状态的订单，通过 upsertOrderFromSource 转化为平台标准订单，
+ *   自动匹配商品规则并分发打手工单，成功后回填 synced_order_id 与 sync_status = 'synced'。
 */

 import { PddClient, type QueryOrdersOptions } from './pdd-client.js'
-import { mapPddOrderToSourceEvent } from './pdd-order-mapper.js'
+import { mapPddOrderToSourceEvent, type PddRawOrderItem } from './pdd-order-mapper.js'
 import {
  getPddSourcesConfig,
  updatePddStoreCheckStatus,
  type PddStoreConfigItem,
 } from './pdd-source-config-service.js'
-import { findExistingPlatformOrderIds } from '../../../repositories/order-repo.js'
+import {
+  batchUpsertPddRawOrders,
+  findPendingPddRawOrders,
+  markPddRawOrderSynced,
+  markPddRawOrderIgnored,
+  markPddRawOrderFailed,
+  resetPddRawOrdersToPending,
+} from '../../../repositories/platforms/pdd-raw-order-repo.js'
 import { notifyPddCookieExpired } from '../../notification/domain-notifications.js'
 import { upsertOrderFromSource } from '../../order/order-service.js'
 import { logDebug, logError, logInfo, logWarn } from '../../../utils/logger.js'
@@ -34,8 +37,9 @@ export interface SyncStoreOrdersOptions {
  orderType?: number | undefined // 2: 已发货 (默认), 1: 待发货, 0: 全部
  startTime?: number | undefined
  endTime?: number | undefined
-  checkExistingOrderIds?: ((orderSns: string[]) => Promise<Set<string>>) | undefined
  upsertFn?: typeof upsertOrderFromSource | undefined
+  batchUpsertRawFn?: typeof batchUpsertPddRawOrders | undefined
+  processPendingFn?: typeof processPendingPddRawOrders | undefined
 }

 export interface SyncStoreResult {
@@ -65,7 +69,7 @@ export interface SyncAllStoresResult {
 */
 export function isPddCookieExpiredError(errorMsg: string): boolean {
  const msg = String(errorMsg || '').toLowerCase()
-  const expiredCodes = ['43001', '40001', '40002', '43002', '43003', '10000']
+  const expiredCodes = ['43001', '40001', '40002', '43003', '10000']
  if (expiredCodes.some((code) => msg.includes(code))) {
    return true
  }
@@ -91,9 +95,67 @@ export function isPddCookieExpiredError(errorMsg: string): boolean {
  return expiredKeywords.some((kw) => msg.includes(kw))
 }

+/**
+ * Stage 2: 消费并转化指定店铺待处理的拼多多原始订单入库
+ */
+export async function processPendingPddRawOrders(
+  shopId: string,
+  shopName: string,
+  options: {
+    limit?: number | undefined
+    upsertFn?: typeof upsertOrderFromSource | undefined
+  } = {},
+): Promise<{
+  upsertedCount: number
+  ignoredCount: number
+  failedCount: number
+}> {
+  const limit = options.limit || 50
+  const doUpsert = options.upsertFn || upsertOrderFromSource
+  const pendingRows = await findPendingPddRawOrders({ shopId, limit })
+
+  let upsertedCount = 0
+  let ignoredCount = 0
+  let failedCount = 0
+
+  for (const rawRow of pendingRows) {
+    try {
+      const rawPayload = (rawRow.raw_payload_json || {}) as PddRawOrderItem
+      const sourceEvent = mapPddOrderToSourceEvent(rawPayload, { shopId, shopName })
+
+      const result = await doUpsert(sourceEvent, {
+        sourceLabel: `pdd_${shopId}`,
+      })
+
+      if (result.ignored) {
+        ignoredCount++
+        await markPddRawOrderIgnored(rawRow.order_sn, result.ignoreReason || 'no_rule_matched')
+      } else if (result.order) {
+        upsertedCount++
+        await markPddRawOrderSynced(rawRow.order_sn, result.order.id)
+      }
+    } catch (upsertErr) {
+      failedCount++
+      const errMsg = upsertErr instanceof Error ? upsertErr.message : String(upsertErr)
+      logError('[pdd-sync]', `拼多多订单进单失败: ${rawRow.order_sn}`, {
+        shopId,
+        orderSn: rawRow.order_sn,
+        error: errMsg,
+      })
+      await markPddRawOrderFailed(rawRow.order_sn, errMsg)
+    }
+  }
+
+  return {
+    upsertedCount,
+    ignoredCount,
+    failedCount,
+  }
+}
+
 /**
 * Synchronize orders for a single PDD store with adaptive pagination.
- * 同步单个拼多多店铺的订单并触发进单流程（支持自适应翻页拉取）。
+ * 同步单个拼多多店铺的订单并触发两段式流水线（采集暂存 -> 业务分发）。
 */
 export async function syncStoreOrders(
  store: PddStoreConfigItem,
@@ -128,7 +190,6 @@ export async function syncStoreOrders(
  const orderType = options.orderType ?? sourcesConfig.orderType ?? 2 // 2: 已发货 (Shipped, default)

  // Determine whether to allow adaptive multi-page pulling
-  // 若传入明确的指定页码且 > 1，则仅拉取该指定页；否则执行自适应翻页（默认上限 5 页，最大支持 20 页）
  const isExplicitSinglePage = options.pageNumber !== undefined && options.pageNumber > 1
  const maxPages = isExplicitSinglePage
    ? 1
@@ -138,12 +199,15 @@ export async function syncStoreOrders(
  let pagesFetched = 0
  let totalReported = 0
  let totalFetched = 0
-  let totalUpserted = 0
-  let totalNewOrders = 0
-  let totalIgnored = 0
-  let totalFailed = 0
+  let totalNewRawOrders = 0

  try {
+    const doBatchUpsert = options.batchUpsertRawFn || batchUpsertPddRawOrders
+    const doProcessPending = options.processPendingFn || processPendingPddRawOrders
+
+    // -------------------------------------------------------------------------
+    // 阶段 1: 采集并批量暂存入 pdd_raw_orders
+    // -------------------------------------------------------------------------
    while (pagesFetched < maxPages) {
      const queryOptions: QueryOrdersOptions = {
        pageNumber: currentPage,
@@ -168,85 +232,41 @@ export async function syncStoreOrders(

      totalFetched += items.length

-      // 提取本页订单号并批量排查数据库已有状态
-      const pageOrderSns = items.map((item) => String(item.order_sn || '').trim()).filter(Boolean)
+      // 批量写入 pdd_raw_orders 暂存仓
+      const upsertRes = await doBatchUpsert(items, { shopId, shopName })
+      totalNewRawOrders += upsertRes.insertedCount

-      const checkExisting =
-        options.checkExistingOrderIds ||
-        ((sns: string[]) =>
-          findExistingPlatformOrderIds({
-            provider: 'pdd',
-            platform: 'pdd',
-            platformOrderIds: sns,
-          }))
-      const existingOrderSns = await checkExisting(pageOrderSns)
-
-      const existingCountOnPage = existingOrderSns.size
-      const newCountOnPage = pageOrderSns.length - existingCountOnPage
-      totalNewOrders += Math.max(0, newCountOnPage)
-
-      const doUpsert = options.upsertFn || upsertOrderFromSource
-
-      // 执行当前页订单进单入库
-      for (const rawItem of items) {
-        try {
-          const sourceEvent = mapPddOrderToSourceEvent(rawItem, { shopId, shopName })
-          const result = await doUpsert(sourceEvent, {
-            sourceLabel: `pdd_${shopId}`,
-          })
-
-          if (result.ignored) {
-            totalIgnored++
-          } else {
-            totalUpserted++
-          }
-        } catch (upsertErr) {
-          totalFailed++
-          logError('[pdd-sync]', `拼多多订单进单失败: ${rawItem.order_sn}`, {
-            shopId,
-            orderSn: rawItem.order_sn,
-            error: upsertErr instanceof Error ? upsertErr.message : String(upsertErr),
-          })
-        }
-      }
-
-      // 本页无新单时降到 debug，避免每 20s 一轮把日志打满。
-      const pageLog = newCountOnPage > 0 || totalFailed > 0 ? logInfo : logDebug
+      const pageLog = upsertRes.insertedCount > 0 ? logInfo : logDebug
      pageLog(
        '[pdd-sync]',
-        `店铺 ${shopName} 第 ${currentPage} 页处理完成 (本页拉取: ${items.length}, 新单: ${newCountOnPage}, 库中已有: ${existingCountOnPage})`,
+        `店铺 ${shopName} 第 ${currentPage} 页采集暂存完成 (本页拉取: ${items.length}, 新增暂存: ${upsertRes.insertedCount}, 更新: ${upsertRes.updatedCount})`,
        {
          shopId,
          currentPage,
          pageItems: items.length,
-          newCountOnPage,
-          existingCountOnPage,
+          insertedCount: upsertRes.insertedCount,
+          updatedCount: upsertRes.updatedCount,
        },
      )

      // 自适应翻页终止条件判断：
-      // 1. 若外部指定单页拉取，立即终止
      if (isExplicitSinglePage) {
        break
      }

-      // 2. 若当前页中已经出现了数据库中已存在的旧订单，说明拉单时间线已触及上次同步断点
-      // （拼多多按时间倒序排列，后续页码必然全为更早的已存历史订单，无需继续翻页）
-      if (existingCountOnPage > 0) {
+      // 若当前页中所有单据均已存在于暂存库中（insertedCount === 0），说明已触及上次同步断点
+      if (upsertRes.insertedCount === 0 && upsertRes.updatedCount > 0) {
        break
      }

-      // 3. 若本页订单数不足 pageSize，说明拼多多后台当前条件已无更多订单
      if (items.length < pageSize) {
        break
      }

-      // 4. 若已拉取条数达到店铺总订单数上限，终止
      if (totalFetched >= totalReported) {
        break
      }

-      // 5. 若已达最大翻页限制，终止并记录日志
      if (pagesFetched >= maxPages) {
        logInfo('[pdd-sync]', `店铺 ${shopName} 达到自适应翻页上限 (${maxPages}页)，停止继续翻页`, {
          shopId,
@@ -256,10 +276,9 @@ export async function syncStoreOrders(
        break
      }

-      // 本页全为全新订单！说明有积压新单未拉完，自适应翻页拉取下一页
      logInfo(
        '[pdd-sync]',
-        `店铺 ${shopName} 第 ${currentPage} 页全为新订单 (${newCountOnPage}条)，自适应翻页拉取第 ${currentPage + 1} 页...`,
+        `店铺 ${shopName} 第 ${currentPage} 页有新订单 (${upsertRes.insertedCount}条)，自适应翻页拉取第 ${currentPage + 1} 页...`,
        { shopId, nextPage: currentPage + 1 },
      )
      currentPage++
@@ -268,6 +287,14 @@ export async function syncStoreOrders(
      await new Promise((resolve) => setTimeout(resolve, 300))
    }

+    // -------------------------------------------------------------------------
+    // 阶段 2: 消费并转化 pending 状态订单
+    // -------------------------------------------------------------------------
+    const processRes = await doProcessPending(shopId, shopName, {
+      limit: 100,
+      upsertFn: options.upsertFn,
+    })
+
    // Update persistent check status to valid / 更新持久化校验状态为有效
    await updatePddStoreCheckStatus(shopId, {
      cookieValid: true,
@@ -276,18 +303,17 @@ export async function syncStoreOrders(
      totalOrders: totalReported,
    }).catch(() => null)

-    // 仅「真正新单 / 失败 / 多页回填」打 info；常态触达旧单后停页的空跑降到 debug。
    const storeDoneLog =
-      totalNewOrders > 0 || totalFailed > 0 || pagesFetched > 1 ? logInfo : logDebug
+      totalNewRawOrders > 0 || processRes.failedCount > 0 || pagesFetched > 1 ? logInfo : logDebug
    storeDoneLog('[pdd-sync]', `店铺 ${shopName} 同步完成 (共拉取 ${pagesFetched} 页)`, {
      shopId,
      total: totalReported,
      pagesFetched,
      fetchedCount: totalFetched,
-      upsertedCount: totalUpserted,
-      newOrderCount: totalNewOrders,
-      ignoredCount: totalIgnored,
-      failedCount: totalFailed,
+      newOrderCount: totalNewRawOrders,
+      upsertedCount: processRes.upsertedCount,
+      ignoredCount: processRes.ignoredCount,
+      failedCount: processRes.failedCount,
    })

    return {
@@ -295,10 +321,10 @@ export async function syncStoreOrders(
      shopName,
      total: totalReported,
      fetchedCount: totalFetched,
-      upsertedCount: totalUpserted,
-      newOrderCount: totalNewOrders,
-      ignoredCount: totalIgnored,
-      failedCount: totalFailed,
+      upsertedCount: processRes.upsertedCount,
+      newOrderCount: totalNewRawOrders,
+      ignoredCount: processRes.ignoredCount,
+      failedCount: processRes.failedCount,
      pagesFetched,
    }
  } catch (err) {
@@ -309,8 +335,6 @@ export async function syncStoreOrders(
      pagesFetched,
    })

-    // If login expired or risk-controlled, mark cookie as invalid and notify admins
-    // 若登录过期或风控拦截，标记为失效并推送告警通知
    if (isPddCookieExpiredError(errMsg)) {
      await updatePddStoreCheckStatus(shopId, {
        cookieValid: false,
@@ -330,10 +354,10 @@ export async function syncStoreOrders(
      shopName,
      total: totalReported,
      fetchedCount: totalFetched,
-      upsertedCount: totalUpserted,
-      newOrderCount: totalNewOrders,
-      ignoredCount: totalIgnored,
-      failedCount: totalFailed,
+      upsertedCount: 0,
+      newOrderCount: totalNewRawOrders,
+      ignoredCount: 0,
+      failedCount: 1,
      pagesFetched,
      error: errMsg,
    }
@@ -366,7 +390,6 @@ export async function syncAllPddStores(
    const res = await syncStoreOrders(store, options)
    results.push(res)

-    // Stagger delay between stores (500ms) to avoid concurrent spikes
    // 店铺之间间隔 500ms，避免并发峰值
    if (i + 1 < activeStores.length) {
      await new Promise((resolve) => setTimeout(resolve, 500))
@@ -380,3 +403,30 @@ export async function syncAllPddStores(
    results,
  }
 }
+
+/**
+ * 一键重放或重新处理指定店铺的历史暂存单
+ */
+export async function replayPddRawOrders(
+  options: { shopId?: string; orderSns?: string[] } = {},
+): Promise<{ resetCount: number; processedCount: number }> {
+  const resetCount = await resetPddRawOrdersToPending(options)
+  const config = getPddSourcesConfig()
+  let processedCount = 0
+
+  const targetStores = options.shopId
+    ? config.stores.filter((s) => s.shopId === options.shopId)
+    : config.stores.filter((s) => s.enabled !== false)
+
+  for (const store of targetStores) {
+    const res = await processPendingPddRawOrders(store.shopId, store.shopName || store.shopId, {
+      limit: 100,
+    })
+    processedCount += res.upsertedCount + res.ignoredCount
+  }
+
+  return {
+    resetCount,
+    processedCount,
+  }
+}