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

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.

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 async events worker (processes handlers enqueued with execution: async)')]16class QueueWorkCommand extends Command17{18    protected function configure(): void19    {20        $this->setName('queue:work')21            ->setDescription('Run async events worker (processes handlers enqueued with execution: async)')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 async events worker (processes handlers enqueued with execution: async)')]
class QueueWorkCommand extends Command
{
    protected function configure(): void
    {
        $this->setName('queue:work')
            ->setDescription('Run async events worker (processes handlers enqueued with execution: async)')
            ->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                    $nextStr = $next->format('Y-m-d H:i:s') . ' (' . $def->getTimezone() . ')';50                } catch (\Throwable) {51                    $nextStr = '(invalid expression)';52                }5354                $rows[] = [55                    $def->getScheduleKey(),56                    $def->getCronExpression(),57                    $nextStr,58                    $def->getPool(),59                    $def->getOverlapPolicy(),60                    $def->getMisfirePolicy(),61                    $def->getTenantMode(),62                    $def->getPlanningCursorAt()?->format('Y-m-d H:i:s') ?? 'never',63                ];64            }6566            $io->table(67                ['Key', 'Cron', 'Next Due', 'Pool', 'Overlap', 'Misfire', 'Tenant Mode', 'Last Planned'],68                $rows,69            );70        } catch (\Throwable $e) {71            $io->error('scheduler:list failed: ' . $e->getMessage());72            return Command::FAILURE;73        }7475        return Command::SUCCESS;76    }77}78

<?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());
                    $nextStr = $next->format('Y-m-d H:i:s') . ' (' . $def->getTimezone() . ')';
                } catch (\Throwable) {
                    $nextStr = '(invalid expression)';
                }

                $rows[] = [
                    $def->getScheduleKey(),
                    $def->getCronExpression(),
                    $nextStr,
                    $def->getPool(),
                    $def->getOverlapPolicy(),
                    $def->getMisfirePolicy(),
                    $def->getTenantMode(),
                    $def->getPlanningCursorAt()?->format('Y-m-d H:i:s') ?? '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);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);

            $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\Event\EventDispatcherInterface;10use Semitexa\Orm\OrmManager;11use Semitexa\Scheduler\Application\Db\MySQL\Repository\SchedulerRunHistoryRepository;12use Semitexa\Scheduler\Configuration\SchedulerConfig;13use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;14use Semitexa\Scheduler\Domain\Contract\ScheduledRunRepositoryInterface;15use Semitexa\Scheduler\Domain\Contract\SchedulerLockRepositoryInterface;16use Semitexa\Scheduler\Application\Service\RunLeaseManager;17use Semitexa\Scheduler\Application\Service\SchedulerLockManager;18use Semitexa\Scheduler\Application\Service\OverlapPolicyHandler;19use Semitexa\Scheduler\Application\Service\RetryScheduler;20use Semitexa\Scheduler\Application\Service\RunExecutor;21use Semitexa\Scheduler\Application\Service\SchedulerWorker;22use Symfony\Component\Console\Command\Command;23use Symfony\Component\Console\Input\InputArgument;24use Symfony\Component\Console\Input\InputInterface;25use Symfony\Component\Console\Output\OutputInterface;26use Symfony\Component\Console\Style\SymfonyStyle;2728#[AsCommand(name: 'scheduler:work', description: 'Run the scheduler worker for a given pool')]29final class SchedulerWorkCommand extends Command30{31    #[InjectAsReadonly]32    protected ScheduledRunRepositoryInterface $runRepo;3334    #[InjectAsReadonly]35    protected SchedulerLockRepositoryInterface $lockRepo;3637    #[InjectAsReadonly]38    protected ScheduleDefinitionRepositoryInterface $definitionRepo;3940    #[InjectAsReadonly]41    protected EventDispatcherInterface $events;4243    protected function configure(): void44    {45        $this->setName('scheduler:work')46             ->setDescription('Run the scheduler worker for a given pool')47             ->addArgument(48                 name:        'pool',49                 mode:        InputArgument::OPTIONAL,50                 description: 'Worker pool name (default: from SCHEDULER_DEFAULT_POOL env or "default")',51                 default:     null,52             );53    }5455    protected function execute(InputInterface $input, OutputInterface $output): int56    {57        $io   = new SymfonyStyle($input, $output);58        $pool = $input->getArgument('pool');5960        $io->title('Scheduler Worker');6162        try {63            // CLI parity with the Swoole WorkerStart bootstrap: register the64            // ORM's default dispatcher resolver BEFORE any job runs, so every65            // scheduled job's ORM writes auto-publish their invalidation66            // signals without each job carrying its own CLI bootstrap.67            OrmManager::setDefaultEventDispatcherResolver(68                fn (): EventDispatcherInterface => $this->events,69            );7071            $config      = SchedulerConfig::create();72            $historyRepo = new SchedulerRunHistoryRepository();7374            $leaseManager   = new RunLeaseManager($this->runRepo, $config->leaseTtlSeconds);75            $lockManager    = new SchedulerLockManager($this->lockRepo, $config->lockTtlSeconds);76            $overlapHandler = new OverlapPolicyHandler($this->runRepo, $lockManager, $this->definitionRepo, $historyRepo);77            $executor       = new RunExecutor($this->runRepo, $historyRepo);78            $retryScheduler = new RetryScheduler($this->runRepo, $historyRepo);7980            $worker = new SchedulerWorker(81                leaseManager:    $leaseManager,82                lockManager:     $lockManager,83                runRepository:   $this->runRepo,84                overlapHandler:  $overlapHandler,85                executor:        $executor,86                retryScheduler:  $retryScheduler,87                historyRepository: $historyRepo,88                config:          $config,89            );90            $worker->setOutput($output);91            $worker->run($pool);92        } catch (\Throwable $e) {93            $io->error('Scheduler worker failed: ' . $e->getMessage());94            return Command::FAILURE;95        }9697        return Command::SUCCESS;98    }99}100

<?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\Application\Db\MySQL\Repository\SchedulerRunHistoryRepository;
use Semitexa\Scheduler\Configuration\SchedulerConfig;
use Semitexa\Scheduler\Domain\Contract\ScheduleDefinitionRepositoryInterface;
use Semitexa\Scheduler\Domain\Contract\ScheduledRunRepositoryInterface;
use Semitexa\Scheduler\Domain\Contract\SchedulerLockRepositoryInterface;
use Semitexa\Scheduler\Application\Service\RunLeaseManager;
use Semitexa\Scheduler\Application\Service\SchedulerLockManager;
use Semitexa\Scheduler\Application\Service\OverlapPolicyHandler;
use Semitexa\Scheduler\Application\Service\RetryScheduler;
use Semitexa\Scheduler\Application\Service\RunExecutor;
use Semitexa\Scheduler\Application\Service\SchedulerWorker;
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,
            );
            $worker->setOutput($output);
            $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                substr($d->getId(), 0, 8) . '...',175                $d->getEndpointKey(),176                $d->getStatus()->value,177                $d->getParsedEventType() ?? '-',178                $d->getFirstReceivedAt()->format('Y-m-d H:i:s'),179            ];180        }181182        $io->table(['ID', 'Endpoint', 'Status', 'Event', 'Received'], $rows);183        return Command::SUCCESS;184    }185186    private function showOutbox(object $container, SymfonyStyle $io, ?string $id, InputInterface $input): int187    {188        $repo = $container->get(OutboundDeliveryRepositoryInterface::class);189190        if ($id !== null) {191            $delivery = $repo->findById($id);192            if ($delivery === null) {193                $io->error("Outbound delivery not found: {$id}");194                return Command::FAILURE;195            }196            $io->definitionList(197                ['ID' => $delivery->getId()],198                ['Endpoint' => $delivery->getEndpointKey()],199                ['Event Type' => $delivery->getEventType()],200                ['Status' => $delivery->getStatus()->value],201                ['Attempts' => $delivery->getAttemptCount() . '/' . $delivery->getMaxAttempts()],202                ['Next Attempt' => $delivery->getNextAttemptAt()->format('Y-m-d H:i:s')],203                ['Last Response' => $delivery->getLastResponseStatus() !== null ? "HTTP {$delivery->getLastResponseStatus()}" : '(none)'],204                ['Error' => $delivery->getLastError() ?? '(none)'],205            );206            return Command::SUCCESS;207        }208209        $status = $input->getOption('status');210        $limit = (int) $input->getOption('limit');211212        if ($status !== null) {213            $deliveries = $repo->findByStatus($status, $limit);214        } else {215            $deliveries = $repo->findByStatus('pending', $limit);216        }217218        if ($deliveries === []) {219            $io->info('No outbox entries found.');220            return Command::SUCCESS;221        }222223        $rows = [];224        foreach ($deliveries as $d) {225            $rows[] = [226                substr($d->getId(), 0, 8) . '...',227                $d->getEndpointKey(),228                $d->getEventType(),229                $d->getStatus()->value,230                $d->getAttemptCount() . '/' . $d->getMaxAttempts(),231            ];232        }233234        $io->table(['ID', 'Endpoint', 'Event', 'Status', 'Attempts'], $rows);235        return Command::SUCCESS;236    }237238    private function showAttempts(object $container, SymfonyStyle $io, ?string $id): int239    {240        if ($id === null) {241            $io->error('An inbox or outbox ID is required for showing attempts.');242            return Command::FAILURE;243        }244245        $repo = $container->get(WebhookAttemptRepositoryInterface::class);246247        // Try inbound first, then outbound248        $attempts = $repo->findByInboxId($id);249        if ($attempts === []) {250            $attempts = $repo->findByOutboxId($id);251        }252253        if ($attempts === []) {254            $io->info("No attempts found for ID: {$id}");255            return Command::SUCCESS;256        }257258        $rows = [];259        foreach ($attempts as $a) {260            $rows[] = [261                $a->getDirection()->value,262                $a->getEventType(),263                $a->getStatusBefore() ?? '-',264                $a->getStatusAfter() ?? '-',265                $a->getHttpStatus() !== null ? (string) $a->getHttpStatus() : '-',266                $a->getMessage() ?? '-',267                $a->getCreatedAt()->format('Y-m-d H:i:s'),268            ];269        }270271        $io->table(['Dir', 'Event', 'Before', 'After', 'HTTP', 'Message', 'At'], $rows);272        return Command::SUCCESS;273    }274275    private function invalidType(SymfonyStyle $io, string $type): int276    {277        $io->error("Unknown type: {$type}. Use: endpoints, inbox, outbox, attempts");278        return Command::FAILURE;279    }280}281

<?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[] = [
                substr($d->getId(), 0, 8) . '...',
                $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[] = [
                substr($d->getId(), 0, 8) . '...',
                $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\Orm\Application\Service\Uuid7;11use Semitexa\Webhooks\Application\Service\Inbound\InboundWebhookReceiver;12use Semitexa\Webhooks\Domain\Contract\InboundDeliveryRepositoryInterface;13use Semitexa\Webhooks\Domain\Contract\WebhookAttemptRepositoryInterface;14use Semitexa\Webhooks\Domain\Enum\WebhookDirection;15use Semitexa\Webhooks\Domain\Model\InboundWebhookEnvelope;16use Semitexa\Webhooks\Domain\Model\WebhookAttempt;17use Symfony\Component\Console\Command\Command;18use Symfony\Component\Console\Input\InputArgument;19use Symfony\Component\Console\Input\InputInterface;20use Symfony\Component\Console\Output\OutputInterface;21use Symfony\Component\Console\Style\SymfonyStyle;2223#[AsCommand(name: 'webhook:replay:inbound', description: 'Replay an inbound webhook delivery by ID')]24final class WebhookReplayInboundCommand extends Command25{26    /**27     * Injected rather than reached for statically. The services below are still28     * resolved lazily inside the command body, and deliberately so: every29     * #[AsCommand] class is instantiated and injected at console boot, so30     * injecting a repository directly would open its connection on every31     * `bin/semitexa` invocation — and a dependency that failed to build would32     * make this command vanish from the list instead of reporting the failure.33     */34    #[InjectAsReadonly]35    protected ContainerInterface $container;3637    protected function configure(): void38    {39        $this40            ->setName('webhook:replay:inbound')41            ->setDescription('Replay an inbound webhook delivery by ID')42            ->addArgument(43                name: 'id',44                mode: InputArgument::REQUIRED,45                description: 'UUID of the inbound delivery to replay',46            );47    }4849    protected function execute(InputInterface $input, OutputInterface $output): int50    {51        $io = new SymfonyStyle($input, $output);52        $id = $input->getArgument('id');5354        try {55            $container = $this->container;56            $inboxRepo = $container->get(InboundDeliveryRepositoryInterface::class);57            $attemptRepo = $container->get(WebhookAttemptRepositoryInterface::class);58            $receiver = $container->get(InboundWebhookReceiver::class);5960            $delivery = $inboxRepo->findById($id);61            if ($delivery === null) {62                $io->error("Inbound delivery not found: {$id}");63                return Command::FAILURE;64            }6566            $io->info("Replaying inbound delivery {$id} (endpoint: {$delivery->getEndpointKey()})");6768            // Record replay attempt69            $attemptRepo->save(new WebhookAttempt(70                id: Uuid7::generate(),71                direction: WebhookDirection::Inbound,72                inboxId: $delivery->getId(),73                outboxId: null,74                eventType: 'replayed',75                attemptNumber: null,76                statusBefore: $delivery->getStatus()->value,77                statusAfter: $delivery->getStatus()->value,78                workerId: 'cli:webhook:replay:inbound',79                httpStatus: null,80                message: 'Manual replay initiated',81                details: null,82            ));8384            // Reconstruct envelope and re-receive85            $envelope = new InboundWebhookEnvelope(86                endpointKey: $delivery->getEndpointKey(),87                httpMethod: $delivery->getHttpMethod(),88                requestUri: $delivery->getRequestUri(),89                headers: $delivery->getHeaders() ?? [],90                rawBody: $delivery->getRawBody() ?? '',91                contentType: $delivery->getContentType(),92            );9394            $result = $receiver->receive($envelope);95            $io->success("Replay completed. Status: {$result->getStatus()->value}");96        } catch (\Throwable $e) {97            $io->error("Replay failed: {$e->getMessage()}");98            return Command::FAILURE;99        }100101        return Command::SUCCESS;102    }103}104

<?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\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\Enum\WebhookDirection;
use Semitexa\Webhooks\Domain\Model\InboundWebhookEnvelope;
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\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',
            );
    }

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

            $io->info("Replaying inbound delivery {$id} (endpoint: {$delivery->getEndpointKey()})");

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

            // Reconstruct envelope and re-receive
            $envelope = new InboundWebhookEnvelope(
                endpointKey: $delivery->getEndpointKey(),
                httpMethod: $delivery->getHttpMethod(),
                requestUri: $delivery->getRequestUri(),
                headers: $delivery->getHeaders() ?? [],
                rawBody: $delivery->getRawBody() ?? '',
                contentType: $delivery->getContentType(),
            );

            $result = $receiver->receive($envelope);
            $io->success("Replay completed. Status: {$result->getStatus()->value}");
        } catch (\Throwable $e) {
            $io->error("Replay 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\Application\Service\Outbound\WebhookDeliveryWorker;11use Symfony\Component\Console\Command\Command;12use Symfony\Component\Console\Input\InputArgument;13use Symfony\Component\Console\Input\InputInterface;14use Symfony\Component\Console\Output\OutputInterface;15use Symfony\Component\Console\Style\SymfonyStyle;1617#[AsCommand(name: 'webhook:work', description: 'Run the webhook outbound delivery worker')]18final class WebhookWorkCommand extends Command19{20    /**21     * Injected rather than reached for statically. The services below are still22     * resolved lazily inside the command body, and deliberately so: every23     * #[AsCommand] class is instantiated and injected at console boot, so24     * injecting a repository directly would open its connection on every25     * `bin/semitexa` invocation — and a dependency that failed to build would26     * make this command vanish from the list instead of reporting the failure.27     */28    #[InjectAsReadonly]29    protected ContainerInterface $container;3031    protected function configure(): void32    {33        $this34            ->setName('webhook:work')35            ->setDescription('Run the webhook outbound delivery worker')36            ->addArgument(37                name: 'worker-id',38                mode: InputArgument::OPTIONAL,39                description: 'Unique worker identifier (default: auto-generated)',40                default: null,41            )42            ->addArgument(43                name: 'poll-interval',44                mode: InputArgument::OPTIONAL,45                description: 'Poll interval in seconds (default: 5)',46                default: '5',47            );48    }4950    protected function execute(InputInterface $input, OutputInterface $output): int51    {52        $io = new SymfonyStyle($input, $output);53        $workerId = $input->getArgument('worker-id') ?? 'webhook-worker-' . gethostname() . '-' . getmypid();54        $pollInterval = (int) $input->getArgument('poll-interval');5556        $io->title('Webhook delivery worker');5758        try {59            $container = $this->container;60            $worker = $container->get(WebhookDeliveryWorker::class);61            $worker->setOutput($output);62            $worker->run($workerId, $pollInterval);63        } catch (\Throwable $e) {64            $io->error('Webhook worker failed: ' . $e->getMessage());65            return Command::FAILURE;66        }6768        return Command::SUCCESS;69    }70}71

<?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\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);
            $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 Symfony\Component\Console\Command\Command;16use Symfony\Component\Console\Input\InputArgument;17use Symfony\Component\Console\Input\InputInterface;18use Symfony\Component\Console\Output\OutputInterface;19use Symfony\Component\Console\Style\SymfonyStyle;2021#[AsCommand(name: 'mail:work', description: 'Run the dedicated mail delivery worker')]22final class MailWorkCommand 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('mail:work')39            ->setDescription('Run the dedicated mail delivery worker')40            ->addArgument(41                name:        'transport',42                mode:        InputArgument::OPTIONAL,43                description: 'Queue transport: nats or in-memory (default from EVENTS_ASYNC)',44                default:     null,45            )46            ->addArgument(47                name:        'queue',48                mode:        InputArgument::OPTIONAL,49                description: 'Queue name (default: mail)',50                default:     null,51            );52    }5354    protected function execute(InputInterface $input, OutputInterface $output): int55    {56        $io        = new SymfonyStyle($input, $output);57        $transport = $input->getArgument('transport');58        $queue     = $input->getArgument('queue');5960        $io->title('Mail worker');6162        try {63            $container      = $this->container;64            $mailRepository = $container->get(MailRepositoryInterface::class);65            $attemptRepo    = $container->get(MailAttemptRepositoryInterface::class);66            $configResolver = $container->get(MailerConfigResolverInterface::class);67            $attachResolver = $container->get(AttachmentResolver::class);6869            $worker = new MailWorker($mailRepository, $attemptRepo, $configResolver, $attachResolver);70            $worker->setOutput($output);71            $worker->run($transport, $queue);72        } catch (\Throwable $e) {73            $io->error('Mail worker failed: ' . $e->getMessage());74            return Command::FAILURE;75        }7677        return Command::SUCCESS;78    }79}80

<?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 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: nats or in-memory (default from EVENTS_ASYNC)',
                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);

            $worker = new MailWorker($mailRepository, $attemptRepo, $configResolver, $attachResolver);
            $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.