Refactor/scheduler

packages/systems/package.json

@@ -367,6 +367,7 @@
     "@fluxcore/database": "workspace:*",
     "@fluxcore/utils": "workspace:*",
     "@napi-rs/canvas": "^0.1.97",
+    "cron-parser": "^5.6.2",
     "discord.js": "catalog:",
     "zod": "^3.25.76"
   },

packages/systems/src/index.ts

@@ -64,9 +64,13 @@ export { checkAndGrantRewards } from "./leveling/rewards.js";
 
 // Scheduled Messages
 export { getScheduledMessages, getScheduledMessageById, createScheduledMessage, updateScheduledMessage, deleteScheduledMessage, getDueMessages, markMessageExecuted } from "./scheduled-messages/persistence.js";
-export { startScheduledMessageScheduler, stopScheduledMessageScheduler, processScheduledMessages } from "./scheduled-messages/scheduler.js";
-export { parseCronExpression, validateCronExpression, getNextCronRun, describeCron } from "./scheduled-messages/cron.js";
+export { startScheduledMessageScheduler, stopScheduledMessageScheduler } from "./scheduled-messages/scheduler.js";
+export { validateCronExpression, getNextCronRun, describeCron } from "./scheduled-messages/cron.js";
 export { SCHEDULER_CHECK_INTERVAL_MS, MAX_SCHEDULED_MESSAGES_PER_GUILD, COMMON_TIMEZONES } from "./scheduled-messages/constants.js";
+
+// Job Queue
+export { JobQueue } from "./queue/index.js";
+
 // Custom Commands
 export {
   getCustomCommands,

packages/systems/src/queue/JobQueue.ts

@@ -0,0 +1,73 @@
+export interface Identifiable {
+  id: string | number;
+}
+
+export class JobQueue<T extends Identifiable> {
+  private buffer: T[] = [];
+  private inFlight: Set<string | number> = new Set();
+  private consumer: (item: T) => Promise<void>;
+  private _running = false;
+  private _processing = false;
+
+  constructor(consumer: (item: T) => Promise<void>) {
+    this.consumer = consumer;
+  }
+
+  enqueue(items: T[]): void {
+    for (const item of items) {
+      if (!this.inFlight.has(item.id)) {
+        this.inFlight.add(item.id);
+        this.buffer.push(item);
+      }
+    }
+    if (this._running && !this._processing) {
+      this.processNext();
+    }
+  }
+
+  private async processNext(): Promise<void> {
+    if (!this._running || this.buffer.length === 0) {
+      this._processing = false;
+      return;
+    }
+
+    this._processing = true;
+    const item = this.buffer.shift()!;
+
+    try {
+      await this.consumer(item);
+    } catch {
+      // Consumer errors are the consumer's responsibility to handle.
+      // We suppress here to prevent unhandled rejections.
+    } finally {
+      this.inFlight.delete(item.id);
+      if (this._running && this.buffer.length > 0) {
+        this.processNext();
+      } else {
+        this._processing = false;
+      }
+    }
+  }
+
+  start(): void {
+    this._running = true;
+    if (this.buffer.length > 0 && !this._processing) {
+      this.processNext();
+    }
+  }
+
+  stop(): void {
+    this._running = false;
+    this._processing = false;
+    this.buffer = [];
+    this.inFlight.clear();
+  }
+
+  get pending(): number {
+    return this.buffer.length;
+  }
+
+  get isProcessing(): boolean {
+    return this._processing;
+  }
+}

packages/systems/src/queue/index.ts

@@ -0,0 +1,2 @@
+export { JobQueue } from "./JobQueue.js";
+export type { Identifiable } from "./JobQueue.js";

packages/systems/src/scheduled-messages/constants.ts

@@ -1,5 +1,5 @@
 /** Interval in milliseconds for checking due scheduled messages */
-export const SCHEDULER_CHECK_INTERVAL_MS = 60_000; // 60 seconds
+export const SCHEDULER_CHECK_INTERVAL_MS = 1_000; // 1 second
 
 /** Maximum number of scheduled messages per guild */
 export const MAX_SCHEDULED_MESSAGES_PER_GUILD = 25;

packages/systems/src/scheduled-messages/cron.ts

@@ -1,233 +1,27 @@
-/**
- * Lightweight cron expression utilities.
- *
- * Supports standard 5-field cron: minute hour day-of-month month day-of-week
- * Uses pure date math instead of pulling in cron-parser at runtime.
- */
+import { CronExpressionParser } from "cron-parser";
 
-interface CronFields {
-  minutes: number[];
-  hours: number[];
-  daysOfMonth: number[];
-  months: number[];
-  daysOfWeek: number[];
-}
-
-function parseField(field: string, min: number, max: number): number[] {
-  const values: Set<number> = new Set();
-
-  for (const part of field.split(",")) {
-    const stepMatch = part.match(/^(.+)\/(\d+)$/);
-    const step = stepMatch ? parseInt(stepMatch[2], 10) : 1;
-    const range = stepMatch ? stepMatch[1] : part;
-
-    if (range === "*") {
-      for (let i = min; i <= max; i += step) values.add(i);
-    } else if (range.includes("-")) {
-      const [startStr, endStr] = range.split("-");
-      const start = parseInt(startStr, 10);
-      const end = parseInt(endStr, 10);
-      if (Number.isNaN(start) || Number.isNaN(end)) throw new Error(`Invalid cron range: ${range}`);
-      for (let i = start; i <= end; i += step) values.add(i);
-    } else {
-      const val = parseInt(range, 10);
-      if (Number.isNaN(val)) throw new Error(`Invalid cron value: ${range}`);
-      values.add(val);
-    }
-  }
-
-  return [...values].sort((a, b) => a - b);
-}
-
-export function parseCronExpression(expr: string): CronFields {
-  const parts = expr.trim().split(/\s+/);
-  if (parts.length !== 5) {
-    throw new Error(`Invalid cron expression: expected 5 fields, got ${parts.length}`);
-  }
-
-  return {
-    minutes: parseField(parts[0], 0, 59),
-    hours: parseField(parts[1], 0, 23),
-    daysOfMonth: parseField(parts[2], 1, 31),
-    months: parseField(parts[3], 1, 12),
-    daysOfWeek: parseField(parts[4], 0, 6),
-  };
-}
-
-/**
- * Validate a cron expression string.
- * Returns null if valid, or an error message string.
- */
 export function validateCronExpression(expr: string): string | null {
+  if (!expr.trim()) return "Cron expression is empty";
   try {
-    parseCronExpression(expr);
+    CronExpressionParser.parse(expr);
     return null;
   } catch (error) {
     return error instanceof Error ? error.message : "Invalid cron expression";
   }
 }
 
-/**
- * Calculate the next run time for a cron expression after the given date.
- * Timezone is handled by adjusting the reference date.
- */
-export function getNextCronRun(cronExpr: string, timezone: string = "UTC", after?: Date): Date {
-  const fields = parseCronExpression(cronExpr);
-  const now = after ?? new Date();
-
-  // Work in the specified timezone by using Intl.DateTimeFormat
-  const formatter = new Intl.DateTimeFormat("en-US", {
-    timeZone: timezone,
-    year: "numeric",
-    month: "2-digit",
-    day: "2-digit",
-    hour: "2-digit",
-    minute: "2-digit",
-    second: "2-digit",
-    hour12: false,
+export function getNextCronRun(
+  cronExpr: string,
+  timezone: string = "UTC",
+  after?: Date,
+): Date {
+  const interval = CronExpressionParser.parse(cronExpr, {
+    tz: timezone,
+    currentDate: after ?? new Date(),
   });
-
-  const parts = formatter.formatToParts(now);
-  const getValue = (type: string): number => {
-    const part = parts.find((p) => p.type === type);
-    return part ? parseInt(part.value, 10) : 0;
-  };
-
-  let year = getValue("year");
-  let month = getValue("month");
-  let day = getValue("day");
-  let hour = getValue("hour");
-  let minute = getValue("minute") + 1; // Start from next minute
-
-  // Search up to 366 days ahead
-  const maxIterations = 366 * 24 * 60;
-  for (let i = 0; i < maxIterations; i++) {
-    // Handle minute overflow
-    if (minute >= 60) {
-      minute = 0;
-      hour++;
-    }
-    if (hour >= 24) {
-      hour = 0;
-      day++;
-    }
-
-    // Handle month/day overflow
-    const daysInMonth = new Date(year, month, 0).getDate();
-    if (day > daysInMonth) {
-      day = 1;
-      month++;
-    }
-    if (month > 12) {
-      month = 1;
-      year++;
-    }
-
-    // Check if this time matches the cron expression
-    if (!fields.months.includes(month)) {
-      // Skip to the next valid month
-      day = 1;
-      hour = 0;
-      minute = 0;
-      month++;
-      continue;
-    }
-
-    if (!fields.daysOfMonth.includes(day)) {
-      hour = 0;
-      minute = 0;
-      day++;
-      continue;
-    }
-
-    // Check day of week
-    const candidateDate = new Date(year, month - 1, day);
-    const dow = candidateDate.getDay();
-    if (!fields.daysOfWeek.includes(dow)) {
-      hour = 0;
-      minute = 0;
-      day++;
-      continue;
-    }
-
-    if (!fields.hours.includes(hour)) {
-      minute = 0;
-      hour++;
-      continue;
-    }
-
-    if (!fields.minutes.includes(minute)) {
-      minute++;
-      continue;
-    }
-
-    // We found a match — convert back from timezone to UTC
-    const tzDateStr = `${year}-${String(month).padStart(2, "0")}-${String(day).padStart(2, "0")}T${String(hour).padStart(2, "0")}:${String(minute).padStart(2, "0")}:00`;
-
-    // Use a trick: create a date in the target timezone and find the UTC equivalent
-    const utcDate = zonedToUtc(tzDateStr, timezone);
-
-    // Ensure result is after 'now'
-    if (utcDate.getTime() > now.getTime()) {
-      return utcDate;
-    }
-
-    minute++;
-  }
-
-  // Fallback: 24 hours from now (should not happen with valid cron)
-  return new Date(now.getTime() + 24 * 60 * 60 * 1000);
+  return interval.next().toDate();
 }
 
-/**
- * Convert a date string in a specific timezone to a UTC Date.
- */
-function zonedToUtc(dateStr: string, timezone: string): Date {
-  if (timezone === "UTC") {
-    return new Date(dateStr + "Z");
-  }
-
-  // Create a date assuming UTC first
-  const utcGuess = new Date(dateStr + "Z");
-
-  // Find the offset of the target timezone at this point in time
-  const formatter = new Intl.DateTimeFormat("en-US", {
-    timeZone: timezone,
-    year: "numeric",
-    month: "2-digit",
-    day: "2-digit",
-    hour: "2-digit",
-    minute: "2-digit",
-    second: "2-digit",
-    hour12: false,
-  });
-
-  const parts = formatter.formatToParts(utcGuess);
-  const getValue = (type: string): number => {
-    const part = parts.find((p) => p.type === type);
-    return part ? parseInt(part.value, 10) : 0;
-  };
-
-  const tzYear = getValue("year");
-  const tzMonth = getValue("month");
-  const tzDay = getValue("day");
-  const tzHour = getValue("hour");
-  const tzMinute = getValue("minute");
-
-  // The difference between what we wanted and what the timezone shows
-  // tells us the offset
-  const wanted = new Date(dateStr + "Z");
-  const actual = new Date(
-    Date.UTC(tzYear, tzMonth - 1, tzDay, tzHour, tzMinute, 0),
-  );
-
-  const offsetMs = actual.getTime() - wanted.getTime();
-  return new Date(wanted.getTime() + offsetMs);
-}
-
-/**
- * Get a human-readable description of a cron expression.
- */
 export function describeCron(cronExpr: string): string {
   const presets: Record<string, string> = {
     "0 * * * *": "Every hour",
@@ -241,24 +35,30 @@ export function describeCron(cronExpr: string): string {
   if (presets[cronExpr]) return presets[cronExpr];
 
   try {
-    const fields = parseCronExpression(cronExpr);
+    const interval = CronExpressionParser.parse(cronExpr);
+    const fields = interval.fields;
     const parts: string[] = [];
 
-    if (fields.minutes.length === 1 && fields.hours.length === 1) {
+    const minutes = Array.from(fields.minute.values);
+    const hours = Array.from(fields.hour.values);
+    const daysOfMonth = Array.from(fields.dayOfMonth.values);
+    const daysOfWeek = Array.from(fields.dayOfWeek.values);
+
+    if (minutes.length === 1 && hours.length === 1) {
       parts.push(
-        `at ${String(fields.hours[0]).padStart(2, "0")}:${String(fields.minutes[0]).padStart(2, "0")}`,
+        `at ${String(hours[0]).padStart(2, "0")}:${String(minutes[0]).padStart(2, "0")}`,
       );
-    } else if (fields.minutes.length === 1) {
-      parts.push(`at minute ${fields.minutes[0]}`);
+    } else if (minutes.length === 1) {
+      parts.push(`at minute ${minutes[0]}`);
     }
 
-    if (fields.daysOfWeek.length < 7) {
+    if (daysOfWeek.length < 7) {
       const dayNames = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"];
-      parts.push(`on ${fields.daysOfWeek.map((d) => dayNames[d]).join(", ")}`);
+      parts.push(`on ${daysOfWeek.map((d) => dayNames[Number(d)]).join(", ")}`);
     }
 
-    if (fields.daysOfMonth.length < 31) {
-      parts.push(`on day ${fields.daysOfMonth.join(", ")}`);
+    if (daysOfMonth.length < 31) {
+      parts.push(`on day ${daysOfMonth.join(", ")}`);
     }
 
     return parts.length > 0 ? parts.join(" ") : cronExpr;