Test queue

backend/package.json

@@ -10,17 +10,19 @@
 	"dependencies": {
 		"@bcgov/citz-imb-express-utilities": "1.0.0",
 		"@bcgov/citz-imb-sso-js-core": "1.0.0",
+		"amqplib": "0.10.4",
 		"cookie-parser": "1.4.7",
 		"cors": "2.8.5",
 		"express": "4.21.0",
 		"express-rate-limit": "7.4.1",
 		"mongoose": "8.7.0",
-		"zod": "3.23.8",
 		"tsconfig-paths": "4.2.0",
-		"typescript": "5.6.2"
+		"typescript": "5.6.2",
+		"zod": "3.23.8"
 	},
 	"devDependencies": {
 		"@biomejs/biome": "1.9.3",
+		"@types/amqplib": "0.10.5",
 		"@types/cookie-parser": "1.4.7",
 		"@types/cors": "2.8.17",
 		"@types/express": "5.0.0",

backend/src/config/env.ts

@@ -1,13 +1,14 @@
 // Environment variables set in the .env file.
 const {
 	NODE_ENV,
-	ENVIRONMENT,
+	ENVIRONMENT = "dev",
 	BACKEND_URL,
+	RABBITMQ_URL,
 	BACKEND_PORT,
-	MONGO_USER,
-	MONGO_PASSWORD,
-	MONGO_DATABASE_NAME,
-	MONGO_HOST,
+	MONGO_USER = "admin",
+	MONGO_PASSWORD = "password",
+	MONGO_DATABASE_NAME = "dats",
+	MONGO_HOST = "mongo",
 	SSO_ENVIRONMENT = "dev",
 	SSO_REALM = "standard",
 	SSO_PROTOCOL = "openid-connect",
@@ -22,6 +23,7 @@ export default {
 	NODE_ENV,
 	ENVIRONMENT,
 	BACKEND_URL,
+	RABBITMQ_URL,
 	MONGO_USER,
 	MONGO_PASSWORD,
 	MONGO_DATABASE_NAME,

backend/src/express.ts

@@ -9,7 +9,8 @@ import rateLimit from "express-rate-limit";
 import cookieParser from "cookie-parser";
 import { CORS_OPTIONS, RATE_LIMIT_OPTIONS } from "./config";
 import { ENV } from "./config";
-import { router } from "./modules/auth";
+import { router as authRouter } from "./modules/auth";
+import { router as testRouter } from "./modules/test";
 import { protectedRoute } from "./modules/auth/middleware";
 import type { Request, Response } from "express";
 
@@ -29,7 +30,8 @@ app.set("view engine", "ejs");
 // Disabled because it exposes information about the used framework to potential attackers.
 app.disable("x-powered-by");
 
-app.use("/auth", router);
+app.use("/auth", authRouter);
+app.use("/test", testRouter);
 
 // Add express utils middleware.
 app.use(expressUtilitiesMiddleware);

backend/src/modules/test/controllers/addToRabbitQueue.ts

@@ -0,0 +1,26 @@
+import type { Request, Response } from "express";
+import { errorWrapper } from "@bcgov/citz-imb-express-utilities";
+import { addToQueue } from "../queue";
+
+// Adds to test queue which processes one request per 10 seconds.
+export const addToRabbitQueue = errorWrapper(async (req: Request, res: Response) => {
+	try {
+		const jobID = `job-${Date.now()}`;
+
+		// Add the job ID to the RabbitMQ queue
+		await addToQueue(jobID);
+
+		// Respond with job ID
+		res
+			.status(200)
+			.json({ success: true, message: "Queue will process a job every 10 seconds.", jobID });
+	} catch (error) {
+		res.status(500).json({
+			success: false,
+			error:
+				error instanceof Error
+					? error.message
+					: "An unknown error occurred during addToRabbitQueue.",
+		});
+	}
+});

backend/src/modules/test/controllers/index.ts

@@ -0,0 +1 @@
+export * from "./addToRabbitQueue";

backend/src/modules/test/index.ts

@@ -0,0 +1 @@
+export { default as router } from "./router";

backend/src/modules/test/queue.ts

@@ -0,0 +1,60 @@
+import amqp from "amqplib";
+import { ENV } from "src/config";
+
+const { RABBITMQ_URL } = ENV;
+const QUEUE_NAME = "test_queue";
+
+// Connect to RabbitMQ and create a channel
+const connectToRabbitMQ = async () => {
+	try {
+		if (!RABBITMQ_URL) throw new Error("RABBITMQ_URL env variable is undefined.");
+		const connection = await amqp.connect(RABBITMQ_URL);
+		const channel = await connection.createChannel();
+		await channel.assertQueue(QUEUE_NAME, { durable: false });
+		return channel;
+	} catch (error) {
+		console.error("Failed to connect to RabbitMQ:", error);
+		throw error;
+	}
+};
+
+// Create a channel instance for reuse
+let channelPromise: Promise<amqp.Channel>;
+
+// Get or create a channel
+export const getChannel = async () => {
+	if (!channelPromise) {
+		channelPromise = connectToRabbitMQ();
+	}
+	return channelPromise;
+};
+
+// Add a message to the queue
+export const addToQueue = async (message: string) => {
+	const channel = await getChannel();
+	channel.sendToQueue(QUEUE_NAME, Buffer.from(message));
+};
+
+// Start consuming messages from the queue
+export const startQueueConsumer = async () => {
+	try {
+		const channel = await getChannel();
+		channel.prefetch(1); // Only process one message at a time
+		channel.consume(
+			QUEUE_NAME,
+			(msg) => {
+				if (msg) {
+					const jobID = msg.content.toString();
+					console.log(`Processed job: ${jobID}`);
+					setTimeout(() => channel.ack(msg), 10 * 1000);
+				}
+			},
+			{ noAck: false },
+		);
+	} catch (error) {
+		console.error("Failed to consume messages from RabbitMQ:", error);
+	}
+};
+
+// Start the consumer immediately
+startQueueConsumer();

backend/src/modules/test/router.ts

@@ -0,0 +1,8 @@
+import { Router } from "express";
+import { addToRabbitQueue } from "./controllers";
+
+const router = Router();
+
+router.post("/addToQueue", addToRabbitQueue);
+
+export default router;

compose.yaml

@@ -21,6 +21,7 @@ services:
     environment:
       - MONGO_HOST=mongo
       - BACKEND_URL=http://localhost:${BACKEND_PORT:-3200}
+      - RABBITMQ_URL=amqp://${RABBITMQ_DEFAULT_USER:-admin}:${RABBITMQ_DEFAULT_PASS:-pass}@rabbitmq:5672
     ports:
       - ${BACKEND_PORT:-3200}:3200
     volumes:
@@ -44,7 +45,7 @@ services:
     volumes:
       - mongo_data:/data/db
     ports:
-      - ${MONGO_EXTERNAL_PORT}:27017
+      - ${MONGO_EXTERNAL_PORT:-3300}:27017
     networks:
       - app-network
     healthcheck:
@@ -60,12 +61,14 @@ services:
     container_name: rabbitmq
     restart: always
     ports:
-      - ${RABBITMQ_EXTERNAL_PORT}:5672        # RabbitMQ default port for clients
-      - '15672:15672'      # RabbitMQ Management UI port
+      - ${RABBITMQ_EXTERNAL_PORT:-3400}:5672    # RabbitMQ default port for clients
+      - '15672:15672'                           # RabbitMQ Management UI port
     env_file: .env
     volumes:
       - rabbitmq_data:/var/lib/rabbitmq
     command: ["bash", "-c", "chmod 400 /var/lib/rabbitmq/.erlang.cookie; rabbitmq-server"]
+    networks:
+      - app-network
     healthcheck:
       test: ["CMD-SHELL", "rabbitmqctl status || exit 1"]
       interval: 30s # Perform the health check every 30 seconds.