gongxiangp2/lib/work-review-notifications.ts

import { randomUUID } from 'crypto'
import { prisma } from '@/lib/prisma'
import {
  getDingtalkPerformanceSurveyNotificationChannelForUser,
  sendDingtalkRobotOtoMessage,
  sendDingtalkWorkNotice,
} from '@/lib/dingtalk-client'
import { submitYouyidianNotification } from '@/lib/youyidian-notification-center'

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

const MAX_ATTEMPTS = 5
const RETRY_DELAYS_MS = [0, 60_000, 5 * 60_000, 15 * 60_000, 30 * 60_000, 60 * 60_000]
const RECOVERY_INTERVAL_MS = 60_000
const STALE_PROCESSING_MS = 5 * 60_000

type QueueWorkReviewNotificationInput = {
  projectId: string
  versionId: string
  recipientEmployeeId?: string | null
  recipientUserId?: string | null
  dingtalkUserId?: string | null
  recipientName: string
  eventType: string
  businessKey: string
  title: string
  content: string
  actionUrl?: string | null
}

let workerStarted = false
let recoveryTimer: NodeJS.Timeout | null = null
const deliveryTimers = new Map<string, NodeJS.Timeout>()

function clean(value: unknown) {
  const text = String(value ?? '').trim()
  return text || null
}

async function resolveDingtalkUserId(notification: {
  dingtalkUserId?: string | null
  recipientUserId?: string | null
  recipientEmployeeId?: string | null
}) {
  const current = clean(notification.dingtalkUserId)
  if (current) return current

  if (notification.recipientUserId) {
    const user = await prisma.user.findUnique({
      where: { id: notification.recipientUserId },
      select: { dingtalkUserId: true },
    })
    const userDingtalkUserId = clean(user?.dingtalkUserId)
    if (userDingtalkUserId) return userDingtalkUserId
  }

  if (notification.recipientEmployeeId) {
    const employee = await prisma.performanceSurveyEmployeeProfile.findUnique({
      where: { id: notification.recipientEmployeeId },
      select: { dingtalkUserId: true, dingEmail: true },
    })
    const employeeDingtalkUserId = clean(employee?.dingtalkUserId)
    if (employeeDingtalkUserId) return employeeDingtalkUserId
    if (employee?.dingEmail) {
      const user = await prisma.user.findUnique({
        where: { email: employee.dingEmail },
        select: { dingtalkUserId: true },
      })
      return clean(user?.dingtalkUserId)
    }
  }
  return null
}

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

async function processWorkReviewNotification(notificationId: string) {
  const claimed = await prisma.workReviewNotification.updateMany({
    where: {
      id: notificationId,
      status: { in: [STATUS.PENDING, STATUS.RETRYING] },
    },
    data: { status: STATUS.PROCESSING },
  })
  if (claimed.count === 0) return null

  const notification = await prisma.workReviewNotification.findUnique({ where: { id: notificationId } })
  if (!notification) return null
  const dingtalkUserId = await resolveDingtalkUserId(notification)
  if (!dingtalkUserId) {
    await prisma.workReviewNotification.update({
      where: { id: notificationId },
      data: {
        status: STATUS.SKIPPED,
        lastError: '收件人缺少钉钉 userId',
        nextRetryAt: null,
        completedAt: new Date(),
      },
    })
    return { success: false, status: STATUS.SKIPPED }
  }

  if (dingtalkUserId !== notification.dingtalkUserId) {
    await prisma.workReviewNotification.update({
      where: { id: notificationId },
      data: { dingtalkUserId },
    })
  }

  try {
    const providerType = getDingtalkPerformanceSurveyNotificationChannelForUser(dingtalkUserId)
    const input = {
      userIds: [dingtalkUserId],
      title: notification.title,
      text: notification.content,
      actionUrl: notification.actionUrl,
      actionTitle: '查看作品',
    }
    const result = providerType === 'notification_center'
      ? await submitYouyidianNotification({
          idempotencyKey: `${notification.eventType}:${notification.businessKey}`,
          recipientUserId: dingtalkUserId,
          title: notification.title,
          content: notification.content,
          actionUrl: notification.actionUrl || '',
          actionTitle: '查看作品',
        })
      : providerType === 'robot_oto'
        ? await sendDingtalkRobotOtoMessage(input)
        : await sendDingtalkWorkNotice(input)

    if ('skipped' in result && result.skipped) {
      await prisma.workReviewNotification.update({
        where: { id: notificationId },
        data: {
          status: STATUS.SKIPPED,
          providerType,
          lastError: result.reason || '钉钉通知未发送',
          nextRetryAt: null,
          completedAt: new Date(),
        },
      })
      return { success: false, status: STATUS.SKIPPED }
    }

    const providerMessageId = providerType === 'notification_center'
      ? ('id' in result ? result.id || null : null)
      : providerType === 'robot_oto'
        ? ('processQueryKey' in result ? result.processQueryKey || null : null)
        : ('taskId' in result ? result.taskId || result.requestId || null : null)
    await prisma.workReviewNotification.update({
      where: { id: notificationId },
      data: {
        status: STATUS.SUCCEEDED,
        attemptCount: { increment: 1 },
        providerType,
        providerMessageId,
        lastError: null,
        sentAt: new Date(),
        completedAt: new Date(),
        nextRetryAt: null,
      },
    })
    return { success: true, status: STATUS.SUCCEEDED }
  } catch (error) {
    const nextAttempt = notification.attemptCount + 1
    const exhausted = nextAttempt >= notification.maxAttempts
    const retryDelayMs = RETRY_DELAYS_MS[Math.min(nextAttempt, RETRY_DELAYS_MS.length - 1)]
    await prisma.workReviewNotification.update({
      where: { id: notificationId },
      data: {
        status: exhausted ? STATUS.EXHAUSTED : STATUS.RETRYING,
        attemptCount: { increment: 1 },
        lastError: error instanceof Error ? error.message : '钉钉通知发送失败',
        nextRetryAt: exhausted ? null : new Date(Date.now() + retryDelayMs),
        completedAt: exhausted ? new Date() : null,
      },
    })
    if (!exhausted) scheduleNotification(notificationId, retryDelayMs)
    return { success: false, status: exhausted ? STATUS.EXHAUSTED : STATUS.RETRYING }
  }
}

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

export function ensureWorkReviewNotificationWorkerStarted() {
  if (workerStarted) return
  workerStarted = true
  void recoverDueNotifications().catch((error) => {
    console.error('[work-review] 恢复待发送通知失败:', error)
  })
  recoveryTimer = setInterval(() => {
    void recoverDueNotifications().catch((error) => {
      console.error('[work-review] 恢复待发送通知失败:', error)
    })
  }, RECOVERY_INTERVAL_MS)
  recoveryTimer.unref?.()
}

export async function queueWorkReviewNotification(input: QueueWorkReviewNotificationInput) {
  ensureWorkReviewNotificationWorkerStarted()
  const notification = await prisma.workReviewNotification.upsert({
    where: {
      eventType_businessKey: {
        eventType: input.eventType,
        businessKey: input.businessKey,
      },
    },
    update: {},
    create: {
      id: randomUUID(),
      projectId: input.projectId,
      versionId: input.versionId,
      recipientEmployeeId: input.recipientEmployeeId || null,
      recipientUserId: input.recipientUserId || null,
      dingtalkUserId: input.dingtalkUserId || null,
      recipientName: input.recipientName,
      eventType: input.eventType,
      businessKey: input.businessKey,
      title: input.title,
      content: input.content,
      actionUrl: input.actionUrl || null,
      status: STATUS.PENDING,
      maxAttempts: MAX_ATTEMPTS,
      nextRetryAt: new Date(),
    },
    select: { id: true, status: true },
  })
  if ([STATUS.PENDING, STATUS.RETRYING].includes(notification.status as any)) {
    scheduleNotification(notification.id)
  }
  return notification.id
}