Replay Strategy #

// ANTI-PATTERN: retry tanpa batas, tanpa delay — membombardir downstream
func processEvent(event Event) error {
    for {
        err := handler.Process(event)
        if err == nil {
            return nil
        }
        // jika downstream down, ini loop selamanya dan memperparah kondisi
    }
}

// BENAR: retry dengan batas maksimum yang jelas
func processWithRetry(event Event, maxAttempts int) error {
    var lastErr error
    for attempt := 1; attempt <= maxAttempts; attempt++ {
        lastErr = handler.Process(event)
        if lastErr == nil {
            return nil
        }
        log.Warnf("attempt %d/%d failed for event %s: %v",
            attempt, maxAttempts, event.ID, lastErr)
    }
    return fmt.Errorf("all %d attempts failed: %w", maxAttempts, lastErr)
}

// ANTI-PATTERN: fixed delay — tidak adaptif terhadap lama kegagalan
func processWithFixedRetry(event Event) error {
    for i := 0; i < 5; i++ {
        if err := handler.Process(event); err == nil {
            return nil
        }
        time.Sleep(1 * time.Second) // selalu 1 detik, tidak peduli berapa lama down
    }
    return errors.New("max retries exceeded")
}

// BENAR: exponential backoff dengan jitter mencegah thundering herd
func processWithBackoff(ctx context.Context, event Event) error {
    baseDelay := 1 * time.Second
    maxDelay  := 5 * time.Minute
    maxAttempts := 8

    for attempt := 1; attempt <= maxAttempts; attempt++ {
        err := handler.Process(event)
        if err == nil {
            return nil
        }

        if attempt == maxAttempts {
            return fmt.Errorf("exhausted %d attempts: %w", maxAttempts, err)
        }

        // Delay: 1s, 2s, 4s, 8s, 16s, 32s, 60s (capped)...
        delay := baseDelay * time.Duration(1<<uint(attempt-1))
        if delay > maxDelay {
            delay = maxDelay
        }

        // Jitter ±25%: agar tidak semua consumer retry pada waktu yang persis sama
        jitter := time.Duration(rand.Int63n(int64(delay / 4)))
        if rand.Intn(2) == 0 {
            delay += jitter
        } else {
            delay -= jitter
        }

        log.Infof("retry attempt %d in %v for event %s", attempt+1, delay, event.ID)

        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(delay):
        }
    }
    return nil
}

// ANTI-PATTERN: buang event jika gagal — kehilangan data
func (c *Consumer) handleMessage(msg Message) error {
    if err := c.process(msg); err != nil {
        log.Errorf("failed to process %s, dropping: %v", msg.ID, err)
        return nil // event hilang selamanya
    }
    return nil
}

// BENAR: kirim ke DLQ dengan metadata lengkap, jangan buang event
func (c *Consumer) handleWithDLQ(ctx context.Context, msg Message) error {
    err := c.processWithBackoff(ctx, msg)
    if err == nil {
        return nil
    }

    // Enrich dengan metadata untuk debugging nanti
    dlqMsg := DLQMessage{
        OriginalMessage: msg,
        FailureReason:   err.Error(),
        AttemptCount:    msg.ApproximateReceiveCount,
        FailedAt:        time.Now(),
        ConsumerVersion: c.version,
        StackTrace:      debug.Stack(),
    }

    if dlqErr := c.dlqQueue.Send(ctx, dlqMsg); dlqErr != nil {
        // DLQ send gagal juga — log CRITICAL, butuh perhatian segera
        log.Criticalf("FAILED to send to DLQ, event %s will be lost: %v",
            msg.ID, dlqErr)
        return dlqErr
    }

    log.Warnf("event %s sent to DLQ after %d attempts: %v",
        msg.ID, msg.ApproximateReceiveCount, err)
    return nil // ack message utama agar tidak looping di queue utama
}

// ANTI-PATTERN: replay tanpa throttle — membanjiri downstream sekaligus
func replayAllFromDLQ() {
    events := dlq.GetAll() // bisa jutaan event
    for _, e := range events {
        process(e) // semua diproses secepat mungkin — sistem bisa down
    }
}

// BENAR: replay terkontrol dengan rate limiter dan progress tracking
func (r *ReplayService) ReplayFromDLQ(ctx context.Context, opts ReplayOptions) error {
    log.Infof("starting replay: from=%v to=%v filter=%s rate=%f/s dryRun=%v",
        opts.From, opts.To, opts.Filter, opts.RatePerSecond, opts.DryRun)

    events, err := r.dlqStore.Query(ctx, opts.From, opts.To, opts.Filter)
    if err != nil {
        return fmt.Errorf("query DLQ: %w", err)
    }

    log.Infof("found %d events to replay", len(events))

    limiter := rate.NewLimiter(rate.Limit(opts.RatePerSecond), 1)
    var succeeded, failed, skipped int

    for i, event := range events {
        // Bisa dibatalkan kapan saja lewat context
        if err := limiter.Wait(ctx); err != nil {
            log.Infof("replay cancelled at event %d/%d", i, len(events))
            return err
        }

        if opts.DryRun {
            log.Infof("[DRY RUN] would replay event %s", event.ID)
            skipped++
            continue
        }

        if err := r.processor.Process(ctx, event); err != nil {
            log.Errorf("replay failed for event %s: %v", event.ID, err)
            failed++
            continue
        }
        succeeded++

        // Progress log setiap 100 event
        if succeeded%100 == 0 {
            log.Infof("replay progress: %d/%d (success=%d fail=%d skip=%d)",
                i+1, len(events), succeeded, failed, skipped)
        }
    }

    log.Infof("replay complete: total=%d succeeded=%d failed=%d skipped=%d",
        len(events), succeeded, failed, skipped)
    return nil
}

// ANTI-PATTERN: rebuild langsung di tabel production tanpa backup
func rebuildProjection() {
    db.Exec("DELETE FROM order_summary") // bahaya! tidak ada rollback jika gagal
    for _, event := range eventLog.GetAll() {
        applyEvent(event)
    }
}

// BENAR: rebuild ke shadow table, swap atomis setelah selesai
func (p *ProjectionRebuilder) RebuildOrderProjection(ctx context.Context) error {
    shadowTable := fmt.Sprintf("order_summary_rebuild_%d", time.Now().Unix())

    // Buat shadow table terlebih dahulu
    if err := p.db.CreateTableLike(shadowTable, "order_summary"); err != nil {
        return fmt.Errorf("create shadow table: %w", err)
    }

    // Stream semua event dari awal ke shadow table
    events, err := p.eventLog.StreamAll(ctx, "orders", StreamOptions{
        FromOffset: 0,
        BatchSize:  500,
    })
    if err != nil {
        p.db.DropTable(shadowTable) // cleanup
        return fmt.Errorf("stream events: %w", err)
    }

    var count int
    for event := range events {
        if err := p.applyEventToTable(ctx, shadowTable, event); err != nil {
            p.db.DropTable(shadowTable)
            return fmt.Errorf("apply event %d: %w", event.Sequence, err)
        }
        count++
        if count%1000 == 0 {
            log.Infof("rebuilt %d events...", count)
        }
    }

    // Atomic swap: rename shadow → production
    if err := p.db.RenameTable("order_summary", "order_summary_old"); err != nil {
        return err
    }
    if err := p.db.RenameTable(shadowTable, "order_summary"); err != nil {
        return err
    }
    p.db.DropTable("order_summary_old")

    log.Infof("projection rebuild complete: %d events applied", count)
    return nil
}

// ANTI-PATTERN: consumer yang tidak idempotent — replay = bencana
func (c *Consumer) handlePayment(event PaymentRequestedEvent) error {
    // Jika event ini di-replay, payment diproses dua kali → double charge
    return c.paymentGateway.Charge(event.UserID, event.Amount)
}

// BENAR: consumer idempotent — cek event_id sebelum proses
func (c *Consumer) handlePayment(ctx context.Context, event PaymentRequestedEvent) error {
    processed, err := c.processedEvents.Exists(ctx, event.EventID)
    if err != nil {
        return fmt.Errorf("check processed: %w", err)
    }
    if processed {
        log.Infof("event %s already processed, skipping", event.EventID)
        return nil // aman di-skip saat replay
    }

    // Proses dalam transaction: charge + tandai sebagai processed — atomic
    return c.db.Transaction(func(tx *gorm.DB) error {
        if err := c.paymentGateway.Charge(event.UserID, event.Amount); err != nil {
            return err
        }
        return tx.Create(&ProcessedEvent{
            EventID:     event.EventID,
            ProcessedAt: time.Now(),
        }).Error
    })
}

// ANTI-PATTERN: publish tanpa partition key — urutan tidak terjamin lintas partition
func (p *Producer) publishOrderEvent(event OrderEvent) error {
    msg := &sarama.ProducerMessage{
        Topic: "order-events",
        // Tanpa key: Kafka distribusi ke partition secara round-robin
        // OrderCreated (partition 0) bisa diproses setelah OrderShipped (partition 1)
        Value: sarama.ByteEncoder(mustMarshal(event)),
    }
    _, _, err := p.client.SendMessage(msg)
    return err
}

// BENAR: gunakan entity ID sebagai partition key
// Semua event untuk order yang sama dijamin masuk partition yang sama
func (p *Producer) publishOrderEvent(event OrderEvent) error {
    msg := &sarama.ProducerMessage{
        Topic: "order-events",
        Key:   sarama.StringEncoder(event.OrderID), // ← kunci ordering
        Value: sarama.ByteEncoder(mustMarshal(event)),
    }
    _, _, err := p.client.SendMessage(msg)
    return err
}

// ANTI-PATTERN: side effect langsung di handler tanpa cek state
func (h *OrderHandler) handleOrderConfirmed(event OrderConfirmedEvent) error {
    h.updateDatabase(event)
    h.emailService.SendConfirmation(event.UserEmail) // dikirim ulang setiap replay!
    h.webhookService.Notify(event)                   // dipanggil ulang setiap replay!
    return nil
}

// BENAR: cek state dulu — side effect hanya dipanggil jika state belum berubah
func (h *OrderHandler) handleOrderConfirmed(ctx context.Context, event OrderConfirmedEvent) error {
    order, err := h.orderRepo.FindByID(event.OrderID)
    if err != nil {
        return err
    }

    if order.Status == "CONFIRMED" {
        // State sudah benar — semua side effect sudah dijalankan sebelumnya
        // Replay aman: tidak ada email duplikat, tidak ada webhook duplikat
        return nil
    }

    // Update state terlebih dahulu
    if err := h.orderRepo.UpdateStatus(event.OrderID, "CONFIRMED"); err != nil {
        return err
    }

    // Side effects hanya dieksekusi sekali karena state sudah berubah
    h.emailService.SendConfirmation(event.UserEmail)
    h.webhookService.Notify(event)
    return nil
}

// ANTI-PATTERN: replay semua DLQ sekaligus — flood sistem
func replayAll() {
    events := dlq.GetAll() // bisa jutaan event
    for _, e := range events {
        process(e) // makan semua resource downstream sekaligus
    }
}

// BENAR: throttle + dry run + kill switch lewat context
type ReplayConfig struct {
    RatePerSecond float64       // berapa event per detik diproses
    BatchSize     int           // berapa event per batch
    DryRun        bool          // simulasi tanpa eksekusi nyata
    From          time.Time     // filter dari waktu tertentu
    To            time.Time     // filter sampai waktu tertentu
    Filter        string        // filter berdasarkan event type atau criteria lain
}

func (r *ReplayService) ReplayWithControl(ctx context.Context, cfg ReplayConfig) error {
    if cfg.DryRun {
        log.Info("[DRY RUN] no events will actually be processed")
    }
    limiter := rate.NewLimiter(rate.Limit(cfg.RatePerSecond), cfg.BatchSize)

    for _, event := range r.fetchEvents(cfg) {
        // Kill switch: ctx.Cancel() dari luar menghentikan replay kapan saja
        if err := limiter.Wait(ctx); err != nil {
            log.Info("replay stopped by cancellation")
            return err
        }
        if !cfg.DryRun {
            r.processor.Process(ctx, event)
        }
    }
    return nil
}

// BENAR: log dan metrics yang informatif untuk setiap replay operation
type ReplayMetrics struct {
    TotalEvents  int
    Succeeded    int
    Failed       int
    Skipped      int           // idempotency skip
    Duration     time.Duration
    TriggeredBy  string        // siapa yang trigger: "alerting-system", "engineer-unis"
    ReplayReason string        // mengapa: "bug-fix-deploy-v2.3.1", "db-migration"
}

func (r *ReplayService) logCompletion(m ReplayMetrics) {
    log.Infof("[REPLAY COMPLETE] reason=%q triggered_by=%q "+
        "total=%d succeeded=%d failed=%d skipped=%d duration=%v",
        m.ReplayReason, m.TriggeredBy,
        m.TotalEvents, m.Succeeded, m.Failed, m.Skipped, m.Duration)

    // Kirim ke metrics system untuk dashboard dan alerting
    metrics.Gauge("replay.success_rate",
        float64(m.Succeeded)/float64(m.TotalEvents)*100)
    metrics.Histogram("replay.duration_seconds", m.Duration.Seconds())

    // Alert jika failure rate terlalu tinggi
    failureRate := float64(m.Failed) / float64(m.TotalEvents)
    if failureRate > 0.05 { // > 5% gagal
        alerting.Send(fmt.Sprintf("High replay failure rate: %.1f%%", failureRate*100))
    }
}

// ✗ Replay tanpa idempotency — menghasilkan double processing
for _, e := range dlqEvents {
    processPayment(e) // tidak ada pengecekan apakah sudah diproses!
}
// ✓ Selalu cek event_id sebelum memproses ulang

// ✗ DLQ tanpa metadata — susah debugging saat investigasi
dlq.Send(Message{Body: originalMsg.Body})
// ✓ Sertakan failure reason, attempt count, dan timestamp lengkap
dlq.Send(DLQMessage{
    Body:          originalMsg.Body,
    FailureReason: err.Error(),
    AttemptCount:  originalMsg.ReceiveCount,
    FailedAt:      time.Now(),
    StackTrace:    debug.Stack(),
})

// ✗ Replay massal tanpa kill switch — tidak bisa dihentikan darurat
func replayAllDLQ() {
    for _, e := range getAllEvents() {
        process(e) // tidak ada cara menghentikan jika sistem mulai kewalahan
    }
}
// ✓ Selalu gunakan context sebagai kill switch
func replayAllDLQ(ctx context.Context) {
    for _, e := range getAllEvents() {
        if ctx.Err() != nil {
            log.Info("replay cancelled")
            return
        }
        process(ctx, e)
    }
}

// ✗ Event disimpan hanya di in-memory queue — hilang saat restart
queue := make(chan Event, 1000) // tidak persistent
// ✓ Event store yang durable: PostgreSQL, S3, atau Kafka dengan retention panjang

// ✗ Side effect tanpa guard — email dikirim setiap replay
func handleOrderConfirmed(event Event) {
    emailService.Send(event.UserEmail) // kirim setiap kali dipanggil
}
// ✓ Cek state dulu, side effect hanya jika state belum berubah