Hướng Dẫn Dapr Workflow Go: Orchestrated Saga Pattern

go get github.com/dapr/go-sdk@v1.11.0

package main

import (
    "context"
    "log"

    dapr "github.com/dapr/go-sdk/client"
    "github.com/dapr/go-sdk/workflow"
)

func main() {
    // Khởi tạo Dapr workflow worker
    w, err := workflow.NewWorker()
    if err != nil {
        log.Fatalf("failed to create workflow worker: %v", err)
    }

    // Đăng ký orchestrator và tất cả các hàm activity
    if err := w.RegisterWorkflow(FundTransferWorkflow); err != nil {
        log.Fatalf("failed to register workflow: %v", err)
    }
    if err := w.RegisterActivity(DebitSourceAccount); err != nil {
        log.Fatalf("failed to register activity: %v", err)
    }
    if err := w.RegisterActivity(CreditTargetAccount); err != nil {
        log.Fatalf("failed to register activity: %v", err)
    }
    if err := w.RegisterActivity(RecordLedgerEntry); err != nil {
        log.Fatalf("failed to register activity: %v", err)
    }
    if err := w.RegisterActivity(CompensateDebitSourceAccount); err != nil {
        log.Fatalf("failed to register activity: %v", err)
    }

    // Khởi động worker (kết nối với Dapr sidecar)
    if err := w.Start(); err != nil {
        log.Fatalf("failed to start workflow worker: %v", err)
    }
    defer w.Shutdown()

    // ... khởi động server HTTP/gRPC của bạn
}

// FundTransferInput mang các tham số đầu vào của saga
type FundTransferInput struct {
    TransactionID string  `json:"transaction_id"`
    SourceAccount string  `json:"source_account"`
    TargetAccount string  `json:"target_account"`
    Amount        float64 `json:"amount"`
    Currency      string  `json:"currency"`
}

// FundTransferWorkflow là Saga orchestrator.
// Nó BẮT BUỘC phải mang tính tất định (deterministic): không dùng time.Now(), không dùng rand, không dùng env vars.
func FundTransferWorkflow(ctx *workflow.WorkflowContext) (any, error) {
    var input FundTransferInput
    if err := ctx.GetInput(&input); err != nil {
        return nil, fmt.Errorf("failed to get workflow input: %w", err)
    }

    // --- Bước 1: Trừ tiền tài khoản nguồn (Debit) ---
    var debitResult DebitResult
    if err := ctx.CallActivity(DebitSourceAccount, workflow.ActivityInput(input)).Await(&debitResult); err != nil {
        // Trừ tiền thất bại — không cần bù trừ (chưa có gì bị trừ)
        return nil, fmt.Errorf("debit failed: %w", err)
    }

    // --- Bước 2: Cộng tiền tài khoản đích (Credit) ---
    var creditResult CreditResult
    if err := ctx.CallActivity(CreditTargetAccount, workflow.ActivityInput(input)).Await(&creditResult); err != nil {
        // Cộng tiền thất bại — phải bù trừ cho bước trừ tiền (debit)
        ctx.GetLogger().Warn("CreditTargetAccount failed, compensating debit", "error", err)
        
        compensateInput := CompensateInput{
            TransactionID: input.TransactionID,
            Account:       input.SourceAccount,
            Amount:        input.Amount,
            Reason:        fmt.Sprintf("credit_failed: %v", err),
        }
        var compensateResult CompensateResult
        if compErr := ctx.CallActivity(CompensateDebitSourceAccount, 
            workflow.ActivityInput(compensateInput)).Await(&compensateResult); compErr != nil {
            // Bản thân việc bù trừ bị lỗi — đây là tình huống báo động nguy cấp
            return nil, fmt.Errorf("CRITICAL: compensation failed after credit failure: debit_err=%v, comp_err=%w", err, compErr)
        }
        return nil, fmt.Errorf("transfer failed and compensated: %w", err)
    }

    // --- Bước 3: Ghi nhận bản ghi sổ cái (Nhật ký kiểm toán / Audit trail) ---
    ledgerInput := LedgerInput{
        TransactionID: input.TransactionID,
        DebitResult:   debitResult,
        CreditResult:  creditResult,
        Timestamp:     ctx.CurrentUTCDateTime(), // thời gian tất định (deterministic time)
    }
    var ledgerResult LedgerResult
    if err := ctx.CallActivity(RecordLedgerEntry, workflow.ActivityInput(ledgerInput)).Await(&ledgerResult); err != nil {
        // Việc ghi sổ cái thất bại — đây là một vấn đề về tính nhất quán nhưng tiền đã di chuyển chính xác.
        // Kích hoạt một cảnh báo đối soát (reconciliation alert) thay vì bù trừ toàn bộ giao dịch chuyển tiền.
        ctx.GetLogger().Error("LedgerEntry failed after successful transfer — reconciliation required",
            "transaction_id", input.TransactionID,
            "error", err)
        // Trả về thành công một phần (partial success) để biểu thị giao dịch đã hoàn thành nhưng nhật ký kiểm toán cần được sửa chữa
    }

    return &FundTransferResult{
        TransactionID:  input.TransactionID,
        Status:         "completed",
        LedgerRecorded: err == nil,
    }, nil
}

// DebitSourceAccount trừ tiền tài khoản nguồn một cách nguyên tử (atomically).
// Hàm này BẮT BUỘC phải là idempotent (lũy đẳng): nếu được gọi hai lần với cùng một TransactionID, 
// lần gọi thứ hai phải là một no-op (chứ không phải trừ tiền hai lần).
func DebitSourceAccount(ctx context.Context, input FundTransferInput) (DebitResult, error) {
    db := dbFromContext(ctx) // retrieve GORM *gorm.DB from context

    var result DebitResult
    err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        // Kiểm tra tính lũy đẳng (Idempotency check): giao dịch này đã được xử lý chưa?
        var existing LedgerTransaction
        if err := tx.Where("external_id = ? AND type = ?", 
            input.TransactionID, "debit").First(&existing).Error; err == nil {
            // Đã được xử lý — trả về kết quả đã lưu trữ (cached)
            result = DebitResult{
                DebitID:   existing.ID,
                NewBalance: existing.PostBalance,
                Idempotent: true,
            }
            return nil
        } else if !errors.Is(err, gorm.ErrRecordNotFound) {
            return fmt.Errorf("idempotency check failed: %w", err)
        }

        // Khóa hàng tài khoản và kiểm tra số dư
        var account Account
        if err := tx.Set("gorm:query_option", "FOR UPDATE").
            Where("account_number = ?", input.SourceAccount).
            First(&account).Error; err != nil {
            return fmt.Errorf("account not found: %w", err)
        }
        if account.Balance < input.Amount {
            return ErrInsufficientFunds
        }

        // Trừ tiền
        newBalance := account.Balance - input.Amount
        if err := tx.Model(&account).Update("balance", newBalance).Error; err != nil {
            return fmt.Errorf("balance update failed: %w", err)
        }

        // Ghi lại bản ghi sổ cái (ledger entry)
        ledger := LedgerTransaction{
            ExternalID:  input.TransactionID,
            Type:        "debit",
            AccountNo:   input.SourceAccount,
            Amount:      input.Amount,
            PostBalance: newBalance,
            CreatedAt:   time.Now(),
        }
        if err := tx.Create(&ledger).Error; err != nil {
            return fmt.Errorf("ledger insert failed: %w", err)
        }

        result = DebitResult{DebitID: ledger.ID, NewBalance: newBalance}
        return nil
    })

    return result, err
}

func CompensateDebitSourceAccount(ctx context.Context, input CompensateInput) (CompensateResult, error) {
    db := dbFromContext(ctx)

    var result CompensateResult
    err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
        // Kiểm tra tính lũy đẳng cho chính quy trình bù trừ này
        var existing LedgerTransaction
        compensateID := input.TransactionID + ":compensate"
        if err := tx.Where("external_id = ? AND type = ?", 
            compensateID, "compensation").First(&existing).Error; err == nil {
            result = CompensateResult{CompensationID: existing.ID, Idempotent: true}
            return nil
        }

        // Tìm bản ghi trừ tiền (debit) ban đầu
        var originalDebit LedgerTransaction
        if err := tx.Where("external_id = ? AND type = ?", 
            input.TransactionID, "debit").First(&originalDebit).Error; err != nil {
            return fmt.Errorf("original debit not found for compensation: %w", err)
        }

        // Đảo ngược lại số dư (balance) đã thay đổi
        var account Account
        if err := tx.Set("gorm:query_option", "FOR UPDATE").
            Where("account_number = ?", input.Account).
            First(&account).Error; err != nil {
            return fmt.Errorf("account not found during compensation: %w", err)
        }

        newBalance := account.Balance + input.Amount
        if err := tx.Model(&account).Update("balance", newBalance).Error; err != nil {
            return fmt.Errorf("compensation balance update failed: %w", err)
        }

        // Ghi lại giao dịch bù trừ thành một bản ghi sổ cái tách biệt
        compensation := LedgerTransaction{
            ExternalID:       compensateID,
            Type:             "compensation",
            AccountNo:        input.Account,
            Amount:           input.Amount,
            PostBalance:      newBalance,
            LinkedExternalID: input.TransactionID,
            Reason:           input.Reason,
            CreatedAt:        time.Now(),
        }
        if err := tx.Create(&compensation).Error; err != nil {
            return fmt.Errorf("compensation ledger insert failed: %w", err)
        }

        result = CompensateResult{CompensationID: compensation.ID}
        return nil
    })
    return result, err
}

// Truy vấn trạng thái workflow từ bất kỳ service nào bằng Dapr client
func GetTransactionStatus(ctx context.Context, transactionID string) (*WorkflowStatus, error) {
    client, err := dapr.NewClient()
    if err != nil {
        return nil, err
    }
    defer client.Close()

    resp, err := client.GetWorkflowBeta1(ctx, &dapr.GetWorkflowRequest{
        InstanceID:        transactionID,
        WorkflowComponent: "dapr",
    })
    if err != nil {
        return nil, fmt.Errorf("failed to get workflow state: %w", err)
    }

    return &WorkflowStatus{
        InstanceID:     resp.InstanceID,
        RuntimeStatus:  resp.RuntimeStatus,    // RUNNING, COMPLETED, FAILED
        CreatedAt:      resp.CreatedAt,
        LastUpdated:    resp.LastUpdatedAt,
        FailureDetails: resp.FailureDetails,
    }, nil
}

// Yêu cầu: github.com/dapr/go-sdk v1.9+ (workflow.ActivityOptions.RetryPolicy được thêm vào ở v1.9)
// import "github.com/dapr/go-sdk/workflow"

// Thiết lập timeout 30 giây cho mỗi lệnh gọi activity
opts := workflow.ActivityOptions{
    StartToCloseTimeout: 30 * time.Second,
    RetryPolicy: &workflow.RetryPolicy{
        MaxAttempts:        3,
        InitialInterval:    time.Second,
        BackoffCoefficient: 2.0,
        MaxInterval:        10 * time.Second,
    },
}
if err := ctx.CallActivity(CreditTargetAccount, 
    workflow.ActivityInput(input), 
    workflow.WithActivityOptions(opts)).Await(&creditResult); err != nil {
    // Xử lý timeout hoặc hết số lần thử lại (retry exhaustion)
}

// Khởi chạy workflow với tổng thời gian chờ tối đa 5 phút
resp, err := client.StartWorkflowBeta1(ctx, &dapr.StartWorkflowRequest{
    WorkflowComponent: "dapr",
    WorkflowName:      "FundTransferWorkflow",
    InstanceID:        transferRequest.TransactionID,
    Options: map[string]string{
        "workflow-timeout": "300s", // 5 phút
    },
    Input: inputBytes,
})

func ReconcilationWorker(ctx context.Context, client dapr.Client) {
    ticker := time.NewTicker(1 * time.Minute)
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            // Liệt kê danh sách workflow đang ở trạng thái FAILED (logic triển khai tùy thuộc vào backend trạng thái của bạn)
            failedInstances := queryFailedWorkflows(ctx)
            for _, instance := range failedInstances {
                routeToDLQ(ctx, instance)
                alertOpsTeam(ctx, instance)
            }
        }
    }
}