diff --git a/apps/backend/src/db/migrations/081_pdd_raw_orders.sql b/apps/backend/src/db/migrations/081_pdd_raw_orders.sql new file mode 100644 index 00000000..5e61720a --- /dev/null +++ b/apps/backend/src/db/migrations/081_pdd_raw_orders.sql @@ -0,0 +1,47 @@ +-- 081_pdd_raw_orders.sql —— 拼多多原始订单暂存仓与两段式同步支持。 + +CREATE TABLE IF NOT EXISTS pdd_raw_orders ( + id BIGSERIAL PRIMARY KEY, + order_sn TEXT NOT NULL UNIQUE, + shop_id TEXT NOT NULL DEFAULT '', + shop_name TEXT NOT NULL DEFAULT '', + order_status INTEGER NOT NULL DEFAULT 1, + order_status_str TEXT NOT NULL DEFAULT '', + goods_id TEXT NOT NULL DEFAULT '', + goods_name TEXT NOT NULL DEFAULT '', + out_sku_sn TEXT NOT NULL DEFAULT '', + spec TEXT NOT NULL DEFAULT '', + quantity INTEGER NOT NULL DEFAULT 1, + order_amount_fen BIGINT NOT NULL DEFAULT 0, + buyer_nickname TEXT NOT NULL DEFAULT '', + buyer_memo TEXT NOT NULL DEFAULT '', + mall_remark TEXT NOT NULL DEFAULT '', + order_time TIMESTAMPTZ, + confirm_time TIMESTAMPTZ, + raw_payload_json JSONB NOT NULL DEFAULT '{}'::jsonb, + sync_status TEXT NOT NULL DEFAULT 'pending' + CHECK (sync_status IN ('pending', 'synced', 'ignored', 'failed')), + sync_error TEXT NOT NULL DEFAULT '', + synced_order_id BIGINT REFERENCES orders(id) ON DELETE SET NULL, + synced_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- 店铺与订单时间复合索引（用于增量拉单断点极速定位） +CREATE INDEX IF NOT EXISTS idx_pdd_raw_orders_shop_order_time + ON pdd_raw_orders (shop_id, order_time DESC NULLS LAST); + +-- 待处理订单部分索引（用于业务引擎快速扫描待进单数据） +CREATE INDEX IF NOT EXISTS idx_pdd_raw_orders_pending_sync + ON pdd_raw_orders (created_at ASC) + WHERE sync_status = 'pending'; + +-- 商家外部编码索引（用于按 out_sku_sn 快速对账和重放） +CREATE INDEX IF NOT EXISTS idx_pdd_raw_orders_out_sku_sn + ON pdd_raw_orders (out_sku_sn); + +-- 关联的平台订单索引 +CREATE INDEX IF NOT EXISTS idx_pdd_raw_orders_synced_order_id + ON pdd_raw_orders (synced_order_id) + WHERE synced_order_id IS NOT NULL; diff --git a/apps/backend/src/repositories/platforms/pdd-raw-order-repo.test.ts b/apps/backend/src/repositories/platforms/pdd-raw-order-repo.test.ts new file mode 100644 index 00000000..9afbb425 --- /dev/null +++ b/apps/backend/src/repositories/platforms/pdd-raw-order-repo.test.ts @@ -0,0 +1,67 @@ +import test from 'node:test' +import assert from 'node:assert/strict' + +import { batchUpsertPddRawOrders, type QueryExecutor } from './pdd-raw-order-repo.js' +import type { PddRawOrderItem } from '../../services/platforms/pdd/pdd-order-mapper.js' + +test('batchUpsertPddRawOrders formats and passes parameters safely to query executor', async () => { + const capturedQueries: Array<{ text: string; params: unknown[] }> = [] + + const mockExecutor: QueryExecutor = async (text: string, params?: unknown[]) => { + capturedQueries.push({ text, params: params || [] }) + return { + rows: [ + { + id: 1, + order_sn: '260919-999999', + shop_id: 'test_shop', + shop_name: '测试店铺', + sync_status: 'pending', + is_inserted: true, + }, + ], + rowCount: 1, + command: 'INSERT', + oid: 0, + fields: [], + } + } + + const rawOrder: PddRawOrderItem = { + order_sn: '260919-999999', + order_status: 1, + order_status_str: '待发货', + order_amount: 5000, + goods_id: 12345, + goods_name: '测试商品', + out_sku_sn: '安卓血宴3钻', + spec: '暗潮1钻', + goods_number: 1, + nickname: '买家\ud835*test', // Lone surrogate + buyer_memo: '游戏ID: 8888\0null', // Null byte + order_time: 1726718000, + confirm_time: 1726718005, + } + + const result = await batchUpsertPddRawOrders( + [rawOrder], + { shopId: 'test_shop', shopName: '测试店铺' }, + mockExecutor, + ) + + assert.equal(result.totalCount, 1) + assert.equal(result.insertedCount, 1) + assert.equal(capturedQueries.length, 1) + + const [q] = capturedQueries + assert.ok(q.text.includes('INSERT INTO pdd_raw_orders')) + assert.ok(q.text.includes('ON CONFLICT (order_sn) DO UPDATE')) + + // Verify parameters are properly sanitized + const params = q.params + assert.equal(params[0], '260919-999999') // orderSn + assert.equal(params[1], 'test_shop') // shopId + assert.equal(params[2], '测试店铺') // shopName + assert.equal(params[11], '买家\uFFFD*test') // buyerNickname (cleaned surrogate) + assert.equal(params[12], '游戏ID: 8888null') // buyerMemo (removed null byte) +}) diff --git a/apps/backend/src/repositories/platforms/pdd-raw-order-repo.ts b/apps/backend/src/repositories/platforms/pdd-raw-order-repo.ts new file mode 100644 index 00000000..cd45d858 --- /dev/null +++ b/apps/backend/src/repositories/platforms/pdd-raw-order-repo.ts @@ -0,0 +1,441 @@ +/** + * Pinduoduo Raw Orders Staging Repository + * 拼多多原始订单暂存仓数据库仓储 + */ + +import { query, withTransaction } from '../../db/client.js' +import type { QueryResult } from 'pg' +import { safeJsonStringify, sanitizeWellFormedUnicode } from '../../utils/json.js' +import type { PddRawOrderItem } from '../../services/platforms/pdd/pdd-order-mapper.js' +import { + extractPddGoodsLine, + extractPddOutSkuSn, +} from '../../services/platforms/pdd/pdd-order-mapper.js' + +export type QueryExecutor = = any>( + text: string, + params?: unknown[], +) => Promise> + +export interface PddRawOrderRow { + id: number + order_sn: string + shop_id: string + shop_name: string + order_status: number + order_status_str: string + goods_id: string + goods_name: string + out_sku_sn: string + spec: string + quantity: number + order_amount_fen: number + buyer_nickname: string + buyer_memo: string + mall_remark: string + order_time: string | null + confirm_time: string | null + raw_payload_json: Record + sync_status: 'pending' | 'synced' | 'ignored' | 'failed' + sync_error: string + synced_order_id: number | null + synced_at: string | null + created_at: string + updated_at: string +} + +export interface PddRawOrderUpsertResult { + insertedCount: number + updatedCount: number + totalCount: number + rows: PddRawOrderRow[] +} + +/** + * 批量暂存拼多多原始订单数据（ON CONFLICT DO UPDATE） + */ +export async function batchUpsertPddRawOrders( + rawOrders: PddRawOrderItem[], + options: { shopId: string; shopName: string }, + executor: QueryExecutor = query, +): Promise { + if (!rawOrders || rawOrders.length === 0) { + return { insertedCount: 0, updatedCount: 0, totalCount: 0, rows: [] } + } + + const shopId = options.shopId.trim() || 'pdd_default' + const shopName = options.shopName.trim() || '拼多多店铺' + const rows: PddRawOrderRow[] = [] + let insertedCount = 0 + let updatedCount = 0 + + for (const rawItem of rawOrders) { + const orderSn = String(rawItem.order_sn || '').trim() + if (!orderSn) continue + + const goodsLine = extractPddGoodsLine(rawItem) + const outSkuSn = extractPddOutSkuSn(rawItem) + const orderStatus = Number(rawItem.order_status || 1) + const orderStatusStr = sanitizeWellFormedUnicode(rawItem.order_status_str || '') + const goodsId = sanitizeWellFormedUnicode(goodsLine.goodsId) + const goodsName = sanitizeWellFormedUnicode(goodsLine.goodsName) + const spec = sanitizeWellFormedUnicode(goodsLine.spec) + const quantity = Math.max(1, Number(goodsLine.quantity || 1)) + const orderAmountFen = Math.max(0, Number(rawItem.order_amount || 0)) + const buyerNickname = sanitizeWellFormedUnicode(rawItem.nickname || '') + const buyerMemo = sanitizeWellFormedUnicode(rawItem.buyer_memo || '').trim() + const mallRemark = sanitizeWellFormedUnicode(rawItem.mall_remark || '').trim() + + const orderTime = rawItem.order_time + ? new Date(Number(rawItem.order_time) * 1000).toISOString() + : null + const confirmTime = rawItem.confirm_time + ? new Date(Number(rawItem.confirm_time) * 1000).toISOString() + : null + + const rawPayloadJson = safeJsonStringify(rawItem) + + const result = await executor( + ` + INSERT INTO pdd_raw_orders ( + order_sn, + shop_id, + shop_name, + order_status, + order_status_str, + goods_id, + goods_name, + out_sku_sn, + spec, + quantity, + order_amount_fen, + buyer_nickname, + buyer_memo, + mall_remark, + order_time, + confirm_time, + raw_payload_json, + sync_status, + created_at, + updated_at + ) VALUES ( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, + $11, $12, $13, $14, $15, $16, $17::jsonb, 'pending', NOW(), NOW() + ) + ON CONFLICT (order_sn) DO UPDATE SET + shop_id = EXCLUDED.shop_id, + shop_name = EXCLUDED.shop_name, + order_status = EXCLUDED.order_status, + order_status_str = EXCLUDED.order_status_str, + goods_id = EXCLUDED.goods_id, + goods_name = EXCLUDED.goods_name, + out_sku_sn = EXCLUDED.out_sku_sn, + spec = EXCLUDED.spec, + quantity = EXCLUDED.quantity, + order_amount_fen = EXCLUDED.order_amount_fen, + buyer_nickname = EXCLUDED.buyer_nickname, + buyer_memo = EXCLUDED.buyer_memo, + mall_remark = EXCLUDED.mall_remark, + order_time = COALESCE(EXCLUDED.order_time, pdd_raw_orders.order_time), + confirm_time = COALESCE(EXCLUDED.confirm_time, pdd_raw_orders.confirm_time), + raw_payload_json = EXCLUDED.raw_payload_json, + updated_at = NOW() + RETURNING *, (xmax = 0) AS is_inserted + `, + [ + orderSn, + shopId, + shopName, + orderStatus, + orderStatusStr, + goodsId, + goodsName, + outSkuSn, + spec, + quantity, + orderAmountFen, + buyerNickname, + buyerMemo, + mallRemark, + orderTime, + confirmTime, + rawPayloadJson, + ], + ) + + const row = result.rows[0] + if (row) { + rows.push(row) + if ((row as any).is_inserted) { + insertedCount++ + } else { + updatedCount++ + } + } + } + + return { + insertedCount, + updatedCount, + totalCount: rows.length, + rows, + } +} + +/** + * 获取指定店铺最近一次订单成团/支付时间，为增量同步提供断点 + */ +export async function getLatestPddOrderTimeByShopId( + shopId: string, + executor: QueryExecutor = query, +): Promise<{ latestOrderTime: string | null; latestConfirmTime: string | null }> { + const result = await executor<{ + latest_order_time: string | null + latest_confirm_time: string | null + }>( + ` + SELECT + order_time AS latest_order_time, + confirm_time AS latest_confirm_time + FROM pdd_raw_orders + WHERE shop_id = $1 + ORDER BY COALESCE(confirm_time, order_time, created_at) DESC NULLS LAST + LIMIT 1 + `, + [shopId], + ) + + const row = result.rows[0] + return { + latestOrderTime: row?.latest_order_time || null, + latestConfirmTime: row?.latest_confirm_time || null, + } +} + +/** + * 扫描待转化的原始拼多多订单 + */ +export async function findPendingPddRawOrders( + options: { shopId?: string; limit?: number } = {}, + executor: QueryExecutor = query, +): Promise { + const limit = Math.min(100, Math.max(1, options.limit || 50)) + const params: unknown[] = [limit] + let whereClause = "WHERE sync_status = 'pending'" + + if (options.shopId) { + params.push(options.shopId) + whereClause += ` AND shop_id = $${params.length}` + } + + const result = await executor( + ` + SELECT * + FROM pdd_raw_orders + ${whereClause} + ORDER BY COALESCE(confirm_time, order_time, created_at) ASC, id ASC + LIMIT $1 + `, + params, + ) + + return result.rows +} + +/** + * 标记拼多多原始订单已成功进单落库 + */ +export async function markPddRawOrderSynced( + orderSn: string, + syncedOrderId: number, + executor: QueryExecutor = query, +): Promise { + const result = await executor( + ` + UPDATE pdd_raw_orders + SET + sync_status = 'synced', + synced_order_id = $2, + synced_at = NOW(), + sync_error = '', + updated_at = NOW() + WHERE order_sn = $1 + `, + [orderSn, syncedOrderId], + ) + return (result.rowCount ?? 0) > 0 +} + +/** + * 标记拼多多原始订单因规则未命中等原因被忽略 + */ +export async function markPddRawOrderIgnored( + orderSn: string, + reason: string, + executor: QueryExecutor = query, +): Promise { + const result = await executor( + ` + UPDATE pdd_raw_orders + SET + sync_status = 'ignored', + sync_error = $2, + updated_at = NOW() + WHERE order_sn = $1 + `, + [orderSn, sanitizeWellFormedUnicode(reason)], + ) + return (result.rowCount ?? 0) > 0 +} + +/** + * 标记拼多多原始订单进单异常失败 + */ +export async function markPddRawOrderFailed( + orderSn: string, + error: string, + executor: QueryExecutor = query, +): Promise { + const result = await executor( + ` + UPDATE pdd_raw_orders + SET + sync_status = 'failed', + sync_error = $2, + updated_at = NOW() + WHERE order_sn = $1 + `, + [orderSn, sanitizeWellFormedUnicode(error)], + ) + return (result.rowCount ?? 0) > 0 +} + +/** + * 重置指定订单为 pending 状态，用于一键重新扫描派单 + */ +export async function resetPddRawOrdersToPending( + options: { shopId?: string; orderSns?: string[] } = {}, + executor: QueryExecutor = query, +): Promise { + const params: unknown[] = [] + const filters: string[] = ["sync_status IN ('ignored', 'failed')"] + + if (options.shopId) { + params.push(options.shopId) + filters.push(`shop_id = $${params.length}`) + } + + if (options.orderSns && options.orderSns.length > 0) { + params.push(options.orderSns) + filters.push(`order_sn = ANY($${params.length}::text[])`) + } + + const whereClause = `WHERE ${filters.join(' AND ')}` + const result = await executor( + ` + UPDATE pdd_raw_orders + SET + sync_status = 'pending', + sync_error = '', + updated_at = NOW() + ${whereClause} + `, + params, + ) + + return result.rowCount ?? 0 +} + +export interface ListPddRawOrdersQuery { + page?: number | undefined + pageSize?: number | undefined + shopId?: string | undefined + orderSn?: string | undefined + syncStatus?: string | undefined + outSkuSn?: string | undefined + dateFrom?: string | undefined + dateTo?: string | undefined +} + +export interface ListPddRawOrdersResult { + items: PddRawOrderRow[] + total: number + page: number + pageSize: number +} + +/** + * 管理后台查询拼多多原始订单流水列表 + */ +export async function listPddRawOrders( + input: ListPddRawOrdersQuery = {}, + executor: QueryExecutor = query, +): Promise { + const page = Math.max(1, Number(input.page || 1)) + const pageSize = Math.min(100, Math.max(1, Number(input.pageSize || 20))) + const offset = (page - 1) * pageSize + + const filters: string[] = [] + const params: unknown[] = [] + + if (input.shopId) { + params.push(input.shopId.trim()) + filters.push(`shop_id = $${params.length}`) + } + + if (input.orderSn) { + params.push(`%${input.orderSn.trim()}%`) + filters.push(`order_sn ILIKE $${params.length}`) + } + + if (input.syncStatus) { + params.push(input.syncStatus.trim()) + filters.push(`sync_status = $${params.length}`) + } + + if (input.outSkuSn) { + params.push(`%${input.outSkuSn.trim()}%`) + filters.push(`out_sku_sn ILIKE $${params.length}`) + } + + if (input.dateFrom) { + params.push(input.dateFrom.trim().slice(0, 10)) + filters.push(`COALESCE(confirm_time, order_time, created_at) >= ($${params.length}::date)`) + } + + if (input.dateTo) { + params.push(input.dateTo.trim().slice(0, 10)) + filters.push( + `COALESCE(confirm_time, order_time, created_at) < (($${params.length}::date + 1)::timestamp)`, + ) + } + + const whereClause = filters.length > 0 ? `WHERE ${filters.join(' AND ')}` : '' + + const countResult = await executor<{ total: number }>( + `SELECT COUNT(*)::int AS total FROM pdd_raw_orders ${whereClause}`, + params, + ) + const total = Number(countResult.rows[0]?.total || 0) + + params.push(pageSize) + params.push(offset) + + const itemsResult = await executor( + ` + SELECT * + FROM pdd_raw_orders + ${whereClause} + ORDER BY COALESCE(confirm_time, order_time, created_at) DESC, id DESC + LIMIT $${params.length - 1} OFFSET $${params.length} + `, + params, + ) + + return { + items: itemsResult.rows, + total, + page, + pageSize, + } +} diff --git a/apps/backend/src/routes/admin/platform-config/pdd.ts b/apps/backend/src/routes/admin/platform-config/pdd.ts index edddf364..e2ae227f 100644 --- a/apps/backend/src/routes/admin/platform-config/pdd.ts +++ b/apps/backend/src/routes/admin/platform-config/pdd.ts @@ -19,12 +19,69 @@ import { notifyPddCookieExpired } from '../../../services/notification/domain-no import { syncAllPddStores, syncStoreOrders, + replayPddRawOrders, } from '../../../services/platforms/pdd/pdd-sync-service.js' +import { listPddRawOrders } from '../../../repositories/platforms/pdd-raw-order-repo.js' import { testPddCookie } from '../../../services/platforms/pdd/pdd-client.js' import { createHttpError } from '../../../utils/http.js' const router = Router() +/** + * GET /platform-config/pdd/raw-orders + * 查询拼多多原始流水暂存记录 + */ +router.get( + '/pdd/raw-orders', + createJsonHandler( + (req) => { + const q = req.query as Record + return listPddRawOrders({ + page: q.page ? Number(q.page) : 1, + pageSize: q.pageSize ? Number(q.pageSize) : 20, + shopId: q.shopId, + orderSn: q.orderSn, + syncStatus: q.syncStatus, + outSkuSn: q.outSkuSn, + dateFrom: q.dateFrom, + dateTo: q.dateTo, + }) + }, + { + successMessage: 'ok', + errorMessage: '查询拼多多暂存订单失败', + scope: '[admin/platform-config/pdd/raw-orders]', + }, + ), +) + +/** + * POST /platform-config/pdd/raw-orders/replay + * 一键重置未匹配/失败的暂存单据为 pending 并触发转化入库 + */ +router.post( + '/pdd/raw-orders/replay', + createJsonHandler( + async (req) => { + const body = (req.body || {}) as { + shopId?: string + orderSns?: string[] + } + return replayPddRawOrders(body) + }, + { + successMessage: '已重置暂存订单并触发重放', + errorMessage: '重放暂存订单失败', + scope: '[admin/platform-config/pdd/raw-orders/replay]', + audit: (req) => ({ + action: 'platform_pdd_raw_orders_replayed', + targetType: 'platform_raw_orders', + targetId: String((req.body as { shopId?: string })?.shopId || 'all'), + }), + }, + ), +) + /** * GET /platform-config/pdd * Get complete PDD configuration and store list. diff --git a/apps/backend/src/services/platforms/pdd/pdd-sync-service.test.ts b/apps/backend/src/services/platforms/pdd/pdd-sync-service.test.ts index ed6f35ba..7338e8ea 100644 --- a/apps/backend/src/services/platforms/pdd/pdd-sync-service.test.ts +++ b/apps/backend/src/services/platforms/pdd/pdd-sync-service.test.ts @@ -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() // 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() - 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') diff --git a/apps/backend/src/services/platforms/pdd/pdd-sync-service.ts b/apps/backend/src/services/platforms/pdd/pdd-sync-service.ts index fe3c9321..6bca03b3 100644 --- a/apps/backend/src/services/platforms/pdd/pdd-sync-service.ts +++ b/apps/backend/src/services/platforms/pdd/pdd-sync-service.ts @@ -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>) | 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, + } +}