diff --git a/Dockerfile b/Dockerfile index 39501d7..b1c49e8 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/components/department-dashboard/PerformanceConversionDashboard.test.tsx b/components/department-dashboard/PerformanceConversionDashboard.test.tsx index 1fc1cb2..080a84d 100644 --- a/components/department-dashboard/PerformanceConversionDashboard.test.tsx +++ b/components/department-dashboard/PerformanceConversionDashboard.test.tsx @@ -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() + + 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({ diff --git a/components/department-dashboard/PerformanceConversionDashboard.tsx b/components/department-dashboard/PerformanceConversionDashboard.tsx index 62757c2..f1bcb9b 100644 --- a/components/department-dashboard/PerformanceConversionDashboard.tsx +++ b/components/department-dashboard/PerformanceConversionDashboard.tsx @@ -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 ( +
+

+

+

{view.detail}{sync.lastSourceDate ? ` · 已扫描至 ${formatDate(sync.lastSourceDate)}` : ''}

+
+ ) +} + function WorkflowTable({ items, detailHref }: { items: PerformanceConversionWorkflowItem[] detailHref: (item: PerformanceConversionWorkflowItem) => string @@ -217,7 +253,7 @@ export default function PerformanceConversionDashboard() {
-

当前品牌

{selectedBrand}

{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}

{listState.data && }
+

当前品牌

{selectedBrand}

{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}

{listState.data &&
}
完结日期 diff --git a/components/department-dashboard/performance-conversion.ts b/components/department-dashboard/performance-conversion.ts index 937712f..fb25f60 100644 --- a/components/department-dashboard/performance-conversion.ts +++ b/components/department-dashboard/performance-conversion.ts @@ -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 = { diff --git a/docs/生产数据库安全增量迁移SOP-2026-08-12.md b/docs/生产数据库安全增量迁移SOP-2026-08-12.md index 0e89490..faba7b5 100644 --- a/docs/生产数据库安全增量迁移SOP-2026-08-12.md +++ b/docs/生产数据库安全增量迁移SOP-2026-08-12.md @@ -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` 仍通过； diff --git a/instrumentation.ts b/instrumentation.ts index 708baa5..fe8c376 100644 --- a/instrumentation.ts +++ b/instrumentation.ts @@ -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 { diff --git a/lib/__tests__/performance-conversion-sync.test.ts b/lib/__tests__/performance-conversion-sync.test.ts new file mode 100644 index 0000000..5152f4f --- /dev/null +++ b/lib/__tests__/performance-conversion-sync.test.ts @@ -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('经营数据同步租约已失效') + }) +}) diff --git a/lib/performance-conversion.ts b/lib/performance-conversion.ts index 83519ac..b92b1c7 100644 --- a/lib/performance-conversion.ts +++ b/lib/performance-conversion.ts @@ -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(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() +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(` 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 + 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 { + // 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() + 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> => 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 diff --git a/prisma/migrations/20260918_add_performance_conversion_sync/migration.sql b/prisma/migrations/20260918_add_performance_conversion_sync/migration.sql new file mode 100644 index 0000000..e59db49 --- /dev/null +++ b/prisma/migrations/20260918_add_performance_conversion_sync/migration.sql @@ -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"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index beab4d3..2591d06 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -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") diff --git a/scripts/git-incremental-deploy.ps1 b/scripts/git-incremental-deploy.ps1 index d674f65..112fd3a 100644 --- a/scripts/git-incremental-deploy.ps1 +++ b/scripts/git-incremental-deploy.ps1 @@ -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) diff --git a/scripts/git-incremental-deploy.test.mjs b/scripts/git-incremental-deploy.test.mjs index c1a1631..e9001ff 100644 --- a/scripts/git-incremental-deploy.test.mjs +++ b/scripts/git-incremental-deploy.test.mjs @@ -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', () => { diff --git a/scripts/safe-production-migrate.mjs b/scripts/safe-production-migrate.mjs index 97c7096..f327469 100644 --- a/scripts/safe-production-migrate.mjs +++ b/scripts/safe-production-migrate.mjs @@ -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', diff --git a/scripts/safe-production-migrate.test.mjs b/scripts/safe-production-migrate.test.mjs index ebace15..3ec2b07 100644 --- a/scripts/safe-production-migrate.test.mjs +++ b/scripts/safe-production-migrate.test.mjs @@ -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)