diff --git a/apps/backend/src/services/scheduler/config-service.test.ts b/apps/backend/src/services/scheduler/config-service.test.ts index e756522c..dacfc066 100644 --- a/apps/backend/src/services/scheduler/config-service.test.ts +++ b/apps/backend/src/services/scheduler/config-service.test.ts @@ -43,3 +43,72 @@ test('normalizeScheduledJobsConfig ensures default jobs exist', () => { assert.ok(ids.includes('pdd_order_sync')) assert.ok(ids.includes('cost-finance-sync')) }) + +test('normalizeScheduledJobsConfig 收录 v2 六个管道作业并保持各自配置', () => { + const config = normalizeScheduledJobsConfig({ + enabled: true, + jobs: [ + { + id: 'channel-assign', + type: 'channel_assign', + enabled: true, + intervalSeconds: 60, + config: { batchSize: 800, maxBatches: 5 }, + }, + { + id: 'carrier-dispatch', + type: 'carrier_dispatch', + enabled: false, + intervalSeconds: 60, + config: { limit: 50 }, + }, + { + id: 'product-normalize', + type: 'product_normalize', + enabled: true, + intervalSeconds: 300, + config: {}, + }, + { + id: 'cost-snapshot', + type: 'cost_snapshot', + enabled: true, + intervalSeconds: 300, + config: {}, + }, + { + id: 'sales-facts', + type: 'sales_facts', + enabled: true, + intervalSeconds: 600, + config: { days: 7 }, + }, + { + id: 'ledger-post', + type: 'ledger_post', + enabled: true, + intervalSeconds: 600, + config: { days: 7 }, + }, + ], + }) + + const byType = new Map(config.jobs.map((item) => [item.type, item])) + assert.equal(byType.get('channel_assign')?.config.batchSize, 800) + assert.equal(byType.get('carrier_dispatch')?.enabled, false) + assert.equal(byType.get('sales_facts')?.config.days, 7) + assert.equal(byType.get('ledger_post')?.config.days, 7) + assert.ok(byType.has('product_normalize')) + assert.ok(byType.has('cost_snapshot')) +}) + +test('normalizeScheduledJobsConfig 丢弃未知任务类型', () => { + const config = normalizeScheduledJobsConfig({ + enabled: true, + jobs: [{ id: 'nope', type: 'unknown_type', enabled: true, intervalSeconds: 60, config: {} }], + }) + assert.equal( + config.jobs.some((item) => item.type === 'unknown_type'), + false, + ) +}) diff --git a/apps/backend/src/services/scheduler/config-service.ts b/apps/backend/src/services/scheduler/config-service.ts index f03fd477..4a4afb5f 100644 --- a/apps/backend/src/services/scheduler/config-service.ts +++ b/apps/backend/src/services/scheduler/config-service.ts @@ -12,6 +12,14 @@ const DATA_RETENTION_JOB_ID = 'data-retention' const PDD_ORDER_SYNC_JOB_ID = 'pdd_order_sync' const COST_FINANCE_SYNC_JOB_ID = 'cost-finance-sync' +/** v2 管道作业:各自只负责一层,可单独启停。 */ +const CHANNEL_ASSIGN_JOB_ID = 'channel-assign' +const CARRIER_DISPATCH_JOB_ID = 'carrier-dispatch' +const PRODUCT_NORMALIZE_JOB_ID = 'product-normalize' +const COST_SNAPSHOT_JOB_ID = 'cost-snapshot' +const SALES_FACTS_JOB_ID = 'sales-facts' +const LEDGER_POST_JOB_ID = 'ledger-post' + export function getScheduledJobsFilePath() { return SCHEDULED_JOBS_FILE_PATH } @@ -125,6 +133,24 @@ function normalizeScheduledJob(rawValue: unknown) { if (type === 'cost_finance_sync') { return normalizeCostFinanceSyncJob(rawValue) } + if (type === 'channel_assign') { + return normalizeChannelAssignJob(rawValue) + } + if (type === 'carrier_dispatch') { + return normalizeCarrierDispatchJob(rawValue) + } + if (type === 'product_normalize') { + return normalizeProductNormalizeJob(rawValue) + } + if (type === 'cost_snapshot') { + return normalizeCostSnapshotJob(rawValue) + } + if (type === 'sales_facts') { + return normalizeSalesFactsJob(rawValue) + } + if (type === 'ledger_post') { + return normalizeLedgerPostJob(rawValue) + } return null } @@ -204,10 +230,157 @@ function createDefaultScheduledJobsConfig() { createDefaultDataRetentionJob(), createDefaultPddOrderSyncJob(), createDefaultCostFinanceSyncJob(), + createDefaultChannelAssignJob(), + createDefaultCarrierDispatchJob(), + createDefaultProductNormalizeJob(), + createDefaultCostSnapshotJob(), + createDefaultSalesFactsJob(), + createDefaultLedgerPostJob(), ], } } +function normalizeChannelAssignJob(rawValue: JsonObject) { + const config = isPlainObject(rawValue.config) ? rawValue.config : {} + return { + id: CHANNEL_ASSIGN_JOB_ID, + type: 'channel_assign', + enabled: rawValue.enabled !== false, + intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 60, 30, 3600), + config: { + batchSize: normalizeRangeInteger(config.batchSize, 500, 50, 5000), + maxBatches: normalizeRangeInteger(config.maxBatches, 20, 1, 200), + }, + } +} + +function normalizeCarrierDispatchJob(rawValue: JsonObject) { + const config = isPlainObject(rawValue.config) ? rawValue.config : {} + return { + id: CARRIER_DISPATCH_JOB_ID, + type: 'carrier_dispatch', + enabled: rawValue.enabled !== false, + intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 60, 30, 3600), + config: { + limit: normalizeRangeInteger(config.limit, 200, 10, 2000), + }, + } +} + +function normalizeProductNormalizeJob(rawValue: JsonObject) { + const config = isPlainObject(rawValue.config) ? rawValue.config : {} + return { + id: PRODUCT_NORMALIZE_JOB_ID, + type: 'product_normalize', + enabled: rawValue.enabled !== false, + intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 3600), + config: { + batchSize: normalizeRangeInteger(config.batchSize, 500, 50, 5000), + maxBatches: normalizeRangeInteger(config.maxBatches, 20, 1, 200), + }, + } +} + +function normalizeCostSnapshotJob(rawValue: JsonObject) { + const config = isPlainObject(rawValue.config) ? rawValue.config : {} + return { + id: COST_SNAPSHOT_JOB_ID, + type: 'cost_snapshot', + enabled: rawValue.enabled !== false, + intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 3600), + config: { + batchSize: normalizeRangeInteger(config.batchSize, 500, 50, 5000), + maxBatches: normalizeRangeInteger(config.maxBatches, 20, 1, 200), + }, + } +} + +function normalizeSalesFactsJob(rawValue: JsonObject) { + const config = isPlainObject(rawValue.config) ? rawValue.config : {} + return { + id: SALES_FACTS_JOB_ID, + type: 'sales_facts', + enabled: rawValue.enabled !== false, + intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 600, 60, 3600), + config: { + days: normalizeRangeInteger(config.days, 3, 1, 60), + }, + } +} + +function normalizeLedgerPostJob(rawValue: JsonObject) { + const config = isPlainObject(rawValue.config) ? rawValue.config : {} + return { + id: LEDGER_POST_JOB_ID, + type: 'ledger_post', + enabled: rawValue.enabled !== false, + intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 600, 60, 3600), + config: { + days: normalizeRangeInteger(config.days, 3, 1, 60), + }, + } +} + +function createDefaultChannelAssignJob() { + return { + id: CHANNEL_ASSIGN_JOB_ID, + type: 'channel_assign', + enabled: true, + intervalSeconds: 60, + config: { batchSize: 500, maxBatches: 20 }, + } +} + +function createDefaultCarrierDispatchJob() { + return { + id: CARRIER_DISPATCH_JOB_ID, + type: 'carrier_dispatch', + enabled: true, + intervalSeconds: 60, + config: { limit: 200 }, + } +} + +function createDefaultProductNormalizeJob() { + return { + id: PRODUCT_NORMALIZE_JOB_ID, + type: 'product_normalize', + enabled: true, + intervalSeconds: 300, + config: { batchSize: 500, maxBatches: 20 }, + } +} + +function createDefaultCostSnapshotJob() { + return { + id: COST_SNAPSHOT_JOB_ID, + type: 'cost_snapshot', + enabled: true, + intervalSeconds: 300, + config: { batchSize: 500, maxBatches: 20 }, + } +} + +function createDefaultSalesFactsJob() { + return { + id: SALES_FACTS_JOB_ID, + type: 'sales_facts', + enabled: true, + intervalSeconds: 600, + config: { days: 3 }, + } +} + +function createDefaultLedgerPostJob() { + return { + id: LEDGER_POST_JOB_ID, + type: 'ledger_post', + enabled: true, + intervalSeconds: 600, + config: { days: 3 }, + } +} + function createDefaultPddOrderSyncJob() { return { id: PDD_ORDER_SYNC_JOB_ID, diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts new file mode 100644 index 00000000..9c1f4692 --- /dev/null +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -0,0 +1,181 @@ +import type { JsonObject } from '../../types/json.js' +import { query } from '../../db/client.js' +import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js' +import { normalizeOrderItemProducts } from '../cost-center/product-normalize-service.js' +import { syncChannelCostSnapshots } from '../cost-center/cost-snapshot-service.js' +import { upsertAggregatedException } from '../cost-center/cost-exception-service.js' +import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js' +import { syncFinanceLedgerFromFacts } from '../finance/finance-ledger-service.js' +import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js' + +/** + * 财务与履约管道作业。 + * + * 每个作业只做一件事、可单独重跑,彼此不共享事务: + * channel_assign → carrier_dispatch → product_normalize + * → cost_snapshot → sales_facts → ledger_post + * + * 与旧 `cost_finance_sync` 的区别:旧作业把六件事串在一个函数里, + * 任一步失败就半途,而且会顺手写履约字段;这里每个作业只碰自己那一列。 + */ + +function jobConfig(job: JsonObject): JsonObject { + return job.config && typeof job.config === 'object' ? (job.config as JsonObject) : {} +} + +function clampInt(value: unknown, fallback: number, min: number, max: number): number { + const parsed = Number(value) + if (!Number.isInteger(parsed)) return fallback + return Math.min(max, Math.max(min, parsed)) +} + +export async function runChannelAssignJob(job: JsonObject = {}) { + const config = jobConfig(job) + const result = await assignPendingOrderItemChannels({ + batchSize: clampInt(config.batchSize, 500, 1, 5000), + maxBatches: clampInt(config.maxBatches, 20, 1, 200), + afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER), + }) + + return { + status: 'ok', + message: + `渠道判定:扫描 ${result.scanned},已定 ${result.assigned},` + + `冲突 ${result.conflicted},仍待定 ${result.stillPending}`, + ...result, + } +} + +export async function runCarrierDispatchJob(job: JsonObject = {}) { + const config = jobConfig(job) + const limit = clampInt(config.limit, 200, 1, 2000) + + const manual = await query<{ id: number; sku_key: string; config_id: number }>( + ` + SELECT oi.id, oi.manual_sku_key AS sku_key, cfg.id AS config_id + FROM order_items oi + JOIN manual_fulfillment_configs cfg ON cfg.sku_key = oi.manual_sku_key AND cfg.enabled + WHERE oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel = 'manual' + AND oi.fulfillment_config_id IS NULL + ORDER BY oi.id + LIMIT $1 + `, + [limit], + ) + + let attached = 0 + for (const row of manual.rows) { + const updated = await query( + ` + UPDATE order_items + SET fulfillment_config_id = $2, updated_at = NOW() + WHERE id = $1 AND fulfillment_config_id IS NULL + `, + [row.id, row.config_id], + ) + attached += updated.rowCount || 0 + } + + // 人工渠道但没有配置:进「人工 SKU 未配置」队列。 + const missingManual = await query<{ id: number; sku_key: string }>( + ` + SELECT oi.id, oi.manual_sku_key AS sku_key + FROM order_items oi + WHERE oi.fulfillment_status = 'assigned' + AND oi.fulfillment_channel = 'manual' + AND oi.fulfillment_config_id IS NULL + AND NOT EXISTS ( + SELECT 1 FROM manual_fulfillment_configs cfg + WHERE cfg.sku_key = oi.manual_sku_key AND cfg.enabled + ) + ORDER BY oi.id + LIMIT $1 + `, + [limit], + ) + + for (const row of missingManual.rows) { + await upsertAggregatedException({ + reason: 'manual_config_missing', + scopeKey: `manual_config_missing|-|-|${row.sku_key}|manual`, + orderItemId: row.id, + detail: { skuKey: row.sku_key }, + }) + } + + return { + status: 'ok', + message: `人工载体挂载 ${attached},缺配置 ${missingManual.rows.length}`, + attached, + missingManual: missingManual.rows.length, + } +} + +export async function runProductNormalizeJob(job: JsonObject = {}) { + const config = jobConfig(job) + const result = await normalizeOrderItemProducts({ + batchSize: clampInt(config.batchSize, 500, 1, 5000), + maxBatches: clampInt(config.maxBatches, 20, 1, 200), + }) + + return { + status: 'ok', + message: `商品归一:扫描 ${result.scanned},命中 ${result.matched},未命中 ${result.unmatched}`, + ...result, + } +} + +export async function runCostSnapshotJob(job: JsonObject = {}) { + const config = jobConfig(job) + const result = await syncChannelCostSnapshots({ + batchSize: clampInt(config.batchSize, 500, 1, 5000), + maxBatches: clampInt(config.maxBatches, 20, 1, 200), + }) + + return { + status: 'ok', + message: `成本快照:扫描 ${result.scanned},生成 ${result.created},缺失 ${result.missing}`, + ...result, + } +} + +export async function runSalesFactsJob(job: JsonObject = {}) { + const config = jobConfig(job) + const days = clampInt(config.days, 3, 1, 60) + const dateTo = formatShanghaiDate(new Date()) + const dateFrom = shiftDate(dateTo, -(days - 1)) + const result = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo }) + + return { + status: 'ok', + message: `销售事实:${result.dateFrom}~${result.dateTo} 写入 ${result.inserted} 行`, + ...result, + } +} + +export async function runLedgerPostJob(job: JsonObject = {}) { + const config = jobConfig(job) + const days = clampInt(config.days, 3, 1, 60) + const dateTo = formatShanghaiDate(new Date()) + const dateFrom = shiftDate(dateTo, -(days - 1)) + const result = await syncFinanceLedgerFromFacts({ dateFrom, dateTo }) + + return { + status: 'ok', + message: `总账:${dateFrom}~${dateTo} 新增 ${result.created} 条,跳过已关账 ${result.skipped}`, + ...result, + } +} + +/** 顺序执行全部管道,供管理端「一键重算」使用。 */ +export async function runFinancePipelineOnce(input: { days?: number } = {}) { + const channel = await runChannelAssignJob({ config: { batchSize: 1000, maxBatches: 50 } }) + const carrier = await runCarrierDispatchJob({ config: { limit: 500 } }) + const product = await runProductNormalizeJob({ config: { batchSize: 1000, maxBatches: 50 } }) + const cost = await runCostSnapshotJob({ config: { batchSize: 1000, maxBatches: 50 } }) + const facts = await runSalesFactsJob({ config: { days: input.days ?? 3 } }) + const ledger = await runLedgerPostJob({ config: { days: input.days ?? 3 } }) + + return { channel, carrier, product, cost, facts, ledger } +} diff --git a/apps/backend/src/services/scheduler/scheduler-service.ts b/apps/backend/src/services/scheduler/scheduler-service.ts index ceddd183..ccafbc0f 100644 --- a/apps/backend/src/services/scheduler/scheduler-service.ts +++ b/apps/backend/src/services/scheduler/scheduler-service.ts @@ -7,6 +7,14 @@ import { runWorkOrderTimeoutJob } from './work-order-timeout-job.js' import { runDataRetentionJob } from './data-retention-job.js' import { runPddOrderSyncJob } from './pdd-order-sync-job.js' import { runCostFinanceSyncJob } from './cost-finance-sync-job.js' +import { + runCarrierDispatchJob, + runChannelAssignJob, + runCostSnapshotJob, + runLedgerPostJob, + runProductNormalizeJob, + runSalesFactsJob, +} from './finance-pipeline-jobs.js' const timers = new Map() const jobStates = new Map() @@ -208,6 +216,24 @@ function dispatchJob(job: JsonObject) { if (job.type === 'cost_finance_sync') { return runCostFinanceSyncJob(job) } + if (job.type === 'channel_assign') { + return runChannelAssignJob(job) + } + if (job.type === 'carrier_dispatch') { + return runCarrierDispatchJob(job) + } + if (job.type === 'product_normalize') { + return runProductNormalizeJob(job) + } + if (job.type === 'cost_snapshot') { + return runCostSnapshotJob(job) + } + if (job.type === 'sales_facts') { + return runSalesFactsJob(job) + } + if (job.type === 'ledger_post') { + return runLedgerPostJob(job) + } throw createHttpError(`不支持的定时任务类型:${job.type}`, { statusCode: 400, diff --git a/apps/backend/src/services/worker-platform/admin-worker-catalog-service.ts b/apps/backend/src/services/worker-platform/admin-worker-catalog-service.ts index 9e09b548..e019df3f 100644 --- a/apps/backend/src/services/worker-platform/admin-worker-catalog-service.ts +++ b/apps/backend/src/services/worker-platform/admin-worker-catalog-service.ts @@ -27,6 +27,7 @@ import { upsertWorkProductRule, } from '../../repositories/worker-platform/work-product-rule-repo.js' import type { WorkOrderRow, WorkProductRuleRow } from '../../repositories/worker-platform/types.js' +import { syncAdminWorkOrderTemplateCostVersions } from '../cost-center/work-order-product-service.js' import { normalizeWorkOrderAcceptanceMode, normalizeWorkOrderGiftCooldownHours, @@ -531,6 +532,17 @@ export async function saveAdminWorkProductRule(payload: JsonObject = {}) { sortOrder: normalizeInteger(payload.sortOrder, 100), now: nowIso(), }) + + // 模板价格变更即同步成本版本(事件驱动);失败不阻塞模板保存, + // 夜间巡检与后台「同步发单金额」按钮仍可修复。 + if (rule) { + try { + await syncAdminWorkOrderTemplateCostVersions(rule.id) + } catch { + // 成本版本同步失败不允许影响模板本身;由巡检补齐。 + } + } + return { rule: mapWorkProductRule(rule) } }