refactor: every driver handles serializing the job payload

Before all drivers would just get a payload string, but now each get payload unknown and depending on their capabilities they can decide how to store it.

mongo db: an object
sqlite: jsonb column
memory: just an object
file: json string
redis: json string
sqs: json string

Author: muniter

src/core/queue.ts

@@ -1,6 +1,6 @@
 import { Queue } from '../core/queue.ts';
 import type { JobStatus, JobMeta, QueueMessage, BaseJobOptions, WithPriority, WithDelay } from '../interfaces/job.ts';
-import type { DatabaseAdapter, QueueJobRecord } from '../interfaces/database.ts';
+import type { DatabaseAdapter } from '../interfaces/database.ts';
 import type { QueueOptions } from '../interfaces/plugin.ts';
 
 // Driver-specific job request interface
@@ -23,8 +23,8 @@ export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJob
     return this.db;
   }
 
-  protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
-    return await this.db.insertJob(Buffer.from(payload), meta);
+  protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
+    return await this.db.insertJob(payload, meta);
   }
 
   protected async reserve(timeout: number): Promise<QueueMessage | null> {
@@ -34,15 +34,10 @@ 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,
-      name: jobData.name,
-      payload,
-      meta: record.meta
+      payload: record.payload,
+      meta: record.meta,
     };
   }
 

src/drivers/db.ts

@@ -62,12 +62,12 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
     }
   }
 
-  protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
+  protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
     const id = await this.touchIndex(async (data) => {
       const jobId = String(++data.lastId);
       const jobPath = path.join(this.path, `job${jobId}.data`);
 
-      await fs.writeFile(jobPath, payload, 'utf8');
+      await fs.writeFile(jobPath, JSON.stringify({ payload, meta }), 'utf8');
       if (this.fileMode !== undefined) {
         await fs.chmod(jobPath, this.fileMode);
       }
@@ -137,17 +137,14 @@ 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);
+        const { payload, meta } = JSON.parse(await fs.readFile(jobPath, 'utf8'));
 
         return {
           id: reserved.id,
-          name: jobData.name,
           payload,
           meta: {
-            ttr: reserved.ttr
+            ttr: reserved.ttr,
+            name: meta.name,
           }
         };
       }

src/drivers/file.ts

@@ -11,7 +11,7 @@ export interface InMemoryJobRequest<TPayload> extends BaseJobOptions, WithPriori
 
 interface InMemoryJobRecord {
   id: string;
-  payload: string;
+  payload: unknown;
   meta: JobMeta;
   status: 'waiting' | 'reserved' | 'done' | 'failed';
   pushedAt: Date;
@@ -86,7 +86,7 @@ export class InMemoryQueue<TJobMap = Record<string, any>> extends Queue<TJobMap,
     this.maxJobs = options.maxJobs || 1000;
   }
 
-  protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
+  protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
     const id = (this.nextJobId++).toString();
     const now = new Date();
 
@@ -154,12 +154,8 @@ 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/memory.ts

@@ -25,6 +25,7 @@ export interface MongooseJobRequest<TPayload>
 // MongoDB document structure for queue jobs
 export interface IQueueJobDocument {
   payload: any;
+  name: string;
   ttr: number;
   delaySeconds: number;
   priority: number;
@@ -46,6 +47,7 @@ export interface IQueueJob extends IQueueJobDocument, Document {
 // Queue job schema
 export const QueueJobSchema = new Schema<IQueueJob>(
   {
+    name: { type: String, required: true },
     payload: { type: Schema.Types.Mixed, required: true },
     ttr: { type: Number, required: true, default: 300 },
     delaySeconds: { type: Number, required: true, default: 0 },
@@ -79,12 +81,13 @@ QueueJobSchema.index({ _id: 1, status: 1 });
 export class MongooseDatabaseAdapter implements DatabaseAdapter {
   constructor(private model: Model<IQueueJob>) {}
 
-  async insertJob(payload: any, meta: JobMeta): Promise<string> {
+  async insertJob(payload: unknown, meta: JobMeta): Promise<string> {
     const now = new Date();
 
     const doc = await this.model
       .create({
         payload,
+        name: meta.name,
         ttr: meta.ttr ?? 300,
         delaySeconds: meta.delaySeconds ?? 0,
         priority: meta.priority ?? 0,
@@ -166,14 +169,15 @@ export class MongooseDatabaseAdapter implements DatabaseAdapter {
 
     return {
       id: doc._id.toHexString(),
-      payload: doc.payload,
       meta: {
+        name: doc.name,
         ttr: doc.ttr,
         delaySeconds: doc.delaySeconds,
         priority: doc.priority,
         pushedAt: doc.pushTime,
         reservedAt: now,
       },
+      payload: doc.payload,
       pushedAt: doc.pushTime,
       reservedAt: now,
     };

src/drivers/mongoose.ts

@@ -69,13 +69,11 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
     this.redis = redisClient;
   }
 
-  protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
+  protected async pushMessage(payload: unknown, meta: JobMeta): Promise<string> {
     const id = (await this.redis.incr(this.idKey)).toString();
-    const ttr = meta.ttr || this.ttrDefault;
     const now = Math.floor(Date.now() / 1000);
 
-    // Store message in Yii2 format: "ttr;jsonPayload"
-    const message = `${ttr};${payload}`;
+    const message = JSON.stringify({ payload, meta });
     await this.redis.hSet(this.messagesKey, id, message);
 
     if (meta.delaySeconds && meta.delaySeconds > 0) {
@@ -142,32 +140,21 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
       return null;
     }
 
-    // Parse Yii2 format: "ttr;jsonPayload"
-    const separatorIndex = message.indexOf(';');
-    if (separatorIndex === -1) {
-      return null;
-    }
-    
-    const ttr = parseInt(message.substring(0, separatorIndex));
-    const payloadStr = message.substring(separatorIndex + 1);
+    const { payload, meta } = JSON.parse(message) as { payload: unknown, meta: JobMeta };
 
     // Reserve the job
-    const expireAt = now + ttr;
+    const expireAt = now + (meta.ttr || this.ttrDefault);
     await this.redis.zAdd(this.reservedKey, { score: expireAt, value: id });
 
     // 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,
+      payload,
       meta: {
-        ttr,
-        pushedAt: new Date()
+        ttr: meta.ttr || this.ttrDefault,
+        name: meta.name,
       }
     };
   }