🔷 Microservices Architecture Course

// User Service Structure user-service/ ├── src/ │ ├── api/ │ │ ├── routes/ │ │ │ └── user.routes.js │ │ └── controllers/ │ │ └── user.controller.js │ ├── domain/ │ │ ├── models/ │ │ │ └── user.model.js │ │ └── services/ │ │ └── user.service.js │ ├── infrastructure/ │ │ ├── database/ │ │ │ └── connection.js │ │ └── messaging/ │ │ └── rabbitmq.js │ └── server.js ├── tests/ ├── Dockerfile ├── docker-compose.yml └── package.json // Service Implementation // user.service.js class UserService { constructor(userRepository, eventPublisher) { this.userRepository = userRepository; this.eventPublisher = eventPublisher; } async createUser(userData) { // Validate data this.validateUserData(userData); // Create user const user = await this.userRepository.create(userData); // Publish event await this.eventPublisher.publish('user.created', { userId: user.id, email: user.email, timestamp: new Date() }); return user; } async getUserById(userId) { const user = await this.userRepository.findById(userId); if (!user) { throw new Error('User not found'); } return user; } validateUserData(userData) { if (!userData.email || !userData.name) { throw new Error('Email and name are required'); } } } module.exports = UserService;

// Service-to-Service HTTP Communication const axios = require('axios'); class OrderService { constructor() { this.userServiceUrl = process.env.USER_SERVICE_URL; this.inventoryServiceUrl = process.env.INVENTORY_SERVICE_URL; } async createOrder(orderData) { try { // Call User Service const userResponse = await axios.get( `${this.userServiceUrl}/users/${orderData.userId}`, { headers: { 'Authorization': `Bearer ${orderData.token}` }, timeout: 5000 } ); const user = userResponse.data; // Call Inventory Service const inventoryResponse = await axios.post( `${this.inventoryServiceUrl}/inventory/reserve`, { items: orderData.items }, { timeout: 5000 } ); // Create order const order = await this.orderRepository.create({ userId: user.id, items: orderData.items, total: this.calculateTotal(orderData.items), status: 'pending' }); return order; } catch (error) { // Handle service failures if (error.code === 'ECONNREFUSED') { throw new Error('Service unavailable'); } throw error; } } }

// RabbitMQ Message Publisher const amqp = require('amqplib'); class EventPublisher { constructor() { this.connection = null; this.channel = null; } async connect() { this.connection = await amqp.connect(process.env.RABBITMQ_URL); this.channel = await this.connection.createChannel(); } async publish(eventType, data) { const exchange = 'events'; await this.channel.assertExchange(exchange, 'topic', { durable: true }); const message = JSON.stringify({ eventType, data, timestamp: new Date(), serviceId: process.env.SERVICE_ID }); this.channel.publish( exchange, eventType, Buffer.from(message), { persistent: true } ); } } // Message Consumer class EventConsumer { constructor(eventHandlers) { this.eventHandlers = eventHandlers; } async connect() { const connection = await amqp.connect(process.env.RABBITMQ_URL); const channel = await connection.createChannel(); const exchange = 'events'; await channel.assertExchange(exchange, 'topic', { durable: true }); const queue = await channel.assertQueue('', { exclusive: true }); // Subscribe to events channel.bindQueue(queue.queue, exchange, 'user.*'); channel.bindQueue(queue.queue, exchange, 'order.*'); channel.consume(queue.queue, async (msg) => { const event = JSON.parse(msg.content.toString()); // Handle event const handler = this.eventHandlers[event.eventType]; if (handler) { try { await handler(event.data); channel.ack(msg); } catch (error) { console.error('Error handling event:', error); channel.nack(msg, false, true); // Requeue } } }); } } // Usage in Order Service const eventHandlers = { 'user.created': async (data) => { console.log('New user created:', data.userId); // Handle user creation }, 'payment.completed': async (data) => { // Update order status await orderService.updateOrderStatus(data.orderId, 'paid'); } }; const consumer = new EventConsumer(eventHandlers); consumer.connect();

// Circuit Breaker Implementation class CircuitBreaker { constructor(options = {}) { this.failureThreshold = options.failureThreshold || 5; this.timeout = options.timeout || 60000; this.resetTimeout = options.resetTimeout || 30000; this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN this.failureCount = 0; this.nextAttempt = Date.now(); } async execute(fn) { if (this.state === 'OPEN') { if (Date.now() < this.nextAttempt) { throw new Error('Circuit breaker is OPEN'); } this.state = 'HALF_OPEN'; } try { const result = await Promise.race([ fn(), new Promise((_, reject) => setTimeout(() => reject(new Error('Timeout')), this.timeout) ) ]); this.onSuccess(); return result; } catch (error) { this.onFailure(); throw error; } } onSuccess() { this.failureCount = 0; this.state = 'CLOSED'; } onFailure() { this.failureCount++; if (this.failureCount >= this.failureThreshold) { this.state = 'OPEN'; this.nextAttempt = Date.now() + this.resetTimeout; } } } // Usage const breaker = new CircuitBreaker({ failureThreshold: 3, timeout: 5000, resetTimeout: 30000 }); async function callExternalService() { return breaker.execute(async () => { return await axios.get('http://external-service/api/data'); }); }

// Saga Pattern for Distributed Transactions class OrderSaga { constructor(services) { this.userService = services.userService; this.inventoryService = services.inventoryService; this.paymentService = services.paymentService; this.orderService = services.orderService; } async createOrder(orderData) { const compensations = []; try { // Step 1: Validate user const user = await this.userService.validateUser(orderData.userId); compensations.push(() => this.userService.releaseUser(user.id)); // Step 2: Reserve inventory const reservation = await this.inventoryService.reserve(orderData.items); compensations.push(() => this.inventoryService.cancelReservation(reservation.id)); // Step 3: Process payment const payment = await this.paymentService.charge({ userId: user.id, amount: orderData.total }); compensations.push(() => this.paymentService.refund(payment.id)); // Step 4: Create order const order = await this.orderService.create({ userId: user.id, items: orderData.items, paymentId: payment.id, reservationId: reservation.id }); return order; } catch (error) { // Compensate in reverse order console.error('Saga failed, compensating...', error); for (const compensate of compensations.reverse()) { try { await compensate(); } catch (compError) { console.error('Compensation failed:', compError); } } throw error; } } }

// Event Store class EventStore { constructor(database) { this.db = database; } async saveEvent(aggregateId, eventType, eventData) { const event = { aggregateId, eventType, eventData, version: await this.getNextVersion(aggregateId), timestamp: new Date() }; await this.db.collection('events').insertOne(event); return event; } async getEvents(aggregateId) { return await this.db.collection('events') .find({ aggregateId }) .sort({ version: 1 }) .toArray(); } async getNextVersion(aggregateId) { const lastEvent = await this.db.collection('events') .findOne({ aggregateId }, { sort: { version: -1 } }); return lastEvent ? lastEvent.version + 1 : 1; } } // Aggregate class OrderAggregate { constructor(orderId) { this.orderId = orderId; this.events = []; this.state = { status: 'pending', items: [], total: 0 }; } createOrder(items, total) { this.applyEvent({ type: 'OrderCreated', data: { items, total } }); } confirmPayment(paymentId) { this.applyEvent({ type: 'PaymentConfirmed', data: { paymentId } }); } applyEvent(event) { this.events.push(event); this.apply(event); } apply(event) { switch (event.type) { case 'OrderCreated': this.state.items = event.data.items; this.state.total = event.data.total; break; case 'PaymentConfirmed': this.state.status = 'paid'; this.state.paymentId = event.data.paymentId; break; } } async save(eventStore) { for (const event of this.events) { await eventStore.saveEvent(this.orderId, event.type, event.data); } this.events = []; } static async load(orderId, eventStore) { const aggregate = new OrderAggregate(orderId); const events = await eventStore.getEvents(orderId); for (const event of events) { aggregate.apply({ type: event.eventType, data: event.eventData }); } return aggregate; } }

# docker-compose.yml version: '3.8' services: user-service: build: ./user-service ports: - "3001:3000" environment: - DATABASE_URL=mongodb://mongo:27017/users - RABBITMQ_URL=amqp://rabbitmq:5672 depends_on: - mongo - rabbitmq networks: - microservices order-service: build: ./order-service ports: - "3002:3000" environment: - DATABASE_URL=mongodb://mongo:27017/orders - RABBITMQ_URL=amqp://rabbitmq:5672 - USER_SERVICE_URL=http://user-service:3000 depends_on: - mongo - rabbitmq networks: - microservices api-gateway: build: ./api-gateway ports: - "8080:8080" environment: - USER_SERVICE_URL=http://user-service:3000 - ORDER_SERVICE_URL=http://order-service:3000 depends_on: - user-service - order-service networks: - microservices mongo: image: mongo:5 ports: - "27017:27017" volumes: - mongo-data:/data/db networks: - microservices rabbitmq: image: rabbitmq:3-management ports: - "5672:5672" - "15672:15672" networks: - microservices networks: microservices: driver: bridge volumes: mongo-data:

// OpenTelemetry Setup const { NodeTracerProvider } = require('@opentelemetry/sdk-trace-node'); const { registerInstrumentations } = require('@opentelemetry/instrumentation'); const { HttpInstrumentation } = require('@opentelemetry/instrumentation-http'); const { ExpressInstrumentation } = require('@opentelemetry/instrumentation-express'); // Initialize tracer const provider = new NodeTracerProvider(); provider.register(); registerInstrumentations({ instrumentations: [ new HttpInstrumentation(), new ExpressInstrumentation() ] }); // Add trace ID to logs const winston = require('winston'); const { trace } = require('@opentelemetry/api'); const logger = winston.createLogger({ format: winston.format.combine( winston.format.timestamp(), winston.format.printf(info => { const span = trace.getActiveSpan(); const traceId = span ? span.spanContext().traceId : 'no-trace'; return `${info.timestamp} [${traceId}] ${info.level}: ${info.message}`; }) ), transports: [new winston.transports.Console()] });