refactor: update MongoDB and SQS adapters to use AWS SDK v3, improve index initialization handling, and enhance test reliability

Author: muniter

src/adapters/mongodb.ts

@@ -21,8 +21,10 @@ export interface MongoConfig {
 }
 
 export class MongoDatabaseAdapter implements DatabaseAdapter {
+  private indexesInitialized: Promise<void>;
+
   constructor(private col: MongoCollection) {
-    this.initializeIndexes();
+    this.indexesInitialized = this.initializeIndexes();
   }
 
   private async initializeIndexes(): Promise<void> {
@@ -48,6 +50,10 @@ export class MongoDatabaseAdapter implements DatabaseAdapter {
     }
   }
 
+  async ensureIndexes(): Promise<void> {
+    await this.indexesInitialized;
+  }
+
   async insertJob(payload: Buffer, meta: JobMeta): Promise<string> {
     const now = new Date();
 

src/adapters/sqs.ts

@@ -1,4 +1,4 @@
-import { SQS as AWSClient, type SQSClientConfig, type SQS } from '@aws-sdk/client-sqs';
+import { SQSClient, type SQSClientConfig } from '@aws-sdk/client-sqs';
 import { SqsQueue } from '../drivers/sqs.ts';
 
 export interface SQSConfig extends SQSClientConfig { 
@@ -7,17 +7,17 @@ export interface SQSConfig extends SQSClientConfig {
 
 // Main export - constructor pattern
 export class SQSQueue<T = Record<string, any>> extends SqsQueue<T> {
-  constructor(config: { client: SQS; queueUrl: string }) {
+  constructor(config: { client: SQSClient; queueUrl: string }) {
     super(config.client, config.queueUrl);
   }
 }
 
-// Convenience factory for AWS SDK
+// Convenience factory for AWS SDK v3
 export function createSQSQueue<T = Record<string, any>>(queueUrl: string, sqsConfig?: SQSConfig): SQSQueue<T> {
-  const client = new AWSClient({
+  const client = new SQSClient({
     region: sqsConfig?.region || process.env.AWS_REGION || 'us-east-1',
     ...sqsConfig
-  }) as any;
+  });
 
   return new SQSQueue<T>({ client, queueUrl });
 }

src/drivers/db.ts

@@ -10,6 +10,10 @@ export class DbQueue<TJobMap = Record<string, any>> extends Queue<TJobMap, DbJob
     super(options);
   }
 
+  get adapter(): DatabaseAdapter {
+    return this.db;
+  }
+
   protected async pushMessage(payload: string, meta: JobMeta): Promise<string> {
     return await this.db.insertJob(Buffer.from(payload), meta);
   }

src/drivers/sqs.ts

@@ -1,4 +1,14 @@
-import type { SQS } from "@aws-sdk/client-sqs";
+import {
+  SQSClient,
+  SendMessageCommand,
+  ReceiveMessageCommand,
+  DeleteMessageCommand,
+  ChangeMessageVisibilityCommand,
+  type SendMessageCommandInput,
+  type ReceiveMessageCommandInput,
+  type DeleteMessageCommandInput,
+  type ChangeMessageVisibilityCommandInput,
+} from "@aws-sdk/client-sqs";
 import { Queue } from "../core/queue.ts";
 import type {
   JobStatus,
@@ -7,17 +17,12 @@ import type {
   SqsJobRequest,
 } from "../interfaces/job.ts";
 
-export type SimplifiedSQSClient = Pick<
-  SQS,
-  "sendMessage" | "receiveMessage" | "changeMessageVisibility" | "deleteMessage"
->;
-
 export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
   TJobMap,
   SqsJobRequest<any>
 > {
   constructor(
-    private client: SimplifiedSQSClient,
+    private client: SQSClient,
     private queueUrl: string,
     options: { ttrDefault?: number } = {}
   ) {
@@ -43,12 +48,14 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
       };
     }
 
-    const result = await this.client.sendMessage({
+    const command = new SendMessageCommand({
       QueueUrl: this.queueUrl,
       MessageBody: payload,
       DelaySeconds: meta.delay || 0,
       MessageAttributes: messageAttributes,
     });
+    
+    const result = await this.client.send(command);
 
     if (!result.MessageId) {
       throw new Error("Failed to send message - no MessageId returned");
@@ -57,12 +64,14 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
   }
 
   protected async reserve(timeout: number): Promise<QueueMessage | null> {
-    const result = await this.client.receiveMessage({
+    const command = new ReceiveMessageCommand({
       QueueUrl: this.queueUrl,
       MaxNumberOfMessages: 1,
       WaitTimeSeconds: timeout,
       MessageAttributeNames: ["All"],
     });
+    
+    const result = await this.client.send(command);
 
     if (!result.Messages || result.Messages.length === 0) {
       return null;
@@ -88,11 +97,13 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
     }
 
     if (meta.ttr) {
-      await this.client.changeMessageVisibility({
+      const visibilityCommand = new ChangeMessageVisibilityCommand({
         QueueUrl: this.queueUrl,
         ReceiptHandle: message.ReceiptHandle,
         VisibilityTimeout: meta.ttr,
       });
+      
+      await this.client.send(visibilityCommand);
     }
 
     return {
@@ -112,10 +123,12 @@ export class SqsQueue<TJobMap = Record<string, any>> extends Queue<
       );
     }
 
-    await this.client.deleteMessage({
+    const deleteCommand = new DeleteMessageCommand({
       QueueUrl: this.queueUrl,
       ReceiptHandle: message.meta.receiptHandle,
     });
+    
+    await this.client.send(deleteCommand);
   }
 
   async status(id: string): Promise<JobStatus> {

tests/drivers/sqs.test.ts

@@ -13,7 +13,7 @@ describe('SqsQueue', () => {
 
   beforeEach(() => {
     sqsClient = new TestSQSClient();
-    queue = new SqsQueue<TestJobs>(sqsClient, 'https://sqs.us-east-1.amazonaws.com/123456789/test-queue');
+    queue = new SqsQueue<TestJobs>(sqsClient as any, 'https://sqs.us-east-1.amazonaws.com/123456789/test-queue');
   });
 
   describe('addJob and reserve cycle', () => {

tests/integration/mongodb.test.ts

@@ -67,8 +67,8 @@ describe('MongoDB Integration Tests (with TestContainers)', () => {
       // Add a job to trigger index creation
       await queue.addJob('simple-job', { payload: { data: 'test' } });
 
-      // Wait a bit for async index creation to complete
-      await new Promise(resolve => setTimeout(resolve, 100));
+      // Wait for indexes to be created
+      await (queue as any).adapter.ensureIndexes();
 
       // Check that indexes exist
       const db = client.db(testDatabase);

tests/integration/sqs-localstack.test.ts

@@ -2,7 +2,6 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { GenericContainer, type StartedTestContainer } from 'testcontainers';
 import { SQSClient, CreateQueueCommand, DeleteQueueCommand, PurgeQueueCommand } from '@aws-sdk/client-sqs';
 import { SqsQueue } from '../../src/drivers/sqs.ts';
-import { SQSClientV3Adapter } from '../utils/sqs-v3-adapter.ts';
 
 interface TestJobs {
   'process-data': { id: number; data: string };
@@ -73,9 +72,8 @@ describe('SQS Integration Tests (LocalStack)', () => {
       // Queue might be empty, ignore
     }
 
-    // Create fresh queue instance using the v3 adapter
-    const adapter = new SQSClientV3Adapter(sqsClient);
-    queue = new SqsQueue<TestJobs>(adapter, queueUrl);
+    // Create fresh queue instance using SQSClient directly
+    queue = new SqsQueue<TestJobs>(sqsClient, queueUrl);
   });
 
   describe('Real SQS Operations', () => {
@@ -105,7 +103,7 @@ describe('SQS Integration Tests (LocalStack)', () => {
       // Register job handler
       queue.onJob('process-data', async (payload) => {
         processedJobs.push(payload);
-        return `Processed: ${payload.data}`;
+        // Just process without returning string to match void return type
       });
 
       // Add job

tests/mocks/test-sqs-client.ts

@@ -1,8 +1,12 @@
-import type { SimplifiedSQSClient } from "../../src/drivers/sqs.js";
-import type { 
-  ReceiveMessageCommandOutput, 
-  DeleteMessageCommandOutput, 
-  ChangeMessageVisibilityCommandOutput 
+import { 
+  SendMessageCommand,
+  ReceiveMessageCommand,
+  DeleteMessageCommand,
+  ChangeMessageVisibilityCommand,
+  type SendMessageCommandOutput,
+  type ReceiveMessageCommandOutput, 
+  type DeleteMessageCommandOutput, 
+  type ChangeMessageVisibilityCommandOutput 
 } from "@aws-sdk/client-sqs";
 
 interface StoredMessage {
@@ -16,7 +20,7 @@ interface StoredMessage {
   visibilityTimeoutUntil?: Date;
 }
 
-export class TestSQSClient implements SimplifiedSQSClient {
+export class TestSQSClient {
   private messages: Map<string, StoredMessage> = new Map();
   private nextMessageId = 1;
   private nextReceiptHandle = 1;
@@ -31,12 +35,25 @@ export class TestSQSClient implements SimplifiedSQSClient {
 
   public deletedMessages: Array<{ MessageId: string; ReceiptHandle: string }> = [];
 
-  async sendMessage(params: {
+  async send(command: SendMessageCommand | ReceiveMessageCommand | DeleteMessageCommand | ChangeMessageVisibilityCommand): Promise<any> {
+    if (command instanceof SendMessageCommand) {
+      return this.handleSendMessage(command.input as any);
+    } else if (command instanceof ReceiveMessageCommand) {
+      return this.handleReceiveMessage(command.input as any);
+    } else if (command instanceof DeleteMessageCommand) {
+      return this.handleDeleteMessage(command.input as any);
+    } else if (command instanceof ChangeMessageVisibilityCommand) {
+      return this.handleChangeMessageVisibility(command.input as any);
+    }
+    throw new Error(`Unsupported command type: ${(command as any).constructor.name}`);
+  }
+
+  private async handleSendMessage(params: {
     QueueUrl: string;
     MessageBody: string;
     DelaySeconds?: number;
     MessageAttributes?: Record<string, { StringValue: string; DataType: string }>;
-  }): Promise<{ MessageId: string, $metadata: any }> {
+  }): Promise<SendMessageCommandOutput> {
     const messageId = this.nextMessageId.toString();
     this.nextMessageId++;
 
@@ -65,11 +82,14 @@ export class TestSQSClient implements SimplifiedSQSClient {
 
     return { 
       MessageId: messageId,
-      $metadata: {}
+      $metadata: {
+        httpStatusCode: 200,
+        requestId: 'test-request-id'
+      }
     };
   }
 
-  async receiveMessage(params: {
+  private async handleReceiveMessage(params: {
     QueueUrl: string;
     MaxNumberOfMessages?: number;
     WaitTimeSeconds?: number;
@@ -106,7 +126,7 @@ export class TestSQSClient implements SimplifiedSQSClient {
     } as ReceiveMessageCommandOutput;
   }
 
-  async deleteMessage(params: {
+  private async handleDeleteMessage(params: {
     QueueUrl: string;
     ReceiptHandle: string;
   }): Promise<DeleteMessageCommandOutput> {
@@ -127,7 +147,7 @@ export class TestSQSClient implements SimplifiedSQSClient {
     } as DeleteMessageCommandOutput;
   }
 
-  async changeMessageVisibility(params: {
+  private async handleChangeMessageVisibility(params: {
     QueueUrl: string;
     ReceiptHandle: string;
     VisibilityTimeout: number;