Enhance scheduler functionality with timezone support and error handling
- Updated scheduling logic to utilize the guild's timezone for cron expressions and one-time scheduling. - Introduced a new `getGuildTimezone` function to retrieve the timezone from guild settings. - Enhanced error handling for scheduling inputs, providing localized error messages for invalid cron expressions and missing schedule times. - Improved the WebUI to support a new timing mode selection (once vs. cron) and updated input fields accordingly. - Added localization entries for new scheduling-related messages in both English and German. - Refactored job scheduling and removal logic to ensure proper handling of scheduled jobs in the queue.
This commit is contained in:
TheOnlyMace
@@ -4,11 +4,20 @@ import {
|
||||
type MessageCreateOptions,
|
||||
type TextChannel
|
||||
} from 'discord.js';
|
||||
import { WelcomeEmbedSchema, expandBracketChannelMentions, parseMessageComponentsV2, t, tf } from '@nexumi/shared';
|
||||
import {
|
||||
DEFAULT_GUILD_TIMEZONE,
|
||||
WelcomeEmbedSchema,
|
||||
expandBracketChannelMentions,
|
||||
parseMessageComponentsV2,
|
||||
resolveScheduleRunAt,
|
||||
t,
|
||||
tf
|
||||
} from '@nexumi/shared';
|
||||
import type { ScheduledMessage } from '@prisma/client';
|
||||
import { ensureGuild } from '../../guild.js';
|
||||
import { buildEmbedFromPayload } from '../../lib/embed-payload.js';
|
||||
import { buildComponentsV2Payload } from '../../lib/components-v2-payload.js';
|
||||
import { safeRemoveBullJob } from '../../lib/safe-remove-job.js';
|
||||
import { logger } from '../../logger.js';
|
||||
import { scheduleQueue } from '../../queues.js';
|
||||
import type { BotContext } from '../../types.js';
|
||||
@@ -36,7 +45,7 @@ export type ScheduleCreateInput = AnnouncementInput & {
|
||||
cron?: string | null;
|
||||
};
|
||||
|
||||
function parseScheduleWhen(input: string): Date {
|
||||
function parseScheduleWhen(input: string, timeZone: string): Date {
|
||||
const trimmed = input.trim();
|
||||
if (/^\d+[smhd]$/i.test(trimmed)) {
|
||||
const ms = parseDuration(trimmed);
|
||||
@@ -46,11 +55,27 @@ function parseScheduleWhen(input: string): Date {
|
||||
return new Date(Date.now() + ms);
|
||||
}
|
||||
|
||||
const parsed = Date.parse(trimmed);
|
||||
if (Number.isNaN(parsed) || parsed <= Date.now()) {
|
||||
try {
|
||||
const parsed = resolveScheduleRunAt(trimmed, timeZone);
|
||||
if (parsed.getTime() <= Date.now()) {
|
||||
throw new SchedulerError('invalid_when');
|
||||
}
|
||||
return parsed;
|
||||
} catch {
|
||||
throw new SchedulerError('invalid_when');
|
||||
}
|
||||
return new Date(parsed);
|
||||
}
|
||||
|
||||
async function getGuildTimezone(prisma: BotContext['prisma'], guildId: string): Promise<string> {
|
||||
const settings = await prisma.guildSettings.findUnique({
|
||||
where: { guildId },
|
||||
select: { timezone: true }
|
||||
});
|
||||
return settings?.timezone || DEFAULT_GUILD_TIMEZONE;
|
||||
}
|
||||
|
||||
export function scheduleJobId(scheduleId: string): string {
|
||||
return `schedule-${scheduleId}`;
|
||||
}
|
||||
|
||||
function isValidCron(pattern: string): boolean {
|
||||
@@ -157,53 +182,74 @@ async function assertSendableChannel(
|
||||
async function enqueueScheduleJob(
|
||||
schedule: ScheduledMessage,
|
||||
cron: string | null,
|
||||
runAt: Date | null
|
||||
runAt: Date | null,
|
||||
timeZone: string
|
||||
): Promise<string> {
|
||||
const jobOptions: {
|
||||
jobId: string;
|
||||
removeOnComplete: boolean;
|
||||
removeOnFail: number;
|
||||
delay?: number;
|
||||
repeat?: { pattern: string };
|
||||
} = {
|
||||
jobId: `schedule-${schedule.id}`,
|
||||
removeOnComplete: true,
|
||||
removeOnFail: 50
|
||||
};
|
||||
const jobId = scheduleJobId(schedule.id);
|
||||
|
||||
if (cron) {
|
||||
jobOptions.repeat = { pattern: cron };
|
||||
} else if (runAt) {
|
||||
jobOptions.delay = Math.max(0, runAt.getTime() - Date.now());
|
||||
// BullMQ 5 job schedulers honor guild timezone for cron patterns.
|
||||
await scheduleQueue.upsertJobScheduler(
|
||||
jobId,
|
||||
{ pattern: cron, tz: timeZone },
|
||||
{
|
||||
name: 'scheduleSend',
|
||||
data: { scheduleId: schedule.id },
|
||||
opts: {
|
||||
removeOnComplete: true,
|
||||
removeOnFail: 50
|
||||
}
|
||||
}
|
||||
);
|
||||
return jobId;
|
||||
}
|
||||
|
||||
const job = await scheduleQueue.add('scheduleSend', { scheduleId: schedule.id }, jobOptions);
|
||||
if (!runAt) {
|
||||
throw new SchedulerError('missing_schedule_time');
|
||||
}
|
||||
|
||||
const job = await scheduleQueue.add(
|
||||
'scheduleSend',
|
||||
{ scheduleId: schedule.id },
|
||||
{
|
||||
jobId,
|
||||
removeOnComplete: true,
|
||||
removeOnFail: 50,
|
||||
delay: Math.max(0, runAt.getTime() - Date.now())
|
||||
}
|
||||
);
|
||||
return String(job.id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes a scheduled/failed job if present. Locked (active) jobs cannot be
|
||||
* removed — do not call this from inside the scheduleSend worker for the
|
||||
* job that is currently running.
|
||||
* Removes a one-shot delayed job and/or cron job scheduler. Locked (active)
|
||||
* jobs cannot be removed — do not call this from inside the scheduleSend
|
||||
* worker for the job that is currently running.
|
||||
*/
|
||||
export async function removeScheduleJob(jobId: string | null): Promise<void> {
|
||||
if (!jobId) {
|
||||
return;
|
||||
export async function removeScheduleJob(
|
||||
jobId: string | null,
|
||||
options?: { scheduleId?: string; cron?: string | null }
|
||||
): Promise<void> {
|
||||
const schedulerId = options?.scheduleId ? scheduleJobId(options.scheduleId) : jobId;
|
||||
if (schedulerId && options?.cron) {
|
||||
try {
|
||||
await scheduleQueue.removeJobScheduler(schedulerId);
|
||||
} catch (error) {
|
||||
logger.warn({ schedulerId, error }, 'Failed to remove schedule job scheduler');
|
||||
}
|
||||
}
|
||||
|
||||
const job = await scheduleQueue.getJob(jobId);
|
||||
if (!job) {
|
||||
return;
|
||||
const candidateIds = new Set<string>();
|
||||
if (jobId) {
|
||||
candidateIds.add(jobId);
|
||||
}
|
||||
try {
|
||||
await job.remove();
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
if (message.includes('locked') || message.includes('could not be removed')) {
|
||||
logger.warn({ jobId, error }, 'Schedule job is locked; skipping remove');
|
||||
return;
|
||||
}
|
||||
throw error;
|
||||
if (schedulerId) {
|
||||
candidateIds.add(schedulerId);
|
||||
}
|
||||
|
||||
for (const id of candidateIds) {
|
||||
const job = await scheduleQueue.getJob(id);
|
||||
await safeRemoveBullJob(job, { label: 'schedule', jobId: id });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -235,6 +281,7 @@ export async function createScheduledMessage(
|
||||
|
||||
let runAt: Date | null = null;
|
||||
let cronPattern: string | null = null;
|
||||
const timeZone = await getGuildTimezone(context.prisma, input.guildId);
|
||||
|
||||
if (cron) {
|
||||
if (!isValidCron(cron)) {
|
||||
@@ -242,7 +289,7 @@ export async function createScheduledMessage(
|
||||
}
|
||||
cronPattern = cron;
|
||||
} else if (whenInput) {
|
||||
runAt = parseScheduleWhen(whenInput);
|
||||
runAt = parseScheduleWhen(whenInput, timeZone);
|
||||
}
|
||||
|
||||
await assertSendableChannel(context, input.channelId);
|
||||
@@ -262,7 +309,7 @@ export async function createScheduledMessage(
|
||||
}
|
||||
});
|
||||
|
||||
const jobId = await enqueueScheduleJob(schedule, cronPattern, runAt);
|
||||
const jobId = await enqueueScheduleJob(schedule, cronPattern, runAt, timeZone);
|
||||
return context.prisma.scheduledMessage.update({
|
||||
where: { id: schedule.id },
|
||||
data: { jobId }
|
||||
@@ -322,7 +369,7 @@ export async function deleteScheduledMessage(
|
||||
return false;
|
||||
}
|
||||
|
||||
await removeScheduleJob(match.jobId);
|
||||
await removeScheduleJob(match.jobId, { scheduleId: match.id, cron: match.cron });
|
||||
await context.prisma.scheduledMessage.delete({ where: { id: match.id } });
|
||||
return true;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user