# 创建 Webhook Topic（6 个分区，3 副本）
kafka-topics.sh --create \
  --topic webhook-events \
  --partitions 6 \
  --replication-factor 3 \
  --bootstrap-server kafka:9092

# 关键配置
topic config:
  retention.ms=604800000       # 保留 7 天
  min.insync.replicas=2        # 至少 2 副本确认写入
  cleanup.policy=delete        # 过期删除

package main

import (
    "context"
    "encoding/json"
    "log"
    "time"
    "github.com/segmentio/kafka-go"
)

var writer = kafka.NewWriter(kafka.WriterConfig{
    Brokers:      []string{"kafka:9092"},
    Topic:        "webhook-events",
    Balancer:     &kafka.LeastBytes{},
    BatchSize:    100,
    BatchTimeout: 10 * time.Millisecond,
    Async:        true,
})

type WebhookEvent struct {
    EventID     string          `json:"event_id"`
    EventType   string          `json:"event_type"`
    Payload     json.RawMessage `json:"payload"`
    Signature   string          `json:"signature"`
    Timestamp   int64           `json:"timestamp"`
    ReceivedAt  time.Time       `json:"received_at"`
}

func enqueueWebhook(ctx context.Context, event *WebhookEvent) error {
    event.ReceivedAt = time.Now()
    data, _ := json.Marshal(event)
    return writer.WriteMessages(ctx, kafka.Message{
        Key:   []byte(event.EventID),
        Value: data,
    })
}

// HTTP Handler（Gateway 层）
func webhookHandler(w http.ResponseWriter, r *http.Request) {
    body, _ := io.ReadAll(r.Body)
    
    event := &WebhookEvent{
        EventID:   r.Header.Get("X-Event-ID"),
        EventType: r.Header.Get("X-Event-Type"),
        Payload:   body,
        Signature: r.Header.Get("X-Signature"),
        Timestamp: time.Now().Unix(),
    }
    
    if err := enqueueWebhook(r.Context(), event); err != nil {
        log.Printf("Failed to enqueue: %v", err)
        http.Error(w, "Service Unavailable", http.StatusServiceUnavailable)
        return
    }
    
    w.WriteHeader(http.StatusOK)
}

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "sync"
    "time"
    "github.com/segmentio/kafka-go"
)

const (
    workerCount   = 8
    maxRetries    = 3
    retryDelay    = 5 * time.Second
)

type WorkerPool struct {
    reader  *kafka.Reader
    workers int
    wg      sync.WaitGroup
    handler func(*WebhookEvent) error
}

func NewWorkerPool(brokers []string, topic string, groupID string, workers int) *WorkerPool {
    return &WorkerPool{
        reader: kafka.NewReader(kafka.ReaderConfig{
            Brokers:         brokers,
            Topic:           topic,
            GroupID:         groupID,
            MinBytes:        1,
            MaxBytes:        10e6, // 10MB
            MaxWait:         500 * time.Millisecond,
            ReadLagInterval: 1 * time.Second,
        }),
        workers: workers,
    }
}

func (p *WorkerPool) Start(ctx context.Context) {
    for i := 0; i < p.workers; i++ {
        p.wg.Add(1)
        go p.worker(ctx, i)
    }
}

func (p *WorkerPool) worker(ctx context.Context, id int) {
    defer p.wg.Done()
    log.Printf("Worker %d started", id)
    
    for {
        select {
        case <-ctx.Done():
            return
        default:
        }
        
        msg, err := p.reader.ReadMessage(ctx)
        if err != nil {
            if ctx.Err() != nil {
                return
            }
            log.Printf("Worker %d read error: %v", id, err)
            continue
        }
        
        var event WebhookEvent
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            log.Printf("Worker %d unmarshal error: %v", id, err)
            continue
        }
        
        if err := p.processWithRetry(&event); err != nil {
            log.Printf("Worker %d failed to process %s: %v", id, event.EventID, err)
            // 发送到 DLQ
            sendToDLQ(&event, err)
        }
    }
}

func (p *WorkerPool) processWithRetry(event *WebhookEvent) error {
    var err error
    for attempt := 0; attempt <= maxRetries; attempt++ {
        if err = p.handler(event); err == nil {
            return nil
        }
        if attempt < maxRetries {
            time.Sleep(retryDelay * time.Duration(attempt+1))
        }
    }
    return err
}

func (p *WorkerPool) Stop() {
    p.reader.Close()
    p.wg.Wait()
}

// 业务处理函数
func handleWebhook(event *WebhookEvent) error {
    // 1. 签名校验
    if !verifySignature(event.Payload, event.Signature, secret) {
        return fmt.Errorf("signature mismatch")
    }
    
    // 2. 幂等性检查
    if isDuplicate(event.EventID) {
        return nil
    }
    
    // 3. 业务处理
    return doBusinessLogic(event.Payload)
}

func sendToDLQ(event *WebhookEvent, err error) {
    // 实现：写入 DLQ topic 或数据库
    log.Printf("[DLQ] event=%s error=%v", event.EventID, err)
}

func main() {
    pool := NewWorkerPool(
        []string{"kafka:9092"},
        "webhook-events",
        "webhook-consumer-group",
        workerCount,
    )
    pool.handler = handleWebhook
    
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    
    pool.Start(ctx)
    
    // 等待信号退出
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    <-sig
    
    log.Println("Shutting down...")
    pool.Stop()
}

import "golang.org/x/time/rate"

var limiter = rate.NewLimiter(rate.Limit(1000), 2000) // 1000 req/s，burst 2000

func rateLimitMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if !limiter.Allow() {
            http.Error(w, "Rate Limited", http.StatusTooManyRequests)
            return
        }
        next.ServeHTTP(w, r)
    })
}

type CircuitBreaker struct {
    failures    int
    threshold   int
    timeout     time.Duration
    lastFailure time.Time
    mu          sync.Mutex
}

func (cb *CircuitBreaker) Call(fn func() error) error {
    cb.mu.Lock()
    if cb.failures >= cb.threshold {
        if time.Since(cb.lastFailure) < cb.timeout {
            cb.mu.Unlock()
            return fmt.Errorf("circuit breaker open")
        }
        cb.failures = 0 // 半开状态
    }
    cb.mu.Unlock()
    
    err := fn()
    
    cb.mu.Lock()
    defer cb.mu.Unlock()
    if err != nil {
        cb.failures++
        cb.lastFailure = time.Now()
    } else {
        cb.failures = 0
    }
    return err
}

// 使用
var cb = &CircuitBreaker{threshold: 5, timeout: 30 * time.Second}

func handleWithCircuitBreaker(event *WebhookEvent) error {
    return cb.Call(func() error {
        return doBusinessLogic(event.Payload)
    })
}

// webhook-load-test.js
import http from 'k6/http';
import { check } from 'k6';

export let options = {
    stages: [
        { duration: '1m', target: 1000 },   // 1 分钟 ramp up 到 1000 RPS
        { duration: '3m', target: 1000 },   // 持续 3 分钟
        { duration: '1m', target: 0 },      // 1 分钟 ramp down
    ],
    thresholds: {
        http_req_duration: ['p(95)<100'],    // 95% 请求 < 100ms
        http_req_failed: ['rate<0.01'],      // 错误率 < 1%
    },
};

export default function() {
    let payload = JSON.stringify({
        id: `evt_${__VU}_${__ITER}`,
        type: 'invoice.paid',
        data: { amount: 2000 }
    });
    
    let headers = {
        'Content-Type': 'application/json',
        'X-Signature': 'sha256=abc...',
    };
    
    let res = http.post('https://api.example.com/webhook', payload, { headers });
    
    check(res, {
        'status is 200': (r) => r.status === 200,
        'response time < 100ms': (r) => r.timings.duration < 100,
    });
}

# 运行压测
k6 run --out quick webhook-load-test.js

# Queue Lag
kafka_consumer_group_lag{group="webhook-consumer-group"}

# Processing Rate
rate(webhook_processed_total[1m])

# Error Rate
rate(webhook_errors_total[1m]) / rate(webhook_received_total[1m])

# P99 Latency
histogram_quantile(0.99, rate(webhook_processing_duration_bucket[5m]))