Create an internal queue manager library (#446)

* feat: create an internal queue manager lib

* wip: tests

* fix: basic unit tests including mocks

* feat: use create a basic producer to send messages over to memphis cloud

* fix: linter

* fix: remove novu module from app

* fix: tests

* fix: linter

---------

Co-authored-by: orig <[email protected]>
Co-authored-by: orig <[email protected]>

.example.env

@@ -33,3 +33,10 @@ JWT_REFRESH_SECRET=abc1234
 # NOVU - You don't need this when running locally (just verify your email from the database)
 NOVU_API_KEY=Get it from https://novu.co/
 
+
+# Memphis (Event streaming service)
+MEMPHIS_ENABLE=false # Set this to true if you want to use the event streaming service
+MEMPHIS_HOST=Get it from https://memphis.dev/
+MEMPHIS_USERNAME=Get it from https://memphis.dev/
+MEMPHIS_PASSWORD=Get it from https://memphis.dev/
+MEMPHIS_ACCOUNT_ID=Get it from https://memphis.dev/

apps/backend/src/app.module.ts

@@ -5,7 +5,6 @@ import { AuthModule } from './auth/auth.module';
 import { AppCacheModule } from './cache/cache.module';
 import { AppConfigModule, AppConfigService } from '@reduced.to/config';
 import { AppLoggerModule } from '@reduced.to/logger';
-import { NovuModule } from './novu/novu.module';
 import { PrismaService } from '@reduced.to/prisma';
 import { UniqueConstraint } from './shared/decorators/unique/unique.decorator';
 import { CustomThrottlerGuard } from './shared/guards/custom-throttler/custom-throttler';
@@ -26,7 +25,6 @@ import { UsersModule } from './users/users.module';
     }),
     ShortenerModule,
     AuthModule,
-    NovuModule,
     UsersModule,
   ],
   providers: [

apps/backend/src/shared/decorators/ip/ip.decorator.ts

@@ -0,0 +1,7 @@
+import { createParamDecorator, ExecutionContext } from '@nestjs/common';
+
+export const Ip = createParamDecorator((data: unknown, ctx: ExecutionContext) => {
+  const request = ctx.switchToHttp().getRequest();
+  const ip = (request.headers['x-forwarded-for'] as string) || request.socket.remoteAddress;
+  return ip;
+});

apps/backend/src/shortener/producer/shortener.producer.ts

@@ -0,0 +1,12 @@
+import { Injectable } from '@nestjs/common';
+import { ProducerService } from '@reduced.to/queue-manager';
+
+const SHORTENER_PRODUCER_NAME = 'shortener';
+const SHORTENER_QUEUE_NAME = 'stats';
+
+@Injectable()
+export class ShortenerProducer extends ProducerService {
+  constructor() {
+    super(SHORTENER_PRODUCER_NAME, SHORTENER_QUEUE_NAME);
+  }
+}

apps/backend/src/shortener/shortener.controller.spec.ts

@@ -6,14 +6,16 @@ import { Test } from '@nestjs/testing';
 import { ShortenerDto } from './dto';
 import { Request } from 'express';
 import { AppLoggerModule } from '@reduced.to/logger';
+import { ShortenerProducer } from './producer/shortener.producer';
+import { QueueManagerModule, QueueManagerService } from '@reduced.to/queue-manager';
 
 describe('ShortenerController', () => {
   let shortenerController: ShortenerController;
   let shortenerService: ShortenerService;
 
   beforeEach(async () => {
     const moduleRef = await Test.createTestingModule({
-      imports: [AppConfigModule, AppLoggerModule, AppCacheModule],
+      imports: [AppConfigModule, AppLoggerModule, AppCacheModule, QueueManagerModule],
       controllers: [ShortenerController],
       providers: [
         {
@@ -24,6 +26,8 @@ describe('ShortenerController', () => {
             createShortenedUrl: jest.fn(),
           },
         },
+        QueueManagerService,
+        ShortenerProducer,
       ],
     }).compile();
 
@@ -81,15 +85,17 @@ describe('ShortenerController', () => {
     it('should return the original URL when given a valid short URL', async () => {
       jest.spyOn(shortenerService, 'getOriginalUrl').mockResolvedValue('https://github.com/origranot/reduced.to');
       const shortUrl = 'best';
-      const originalUrl = await shortenerController.findOne(shortUrl);
+      const ip = '1.2.3.4';
+      const originalUrl = await shortenerController.findOne(ip, shortUrl);
       expect(originalUrl).toBe('https://github.com/origranot/reduced.to');
     });
 
     it('should return an error if the short URL is not found in the database', async () => {
       jest.spyOn(shortenerService, 'getOriginalUrl').mockResolvedValue(null);
       const shortUrl = 'not-found';
+      const ip = '1.2.3.4';
       try {
-        await shortenerController.findOne(shortUrl);
+        await shortenerController.findOne(ip, shortUrl);
         throw new Error('Expected an error to be thrown!');
       } catch (err) {
         expect(err.message).toBe('Shortened url is wrong or expired');

apps/backend/src/shortener/shortener.controller.ts

@@ -5,20 +5,34 @@ import { ShortenerService } from './shortener.service';
 import { UserContext } from '../auth/interfaces/user-context';
 import { OptionalJwtAuthGuard } from '../auth/guards/optional-jwt-auth.guard';
 import { AppLoggerSerivce } from '@reduced.to/logger';
+import { ShortenerProducer } from './producer/shortener.producer';
+import { Ip } from '../shared/decorators/ip/ip.decorator';
 
 @Controller({
   path: 'shortener',
   version: '1',
 })
 export class ShortenerController {
-  constructor(private readonly logger: AppLoggerSerivce, private readonly shortenerService: ShortenerService) {}
+  constructor(
+    private readonly logger: AppLoggerSerivce,
+    private readonly shortenerService: ShortenerService,
+    private readonly shortenerProducer: ShortenerProducer
+  ) {}
 
   @Get(':shortenedUrl')
-  async findOne(@Param('shortenedUrl') shortenedUrl: string): Promise<string> {
+  async findOne(@Ip() ip: string, @Param('shortenedUrl') shortenedUrl: string): Promise<string> {
     const originalUrl = await this.shortenerService.getOriginalUrl(shortenedUrl);
     if (!originalUrl) {
       throw new BadRequestException('Shortened url is wrong or expired');
     }
+
+    // Send an event to the queue to update the shortened url's stats
+    await this.shortenerProducer.publish({
+      ip,
+      shortenedUrl,
+      originalUrl: originalUrl,
+    });
+
     return originalUrl;
   }
 

apps/backend/src/shortener/shortener.module.ts

@@ -2,11 +2,13 @@ import { Module } from '@nestjs/common';
 import { ShortenerController } from './shortener.controller';
 import { ShortenerService } from './shortener.service';
 import { PrismaModule } from '@reduced.to/prisma';
+import { ShortenerProducer } from './producer/shortener.producer';
+import { QueueManagerModule, QueueManagerService } from '@reduced.to/queue-manager';
 
 @Module({
-  imports: [PrismaModule],
+  imports: [PrismaModule, QueueManagerModule],
   controllers: [ShortenerController],
-  providers: [ShortenerService],
+  providers: [ShortenerService, ShortenerProducer, QueueManagerService],
   exports: [ShortenerService],
 })
 export class ShortenerModule {}

jest.preset.js

@@ -1,3 +1,8 @@
 const nxPreset = require('@nx/jest/preset').default;
 
 module.exports = { ...nxPreset };
+
+// Set the NODE_ENV to test
+process.env = Object.assign(process.env, {
+  NODE_ENV: 'test',
+});

libs/config/src/lib/config.factory.ts

@@ -37,6 +37,13 @@ export const configFactory: ConfigFactory<{ config: Configuration }> = () => {
       novu: {
         apiKey: process.env.NOVU_API_KEY,
       },
+      memphis: {
+        enable: process.env.MEMPHIS_ENABLE === 'true' || false,
+        host: process.env.MEMPHIS_HOST,
+        username: process.env.MEMPHIS_USERNAME,
+        password: process.env.MEMPHIS_PASSWORD,
+        accountId: +process.env.MEMPHIS_ACCOUNT_ID,
+      },
     },
   };
 };
@@ -81,6 +88,14 @@ export interface NovuConfig {
   apiKey: string;
 }
 
+export interface MemphisConfig {
+  enable: boolean;
+  host: string;
+  username: string;
+  password: string;
+  accountId: number;
+}
+
 export interface Configuration {
   general: GeneralConfig;
   logger: LoggerConfig;
@@ -89,4 +104,5 @@ export interface Configuration {
   redis: RedisConfig;
   jwt: JWTConfig;
   novu: NovuConfig;
+  memphis: MemphisConfig;
 }

libs/queue-manager/.eslintrc.json

@@ -0,0 +1,18 @@
+{
+  "extends": ["../../.eslintrc.json"],
+  "ignorePatterns": ["!**/*"],
+  "overrides": [
+    {
+      "files": ["*.ts", "*.tsx", "*.js", "*.jsx"],
+      "rules": {}
+    },
+    {
+      "files": ["*.ts", "*.tsx"],
+      "rules": {}
+    },
+    {
+      "files": ["*.js", "*.jsx"],
+      "rules": {}
+    }
+  ]
+}

libs/queue-manager/README.md

@@ -0,0 +1,7 @@
+# queue-manager
+
+This library was generated with [Nx](https://nx.dev).
+
+## Running unit tests
+
+Run `nx test queue-manager` to execute the unit tests via [Jest](https://jestjs.io).

libs/queue-manager/jest.config.ts

@@ -0,0 +1,11 @@
+/* eslint-disable */
+export default {
+  displayName: 'queue-manager',
+  preset: '../../jest.preset.js',
+  testEnvironment: 'node',
+  transform: {
+    '^.+\\.[tj]s$': ['ts-jest', { tsconfig: '<rootDir>/tsconfig.spec.json' }],
+  },
+  moduleFileExtensions: ['ts', 'js', 'html'],
+  coverageDirectory: '../../coverage/libs/queue-manager',
+};

libs/queue-manager/project.json

@@ -0,0 +1,30 @@
+{
+  "name": "queue-manager",
+  "$schema": "../../node_modules/nx/schemas/project-schema.json",
+  "sourceRoot": "libs/queue-manager/src",
+  "projectType": "library",
+  "targets": {
+    "lint": {
+      "executor": "@nx/linter:eslint",
+      "outputs": ["{options.outputFile}"],
+      "options": {
+        "lintFilePatterns": ["libs/queue-manager/**/*.ts"]
+      }
+    },
+    "test": {
+      "executor": "@nx/jest:jest",
+      "outputs": ["{workspaceRoot}/coverage/{projectRoot}"],
+      "options": {
+        "jestConfig": "libs/queue-manager/jest.config.ts",
+        "passWithNoTests": true
+      },
+      "configurations": {
+        "ci": {
+          "ci": true,
+          "codeCoverage": true
+        }
+      }
+    }
+  },
+  "tags": []
+}

libs/queue-manager/src/index.ts

@@ -0,0 +1,3 @@
+export * from './lib/queue-manager.module';
+export * from './lib/queue-manager.service';
+export * from './lib/producer/producer.service';

libs/queue-manager/src/lib/__mocks__/queue-manager.service.ts

@@ -0,0 +1,19 @@
+export class MemphisMock {
+  produce(payload: { stationName: string; producerName: string; message: any }) {
+    console.log(`Producing message to ${payload.stationName}`);
+  }
+}
+
+// Mock implementation for the queue manager service
+export class QueueManagerService {
+  private readonly queueManager: MemphisMock;
+
+  constructor() {
+    this.queueManager = new MemphisMock();
+  }
+
+  get client() {
+    // Mock implementation for getting the queue manager
+    return this.queueManager;
+  }
+}

libs/queue-manager/src/lib/producer/producer.service.spec.ts

@@ -0,0 +1,76 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { ProducerService } from './producer.service';
+import { AppConfigModule, AppConfigService } from '@reduced.to/config';
+import { Injectable } from '@nestjs/common';
+import { AppLoggerModule } from '@reduced.to/logger';
+import { QueueManagerService } from '../queue-manager.service';
+import { QueueManagerModule } from '../queue-manager.module';
+
+jest.mock('../queue-manager.service');
+
+describe('ProducerService', () => {
+  const TEST_PRODUCER_NAME = 'test-producer';
+  const TEST_QUEUE_NAME = 'test-queue';
+
+  @Injectable()
+  class TestProducerService extends ProducerService {
+    constructor() {
+      super(TEST_PRODUCER_NAME, TEST_QUEUE_NAME);
+    }
+  }
+
+  let service: TestProducerService;
+  let queueManager: QueueManagerService;
+  let configService: AppConfigService;
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      imports: [AppConfigModule, AppLoggerModule, QueueManagerModule],
+      providers: [TestProducerService, QueueManagerService],
+    }).compile();
+
+    service = module.get<TestProducerService>(TestProducerService);
+    queueManager = module.get<QueueManagerService>(QueueManagerService);
+    configService = module.get<AppConfigService>(AppConfigService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  it('should get the producer name', () => {
+    expect(service.name).toBe(TEST_PRODUCER_NAME);
+  });
+
+  it('should get the queue name', () => {
+    expect(service.queueName).toBe(TEST_QUEUE_NAME);
+  });
+
+  it('should not publish a message to the queue if we are in test environment', async () => {
+    const queueManagerSpy = jest.spyOn(queueManager.client, 'produce');
+
+    const PAYLOAD = { message: 'test', 1: 2 };
+
+    await service.publish(PAYLOAD);
+    expect(queueManagerSpy).toBeCalledTimes(0);
+  });
+
+  // It is not actually going to publish a message to the queue, but it is going to call the produce method of the queue-manager mock
+  it('should publish a message to the queue if we are not in test environment', async () => {
+    // Mock the config service to return the development environment
+    const configMock = jest.spyOn(configService, 'getConfig');
+    configMock.mockReturnValue({ general: { env: 'development' }, memphis: { enable: true } } as any);
+
+    const queueManagerSpy = jest.spyOn(queueManager.client, 'produce');
+
+    const PAYLOAD = { message: 'test', 1: 2 };
+
+    await service.publish(PAYLOAD);
+    expect(queueManagerSpy).toBeCalledTimes(1);
+    expect(queueManagerSpy).toBeCalledWith({
+      stationName: TEST_QUEUE_NAME,
+      producerName: TEST_PRODUCER_NAME,
+      message: PAYLOAD,
+    });
+  });
+});

libs/queue-manager/src/lib/producer/producer.service.ts

@@ -0,0 +1,34 @@
+import { Inject } from '@nestjs/common';
+import { QueueManagerService } from '../queue-manager.service';
+import { AppConfigService } from '@reduced.to/config';
+import { AppLoggerSerivce } from '@reduced.to/logger';
+
+export abstract class ProducerService {
+  @Inject(AppLoggerSerivce) private readonly logger: AppLoggerSerivce;
+  @Inject(AppConfigService) private readonly config: AppConfigService;
+  @Inject(QueueManagerService) private readonly queueManager: QueueManagerService;
+
+  constructor(private readonly producerName: string, private readonly queue: string) {}
+
+  get name() {
+    return this.producerName;
+  }
+
+  get queueName() {
+    return this.queue;
+  }
+
+  async publish(message: any) {
+    // Do not publish if Memphis is disabled or if we are in test environment
+    if (this.config.getConfig().general.env === 'test' || !this.config.getConfig().memphis.enable) {
+      return;
+    }
+
+    this.logger.log(`Publishing message to ${this.queueName} with producer ${this.producerName}`);
+    return this.queueManager.client.produce({
+      stationName: this.queue,
+      producerName: this.name,
+      message,
+    });
+  }
+}