base.listener.ts

import { Consumer, SQSMessage } from 'sqs-consumer';
import { Agent } from 'https';
import * as Sentry from '@sentry/node';
import { Event } from './base.event';
import AWS from '../../utils/awsConfig';
import { APP_LOGGER } from '../../utils/logger';

export abstract class BaseListener<T extends Event> {
  protected abstract queueUrl: string;
  protected abstract visibilityTimeout: number;
  protected abstract workerName: string;

  protected getMessage(message: SQSMessage): T {
    return JSON.parse(message.Body) as T;
  }

  listen(): void {
    APP_LOGGER.info(`🚀 ${this.workerName} worker listening`);

    if (process.env.NODE_ENV === 'production') {
      Sentry.init({
        dsn: process.env.SENTRY_DSN,
        tracesSampleRate: 1.0,
        environment: process.env.NODE_ENV,
      });
    }

    const sqs =
      process.env.NODE_ENV === 'production'
        ? new AWS.SQS({ httpOptions: { agent: new Agent({ keepAlive: true }) } })
        : new AWS.SQS();

    const app = Consumer.create({
      sqs,
      queueUrl: this.queueUrl,
      visibilityTimeout: this.visibilityTimeout,
      handleMessage: async (message: SQSMessage) => {
        APP_LOGGER.info(`Message ${message.MessageId} received`);
        this.onMessage(this.getMessage(message));
        APP_LOGGER.info(`Message ${message.MessageId} successfully processed`);
      },
    });

    app.start();
  }

  // eslint-disable-next-line no-unused-vars
  abstract onMessage(message: T): Promise<void>;
}