refactor: change job payload handling from Buffer to string across queue implementations

Author: muniter

src/core/queue.ts

@@ -86,7 +86,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
     this.emit('beforePush', event);
 
     const jobData: JobData = { name: name as string, payload };
-    const serializedPayload = Buffer.from(JSON.stringify(jobData));
+    const serializedPayload = JSON.stringify(jobData);
     const id = await this.pushMessage(serializedPayload, meta);
 
     const afterEvent: QueueEvent = { type: 'afterPush', id, name: name as string, payload, meta };
@@ -176,7 +176,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    */
   protected async handleMessage(message: QueueMessage): Promise<boolean> {
     try {
-      const jobData: JobData = JSON.parse(message.payload.toString());
+      const jobData: JobData = JSON.parse(message.payload);
       const { name, payload } = jobData;
 
       const beforeEvent: QueueEvent = { type: 'beforeExec', id: message.id, name, payload, meta: message.meta };
@@ -214,7 +214,7 @@ export abstract class Queue<TJobMap = Record<string, any>, TJobRequest extends B
    */
   protected async handleError(message: QueueMessage, error: unknown): Promise<boolean> {
     try {
-      const jobData: JobData = JSON.parse(message.payload.toString());
+      const jobData: JobData = JSON.parse(message.payload);
       const { name, payload } = jobData;
 
       const errorEvent: QueueEvent = { type: 'afterError', id: message.id, name, payload, meta: message.meta, error };
@@ -237,13 +237,13 @@ 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 Buffer
+   * @param payload - Serialized job data as string
    * @param meta - Job metadata including TTR, delay, priority
    * @returns Promise resolving to unique job ID
    * @protected
    * @abstract
    */
-  protected abstract pushMessage(payload: Buffer, meta: JobMeta): Promise<string>;
+  protected abstract pushMessage(payload: string, meta: JobMeta): Promise<string>;
 
   /**
    * Reserves the next available job from the queue for processing.

src/drivers/db.ts

@@ -10,8 +10,8 @@ export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJob
     super(options);
   }
 
-  protected async pushMessage(payload: Buffer, meta: JobMeta): Promise<string> {
-    return await this.db.insertJob(payload, meta);
+  protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
+    return await this.db.insertJob(Buffer.from(payload), meta);
   }
 
   protected async reserve(timeout: number): Promise<QueueMessage | null> {
@@ -23,7 +23,7 @@ export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJob
 
     return {
       id: record.id,
-      payload: record.payload,
+      payload: record.payload.toString(),
       meta: record.meta
     };
   }

src/drivers/file.ts

@@ -54,12 +54,12 @@ export class FileQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Fil
     }
   }
 
-  protected async pushMessage(payload: Buffer, meta: JobMeta): Promise<string> {
+  protected async pushMessage(payload: string, 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);
+      await fs.writeFile(jobPath, payload, 'utf8');
       if (this.fileMode !== undefined) {
         await fs.chmod(jobPath, this.fileMode);
       }
@@ -129,7 +129,7 @@ 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);
+        const payload = await fs.readFile(jobPath, 'utf8');
         return {
           id: reserved.id,
           payload,

src/drivers/redis.ts

@@ -136,13 +136,13 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
     this.redis = new RedisAdapter(redisClient);
   }
 
-  protected async pushMessage(payload: Buffer, meta: JobMeta): Promise<string> {
+  protected async pushMessage(payload: string, 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.toString('utf8')}`;
+    const message = `${ttr};${payload}`;
     await this.redis.hset(this.messagesKey, id, message);
 
     if (meta.delay && meta.delay > 0) {
@@ -208,7 +208,7 @@ export class RedisQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, Re
 
     return {
       id,
-      payload: Buffer.from(payloadStr, 'utf8'),
+      payload: payloadStr,
       meta: {
         ttr,
         pushedAt: new Date()

src/drivers/sqs.ts

@@ -42,7 +42,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, SqsJ
     super(options);
   }
 
-  protected async pushMessage(payload: Buffer, meta: JobMeta): Promise<string> {
+  protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
     const messageAttributes: Record<string, { StringValue: string; DataType: string }> = {};
 
     if (meta.ttr) {
@@ -54,7 +54,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, SqsJ
 
     const result = await this.client.sendMessage({
       QueueUrl: this.queueUrl,
-      MessageBody: payload.toString('utf8'),
+      MessageBody: payload,
       DelaySeconds: meta.delay || 0,
       MessageAttributes: messageAttributes
     });
@@ -81,7 +81,7 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, SqsJ
     if (!message || !message.Body || !message.MessageId || !message.ReceiptHandle) {
       return null;
     }
-    const payload = Buffer.from(message.Body, 'utf8');
+    const payload = message.Body;
 
     const meta: JobMeta = {};
     if (message.MessageAttributes?.ttr?.StringValue) {

src/interfaces/job.ts

@@ -12,7 +12,7 @@ export interface JobMeta {
 
 export interface QueueMessage {
   id: string;
-  payload: Buffer;
+  payload: string;
   meta: JobMeta;
 }