projects/Dockers/socket-broker/index.js

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);
    });
});