// CompleteLessonUseCase завершает урок и записывает событие в outbox.
type CompleteLessonUseCase struct {
    db *sql.DB
}

// Execute выполняет завершение урока атомарно с outbox-записью.
func (uc *CompleteLessonUseCase) Execute(ctx context.Context, userID, lessonID int64) error {
    tx, err := uc.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    // 1. Бизнес-логика: обновляем прогресс
    _, err = tx.ExecContext(ctx,
        `UPDATE lesson_progress
         SET completed = true, completed_at = now()
         WHERE user_id = $1 AND lesson_id = $2`,
        userID, lessonID,
    )
    if err != nil {
        return fmt.Errorf("update progress: %w", err)
    }

    // 2. Записываем событие в outbox (в той же транзакции!)
    eventID := ulid.Make().String()
    payload, _ := json.Marshal(LessonCompletedPayload{
        UserID:   userID,
        LessonID: lessonID,
    })

    _, err = tx.ExecContext(ctx,
        `INSERT INTO outbox (event_id, event_type, payload)
         VALUES ($1, $2, $3)`,
        eventID, "lesson.completed", payload,
    )
    if err != nil {
        return fmt.Errorf("insert outbox: %w", err)
    }

    // Коммит: и прогресс, и событие сохранены атомарно
    return tx.Commit()
}

<?php
// src/Application/UseCase/CompleteLessonUseCase.php
declare(strict_types=1);

namespace App\Application\UseCase;

use Doctrine\DBAL\Connection;
use Symfony\Component\Uid\Uuid;

final class CompleteLessonUseCase
{
    public function __construct(
        private readonly Connection $connection,
    ) {}

    public function execute(int $userId, int $lessonId): void
    {
        $this->connection->transactional(function (Connection $tx) use ($userId, $lessonId): void {
            // 1. Бизнес-логика: обновляем прогресс
            $tx->executeStatement(
                'UPDATE lesson_progress SET completed = true, completed_at = now()
                 WHERE user_id = :user_id AND lesson_id = :lesson_id',
                ['user_id' => $userId, 'lesson_id' => $lessonId],
            );

            // 2. Записываем событие в outbox (в той же транзакции!)
            $tx->insert('outbox', [
                'event_id' => Uuid::v4()->toRfc4122(),
                'event_type' => 'lesson.completed',
                'payload' => json_encode([
                    'user_id' => $userId,
                    'lesson_id' => $lessonId,
                ], JSON_THROW_ON_ERROR),
            ]);
        });
        // Коммит автоматический: и прогресс, и событие сохранены атомарно
    }
}

// Poller досылает события из outbox в брокер.
type Poller struct {
	db        *sql.DB
	publisher Publisher
	batchSize int
}

func NewPoller(db *sql.DB, publisher Publisher, batchSize int) *Poller {
	return &Poller{db: db, publisher: publisher, batchSize: batchSize}
}

// outboxEvent - строка, вычитанная из outbox.
type outboxEvent struct {
	id        int64
	eventID   string
	eventType string
	payload   []byte
}

// PollBatch публикует одну пачку событий.
//
// Всё - в одной транзакции, и это не небрежность. `FOR UPDATE SKIP LOCKED`
// держит строки занятыми **до конца транзакции**: закоммить её сразу после
// выборки, и второй экземпляр поллера тут же увидит те же события. Тогда
// блокировка защищала бы ровно на время чтения, то есть ни от чего.
//
// Порядок шагов внутри:
//
//  1. пачка целиком считывается и `rows` закрывается, и только потом идут
//     публикации. Запрос к базе внутри цикла `rows.Next()` занял бы второе
//     соединение (первое держит открытый набор строк), а на пуле из одного
//     это тупик: процесс просто встанет, без ошибки;
//  2. публикация в брокер - внешний эффект, транзакция его не откатит.
//     Поэтому упасть между `Publish` и отметкой значит отправить событие
//     второй раз на следующем круге. Это at-least-once, и именно поэтому
//     подписчик обязан быть идемпотентным - см. inbox.
func (p *Poller) PollBatch(ctx context.Context) error {
	tx, err := p.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	defer tx.Rollback()

	events, err := claimBatch(ctx, tx, p.batchSize)
	if err != nil {
		return err
	}
	if len(events) == 0 {
		return tx.Commit()
	}

	for _, e := range events {
		if err := p.publisher.Publish(ctx, e.eventType, e.payload); err != nil {
			// Публикация не удалась: считаем попытку и идём дальше. Событие
			// останется неопубликованным и попадёт в следующую пачку.
			if _, uErr := tx.ExecContext(ctx,
				`UPDATE outbox SET attempts = attempts + 1 WHERE id = $1`, e.id,
			); uErr != nil {
				return fmt.Errorf("bump attempts for %s: %w", e.eventID, uErr)
			}
			continue
		}

		if _, err := tx.ExecContext(ctx,
			`UPDATE outbox SET published_at = now() WHERE id = $1`, e.id,
		); err != nil {
			return fmt.Errorf("mark published %s: %w", e.eventID, err)
		}
	}

	return tx.Commit()
}

// claimBatch забирает пачку неопубликованных событий, блокируя строки
// от других экземпляров поллера. Транзакцию не коммитит - её жизнью
// управляет вызывающий, и от этого зависит смысл блокировки.
func claimBatch(ctx context.Context, tx *sql.Tx, batchSize int) ([]outboxEvent, error) {
	rows, err := tx.QueryContext(ctx,
		`SELECT id, event_id, event_type, payload
		   FROM outbox
		  WHERE published_at IS NULL
		  ORDER BY created_at
		    FOR UPDATE SKIP LOCKED
		  LIMIT $1`,
		batchSize,
	)
	if err != nil {
		return nil, fmt.Errorf("query outbox: %w", err)
	}
	defer rows.Close()

	var events []outboxEvent
	for rows.Next() {
		var e outboxEvent
		if err := rows.Scan(&e.id, &e.eventID, &e.eventType, &e.payload); err != nil {
			return nil, fmt.Errorf("scan row: %w", err)
		}
		events = append(events, e)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate rows: %w", err)
	}
	return events, nil
}

<?php
// src/Application/Cron/OutboxPollCommand.php
declare(strict_types=1);

namespace App\Application\Cron;

use App\Application\Port\EventPublisherPort;
use Doctrine\DBAL\Connection;
use Psr\Log\LoggerInterface;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Throwable;

// В PHP-FPM долгоживущий цикл не подходит - используй один из двух путей:
// 1) Symfony Messenger worker (`bin/console messenger:consume outbox`) -
//    отдельный процесс с встроенным циклом и graceful shutdown.
// 2) Однопроходная команда из cron каждые N секунд (вариант ниже).
#[AsCommand(name: 'app:outbox-poll')]
final class OutboxPollCommand extends Command
{
    private const BATCH_SIZE = 100;

    public function __construct(
        private readonly Connection $connection,
        private readonly EventPublisherPort $publisher,
        private readonly LoggerInterface $logger,
    ) {
        parent::__construct();
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        // FOR UPDATE SKIP LOCKED - несколько параллельных воркеров не пересекаются
        $rows = $this->connection->fetchAllAssociative(
            'SELECT id, event_id, event_type, payload
             FROM outbox
             WHERE published_at IS NULL
             ORDER BY created_at
             LIMIT :lim
             FOR UPDATE SKIP LOCKED',
            ['lim' => self::BATCH_SIZE],
        );

        foreach ($rows as $row) {
            try {
                $this->publisher->publishRaw($row['event_type'], (string) $row['payload']);
                $this->connection->executeStatement(
                    'UPDATE outbox SET published_at = now() WHERE id = :id',
                    ['id' => $row['id']],
                );
            } catch (Throwable $e) {
                $this->connection->executeStatement(
                    'UPDATE outbox SET attempts = attempts + 1 WHERE id = :id',
                    ['id' => $row['id']],
                );
                $this->logger->error('publish failed', [
                    'event_id' => $row['event_id'],
                    'err' => $e->getMessage(),
                ]);
            }
        }

        return Command::SUCCESS;
    }
}