diff --git a/Dockerfile b/Dockerfile
index e45f675..580f621 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -81,7 +81,8 @@ RUN test -s "/app/public/brand/header-logo.png" && \
test -f "/app/prisma/migrations/20260817_add_material_is_incomplete_shell/migration.sql" && \
test -f "/app/prisma/migrations/20260819_add_user_permission_inherit/migration.sql" && \
test -f "/app/prisma/migrations/20260820_add_permission_group_members/migration.sql" && \
- test -f "/app/prisma/migrations/20260820_unique_permission_group_department/migration.sql"
+ test -f "/app/prisma/migrations/20260820_unique_permission_group_department/migration.sql" && \
+ test -f "/app/prisma/migrations/20260820_add_ding_workflow_snapshots/migration.sql"
# 这两个动态路由负责兜底访问本地上传文件;如果它们在构建上下文里丢失,
# 线上音频/视频就会直接 404。这里提前失败,避免再把缺路由的镜像发出去。
@@ -194,6 +195,7 @@ COPY --from=builder --chown=root:root /app/prisma/migrations/20260817_add_materi
COPY --from=builder --chown=root:root /app/prisma/migrations/20260819_add_user_permission_inherit ./prisma/migrations/20260819_add_user_permission_inherit
COPY --from=builder --chown=root:root /app/prisma/migrations/20260820_add_permission_group_members ./prisma/migrations/20260820_add_permission_group_members
COPY --from=builder --chown=root:root /app/prisma/migrations/20260820_unique_permission_group_department ./prisma/migrations/20260820_unique_permission_group_department
+COPY --from=builder --chown=root:root /app/prisma/migrations/20260820_add_ding_workflow_snapshots ./prisma/migrations/20260820_add_ding_workflow_snapshots
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/app/api/department-dashboard/core-metrics/ongoing-projects/route.ts b/app/api/department-dashboard/core-metrics/ongoing-projects/route.ts
new file mode 100644
index 0000000..ad8d3b3
--- /dev/null
+++ b/app/api/department-dashboard/core-metrics/ongoing-projects/route.ts
@@ -0,0 +1,36 @@
+import { getServerSession } from 'next-auth'
+import { NextResponse } from 'next/server'
+import { authOptions } from '@/lib/auth'
+import {
+ DingWorkflowMetricApiError,
+ getOngoingDingWorkflowSummary,
+} from '@/lib/ding-workflow-sync'
+
+export const runtime = 'nodejs'
+export const dynamic = 'force-dynamic'
+export const revalidate = 0
+
+const PRIVATE_HEADERS = {
+ 'Cache-Control': 'private, no-store, max-age=0',
+}
+
+export async function GET() {
+ try {
+ const session = await getServerSession(authOptions)
+ const data = await getOngoingDingWorkflowSummary(session)
+ return NextResponse.json({ success: true, data }, { headers: PRIVATE_HEADERS })
+ } catch (error) {
+ if (error instanceof DingWorkflowMetricApiError) {
+ return NextResponse.json(
+ { success: false, error: error.message },
+ { status: error.status, headers: PRIVATE_HEADERS },
+ )
+ }
+
+ console.error('[department-dashboard] 进行中流程指标读取失败:', error)
+ return NextResponse.json(
+ { success: false, error: '进行中流程指标读取失败,请稍后重试' },
+ { status: 500, headers: PRIVATE_HEADERS },
+ )
+ }
+}
diff --git a/app/workspaces/[section]/[item]/[metric]/page.tsx b/app/workspaces/[section]/[item]/[metric]/page.tsx
index bdbab8c..19da000 100644
--- a/app/workspaces/[section]/[item]/[metric]/page.tsx
+++ b/app/workspaces/[section]/[item]/[metric]/page.tsx
@@ -4,9 +4,9 @@ import { notFound } from 'next/navigation'
import CoreMetricDetail from '@/components/department-dashboard/CoreMetricDetail'
import { getCoreMetric } from '@/components/department-dashboard/core-metrics'
import { authOptions } from '@/lib/auth'
+import { getOngoingDingWorkflowProjects } from '@/lib/ding-workflow-sync'
import {
getFirstPassReviewProjects,
- getOngoingWorkReviewProjects,
} from '@/lib/work-review-service'
type CoreMetricPageProps = {
@@ -34,7 +34,7 @@ export default async function CoreMetricPage({ params, searchParams }: CoreMetri
if (metric.key === 'ongoing-projects') {
const session = await getServerSession(authOptions)
- const projectPage = await getOngoingWorkReviewProjects(session, { page, pageSize: 20, query })
+ const projectPage = await getOngoingDingWorkflowProjects(session, { page, pageSize: 20, query })
return
}
diff --git a/components/department-dashboard/CoreMetricDetail.test.tsx b/components/department-dashboard/CoreMetricDetail.test.tsx
index 34a5b14..b393daf 100644
--- a/components/department-dashboard/CoreMetricDetail.test.tsx
+++ b/components/department-dashboard/CoreMetricDetail.test.tsx
@@ -16,25 +16,28 @@ describe('CoreMetricDetail', () => {
projectPage={{
items: [{
id: 'project-1',
+ businessId: 'BUSINESS-1',
title: '夏季主视觉过稿',
- description: null,
- status: 'pending_review',
- workDate: '2026-08-10T00:00:00.000Z',
- latestVersion: 2,
- submitterName: '林一',
- managerName: '陈主管',
- finalReviewerName: '元泽',
- updatedAt: '2026-08-14T02:30:00.000Z',
+ status: 'RUNNING',
+ originatorUserId: 'ding-1',
+ originatorDeptName: '品牌中心',
+ sourceCreatedAt: '2026-08-10T00:00:00.000Z',
+ sourceUpdatedAt: '2026-08-14T02:30:00.000Z',
+ activeTaskCount: 1,
+ taskCount: 3,
+ schemeFieldCount: 12,
+ latestTaskStatus: 'RUNNING',
}],
pagination: { page: 1, pageSize: 20, total: 1, totalPages: 1 },
+ sync: { total: 1, lastSyncedAt: '2026-08-14T02:30:00.000Z', syncStatus: 'ready' },
}}
/>,
)
expect(screen.getByRole('heading', { level: 1, name: '进行中项目' })).toBeInTheDocument()
expect(screen.getAllByText('夏季主视觉过稿')).toHaveLength(2)
- expect(screen.getAllByText('评审中').length).toBeGreaterThan(0)
- expect(screen.getByText(/数据来自作品过稿项目库/)).toBeInTheDocument()
+ expect(screen.getAllByText('流转中').length).toBeGreaterThan(0)
+ expect(screen.getByText(/数据来自钉钉审批流程本地快照/)).toBeInTheDocument()
expect(screen.getByRole('link', { name: /返回核心指标/ })).toHaveAttribute('href', '/workspaces/dashboard/core-metrics')
})
diff --git a/components/department-dashboard/CoreMetricDetail.tsx b/components/department-dashboard/CoreMetricDetail.tsx
index 3aa888c..357941e 100644
--- a/components/department-dashboard/CoreMetricDetail.tsx
+++ b/components/department-dashboard/CoreMetricDetail.tsx
@@ -13,9 +13,11 @@ import {
import type {
FirstPassReviewProjectListItem,
FirstPassReviewProjectPage,
- OngoingWorkReviewProjectListItem,
- OngoingWorkReviewProjectPage,
} from '@/lib/work-review-service'
+import type {
+ OngoingDingWorkflowListItem,
+ OngoingDingWorkflowPage,
+} from '@/lib/ding-workflow-sync'
import DepartmentShell from './DepartmentShell'
import OnTimeDeliveryRate from './OnTimeDeliveryRate'
import type { CoreMetric } from './core-metrics'
@@ -23,16 +25,19 @@ import type { CoreMetric } from './core-metrics'
type CoreMetricDetailProps = {
metric: CoreMetric
query?: string
- projectPage?: OngoingWorkReviewProjectPage
+ projectPage?: OngoingDingWorkflowPage
firstPassPage?: FirstPassReviewProjectPage
}
-const STATUS_LABELS: Record = {
- pending_review: '评审中',
- revision_required: '待修改',
+const WORKFLOW_STATUS_LABELS: Record = {
+ RUNNING: '流转中',
+ NEW: '待处理',
+ PENDING: '待处理',
+ PROCESSING: '处理中',
}
-function formatDate(value: string, includeTime = false) {
+function formatDate(value: string | null, includeTime = false) {
+ if (!value) return '—'
return new Intl.DateTimeFormat('zh-CN', {
year: 'numeric',
month: '2-digit',
@@ -55,11 +60,10 @@ function formatRate(rate: number | null) {
return `${Number.isInteger(rate) ? rate.toFixed(0) : rate.toFixed(1)}%`
}
-function StatusBadge({ status }: { status: OngoingWorkReviewProjectListItem['status'] }) {
- const isRevision = status === 'revision_required'
+function StatusBadge({ status }: { status: OngoingDingWorkflowListItem['status'] }) {
return (
-
- {STATUS_LABELS[status]}
+
+ {WORKFLOW_STATUS_LABELS[status] || '未完结'}
)
}
@@ -73,14 +77,14 @@ function ProjectList({ metric, query, projectPage }: Required
进行中项目明细
- 数据来自作品过稿项目库;“评审中”和“待修改”均计入进行中。
+ 数据来自钉钉审批流程本地快照;所有未完结流程均计入进行中,每小时增量同步一次。
共 {pagination.total} 个项目
) : (
@@ -116,12 +122,12 @@ function ProjectList({ metric, query, projectPage }: Required进行中项目数据表
- | 项目 |
- 状态 |
- 提交人 |
- 负责人 |
- 作品日期 |
- 最近更新 |
+ 流程 |
+ 状态 |
+ 发起部门 |
+ 待办任务 |
+ 发起时间 |
+ 最近更新 |
@@ -129,13 +135,13 @@ function ProjectList({ metric, query, projectPage }: Required
{project.title}
- 第 {project.latestVersion} 版 · 终审 {project.finalReviewerName}
+ 流程编号 {project.businessId || project.id}
|
|
- {project.submitterName} |
- {project.managerName} |
- {formatDate(project.workDate)} |
- {formatDate(project.updatedAt, true)} |
+ {project.originatorDeptName || '未记录'} |
+ {project.activeTaskCount} / {project.taskCount} |
+ {formatDate(project.sourceCreatedAt, true)} |
+ {formatDate(project.sourceUpdatedAt, true)} |
))}
@@ -148,15 +154,15 @@ function ProjectList({ metric, query, projectPage }: Required
{project.title}
-
第 {project.latestVersion} 版 · 终审 {project.finalReviewerName}
+
流程编号 {project.businessId || project.id}
- - 提交人
- {project.submitterName}
- - 负责人
- {project.managerName}
- - 作品日期
- {formatDate(project.workDate)}
- - 最近更新
- {formatDate(project.updatedAt, true)}
+ - 发起部门
- {project.originatorDeptName || '未记录'}
+ - 待办 / 任务
- {project.activeTaskCount} / {project.taskCount}
+ - 发起时间
- {formatDate(project.sourceCreatedAt, true)}
+ - 最近更新
- {formatDate(project.sourceUpdatedAt, true)}
))}
@@ -371,33 +377,40 @@ function PendingDataSource({ metric }: { metric: CoreMetric }) {
}
export default function CoreMetricDetail({ metric, query, projectPage, firstPassPage }: CoreMetricDetailProps) {
- const Icon = metric.icon
const hasRealProjectSource = metric.key === 'ongoing-projects' && projectPage
const hasFirstPassSource = metric.key === 'first-pass-rate' && firstPassPage
const isOnTimeDelivery = metric.key === 'on-time-delivery'
- const hasRealSource = hasRealProjectSource || hasFirstPassSource
return (
-
{hasFirstPassSource ? '当前首轮通过率' : hasRealProjectSource ? '当前项目数' : '总览示例值'}
+
{hasFirstPassSource ? '当前首轮通过率' : hasRealProjectSource ? '当前未完结流程' : '总览示例值'}
{isOnTimeDelivery && metric.exampleNumericValue !== undefined
?
:
{hasFirstPassSource ? formatRate(firstPassPage.summary.rate) : hasRealProjectSource ? projectPage.pagination.total : metric.exampleValue}
}
明细数据
-
{hasFirstPassSource ? `V1通过 ${firstPassPage.summary.firstPassApproved} / 全部 ${firstPassPage.summary.totalUploads}` : hasRealProjectSource ? `${projectPage.pagination.total} 条真实记录` : '等待业务数据源'}
+
{hasFirstPassSource ? `V1通过 ${firstPassPage.summary.firstPassApproved} / 全部 ${firstPassPage.summary.totalUploads}` : hasRealProjectSource ? `${projectPage.pagination.total} 条流程快照` : '等待业务数据源'}
+
+
+
同步状态
+
+
+ {hasRealProjectSource
+ ? projectPage.sync.lastSyncedAt
+ ? `每小时增量同步 · ${formatDate(projectPage.sync.lastSyncedAt, true)}`
+ : projectPage.sync.syncStatus === 'error' ? '首轮同步失败' : '等待首轮同步'
+ : hasFirstPassSource ? '打开页面时读取' : '每日同步待接入'}
+
- 同步状态
{hasRealSource ? '打开页面时读取' : '每日同步待接入'}
{hasRealProjectSource
@@ -408,7 +421,6 @@ export default function CoreMetricDetail({ metric, query, projectPage, firstPass
diff --git a/components/department-dashboard/CoreMetricsDashboard.test.tsx b/components/department-dashboard/CoreMetricsDashboard.test.tsx
index 36a1f2b..223bb29 100644
--- a/components/department-dashboard/CoreMetricsDashboard.test.tsx
+++ b/components/department-dashboard/CoreMetricsDashboard.test.tsx
@@ -8,26 +8,35 @@ jest.mock('next-auth/react', () => ({
describe('CoreMetricsDashboard', () => {
beforeEach(() => {
- global.fetch = jest.fn().mockResolvedValue({
+ global.fetch = jest.fn().mockImplementation(async (url: string) => ({
ok: true,
- json: async () => ({
- success: true,
- data: {
- totalUploads: 30,
- firstPassApproved: 11,
- rate: 36.7,
- statusBreakdown: { approved: 11, revisionRequired: 7, pendingReview: 10, unknown: 2 },
- readAt: '2026-08-14T02:00:00.000Z',
- },
- }),
- }) as jest.Mock
+ json: async () => url.includes('ongoing-projects')
+ ? {
+ success: true,
+ data: {
+ total: 2338,
+ lastSyncedAt: '2026-08-14T02:00:00.000Z',
+ syncStatus: 'ready',
+ },
+ }
+ : {
+ success: true,
+ data: {
+ totalUploads: 30,
+ firstPassApproved: 11,
+ rate: 36.7,
+ statusBreakdown: { approved: 11, revisionRequired: 7, pendingReview: 10, unknown: 2 },
+ readAt: '2026-08-14T02:00:00.000Z',
+ },
+ },
+ })) as jest.Mock
})
afterEach(() => {
jest.restoreAllMocks()
})
- it('展示真实首轮通过率与其余五项诚实的示例状态', async () => {
+ it('展示真实进行中流程、首轮通过率与其余四项诚实的示例状态', async () => {
render()
expect(screen.getByRole('heading', { level: 1, name: '核心指标' })).toBeInTheDocument()
@@ -35,17 +44,19 @@ describe('CoreMetricsDashboard', () => {
expect(await screen.findByText('36.7%')).toBeInTheDocument()
expect(screen.getByText('V1通过 11 / 全部 30')).toBeInTheDocument()
expect(screen.getByText('实时读取')).toBeInTheDocument()
- expect(screen.getAllByText('示例值')).toHaveLength(5)
+ expect(screen.getByText('每小时同步')).toBeInTheDocument()
+ expect(screen.getAllByText('示例值')).toHaveLength(4)
for (const label of ['进行中项目', '本月上新', '按时交付', '首轮通过', '团队负荷', '风险']) {
expect(screen.getByRole('heading', { level: 3, name: label })).toBeInTheDocument()
}
- for (const value of ['42', '18/24', '94.6%', '36.7%', '87%', '4项']) {
+ for (const value of ['2338', '18/24', '94.6%', '36.7%', '87%', '4项']) {
expect(screen.getByText(value)).toBeInTheDocument()
}
- expect(screen.getAllByText('公式待接入')).toHaveLength(5)
+ expect(screen.getAllByText('公式待接入')).toHaveLength(4)
+ expect(screen.getByText('钉钉流程快照')).toBeInTheDocument()
expect(screen.getByText('口径已定义')).toBeInTheDocument()
expect(screen.getAllByText('查看明细')).toHaveLength(6)
const onTimeDelivery = screen.getByRole('link', { name: '查看按时交付明细' })
@@ -66,8 +77,8 @@ describe('CoreMetricsDashboard', () => {
global.fetch = jest.fn().mockResolvedValue({ ok: false, json: async () => ({ success: false }) }) as jest.Mock
render()
- expect(await screen.findByText('读取失败')).toBeInTheDocument()
+ expect(await screen.findAllByText('读取失败')).toHaveLength(2)
expect(screen.queryByText('82.3%')).not.toBeInTheDocument()
- expect(screen.getByText('数据未加载')).toBeInTheDocument()
+ expect(screen.getAllByText('数据未加载')).toHaveLength(2)
})
})
diff --git a/components/department-dashboard/CoreMetricsDashboard.tsx b/components/department-dashboard/CoreMetricsDashboard.tsx
index 672dad3..1146d77 100644
--- a/components/department-dashboard/CoreMetricsDashboard.tsx
+++ b/components/department-dashboard/CoreMetricsDashboard.tsx
@@ -7,6 +7,7 @@ import {
CalendarClock,
} from 'lucide-react'
import type { FirstPassReviewSummary } from '@/lib/work-review-metrics'
+import type { OngoingDingWorkflowSummary } from '@/lib/ding-workflow-sync'
import DepartmentShell from './DepartmentShell'
import { CORE_METRICS, getCoreMetricHref } from './core-metrics'
import OnTimeDeliveryRate from './OnTimeDeliveryRate'
@@ -16,13 +17,31 @@ type FirstPassState =
| { status: 'success'; data: FirstPassReviewSummary }
| { status: 'error'; data: null }
+type OngoingWorkflowState =
+ | { status: 'loading'; data: null }
+ | { status: 'success'; data: OngoingDingWorkflowSummary }
+ | { status: 'error'; data: null }
+
function formatRate(rate: number | null) {
if (rate === null) return '—'
return `${Number.isInteger(rate) ? rate.toFixed(0) : rate.toFixed(1)}%`
}
+function formatSyncTime(value: string | null) {
+ if (!value) return '等待首轮同步'
+ return new Intl.DateTimeFormat('zh-CN', {
+ month: '2-digit',
+ day: '2-digit',
+ hour: '2-digit',
+ minute: '2-digit',
+ hour12: false,
+ timeZone: 'Asia/Shanghai',
+ }).format(new Date(value))
+}
+
export default function CoreMetricsDashboard() {
const [firstPass, setFirstPass] = useState({ status: 'loading', data: null })
+ const [ongoingWorkflows, setOngoingWorkflows] = useState({ status: 'loading', data: null })
useEffect(() => {
const controller = new AbortController()
@@ -41,7 +60,22 @@ export default function CoreMetricsDashboard() {
}
}
+ async function loadOngoingWorkflowSummary() {
+ try {
+ const response = await fetch('/api/department-dashboard/core-metrics/ongoing-projects', {
+ cache: 'no-store',
+ signal: controller.signal,
+ })
+ const payload = await response.json().catch(() => null)
+ if (!response.ok || !payload?.success || !payload.data) throw new Error('指标读取失败')
+ setOngoingWorkflows({ status: 'success', data: payload.data })
+ } catch (error) {
+ if ((error as Error).name !== 'AbortError') setOngoingWorkflows({ status: 'error', data: null })
+ }
+ }
+
void loadFirstPassSummary()
+ void loadOngoingWorkflowSummary()
return () => controller.abort()
}, [])
@@ -73,12 +107,25 @@ export default function CoreMetricsDashboard() {
{CORE_METRICS.map((metric) => {
const Icon = metric.icon
+ const isOngoingWorkflow = metric.key === 'ongoing-projects'
const isFirstPass = metric.key === 'first-pass-rate'
const isOnTimeDelivery = metric.key === 'on-time-delivery'
- const badge = isFirstPass
+ const badge = isOngoingWorkflow
+ ? ongoingWorkflows.status === 'success'
+ ? ongoingWorkflows.data.syncStatus === 'ready'
+ ? '每小时同步'
+ : ongoingWorkflows.data.syncStatus === 'syncing'
+ ? '正在同步'
+ : ongoingWorkflows.data.syncStatus === 'error'
+ ? '同步异常'
+ : '等待首轮同步'
+ : ongoingWorkflows.status === 'error' ? '读取失败' : '正在读取'
+ : isFirstPass
? firstPass.status === 'success' ? '实时读取' : firstPass.status === 'error' ? '读取失败' : '正在读取'
: '示例值'
- const value = isFirstPass
+ const value = isOngoingWorkflow
+ ? ongoingWorkflows.status === 'success' ? String(ongoingWorkflows.data.total) : '—'
+ : isFirstPass
? firstPass.status === 'success' ? formatRate(firstPass.data.rate) : '—'
: metric.exampleValue
@@ -99,11 +146,11 @@ export default function CoreMetricsDashboard() {
{metric.label}
-
+
{isOnTimeDelivery && metric.exampleNumericValue !== undefined
?
: (
-
+
{value}
)}
@@ -115,6 +162,14 @@ export default function CoreMetricsDashboard() {
{isFirstPass && firstPass.status === 'loading' && (
正在连接作品过稿
)}
+ {isOngoingWorkflow && ongoingWorkflows.status === 'success' && (
+
+ 最近同步 {formatSyncTime(ongoingWorkflows.data.lastSyncedAt)}
+
+ )}
+ {isOngoingWorkflow && ongoingWorkflows.status === 'loading' && (
+
正在读取本地流程快照
+ )}
{metric.description}
@@ -122,10 +177,14 @@ export default function CoreMetricsDashboard() {
- {isFirstPass ? (firstPass.status === 'error' ? '刷新页面重试' : '打开页面时读取') : '每日同步'}
+ {isOngoingWorkflow
+ ? ongoingWorkflows.status === 'error' ? '刷新页面重试' : '每小时增量同步'
+ : isFirstPass ? (firstPass.status === 'error' ? '刷新页面重试' : '打开页面时读取') : '每日同步'}
·
- {isFirstPass ? (firstPass.status === 'error' ? '数据未加载' : '口径已定义') : '公式待接入'}
+ {isOngoingWorkflow
+ ? ongoingWorkflows.status === 'error' ? '数据未加载' : '钉钉流程快照'
+ : isFirstPass ? (firstPass.status === 'error' ? '数据未加载' : '口径已定义') : '公式待接入'}
查看明细
@@ -137,7 +196,7 @@ export default function CoreMetricsDashboard() {
})}
- “首轮通过”已读取作品过稿 V1 审核结果;其余示例值将在对应数据源与公式确认后替换。
+ “进行中项目”每小时读取钉钉审批流程快照,“首轮通过”读取作品过稿 V1 审核结果;其余示例值将在对应数据源与公式确认后替换。
diff --git a/components/department-dashboard/core-metrics.ts b/components/department-dashboard/core-metrics.ts
index 273931b..5bf50e0 100644
--- a/components/department-dashboard/core-metrics.ts
+++ b/components/department-dashboard/core-metrics.ts
@@ -32,10 +32,10 @@ export const CORE_METRICS: CoreMetric[] = [
key: 'ongoing-projects',
label: '进行中项目',
exampleValue: '42',
- description: '当前处于进行中状态的项目总量。',
+ description: '当前未完结的钉钉流程总量。',
icon: BriefcaseBusiness,
- detailDescription: '展示当前仍在流转中的项目,并按最近更新时间排序。',
- plannedFields: ['项目名称', '当前状态', '提交人', '负责人', '作品日期', '最近更新'],
+ detailDescription: '展示当前仍在流转中的钉钉审批流程,并按最近更新时间排序。',
+ plannedFields: ['流程名称', '当前状态', '发起部门', '待办任务', '发起时间', '最近更新'],
},
{
key: 'monthly-launches',
diff --git a/docs/生产数据库安全增量迁移SOP-2026-08-12.md b/docs/生产数据库安全增量迁移SOP-2026-08-12.md
index 927fc21..648686c 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:
@@ -13,6 +13,7 @@
- `20260813_add_material_custody_department`
- `20260813_add_organization_units_and_employment_state`
- `20260813_add_user_employee_binding`
+- `20260820_add_ding_workflow_snapshots`
它们新增可空租约字段/索引、修改未来 Logo 默认值、精确把 `system_settings` 品牌两列中的退役 `s3.1080p.hk` 值改为本地品牌图,并把 `materials_uploadId_key` 重建为“`uploadId` 非空且素材未软删除”范围内的唯一 btree 索引;同时新增人员组织树、素材托管字段和账号—人员绑定表,并在同一个事务中按钉钉唯一值做确定性回填。索引迁移会先检测未删除素材中的重复 `uploadId`;人员绑定迁移会先检测一个钉钉身份对应多个账号的冲突。任一预检失败都会直接停止,不会猜测性合并、删除素材或留下一半回填。迁移没有自动 down;执行失败时停止发布,恢复依赖执行前的数据库备份。
@@ -29,7 +30,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。
@@ -353,7 +354,7 @@ docker run --rm \
"$MIGRATOR_IMAGE" --apply
```
-工具会在修改数据库前先验证审计目录可执行“新建→原子改名→删除”,再依次完成:核对镜像 SHA、配置端口与数据库实际端口、核对机器生成的 receipt 和完整备份产物、拒绝未知待执行/失败 migration、执行 `prisma migrate deploy`,最后核对七条 migration 的数据库 checksum、完成状态、关键人员组织表/字段类型与可空性、租约索引的精确列顺序、`materials_uploadId_key` 的唯一性/btree/单列/有效就绪状态/active-only 谓词、Logo 默认值,以及每条现存 Logo 地址均为本地品牌图或当前可信桶地址,并原子写审计 JSON。
+工具会在修改数据库前先验证审计目录可执行“新建→原子改名→删除”,再依次完成:核对镜像 SHA、配置端口与数据库实际端口、核对机器生成的 receipt 和完整备份产物、拒绝未知待执行/失败 migration、执行 `prisma migrate deploy`,最后核对八条目标 migration 的数据库 checksum、完成状态、关键人员组织表、钉钉流程快照表、字段类型与可空性、租约索引的精确列顺序、`materials_uploadId_key` 的唯一性/btree/单列/有效就绪状态/active-only 谓词、Logo 默认值,以及每条现存 Logo 地址均为本地品牌图或当前可信桶地址,并原子写审计 JSON。
apply 成功后立即固定审计文件本身;后续“迁移批次部署”读取这一个文件,不能手工重写:
@@ -369,7 +370,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 33046ff..d677f03 100644
--- a/instrumentation.ts
+++ b/instrumentation.ts
@@ -80,5 +80,15 @@ export async function register() {
} catch (error) {
console.error('[Instrumentation] 调度组织架构每日同步任务失败:', error)
}
+
+ // 钉钉审批流程快照:启动后建立基线,之后每小时增量同步。
+ // 看板只读本地 PostgreSQL 快照,避免用户请求直接扫描上游无索引大表。
+ try {
+ const { ensureHourlyDingWorkflowSyncStarted } = await import('./lib/ding-workflow-sync')
+ ensureHourlyDingWorkflowSyncStarted()
+ console.log('[Instrumentation] 钉钉流程每小时增量同步任务已调度')
+ } catch (error) {
+ console.error('[Instrumentation] 调度钉钉流程每小时同步任务失败:', error)
+ }
}
}
diff --git a/lib/__tests__/ding-workflow-sync.test.ts b/lib/__tests__/ding-workflow-sync.test.ts
new file mode 100644
index 0000000..5a10fb9
--- /dev/null
+++ b/lib/__tests__/ding-workflow-sync.test.ts
@@ -0,0 +1,32 @@
+import {
+ isDingWorkflowOngoing,
+ sourceWorkflowTimestampToIso,
+} from '@/lib/ding-workflow-sync'
+
+jest.mock('mssql', () => ({
+ __esModule: true,
+ default: {},
+}))
+
+jest.mock('@/lib/prisma', () => ({ prisma: {} }))
+jest.mock('@/lib/brand-center-access', () => ({ resolveBrandCenterAccessForUser: jest.fn() }))
+
+describe('ding workflow snapshot sync semantics', () => {
+ test.each([
+ ['RUNNING', null, true],
+ ['NEW', '', true],
+ ['COMPLETED', null, false],
+ ['TERMINATED', null, false],
+ ['CANCELED', null, false],
+ ['RUNNING', '2026-08-20 10:30:00', false],
+ ['', null, false],
+ ])('status %s and finish time %s resolves ongoing=%s', (status, finishTime, expected) => {
+ expect(isDingWorkflowOngoing(status, finishTime)).toBe(expected)
+ })
+
+ it('interprets upstream timestamps as Asia/Shanghai when no zone is present', () => {
+ expect(sourceWorkflowTimestampToIso('2026-08-20 10:30:00')).toBe('2026-08-20T02:30:00.000Z')
+ expect(sourceWorkflowTimestampToIso('2026-08-20T10:30:00Z')).toBe('2026-08-20T10:30:00.000Z')
+ expect(sourceWorkflowTimestampToIso('invalid')).toBeNull()
+ })
+})
diff --git a/lib/ding-workflow-sync.ts b/lib/ding-workflow-sync.ts
new file mode 100644
index 0000000..1ea191d
--- /dev/null
+++ b/lib/ding-workflow-sync.ts
@@ -0,0 +1,617 @@
+import { randomUUID } from 'node:crypto'
+import { hostname } from 'node:os'
+import type { Session } from 'next-auth'
+import mssql, { type ConnectionPool } from 'mssql'
+import { resolveBrandCenterAccessForUser } from '@/lib/brand-center-access'
+import { prisma } from '@/lib/prisma'
+
+const SYNC_STATE_ID = 'ding-workflow-hourly'
+const TERMINAL_WORKFLOW_STATUSES = new Set(['COMPLETED', 'TERMINATED', 'CANCELED'])
+const LEASE_DURATION_MS = 20 * 60 * 1000
+const DEFAULT_SYNC_INTERVAL_MS = 60 * 60 * 1000
+const INITIAL_SYNC_DELAY_MS = 5_000
+const UPSERT_BATCH_SIZE = 750
+
+const WORKFLOW_TABLE = {
+ schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt',
+ table: process.env.REPORT_DB_WORKFLOW_TABLE || 'ding_Workflow',
+}
+const WORKFLOW_SCHEME_TABLE = {
+ schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt',
+ table: process.env.REPORT_DB_WORKFLOW_SCHEME_TABLE || 'ding_WorkflowScheme',
+}
+const WORKFLOW_TASK_TABLE = {
+ schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt',
+ table: process.env.REPORT_DB_WORKFLOW_TASK_TABLE || 'ding_WorkflowTask',
+}
+
+type SourceWorkflowRow = {
+ processInstanceId: unknown
+ businessId: unknown
+ title: unknown
+ result: unknown
+ originatorUserId: unknown
+ originatorDeptId: unknown
+ originatorDeptName: unknown
+ status: unknown
+ bizAction: unknown
+ processCode: unknown
+ flowType: unknown
+ joinedAtRaw: unknown
+ sourceCreatedAtRaw: unknown
+ sourceUpdatedAtRaw: unknown
+ sourceFinishedAtRaw: unknown
+}
+
+type SourceTaskAggregate = {
+ processInstanceId: unknown
+ taskCount: unknown
+ activeTaskCount: unknown
+ latestTaskStatus: unknown
+ latestTaskAtRaw: unknown
+}
+
+type SourceSchemeAggregate = {
+ processInstanceId: unknown
+ schemeFieldCount: unknown
+}
+
+type WorkflowSnapshotPayload = {
+ processInstanceId: string
+ businessId: string | null
+ title: string | null
+ result: string | null
+ originatorUserId: string | null
+ originatorDeptId: string | null
+ originatorDeptName: string | null
+ status: string
+ bizAction: string | null
+ processCode: string | null
+ flowType: string | null
+ joinedAtRaw: string | null
+ sourceCreatedAtRaw: string | null
+ sourceUpdatedAtRaw: string | null
+ sourceFinishedAtRaw: string | null
+ isOngoing: boolean
+ schemeFieldCount: number
+ taskCount: number
+ activeTaskCount: number
+ latestTaskStatus: string | null
+ latestTaskAtRaw: string | null
+}
+
+export type DingWorkflowSyncStatus = 'pending' | 'syncing' | 'ready' | 'error'
+
+export type OngoingDingWorkflowSummary = {
+ total: number
+ lastSyncedAt: string | null
+ syncStatus: DingWorkflowSyncStatus
+}
+
+export type OngoingDingWorkflowListItem = {
+ id: string
+ businessId: string | null
+ title: string
+ status: string
+ originatorUserId: string | null
+ originatorDeptName: string | null
+ sourceCreatedAt: string | null
+ sourceUpdatedAt: string | null
+ activeTaskCount: number
+ taskCount: number
+ schemeFieldCount: number
+ latestTaskStatus: string | null
+}
+
+export type OngoingDingWorkflowPage = {
+ items: OngoingDingWorkflowListItem[]
+ pagination: {
+ page: number
+ pageSize: number
+ total: number
+ totalPages: number
+ }
+ sync: OngoingDingWorkflowSummary
+}
+
+export class DingWorkflowMetricApiError extends Error {
+ status: number
+
+ constructor(message: string, status = 400) {
+ super(message)
+ this.status = status
+ }
+}
+
+function clean(value: unknown): string | null {
+ if (value === null || value === undefined) return null
+ const normalized = String(value).trim()
+ return normalized || null
+}
+
+function nonNegativeInteger(value: unknown): number {
+ const number = Number(value)
+ return Number.isFinite(number) && number > 0 ? Math.floor(number) : 0
+}
+
+function quoteIdentifier(value: string) {
+ return '[' + String(value).replace(/]/g, ']]') + ']'
+}
+
+function tableRef(table: { schema: string; table: string }) {
+ return `${quoteIdentifier(table.schema)}.${quoteIdentifier(table.table)}`
+}
+
+export function isDingWorkflowOngoing(status: unknown, finishedAt: unknown): boolean {
+ const normalizedStatus = String(status || '').trim().toUpperCase()
+ return Boolean(normalizedStatus)
+ && !TERMINAL_WORKFLOW_STATUSES.has(normalizedStatus)
+ && !clean(finishedAt)
+}
+
+export function sourceWorkflowTimestampToIso(value: unknown): string | null {
+ const normalized = clean(value)
+ if (!normalized) return null
+ const hasTimeZone = /(?:Z|[+-]\d{2}:?\d{2})$/i.test(normalized)
+ const candidate = hasTimeZone
+ ? normalized
+ : `${normalized.replace(' ', 'T')}+08:00`
+ const parsed = new Date(candidate)
+ return Number.isNaN(parsed.getTime()) ? null : parsed.toISOString()
+}
+
+function maxSourceCursor(rows: WorkflowSnapshotPayload[], current: string | null) {
+ return rows.reduce((latest, row) => {
+ const candidate = row.sourceUpdatedAtRaw
+ if (!candidate) return latest
+ return !latest || candidate > latest ? candidate : latest
+ }, current)
+}
+
+export function isReportWorkflowDbConfigured() {
+ return Boolean(
+ process.env.REPORT_DB_SERVER
+ && process.env.REPORT_DB_DATABASE
+ && process.env.REPORT_DB_USER
+ && process.env.REPORT_DB_PASSWORD,
+ )
+}
+
+function buildSourceConfig() {
+ return {
+ server: process.env.REPORT_DB_SERVER as string,
+ database: process.env.REPORT_DB_DATABASE as string,
+ user: process.env.REPORT_DB_USER as string,
+ password: process.env.REPORT_DB_PASSWORD as string,
+ port: Number(process.env.REPORT_DB_PORT || 1433),
+ options: {
+ encrypt: process.env.REPORT_DB_ENCRYPT === 'true',
+ trustServerCertificate: process.env.REPORT_DB_TRUST_SERVER_CERTIFICATE !== 'false',
+ },
+ connectionTimeout: 20_000,
+ requestTimeout: 300_000,
+ pool: { max: 3, min: 0, idleTimeoutMillis: 30_000 },
+ }
+}
+
+async function assertDashboardAccess(session: Session | null) {
+ const userId = session?.user?.id
+ if (!userId) throw new DingWorkflowMetricApiError('请先登录', 401)
+
+ const user = await prisma.user.findUnique({
+ where: { id: userId },
+ select: { role: true, isSuperAdmin: true, isActive: true, status: true },
+ })
+ if (!user || !user.isActive || user.status === 'archived' || user.status === 'suspended') {
+ throw new DingWorkflowMetricApiError('账号不可用,请重新登录', 401)
+ }
+
+ const allowed = await resolveBrandCenterAccessForUser(userId, {
+ role: user.role,
+ isSuperAdmin: user.isSuperAdmin,
+ })
+ if (!allowed) throw new DingWorkflowMetricApiError('无权查看品牌中心流程指标', 403)
+}
+
+async function fetchSourceWorkflows(pool: ConnectionPool, cursor: string | null) {
+ const request = pool.request()
+ request.input('cursor', mssql.NVarChar(64), cursor)
+ const result = await request.query(`
+ SELECT
+ CONVERT(nvarchar(255), processInstanceId) AS processInstanceId,
+ CONVERT(nvarchar(255), businessId) AS businessId,
+ CONVERT(nvarchar(max), title) AS title,
+ CONVERT(nvarchar(100), result) AS result,
+ CONVERT(nvarchar(255), originatorUserId) AS originatorUserId,
+ CONVERT(nvarchar(255), originatorDeptId) AS originatorDeptId,
+ CONVERT(nvarchar(500), originatorDeptName) AS originatorDeptName,
+ CONVERT(nvarchar(100), status) AS status,
+ CONVERT(nvarchar(100), bizAction) AS bizAction,
+ CONVERT(nvarchar(255), processCode) AS processCode,
+ CONVERT(nvarchar(100), flowType) AS flowType,
+ CONVERT(nvarchar(64), joinAt) AS joinedAtRaw,
+ CONVERT(nvarchar(64), createAt) AS sourceCreatedAtRaw,
+ CONVERT(nvarchar(64), updateAt) AS sourceUpdatedAtRaw,
+ CONVERT(nvarchar(64), finishTime) AS sourceFinishedAtRaw
+ FROM ${tableRef(WORKFLOW_TABLE)}
+ WHERE
+ @cursor IS NULL
+ OR UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(100), status), '')))) NOT IN ('COMPLETED', 'TERMINATED', 'CANCELED')
+ OR TRY_CONVERT(datetime2, NULLIF(CONVERT(nvarchar(64), updateAt), '')) >=
+ DATEADD(minute, -5, TRY_CONVERT(datetime2, @cursor))
+ `)
+ return result.recordset
+}
+
+async function fetchOngoingAggregates(pool: ConnectionPool, processInstanceIds: string[]) {
+ if (processInstanceIds.length === 0) {
+ return {
+ taskByWorkflow: new Map(),
+ schemeByWorkflow: new Map(),
+ }
+ }
+
+ const request = pool.request()
+ request.multiple = true
+ request.input('workflowIds', mssql.NVarChar(mssql.MAX), JSON.stringify(processInstanceIds))
+ const result = await request.query(`
+ ;WITH workflow_ids AS (
+ SELECT CONVERT(nvarchar(255), [value]) AS processInstanceId
+ FROM OPENJSON(@workflowIds)
+ ), task_rows AS (
+ SELECT
+ CONVERT(nvarchar(255), task.processInstanceId) AS processInstanceId,
+ CONVERT(nvarchar(100), task.status) AS status,
+ CONVERT(nvarchar(64), task.createTime) AS createTime,
+ ROW_NUMBER() OVER (
+ PARTITION BY task.processInstanceId
+ ORDER BY TRY_CONVERT(datetime2, NULLIF(CONVERT(nvarchar(64), task.createTime), '')) DESC, task.id DESC
+ ) AS rowNumber
+ FROM ${tableRef(WORKFLOW_TASK_TABLE)} task
+ INNER JOIN workflow_ids ids
+ ON ids.processInstanceId = CONVERT(nvarchar(255), task.processInstanceId)
+ )
+ SELECT
+ processInstanceId,
+ COUNT_BIG(*) AS taskCount,
+ SUM(CASE WHEN UPPER(LTRIM(RTRIM(COALESCE(status, '')))) NOT IN ('COMPLETED', 'TERMINATED', 'CANCELED') THEN 1 ELSE 0 END) AS activeTaskCount,
+ MAX(CASE WHEN rowNumber = 1 THEN status END) AS latestTaskStatus,
+ MAX(CASE WHEN rowNumber = 1 THEN createTime END) AS latestTaskAtRaw
+ FROM task_rows
+ GROUP BY processInstanceId;
+
+ ;WITH workflow_ids AS (
+ SELECT CONVERT(nvarchar(255), [value]) AS processInstanceId
+ FROM OPENJSON(@workflowIds)
+ )
+ SELECT
+ CONVERT(nvarchar(255), scheme.processInstanceId) AS processInstanceId,
+ COUNT_BIG(*) AS schemeFieldCount
+ FROM ${tableRef(WORKFLOW_SCHEME_TABLE)} scheme
+ INNER JOIN workflow_ids ids
+ ON ids.processInstanceId = CONVERT(nvarchar(255), scheme.processInstanceId)
+ GROUP BY scheme.processInstanceId;
+ `)
+
+ const taskRows = (result.recordsets?.[0] || []) as SourceTaskAggregate[]
+ const schemeRows = (result.recordsets?.[1] || []) as SourceSchemeAggregate[]
+ return {
+ taskByWorkflow: new Map(taskRows.map((row) => [String(row.processInstanceId), row])),
+ schemeByWorkflow: new Map(schemeRows.map((row) => [String(row.processInstanceId), row])),
+ }
+}
+
+function buildSnapshotPayload(
+ sourceRows: SourceWorkflowRow[],
+ taskByWorkflow: Map,
+ schemeByWorkflow: Map,
+): WorkflowSnapshotPayload[] {
+ return sourceRows.flatMap((row) => {
+ const processInstanceId = clean(row.processInstanceId)
+ const status = clean(row.status)
+ if (!processInstanceId || !status) return []
+ const ongoing = isDingWorkflowOngoing(status, row.sourceFinishedAtRaw)
+ const task = taskByWorkflow.get(processInstanceId)
+ const scheme = schemeByWorkflow.get(processInstanceId)
+
+ return [{
+ processInstanceId,
+ businessId: clean(row.businessId),
+ title: clean(row.title),
+ result: clean(row.result),
+ originatorUserId: clean(row.originatorUserId),
+ originatorDeptId: clean(row.originatorDeptId),
+ originatorDeptName: clean(row.originatorDeptName),
+ status: status.toUpperCase(),
+ bizAction: clean(row.bizAction),
+ processCode: clean(row.processCode),
+ flowType: clean(row.flowType),
+ joinedAtRaw: clean(row.joinedAtRaw),
+ sourceCreatedAtRaw: clean(row.sourceCreatedAtRaw),
+ sourceUpdatedAtRaw: clean(row.sourceUpdatedAtRaw),
+ sourceFinishedAtRaw: clean(row.sourceFinishedAtRaw),
+ isOngoing: ongoing,
+ schemeFieldCount: ongoing ? nonNegativeInteger(scheme?.schemeFieldCount) : 0,
+ taskCount: ongoing ? nonNegativeInteger(task?.taskCount) : 0,
+ activeTaskCount: ongoing ? nonNegativeInteger(task?.activeTaskCount) : 0,
+ latestTaskStatus: ongoing ? clean(task?.latestTaskStatus)?.toUpperCase() || null : null,
+ latestTaskAtRaw: ongoing ? clean(task?.latestTaskAtRaw) : null,
+ }]
+ })
+}
+
+async function upsertSnapshotBatch(rows: WorkflowSnapshotPayload[], syncedAt: Date) {
+ if (rows.length === 0) return
+ await prisma.$executeRawUnsafe(`
+ INSERT INTO "ding_workflow_snapshots" (
+ "process_instance_id", "business_id", "title", "result",
+ "originator_user_id", "originator_dept_id", "originator_dept_name",
+ "status", "biz_action", "process_code", "flow_type", "joined_at_raw",
+ "source_created_at_raw", "source_updated_at_raw", "source_finished_at_raw",
+ "is_ongoing", "scheme_field_count", "task_count", "active_task_count",
+ "latest_task_status", "latest_task_at_raw", "synced_at", "updated_at"
+ )
+ SELECT
+ incoming."processInstanceId", incoming."businessId", incoming."title", incoming."result",
+ incoming."originatorUserId", incoming."originatorDeptId", incoming."originatorDeptName",
+ incoming."status", incoming."bizAction", incoming."processCode", incoming."flowType", incoming."joinedAtRaw",
+ incoming."sourceCreatedAtRaw", incoming."sourceUpdatedAtRaw", incoming."sourceFinishedAtRaw",
+ incoming."isOngoing", incoming."schemeFieldCount", incoming."taskCount", incoming."activeTaskCount",
+ incoming."latestTaskStatus", incoming."latestTaskAtRaw", $2::timestamp, $2::timestamp
+ FROM jsonb_to_recordset($1::jsonb) AS incoming(
+ "processInstanceId" text, "businessId" text, "title" text, "result" text,
+ "originatorUserId" text, "originatorDeptId" text, "originatorDeptName" text,
+ "status" text, "bizAction" text, "processCode" text, "flowType" text, "joinedAtRaw" text,
+ "sourceCreatedAtRaw" text, "sourceUpdatedAtRaw" text, "sourceFinishedAtRaw" text,
+ "isOngoing" boolean, "schemeFieldCount" integer, "taskCount" integer, "activeTaskCount" integer,
+ "latestTaskStatus" text, "latestTaskAtRaw" text
+ )
+ ON CONFLICT ("process_instance_id") DO UPDATE SET
+ "business_id" = EXCLUDED."business_id",
+ "title" = EXCLUDED."title",
+ "result" = EXCLUDED."result",
+ "originator_user_id" = EXCLUDED."originator_user_id",
+ "originator_dept_id" = EXCLUDED."originator_dept_id",
+ "originator_dept_name" = EXCLUDED."originator_dept_name",
+ "status" = EXCLUDED."status",
+ "biz_action" = EXCLUDED."biz_action",
+ "process_code" = EXCLUDED."process_code",
+ "flow_type" = EXCLUDED."flow_type",
+ "joined_at_raw" = EXCLUDED."joined_at_raw",
+ "source_created_at_raw" = EXCLUDED."source_created_at_raw",
+ "source_updated_at_raw" = EXCLUDED."source_updated_at_raw",
+ "source_finished_at_raw" = EXCLUDED."source_finished_at_raw",
+ "is_ongoing" = EXCLUDED."is_ongoing",
+ "scheme_field_count" = EXCLUDED."scheme_field_count",
+ "task_count" = EXCLUDED."task_count",
+ "active_task_count" = EXCLUDED."active_task_count",
+ "latest_task_status" = EXCLUDED."latest_task_status",
+ "latest_task_at_raw" = EXCLUDED."latest_task_at_raw",
+ "synced_at" = EXCLUDED."synced_at",
+ "updated_at" = EXCLUDED."updated_at"
+ `, JSON.stringify(rows), syncedAt)
+}
+
+async function acquireSyncLease(leaseOwner: string, startedAt: Date) {
+ await prisma.dingWorkflowSyncState.upsert({
+ where: { id: SYNC_STATE_ID },
+ create: { id: SYNC_STATE_ID },
+ update: {},
+ })
+ const claimed = await prisma.dingWorkflowSyncState.updateMany({
+ where: {
+ id: SYNC_STATE_ID,
+ OR: [
+ { leaseExpiresAt: null },
+ { leaseExpiresAt: { lt: startedAt } },
+ { leaseOwner },
+ ],
+ },
+ data: {
+ status: 'syncing',
+ lastStartedAt: startedAt,
+ lastError: null,
+ leaseOwner,
+ leaseExpiresAt: new Date(startedAt.getTime() + LEASE_DURATION_MS),
+ },
+ })
+ if (claimed.count !== 1) return null
+ return prisma.dingWorkflowSyncState.findUniqueOrThrow({ where: { id: SYNC_STATE_ID } })
+}
+
+export async function syncDingWorkflowSnapshots() {
+ if (!isReportWorkflowDbConfigured()) {
+ throw new Error('未配置上游报表库(REPORT_DB_*),无法同步钉钉流程')
+ }
+
+ const startedAt = new Date()
+ const leaseOwner = `${hostname()}:${process.pid}:${randomUUID()}`
+ const state = await acquireSyncLease(leaseOwner, startedAt)
+ if (!state) return { skipped: true, rowsUpserted: 0, ongoingCount: null }
+
+ const pool = new mssql.ConnectionPool(buildSourceConfig())
+ try {
+ await pool.connect()
+ const sourceRows = await fetchSourceWorkflows(pool, state.sourceCursorRaw)
+ const ongoingIds = sourceRows.flatMap((row) => {
+ const id = clean(row.processInstanceId)
+ return id && isDingWorkflowOngoing(row.status, row.sourceFinishedAtRaw) ? [id] : []
+ })
+ const { taskByWorkflow, schemeByWorkflow } = await fetchOngoingAggregates(pool, ongoingIds)
+ const rows = buildSnapshotPayload(sourceRows, taskByWorkflow, schemeByWorkflow)
+ const syncedAt = new Date()
+
+ for (let index = 0; index < rows.length; index += UPSERT_BATCH_SIZE) {
+ await upsertSnapshotBatch(rows.slice(index, index + UPSERT_BATCH_SIZE), syncedAt)
+ }
+
+ const ongoingCount = await prisma.dingWorkflowSnapshot.count({ where: { isOngoing: true } })
+ const sourceCursorRaw = maxSourceCursor(rows, state.sourceCursorRaw)
+ await prisma.dingWorkflowSyncState.updateMany({
+ where: { id: SYNC_STATE_ID, leaseOwner },
+ data: {
+ status: 'ready',
+ sourceCursorRaw,
+ lastCompletedAt: syncedAt,
+ lastError: null,
+ leaseOwner: null,
+ leaseExpiresAt: null,
+ rowsUpserted: rows.length,
+ currentOngoingCount: ongoingCount,
+ },
+ })
+ return { skipped: false, rowsUpserted: rows.length, ongoingCount }
+ } catch (error) {
+ const message = error instanceof Error ? error.message : String(error)
+ await prisma.dingWorkflowSyncState.updateMany({
+ where: { id: 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)
+ }
+}
+
+function normalizeSyncStatus(state: { status: string; lastCompletedAt: Date | null } | null): DingWorkflowSyncStatus {
+ 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'
+}
+
+async function readSummaryWithoutAccessCheck(): Promise {
+ const [total, state] = await prisma.$transaction([
+ prisma.dingWorkflowSnapshot.count({ where: { isOngoing: true } }),
+ prisma.dingWorkflowSyncState.findUnique({
+ where: { id: SYNC_STATE_ID },
+ select: { status: true, lastCompletedAt: true },
+ }),
+ ])
+ return {
+ total,
+ lastSyncedAt: state?.lastCompletedAt?.toISOString() || null,
+ syncStatus: normalizeSyncStatus(state),
+ }
+}
+
+export async function getOngoingDingWorkflowSummary(session: Session | null) {
+ await assertDashboardAccess(session)
+ return readSummaryWithoutAccessCheck()
+}
+
+export async function getOngoingDingWorkflowProjects(
+ session: Session | null,
+ params?: { page?: number; pageSize?: number; query?: string },
+): Promise {
+ await assertDashboardAccess(session)
+ const page = Math.max(1, Number(params?.page || 1) || 1)
+ const pageSize = Math.min(50, Math.max(1, Number(params?.pageSize || 20) || 20))
+ const query = clean(params?.query)
+ const where = {
+ isOngoing: true,
+ ...(query ? {
+ OR: [
+ { title: { contains: query, mode: 'insensitive' as const } },
+ { businessId: { contains: query, mode: 'insensitive' as const } },
+ { processInstanceId: { contains: query, mode: 'insensitive' as const } },
+ { originatorDeptName: { contains: query, mode: 'insensitive' as const } },
+ { originatorUserId: { contains: query, mode: 'insensitive' as const } },
+ ],
+ } : {}),
+ }
+
+ const [total, rows, sync] = await Promise.all([
+ prisma.dingWorkflowSnapshot.count({ where }),
+ prisma.dingWorkflowSnapshot.findMany({
+ where,
+ select: {
+ processInstanceId: true,
+ businessId: true,
+ title: true,
+ status: true,
+ originatorUserId: true,
+ originatorDeptName: true,
+ sourceCreatedAtRaw: true,
+ sourceUpdatedAtRaw: true,
+ activeTaskCount: true,
+ taskCount: true,
+ schemeFieldCount: true,
+ latestTaskStatus: true,
+ },
+ orderBy: [
+ { sourceUpdatedAtRaw: { sort: 'desc', nulls: 'last' } },
+ { processInstanceId: 'desc' },
+ ],
+ skip: (page - 1) * pageSize,
+ take: pageSize,
+ }),
+ readSummaryWithoutAccessCheck(),
+ ])
+
+ return {
+ items: rows.map((row) => ({
+ id: row.processInstanceId,
+ businessId: row.businessId,
+ title: row.title || '未命名流程',
+ status: row.status,
+ originatorUserId: row.originatorUserId,
+ originatorDeptName: row.originatorDeptName,
+ sourceCreatedAt: sourceWorkflowTimestampToIso(row.sourceCreatedAtRaw),
+ sourceUpdatedAt: sourceWorkflowTimestampToIso(row.sourceUpdatedAtRaw),
+ activeTaskCount: row.activeTaskCount,
+ taskCount: row.taskCount,
+ schemeFieldCount: row.schemeFieldCount,
+ latestTaskStatus: row.latestTaskStatus,
+ })),
+ pagination: {
+ page,
+ pageSize,
+ total,
+ totalPages: Math.max(1, Math.ceil(total / pageSize)),
+ },
+ sync,
+ }
+}
+
+type DingWorkflowSchedulerGlobal = typeof globalThis & {
+ __dingWorkflowHourlySyncStarted?: boolean
+}
+
+export function ensureHourlyDingWorkflowSyncStarted() {
+ const schedulerGlobal = globalThis as DingWorkflowSchedulerGlobal
+ if (schedulerGlobal.__dingWorkflowHourlySyncStarted) return
+ schedulerGlobal.__dingWorkflowHourlySyncStarted = true
+
+ if (!isReportWorkflowDbConfigured()) {
+ console.info('[ding-workflow-sync] 未配置上游报表库,跳过每小时同步')
+ return
+ }
+
+ const configuredHours = Number(process.env.DING_WORKFLOW_SYNC_INTERVAL_HOURS || 1)
+ const intervalMs = Number.isFinite(configuredHours) && configuredHours > 0
+ ? Math.max(5 * 60 * 1000, configuredHours * 60 * 60 * 1000)
+ : DEFAULT_SYNC_INTERVAL_MS
+ const runOnce = () => {
+ void syncDingWorkflowSnapshots()
+ .then((result) => {
+ if (result.skipped) {
+ console.info('[ding-workflow-sync] 另一实例正在同步,本轮跳过')
+ return
+ }
+ console.info(`[ding-workflow-sync] 同步完成:更新 ${result.rowsUpserted} 条,进行中 ${result.ongoingCount} 条`)
+ })
+ .catch((error) => console.warn('[ding-workflow-sync] 同步失败:', error))
+ }
+
+ const initialTimer = setTimeout(runOnce, INITIAL_SYNC_DELAY_MS)
+ initialTimer.unref?.()
+ const interval = setInterval(runOnce, intervalMs)
+ interval.unref?.()
+ console.info(`[ding-workflow-sync] 已调度每小时增量同步,周期 ${Math.round(intervalMs / 60_000)} 分钟`)
+}
diff --git a/lib/env-validation.ts b/lib/env-validation.ts
index 4b708a8..9f4b518 100644
--- a/lib/env-validation.ts
+++ b/lib/env-validation.ts
@@ -94,6 +94,11 @@ const OPTIONAL_ENV_VARS = [
'REPORT_DB_REGISTER_TABLE',
'REPORT_DB_FLOWER_NAME_SCHEMA',
'REPORT_DB_FLOWER_NAME_TABLE',
+ 'REPORT_DB_WORKFLOW_SCHEMA',
+ 'REPORT_DB_WORKFLOW_TABLE',
+ 'REPORT_DB_WORKFLOW_SCHEME_TABLE',
+ 'REPORT_DB_WORKFLOW_TASK_TABLE',
+ 'DING_WORKFLOW_SYNC_INTERVAL_HOURS',
'PERFORMANCE_SURVEY_PROFILE_SYNC_HOURS',
'PERFORMANCE_SURVEY_DAILY_SYNC_INTERVAL_HOURS',
'PERFORMANCE_SURVEY_DAILY_SYNC_INITIAL_DELAY_SECONDS',
diff --git a/prisma/migrations/20260820_add_ding_workflow_snapshots/migration.sql b/prisma/migrations/20260820_add_ding_workflow_snapshots/migration.sql
new file mode 100644
index 0000000..c471044
--- /dev/null
+++ b/prisma/migrations/20260820_add_ding_workflow_snapshots/migration.sql
@@ -0,0 +1,57 @@
+CREATE TABLE "ding_workflow_snapshots" (
+ "process_instance_id" TEXT NOT NULL,
+ "business_id" TEXT,
+ "title" TEXT,
+ "result" TEXT,
+ "originator_user_id" TEXT,
+ "originator_dept_id" TEXT,
+ "originator_dept_name" TEXT,
+ "status" TEXT NOT NULL,
+ "biz_action" TEXT,
+ "process_code" TEXT,
+ "flow_type" TEXT,
+ "joined_at_raw" TEXT,
+ "source_created_at_raw" TEXT,
+ "source_updated_at_raw" TEXT,
+ "source_finished_at_raw" TEXT,
+ "is_ongoing" BOOLEAN NOT NULL DEFAULT false,
+ "scheme_field_count" INTEGER NOT NULL DEFAULT 0,
+ "task_count" INTEGER NOT NULL DEFAULT 0,
+ "active_task_count" INTEGER NOT NULL DEFAULT 0,
+ "latest_task_status" TEXT,
+ "latest_task_at_raw" TEXT,
+ "synced_at" TIMESTAMP(3) NOT NULL,
+ "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ "updated_at" TIMESTAMP(3) NOT NULL,
+
+ CONSTRAINT "ding_workflow_snapshots_pkey" PRIMARY KEY ("process_instance_id")
+);
+
+CREATE TABLE "ding_workflow_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,
+ "current_ongoing_count" INTEGER NOT NULL DEFAULT 0,
+ "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ "updated_at" TIMESTAMP(3) NOT NULL,
+
+ CONSTRAINT "ding_workflow_sync_states_pkey" PRIMARY KEY ("id")
+);
+
+CREATE INDEX "ding_workflow_snapshots_ongoing_updated_idx"
+ ON "ding_workflow_snapshots"("is_ongoing", "source_updated_at_raw");
+
+CREATE INDEX "ding_workflow_snapshots_status_idx"
+ ON "ding_workflow_snapshots"("status");
+
+CREATE INDEX "ding_workflow_snapshots_department_idx"
+ ON "ding_workflow_snapshots"("originator_dept_name");
+
+CREATE INDEX "ding_workflow_sync_states_lease_idx"
+ ON "ding_workflow_sync_states"("lease_expires_at");
diff --git a/prisma/schema.prisma b/prisma/schema.prisma
index 30ee795..1320fa4 100644
--- a/prisma/schema.prisma
+++ b/prisma/schema.prisma
@@ -1277,6 +1277,63 @@ model WorkReviewNotification {
@@map("work_review_notifications")
}
+
+/// 钉钉审批流程的本地只读快照。
+///
+/// 上游 rpt.ding_Workflow / ding_WorkflowTask / ding_WorkflowScheme 均为无索引大表,
+/// 部门看板不得在用户请求链路中直连扫描。后台任务首次建立全量基线,之后每小时
+/// 增量同步主流程,并为全部未完结流程刷新任务与表单字段汇总。
+model DingWorkflowSnapshot {
+ processInstanceId String @id @map("process_instance_id")
+ businessId String? @map("business_id")
+ title String?
+ result String?
+ originatorUserId String? @map("originator_user_id")
+ originatorDeptId String? @map("originator_dept_id")
+ originatorDeptName String? @map("originator_dept_name")
+ status String
+ bizAction String? @map("biz_action")
+ processCode String? @map("process_code")
+ flowType String? @map("flow_type")
+ joinedAtRaw String? @map("joined_at_raw")
+ sourceCreatedAtRaw String? @map("source_created_at_raw")
+ sourceUpdatedAtRaw String? @map("source_updated_at_raw")
+ sourceFinishedAtRaw String? @map("source_finished_at_raw")
+ isOngoing Boolean @default(false) @map("is_ongoing")
+ schemeFieldCount Int @default(0) @map("scheme_field_count")
+ taskCount Int @default(0) @map("task_count")
+ activeTaskCount Int @default(0) @map("active_task_count")
+ latestTaskStatus String? @map("latest_task_status")
+ latestTaskAtRaw String? @map("latest_task_at_raw")
+ syncedAt DateTime @map("synced_at")
+ createdAt DateTime @default(now()) @map("created_at")
+ updatedAt DateTime @updatedAt @map("updated_at")
+
+ @@index([isOngoing, sourceUpdatedAtRaw], map: "ding_workflow_snapshots_ongoing_updated_idx")
+ @@index([status], map: "ding_workflow_snapshots_status_idx")
+ @@index([originatorDeptName], map: "ding_workflow_snapshots_department_idx")
+ @@map("ding_workflow_snapshots")
+}
+
+/// 钉钉审批流程同步游标、租约与可观测状态。
+model DingWorkflowSyncState {
+ 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")
+ currentOngoingCount Int @default(0) @map("current_ongoing_count")
+ createdAt DateTime @default(now()) @map("created_at")
+ updatedAt DateTime @updatedAt @map("updated_at")
+
+ @@index([leaseExpiresAt], map: "ding_workflow_sync_states_lease_idx")
+ @@map("ding_workflow_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 67c57b5..0d0bd24 100644
--- a/scripts/git-incremental-deploy.ps1
+++ b/scripts/git-incremental-deploy.ps1
@@ -1498,6 +1498,7 @@ const expectedMigrationNames = new Set([
'20260813_add_material_custody_department',
'20260813_add_organization_units_and_employment_state',
'20260813_add_user_employee_binding',
+ '20260820_add_ding_workflow_snapshots',
]);
if (migrations.length !== expectedMigrationNames.size
|| migrations.some((item) => !expectedMigrationNames.has(item && item.name)
diff --git a/scripts/safe-production-migrate.mjs b/scripts/safe-production-migrate.mjs
index bc32f20..3857f04 100644
--- a/scripts/safe-production-migrate.mjs
+++ b/scripts/safe-production-migrate.mjs
@@ -1,6 +1,6 @@
import { createHash } from 'node:crypto'
import { spawnSync } from 'node:child_process'
-import { createReadStream } from 'node:fs'
+import { createReadStream, realpathSync } from 'node:fs'
import { lstat, mkdir, readFile, readdir, rename, unlink, writeFile } from 'node:fs/promises'
import path from 'node:path'
import { fileURLToPath } from 'node:url'
@@ -16,6 +16,7 @@ const TARGET_MIGRATIONS = Object.freeze([
'20260813_add_material_custody_department',
'20260813_add_organization_units_and_employment_state',
'20260813_add_user_employee_binding',
+ '20260820_add_ding_workflow_snapshots',
])
const VERIFIED_BACKUP_GENERATOR = 'scripts/create-verified-production-backup.mjs'
const VERIFIED_BACKUP_MAINTENANCE_CONFIRMATION = 'NO_WRITES_DURING_SNAPSHOT'
@@ -256,7 +257,9 @@ async function queryEvidence(prisma, localMigrations) {
OR (table_name = 'system_settings' AND column_name IN ('headerLogoUrl', 'loginLogoUrl', 'logoUrl'))
OR (table_name = 'organization_units' AND column_name IN ('id', 'source_dept_id', 'parent_id', 'path', 'level', 'is_active'))
OR (table_name = 'performance_survey_employee_profiles' AND column_name IN ('employment_state', 'resigned_at', 'handover_dingtalk_user_id'))
- OR (table_name = 'user_employee_bindings' AND column_name IN ('id', 'user_id', 'employee_profile_id', 'binding_source', 'status')))
+ OR (table_name = 'user_employee_bindings' AND column_name IN ('id', 'user_id', 'employee_profile_id', 'binding_source', 'status'))
+ OR (table_name = 'ding_workflow_snapshots' AND column_name IN ('process_instance_id', 'status', 'is_ongoing', 'source_updated_at_raw', 'scheme_field_count', 'task_count', 'active_task_count', 'synced_at'))
+ OR (table_name = 'ding_workflow_sync_states' AND column_name IN ('id', 'status', 'source_cursor_raw', 'last_completed_at', 'lease_owner', 'lease_expires_at', 'current_ongoing_count')))
ORDER BY table_name, column_name`,
)
const indexes = await prisma.$queryRawUnsafe(
@@ -283,7 +286,9 @@ async function queryEvidence(prisma, localMigrations) {
AND idx.relname IN (
'materials_transcode_lease_idx',
'materials_uploadId_key',
- 'product_package_upload_jobs_lease_idx'
+ 'product_package_upload_jobs_lease_idx',
+ 'ding_workflow_snapshots_ongoing_updated_idx',
+ 'ding_workflow_sync_states_lease_idx'
)
ORDER BY idx.relname`,
)
@@ -349,6 +354,21 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) {
requireColumn('user_employee_bindings', 'employee_profile_id', 'text', 'NO')
requireColumn('user_employee_bindings', 'binding_source', 'text', 'NO')
requireColumn('user_employee_bindings', 'status', 'text', 'NO')
+ requireColumn('ding_workflow_snapshots', 'process_instance_id', 'text', 'NO')
+ requireColumn('ding_workflow_snapshots', 'status', 'text', 'NO')
+ requireColumn('ding_workflow_snapshots', 'is_ongoing', 'boolean', 'NO')
+ requireColumn('ding_workflow_snapshots', 'source_updated_at_raw', 'text', 'YES')
+ requireColumn('ding_workflow_snapshots', 'scheme_field_count', 'integer', 'NO')
+ requireColumn('ding_workflow_snapshots', 'task_count', 'integer', 'NO')
+ requireColumn('ding_workflow_snapshots', 'active_task_count', 'integer', 'NO')
+ requireColumn('ding_workflow_snapshots', 'synced_at', 'timestamp without time zone', 'NO')
+ requireColumn('ding_workflow_sync_states', 'id', 'text', 'NO')
+ requireColumn('ding_workflow_sync_states', 'status', 'text', 'NO')
+ requireColumn('ding_workflow_sync_states', 'source_cursor_raw', 'text', 'YES')
+ requireColumn('ding_workflow_sync_states', 'last_completed_at', 'timestamp without time zone', 'YES')
+ requireColumn('ding_workflow_sync_states', 'lease_owner', 'text', 'YES')
+ requireColumn('ding_workflow_sync_states', 'lease_expires_at', 'timestamp without time zone', 'YES')
+ requireColumn('ding_workflow_sync_states', 'current_ongoing_count', 'integer', 'NO')
requireColumn('system_settings', 'headerLogoUrl', 'text', 'NO')
requireColumn('system_settings', 'loginLogoUrl', 'text', 'NO')
@@ -382,6 +402,16 @@ export function assertTargetMigrationEvidence(localMigrations, evidence) {
assertExactActiveUploadIdIndex(
evidence.indexes.find((row) => String(row.indexname) === 'materials_uploadId_key'),
)
+ assertExactBtreeIndex(
+ indexByName.get('ding_workflow_snapshots_ongoing_updated_idx'),
+ 'ding_workflow_snapshots',
+ ['is_ongoing', 'source_updated_at_raw'],
+ )
+ assertExactBtreeIndex(
+ indexByName.get('ding_workflow_sync_states_lease_idx'),
+ 'ding_workflow_sync_states',
+ ['lease_expires_at'],
+ )
const isTrustedBrandValue = (value, localDefault) => {
const normalized = String(value || '').trim()
@@ -756,7 +786,10 @@ async function main() {
else process.stdout.write('Read-only production migration audit completed.\n')
}
-if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
+if (
+ process.argv[1]
+ && realpathSync(process.argv[1]) === realpathSync(fileURLToPath(import.meta.url))
+) {
main().catch((error) => {
process.stderr.write(`[SAFE_MIGRATION_ERROR] ${error instanceof Error ? error.message : String(error)}\n`)
process.exitCode = 1
diff --git a/scripts/safe-production-migrate.test.mjs b/scripts/safe-production-migrate.test.mjs
index 1dbea0d..e5cf452 100644
--- a/scripts/safe-production-migrate.test.mjs
+++ b/scripts/safe-production-migrate.test.mjs
@@ -35,18 +35,22 @@ test('entrypoint validates only user arguments and still rejects unknown flags',
cpSync(new URL('./safe-production-migrate.mjs', import.meta.url), path.join(scripts, 'safe-production-migrate.mjs'))
const commit = 'f'.repeat(40)
writeFileSync(path.join(fixture, 'DEPLOYMENT_COMMIT'), `${commit}\n`)
+ // Node 24 会把内建 test runner 的子进程标记继续传给 spawn;迁移入口必须按普通脚本执行。
+ const childEnv = Object.fromEntries(
+ Object.entries(process.env).filter(([name]) => !name.startsWith('NODE_TEST_') && name !== 'NODE_V8_COVERAGE'),
+ )
const run = (args) => spawnSync(
process.execPath,
[path.join(scripts, 'safe-production-migrate.mjs'), ...args],
{
cwd: fixture,
- env: { ...process.env, EXPECTED_COMMIT_SHA: commit, DATABASE_URL: '' },
+ env: { ...childEnv, EXPECTED_COMMIT_SHA: commit, DATABASE_URL: '' },
encoding: 'utf8',
},
)
for (const args of [['--audit'], ['--apply']]) {
const result = run(args)
- assert.notEqual(result.status, 0)
+ assert.notEqual(result.status, 0, `stdout=${result.stdout}\nstderr=${result.stderr}`)
assert.match(result.stderr, /DATABASE_URL is required/)
assert.doesNotMatch(result.stderr, /Only --audit/)
}
@@ -328,6 +332,7 @@ test('target evidence rejects nullable logo columns and structurally wrong index
['20260813_add_material_custody_department', { checksum: 'e' }],
['20260813_add_organization_units_and_employment_state', { checksum: 'f' }],
['20260813_add_user_employee_binding', { checksum: 'g' }],
+ ['20260820_add_ding_workflow_snapshots', { checksum: 'h' }],
])
const evidence = {
migrationRows: [
@@ -338,6 +343,7 @@ test('target evidence rejects nullable logo columns and structurally wrong index
{ migration_name: '20260813_add_material_custody_department', checksum: 'e', finished_at: 'now', rolled_back_at: null },
{ migration_name: '20260813_add_organization_units_and_employment_state', checksum: 'f', finished_at: 'now', rolled_back_at: null },
{ migration_name: '20260813_add_user_employee_binding', checksum: 'g', finished_at: 'now', rolled_back_at: null },
+ { migration_name: '20260820_add_ding_workflow_snapshots', checksum: 'h', finished_at: 'now', rolled_back_at: null },
],
columns: [
{ table_name: 'materials', column_name: 'transcode_lease_token', data_type: 'text', is_nullable: 'YES' },
@@ -363,6 +369,21 @@ test('target evidence rejects nullable logo columns and structurally wrong index
{ table_name: 'user_employee_bindings', column_name: 'employee_profile_id', data_type: 'text', is_nullable: 'NO' },
{ table_name: 'user_employee_bindings', column_name: 'binding_source', data_type: 'text', is_nullable: 'NO' },
{ table_name: 'user_employee_bindings', column_name: 'status', data_type: 'text', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'process_instance_id', data_type: 'text', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'status', data_type: 'text', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'is_ongoing', data_type: 'boolean', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'source_updated_at_raw', data_type: 'text', is_nullable: 'YES' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'scheme_field_count', data_type: 'integer', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'task_count', data_type: 'integer', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'active_task_count', data_type: 'integer', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_snapshots', column_name: 'synced_at', data_type: 'timestamp without time zone', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'id', data_type: 'text', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'status', data_type: 'text', is_nullable: 'NO' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'source_cursor_raw', data_type: 'text', is_nullable: 'YES' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'last_completed_at', data_type: 'timestamp without time zone', is_nullable: 'YES' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'lease_owner', data_type: 'text', is_nullable: 'YES' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'lease_expires_at', data_type: 'timestamp without time zone', is_nullable: 'YES' },
+ { table_name: 'ding_workflow_sync_states', column_name: 'current_ongoing_count', data_type: 'integer', is_nullable: 'NO' },
{ 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" },
@@ -382,6 +403,8 @@ test('target evidence rejects nullable logo columns and structurally wrong index
predicate: '(("uploadId" IS NOT NULL) AND ("deletedAt" IS NULL))',
},
{ indexname: 'product_package_upload_jobs_lease_idx', indexdef: 'CREATE INDEX product_package_upload_jobs_lease_idx ON public.product_package_upload_jobs USING btree (status, lease_expires_at)' },
+ { indexname: 'ding_workflow_snapshots_ongoing_updated_idx', indexdef: 'CREATE INDEX ding_workflow_snapshots_ongoing_updated_idx ON public.ding_workflow_snapshots USING btree (is_ongoing, source_updated_at_raw)' },
+ { indexname: 'ding_workflow_sync_states_lease_idx', indexdef: 'CREATE INDEX ding_workflow_sync_states_lease_idx ON public.ding_workflow_sync_states USING btree (lease_expires_at)' },
],
brandRows: [
{ id: 'local', headerLogoUrl: '/brand/header-logo.png', loginLogoUrl: '/brand/login-logo.png' },
@@ -486,6 +509,7 @@ test('container supply chain and migrator image are immutable while SOP keeps co
'20260813_add_material_custody_department',
'20260813_add_organization_units_and_employment_state',
'20260813_add_user_employee_binding',
+ '20260820_add_ding_workflow_snapshots',
]) {
assert.match(dockerfile, new RegExp(migrationName))
assert.match(migratorSource, new RegExp(`'${migrationName}'`))
diff --git a/types/mssql.d.ts b/types/mssql.d.ts
index 3457511..e08706b 100644
--- a/types/mssql.d.ts
+++ b/types/mssql.d.ts
@@ -1,22 +1,28 @@
declare module 'mssql' {
export interface Request {
+ multiple: boolean
input(name: string, type: unknown, value: unknown): Request
- query(sql: string): Promise<{ recordset: T[] }>
+ query(sql: string): Promise<{ recordset: T[]; recordsets?: any[][] }>
}
- export interface ConnectionPool {
+ export class ConnectionPool {
+ constructor(config: unknown)
+ connect(): Promise
request(): Request
close(): Promise
}
export const Int: unknown
+ export const MAX: number
export function NVarChar(length?: number): unknown
export function connect(config: unknown): Promise
const sql: {
Int: typeof Int
+ MAX: typeof MAX
NVarChar: typeof NVarChar
connect: typeof connect
+ ConnectionPool: typeof ConnectionPool
}
export default sql