gongxiangp2/lib/task-detail-change-notifications.ts

import { randomUUID } from 'crypto'
import type { TaskDetailChangeNotification } from '@prisma/client'

import { isRosterEmployed } from '@/lib/auto-personnel-rules'
import { getEmployeeByUserId } from '@/lib/employee-directory'
import { prisma } from '@/lib/prisma'
import type { StoredTaskDetailChangeOperation } from '@/lib/task-detail-change-approval'
import { buildTaskDetailNotificationReviewImage, hasCurrentTaskDetailReviewImages } from '@/lib/task-detail-change-review'
import { canAccessMaterialPrivacy, TASK_DETAIL_PRIVACY_SELECT } from '@/lib/task-detail-privacy'
import { submitYouyidianNotification, uploadYouyidianNotificationImage } from '@/lib/youyidian-notification-center'

const STATUS = {
  PENDING: 'pending',
  PROCESSING: 'processing',
  RETRYING: 'retrying',
  SUCCEEDED: 'succeeded',
  EXHAUSTED: 'exhausted',
} as const

const RETRY_DELAYS_MS = [0, 60_000, 5 * 60_000, 15 * 60_000, 30 * 60_000]
const RECOVERY_INTERVAL_MS = 60_000
const STALE_PROCESSING_MS = 5 * 60_000
const deliveryTimers = new Map<string, NodeJS.Timeout>()
let workerStarted = false

const REVIEW_HINT = '打开审批页查看全部变更前后的图片。'
const REVIEW_IMAGE = '![图片变更前后对照]('
export const TASK_DETAIL_REVIEW_PROMPT = '请核对变更前后的图片再审批。'

type QueueInput = {
  requestId: string
  recipientUserId?: string | null
  dingtalkUserId: string
  recipientName: string
  eventType: string
  businessKey: string
  title: string
  content: string
  actionUrl: string
  actionTitle: string
}

function schedule(id: string, delayMs = 0) {
  const previous = deliveryTimers.get(id)
  if (previous) clearTimeout(previous)
  const timer = setTimeout(() => {
    deliveryTimers.delete(id)
    void processTaskDetailChangeNotification(id)
  }, Math.max(0, delayMs))
  timer.unref?.()
  deliveryTimers.set(id, timer)
}

async function prepareReviewNotice(notification: TaskDetailChangeNotification): Promise<string | null> {
  if (notification.eventType !== 'task_detail_change.requested') return notification.content
  const includesImage = notification.content.includes(REVIEW_IMAGE)
  if (!includesImage && !notification.content.includes(TASK_DETAIL_REVIEW_PROMPT)) {
    return notification.content
  }

  const request = await prisma.taskDetailChangeRequest.findUnique({
    where: { id: notification.requestId },
    include: { material: { select: { updatedAt: true, deletedAt: true, ...TASK_DETAIL_PRIVACY_SELECT } } },
  })
  const managerId = request?.managerUserId
  const employee = managerId && managerId === notification.recipientUserId
    ? await getEmployeeByUserId(managerId)
    : null
  const canSendImage = request
    && employee
    && request.managerDingtalkUserId === notification.dingtalkUserId
    && employee.dingtalkUserId === notification.dingtalkUserId
    && isRosterEmployed(employee.employmentState)
    && hasCurrentTaskDetailReviewImages(request)
    && await canAccessMaterialPrivacy(request.material, managerId)

  if (includesImage) return canSendImage ? notification.content : null

  const parts = [notification.content.replace(TASK_DETAIL_REVIEW_PROMPT, REVIEW_HINT)]
  if (canSendImage && request && Array.isArray(request.operations)) {
    try {
      const preview = await buildTaskDetailNotificationReviewImage(
        request.id, request.materialId, request.operations as StoredTaskDetailChangeOperation[],
      )
      if (preview) {
        const mediaId = await uploadYouyidianNotificationImage(preview.image)
        parts.push(`预览从上到下：${preview.rowLabels.join('、')}。左侧为变更前，右侧为变更后，长图显示顶部。`)
        parts.push(`${REVIEW_IMAGE}${mediaId})`)
      }
    } catch (error) {
      console.error('[task-detail-change] 通知对照图准备失败:', error instanceof Error ? error.message : '未知错误')
    }
  }
  const content = parts.join('\n\n')
  // 保存实际正文，使同一审批通知重试时复用媒体 ID 和幂等请求体。
  await prisma.taskDetailChangeNotification.update({ where: { id: notification.id }, data: { content } })
  return content
}

async function processTaskDetailChangeNotification(id: string) {
  const claimed = await prisma.taskDetailChangeNotification.updateMany({
    where: { id, status: { in: [STATUS.PENDING, STATUS.RETRYING] } },
    data: { status: STATUS.PROCESSING },
  })
  if (claimed.count !== 1) return null

  const notification = await prisma.taskDetailChangeNotification.findUnique({ where: { id } })
  if (!notification) return null
  try {
    const content = await prepareReviewNotice(notification)
    if (content === null) {
      await prisma.taskDetailChangeNotification.update({
        where: { id },
        data: {
          status: STATUS.EXHAUSTED, completedAt: new Date(), nextRetryAt: null,
          lastError: '审批状态或主管权限已变化，已停止图片通知投递',
        },
      })
      return { success: false }
    }
    const result = await submitYouyidianNotification({
      idempotencyKey: `${notification.eventType}:${notification.businessKey}`,
      recipientUserId: notification.dingtalkUserId,
      title: notification.title,
      content,
      actionUrl: notification.actionUrl,
      actionTitle: notification.eventType === 'task_detail_change.requested'
        ? (content.includes(REVIEW_IMAGE) ? '查看对比并审批' : '立即确认')
        : '查看结果',
    })
    await prisma.taskDetailChangeNotification.update({
      where: { id },
      data: {
        status: STATUS.SUCCEEDED,
        attemptCount: { increment: 1 },
        providerMessageId: result.id || null,
        lastError: null,
        nextRetryAt: null,
        sentAt: new Date(),
        completedAt: new Date(),
      },
    })
    return { success: true }
  } catch (error) {
    const nextAttempt = notification.attemptCount + 1
    const exhausted = nextAttempt >= notification.maxAttempts
    const retryDelay = RETRY_DELAYS_MS[Math.min(nextAttempt, RETRY_DELAYS_MS.length - 1)]
    await prisma.taskDetailChangeNotification.update({
      where: { id },
      data: {
        status: exhausted ? STATUS.EXHAUSTED : STATUS.RETRYING,
        attemptCount: { increment: 1 },
        lastError: error instanceof Error ? error.message : '统一通知中心投递失败',
        nextRetryAt: exhausted ? null : new Date(Date.now() + retryDelay),
        completedAt: exhausted ? new Date() : null,
      },
    })
    if (!exhausted) schedule(id, retryDelay)
    return { success: false }
  }
}

async function recoverDueNotifications() {
  const staleBefore = new Date(Date.now() - STALE_PROCESSING_MS)
  await prisma.taskDetailChangeNotification.updateMany({
    where: { status: STATUS.PROCESSING, updatedAt: { lt: staleBefore } },
    data: { status: STATUS.RETRYING, nextRetryAt: new Date(), lastError: '通知处理超时，已自动重试' },
  })
  const rows = await prisma.taskDetailChangeNotification.findMany({
    where: {
      status: { in: [STATUS.PENDING, STATUS.RETRYING] },
      OR: [{ nextRetryAt: null }, { nextRetryAt: { lte: new Date() } }],
    },
    select: { id: true },
    orderBy: { createdAt: 'asc' },
    take: 20,
  })
  rows.forEach((row) => schedule(row.id))
  return rows.length
}

export function ensureTaskDetailChangeNotificationWorkerStarted() {
  if (workerStarted) return
  workerStarted = true
  const bootstrap = setTimeout(() => {
    void recoverDueNotifications().catch((error) => {
      console.error('[task-detail-change] 通知恢复失败:', error)
    })
  }, 1000)
  bootstrap.unref?.()
  const recovery = setInterval(() => {
    void recoverDueNotifications().catch((error) => {
      console.error('[task-detail-change] 通知周期恢复失败:', error)
    })
  }, RECOVERY_INTERVAL_MS)
  recovery.unref?.()
}

export async function queueTaskDetailChangeNotification(input: QueueInput) {
  ensureTaskDetailChangeNotificationWorkerStarted()
  const row = await prisma.taskDetailChangeNotification.upsert({
    where: { eventType_businessKey: { eventType: input.eventType, businessKey: input.businessKey } },
    update: {},
    create: {
      id: randomUUID(),
      requestId: input.requestId,
      recipientUserId: input.recipientUserId || null,
      dingtalkUserId: input.dingtalkUserId,
      recipientName: input.recipientName,
      eventType: input.eventType,
      businessKey: input.businessKey,
      title: input.title,
      content: input.content,
      actionUrl: input.actionUrl,
      status: STATUS.PENDING,
      nextRetryAt: new Date(),
    },
    select: { id: true, status: true },
  })
  if ([STATUS.PENDING, STATUS.RETRYING].includes(row.status as any)) schedule(row.id)
  return row.id
}