From e5ad6f4918fc79ad46717dacbcf4ac70119743f1 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 29 May 2026 22:17:51 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A2=86=E5=8F=96/=E5=8F=91=E8=B4=A7=E9=93=BE?= =?UTF-8?q?=E8=B7=AF=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../fulfillment/kuaishou-cloud/index.ts | 101 ++----- .../kuaishou-cloud/task-finalization.test.ts | 67 +++++ .../kuaishou-cloud/task-finalization.ts | 248 ++++++++++++++++-- .../platforms/cloudtentacles/http-client.ts | 2 +- 4 files changed, 317 insertions(+), 101 deletions(-) create mode 100644 apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.test.ts diff --git a/apps/backend/src/services/fulfillment/kuaishou-cloud/index.ts b/apps/backend/src/services/fulfillment/kuaishou-cloud/index.ts index 0b95016f..c9b52091 100644 --- a/apps/backend/src/services/fulfillment/kuaishou-cloud/index.ts +++ b/apps/backend/src/services/fulfillment/kuaishou-cloud/index.ts @@ -2,8 +2,6 @@ import { createTaskEvent } from "../../../repositories/task-event-repo.js"; import { updateTask } from "../../../repositories/task-repo.js"; import { buildClaimUrl, createTaskClaimToken } from "../../claim/claim-service.js"; import { - buyCloudtentaclesSku, - getCloudtentaclesAsset, listCloudtentaclesSku, } from "../../platforms/cloudtentacles/catalog-service.js"; import { getCloudtentaclesKnapsack } from "../../platforms/cloudtentacles/knapsack-service.js"; @@ -14,7 +12,6 @@ import { } from "../../platforms/cloudtentacles/virtual-number-service.js"; import { resolveCloudtentaclesConfig } from "../../platforms/cloudtentacles/helpers.js"; import { - notifyKuaishouCloudAssetNotEnough, notifyKuaishouCloudBindUrlRefreshFailed, } from "../../notification/domain-notifications.js"; import { createHttpError } from "../../../utils/http.js"; @@ -189,78 +186,28 @@ export async function prepareKuaishouCloudFulfillmentTask( }); } - let usedKnapsack = deliveryPlan.items.every((item) => item.missingCount <= 0); - let purchaseTriggered = false; - let assetBefore = 0; - let assetAfter = 0; - - if (!usedKnapsack) { - if (!flowWithResolvedBinding.purchase.autoBuyEnabled) { - throw createHttpError("背包中没有现成库存，且当前配置未开启自动购买", { - statusCode: 409, - errorCode: "kuaishou_cloud_auto_buy_disabled", - }); - } - - const asset = await getCloudtentaclesAsset(cloudContext); - const missingSku = deliveryPlan.items.find( - (item) => item.missingCount > 0 && !item.skuItem - ); - if (missingSku) { - throw createHttpError( - `cloudtentacles 未找到 SKU ${missingSku.cloudSkuId}`, - { - statusCode: 404, - errorCode: "kuaishou_cloud_sku_not_found", - } - ); - } - - assetBefore = Number(asset.asset || 0) || 0; - const targetPrice = deliveryPlan.items.reduce( - (sum, item) => - sum + Math.max(0, item.missingCount) * Number(item.skuItem?.price || 0), - 0 - ); - const requiredAsset = - targetPrice + flowWithResolvedBinding.purchase.minAssetReserve; - if (assetBefore < requiredAsset) { - await notifyKuaishouCloudAssetNotEnough({ - task, - flow: flowWithResolvedBinding, - assetBefore, - requiredAsset, - skuName: deliveryPlan.items - .filter((item) => item.missingCount > 0) - .map((item) => item.cloudSkuName || `SKU ${item.cloudSkuId}`) - .join("、"), - }); - throw createHttpError( - `余额不足，当前 ${assetBefore}，至少需要 ${requiredAsset}`, - { - statusCode: 409, - errorCode: "kuaishou_cloud_asset_not_enough", - } - ); - } - - for (const item of deliveryPlan.items) { - if (item.missingCount <= 0) { - continue; - } - - await buyCloudtentaclesSku({ - ...cloudContext, - id: item.cloudSkuId, - count: item.missingCount, - }); - } - purchaseTriggered = true; - - const assetResult = await getCloudtentaclesAsset(cloudContext); - assetAfter = Number(assetResult.asset || 0) || 0; + const usedKnapsack = deliveryPlan.items.every((item) => item.missingCount <= 0); + if (!usedKnapsack && !flowWithResolvedBinding.purchase.autoBuyEnabled) { + throw createHttpError("背包中没有现成库存，且当前配置未开启自动购买", { + statusCode: 409, + errorCode: "kuaishou_cloud_auto_buy_disabled", + }); } + const missingSku = deliveryPlan.items.find( + (item) => item.missingCount > 0 && !item.skuItem + ); + if (missingSku) { + throw createHttpError(`cloudtentacles 未找到 SKU ${missingSku.cloudSkuId}`, { + statusCode: 404, + errorCode: "kuaishou_cloud_sku_not_found", + }); + } + + const purchaseTriggered = false; + const assetBefore = 0; + const assetAfter = 0; + const preparedBinding = await prepareKuaishouCloudBindResourceWithFallback({ cloudContext, vnKeyCandidates, @@ -305,11 +252,9 @@ export async function prepareKuaishouCloudFulfillmentTask( cloudSkuName: item.cloudSkuName, requiredCount: item.quantity, knapsackCount: item.knapsackCount, - purchasedCount: Math.max(0, item.missingCount), + purchasedCount: 0, })), - purchaseAt: purchaseTriggered - ? now - : flowWithResolvedBinding.purchase.purchaseAt, + purchaseAt: flowWithResolvedBinding.purchase.purchaseAt, }, }, }; @@ -343,7 +288,7 @@ export async function prepareKuaishouCloudFulfillmentTask( cloudSkuName: item.cloudSkuName, quantity: item.quantity, knapsackCount: item.knapsackCount, - purchasedCount: Math.max(0, item.missingCount), + purchasedCount: 0, })), resolvedByName: resolvedBinding.resolvedByName, actor, diff --git a/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.test.ts b/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.test.ts new file mode 100644 index 00000000..1844b674 --- /dev/null +++ b/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.test.ts @@ -0,0 +1,67 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { buildDispatchStockItems } from "./task-finalization.js"; + +test("buildDispatchStockItems 只计算兑换阶段需要补买的库存缺口", () => { + const result = buildDispatchStockItems( + [ + { + cloudSkuId: 101, + cloudSkuName: "", + quantity: 3, + }, + ], + { + skuItems: [ + { + id: 101, + name: "套装-暗影哥特", + price: 88, + }, + ], + knapsackItems: [ + { + id: 101, + name: "套装-暗影哥特", + count: 1, + }, + ], + } + ); + + assert.deepEqual(result, [ + { + cloudSkuId: 101, + cloudSkuName: "套装-暗影哥特", + quantity: 3, + requiredCount: 3, + knapsackCount: 1, + purchasedCount: 2, + }, + ]); +}); + +test("buildDispatchStockItems 背包足够时不需要补买", () => { + const result = buildDispatchStockItems( + [ + { + cloudSkuId: 202, + cloudSkuName: "套装-浪漫天命", + quantity: 1, + }, + ], + { + knapsackItems: [ + { + id: 202, + name: "套装-浪漫天命", + count: 5, + }, + ], + } + ); + + assert.equal(result[0]?.knapsackCount, 5); + assert.equal(result[0]?.purchasedCount, 0); +}); diff --git a/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.ts b/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.ts index 2338348a..aeb2fb7e 100644 --- a/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.ts +++ b/apps/backend/src/services/fulfillment/kuaishou-cloud/task-finalization.ts @@ -4,8 +4,17 @@ import { updateTask } from "../../../repositories/task-repo.js"; import { createHttpError } from "../../../utils/http.js"; import { nowIso } from "../../../utils/time.js"; import { TASK_STATUS, normalizeTaskStatus, type TaskStatus } from "../../../domain/task-status.js"; -import { notifyKuaishouCloudConsumeFailed } from "../../notification/domain-notifications.js"; -import { useCloudtentaclesSku } from "../../platforms/cloudtentacles/catalog-service.js"; +import { + notifyKuaishouCloudAssetNotEnough, + notifyKuaishouCloudConsumeFailed, +} from "../../notification/domain-notifications.js"; +import { + buyCloudtentaclesSku, + getCloudtentaclesAsset, + listCloudtentaclesSku, + useCloudtentaclesSku, +} from "../../platforms/cloudtentacles/catalog-service.js"; +import { getCloudtentaclesKnapsack } from "../../platforms/cloudtentacles/knapsack-service.js"; import { backCloudtentaclesVirtualNumber } from "../../platforms/cloudtentacles/virtual-number-service.js"; import { consumeKuaishouEticket } from "../../platforms/kuaishou-eticket/consume-service.js"; import { isKuaishouEticketMockTicketCode } from "../../platforms/kuaishou-eticket/mock-ticket-service.js"; @@ -38,6 +47,22 @@ type DispatchResultItem = DispatchDeliveryItem & { responseMessage: string; }; +type DispatchStockItem = DispatchDeliveryItem & { + requiredCount: number; + knapsackCount: number; + purchasedCount: number; +}; + +type DispatchStockResult = { + usedKnapsack: boolean; + purchaseTriggered: boolean; + assetBefore: number; + assetAfter: number; + items: DispatchStockItem[]; +}; + +const cloudSkuDispatchLocks = new Map>(); + export async function dispatchKuaishouCloudFulfillmentTask( task: TaskRow, options: JsonObject = {} @@ -103,26 +128,15 @@ export async function dispatchKuaishouCloudFulfillmentTask( }); } - const dispatchResults: DispatchResultItem[] = []; - for (const item of deliveryItems) { - for (let index = 0; index < item.quantity; index += 1) { - const dispatchResult = await useCloudtentaclesSku({ - ...cloudContext, - id: item.cloudSkuId, - virtualNumberId: syncedFlow.binding.vnId, - phone: syncedFlow.binding.vnPhone, - }); - dispatchResults.push({ - cloudSkuId: item.cloudSkuId, - cloudSkuName: item.cloudSkuName, - unitIndex: index + 1, - quantity: item.quantity, - sendType: Number(dispatchResult.sendType || 0) || 0, - note: String(dispatchResult.note || "").trim(), - responseMessage: String(dispatchResult.responseMessage || "").trim(), - }); - } - } + const { dispatchResults, stockResult } = await withCloudSkuDispatchLocks( + resolveCloudSkuDispatchLockKeys(cloudContext.resolvedSourceKey, deliveryItems), + () => prepareStockAndDispatch({ + task, + flow: syncedFlow, + cloudContext, + deliveryItems, + }) + ); const firstDispatchResult: DispatchResultItem = dispatchResults[0] || { cloudSkuId: syncedFlow.binding.skuId, cloudSkuName: syncedFlow.binding.skuName, @@ -160,6 +174,17 @@ export async function dispatchKuaishouCloudFulfillmentTask( note: dispatchSummary, items: dispatchResults, }, + purchase: { + ...syncedFlow.purchase, + usedKnapsack: stockResult.usedKnapsack, + purchaseTriggered: stockResult.purchaseTriggered, + assetBefore: stockResult.assetBefore, + assetAfter: stockResult.assetAfter, + purchaseAt: stockResult.purchaseTriggered + ? now + : syncedFlow.purchase.purchaseAt, + items: stockResult.items, + }, }, }; @@ -193,6 +218,7 @@ export async function dispatchKuaishouCloudFulfillmentTask( note: dispatchSummary, deliveryItems, dispatchResults, + stockResult, actor, }, now @@ -222,6 +248,184 @@ export async function dispatchKuaishouCloudFulfillmentTask( }; } +async function prepareStockAndDispatch({ + task, + flow, + cloudContext, + deliveryItems, +}: { + task: TaskRow; + flow: JsonObject; + cloudContext: JsonObject; + deliveryItems: DispatchDeliveryItem[]; +}) { + const [knapsack, skuList] = await Promise.all([ + getCloudtentaclesKnapsack(cloudContext), + listCloudtentaclesSku(cloudContext), + ]); + const skuItems = Array.isArray(skuList.items) ? skuList.items : []; + const knapsackItems = Array.isArray(knapsack.items) ? knapsack.items : []; + const stockItems = buildDispatchStockItems(deliveryItems, { + skuItems, + knapsackItems, + }); + const missingItems = stockItems.filter((item) => item.purchasedCount > 0); + + let assetBefore = 0; + let assetAfter = 0; + let purchaseTriggered = false; + + if (missingItems.length > 0) { + if (flow.purchase?.autoBuyEnabled === false) { + throw createHttpError("背包中没有现成库存，且当前配置未开启自动购买", { + statusCode: 409, + errorCode: "kuaishou_cloud_auto_buy_disabled", + }); + } + + const missingSku = missingItems.find((item) => !findCloudSkuItem(skuItems, item.cloudSkuId)); + if (missingSku) { + throw createHttpError(`cloudtentacles 未找到 SKU ${missingSku.cloudSkuId}`, { + statusCode: 404, + errorCode: "kuaishou_cloud_sku_not_found", + }); + } + + const asset = await getCloudtentaclesAsset(cloudContext); + assetBefore = Number(asset.asset || 0) || 0; + const targetPrice = missingItems.reduce((sum, item) => { + const skuItem = findCloudSkuItem(skuItems, item.cloudSkuId); + return sum + item.purchasedCount * Number(skuItem?.price || 0); + }, 0); + const requiredAsset = targetPrice + Number(flow.purchase?.minAssetReserve || 0); + + if (assetBefore < requiredAsset) { + await notifyKuaishouCloudAssetNotEnough({ + task, + flow, + assetBefore, + requiredAsset, + skuName: missingItems + .map((item) => item.cloudSkuName || `SKU ${item.cloudSkuId}`) + .join("、"), + }); + throw createHttpError(`余额不足，当前 ${assetBefore}，至少需要 ${requiredAsset}`, { + statusCode: 409, + errorCode: "kuaishou_cloud_asset_not_enough", + }); + } + + for (const item of missingItems) { + await buyCloudtentaclesSku({ + ...cloudContext, + id: item.cloudSkuId, + count: item.purchasedCount, + }); + } + + purchaseTriggered = true; + const assetResult = await getCloudtentaclesAsset(cloudContext); + assetAfter = Number(assetResult.asset || 0) || 0; + } + + const dispatchResults: DispatchResultItem[] = []; + for (const item of deliveryItems) { + for (let index = 0; index < item.quantity; index += 1) { + const dispatchResult = await useCloudtentaclesSku({ + ...cloudContext, + id: item.cloudSkuId, + virtualNumberId: flow.binding.vnId, + phone: flow.binding.vnPhone, + }); + dispatchResults.push({ + cloudSkuId: item.cloudSkuId, + cloudSkuName: item.cloudSkuName, + unitIndex: index + 1, + quantity: item.quantity, + sendType: Number(dispatchResult.sendType || 0) || 0, + note: String(dispatchResult.note || "").trim(), + responseMessage: String(dispatchResult.responseMessage || "").trim(), + }); + } + } + + return { + dispatchResults, + stockResult: { + usedKnapsack: stockItems.every((item) => item.purchasedCount <= 0), + purchaseTriggered, + assetBefore, + assetAfter, + items: stockItems, + }, + }; +} + +export function buildDispatchStockItems( + deliveryItems: DispatchDeliveryItem[], + { skuItems = [], knapsackItems = [] }: { skuItems?: JsonObject[]; knapsackItems?: JsonObject[] } = {} +): DispatchStockItem[] { + return deliveryItems.map((item) => { + const knapsackItem = findCloudSkuItem(knapsackItems, item.cloudSkuId); + const skuItem = findCloudSkuItem(skuItems, item.cloudSkuId); + const knapsackCount = Math.max(0, Number(knapsackItem?.count || 0) || 0); + + return { + ...item, + cloudSkuName: item.cloudSkuName || String(skuItem?.name || knapsackItem?.name || "").trim(), + requiredCount: item.quantity, + knapsackCount, + purchasedCount: Math.max(0, item.quantity - knapsackCount), + }; + }); +} + +function findCloudSkuItem(items: JsonObject[], cloudSkuId: number) { + return items.find((item) => Number(item.id || 0) === cloudSkuId) || null; +} + +function resolveCloudSkuDispatchLockKeys(sourceKey: unknown, deliveryItems: DispatchDeliveryItem[]) { + const normalizedSourceKey = String(sourceKey || "default").trim() || "default"; + return deliveryItems.map((item) => `${normalizedSourceKey}:${item.cloudSkuId}`); +} + +async function withCloudSkuDispatchLocks(keys: string[], callback: () => Promise): Promise { + const normalizedKeys = [...new Set(keys.map((key) => String(key || "").trim()).filter(Boolean))].sort(); + + async function run(index: number): Promise { + const key = normalizedKeys[index]; + if (!key) { + return callback(); + } + + return withCloudSkuDispatchLock(key, () => run(index + 1)); + } + + return run(0); +} + +async function withCloudSkuDispatchLock(key: string, callback: () => Promise): Promise { + const previous = cloudSkuDispatchLocks.get(key) || Promise.resolve(); + let release: () => void = () => {}; + const current = new Promise((resolve) => { + release = resolve; + }); + const queued = previous.catch(() => {}).then(() => current); + + cloudSkuDispatchLocks.set(key, queued); + + await previous.catch(() => {}); + + try { + return await callback(); + } finally { + release(); + if (cloudSkuDispatchLocks.get(key) === queued) { + cloudSkuDispatchLocks.delete(key); + } + } +} + function resolveDispatchDeliveryItems(flow: JsonObject): DispatchDeliveryItem[] { const rawItems = Array.isArray(flow.deliveryItems) ? flow.deliveryItems : []; const items = rawItems diff --git a/apps/backend/src/services/platforms/cloudtentacles/http-client.ts b/apps/backend/src/services/platforms/cloudtentacles/http-client.ts index 0f903759..c90fad65 100644 --- a/apps/backend/src/services/platforms/cloudtentacles/http-client.ts +++ b/apps/backend/src/services/platforms/cloudtentacles/http-client.ts @@ -57,7 +57,7 @@ export async function cloudtentaclesRequest(pathname: unknown, options: JsonObje const statusCode = Number(options.businessErrorStatusCode || 400) const errorCode = String(options.businessErrorCode || 'cloudtentacles_business_error') - if (statusCode === 401 || isCloudtentaclesExpiredMessage(message)) { + if (isCloudtentaclesExpiredMessage(message)) { await notifyCloudtentaclesAuthExpired({ pathname: normalizedPathname, errorCode,