Olay Güdümlü Mimari Araçları: Kafka, SQS, EventBridge ve Cloud Alternatifleri Üzerine Kapsamlı Rehber

// Uygun error handling ve DLQ ile SQS
const params = {
  QueueUrl: 'https://sqs.us-east-1.amazonaws.com/123/my-queue',
  ReceiveMessageWaitTimeSeconds: 20,  // Long polling
  MaxNumberOfMessages: 10,
  VisibilityTimeout: 30,  // İşleme penceresi
  MessageAttributeNames: ['All']
};

// DLQ konfigürasyonu
const dlqParams = {
  QueueName: 'my-queue-dlq',
  Attributes: {
    MessageRetentionPeriod: '1209600',  // 14 gün
    RedrivePolicy: JSON.stringify({
      deadLetterTargetArn: dlqArn,
      maxReceiveCount: 3  // DLQ'ya gitmeden önce 3 kez dene
    })
  }
};

// Session'lar ve DLQ handling ile Service Bus
var client = new ServiceBusClient(connectionString);
var processor = client.CreateProcessor(queueName, new ServiceBusProcessorOptions
{
    MaxConcurrentCalls = 10,
    AutoCompleteMessages = false,
    MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5),
    SubQueue = SubQueue.DeadLetter  // DLQ'ya erişim
});

// Duplikasyon algılamalı mesaj
var message = new ServiceBusMessage(body)
{
    MessageId = Guid.NewGuid().ToString(),  // Deduplication için
    SessionId = sessionId,  // Sıralı işleme için
    TimeToLive = TimeSpan.FromMinutes(5)
};

// Akıllı routing için filter policy'li SNS
const publishParams = {
  TopicArn: 'arn:aws:sns:us-east-1:123:my-topic',
  Message: JSON.stringify(event),
  MessageAttributes: {
    eventType: { DataType: 'String', StringValue: 'ORDER_CREATED' },
    priority: { DataType: 'Number', StringValue: '1' }
  }
};

// Filter'lı subscription
const subscriptionPolicy = {
  eventType: ['ORDER_CREATED', 'ORDER_UPDATED'],
  priority: [{ numeric: ['>', 0] }]
};

class MessageProcessor {
  async processWithRetry(message: Message, maxRetries = 3) {
    let retryCount = 0;
    let lastError;

    while (retryCount < maxRetries) {
      try {
        return await this.process(message);
      } catch (error) {
        lastError = error;
        retryCount++;

        // Exponential backoff: 1s, 2s, 4s
        const delay = Math.pow(2, retryCount - 1) * 1000;
        await new Promise(resolve => setTimeout(resolve, delay));

        // Retry metadata ekle
        message.metadata = {
          ...message.metadata,
          retryCount,
          lastError: error.message,
          lastRetryTimestamp: new Date().toISOString()
        };
      }
    }

    // Max retry'dan sonra DLQ'ya gönder
    await this.sendToDLQ(message, lastError);
  }
}

class CircuitBreakerDLQ {
  private failureCount = 0;
  private state: 'CLOSED' | 'OPEN' | 'HALF_OPEN' = 'CLOSED';

  async processMessage(message: Message) {
    if (this.state === 'OPEN') {
      // Circuit açık, direkt DLQ'ya
      return this.sendToDLQ(message, 'Circuit breaker open');
    }

    try {
      const result = await this.process(message);
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();

      if (this.state === 'OPEN') {
        await this.sendToDLQ(message, error);
      } else {
        // Retry mantığı
        await this.retryQueue.send(message);
      }
    }
  }

  private onFailure() {
    this.failureCount++;

    if (this.failureCount >= 5) {
      this.state = 'OPEN';
      console.error('Circuit breaker hatalar nedeniyle açıldı');
    }
  }
}

// Edge-optimize edilmiş event işleme
class EdgeEventProcessor {
  private localQueue: Queue[] = [];
  private cloudBuffer: Message[] = [];

  async processEvent(event: Event) {
    // Önce lokal işle
    const processed = await this.localProcess(event);

    // Cloud sync için batch'le
    if (this.shouldSyncToCloud(processed)) {
      this.cloudBuffer.push(processed);

      if (this.cloudBuffer.length >= 100 ||
          Date.now() - this.lastSync > 60000) {
        await this.syncToCloud();
      }
    }
  }

  private async syncToCloud() {
    try {
      // Sıkıştır ve batch gönder
      const compressed = this.compress(this.cloudBuffer);
      await this.cloudClient.sendBatch(compressed);
      this.cloudBuffer = [];
    } catch (error) {
      // Cloud ulaşılamazsa lokal sakla
      await this.localStorage.store(this.cloudBuffer);
    }
  }
}

// Cloudflare Workers Queue Handler
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        // Edge'de işle
        const result = await processMessage(message.body);

        // Durable Objects veya KV'de sakla
        await env.KV.put(
          `processed:${message.id}`,
          JSON.stringify(result),
          { expirationTtl: 3600 }
        );

        message.ack();
      } catch (error) {
        // Backoff ile retry
        message.retry({ delaySeconds: 30 });
      }
    }
  }
};

// Çözüm: Claim check pattern
class LargeMessageHandler {
  async send(largePayload: any) {
    if (JSON.stringify(largePayload).length > 256000) {
      // S3'e sakla
      const s3Key = await this.uploadToS3(largePayload);

      // Referans gönder
      return this.queue.send({
        type: 'large_message',
        s3Key,
        size: largePayload.length
      });
    }

    return this.queue.send(largePayload);
  }
}

// Çözüm: Zehirli mesaj algılama
class PoisonMessageDetector {
  private messageAttempts = new Map<string, number>();

  async process(message: Message) {
    const attempts = this.messageAttempts.get(message.id) || 0;

    if (attempts >= 3) {
      // Zehirli mesaj olarak tanımlandı
      await this.quarantine(message);
      return;
    }

    try {
      await this.processMessage(message);
      this.messageAttempts.delete(message.id);
    } catch (error) {
      this.messageAttempts.set(message.id, attempts + 1);

      if (this.isPoisonPattern(error)) {
        await this.quarantine(message);
      } else {
        throw error; // Retry
      }
    }
  }
}

// Çözüm: Partition key stratejisi
class OrderedEventProcessor {
  async publishOrdered(events: Event[]) {
    // Entity ID'ye göre sıralama için grupla
    const grouped = this.groupBy(events, e => e.entityId);

    for (const [entityId, entityEvents] of grouped) {
      // Timestamp'e göre sırala
      entityEvents.sort((a, b) => a.timestamp - b.timestamp);

      // Aynı partition key ile gönder
      for (const event of entityEvents) {
        await this.kafka.send({
          topic: 'events',
          key: entityId,  // Sıralamayı garanti eder
          value: event
        });
      }
    }
  }
}

// Kapsamlı metrik toplama
class EventMetrics {
  private metrics = {
    messagesPublished: new Counter('messages_published_total'),
    messagesConsumed: new Counter('messages_consumed_total'),
    messagesFailed: new Counter('messages_failed_total'),
    processingDuration: new Histogram('message_processing_duration_seconds'),
    queueDepth: new Gauge('queue_depth'),
    consumerLag: new Gauge('consumer_lag'),
    dlqDepth: new Gauge('dlq_depth')
  };

  async recordProcessing(message: Message, processor: Function) {
    const timer = this.metrics.processingDuration.startTimer();

    try {
      const result = await processor(message);
      this.metrics.messagesConsumed.inc();
      return result;
    } catch (error) {
      this.metrics.messagesFailed.inc({
        error_type: error.constructor.name,
        queue: message.source
      });
      throw error;
    } finally {
      timer();
    }
  }
}