修复错误

2026-07-08 20:41:29 +08:00
parent 487d586492
commit 7b215a66a5
6 changed files with 436 additions and 27 deletions
@@ -2,10 +2,13 @@ import { findLatestOrderByPlatformOrderId } from '../../../repositories/order-re
 import { listTasksByOrderId } from '../../../repositories/task-repo.js'
 import {
  findKuaishouIndustryVoucherByCode,
+  listKuaishouIndustryVouchersByOid,
+  listKuaishouIndustryVouchersForSendCallbackRetry,
  updateKuaishouIndustryVoucherByCode,
  upsertKuaishouIndustryVoucher,
 } from '../../../repositories/kuaishou-industry-voucher-repo.js'
 import { OPEN_91_PLATFORM, OPEN_91_PROVIDER } from '../../open-91/config.js'
+import { retryOpen91Order } from '../ninetyone/order-service.js'
 import { normalizeTimestampIso } from '../../../utils/time.js'
 import { logIntegration } from '../../../utils/logger.js'
 import {
@@ -15,7 +18,6 @@ import {
 import { normalizeSendCodePayload, assertSendCodePayload } from './payload.js'
 import { assertKuaishouIndustrySignature } from './crypto.js'
 import {
-  buildIndustryErrorResponse,
  buildIndustrySuccessResponse,
  buildIndustryEticketItem,
  buildIndustrySendCodeData,
@@ -23,6 +25,7 @@ import {
 import { sendCallback } from './send-callback-service.js'
 import {
  buildKuaishouIndustryEticketFromVoucher,
+  isKuaishouIndustryVoucherSendCallbackSuccess,
  KUAISHOU_INDUSTRY_SEND_CALLBACK_STATUS,
  resolveKuaishouIndustryVoucherValidity,
 } from './voucher-service.js'
@@ -41,6 +44,15 @@ type SendCodeCallbackParams = {
  ext?: string
 }

+const SEND_CALLBACK_INITIAL_DELAY_MS = 3_000
+const SEND_CALLBACK_SCAN_INTERVAL_MS = 15_000
+const SEND_CALLBACK_MAX_ATTEMPTS = 8
+const SEND_CALLBACK_BACKOFF_MS = [3_000, 10_000, 30_000, 60_000, 120_000, 300_000]
+
+const sendCallbackRetryTimers = new Map<string, NodeJS.Timeout>()
+let sendCallbackWorkerTimer: NodeJS.Timeout | null = null
+let sendCallbackWorkerRunning = false
+
 export async function handleSendCode(rawBody: JsonObject = {}) {
  const config = getKuaishouIndustryConfig()

@@ -99,25 +111,6 @@ export async function handleSendCode(rawBody: JsonObject = {}) {
    buildKuaishouIndustryEticketFromVoucher(voucher, params.eticketType),
  )

-  const callbackResult = await sendAndRecordCallback({
-    oid: normalizedOid,
-    sendType: params.sendType,
-    etickets,
-    vouchers,
-    sellerId: params.sellerId,
-    token: params.token,
-    eticketType: params.eticketType,
-    ext: params.ext,
-    preferredTotalGoodsValue: resolveSendCallbackPreferredTotalGoodsValue(order),
-  })
-
-  if (!callbackResult.success) {
-    return buildIndustryErrorResponse(
-      4010003,
-      resolveSendCallbackFailureMessage(callbackResult),
-    )
-  }
-
  if (order) {
    await bindKuaishouIndustryVouchersToOrderTasks(order, tasks, {
      source: 'kuaishou_industry_send_code',
@@ -125,6 +118,10 @@ export async function handleSendCode(rawBody: JsonObject = {}) {
    })
  }

+  scheduleKuaishouIndustrySendCallbackRetry(normalizedOid, SEND_CALLBACK_INITIAL_DELAY_MS, {
+    source: 'send_code_accepted',
+  })
+
  return buildIndustrySuccessResponse(
    buildIndustrySendCodeData({
      oid: normalizedOid,
@@ -135,6 +132,26 @@ export async function handleSendCode(rawBody: JsonObject = {}) {
  )
 }

+export function startKuaishouIndustrySendCallbackRetryWorker() {
+  if (sendCallbackWorkerTimer) {
+    return
+  }
+
+  scheduleSendCallbackWorkerScan(5_000)
+}
+
+export function stopKuaishouIndustrySendCallbackRetryWorker() {
+  if (sendCallbackWorkerTimer) {
+    clearTimeout(sendCallbackWorkerTimer)
+    sendCallbackWorkerTimer = null
+  }
+
+  for (const timer of sendCallbackRetryTimers.values()) {
+    clearTimeout(timer)
+  }
+  sendCallbackRetryTimers.clear()
+}
+
 export async function resendKuaishouIndustryVoucherSendCallback(input: {
  voucherCode: string
  oid?: string
@@ -170,12 +187,273 @@ export async function resendKuaishouIndustryVoucherSendCallback(input: {
  })
  const updatedVoucher = await findKuaishouIndustryVoucherByCode(voucher.voucher_code, voucher.oid)

+  if (result.success) {
+    await syncOpen91OrderAfterSendCallbackSuccess(params.oid, {
+      source: 'admin_resend',
+    })
+  }
+
  return {
    ...result,
    voucher: updatedVoucher || voucher,
  }
 }

+async function runKuaishouIndustrySendCallbackForOid(
+  oid: string,
+  { source = 'async_retry' }: { source?: string } = {},
+) {
+  const normalizedOid = String(oid || '').trim()
+  if (!normalizedOid) {
+    return {
+      success: false,
+      error: '缺少订单号，无法发起电子凭证发货回调',
+    }
+  }
+
+  const vouchers = await listKuaishouIndustryVouchersByOid(normalizedOid)
+  const pendingVouchers = vouchers
+    .filter((voucher) => !isKuaishouIndustryVoucherSendCallbackSuccess(voucher))
+    .filter((voucher) =>
+      normalizePositiveInteger(voucher.send_callback_attempt_count) < SEND_CALLBACK_MAX_ATTEMPTS,
+    )
+
+  if (pendingVouchers.length === 0) {
+    return {
+      success: vouchers.length > 0,
+      skipped: true,
+      reason: vouchers.length > 0 ? 'send_callback_already_success_or_max_attempts' : 'voucher_missing',
+    }
+  }
+
+  const rawPayload = parseJsonObject(pendingVouchers[0]?.raw_payload_json)
+  const body = parseJsonObject(rawPayload.body)
+  const firstVoucher = pendingVouchers[0]
+  if (!firstVoucher) {
+    return {
+      success: false,
+      error: '缺少待回调电子凭证',
+    }
+  }
+
+  const params = resolveSendCodeCallbackParams(firstVoucher, body)
+  const eticketType = String(params.eticketType || '').trim()
+  const etickets = pendingVouchers.map((voucher) =>
+    buildKuaishouIndustryEticketFromVoucher(voucher, eticketType),
+  )
+  const order = await findLatestOrderByPlatformOrderId({
+    provider: OPEN_91_PROVIDER,
+    platform: OPEN_91_PLATFORM,
+    platformOrderId: normalizedOid,
+  })
+  const result = await sendAndRecordCallback({
+    oid: params.oid,
+    sendType: params.sendType,
+    etickets,
+    vouchers: pendingVouchers,
+    sellerId: params.sellerId,
+    token: params.token,
+    eticketType,
+    source,
+    preferredTotalGoodsValue: resolveSendCallbackPreferredTotalGoodsValue(order),
+    ...(params.ext ? { ext: params.ext } : {}),
+  })
+
+  if (result.success) {
+    await syncOpen91OrderAfterSendCallbackSuccess(normalizedOid, { source })
+    return result
+  }
+
+  const nextAttemptCount =
+    Math.max(...pendingVouchers.map((voucher) =>
+      normalizePositiveInteger(voucher.send_callback_attempt_count),
+    )) + 1
+  if (
+    nextAttemptCount < SEND_CALLBACK_MAX_ATTEMPTS &&
+    isRetriableSendCallbackFailure(result)
+  ) {
+    scheduleKuaishouIndustrySendCallbackRetry(normalizedOid, resolveSendCallbackBackoffMs(nextAttemptCount), {
+      source: `${source}_retry`,
+    })
+  }
+
+  return result
+}
+
+function scheduleKuaishouIndustrySendCallbackRetry(
+  oid: string,
+  delayMs: number,
+  { source = 'async_retry' }: { source?: string } = {},
+) {
+  const normalizedOid = String(oid || '').trim()
+  if (!normalizedOid || sendCallbackRetryTimers.has(normalizedOid)) {
+    return false
+  }
+
+  const normalizedDelayMs = Math.max(0, Number(delayMs) || 0)
+  const timer = setTimeout(() => {
+    sendCallbackRetryTimers.delete(normalizedOid)
+    void runKuaishouIndustrySendCallbackForOid(normalizedOid, { source }).catch((error) => {
+      logIntegration('[kuaishou-industry/send-code]', '异步电子凭证发货回调执行异常', {
+        source,
+        oid: normalizedOid,
+        error: error instanceof Error ? error.message : String(error),
+      }, { level: 'warn' })
+      scheduleKuaishouIndustrySendCallbackRetry(normalizedOid, resolveSendCallbackBackoffMs(1), {
+        source: `${source}_exception_retry`,
+      })
+    })
+  }, normalizedDelayMs)
+
+  sendCallbackRetryTimers.set(normalizedOid, timer)
+  logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调已进入异步队列', {
+    source,
+    oid: normalizedOid,
+    delayMs: normalizedDelayMs,
+  })
+  return true
+}
+
+function scheduleSendCallbackWorkerScan(delayMs: number) {
+  if (sendCallbackWorkerTimer) {
+    return
+  }
+
+  sendCallbackWorkerTimer = setTimeout(() => {
+    sendCallbackWorkerTimer = null
+    void runSendCallbackWorkerScan().finally(() => {
+      scheduleSendCallbackWorkerScan(SEND_CALLBACK_SCAN_INTERVAL_MS)
+    })
+  }, Math.max(0, Number(delayMs) || 0))
+}
+
+async function runSendCallbackWorkerScan() {
+  if (sendCallbackWorkerRunning) {
+    return
+  }
+
+  sendCallbackWorkerRunning = true
+  try {
+    const vouchers = await listKuaishouIndustryVouchersForSendCallbackRetry(200)
+    const dueOids = new Set<string>()
+    for (const voucher of vouchers) {
+      if (isVoucherDueForSendCallbackRetry(voucher)) {
+        dueOids.add(String(voucher.oid || '').trim())
+      }
+    }
+
+    for (const oid of dueOids) {
+      scheduleKuaishouIndustrySendCallbackRetry(oid, 0, {
+        source: 'worker_scan',
+      })
+    }
+  } catch (error) {
+    logIntegration('[kuaishou-industry/send-code]', '扫描待重试电子凭证发货回调失败', {
+      error: error instanceof Error ? error.message : String(error),
+    }, { level: 'warn' })
+  } finally {
+    sendCallbackWorkerRunning = false
+  }
+}
+
+function isVoucherDueForSendCallbackRetry(voucher: KuaishouIndustryVoucherRow) {
+  if (isKuaishouIndustryVoucherSendCallbackSuccess(voucher)) {
+    return false
+  }
+
+  const attemptCount = normalizePositiveInteger(voucher.send_callback_attempt_count)
+  if (attemptCount >= SEND_CALLBACK_MAX_ATTEMPTS) {
+    return false
+  }
+
+  const baseAt = voucher.send_callback_sent_at || voucher.created_at
+  const baseTime = Date.parse(String(baseAt || ''))
+  const elapsedMs = Date.now() - (Number.isFinite(baseTime) ? baseTime : 0)
+  const requiredDelayMs = attemptCount === 0
+    ? SEND_CALLBACK_INITIAL_DELAY_MS
+    : resolveSendCallbackBackoffMs(attemptCount)
+
+  return elapsedMs >= requiredDelayMs
+}
+
+function resolveSendCallbackBackoffMs(attemptCount: unknown) {
+  const attempt = Math.max(1, normalizePositiveInteger(attemptCount))
+  return SEND_CALLBACK_BACKOFF_MS[Math.min(attempt, SEND_CALLBACK_BACKOFF_MS.length - 1)] ||
+    SEND_CALLBACK_BACKOFF_MS[SEND_CALLBACK_BACKOFF_MS.length - 1] ||
+    SEND_CALLBACK_INITIAL_DELAY_MS
+}
+
+async function syncOpen91OrderAfterSendCallbackSuccess(
+  oid: string,
+  { source = 'send_callback_success' }: { source?: string } = {},
+) {
+  const normalizedOid = String(oid || '').trim()
+  if (!normalizedOid) {
+    return
+  }
+
+  const order = await findLatestOrderByPlatformOrderId({
+    provider: OPEN_91_PROVIDER,
+    platform: OPEN_91_PLATFORM,
+    platformOrderId: normalizedOid,
+  })
+  if (!order) {
+    return
+  }
+
+  const tasks = await listTasksByOrderId(order.id)
+  if (tasks.length > 0) {
+    await bindKuaishouIndustryVouchersToOrderTasks(order, tasks, {
+      source,
+      now: normalizeTimestampIso(new Date().toISOString()),
+    })
+    return
+  }
+
+  try {
+    const result = await retryOpen91Order(order.id)
+    logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调成功后已触发 91 订单重试', {
+      source,
+      oid: normalizedOid,
+      orderId: order.id,
+      taskCount: result.tasks.length,
+    })
+  } catch (error) {
+    logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调成功后重试 91 订单失败', {
+      source,
+      oid: normalizedOid,
+      orderId: order.id,
+      error: error instanceof Error ? error.message : String(error),
+    }, { level: 'warn' })
+  }
+}
+
+function isRetriableSendCallbackFailure(result: Awaited<ReturnType<typeof sendCallback>>) {
+  if (result.success) {
+    return false
+  }
+
+  if (String(result.error || '').trim()) {
+    return true
+  }
+
+  const response = parseJsonObject(result.response)
+  const marker = [
+    response.result,
+    response.code,
+    response.sub_code,
+    response.error_msg,
+    response.sub_msg,
+    response.msg,
+  ].map((item) => String(item || '').trim()).join(' ')
+
+  return (
+    marker.includes('9994') ||
+    marker.includes('807000') ||
+    marker.includes('并发冲突')
+  )
+}
+
 async function sendAndRecordCallback(input: {
  oid: string
  sendType: string
@@ -254,7 +532,7 @@ async function sendAndRecordCallback(input: {
  ))

  if (!result.success) {
-    logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调执行失败，发码请求已拒绝继续', {
+    logIntegration('[kuaishou-industry/send-code]', '电子凭证发货回调执行失败，等待异步重试', {
      source: input.source || 'send_code',
      oid: input.oid,
      error: result.error || '',