create shipment for order creation using kafka consumer. (#15)

* add createShipment method to simulate the shipment process.
* commit the offset to mark the message as processed in consumeOrder method.
* add consumeOrders inside idnex.js.

config/kafka.js

@@ -25,9 +25,11 @@ const produceOrder = async (order) => {
 
 const consumeOrders = async (processOrderCallback) => {
   await kafkaConsumer.run({
-    eachMessage: async ({ message }) => {
+    eachMessage: async ({ message, partition }) => {
       const order = JSON.parse(message.value.toString());
       await processOrderCallback(order);
+      // Commit the offset to mark the message as processed
+      await kafkaConsumer.commitOffsets([{ topic: 'order-queue-topic', partition, offset: message.offset }]);
     },
   });
 };

index.js

@@ -2,6 +2,7 @@ const { setupDatabase } = require('./config/db.js');
 const { userRoutes } = require('./routes/users.js');
 const { kafka } = require('./config');
 const {checkPayment} = require("./config/payment");
+const {createShipment} = require("./services/shipment");
 
 const fastify = require('fastify')({
   logger: true,
@@ -15,6 +16,12 @@ async function runServer() {
     // Initialize Kafka
     await kafka.initializeKafka();
 
+    // consume orders
+    await kafka.consumeOrders(async (order) => {
+      const shipment = await createShipment(order)
+      console.log(shipment)
+    })
+
     fastify.get('/health', async (request, reply) => {
       reply.type('application/json').code(200);
       return { status: 'ok', uptime: process.uptime() };

services/shipment.js

@@ -0,0 +1,21 @@
+const {delay} = require("../utility");
+
+async function createShipment(order) {
+  if (!order || !order.orderId || !order.txnId || !order?.items || order?.items?.length === 0) {
+    console.error("Invalid order details. Shipment creation failed.");
+    return null;
+  }
+  await delay(1000);
+  return {
+    shipmentId: "SHIP" + Math.floor(Math.random() * 10000),
+    orderId: order.orderId,
+    items: order.items,
+    shippingAddress: order.shippingAddress || null,
+    shipmentDate: new Date(),
+    status: "Pending",
+  };
+}
+
+module.exports = {
+  createShipment
+}