479 lines
15 KiB
TypeScript
479 lines
15 KiB
TypeScript
import { randomUUID } from 'crypto'
|
||
import { prisma } from '@/lib/prisma'
|
||
import { getPublicAppOrigin } from '@/lib/public-app-url'
|
||
import { buildDingtalkEntryPath } from '@/lib/safe-redirect'
|
||
import { sendDingtalkWorkNotice } from '@/lib/dingtalk-client'
|
||
|
||
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 REMINDER_INTERVAL_MS = 10 * 60_000
|
||
const STALE_PROCESSING_MS = 5 * 60_000
|
||
|
||
let workerStarted = false
|
||
let recoveryTimer: NodeJS.Timeout | null = null
|
||
let reminderTimer: NodeJS.Timeout | null = null
|
||
const deliveryTimers = new Map<string, NodeJS.Timeout>()
|
||
|
||
type QueueNotificationInput = {
|
||
surveyId?: string | null
|
||
targetId?: string | null
|
||
assignmentId?: string | null
|
||
recipientEmployeeId?: string | null
|
||
recipientUserId?: string | null
|
||
dingtalkUserId?: string | null
|
||
recipientName: string
|
||
eventType: string
|
||
businessKey: string
|
||
title: string
|
||
content: string
|
||
actionUrl?: string | null
|
||
}
|
||
|
||
function truncateBody(value: string | null | undefined, limit = 2000) {
|
||
const text = String(value || '')
|
||
return text.length > limit ? text.slice(0, limit) : text
|
||
}
|
||
|
||
function getSurveyUrl(path: string) {
|
||
return `${getPublicAppOrigin()}${buildDingtalkEntryPath(path)}`
|
||
}
|
||
|
||
function clean(value: unknown) {
|
||
const normalized = String(value ?? '').trim()
|
||
return normalized.length > 0 ? normalized : null
|
||
}
|
||
|
||
async function resolveNotificationDingtalkUserId(notification: {
|
||
dingtalkUserId?: string | null
|
||
recipientUserId?: string | null
|
||
recipientEmployeeId?: string | null
|
||
}) {
|
||
const current = clean(notification.dingtalkUserId)
|
||
if (current) return current
|
||
|
||
const recipientUserId = clean(notification.recipientUserId)
|
||
if (recipientUserId) {
|
||
const user = await prisma.user.findUnique({
|
||
where: { id: recipientUserId },
|
||
select: { dingtalkUserId: true },
|
||
})
|
||
const userDingtalkUserId = clean(user?.dingtalkUserId)
|
||
if (userDingtalkUserId) return userDingtalkUserId
|
||
}
|
||
|
||
const recipientEmployeeId = clean(notification.recipientEmployeeId)
|
||
if (!recipientEmployeeId) return null
|
||
|
||
const employee = await prisma.performanceSurveyEmployeeProfile.findUnique({
|
||
where: { id: recipientEmployeeId },
|
||
select: { dingtalkUserId: true, dingEmail: true },
|
||
})
|
||
const employeeDingtalkUserId = clean(employee?.dingtalkUserId)
|
||
if (employeeDingtalkUserId) return employeeDingtalkUserId
|
||
|
||
const email = clean(employee?.dingEmail)
|
||
if (!email) return null
|
||
|
||
const user = await prisma.user.findFirst({
|
||
where: { email },
|
||
select: { dingtalkUserId: true },
|
||
})
|
||
return clean(user?.dingtalkUserId)
|
||
}
|
||
|
||
export function ensurePerformanceSurveyNotificationWorkerStarted() {
|
||
if (workerStarted) return
|
||
workerStarted = true
|
||
|
||
const bootstrap = setTimeout(() => {
|
||
void recoverDuePerformanceSurveyNotifications()
|
||
void queueDuePerformanceSurveyReminders()
|
||
}, 1000)
|
||
bootstrap.unref?.()
|
||
|
||
recoveryTimer = setInterval(() => {
|
||
void recoverDuePerformanceSurveyNotifications()
|
||
}, RECOVERY_INTERVAL_MS)
|
||
recoveryTimer.unref?.()
|
||
|
||
reminderTimer = setInterval(() => {
|
||
void queueDuePerformanceSurveyReminders()
|
||
}, REMINDER_INTERVAL_MS)
|
||
reminderTimer.unref?.()
|
||
}
|
||
|
||
function scheduleNotificationProcessing(notificationId: string, delayMs: number) {
|
||
const existing = deliveryTimers.get(notificationId)
|
||
if (existing) {
|
||
clearTimeout(existing)
|
||
}
|
||
|
||
const timer = setTimeout(() => {
|
||
deliveryTimers.delete(notificationId)
|
||
void processPerformanceSurveyNotification(notificationId)
|
||
}, Math.max(0, delayMs))
|
||
timer.unref?.()
|
||
deliveryTimers.set(notificationId, timer)
|
||
}
|
||
|
||
export async function queuePerformanceSurveyNotification(input: QueueNotificationInput) {
|
||
ensurePerformanceSurveyNotificationWorkerStarted()
|
||
|
||
const existing = await prisma.performanceSurveyNotification.findUnique({
|
||
where: {
|
||
eventType_businessKey: {
|
||
eventType: input.eventType,
|
||
businessKey: input.businessKey,
|
||
},
|
||
},
|
||
})
|
||
|
||
if (existing) {
|
||
return existing.id
|
||
}
|
||
|
||
const created = await prisma.performanceSurveyNotification.create({
|
||
data: {
|
||
id: randomUUID(),
|
||
surveyId: input.surveyId || null,
|
||
targetId: input.targetId || null,
|
||
assignmentId: input.assignmentId || null,
|
||
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 },
|
||
})
|
||
|
||
scheduleNotificationProcessing(created.id, 0)
|
||
return created.id
|
||
}
|
||
|
||
export async function queueSurveyAssignmentNotifications(surveyId: string) {
|
||
const survey = await prisma.performanceSurvey.findUnique({
|
||
where: { id: surveyId },
|
||
include: {
|
||
assignments: {
|
||
where: { status: 'pending' },
|
||
include: { target: true, evaluatorEmployee: true },
|
||
},
|
||
},
|
||
})
|
||
|
||
if (!survey) return []
|
||
|
||
const ids: string[] = []
|
||
for (const assignment of survey.assignments) {
|
||
const targetName = assignment.target.targetName
|
||
ids.push(
|
||
await queuePerformanceSurveyNotification({
|
||
surveyId: survey.id,
|
||
targetId: assignment.targetId,
|
||
assignmentId: assignment.id,
|
||
recipientEmployeeId: assignment.evaluatorEmployeeId,
|
||
recipientUserId: assignment.evaluatorUserId,
|
||
dingtalkUserId: assignment.evaluatorDingtalkUserId,
|
||
recipientName: assignment.evaluatorName,
|
||
eventType: 'survey.assignment.created',
|
||
businessKey: `assignment:${assignment.id}`,
|
||
title: '你有新的考核调研待填写',
|
||
content: `你有一份新的考核调研需要填写。\n\n调研：${survey.title}\n对象：${targetName}${survey.dueAt ? `\n截止：${formatDateTime(survey.dueAt)}` : ''}`,
|
||
actionUrl: getSurveyUrl(`/performance-survey/survey/${survey.id}?targetId=${assignment.targetId}`),
|
||
}),
|
||
)
|
||
}
|
||
|
||
return ids
|
||
}
|
||
|
||
export async function queueSurveyResultNotifications(surveyId: string) {
|
||
const survey = await prisma.performanceSurvey.findUnique({
|
||
where: { id: surveyId },
|
||
include: {
|
||
targets: { include: { employeeProfile: true } },
|
||
},
|
||
})
|
||
|
||
if (!survey) return []
|
||
|
||
const ids: string[] = []
|
||
for (const target of survey.targets) {
|
||
if (target.targetType !== 'employee' || !target.employeeProfile) continue
|
||
ids.push(
|
||
await queuePerformanceSurveyNotification({
|
||
surveyId: survey.id,
|
||
targetId: target.id,
|
||
recipientEmployeeId: target.employeeProfileId,
|
||
dingtalkUserId: target.employeeProfile.dingtalkUserId,
|
||
recipientName: target.targetName,
|
||
eventType: 'survey.result.available',
|
||
businessKey: `result:${survey.id}:${target.id}`,
|
||
title: '考核调研结果已可查看',
|
||
content: `你参与的考核调研结果已经生成。\n\n调研：${survey.title}\n对象：${target.targetName}`,
|
||
actionUrl: getSurveyUrl(`/performance-survey/results/${survey.id}?targetId=${target.id}`),
|
||
}),
|
||
)
|
||
}
|
||
|
||
return ids
|
||
}
|
||
|
||
export async function queueSurveyCreatorNotification(surveyId: string, eventType = 'survey.creator.completed') {
|
||
const survey = await prisma.performanceSurvey.findUnique({
|
||
where: { id: surveyId },
|
||
include: {
|
||
assignments: true,
|
||
},
|
||
})
|
||
|
||
if (!survey) return null
|
||
|
||
const creator = await prisma.user.findUnique({
|
||
where: { id: survey.creatorUserId },
|
||
select: { id: true, username: true, realName: true, dingtalkUserId: true },
|
||
})
|
||
|
||
const total = survey.assignments.length
|
||
const submitted = survey.assignments.filter((assignment) => assignment.status === 'submitted').length
|
||
|
||
return queuePerformanceSurveyNotification({
|
||
surveyId: survey.id,
|
||
recipientUserId: creator?.id || survey.creatorUserId,
|
||
dingtalkUserId: creator?.dingtalkUserId || null,
|
||
recipientName: creator?.realName || creator?.username || survey.creatorName,
|
||
eventType,
|
||
businessKey: `${eventType}:${survey.id}`,
|
||
title: eventType === 'survey.creator.closed' ? '考核调研已关闭' : '考核调研已完成',
|
||
content: `你发起的考核调研有新的进度。\n\n调研：${survey.title}\n提交进度：${submitted}/${total}`,
|
||
actionUrl: getSurveyUrl(`/performance-survey/results/${survey.id}`),
|
||
})
|
||
}
|
||
|
||
export async function queueDuePerformanceSurveyReminders() {
|
||
ensurePerformanceSurveyNotificationWorkerStarted()
|
||
|
||
const now = new Date()
|
||
const before = new Date(Date.now() + 24 * 60 * 60 * 1000)
|
||
const surveys = await prisma.performanceSurvey.findMany({
|
||
where: {
|
||
status: 'open',
|
||
dueAt: {
|
||
gt: now,
|
||
lte: before,
|
||
},
|
||
},
|
||
include: {
|
||
assignments: {
|
||
where: { status: 'pending' },
|
||
include: { target: true },
|
||
},
|
||
},
|
||
})
|
||
|
||
const ids: string[] = []
|
||
for (const survey of surveys) {
|
||
const dueKey = survey.dueAt ? survey.dueAt.toISOString().slice(0, 10) : 'no-date'
|
||
for (const assignment of survey.assignments) {
|
||
ids.push(
|
||
await queuePerformanceSurveyNotification({
|
||
surveyId: survey.id,
|
||
targetId: assignment.targetId,
|
||
assignmentId: assignment.id,
|
||
recipientEmployeeId: assignment.evaluatorEmployeeId,
|
||
recipientUserId: assignment.evaluatorUserId,
|
||
dingtalkUserId: assignment.evaluatorDingtalkUserId,
|
||
recipientName: assignment.evaluatorName,
|
||
eventType: 'survey.assignment.due_reminder',
|
||
businessKey: `due:${dueKey}:${assignment.id}`,
|
||
title: '考核调研即将截止',
|
||
content: `你还有一份考核调研未填写。\n\n调研：${survey.title}\n对象：${assignment.target.targetName}\n截止：${formatDateTime(survey.dueAt)}`,
|
||
actionUrl: getSurveyUrl(`/performance-survey/survey/${survey.id}?targetId=${assignment.targetId}`),
|
||
}),
|
||
)
|
||
}
|
||
}
|
||
|
||
return ids
|
||
}
|
||
|
||
export async function recoverDuePerformanceSurveyNotifications(limit = 20) {
|
||
ensurePerformanceSurveyNotificationWorkerStarted()
|
||
|
||
await prisma.performanceSurveyNotification.updateMany({
|
||
where: {
|
||
status: STATUS.PROCESSING,
|
||
updatedAt: { lte: new Date(Date.now() - STALE_PROCESSING_MS) },
|
||
},
|
||
data: {
|
||
status: STATUS.RETRYING,
|
||
nextRetryAt: new Date(),
|
||
lastError: '上次投递处理中断，已重新入队',
|
||
},
|
||
})
|
||
|
||
const rows = await prisma.performanceSurveyNotification.findMany({
|
||
where: {
|
||
status: { in: [STATUS.PENDING, STATUS.RETRYING] },
|
||
OR: [{ nextRetryAt: null }, { nextRetryAt: { lte: new Date() } }],
|
||
},
|
||
select: { id: true },
|
||
orderBy: { createdAt: 'asc' },
|
||
take: limit,
|
||
})
|
||
|
||
rows.forEach((row) => scheduleNotificationProcessing(row.id, 0))
|
||
return rows.length
|
||
}
|
||
|
||
export async function retryPerformanceSurveyNotification(notificationId: string) {
|
||
ensurePerformanceSurveyNotificationWorkerStarted()
|
||
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
status: STATUS.PENDING,
|
||
nextRetryAt: new Date(),
|
||
completedAt: null,
|
||
attemptCount: 0,
|
||
maxAttempts: MAX_ATTEMPTS,
|
||
lastStatusCode: null,
|
||
lastResponseBody: null,
|
||
lastError: null,
|
||
providerMessageId: null,
|
||
sentAt: null,
|
||
},
|
||
})
|
||
|
||
scheduleNotificationProcessing(notificationId, 0)
|
||
}
|
||
|
||
export async function processPerformanceSurveyNotification(notificationId: string) {
|
||
ensurePerformanceSurveyNotificationWorkerStarted()
|
||
|
||
const notification = await prisma.performanceSurveyNotification.findUnique({
|
||
where: { id: notificationId },
|
||
})
|
||
|
||
if (!notification) return null
|
||
if (![STATUS.PENDING, STATUS.RETRYING, STATUS.PROCESSING].includes(notification.status as any)) {
|
||
return null
|
||
}
|
||
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
status: STATUS.PROCESSING,
|
||
updatedAt: new Date(),
|
||
},
|
||
})
|
||
|
||
const resolvedDingtalkUserId = await resolveNotificationDingtalkUserId(notification)
|
||
if (!resolvedDingtalkUserId) {
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
status: STATUS.SKIPPED,
|
||
lastError: '收件人缺少钉钉 userId',
|
||
completedAt: new Date(),
|
||
},
|
||
})
|
||
return { success: false, status: STATUS.SKIPPED }
|
||
}
|
||
|
||
if (resolvedDingtalkUserId !== notification.dingtalkUserId) {
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
dingtalkUserId: resolvedDingtalkUserId,
|
||
updatedAt: new Date(),
|
||
},
|
||
})
|
||
}
|
||
|
||
try {
|
||
const result = await sendDingtalkWorkNotice({
|
||
userIds: [resolvedDingtalkUserId],
|
||
title: notification.title,
|
||
text: notification.content,
|
||
actionUrl: notification.actionUrl,
|
||
})
|
||
|
||
if (result.skipped) {
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
status: STATUS.SKIPPED,
|
||
lastError: result.reason || '钉钉工作通知未发送',
|
||
completedAt: new Date(),
|
||
},
|
||
})
|
||
return { success: false, status: STATUS.SKIPPED }
|
||
}
|
||
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
status: STATUS.SUCCEEDED,
|
||
attemptCount: { increment: 1 },
|
||
providerMessageId: result.taskId || result.requestId || null,
|
||
lastStatusCode: 200,
|
||
lastResponseBody: truncateBody(JSON.stringify(result.raw || result)),
|
||
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)]
|
||
const nextRetryAt = new Date(Date.now() + retryDelayMs)
|
||
|
||
await prisma.performanceSurveyNotification.update({
|
||
where: { id: notificationId },
|
||
data: {
|
||
status: exhausted ? STATUS.EXHAUSTED : STATUS.RETRYING,
|
||
attemptCount: { increment: 1 },
|
||
lastError: error instanceof Error ? error.message : '钉钉工作通知发送失败',
|
||
nextRetryAt: exhausted ? null : nextRetryAt,
|
||
completedAt: exhausted ? new Date() : null,
|
||
},
|
||
})
|
||
|
||
if (!exhausted) {
|
||
scheduleNotificationProcessing(notificationId, retryDelayMs)
|
||
}
|
||
|
||
return { success: false, status: exhausted ? STATUS.EXHAUSTED : STATUS.RETRYING }
|
||
}
|
||
}
|
||
|
||
function formatDateTime(value: Date | string | null | undefined) {
|
||
if (!value) return ''
|
||
const date = new Date(value)
|
||
if (Number.isNaN(date.getTime())) return ''
|
||
return date.toLocaleString('zh-CN', { hour12: false })
|
||
}
|