1130 lines
42 KiB
TypeScript
1130 lines
42 KiB
TypeScript
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 { isBrandCenterOrganizationPath } from '@/lib/brand-center-access-policy'
|
||
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
|
||
|
||
export const DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL = '2026-04-01 00:00:00'
|
||
export const DING_WORKFLOW_STARTED_AT_CUTOFF_ISO = '2026-03-31T16:00:00.000Z'
|
||
|
||
export const DING_WORKFLOW_ALLOWED_PROCESS_CODES = [
|
||
'PROC-3571C915-0506-4490-A91A-8AA7EA33F838',
|
||
'PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673',
|
||
] as const
|
||
export const DING_WORKFLOW_BRANDS = ['亿家达', '卓禾', '欧意朗', '顾家', '索菲亚', '朴美'] as const
|
||
const ALLOWED_PROCESS_CODE_VALUES: string[] = [...DING_WORKFLOW_ALLOWED_PROCESS_CODES]
|
||
const ALLOWED_PROCESS_CODE_SET = new Set<string>(DING_WORKFLOW_ALLOWED_PROCESS_CODES)
|
||
const PROCESS_TYPE_LABELS: Record<string, string> = {
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[0]]: '美术设计流程',
|
||
[DING_WORKFLOW_ALLOWED_PROCESS_CODES[1]]: '内容视频和图片拍摄',
|
||
}
|
||
|
||
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 SourceTaskRow = {
|
||
processInstanceId: unknown
|
||
taskId: unknown
|
||
activityId: unknown
|
||
userId: unknown
|
||
status: unknown
|
||
createTime: unknown
|
||
}
|
||
|
||
type SourceSchemeAggregate = {
|
||
processInstanceId: unknown
|
||
schemeFieldCount: unknown
|
||
}
|
||
|
||
type SourceSchemeField = {
|
||
processInstanceId: unknown
|
||
componentType: unknown
|
||
name: unknown
|
||
value: unknown
|
||
extValue: unknown
|
||
}
|
||
|
||
type ContactAssignment = {
|
||
userId: string
|
||
name: string | null
|
||
nodeName: string
|
||
}
|
||
|
||
export type OngoingDingWorkflowActiveTask = {
|
||
taskId: string
|
||
activityId: string | null
|
||
nodeName: string
|
||
assigneeUserId: string | null
|
||
assigneeName: string
|
||
assigneeDepartment: string | null
|
||
isBrandCenter: boolean
|
||
createdAt: string | null
|
||
}
|
||
|
||
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
|
||
brand: string
|
||
requesterName: string
|
||
requesterDepartment: string
|
||
projectType: string
|
||
activeTasks: OngoingDingWorkflowActiveTask[]
|
||
}
|
||
|
||
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
|
||
processCode: string
|
||
projectType: string
|
||
brand: string
|
||
requesterName: string
|
||
requesterDepartment: string
|
||
activeTasks: OngoingDingWorkflowActiveTask[]
|
||
}
|
||
|
||
export type OngoingDingWorkflowPage = {
|
||
items: OngoingDingWorkflowListItem[]
|
||
pagination: {
|
||
page: number
|
||
pageSize: number
|
||
total: number
|
||
totalPages: number
|
||
}
|
||
sync: OngoingDingWorkflowSummary
|
||
}
|
||
|
||
export type OngoingDingWorkflowInsights = {
|
||
items: OngoingDingWorkflowListItem[]
|
||
total: 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
|
||
}
|
||
|
||
export function normalizeDingWorkflowProcessCode(value: unknown): string | null {
|
||
return clean(value)?.toUpperCase() || null
|
||
}
|
||
|
||
export function isAllowedDingWorkflowProcessCode(value: unknown): boolean {
|
||
const processCode = normalizeDingWorkflowProcessCode(value)
|
||
return Boolean(processCode && ALLOWED_PROCESS_CODE_SET.has(processCode))
|
||
}
|
||
|
||
export function getDingWorkflowProjectType(value: unknown): string {
|
||
const processCode = normalizeDingWorkflowProcessCode(value)
|
||
return processCode ? PROCESS_TYPE_LABELS[processCode] || '其他流程' : '其他流程'
|
||
}
|
||
|
||
export function normalizeDingWorkflowBrand(value: unknown): string {
|
||
const normalized = clean(value)
|
||
if (!normalized) return '其他品牌'
|
||
return DING_WORKFLOW_BRANDS.find((brand) => normalized.includes(brand)) || '其他品牌'
|
||
}
|
||
|
||
function collectDingWorkflowDepartmentNames(value: unknown): string[] {
|
||
if (value === null || value === undefined) return []
|
||
if (Array.isArray(value)) {
|
||
return value.flatMap(collectDingWorkflowDepartmentNames)
|
||
}
|
||
if (typeof value === 'object') {
|
||
const record = value as Record<string, unknown>
|
||
return ['name', 'label', 'value', 'text', 'title']
|
||
.flatMap((key) => collectDingWorkflowDepartmentNames(record[key]))
|
||
}
|
||
|
||
const normalized = clean(value)
|
||
if (!normalized) return []
|
||
if (normalized.startsWith('[') || normalized.startsWith('{')) {
|
||
try {
|
||
return collectDingWorkflowDepartmentNames(JSON.parse(normalized))
|
||
} catch {
|
||
// Keep non-JSON department names unchanged.
|
||
}
|
||
}
|
||
return [normalized]
|
||
}
|
||
|
||
export function normalizeDingWorkflowDepartment(value: unknown): string | null {
|
||
const names = Array.from(new Set(collectDingWorkflowDepartmentNames(value)))
|
||
return names.length > 0 ? names.join(' / ') : null
|
||
}
|
||
|
||
function parseJson(value: unknown): unknown {
|
||
const normalized = clean(value)
|
||
if (!normalized) return null
|
||
try {
|
||
return JSON.parse(normalized)
|
||
} catch {
|
||
return null
|
||
}
|
||
}
|
||
|
||
function collectContactAssignments(value: unknown, fallbackNodeName: string): ContactAssignment[] {
|
||
if (!value) return []
|
||
if (Array.isArray(value)) {
|
||
return value.flatMap((entry) => collectContactAssignments(entry, fallbackNodeName))
|
||
}
|
||
if (typeof value !== 'object') return []
|
||
|
||
const record = value as Record<string, unknown>
|
||
const userId = clean(record.emplId) || clean(record.userId)
|
||
const contactName = clean(record.name) || clean(record.value)
|
||
const currentNodeName = clean(record.label) || fallbackNodeName
|
||
const direct = userId ? [{ userId, name: contactName, nodeName: currentNodeName }] : []
|
||
const rowValues = Array.isArray(record.rowValue)
|
||
? record.rowValue.flatMap((entry) => {
|
||
if (!entry || typeof entry !== 'object') return []
|
||
const cell = entry as Record<string, unknown>
|
||
const nodeName = clean(cell.label) || fallbackNodeName
|
||
return [cell.extendValue, cell.value]
|
||
.flatMap((candidate) => collectContactAssignments(candidate, nodeName))
|
||
})
|
||
: []
|
||
return [...direct, ...rowValues]
|
||
}
|
||
|
||
function extractContactAssignments(fields: SourceSchemeField[]): ContactAssignment[] {
|
||
const assignments = fields.flatMap((field) => {
|
||
const componentType = clean(field.componentType)
|
||
const nodeName = clean(field.name) || '当前处理节点'
|
||
if (componentType === 'InnerContactField') {
|
||
return [parseJson(field.extValue), parseJson(field.value)]
|
||
.flatMap((value) => collectContactAssignments(value, nodeName))
|
||
}
|
||
if (componentType === 'TableField') {
|
||
return collectContactAssignments(parseJson(field.value), nodeName)
|
||
}
|
||
return []
|
||
})
|
||
|
||
return Array.from(new Map(assignments.map((assignment) => [
|
||
`${assignment.userId}:${assignment.nodeName}`,
|
||
assignment,
|
||
])).values())
|
||
}
|
||
|
||
function firstSchemeValue(fields: SourceSchemeField[], names: string[]): string | null {
|
||
for (const name of names) {
|
||
const value = fields.find((field) => clean(field.name) === name && clean(field.value))?.value
|
||
const normalized = clean(value)
|
||
if (normalized) return normalized
|
||
}
|
||
return null
|
||
}
|
||
|
||
function titleOriginatorName(value: unknown): string | null {
|
||
return clean(value)?.match(/^(.+?)提交的/)?.[1]?.trim() || null
|
||
}
|
||
|
||
function normalizeActiveTasks(value: unknown): OngoingDingWorkflowActiveTask[] {
|
||
if (!Array.isArray(value)) return []
|
||
return value.flatMap((task) => {
|
||
if (!task || typeof task !== 'object') return []
|
||
const row = task as Record<string, unknown>
|
||
const taskId = clean(row.taskId)
|
||
if (!taskId) return []
|
||
return [{
|
||
taskId,
|
||
activityId: clean(row.activityId),
|
||
nodeName: clean(row.nodeName) || '当前处理节点',
|
||
assigneeUserId: clean(row.assigneeUserId),
|
||
assigneeName: clean(row.assigneeName) || '待同步人员',
|
||
assigneeDepartment: clean(row.assigneeDepartment),
|
||
isBrandCenter: row.isBrandCenter === true,
|
||
createdAt: clean(row.createdAt),
|
||
}]
|
||
})
|
||
}
|
||
|
||
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()
|
||
}
|
||
|
||
export function isDingWorkflowStartedOnOrAfterCutoff(value: unknown): boolean {
|
||
const startedAt = sourceWorkflowTimestampToIso(value)
|
||
return Boolean(startedAt && startedAt >= DING_WORKFLOW_STARTED_AT_CUTOFF_ISO)
|
||
}
|
||
|
||
function maxSourceCursor(rows: WorkflowSnapshotPayload[], current: string | null) {
|
||
return rows.reduce<string | null>((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)
|
||
request.input('startedAtCutoff', mssql.NVarChar(64), DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL)
|
||
DING_WORKFLOW_ALLOWED_PROCESS_CODES.forEach((processCode, index) => {
|
||
request.input(`allowedProcessCode${index}`, mssql.NVarChar(255), processCode)
|
||
})
|
||
const result = await request.query<SourceWorkflowRow>(`
|
||
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
|
||
UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(255), processCode), '')))) IN (
|
||
@allowedProcessCode0,
|
||
@allowedProcessCode1
|
||
)
|
||
AND TRY_CONVERT(datetime2, NULLIF(CONVERT(nvarchar(64), createAt), '')) >=
|
||
TRY_CONVERT(datetime2, @startedAtCutoff)
|
||
AND (
|
||
@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<string, SourceTaskAggregate>(),
|
||
taskRowsByWorkflow: new Map<string, SourceTaskRow[]>(),
|
||
schemeByWorkflow: new Map<string, SourceSchemeAggregate>(),
|
||
schemeRowsByWorkflow: new Map<string, SourceSchemeField[]>(),
|
||
}
|
||
}
|
||
|
||
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)
|
||
)
|
||
SELECT
|
||
CONVERT(nvarchar(255), task.processInstanceId) AS processInstanceId,
|
||
CONVERT(nvarchar(255), task.taskId) AS taskId,
|
||
CONVERT(nvarchar(255), task.activityId) AS activityId,
|
||
CONVERT(nvarchar(255), task.userId) AS userId,
|
||
CONVERT(nvarchar(100), task.status) AS status,
|
||
CONVERT(nvarchar(64), task.createTime) AS createTime,
|
||
ROW_NUMBER() OVER (
|
||
PARTITION BY task.processInstanceId, task.taskId
|
||
ORDER BY task.id DESC
|
||
) AS duplicateRow,
|
||
task.id AS sourceRowId
|
||
INTO #workflow_task_rows
|
||
FROM ${tableRef(WORKFLOW_TASK_TABLE)} task
|
||
INNER JOIN workflow_ids ids
|
||
ON ids.processInstanceId = CONVERT(nvarchar(255), task.processInstanceId)
|
||
;
|
||
|
||
;WITH ranked_tasks AS (
|
||
SELECT *, ROW_NUMBER() OVER (
|
||
PARTITION BY processInstanceId
|
||
ORDER BY TRY_CONVERT(datetime2, NULLIF(createTime, '')) DESC, sourceRowId DESC
|
||
) AS latestRow
|
||
FROM #workflow_task_rows
|
||
WHERE duplicateRow = 1
|
||
)
|
||
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 latestRow = 1 THEN status END) AS latestTaskStatus,
|
||
MAX(CASE WHEN latestRow = 1 THEN createTime END) AS latestTaskAtRaw
|
||
FROM ranked_tasks
|
||
GROUP BY processInstanceId;
|
||
|
||
SELECT processInstanceId, taskId, activityId, userId, status, createTime
|
||
FROM #workflow_task_rows
|
||
WHERE duplicateRow = 1;
|
||
|
||
;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;
|
||
|
||
;WITH workflow_ids AS (
|
||
SELECT CONVERT(nvarchar(255), [value]) AS processInstanceId
|
||
FROM OPENJSON(@workflowIds)
|
||
)
|
||
SELECT
|
||
CONVERT(nvarchar(255), scheme.processInstanceId) AS processInstanceId,
|
||
CONVERT(nvarchar(500), scheme.componentType) AS componentType,
|
||
CONVERT(nvarchar(2000), scheme.name) AS name,
|
||
CONVERT(nvarchar(max), scheme.value) AS value,
|
||
CONVERT(nvarchar(max), scheme.extValue) AS extValue
|
||
FROM ${tableRef(WORKFLOW_SCHEME_TABLE)} scheme
|
||
INNER JOIN workflow_ids ids
|
||
ON ids.processInstanceId = CONVERT(nvarchar(255), scheme.processInstanceId)
|
||
WHERE
|
||
LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(2000), scheme.name), ''))) IN ('品牌', '需求部门', '需求人', '任务类型')
|
||
OR CONVERT(nvarchar(500), scheme.componentType) IN ('InnerContactField', 'TableField');
|
||
`)
|
||
|
||
const taskRows = (result.recordsets?.[0] || []) as SourceTaskAggregate[]
|
||
const activeTaskRows = (result.recordsets?.[1] || []) as SourceTaskRow[]
|
||
const schemeRows = (result.recordsets?.[2] || []) as SourceSchemeAggregate[]
|
||
const schemeFieldRows = (result.recordsets?.[3] || []) as SourceSchemeField[]
|
||
const groupByWorkflow = <T extends { processInstanceId: unknown }>(rows: T[]) => {
|
||
const grouped = new Map<string, T[]>()
|
||
rows.forEach((row) => {
|
||
const processInstanceId = clean(row.processInstanceId)
|
||
if (!processInstanceId) return
|
||
grouped.set(processInstanceId, [...(grouped.get(processInstanceId) || []), row])
|
||
})
|
||
return grouped
|
||
}
|
||
return {
|
||
taskByWorkflow: new Map(taskRows.map((row) => [String(row.processInstanceId), row])),
|
||
taskRowsByWorkflow: groupByWorkflow(activeTaskRows),
|
||
schemeByWorkflow: new Map(schemeRows.map((row) => [String(row.processInstanceId), row])),
|
||
schemeRowsByWorkflow: groupByWorkflow(schemeFieldRows),
|
||
}
|
||
}
|
||
|
||
function sourceTaskIsOngoing(status: unknown): boolean {
|
||
const normalized = clean(status)?.toUpperCase()
|
||
return Boolean(normalized && !TERMINAL_WORKFLOW_STATUSES.has(normalized))
|
||
}
|
||
|
||
type EmployeeDirectoryEntry = {
|
||
name: string
|
||
departmentName: string | null
|
||
isBrandCenter: boolean
|
||
}
|
||
|
||
async function readEmployeeDirectory(
|
||
sourceRows: SourceWorkflowRow[],
|
||
taskRowsByWorkflow: Map<string, SourceTaskRow[]>,
|
||
) {
|
||
const ids = Array.from(new Set([
|
||
...sourceRows.map((row) => clean(row.originatorUserId)),
|
||
...Array.from(taskRowsByWorkflow.values()).flatMap((rows) => rows.map((row) => clean(row.userId))),
|
||
].filter((value): value is string => Boolean(value) && value !== 'bpms_system')))
|
||
if (ids.length === 0) return new Map<string, EmployeeDirectoryEntry>()
|
||
const profiles = await prisma.performanceSurveyEmployeeProfile.findMany({
|
||
where: { dingtalkUserId: { in: ids } },
|
||
select: {
|
||
dingtalkUserId: true,
|
||
displayName: true,
|
||
departmentName: true,
|
||
departmentPath: true,
|
||
employmentState: true,
|
||
departmentMemberships: { select: { departmentPath: true } },
|
||
},
|
||
})
|
||
return new Map(profiles.flatMap((profile) => (
|
||
profile.dingtalkUserId ? [[profile.dingtalkUserId, {
|
||
name: profile.displayName,
|
||
departmentName: profile.departmentName,
|
||
isBrandCenter: profile.employmentState !== 'RESIGNED' && [
|
||
profile.departmentPath,
|
||
...profile.departmentMemberships.map((membership) => membership.departmentPath),
|
||
].some(isBrandCenterOrganizationPath),
|
||
}] as const] : []
|
||
)))
|
||
}
|
||
|
||
function inferActivityNodeNames(
|
||
sourceRows: SourceWorkflowRow[],
|
||
taskRowsByWorkflow: Map<string, SourceTaskRow[]>,
|
||
schemeRowsByWorkflow: Map<string, SourceSchemeField[]>,
|
||
) {
|
||
const processCodeByWorkflow = new Map(sourceRows.flatMap((row) => {
|
||
const id = clean(row.processInstanceId)
|
||
const code = normalizeDingWorkflowProcessCode(row.processCode)
|
||
return id && code ? [[id, code] as const] : []
|
||
}))
|
||
const votes = new Map<string, Map<string, number>>()
|
||
|
||
taskRowsByWorkflow.forEach((taskRows, processInstanceId) => {
|
||
const processCode = processCodeByWorkflow.get(processInstanceId)
|
||
if (!processCode) return
|
||
const assignments = extractContactAssignments(schemeRowsByWorkflow.get(processInstanceId) || [])
|
||
taskRows.forEach((task) => {
|
||
const activityId = clean(task.activityId)
|
||
const userId = clean(task.userId)
|
||
if (!activityId || !userId) return
|
||
const nodeNames = Array.from(new Set(
|
||
assignments.filter((assignment) => assignment.userId === userId).map((assignment) => assignment.nodeName),
|
||
))
|
||
if (nodeNames.length !== 1) return
|
||
const key = `${processCode}:${activityId}`
|
||
const activityVotes = votes.get(key) || new Map<string, number>()
|
||
activityVotes.set(nodeNames[0], (activityVotes.get(nodeNames[0]) || 0) + 1)
|
||
votes.set(key, activityVotes)
|
||
})
|
||
})
|
||
|
||
return new Map(Array.from(votes.entries()).map(([key, activityVotes]) => {
|
||
const [nodeName] = Array.from(activityVotes.entries()).sort((a, b) => b[1] - a[1])[0]
|
||
return [key, nodeName]
|
||
}))
|
||
}
|
||
|
||
function buildSnapshotPayload(
|
||
sourceRows: SourceWorkflowRow[],
|
||
taskByWorkflow: Map<string, SourceTaskAggregate>,
|
||
taskRowsByWorkflow: Map<string, SourceTaskRow[]>,
|
||
schemeByWorkflow: Map<string, SourceSchemeAggregate>,
|
||
schemeRowsByWorkflow: Map<string, SourceSchemeField[]>,
|
||
employeeDirectory: Map<string, EmployeeDirectoryEntry>,
|
||
activityNodeNames: Map<string, string>,
|
||
): WorkflowSnapshotPayload[] {
|
||
return sourceRows.flatMap((row) => {
|
||
const processInstanceId = clean(row.processInstanceId)
|
||
const status = clean(row.status)
|
||
const sourceCreatedAtRaw = sourceWorkflowTimestampToIso(row.sourceCreatedAtRaw)
|
||
if (!processInstanceId || !status || !isDingWorkflowStartedOnOrAfterCutoff(row.sourceCreatedAtRaw)) return []
|
||
const ongoing = isDingWorkflowOngoing(status, row.sourceFinishedAtRaw)
|
||
const task = taskByWorkflow.get(processInstanceId)
|
||
const scheme = schemeByWorkflow.get(processInstanceId)
|
||
const processCode = normalizeDingWorkflowProcessCode(row.processCode)
|
||
const schemeFields = schemeRowsByWorkflow.get(processInstanceId) || []
|
||
const assignments = extractContactAssignments(schemeFields)
|
||
const brand = normalizeDingWorkflowBrand(firstSchemeValue(schemeFields, ['品牌']))
|
||
const requesterDepartment = normalizeDingWorkflowDepartment(firstSchemeValue(schemeFields, ['需求部门']))
|
||
|| normalizeDingWorkflowDepartment(row.originatorDeptName)
|
||
|| '未记录部门'
|
||
const originatorUserId = clean(row.originatorUserId)
|
||
const requesterName = firstSchemeValue(schemeFields, ['需求人'])
|
||
|| (originatorUserId ? employeeDirectory.get(originatorUserId)?.name : null)
|
||
|| titleOriginatorName(row.title)
|
||
|| '未记录'
|
||
const activeTasks = ongoing ? (taskRowsByWorkflow.get(processInstanceId) || []).flatMap((sourceTask) => {
|
||
if (!sourceTaskIsOngoing(sourceTask.status)) return []
|
||
const taskId = clean(sourceTask.taskId)
|
||
if (!taskId) return []
|
||
const activityId = clean(sourceTask.activityId)
|
||
const assigneeUserId = clean(sourceTask.userId)
|
||
const matchingAssignments = assigneeUserId
|
||
? assignments.filter((assignment) => assignment.userId === assigneeUserId)
|
||
: []
|
||
const employee = assigneeUserId ? employeeDirectory.get(assigneeUserId) : null
|
||
const uniqueAssignmentNodeNames = Array.from(new Set(matchingAssignments.map((assignment) => assignment.nodeName)))
|
||
const inferredNodeName = processCode && activityId
|
||
? activityNodeNames.get(`${processCode}:${activityId}`)
|
||
: null
|
||
return [{
|
||
taskId,
|
||
activityId,
|
||
nodeName: inferredNodeName
|
||
|| (uniqueAssignmentNodeNames.length === 1 ? uniqueAssignmentNodeNames[0] : null)
|
||
|| '当前处理节点',
|
||
assigneeUserId,
|
||
assigneeName: matchingAssignments.find((assignment) => assignment.name)?.name
|
||
|| employee?.name
|
||
|| (assigneeUserId === 'bpms_system' ? '系统处理' : '待同步人员'),
|
||
assigneeDepartment: employee?.departmentName || null,
|
||
isBrandCenter: employee?.isBrandCenter === true,
|
||
createdAt: sourceWorkflowTimestampToIso(sourceTask.createTime),
|
||
}]
|
||
}) : []
|
||
|
||
return [{
|
||
processInstanceId,
|
||
businessId: clean(row.businessId),
|
||
title: clean(row.title),
|
||
result: clean(row.result),
|
||
originatorUserId,
|
||
originatorDeptId: clean(row.originatorDeptId),
|
||
originatorDeptName: clean(row.originatorDeptName),
|
||
status: status.toUpperCase(),
|
||
bizAction: clean(row.bizAction),
|
||
processCode,
|
||
flowType: clean(row.flowType),
|
||
joinedAtRaw: clean(row.joinedAtRaw),
|
||
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,
|
||
brand,
|
||
requesterName,
|
||
requesterDepartment,
|
||
projectType: getDingWorkflowProjectType(processCode),
|
||
activeTasks,
|
||
}]
|
||
})
|
||
}
|
||
|
||
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", "brand", "requester_name",
|
||
"requester_department", "project_type", "active_tasks", "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", incoming."brand", incoming."requesterName",
|
||
incoming."requesterDepartment", incoming."projectType", incoming."activeTasks", $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, "brand" text, "requesterName" text,
|
||
"requesterDepartment" text, "projectType" text, "activeTasks" jsonb
|
||
)
|
||
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",
|
||
"brand" = EXCLUDED."brand",
|
||
"requester_name" = EXCLUDED."requester_name",
|
||
"requester_department" = EXCLUDED."requester_department",
|
||
"project_type" = EXCLUDED."project_type",
|
||
"active_tasks" = EXCLUDED."active_tasks",
|
||
"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, rowsDeleted: 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,
|
||
taskRowsByWorkflow,
|
||
schemeByWorkflow,
|
||
schemeRowsByWorkflow,
|
||
} = await fetchOngoingAggregates(pool, ongoingIds)
|
||
const employeeDirectory = await readEmployeeDirectory(sourceRows, taskRowsByWorkflow)
|
||
const activityNodeNames = inferActivityNodeNames(sourceRows, taskRowsByWorkflow, schemeRowsByWorkflow)
|
||
const rows = buildSnapshotPayload(
|
||
sourceRows,
|
||
taskByWorkflow,
|
||
taskRowsByWorkflow,
|
||
schemeByWorkflow,
|
||
schemeRowsByWorkflow,
|
||
employeeDirectory,
|
||
activityNodeNames,
|
||
)
|
||
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 deleted = await prisma.dingWorkflowSnapshot.deleteMany({
|
||
where: {
|
||
OR: [
|
||
{ processCode: null },
|
||
{ processCode: { notIn: ALLOWED_PROCESS_CODE_VALUES } },
|
||
{ sourceCreatedAtRaw: null },
|
||
{ sourceCreatedAtRaw: { lt: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO } },
|
||
],
|
||
},
|
||
})
|
||
const ongoingCount = await prisma.dingWorkflowSnapshot.count({
|
||
where: {
|
||
isOngoing: true,
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
},
|
||
})
|
||
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, rowsDeleted: deleted.count, 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<OngoingDingWorkflowSummary> {
|
||
const [total, state] = await prisma.$transaction([
|
||
prisma.dingWorkflowSnapshot.count({
|
||
where: {
|
||
isOngoing: true,
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
},
|
||
}),
|
||
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()
|
||
}
|
||
|
||
const ONGOING_WORKFLOW_SELECT = {
|
||
processInstanceId: true,
|
||
businessId: true,
|
||
title: true,
|
||
status: true,
|
||
processCode: true,
|
||
projectType: true,
|
||
brand: true,
|
||
requesterName: true,
|
||
requesterDepartment: true,
|
||
activeTasks: true,
|
||
originatorUserId: true,
|
||
originatorDeptName: true,
|
||
sourceCreatedAtRaw: true,
|
||
sourceUpdatedAtRaw: true,
|
||
activeTaskCount: true,
|
||
taskCount: true,
|
||
schemeFieldCount: true,
|
||
latestTaskStatus: true,
|
||
} as const
|
||
|
||
function mapSnapshotListItem(row: {
|
||
processInstanceId: string
|
||
businessId: string | null
|
||
title: string | null
|
||
status: string
|
||
processCode: string | null
|
||
projectType: string | null
|
||
brand: string | null
|
||
requesterName: string | null
|
||
requesterDepartment: string | null
|
||
activeTasks: unknown
|
||
originatorUserId: string | null
|
||
originatorDeptName: string | null
|
||
sourceCreatedAtRaw: string | null
|
||
sourceUpdatedAtRaw: string | null
|
||
activeTaskCount: number
|
||
taskCount: number
|
||
schemeFieldCount: number
|
||
latestTaskStatus: string | null
|
||
}): OngoingDingWorkflowListItem {
|
||
const processCode = normalizeDingWorkflowProcessCode(row.processCode) || 'UNKNOWN'
|
||
return {
|
||
id: row.processInstanceId,
|
||
businessId: row.businessId,
|
||
title: row.title || '未命名流程',
|
||
status: row.status,
|
||
processCode,
|
||
projectType: row.projectType || getDingWorkflowProjectType(processCode),
|
||
brand: row.brand || '其他品牌',
|
||
requesterName: row.requesterName || '未记录',
|
||
requesterDepartment: normalizeDingWorkflowDepartment(row.requesterDepartment)
|
||
|| normalizeDingWorkflowDepartment(row.originatorDeptName)
|
||
|| '未记录部门',
|
||
activeTasks: normalizeActiveTasks(row.activeTasks),
|
||
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,
|
||
}
|
||
}
|
||
|
||
export async function getOngoingDingWorkflowInsights(session: Session | null): Promise<OngoingDingWorkflowInsights> {
|
||
await assertDashboardAccess(session)
|
||
const where = {
|
||
isOngoing: true,
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
}
|
||
const [rows, sync] = await Promise.all([
|
||
prisma.dingWorkflowSnapshot.findMany({
|
||
where,
|
||
select: ONGOING_WORKFLOW_SELECT,
|
||
orderBy: [
|
||
{ sourceUpdatedAtRaw: { sort: 'desc', nulls: 'last' } },
|
||
{ processInstanceId: 'desc' },
|
||
],
|
||
}),
|
||
readSummaryWithoutAccessCheck(),
|
||
])
|
||
return {
|
||
items: rows.map(mapSnapshotListItem),
|
||
total: rows.length,
|
||
sync,
|
||
}
|
||
}
|
||
|
||
export async function getOngoingDingWorkflowProjects(
|
||
session: Session | null,
|
||
params?: { page?: number; pageSize?: number; query?: string },
|
||
): Promise<OngoingDingWorkflowPage> {
|
||
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,
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
sourceCreatedAtRaw: { gte: DING_WORKFLOW_STARTED_AT_CUTOFF_ISO },
|
||
...(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: ONGOING_WORKFLOW_SELECT,
|
||
orderBy: [
|
||
{ sourceUpdatedAtRaw: { sort: 'desc', nulls: 'last' } },
|
||
{ processInstanceId: 'desc' },
|
||
],
|
||
skip: (page - 1) * pageSize,
|
||
take: pageSize,
|
||
}),
|
||
readSummaryWithoutAccessCheck(),
|
||
])
|
||
|
||
return {
|
||
items: rows.map(mapSnapshotListItem),
|
||
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.rowsDeleted} 条，进行中 ${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)} 分钟`)
|
||
}
|