diff --git a/apps/backend/src/db/migrations/003_message_delivery_dedupe_index.sql b/apps/backend/src/db/migrations/003_message_delivery_dedupe_index.sql new file mode 100644 index 00000000..a201d627 --- /dev/null +++ b/apps/backend/src/db/migrations/003_message_delivery_dedupe_index.sql @@ -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'; diff --git a/apps/backend/src/repositories/message-delivery-repo.js b/apps/backend/src/repositories/message-delivery-repo.js index 1d419659..824e7454 100644 --- a/apps/backend/src/repositories/message-delivery-repo.js +++ b/apps/backend/src/repositories/message-delivery-repo.js @@ -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, diff --git a/apps/backend/src/repositories/order-item-repo.js b/apps/backend/src/repositories/order-item-repo.js index e3e61cfa..8f30372a 100644 --- a/apps/backend/src/repositories/order-item-repo.js +++ b/apps/backend/src/repositories/order-item-repo.js @@ -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} */ export async function listOrderItemsByOrderId(orderId) { - const result = await query( + return listOrderItemsByOrderIdWithExecutor(query, orderId) +} + +/** @returns {Promise} */ +/** @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} */ -/** @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} */ export async function getOrderItemById(orderItemId) { const result = await query('SELECT * FROM order_items WHERE id = $1 LIMIT 1', [Number(orderItemId)]) diff --git a/apps/backend/src/repositories/order-item-repo.test.js b/apps/backend/src/repositories/order-item-repo.test.js new file mode 100644 index 00000000..f9d5de1c --- /dev/null +++ b/apps/backend/src/repositories/order-item-repo.test.js @@ -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]) +}) diff --git a/apps/backend/src/services/order/delivery-task-service.js b/apps/backend/src/services/order/delivery-task-service.js index 17e76433..9b2a6f1e 100644 --- a/apps/backend/src/services/order/delivery-task-service.js +++ b/apps/backend/src/services/order/delivery-task-service.js @@ -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 } diff --git a/apps/backend/src/services/platforms/agiso/xianyu/message-service.js b/apps/backend/src/services/platforms/agiso/xianyu/message-service.js index d6ffda63..39967ad3 100644 --- a/apps/backend/src/services/platforms/agiso/xianyu/message-service.js +++ b/apps/backend/src/services/platforms/agiso/xianyu/message-service.js @@ -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 { diff --git a/apps/backend/src/services/platforms/agiso/xianyu/message-service.test.js b/apps/backend/src/services/platforms/agiso/xianyu/message-service.test.js new file mode 100644 index 00000000..39d05b14 --- /dev/null +++ b/apps/backend/src/services/platforms/agiso/xianyu/message-service.test.js @@ -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 + } +}) diff --git a/apps/frontend/src/views/admin/AdminTasksView.vue b/apps/frontend/src/views/admin/AdminTasksView.vue index 2293d2ab..9c76466c 100644 --- a/apps/frontend/src/views/admin/AdminTasksView.vue +++ b/apps/frontend/src/views/admin/AdminTasksView.vue @@ -132,16 +132,7 @@ async function runAction( } } -function handleActionCommand(command: `${AdminTaskActionKey}:${number}`) { - const [actionKey, rawTaskId] = command.split(':') as [AdminTaskActionKey, string] - const taskId = Number(rawTaskId) - const item = items.value.find((entry) => entry.taskId === taskId) - - if (!item) { - showError('未找到对应任务') - return - } - +function handleActionCommand(actionKey: AdminTaskActionKey, item: AdminTaskListItem) { if (actionKey === 'retry') { void runAction(item.taskId, () => retryAdminTask(item.taskId), '任务已重试', `确认重试任务 ${item.taskNo} 吗？`) return @@ -306,7 +297,7 @@ function handleActionCommand(command: `${AdminTaskActionKey}:${number}`) { v-if="canManageTaskLifecycle && item.executorKey !== 'manual_dispatch'" class="menu-action-button" type="button" - @click="handleActionCommand(`regenerate_claim_link:${item.taskId}`)" + @click="handleActionCommand('regenerate_claim_link', item)" > 重发链接 @@ -314,7 +305,7 @@ function handleActionCommand(command: `${AdminTaskActionKey}:${number}`) { v-if="canManageTaskLifecycle" class="menu-action-button" type="button" - @click="handleActionCommand(`manual_review:${item.taskId}`)" + @click="handleActionCommand('manual_review', item)" > 转人工 @@ -322,7 +313,7 @@ function handleActionCommand(command: `${AdminTaskActionKey}:${number}`) { v-if="!['redeemed', 'closed'].includes(item.status)" class="menu-action-button menu-action-button-danger" type="button" - @click="handleActionCommand(`close:${item.taskId}`)" + @click="handleActionCommand('close', item)" > 关闭任务