fix(#12): add integration test + feedback

docker-compose.yml

@@ -11,8 +11,9 @@ services:
       - NODE_ENV=${NODE_ENV:-production}
       - EXTERNAL_PORT=${EXTERNAL_PORT:-3000}
       - COOKIE_PRIVATE_KEY=${COOKIE_PRIVATE_KEY}
-      - QUEUE_PRIVATE_KEY=${QUEUE_PRIVATE_KEY}
+      - WORKER_PRIVATE_KEY=${WORKER_PRIVATE_KEY}
       - CONFIG_NAME=${CONFIG_NAME}
+      - CHT_DEV_HTTP=${CHT_DEV_HTTP}
       - CHT_DEV_URL_PORT=${CHT_DEV_URL_PORT}
       - REDIS_HOST=${REDIS_HOST:-redis}
       - REDIS_PORT=${REDIS_PORT:-6379}
@@ -31,7 +32,7 @@ services:
       - NODE_ENV=${NODE_ENV:-production}
       - REDIS_HOST=${REDIS_HOST:-redis}
       - REDIS_PORT=${REDIS_PORT:-6379}
-      - QUEUE_PRIVATE_KEY=${QUEUE_PRIVATE_KEY}
+      - WORKER_PRIVATE_KEY=${WORKER_PRIVATE_KEY}
     depends_on:
       - redis
 

env.example

@@ -1,6 +1,6 @@
 NODE_ENV=                                             # set to "dev" to use CHT_DEV_URL_PORT below, leave empty for production
 COOKIE_PRIVATE_KEY=                                   # unique random key, use uuidgen to populate  
-QUEUE_PRIVATE_KEY=                                    # unique random key, use uuidgen to populate. different from COOKIE_PRIVATE_KEY
+WORKER_PRIVATE_KEY=                                   # unique random key, use uuidgen to populate. different from COOKIE_PRIVATE_KEY
 CONFIG_NAME=chis-ke                                   # Name of the configuration
 PORT=3000                                             # for development environmentcontainer)
 EXTERNAL_PORT=3000                                    # for docker

package.json

@@ -28,6 +28,7 @@
     "fastify": "^4.27.0",
     "fastify-sse-v2": "^3.1.2",
     "html-minifier": "^4.0.0",
+    "ioredis": "^5.4.1",
     "jsonwebtoken": "^9.0.2",
     "jszip": "^3.10.1",
     "libphonenumber-js": "^1.10.48",
@@ -64,15 +65,18 @@
   },
   "scripts": {
     "cp-package-json": "cp package.json ./src",
-    "test": "npx ts-mocha test/{,**}/*.spec.ts",
+    "test": "npx ts-mocha test/{,**}/*.spec.ts --exclude test/integration/{,**}/*.spec.ts",
     "build": "npm run cp-package-json && npx tsc",
     "lint": "npx eslint --color --cache .",
     "start": "node dist/index.js",
     "dev": "tsc-watch --onSuccess \"node dist/index.js\"",
     "publish:cht-user-management": "SERVICE=cht-user-management node scripts/publish.js",
     "publish:cht-user-management-worker": "SERVICE=cht-user-management-worker node scripts/publish.js",
     "publish": "npm run publish:cht-user-management && npm run publish:cht-user-management-worker",
-    "start:worker": "node dist/worker/main.js"
+    "start:worker": "node dist/worker/main.js",
+    "docker-start-redis": "npm run docker-stop-redis && docker run -d --rm --name test-redis -p 6363:6379 redis && sh test/scripts/wait_for_redis.sh 6363",
+    "docker-stop-redis": "docker stop test-redis || true",
+    "integration-test": "npm run docker-start-redis && npx ts-mocha test/integration/{,**}/*.spec.ts && npm run docker-stop-redis"
   },
   "repository": {
     "type": "git",

src/config/config-worker.ts

@@ -0,0 +1,39 @@
+import Redis from 'ioredis';
+import { env } from 'process';
+
+const environment = env as unknown as { 
+  REDIS_HOST: string; 
+  REDIS_PORT: string; 
+};
+
+export const WorkerConfig = {
+  redisConnection: {
+    host: environment.REDIS_HOST,
+    port: Number(environment.REDIS_PORT),
+  },
+  moveContactQueue: 'MOVE_CONTACT_QUEUE',
+};
+
+const assertRedisConfig = () => {
+  const {host, port} = WorkerConfig.redisConnection;
+  if (!host) {
+    throw new Error('REDIS_HOST is not defined');
+  }
+  if (!port || isNaN(port) || port <= 0) {
+    throw new Error('REDIS_PORT is not defined or invalid');
+  }
+};
+
+export const checkRedisConnection = async () => {
+  assertRedisConfig();
+
+  const config = WorkerConfig.redisConnection;
+  const redis = new Redis(config.port, config.host, {lazyConnect: true});
+  try {
+    await redis.connect();
+  } catch (error : any) {
+    throw new Error(`Failed to connect to Redis at ${config.host}:${config.port}: ${error}`);
+  } finally {
+    redis?.disconnect();
+  }
+};

src/lib/authentication.ts

@@ -4,10 +4,9 @@ import ChtSession from './cht-session';
 
 const LOGIN_EXPIRES_AFTER_MS = 2 * 24 * 60 * 60 * 1000;
 const QUEUE_SESSION_EXPIRATION = '48h';
-const { COOKIE_PRIVATE_KEY, QUEUE_PRIVATE_KEY } = process.env;
+const { COOKIE_PRIVATE_KEY, WORKER_PRIVATE_KEY } = process.env;
 const PRIVATE_KEY_SALT = '_'; // change to logout all users
 const COOKIE_SIGNING_KEY = COOKIE_PRIVATE_KEY + PRIVATE_KEY_SALT;
-const QUEUE_SIGNING_KEY = QUEUE_PRIVATE_KEY + PRIVATE_KEY_SALT;
 
 export default class Auth {
   public static AUTH_COOKIE_NAME = 'AuthToken';
@@ -16,8 +15,11 @@ export default class Auth {
     if (!COOKIE_PRIVATE_KEY) {
       throw new Error('.env missing COOKIE_PRIVATE_KEY');
     }
-    if (!QUEUE_PRIVATE_KEY) {
-      throw new Error('.env missing QUEUE_PRIVATE_KEY');
+    if (!WORKER_PRIVATE_KEY) {
+      throw new Error('.env missing WORKER_PRIVATE_KEY');
+    }
+    if (COOKIE_PRIVATE_KEY === WORKER_PRIVATE_KEY) {
+      throw new Error('COOKIE_PRIVATE_KEY should be different from WORKER_PRIVATE_KEY');
     }
   }
 
@@ -43,12 +45,12 @@ export default class Auth {
     return this.decodeToken(token, COOKIE_SIGNING_KEY);
   }
 
-  public static encodeTokenForQueue(session: ChtSession) {
-    return this.encodeToken(session, QUEUE_SIGNING_KEY, QUEUE_SESSION_EXPIRATION);
+  public static encodeTokenForWorker(session: ChtSession) {
+    return this.encodeToken(session, `${WORKER_PRIVATE_KEY}`, QUEUE_SESSION_EXPIRATION);
   }
 
-  public static decodeTokenForQueue(token: string): ChtSession {
-    return this.decodeToken(token, QUEUE_SIGNING_KEY);
+  public static decodeTokenForWorker(token: string): ChtSession {
+    return this.decodeToken(token, `${WORKER_PRIVATE_KEY}`);
   }
 
   public static cookieExpiry() {

src/lib/move.ts

@@ -4,14 +4,16 @@ import { ChtApi } from './cht-api';
 import RemotePlaceResolver from './remote-place-resolver';
 import Place from '../services/place';
 
-import { JobParams, IQueue } from '../lib/queues';
+import { JobParams, IQueue, getMoveContactQueue } from '../lib/queues';
 import Auth from './authentication';
 import { MoveContactData } from '../worker/move-contact-worker';
 
 export default class MoveLib {
   constructor() { }
 
-  public static async move(formData: any, contactType: ContactType, sessionCache: SessionCache, chtApi: ChtApi, moveContactQueue: IQueue) {
+  public static async move(
+    formData: any, contactType: ContactType, sessionCache: SessionCache, chtApi: ChtApi, moveContactQueue: IQueue = getMoveContactQueue()
+  ) {
     const fromLineage = await resolve('from_', formData, contactType, sessionCache, chtApi);
     const toLineage = await resolve('to_', formData, contactType, sessionCache, chtApi);
 
@@ -50,7 +52,7 @@ export default class MoveLib {
       contactId: fromId,
       parentId: toId,
       instanceUrl: `http${authInfo.useHttp ? '' : 's'}://${authInfo.domain}`,
-      sessionToken: Auth.encodeTokenForQueue(chtApi.chtSession),
+      sessionToken: Auth.encodeTokenForWorker(chtApi.chtSession),
     };
   }
 }

src/lib/queues.ts

@@ -1,21 +1,6 @@
 import { v4 } from 'uuid';
-import { env } from 'process';
 import { JobsOptions, Queue, ConnectionOptions } from 'bullmq';
-
-const MOVE_CONTACT_QUEUE = 'MOVE_CONTACT_QUEUE';
-
-const environment = env as unknown as { 
-  REDIS_HOST: string; 
-  REDIS_PORT: string; 
-  QUEUE_PRIVATE_KEY: string;
-};
-
-export const QUEUE_PRIVATE_KEY = environment.QUEUE_PRIVATE_KEY;
-
-export const redisConnection: ConnectionOptions = {
-  host: environment.REDIS_HOST,
-  port: Number(environment.REDIS_PORT)
-};
+import { WorkerConfig } from '../config/config-worker';
 
 export interface IQueue {
   name: string;
@@ -32,9 +17,9 @@ export class BullQueue implements IQueue {
   public readonly name: string;
   public readonly bullQueue: Queue;
 
-  constructor(queueName: string) {
+  constructor(queueName: string, connection: ConnectionOptions) {
     this.name = queueName;
-    this.bullQueue = new Queue(queueName, { connection: redisConnection });
+    this.bullQueue = new Queue(queueName, { connection });
   }
 
   public async add(jobParams: JobParams): Promise<string> {
@@ -44,7 +29,13 @@ export class BullQueue implements IQueue {
     await this.bullQueue.add(jobName, jobData, { jobId, ...jobOpts });
     return jobId;
   }
+
+  public async close(): Promise<void> {
+    await this.bullQueue.close();
+  }
 }
 
-// Create a singleton instance of QueueManager
-export const moveContactQueue = new BullQueue(MOVE_CONTACT_QUEUE);
+export const getMoveContactQueue = () => new BullQueue(
+  WorkerConfig.moveContactQueue, 
+  WorkerConfig.redisConnection
+);

src/plugins/bullmq.ts

@@ -5,7 +5,7 @@ import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
 import { createBullBoard } from '@bull-board/api';
 import { FastifyAdapter } from '@bull-board/fastify';
 
-import { moveContactQueue } from '../lib/queues';
+import { getMoveContactQueue } from '../lib/queues';
 
 
 async function bullMQBoardPlugin(fastify: FastifyInstance) {
@@ -15,7 +15,7 @@ async function bullMQBoardPlugin(fastify: FastifyInstance) {
   createBullBoard({
     queues: [
       new BullMQAdapter(
-        moveContactQueue.bullQueue
+        getMoveContactQueue().bullQueue
       ),
     ],
     serverAdapter,

src/routes/move.ts

@@ -5,7 +5,6 @@ import { ChtApi } from '../lib/cht-api';
 import { FastifyInstance } from 'fastify';
 import MoveLib from '../lib/move';
 import SessionCache from '../services/session-cache';
-import { moveContactQueue } from '../lib/queues';
 
 export default async function sessionCache(fastify: FastifyInstance) {
   fastify.get('/move/:placeType', async (req, resp) => {
@@ -35,7 +34,7 @@ export default async function sessionCache(fastify: FastifyInstance) {
     const chtApi = new ChtApi(req.chtSession);
 
     try {
-      const result = await MoveLib.move(formData, contactType, sessionCache, chtApi, moveContactQueue);
+      const result = await MoveLib.move(formData, contactType, sessionCache, chtApi);
 
       const tmplData = {
         view: 'move',

src/server.ts

@@ -12,6 +12,7 @@ import path from 'path';
 
 import Auth from './lib/authentication';
 import SessionCache from './services/session-cache';
+import { checkRedisConnection } from './config/config-worker';
 
 const build = (opts: FastifyServerOptions): FastifyInstance => {
   const fastify = Fastify(opts);
@@ -44,6 +45,7 @@ const build = (opts: FastifyServerOptions): FastifyInstance => {
   });
 
   Auth.assertEnvironmentSetup();
+  checkRedisConnection();
 
   fastify.addHook('preValidation', async (req: FastifyRequest, reply: FastifyReply) => {
     if (req.unauthenticated || req.routeOptions.url === '/public/*') {

src/worker/main.ts

@@ -2,10 +2,15 @@ import { config } from 'dotenv';
 
 config();
 
-import { moveContactQueue } from '../lib/queues';
 import { MoveContactWorker } from './move-contact-worker';
+import { WorkerConfig, checkRedisConnection } from '../config/config-worker';
 
-(() => {
-  new MoveContactWorker(moveContactQueue.name);
+(async () => {
+  const { moveContactQueue, redisConnection} = WorkerConfig;
+  await checkRedisConnection();
+  MoveContactWorker.processQueue(
+    moveContactQueue, 
+    redisConnection
+  );
   console.log(`🚀 Move Contact Worker is listening`);
 })();