662 lines
24 KiB
TypeScript
662 lines
24 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 { 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_ALLOWED_PROCESS_CODES = [
|
||
'PROC-3571C915-0506-4490-A91A-8AA7EA33F838',
|
||
'PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673',
|
||
] 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 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
|
||
}
|
||
|
||
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))
|
||
}
|
||
|
||
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<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)
|
||
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 (
|
||
@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>(),
|
||
schemeByWorkflow: new Map<string, SourceSchemeAggregate>(),
|
||
}
|
||
}
|
||
|
||
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<string, SourceTaskAggregate>,
|
||
schemeByWorkflow: Map<string, SourceSchemeAggregate>,
|
||
): 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: normalizeDingWorkflowProcessCode(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, 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, 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 deleted = await prisma.dingWorkflowSnapshot.deleteMany({
|
||
where: {
|
||
OR: [
|
||
{ processCode: null },
|
||
{ processCode: { notIn: ALLOWED_PROCESS_CODE_VALUES } },
|
||
],
|
||
},
|
||
})
|
||
const ongoingCount = await prisma.dingWorkflowSnapshot.count({
|
||
where: {
|
||
isOngoing: true,
|
||
processCode: { in: ALLOWED_PROCESS_CODE_VALUES },
|
||
},
|
||
})
|
||
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 },
|
||
},
|
||
}),
|
||
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<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 },
|
||
...(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.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)} 分钟`)
|
||
}
|