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

import { randomUUID } from 'crypto'

import { prisma } from '@/lib/prisma'
import { submitYouyidianNotification } 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

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 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 result = await submitYouyidianNotification({
      idempotencyKey: `${notification.eventType}:${notification.businessKey}`,
      recipientUserId: notification.dingtalkUserId,
      title: notification.title,
      content: notification.content,
      actionUrl: notification.actionUrl,
      actionTitle: notification.eventType === 'task_detail_change.requested' ? '立即确认' : '查看结果',
    })
    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
}