Go 마이크로서비스 Saga 보상 트랜잭션 설계

package saga

import (
	"context"
	"errors"
	"fmt"
	"time"
)

type Status string

const (
	StatusPending       Status = "PENDING"
	StatusRunning       Status = "RUNNING"
	StatusSucceeded     Status = "SUCCEEDED"
	StatusFailed        Status = "FAILED"
	StatusCompensating  Status = "COMPENSATING"
	StatusCompensated   Status = "COMPENSATED"
	StatusManualAction  Status = "MANUAL_ACTION"
)

type StepStatus string

const (
	StepPending      StepStatus = "PENDING"
	StepDone         StepStatus = "DONE"
	StepFailed       StepStatus = "FAILED"
	StepCompensated  StepStatus = "COMPENSATED"
)

type Step struct {
	Name       string
	Do         func(ctx context.Context, sagaID string) error
	Compensate func(ctx context.Context, sagaID string) error
}

type Store interface {
	CreateSaga(ctx context.Context, sagaID, sagaType string) error
	SetSagaStatus(ctx context.Context, sagaID string, st Status) error
	GetSagaStatus(ctx context.Context, sagaID string) (Status, error)

	InitStep(ctx context.Context, sagaID, stepName string) error
	SetStepStatus(ctx context.Context, sagaID, stepName string, st StepStatus, lastErr string) error
	GetStepStatus(ctx context.Context, sagaID, stepName string) (StepStatus, error)
}

var ErrRetryable = errors.New("retryable")

func isRetryable(err error) bool {
	return errors.Is(err, ErrRetryable)
}

package saga

import (
	"context"
	"fmt"
	"time"
)

type Orchestrator struct {
	Store       Store
	Steps       []Step
	MaxAttempts int
	Backoff     func(attempt int) time.Duration
}

func (o *Orchestrator) Run(ctx context.Context, sagaID, sagaType string) error {
	if o.MaxAttempts <= 0 {
		o.MaxAttempts = 5
	}
	if o.Backoff == nil {
		o.Backoff = func(attempt int) time.Duration {
			return time.Duration(attempt) * 200 * time.Millisecond
		}
	}

	if err := o.Store.CreateSaga(ctx, sagaID, sagaType); err != nil {
		return err
	}
	_ = o.Store.SetSagaStatus(ctx, sagaID, StatusRunning)

	// 정방향 실행
	for _, step := range o.Steps {
		_ = o.Store.InitStep(ctx, sagaID, step.Name)

		st, _ := o.Store.GetStepStatus(ctx, sagaID, step.Name)
		if st == StepDone {
			continue // 재실행 시 멱등
		}

		err := o.runWithRetry(ctx, sagaID, step.Name, step.Do)
		if err != nil {
			_ = o.Store.SetSagaStatus(ctx, sagaID, StatusFailed)
			// 보상 시작
			_ = o.Store.SetSagaStatus(ctx, sagaID, StatusCompensating)
			compErr := o.compensate(ctx, sagaID)
			if compErr != nil {
				_ = o.Store.SetSagaStatus(ctx, sagaID, StatusManualAction)
				return fmt.Errorf("saga failed: %w; compensation failed: %v", err, compErr)
			}
			_ = o.Store.SetSagaStatus(ctx, sagaID, StatusCompensated)
			return fmt.Errorf("saga failed and compensated: %w", err)
		}

		_ = o.Store.SetStepStatus(ctx, sagaID, step.Name, StepDone, "")
	}

	_ = o.Store.SetSagaStatus(ctx, sagaID, StatusSucceeded)
	return nil
}

func (o *Orchestrator) runWithRetry(
	ctx context.Context,
	sagaID, stepName string,
	fn func(context.Context, string) error,
) error {
	var last error
	for attempt := 1; attempt <= o.MaxAttempts; attempt++ {
		err := fn(ctx, sagaID)
		if err == nil {
			return nil
		}
		last = err
		_ = o.Store.SetStepStatus(ctx, sagaID, stepName, StepFailed, err.Error())

		if !isRetryable(err) {
			return err
		}
		select {
		case <-time.After(o.Backoff(attempt)):
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	return last
}

func (o *Orchestrator) compensate(ctx context.Context, sagaID string) error {
	// 역순 보상
	for i := len(o.Steps) - 1; i >= 0; i-- {
		step := o.Steps[i]
		st, _ := o.Store.GetStepStatus(ctx, sagaID, step.Name)
		if st != StepDone {
			continue // 완료된 것만 보상
		}
		if step.Compensate == nil {
			continue
		}
		err := o.runWithRetry(ctx, sagaID, step.Name+"_COMP", step.Compensate)
		if err != nil {
			return err
		}
		_ = o.Store.SetStepStatus(ctx, sagaID, step.Name, StepCompensated, "")
	}
	return nil
}

package payment

import (
	"context"
	"database/sql"
	"errors"
)

type Service struct {
	DB *sql.DB
}

// idempotencyKey 예: sagaID + ":PAYMENT_AUTHORIZE"
func (s *Service) Authorize(ctx context.Context, idempotencyKey, orderID string, amount int64) error {
	tx, err := s.DB.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	// 1) 이미 처리했는지 확인
	var exists int
	err = tx.QueryRowContext(ctx,
		"select 1 from payment_idempotency where key = $1",
		idempotencyKey,
	).Scan(&exists)
	if err == nil {
		// 이미 승인 처리됨: 멱등 성공
		return tx.Commit()
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return err
	}

	// 2) 결제 승인 처리(외부 PG 호출은 보통 트랜잭션 밖에서 해야 함)
	// 여기서는 예시로 DB 처리만 표현
	_, err = tx.ExecContext(ctx,
		"insert into payments(order_id, amount, status) values ($1,$2,'AUTHORIZED')",
		orderID, amount,
	)
	if err != nil {
		return err
	}

	// 3) 멱등 키 기록
	_, err = tx.ExecContext(ctx,
		"insert into payment_idempotency(key, created_at) values ($1, now())",
		idempotencyKey,
	)
	if err != nil {
		return err
	}

	return tx.Commit()
}

package outbox

import (
	"context"
	"database/sql"
	"time"
)

type Publisher interface {
	Publish(ctx context.Context, topic string, key string, value []byte) error
}

type Worker struct {
	DB        *sql.DB
	Publisher Publisher
}

func (w *Worker) Run(ctx context.Context) error {
	ticker := time.NewTicker(500 * time.Millisecond)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
			if err := w.flushOnce(ctx); err != nil {
				// 로깅 후 계속 (재시도는 다음 tick에서)
			}
		}
	}
}

func (w *Worker) flushOnce(ctx context.Context) error {
	tx, err := w.DB.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	rows, err := tx.QueryContext(ctx,
		"select id, topic, key, payload from outbox where published_at is null order by id limit 50 for update skip locked",
	)
	if err != nil {
		return err
	}
	defer rows.Close()

	type msg struct {
		id      int64
		topic   string
		key     string
		payload []byte
	}
	msgs := make([]msg, 0, 50)

	for rows.Next() {
		var m msg
		if err := rows.Scan(&m.id, &m.topic, &m.key, &m.payload); err != nil {
			return err
		}
		msgs = append(msgs, m)
	}

	for _, m := range msgs {
		if err := w.Publisher.Publish(ctx, m.topic, m.key, m.payload); err != nil {
			return err
		}
		_, err := tx.ExecContext(ctx,
			"update outbox set published_at = now() where id = $1",
			m.id,
		)
		if err != nil {
			return err
		}
	}

	return tx.Commit()
}