yml/order_site
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

修复订单重复推送走兑换逻辑bug, 修复任务关闭失败bug

Browse Source
This commit is contained in:
yml
2026-04-14 21:32:55 +08:00
parent 4d03b4ef8d
commit 52c190c575
8 changed files with 414 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/src/db/migrations/003_message_delivery_dedupe_index.sql
+3
View File
@@ -0,0 +1,3 @@
CREATE INDEX IF NOT EXISTS idx_message_deliveries_dedupe_success
ON message_deliveries(provider, platform, shop_id, platform_order_id, channel, claim_url, created_at DESC)
WHERE status = 'success';
apps/backend/src/repositories/message-delivery-repo.js
+35
View File
@@ -116,6 +116,41 @@ export async function findLatestSuccessfulMessageDeliveryByTask(taskId, channel)
return result.rows[0] || null
}
export async function findLatestSuccessfulMessageDelivery({
provider = '',
platform = '',
shopId = '',
platformOrderId = '',
channel = '',
claimUrl = '',
} = {}) {
const result = await query(
`
SELECT *
FROM message_deliveries
WHERE provider = $1
AND platform = $2
AND shop_id = $3
AND platform_order_id = $4
AND channel = $5
AND claim_url = $6
AND status = 'success'
ORDER BY id DESC
LIMIT 1
`,
[
provider,
platform,
shopId,
platformOrderId,
channel,
claimUrl,
],
)
return result.rows[0] || null
}
export async function listMessageDeliveries({
page = 1,
pageSize = 20,
apps/backend/src/repositories/order-item-repo.js
+162 -37
View File
@@ -1,13 +1,173 @@
// @ts-check
import { query } from '../db/client.js'
import { query, withTransaction } from '../db/client.js'
/** @typedef {import('../types/repository-inputs.js').OrderItemReplaceInput} OrderItemReplaceInput */
/** @typedef {import('../types/repository-rows.js').OrderItemRow} OrderItemRow */
/** @returns {Promise<OrderItemRow[]>} */
export async function listOrderItemsByOrderId(orderId) {
const result = await query(
return listOrderItemsByOrderIdWithExecutor(query, orderId)
}
/** @returns {Promise<OrderItemRow[]>} */
/** @param {OrderItemReplaceInput[]} items */
export async function replaceOrderItems(orderId, items) {
return withTransaction(async (client) => {
const executor = client.query.bind(client)
const existingItems = await listOrderItemsByOrderIdWithExecutor(executor, orderId)
const plan = resolveOrderItemSyncPlan(existingItems, items)
for (const update of plan.updates) {
await executor(
`
UPDATE order_items
SET
sku_code = $1,
sku_name = $2,
quantity = $3,
spec_json = $4::jsonb,
item_snapshot_json = $5::jsonb,
updated_at = $6
WHERE id = $7
`,
[
update.item.skuCode,
update.item.skuName,
update.item.quantity,
update.item.specJson || '{}',
update.item.itemSnapshotJson || update.item.specJson || '{}',
update.item.updatedAt,
update.orderItemId,
],
)
}
for (const create of plan.creates) {
await executor(
`
INSERT INTO order_items (
order_id,
sku_code,
sku_name,
quantity,
spec_json,
item_snapshot_json,
created_at,
updated_at
) VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7, $8)
`,
[
Number(orderId),
create.skuCode,
create.skuName,
create.quantity,
create.specJson || '{}',
create.itemSnapshotJson || create.specJson || '{}',
create.createdAt,
create.updatedAt,
],
)
}
if (plan.deletes.length > 0) {
const deletableIds = await listDeletableOrderItemIdsWithExecutor(executor, plan.deletes)
if (deletableIds.length > 0) {
await executor(
'DELETE FROM order_items WHERE id = ANY($1::bigint[])',
[deletableIds],
)
}
}
return listOrderItemsByOrderIdWithExecutor(executor, orderId)
})
}
/**
* @param {OrderItemRow[]} existingItems
* @param {OrderItemReplaceInput[]} nextItems
*/
export function resolveOrderItemSyncPlan(existingItems, nextItems) {
const normalizedExisting = Array.isArray(existingItems) ? existingItems : []
const normalizedNext = Array.isArray(nextItems) ? nextItems : []
const unmatchedExisting = [...normalizedExisting]
const updates = []
const creates = []
for (const item of normalizedNext) {
const matchedIndex = findMatchingExistingOrderItemIndex(unmatchedExisting, item)
if (matchedIndex >= 0) {
const matched = unmatchedExisting.splice(matchedIndex, 1)[0]
updates.push({
orderItemId: matched.id,
item,
})
continue
}
if (unmatchedExisting.length > 0) {
const matched = unmatchedExisting.shift()
updates.push({
orderItemId: matched.id,
item,
})
continue
}
creates.push(item)
}
return {
updates,
creates,
deletes: unmatchedExisting.map((item) => Number(item.id)).filter((value) => value > 0),
}
}
function findMatchingExistingOrderItemIndex(existingItems, nextItem) {
const nextIdentity = buildOrderItemIdentity(nextItem)
return existingItems.findIndex((item) => buildOrderItemIdentity(item) === nextIdentity)
}
function buildOrderItemIdentity(item) {
return [
String(item?.skuCode ?? item?.sku_code ?? '').trim(),
String(item?.skuName ?? item?.sku_name ?? '').trim(),
String(Math.max(1, Number(item?.quantity || 1))),
].join('::')
}
async function listDeletableOrderItemIdsWithExecutor(executor, orderItemIds) {
const normalizedIds = Array.isArray(orderItemIds)
? orderItemIds.map((value) => Number(value)).filter((value) => value > 0)
: []
if (normalizedIds.length === 0) {
return []
}
const result = await executor(
`
SELECT oi.id
FROM order_items oi
LEFT JOIN fulfillment_tasks ft ON ft.order_item_id = oi.id
WHERE oi.id = ANY($1::bigint[])
GROUP BY oi.id
HAVING COUNT(ft.id) = 0
ORDER BY oi.id ASC
`,
[normalizedIds],
)
return result.rows.map((row) => Number(row.id)).filter((value) => value > 0)
}
async function listOrderItemsByOrderIdWithExecutor(executor, orderId) {
const result = await executor(
`
SELECT *
FROM order_items
@@ -20,41 +180,6 @@ export async function listOrderItemsByOrderId(orderId) {
return /** @type {OrderItemRow[]} */ (result.rows)
}
/** @returns {Promise<OrderItemRow[]>} */
/** @param {OrderItemReplaceInput[]} items */
export async function replaceOrderItems(orderId, items) {
await query('DELETE FROM order_items WHERE order_id = $1', [Number(orderId)])
for (const item of items) {
await query(
`
INSERT INTO order_items (
order_id,
sku_code,
sku_name,
quantity,
spec_json,
item_snapshot_json,
created_at,
updated_at
) VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7, $8)
`,
[
Number(orderId),
item.skuCode,
item.skuName,
item.quantity,
item.specJson || '{}',
item.itemSnapshotJson || item.specJson || '{}',
item.createdAt,
item.updatedAt,
],
)
}
return listOrderItemsByOrderId(orderId)
}
/** @returns {Promise<OrderItemRow | null>} */
export async function getOrderItemById(orderItemId) {
const result = await query('SELECT * FROM order_items WHERE id = $1 LIMIT 1', [Number(orderItemId)])
apps/backend/src/repositories/order-item-repo.test.js
+94
View File
@@ -0,0 +1,94 @@
import test from 'node:test'
import assert from 'node:assert/strict'
import { resolveOrderItemSyncPlan } from './order-item-repo.js'
test('resolveOrderItemSyncPlan preserves matching order item ids across repeated webhook syncs', () => {
const existingItems = [
{
id: 15,
sku_code: 'sku-a',
sku_name: '礼包 A',
quantity: 1,
},
]
const nextItems = [
{
skuCode: 'sku-a',
skuName: '礼包 A',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: '2026-04-14T12:00:00.000Z',
updatedAt: '2026-04-14T12:00:00.000Z',
},
]
const plan = resolveOrderItemSyncPlan(existingItems, nextItems)
assert.deepEqual(plan.updates, [
{
orderItemId: 15,
item: nextItems[0],
},
])
assert.deepEqual(plan.creates, [])
assert.deepEqual(plan.deletes, [])
})
test('resolveOrderItemSyncPlan matches by identity before falling back to position', () => {
const existingItems = [
{ id: 21, sku_code: 'sku-a', sku_name: '礼包 A', quantity: 1 },
{ id: 22, sku_code: 'sku-b', sku_name: '礼包 B', quantity: 1 },
]
const nextItems = [
{
skuCode: 'sku-b',
skuName: '礼包 B',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: '2026-04-14T12:00:00.000Z',
updatedAt: '2026-04-14T12:00:00.000Z',
},
{
skuCode: 'sku-a',
skuName: '礼包 A',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: '2026-04-14T12:00:00.000Z',
updatedAt: '2026-04-14T12:00:00.000Z',
},
]
const plan = resolveOrderItemSyncPlan(existingItems, nextItems)
assert.deepEqual(plan.updates.map((item) => item.orderItemId), [22, 21])
assert.deepEqual(plan.creates, [])
assert.deepEqual(plan.deletes, [])
})
test('resolveOrderItemSyncPlan keeps surplus existing ids for conditional cleanup instead of recreating everything', () => {
const existingItems = [
{ id: 31, sku_code: 'sku-a', sku_name: '礼包 A', quantity: 1 },
{ id: 32, sku_code: 'sku-b', sku_name: '礼包 B', quantity: 1 },
]
const nextItems = [
{
skuCode: 'sku-a',
skuName: '礼包 A',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: '2026-04-14T12:00:00.000Z',
updatedAt: '2026-04-14T12:00:00.000Z',
},
]
const plan = resolveOrderItemSyncPlan(existingItems, nextItems)
assert.deepEqual(plan.updates.map((item) => item.orderItemId), [31])
assert.deepEqual(plan.creates, [])
assert.deepEqual(plan.deletes, [32])
})
apps/backend/src/services/order/delivery-task-service.js
+1 -1
View File
@@ -110,7 +110,7 @@ async function preparePaidTask(task) {
const primaryRequirement = taskContext.primaryRequirement || null
const inventorySkuCode = String(taskContext.inventorySkuCode || task.skuCode || '').trim()
if (['link_generated', 'claimed', 'role_confirmed', 'redeeming', 'redeemed'].includes(task.task_status)) {
if (['link_generated', 'claimed', 'role_confirmed', 'redeeming', 'redeemed', 'closed'].includes(task.task_status)) {
return task
}
apps/backend/src/services/platforms/agiso/xianyu/message-service.js
+39 -10
View File
@@ -4,7 +4,7 @@ import { runtimeConfig } from '../../../../config/runtime.js'
import { getAgisoMessagingDefaults, getAgisoShopConfigMap } from '../shop-config-service.js'
import {
createMessageDelivery,
findLatestSuccessfulMessageDeliveryByTask,
findLatestSuccessfulMessageDelivery,
updateMessageDelivery,
} from '../../../../repositories/message-delivery-repo.js'
import { nowIso } from '../../../../utils/time.js'
@@ -56,6 +56,28 @@ async function deliverAgisoXianyuMessageForTask({
messageContent,
claimUrl = '',
} = {}) {
return deliverAgisoXianyuMessageForTaskWithDeps({
order,
task,
channel,
messageContent,
claimUrl,
})
}
export async function deliverAgisoXianyuMessageForTaskWithDeps({
order,
task,
channel,
messageContent,
claimUrl = '',
} = {}, deps = {}) {
const now = deps.nowIso || nowIso
const findSuccessfulDelivery = deps.findLatestSuccessfulMessageDelivery || findLatestSuccessfulMessageDelivery
const insertMessageDelivery = deps.createMessageDelivery || createMessageDelivery
const patchMessageDelivery = deps.updateMessageDelivery || updateMessageDelivery
const sendRequest = deps.fetch || fetch
if (!order || !task || !messageContent) {
return { sent: false, skipped: true, reason: 'missing_message_context' }
}
@@ -72,7 +94,14 @@ async function deliverAgisoXianyuMessageForTask({
return { sent: false, skipped: true, reason: 'missing_endpoint_or_access_token_or_app_secret' }
}
const successful = await findLatestSuccessfulMessageDeliveryByTask(task.id, channel)
const successful = await findSuccessfulDelivery({
provider: 'agiso',
platform: String(order.platform || '').trim() || 'unknown',
shopId: String(order.shop_id || '').trim(),
platformOrderId: String(order.platform_order_id || '').trim(),
channel,
claimUrl: String(claimUrl || ''),
})
if (successful) {
return { sent: false, skipped: true, reason: 'already_sent', deliveryId: successful.id }
}
@@ -87,8 +116,8 @@ async function deliverAgisoXianyuMessageForTask({
accessToken,
apiVersion: String(config.apiVersion || '1').trim() || '1',
})
const createdAt = nowIso()
const delivery = await createMessageDelivery({
const createdAt = now()
const delivery = await insertMessageDelivery({
provider: 'agiso',
platform: String(order.platform || '').trim() || 'unknown',
shopId: String(order.shop_id || '').trim(),
@@ -113,7 +142,7 @@ async function deliverAgisoXianyuMessageForTask({
})
try {
const response = await fetch(url, {
const response = await sendRequest(url, {
method: 'POST',
headers: requestHeaders,
body: new URLSearchParams(requestBody).toString(),
@@ -122,13 +151,13 @@ async function deliverAgisoXianyuMessageForTask({
const parsed = safeParseJson(rawText)
const success = isAgisoSendSuccess(response.status, parsed)
const errorMessage = success ? '' : resolveAgisoErrorMessage(parsed, rawText, response.status)
const updated = await updateMessageDelivery(delivery.id, {
const updated = await patchMessageDelivery(delivery.id, {
status: success ? 'success' : 'failed',
response_status: response.status,
response_json: JSON.stringify(parsed ?? { rawText }),
error_message: errorMessage,
sent_at: success ? nowIso() : null,
updated_at: nowIso(),
sent_at: success ? now() : null,
updated_at: now(),
})
return {
@@ -141,13 +170,13 @@ async function deliverAgisoXianyuMessageForTask({
}
} catch (error) {
const message = error instanceof Error ? error.message : String(error || '发送消息失败')
await updateMessageDelivery(delivery.id, {
await patchMessageDelivery(delivery.id, {
status: 'failed',
response_status: 0,
response_json: '{}',
error_message: message,
sent_at: null,
updated_at: nowIso(),
updated_at: now(),
})
return {
apps/backend/src/services/platforms/agiso/xianyu/message-service.test.js
+76
View File
@@ -0,0 +1,76 @@
import test from 'node:test'
import assert from 'node:assert/strict'
import { runtimeConfig } from '../../../../config/runtime.js'
import { deliverAgisoXianyuMessageForTaskWithDeps } from './message-service.js'
test('deliverAgisoXianyuMessageForTaskWithDeps skips duplicate successful claim message by order scope', async () => {
const originalMessaging = runtimeConfig.platforms.agiso.messaging
const originalAppSecret = runtimeConfig.platforms.agiso.appSecret
runtimeConfig.platforms.agiso.messaging = {
...originalMessaging,
enabled: true,
sendMessageEndpoint: 'https://example.com/send',
apiVersion: '1',
accessToken: 'access-token',
}
runtimeConfig.platforms.agiso.appSecret = 'app-secret'
const calls = {
find: [],
create: 0,
fetch: 0,
}
try {
const result = await deliverAgisoXianyuMessageForTaskWithDeps({
order: {
id: 15,
platform: 'xianyu',
shop_id: '2209880145223',
platform_order_id: '4502280133178028841',
},
task: {
id: 36,
task_no: 'DT7af76e9b2438',
},
channel: 'agiso_im',
messageContent: '测试消息',
claimUrl: 'https://221329.cc.cd/#/claim/same-token',
}, {
findLatestSuccessfulMessageDelivery: async (input) => {
calls.find.push(input)
return { id: 16, task_id: null }
},
createMessageDelivery: async () => {
calls.create += 1
return { id: 999 }
},
fetch: async () => {
calls.fetch += 1
return { status: 200, text: async () => '{}' }
},
})
assert.deepEqual(calls.find, [
{
provider: 'agiso',
platform: 'xianyu',
shopId: '2209880145223',
platformOrderId: '4502280133178028841',
channel: 'agiso_im',
claimUrl: 'https://221329.cc.cd/#/claim/same-token',
},
])
assert.equal(calls.create, 0)
assert.equal(calls.fetch, 0)
assert.equal(result.sent, false)
assert.equal(result.skipped, true)
assert.equal(result.reason, 'already_sent')
assert.equal(result.deliveryId, 16)
} finally {
runtimeConfig.platforms.agiso.messaging = originalMessaging
runtimeConfig.platforms.agiso.appSecret = originalAppSecret
}
})