Microservices Architecture Patterns: Complete Implementation Guide

const express = require('express');
const httpProxy = require('http-proxy-middleware');
const jwt = require('jsonwebtoken');
const rateLimit = require('express-rate-limit');

const app = express();

// Rate limiting middleware
const limiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100, // limit each IP to 100 requests per windowMs
  message: 'Too many requests from this IP'
});

// Authentication middleware
const authenticateToken = (req, res, next) => {
  const authHeader = req.headers['authorization'];
  const token = authHeader && authHeader.split(' ')[1];
  
  if (!token) {
    return res.sendStatus(401);
  }
  
  jwt.verify(token, process.env.JWT_SECRET, (err, user) => {
    if (err) return res.sendStatus(403);
    req.user = user;
    next();
  });
};

// Service routing configuration
const services = {
  '/api/users': {
    target: 'http://user-service:3001',
    changeOrigin: true,
    pathRewrite: { '^/api/users': '' }
  },
  '/api/orders': {
    target: 'http://order-service:3002',
    changeOrigin: true,
    pathRewrite: { '^/api/orders': '' }
  },
  '/api/payments': {
    target: 'http://payment-service:3003',
    changeOrigin: true,
    pathRewrite: { '^/api/payments': '' }
  }
};

// Apply middleware
app.use(limiter);
app.use(express.json());

// Health check endpoint
app.get('/health', (req, res) => {
  res.status(200).json({ status: 'OK', timestamp: new Date().toISOString() });
});

// Setup proxy routes
Object.keys(services).forEach(route => {
  app.use(route, authenticateToken, httpProxy(services[route]));
});

// Error handling middleware
app.use((err, req, res, next) => {
  console.error('Gateway Error:', err);
  res.status(500).json({ 
    error: 'Internal Gateway Error',
    requestId: req.headers['x-request-id'] || 'unknown'
  });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
  console.log(`API Gateway running on port ${PORT}`);
});

module.exports = app;

# Virtual Service for traffic routing
apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: user-service
spec:
  http:
  - match:
    - headers:
        version:
          exact: v2
    route:
    - destination:
        host: user-service
        subset: v2
      weight: 100
  - route:
    - destination:
        host: user-service
        subset: v1
      weight: 80
    - destination:
        host: user-service
        subset: v2
      weight: 20

---
# Destination Rule for load balancing
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: user-service
spec:
  host: user-service
  trafficPolicy:
    loadBalancer:
      simple: LEAST_CONN
  subsets:
  - name: v1
    labels:
      version: v1
    trafficPolicy:
      connectionPool:
        tcp:
          maxConnections: 100
        http:
          http1MaxPendingRequests: 50
          maxRequestsPerConnection: 10
  - name: v2
    labels:
      version: v2

---
# Circuit Breaker Configuration
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: payment-service-circuit-breaker
spec:
  host: payment-service
  trafficPolicy:
    outlierDetection:
      consecutiveErrors: 3
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50

const amqp = require('amqplib');
const EventEmitter = require('events');

class EventPublisher extends EventEmitter {
  constructor(rabbitmqUrl) {
    super();
    this.rabbitmqUrl = rabbitmqUrl;
    this.connection = null;
    this.channel = null;
  }

  async connect() {
    try {
      this.connection = await amqp.connect(this.rabbitmqUrl);
      this.channel = await this.connection.createChannel();
      
      // Setup exchange
      await this.channel.assertExchange('microservices.events', 'topic', {
        durable: true
      });
      
      console.log('Connected to RabbitMQ');
      this.emit('connected');
    } catch (error) {
      console.error('Failed to connect to RabbitMQ:', error);
      this.emit('error', error);
    }
  }

  async publishEvent(eventType, data, routingKey = null) {
    if (!this.channel) {
      throw new Error('Not connected to RabbitMQ');
    }

    const event = {
      id: this.generateUUID(),
      timestamp: new Date().toISOString(),
      type: eventType,
      data: data,
      version: '1.0'
    };

    const key = routingKey || eventType.toLowerCase().replace(/\s+/g, '.');
    
    try {
      await this.channel.publish(
        'microservices.events',
        key,
        Buffer.from(JSON.stringify(event)),
        {
          persistent: true,
          contentType: 'application/json',
          headers: {
            eventType: eventType,
            source: process.env.SERVICE_NAME || 'unknown'
          }
        }
      );
      
      console.log(`Event published: ${eventType}`);
      this.emit('published', event);
    } catch (error) {
      console.error('Failed to publish event:', error);
      this.emit('publishError', error, event);
    }
  }

  generateUUID() {
    return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {
      const r = Math.random() * 16 | 0;
      const v = c == 'x' ? r : (r & 0x3 | 0x8);
      return v.toString(16);
    });
  }

  async close() {
    if (this.connection) {
      await this.connection.close();
      this.emit('disconnected');
    }
  }
}

// Usage example
async function setupEventPublisher() {
  const publisher = new EventPublisher(process.env.RABBITMQ_URL);
  
  publisher.on('connected', () => {
    console.log('Event publisher ready');
  });
  
  publisher.on('error', (error) => {
    console.error('Publisher error:', error);
    process.exit(1);
  });
  
  await publisher.connect();
  
  // Publish user created event
  await publisher.publishEvent('UserCreated', {
    userId: '12345',
    email: 'user@example.com',
    createdAt: new Date().toISOString()
  }, 'user.created');
  
  return publisher;
}

module.exports = { EventPublisher, setupEventPublisher };

// Event base class
@Entity
@Table(name = "event_store")
public abstract class BaseEvent {
    @Id
    private String eventId;
    
    @Column(name = "aggregate_id")
    private String aggregateId;
    
    @Column(name = "event_type")
    private String eventType;
    
    @Column(name = "event_data", columnDefinition = "TEXT")
    private String eventData;
    
    @Column(name = "timestamp")
    private LocalDateTime timestamp;
    
    @Column(name = "version")
    private Long version;
    
    // constructors, getters, setters
    public BaseEvent() {
        this.eventId = UUID.randomUUID().toString();
        this.timestamp = LocalDateTime.now();
    }
}

// Specific event implementation
public class OrderCreatedEvent extends BaseEvent {
    private String customerId;
    private BigDecimal totalAmount;
    private List items;
    
    public OrderCreatedEvent(String orderId, String customerId, 
                           BigDecimal totalAmount, List items) {
        super();
        setAggregateId(orderId);
        setEventType("OrderCreated");
        this.customerId = customerId;
        this.totalAmount = totalAmount;
        this.items = items;
        
        // Serialize event data to JSON
        ObjectMapper mapper = new ObjectMapper();
        try {
            Map data = Map.of(
                "customerId", customerId,
                "totalAmount", totalAmount,
                "items", items
            );
            setEventData(mapper.writeValueAsString(data));
        } catch (JsonProcessingException e) {
            throw new RuntimeException("Failed to serialize event data", e);
        }
    }
    
    // getters and setters
}

// Event Store implementation
@Service
public class EventStore {
    
    @Autowired
    private EventRepository eventRepository;
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    @Transactional
    public void saveEvent(BaseEvent event) {
        // Get current version for aggregate
        Long currentVersion = eventRepository
            .findMaxVersionByAggregateId(event.getAggregateId())
            .orElse(0L);
        
        event.setVersion(currentVersion + 1);
        
        // Save event
        eventRepository.save(event);
        
        // Publish event for projections and other services
        eventPublisher.publishEvent(new EventSavedEvent(event));
    }
    
    public List getEventsForAggregate(String aggregateId) {
        return eventRepository.findByAggregateIdOrderByVersionAsc(aggregateId);
    }
    
    public List getEventsFromVersion(String aggregateId, Long fromVersion) {
        return eventRepository
            .findByAggregateIdAndVersionGreaterThanOrderByVersionAsc(
                aggregateId, fromVersion);
    }
}

// Aggregate root that can be rebuilt from events
public class Order {
    private String orderId;
    private String customerId;
    private OrderStatus status;
    private BigDecimal totalAmount;
    private List items;
    private LocalDateTime createdAt;
    private Long version;
    
    // Apply events to rebuild state
    public void apply(BaseEvent event) {
        switch (event.getEventType()) {
            case "OrderCreated":
                applyOrderCreated((OrderCreatedEvent) event);
                break;
            case "OrderStatusChanged":
                applyOrderStatusChanged((OrderStatusChangedEvent) event);
                break;
            case "OrderItemAdded":
                applyOrderItemAdded((OrderItemAddedEvent) event);
                break;
        }
        this.version = event.getVersion();
    }
    
    private void applyOrderCreated(OrderCreatedEvent event) {
        this.orderId = event.getAggregateId();
        this.customerId = event.getCustomerId();
        this.totalAmount = event.getTotalAmount();
        this.items = new ArrayList<>(event.getItems());
        this.status = OrderStatus.CREATED;
        this.createdAt = event.getTimestamp();
    }
    
    // Other apply methods...
}

// Repository for rebuilding aggregates from events
@Service
public class OrderEventSourcingRepository {
    
    @Autowired
    private EventStore eventStore;
    
    public Order findById(String orderId) {
        List events = eventStore.getEventsForAggregate(orderId);
        
        if (events.isEmpty()) {
            return null;
        }
        
        Order order = new Order();
        events.forEach(order::apply);
        
        return order;
    }
    
    public void save(Order order, BaseEvent event) {
        eventStore.saveEvent(event);
    }
}

from abc import ABC, abstractmethod
from typing import Optional, List
from dataclasses import dataclass
from datetime import datetime
import uuid

# Command side - Write model
@dataclass
class CreateOrderCommand:
    customer_id: str
    items: List[dict]
    total_amount: float

@dataclass
class UpdateOrderStatusCommand:
    order_id: str
    status: str
    updated_by: str

# Query side - Read model
@dataclass
class OrderSummaryQuery:
    order_id: str

@dataclass
class CustomerOrdersQuery:
    customer_id: str
    status: Optional[str] = None
    limit: int = 10
    offset: int = 0

# Command handlers
class CommandHandler(ABC):
    @abstractmethod
    async def handle(self, command):
        pass

class CreateOrderCommandHandler(CommandHandler):
    def __init__(self, event_store, event_publisher):
        self.event_store = event_store
        self.event_publisher = event_publisher
    
    async def handle(self, command: CreateOrderCommand):
        order_id = str(uuid.uuid4())
        
        # Create and save event
        event = {
            'event_id': str(uuid.uuid4()),
            'aggregate_id': order_id,
            'event_type': 'OrderCreated',
            'event_data': {
                'customer_id': command.customer_id,
                'items': command.items,
                'total_amount': command.total_amount
            },
            'timestamp': datetime.utcnow().isoformat(),
            'version': 1
        }
        
        await self.event_store.save_event(event)
        await self.event_publisher.publish(event)
        
        return order_id

# Query handlers
class QueryHandler(ABC):
    @abstractmethod
    async def handle(self, query):
        pass

class OrderSummaryQueryHandler(QueryHandler):
    def __init__(self, read_model_repository):
        self.read_model_repository = read_model_repository
    
    async def handle(self, query: OrderSummaryQuery):
        return await self.read_model_repository.get_order_summary(query.order_id)

class CustomerOrdersQueryHandler(QueryHandler):
    def __init__(self, read_model_repository):
        self.read_model_repository = read_model_repository
    
    async def handle(self, query: CustomerOrdersQuery):
        return await self.read_model_repository.get_customer_orders(
            query.customer_id,
            query.status,
            query.limit,
            query.offset
        )

# CQRS Bus implementation
class CommandBus:
    def __init__(self):
        self.handlers = {}
    
    def register_handler(self, command_type, handler):
        self.handlers[command_type] = handler
    
    async def dispatch(self, command):
        handler = self.handlers.get(type(command))
        if not handler:
            raise ValueError(f"No handler registered for {type(command)}")
        return await handler.handle(command)

class QueryBus:
    def __init__(self):
        self.handlers = {}
    
    def register_handler(self, query_type, handler):
        self.handlers[query_type] = handler
    
    async def dispatch(self, query):
        handler = self.handlers.get(type(query))
        if not handler:
            raise ValueError(f"No handler registered for {type(query)}")
        return await handler.handle(query)

# FastAPI endpoints
from fastapi import FastAPI, HTTPException, Depends

app = FastAPI(title="CQRS Order Service")

# Dependency injection setup
async def get_command_bus():
    # Setup command bus with handlers
    bus = CommandBus()
    # Register handlers...
    return bus

async def get_query_bus():
    # Setup query bus with handlers
    bus = QueryBus()
    # Register handlers...
    return bus

@app.post("/orders/")
async def create_order(
    customer_id: str,
    items: List[dict],
    total_amount: float,
    command_bus: CommandBus = Depends(get_command_bus)
):
    try:
        command = CreateOrderCommand(
            customer_id=customer_id,
            items=items,
            total_amount=total_amount
        )
        order_id = await command_bus.dispatch(command)
        return {"order_id": order_id, "status": "created"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/orders/{order_id}")
async def get_order(
    order_id: str,
    query_bus: QueryBus = Depends(get_query_bus)
):
    try:
        query = OrderSummaryQuery(order_id=order_id)
        result = await query_bus.dispatch(query)
        if not result:
            raise HTTPException(status_code=404, detail="Order not found")
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/customers/{customer_id}/orders")
async def get_customer_orders(
    customer_id: str,
    status: Optional[str] = None,
    limit: int = 10,
    offset: int = 0,
    query_bus: QueryBus = Depends(get_query_bus)
):
    try:
        query = CustomerOrdersQuery(
            customer_id=customer_id,
            status=status,
            limit=limit,
            offset=offset
        )
        return await query_bus.dispatch(query)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

version: '3.8'

services:
  # API Gateway
  api-gateway:
    build:
      context: ./api-gateway
      dockerfile: Dockerfile
    ports:
      - "3000:3000"
    environment:
      - NODE_ENV=production
      - JWT_SECRET=your-super-secret-key
      - RABBITMQ_URL=amqp://rabbitmq:5672
    depends_on:
      - rabbitmq
      - user-service
      - order-service
      - payment-service
    networks:
      - microservices-network
    restart: unless-stopped

  # User Service
  user-service:
    build:
      context: ./user-service
      dockerfile: Dockerfile
    environment:
      - NODE_ENV=production
      - DB_HOST=user-db
      - DB_PORT=5432
      - DB_NAME=users
      - DB_USER=postgres
      - DB_PASSWORD=password
      - RABBITMQ_URL=amqp://rabbitmq:5672
    depends_on:
      - user-db
      - rabbitmq
    networks:
      - microservices-network
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:3001/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  user-db:
    image: postgres:13
    environment:
      - POSTGRES_DB=users
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
    volumes:
      - user-db-data:/var/lib/postgresql/data
      - ./user-service/db/init.sql:/docker-entrypoint-initdb.d/init.sql
    networks:
      - microservices-network
    restart: unless-stopped

  # Order Service
  order-service:
    build:
      context: ./order-service
      dockerfile: Dockerfile
    environment:
      - SPRING_PROFILES_ACTIVE=production
      - DB_HOST=order-db
      - DB_PORT=5432
      - DB_NAME=orders
      - DB_USER=postgres
      - DB_PASSWORD=password
      - RABBITMQ_HOST=rabbitmq
      - RABBITMQ_PORT=5672
    depends_on:
      - order-db
      - rabbitmq
    networks:
      - microservices-network
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/actuator/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  order-db:
    image: postgres:13
    environment:
      - POSTGRES_DB=orders
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
    volumes:
      - order-db-data:/var/lib/postgresql/data
    networks:
      - microservices-network
    restart: unless-stopped

  # Payment Service
  payment-service:
    build:
      context: ./payment-service
      dockerfile: Dockerfile
    environment:
      - FLASK_ENV=production
      - DATABASE_URL=postgresql://postgres:password@payment-db:5432/payments
      - RABBITMQ_URL=amqp://rabbitmq:5672
      - STRIPE_API_KEY=your-stripe-key
    depends_on:
      - payment-db
      - rabbitmq
    networks:
      - microservices-network
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:5000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  payment-db:
    image: postgres:13
    environment:
      - POSTGRES_DB=payments
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
    volumes:
      - payment-db-data:/var/lib/postgresql/data
    networks:
      - microservices-network
    restart: unless-stopped

  # Message Broker
  rabbitmq:
    image: rabbitmq:3.9-management
    environment:
      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=password
    ports:
      - "5672:5672"
      - "15672:15672"  # Management UI
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    networks:
      - microservices-network
    restart: unless-stopped

  # Redis for caching
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    networks:
      - microservices-network
    restart: unless-stopped

  # Monitoring
  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus-data:/prometheus
    networks:
      - microservices-network
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3001:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana-data:/var/lib/grafana
      - ./monitoring/grafana/dashboards:/etc/grafana/provisioning/dashboards
    networks:
      - microservices-network
    restart: unless-stopped

networks:
  microservices-network:
    driver: bridge

volumes:
  user-db-data:
  order-db-data:
  payment-db-data:
  rabbitmq-data:
  redis-data:
  prometheus-data:
  grafana-data:

# user-service-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-service
  labels:
    app: user-service
    version: v1
spec:
  replicas: 3
  selector:
    matchLabels:
      app: user-service
      version: v1
  template:
    metadata:
      labels:
        app: user-service
        version: v1
    spec:
      containers:
      - name: user-service
        image: your-registry/user-service:v1.0.0
        ports:
        - containerPort: 3001
        env:
        - name: NODE_ENV
          value: "production"
        - name: DB_HOST
          valueFrom:
            secretKeyRef:
              name: user-service-secrets
              key: db-host
        - name: DB_PASSWORD
          valueFrom:
            secretKeyRef:
              name: user-service-secrets
              key: db-password
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 3001
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /ready
            port: 3001
          initialDelaySeconds: 5
          periodSeconds: 5
          timeoutSeconds: 3
          failureThreshold: 3

---
# user-service-service.yaml
apiVersion: v1
kind: Service
metadata:
  name: user-service
  labels:
    app: user-service
spec:
  selector:
    app: user-service
  ports:
  - name: http
    port: 80
    targetPort: 3001
    protocol: TCP
  type: ClusterIP

---
# user-service-hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: user-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: user-service
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 15
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60

---
# ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: microservices-ingress
  annotations:
    kubernetes.io/ingress.class: "nginx"
    nginx.ingress.kubernetes.io/rewrite-target: /$2
    nginx.ingress.kubernetes.io/rate-limit: "100"
    nginx.ingress.kubernetes.io/rate-limit-window: "1m"
    cert-manager.io/cluster-issuer: "letsencrypt-prod"
spec:
  tls:
  - hosts:
    - api.yourdomain.com
    secretName: api-tls-secret
  rules:
  - host: api.yourdomain.com
    http:
      paths:
      - path: /api/users(/|$)(.*)
        pathType: Prefix
        backend:
          service:
            name: user-service
            port:
              number: 80
      - path: /api/orders(/|$)(.*)
        pathType: Prefix
        backend:
          service:
            name: order-service
            port:
              number: 80
      - path: /api/payments(/|$)(.*)
        pathType: Prefix
        backend:
          service:
            name: payment-service
            port:
              number: 80

const { NodeSDK } = require('@opentelemetry/sdk-node');
const { ConsoleSpanExporter } = require('@opentelemetry/sdk-trace-node');
const { getNodeAutoInstrumentations } = require('@opentelemetry/auto-instrumentations-node');
const { PeriodicExportingMetricReader, ConsoleMetricExporter } = require('@opentelemetry/sdk-metrics');
const { JaegerExporter } = require('@opentelemetry/exporter-jaeger');

// Initialize OpenTelemetry SDK
const sdk = new NodeSDK({
  spanProcessor: new SimpleSpanProcessor(new JaegerExporter({
    endpoint: process.env.JAEGER_ENDPOINT || 'http://localhost:14268/api/traces',
  })),
  metricReader: new PeriodicExportingMetricReader({
    exporter: new ConsoleMetricExporter(),
    exportIntervalMillis: 5000,
  }),
  instrumentations: [getNodeAutoInstrumentations()],
  serviceName: process.env.SERVICE_NAME || 'user-service',
  serviceVersion: process.env.SERVICE_VERSION || '1.0.0',
});

sdk.start();

// Custom tracing middleware for Express
const { trace, context, SpanStatusCode } = require('@opentelemetry/api');

function tracingMiddleware(serviceName) {
  const tracer = trace.getTracer(serviceName);
  
  return (req, res, next) => {
    const span = tracer.startSpan(`${req.method} ${req.route?.path || req.path}`, {
      kind: SpanKind.SERVER,
      attributes: {
        'http.method': req.method,
        'http.url': req.url,
        'http.user_agent': req.get('user-agent'),
        'service.name': serviceName,
        'service.version': process.env.SERVICE_VERSION || '1.0.0'
      }
    });

    // Add request context
    const activeContext = trace.setSpan(context.active(), span);
    
    // Inject correlation ID
    const correlationId = req.headers['x-correlation-id'] || 
                         trace.getActiveSpan()?.spanContext().traceId;
    req.correlationId = correlationId;
    res.setHeader('x-correlation-id', correlationId);

    // Override res.send to capture response data
    const originalSend = res.send;
    res.send = function(data) {
      span.setAttributes({
        'http.status_code': res.statusCode,
        'http.response.size': Buffer.byteLength(data, 'utf8')
      });
      
      if (res.statusCode >= 400) {
        span.setStatus({
          code: SpanStatusCode.ERROR,
          message: `HTTP ${res.statusCode}`
        });
      }
      
      span.end();
      return originalSend.call(this, data);
    };

    context.with(activeContext, () => {
      next();
    });
  };
}

// Usage in Express app
const express = require('express');
const app = express();

app.use(tracingMiddleware('user-service'));

// Custom span creation for business logic
async function getUserById(userId) {
  const tracer = trace.getTracer('user-service');
  const span = tracer.startSpan('getUserById', {
    attributes: {
      'user.id': userId,
      'operation': 'database.query'
    }
  });

  try {
    // Simulated database call
    const user = await db.query('SELECT * FROM users WHERE id = $1', [userId]);
    
    span.setAttributes({
      'db.statement': 'SELECT * FROM users WHERE id = $1',
      'db.rows_affected': user.rowCount
    });
    
    if (!user.rows.length) {
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: 'User not found'
      });
      throw new Error('User not found');
    }
    
    span.setStatus({ code: SpanStatusCode.OK });
    return user.rows[0];
  } catch (error) {
    span.recordException(error);
    span.setStatus({
      code: SpanStatusCode.ERROR,
      message: error.message
    });
    throw error;
  } finally {
    span.end();
  }
}

// HTTP client instrumentation
const axios = require('axios');

async function callOrderService(userId) {
  const tracer = trace.getTracer('user-service');
  const span = tracer.startSpan('callOrderService', {
    kind: SpanKind.CLIENT,
    attributes: {
      'http.method': 'GET',
      'http.url': `${process.env.ORDER_SERVICE_URL}/orders/user/${userId}`,
      'service.name': 'order-service'
    }
  });

  const activeContext = trace.setSpan(context.active(), span);
  
  try {
    const response = await context.with(activeContext, async () => {
      return axios.get(`${process.env.ORDER_SERVICE_URL}/orders/user/${userId}`, {
        headers: {
          'x-correlation-id': trace.getActiveSpan()?.spanContext().traceId
        }
      });
    });
    
    span.setAttributes({
      'http.status_code': response.status,
      'http.response.size': JSON.stringify(response.data).length
    });
    
    span.setStatus({ code: SpanStatusCode.OK });
    return response.data;
  } catch (error) {
    span.recordException(error);
    span.setStatus({
      code: SpanStatusCode.ERROR,
      message: error.message
    });
    throw error;
  } finally {
    span.end();
  }
}

module.exports = { tracingMiddleware, getUserById, callOrderService };

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/gin-gonic/gin"
    "github.com/go-redis/redis/v8"
    "github.com/google/uuid"
    "github.com/streadway/amqp"
    "gorm.io/driver/postgres"
    "gorm.io/gorm"
    "gorm.io/gorm/logger"
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// Domain models
type Product struct {
    ID          uint      `gorm:"primaryKey" json:"id"`
    Name        string    `json:"name" binding:"required"`
    Description string    `json:"description"`
    Price       float64   `json:"price" binding:"required,min=0"`
    Stock       int       `json:"stock" binding:"required,min=0"`
    Category    string    `json:"category" binding:"required"`
    CreatedAt   time.Time `json:"created_at"`
    UpdatedAt   time.Time `json:"updated_at"`
}

type ProductService struct {
    db       *gorm.DB
    cache    *redis.Client
    publisher *EventPublisher
    metrics  *Metrics
}

// Event structures
type ProductCreatedEvent struct {
    EventID   string    `json:"event_id"`
    ProductID uint      `json:"product_id"`
    Name      string    `json:"name"`
    Price     float64   `json:"price"`
    Category  string    `json:"category"`
    Timestamp time.Time `json:"timestamp"`
}

type ProductUpdatedEvent struct {
    EventID     string    `json:"event_id"`
    ProductID   uint      `json:"product_id"`
    Name        string    `json:"name"`
    Price       float64   `json:"price"`
    OldPrice    float64   `json:"old_price"`
    Stock       int       `json:"stock"`
    Timestamp   time.Time `json:"timestamp"`
}

// Event publisher
type EventPublisher struct {
    conn    *amqp.Connection
    channel *amqp.Channel
}

func NewEventPublisher(rabbitmqURL string) (*EventPublisher, error) {
    conn, err := amqp.Dial(rabbitmqURL)
    if err != nil {
        return nil, fmt.Errorf("failed to connect to RabbitMQ: %v", err)
    }
    
    channel, err := conn.Channel()
    if err != nil {
        conn.Close()
        return nil, fmt.Errorf("failed to open channel: %v", err)
    }
    
    // Declare exchange
    err = channel.ExchangeDeclare(
        "ecommerce.events",
        "topic",
        true,  // durable
        false, // auto-deleted
        false, // internal
        false, // no-wait
        nil,   // arguments
    )
    if err != nil {
        channel.Close()
        conn.Close()
        return nil, fmt.Errorf("failed to declare exchange: %v", err)
    }
    
    return &EventPublisher{
        conn:    conn,
        channel: channel,
    }, nil
}

func (ep *EventPublisher) PublishEvent(eventType string, event interface{}) error {
    body, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("failed to marshal event: %v", err)
    }
    
    routingKey := fmt.Sprintf("product.%s", eventType)
    
    return ep.channel.Publish(
        "ecommerce.events",
        routingKey,
        false, // mandatory
        false, // immediate
        amqp.Publishing{
            ContentType:  "application/json",
            Body:         body,
            Timestamp:    time.Now(),
            MessageId:    uuid.New().String(),
            Type:         eventType,
            AppId:        "product-service",
            DeliveryMode: amqp.Persistent,
        },
    )
}

// Metrics
type Metrics struct {
    RequestDuration *prometheus.HistogramVec
    RequestTotal    *prometheus.CounterVec
    CacheHits       prometheus.Counter
    CacheMisses     prometheus.Counter
}

func NewMetrics() *Metrics {
    metrics := &Metrics{
        RequestDuration: prometheus.NewHistogramVec(
            prometheus.HistogramOpts{
                Name: "http_request_duration_seconds",
                Help: "Duration of HTTP requests in seconds",
            },
            []string{"method", "endpoint", "status_code"},
        ),
        RequestTotal: prometheus.NewCounterVec(
            prometheus.CounterOpts{
                Name: "http_requests_total",
                Help: "Total number of HTTP requests",
            },
            []string{"method", "endpoint", "status_code"},
        ),
        CacheHits: prometheus.NewCounter(prometheus.CounterOpts{
            Name: "cache_hits_total",
            Help: "Total number of cache hits",
        }),
        CacheMisses: prometheus.NewCounter(prometheus.CounterOpts{
            Name: "cache_misses_total",
            Help: "Total number of cache misses",
        }),
    }
    
    prometheus.MustRegister(
        metrics.RequestDuration,
        metrics.RequestTotal,
        metrics.CacheHits,
        metrics.CacheMisses,
    )
    
    return metrics
}

// Service implementation
func NewProductService(db *gorm.DB, cache *redis.Client, publisher *EventPublisher) *ProductService {
    return &ProductService{
        db:        db,
        cache:     cache,
        publisher: publisher,
        metrics:   NewMetrics(),
    }
}

func (ps *ProductService) CreateProduct(product *Product) error {
    if err := ps.db.Create(product).Error; err != nil {
        return fmt.Errorf("failed to create product: %v", err)
    }
    
    // Publish event
    event := ProductCreatedEvent{
        EventID:   uuid.New().String(),
        ProductID: product.ID,
        Name:      product.Name,
        Price:     product.Price,
        Category:  product.Category,
        Timestamp: time.Now(),
    }
    
    if err := ps.publisher.PublishEvent("created", event); err != nil {
        log.Printf("Failed to publish product created event: %v", err)
    }
    
    // Invalidate cache
    ps.cache.Del(context.Background(), fmt.Sprintf("product:%d", product.ID))
    ps.cache.Del(context.Background(), "products:all")
    
    return nil
}

func (ps *ProductService) GetProduct(id uint) (*Product, error) {
    ctx := context.Background()
    cacheKey := fmt.Sprintf("product:%d", id)
    
    // Try cache first
    cached, err := ps.cache.Get(ctx, cacheKey).Result()
    if err == nil {
        ps.metrics.CacheHits.Inc()
        var product Product
        if err := json.Unmarshal([]byte(cached), &product); err == nil {
            return &product, nil
        }
    }
    
    ps.metrics.CacheMisses.Inc()
    
    // Get from database
    var product Product
    if err := ps.db.First(&product, id).Error; err != nil {
        return nil, fmt.Errorf("product not found: %v", err)
    }
    
    // Cache the result
    productJSON, _ := json.Marshal(product)
    ps.cache.Set(ctx, cacheKey, productJSON, time.Hour)
    
    return &product, nil
}

func (ps *ProductService) UpdateProduct(id uint, updates *Product) error {
    var existingProduct Product
    if err := ps.db.First(&existingProduct, id).Error; err != nil {
        return fmt.Errorf("product not found: %v", err)
    }
    
    oldPrice := existingProduct.Price
    
    if err := ps.db.Model(&existingProduct).Updates(updates).Error; err != nil {
        return fmt.Errorf("failed to update product: %v", err)
    }
    
    // Publish event
    event := ProductUpdatedEvent{
        EventID:   uuid.New().String(),
        ProductID: id,
        Name:      existingProduct.Name,
        Price:     existingProduct.Price,
        OldPrice:  oldPrice,
        Stock:     existingProduct.Stock,
        Timestamp: time.Now(),
    }
    
    if err := ps.publisher.PublishEvent("updated", event); err != nil {
        log.Printf("Failed to publish product updated event: %v", err)
    }
    
    // Invalidate cache
    ps.cache.Del(context.Background(), fmt.Sprintf("product:%d", id))
    ps.cache.Del(context.Background(), "products:all")
    
    return nil
}

// HTTP handlers
func (ps *ProductService) SetupRoutes() *gin.Engine {
    router := gin.New()
    
    // Middleware
    router.Use(gin.Logger())
    router.Use(gin.Recovery())
    router.Use(ps.metricsMiddleware())
    router.Use(ps.corsMiddleware())
    
    // Health check
    router.GET("/health", func(c *gin.Context) {
        c.JSON(http.StatusOK, gin.H{
            "status":    "healthy",
            "timestamp": time.Now().ISO8601(),
            "service":   "product-service",
            "version":   os.Getenv("SERVICE_VERSION"),
        })
    })
    
    // Metrics endpoint
    router.GET("/metrics", gin.WrapH(promhttp.Handler()))
    
    // API routes
    v1 := router.Group("/api/v1")
    {
        v1.POST("/products", ps.createProductHandler)
        v1.GET("/products/:id", ps.getProductHandler)
        v1.PUT("/products/:id", ps.updateProductHandler)
        v1.GET("/products", ps.listProductsHandler)
    }
    
    return router
}

func (ps *ProductService) createProductHandler(c *gin.Context) {
    var product Product
    if err := c.ShouldBindJSON(&product); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }
    
    if err := ps.CreateProduct(&product); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
        return
    }
    
    c.JSON(http.StatusCreated, product)
}

func (ps *ProductService) getProductHandler(c *gin.Context) {
    var id uint
    if err := c.ShouldBindUri(struct{ ID uint `uri:"id" binding:"required"`}{&id}); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
        return
    }
    
    product, err := ps.GetProduct(id)
    if err != nil {
        c.JSON(http.StatusNotFound, gin.H{"error": "Product not found"})
        return
    }
    
    c.JSON(http.StatusOK, product)
}

// Middleware
func (ps *ProductService) metricsMiddleware() gin.HandlerFunc {
    return gin.HandlerFunc(func(c *gin.Context) {
        start := time.Now()
        
        c.Next()
        
        duration := time.Since(start).Seconds()
        statusCode := fmt.Sprintf("%d", c.Writer.Status())
        
        ps.metrics.RequestDuration.WithLabelValues(
            c.Request.Method,
            c.FullPath(),
            statusCode,
        ).Observe(duration)
        
        ps.metrics.RequestTotal.WithLabelValues(
            c.Request.Method,
            c.FullPath(),
            statusCode,
        ).Inc()
    })
}

func (ps *ProductService) corsMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        c.Header("Access-Control-Allow-Origin", "*")
        c.Header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
        c.Header("Access-Control-Allow-Headers", "Content-Type, Authorization")
        
        if c.Request.Method == "OPTIONS" {
            c.AbortWithStatus(http.StatusOK)
            return
        }
        
        c.Next()
    }
}

// Main function
func main() {
    // Database connection
    dsn := fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable",
        os.Getenv("DB_HOST"),
        os.Getenv("DB_PORT"),
        os.Getenv("DB_USER"),
        os.Getenv("DB_PASSWORD"),
        os.Getenv("DB_NAME"),
    )
    
    db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
        Logger: logger.Default.LogMode(logger.Info),
    })
    if err != nil {
        log.Fatal("Failed to connect to database:", err)
    }
    
    // Auto-migrate
    db.AutoMigrate(&Product{})
    
    // Redis connection
    rdb := redis.NewClient(&redis.Options{
        Addr:     os.Getenv("REDIS_ADDR"),
        Password: os.Getenv("REDIS_PASSWORD"),
        DB:       0,
    })
    
    // Event publisher
    publisher, err := NewEventPublisher(os.Getenv("RABBITMQ_URL"))
    if err != nil {
        log.Fatal("Failed to create event publisher:", err)
    }
    defer publisher.conn.Close()
    defer publisher.channel.Close()
    
    // Initialize service
    productService := NewProductService(db, rdb, publisher)
    router := productService.SetupRoutes()
    
    // Start server
    server := &http.Server{
        Addr:    ":8080",
        Handler: router,
    }
    
    // Graceful shutdown
    go func() {
        if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatal("Failed to start server:", err)
        }
    }()
    
    // Wait for interrupt signal
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit
    log.Println("Shutting down server...")
    
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    if err := server.Shutdown(ctx); err != nil {
        log.Fatal("Server forced to shutdown:", err)
    }
    
    log.Println("Server exited")
}