Feature Guide

A quick orientation block that answers the essential questions: what this feature does, how it works, why it matters, and the key concepts behind it.

What this does

Semitexa CLI also owns the long-running side of the platform: queues, schedules, webhook delivery, mail workers, and tenant-context execution.

How it works

Dedicated commands expose each stage separately: inspect schedules, plan due runs, start workers, inspect webhook inbox/outbox state, replay deliveries, and run commands in tenant scope.

Why it matters

This makes background systems operable. Operators can inspect, intervene, and replay explicitly instead of treating async infrastructure as invisible magic behind the web server.

Key concepts

queue:work: Runs the async events worker for queued handlers.
scheduler:plan: Materializes due schedule occurrences into concrete run rows before workers execute them.
tenant:run: Executes any command inside a specific tenant context.

Operator Runtime

Long-running platform work is part of the same command surface

Async queues, scheduler pools, outbound webhooks, mail delivery, and tenant-scoped execution all surface through explicit commands. That makes the runtime operable without custom one-off scripts.

Dedicated workers. queue:work, webhook:work, and mail:work turn background processing into explicit operator processes rather than hidden side-effects.

Planner plus executor. The scheduler surface is separated into list, plan, run-now, and work so teams can inspect and intervene before blindly starting daemons.

Context-aware execution. tenant:run lets operators execute commands inside a concrete tenant context instead of manually injecting environment assumptions.

Command	Purpose	Why it matters
`bin/semitexa queue:work nats async`	Run the async events worker against a chosen transport and queue.	Keeps event-driven background work explicit and separately operable.
`bin/semitexa scheduler:list && bin/semitexa scheduler:plan`	Inspect configured schedules, then materialize due runs.	Good operational sequence before starting or debugging the scheduler worker.
`bin/semitexa webhook:show outbox --status=pending && bin/semitexa webhook:work`	Inspect webhook backlog, then run delivery worker.	Makes outbound integration behavior reviewable rather than opaque.
`bin/semitexa tenant:run acme cache:clear --twig`	Run a command inside a tenant context.	Critical when the platform behavior depends on tenant-aware configuration or data isolation.

Bring up the scheduler loop deliberately

bin/semitexa scheduler:list
bin/semitexa scheduler:plan
bin/semitexa scheduler:work default

Inspect and replay webhook traffic

bin/semitexa webhook:show inbox --limit=10
bin/semitexa webhook:replay:inbound <delivery-uuid>

Operate background work in one tenant

bin/semitexa tenant:run acme queue:work
bin/semitexa tenant:run acme cache:clear --twig

worker:processes scheduler:list scheduler:work webhook:work

Verified against Semitexa Ultimate 2026.09.19.1020

Workers & Scheduling

Semitexa is not only request-response code. The CLI also owns the long-running workers and operator interventions that keep the platform moving.

How it works

queue:work, webhook:work, and mail:work are separate explicit processes rather than hidden side-effects of the web runtime. The scheduler surface is split into scheduler:list (inspect), scheduler:plan (materialize), and scheduler:work (run the loop) so teams can inspect state before pushing harder. tenant:run executes any command inside a concrete tenant context, including queue and cache operations.

In a project started with bin/semitexa server:start, these workers are already running: a worker service, in a container of its own so it can be scaled apart from the web server, runs every line bin/semitexa worker:processes prints. Each package names what it needs with #[AsWorkerProcessSource]: core one queue:work per queue its listeners and handlers use, mail mail:work, media media:work, the scheduler scheduler:plan (once a minute) and one scheduler:work for each pool its jobs name (scheduler:work reads one pool; a pool chosen only when a job is dispatched is named by the package that dispatches it, as workflow does for its timeout pools), webhooks webhook:work. It runs whatever EVENTS_ASYNC says: with EVENTS_ASYNC=0 and no EVENTS_TRANSPORT the queue consumers name nothing (the transport is in-memory, one per process), and the scheduler and webhooks still run; EVENTS_TRANSPORT=nats or database gives them their queue:work all the same. WORKER_REPLICAS in .env sets how many worker containers run.

Why this matters

Operability matters as much as functionality. Separate inspect, plan, execute, and replay actions mean the platform can be observed and intervened on deliberately instead of forcing operators to restart processes and hope for the best.

1<?php23declare(strict_types=1);45namespace Semitexa\Core\Application\Console\Command;67use Semitexa\Core\Attribute\AsCommand;8use Symfony\Component\Console\Command\Command;9use Symfony\Component\Console\Input\InputArgument;10use Symfony\Component\Console\Input\InputInterface;11use Symfony\Component\Console\Input\InputOption;12use Symfony\Component\Console\Output\OutputInterface;13use Symfony\Component\Console\Style\SymfonyStyle;1415#[AsCommand(name: 'queue:work', description: 'Run a queue worker (processes queued listeners and handlers enqueued with execution: queued)')]16class QueueWorkCommand extends Command17{18    protected function configure(): void19    {20        $this->setName('queue:work')21            ->setDescription('Run a queue worker (processes queued listeners and handlers enqueued with execution: queued)')22            ->addArgument('transport', InputArgument::OPTIONAL, 'Transport: nats or in-memory (default from EVENTS_ASYNC)', null)23            ->addArgument('queue', InputArgument::OPTIONAL, 'Queue name (default from EVENTS_QUEUE_DEFAULT)', null)24            ->addOption('timeout', 't', InputOption::VALUE_OPTIONAL, 'Worker timeout in seconds', null);25    }2627    protected function execute(InputInterface $input, OutputInterface $output): int28    {29        $io = new SymfonyStyle($input, $output);30        $transport = $input->getArgument('transport');31        $queue = $input->getArgument('queue');3233        $io->title('Events worker (queue)');3435        try {36            $worker = new \Semitexa\Core\Queue\QueueWorker();37            $worker->setOutput($output);38            $worker->run($transport, $queue);39        } catch (\Throwable $e) {40            $io->error('Worker failed: ' . $e->getMessage());41            return Command::FAILURE;42        }4344        return Command::SUCCESS;45    }46}4748

<?php

declare(strict_types=1);

namespace Semitexa\Core\Application\Console\Command;

use Semitexa\Core\Attribute\AsCommand;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'queue:work', description: 'Run a queue worker (processes queued listeners and handlers enqueued with execution: queued)')]
class QueueWorkCommand extends Command
{
    protected function configure(): void
    {
        $this->setName('queue:work')
            ->setDescription('Run a queue worker (processes queued listeners and handlers enqueued with execution: queued)')
            ->addArgument('transport', InputArgument::OPTIONAL, 'Transport: nats or in-memory (default from EVENTS_ASYNC)', null)
            ->addArgument('queue', InputArgument::OPTIONAL, 'Queue name (default from EVENTS_QUEUE_DEFAULT)', null)
            ->addOption('timeout', 't', InputOption::VALUE_OPTIONAL, 'Worker timeout in seconds', null);
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);
        $transport = $input->getArgument('transport');
        $queue = $input->getArgument('queue');

        $io->title('Events worker (queue)');

        try {
            $worker = new \Semitexa\Core\Queue\QueueWorker();
            $worker->setOutput($output);
            $worker->run($transport, $queue);
        } catch (\Throwable $e) {
            $io->error('Worker failed: ' . $e->getMessage());
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Scheduler\Application\Console\Command;67use Semitexa\Core\Attribute\AsCommand;8use Semitexa\Core\Attribute\InjectAsReadonly;9use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;10use Semitexa\Scheduler\Application\Service\CronOccurrenceCalculator;11use Symfony\Component\Console\Command\Command;12use Symfony\Component\Console\Input\InputInterface;13use Symfony\Component\Console\Output\OutputInterface;14use Symfony\Component\Console\Style\SymfonyStyle;1516#[AsCommand(name: 'scheduler:list', description: 'List all enabled schedule definitions and their next due time')]17final class SchedulerListCommand extends Command18{19    #[InjectAsReadonly]20    protected ScheduleDefinitionRepositoryInterface $definitionRepo;2122    protected function configure(): void23    {24        $this->setName('scheduler:list')25             ->setDescription('List all enabled schedule definitions and their next due time');26    }2728    protected function execute(InputInterface $input, OutputInterface $output): int29    {30        $io = new SymfonyStyle($input, $output);31        $io->title('Schedule Definitions');3233        try {34            $calculator = new CronOccurrenceCalculator();3536            $definitions = $this->definitionRepo->findAllEnabled();3738            if ($definitions === []) {39                $io->note('No enabled schedule definitions found.');40                return Command::SUCCESS;41            }4243            $rows = [];44            $now  = new \DateTimeImmutable();4546            foreach ($definitions as $def) {47                try {48                    $next = $calculator->getNextOccurrence($def->getCronExpression(), $now, $def->getTimezone());49                    // The calculator answers in UTC; it was printed under the50                    // row's timezone label, so a 03:00 Kyiv job read 00:00.51                    $nextStr = $def->getTimezone() === 'UTC'52                        ? $next->format('Y-m-d H:i:s') . ' (UTC)'53                        : $next->setTimezone(new \DateTimeZone($def->getTimezone()))->format('Y-m-d H:i:s')54                            . ' (' . $def->getTimezone() . '; '55                            . $next->format($next->format('Y-m-d') === $next->setTimezone(new \DateTimeZone($def->getTimezone()))->format('Y-m-d') ? 'H:i' : 'Y-m-d H:i')56                            . ' UTC)';57                } catch (\Throwable) {58                    $nextStr = '(invalid expression)';59                }6061                $rows[] = [62                    $def->getScheduleKey(),63                    $def->getCronExpression(),64                    $nextStr,65                    $def->getPool(),66                    $def->getOverlapPolicy(),67                    $def->getMisfirePolicy(),68                    $def->getTenantMode(),69                    $def->getPlanningCursorAt() !== null ? $def->getPlanningCursorAt()->format('Y-m-d H:i:s') . ' UTC' : 'never',70                ];71            }7273            $io->table(74                ['Key', 'Cron', 'Next Due', 'Pool', 'Overlap', 'Misfire', 'Tenant Mode', 'Last Planned'],75                $rows,76            );77        } catch (\Throwable $e) {78            $io->error('scheduler:list failed: ' . $e->getMessage());79            return Command::FAILURE;80        }8182        return Command::SUCCESS;83    }84}85

<?php

declare(strict_types=1);

namespace Semitexa\Scheduler\Application\Console\Command;

use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;
use Semitexa\Scheduler\Application\Service\CronOccurrenceCalculator;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'scheduler:list', description: 'List all enabled schedule definitions and their next due time')]
final class SchedulerListCommand extends Command
{
    #[InjectAsReadonly]
    protected ScheduleDefinitionRepositoryInterface $definitionRepo;

    protected function configure(): void
    {
        $this->setName('scheduler:list')
             ->setDescription('List all enabled schedule definitions and their next due time');
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);
        $io->title('Schedule Definitions');

        try {
            $calculator = new CronOccurrenceCalculator();

            $definitions = $this->definitionRepo->findAllEnabled();

            if ($definitions === []) {
                $io->note('No enabled schedule definitions found.');
                return Command::SUCCESS;
            }

            $rows = [];
            $now  = new \DateTimeImmutable();

            foreach ($definitions as $def) {
                try {
                    $next = $calculator->getNextOccurrence($def->getCronExpression(), $now, $def->getTimezone());
                    // The calculator answers in UTC; it was printed under the
                    // row's timezone label, so a 03:00 Kyiv job read 00:00.
                    $nextStr = $def->getTimezone() === 'UTC'
                        ? $next->format('Y-m-d H:i:s') . ' (UTC)'
                        : $next->setTimezone(new \DateTimeZone($def->getTimezone()))->format('Y-m-d H:i:s')
                            . ' (' . $def->getTimezone() . '; '
                            . $next->format($next->format('Y-m-d') === $next->setTimezone(new \DateTimeZone($def->getTimezone()))->format('Y-m-d') ? 'H:i' : 'Y-m-d H:i')
                            . ' UTC)';
                } catch (\Throwable) {
                    $nextStr = '(invalid expression)';
                }

                $rows[] = [
                    $def->getScheduleKey(),
                    $def->getCronExpression(),
                    $nextStr,
                    $def->getPool(),
                    $def->getOverlapPolicy(),
                    $def->getMisfirePolicy(),
                    $def->getTenantMode(),
                    $def->getPlanningCursorAt() !== null ? $def->getPlanningCursorAt()->format('Y-m-d H:i:s') . ' UTC' : 'never',
                ];
            }

            $io->table(
                ['Key', 'Cron', 'Next Due', 'Pool', 'Overlap', 'Misfire', 'Tenant Mode', 'Last Planned'],
                $rows,
            );
        } catch (\Throwable $e) {
            $io->error('scheduler:list failed: ' . $e->getMessage());
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Scheduler\Application\Console\Command;67use Semitexa\Core\Attribute\AsCommand;8use Semitexa\Core\Attribute\InjectAsReadonly;9use Semitexa\Core\Event\EventDispatcherInterface;10use Semitexa\Orm\OrmManager;11use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;12use Semitexa\Scheduler\Domain\Contract\ScheduledRunRepositoryInterface;13use Semitexa\Scheduler\Application\Service\CronOccurrenceCalculator;14use Semitexa\Scheduler\Application\Service\MisfirePolicyResolver;15use Semitexa\Scheduler\Application\Service\SchedulePlanner;16use Semitexa\Scheduler\Application\Service\TenantOccurrenceExpander;17use Semitexa\Scheduler\Application\Service\ScheduleDefinitionRegistry;18use Semitexa\Tenancy\Domain\Contract\TenantRepositoryInterface;19use Symfony\Component\Console\Command\Command;20use Symfony\Component\Console\Input\InputInterface;21use Symfony\Component\Console\Output\OutputInterface;22use Symfony\Component\Console\Style\SymfonyStyle;2324#[AsCommand(name: 'scheduler:plan', description: 'Materialize due recurring schedule occurrences into run rows')]25final class SchedulerPlanCommand extends Command26{27    #[InjectAsReadonly]28    protected ScheduleDefinitionRepositoryInterface $definitionRepo;2930    #[InjectAsReadonly]31    protected ScheduledRunRepositoryInterface $runRepo;3233    /**34     * Resolves to the environment-configured repository35     * (`EnvironmentTenantRepository`); tenancy is a hard dependency of this36     * package, so per-tenant expansion always has a repository to fan out37     * over — the old "could not be resolved" degradation path is gone.38     */39    #[InjectAsReadonly]40    protected TenantRepositoryInterface $tenantRepo;4142    #[InjectAsReadonly]43    protected ScheduleDefinitionRegistry $registry;4445    #[InjectAsReadonly]46    protected EventDispatcherInterface $events;4748    protected function configure(): void49    {50        $this->setName('scheduler:plan')51             ->setDescription('Materialize due recurring schedule occurrences into run rows');52    }5354    protected function execute(InputInterface $input, OutputInterface $output): int55    {56        $io = new SymfonyStyle($input, $output);57        $io->title('Scheduler Planner');5859        try {60            // CLI parity with the Swoole WorkerStart bootstrap (same as61            // scheduler:work): the planner's own ORM writes — definition62            // sync + run materialization — publish their invalidation63            // signals like any other write.64            OrmManager::setDefaultEventDispatcherResolver(65                fn (): EventDispatcherInterface => $this->events,66            );6768            // Sync code-discovered schedules to DB69            foreach ($this->registry->sync() as $problem) {70                $io->warning($problem);71            }7273            $planner = new SchedulePlanner(74                definitionRepository: $this->definitionRepo,75                runRepository:        $this->runRepo,76                calculator:           new CronOccurrenceCalculator(),77                misfireResolver:      new MisfirePolicyResolver(),78                tenantExpander:       new TenantOccurrenceExpander($this->tenantRepo),79            );8081            $now     = new \DateTimeImmutable();82            $planned = $planner->plan($now, $output, $this->registry->plannableKeys());8384            $io->success("Planned {$planned} run(s).");85        } catch (\Throwable $e) {86            $io->error('Scheduler plan failed: ' . $e->getMessage());87            return Command::FAILURE;88        }8990        return Command::SUCCESS;91    }92}93

<?php

declare(strict_types=1);

namespace Semitexa\Scheduler\Application\Console\Command;

use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Core\Event\EventDispatcherInterface;
use Semitexa\Orm\OrmManager;
use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;
use Semitexa\Scheduler\Domain\Contract\ScheduledRunRepositoryInterface;
use Semitexa\Scheduler\Application\Service\CronOccurrenceCalculator;
use Semitexa\Scheduler\Application\Service\MisfirePolicyResolver;
use Semitexa\Scheduler\Application\Service\SchedulePlanner;
use Semitexa\Scheduler\Application\Service\TenantOccurrenceExpander;
use Semitexa\Scheduler\Application\Service\ScheduleDefinitionRegistry;
use Semitexa\Tenancy\Domain\Contract\TenantRepositoryInterface;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'scheduler:plan', description: 'Materialize due recurring schedule occurrences into run rows')]
final class SchedulerPlanCommand extends Command
{
    #[InjectAsReadonly]
    protected ScheduleDefinitionRepositoryInterface $definitionRepo;

    #[InjectAsReadonly]
    protected ScheduledRunRepositoryInterface $runRepo;

    /**
     * Resolves to the environment-configured repository
     * (`EnvironmentTenantRepository`); tenancy is a hard dependency of this
     * package, so per-tenant expansion always has a repository to fan out
     * over — the old "could not be resolved" degradation path is gone.
     */
    #[InjectAsReadonly]
    protected TenantRepositoryInterface $tenantRepo;

    #[InjectAsReadonly]
    protected ScheduleDefinitionRegistry $registry;

    #[InjectAsReadonly]
    protected EventDispatcherInterface $events;

    protected function configure(): void
    {
        $this->setName('scheduler:plan')
             ->setDescription('Materialize due recurring schedule occurrences into run rows');
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);
        $io->title('Scheduler Planner');

        try {
            // CLI parity with the Swoole WorkerStart bootstrap (same as
            // scheduler:work): the planner's own ORM writes — definition
            // sync + run materialization — publish their invalidation
            // signals like any other write.
            OrmManager::setDefaultEventDispatcherResolver(
                fn (): EventDispatcherInterface => $this->events,
            );

            // Sync code-discovered schedules to DB
            foreach ($this->registry->sync() as $problem) {
                $io->warning($problem);
            }

            $planner = new SchedulePlanner(
                definitionRepository: $this->definitionRepo,
                runRepository:        $this->runRepo,
                calculator:           new CronOccurrenceCalculator(),
                misfireResolver:      new MisfirePolicyResolver(),
                tenantExpander:       new TenantOccurrenceExpander($this->tenantRepo),
            );

            $now     = new \DateTimeImmutable();
            $planned = $planner->plan($now, $output, $this->registry->plannableKeys());

            $io->success("Planned {$planned} run(s).");
        } catch (\Throwable $e) {
            $io->error('Scheduler plan failed: ' . $e->getMessage());
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Scheduler\Application\Console\Command;67use Semitexa\Core\Attribute\AsCommand;8use Semitexa\Core\Attribute\InjectAsReadonly;9use Semitexa\Core\Lifecycle\StopSignals;10use Semitexa\Core\Event\EventDispatcherInterface;11use Semitexa\Orm\OrmManager;12use Semitexa\Scheduler\Application\Db\MySQL\Repository\SchedulerRunHistoryRepository;13use Semitexa\Scheduler\Application\Service\OverlapPolicyHandler;14use Semitexa\Scheduler\Application\Service\RetryScheduler;15use Semitexa\Scheduler\Application\Service\RunExecutor;16use Semitexa\Scheduler\Application\Service\RunLeaseManager;17use Semitexa\Scheduler\Application\Service\SchedulerLockManager;18use Semitexa\Scheduler\Application\Service\SchedulerWorker;19use Semitexa\Scheduler\Configuration\SchedulerConfig;20use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;21use Semitexa\Scheduler\Domain\Contract\ScheduledRunRepositoryInterface;22use Semitexa\Scheduler\Domain\Contract\ScheduledRunRetentionInterface;23use Semitexa\Scheduler\Domain\Contract\SchedulerLockRepositoryInterface;24use Symfony\Component\Console\Command\Command;25use Symfony\Component\Console\Input\InputArgument;26use Symfony\Component\Console\Input\InputInterface;27use Symfony\Component\Console\Output\OutputInterface;28use Symfony\Component\Console\Style\SymfonyStyle;2930#[AsCommand(name: 'scheduler:work', description: 'Run the scheduler worker for a given pool')]31final class SchedulerWorkCommand extends Command32{33    #[InjectAsReadonly]34    protected ScheduledRunRepositoryInterface $runRepo;3536    #[InjectAsReadonly]37    protected SchedulerLockRepositoryInterface $lockRepo;3839    #[InjectAsReadonly]40    protected ScheduleDefinitionRepositoryInterface $definitionRepo;4142    #[InjectAsReadonly]43    protected EventDispatcherInterface $events;4445    protected function configure(): void46    {47        $this->setName('scheduler:work')48             ->setDescription('Run the scheduler worker for a given pool')49             ->addArgument(50                 name:        'pool',51                 mode:        InputArgument::OPTIONAL,52                 description: 'Worker pool name (default: from SCHEDULER_DEFAULT_POOL env or "default")',53                 default:     null,54             );55    }5657    protected function execute(InputInterface $input, OutputInterface $output): int58    {59        $io   = new SymfonyStyle($input, $output);60        $pool = $input->getArgument('pool');6162        $io->title('Scheduler Worker');6364        try {65            // CLI parity with the Swoole WorkerStart bootstrap: register the66            // ORM's default dispatcher resolver BEFORE any job runs, so every67            // scheduled job's ORM writes auto-publish their invalidation68            // signals without each job carrying its own CLI bootstrap.69            OrmManager::setDefaultEventDispatcherResolver(70                fn (): EventDispatcherInterface => $this->events,71            );7273            $config      = SchedulerConfig::create();74            $historyRepo = new SchedulerRunHistoryRepository();7576            $leaseManager   = new RunLeaseManager($this->runRepo, $config->leaseTtlSeconds);77            $lockManager    = new SchedulerLockManager($this->lockRepo, $config->lockTtlSeconds);78            $overlapHandler = new OverlapPolicyHandler($this->runRepo, $lockManager, $this->definitionRepo, $historyRepo);79            $executor       = new RunExecutor($this->runRepo, $historyRepo);80            $retryScheduler = new RetryScheduler($this->runRepo, $historyRepo);8182            $worker = new SchedulerWorker(83                leaseManager:    $leaseManager,84                lockManager:     $lockManager,85                runRepository:   $this->runRepo,86                overlapHandler:  $overlapHandler,87                executor:        $executor,88                retryScheduler:  $retryScheduler,89                historyRepository: $historyRepo,90                config:          $config,91                retention:       $this->runRepo instanceof ScheduledRunRetentionInterface ? $this->runRepo : null,92            );93            $worker->setOutput($output);94            // docker stop: the run in hand finishes and is recorded, then the loop ends.95            StopSignals::install(static fn () => $worker->stop(), static fn (string $m) => $output->writeln($m), 'the run in hand');96            if (!$this->runRepo instanceof ScheduledRunRetentionInterface && $config->runRetentionDays > 0) {97                $io->warning('Finished runs are not pruned: the run repository bound here does not implement ScheduledRunRetentionInterface.');98            }99            $worker->run($pool);100        } catch (\Throwable $e) {101            $io->error('Scheduler worker failed: ' . $e->getMessage());102            return Command::FAILURE;103        }104105        return Command::SUCCESS;106    }107}108

<?php

declare(strict_types=1);

namespace Semitexa\Scheduler\Application\Console\Command;

use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Core\Lifecycle\StopSignals;
use Semitexa\Core\Event\EventDispatcherInterface;
use Semitexa\Orm\OrmManager;
use Semitexa\Scheduler\Application\Db\MySQL\Repository\SchedulerRunHistoryRepository;
use Semitexa\Scheduler\Application\Service\OverlapPolicyHandler;
use Semitexa\Scheduler\Application\Service\RetryScheduler;
use Semitexa\Scheduler\Application\Service\RunExecutor;
use Semitexa\Scheduler\Application\Service\RunLeaseManager;
use Semitexa\Scheduler\Application\Service\SchedulerLockManager;
use Semitexa\Scheduler\Application\Service\SchedulerWorker;
use Semitexa\Scheduler\Configuration\SchedulerConfig;
use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;
use Semitexa\Scheduler\Domain\Contract\ScheduledRunRepositoryInterface;
use Semitexa\Scheduler\Domain\Contract\ScheduledRunRetentionInterface;
use Semitexa\Scheduler\Domain\Contract\SchedulerLockRepositoryInterface;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'scheduler:work', description: 'Run the scheduler worker for a given pool')]
final class SchedulerWorkCommand extends Command
{
    #[InjectAsReadonly]
    protected ScheduledRunRepositoryInterface $runRepo;

    #[InjectAsReadonly]
    protected SchedulerLockRepositoryInterface $lockRepo;

    #[InjectAsReadonly]
    protected ScheduleDefinitionRepositoryInterface $definitionRepo;

    #[InjectAsReadonly]
    protected EventDispatcherInterface $events;

    protected function configure(): void
    {
        $this->setName('scheduler:work')
             ->setDescription('Run the scheduler worker for a given pool')
             ->addArgument(
                 name:        'pool',
                 mode:        InputArgument::OPTIONAL,
                 description: 'Worker pool name (default: from SCHEDULER_DEFAULT_POOL env or "default")',
                 default:     null,
             );
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io   = new SymfonyStyle($input, $output);
        $pool = $input->getArgument('pool');

        $io->title('Scheduler Worker');

        try {
            // CLI parity with the Swoole WorkerStart bootstrap: register the
            // ORM's default dispatcher resolver BEFORE any job runs, so every
            // scheduled job's ORM writes auto-publish their invalidation
            // signals without each job carrying its own CLI bootstrap.
            OrmManager::setDefaultEventDispatcherResolver(
                fn (): EventDispatcherInterface => $this->events,
            );

            $config      = SchedulerConfig::create();
            $historyRepo = new SchedulerRunHistoryRepository();

            $leaseManager   = new RunLeaseManager($this->runRepo, $config->leaseTtlSeconds);
            $lockManager    = new SchedulerLockManager($this->lockRepo, $config->lockTtlSeconds);
            $overlapHandler = new OverlapPolicyHandler($this->runRepo, $lockManager, $this->definitionRepo, $historyRepo);
            $executor       = new RunExecutor($this->runRepo, $historyRepo);
            $retryScheduler = new RetryScheduler($this->runRepo, $historyRepo);

            $worker = new SchedulerWorker(
                leaseManager:    $leaseManager,
                lockManager:     $lockManager,
                runRepository:   $this->runRepo,
                overlapHandler:  $overlapHandler,
                executor:        $executor,
                retryScheduler:  $retryScheduler,
                historyRepository: $historyRepo,
                config:          $config,
                retention:       $this->runRepo instanceof ScheduledRunRetentionInterface ? $this->runRepo : null,
            );
            $worker->setOutput($output);
            // docker stop: the run in hand finishes and is recorded, then the loop ends.
            StopSignals::install(static fn () => $worker->stop(), static fn (string $m) => $output->writeln($m), 'the run in hand');
            if (!$this->runRepo instanceof ScheduledRunRetentionInterface && $config->runRetentionDays > 0) {
                $io->warning('Finished runs are not pruned: the run repository bound here does not implement ScheduledRunRetentionInterface.');
            }
            $worker->run($pool);
        } catch (\Throwable $e) {
            $io->error('Scheduler worker failed: ' . $e->getMessage());
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Webhooks\Application\Console\Command;67use Psr\Container\ContainerInterface;8use Semitexa\Core\Attribute\AsCommand;9use Semitexa\Core\Attribute\InjectAsReadonly;10use Semitexa\Webhooks\Domain\Contract\InboundDeliveryRepositoryInterface;11use Semitexa\Webhooks\Domain\Contract\OutboundDeliveryRepositoryInterface;12use Semitexa\Webhooks\Domain\Contract\WebhookAttemptRepositoryInterface;13use Semitexa\Webhooks\Domain\Contract\WebhookEndpointDefinitionRepositoryInterface;14use Symfony\Component\Console\Command\Command;15use Symfony\Component\Console\Input\InputArgument;16use Symfony\Component\Console\Input\InputInterface;17use Symfony\Component\Console\Input\InputOption;18use Symfony\Component\Console\Output\OutputInterface;19use Symfony\Component\Console\Style\SymfonyStyle;2021#[AsCommand(name: 'webhook:show', description: 'Display webhook endpoint, inbox, or outbox details')]22final class WebhookShowCommand extends Command23{24    /**25     * Injected rather than reached for statically. The services below are still26     * resolved lazily inside the command body, and deliberately so: every27     * #[AsCommand] class is instantiated and injected at console boot, so28     * injecting a repository directly would open its connection on every29     * `bin/semitexa` invocation — and a dependency that failed to build would30     * make this command vanish from the list instead of reporting the failure.31     */32    #[InjectAsReadonly]33    protected ContainerInterface $container;3435    protected function configure(): void36    {37        $this38            ->setName('webhook:show')39            ->setDescription('Display webhook endpoint, inbox, or outbox details')40            ->addArgument(41                name: 'type',42                mode: InputArgument::REQUIRED,43                description: 'What to show: endpoints, inbox, outbox, or attempts',44            )45            ->addArgument(46                name: 'id',47                mode: InputArgument::OPTIONAL,48                description: 'Specific ID to show details for',49            )50            ->addOption(51                name: 'status',52                shortcut: 's',53                mode: InputOption::VALUE_OPTIONAL,54                description: 'Filter by status',55            )56            ->addOption(57                name: 'limit',58                shortcut: 'l',59                mode: InputOption::VALUE_OPTIONAL,60                description: 'Max rows to display',61                default: '20',62            );63    }6465    protected function execute(InputInterface $input, OutputInterface $output): int66    {67        $io = new SymfonyStyle($input, $output);68        $type = $input->getArgument('type');69        $id = $input->getArgument('id');7071        try {72            $container = $this->container;7374            return match ($type) {75                'endpoints' => $this->showEndpoints($container, $io, $id),76                'inbox' => $this->showInbox($container, $io, $id, $input),77                'outbox' => $this->showOutbox($container, $io, $id, $input),78                'attempts' => $this->showAttempts($container, $io, $id),79                default => $this->invalidType($io, $type),80            };81        } catch (\Throwable $e) {82            $io->error($e->getMessage());83            return Command::FAILURE;84        }85    }8687    private function showEndpoints(object $container, SymfonyStyle $io, ?string $id): int88    {89        $repo = $container->get(WebhookEndpointDefinitionRepositoryInterface::class);9091        if ($id !== null) {92            $endpoint = $repo->findByEndpointKey($id) ?? $repo->findById($id);93            if ($endpoint === null) {94                $io->error("Endpoint not found: {$id}");95                return Command::FAILURE;96            }97            $io->definitionList(98                ['ID' => $endpoint->getId()],99                ['Endpoint Key' => $endpoint->getEndpointKey()],100                ['Direction' => $endpoint->getDirection()->value],101                ['Provider' => $endpoint->getProviderKey()],102                ['Enabled' => $endpoint->isEnabled() ? 'Yes' : 'No'],103                ['Target URL' => $endpoint->getTargetUrl() ?? '(none)'],104                ['Verification' => $endpoint->getVerificationMode() ?? '(none)'],105                ['Secret Ref' => $endpoint->getSecretRef() !== null ? '***' . substr($endpoint->getSecretRef(), -4) : '(none)'],106                ['Max Attempts' => (string) $endpoint->getMaxAttempts()],107                ['Timeout' => $endpoint->getTimeoutSeconds() . 's'],108            );109            return Command::SUCCESS;110        }111112        $endpoints = $repo->findAll();113        if ($endpoints === []) {114            $io->info('No webhook endpoints defined.');115            return Command::SUCCESS;116        }117118        $rows = [];119        foreach ($endpoints as $ep) {120            $rows[] = [121                $ep->getEndpointKey(),122                $ep->getDirection()->value,123                $ep->getProviderKey(),124                $ep->isEnabled() ? 'Yes' : 'No',125                $ep->getTargetUrl() ?? '-',126            ];127        }128129        $io->table(['Endpoint Key', 'Direction', 'Provider', 'Enabled', 'Target URL'], $rows);130        return Command::SUCCESS;131    }132133    private function showInbox(object $container, SymfonyStyle $io, ?string $id, InputInterface $input): int134    {135        $repo = $container->get(InboundDeliveryRepositoryInterface::class);136137        if ($id !== null) {138            $delivery = $repo->findById($id);139            if ($delivery === null) {140                $io->error("Inbound delivery not found: {$id}");141                return Command::FAILURE;142            }143            $io->definitionList(144                ['ID' => $delivery->getId()],145                ['Endpoint' => $delivery->getEndpointKey()],146                ['Provider' => $delivery->getProviderKey()],147                ['Status' => $delivery->getStatus()->value],148                ['Signature' => $delivery->getSignatureStatus()->value],149                ['Event Type' => $delivery->getParsedEventType() ?? '(unknown)'],150                ['First Received' => $delivery->getFirstReceivedAt()->format('Y-m-d H:i:s')],151                ['Duplicates' => (string) $delivery->getDuplicateCount()],152                ['Error' => $delivery->getLastError() ?? '(none)'],153            );154            return Command::SUCCESS;155        }156157        $status = $input->getOption('status');158        $limit = (int) $input->getOption('limit');159160        if ($status !== null) {161            $deliveries = $repo->findByStatus($status, $limit);162        } else {163            $deliveries = $repo->findByStatus('received', $limit);164        }165166        if ($deliveries === []) {167            $io->info('No inbox entries found.');168            return Command::SUCCESS;169        }170171        $rows = [];172        foreach ($deliveries as $d) {173            $rows[] = [174                // The whole id: webhook:replay:* takes nothing shorter, and a UUIDv7's175                // first characters are its timestamp, shared by rows of the same minute.176                $d->getId(),177                $d->getEndpointKey(),178                $d->getStatus()->value,179                $d->getParsedEventType() ?? '-',180                $d->getFirstReceivedAt()->format('Y-m-d H:i:s'),181            ];182        }183184        $io->table(['ID', 'Endpoint', 'Status', 'Event', 'Received'], $rows);185        return Command::SUCCESS;186    }187188    private function showOutbox(object $container, SymfonyStyle $io, ?string $id, InputInterface $input): int189    {190        $repo = $container->get(OutboundDeliveryRepositoryInterface::class);191192        if ($id !== null) {193            $delivery = $repo->findById($id);194            if ($delivery === null) {195                $io->error("Outbound delivery not found: {$id}");196                return Command::FAILURE;197            }198            $io->definitionList(199                ['ID' => $delivery->getId()],200                ['Endpoint' => $delivery->getEndpointKey()],201                ['Event Type' => $delivery->getEventType()],202                ['Status' => $delivery->getStatus()->value],203                ['Attempts' => $delivery->getAttemptCount() . '/' . $delivery->getMaxAttempts()],204                ['Next Attempt' => $delivery->getNextAttemptAt()->format('Y-m-d H:i:s')],205                ['Last Response' => $delivery->getLastResponseStatus() !== null ? "HTTP {$delivery->getLastResponseStatus()}" : '(none)'],206                ['Error' => $delivery->getLastError() ?? '(none)'],207            );208            return Command::SUCCESS;209        }210211        $status = $input->getOption('status');212        $limit = (int) $input->getOption('limit');213214        if ($status !== null) {215            $deliveries = $repo->findByStatus($status, $limit);216        } else {217            $deliveries = $repo->findByStatus('pending', $limit);218        }219220        if ($deliveries === []) {221            $io->info('No outbox entries found.');222            return Command::SUCCESS;223        }224225        $rows = [];226        foreach ($deliveries as $d) {227            $rows[] = [228                // The whole id: webhook:replay:* takes nothing shorter, and a UUIDv7's229                // first characters are its timestamp, shared by rows of the same minute.230                $d->getId(),231                $d->getEndpointKey(),232                $d->getEventType(),233                $d->getStatus()->value,234                $d->getAttemptCount() . '/' . $d->getMaxAttempts(),235            ];236        }237238        $io->table(['ID', 'Endpoint', 'Event', 'Status', 'Attempts'], $rows);239        return Command::SUCCESS;240    }241242    private function showAttempts(object $container, SymfonyStyle $io, ?string $id): int243    {244        if ($id === null) {245            $io->error('An inbox or outbox ID is required for showing attempts.');246            return Command::FAILURE;247        }248249        $repo = $container->get(WebhookAttemptRepositoryInterface::class);250251        // Try inbound first, then outbound252        $attempts = $repo->findByInboxId($id);253        if ($attempts === []) {254            $attempts = $repo->findByOutboxId($id);255        }256257        if ($attempts === []) {258            $io->info("No attempts found for ID: {$id}");259            return Command::SUCCESS;260        }261262        $rows = [];263        foreach ($attempts as $a) {264            $rows[] = [265                $a->getDirection()->value,266                $a->getEventType(),267                $a->getStatusBefore() ?? '-',268                $a->getStatusAfter() ?? '-',269                $a->getHttpStatus() !== null ? (string) $a->getHttpStatus() : '-',270                $a->getMessage() ?? '-',271                $a->getCreatedAt()->format('Y-m-d H:i:s'),272            ];273        }274275        $io->table(['Dir', 'Event', 'Before', 'After', 'HTTP', 'Message', 'At'], $rows);276        return Command::SUCCESS;277    }278279    private function invalidType(SymfonyStyle $io, string $type): int280    {281        $io->error("Unknown type: {$type}. Use: endpoints, inbox, outbox, attempts");282        return Command::FAILURE;283    }284}285

<?php

declare(strict_types=1);

namespace Semitexa\Webhooks\Application\Console\Command;

use Psr\Container\ContainerInterface;
use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Webhooks\Domain\Contract\InboundDeliveryRepositoryInterface;
use Semitexa\Webhooks\Domain\Contract\OutboundDeliveryRepositoryInterface;
use Semitexa\Webhooks\Domain\Contract\WebhookAttemptRepositoryInterface;
use Semitexa\Webhooks\Domain\Contract\WebhookEndpointDefinitionRepositoryInterface;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'webhook:show', description: 'Display webhook endpoint, inbox, or outbox details')]
final class WebhookShowCommand extends Command
{
    /**
     * Injected rather than reached for statically. The services below are still
     * resolved lazily inside the command body, and deliberately so: every
     * #[AsCommand] class is instantiated and injected at console boot, so
     * injecting a repository directly would open its connection on every
     * `bin/semitexa` invocation — and a dependency that failed to build would
     * make this command vanish from the list instead of reporting the failure.
     */
    #[InjectAsReadonly]
    protected ContainerInterface $container;

    protected function configure(): void
    {
        $this
            ->setName('webhook:show')
            ->setDescription('Display webhook endpoint, inbox, or outbox details')
            ->addArgument(
                name: 'type',
                mode: InputArgument::REQUIRED,
                description: 'What to show: endpoints, inbox, outbox, or attempts',
            )
            ->addArgument(
                name: 'id',
                mode: InputArgument::OPTIONAL,
                description: 'Specific ID to show details for',
            )
            ->addOption(
                name: 'status',
                shortcut: 's',
                mode: InputOption::VALUE_OPTIONAL,
                description: 'Filter by status',
            )
            ->addOption(
                name: 'limit',
                shortcut: 'l',
                mode: InputOption::VALUE_OPTIONAL,
                description: 'Max rows to display',
                default: '20',
            );
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);
        $type = $input->getArgument('type');
        $id = $input->getArgument('id');

        try {
            $container = $this->container;

            return match ($type) {
                'endpoints' => $this->showEndpoints($container, $io, $id),
                'inbox' => $this->showInbox($container, $io, $id, $input),
                'outbox' => $this->showOutbox($container, $io, $id, $input),
                'attempts' => $this->showAttempts($container, $io, $id),
                default => $this->invalidType($io, $type),
            };
        } catch (\Throwable $e) {
            $io->error($e->getMessage());
            return Command::FAILURE;
        }
    }

    private function showEndpoints(object $container, SymfonyStyle $io, ?string $id): int
    {
        $repo = $container->get(WebhookEndpointDefinitionRepositoryInterface::class);

        if ($id !== null) {
            $endpoint = $repo->findByEndpointKey($id) ?? $repo->findById($id);
            if ($endpoint === null) {
                $io->error("Endpoint not found: {$id}");
                return Command::FAILURE;
            }
            $io->definitionList(
                ['ID' => $endpoint->getId()],
                ['Endpoint Key' => $endpoint->getEndpointKey()],
                ['Direction' => $endpoint->getDirection()->value],
                ['Provider' => $endpoint->getProviderKey()],
                ['Enabled' => $endpoint->isEnabled() ? 'Yes' : 'No'],
                ['Target URL' => $endpoint->getTargetUrl() ?? '(none)'],
                ['Verification' => $endpoint->getVerificationMode() ?? '(none)'],
                ['Secret Ref' => $endpoint->getSecretRef() !== null ? '***' . substr($endpoint->getSecretRef(), -4) : '(none)'],
                ['Max Attempts' => (string) $endpoint->getMaxAttempts()],
                ['Timeout' => $endpoint->getTimeoutSeconds() . 's'],
            );
            return Command::SUCCESS;
        }

        $endpoints = $repo->findAll();
        if ($endpoints === []) {
            $io->info('No webhook endpoints defined.');
            return Command::SUCCESS;
        }

        $rows = [];
        foreach ($endpoints as $ep) {
            $rows[] = [
                $ep->getEndpointKey(),
                $ep->getDirection()->value,
                $ep->getProviderKey(),
                $ep->isEnabled() ? 'Yes' : 'No',
                $ep->getTargetUrl() ?? '-',
            ];
        }

        $io->table(['Endpoint Key', 'Direction', 'Provider', 'Enabled', 'Target URL'], $rows);
        return Command::SUCCESS;
    }

    private function showInbox(object $container, SymfonyStyle $io, ?string $id, InputInterface $input): int
    {
        $repo = $container->get(InboundDeliveryRepositoryInterface::class);

        if ($id !== null) {
            $delivery = $repo->findById($id);
            if ($delivery === null) {
                $io->error("Inbound delivery not found: {$id}");
                return Command::FAILURE;
            }
            $io->definitionList(
                ['ID' => $delivery->getId()],
                ['Endpoint' => $delivery->getEndpointKey()],
                ['Provider' => $delivery->getProviderKey()],
                ['Status' => $delivery->getStatus()->value],
                ['Signature' => $delivery->getSignatureStatus()->value],
                ['Event Type' => $delivery->getParsedEventType() ?? '(unknown)'],
                ['First Received' => $delivery->getFirstReceivedAt()->format('Y-m-d H:i:s')],
                ['Duplicates' => (string) $delivery->getDuplicateCount()],
                ['Error' => $delivery->getLastError() ?? '(none)'],
            );
            return Command::SUCCESS;
        }

        $status = $input->getOption('status');
        $limit = (int) $input->getOption('limit');

        if ($status !== null) {
            $deliveries = $repo->findByStatus($status, $limit);
        } else {
            $deliveries = $repo->findByStatus('received', $limit);
        }

        if ($deliveries === []) {
            $io->info('No inbox entries found.');
            return Command::SUCCESS;
        }

        $rows = [];
        foreach ($deliveries as $d) {
            $rows[] = [
                // The whole id: webhook:replay:* takes nothing shorter, and a UUIDv7's
                // first characters are its timestamp, shared by rows of the same minute.
                $d->getId(),
                $d->getEndpointKey(),
                $d->getStatus()->value,
                $d->getParsedEventType() ?? '-',
                $d->getFirstReceivedAt()->format('Y-m-d H:i:s'),
            ];
        }

        $io->table(['ID', 'Endpoint', 'Status', 'Event', 'Received'], $rows);
        return Command::SUCCESS;
    }

    private function showOutbox(object $container, SymfonyStyle $io, ?string $id, InputInterface $input): int
    {
        $repo = $container->get(OutboundDeliveryRepositoryInterface::class);

        if ($id !== null) {
            $delivery = $repo->findById($id);
            if ($delivery === null) {
                $io->error("Outbound delivery not found: {$id}");
                return Command::FAILURE;
            }
            $io->definitionList(
                ['ID' => $delivery->getId()],
                ['Endpoint' => $delivery->getEndpointKey()],
                ['Event Type' => $delivery->getEventType()],
                ['Status' => $delivery->getStatus()->value],
                ['Attempts' => $delivery->getAttemptCount() . '/' . $delivery->getMaxAttempts()],
                ['Next Attempt' => $delivery->getNextAttemptAt()->format('Y-m-d H:i:s')],
                ['Last Response' => $delivery->getLastResponseStatus() !== null ? "HTTP {$delivery->getLastResponseStatus()}" : '(none)'],
                ['Error' => $delivery->getLastError() ?? '(none)'],
            );
            return Command::SUCCESS;
        }

        $status = $input->getOption('status');
        $limit = (int) $input->getOption('limit');

        if ($status !== null) {
            $deliveries = $repo->findByStatus($status, $limit);
        } else {
            $deliveries = $repo->findByStatus('pending', $limit);
        }

        if ($deliveries === []) {
            $io->info('No outbox entries found.');
            return Command::SUCCESS;
        }

        $rows = [];
        foreach ($deliveries as $d) {
            $rows[] = [
                // The whole id: webhook:replay:* takes nothing shorter, and a UUIDv7's
                // first characters are its timestamp, shared by rows of the same minute.
                $d->getId(),
                $d->getEndpointKey(),
                $d->getEventType(),
                $d->getStatus()->value,
                $d->getAttemptCount() . '/' . $d->getMaxAttempts(),
            ];
        }

        $io->table(['ID', 'Endpoint', 'Event', 'Status', 'Attempts'], $rows);
        return Command::SUCCESS;
    }

    private function showAttempts(object $container, SymfonyStyle $io, ?string $id): int
    {
        if ($id === null) {
            $io->error('An inbox or outbox ID is required for showing attempts.');
            return Command::FAILURE;
        }

        $repo = $container->get(WebhookAttemptRepositoryInterface::class);

        // Try inbound first, then outbound
        $attempts = $repo->findByInboxId($id);
        if ($attempts === []) {
            $attempts = $repo->findByOutboxId($id);
        }

        if ($attempts === []) {
            $io->info("No attempts found for ID: {$id}");
            return Command::SUCCESS;
        }

        $rows = [];
        foreach ($attempts as $a) {
            $rows[] = [
                $a->getDirection()->value,
                $a->getEventType(),
                $a->getStatusBefore() ?? '-',
                $a->getStatusAfter() ?? '-',
                $a->getHttpStatus() !== null ? (string) $a->getHttpStatus() : '-',
                $a->getMessage() ?? '-',
                $a->getCreatedAt()->format('Y-m-d H:i:s'),
            ];
        }

        $io->table(['Dir', 'Event', 'Before', 'After', 'HTTP', 'Message', 'At'], $rows);
        return Command::SUCCESS;
    }

    private function invalidType(SymfonyStyle $io, string $type): int
    {
        $io->error("Unknown type: {$type}. Use: endpoints, inbox, outbox, attempts");
        return Command::FAILURE;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Webhooks\Application\Console\Command;67use Psr\Container\ContainerInterface;8use Semitexa\Core\Attribute\AsCommand;9use Semitexa\Core\Attribute\InjectAsReadonly;10use Semitexa\Core\Tenant\TenantContextAccess;11use Semitexa\Core\Tenant\TenantContextStoreInterface;12use Semitexa\Orm\Application\Service\Uuid7;13use Semitexa\Webhooks\Application\Service\Inbound\InboundWebhookReceiver;14use Semitexa\Webhooks\Domain\Contract\InboundDeliveryRepositoryInterface;15use Semitexa\Webhooks\Domain\Contract\WebhookAttemptRepositoryInterface;16use Semitexa\Webhooks\Domain\Contract\WebhookEndpointDefinitionRepositoryInterface;17use Semitexa\Webhooks\Domain\Contract\WebhookInboundProcessorInterface;18use Semitexa\Webhooks\Domain\Enum\InboundStatus;19use Semitexa\Webhooks\Domain\Enum\WebhookDirection;20use Semitexa\Webhooks\Domain\Model\WebhookAttempt;21use Symfony\Component\Console\Command\Command;22use Symfony\Component\Console\Input\InputArgument;23use Symfony\Component\Console\Input\InputInterface;24use Symfony\Component\Console\Input\InputOption;25use Symfony\Component\Console\Output\OutputInterface;26use Symfony\Component\Console\Style\SymfonyStyle;2728#[AsCommand(name: 'webhook:replay:inbound', description: 'Replay an inbound webhook delivery by ID')]29final class WebhookReplayInboundCommand extends Command30{31    /**32     * Injected rather than reached for statically. The services below are still33     * resolved lazily inside the command body, and deliberately so: every34     * #[AsCommand] class is instantiated and injected at console boot, so35     * injecting a repository directly would open its connection on every36     * `bin/semitexa` invocation — and a dependency that failed to build would37     * make this command vanish from the list instead of reporting the failure.38     */39    #[InjectAsReadonly]40    protected ContainerInterface $container;4142    protected function configure(): void43    {44        $this45            ->setName('webhook:replay:inbound')46            ->setDescription('Replay an inbound webhook delivery by ID')47            ->addArgument(48                name: 'id',49                mode: InputArgument::REQUIRED,50                description: 'UUID of the inbound delivery to replay',51            )52            ->addOption('force', null, InputOption::VALUE_NONE, 'Also replay a delivery that was already processed');53    }5455    protected function execute(InputInterface $input, OutputInterface $output): int56    {57        $io = new SymfonyStyle($input, $output);58        $id = $input->getArgument('id');5960        try {61            $container = $this->container;62            $inboxRepo = $container->get(InboundDeliveryRepositoryInterface::class);63            $attemptRepo = $container->get(WebhookAttemptRepositoryInterface::class);64            $receiver = $container->get(InboundWebhookReceiver::class);6566            $delivery = $inboxRepo->findById($id);67            if ($delivery === null) {68                $io->error("Inbound delivery not found: {$id}");69                return Command::FAILURE;70            }7172            // Not received again: that re-checked a signature whose timestamp73            // had long expired, and ran no processor. The stored delivery goes74            // to the processor its endpoint names.75            $endpoint = $container->get(WebhookEndpointDefinitionRepositoryInterface::class)->findById($delivery->getEndpointDefinitionId());76            if ($endpoint === null) {77                $io->error("The endpoint of delivery {$id} ({$delivery->getEndpointKey()}) no longer exists.");78                return Command::FAILURE;79            }80            if (!$endpoint->isEnabled()) {81                $io->error("Endpoint {$endpoint->getEndpointKey()} is disabled: receive() would refuse this delivery, and so does replay.");82                return Command::FAILURE;83            }84            $handlerClass = $endpoint->getHandlerClass();85            if ($handlerClass === null || $handlerClass === '') {86                $io->error("Endpoint {$endpoint->getEndpointKey()} names no handler_class: there is nothing to replay the delivery through.");87                return Command::FAILURE;88            }89            $processor = $container->get($handlerClass);90            if (!$processor instanceof WebhookInboundProcessorInterface) {91                $io->error("{$handlerClass} is not a WebhookInboundProcessorInterface.");92                return Command::FAILURE;93            }94            // A processor scoped to the delivery's tenant would run in no95            // tenant at all from a plain CLI call.96            $tenant = $delivery->getTenantId();97            if ($tenant !== null && $tenant !== $this->currentTenant()) {98                $io->error("Delivery {$id} belongs to tenant {$tenant}: run bin/semitexa tenant:run {$tenant} webhook:replay:inbound {$id}");99                return Command::FAILURE;100            }101102            $statusBefore = $delivery->getStatus()->value;103            $io->info("Replaying inbound delivery {$id} (endpoint: {$delivery->getEndpointKey()})");104            $result = $receiver->replay($delivery, $processor, (bool) $input->getOption('force'));105106            $attemptRepo->save(new WebhookAttempt(107                id: Uuid7::generate(),108                direction: WebhookDirection::Inbound,109                inboxId: $delivery->getId(),110                outboxId: null,111                eventType: 'replayed',112                attemptNumber: null,113                statusBefore: $statusBefore,114                statusAfter: $result->getStatus()->value,115                workerId: 'cli:webhook:replay:inbound',116                httpStatus: null,117                message: 'Manual replay',118                details: null,119            ));120121            if ($result->getStatus() === InboundStatus::Failed) {122                $io->error("Replay ran and the processor failed: {$result->getLastError()}");123                return Command::FAILURE;124            }125            $io->success("Replay completed. Status: {$result->getStatus()->value}");126        } catch (\Throwable $e) {127            $io->error("Replay failed: {$e->getMessage()}");128            return Command::FAILURE;129        }130131        return Command::SUCCESS;132    }133134    /** The tenant this process runs in, "default" on a single-tenant install or without tenant:run. */135    private function currentTenant(): string136    {137        $store = $this->container->has(TenantContextStoreInterface::class) ? $this->container->get(TenantContextStoreInterface::class) : null;138139        return TenantContextAccess::tenantIdOrDefault($store instanceof TenantContextStoreInterface ? $store->tryGet() : null);140    }141}142

<?php

declare(strict_types=1);

namespace Semitexa\Webhooks\Application\Console\Command;

use Psr\Container\ContainerInterface;
use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Core\Tenant\TenantContextAccess;
use Semitexa\Core\Tenant\TenantContextStoreInterface;
use Semitexa\Orm\Application\Service\Uuid7;
use Semitexa\Webhooks\Application\Service\Inbound\InboundWebhookReceiver;
use Semitexa\Webhooks\Domain\Contract\InboundDeliveryRepositoryInterface;
use Semitexa\Webhooks\Domain\Contract\WebhookAttemptRepositoryInterface;
use Semitexa\Webhooks\Domain\Contract\WebhookEndpointDefinitionRepositoryInterface;
use Semitexa\Webhooks\Domain\Contract\WebhookInboundProcessorInterface;
use Semitexa\Webhooks\Domain\Enum\InboundStatus;
use Semitexa\Webhooks\Domain\Enum\WebhookDirection;
use Semitexa\Webhooks\Domain\Model\WebhookAttempt;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'webhook:replay:inbound', description: 'Replay an inbound webhook delivery by ID')]
final class WebhookReplayInboundCommand extends Command
{
    /**
     * Injected rather than reached for statically. The services below are still
     * resolved lazily inside the command body, and deliberately so: every
     * #[AsCommand] class is instantiated and injected at console boot, so
     * injecting a repository directly would open its connection on every
     * `bin/semitexa` invocation — and a dependency that failed to build would
     * make this command vanish from the list instead of reporting the failure.
     */
    #[InjectAsReadonly]
    protected ContainerInterface $container;

    protected function configure(): void
    {
        $this
            ->setName('webhook:replay:inbound')
            ->setDescription('Replay an inbound webhook delivery by ID')
            ->addArgument(
                name: 'id',
                mode: InputArgument::REQUIRED,
                description: 'UUID of the inbound delivery to replay',
            )
            ->addOption('force', null, InputOption::VALUE_NONE, 'Also replay a delivery that was already processed');
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);
        $id = $input->getArgument('id');

        try {
            $container = $this->container;
            $inboxRepo = $container->get(InboundDeliveryRepositoryInterface::class);
            $attemptRepo = $container->get(WebhookAttemptRepositoryInterface::class);
            $receiver = $container->get(InboundWebhookReceiver::class);

            $delivery = $inboxRepo->findById($id);
            if ($delivery === null) {
                $io->error("Inbound delivery not found: {$id}");
                return Command::FAILURE;
            }

            // Not received again: that re-checked a signature whose timestamp
            // had long expired, and ran no processor. The stored delivery goes
            // to the processor its endpoint names.
            $endpoint = $container->get(WebhookEndpointDefinitionRepositoryInterface::class)->findById($delivery->getEndpointDefinitionId());
            if ($endpoint === null) {
                $io->error("The endpoint of delivery {$id} ({$delivery->getEndpointKey()}) no longer exists.");
                return Command::FAILURE;
            }
            if (!$endpoint->isEnabled()) {
                $io->error("Endpoint {$endpoint->getEndpointKey()} is disabled: receive() would refuse this delivery, and so does replay.");
                return Command::FAILURE;
            }
            $handlerClass = $endpoint->getHandlerClass();
            if ($handlerClass === null || $handlerClass === '') {
                $io->error("Endpoint {$endpoint->getEndpointKey()} names no handler_class: there is nothing to replay the delivery through.");
                return Command::FAILURE;
            }
            $processor = $container->get($handlerClass);
            if (!$processor instanceof WebhookInboundProcessorInterface) {
                $io->error("{$handlerClass} is not a WebhookInboundProcessorInterface.");
                return Command::FAILURE;
            }
            // A processor scoped to the delivery's tenant would run in no
            // tenant at all from a plain CLI call.
            $tenant = $delivery->getTenantId();
            if ($tenant !== null && $tenant !== $this->currentTenant()) {
                $io->error("Delivery {$id} belongs to tenant {$tenant}: run bin/semitexa tenant:run {$tenant} webhook:replay:inbound {$id}");
                return Command::FAILURE;
            }

            $statusBefore = $delivery->getStatus()->value;
            $io->info("Replaying inbound delivery {$id} (endpoint: {$delivery->getEndpointKey()})");
            $result = $receiver->replay($delivery, $processor, (bool) $input->getOption('force'));

            $attemptRepo->save(new WebhookAttempt(
                id: Uuid7::generate(),
                direction: WebhookDirection::Inbound,
                inboxId: $delivery->getId(),
                outboxId: null,
                eventType: 'replayed',
                attemptNumber: null,
                statusBefore: $statusBefore,
                statusAfter: $result->getStatus()->value,
                workerId: 'cli:webhook:replay:inbound',
                httpStatus: null,
                message: 'Manual replay',
                details: null,
            ));

            if ($result->getStatus() === InboundStatus::Failed) {
                $io->error("Replay ran and the processor failed: {$result->getLastError()}");
                return Command::FAILURE;
            }
            $io->success("Replay completed. Status: {$result->getStatus()->value}");
        } catch (\Throwable $e) {
            $io->error("Replay failed: {$e->getMessage()}");
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }

    /** The tenant this process runs in, "default" on a single-tenant install or without tenant:run. */
    private function currentTenant(): string
    {
        $store = $this->container->has(TenantContextStoreInterface::class) ? $this->container->get(TenantContextStoreInterface::class) : null;

        return TenantContextAccess::tenantIdOrDefault($store instanceof TenantContextStoreInterface ? $store->tryGet() : null);
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Webhooks\Application\Console\Command;67use Psr\Container\ContainerInterface;8use Semitexa\Core\Attribute\AsCommand;9use Semitexa\Core\Attribute\InjectAsReadonly;10use Semitexa\Core\Lifecycle\StopSignals;11use Semitexa\Webhooks\Application\Service\Outbound\WebhookDeliveryWorker;12use Symfony\Component\Console\Command\Command;13use Symfony\Component\Console\Input\InputArgument;14use Symfony\Component\Console\Input\InputInterface;15use Symfony\Component\Console\Output\OutputInterface;16use Symfony\Component\Console\Style\SymfonyStyle;1718#[AsCommand(name: 'webhook:work', description: 'Run the webhook outbound delivery worker')]19final class WebhookWorkCommand extends Command20{21    /**22     * Injected rather than reached for statically. The services below are still23     * resolved lazily inside the command body, and deliberately so: every24     * #[AsCommand] class is instantiated and injected at console boot, so25     * injecting a repository directly would open its connection on every26     * `bin/semitexa` invocation — and a dependency that failed to build would27     * make this command vanish from the list instead of reporting the failure.28     */29    #[InjectAsReadonly]30    protected ContainerInterface $container;3132    protected function configure(): void33    {34        $this35            ->setName('webhook:work')36            ->setDescription('Run the webhook outbound delivery worker')37            ->addArgument(38                name: 'worker-id',39                mode: InputArgument::OPTIONAL,40                description: 'Unique worker identifier (default: auto-generated)',41                default: null,42            )43            ->addArgument(44                name: 'poll-interval',45                mode: InputArgument::OPTIONAL,46                description: 'Poll interval in seconds (default: 5)',47                default: '5',48            );49    }5051    protected function execute(InputInterface $input, OutputInterface $output): int52    {53        $io = new SymfonyStyle($input, $output);54        $workerId = $input->getArgument('worker-id') ?? 'webhook-worker-' . gethostname() . '-' . getmypid();55        $pollInterval = (int) $input->getArgument('poll-interval');5657        $io->title('Webhook delivery worker');5859        try {60            $container = $this->container;61            $worker = $container->get(WebhookDeliveryWorker::class);62            $worker->setOutput($output);63            // docker stop: the delivery in hand is sent and recorded, then the loop ends.64            StopSignals::install(static fn () => $worker->stop(), static fn (string $m) => $output->writeln($m), 'the delivery in hand');65            $worker->run($workerId, $pollInterval);66        } catch (\Throwable $e) {67            $io->error('Webhook worker failed: ' . $e->getMessage());68            return Command::FAILURE;69        }7071        return Command::SUCCESS;72    }73}74

<?php

declare(strict_types=1);

namespace Semitexa\Webhooks\Application\Console\Command;

use Psr\Container\ContainerInterface;
use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Core\Lifecycle\StopSignals;
use Semitexa\Webhooks\Application\Service\Outbound\WebhookDeliveryWorker;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'webhook:work', description: 'Run the webhook outbound delivery worker')]
final class WebhookWorkCommand extends Command
{
    /**
     * Injected rather than reached for statically. The services below are still
     * resolved lazily inside the command body, and deliberately so: every
     * #[AsCommand] class is instantiated and injected at console boot, so
     * injecting a repository directly would open its connection on every
     * `bin/semitexa` invocation — and a dependency that failed to build would
     * make this command vanish from the list instead of reporting the failure.
     */
    #[InjectAsReadonly]
    protected ContainerInterface $container;

    protected function configure(): void
    {
        $this
            ->setName('webhook:work')
            ->setDescription('Run the webhook outbound delivery worker')
            ->addArgument(
                name: 'worker-id',
                mode: InputArgument::OPTIONAL,
                description: 'Unique worker identifier (default: auto-generated)',
                default: null,
            )
            ->addArgument(
                name: 'poll-interval',
                mode: InputArgument::OPTIONAL,
                description: 'Poll interval in seconds (default: 5)',
                default: '5',
            );
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);
        $workerId = $input->getArgument('worker-id') ?? 'webhook-worker-' . gethostname() . '-' . getmypid();
        $pollInterval = (int) $input->getArgument('poll-interval');

        $io->title('Webhook delivery worker');

        try {
            $container = $this->container;
            $worker = $container->get(WebhookDeliveryWorker::class);
            $worker->setOutput($output);
            // docker stop: the delivery in hand is sent and recorded, then the loop ends.
            StopSignals::install(static fn () => $worker->stop(), static fn (string $m) => $output->writeln($m), 'the delivery in hand');
            $worker->run($workerId, $pollInterval);
        } catch (\Throwable $e) {
            $io->error('Webhook worker failed: ' . $e->getMessage());
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Mail\Application\Console\Command;67use Psr\Container\ContainerInterface;8use Semitexa\Core\Attribute\AsCommand;9use Semitexa\Core\Attribute\InjectAsReadonly;10use Semitexa\Mail\Application\Service\AttachmentResolver;11use Semitexa\Mail\Application\Service\MailWorker;12use Semitexa\Mail\Domain\Contract\MailAttemptRepositoryInterface;13use Semitexa\Mail\Domain\Contract\MailerConfigResolverInterface;14use Semitexa\Mail\Domain\Contract\MailRepositoryInterface;15use Semitexa\Mail\Domain\Contract\MailRetentionInterface;16use Symfony\Component\Console\Command\Command;17use Symfony\Component\Console\Input\InputArgument;18use Symfony\Component\Console\Input\InputInterface;19use Symfony\Component\Console\Output\OutputInterface;20use Symfony\Component\Console\Style\SymfonyStyle;2122#[AsCommand(name: 'mail:work', description: 'Run the dedicated mail delivery worker')]23final class MailWorkCommand extends Command24{25    /**26     * Injected rather than reached for statically. The services below are still27     * resolved lazily inside the command body, and deliberately so: every28     * #[AsCommand] class is instantiated and injected at console boot, so29     * injecting a repository directly would open its connection on every30     * `bin/semitexa` invocation — and a dependency that failed to build would31     * make this command vanish from the list instead of reporting the failure.32     */33    #[InjectAsReadonly]34    protected ContainerInterface $container;3536    protected function configure(): void37    {38        $this39            ->setName('mail:work')40            ->setDescription('Run the dedicated mail delivery worker')41            ->addArgument(42                name:        'transport',43                mode:        InputArgument::OPTIONAL,44                description: 'Queue transport: database, nats or in-memory (default from EVENTS_ASYNC / EVENTS_TRANSPORT)',45                default:     null,46            )47            ->addArgument(48                name:        'queue',49                mode:        InputArgument::OPTIONAL,50                description: 'Queue name (default: mail)',51                default:     null,52            );53    }5455    protected function execute(InputInterface $input, OutputInterface $output): int56    {57        $io        = new SymfonyStyle($input, $output);58        $transport = $input->getArgument('transport');59        $queue     = $input->getArgument('queue');6061        $io->title('Mail worker');6263        try {64            $container      = $this->container;65            $mailRepository = $container->get(MailRepositoryInterface::class);66            $attemptRepo    = $container->get(MailAttemptRepositoryInterface::class);67            $configResolver = $container->get(MailerConfigResolverInterface::class);68            $attachResolver = $container->get(AttachmentResolver::class);6970            $retention = $mailRepository instanceof MailRetentionInterface ? $mailRepository : null;71            if ($retention === null && (int) \Semitexa\Core\Environment::getEnvValue('MAIL_RETENTION_DAYS', '0') > 0) {72                $output->writeln('<comment>MAIL_RETENTION_DAYS is set, but the bound mail repository does not implement MailRetentionInterface: nothing is deleted.</comment>');73            }74            $worker = new MailWorker($mailRepository, $attemptRepo, $configResolver, $attachResolver, $retention);75            $worker->setOutput($output);76            $worker->run($transport, $queue);77        } catch (\Throwable $e) {78            $io->error('Mail worker failed: ' . $e->getMessage());79            return Command::FAILURE;80        }8182        return Command::SUCCESS;83    }84}85

<?php

declare(strict_types=1);

namespace Semitexa\Mail\Application\Console\Command;

use Psr\Container\ContainerInterface;
use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Attribute\InjectAsReadonly;
use Semitexa\Mail\Application\Service\AttachmentResolver;
use Semitexa\Mail\Application\Service\MailWorker;
use Semitexa\Mail\Domain\Contract\MailAttemptRepositoryInterface;
use Semitexa\Mail\Domain\Contract\MailerConfigResolverInterface;
use Semitexa\Mail\Domain\Contract\MailRepositoryInterface;
use Semitexa\Mail\Domain\Contract\MailRetentionInterface;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

#[AsCommand(name: 'mail:work', description: 'Run the dedicated mail delivery worker')]
final class MailWorkCommand extends Command
{
    /**
     * Injected rather than reached for statically. The services below are still
     * resolved lazily inside the command body, and deliberately so: every
     * #[AsCommand] class is instantiated and injected at console boot, so
     * injecting a repository directly would open its connection on every
     * `bin/semitexa` invocation — and a dependency that failed to build would
     * make this command vanish from the list instead of reporting the failure.
     */
    #[InjectAsReadonly]
    protected ContainerInterface $container;

    protected function configure(): void
    {
        $this
            ->setName('mail:work')
            ->setDescription('Run the dedicated mail delivery worker')
            ->addArgument(
                name:        'transport',
                mode:        InputArgument::OPTIONAL,
                description: 'Queue transport: database, nats or in-memory (default from EVENTS_ASYNC / EVENTS_TRANSPORT)',
                default:     null,
            )
            ->addArgument(
                name:        'queue',
                mode:        InputArgument::OPTIONAL,
                description: 'Queue name (default: mail)',
                default:     null,
            );
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io        = new SymfonyStyle($input, $output);
        $transport = $input->getArgument('transport');
        $queue     = $input->getArgument('queue');

        $io->title('Mail worker');

        try {
            $container      = $this->container;
            $mailRepository = $container->get(MailRepositoryInterface::class);
            $attemptRepo    = $container->get(MailAttemptRepositoryInterface::class);
            $configResolver = $container->get(MailerConfigResolverInterface::class);
            $attachResolver = $container->get(AttachmentResolver::class);

            $retention = $mailRepository instanceof MailRetentionInterface ? $mailRepository : null;
            if ($retention === null && (int) \Semitexa\Core\Environment::getEnvValue('MAIL_RETENTION_DAYS', '0') > 0) {
                $output->writeln('<comment>MAIL_RETENTION_DAYS is set, but the bound mail repository does not implement MailRetentionInterface: nothing is deleted.</comment>');
            }
            $worker = new MailWorker($mailRepository, $attemptRepo, $configResolver, $attachResolver, $retention);
            $worker->setOutput($output);
            $worker->run($transport, $queue);
        } catch (\Throwable $e) {
            $io->error('Mail worker failed: ' . $e->getMessage());
            return Command::FAILURE;
        }

        return Command::SUCCESS;
    }
}

1<?php23declare(strict_types=1);45namespace Semitexa\Tenancy\Application\Console\Command;67use Semitexa\Core\Attribute\AsCommand;8use Semitexa\Core\Console\BaseCommand;9use Semitexa\Core\Event\EventDispatcherInterface;10use Semitexa\Core\Tenant\TenantContextStoreInterface;11use Semitexa\Tenancy\Context\TenantContext;12use Semitexa\Tenancy\Context\TenantContextStore;13use Semitexa\Tenancy\Domain\Event\TenantSwitched;14use Semitexa\Tenancy\Domain\Contract\TenantRepositoryInterface;15use Semitexa\Tenancy\Application\Service\TenancyBootstrapper;16use Symfony\Component\Console\Command\Command;17use Symfony\Component\Console\Input\InputArgument;18use Symfony\Component\Console\Input\InputInterface;19use Symfony\Component\Console\Input\StringInput;20use Symfony\Component\Console\Output\OutputInterface;21use Symfony\Component\Console\Style\SymfonyStyle;2223/**24 * Execute a CLI command within a specific tenant context.25 *26 * Usage:27 *   bin/semitexa tenant:run acme cache:clear28 *   bin/semitexa tenant:run globex queue:work29 */30#[AsCommand(name: 'tenant:run', description: 'Execute a command in a specific tenant context')]31class TenantRunCommand extends BaseCommand32{33    private TenantRepositoryInterface $repository;34    private ?EventDispatcherInterface $events;35    private TenantContextStoreInterface $tenantContextStore;3637    public function __construct(38        ?TenantRepositoryInterface $repository = null,39        ?EventDispatcherInterface $events = null,40        ?TenantContextStoreInterface $tenantContextStore = null,41    ) {42        parent::__construct();43        $store = $tenantContextStore ?? TenantContextStore::shared();44        $this->repository = $repository ?? (new TenancyBootstrapper($store))->getRepository();45        $this->events = $events;46        $this->tenantContextStore = $store;47    }4849    protected function configure(): void50    {51        $this->setName('tenant:run')52            ->setDescription('Execute a command in a specific tenant context')53            ->addArgument('tenant', InputArgument::REQUIRED, 'Tenant ID')54            ->addArgument('cmd', InputArgument::REQUIRED | InputArgument::IS_ARRAY, 'Command and its arguments');55    }5657    protected function execute(InputInterface $input, OutputInterface $output): int58    {59        $io = new SymfonyStyle($input, $output);6061        $tenantId = $input->getArgument('tenant');62        $cmdParts = $input->getArgument('cmd');6364        // Validate tenant exists and is active65        $tenant = $this->repository->findActive($tenantId);6667        if ($tenant === null) {68            $io->error(sprintf('Tenant "%s" not found or not active.', $tenantId));69            return Command::FAILURE;70        }7172        // Set tenant context for the CLI session, preserving previous73        $context = TenantContext::fromResolution($tenantId, 'cli', 'tenant:run');74        $previous = $this->swapFallback($context);75        $previousEventContext = $previous instanceof TenantContext ? $previous : TenantContext::default();7677        $this->events?->dispatch(new TenantSwitched(78            previous: $previousEventContext,79            current: $context,80        ));8182        $io->text(sprintf('Running in tenant context: %s (%s)', $tenant->name, $tenant->id));8384        try {85            $application = $this->getApplication();8687            if ($application === null) {88                $io->error('Could not access the console application.');89                return Command::FAILURE;90            }9192            $commandInput = new StringInput(implode(' ', $cmdParts));9394            return $application->run($commandInput, $output);95        } finally {96            $this->swapFallback($previous);9798            $this->events?->dispatch(new TenantSwitched(99                previous: $context,100                current: $previousEventContext,101            ));102        }103    }104105    private function swapFallback(?TenantContext $context): ?TenantContext106    {107        $previous = $this->tenantContextStore->tryGet();108109        if ($context instanceof TenantContext) {110            $this->tenantContextStore->set($context);111        } else {112            $this->tenantContextStore->clear();113        }114115        return $previous instanceof TenantContext ? $previous : null;116    }117}118

<?php

declare(strict_types=1);

namespace Semitexa\Tenancy\Application\Console\Command;

use Semitexa\Core\Attribute\AsCommand;
use Semitexa\Core\Console\BaseCommand;
use Semitexa\Core\Event\EventDispatcherInterface;
use Semitexa\Core\Tenant\TenantContextStoreInterface;
use Semitexa\Tenancy\Context\TenantContext;
use Semitexa\Tenancy\Context\TenantContextStore;
use Semitexa\Tenancy\Domain\Event\TenantSwitched;
use Semitexa\Tenancy\Domain\Contract\TenantRepositoryInterface;
use Semitexa\Tenancy\Application\Service\TenancyBootstrapper;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\StringInput;
use Symfony\Component\Console\Output\OutputInterface;
use Symfony\Component\Console\Style\SymfonyStyle;

/**
 * Execute a CLI command within a specific tenant context.
 *
 * Usage:
 *   bin/semitexa tenant:run acme cache:clear
 *   bin/semitexa tenant:run globex queue:work
 */
#[AsCommand(name: 'tenant:run', description: 'Execute a command in a specific tenant context')]
class TenantRunCommand extends BaseCommand
{
    private TenantRepositoryInterface $repository;
    private ?EventDispatcherInterface $events;
    private TenantContextStoreInterface $tenantContextStore;

    public function __construct(
        ?TenantRepositoryInterface $repository = null,
        ?EventDispatcherInterface $events = null,
        ?TenantContextStoreInterface $tenantContextStore = null,
    ) {
        parent::__construct();
        $store = $tenantContextStore ?? TenantContextStore::shared();
        $this->repository = $repository ?? (new TenancyBootstrapper($store))->getRepository();
        $this->events = $events;
        $this->tenantContextStore = $store;
    }

    protected function configure(): void
    {
        $this->setName('tenant:run')
            ->setDescription('Execute a command in a specific tenant context')
            ->addArgument('tenant', InputArgument::REQUIRED, 'Tenant ID')
            ->addArgument('cmd', InputArgument::REQUIRED | InputArgument::IS_ARRAY, 'Command and its arguments');
    }

    protected function execute(InputInterface $input, OutputInterface $output): int
    {
        $io = new SymfonyStyle($input, $output);

        $tenantId = $input->getArgument('tenant');
        $cmdParts = $input->getArgument('cmd');

        // Validate tenant exists and is active
        $tenant = $this->repository->findActive($tenantId);

        if ($tenant === null) {
            $io->error(sprintf('Tenant "%s" not found or not active.', $tenantId));
            return Command::FAILURE;
        }

        // Set tenant context for the CLI session, preserving previous
        $context = TenantContext::fromResolution($tenantId, 'cli', 'tenant:run');
        $previous = $this->swapFallback($context);
        $previousEventContext = $previous instanceof TenantContext ? $previous : TenantContext::default();

        $this->events?->dispatch(new TenantSwitched(
            previous: $previousEventContext,
            current: $context,
        ));

        $io->text(sprintf('Running in tenant context: %s (%s)', $tenant->name, $tenant->id));

        try {
            $application = $this->getApplication();

            if ($application === null) {
                $io->error('Could not access the console application.');
                return Command::FAILURE;
            }

            $commandInput = new StringInput(implode(' ', $cmdParts));

            return $application->run($commandInput, $output);
        } finally {
            $this->swapFallback($previous);

            $this->events?->dispatch(new TenantSwitched(
                previous: $context,
                current: $previousEventContext,
            ));
        }
    }

    private function swapFallback(?TenantContext $context): ?TenantContext
    {
        $previous = $this->tenantContextStore->tryGet();

        if ($context instanceof TenantContext) {
            $this->tenantContextStore->set($context);
        } else {
            $this->tenantContextStore->clear();
        }

        return $previous instanceof TenantContext ? $previous : null;
    }
}

Ops Pattern

Healthy command topology for background systems

The value here is operability: separate inspect, plan, execute, and replay actions so the platform can be observed before it is pushed harder.

Separate “show/list” commands from “work/replay/run-now” commands so operators can inspect state before mutating it.

Treat workers as first-class processes with their own commands, not as accidental sidecars hidden behind the web runtime.

Use tenant:run when operational intent is tenant-specific instead of hoping ambient context is correct.

Scheduler surfaces are stronger when planning and execution remain explicit and individually observable.

How it works

Dedicated commands expose each stage separately: inspect schedules, plan due runs, start workers, inspect webhook inbox/outbox state, replay deliveries, and run commands in tenant scope.

Why it matters

This makes background systems operable. Operators can inspect, intervene, and replay explicitly instead of treating async infrastructure as invisible magic behind the web server.

Key concepts

queue:work: Runs the async events worker for queued handlers.
scheduler:plan: Materializes due schedule occurrences into concrete run rows before workers execute them.
tenant:run: Executes any command inside a specific tenant context.