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_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()
}

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>(),
      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)
    if (!processInstanceId || !status) 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: 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,
      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 } },
        ],
      },
    })
    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()
}

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 },
  }
  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 },
    ...(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)} 分钟`)
}