139 lines
3.6 KiB
TypeScript
139 lines
3.6 KiB
TypeScript
import "server-only";
|
|
|
|
import { and, asc, desc, eq, lte, sql } from "drizzle-orm";
|
|
import { db } from "@/lib/db";
|
|
import { getSystemSetting } from "@/lib/db/queries";
|
|
import { dingTalkNotificationTask } from "@/lib/db/schema";
|
|
import {
|
|
DINGTALK_ROBOT_NOTIFICATION_CONFIG_KEY,
|
|
defaultDingTalkRobotNotificationConfig,
|
|
normalizeDingTalkRobotNotificationConfig,
|
|
} from "@/lib/notifications/dingtalk-robot-config";
|
|
import {
|
|
assertDingTalkRobotNotificationConfigReady,
|
|
sendDingTalkRobotNotification,
|
|
} from "@/lib/notifications/dingtalk-robot";
|
|
|
|
const DUE_STATUSES = ["pending"] as const;
|
|
|
|
export type DingTalkNotificationTaskStatus =
|
|
| "pending"
|
|
| "sending"
|
|
| "sent"
|
|
| "failed"
|
|
| "cancelled";
|
|
|
|
export async function listRecentDingTalkNotificationTasks(limit = 30) {
|
|
return db
|
|
.select()
|
|
.from(dingTalkNotificationTask)
|
|
.orderBy(desc(dingTalkNotificationTask.createdAt))
|
|
.limit(Math.max(1, Math.min(100, Math.floor(limit))));
|
|
}
|
|
|
|
export async function getDueDingTalkNotificationTasks(limit = 20) {
|
|
return db
|
|
.select()
|
|
.from(dingTalkNotificationTask)
|
|
.where(
|
|
and(
|
|
eq(dingTalkNotificationTask.status, DUE_STATUSES[0]),
|
|
lte(dingTalkNotificationTask.scheduledAt, new Date()),
|
|
),
|
|
)
|
|
.orderBy(asc(dingTalkNotificationTask.scheduledAt))
|
|
.limit(Math.max(1, Math.min(100, Math.floor(limit))));
|
|
}
|
|
|
|
async function claimPendingTask(taskId: string) {
|
|
const [claimedTask] = await db
|
|
.update(dingTalkNotificationTask)
|
|
.set({
|
|
status: "sending",
|
|
attempts: sql`${dingTalkNotificationTask.attempts} + 1`,
|
|
lastAttemptAt: new Date(),
|
|
updatedAt: new Date(),
|
|
lastError: null,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(dingTalkNotificationTask.id, taskId),
|
|
eq(dingTalkNotificationTask.status, "pending"),
|
|
),
|
|
)
|
|
.returning();
|
|
|
|
return claimedTask ?? null;
|
|
}
|
|
|
|
export async function runDingTalkNotificationJob(options?: { limit?: number }) {
|
|
const pendingTasks = await getDueDingTalkNotificationTasks(options?.limit ?? 20);
|
|
const setting = await getSystemSetting({
|
|
key: DINGTALK_ROBOT_NOTIFICATION_CONFIG_KEY,
|
|
}).catch(() => null);
|
|
const config = setting?.value
|
|
? normalizeDingTalkRobotNotificationConfig(setting.value)
|
|
: defaultDingTalkRobotNotificationConfig;
|
|
|
|
assertDingTalkRobotNotificationConfigReady(config);
|
|
|
|
const result = {
|
|
pending: pendingTasks.length,
|
|
attempted: 0,
|
|
sent: 0,
|
|
failed: 0,
|
|
skipped: 0,
|
|
errors: [] as Array<{ id: string; error: string }>,
|
|
};
|
|
|
|
for (const task of pendingTasks) {
|
|
const claimedTask = await claimPendingTask(task.id);
|
|
if (!claimedTask) {
|
|
result.skipped += 1;
|
|
continue;
|
|
}
|
|
|
|
result.attempted += 1;
|
|
|
|
try {
|
|
await sendDingTalkRobotNotification({
|
|
config,
|
|
title: claimedTask.title,
|
|
content: claimedTask.content,
|
|
});
|
|
|
|
await db
|
|
.update(dingTalkNotificationTask)
|
|
.set({
|
|
status: "sent",
|
|
sentAt: new Date(),
|
|
updatedAt: new Date(),
|
|
lastError: null,
|
|
})
|
|
.where(eq(dingTalkNotificationTask.id, claimedTask.id));
|
|
|
|
result.sent += 1;
|
|
} catch (error) {
|
|
const errorMessage =
|
|
error instanceof Error ? error.message : "发送钉钉通知失败";
|
|
|
|
await db
|
|
.update(dingTalkNotificationTask)
|
|
.set({
|
|
status: "failed",
|
|
updatedAt: new Date(),
|
|
lastError: errorMessage,
|
|
})
|
|
.where(eq(dingTalkNotificationTask.id, claimedTask.id));
|
|
|
|
result.failed += 1;
|
|
result.errors.push({
|
|
id: claimedTask.id,
|
|
error: errorMessage,
|
|
});
|
|
}
|
|
}
|
|
|
|
return result;
|
|
}
|