feat(scheduler): add cron/interval scheduling with persistent schedules

Author: RomainLanz

.beads/interactions.jsonl

@@ -0,0 +1,4 @@
+{
+  "database": "beads.db",
+  "jsonl_export": "issues.jsonl"
+}

.beads/metadata.json

@@ -29,7 +29,7 @@ npm install @boringnode/queue
 - **Priority Queues**: Process high-priority jobs first
 - **Retry with Backoff**: Automatic retries with exponential, linear, or fixed backoff strategies
 - **Job Timeout**: Automatically fail or retry jobs that exceed a time limit
-- **Repeating Jobs**: Schedule jobs to repeat at fixed intervals
+- **Scheduled Jobs**: Cron-based or interval-based job scheduling with pause/resume support
 
 ## Quick Start
 
@@ -355,18 +355,15 @@ export default class MyJob extends Job<Payload> {
 
 ### Context Properties
 
-| Property          | Type                | Description                                       |
-|-------------------|---------------------|---------------------------------------------------|
-| `jobId`           | string              | Unique identifier for this job                    |
-| `name`            | string              | Job class name                                    |
-| `attempt`         | number              | Current attempt number (1-based)                  |
-| `queue`           | string              | Queue name this job is being processed from       |
-| `priority`        | number              | Job priority (lower = higher priority)            |
-| `acquiredAt`      | Date                | When this job was acquired by the worker          |
-| `stalledCount`    | number              | Times this job was recovered from stalled state   |
-| `isRepeating`     | boolean             | Whether this job is configured to repeat          |
-| `repeatRemaining` | number \| undefined | Remaining repetitions (undefined = infinite)      |
-| `repeatId`        | string \| undefined | Unique ID for the repeat chain (for cancellation) |
+| Property       | Type   | Description                                     |
+|----------------|--------|-------------------------------------------------|
+| `jobId`        | string | Unique identifier for this job                  |
+| `name`         | string | Job class name                                  |
+| `attempt`      | number | Current attempt number (1-based)                |
+| `queue`        | string | Queue name this job is being processed from     |
+| `priority`     | number | Job priority (lower = higher priority)          |
+| `acquiredAt`   | Date   | When this job was acquired by the worker        |
+| `stalledCount` | number | Times this job was recovered from stalled state |
 
 ## Dependency Injection
 
@@ -417,87 +414,96 @@ export default class SendEmailJob extends Job<SendEmailPayload> {
 
 Without a `jobFactory`, jobs are instantiated with `new JobClass(payload, context)`.
 
-## Repeating Jobs
+## Scheduled Jobs
 
-Schedule jobs to repeat automatically at fixed intervals:
+Schedule jobs to run on a recurring basis using cron expressions or fixed intervals. Schedules are persisted and survive worker restarts.
 
-```typescript
-// Repeat every 5 seconds indefinitely
-await SyncJob.dispatch({ source: 'api' }).every('5s')
-
-// Repeat every hour, 10 times total
-await CleanupJob.dispatch({ days: 30 }).every('1h').times(10)
+### Creating a Schedule
 
-// Combine with delay (start after 30 seconds, then repeat every minute)
-await ReportJob.dispatch({ type: 'daily' }).in('30s').every('1m')
+```typescript
+import { Schedule } from '@boringnode/queue'
+
+// Run every 10 seconds (uses job name as schedule ID by default)
+const { scheduleId } = await MetricsJob.schedule({ endpoint: '/api/health' }).every('10s').run()
+
+// Run on a cron schedule with custom ID
+await CleanupJob.schedule({ days: 30 })
+  .id('daily-cleanup') // Custom ID (optional, defaults to job name)
+  .cron('0 * * * *') // Every hour at minute 0
+  .timezone('Europe/Paris') // Optional timezone (default: UTC)
+  .run()
+
+// Schedule with constraints
+await ReportJob.schedule({ type: 'weekly' })
+  .id('weekly-report')
+  .cron('0 9 * * MON') // Every Monday at 9am
+  .from(new Date('2024-01-01')) // Start date
+  .to(new Date('2024-12-31')) // End date
+  .limit(52) // Maximum 52 runs
+  .run()
 ```
 
-### Cancelling a Repeating Job
-
-When dispatching a repeating job, you receive a `repeatId` that can be used to cancel the entire repeat chain from anywhere:
+### Managing Schedules
 
 ```typescript
-import { QueueManager } from '@boringnode/queue'
+import { Schedule } from '@boringnode/queue'
 
-// Dispatch returns jobId and repeatId
-const { jobId, repeatId } = await SyncJob.dispatch({ source: 'api' }).every('5s')
+// Find a schedule by ID
+const schedule = await Schedule.find('health-check')
 
-console.log(`Started repeating job ${jobId} with repeat chain ${repeatId}`)
+if (schedule) {
+  console.log(`Status: ${schedule.status}`) // 'active' or 'paused'
+  console.log(`Run count: ${schedule.runCount}`)
+  console.log(`Next run: ${schedule.nextRunAt}`)
+  console.log(`Last run: ${schedule.lastRunAt}`)
 
-// Later, cancel the repeat chain from anywhere
-if (repeatId) {
-  await QueueManager.cancelRepeat(repeatId)
-}
-```
-
-The `repeatId` is also available inside the job via `this.context.repeatId`.
-
-### Stopping from Within the Job
+  // Pause the schedule
+  await schedule.pause()
 
-A job can stop its own repetition by calling `this.stopRepeating()`:
+  // Resume the schedule
+  await schedule.resume()
 
-```typescript
-import { Job } from '@boringnode/queue'
-import type { JobContext } from '@boringnode/queue/types'
+  // Trigger an immediate run (outside of the normal schedule)
+  await schedule.trigger()
 
-export default class SyncJob extends Job<SyncPayload> {
-  static readonly jobName = 'SyncJob'
-
-  async execute(): Promise<void> {
-    const result = await this.syncData()
-
-    // Stop repeating when sync is complete
-    if (result.isComplete) {
-      this.stopRepeating()
-    }
-  }
+  // Delete the schedule
+  await schedule.delete()
 }
 ```
 
-### Repeat Context
-
-Jobs have access to repeat information via `this.context`:
+### Listing Schedules
 
 ```typescript
-async execute(): Promise<void> {
-  if (this.context.isRepeating) {
-    console.log(`Repeating job, ${this.context.repeatRemaining ?? 'infinite'} runs remaining`)
-  }
-}
-```
+import { Schedule } from '@boringnode/queue'
 
-| Property          | Type                | Description                                       |
-|-------------------|---------------------|---------------------------------------------------|
-| `isRepeating`     | boolean             | Whether this job is configured to repeat          |
-| `repeatRemaining` | number \| undefined | Remaining repetitions (undefined = infinite)      |
-| `repeatId`        | string \| undefined | Unique ID for the repeat chain (for cancellation) |
+// List all schedules
+const all = await Schedule.list()
 
-### How Repeating Works
+// Filter by status
+const active = await Schedule.list({ status: 'active' })
+const paused = await Schedule.list({ status: 'paused' })
+```
 
-- Each repeat creates a **new job** with a new ID
-- The payload is **preserved** across repeats
-- Failed jobs do **not** repeat (only successful completions trigger the next run)
-- The repeat interval is the delay **between** job completions
+### Schedule Options
+
+| Method               | Description                                     |
+|----------------------|-------------------------------------------------|
+| `.id(string)`        | Unique identifier (defaults to job name)        |
+| `.every(duration)`   | Run at fixed intervals ('5s', '1m', '1h', '1d') |
+| `.cron(expression)`  | Run on a cron schedule                          |
+| `.timezone(tz)`      | Timezone for cron expressions (default: 'UTC')  |
+| `.from(date)`        | Don't run before this date                      |
+| `.to(date)`          | Don't run after this date                       |
+| `.between(from, to)` | Shorthand for `.from().to()`                    |
+| `.limit(n)`          | Maximum number of runs                          |
+
+### How Scheduling Works
+
+- Schedules are **persisted** in the database (via the adapter)
+- The **Worker** polls for due schedules and dispatches jobs automatically
+- Each schedule run creates a **new job** with a unique ID
+- Multiple workers can run concurrently - only one will claim each due schedule
+- Failed jobs do **not** affect the schedule (the next run will still occur)
 
 ## Job Discovery
 

README.md

@@ -1,5 +1,6 @@
 import { config } from './config.js'
 import { QueueManager } from '../src/queue_manager.js'
+import { Schedule } from '../src/schedule.js'
 import SendEmailJob from './jobs/send_email_job.js'
 import SyncJob from './jobs/sync_job.js'
 import MetricsJob from './jobs/metrics_job.js'
@@ -14,13 +15,25 @@ for (let i = 0; i < 10; i++) {
   await SendEmailJob.dispatch({ to: 'romain.lanz@pm.me' + i })
 }
 
-// Example: Dispatch a repeating job and get the repeatId for later cancellation
-const { jobId, repeatId } = await MetricsJob.dispatch({ endpoint: '/api/health' }).every('10s')
+// Example: Schedule a recurring metrics job every 10 seconds
+// By default, the schedule ID is the job name ('MetricsJob')
+const { scheduleId } = await MetricsJob.schedule({ endpoint: '/api/health' }).every('10s').run()
 
-console.log(`Started metrics collection job ${jobId}`)
-console.log(`To cancel this repeating job, use: await QueueManager.cancelRepeat('${repeatId}')`)
+console.log(`Created schedule: ${scheduleId}`) // 'MetricsJob'
 
-// Example: Cancel a repeating job after some condition
-// await QueueManager.cancelRepeat(repeatId)
+// Manage the schedule using its ID
+const schedule = await Schedule.find('MetricsJob')
+if (schedule) {
+  console.log(`Schedule status: ${schedule.status}, run count: ${schedule.runCount}`)
+
+  // Pause, resume, or delete
+  // await schedule.pause()
+  // await schedule.resume()
+  // await schedule.delete()
+}
+
+// List all active schedules
+const activeSchedules = await Schedule.list({ status: 'active' })
+console.log(`Active schedules: ${activeSchedules.length}`)
 
 await QueueManager.destroy()

examples/app.ts

@@ -6,8 +6,15 @@ interface MetricsJobPayload {
 }
 
 /**
- * Example job that collects metrics at regular intervals.
- * Demonstrates repeating jobs with external cancellation.
+ * Example job that collects metrics.
+ * For scheduled/repeating execution, use the Schedule API:
+ *
+ * ```typescript
+ * await MetricsJob.schedule({ endpoint: '/api/health' })
+ *   .id('health-check')
+ *   .every('10s')
+ *   .run()
+ * ```
  */
 export default class MetricsJob extends Job<MetricsJobPayload> {
   static readonly jobName = 'MetricsJob'
@@ -17,13 +24,7 @@ export default class MetricsJob extends Job<MetricsJobPayload> {
   }
 
   async execute(): Promise<void> {
-    const repeatInfo = this.context.isRepeating
-      ? ` (repeat ${this.context.repeatRemaining ?? '∞'} remaining, repeatId: ${this.context.repeatId})`
-      : ''
-
-    console.log(
-      `[Job ${this.context.jobId}] Collecting metrics from ${this.payload.endpoint}${repeatInfo}`
-    )
+    console.log(`[Job ${this.context.jobId}] Collecting metrics from ${this.payload.endpoint}`)
 
     // Simulate metrics collection
     const metrics = {

examples/jobs/metrics_job.ts

@@ -2,6 +2,8 @@ export { Job } from './src/job.js'
 export { Worker } from './src/worker.js'
 export { QueueManager } from './src/queue_manager.js'
 export { Locator } from './src/locator.js'
+export { Schedule } from './src/schedule.js'
+export { ScheduleBuilder } from './src/schedule_builder.js'
 export {
   customBackoff,
   linearBackoff,

index.ts

@@ -28,7 +28,8 @@
   },
   "dependencies": {
     "@lukeed/ms": "^2.0.2",
-    "@poppinss/utils": "^6.10.1"
+    "@poppinss/utils": "^6.10.1",
+    "cron-parser": "^5.4.0"
   },
   "devDependencies": {
     "@adonisjs/eslint-config": "^2.1.2",

package.json

@@ -1,4 +1,4 @@
-import type { JobData } from '../types/main.js'
+import type { JobData, ScheduleConfig, ScheduleData, ScheduleListOptions } from '../types/main.js'
 
 /**
  * A job that has been acquired by a worker for processing.
@@ -151,20 +151,62 @@ export interface Adapter {
   destroy(): Promise<void>
 
   /**
-   * Cancel a repeating job chain.
+   * Create or update a schedule.
    *
-   * After calling this, `isRepeatCancelled` will return true for this groupId,
-   * and the worker will not re-dispatch jobs with this groupId.
+   * If a schedule with the given id exists, it will be updated (upsert).
+   * Otherwise, a new schedule is created.
    *
-   * @param groupId - The repeat chain identifier (from RepeatConfig.groupId)
+   * @param config - The schedule configuration
+   * @returns The schedule ID
    */
-  cancelRepeat(groupId: string): Promise<void>
+  createSchedule(config: ScheduleConfig): Promise<string>
 
   /**
-   * Check if a repeat chain has been cancelled.
+   * Get a schedule by ID.
    *
-   * @param groupId - The repeat chain identifier to check
-   * @returns True if the repeat chain has been cancelled
+   * @param id - The schedule ID
+   * @returns The schedule data, or null if not found
    */
-  isRepeatCancelled(groupId: string): Promise<boolean>
+  getSchedule(id: string): Promise<ScheduleData | null>
+
+  /**
+   * List all schedules matching the given options.
+   *
+   * @param options - Optional filters for listing
+   * @returns Array of schedule data
+   */
+  listSchedules(options?: ScheduleListOptions): Promise<ScheduleData[]>
+
+  /**
+   * Update a schedule's status or run metadata.
+   *
+   * @param id - The schedule ID
+   * @param updates - The fields to update
+   */
+  updateSchedule(
+    id: string,
+    updates: Partial<Pick<ScheduleData, 'status' | 'nextRunAt' | 'lastRunAt' | 'runCount'>>
+  ): Promise<void>
+
+  /**
+   * Delete a schedule permanently.
+   *
+   * @param id - The schedule ID to delete
+   */
+  deleteSchedule(id: string): Promise<void>
+
+  /**
+   * Atomically claim a due schedule for execution.
+   *
+   * This method:
+   * 1. Finds ONE schedule where nextRunAt <= now AND status = 'active'
+   * 2. Calculates and updates its nextRunAt to the next occurrence
+   * 3. Increments runCount and sets lastRunAt
+   * 4. Returns the schedule data for job dispatching
+   *
+   * The atomic nature prevents multiple workers from claiming the same schedule.
+   *
+   * @returns The claimed schedule, or null if no schedules are due
+   */
+  claimDueSchedule(): Promise<ScheduleData | null>
 }