Add streams initializer to redis model

src/models/index.js

@@ -1,3 +1,4 @@
+const config = require("../config");
 const dbConfig = require("../config/db.config");
 const redisConfig = require("../config/redis.config");
 
@@ -51,15 +52,14 @@ redis.client = Redis.createClient({
 
 redis.client.on('error', (err) => console.error('Redis error:', err));
 
-
+redis.initializeStreams = async () => {
-
-const initializeStreams = async () => {
     try {
         await redis.client.connect().catch(console.error);
         
         console.log("Initializing Redis Streams...");
 
         // Create transcriptions streams consumer group
+        try {
             await client.xGroupCreate("communication", "mtf-devices", '0', {MKSTREAM: true});
 
             console.log("Created consumer group: mtf-devices (communication)");
@@ -71,7 +71,27 @@ const initializeStreams = async () => {
             }
         }
 
-    const deviceIds = [""]
+        const deviceIds = [config.DEVICE_ID];
+
+        for (const deviceId of deviceIds) {
+            const groupName = `${deviceId}_group`;
+            try {
+                await client.xGroupCreate('communication', groupName, '0', {MKSTREAM: true});
+                console.log(`Created consumer group: ${groupName} (communication)`);
+            } catch (err) {
+                if (err.message.includes('BUSYGROUP')) {
+                    console.log(`Consumer group already exists: ${groupName}`);
+                } else {
+                    console.error(err);
+                }
+            }
+        }
+
+        console.log("Streams initialized successfully");
+        await client.quit();
+    } catch (err) {
+        console.error("Error initializing streams:", err.message);
+    }
 }
 
 module.exports = {

src/server.js

@@ -9,6 +9,7 @@ const rateLimit = require("express-rate-limit");
 const morgan = require("morgan");
 const models = require('./models');
 const db = models.db;
+const redis = models.redis;
 const {sequelize, Sequelize} = db;
 const Role = db.role;
 
@@ -16,6 +17,7 @@ const authConfig = require("./config/auth.config");
 const config = require("./config");
 
 const createApp = () => {
+    redis.initializeStreams();
     const app = express();
 
     // Security headers