Building Asynchronous Tasks with NestJS, AWS SNS, and SQS

graph LR
    A[Client / API Request] --> B(NestJS API / Publisher);
    B -- Publish Message --> C{AWS SNS Topic};
    C -- Fan-out --> D[AWS SQS Queue];
    E(NestJS Consumer Service) -- Polls Queue --> D;
    E -- Processes Message --> F[(SQLite Database)];
    E -- Deletes Message --> D;

    style B fill:#f9f,stroke:#333,stroke-width:2px
    style E fill:#f9f,stroke:#333,stroke-width:2px
    style C fill:#FF9900,stroke:#333,stroke-width:2px
    style D fill:#FF9900,stroke:#333,stroke-width:2px
    style F fill:#ccf,stroke:#333,stroke-width:2px

npx @nestjs/cli new nestjs-sns-sqs-reminders

cd nestjs-sns-sqs-reminders

# AWS SDK v3 Clients for SNS and SQS
npm install @aws-sdk/client-sns @aws-sdk/client-sqs

# SQS Consumer library
npm install sqs-consumer

# TypeORM for database interaction and SQLite driver
npm install @nestjs/typeorm typeorm sqlite3

# Configuration management and UUID generation
npm install @nestjs/config uuid dotenv
npm install --save-dev @types/uuid # Types for uuid

src/
├── app.controller.ts
├── app.module.ts
├── app.service.ts
├── main.ts
├── config/                 # Configuration setup (optional, but good practice)
│   └── configuration.ts
├── database/               # Database related files
│   ├── entities/
│   │   └── processed-message.entity.ts
│   └── daos/
│       └── processed-message.dao.ts
├── sns/                    # SNS Publisher module
│   ├── sns.module.ts
│   ├── sns.publisher.service.ts
│   └── sns.controller.ts   # Example controller to trigger publishing
└── sqs/                    # SQS Consumer module
    ├── sqs.module.ts
    ├── sqs.consumer.service.ts
    └── utils/
        └── constants.ts    # Shared constants like MessageSourceType

{
  ""Sid"": ""AllowSNSPublishToSQS"",
  ""Effect"": ""Allow"",
  ""Principal"": {
    ""Service"": ""sns.amazonaws.com""
  },
  ""Action"": ""SQS:SendMessage"",
  ""Resource"": ""YOUR_SQS_QUEUE_ARN"",
  ""Condition"": {
    ""ArnEquals"": {
      ""aws:SourceArn"": ""YOUR_SNS_TOPIC_ARN""
    }
  }
}

# .env

# AWS Credentials - Use IAM user keys, NOT root keys
AWS_ACCESS_KEY_ID=YOUR_AWS_ACCESS_KEY_ID
AWS_SECRET_ACCESS_KEY=YOUR_AWS_SECRET_ACCESS_KEY
AWS_REGION=us-east-1 # Or your preferred AWS region

# AWS Resource Identifiers
SNS_TOPIC_ARN=YOUR_SNS_TOPIC_ARN # ARN copied from SNS console
SQS_QUEUE_URL=YOUR_SQS_QUEUE_URL # URL copied from SQS console

// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { AppController } from './app.controller';
import { AppService } from './app.service';
// Import other modules later (SnsModule, SqsModule, TypeOrmModule)

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true, // Make ConfigService available globally
      envFilePath: '.env', // Specify the env file path
    }),
    // ... other modules will be added here
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

// src/database/entities/processed-message.entity.ts
import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn, Index } from 'typeorm';

@Entity({ name: 'processed_messages' })
export class ProcessedMessage {
  @PrimaryGeneratedColumn()
  id: number;

  @Index({ unique: true }) // Ensure messageId is unique
  @Column({ type: 'varchar', length: 255 })
  messageId: string; // The unique ID from the SQS message

  @Column({ type: 'text' })
  messageBody: string; // Store the original message body for reference

  @Column({ type: 'varchar', length: 50, nullable: true })
  sourceType?: string; // e.g., 'SNS', 'SQS' - optional tracking

  @CreateDateColumn()
  processedAt: Date;
}

// src/database/daos/processed-message.dao.ts
import { Injectable, Logger } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository, QueryFailedError } from 'typeorm';
import { ProcessedMessage } from '../entities/processed-message.entity';

@Injectable()
export class ProcessedMessageDao {
  private readonly logger = new Logger(ProcessedMessageDao.name);

  constructor(
    @InjectRepository(ProcessedMessage)
    private readonly repository: Repository<ProcessedMessage>,
  ) {}

  async recordMessageProcessed(
    messageId: string,
    messageBody: string,
    sourceType?: string,
  ): Promise<ProcessedMessage | null> {
    try {
      const newMessage = this.repository.create({
        messageId,
        messageBody,
        sourceType,
      });
      return await this.repository.save(newMessage);
    } catch (error) {
      // Check if it's a unique constraint violation (already processed)
      // Note: Specific error code ('SQLITE_CONSTRAINT') might differ for other DBs (e.g., '23505' for PostgreSQL)
      if (error instanceof QueryFailedError && error.driverError?.code === 'SQLITE_CONSTRAINT') {
        this.logger.warn(`Message ID ${messageId} has already been processed.`);
        return null; // Indicate already processed
      }
      this.logger.error(`Failed to save processed message ${messageId}: ${error.message}`, error.stack);
      throw error; // Re-throw other errors
    }
  }

  async hasBeenProcessed(messageId: string): Promise<boolean> {
    const count = await this.repository.count({ where: { messageId } });
    return count > 0;
  }
}

// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { TypeOrmModule } from '@nestjs/typeorm'; // Import TypeOrmModule
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { ProcessedMessage } from './database/entities/processed-message.entity'; // Import Entity
import { ProcessedMessageDao } from './database/daos/processed-message.dao'; // Import DAO
// Import other modules later (SnsModule, SqsModule)

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true,
      envFilePath: '.env',
    }),
    TypeOrmModule.forRootAsync({ // Use forRootAsync for dependency injection
      imports: [ConfigModule], // Import ConfigModule here
      useFactory: (configService: ConfigService) => ({
        type: 'sqlite',
        database: 'processed_messages.db', // Name of the SQLite file
        entities: [ProcessedMessage],      // Register the entity
        synchronize: true, // DEV ONLY: Auto-creates schema. Disable in prod and use migrations.
        logging: configService.get<string>('NODE_ENV') !== 'production', // Log SQL in dev
      }),
      inject: [ConfigService], // Inject ConfigService into the factory
    }),
    TypeOrmModule.forFeature([ProcessedMessage]), // Make repository available
    // ... other modules will be added here
  ],
  controllers: [AppController],
  providers: [AppService, ProcessedMessageDao], // Provide the DAO
  exports: [ProcessedMessageDao], // Export DAO if needed by other modules
})
export class AppModule {}

// src/sns/sns.publisher.service.ts
import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { SNSClient, PublishCommand, PublishCommandInput } from '@aws-sdk/client-sns';
import { v4 as uuid } from 'uuid';

@Injectable()
export class SnsPublisherService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(SnsPublisherService.name);
  private snsClient: SNSClient;
  private topicArn: string;

  constructor(private configService: ConfigService) {}

  onModuleInit() {
    const region = this.configService.get<string>('AWS_REGION');
    this.topicArn = this.configService.get<string>('SNS_TOPIC_ARN');

    if (!region || !this.topicArn) {
        throw new Error('AWS Region or SNS Topic ARN not configured in environment variables.');
    }

    this.snsClient = new SNSClient({
      region: region,
      credentials: {
        accessKeyId: this.configService.get<string>('AWS_ACCESS_KEY_ID'),
        secretAccessKey: this.configService.get<string>('AWS_SECRET_ACCESS_KEY'),
      },
    });
    this.logger.log('SNS Client Initialized');
  }

  async publish<T>(message: T, messageGroupId?: string): Promise<string | undefined> {
    // Ensure message is stringified JSON for structured data
    const messageBody = typeof message === 'string' ? message : JSON.stringify(message);

    const params: PublishCommandInput = {
      TopicArn: this.topicArn,
      Message: messageBody,
      // MessageGroupId and MessageDeduplicationId are required for FIFO topics
      // If using a Standard topic, they are optional but can help with ordering/deduplication within SQS FIFO subscribers
      // MessageGroupId: messageGroupId || 'default-group', // Example group ID
      // MessageDeduplicationId: uuid(), // Use content-based hashing or a unique ID if needed for FIFO
    };

    // Remove FIFO parameters if not needed (e.g., for Standard topics)
    // Check if your topic ARN ends with .fifo before including these
    // if (!this.topicArn.endsWith('.fifo')) {
    //   delete params.MessageGroupId;
    //   delete params.MessageDeduplicationId;
    // }

    try {
      const command = new PublishCommand(params);
      const response = await this.snsClient.send(command);
      this.logger.log(`Message published to SNS Topic ${this.topicArn}. Message ID: ${response.MessageId}`);
      return response.MessageId;
    } catch (error) {
      this.logger.error(`Error publishing message to SNS Topic ${this.topicArn}: ${error.message}`, error.stack);
      // Implement retry logic or specific error handling here if needed
      throw error; // Re-throw to allow upstream handling
    }
  }

  onModuleDestroy() {
    this.snsClient?.destroy();
    this.logger.log('SNS Client Destroyed');
  }
}

// src/sns/sns.controller.ts
import { Controller, Post, Body, HttpCode, HttpStatus } from '@nestjs/common';
import { SnsPublisherService } from './sns.publisher.service';

interface PublishMessageDto {
  message: any; // Can be any structure you want to send
  type: 'reminder' | 'notification' | 'task'; // Example field
}

@Controller('messages') // Example endpoint prefix
export class SnsController {
  constructor(private readonly snsPublisherService: SnsPublisherService) {}

  @Post('publish')
  @HttpCode(HttpStatus.ACCEPTED) // Use 202 Accepted for async operations
  async publishMessage(@Body() body: PublishMessageDto) {
    // You might add validation using class-validator here
    const messageId = await this.snsPublisherService.publish(body);
    return {
      status: 'Message submitted for publishing',
      messageId: messageId,
    };
  }
}

// src/sns/sns.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config'; // Import if not global
import { SnsPublisherService } from './sns.publisher.service';
import { SnsController } from './sns.controller';

@Module({
  imports: [ConfigModule], // Only needed if ConfigModule isn't global
  controllers: [SnsController],
  providers: [SnsPublisherService],
  exports: [SnsPublisherService], // Export service if needed elsewhere
})
export class SnsModule {}

// src/app.module.ts
// ... other imports
import { SnsModule } from './sns/sns.module'; // Import SNS module
import { SqsModule } from './sqs/sqs.module'; // Placeholder for later
import { TypeOrmModule } from '@nestjs/typeorm';
import { ConfigModule } from '@nestjs/config';
import { ProcessedMessage } from './database/entities/processed-message.entity';
import { ProcessedMessageDao } from './database/daos/processed-message.dao';
import { AppController } from './app.controller';
import { AppService } from './app.service';


@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true, envFilePath: '.env' }),
    TypeOrmModule.forRootAsync({ /* ... TypeORM config ... */ }),
    TypeOrmModule.forFeature([ProcessedMessage]),
    SnsModule, // Add SnsModule
    // SqsModule, // Add SqsModule later
  ],
  controllers: [AppController],
  providers: [AppService, ProcessedMessageDao],
  exports: [ProcessedMessageDao],
})
export class AppModule {}

// src/sqs/utils/constants.ts
export enum MessageSourceType {
  SQS = 'SQS', // Indicates direct SQS message or Raw SNS message
  SNS = 'SNS', // Indicates message originated from SNS (non-raw delivery)
  UNKNOWN = 'UNKNOWN',
}

// src/sqs/sqs.consumer.service.ts
import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { SQSClient, Message, DeleteMessageCommand } from '@aws-sdk/client-sqs';
import { Consumer } from 'sqs-consumer';
import { ProcessedMessageDao } from '../database/daos/processed-message.dao';
import { MessageSourceType } from './utils/constants';

@Injectable()
export class SqsConsumerService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(SqsConsumerService.name);
  private consumer: Consumer;
  private sqsClient: SQSClient;
  private queueUrl: string;

  constructor(
    private configService: ConfigService,
    private processedMessageDao: ProcessedMessageDao, // Inject DAO
  ) {}

  onModuleInit() {
    const region = this.configService.get<string>('AWS_REGION');
    this.queueUrl = this.configService.get<string>('SQS_QUEUE_URL');

    if (!region || !this.queueUrl) {
      throw new Error('AWS Region or SQS Queue URL not configured.');
    }

    // Create an SQS client instance for the consumer and for manual deletion.
    this.sqsClient = new SQSClient({
      region: region,
      credentials: {
        accessKeyId: this.configService.get<string>('AWS_ACCESS_KEY_ID'),
        secretAccessKey: this.configService.get<string>('AWS_SECRET_ACCESS_KEY'),
      },
    });

    this.initializeConsumer();
    this.consumer.start();
    this.logger.log(`SQS Consumer started polling queue: ${this.queueUrl}`);
  }

  private initializeConsumer() {
    this.consumer = Consumer.create({
      queueUrl: this.queueUrl,
      sqs: this.sqsClient, // Pass the initialized SQS client
      handleMessage: async (message: Message) => {
        await this.processMessage(message);
      },
      batchSize: 10, // Process up to 10 messages concurrently (adjust as needed)
      visibilityTimeout: 60, // Must be >= handleMessage timeout. In seconds.
      waitTimeSeconds: 10, // Use long polling (reduces costs, improves efficiency)
      // --- IMPORTANT ---
      // Set to false: We will manually delete messages only after successful processing.
      shouldDeleteMessages: false,
    });

    this.consumer.on('error', (err, message) => {
      this.logger.error(`SQS Consumer error: ${err.message}`, { messageId: message?.MessageId });
      // Add more specific error handling/alerting here
    });

    this.consumer.on('processing_error', (err, message) => {
      this.logger.error(`SQS Consumer processing error: ${err.message}`, { messageId: message?.MessageId });
      // Decide if the message should be retried or sent to a Dead Letter Queue (DLQ)
      // Note: If processing fails repeatedly, it will eventually go to DLQ if configured.
    });

    this.consumer.on('timeout_error', (err, message) => {
        this.logger.error(`SQS Consumer message visibility timeout error: ${err.message}`, { messageId: message?.MessageId });
        // This usually means processing took longer than the visibility timeout.
        // Increase visibilityTimeout if needed, or optimize processing.
    });

    this.consumer.on('stopped', () => {
        this.logger.log('SQS Consumer stopped.');
    });
  }

  private async processMessage(message: Message): Promise<void> {
    this.logger.debug(`Received message: ${message.MessageId}`);

    if (!message.MessageId || !message.Body || !message.ReceiptHandle) {
        this.logger.error('Received malformed message, skipping.', message);
        // Consider sending to DLQ or logging more details
        // We cannot delete without ReceiptHandle, so let it timeout and potentially retry/DLQ.
        return;
    }

    const messageId = message.MessageId;
    const messageBody = message.Body;
    const receiptHandle = message.ReceiptHandle; // Capture receipt handle early

    try {
      // 1. Check for Duplicates (Idempotency Check)
      const alreadyProcessed = await this.processedMessageDao.hasBeenProcessed(messageId);
      if (alreadyProcessed) {
        this.logger.warn(`Message ${messageId} already processed. Deleting from queue.`);
        await this.deleteMessageFromQueue(receiptHandle);
        return;
      }

      // 2. Parse and Process the Message Content
      // Attempts to handle both raw SNS delivery (enabled) and non-raw (disabled).
      // The primary path assumes raw delivery is enabled, where message.Body is the direct payload.
      let payload: any;
      let sourceType = MessageSourceType.UNKNOWN; // Default
      try {
          payload = JSON.parse(messageBody);
          // Check if it looks like an SNS notification wrapper (non-raw delivery)
          if (payload.Type === 'Notification' && payload.TopicArn && payload.Message) {
              sourceType = MessageSourceType.SNS;
              // Re-parse the actual message content from the wrapper
              try {
                payload = JSON.parse(payload.Message);
              } catch (innerParseError) {
                 this.logger.warn(`Could not JSON parse inner SNS message for ${messageId}. Processing inner content as raw string.`);
                 payload = payload.Message; // Treat inner message as raw string
              }
          } else {
              // Assume it's a direct SQS message or a raw SNS message (valid JSON)
              sourceType = MessageSourceType.SQS;
          }
      } catch (parseError) {
        // Not valid JSON - treat as a plain text SQS message or raw SNS message
        this.logger.warn(`Message ${messageId} body is not valid JSON. Processing as raw string.`);
        payload = messageBody;
        sourceType = MessageSourceType.SQS;
      }

      this.logger.log(`Processing message ${messageId} of type ${sourceType}`);
      // --- YOUR BUSINESS LOGIC HERE ---
      // Example: Handle different message types based on payload content
      if (payload?.type === 'reminder') {
        this.logger.log(`Handling reminder: ${payload.message}`);
        // await sendReminderNotification(payload);
      } else if (payload?.type === 'notification') {
         this.logger.log(`Handling notification: ${payload.message}`);
         // await sendGenericNotification(payload);
      } else {
         this.logger.log(`Handling generic task:`, payload);
         // await processGenericTask(payload);
      }
      // Simulate work
      await new Promise(resolve => setTimeout(resolve, 500));
      // --- END BUSINESS LOGIC ---


      // 3. Record Successful Processing (Idempotency)
      // Use the original SQS message ID and body for tracking
      const recorded = await this.processedMessageDao.recordMessageProcessed(messageId, messageBody, sourceType);
      if (recorded === null) {
        // This means it was processed by another concurrent consumer between step 1 and now.
        this.logger.warn(`Message ${messageId} was processed concurrently. Skipping delete.`);
        // Do not delete, let the other consumer handle deletion.
        return;
      }
      this.logger.log(`Successfully processed and recorded message ${messageId}.`);

      // 4. Delete Message from Queue
      await this.deleteMessageFromQueue(receiptHandle);
      this.logger.debug(`Deleted message ${messageId} from queue.`);

    } catch (error) {
      // If any step fails (business logic, DB recording), DO NOT delete the message.
      // It will become visible again after the visibility timeout for retry.
      // Configure a Dead Letter Queue (DLQ) in SQS to handle messages that fail repeatedly.
      this.logger.error(`Failed to process message ${messageId}: ${error.message}`, error.stack);
      // Re-throwing might trigger 'processing_error' event depending on `sqs-consumer` version/handling.
      // throw error; // Or handle specific errors without re-throwing if appropriate
    }
  }

  private async deleteMessageFromQueue(receiptHandle: string): Promise<void> {
    try {
      const deleteCommand = new DeleteMessageCommand({
        QueueUrl: this.queueUrl,
        ReceiptHandle: receiptHandle,
      });
      await this.sqsClient.send(deleteCommand);
    } catch (error) {
      this.logger.error(`Failed to delete message with receipt handle ${receiptHandle.substring(0, 10)}... from queue ${this.queueUrl}: ${error.message}`, error.stack);
      // This is problematic - the message was processed but not deleted.
      // It will likely be processed again after the visibility timeout.
      // Idempotency check should prevent duplicate *actions*, but logging this error is crucial.
      // Consider adding monitoring/alerting for delete failures.
    }
  }

  onModuleDestroy() {
    this.consumer?.stop();
    this.sqsClient?.destroy();
    this.logger.log('SQS Consumer and Client Destroyed');
  }
}

// src/sqs/sqs.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config'; // Import if not global
import { TypeOrmModule } from '@nestjs/typeorm'; // Needed for DAO injection
import { ProcessedMessage } from '../database/entities/processed-message.entity';
import { ProcessedMessageDao } from '../database/daos/processed-message.dao';
import { SqsConsumerService } from './sqs.consumer.service';

@Module({
  imports: [
    ConfigModule, // Only needed if ConfigModule isn't global
    TypeOrmModule.forFeature([ProcessedMessage]), // Make repository available for DAO
  ],
  providers: [SqsConsumerService, ProcessedMessageDao], // Provide Consumer and DAO
  exports: [SqsConsumerService], // Export if needed elsewhere
})
export class SqsModule {}

// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { TypeOrmModule } from '@nestjs/typeorm';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { ProcessedMessage } from './database/entities/processed-message.entity';
// Remove DAO import/providers/exports if handled solely by SqsModule
// import { ProcessedMessageDao } from './database/daos/processed-message.dao';
import { SnsModule } from './sns/sns.module';
import { SqsModule } from './sqs/sqs.module'; // Import SQS module

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true,
      envFilePath: '.env',
    }),
    TypeOrmModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (configService: ConfigService) => ({
        type: 'sqlite',
        database: 'processed_messages.db',
        entities: [ProcessedMessage],
        synchronize: configService.get<string>('NODE_ENV') !== 'production', // Safer default
        logging: configService.get<string>('NODE_ENV') !== 'production',
      }),
      inject: [ConfigService],
    }),
    // No need for TypeOrmModule.forFeature here if DAO handles it internally
    SnsModule,
    SqsModule, // Add SqsModule
  ],
  controllers: [AppController],
  providers: [AppService], // DAO is now managed within SqsModule
  // exports: [], // No need to export DAO from here anymore
})
export class AppModule {}