feat: sync performance conversion metrics daily

This commit is contained in:
江林铖 2026-09-18 11:31:40 +08:00
parent 88f6c10799
commit 3be02ff92e
14 changed files with 860 additions and 23 deletions

View File

@ -98,7 +98,8 @@ RUN test -s "/app/public/brand/header-logo.png" && \
test -f "/app/prisma/migrations/20260914_add_brand_knowledge_engagement/migration.sql" && \
test -f "/app/prisma/migrations/20260914_expand_brand_knowledge_download_count/migration.sql" && \
test -f "/app/prisma/migrations/20260914_expand_brand_knowledge_preview_slices/migration.sql" && \
test -f "/app/prisma/migrations/20260915_add_brand_knowledge_soft_delete/migration.sql"
test -f "/app/prisma/migrations/20260915_add_brand_knowledge_soft_delete/migration.sql" && \
test -f "/app/prisma/migrations/20260918_add_performance_conversion_sync/migration.sql"
# 这两个动态路由负责兜底访问本地上传文件;如果它们在构建上下文里丢失,
# 线上音频/视频就会直接 404。这里提前失败,避免再把缺路由的镜像发出去。
@ -230,6 +231,7 @@ COPY --from=builder --chown=root:root /app/prisma/migrations/20260914_add_brand_
COPY --from=builder --chown=root:root /app/prisma/migrations/20260914_expand_brand_knowledge_download_count ./prisma/migrations/20260914_expand_brand_knowledge_download_count
COPY --from=builder --chown=root:root /app/prisma/migrations/20260914_expand_brand_knowledge_preview_slices ./prisma/migrations/20260914_expand_brand_knowledge_preview_slices
COPY --from=builder --chown=root:root /app/prisma/migrations/20260915_add_brand_knowledge_soft_delete ./prisma/migrations/20260915_add_brand_knowledge_soft_delete
COPY --from=builder --chown=root:root /app/prisma/migrations/20260918_add_performance_conversion_sync ./prisma/migrations/20260918_add_performance_conversion_sync
COPY --from=builder --chown=nextjs:nodejs /app/lib ./lib
COPY --from=builder --chown=nextjs:nodejs /app/scripts/public-traffic-rollup-from-nginx.py ./scripts/public-traffic-rollup-from-nginx.py
COPY --from=builder --chown=nextjs:nodejs /app/scripts/collapse-performance-survey-template-versions.mjs ./scripts/collapse-performance-survey-template-versions.mjs

View File

@ -99,6 +99,28 @@ describe('PerformanceConversionDashboard filters', () => {
expect(await screen.findByText(label)).toBeInTheDocument()
})
it('shows the business data incremental synchronization state', async () => {
global.fetch = jest.fn().mockResolvedValue({
ok: true,
json: async () => ({
success: true,
data: {
...emptyList,
performanceSync: {
status: 'ready',
lastCompletedAt: '2099-09-17T01:00:00.000Z',
lastSourceDate: '2099-09-17',
},
},
}),
} as Response)
render(<PerformanceConversionDashboard />)
expect(await screen.findByText('经营数据每日增量同步')).toBeInTheDocument()
expect(screen.getByText(/已扫描至 2099\/09\/17/)).toBeInTheDocument()
})
it('opens each product ID on a separate detail page and preserves the list context', async () => {
window.history.replaceState(null, '', '/workspaces/business/performance-conversion?brand=%E4%BA%BF%E5%AE%B6%E8%BE%BE&q=%E4%B8%BB%E5%9B%BE&completedFrom=2026-07-01&completedTo=2026-07-31&page=3')
const fetchMock = jest.fn().mockResolvedValue({

View File

@ -16,6 +16,7 @@ import {
import DepartmentShell from './DepartmentShell'
import {
BUSINESS_TREND_BRANDS,
type PerformanceConversionSourceSync,
type PerformanceConversionWorkflowSync,
type PerformanceConversionWorkflowItem,
type PerformanceConversionWorkflowList,
@ -82,6 +83,41 @@ function WorkflowSyncStatus({ sync }: { sync: PerformanceConversionWorkflowSync
)
}
function PerformanceDataSyncStatus({ sync }: { sync?: PerformanceConversionSourceSync }) {
if (!sync) return null
const latestSuccess = sync.lastCompletedAt
const latestSuccessLabel = latestSuccess ? formatDate(latestSuccess, true) : null
const isDelayed = sync.status === 'ready'
&& latestSuccess !== null
&& Date.now() - new Date(latestSuccess).getTime() > 48 * 60 * 60 * 1000
const view = sync.status === 'syncing'
? { label: '正在同步经营数据', detail: latestSuccessLabel ? `上次成功 ${latestSuccessLabel}` : '首次同步进行中', tone: 'text-[#0050cb]', icon: RefreshCw, spin: true }
: sync.status === 'error'
? { label: '经营数据同步异常', detail: latestSuccessLabel ? `暂显示 ${latestSuccessLabel} 的数据` : '暂无可用同步记录', tone: 'text-amber-700', icon: AlertTriangle, spin: false }
: sync.status === 'pending'
? { label: '等待首次经营数据同步', detail: '后台任务尚未完成', tone: 'text-slate-500', icon: Clock3, spin: false }
: isDelayed
? { label: '经营数据每日同步延迟', detail: `最近成功 ${latestSuccessLabel}`, tone: 'text-amber-700', icon: AlertTriangle, spin: false }
: { label: '经营数据每日增量同步', detail: latestSuccessLabel ? `最近成功 ${latestSuccessLabel}` : '等待首次成功记录', tone: 'text-emerald-700', icon: CheckCircle2, spin: false }
const Icon = view.icon
return (
<div
role="status"
aria-live="polite"
className={`min-w-0 text-left text-xs leading-5 lg:text-right ${view.tone}`}
title={latestSuccess ? `最近同步成功:${latestSuccessLabel}(北京时间)` : undefined}
>
<p className="inline-flex items-center gap-1.5 font-bold">
<Icon className={`h-3.5 w-3.5 shrink-0 ${view.spin ? 'animate-spin' : ''}`} aria-hidden="true" />
<span>{view.label}</span>
</p>
<p className="text-slate-500">{view.detail}{sync.lastSourceDate ? ` · 已扫描至 ${formatDate(sync.lastSourceDate)}` : ''}</p>
</div>
)
}
function WorkflowTable({ items, detailHref }: {
items: PerformanceConversionWorkflowItem[]
detailHref: (item: PerformanceConversionWorkflowItem) => string
@ -217,7 +253,7 @@ export default function PerformanceConversionDashboard() {
<aside aria-label="品牌导航" className="border-b border-slate-200 bg-slate-50 lg:border-b-0 lg:border-r"><div className="border-b border-slate-200 px-4 py-4"><p className="text-xs font-bold text-slate-500">美术申请流程</p><p className="mt-1 text-sm font-bold text-slate-900">六品牌</p></div><nav className="flex gap-1 overflow-x-auto p-2 lg:block lg:space-y-1 lg:overflow-visible" aria-label="选择品牌">{BUSINESS_TREND_BRANDS.map((brand) => { const count = listState.data?.brands.find((item) => item.brand === brand)?.count; return <button key={brand} type="button" aria-current={selectedBrand === brand ? 'page' : undefined} onClick={() => { const nextFilters = { ...appliedFilters, query: '' }; setSelectedBrand(brand); setQuery(''); setCompletedFrom(nextFilters.completedFrom); setCompletedTo(nextFilters.completedTo); setAppliedFilters(nextFilters); setFilterError(null); setPage(1); void loadList(brand, 1, nextFilters) }} className={`flex min-h-11 shrink-0 items-center gap-2 rounded-lg px-3 text-left text-sm font-bold focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-[#1677ff] lg:w-full ${selectedBrand === brand ? 'bg-[#1677ff] text-white' : 'text-slate-600 hover:bg-slate-200/70 hover:text-slate-950'}`}><span className="min-w-0 flex-1 truncate">{brand}</span>{count !== undefined && <span className={`tabular-nums text-xs ${selectedBrand === brand ? 'text-white/80' : 'text-slate-400'}`}>{count}</span>}</button> })}</nav></aside>
<section className="min-w-0 bg-white" aria-labelledby="performance-brand-title" aria-busy={listState.loading}>
<header className="flex flex-col gap-4 border-b border-slate-200 px-5 py-5 sm:px-7 lg:flex-row lg:items-end lg:justify-between"><div><p className="text-xs font-bold text-[#9a4d08]">当前品牌</p><h2 id="performance-brand-title" className="mt-1 text-xl font-bold tracking-normal text-slate-950">{selectedBrand}</h2><p className="mt-1 text-xs text-slate-500">{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}</p></div>{listState.data && <WorkflowSyncStatus sync={listState.data.workflowSync} />}</header>
<header className="flex flex-col gap-4 border-b border-slate-200 px-5 py-5 sm:px-7 lg:flex-row lg:items-end lg:justify-between"><div><p className="text-xs font-bold text-[#9a4d08]">当前品牌</p><h2 id="performance-brand-title" className="mt-1 text-xl font-bold tracking-normal text-slate-950">{selectedBrand}</h2><p className="mt-1 text-xs text-slate-500">{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}</p></div>{listState.data && <div className="grid gap-2"><WorkflowSyncStatus sync={listState.data.workflowSync} /><PerformanceDataSyncStatus sync={listState.data.performanceSync} /></div>}</header>
<form onSubmit={submitSearch} className="grid gap-3 border-b border-slate-200 bg-slate-50 px-5 py-4 sm:px-7 md:grid-cols-[minmax(0,1fr)_auto] md:items-end xl:grid-cols-[minmax(280px,0.9fr)_minmax(220px,1fr)_auto]">
<fieldset className="min-w-0 md:col-span-2 xl:col-span-1">
<legend className="mb-1 text-xs font-bold text-slate-600">完结日期</legend>

View File

@ -8,6 +8,12 @@ export type PerformanceConversionWorkflowSync = {
lastFullReconciledAt: string | null
}
export type PerformanceConversionSourceSync = {
status: 'pending' | 'syncing' | 'ready' | 'error'
lastCompletedAt: string | null
lastSourceDate: string | null
}
export const PERFORMANCE_CONVERSION_METRICS = [
{ key: 'impressions', label: '展现量', group: '流量', format: 'integer' },
{ key: 'visitors', label: '商品访客数', group: '流量', format: 'integer' },
@ -45,6 +51,7 @@ export type PerformanceConversionWorkflowList = {
pageSize: number
hasMore: boolean
workflowSync: PerformanceConversionWorkflowSync
performanceSync?: PerformanceConversionSourceSync
}
export type PerformanceConversionDailyMetric = {

View File

@ -2,7 +2,7 @@
## 结论
当前应用发布工具验证二十五份目标 migration。普通部署脚本**只读验证,不会自动修改数据库**。只有独立 migrator 在同时拿到精确 Git 提交、数据库身份和备份凭证后,才允许执行 `prisma migrate deploy`。
当前应用发布工具验证二十六份目标 migration。普通部署脚本**只读验证,不会自动修改数据库**。只有独立 migrator 在同时拿到精确 Git 提交、数据库身份和备份凭证后,才允许执行 `prisma migrate deploy`。
目标 migration:
@ -31,6 +31,9 @@
- `20260914_expand_brand_knowledge_download_count`
- `20260914_expand_brand_knowledge_preview_slices`
- `20260915_add_brand_knowledge_soft_delete`
- `20260918_add_performance_conversion_sync`
`20260918_add_performance_conversion_sync` 只新增业绩转化经营数据快照表和每日增量同步状态表。首次同步按已完结美术流程的商品 ID 补齐六个月监控区间,后续每日重刷最近两天;租约索引用于防止多实例重复执行,不改写流程快照或上游报表库。
`20260915_add_brand_knowledge_soft_delete` 只为品牌知识增加可空删除时间和查询索引。删除操作仅写入删除时间,主文件和预览图继续保留;列表、详情、下载、预览、点赞和收藏均排除已删除内容。
@ -55,7 +58,7 @@
1. 构建并固定同一 exact SHA/digest 的 runner 与 migrator;
2. legacy bootstrap `Audit`;
3. legacy bootstrap `Apply`,外部验收 503,确认旧容器停止且保留;
4. 在同一维护窗内运行本文的真实备份、隔离恢复、二十五条目标 migration 和迁移后审计;
4. 在同一维护窗内运行本文的真实备份、隔离恢复、二十六条目标 migration 和迁移后审计;
5. 仅在所有门禁通过后,运行普通同 SHA 应用部署。
顺序不得改为“先 migration,后 bootstrap”。本文备份工具的生产前后摘要若有变化,说明维护窗内仍可能存在写入,会直接失败且不生成 receipt。
@ -379,7 +382,7 @@ docker run --rm \
"$MIGRATOR_IMAGE" --apply
```
工具会在修改数据库前先验证审计目录可执行“新建→原子改名→删除”,再依次完成:核对镜像 SHA、配置端口与数据库实际端口、核对机器生成的 receipt 和完整备份产物、拒绝未知待执行/失败 migration、执行 `prisma migrate deploy`,最后核对二十五条目标 migration 的数据库 checksum、完成状态、关键人员组织表、素材空壳标记、钉钉流程快照表及需求类型和表单字段、员工花名册历史快照表、字段类型与可空性、`api_keys` 三个可空加密列、`api_key_reveal_audits` 关键列、租约/需求类型/API Key reveal 审计索引的精确列顺序、`materials_uploadId_key` 的唯一性/btree/单列/有效就绪状态/active-only 谓词、Logo 默认值,以及 `materials.task_detail_demand_type` 无默认值、`materials.task_detail_private` / `material_permission_pack_items.can_access_private` 默认值为 `false`,并用稳定语义核对 `materials_private_optimization_check`,确认私密任务详情仍被精确限制在“任务-详情 + 优化”条件下,同时确认品牌知识预览切片列为非空文本数组且默认空数组;最后原子写审计 JSON。
工具会在修改数据库前先验证审计目录可执行“新建→原子改名→删除”,再依次完成:核对镜像 SHA、配置端口与数据库实际端口、核对机器生成的 receipt 和完整备份产物、拒绝未知待执行/失败 migration、执行 `prisma migrate deploy`,最后核对二十六条目标 migration 的数据库 checksum、完成状态、关键人员组织表、素材空壳标记、钉钉流程快照表及需求类型和表单字段、员工花名册历史快照表、业绩转化经营数据快照及同步状态表、字段类型与可空性、`api_keys` 三个可空加密列、`api_key_reveal_audits` 关键列、租约/需求类型/API Key reveal 审计索引的精确列顺序、`materials_uploadId_key` 的唯一性/btree/单列/有效就绪状态/active-only 谓词、Logo 默认值,以及 `materials.task_detail_demand_type` 无默认值、`materials.task_detail_private` / `material_permission_pack_items.can_access_private` 默认值为 `false`,并用稳定语义核对 `materials_private_optimization_check`,确认私密任务详情仍被精确限制在“任务-详情 + 优化”条件下,同时确认品牌知识预览切片列为非空文本数组且默认空数组;最后原子写审计 JSON。
apply 成功后立即固定审计文件本身;后续“迁移批次部署”读取这一个文件,不能手工重写:
@ -395,7 +398,7 @@ test -n "$MIGRATION_AUDIT_SHA256"
只有以下全部成立,才进入应用首次安全上线步骤:
- `applied.json` 的 `verified=true`;
- 二十五条 target migration 均 `completed=true`,source/database checksum 一致;
- 二十六条 target migration 均 `completed=true`,source/database checksum 一致;
- `_prisma_migrations.finished_at` 非空、`rolled_back_at` 为空;
- 视频租约列/索引、任务详情作业租约列/索引、active-only `uploadId` 唯一索引、品牌默认值精确匹配;
- 再次运行 `--audit` 仍通过;

View File

@ -130,6 +130,16 @@ export async function register() {
console.error('[Instrumentation] 调度钉钉流程每小时同步任务失败:', error)
}
// 业绩转化经营数据:首次按已完结美术流程商品建立快照,之后每日只重刷
// 最近两天,覆盖报表库对前一天数据的迟到写入或更正。
try {
const { ensureDailyPerformanceConversionSyncStarted } = await import('./lib/performance-conversion')
ensureDailyPerformanceConversionSyncStarted()
console.log('[Instrumentation] 业绩转化经营数据每日增量同步任务已调度')
} catch (error) {
console.error('[Instrumentation] 调度业绩转化经营数据每日同步任务失败:', error)
}
// 作品过稿后台是首轮通过指标的唯一权威源;应用启动后增量检查项目、
// V1 版本和评审记录,避免看板退化为只在有人打开时才知道数据是否更新。
try {

View File

@ -0,0 +1,163 @@
/** @jest-environment node */
import mssql from 'mssql'
import { prisma } from '@/lib/prisma'
import { syncPerformanceConversionMetrics } from '@/lib/performance-conversion'
jest.mock('mssql', () => {
const request = { input: jest.fn(), query: jest.fn() }
request.input.mockReturnValue(request)
const pool = { connect: jest.fn(), close: jest.fn(), request: jest.fn(() => request) }
return {
__esModule: true,
default: {
ConnectionPool: jest.fn(() => pool),
NVarChar: jest.fn((length: number) => `NVarChar(${length})`),
__pool: pool,
__request: request,
},
}
})
jest.mock('@/lib/prisma', () => ({
prisma: {
dingWorkflowSnapshot: { findMany: jest.fn() },
performanceConversionSyncState: {
upsert: jest.fn(),
updateMany: jest.fn(),
findUniqueOrThrow: jest.fn(),
},
performanceConversionMetricSnapshot: { findFirst: jest.fn() },
$transaction: jest.fn(),
},
}))
jest.mock('@/lib/ding-workflow-sync', () => ({
DING_WORKFLOW_ALLOWED_PROCESS_CODES: ['art-process'],
DING_WORKFLOW_STARTED_AT_CUTOFF_ISO: '2026-03-31T16:00:00.000Z',
getDingWorkflowSyncMetadata: jest.fn(),
isDingWorkflowApprovedCompletion: jest.fn(() => true),
}))
const workflowFindMany = prisma.dingWorkflowSnapshot.findMany as jest.Mock
const syncState = prisma.performanceConversionSyncState as unknown as {
upsert: jest.Mock
updateMany: jest.Mock
findUniqueOrThrow: jest.Mock
}
const metricSnapshot = prisma.performanceConversionMetricSnapshot as unknown as { findFirst: jest.Mock }
const runTransaction = prisma.$transaction as jest.Mock
const transactionSnapshot = { deleteMany: jest.fn(), createMany: jest.fn() }
const mssqlMock = mssql as typeof mssql & {
__pool: { connect: jest.Mock; close: jest.Mock }
__request: { input: jest.Mock; query: jest.Mock }
}
function completedWorkflow() {
return {
processInstanceId: 'flow-1',
businessId: 'approval-1',
title: '主图优化',
result: 'agree',
bizAction: 'finish',
status: 'completed',
processCode: 'art-process',
brand: '亿家达',
sourceFinishedAtRaw: '2026-09-01T08:00:00.000Z',
formFields: { productId: '1084275176878', platform: '天猫' },
}
}
describe('performance conversion daily incremental sync', () => {
beforeAll(() => {
jest.useFakeTimers().setSystemTime(new Date('2026-09-18T04:00:00.000Z'))
})
afterAll(() => {
jest.useRealTimers()
})
beforeEach(() => {
jest.clearAllMocks()
process.env.REPORT_DB_SERVER = 'report.internal'
process.env.REPORT_DB_DATABASE = 'report'
process.env.REPORT_DB_USER = 'reader'
process.env.REPORT_DB_PASSWORD = 'secret'
mssqlMock.__pool.connect.mockResolvedValue(undefined)
mssqlMock.__pool.close.mockResolvedValue(undefined)
syncState.upsert.mockResolvedValue({})
syncState.updateMany.mockResolvedValue({ count: 1 })
workflowFindMany.mockResolvedValue([completedWorkflow()])
runTransaction.mockImplementation(async (callback: (tx: unknown) => unknown) => callback({
performanceConversionMetricSnapshot: transactionSnapshot,
}))
transactionSnapshot.deleteMany.mockResolvedValue({ count: 0 })
transactionSnapshot.createMany.mockResolvedValue({ count: 1 })
})
it('backfills the full workflow monitoring window on the first product sync', async () => {
syncState.findUniqueOrThrow.mockResolvedValue({ sourceCursorRaw: null })
metricSnapshot.findFirst.mockResolvedValue(null)
mssqlMock.__request.query.mockResolvedValue({
recordset: [{
sourceRowId: 'row-1',
metricDate: new Date('2026-09-17T00:00:00.000Z'),
productId: '1084275176878',
paymentAmount: '100.00',
}],
})
await expect(syncPerformanceConversionMetrics()).resolves.toEqual({
skipped: false,
rowsUpserted: 1,
productsSynced: 1,
})
expect(mssqlMock.__request.input).toHaveBeenCalledWith('startDate', 'NVarChar(10)', '2026-09-01')
expect(mssqlMock.__request.input).toHaveBeenCalledWith('endDate', 'NVarChar(10)', '2026-09-18')
expect(transactionSnapshot.deleteMany).toHaveBeenCalledWith({
where: {
productId: '1084275176878',
metricDate: {
gte: new Date('2026-09-01T00:00:00.000Z'),
lte: new Date('2026-09-18T00:00:00.000Z'),
},
},
})
expect(transactionSnapshot.createMany).toHaveBeenCalledWith({
data: [expect.objectContaining({
sourceKey: '1084275176878:2026-09-17:row-1',
productId: '1084275176878',
paymentAmount: '100.00',
})],
})
})
it('refreshes only the rolling correction window after a completed sync', async () => {
syncState.findUniqueOrThrow.mockResolvedValue({
sourceCursorRaw: JSON.stringify({ version: 1, lastSourceDate: '2026-09-17' }),
})
metricSnapshot.findFirst.mockResolvedValue({ id: 'existing-row' })
mssqlMock.__request.query.mockResolvedValue({ recordset: [] })
await syncPerformanceConversionMetrics()
expect(mssqlMock.__request.input).toHaveBeenCalledWith('startDate', 'NVarChar(10)', '2026-09-15')
expect(mssqlMock.__request.input).toHaveBeenCalledWith('endDate', 'NVarChar(10)', '2026-09-18')
expect(transactionSnapshot.createMany).not.toHaveBeenCalled()
})
it('does not report success after ownership of the sync lease is lost', async () => {
syncState.findUniqueOrThrow.mockResolvedValue({ sourceCursorRaw: null })
metricSnapshot.findFirst.mockResolvedValue(null)
mssqlMock.__request.query.mockResolvedValue({ recordset: [] })
syncState.updateMany
.mockResolvedValueOnce({ count: 1 })
.mockResolvedValueOnce({ count: 1 })
.mockResolvedValueOnce({ count: 0 })
.mockResolvedValueOnce({ count: 0 })
await expect(syncPerformanceConversionMetrics()).rejects.toThrow('经营数据同步租约已失效')
})
})

View File

@ -1,5 +1,7 @@
import { randomUUID } from 'node:crypto'
import { hostname } from 'node:os'
import type { Session } from 'next-auth'
import mssql from 'mssql'
import mssql, { type ConnectionPool } from 'mssql'
import { assertBrandCenterDashboardAccess } from '@/lib/brand-center-dashboard-access'
import { prisma } from '@/lib/prisma'
import {
@ -28,6 +30,11 @@ const VALID_BRANDS = new Set<string>(BUSINESS_TREND_BRANDS)
const DEFAULT_PAGE_SIZE = 20
const MAX_PAGE_SIZE = 50
const SOURCE_CACHE_TTL_MS = 5 * 60 * 1000
const PERFORMANCE_CONVERSION_SYNC_STATE_ID = 'performance-conversion-daily-v1'
const SYNC_LEASE_DURATION_MS = 20 * 60 * 1000
const SYNC_INITIAL_DELAY_MS = 30 * 1000
const SYNC_INTERVAL_MS = 24 * 60 * 60 * 1000
const SYNC_LOOKBACK_DAYS = 2
export class PerformanceConversionError extends Error {
status: number
@ -53,6 +60,7 @@ type SnapshotRow = {
}
type SourceMetricRow = {
sourceRowId: string | null
metricDate: Date | string | null
productId: string | null
productName: string | null
@ -77,6 +85,19 @@ type SourceMetricRow = {
type CachedSourceRows = { expiresAt: number; rows: SourceMetricRow[] }
const sourceCache = new Map<string, CachedSourceRows>()
type PerformanceConversionSourceCursor = {
version: 1
lastSourceDate: string | null
}
export type PerformanceConversionSyncStatus = 'pending' | 'syncing' | 'ready' | 'error'
export type PerformanceConversionSyncMetadata = {
status: PerformanceConversionSyncStatus
lastCompletedAt: string | null
lastSourceDate: string | null
}
function clean(value: unknown) {
const normalized = String(value ?? '').trim()
return normalized && normalized.toUpperCase() !== 'NULL' && normalized !== '--' ? normalized : null
@ -172,9 +193,10 @@ export async function getPerformanceConversionWorkflows(
if (completedFrom && completedTo && compareDateKeys(completedFrom, completedTo) > 0) {
throw new PerformanceConversionError('完结开始日期不能晚于结束日期')
}
const [snapshots, workflowSync] = await Promise.all([
const [snapshots, workflowSync, performanceSync] = await Promise.all([
readEligibleSnapshots(),
getDingWorkflowSyncMetadata(),
getPerformanceConversionSyncMetadata(),
])
const rangeSnapshots = snapshots.filter((row) => {
const completedDate = row.sourceFinishedAtRaw ? toShanghaiDateKey(row.sourceFinishedAtRaw) : null
@ -207,6 +229,7 @@ export async function getPerformanceConversionWorkflows(
pageSize,
hasMore: offset + pageSize < filtered.length,
workflowSync,
performanceSync,
}
}
@ -234,24 +257,18 @@ function sourceConfig() {
},
connectionTimeout: 15_000,
requestTimeout: 60_000,
pool: { max: 1, min: 0, idleTimeoutMillis: 10_000 },
pool: { max: 4, min: 0, idleTimeoutMillis: 10_000 },
}
}
async function readSourceRows(productId: string, startDate: string, endDate: string) {
const cacheKey = `${productId}:${startDate}:${endDate}`
const cached = sourceCache.get(cacheKey)
if (cached && cached.expiresAt > Date.now()) return cached.rows
const pool = new mssql.ConnectionPool(sourceConfig())
try {
await pool.connect()
const result = await pool.request()
async function fetchSourceRows(pool: ConnectionPool, productId: string, startDate: string, endDate: string) {
const result = await pool.request()
.input('productId', mssql.NVarChar(255), productId)
.input('startDate', mssql.NVarChar(10), startDate)
.input('endDate', mssql.NVarChar(10), endDate)
.query<SourceMetricRow>(`
SELECT TOP (5000)
CONVERT(nvarchar(255), [RowId]) AS sourceRowId,
[统计日期] AS metricDate,
CONVERT(nvarchar(255), [商品ID]) AS productId,
CONVERT(nvarchar(2000), [商品名称]) AS productName,
@ -277,7 +294,122 @@ async function readSourceRows(productId: string, startDate: string, endDate: str
AND [统计日期] <= TRY_CONVERT(date, @endDate)
ORDER BY [统计日期] ASC, [RowId] ASC
`)
const rows = result.recordset || []
return result.recordset || []
}
function dateKeyToUtcDate(value: string) {
return new Date(`${value}T00:00:00.000Z`)
}
function shiftDateKey(value: string, days: number) {
const date = dateKeyToUtcDate(value)
date.setUTCDate(date.getUTCDate() + days)
return date.toISOString().slice(0, 10)
}
function maxDateKey(left: string, right: string) {
return compareDateKeys(left, right) >= 0 ? left : right
}
function parsePerformanceConversionSourceCursor(value: string | null): PerformanceConversionSourceCursor {
if (!value) return { version: 1, lastSourceDate: null }
try {
const parsed = JSON.parse(value) as Partial<PerformanceConversionSourceCursor>
return {
version: 1,
lastSourceDate: typeof parsed.lastSourceDate === 'string' && isValidDateKey(parsed.lastSourceDate)
? parsed.lastSourceDate
: null,
}
} catch {
return { version: 1, lastSourceDate: null }
}
}
function serializePerformanceConversionSourceCursor(cursor: PerformanceConversionSourceCursor) {
return JSON.stringify(cursor)
}
function normalizePerformanceConversionSyncStatus(state: { status: string; lastCompletedAt: Date | null } | null): PerformanceConversionSyncStatus {
if (!state?.lastCompletedAt) {
return state?.status === 'syncing' ? 'syncing' : state?.status === 'error' ? 'error' : 'pending'
}
if (state.status === 'syncing') return 'syncing'
if (state.status === 'error') return 'error'
return 'ready'
}
export async function getPerformanceConversionSyncMetadata(): Promise<PerformanceConversionSyncMetadata> {
// Keep isolated unit tests and pre-migration local shells readable while the
// additive table is being introduced. Production Prisma clients always have
// this delegate after the migration has been applied.
const syncState = (prisma as typeof prisma & {
performanceConversionSyncState?: typeof prisma.performanceConversionSyncState
}).performanceConversionSyncState
if (!syncState) {
return { status: 'pending', lastCompletedAt: null, lastSourceDate: null }
}
try {
const state = await syncState.findUnique({
where: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID },
select: { status: true, lastCompletedAt: true, sourceCursorRaw: true },
})
const cursor = parsePerformanceConversionSourceCursor(state?.sourceCursorRaw || null)
return {
status: normalizePerformanceConversionSyncStatus(state),
lastCompletedAt: state?.lastCompletedAt?.toISOString() || null,
lastSourceDate: cursor.lastSourceDate,
}
} catch (error) {
console.warn('[performance-conversion-sync] 同步状态读取失败,暂按首次同步处理:', error)
return { status: 'pending', lastCompletedAt: null, lastSourceDate: null }
}
}
async function readSnapshotRows(productId: string, startDate: string, endDate: string) {
const rows = await prisma.performanceConversionMetricSnapshot.findMany({
where: {
productId,
metricDate: {
gte: dateKeyToUtcDate(startDate),
lte: dateKeyToUtcDate(endDate),
},
},
orderBy: [{ metricDate: 'asc' }, { sourceRowId: 'asc' }],
})
return rows.map((row) => ({
sourceRowId: row.sourceRowId,
metricDate: row.metricDate,
productId: row.productId,
productName: row.productName,
sourceBrand: row.sourceBrand,
category1: row.category1,
category2: row.category2,
category3: row.category3,
paymentAmount: row.paymentAmount,
paymentUsers: row.paymentUsers,
cartUsers: row.cartUsers,
visitors: row.visitors,
conversionRate: row.conversionRate,
favorites: row.favorites,
impressions: row.impressions,
clicks: row.clicks,
clickRate: row.clickRate,
promotionSpend: row.promotionSpend,
promotionRevenue: row.promotionRevenue,
roi: row.roi,
})) satisfies SourceMetricRow[]
}
async function readUpstreamSourceRows(productId: string, startDate: string, endDate: string) {
const cacheKey = `upstream:${productId}:${startDate}:${endDate}`
const cached = sourceCache.get(cacheKey)
if (cached && cached.expiresAt > Date.now()) return cached.rows
const pool = new mssql.ConnectionPool(sourceConfig())
try {
await pool.connect()
const rows = await fetchSourceRows(pool, productId, startDate, endDate)
sourceCache.set(cacheKey, { rows, expiresAt: Date.now() + SOURCE_CACHE_TTL_MS })
return rows
} finally {
@ -285,6 +417,233 @@ async function readSourceRows(productId: string, startDate: string, endDate: str
}
}
async function readSourceRows(productId: string, startDate: string, endDate: string) {
const sync = await getPerformanceConversionSyncMetadata()
if (sync.lastCompletedAt) {
const localRows = await readSnapshotRows(productId, startDate, endDate)
if (localRows.length > 0) {
return localRows
}
// A newly synced workflow may not be included in the previous daily run.
// Keep the old direct read only for that bootstrap gap; the next scheduled
// run will persist the same product into the local snapshot.
}
return readUpstreamSourceRows(productId, startDate, endDate)
}
type MetricProductWindow = {
productId: string
startDate: string
endDate: string
}
function getMetricProductWindows(snapshots: SnapshotRow[], today: string): MetricProductWindow[] {
const windows = new Map<string, MetricProductWindow>()
snapshots.flatMap(workflowItems).forEach((item) => {
const endDate = compareDateKeys(today, item.monitorEndDate) <= 0 ? today : item.monitorEndDate
const existing = windows.get(item.productId)
if (!existing) {
windows.set(item.productId, { productId: item.productId, startDate: item.monitorStartDate, endDate })
return
}
existing.startDate = compareDateKeys(item.monitorStartDate, existing.startDate) < 0
? item.monitorStartDate
: existing.startDate
existing.endDate = maxDateKey(existing.endDate, endDate)
})
return Array.from(windows.values()).filter((item) => compareDateKeys(item.startDate, item.endDate) <= 0)
}
async function acquirePerformanceConversionSyncLease(leaseOwner: string, startedAt: Date) {
await prisma.performanceConversionSyncState.upsert({
where: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID },
create: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID },
update: {},
})
const claimed = await prisma.performanceConversionSyncState.updateMany({
where: {
id: PERFORMANCE_CONVERSION_SYNC_STATE_ID,
OR: [
{ leaseExpiresAt: null },
{ leaseExpiresAt: { lt: startedAt } },
{ leaseOwner },
],
},
data: {
status: 'syncing',
lastStartedAt: startedAt,
lastError: null,
leaseOwner,
leaseExpiresAt: new Date(startedAt.getTime() + SYNC_LEASE_DURATION_MS),
},
})
if (claimed.count !== 1) return null
return prisma.performanceConversionSyncState.findUniqueOrThrow({
where: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID },
})
}
async function renewPerformanceConversionSyncLease(leaseOwner: string) {
const renewed = await prisma.performanceConversionSyncState.updateMany({
where: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID, leaseOwner },
data: { leaseExpiresAt: new Date(Date.now() + SYNC_LEASE_DURATION_MS) },
})
if (renewed.count !== 1) throw new Error('经营数据同步租约已失效,本轮停止写入')
}
function toMetricSnapshotData(row: SourceMetricRow, productId: string, syncedAt: Date, index: number) {
const metricDate = metricDateKey(row.metricDate)
if (!metricDate) return null
const sourceRowId = clean(row.sourceRowId)
return {
id: randomUUID(),
sourceKey: `${productId}:${metricDate}:${sourceRowId || `row-${index}`}`,
sourceRowId,
productId,
metricDate: dateKeyToUtcDate(metricDate),
productName: clean(row.productName),
sourceBrand: clean(row.sourceBrand),
category1: clean(row.category1),
category2: clean(row.category2),
category3: clean(row.category3),
paymentAmount: clean(row.paymentAmount),
paymentUsers: clean(row.paymentUsers),
cartUsers: clean(row.cartUsers),
visitors: clean(row.visitors),
conversionRate: clean(row.conversionRate),
favorites: clean(row.favorites),
impressions: clean(row.impressions),
clicks: clean(row.clicks),
clickRate: clean(row.clickRate),
promotionSpend: clean(row.promotionSpend),
promotionRevenue: clean(row.promotionRevenue),
roi: clean(row.roi),
syncedAt,
createdAt: syncedAt,
updatedAt: syncedAt,
}
}
export async function syncPerformanceConversionMetrics() {
if (!sourceDbConfigured()) throw new Error('未配置上游报表库(REPORT_DB_*),无法同步经营数据')
const startedAt = new Date()
const leaseOwner = `${hostname()}:${process.pid}:${randomUUID()}`
const state = await acquirePerformanceConversionSyncLease(leaseOwner, startedAt)
if (!state) return { skipped: true, rowsUpserted: 0, productsSynced: 0 }
const pool = new mssql.ConnectionPool(sourceConfig())
let rowsUpserted = 0
let productsSynced = 0
try {
await pool.connect()
const cursor = parsePerformanceConversionSourceCursor(state.sourceCursorRaw)
const today = toShanghaiDateKey(new Date())
if (!today) throw new Error('无法确定上海时区的当前日期')
const snapshots = await readEligibleSnapshots()
const windows = getMetricProductWindows(snapshots, today)
const syncStartedAt = new Date()
for (const window of windows) {
await renewPerformanceConversionSyncLease(leaseOwner)
const existing = await prisma.performanceConversionMetricSnapshot.findFirst({
where: { productId: window.productId },
select: { id: true },
})
const incrementalStart = cursor.lastSourceDate
? shiftDateKey(cursor.lastSourceDate, -SYNC_LOOKBACK_DAYS)
: window.startDate
const startDate = existing ? maxDateKey(window.startDate, incrementalStart) : window.startDate
const endDate = window.endDate
if (compareDateKeys(startDate, endDate) > 0) continue
const sourceRows = await fetchSourceRows(pool, window.productId, startDate, endDate)
const snapshotsToWrite = sourceRows
.map((row, index) => toMetricSnapshotData(row, window.productId, syncStartedAt, index))
.filter((row): row is NonNullable<ReturnType<typeof toMetricSnapshotData>> => row !== null)
await prisma.$transaction(async (tx) => {
await tx.performanceConversionMetricSnapshot.deleteMany({
where: {
productId: window.productId,
metricDate: { gte: dateKeyToUtcDate(startDate), lte: dateKeyToUtcDate(endDate) },
},
})
if (snapshotsToWrite.length > 0) {
await tx.performanceConversionMetricSnapshot.createMany({ data: snapshotsToWrite })
}
})
rowsUpserted += snapshotsToWrite.length
productsSynced += 1
}
sourceCache.clear()
const sourceCursorRaw = serializePerformanceConversionSourceCursor({ version: 1, lastSourceDate: today })
const completed = await prisma.performanceConversionSyncState.updateMany({
where: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID, leaseOwner },
data: {
status: 'ready',
sourceCursorRaw,
lastCompletedAt: syncStartedAt,
lastError: null,
leaseOwner: null,
leaseExpiresAt: null,
rowsUpserted,
productsSynced,
},
})
if (completed.count !== 1) throw new Error('经营数据同步租约已失效,本轮不能标记为成功')
return { skipped: false, rowsUpserted, productsSynced }
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
await prisma.performanceConversionSyncState.updateMany({
where: { id: PERFORMANCE_CONVERSION_SYNC_STATE_ID, leaseOwner },
data: {
status: 'error',
lastError: message.slice(0, 2_000),
leaseOwner: null,
leaseExpiresAt: null,
},
}).catch(() => undefined)
throw error
} finally {
await pool.close().catch(() => undefined)
}
}
type PerformanceConversionSyncGlobal = typeof globalThis & {
__performanceConversionDailySyncStarted?: boolean
}
export function ensureDailyPerformanceConversionSyncStarted() {
const schedulerGlobal = globalThis as PerformanceConversionSyncGlobal
if (schedulerGlobal.__performanceConversionDailySyncStarted) return
schedulerGlobal.__performanceConversionDailySyncStarted = true
if (!sourceDbConfigured()) {
console.info('[performance-conversion-sync] 未配置上游报表库,跳过每日经营数据同步')
return
}
const runOnce = () => {
void syncPerformanceConversionMetrics()
.then((result) => {
if (result.skipped) {
console.info('[performance-conversion-sync] 另一实例正在同步,本轮跳过')
return
}
console.info(`[performance-conversion-sync] 同步完成:更新 ${result.rowsUpserted} 条,商品 ${result.productsSynced} 个`)
})
.catch((error) => console.warn('[performance-conversion-sync] 同步失败:', error))
}
const initialTimer = setTimeout(runOnce, SYNC_INITIAL_DELAY_MS)
initialTimer.unref?.()
const interval = setInterval(runOnce, SYNC_INTERVAL_MS)
interval.unref?.()
console.info('[performance-conversion-sync] 已调度经营数据每日增量同步,周期 24 小时,按最近两天增量校准')
}
function numberValue(value: unknown) {
const normalized = clean(value)?.replace(/[,%¥¥\s]/g, '')
if (!normalized) return null

View File

@ -0,0 +1,58 @@
CREATE TABLE "performance_conversion_metric_snapshots" (
"id" TEXT NOT NULL,
"source_key" TEXT NOT NULL,
"source_row_id" TEXT,
"product_id" TEXT NOT NULL,
"metric_date" DATE NOT NULL,
"product_name" TEXT,
"source_brand" TEXT,
"category1" TEXT,
"category2" TEXT,
"category3" TEXT,
"payment_amount" TEXT,
"payment_users" TEXT,
"cart_users" TEXT,
"visitors" TEXT,
"conversion_rate" TEXT,
"favorites" TEXT,
"impressions" TEXT,
"clicks" TEXT,
"click_rate" TEXT,
"promotion_spend" TEXT,
"promotion_revenue" TEXT,
"roi" TEXT,
"synced_at" TIMESTAMP(3) NOT NULL,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "performance_conversion_metric_snapshots_pkey" PRIMARY KEY ("id")
);
CREATE UNIQUE INDEX "performance_conversion_metric_snapshots_source_key_key"
ON "performance_conversion_metric_snapshots"("source_key");
CREATE INDEX "performance_conversion_metric_product_date_idx"
ON "performance_conversion_metric_snapshots"("product_id", "metric_date");
CREATE INDEX "performance_conversion_metric_date_idx"
ON "performance_conversion_metric_snapshots"("metric_date");
CREATE TABLE "performance_conversion_sync_states" (
"id" TEXT NOT NULL,
"status" TEXT NOT NULL DEFAULT 'idle',
"source_cursor_raw" TEXT,
"last_started_at" TIMESTAMP(3),
"last_completed_at" TIMESTAMP(3),
"last_error" TEXT,
"lease_owner" TEXT,
"lease_expires_at" TIMESTAMP(3),
"rows_upserted" INTEGER NOT NULL DEFAULT 0,
"products_synced" INTEGER NOT NULL DEFAULT 0,
"created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updated_at" TIMESTAMP(3) NOT NULL,
CONSTRAINT "performance_conversion_sync_states_pkey" PRIMARY KEY ("id")
);
CREATE INDEX "performance_conversion_sync_states_lease_idx"
ON "performance_conversion_sync_states"("lease_expires_at");

View File

@ -1531,6 +1531,61 @@ model DingWorkflowSyncState {
@@map("ding_workflow_sync_states")
}
/// 业绩转化商品经营数据的本地只读快照。
///
/// 上游经营报表按商品和统计日期保存多行来源记录。快照保留来源行,
/// 详情页再按日期聚合,避免把来源表的修订值固化成不可追溯的汇总值。
model PerformanceConversionMetricSnapshot {
id String @id @default(uuid())
sourceKey String @unique @map("source_key")
sourceRowId String? @map("source_row_id")
productId String @map("product_id")
metricDate DateTime @db.Date @map("metric_date")
productName String? @map("product_name")
sourceBrand String? @map("source_brand")
category1 String?
category2 String?
category3 String?
paymentAmount String? @map("payment_amount")
paymentUsers String? @map("payment_users")
cartUsers String? @map("cart_users")
visitors String?
conversionRate String? @map("conversion_rate")
favorites String?
impressions String?
clicks String?
clickRate String? @map("click_rate")
promotionSpend String? @map("promotion_spend")
promotionRevenue String? @map("promotion_revenue")
roi String?
syncedAt DateTime @map("synced_at")
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @updatedAt @map("updated_at")
@@index([productId, metricDate], map: "performance_conversion_metric_product_date_idx")
@@index([metricDate], map: "performance_conversion_metric_date_idx")
@@map("performance_conversion_metric_snapshots")
}
/// 业绩转化经营数据每日增量同步游标、租约与可观测状态。
model PerformanceConversionSyncState {
id String @id
status String @default("idle")
sourceCursorRaw String? @map("source_cursor_raw")
lastStartedAt DateTime? @map("last_started_at")
lastCompletedAt DateTime? @map("last_completed_at")
lastError String? @map("last_error")
leaseOwner String? @map("lease_owner")
leaseExpiresAt DateTime? @map("lease_expires_at")
rowsUpserted Int @default(0) @map("rows_upserted")
productsSynced Int @default(0) @map("products_synced")
createdAt DateTime @default(now()) @map("created_at")
updatedAt DateTime @updatedAt @map("updated_at")
@@index([leaseExpiresAt], map: "performance_conversion_sync_states_lease_idx")
@@map("performance_conversion_sync_states")
}
model VideoTranscodeProgress {
id String @id @default(uuid())
materialId String @map("material_id")

View File

@ -1518,6 +1518,7 @@ const expectedMigrationNames = new Set([
'20260914_expand_brand_knowledge_download_count',
'20260914_expand_brand_knowledge_preview_slices',
'20260915_add_brand_knowledge_soft_delete',
'20260918_add_performance_conversion_sync',
]);
if (migrations.length !== expectedMigrationNames.size
|| migrations.some((item) => !expectedMigrationNames.has(item && item.name)

View File

@ -71,6 +71,7 @@ test('migration deployment gate recognizes the latest approved schema migration'
assert.match(gate, /20260914_expand_brand_knowledge_download_count/)
assert.match(gate, /20260914_expand_brand_knowledge_preview_slices/)
assert.match(gate, /20260915_add_brand_knowledge_soft_delete/)
assert.match(gate, /20260918_add_performance_conversion_sync/)
})
test('direct replacement runs database and dependency gates before switching traffic', () => {

View File

@ -34,6 +34,7 @@ const TARGET_MIGRATIONS = Object.freeze([
'20260914_expand_brand_knowledge_download_count',
'20260914_expand_brand_knowledge_preview_slices',
'20260915_add_brand_knowledge_soft_delete',
'20260918_add_performance_conversion_sync',
])
const VERIFIED_BACKUP_GENERATOR = 'scripts/create-verified-production-backup.mjs'
const VERIFIED_BACKUP_MAINTENANCE_CONFIRMATION = 'NO_WRITES_DURING_SNAPSHOT'
@ -358,7 +359,9 @@ async function queryEvidence(prisma, localMigrations) {
OR (table_name = 'task_detail_change_requests' AND column_name IN ('id', 'material_id', 'requester_user_id', 'manager_dingtalk_user_id', 'status', 'operations', 'base_material_updated_at', 'expires_at'))
OR (table_name = 'task_detail_change_notifications' AND column_name IN ('id', 'request_id', 'event_type', 'business_key', 'status', 'next_retry_at'))
OR (table_name = 'brand_knowledge' AND column_name IN ('id', 'title', 'knowledge_type', 'brand', 'product_category_name', 'file_name', 'file_path', 'mime_type', 'file_size', 'download_count', 'preview_image_name', 'preview_image_path', 'preview_image_slices', 'preview_image_mime_type', 'preview_image_size', 'uploaded_by', 'created_at', 'updated_at', 'deleted_at'))
OR (table_name IN ('brand_knowledge_likes', 'brand_knowledge_favorites') AND column_name IN ('id', 'knowledge_id', 'user_id', 'created_at')))
OR (table_name IN ('brand_knowledge_likes', 'brand_knowledge_favorites') AND column_name IN ('id', 'knowledge_id', 'user_id', 'created_at'))
OR (table_name = 'performance_conversion_metric_snapshots' AND column_name IN ('id', 'source_key', 'source_row_id', 'product_id', 'metric_date', 'product_name', 'source_brand', 'category1', 'category2', 'category3', 'payment_amount', 'payment_users', 'cart_users', 'visitors', 'conversion_rate', 'favorites', 'impressions', 'clicks', 'click_rate', 'promotion_spend', 'promotion_revenue', 'roi', 'synced_at', 'created_at', 'updated_at'))
OR (table_name = 'performance_conversion_sync_states' AND column_name IN ('id', 'status', 'source_cursor_raw', 'last_started_at', 'last_completed_at', 'last_error', 'lease_owner', 'lease_expires_at', 'rows_upserted', 'products_synced', 'created_at', 'updated_at')))
ORDER BY table_name, column_name`,
)
const indexes = await prisma.$queryRawUnsafe(
@ -404,7 +407,11 @@ async function queryEvidence(prisma, localMigrations) {
'task_detail_change_requests_status_updated_idx',
'task_detail_change_notifications_status_retry_idx',
'api_key_reveal_audits_api_key_id_created_at_idx',
'api_key_reveal_audits_viewer_user_id_created_at_idx'
'api_key_reveal_audits_viewer_user_id_created_at_idx',
'performance_conversion_metric_snapshots_source_key_key',
'performance_conversion_metric_product_date_idx',
'performance_conversion_metric_date_idx',
'performance_conversion_sync_states_lease_idx'
)
ORDER BY idx.relname`,
)
@ -592,6 +599,34 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) {
requireColumn('ding_workflow_sync_states', 'current_ongoing_count', 'integer', 'NO')
requireColumn('system_settings', 'headerLogoUrl', 'text', 'NO')
requireColumn('system_settings', 'loginLogoUrl', 'text', 'NO')
requireColumn('performance_conversion_metric_snapshots', 'id', 'text', 'NO')
requireColumn('performance_conversion_metric_snapshots', 'source_key', 'text', 'NO')
requireColumn('performance_conversion_metric_snapshots', 'source_row_id', 'text', 'YES')
requireColumn('performance_conversion_metric_snapshots', 'product_id', 'text', 'NO')
requireColumn('performance_conversion_metric_snapshots', 'metric_date', 'date', 'NO')
for (const column of [
'product_name', 'source_brand', 'category1', 'category2', 'category3',
'payment_amount', 'payment_users', 'cart_users', 'visitors', 'conversion_rate',
'favorites', 'impressions', 'clicks', 'click_rate', 'promotion_spend',
'promotion_revenue', 'roi',
]) {
requireColumn('performance_conversion_metric_snapshots', column, 'text', 'YES')
}
requireColumn('performance_conversion_metric_snapshots', 'synced_at', 'timestamp without time zone', 'NO')
requireColumn('performance_conversion_metric_snapshots', 'created_at', 'timestamp without time zone', 'NO')
requireColumn('performance_conversion_metric_snapshots', 'updated_at', 'timestamp without time zone', 'NO')
requireColumn('performance_conversion_sync_states', 'id', 'text', 'NO')
requireColumn('performance_conversion_sync_states', 'status', 'text', 'NO')
requireColumn('performance_conversion_sync_states', 'source_cursor_raw', 'text', 'YES')
requireColumn('performance_conversion_sync_states', 'last_started_at', 'timestamp without time zone', 'YES')
requireColumn('performance_conversion_sync_states', 'last_completed_at', 'timestamp without time zone', 'YES')
requireColumn('performance_conversion_sync_states', 'last_error', 'text', 'YES')
requireColumn('performance_conversion_sync_states', 'lease_owner', 'text', 'YES')
requireColumn('performance_conversion_sync_states', 'lease_expires_at', 'timestamp without time zone', 'YES')
requireColumn('performance_conversion_sync_states', 'rows_upserted', 'integer', 'NO')
requireColumn('performance_conversion_sync_states', 'products_synced', 'integer', 'NO')
requireColumn('performance_conversion_sync_states', 'created_at', 'timestamp without time zone', 'NO')
requireColumn('performance_conversion_sync_states', 'updated_at', 'timestamp without time zone', 'NO')
const header = columnByKey.get('system_settings.headerLogoUrl')
const login = columnByKey.get('system_settings.loginLogoUrl')
@ -601,6 +636,9 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) {
const canAccessPrivate = columnByKey.get('material_permission_pack_items.can_access_private')
const brandKnowledgeDownloadCount = columnByKey.get('brand_knowledge.download_count')
const brandKnowledgePreviewSlices = columnByKey.get('brand_knowledge.preview_image_slices')
const performanceSyncStatus = columnByKey.get('performance_conversion_sync_states.status')
const performanceSyncRowsUpserted = columnByKey.get('performance_conversion_sync_states.rows_upserted')
const performanceSyncProductsSynced = columnByKey.get('performance_conversion_sync_states.products_synced')
if (!header || normalizeColumnDefault(header.column_default) !== '/brand/header-logo.png') {
fail('Database default mismatch: system_settings.headerLogoUrl')
}
@ -622,6 +660,15 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) {
if (!brandKnowledgePreviewSlices || String(brandKnowledgePreviewSlices.column_default || '').replace(/\s+/g, '').toLowerCase() !== 'array[]::text[]') {
fail('Database default mismatch: brand_knowledge.preview_image_slices')
}
if (!performanceSyncStatus || normalizeColumnDefault(performanceSyncStatus.column_default) !== 'idle') {
fail('Database default mismatch: performance_conversion_sync_states.status')
}
if (!performanceSyncRowsUpserted || normalizeColumnDefault(performanceSyncRowsUpserted.column_default) !== '0') {
fail('Database default mismatch: performance_conversion_sync_states.rows_upserted')
}
if (!performanceSyncProductsSynced || normalizeColumnDefault(performanceSyncProductsSynced.column_default) !== '0') {
fail('Database default mismatch: performance_conversion_sync_states.products_synced')
}
if (legacy && (legacy.data_type !== 'text' || legacy.is_nullable !== 'NO')) {
fail('Database column mismatch: system_settings.logoUrl')
}
@ -718,6 +765,27 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) {
'api_key_reveal_audits',
['viewer_user_id', 'created_at'],
)
assertExactUniqueBtreeIndex(
evidence.indexes.find((row) => String(row.indexname) === 'performance_conversion_metric_snapshots_source_key_key'),
'performance_conversion_metric_snapshots',
'performance_conversion_metric_snapshots_source_key_key',
['source_key'],
)
assertExactBtreeIndex(
indexByName.get('performance_conversion_metric_product_date_idx'),
'performance_conversion_metric_snapshots',
['product_id', 'metric_date'],
)
assertExactBtreeIndex(
indexByName.get('performance_conversion_metric_date_idx'),
'performance_conversion_metric_snapshots',
['metric_date'],
)
assertExactBtreeIndex(
indexByName.get('performance_conversion_sync_states_lease_idx'),
'performance_conversion_sync_states',
['lease_expires_at'],
)
assertExactCheckConstraint(
(evidence.constraints || []).find((row) => String(row.constraint_name) === 'materials_private_optimization_check'),
'materials',

View File

@ -350,6 +350,7 @@ test('target evidence rejects nullable logo columns and structurally wrong index
['20260914_expand_brand_knowledge_download_count', { checksum: 'w' }],
['20260914_expand_brand_knowledge_preview_slices', { checksum: 'x' }],
['20260915_add_brand_knowledge_soft_delete', { checksum: 'y' }],
['20260918_add_performance_conversion_sync', { checksum: 'z' }],
])
const evidence = {
migrationRows: [
@ -378,6 +379,7 @@ test('target evidence rejects nullable logo columns and structurally wrong index
{ migration_name: '20260914_expand_brand_knowledge_download_count', checksum: 'w', finished_at: 'now', rolled_back_at: null },
{ migration_name: '20260914_expand_brand_knowledge_preview_slices', checksum: 'x', finished_at: 'now', rolled_back_at: null },
{ migration_name: '20260915_add_brand_knowledge_soft_delete', checksum: 'y', finished_at: 'now', rolled_back_at: null },
{ migration_name: '20260918_add_performance_conversion_sync', checksum: 'z', finished_at: 'now', rolled_back_at: null },
],
columns: [
{ table_name: 'materials', column_name: 'transcode_lease_token', data_type: 'text', is_nullable: 'YES' },
@ -486,6 +488,30 @@ test('target evidence rejects nullable logo columns and structurally wrong index
{ table_name, column_name: 'user_id', data_type: 'text', is_nullable: 'NO' },
{ table_name, column_name: 'created_at', data_type: 'timestamp without time zone', is_nullable: 'NO' },
]),
...[
['id', 'text', 'NO'], ['source_key', 'text', 'NO'], ['source_row_id', 'text', 'YES'],
['product_id', 'text', 'NO'], ['metric_date', 'date', 'NO'],
['product_name', 'text', 'YES'], ['source_brand', 'text', 'YES'], ['category1', 'text', 'YES'],
['category2', 'text', 'YES'], ['category3', 'text', 'YES'], ['payment_amount', 'text', 'YES'],
['payment_users', 'text', 'YES'], ['cart_users', 'text', 'YES'], ['visitors', 'text', 'YES'],
['conversion_rate', 'text', 'YES'], ['favorites', 'text', 'YES'], ['impressions', 'text', 'YES'],
['clicks', 'text', 'YES'], ['click_rate', 'text', 'YES'], ['promotion_spend', 'text', 'YES'],
['promotion_revenue', 'text', 'YES'], ['roi', 'text', 'YES'],
['synced_at', 'timestamp without time zone', 'NO'], ['created_at', 'timestamp without time zone', 'NO'],
['updated_at', 'timestamp without time zone', 'NO'],
].map(([column_name, data_type, is_nullable]) => ({
table_name: 'performance_conversion_metric_snapshots', column_name, data_type, is_nullable,
})),
...[
['id', 'text', 'NO', undefined], ['status', 'text', 'NO', "'idle'::text"], ['source_cursor_raw', 'text', 'YES', undefined],
['last_started_at', 'timestamp without time zone', 'YES'], ['last_completed_at', 'timestamp without time zone', 'YES'],
['last_error', 'text', 'YES'], ['lease_owner', 'text', 'YES'],
['lease_expires_at', 'timestamp without time zone', 'YES'], ['rows_upserted', 'integer', 'NO', '0'],
['products_synced', 'integer', 'NO', '0'], ['created_at', 'timestamp without time zone', 'NO'],
['updated_at', 'timestamp without time zone', 'NO'],
].map(([column_name, data_type, is_nullable, column_default]) => ({
table_name: 'performance_conversion_sync_states', column_name, data_type, is_nullable, column_default,
})),
{ table_name: 'system_settings', column_name: 'headerLogoUrl', data_type: 'text', is_nullable: 'NO', column_default: "'/brand/header-logo.png'::text" },
{ table_name: 'system_settings', column_name: 'loginLogoUrl', data_type: 'text', is_nullable: 'NO', column_default: "'/brand/login-logo.png'::text" },
{ table_name: 'system_settings', column_name: 'logoUrl', data_type: 'text', is_nullable: 'NO', column_default: "'/brand/header-logo.png'::text" },
@ -518,6 +544,21 @@ test('target evidence rejects nullable logo columns and structurally wrong index
{ indexname: 'api_key_reveal_audits_viewer_user_id_created_at_idx', indexdef: 'CREATE INDEX api_key_reveal_audits_viewer_user_id_created_at_idx ON public.api_key_reveal_audits USING btree (viewer_user_id, created_at)' },
{ indexname: 'brand_knowledge_product_category_name_idx', indexdef: 'CREATE INDEX brand_knowledge_product_category_name_idx ON public.brand_knowledge USING btree (product_category_name)' },
{ indexname: 'brand_knowledge_deleted_at_idx', indexdef: 'CREATE INDEX brand_knowledge_deleted_at_idx ON public.brand_knowledge USING btree (deleted_at)' },
{
tablename: 'performance_conversion_metric_snapshots',
indexname: 'performance_conversion_metric_snapshots_source_key_key',
indexdef: 'CREATE UNIQUE INDEX performance_conversion_metric_snapshots_source_key_key ON public.performance_conversion_metric_snapshots USING btree (source_key)',
indisunique: true,
indisprimary: false,
indisvalid: true,
indisready: true,
access_method: 'btree',
columns: ['source_key'],
predicate: '',
},
{ indexname: 'performance_conversion_metric_product_date_idx', indexdef: 'CREATE INDEX performance_conversion_metric_product_date_idx ON public.performance_conversion_metric_snapshots USING btree (product_id, metric_date)' },
{ indexname: 'performance_conversion_metric_date_idx', indexdef: 'CREATE INDEX performance_conversion_metric_date_idx ON public.performance_conversion_metric_snapshots USING btree (metric_date)' },
{ indexname: 'performance_conversion_sync_states_lease_idx', indexdef: 'CREATE INDEX performance_conversion_sync_states_lease_idx ON public.performance_conversion_sync_states USING btree (lease_expires_at)' },
...['brand_knowledge_likes', 'brand_knowledge_favorites'].flatMap((table) => [
{
tablename: table,
@ -703,6 +744,14 @@ test('target evidence rejects nullable logo columns and structurally wrong index
() => assertTargetMigrationEvidence(migrations, missingRevealAuditIndex),
/api_key_reveal_audits/,
)
const missingPerformanceSyncIndex = {
...evidence,
indexes: evidence.indexes.filter((index) => index.indexname !== 'performance_conversion_sync_states_lease_idx'),
}
assert.throws(
() => assertTargetMigrationEvidence(migrations, missingPerformanceSyncIndex),
/performance_conversion_sync_states/,
)
const wrongPrivateConstraint = {
...evidence,
constraints: evidence.constraints.map((constraint) => constraint.constraint_name === 'materials_private_optimization_check'
@ -798,6 +847,7 @@ test('container supply chain and migrator image are immutable while SOP keeps co
'20260914_expand_brand_knowledge_download_count',
'20260914_expand_brand_knowledge_preview_slices',
'20260915_add_brand_knowledge_soft_delete',
'20260918_add_performance_conversion_sync',
]) {
assert.match(dockerfile, new RegExp(migrationName))
assert.match(migratorSource, new RegExp(`'${migrationName}'`))
@ -825,6 +875,8 @@ test('container supply chain and migrator image are immutable while SOP keeps co
assert.ok(sop.includes('`20260914_expand_brand_knowledge_preview_slices`'))
assert.match(deploymentSource, /'20260915_add_brand_knowledge_soft_delete'/)
assert.ok(sop.includes('`20260915_add_brand_knowledge_soft_delete`'))
assert.match(deploymentSource, /'20260918_add_performance_conversion_sync'/)
assert.ok(sop.includes('`20260918_add_performance_conversion_sync`'))
const indexesQueryStart = migratorSource.indexOf('const indexes = await prisma.$queryRawUnsafe(')
const indexesQueryEnd = migratorSource.indexOf('const constraints = await prisma.$queryRawUnsafe(', indexesQueryStart)
assert.ok(indexesQueryStart >= 0)
@ -908,8 +960,8 @@ test('container supply chain and migrator image are immutable while SOP keeps co
assert.match(sop, /-v "\$HOST_RESULT_DIR:\/audit"/)
assert.match(sop, /test -f "\$PREFLIGHT_AUDIT_FILE" && test ! -L "\$PREFLIGHT_AUDIT_FILE"/)
assert.doesNotMatch(sop, /-v \/root\/gongxiangp2-migration-audit:\/audit/)
assert.match(sop, /二十五份目标 migration/)
assert.match(sop, /二十五条目标 migration/)
assert.match(sop, /二十六份目标 migration/)
assert.match(sop, /二十六条目标 migration/)
const applyMigration = migratorSource.indexOf("['migrate', 'deploy'")
const postEvidence = migratorSource.indexOf('after = await queryEvidence(verifyClient, localMigrations)', applyMigration)