// Step - один шаг саги: действие и его компенсация.
//
// Обе функции получают транзакцию: шаг обязан записывать свой эффект и отметку
// о себе одной командой COMMIT. Иначе «состояние саги в базе» разъезжается
// с реальностью ровно так же, как отметка в inbox с бизнес-операцией.
type Step struct {
	Name       string
	Execute    func(ctx context.Context, tx *sql.Tx) error
	Compensate func(ctx context.Context, tx *sql.Tx) error
}

// Orchestrator выполняет шаги по порядку, при сбое откатывает выполненные.
type Orchestrator struct {
	db    *sql.DB
	log   *slog.Logger
	steps []Step
}

func New(db *sql.DB, log *slog.Logger, steps ...Step) *Orchestrator {
	return &Orchestrator{db: db, log: log, steps: steps}
}

// Run выполняет сагу с идентификатором sagaID.
//
// Идентификатор нужен не для логов: он ключ идемпотентности. Повторный запуск
// той же саги не должен повторять уже выполненные шаги, а восстановление
// после перезапуска процесса - находить их в базе, а не в памяти.
func (o *Orchestrator) Run(ctx context.Context, sagaID string) error {
	for _, step := range o.steps {
		if err := o.runStep(ctx, sagaID, step); err != nil {
			o.log.ErrorContext(ctx, "saga step failed, compensating",
				slog.String("saga_id", sagaID),
				slog.String("step", step.Name),
				slog.String("err", err.Error()),
			)

			// Компенсации идут отдельной операцией и **возвращают** ошибку.
			// Проглотить её значит оставить сагу в промежуточном состоянии
			// молча - «оплачено, но не выдано» без единого сигнала.
			if cErr := o.Compensate(ctx, sagaID); cErr != nil {
				return fmt.Errorf("шаг %q упал (%w), откат не завершён: %w",
					step.Name, err, cErr)
			}
			return fmt.Errorf("сага %s: шаг %q: %w", sagaID, step.Name, err)
		}
	}
	return nil
}

// runStep выполняет один шаг: действие и отметку о нём одной транзакцией.
func (o *Orchestrator) runStep(ctx context.Context, sagaID string, step Step) error {
	tx, err := o.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	if err := step.Execute(ctx, tx); err != nil {
		return err
	}

	// Отметка в том же COMMIT, что и эффект шага. Порядок здесь не важен -
	// важно, что они в одной транзакции: иначе после сбоя журнал скажет
	// «шаг сделан», а эффекта не будет (или наоборот).
	if err := markStep(ctx, tx, sagaID, step.Name, faultsql.StepDone); err != nil {
		return err
	}
	return tx.Commit()
}

// Compensate откатывает все шаги саги, помеченные выполненными.
//
// Отдельный экспортируемый метод, а не приватная часть Run: после
// перезапуска процесса откат должен продолжиться, и запускать его будет
// внешний восстановитель, у которого нет ни `executed`, ни стека вызовов -
// только sagaID и база.
func (o *Orchestrator) Compensate(ctx context.Context, sagaID string) error {
	done, err := o.doneSteps(ctx, sagaID)
	if err != nil {
		return err
	}

	var failed []string
	// Обратный порядок: шаги откатываются от последнего к первому.
	for i := len(o.steps) - 1; i >= 0; i-- {
		step := o.steps[i]
		if !done[step.Name] {
			continue
		}
		if err := o.compensateStep(ctx, sagaID, step); err != nil {
			// Не прекращаем откат: остальные шаги тоже надо попытаться
			// откатить. Но и не забываем - список уйдёт вызывающему.
			o.log.ErrorContext(ctx, "compensation failed",
				slog.String("saga_id", sagaID),
				slog.String("step", step.Name),
				slog.String("err", err.Error()),
			)
			failed = append(failed, step.Name)
		}
	}

	if len(failed) > 0 {
		return fmt.Errorf("%w: шаги %v требуют вмешательства", ErrCompensationFailed, failed)
	}
	return nil
}

// compensateStep откатывает шаг и обновляет журнал одной транзакцией.
func (o *Orchestrator) compensateStep(ctx context.Context, sagaID string, step Step) error {
	tx, err := o.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	if err := step.Compensate(ctx, tx); err != nil {
		// Пометку о неудаче пишем **отдельной** транзакцией: та, в которой
		// упала компенсация, откатится и унесёт пометку с собой.
		o.markCompensationFailed(ctx, sagaID, step.Name)
		return err
	}

	if err := markStep(ctx, tx, sagaID, step.Name, faultsql.StepCompensated); err != nil {
		return err
	}
	return tx.Commit()
}

// markCompensationFailed фиксирует, что откат шага не удался.
//
// Ошибку записи здесь только логируем: если и она не прошла, сага уже
// в состоянии, требующем человека, а вернуть мы всё равно должны исходную
// ошибку компенсации - она информативнее.
func (o *Orchestrator) markCompensationFailed(ctx context.Context, sagaID, step string) {
	tx, err := o.db.BeginTx(ctx, nil)
	if err != nil {
		o.log.ErrorContext(ctx, "cannot record compensation failure",
			slog.String("saga_id", sagaID), slog.String("step", step),
			slog.String("err", err.Error()))
		return
	}
	defer tx.Rollback()

	if err := markStep(ctx, tx, sagaID, step, faultsql.StepCompensationFailed); err != nil {
		o.log.ErrorContext(ctx, "cannot record compensation failure",
			slog.String("saga_id", sagaID), slog.String("step", step),
			slog.String("err", err.Error()))
		return
	}
	if err := tx.Commit(); err != nil {
		o.log.ErrorContext(ctx, "cannot commit compensation failure",
			slog.String("saga_id", sagaID), slog.String("step", step),
			slog.String("err", err.Error()))
	}
}

// doneSteps читает из базы шаги, которые выполнены и ещё не откачены.
func (o *Orchestrator) doneSteps(ctx context.Context, sagaID string) (map[string]bool, error) {
	rows, err := o.db.QueryContext(ctx,
		`SELECT step FROM saga_steps WHERE saga_id = $1 AND state = $2`,
		sagaID, string(faultsql.StepDone),
	)
	if err != nil {
		return nil, fmt.Errorf("query saga steps: %w", err)
	}
	defer rows.Close()

	done := map[string]bool{}
	for rows.Next() {
		var step string
		if err := rows.Scan(&step); err != nil {
			return nil, fmt.Errorf("scan step: %w", err)
		}
		done[step] = true
	}
	return done, rows.Err()
}

// markStep пишет состояние шага. UPSERT, потому что шаг может быть записан
// повторно: retry, восстановление после перезапуска, повторная компенсация.
func markStep(ctx context.Context, tx *sql.Tx, sagaID, step string, st faultsql.StepState) error {
	_, err := tx.ExecContext(ctx,
		`INSERT INTO saga_steps (saga_id, step, state) VALUES ($1, $2, $3)
		 ON CONFLICT (saga_id, step) DO UPDATE SET state = excluded.state`,
		sagaID, step, string(st),
	)
	if err != nil {
		return fmt.Errorf("mark step %q as %s: %w", step, st, err)
	}
	return nil
}

<?php
// src/Application/Saga/SagaStep.php
declare(strict_types=1);

namespace App\Application\Saga;

use Closure;
use Throwable;

// Один шаг саги: действие + компенсация. Immutable - readonly properties.
final readonly class SagaStep
{
    /**
     * @param callable():void $execute
     * @param callable():void $compensate
     */
    public function __construct(
        public string $name,
        public Closure $execute,
        public Closure $compensate,
    ) {}
}

// src/Application/Saga/SagaOrchestrator.php
final class SagaOrchestrator
{
    public function __construct(
        private readonly LoggerInterface $logger,
    ) {}

    /** @param list<SagaStep> $steps */
    public function run(array $steps): void
    {
        /** @var list<SagaStep> $executed */
        $executed = [];

        foreach ($steps as $step) {
            $this->logger->info('saga step executing', ['step' => $step->name]);

            try {
                ($step->execute)();
            } catch (Throwable $e) {
                $this->logger->error('saga step failed, starting compensation', [
                    'step' => $step->name,
                    'err' => $e->getMessage(),
                ]);
                $this->compensate($executed);
                throw new SagaFailedException(sprintf('saga failed at step %s', $step->name), 0, $e);
            }

            $executed[] = $step;
        }

        $this->logger->info('saga completed successfully', ['steps' => count($steps)]);
    }

    /** @param list<SagaStep> $executed */
    private function compensate(array $executed): void
    {
        foreach (array_reverse($executed) as $step) {
            $this->logger->info('compensating step', ['step' => $step->name]);
            try {
                ($step->compensate)();
            } catch (Throwable $e) {
                $this->logger->error('compensation failed', [
                    'step' => $step->name,
                    'err' => $e->getMessage(),
                ]);
            }
        }
    }
}

// CreateOrderSaga оформляет заказ: резервирует товар, списывает оплату, выдаёт доступ.
func CreateOrderSaga(
    orderSvc OrderService,
    paymentSvc PaymentService,
    accessSvc AccessService,
    req CreateOrderRequest,
) *SagaOrchestrator {
    var orderID int64

    return NewSaga(
        SagaStep{
            Name: "create_order",
            Execute: func(ctx context.Context) error {
                id, err := orderSvc.Create(ctx, req.UserID, req.CourseID)
                if err != nil {
                    return err
                }
                orderID = id
                return nil
            },
            Compensate: func(ctx context.Context) error {
                return orderSvc.Cancel(ctx, orderID)
            },
        },
        SagaStep{
            Name: "charge_payment",
            Execute: func(ctx context.Context) error {
                return paymentSvc.Charge(ctx, req.UserID, req.Amount, orderID)
            },
            Compensate: func(ctx context.Context) error {
                return paymentSvc.Refund(ctx, orderID)
            },
        },
        SagaStep{
            Name: "grant_access",
            Execute: func(ctx context.Context) error {
                return accessSvc.GrantCourseAccess(ctx, req.UserID, req.CourseID)
            },
            Compensate: func(ctx context.Context) error {
                return accessSvc.RevokeCourseAccess(ctx, req.UserID, req.CourseID)
            },
        },
    )
}

// Использование:
// saga := CreateOrderSaga(orderSvc, paymentSvc, accessSvc, req)
// if err := saga.Run(ctx); err != nil {
//     // Saga откатилась, все компенсации выполнены
//     return fmt.Errorf("order creation failed: %w", err)
// }

<?php
// src/Application/Saga/CreateOrderSaga.php
declare(strict_types=1);

namespace App\Application\Saga;

use App\Application\Port\AccessService;
use App\Application\Port\OrderService;
use App\Application\Port\PaymentService;

final class CreateOrderSaga
{
    public function __construct(
        private readonly OrderService $orders,
        private readonly PaymentService $payments,
        private readonly AccessService $access,
        private readonly SagaOrchestrator $orchestrator,
    ) {}

    public function run(CreateOrderRequest $req): void
    {
        // Захватываем orderId через by-reference замыкания
        $orderId = null;

        $this->orchestrator->run([
            new SagaStep(
                name: 'create_order',
                execute: function () use ($req, &$orderId): void {
                    $orderId = $this->orders->create($req->userId, $req->courseId);
                },
                compensate: function () use (&$orderId): void {
                    if ($orderId !== null) {
                        $this->orders->cancel($orderId);
                    }
                },
            ),
            new SagaStep(
                name: 'charge_payment',
                execute: function () use ($req, &$orderId): void {
                    $this->payments->charge($req->userId, $req->amount, $orderId);
                },
                compensate: function () use (&$orderId): void {
                    $this->payments->refund($orderId);
                },
            ),
            new SagaStep(
                name: 'grant_access',
                execute: function () use ($req): void {
                    $this->access->grantCourseAccess($req->userId, $req->courseId);
                },
                compensate: function () use ($req): void {
                    $this->access->revokeCourseAccess($req->userId, $req->courseId);
                },
            ),
        ]);
    }
}