dealy seconds

Author: muniter

src/adapters/mongodb.ts

@@ -60,10 +60,10 @@ export class MongoDatabaseAdapter implements DatabaseAdapter {
     const res = await this.col.insertOne({
       payload,
       ttr: meta.ttr ?? 300,
-      delay: meta.delay ?? 0,
+      delaySeconds: meta.delaySeconds ?? 0,
       priority: meta.priority ?? 0,
       pushTime: now,
-      delayTime: meta.delay ? new Date(now.getTime() + meta.delay * 1000) : null,
+      delayTime: meta.delaySeconds ? new Date(now.getTime() + meta.delaySeconds * 1000) : null,
       status: 'waiting',
       attempt: 0
     });
@@ -115,7 +115,7 @@ export class MongoDatabaseAdapter implements DatabaseAdapter {
       payload: Buffer.from(doc.payload?.buffer || doc.payload || []), // Handle Binary/Buffer/empty
       meta: {
         ttr: doc.ttr,
-        delay: doc.delay,
+        delaySeconds: doc.delaySeconds,
         priority: doc.priority,
         pushedAt: doc.pushTime,
         reservedAt: now

src/adapters/redis.ts

@@ -77,10 +77,10 @@ export class RedisDatabaseAdapter implements DatabaseAdapter {
       id: jobId,
       payload: payload.toString('base64'),
       ttr: (meta.ttr || 300).toString(),
-      delay: (meta.delay || 0).toString(),
+      delay_seconds: (meta.delaySeconds || 0).toString(),
       priority: (meta.priority || 0).toString(),
       push_time: now.toString(),
-      delay_time: meta.delay ? (now + meta.delay * 1000).toString() : '',
+      delay_time: meta.delaySeconds ? (now + meta.delaySeconds * 1000).toString() : '',
       status: 'waiting',
       attempt: '0'
     };
@@ -89,9 +89,9 @@ export class RedisDatabaseAdapter implements DatabaseAdapter {
     await this.client.hSet(this.getJobKey(jobId), jobData);
 
     // Add to waiting queue based on delay
-    if (meta.delay && meta.delay > 0) {
+    if (meta.delaySeconds && meta.delaySeconds > 0) {
       // Add to delayed set with execution time as score
-      const executeAt = Math.floor((now + meta.delay * 1000) / 1000);
+      const executeAt = Math.floor((now + meta.delaySeconds * 1000) / 1000);
       await this.client.zAdd(`${this.keyPrefix}:delayed`, {
         score: executeAt,
         value: jobId
@@ -157,7 +157,7 @@ export class RedisDatabaseAdapter implements DatabaseAdapter {
       payload: Buffer.from(jobData.payload, 'base64'),
       meta: {
         ttr: parseInt(jobData.ttr || '300'),
-        delay: parseInt(jobData.delay || '0'),
+        delaySeconds: parseInt(jobData.delay_seconds || '0'),
         priority: parseInt(jobData.priority || '0'),
         pushedAt: new Date(parseInt(jobData.push_time || '0')),
         reservedAt: new Date(now * 1000)

src/adapters/sqlite.ts

@@ -33,7 +33,7 @@ export class SQLiteDatabaseAdapter implements DatabaseAdapter {
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         payload BLOB NOT NULL,
         ttr INTEGER DEFAULT 300,
-        delay INTEGER DEFAULT 0,
+        delay_seconds INTEGER DEFAULT 0,
         priority INTEGER DEFAULT 0,
         push_time INTEGER NOT NULL,
         delay_time INTEGER,
@@ -62,18 +62,18 @@ export class SQLiteDatabaseAdapter implements DatabaseAdapter {
     const now = new Date();
     const stmt = this.db.prepare(`
       INSERT INTO jobs (
-        payload, ttr, delay, priority, push_time, 
+        payload, ttr, delay_seconds, priority, push_time, 
         delay_time, status
       ) VALUES (?, ?, ?, ?, ?, ?, ?)
     `);
 
     const result = stmt.run(
       payload,
       meta.ttr || 300,
-      meta.delay || 0,
+      meta.delaySeconds || 0,
       meta.priority || 0,
       now.getTime(),
-      meta.delay ? now.getTime() + meta.delay * 1000 : null,
+      meta.delaySeconds ? now.getTime() + meta.delaySeconds * 1000 : null,
       'waiting'
     );
 
@@ -143,7 +143,7 @@ export class SQLiteDatabaseAdapter implements DatabaseAdapter {
       payload: job.payload,
       meta: {
         ttr: job.ttr,
-        delay: job.delay,
+        delaySeconds: job.delay_seconds,
         priority: job.priority,
         pushedAt: new Date(job.push_time),
         reservedAt: new Date(now)

src/core/queue.ts

@@ -97,7 +97,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    * await queue.addJob('backup', {
    *   payload: { path: '/data' },
    *   ttr: 3600,
-   *   delay: 60
+   *   delaySeconds: 60
    * });
    * ```
    */
@@ -109,7 +109,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
 
     const meta: JobMeta = {
       ttr: options.ttr ?? this.ttrDefault,
-      delay: (options as any).delay ?? 0,
+      delaySeconds: (options as any).delaySeconds ?? 0,
       priority: (options as any).priority ?? 0,
       pushedAt: new Date()
     };
@@ -423,7 +423,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    * Pushes a new message to the queue storage backend.
    * 
    * @param payload - Serialized job data as string
-   * @param meta - Job metadata including TTR, delay, priority
+   * @param meta - Job metadata including TTR, delaySeconds, priority
    * @returns Promise resolving to unique job ID
    * @protected
    * @abstract

src/drivers/db.ts

@@ -26,9 +26,14 @@ export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJob
       return null;
     }
 
+    const payload = record.payload.toString();
+    // Extract job name from payload
+    const jobData = JSON.parse(payload);
+    
     return {
       id: record.id,
-      payload: record.payload.toString(),
+      name: jobData.name,
+      payload,
       meta: record.meta
     };
   }

src/drivers/file.ts

@@ -66,7 +66,7 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
       }
 
       const ttr = meta.ttr ?? 300;
-      const delay = meta.delay ?? 0;
+      const delay = meta.delaySeconds ?? 0;
 
       if (delay === 0) {
         data.waiting.push([jobId, ttr]);
@@ -131,8 +131,13 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
       if (reserved) {
         const jobPath = path.join(this.path, `job${reserved.id}.data`);
         const payload = await fs.readFile(jobPath, 'utf8');
+        
+        // Extract job name from payload
+        const jobData = JSON.parse(payload);
+        
         return {
           id: reserved.id,
+          name: jobData.name,
           payload,
           meta: {
             ttr: reserved.ttr

src/drivers/memory.ts

@@ -59,7 +59,7 @@ export interface InMemoryQueueOptions extends QueueOptions {
  * await queue.addJob('my-job', { 
  *   payload: { data: 'test' },
  *   priority: 5,
- *   delay: 10
+ *   delaySeconds: 10
  * });
  * 
  * await queue.run(true, 1);
@@ -94,15 +94,15 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
     this.jobs.set(id, job);
 
     // Handle delay
-    if (meta.delay && meta.delay > 0) {
-      job.delayTime = Date.now() + (meta.delay * 1000);
+    if (meta.delaySeconds && meta.delaySeconds > 0) {
+      job.delayTime = Date.now() + (meta.delaySeconds * 1000);
       job.status = 'waiting';
 
       // Schedule job to become available after delay
       const timeout = setTimeout(() => {
         this.delayedJobs.delete(id);
         this.addToWaitingQueue(id);
-      }, meta.delay * 1000);
+      }, meta.delaySeconds * 1000);
 
       this.delayedJobs.set(id, timeout);
     } else {
@@ -147,8 +147,12 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
 
     this.ttrTimeouts.set(jobId, ttrTimeout);
 
+    // Extract job name from payload
+    const jobData = JSON.parse(job.payload);
+    
     return {
       id: jobId,
+      name: jobData.name,
       payload: job.payload,
       meta: job.meta
     };

src/drivers/redis.ts

@@ -146,9 +146,9 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
     const message = `${ttr};${payload}`;
     await this.redis.hset(this.messagesKey, id, message);
 
-    if (meta.delay && meta.delay > 0) {
+    if (meta.delaySeconds && meta.delaySeconds > 0) {
       // Add to delayed set with execution time as score
-      const executeAt = now + meta.delay;
+      const executeAt = now + meta.delaySeconds;
       await this.redis.zadd(this.delayedKey, executeAt, id);
     } else {
       // Add to waiting list (FIFO queue)
@@ -207,8 +207,12 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
     // Increment attempt counter
     await this.redis.hincrby(this.attemptsKey, id, 1);
 
+    // Extract job name from payload
+    const jobData = JSON.parse(payloadStr);
+    
     return {
       id,
+      name: jobData.name,
       payload: payloadStr,
       meta: {
         ttr,

src/drivers/sqs.ts

@@ -61,7 +61,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
     const command = new SendMessageCommand({
       QueueUrl: this.queueUrl,
       MessageBody: payload,
-      DelaySeconds: meta.delay || 0,
+      DelaySeconds: meta.delaySeconds || 0,
       MessageAttributes: messageAttributes,
     });
 
@@ -116,8 +116,12 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
       await this.client.send(visibilityCommand);
     }
 
+    // Extract job name from payload
+    const jobData = JSON.parse(payload);
+    
     return {
       id: message.MessageId,
+      name: jobData.name,
       payload,
       meta: {
         ...meta,

src/interfaces/job.ts

@@ -6,7 +6,7 @@ export interface JobMeta {
   /** Time to run - number of seconds to run the job */ 
   ttr?: number;
   /** Number of seconds to delay job execution from now */
-  delay?: number;
+  delaySeconds?: number;
   /** Job priority - higher numbers = higher priority (processed first) */
   priority?: number;
   /** Job pushed at */
@@ -27,7 +27,7 @@ export interface JobContext<T> {
   id: string;
   /** Job payload */
   payload: T;
-  /** Job meta: ttr, delay, priority, pushedAt, reservedAt, doneAt, receiptHandle */
+  /** Job meta: ttr, delaySeconds, priority, pushedAt, reservedAt, doneAt, receiptHandle */
   meta: JobMeta;
   /** Job pushed at */
   pushedAt?: Date;
@@ -54,7 +54,7 @@ export interface QueueMessage {
   name: string;
   /** Job payload */
   payload: string;
-  /** Job meta: ttr, delay, priority, pushedAt, reservedAt, doneAt, receiptHandle */
+  /** Job meta: ttr, delaySeconds, priority, pushedAt, reservedAt, doneAt, receiptHandle */
   meta: JobMeta;
 }
 
@@ -81,7 +81,7 @@ export interface BaseJobOptions {
 // Full options interface (for internal use)
 export interface JobOptions extends BaseJobOptions {
   /** Number of seconds to delay job execution from now */
-  delay?: number;
+  delaySeconds?: number;
   /** Job priority - higher numbers = higher priority (processed first) */
   priority?: number;
 }
@@ -91,24 +91,24 @@ export interface DbJobOptions extends BaseJobOptions {
   // DB adapters may or may not support delay/priority - we allow them for flexibility
   // The specific DatabaseAdapter implementation determines actual support
   /** Number of seconds to delay job execution from now. Support varies by database adapter. */
-  delay?: number;
+  delaySeconds?: number;
   /** Job priority - higher numbers = higher priority. Support varies by database adapter. */
   priority?: number;
 }
 
 export interface SqsJobOptions extends BaseJobOptions {
   /** Number of seconds to delay job execution from now (0-900 seconds max for SQS) */
-  delay?: number;
+  delaySeconds?: number;
 }
 
 export interface FileJobOptions extends BaseJobOptions {
   /** Number of seconds to delay job execution from now */
-  delay?: number;
+  delaySeconds?: number;
 }
 
 export interface InMemoryJobOptions extends BaseJobOptions {
   /** Number of seconds to delay job execution from now */
-  delay?: number;
+  delaySeconds?: number;
   /** Job priority - higher numbers = higher priority (processed first) */
   priority?: number;
 }