| 4 | import { logger } from './utils/logger'; |
| 5 | |
| 6 | async function removeConflictingJobs(schedulerKey: string) { |
| 7 | // Remove any existing jobs that might conflict with the scheduler |
| 8 | // BullMQ scheduler jobs have IDs like "repeat:<key>:<timestamp>" |
| 9 | const jobStates = ['delayed', 'waiting', 'completed', 'failed'] as const; |
| 10 | |
| 11 | for (const state of jobStates) { |
| 12 | try { |
| 13 | const jobs = await cronQueue.getJobs([state]); |
| 14 | for (const job of jobs) { |
| 15 | // Check if this job was created by the scheduler we're about to upsert |
| 16 | if (job.id?.startsWith(`repeat:${schedulerKey}:`)) { |
| 17 | await job.remove(); |
| 18 | logger.info( |
| 19 | { jobId: job.id, schedulerKey }, |
| 20 | 'Removed conflicting scheduler job', |
| 21 | ); |
| 22 | } |
| 23 | } |
| 24 | } catch (error) { |
| 25 | // Ignore errors during cleanup |
| 26 | } |
| 27 | } |
| 28 | } |
| 29 | |
| 30 | export async function bootCron() { |
| 31 | const jobs: { |