chore: wip

Author: chrisbbreuer

docs/cron-expressions.md

@@ -96,7 +96,8 @@ For more complex scheduling needs, you might need to combine multiple cron expre
 ## Using Cron Jobs in Bun Bull
 
 ```typescript
-import { Queue, type CronJobOptions } from 'bun-queue'
+import type { CronJobOptions } from 'bun-queue'
+import { Queue } from 'bun-queue'
 
 // Create a queue
 const queue = new Queue('my-queue')

src/cron-scheduler.ts

@@ -94,7 +94,7 @@ export class CronScheduler {
           limit: options.limit,
           // Need to specify 'every' for backward compatibility
           every: 0, // This value is ignored when cron is specified
-        }
+        },
       }
 
       // Calculate delay until next execution
@@ -107,7 +107,8 @@ export class CronScheduler {
       this.logger.info(`Scheduled cron job ${jobId} with expression "${cronExpression}", next run at ${nextRun.toLocaleString()}`)
 
       return jobId
-    } catch (error) {
+    }
+    catch (error) {
       this.logger.error(`Error scheduling cron job: ${(error as Error).message}`)
       throw error
     }
@@ -155,7 +156,7 @@ export class CronScheduler {
     for (const part of parts) {
       // Handle ranges (e.g. "1-5")
       if (part.includes('-')) {
-        const [start, end] = part.split('-').map(n => parseInt(n, 10))
+        const [start, end] = part.split('-').map(n => Number.parseInt(n, 10))
 
         if (isNaN(start) || isNaN(end) || start < min || end > max || start > end) {
           throw new Error(`Invalid range in cron expression: ${part}`)
@@ -168,17 +169,17 @@ export class CronScheduler {
       // Handle steps (e.g. "*/2", "1/2")
       else if (part.includes('/')) {
         const [range, step] = part.split('/')
-        const stepValue = parseInt(step, 10)
+        const stepValue = Number.parseInt(step, 10)
 
         if (isNaN(stepValue) || stepValue <= 0) {
           throw new Error(`Invalid step in cron expression: ${part}`)
         }
 
         let start = min
-        let end = max
+        const end = max
 
         if (range !== '*') {
-          const rangeValue = parseInt(range, 10)
+          const rangeValue = Number.parseInt(range, 10)
           if (!isNaN(rangeValue)) {
             start = rangeValue
           }
@@ -190,7 +191,7 @@ export class CronScheduler {
       }
       // Handle single values
       else {
-        const value = parseInt(part, 10)
+        const value = Number.parseInt(part, 10)
 
         if (isNaN(value) || value < min || value > max) {
           throw new Error(`Invalid value in cron expression: ${part}`)
@@ -226,15 +227,15 @@ export class CronScheduler {
           hour: 'numeric',
           minute: 'numeric',
           second: 'numeric',
-          hour12: false
+          hour12: false,
         }
 
         const formatter = new Intl.DateTimeFormat('en-US', options)
         const parts = formatter.formatToParts(date)
 
         const getValue = (type: string): number => {
           const part = parts.find(p => p.type === type)
-          return part ? parseInt(part.value, 10) : 0
+          return part ? Number.parseInt(part.value, 10) : 0
         }
 
         const year = getValue('year')
@@ -246,7 +247,8 @@ export class CronScheduler {
         // Create a new date in the target timezone
         date.setFullYear(year, month, day)
         date.setHours(hour, minute, 0, 0)
-      } catch (error) {
+      }
+      catch (error) {
         this.logger.warn(`Invalid timezone: ${timezone}, using system timezone`)
       }
     }
@@ -325,7 +327,8 @@ export class CronScheduler {
 
       this.logger.info(`Unscheduled cron job ${jobId}`)
       return true
-    } catch (error) {
+    }
+    catch (error) {
       this.logger.error(`Error unscheduling cron job ${jobId}: ${(error as Error).message}`)
       return false
     }

src/distributed-lock.ts

@@ -1,6 +1,6 @@
 import type { RedisClient } from 'bun'
-import { generateId } from './utils'
 import { createLogger } from './logger'
+import { generateId } from './utils'
 
 export interface LockOptions {
   /**
@@ -61,8 +61,8 @@ export class DistributedLock {
       const result = await this.redisClient.send('SET', [
         lockKey,
         token,
-        'NX',  // Only set if key doesn't exist
-        'PX',  // Set expiry in milliseconds
+        'NX', // Only set if key doesn't exist
+        'PX', // Set expiry in milliseconds
         duration.toString(),
       ])
 
@@ -108,7 +108,8 @@ export class DistributedLock {
       await this.redisClient.del(lockKey)
       this.logger.debug(`Released lock ${resource} with token ${token}`)
       return true
-    } else {
+    }
+    else {
       this.logger.debug(`Failed to release lock ${resource} with token ${token}`)
       return false
     }
@@ -123,7 +124,7 @@ export class DistributedLock {
     const lockKey = this.getLockKey(resource)
     const result = await this.redisClient.exists(lockKey)
     // Different Redis clients may return different types
-    return result ? true : false
+    return !!result
   }
 
   /**
@@ -166,7 +167,8 @@ export class DistributedLock {
           clearInterval(autoExtendId)
           this.logger.debug(`Stopped auto-extension for lock ${resource}`)
         }
-      } catch (error) {
+      }
+      catch (error) {
         this.logger.error(`Error extending lock ${resource}: ${(error as Error).message}`)
         clearInterval(autoExtendId)
       }
@@ -219,8 +221,9 @@ export class DistributedLock {
 
     try {
       return await fn()
-    } finally {
+    }
+    finally {
       await this.release(resource, token)
     }
   }
-}
+}

src/index.ts

@@ -1,7 +1,7 @@
 export { BatchProcessor } from './batch'
 export { type CleanupOptions, CleanupService } from './cleanup'
 export * from './config'
-export { CronScheduler, type CronJobOptions } from './cron-scheduler'
+export { type CronJobOptions, CronScheduler } from './cron-scheduler'
 export { DistributedLock, type LockOptions } from './distributed-lock'
 export { JobEvents } from './events'
 export { QueueGroup } from './group'

src/priority-queue.ts

@@ -1,12 +1,11 @@
 import type { RedisClient } from 'bun'
-import type { QueueConfig, JobOptions, PriorityLevel, PriorityQueueOptions, JobStatus } from './types'
-import { DEFAULT_PRIORITY_LEVEL, MAX_PRIORITY_LEVELS } from './types'
+import type { JobEvents } from './events'
+import type { JobOptions, JobStatus, PriorityLevel, PriorityQueueOptions, QueueConfig } from './types'
 import { Job } from './job'
-import { Queue } from './queue'
 import { createLogger } from './logger'
-import { generateId, getRedisClient, mergeOptions } from './utils'
-import { Worker } from './worker'
-import { JobEvents } from './events'
+import { Queue } from './queue'
+import { DEFAULT_PRIORITY_LEVEL, MAX_PRIORITY_LEVELS } from './types'
+import { generateId } from './utils'
 
 /**
  * PriorityQueue implements a queue with priority support
@@ -98,15 +97,24 @@ export class PriorityQueue<T = any> {
       // Store the job data with priority
       await this.redisClient.send('HMSET', [
         jobKey,
-        'id', jobId,
-        'name', this.name,
-        'data', JSON.stringify(data),
-        'timestamp', timestamp.toString(),
-        'delay', (opts.delay || 0).toString(),
-        'opts', JSON.stringify(opts),
-        'attemptsMade', '0',
-        'progress', '0',
-        'priority', priority.toString(),
+        'id',
+        jobId,
+        'name',
+        this.name,
+        'data',
+        JSON.stringify(data),
+        'timestamp',
+        timestamp.toString(),
+        'delay',
+        (opts.delay || 0).toString(),
+        'opts',
+        JSON.stringify(opts),
+        'attemptsMade',
+        '0',
+        'progress',
+        '0',
+        'priority',
+        priority.toString(),
       ])
 
       // Handle dependencies if any
@@ -283,7 +291,7 @@ export class PriorityQueue<T = any> {
     for (let priority = 0; priority < this.priorityLevels; priority++) {
       const priorityKey = this.getPriorityKey(priority)
       const countResult = await this.redisClient.send('LLEN', [priorityKey])
-      const count = typeof countResult === 'string' ? parseInt(countResult, 10) : Number(countResult)
+      const count = typeof countResult === 'string' ? Number.parseInt(countResult, 10) : Number(countResult)
       priorityJobs += count
     }
 
@@ -319,7 +327,7 @@ export class PriorityQueue<T = any> {
 
       // See if there are any jobs in this priority queue
       const lengthResult = await this.redisClient.send('LLEN', [priorityKey])
-      const length = typeof lengthResult === 'string' ? parseInt(lengthResult, 10) : Number(lengthResult)
+      const length = typeof lengthResult === 'string' ? Number.parseInt(lengthResult, 10) : Number(lengthResult)
 
       if (length && length > 0) {
         // Get all jobs from this priority level
@@ -350,13 +358,13 @@ export class PriorityQueue<T = any> {
   private startJobMover(): void {
     // Move jobs immediately to start
     this.moveJobsToWaiting().catch(err =>
-      this.logger.error(`Error moving priority jobs to waiting: ${(err as Error).message}`)
+      this.logger.error(`Error moving priority jobs to waiting: ${(err as Error).message}`),
     )
 
     // Then set up interval to do it periodically (faster than the worker tick)
     this.jobMoverTimer = setInterval(() => {
       this.moveJobsToWaiting().catch(err =>
-        this.logger.error(`Error moving priority jobs to waiting: ${(err as Error).message}`)
+        this.logger.error(`Error moving priority jobs to waiting: ${(err as Error).message}`),
       )
     }, 25) // Faster than the standard worker tick (50ms)
   }
@@ -458,4 +466,4 @@ export class PriorityQueue<T = any> {
   private getPriorityKey(priority: PriorityLevel): string {
     return `${this.keyPrefix}:priority:${priority}`
   }
-}
+}

src/queue.ts

@@ -1,19 +1,20 @@
 import type { RedisClient } from 'bun'
+import type { CronJobOptions } from './cron-scheduler'
+import type { RateLimitResult } from './rate-limiter'
 import type { JobOptions, JobStatus, QueueConfig } from './types'
 import { CleanupService } from './cleanup'
 import { scriptLoader } from './commands'
 import { config } from './config'
-import { CronScheduler, type CronJobOptions } from './cron-scheduler'
+import { CronScheduler } from './cron-scheduler'
 import { DistributedLock } from './distributed-lock'
 import { JobEvents } from './events'
 import { Job } from './job'
 import { createLogger } from './logger'
 import { Metrics } from './metrics'
+import { RateLimiter } from './rate-limiter'
 import { StalledJobChecker } from './stalled-checker'
 import { generateId, getRedisClient, mergeOptions } from './utils'
 import { Worker } from './worker'
-import { RateLimiter } from './rate-limiter'
-import type { RateLimitResult } from './rate-limiter'
 
 export class Queue<T = any> {
   name: string
@@ -112,7 +113,7 @@ export class Queue<T = any> {
           const opts = {
             ...this.defaultJobOptions,
             ...options,
-            delay: limiterResult.resetIn
+            delay: limiterResult.resetIn,
           }
 
           return this.add(data, opts)
@@ -540,7 +541,7 @@ export class Queue<T = any> {
   /**
    * Check if the queue is rate limited for a specific key
    */
-  async isRateLimited(key?: string, data?: T): Promise<{ limited: boolean; resetIn: number }> {
+  async isRateLimited(key?: string, data?: T): Promise<{ limited: boolean, resetIn: number }> {
     if (!this.limiter) {
       return { limited: false, resetIn: 0 }
     }
@@ -550,7 +551,8 @@ export class Queue<T = any> {
     if (key) {
       // Use explicit key if provided
       result = await this.limiter.checkByKey(key)
-    } else {
+    }
+    else {
       // Use data with keyPrefix from limiter options
       result = await this.limiter.check(data)
     }
@@ -570,7 +572,8 @@ export class Queue<T = any> {
     // If locks are disabled, just process the job directly
     if (!this.lock) {
       const job = await this.getJob(jobId)
-      if (!job) return null
+      if (!job)
+        return null
       return handler(job)
     }
 
@@ -582,17 +585,19 @@ export class Queue<T = any> {
       return await this.lock.withLock(lockResource, async () => {
         // Now we have the lock, fetch the job and process it
         const job = await this.getJob(jobId)
-        if (!job) return null
+        if (!job)
+          return null
 
         // Process the job with the lock held
         this.logger.debug(`Processing job ${jobId} with distributed lock`)
         return handler(job)
       }, {
         duration: 30000, // 30 second lock
-        retries: 3,      // Try 3 times to get the lock
-        retryDelay: 200  // 200ms between retries
+        retries: 3, // Try 3 times to get the lock
+        retryDelay: 200, // 200ms between retries
       })
-    } catch (error) {
+    }
+    catch (error) {
       this.logger.error(`Failed to acquire lock for job ${jobId}: ${(error as Error).message}`)
       throw error
     }

src/rate-limiter.ts

@@ -31,7 +31,8 @@ export class RateLimiter {
         // Use the function to generate a key from the data
         const keyValue = this.options.keyPrefix(data)
         identifier = `${this.queue.name}:${keyValue}`
-      } else {
+      }
+      else {
         // Use the keyPrefix as a property path in the data object
         const keyValue = data[this.options.keyPrefix]
         if (keyValue) {

src/worker.ts

@@ -130,7 +130,8 @@ export class Worker<T = any> {
         if ('processJobWithLock' in this.queue && typeof this.queue.processJobWithLock === 'function') {
           // Process with distributed lock for better concurrency safety
           await this.queue.processJobWithLock(jobId, async (job) => {
-            if (!job) return
+            if (!job)
+              return
 
             // Set processedOn timestamp
             const jobKey = this.queue.getJobKey(jobId)
@@ -145,7 +146,8 @@ export class Worker<T = any> {
             await job.moveToCompleted(result)
             return result
           })
-        } else {
+        }
+        else {
           // Fallback to traditional processing without distributed lock
           const job = await this.queue.getJob(jobId)
           if (!job)