socket broker

2025-08-21 10:59:09 +02:00
parent 4470eb400a
commit bf404135b5
9 changed files with 2910 additions and 0 deletions
@@ -0,0 +1,281 @@
+const express = require('express');
+const { createServer } = require('http');
+const { Server } = require('socket.io');
+const cors = require('cors');
+
+// Create Express app and HTTP server
+const app = express();
+const server = createServer(app);
+
+// Create Socket.IO server
+const io = new Server(server, {
+    cors: {
+        origin: "*", // Allow all origins for development
+        methods: ["GET", "POST"]
+    }
+});
+
+// Store topic subscriptions
+const topicSubscriptions = new Map(); // topic -> Set of socket IDs
+const socketTopics = new Map(); // socket ID -> Set of topics
+
+// Get API key from environment variable
+const API_KEY = process.env.API_KEY || 'default-api-key-change-me';
+
+// API Key authentication middleware
+const authenticateApiKey = (req, res, next) => {
+    const authHeader = req.headers.authorization;
+    const apiKey = req.headers['x-api-key'] || req.query.api_key;
+
+    if (!authHeader && !apiKey) {
+        return res.status(401).json({
+            error: 'Unauthorized',
+            message: 'API key is required. Use Authorization header or x-api-key header or api_key query parameter.'
+        });
+    }
+
+    // Check Authorization header (Bearer token format)
+    if (authHeader && authHeader.startsWith('Bearer ')) {
+        const token = authHeader.substring(7);
+        if (token === API_KEY) {
+            return next();
+        }
+    }
+
+    // Check x-api-key header
+    if (apiKey && apiKey === API_KEY) {
+        return next();
+    }
+
+    return res.status(401).json({
+        error: 'Unauthorized',
+        message: 'Invalid API key'
+    });
+};
+
+// Middleware
+app.use(cors());
+app.use(express.json());
+
+// WebSocket connection handling (Port 3000)
+io.on('connection', (socket) => {
+    console.log(`Client connected: ${socket.id}`);
+
+    // Handle topic subscription
+    socket.on('subscribe', (topic) => {
+        if (!topic || typeof topic !== 'string') {
+            socket.emit('error', { message: 'Invalid topic' });
+            return;
+        }
+
+        // Add socket to topic
+        if (!topicSubscriptions.has(topic)) {
+            topicSubscriptions.set(topic, new Set());
+        }
+        topicSubscriptions.get(topic).add(socket.id);
+
+        // Add topic to socket
+        if (!socketTopics.has(socket.id)) {
+            socketTopics.set(socket.id, new Set());
+        }
+        socketTopics.get(socket.id).add(topic);
+
+        console.log(`Socket ${socket.id} subscribed to topic: ${topic}`);
+        socket.emit('subscribed', { topic, message: `Successfully subscribed to ${topic}` });
+    });
+
+    // Handle topic unsubscription
+    socket.on('unsubscribe', (topic) => {
+        if (!topic || typeof topic !== 'string') {
+            socket.emit('error', { message: 'Invalid topic' });
+            return;
+        }
+
+        // Remove socket from topic
+        if (topicSubscriptions.has(topic)) {
+            topicSubscriptions.get(topic).delete(socket.id);
+            if (topicSubscriptions.get(topic).size === 0) {
+                topicSubscriptions.delete(topic);
+            }
+        }
+
+        // Remove topic from socket
+        if (socketTopics.has(socket.id)) {
+            socketTopics.get(socket.id).delete(topic);
+            if (socketTopics.get(socket.id).size === 0) {
+                socketTopics.delete(socket.id);
+            }
+        }
+
+        console.log(`Socket ${socket.id} unsubscribed from topic: ${topic}`);
+        socket.emit('unsubscribed', { topic, message: `Successfully unsubscribed from ${topic}` });
+    });
+
+    // Handle disconnect
+    socket.on('disconnect', () => {
+        console.log(`Client disconnected: ${socket.id}`);
+
+        // Clean up subscriptions
+        if (socketTopics.has(socket.id)) {
+            const topics = socketTopics.get(socket.id);
+            topics.forEach(topic => {
+                if (topicSubscriptions.has(topic)) {
+                    topicSubscriptions.get(topic).delete(socket.id);
+                    if (topicSubscriptions.get(topic).size === 0) {
+                        topicSubscriptions.delete(topic);
+                    }
+                }
+            });
+            socketTopics.delete(socket.id);
+        }
+    });
+});
+
+// HTTP API endpoints (Port 8080)
+
+// Public endpoints (no authentication required)
+app.get('/health', (req, res) => {
+    res.json({
+        status: 'healthy',
+        connections: io.engine.clientsCount,
+        topics: topicSubscriptions.size,
+        timestamp: new Date().toISOString(),
+        apiKeyConfigured: !!API_KEY && API_KEY !== 'default-api-key-change-me'
+    });
+});
+
+app.get('/topics', (req, res) => {
+    const topics = Array.from(topicSubscriptions.keys()).map(topic => ({
+        topic,
+        subscribers: topicSubscriptions.get(topic).size
+    }));
+    res.json({ topics });
+});
+
+app.get('/topics/:topic', (req, res) => {
+    const { topic } = req.params;
+    if (!topicSubscriptions.has(topic)) {
+        return res.status(404).json({ error: 'Topic not found' });
+    }
+
+    const subscribers = topicSubscriptions.get(topic).size;
+    res.json({ topic, subscribers });
+});
+
+// Protected endpoints (require API key authentication)
+app.post('/publish', authenticateApiKey, (req, res) => {
+    const { topic, message, data } = req.body;
+
+    if (!topic || typeof topic !== 'string') {
+        return res.status(400).json({ error: 'Topic is required and must be a string' });
+    }
+
+    if (!message && !data) {
+        return res.status(400).json({ error: 'Either message or data is required' });
+    }
+
+    // Check if topic has subscribers
+    if (!topicSubscriptions.has(topic)) {
+        // Return 204 No Content when there are no subscribers
+        // The operation succeeded, there just weren't any recipients
+        return res.status(204).send();
+    }
+
+    const payload = {
+        topic,
+        message: message || null,
+        data: data || null,
+        timestamp: new Date().toISOString()
+    };
+
+    // Emit to all subscribers of the topic
+    const subscribers = topicSubscriptions.get(topic);
+    subscribers.forEach(socketId => {
+        io.to(socketId).emit('message', payload);
+    });
+
+    console.log(`Message published to topic '${topic}' for ${subscribers.size} subscribers`);
+
+    res.json({
+        success: true,
+        message: `Message published to topic '${topic}'`,
+        subscribers: subscribers.size,
+        payload
+    });
+});
+
+// Alternative endpoint with topic in URL (also protected)
+app.post('/publish/:topic', authenticateApiKey, (req, res) => {
+    const { topic } = req.params;
+    const { message, data } = req.body;
+
+    if (!message && !data) {
+        return res.status(400).json({ error: 'Either message or data is required' });
+    }
+
+    // Check if topic has subscribers
+    if (!topicSubscriptions.has(topic)) {
+        // Return 204 No Content when there are no subscribers
+        // The operation succeeded, there just weren't any recipients
+        return res.status(204).send();
+    }
+
+    const payload = {
+        topic,
+        message: message || null,
+        data: data || null,
+        timestamp: new Date().toISOString()
+    };
+
+    // Emit to all subscribers of the topic
+    const subscribers = topicSubscriptions.get(topic);
+    subscribers.forEach(socketId => {
+        io.to(socketId).emit('message', payload);
+    });
+
+    console.log(`Message published to topic '${topic}' for ${subscribers.size} subscribers`);
+
+    res.json({
+        success: true,
+        message: `Message published to topic '${topic}'`,
+        subscribers: subscribers.size,
+        payload
+    });
+});
+
+// Start servers
+const WS_PORT = process.env.WS_PORT || 3000;
+const API_PORT = process.env.API_PORT || 8080;
+
+// Start WebSocket server
+io.listen(WS_PORT);
+console.log(`WebSocket server running on port ${WS_PORT}`);
+
+// Start HTTP API server
+app.listen(API_PORT, () => {
+    console.log(`HTTP API server running on port ${API_PORT}`);
+    console.log(`Health check: http://localhost:${API_PORT}/health`);
+    console.log(`Topics: http://localhost:${API_PORT}/topics`);
+    console.log(`Publish: POST http://localhost:${API_PORT}/publish`);
+    console.log(`API Key configured: ${API_KEY !== 'default-api-key-change-me' ? 'Yes' : 'No (using default)'}`);
+    if (API_KEY === 'default-api-key-change-me') {
+        console.log(`⚠️  WARNING: Using default API key. Set API_KEY environment variable for production!`);
+    }
+});
+
+// Graceful shutdown
+process.on('SIGTERM', () => {
+    console.log('SIGTERM received, shutting down gracefully');
+    server.close(() => {
+        console.log('Process terminated');
+        process.exit(0);
+    });
+});
+
+process.on('SIGINT', () => {
+    console.log('SIGINT received, shutting down gracefully');
+    server.close(() => {
+        console.log('Process terminated');
+        process.exit(0);
+    });
+});