diff --git a/apps/backend/src/routes/admin/fulfillment-channel.ts b/apps/backend/src/routes/admin/fulfillment-channel.ts index cb4b302e..2a7bf804 100644 --- a/apps/backend/src/routes/admin/fulfillment-channel.ts +++ b/apps/backend/src/routes/admin/fulfillment-channel.ts @@ -17,7 +17,14 @@ import { upsertChannelRule, } from '../../services/fulfillment/channel-assign-service.js' import { listExceptionQueue } from '../../services/cost-center/cost-exception-service.js' -import { runFinancePipelineOnce } from '../../services/scheduler/finance-pipeline-jobs.js' +import { + runFinancePipelineOnce, + runSingleFinancePipeline, +} from '../../services/scheduler/finance-pipeline-jobs.js' +import { + getFinancePipelineStatus, + isFinancePipelineName, +} from '../../services/scheduler/finance-pipeline-status.js' import { bodyOf } from './request-helpers.js' import { createJsonHandler, requireAdminPermission, requireAdminRoles } from './session.js' @@ -185,6 +192,16 @@ router.delete( }), ) +router.get( + '/fulfillment/pipeline/status', + requireAdminPermission('finance.view'), + createJsonHandler(() => getFinancePipelineStatus(), { + successMessage: 'ok', + errorMessage: '读取管道状态失败', + scope: '[admin/fulfillment/pipeline/status]', + }), +) + router.post( '/fulfillment/pipeline/run', requireAdminPermission('finance.config'), @@ -192,6 +209,15 @@ router.post( async (req) => { const body = bodyOf(req) const days = Number(body.days) + // 指定 pipeline 时只跑单步，便于排障与补跑。 + if (body.pipeline) { + if (!isFinancePipelineName(body.pipeline)) { + throw new Error(`未知管道：${String(body.pipeline)}`) + } + return runSingleFinancePipeline(body.pipeline, { + config: Number.isFinite(days) ? { days } : {}, + }) + } return runFinancePipelineOnce({ days: Number.isFinite(days) ? days : 3 }) }, { diff --git a/apps/backend/src/services/admin/platform-config/notification-service.ts b/apps/backend/src/services/admin/platform-config/notification-service.ts index fe515c3c..f74232ec 100644 --- a/apps/backend/src/services/admin/platform-config/notification-service.ts +++ b/apps/backend/src/services/admin/platform-config/notification-service.ts @@ -127,7 +127,10 @@ function mapAdminScheduledJobsConfig(config: JsonObject = {}) { enabled: item.enabled === true, intervalSeconds: Number(item.intervalSeconds || 300), cooldownSeconds: Number(item.cooldownSeconds || 1800), + // 保留各任务自己的配置键（管道作业的 batchSize / maxBatches / days 等）， + // 避免管理接口读一次就把没列出来的字段丢掉。 config: { + ...(item.config && typeof item.config === 'object' ? item.config : {}), assetThreshold: normalizeNonNegativeNumber(item.config?.assetThreshold, 500), accounts: Array.isArray(item.config?.accounts) ? item.config.accounts : [], pageSize: Number(item.config?.pageSize || 0) || undefined, diff --git a/apps/backend/src/services/scheduler/config-service.ts b/apps/backend/src/services/scheduler/config-service.ts index 4a4afb5f..cf5860ea 100644 --- a/apps/backend/src/services/scheduler/config-service.ts +++ b/apps/backend/src/services/scheduler/config-service.ts @@ -85,6 +85,12 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) { const hasDataRetention = jobs.some((item) => item.id === DATA_RETENTION_JOB_ID) const hasPddOrderSync = jobs.some((item) => item.id === PDD_ORDER_SYNC_JOB_ID) const hasCostFinanceSync = jobs.some((item) => item.id === COST_FINANCE_SYNC_JOB_ID) + const hasChannelAssign = jobs.some((item) => item.id === CHANNEL_ASSIGN_JOB_ID) + const hasCarrierDispatch = jobs.some((item) => item.id === CARRIER_DISPATCH_JOB_ID) + const hasProductNormalize = jobs.some((item) => item.id === PRODUCT_NORMALIZE_JOB_ID) + const hasCostSnapshot = jobs.some((item) => item.id === COST_SNAPSHOT_JOB_ID) + const hasSalesFacts = jobs.some((item) => item.id === SALES_FACTS_JOB_ID) + const hasLedgerPost = jobs.some((item) => item.id === LEDGER_POST_JOB_ID) if (!hasWorkOrderTimeout) { jobs.push(createDefaultWorkOrderTimeoutJob()) @@ -101,6 +107,13 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) { if (!hasCostFinanceSync) { jobs.push(createDefaultCostFinanceSyncJob()) } + // v2 管道：老配置里没有这些任务时自动补齐，保证升级后无需手工添加。 + if (!hasChannelAssign) jobs.push(createDefaultChannelAssignJob()) + if (!hasCarrierDispatch) jobs.push(createDefaultCarrierDispatchJob()) + if (!hasProductNormalize) jobs.push(createDefaultProductNormalizeJob()) + if (!hasCostSnapshot) jobs.push(createDefaultCostSnapshotJob()) + if (!hasSalesFacts) jobs.push(createDefaultSalesFactsJob()) + if (!hasLedgerPost) jobs.push(createDefaultLedgerPostJob()) return { enabled: typeof source.enabled === 'boolean' ? source.enabled : true, diff --git a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts index 9c1f4692..638d25e1 100644 --- a/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts +++ b/apps/backend/src/services/scheduler/finance-pipeline-jobs.ts @@ -7,6 +7,11 @@ import { upsertAggregatedException } from '../cost-center/cost-exception-service 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' +import { + recordPipelineError, + touchWatermark, + type FinancePipelineName, +} from './finance-pipeline-status.js' /** * 财务与履约管道作业。 @@ -37,6 +42,7 @@ export async function runChannelAssignJob(job: JsonObject = {}) { afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER), }) + await touchWatermark('channel_assign', { lastItemId: result.lastId }) return { status: 'ok', message: @@ -104,6 +110,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) { }) } + await touchWatermark('carrier_dispatch') return { status: 'ok', message: `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}`, @@ -119,6 +126,7 @@ export async function runProductNormalizeJob(job: JsonObject = {}) { maxBatches: clampInt(config.maxBatches, 20, 1, 200), }) + await touchWatermark('product_normalize', { lastItemId: result.lastId }) return { status: 'ok', message: `商品归一：扫描 ${result.scanned}，命中 ${result.matched}，未命中 ${result.unmatched}`, @@ -133,6 +141,7 @@ export async function runCostSnapshotJob(job: JsonObject = {}) { maxBatches: clampInt(config.maxBatches, 20, 1, 200), }) + await touchWatermark('cost_snapshot', { lastItemId: result.lastId }) return { status: 'ok', message: `成本快照：扫描 ${result.scanned}，生成 ${result.created}，缺失 ${result.missing}`, @@ -147,6 +156,7 @@ export async function runSalesFactsJob(job: JsonObject = {}) { const dateFrom = shiftDate(dateTo, -(days - 1)) const result = await rebuildChannelSalesDailyFacts({ dateFrom, dateTo }) + await touchWatermark('sales_facts', { lastStatDate: result.dateTo }) return { status: 'ok', message: `销售事实：${result.dateFrom}~${result.dateTo} 写入 ${result.inserted} 行`, @@ -161,6 +171,7 @@ export async function runLedgerPostJob(job: JsonObject = {}) { const dateFrom = shiftDate(dateTo, -(days - 1)) const result = await syncFinanceLedgerFromFacts({ dateFrom, dateTo }) + await touchWatermark('ledger_post', { lastStatDate: dateTo }) return { status: 'ok', message: `总账：${dateFrom}~${dateTo} 新增 ${result.created} 条，跳过已关账 ${result.skipped}`, @@ -179,3 +190,28 @@ export async function runFinancePipelineOnce(input: { days?: number } = {}) { return { channel, carrier, product, cost, facts, ledger } } + +const PIPELINE_RUNNERS: Record< + FinancePipelineName, + (job?: JsonObject) => Promise> +> = { + channel_assign: runChannelAssignJob, + carrier_dispatch: runCarrierDispatchJob, + product_normalize: runProductNormalizeJob, + cost_snapshot: runCostSnapshotJob, + sales_facts: runSalesFactsJob, + ledger_post: runLedgerPostJob, +} + +/** 单独执行一个管道（排障 / 补跑），失败时把原因写进水位表。 */ +export async function runSingleFinancePipeline( + pipeline: FinancePipelineName, + job: JsonObject = {}, +): Promise> { + try { + return await PIPELINE_RUNNERS[pipeline](job) + } catch (error) { + await recordPipelineError(pipeline, error) + throw error + } +} diff --git a/apps/backend/src/services/scheduler/finance-pipeline-status.ts b/apps/backend/src/services/scheduler/finance-pipeline-status.ts new file mode 100644 index 00000000..6b2ef15b --- /dev/null +++ b/apps/backend/src/services/scheduler/finance-pipeline-status.ts @@ -0,0 +1,130 @@ +import { query } from '../../db/client.js' +import { nowIso } from '../../utils/time.js' +import { getScheduledJobRuntimeStates } from './scheduler-service.js' + +/** + * 财务管道可观测性：水位 + 定时任务运行状态。 + * + * 后台「履约渠道 → 管道状态」用的就是这里的数据，用于判断： + * · 管道有没有在跑（last_success_at） + * · 跑到哪了（last_item_id / last_stat_date） + * · 最近为什么失败（last_error） + */ + +export const FINANCE_PIPELINES = [ + { name: 'channel_assign', label: '渠道判定', jobId: 'channel-assign' }, + { name: 'carrier_dispatch', label: '履约载体', jobId: 'carrier-dispatch' }, + { name: 'product_normalize', label: '商品归一', jobId: 'product-normalize' }, + { name: 'cost_snapshot', label: '成本快照', jobId: 'cost-snapshot' }, + { name: 'sales_facts', label: '销售事实', jobId: 'sales-facts' }, + { name: 'ledger_post', label: '总账过账', jobId: 'ledger-post' }, +] as const + +export type FinancePipelineName = (typeof FINANCE_PIPELINES)[number]['name'] + +export function isFinancePipelineName(value: unknown): value is FinancePipelineName { + return FINANCE_PIPELINES.some((item) => item.name === value) +} + +/** 记录一次成功运行的水位。 */ +export async function touchWatermark( + pipeline: FinancePipelineName, + patch: { lastItemId?: number; lastStatDate?: string | null } = {}, +): Promise { + const now = nowIso() + await query( + ` + INSERT INTO finance_watermarks (pipeline, last_item_id, last_stat_date, last_success_at, updated_at) + VALUES ($1, $2, $3::date, $4, $4) + ON CONFLICT (pipeline) DO UPDATE SET + last_item_id = GREATEST(finance_watermarks.last_item_id, EXCLUDED.last_item_id), + last_stat_date = COALESCE(EXCLUDED.last_stat_date, finance_watermarks.last_stat_date), + last_success_at = EXCLUDED.last_success_at, + last_error = '', + updated_at = EXCLUDED.updated_at + `, + [pipeline, Math.max(0, Math.trunc(patch.lastItemId || 0)), patch.lastStatDate ?? null, now], + ) +} + +/** 记录一次失败原因，不改水位。 */ +export async function recordPipelineError( + pipeline: FinancePipelineName, + error: unknown, +): Promise { + const message = error instanceof Error ? error.message : String(error) + await query( + ` + INSERT INTO finance_watermarks (pipeline, last_error, updated_at) + VALUES ($1, $2, $3) + ON CONFLICT (pipeline) DO UPDATE SET + last_error = EXCLUDED.last_error, + updated_at = EXCLUDED.updated_at + `, + [pipeline, message.slice(0, 500), nowIso()], + ) +} + +export type FinancePipelineStatusRow = { + pipeline: string + label: string + lastItemId: number + lastStatDate: string | null + lastSuccessAt: string | null + lastError: string + jobId: string + scheduler: { + running: boolean + enabled: boolean + lastRunAt: string + lastFinishedAt: string + nextRunAt: string + lastStatus: string + lastMessage: string + } | null +} + +export async function getFinancePipelineStatus(): Promise { + const marks = await query<{ + pipeline: string + last_item_id: string + last_stat_date: string | null + last_success_at: string | null + last_error: string + }>( + ` + SELECT pipeline, last_item_id::text, last_stat_date::text, last_success_at, last_error + FROM finance_watermarks + `, + ) + + const byPipeline = new Map(marks.rows.map((row) => [row.pipeline, row])) + const runtime = new Map( + getScheduledJobRuntimeStates().map((state) => [String(state.id || ''), state]), + ) + + return FINANCE_PIPELINES.map((item) => { + const mark = byPipeline.get(item.name) + const state = runtime.get(item.jobId) + return { + pipeline: item.name, + label: item.label, + lastItemId: Number(mark?.last_item_id || 0), + lastStatDate: mark?.last_stat_date || null, + lastSuccessAt: mark?.last_success_at || null, + lastError: String(mark?.last_error || ''), + jobId: item.jobId, + scheduler: state + ? { + running: state.running === true, + enabled: state.enabled === true, + lastRunAt: String(state.lastRunAt || ''), + lastFinishedAt: String(state.lastFinishedAt || ''), + nextRunAt: String(state.nextRunAt || ''), + lastStatus: String(state.lastStatus || ''), + lastMessage: String(state.lastMessage || ''), + } + : null, + } + }) +} diff --git a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx index 4eb20081..d5e88180 100644 --- a/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx +++ b/apps/frontend/src/pages/admin/AdminFulfillmentChannelPage.tsx @@ -34,7 +34,9 @@ import { fetchChannelExceptions, fetchChannelRules, fetchPendingChannelGroups, + fetchPipelineStatus, runFinancePipeline, + type FinancePipelineStatus, saveChannelRule, type ChannelRule, type PendingChannelGroup, @@ -72,6 +74,7 @@ export default function AdminFulfillmentChannelPage() { { key: 'pending', label: '待分类队列', children: }, { key: 'rules', label: '渠道规则', children: }, { key: 'exceptions', label: '问题队列', children: }, + { key: 'pipelines', label: '管道状态', children: }, ]} /> @@ -406,6 +409,106 @@ function ExceptionsPanel() { ) } +function PipelinesPanel() { + const queryClient = useQueryClient() + const query = useQuery({ + queryKey: ['fulfillment-pipelines'], + queryFn: fetchPipelineStatus, + refetchInterval: 30_000, + }) + + const runMutation = useMutation({ + mutationFn: (pipeline: string) => runFinancePipeline({ pipeline }), + onSuccess: () => { + message.success('管道已执行') + void queryClient.invalidateQueries({ queryKey: ['fulfillment-pipelines'] }) + void queryClient.invalidateQueries({ queryKey: ['fulfillment-pending'] }) + }, + onError: (error) => message.error(error instanceof Error ? error.message : '执行管道失败'), + }) + + const columns = [ + { title: '管道', dataIndex: 'label', width: 130 }, + { + title: '定时任务', + width: 150, + render: (_: unknown, row: FinancePipelineStatus) => + row.scheduler ? ( + row.scheduler.enabled ? ( + 已启用 + ) : ( + 已停用 + ) + ) : ( + 未注册 + ), + }, + { + title: '上次成功', + dataIndex: 'lastSuccessAt', + width: 170, + render: (value: string | null) => (value ? formatDateTime(value) : '从未成功'), + }, + { + title: '水位', + width: 200, + render: (_: unknown, row: FinancePipelineStatus) => + row.lastStatDate + ? `日期 ${row.lastStatDate}` + : row.lastItemId > 0 + ? `订单项 ${row.lastItemId}` + : '—', + }, + { + title: '最近一次执行', + width: 260, + render: (_: unknown, row: FinancePipelineStatus) => row.scheduler?.lastMessage || '—', + }, + { + title: '错误', + render: (_: unknown, row: FinancePipelineStatus) => + row.lastError ? {row.lastError} : '—', + }, + { + title: '操作', + width: 110, + render: (_: unknown, row: FinancePipelineStatus) => ( + + ), + }, + ] + + return ( + } onClick={() => void query.refetch()}> + 刷新 + + } + > + + 六个管道由定时任务自动执行，无需人工干预；这里只用于查看进度、 + 排障和补跑。默认间隔：渠道判定/载体 60s，商品/成本 300s，事实/总账 600s。 + + + + ) +} + function rowKey(row: PendingChannelGroup) { return `${row.provider}|${row.shop_id}|${row.sku_code}` } diff --git a/apps/frontend/src/services/admin/fulfillment-channel.ts b/apps/frontend/src/services/admin/fulfillment-channel.ts index 6b4f0a08..81cbd752 100644 --- a/apps/frontend/src/services/admin/fulfillment-channel.ts +++ b/apps/frontend/src/services/admin/fulfillment-channel.ts @@ -51,6 +51,25 @@ export type ChannelExceptionListResult = { export type PipelineRunResult = Record +export type FinancePipelineStatus = { + pipeline: string + label: string + lastItemId: number + lastStatDate: string | null + lastSuccessAt: string | null + lastError: string + jobId: string + scheduler: { + running: boolean + enabled: boolean + lastRunAt: string + lastFinishedAt: string + nextRunAt: string + lastStatus: string + lastMessage: string + } | null +} + export function fetchPendingChannelGroups( params: { page?: number @@ -112,7 +131,11 @@ export function deleteChannelRule(ruleId: number) { return apiDelete<{ deleted: boolean }>(`/api/v1/admin/fulfillment/rules/${ruleId}`) } -export function runFinancePipeline(payload: { days?: number } = {}) { +export function fetchPipelineStatus() { + return apiGet('/api/v1/admin/fulfillment/pipeline/status') +} + +export function runFinancePipeline(payload: { days?: number; pipeline?: string } = {}) { return apiPost('/api/v1/admin/fulfillment/pipeline/run', payload, { timeout: 300_000, })