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.WS_PORT || 3000; // Use same port for both // Start HTTP API server (Socket.IO will be served on the same port) server.listen(API_PORT, () => { console.log(`HTTP API and WebSocket 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(`WebSocket connections: ws://localhost:${API_PORT}`); 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); }); });