feat(fulfillment): 自动核销任务展示并搜索核销码

This commit is contained in:
yml2213
2026-11-06 10:47:45 +08:00
parent 398eafa911
commit eef4cc1673
7 changed files with 188 additions and 5 deletions
@@ -0,0 +1,82 @@
-- 核销任务直接保存目标券码,供后台“最近核销任务”展示和检索。
-- 任务创建时会先解析候选券码;执行时再刷新一次,兼容券码晚于任务到达的场景。
CREATE EXTENSION IF NOT EXISTS pg_trgm;
ALTER TABLE voucher_consume_jobs
ADD COLUMN IF NOT EXISTS voucher_codes TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[];
COMMENT ON COLUMN voucher_consume_jobs.voucher_codes IS
'核销任务关联的券码;快手电子凭证保存 voucher_code,抖音码速达保存我方与平台侧券码';
-- array_to_string 在 PostgreSQL 中被标记为 STABLE,不能直接用于表达式索引;
-- text[] 的拼接是本项目的确定性展示口径,用 IMMUTABLE 包装支持核销码模糊检索。
CREATE OR REPLACE FUNCTION fulfillment_voucher_codes_search_text(codes TEXT[])
RETURNS TEXT
LANGUAGE sql
IMMUTABLE
PARALLEL SAFE
RETURN ARRAY_TO_STRING(codes, ' ');
COMMENT ON FUNCTION fulfillment_voucher_codes_search_text(TEXT[]) IS
'核销任务券码模糊搜索文本(用于 GIN trigram 表达式索引)';
CREATE INDEX IF NOT EXISTS idx_voucher_consume_jobs_voucher_codes
ON voucher_consume_jobs
USING GIN ((fulfillment_voucher_codes_search_text(voucher_codes)) gin_trgm_ops);
-- 回填已有任务。快手优先按任务/单元定位;没有任务时退回订单号维度。
UPDATE voucher_consume_jobs job
SET voucher_codes = COALESCE((
SELECT ARRAY_AGG(DISTINCT voucher.voucher_code ORDER BY voucher.voucher_code)
FROM kuaishou_industry_vouchers voucher
WHERE job.provider = 'kuaishou_industry'
AND voucher.oid = job.platform_order_id
AND BTRIM(voucher.voucher_code) <> ''
AND (
job.task_id IS NULL
OR voucher.task_id = job.task_id
OR voucher.unit_index = (
SELECT task.unit_index FROM fulfillment_tasks task WHERE task.id = job.task_id
)
)
), ARRAY[]::TEXT[])
WHERE job.provider = 'kuaishou_industry'
AND CARDINALITY(job.voucher_codes) = 0;
-- 抖音码速达可能发生平台侧换码,因此两个券码都保留并可搜索。
UPDATE voucher_consume_jobs job
SET voucher_codes = COALESCE((
SELECT ARRAY_AGG(DISTINCT code ORDER BY code)
FROM (
SELECT certificate.cert_no AS code
FROM redeem_certificates certificate
WHERE certificate.provider = 'douyin_mashida'
AND (
certificate.local_order_id = job.order_id
OR (job.order_id IS NULL AND certificate.order_id = job.platform_order_id)
)
AND BTRIM(certificate.cert_no) <> ''
AND (
job.order_item_id IS NULL
OR certificate.order_item_id IS NULL
OR certificate.order_item_id = job.order_item_id
)
UNION
SELECT certificate.platform_cert_no AS code
FROM redeem_certificates certificate
WHERE certificate.provider = 'douyin_mashida'
AND (
certificate.local_order_id = job.order_id
OR (job.order_id IS NULL AND certificate.order_id = job.platform_order_id)
)
AND BTRIM(certificate.platform_cert_no) <> ''
AND (
job.order_item_id IS NULL
OR certificate.order_item_id IS NULL
OR certificate.order_item_id = job.order_item_id
)
) codes
), ARRAY[]::TEXT[])
WHERE job.provider = 'douyin_mashida'
AND CARDINALITY(job.voucher_codes) = 0;
@@ -3,6 +3,10 @@ import { readFileSync } from 'node:fs'
import test from 'node:test'
const source = readFileSync(new URL('./voucher-consume-queue-service.ts', import.meta.url), 'utf8')
const migration = readFileSync(
new URL('../../db/migrations/142_voucher_consume_job_codes.sql', import.meta.url),
'utf8',
)
test('核销任务列表支持关键词、平台、渠道和状态组合筛选', () => {
assert.ok(source.includes('keyword?: string'))
@@ -14,6 +18,7 @@ test('核销任务列表支持关键词、平台、渠道和状态组合筛选',
assert.ok(source.includes('status = $'))
assert.ok(source.includes('platform_order_id ILIKE'))
assert.ok(source.includes('target_key ILIKE'))
assert.ok(source.includes('fulfillment_voucher_codes_search_text(voucher_codes) ILIKE'))
})
test('核销任务列表使用服务端总数和分页参数', () => {
@@ -22,3 +27,15 @@ test('核销任务列表使用服务端总数和分页参数', () => {
assert.ok(source.includes('OFFSET $'))
assert.ok(source.includes('ORDER BY updated_at DESC, id DESC'))
})
test('核销任务保存并展示核销码,执行时刷新候选券码', () => {
assert.ok(migration.includes('voucher_codes TEXT[]'))
assert.ok(migration.includes('fulfillment_voucher_codes_search_text(voucher_codes)'))
assert.ok(migration.includes('gin_trgm_ops'))
assert.ok(migration.includes('kuaishou_industry_vouchers'))
assert.ok(migration.includes('redeem_certificates'))
assert.ok(source.includes('INSERT INTO voucher_consume_jobs'))
assert.ok(source.includes('candidates.voucherCodes'))
assert.ok(source.includes('updateVoucherConsumeJobCodes'))
assert.ok(source.includes('voucher_codes = $1::text[]'))
})
@@ -6,6 +6,7 @@ import { listKuaishouIndustryVouchersByOid } from '../../repositories/kuaishou-i
import {
findRedeemCertificateByOrder,
listRedeemCertificatesByLocalOrder,
resolveRedeemVerifyCertNo,
type RedeemCertificateRow,
} from '../../repositories/redeem-certificate-repo.js'
import type { WorkOrderRow } from '../../repositories/worker-platform/index.js'
@@ -70,6 +71,7 @@ export type VoucherConsumeJobRow = {
max_attempts: number
locked_until: string | null
last_error: string
voucher_codes: string[]
result_json: string | Record<string, unknown>
created_at: string
updated_at: string
@@ -331,13 +333,14 @@ async function scheduleVoucherConsumeTarget(
order_item_id,
task_id,
work_order_id,
voucher_codes,
due_at,
status,
attempt_count,
max_attempts,
updated_at
) VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10::text[], $11,
'pending', 0, 8, NOW()
)
ON CONFLICT (provider, target_key)
@@ -355,6 +358,7 @@ async function scheduleVoucherConsumeTarget(
target.orderItemId,
target.taskId,
target.workOrderId,
candidates.voucherCodes,
dueAt,
],
)
@@ -488,7 +492,8 @@ export async function listVoucherConsumeJobs(
`(CAST(id AS TEXT) ILIKE $${params.length}
OR platform_order_id ILIKE $${params.length}
OR target_key ILIKE $${params.length}
OR CAST(order_id AS TEXT) ILIKE $${params.length})`,
OR CAST(order_id AS TEXT) ILIKE $${params.length}
OR fulfillment_voucher_codes_search_text(voucher_codes) ILIKE $${params.length})`,
)
}
if (provider) {
@@ -580,6 +585,10 @@ async function executeKuaishouConsumeJob(job: VoucherConsumeJobRow) {
if (vouchers.length === 0) {
throw new Error('未找到待核销的快手电子凭证')
}
await updateVoucherConsumeJobCodes(
job.id,
vouchers.map((voucher) => String(voucher.voucher_code || '').trim()).filter(Boolean),
)
const task = job.task_id ? await getTaskById(job.task_id) : null
let consumedCount = 0
@@ -631,6 +640,18 @@ async function executeDouyinConsumeJob(job: VoucherConsumeJobRow) {
if (certificates.length === 0) {
throw new Error('未找到待核销的抖音券')
}
await updateVoucherConsumeJobCodes(
job.id,
Array.from(
new Set(
certificates.flatMap((certificate) => [
String(certificate.cert_no || '').trim(),
String(certificate.platform_cert_no || '').trim(),
resolveRedeemVerifyCertNo(certificate),
]),
),
).filter(Boolean),
)
let consumedCount = 0
let alreadyConsumedCount = 0
@@ -779,6 +800,7 @@ function buildSyntheticJob(
max_attempts: 8,
locked_until: null,
last_error: '',
voucher_codes: [],
result_json: {},
created_at: '',
updated_at: '',
@@ -837,6 +859,23 @@ async function finishVoucherConsumeJob(
)
}
async function updateVoucherConsumeJobCodes(jobId: number, voucherCodes: string[]) {
const codes = Array.from(new Set(voucherCodes.map((code) => String(code || '').trim()))).filter(
Boolean,
)
if (codes.length === 0) {
return
}
await query(
`
UPDATE voucher_consume_jobs
SET voucher_codes = $1::text[], updated_at = NOW()
WHERE id = $2
`,
[codes, jobId],
)
}
async function failVoucherConsumeJob(job: VoucherConsumeJobRow, error: unknown, retry: boolean) {
const message = error instanceof Error ? error.message : String(error)
if (!retry) {
@@ -315,6 +315,31 @@ export default function VoucherConsumePolicyPanel() {
width: 170,
render: (value: string) => value || '-',
},
{
title: '核销码',
dataIndex: 'voucher_codes',
width: 210,
render: (values: string[] | null | undefined) => {
const codes = Array.isArray(values)
? values.map((value) => String(value || '').trim()).filter(Boolean)
: []
if (codes.length === 0) {
return '-'
}
return (
<Space direction="vertical" size={2}>
{codes.slice(0, 3).map((code) => (
<Typography.Text key={code} copyable={{ text: code }} ellipsis>
{code}
</Typography.Text>
))}
{codes.length > 3 ? (
<Typography.Text type="secondary">另有 {codes.length - 3} 个</Typography.Text>
) : null}
</Space>
)
},
},
{
title: '计划时间',
dataIndex: 'due_at',
@@ -415,7 +440,7 @@ export default function VoucherConsumePolicyPanel() {
<Input.Search
allowClear
prefix={<SearchOutlined />}
placeholder="任务号 / 订单号 / 平台单号"
placeholder="任务号 / 订单号 / 平台单号 / 核销码"
value={keywordDraft}
onChange={(event) => {
const value = event.target.value
@@ -492,7 +517,7 @@ export default function VoucherConsumePolicyPanel() {
setPageSize(nextPageSize)
},
})}
scroll={{ x: 1200 }}
scroll={{ x: 1420 }}
/>
</Space>
)
@@ -209,6 +209,7 @@ export type VoucherConsumeJob = {
attempt_count: number
max_attempts: number
last_error: string
voucher_codes: string[]
created_at: string
updated_at: string
finished_at: string | null
@@ -0,0 +1,19 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
test('自动核销最近任务展示并搜索核销码', () => {
const page = readFileSync(
new URL('../src/pages/admin/panels/VoucherConsumePolicyPanel.tsx', import.meta.url).pathname,
'utf8',
)
const service = readFileSync(
new URL('../src/services/admin/fulfillment-channel.ts', import.meta.url).pathname,
'utf8',
)
assert.ok(page.includes("title: '核销码'"), '最近核销任务必须展示核销码列')
assert.ok(page.includes('copyable={{ text: code }}'), '核销码必须支持一键复制')
assert.ok(page.includes('任务号 / 订单号 / 平台单号 / 核销码'), '搜索框必须明确支持核销码')
assert.ok(service.includes('voucher_codes: string[]'), '任务接口类型必须包含核销码数组')
})
@@ -8,7 +8,7 @@ const panel = readFileSync(
)
test('最近核销任务提供关键词、平台、渠道和状态筛选', () => {
assert.ok(panel.includes('placeholder="任务号 / 订单号 / 平台单号"'))
assert.ok(panel.includes('placeholder="任务号 / 订单号 / 平台单号 / 核销码"'))
assert.ok(panel.includes('placeholder="全部平台"'))
assert.ok(panel.includes('placeholder="全部渠道"'))
assert.ok(panel.includes('placeholder="全部状态"'))