// 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),
            ]);
        });
        // Коммит автоматический: и прогресс, и событие сохранены атомарно
    }
}

// OutboxPoller читает outbox и публикует события в брокер.
type OutboxPoller struct {
    db        *sql.DB
    publisher EventPublisher
    interval  time.Duration
    batchSize int
}

// EventPublisher - порт для отправки событий (Kafka, NATS, in-memory).
type EventPublisher interface {
    Publish(ctx context.Context, eventType string, payload []byte) error
}

// Run запускает polling в бесконечном цикле. Останавливается по ctx.Done().
func (p *OutboxPoller) Run(ctx context.Context) {
    ticker := time.NewTicker(p.interval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            slog.Info("outbox poller stopped")
            return
        case <-ticker.C:
            if err := p.pollBatch(ctx); err != nil {
                slog.ErrorContext(ctx, "outbox poll failed",
                    slog.String("err", err.Error()),
                )
            }
        }
    }
}

// pollBatch выбирает пачку неотправленных событий и публикует их.
func (p *OutboxPoller) pollBatch(ctx context.Context) error {
    rows, err := p.db.QueryContext(ctx,
        `SELECT id, event_id, event_type, payload
         FROM outbox
         WHERE published_at IS NULL
         ORDER BY created_at
         LIMIT $1`,
        p.batchSize,
    )
    if err != nil {
        return fmt.Errorf("query outbox: %w", err)
    }
    defer rows.Close()

    for rows.Next() {
        var id int64
        var eventID, eventType string
        var payload []byte

        if err := rows.Scan(&id, &eventID, &eventType, &payload); err != nil {
            return fmt.Errorf("scan row: %w", err)
        }

        // Публикуем в брокер
        if err := p.publisher.Publish(ctx, eventType, payload); err != nil {
            // Увеличиваем счётчик попыток, но не останавливаемся
            p.db.ExecContext(ctx,
                `UPDATE outbox SET attempts = attempts + 1 WHERE id = $1`, id)
            slog.ErrorContext(ctx, "publish failed",
                slog.String("event_id", eventID),
                slog.String("err", err.Error()),
            )
            continue
        }

        // Отмечаем как отправленное
        _, err := p.db.ExecContext(ctx,
            `UPDATE outbox SET published_at = now() WHERE id = $1`, id)
        if err != nil {
            slog.ErrorContext(ctx, "mark published failed",
                slog.String("event_id", eventID),
                slog.String("err", err.Error()),
            )
        }
    }
    return rows.Err()
}

<?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;

// В 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;
    }
}

// CleanupPublished удаляет отправленные события старше ttl.
func (p *OutboxPoller) CleanupPublished(ctx context.Context, ttl time.Duration) (int64, error) {
    res, err := p.db.ExecContext(ctx,
        `DELETE FROM outbox
         WHERE published_at IS NOT NULL
         AND published_at < $1`,
        time.Now().Add(-ttl),
    )
    if err != nil {
        return 0, err
    }
    return res.RowsAffected()
}

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

namespace App\Application\Cron;

use Doctrine\DBAL\Connection;
use Symfony\Component\Console\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;

// Запуск из cron раз в сутки: `bin/console app:cleanup-outbox`.
#[AsCommand(name: 'app:cleanup-outbox')]
final class CleanupOutboxCommand extends Command
{
    public function __construct(
        private readonly Connection $connection,
    ) {
        parent::__construct();
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $ttlDays = 7;
        $threshold = (new \DateTimeImmutable())->modify(sprintf('-%d days', $ttlDays));

        $affected = $this->connection->executeStatement(
            'DELETE FROM outbox
             WHERE published_at IS NOT NULL AND published_at < :threshold',
            ['threshold' => $threshold->format('Y-m-d H:i:s')],
        );

        $output->writeln(sprintf('deleted %d outbox rows', $affected));
        return Command::SUCCESS;
    }
}