Inter-Service Communication

// HTTP client for service-to-service communication
const axios = require('axios');
const circuitBreaker = require('opossum');

class ServiceClient {
    constructor(serviceUrl, options = {}) {
        this.serviceUrl = serviceUrl;
        this.timeout = options.timeout || 5000;
        this.retries = options.retries || 3;
        
        // Circuit breaker pattern
        this.circuitBreaker = circuitBreaker(this.makeRequest.bind(this), {
            timeout: this.timeout,
            errorThresholdPercentage: 50,
            resetTimeout: 30000
        });
        
        this.circuitBreaker.on('open', () => {
            console.log('Circuit breaker opened for', serviceUrl);
        });
    }
    
    async makeRequest(method, path, data = null) {
        const config = {
            method,
            url: `${this.serviceUrl}${path}`,
            timeout: this.timeout,
            headers: {
                'Content-Type': 'application/json'
            }
        };
        
        if (data) {
            config.data = data;
        }
        
        return await axios(config);
    }
    
    async get(path) {
        return await this.circuitBreaker.fire('GET', path);
    }
    
    async post(path, data) {
        return await this.circuitBreaker.fire('POST', path, data);
    }
}

// Usage
const userService = new ServiceClient('http://user-service:3001');
const productService = new ServiceClient('http://product-service:3002');

// RabbitMQ implementation for async communication
const amqp = require('amqplib');

class MessageBroker {
    constructor(connectionString) {
        this.connectionString = connectionString;
        this.connection = null;
        this.channel = null;
    }
    
    async connect() {
        this.connection = await amqp.connect(this.connectionString);
        this.channel = await this.connection.createChannel();
        
        // Enable publisher confirms
        await this.channel.confirmSelect();
    }
    
    async publish(exchange, routingKey, message) {
        if (!this.channel) {
            throw new Error('Not connected to message broker');
        }
        
        const messageBuffer = Buffer.from(JSON.stringify(message));
        
        return this.channel.publish(
            exchange,
            routingKey,
            messageBuffer,
            {
                persistent: true,
                messageId: require('uuid').v4(),
                timestamp: Date.now()
            }
        );
    }
    
    async subscribe(queue, handler) {
        if (!this.channel) {
            throw new Error('Not connected to message broker');
        }
        
        await this.channel.assertQueue(queue, { durable: true });
        
        this.channel.consume(queue, async (msg) => {
            if (msg) {
                try {
                    const message = JSON.parse(msg.content.toString());
                    await handler(message);
                    this.channel.ack(msg);
                } catch (error) {
                    console.error('Error processing message:', error);
                    this.channel.nack(msg, false, false);
                }
            }
        });
    }
}

// Event-driven service
class OrderService {
    constructor() {
        this.messageBroker = new MessageBroker(process.env.RABBITMQ_URL);
    }
    
    async start() {
        await this.messageBroker.connect();
        
        // Subscribe to events
        await this.messageBroker.subscribe('order.created', this.handleOrderCreated.bind(this));
        await this.messageBroker.subscribe('payment.processed', this.handlePaymentProcessed.bind(this));
    }
    
    async handleOrderCreated(orderData) {
        console.log('Processing order:', orderData);
        // Process order logic
    }
    
    async handlePaymentProcessed(paymentData) {
        console.log('Payment processed:', paymentData);
        // Update order status
    }
}

// Event sourcing implementation
class EventStore {
    constructor() {
        this.events = [];
    }
    
    append(streamId, event) {
        const eventRecord = {
            streamId,
            eventId: require('uuid').v4(),
            eventType: event.constructor.name,
            eventData: event,
            timestamp: new Date(),
            version: this.getNextVersion(streamId)
        };
        
        this.events.push(eventRecord);
        return eventRecord;
    }
    
    getEvents(streamId) {
        return this.events.filter(e => e.streamId === streamId);
    }
    
    getNextVersion(streamId) {
        const streamEvents = this.getEvents(streamId);
        return streamEvents.length;
    }
}

// Aggregate root
class Order {
    constructor(id) {
        this.id = id;
        this.status = 'pending';
        this.items = [];
        this.version = 0;
    }
    
    addItem(productId, quantity, price) {
        const event = new OrderItemAdded(this.id, productId, quantity, price);
        this.apply(event);
        return event;
    }
    
    apply(event) {
        switch (event.constructor.name) {
            case 'OrderItemAdded':
                this.items.push({
                    productId: event.productId,
                    quantity: event.quantity,
                    price: event.price
                });
                break;
            case 'OrderConfirmed':
                this.status = 'confirmed';
                break;
        }
        this.version++;
    }
}

// Events
class OrderItemAdded {
    constructor(orderId, productId, quantity, price) {
        this.orderId = orderId;
        this.productId = productId;
        this.quantity = quantity;
        this.price = price;
    }
}

// Complete event-driven microservices system
const express = require('express');
const { EventEmitter } = require('events');

// Event bus
class EventBus extends EventEmitter {
    constructor() {
        super();
        this.setMaxListeners(100);
    }
    
    publish(event) {
        this.emit(event.type, event);
    }
    
    subscribe(eventType, handler) {
        this.on(eventType, handler);
    }
}

// Event types
const EventTypes = {
    USER_CREATED: 'user.created',
    ORDER_CREATED: 'order.created',
    PAYMENT_PROCESSED: 'payment.processed',
    INVENTORY_UPDATED: 'inventory.updated'
};

// User service
class UserService {
    constructor(eventBus) {
        this.eventBus = eventBus;
        this.users = new Map();
    }
    
    createUser(userData) {
        const user = {
            id: require('uuid').v4(),
            ...userData,
            createdAt: new Date()
        };
        
        this.users.set(user.id, user);
        
        // Publish event
        this.eventBus.publish({
            type: EventTypes.USER_CREATED,
            data: user,
            timestamp: new Date()
        });
        
        return user;
    }
    
    getUser(id) {
        return this.users.get(id);
    }
}

// Order service
class OrderService {
    constructor(eventBus, userService) {
        this.eventBus = eventBus;
        this.userService = userService;
        this.orders = new Map();
        
        // Subscribe to events
        this.eventBus.subscribe(EventTypes.USER_CREATED, this.handleUserCreated.bind(this));
        this.eventBus.subscribe(EventTypes.PAYMENT_PROCESSED, this.handlePaymentProcessed.bind(this));
    }
    
    createOrder(orderData) {
        const order = {
            id: require('uuid').v4(),
            ...orderData,
            status: 'pending',
            createdAt: new Date()
        };
        
        this.orders.set(order.id, order);
        
        // Publish event
        this.eventBus.publish({
            type: EventTypes.ORDER_CREATED,
            data: order,
            timestamp: new Date()
        });
        
        return order;
    }
    
    handleUserCreated(event) {
        console.log('New user created, setting up welcome order:', event.data.id);
    }
    
    handlePaymentProcessed(event) {
        console.log('Payment processed for order:', event.data.orderId);
        // Update order status
    }
}

// Main application
const app = express();
app.use(express.json());

const eventBus = new EventBus();
const userService = new UserService(eventBus);
const orderService = new OrderService(eventBus, userService);

// API endpoints
app.post('/users', (req, res) => {
    const user = userService.createUser(req.body);
    res.status(201).json(user);
});

app.post('/orders', (req, res) => {
    const order = orderService.createOrder(req.body);
    res.status(201).json(order);
});

app.listen(3000, () => {
    console.log('Event-driven microservices running on port 3000');
});