Process managers

AbstractProcessManager is the base for event-driven process managers (also called sagas). A process has a correlation ID, a status, business data, an optional timeout, and a storage version.

Define a process manager

Implement the lifecycle hooks and correlation rule:

use DomainFlow\EventSourcing\Interface\DomainEventInterface;
use DomainFlow\EventSourcing\Interface\EntityIdentifierInterface;
use DomainFlow\EventSourcing\ProcessManager\AbstractProcessManager;
use DomainFlow\EventSourcing\ProcessManager\ProcessManagerState;

final class FulfilOrder extends AbstractProcessManager
{
    protected static function newInstance(): static
    {
        return new static();
    }

    public static function correlationId(DomainEventInterface $event): EntityIdentifierInterface
    {
        return $event->getAggregateId();
    }

    protected function createInitialData(DomainEventInterface $event): array
    {
        return ['startedBy' => $event::class];
    }

    protected function onEvent(DomainEventInterface $event): void
    {
        // Update state, call markCompleted()/markFailed(), or setTimeout().
    }

    public static function getSubscribedTo(): array
    {
        return [OrderCreated::class, OrderCompleted::class];
    }
}

handle() starts a new process on the first event. Later events are ignored when they have a different correlation ID or the process is already complete. The available statuses are WAITING, PROCESSING, COMPLETED, and FAILED.

Persisted processing

Use ProcessManagerRepository when a worker loads and saves state for each event:

use DomainFlow\EventSourcing\ProcessManager\ProcessManagerRepository;

$repository = new ProcessManagerRepository($processManagerStorage);
$manager = $repository->handle(FulfilOrder::class, $event);

The storage write is conditional on the state version that was loaded. A concurrent update raises ProcessManagerConcurrencyException rather than overwriting the other worker's state. Completed processes are retained by default; pass forgetCompletedProcesses: true if completed state should be deleted after handling.

For a single long-lived in-process instance, the process manager can instead be registered directly with EventDispatcher as an event subscriber.

Timeouts

setTimeout() schedules a future check; it does not run the timeout itself. Schedule ProcessManagerTimeoutRunner as a worker:

use DomainFlow\EventSourcing\ProcessManager\ProcessManagerTimeoutRunner;

$runner = new ProcessManagerTimeoutRunner(
    $processManagerStorage,
    static fn (ProcessManagerState $state): ?string => FulfilOrder::class,
);

$result = $runner->run();

The resolver must identify the process-manager class for each state. The runner clears the due timeout before calling onTimeout(). A timeout hook can schedule another timeout, mark the process completed or failed, or update its data. The storage version check ensures that only one competing worker persists the result.