* much more robust ws connection manager

src/wsConnectionManager.ts

@@ -22,46 +22,83 @@ async function main() {
 	const redisClient = new RedisClient(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
 	await redisClient.connect();
 
+	const channelSubscribers = new Map<string, Set<WebSocket>>();
+	const subscribedChannels = new Set<string>();
+
+	redisClient.client.on('message', (subscribedChannel, message) => {
+		const subscribers = channelSubscribers.get(subscribedChannel);
+		subscribers.forEach((ws) => {
+			ws.send(JSON.stringify({channel: subscribedChannel, data: message }));
+		});
+	});
+
 	wss.on('connection', (ws: WebSocket) => {
 		console.log('Client connected');
 
 		ws.on('message', async (msg) => {
 			const parsedMessage = JSON.parse(msg.toString());
-			switch (parsedMessage.type) {
+
+			switch (parsedMessage.type.toLowerCase()) {
 				case 'subscribe': {
 					const channel = parsedMessage.channel;
-					console.log('Subscribing to channel', channel);
+					if (!subscribedChannels.has(channel)) {
-					redisClient.client.subscribe(channel);
+						console.log('Subscribing to channel', channel);
-					redisClient.client.on('message', (subscribedChannel, message) => {
+						redisClient.client.subscribe(channel).then(() => {
-						if (subscribedChannel === channel) {
+							subscribedChannels.add(channel);
-							ws.send(JSON.stringify({channel, data: message}));
+						}).catch(() => {
-						}
+							ws.send(JSON.stringify({channel, error: `Invalid channel: ${channel}`}));
-					});
+							return;
+						});
+					}	
+				
+					if (!channelSubscribers.get(channel)) {
+						const subscribers = new Set<WebSocket>();
+						channelSubscribers.set(channel, subscribers);
+					}
+					channelSubscribers.get(channel).add(ws);
 					break;
 				}
 				case 'unsubscribe': {
 					const channel = parsedMessage.channel;
-					console.log('Unsubscribing from channel', channel);
+					const subscribers = channelSubscribers.get(channel);
-					redisClient.client.unsubscribe(channel);
+					if (subscribers) {
+						channelSubscribers.get(channel).delete(ws);
+					}
+					break;
 				}	
+				default:
+					break;
 			}
 		});
 
 		// Ping/pong connection timeout
+		let pongTimeoutId;
 		let isAlive = true;
-
+		const pingIntervalId = setInterval(() => {
-    ws.on('pong', () => {
+			isAlive = false;
-      isAlive = true;
+			pongTimeoutId = setTimeout(() => {
-    });
+				if (!isAlive) {
-		setInterval(() => {
+					console.log('Disconnecting because of ping/pong timeout');
-			wss.clients.forEach((ws: WebSocket) => {
+					ws.terminate();
-				if (isAlive === false) return ws.terminate();
+				}
-		
+			}, 5000);  // 5 seconds to wait for a pong
-				isAlive = false;
+			ws.ping();
-				ws.ping();
-			});
 		}, 30000);
 
+		// Listen for pong messages
+		ws.on('pong', () => {
+			console.log('poooooooong');
+			isAlive = true;
+			clearTimeout(pongTimeoutId);
+		});
+
+		// Handle disconnection
+		ws.on('close', () => {
+			// Clear any existing intervals and timeouts
+			clearInterval(pingIntervalId);
+			clearTimeout(pongTimeoutId);
+		});
+
 		ws.on('disconnect', () => {
 			console.log('Client disconnected');
 		});