order_site/apps/backend/scripts/backfill-voucher-consume-order-received.ts

#!/usr/bin/env tsx
/**
 * 为部署前已支付、已有可核销券但未进入订单级核销队列的订单补投任务。
 *
 * 只处理明确可核销的券：
 *   - 快手电子凭证：UNUSED 且发码回调成功；
 *   - 抖音码速达：issued/synced 且同步成功。
 *
 * 默认 dry-run。执行必须显式确认。
 */
import process from 'node:process'

import { closeDb, query } from '../src/db/client.js'
import { loadAppConfigCacheFromDatabase } from '../src/services/config/app-config-store.js'
import { getVoucherConsumePolicy } from '../src/services/fulfillment/voucher-consume-policy-service.js'
import { scheduleVoucherConsumeForOrderReceived } from '../src/services/fulfillment/voucher-consume-queue-service.js'

const CONFIRMATION = 'ENQUEUE_ORDER_RECEIVED'
const options = parseOptions(process.argv.slice(2))
const execute = options.execute === 'true'
const confirmation = String(options.confirm || '').trim()
const days = normalizeInteger(options.days, 7, 1, 90)
const batchSize = normalizeInteger(options.batchSize, 500, 1, 5000)
const maxBatches = normalizeInteger(options.maxBatches, 1000, 1, 10000)
const dateTo = normalizeDate(options.to) || today()
const dateFrom = normalizeDate(options.from) || shiftDate(dateTo, -(days - 1))

const ELIGIBLE_ORDER_SQL = `
  o.created_at >= $1::date
  AND o.created_at < ($2::date + INTERVAL '1 day')
  AND o.pay_status = 'paid'
  AND o.order_status NOT IN ('cancelled', 'closed', 'refunded')
  AND (
    (
      $3::boolean
      AND (o.provider IN ('91kaquan', 'open_91', '91') OR o.platform = 'kuaishou')
      AND EXISTS (
        SELECT 1
        FROM kuaishou_industry_vouchers voucher
        WHERE voucher.oid = o.platform_order_id
          AND voucher.status = 'UNUSED'
          AND voucher.send_callback_status = 'success'
      )
    )
    OR
    (
      $4::boolean
      AND o.provider = 'douyin_mashida'
      AND EXISTS (
        SELECT 1
        FROM redeem_certificates certificate
        WHERE certificate.provider = 'douyin_mashida'
          AND (
            certificate.local_order_id = o.id
            OR certificate.order_id = o.platform_order_id
          )
          AND certificate.status IN ('issued', 'synced')
          AND certificate.sync_status = 'success'
      )
    )
  )
  AND NOT EXISTS (
    SELECT 1
    FROM voucher_consume_jobs job
    WHERE job.order_id = o.id
      AND job.status IN ('pending', 'processing', 'succeeded')
  )
`

async function main() {
  if (dateFrom > dateTo) {
    throw new Error(`--from 不能晚于 --to：${dateFrom} > ${dateTo}`)
  }

  await loadAppConfigCacheFromDatabase()
  const policy = getVoucherConsumePolicy()
  const enabledProviders = (['kuaishou_industry', 'douyin_mashida'] as const).filter(
    (provider) => policy.platforms[provider].channels.order_received.enabled,
  )
  if (enabledProviders.length === 0) {
    console.log(
      JSON.stringify(
        {
          mode: execute ? 'execute' : 'dry-run',
          dateFrom,
          dateTo,
          policyDisabled: true,
          enabledProviders,
          pendingOrders: 0,
        },
        null,
        2,
      ),
    )
    return
  }

  const includeKuaishou = enabledProviders.includes('kuaishou_industry')
  const includeDouyin = enabledProviders.includes('douyin_mashida')
  const pending = await countEligibleOrders(includeKuaishou, includeDouyin)
  const startedAt = Date.now()

  if (!execute) {
    const samples = await previewEligibleOrders(includeKuaishou, includeDouyin)
    console.log(
      JSON.stringify(
        {
          mode: 'dry-run',
          dateFrom,
          dateTo,
          pendingOrders: pending,
          batchSize,
          estimatedBatches: Math.ceil(pending / batchSize),
          samples,
          durationMs: Date.now() - startedAt,
          hint: `确认后执行：npm run voucher:backfill-order-received -- --from=${dateFrom} --to=${dateTo} --execute --confirm=${CONFIRMATION}`,
        },
        null,
        2,
      ),
    )
    return
  }

  if (confirmation !== CONFIRMATION) {
    throw new Error(`--confirm 必须是 ${CONFIRMATION}`)
  }

  let cursor = 0
  let scanned = 0
  let queued = 0
  let batches = 0

  while (batches < maxBatches) {
    const orderIds = await loadEligibleOrderIds(cursor, batchSize, includeKuaishou, includeDouyin)
    if (orderIds.length === 0) break

    const scheduled = await scheduleVoucherConsumeForOrderReceived(orderIds)
    scanned += orderIds.length
    queued += scheduled.queuedCount
    cursor = orderIds[orderIds.length - 1]!
    batches += 1

    if (orderIds.length < batchSize) break
  }

  const remaining = await countEligibleOrders(includeKuaishou, includeDouyin)
  console.log(
    JSON.stringify(
      {
        mode: 'execute',
        dateFrom,
        dateTo,
        pendingBefore: pending,
        batches,
        scannedOrders: scanned,
        queuedJobs: queued,
        remainingOrders: remaining,
        durationMs: Date.now() - startedAt,
      },
      null,
      2,
    ),
  )

  if (remaining > 0) {
    console.error(`仍有 ${remaining} 个订单未入队；请增大 --max-batches 后重跑（幂等）。`)
    process.exitCode = 2
  }
}

async function countEligibleOrders(includeKuaishou: boolean, includeDouyin: boolean) {
  const result = await query<{ count: number }>(
    `SELECT COUNT(*)::int AS count
     FROM orders o
     WHERE ${ELIGIBLE_ORDER_SQL}`,
    [dateFrom, dateTo, includeKuaishou, includeDouyin],
  )
  return Number(result.rows[0]?.count || 0)
}

async function loadEligibleOrderIds(
  afterId: number,
  limit: number,
  includeKuaishou: boolean,
  includeDouyin: boolean,
) {
  const result = await query<{ id: number }>(
    `SELECT o.id
     FROM orders o
     WHERE o.id > $5
       AND ${ELIGIBLE_ORDER_SQL}
     ORDER BY o.id
     LIMIT $6`,
    [dateFrom, dateTo, includeKuaishou, includeDouyin, afterId, limit],
  )
  return result.rows.map((row) => Number(row.id))
}

async function previewEligibleOrders(includeKuaishou: boolean, includeDouyin: boolean) {
  const result = await query<{
    id: number
    provider: string
    platform_order_id: string
    created_at: string
  }>(
    `SELECT o.id, o.provider, o.platform_order_id, o.created_at::text
     FROM orders o
     WHERE ${ELIGIBLE_ORDER_SQL}
     ORDER BY o.id
     LIMIT 10`,
    [dateFrom, dateTo, includeKuaishou, includeDouyin],
  )
  return result.rows
}

function parseOptions(args: string[]) {
  const result: Record<string, string> = {}
  for (const arg of args) {
    if (!arg.startsWith('--')) continue
    const body = arg.slice(2)
    const index = body.indexOf('=')
    const key = index === -1 ? body : body.slice(0, index)
    const value = index === -1 ? 'true' : body.slice(index + 1)
    result[key] = value
    result[toCamelCase(key)] = value
  }
  return result
}

function toCamelCase(value: string) {
  return value.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())
}

function normalizeInteger(value: unknown, fallback: number, min: number, max: number) {
  const parsed = Number(value)
  if (!Number.isInteger(parsed)) return fallback
  return Math.min(max, Math.max(min, parsed))
}

function normalizeDate(value: unknown): string {
  const text = String(value || '').trim()
  return /^\d{4}-\d{2}-\d{2}$/.test(text) ? text : ''
}

function today() {
  return new Date().toLocaleDateString('sv-SE', { timeZone: 'Asia/Shanghai' })
}

function shiftDate(value: string, days: number) {
  const date = new Date(`${value}T12:00:00+08:00`)
  date.setUTCDate(date.getUTCDate() + days)
  return date.toLocaleDateString('sv-SE', { timeZone: 'Asia/Shanghai' })
}

main()
  .catch((error) => {
    console.error(error instanceof Error ? error.message : error)
    process.exitCode = 1
  })
  .finally(() => closeDb())