Arquiteturas Distribuídas

# 📁 k8s/istio/virtual-service.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: trpc-services
  namespace: trpc-system
spec:
  hosts:
  - api.trpc-app.com
  http:
  # 🚀 API Gateway routing
  - match:
    - uri:
        prefix: /api/trpc
    route:
    - destination:
        host: api-gateway-service
        port:
          number: 3000
      weight: 90
    - destination:
        host: api-gateway-service-canary
        port:
          number: 3000
      weight: 10
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 5s
    retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx,reset,connect-failure,refused-stream

  # 👥 User Service routing
  - match:
    - uri:
        prefix: /internal/users
    route:
    - destination:
        host: user-service
        port:
          number: 3001
    timeout: 10s
    
  # 📝 Post Service routing  
  - match:
    - uri:
        prefix: /internal/posts
    route:
    - destination:
        host: post-service
        port:
          number: 3002
    timeout: 15s

---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: trpc-services-destination
  namespace: trpc-system
spec:
  host: "*.trpc-system.svc.cluster.local"
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 10
    loadBalancer:
      simple: LEAST_CONN
    circuitBreaker:
      consecutiveErrors: 3
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
    retryPolicy:
      attempts: 3
      perTryTimeout: 2s

# 📁 k8s/istio/security-policies.yaml
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: trpc-system
spec:
  mtls:
    mode: STRICT

---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: trpc-authz
  namespace: trpc-system
spec:
  selector:
    matchLabels:
      app: api-gateway
  rules:
  # 🔐 Allow authenticated users
  - from:
    - source:
        principals: ["cluster.local/ns/trpc-system/sa/api-gateway"]
    to:
    - operation:
        methods: ["GET", "POST"]
        paths: ["/api/trpc/*"]
    when:
    - key: request.headers[authorization]
      values: ["Bearer *"]

  # 🚫 Block specific paths
  - from:
    - source:
        notPrincipals: ["cluster.local/ns/trpc-system/sa/admin"]
    to:
    - operation:
        paths: ["/api/trpc/admin.*"]
    action: DENY

---
apiVersion: security.istio.io/v1beta1
kind: RequestAuthentication
metadata:
  name: jwt-auth
  namespace: trpc-system
spec:
  selector:
    matchLabels:
      app: api-gateway
  jwtRules:
  - issuer: "https://auth.trpc-app.com"
    jwksUri: "https://auth.trpc-app.com/.well-known/jwks.json"
    audiences:
    - "trpc-api"
    forwardOriginalToken: true

// 📁 services/shared/event-bus.ts
import { EventEmitter } from 'events';
import Redis from 'ioredis';
import { z } from 'zod';

// 🎯 Event schemas
export const EventSchemas = {
  UserCreated: z.object({
    userId: z.string(),
    email: z.string(),
    name: z.string(),
    timestamp: z.date(),
  }),
  
  PostPublished: z.object({
    postId: z.string(),
    authorId: z.string(),
    title: z.string(),
    timestamp: z.date(),
  }),
  
  NotificationSent: z.object({
    notificationId: z.string(),
    userId: z.string(),
    type: z.enum(['email', 'push', 'sms']),
    status: z.enum(['sent', 'failed', 'delivered']),
    timestamp: z.date(),
  }),
};

export type Events = {
  'user.created': z.infer<typeof EventSchemas.UserCreated>;
  'post.published': z.infer<typeof EventSchemas.PostPublished>;
  'notification.sent': z.infer<typeof EventSchemas.NotificationSent>;
};

class DistributedEventBus extends EventEmitter {
  private redis: Redis;
  private subscriber: Redis;
  private serviceName: string;

  constructor(serviceName: string) {
    super();
    this.serviceName = serviceName;
    this.redis = new Redis(process.env.REDIS_URL || 'redis://localhost:6379');
    this.subscriber = new Redis(process.env.REDIS_URL || 'redis://localhost:6379');
    this.setupSubscriber();
  }

  // 📡 Publish event to distributed system
  async publish<K extends keyof Events>(
    eventName: K,
    data: Events[K],
    options: {
      retentionHours?: number;
      partitionKey?: string;
    } = {}
  ): Promise<void> {
    const event = {
      id: this.generateEventId(),
      name: eventName,
      data,
      source: this.serviceName,
      timestamp: new Date().toISOString(),
      version: '1.0',
    };

    try {
      // 🔍 Validate event schema
      const schema = EventSchemas[this.getSchemaKey(eventName)];
      if (schema) {
        schema.parse(data);
      }

      // 📊 Store event for event sourcing
      await this.storeEvent(event, options.retentionHours);

      // 📡 Publish to Redis channels
      await this.redis.publish(`events:${eventName}`, JSON.stringify(event));
      await this.redis.publish('events:all', JSON.stringify(event));

      console.log(`📡 Event published: ${eventName}`, { eventId: event.id });
    } catch (error) {
      console.error(`❌ Failed to publish event ${eventName}:`, error);
      throw error;
    }
  }

  // 🎧 Subscribe to specific events
  subscribe<K extends keyof Events>(
    eventName: K,
    handler: (data: Events[K], metadata: EventMetadata) => Promise<void> | void
  ): void {
    this.on(eventName, async (event) => {
      try {
        const startTime = Date.now();
        await handler(event.data, {
          id: event.id,
          source: event.source,
          timestamp: new Date(event.timestamp),
          version: event.version,
        });
        
        const duration = Date.now() - startTime;
        console.log(`✅ Event handled: ${eventName}`, { 
          eventId: event.id, 
          duration: `${duration}ms` 
        });
      } catch (error) {
        console.error(`❌ Event handler failed for ${eventName}:`, error);
        
        // 🔄 Dead letter queue for failed events
        await this.sendToDeadLetterQueue(event, error);
      }
    });
  }

  // 🎧 Subscribe to all events (for debugging/monitoring)
  subscribeToAll(
    handler: (eventName: string, data: any, metadata: EventMetadata) => void
  ): void {
    this.on('*', handler);
  }

  private setupSubscriber(): void {
    this.subscriber.psubscribe('events:*');
    
    this.subscriber.on('pmessage', (pattern, channel, message) => {
      try {
        const event = JSON.parse(message);
        const eventName = channel.replace('events:', '');
        
        if (eventName === 'all') {
          this.emit('*', event.name, event.data, {
            id: event.id,
            source: event.source,
            timestamp: new Date(event.timestamp),
            version: event.version,
          });
        } else {
          this.emit(eventName, event);
        }
      } catch (error) {
        console.error('Failed to parse event message:', error);
      }
    });
  }

  private async storeEvent(
    event: any, 
    retentionHours: number = 24 * 7 // 7 days default
  ): Promise<void> {
    const key = `event:${event.id}`;
    const ttl = retentionHours * 60 * 60; // Convert to seconds
    
    await this.redis.setex(key, ttl, JSON.stringify(event));
    
    // 📊 Add to event stream for event sourcing
    await this.redis.zadd(
      `events:${event.name}:stream`,
      Date.now(),
      event.id
    );
  }

  private async sendToDeadLetterQueue(event: any, error: Error): Promise<void> {
    const dlqEvent = {
      ...event,
      error: {
        message: error.message,
        stack: error.stack,
        timestamp: new Date().toISOString(),
      },
      retryCount: (event.retryCount || 0) + 1,
    };

    await this.redis.lpush('events:dlq', JSON.stringify(dlqEvent));
  }

  private generateEventId(): string {
    return `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
  }

  private getSchemaKey(eventName: string): keyof typeof EventSchemas {
    const mapping: Record<string, keyof typeof EventSchemas> = {
      'user.created': 'UserCreated',
      'post.published': 'PostPublished',
      'notification.sent': 'NotificationSent',
    };
    return mapping[eventName] || 'UserCreated';
  }
}

interface EventMetadata {
  id: string;
  source: string;
  timestamp: Date;
  version: string;
}

// 🚀 Global event bus instance
export const eventBus = new DistributedEventBus(
  process.env.SERVICE_NAME || 'unknown-service'
);

// 🎯 Event handler decorators
export function EventHandler<K extends keyof Events>(eventName: K) {
  return function (
    target: any,
    propertyKey: string,
    descriptor: PropertyDescriptor
  ) {
    const originalMethod = descriptor.value;

    // Register handler when class is instantiated
    setImmediate(() => {
      eventBus.subscribe(eventName, originalMethod.bind(target));
    });

    return descriptor;
  };
}

// 📁 services/user-service/src/cqrs/user-aggregate.ts
import { eventBus } from '@/shared/event-bus';

interface UserState {
  id: string;
  email: string;
  name: string;
  isActive: boolean;
  createdAt: Date;
  updatedAt: Date;
  version: number;
}

class UserAggregate {
  private state: UserState;
  private uncommittedEvents: any[] = [];

  constructor(id: string) {
    this.state = {
      id,
      email: '',
      name: '',
      isActive: false,
      createdAt: new Date(),
      updatedAt: new Date(),
      version: 0,
    };
  }

  // 📝 Commands (Write operations)
  createUser(data: { email: string; name: string; password: string }): void {
    if (this.state.email) {
      throw new Error('User already exists');
    }

    // ✅ Business logic validation
    if (!this.isValidEmail(data.email)) {
      throw new Error('Invalid email format');
    }

    if (data.password.length < 8) {
      throw new Error('Password must be at least 8 characters');
    }

    // 🎉 Generate domain event
    const event = {
      type: 'UserCreated',
      data: {
        userId: this.state.id,
        email: data.email,
        name: data.name,
        timestamp: new Date(),
      },
    };

    this.applyEvent(event);
    this.uncommittedEvents.push(event);
  }

  updateProfile(data: { name?: string; email?: string }): void {
    if (!this.state.isActive) {
      throw new Error('Cannot update inactive user');
    }

    const changes: any = {};
    
    if (data.name && data.name !== this.state.name) {
      changes.name = data.name;
    }
    
    if (data.email && data.email !== this.state.email) {
      if (!this.isValidEmail(data.email)) {
        throw new Error('Invalid email format');
      }
      changes.email = data.email;
    }

    if (Object.keys(changes).length === 0) {
      return; // No changes
    }

    const event = {
      type: 'UserProfileUpdated',
      data: {
        userId: this.state.id,
        changes,
        timestamp: new Date(),
      },
    };

    this.applyEvent(event);
    this.uncommittedEvents.push(event);
  }

  deactivateUser(): void {
    if (!this.state.isActive) {
      throw new Error('User is already inactive');
    }

    const event = {
      type: 'UserDeactivated',
      data: {
        userId: this.state.id,
        timestamp: new Date(),
      },
    };

    this.applyEvent(event);
    this.uncommittedEvents.push(event);
  }

  // 🎭 Event sourcing - Apply events to state
  private applyEvent(event: any): void {
    switch (event.type) {
      case 'UserCreated':
        this.state.email = event.data.email;
        this.state.name = event.data.name;
        this.state.isActive = true;
        this.state.createdAt = event.data.timestamp;
        break;

      case 'UserProfileUpdated':
        Object.assign(this.state, event.data.changes);
        this.state.updatedAt = event.data.timestamp;
        break;

      case 'UserDeactivated':
        this.state.isActive = false;
        this.state.updatedAt = event.data.timestamp;
        break;
    }

    this.state.version++;
    this.state.updatedAt = event.data.timestamp;
  }

  // 💾 Persist events and publish to event bus
  async commit(): Promise<void> {
    if (this.uncommittedEvents.length === 0) {
      return;
    }

    try {
      // 💾 Store events in event store
      for (const event of this.uncommittedEvents) {
        await this.saveEventToStore(event);
        
        // 📡 Publish to event bus for other services
        if (event.type === 'UserCreated') {
          await eventBus.publish('user.created', event.data);
        }
      }

      this.uncommittedEvents = [];
    } catch (error) {
      console.error('Failed to commit events:', error);
      throw error;
    }
  }

  // 🔄 Rebuild state from events (Event Sourcing)
  static async fromHistory(userId: string): Promise<UserAggregate> {
    const aggregate = new UserAggregate(userId);
    const events = await aggregate.loadEventsFromStore(userId);

    for (const event of events) {
      aggregate.applyEvent(event);
    }

    return aggregate;
  }

  // 📊 Getters for read operations
  getState(): Readonly<UserState> {
    return { ...this.state };
  }

  getId(): string {
    return this.state.id;
  }

  isActive(): boolean {
    return this.state.isActive;
  }

  // 🔧 Private helpers
  private isValidEmail(email: string): boolean {
    return /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email);
  }

  private async saveEventToStore(event: any): Promise<void> {
    // Implementation depends on your event store (EventStore, MongoDB, etc.)
    const eventStore = await import('../infrastructure/event-store');
    await eventStore.saveEvent(this.state.id, event, this.state.version);
  }

  private async loadEventsFromStore(userId: string): Promise<any[]> {
    const eventStore = await import('../infrastructure/event-store');
    return eventStore.getEvents(userId);
  }
}

export { UserAggregate };

// 📁 services/shared/tracing.ts
import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { Resource } from '@opentelemetry/resources';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';
import { JaegerExporter } from '@opentelemetry/exporter-jaeger';
import { trace, context, SpanStatusCode, SpanKind } from '@opentelemetry/api';

// 🔧 Initialize OpenTelemetry SDK
const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: process.env.SERVICE_NAME || 'trpc-service',
    [SemanticResourceAttributes.SERVICE_VERSION]: process.env.SERVICE_VERSION || '1.0.0',
    [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: process.env.NODE_ENV || 'development',
  }),
  
  traceExporter: new JaegerExporter({
    endpoint: process.env.JAEGER_ENDPOINT || 'http://jaeger:14268/api/traces',
  }),
  
  instrumentations: [
    getNodeAutoInstrumentations({
      // 🎯 Customize instrumentation
      '@opentelemetry/instrumentation-fs': {
        enabled: false, // Disable filesystem instrumentation
      },
      '@opentelemetry/instrumentation-http': {
        enabled: true,
        requestHook: (span, request) => {
          span.setAttributes({
            'http.request.header.user-agent': request.getHeader('user-agent'),
            'http.request.header.x-forwarded-for': request.getHeader('x-forwarded-for'),
          });
        },
      },
      '@opentelemetry/instrumentation-express': {
        enabled: true,
      },
    }),
  ],
});

// 🚀 Start tracing
sdk.start();

// 🔧 Helper functions for manual instrumentation
export class TracingHelper {
  private static tracer = trace.getTracer('trpc-application');

  // 🎯 Create span for tRPC procedures
  static async traceProcedure<T>(
    procedureName: string,
    operation: () => Promise<T>,
    attributes: Record<string, string | number | boolean> = {}
  ): Promise<T> {
    const span = this.tracer.startSpan(`trpc.${procedureName}`, {
      kind: SpanKind.SERVER,
      attributes: {
        'trpc.procedure': procedureName,
        'trpc.type': 'procedure',
        ...attributes,
      },
    });

    try {
      const result = await context.with(trace.setSpan(context.active(), span), operation);
      span.setStatus({ code: SpanStatusCode.OK });
      return result;
    } catch (error) {
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: error instanceof Error ? error.message : 'Unknown error',
      });
      
      span.setAttributes({
        'error': true,
        'error.name': error instanceof Error ? error.constructor.name : 'UnknownError',
        'error.message': error instanceof Error ? error.message : String(error),
      });
      
      throw error;
    } finally {
      span.end();
    }
  }

  // 📡 Trace external service calls
  static async traceExternalCall<T>(
    serviceName: string,
    operation: string,
    call: () => Promise<T>,
    attributes: Record<string, any> = {}
  ): Promise<T> {
    const span = this.tracer.startSpan(`external.${serviceName}.${operation}`, {
      kind: SpanKind.CLIENT,
      attributes: {
        'service.name': serviceName,
        'service.operation': operation,
        'service.type': 'external',
        ...attributes,
      },
    });

    try {
      const result = await context.with(trace.setSpan(context.active(), span), call);
      span.setStatus({ code: SpanStatusCode.OK });
      return result;
    } catch (error) {
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: error instanceof Error ? error.message : 'External call failed',
      });
      
      span.recordException(error as Error);
      throw error;
    } finally {
      span.end();
    }
  }

  // 🗄️ Trace database operations
  static async traceDatabase<T>(
    operation: string,
    query: string,
    executor: () => Promise<T>,
    attributes: Record<string, any> = {}
  ): Promise<T> {
    const span = this.tracer.startSpan(`db.${operation}`, {
      kind: SpanKind.CLIENT,
      attributes: {
        'db.system': 'postgresql',
        'db.operation': operation,
        'db.statement': query.length > 1000 ? query.substring(0, 1000) + '...' : query,
        ...attributes,
      },
    });

    const startTime = Date.now();

    try {
      const result = await context.with(trace.setSpan(context.active(), span), executor);
      
      span.setAttributes({
        'db.rows_affected': Array.isArray(result) ? result.length : 1,
        'db.duration_ms': Date.now() - startTime,
      });
      
      span.setStatus({ code: SpanStatusCode.OK });
      return result;
    } catch (error) {
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: error instanceof Error ? error.message : 'Database operation failed',
      });
      
      span.recordException(error as Error);
      throw error;
    } finally {
      span.end();
    }
  }

  // 📊 Get current trace context
  static getCurrentTraceId(): string | null {
    const activeSpan = trace.getActiveSpan();
    return activeSpan ? activeSpan.spanContext().traceId : null;
  }

  // 🔗 Inject trace context into headers
  static injectTraceHeaders(): Record<string, string> {
    const headers: Record<string, string> = {};
    
    trace.propagation.inject(context.active(), headers);
    
    return headers;
  }

  // 📥 Extract trace context from headers
  static extractTraceContext(headers: Record<string, string>): void {
    const extractedContext = trace.propagation.extract(context.active(), headers);
    context.with(extractedContext, () => {
      // Context is now active
    });
  }
}

// 🎯 tRPC middleware for automatic tracing
export function createTracingMiddleware() {
  return async (opts: any) => {
    const { path, type, next, ctx } = opts;
    
    return TracingHelper.traceProcedure(
      `${type}.${path}`,
      async () => {
        // 🔗 Add trace ID to context
        const traceId = TracingHelper.getCurrentTraceId();
        if (traceId) {
          ctx.traceId = traceId;
        }
        
        // 📊 Add custom attributes
        const span = trace.getActiveSpan();
        if (span) {
          span.setAttributes({
            'user.id': ctx.session?.user?.id || 'anonymous',
            'request.path': path,
            'request.type': type,
          });
        }
        
        return next();
      },
      {
        'trpc.path': path,
        'trpc.type': type,
        'user.authenticated': !!ctx.session?.user,
      }
    );
  };
}

// 🔧 Enhanced service client with tracing
export function createTracedServiceClient<T>(
  serviceName: string,
  client: T
): T {
  return new Proxy(client as any, {
    get(target, prop) {
      const originalMethod = target[prop];
      
      if (typeof originalMethod === 'function') {
        return function (...args: any[]) {
          return TracingHelper.traceExternalCall(
            serviceName,
            String(prop),
            () => originalMethod.apply(target, args),
            {
              'service.method': String(prop),
              'service.args_count': args.length,
            }
          );
        };
      }
      
      return originalMethod;
    },
  });
}

export { sdk as tracingSDK };

// 📁 services/shared/observability.ts
import { metrics } from '@opentelemetry/api';
import { MeterProvider } from '@opentelemetry/sdk-metrics';
import { PrometheusExporter } from '@opentelemetry/exporter-prometheus';

// 🔧 Setup metrics
const meterProvider = new MeterProvider();
const exporter = new PrometheusExporter({
  port: 9090,
  endpoint: '/metrics',
});

meterProvider.addMetricReader(exporter);
metrics.setGlobalMeterProvider(meterProvider);

const meter = metrics.getMeter('trpc-application', '1.0.0');

// 📊 Custom metrics for tRPC
export const trpcMetrics = {
  // 📈 Request counter
  requestCounter: meter.createCounter('trpc_requests_total', {
    description: 'Total number of tRPC requests',
  }),

  // ⏱️ Request duration histogram
  requestDuration: meter.createHistogram('trpc_request_duration_seconds', {
    description: 'Duration of tRPC requests in seconds',
    boundaries: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 2, 5],
  }),

  // 🔢 Active connections gauge
  activeConnections: meter.createUpDownCounter('trpc_active_connections', {
    description: 'Number of active tRPC connections',
  }),

  // 📊 Cache hit ratio
  cacheHitRatio: meter.createGauge('trpc_cache_hit_ratio', {
    description: 'Cache hit ratio for tRPC operations',
  }),

  // 🗄️ Database query duration
  dbQueryDuration: meter.createHistogram('trpc_db_query_duration_seconds', {
    description: 'Duration of database queries in seconds',
    boundaries: [0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1],
  }),
};

// 🎯 Metrics middleware for tRPC
export function createMetricsMiddleware() {
  return async (opts: any) => {
    const { path, type, next } = opts;
    const startTime = Date.now();

    // 📈 Increment request counter
    trpcMetrics.requestCounter.add(1, {
      procedure: path,
      type: type,
      service: process.env.SERVICE_NAME || 'unknown',
    });

    // 🔢 Track active connections
    trpcMetrics.activeConnections.add(1, {
      service: process.env.SERVICE_NAME || 'unknown',
    });

    try {
      const result = await next();
      
      // ✅ Record successful request duration
      const duration = (Date.now() - startTime) / 1000;
      trpcMetrics.requestDuration.record(duration, {
        procedure: path,
        type: type,
        status: 'success',
        service: process.env.SERVICE_NAME || 'unknown',
      });

      return result;
    } catch (error) {
      // ❌ Record failed request duration
      const duration = (Date.now() - startTime) / 1000;
      trpcMetrics.requestDuration.record(duration, {
        procedure: path,
        type: type,
        status: 'error',
        error_type: error instanceof Error ? error.constructor.name : 'unknown',
        service: process.env.SERVICE_NAME || 'unknown',
      });

      throw error;
    } finally {
      // 📉 Decrement active connections
      trpcMetrics.activeConnections.add(-1, {
        service: process.env.SERVICE_NAME || 'unknown',
      });
    }
  };
}

// 📊 Business metrics collector
export class BusinessMetrics {
  private static userActionCounter = meter.createCounter('business_user_actions_total', {
    description: 'Total number of user actions',
  });

  private static revenueGauge = meter.createGauge('business_revenue_total', {
    description: 'Total revenue in USD',
  });

  static recordUserAction(action: string, userId: string, metadata: Record<string, any> = {}): void {
    this.userActionCounter.add(1, {
      action,
      user_id: userId,
      service: process.env.SERVICE_NAME || 'unknown',
      ...metadata,
    });
  }

  static updateRevenue(amount: number, currency: string = 'USD'): void {
    this.revenueGauge.record(amount, {
      currency,
      service: process.env.SERVICE_NAME || 'unknown',
    });
  }

  static recordConversion(funnel: string, step: string, success: boolean): void {
    const conversionCounter = meter.createCounter('business_conversions_total', {
      description: 'Conversion funnel tracking',
    });

    conversionCounter.add(1, {
      funnel,
      step,
      success: success.toString(),
      service: process.env.SERVICE_NAME || 'unknown',
    });
  }
}

export { meterProvider as metricsProvider };