Download the PHP package hivesper/php-events without Composer

☀️ php-events

A lightweight, framework-agnostic event system for PHP 8.4+ built around the transactional outbox pattern.

Events are first written to a durable store (in-memory or SQL), then dispatched to subscribers by a processor. This decouples publishing from handling and makes event delivery reliable across process boundaries.

Features

• Publish events with a typed payload, a string name, and an optional publishAt timestamp
• Subscribe to event types with any callable
• Process queued events sequentially — each event is routed to every registered subscriber by type
• Two stores out of the box — in-memory for tests/dev, SQL (MySQL / SQLite) for production
• Composable dispatch — a ListenerDispatcher interface with a default void+throw implementation and a redelivering decorator that catches failures and writes them to a RedeliveryStore
• Outbox pattern — events transition pending → processing → processed, with the intermediate state surviving worker crashes
• Per-listener retries — failures persist to a RedeliveryStore and the SequentialRedeliveryProcessor consults a configurable RetryPolicy to decide whether and when to retry; a single failing listener of an event is retried independently while the others continue
• Ignored exceptions — pass a list of Throwable classes to the dispatcher to silently swallow expected domain failures (no retry, no log, no DB row)
• Status audit trail — every event status transition is recorded in event_outbox_status for ops visibility
• Scheduled delivery — set publishAt in the future; the processor only picks up events whose time has come
• Worker-safe — MySQL store uses FOR UPDATE SKIP LOCKED to allow multiple workers without double-processing
• Schema templates — versioned DDL files in migrations/{mysql,sqlite}/ to drop into your migration tool, or a one-line Schema::create() helper for projects that want boot-time setup
• Clean architecture boundaries — EventSerializer and EventHydrator keep RawEvent out of your application layer

Installation

Requires PHP ≥ 8.4, ext-json, ext-pdo.

Core concepts

Quick start

Scheduled jobs

A production deployment runs four scheduled jobs against this library. They touch mostly-disjoint tables and rows, so they can run in parallel without contending.

recoverStuckEvents and recoverStuckRedeliveries live on the EventStore and RedeliveryStore interfaces, so recovery jobs can type-hint against the interface. Only the SQL implementations do meaningful work — InMemoryEventStore::recoverStuckEvents() returns 0 (there is no persisted processing state to recover from).

The four jobs do not compete for the same rows:

• EventProcessor::process() reads event_outbox rows in pending; recoverStuckEvents() reads rows in processing — disjoint by status filter.
• RedeliveryProcessor::process() reads event_outbox_redelivery rows in pending_retry and claims by transitioning to dispatching; recoverStuckRedeliveries() reads rows in dispatching — disjoint by status filter.
• The redelivery and event tables are separate.
• The lock from SELECT ... FOR UPDATE SKIP LOCKED is held through each claim, so concurrent workers running the same job never claim the same row either.

The one cross-job interaction to think about: if a recoverStuck* sweeper resets a row while a slow worker is still actively dispatching it, another worker can pick it up on the next tick and re-fire its listener. Choose thresholds comfortably longer than your longest healthy dispatch, and make sure listeners are idempotent (the at-least-once outbox contract requires that anyway).

RawEvent

RawEvent is an internal value object used by the store layer. Application code does not construct or receive RawEvent directly — use EventSerializer when publishing and EventHydrator when processing (see below).

Events are created via two static factories:

Scheduled events

Pass any CarbonImmutable timestamp as publishAt — the SQL store only dequeues events whose publish_at <= now():

EventSerializer

EventSerializer converts a domain event object into a SerializedEvent (a name string and a payload array) before it is stored. EventPublisher calls it automatically — application code never touches RawEvent directly.

JacksonSerializer (built-in)

Derives the event name from the short class name (PascalCase) and uses Jackson's ArrayObjectMapper to serialize the object to an array payload. Handles constructor-promoted properties, nested objects, and collections automatically:

Warning: Renaming the class silently changes the event name, breaking any consumers subscribed to the old name. Use a custom EventSerializer with explicit, stable names when this matters across deployments.

Custom serializer

Implement EventSerializer for explicit name mapping or complex payload graphs:

EventHydrator

EventHydrator reconstructs a domain event object from the stored name and payload. DefaultListenerDispatcher calls it once per subscriber, passing the subscriber as the third argument so the hydrator can resolve a different type for each listener.

JacksonHydrator (built-in, default)

Inspects the subscriber's first parameter type-hint via reflection and delegates reconstruction to Jackson's ArrayObjectMapper:

• Typed class (OrderPlaced $event) — maps the payload array to a fully hydrated instance of that class, including nested objects
• object or no type-hint — falls back to a plain stdClass cast of the payload

Because reconstruction is driven by each subscriber's own type-hint, different listeners for the same event can each receive a different type with no extra wiring:

Custom hydrator

Implement EventHydrator to return fully typed domain event objects to your subscribers:

With a custom hydrator, subscribers receive typed objects:

Event stores

InMemoryEventStore

Zero-dependency, FIFO queue. Perfect for tests and single-process applications.

SqlEventStore

Production-ready persistent store. Requires a PDO connection to MySQL or SQLite. The store assumes its tables already exist — see Setting up the schema below.

MySQL workers use SELECT … FOR UPDATE SKIP LOCKED on the event_outbox table for safe concurrent processing.

If you wire up a RedeliveryStore (see Automatic retry & failure tracking), a third table event_outbox_redelivery holds per-listener retry state — install it from the same migrations/ directory.

Setting up the schema

Two options. Pick whichever matches how the host application manages schema.

Option 1 — your migration tool runs the shipped DDL (recommended). Versioned templates live in migrations/{mysql,sqlite}/ in this package. Each shipped file is immutable; future schema changes ship as new numbered files. The two initial files are:

• 0001_create_event_outbox.sql — required for SqlEventStore (creates event_outbox and event_outbox_status).
• 0001_create_event_outbox_redelivery.sql — required if you use SqlRedeliveryStore (creates event_outbox_redelivery).

Either copy the SQL into your own migration file, or have your migration read it from vendor/hivesper/php-events/migrations/{driver}/. The DDL the package ships for MySQL looks like:

Option 2 — call the shipped Schema::create() helper at boot. Convenient for prototypes, single-app deployments, or projects without their own migration tool. The helper is idempotent (CREATE … IF NOT EXISTS on SQLite, information_schema check on MySQL), so it's safe to call on every boot — but be aware: if the host application also runs migrations, this can race with them. Prefer Option 1 in that case.

The redelivery table has matching MysqlRedeliverySchema::create() and SqliteRedeliverySchema::create() helpers.

Event lifecycle

The processor advances each event through three states:

1. pending — written by add() inside the caller's transaction, alongside a matching audit row.
2. processing — set by next() when the worker claims the event. The row is claimed but not yet declared finished. The audit table gets a second row.
3. processed — set by markProcessed() once every listener for the event has settled (succeeded, been persisted to the redelivery queue, been swallowed by the ignored-exceptions list, or been marked permanently failed). The audit table gets a third row.

If a worker dies between next() and markProcessed(), the row stays in processing — intentionally. Any redelivery rows that did get persisted before the crash remain durable, so listener-level retries still fire when their time comes. To recover the wedged processing row itself, call EventStore::recoverStuckEvents() from a separate scheduled job (see Recovering stuck events below).

Recovering stuck events

EventStore::recoverStuckEvents(CarbonInterval $olderThan): int resets events that are stuck in processing back to pending so the next worker can claim them again. An event is "stuck" when its most recent processing audit row is older than $olderThan. The recovery writes a pending audit row tagged Recovered from stuck processing state so dashboards can distinguish organic vs. recovered transitions. The method returns the number of events it recovered and is safe to run alongside the main worker.

The threshold should be comfortably longer than the longest dispatch you expect under healthy conditions. Too tight a threshold will pull events that are simply taking a while back into pending and double-dispatch them; listeners are expected to be idempotent regardless, but unnecessary work is unnecessary work.

Re-dispatch can re-fire listeners that already succeeded on the previous run. Listeners must be idempotent for the recovery path to be safe.

Recovering stuck redeliveries

The same shape applies to the redelivery table: RedeliveryStore::recoverStuckRedeliveries(CarbonInterval $olderThan): int resets rows wedged in dispatching back to pending_retry so a worker can claim them again on the next processNextRedelivery() tick. A row is "stuck" when its updated_at is older than $olderThan.

Same threshold guidance and same idempotency requirement as recoverStuckEvents.

EventSubscriberMap

Subscribe any callable to a named event type:

Multiple subscribers for the same type are called in registration order.

EventSubscriberBuilder

A fluent builder that produces a ready-to-use EventSubscriberMap. Useful for wiring up callables in one place before handing the result to SequentialEventProcessor:

Duplicate listener registrations are deduplicated automatically.

Running your own processor

SequentialEventProcessor processes all currently-queued events in a single call. Run it in a scheduled job, queue worker, or after each HTTP request:

Implement EventProcessor to build your own — e.g. a parallel or batched processor:

Custom processors must call $store->markProcessed($event->id) once dispatch for an event is complete. next() only moves the row from pending to processing; the final advance to processed is the processor's responsibility, so it can hold the row in processing while it drives any per-listener retries.

Custom EventStore

If you implement your own EventStore, you need to satisfy add(), next(), markProcessed(), and recoverStuckEvents(). For an in-memory or queue-style store with no persisted processing status, the last two collapse to no-ops:

Fail-fast in dev, durable in prod

In local development and CI, letting a failing listener throw immediately is exactly what you want — fast, noisy feedback. In production the calculus is different: a single listener failure should not prevent the remaining listeners from running, nor should an infrastructure blip (a DB hiccup, a connection drop) leave the worker dead.

The library separates those concerns into two composable layers:

• RedeliveringListenerDispatcher — a decorator on the dispatcher. Catches a listener throw and writes a fresh row to the RedeliveryStore, so the failure is durable but the rest of the event's listeners still run.
• SilentEventProcessor / SilentRedeliveryProcessor — decorators on the processor. Catch anything that escapes process() (infra-level failures, the redelivery store's own writes failing, etc.) and log via PSR-3.

In dev, you skip both decorators — anything that goes wrong propagates to the call site. In prod, you compose them.

The silent decorator logs each escape with the throwable attached so you can diagnose what aborted the batch:

(SilentRedeliveryProcessor logs Redelivery processor aborted. in the same shape.) Per-listener failures are caught one layer down by RedeliveringListenerDispatcher and don't pass through the silent decorator at all.

PSR-3 logger

The silent decorators accept any PSR-3 LoggerInterface. vesper php-events ships no concrete logger — supply whichever one your application already uses:

Choosing a wiring per environment

Per-attempt failure logging

SequentialRedeliveryProcessor exhausts retries silently — when the RetryPolicy returns null, the row moves to failed with last_error set, but nothing is logged. To get per-attempt visibility, wrap the inner dispatcher in LoggingListenerDispatcher. It logs every failed dispatch at error level via PSR-3 with ['exception', 'event', 'listener'] context and rethrows, so upstream redelivery scheduling/rescheduling still fires.

Wrap it outside DefaultListenerDispatcher so its ignoredExceptions never reach the logger, and inside RedeliveringListenerDispatcher (event path) / SequentialRedeliveryProcessor (redelivery path) so every attempt is logged:

With exponential backoff and a deduplicating log backend, per-attempt log volume stays bounded while still giving a trail of differing errors across retries.

Automatic retry & failure tracking

A failed listener should not become a silent loss. This library supports automatic retry and durable failure tracking on a per-(event, listener) basis: if a single listener of an event fails, only that listener is retried — the others continue to run and successful deliveries are not re-fired.

The model is fully async: when a listener throws inside SequentialEventProcessor, the RedeliveringListenerDispatcher catches the exception and persists it to the redelivery table immediately (with attempt_number = 1 and next_retry_at = now); the main outbox worker then moves on to the next listener / next event. A separate scheduled job runs SequentialRedeliveryProcessor::process(), which drains the redelivery table and consults the RetryPolicy per failed attempt to decide whether to reschedule or mark the row permanently failed.

The split keeps the event processor simple — it knows nothing about the redelivery store, retry policies, attempt counts, or permanent failure. The dispatcher decorator owns "failure → row"; the redelivery processor owns "row → retry decision".

Retry policy

A RetryPolicy decides whether a failed dispatch should be retried, and at what time. It is consulted only by SequentialRedeliveryProcessor, not by the event processor:

Two implementations ship out of the box:

• NoRetryPolicy (default) — never retries. The first redelivery attempt that fails is marked permanently failed.
• ExponentialBackoffRetryPolicy — five total attempts (one initial + four retries) with delays of 100ms, 500ms, 1min, 5min by default. Every retry is persisted back to the redelivery table and picked up on a future RedeliveryProcessor::process() run.

Without a RedeliveringListenerDispatcher wrapping the dispatcher, any listener throw propagates out of process() (fail-fast) — there is nowhere to persist the failed attempt. Wrap the dispatcher to enable durable retry.

Redelivery store

The RedeliveryStore interface owns per-listener retry state — when an attempt failed, how many attempts have been made, when the next one should run, what the last error was. Two implementations:

• InMemoryRedeliveryStore — array-backed, for tests and dev.
• SqlRedeliveryStore — durable, MySQL/SQLite-compatible. Worker-safe via FOR UPDATE SKIP LOCKED on MySQL. Assumes its event_outbox_redelivery table already exists — see Setting up the schema.

Wiring it together

Run the two processors as separate scheduled jobs:

Long backoffs (e.g. the default 1min / 5min steps) won't be picked up until a future RedeliveryProcessor::process() call after their next_retry_at passes — which is exactly how outbox workers already poll on a schedule.

Ignored exceptions (skip-list)

Some listener failures are not bugs — they're expected domain outcomes that the application already handles upstream (e.g. UserNotFoundException, OrderAlreadyShipped). Retrying them wastes time and reporting them spams the error tracker.

The ignoredExceptions constructor parameter on DefaultListenerDispatcher takes a list of Throwable class-strings. Matching is instanceof-based, so subclasses are also matched. When a listener throws an ignored exception the dispatcher swallows it and returns normally — indistinguishable to callers from a clean run:

• No retry attempt — RedeliveringListenerDispatcher only catches throws that propagate past DefaultListenerDispatcher, so ignored exceptions never reach the redelivery store.
• No PSR-3 log line — nothing escaped the processor for SilentEventProcessor to log.
• No row written to event_outbox_redelivery by the event flow.
• No exception propagation — the next listener for the same event runs as normal.
• Marked succeeded if encountered during redelivery — the listener has "handled" the event.

The recommended pattern is to share the same list with whatever already configures your application's error reporter (Sentry/Bugsnag/etc.) so behaviour stays consistent across the boundary: anything your app considers "expected and not worth a page" is also considered expected here.

Permanently failed dispatches

When SequentialRedeliveryProcessor runs a due retry and the RetryPolicy returns null (no more retries), the row is marked status = 'failed' in event_outbox_redelivery. The row stays in the table so operators can inspect it. The redelivery processor does not throw — the loop continues to the next due row.

The event processor itself never marks anything failed. The only path to a failed row is through the redelivery processor's retry-policy exhaustion. If you want fail-fast semantics in local/CI, simply omit the RedeliveringListenerDispatcher wrapper — listener throws will then propagate out of process() on the first attempt instead.

Re-triggering a failed dispatch

RedeliveryStore::retryNow($eventId, $listener) re-queues a dispatch for immediate retry, regardless of its current status (including failed). The attempt count is preserved — the retry policy's max-attempts ceiling still applies on subsequent automatic failures. The library ships no CLI; wire retryNow() into whatever admin surface you prefer (admin UI, Slack command, console script, etc.).

For listing failures, query event_outbox_redelivery directly.

Listener identity and closures

The redelivery row's listener column is the class name for class-string subscribers, the class name for invokable objects (get_class($obj)), or the literal string 'Closure' for anonymous closures. Class-string and invokable-object listeners can be reliably retried across processes; closures cannot (their identity is not stable across process boundaries). If your listener registrations and your retry policy together require closure tracking, use a class that implements __invoke() instead.

RawEventStatus

Operational queries

Useful queries against the existing tables:

• Listeners with permanent failures — SELECT * FROM event_outbox_redelivery WHERE status = 'failed'.
• Events recovered by the stuck-events sweeper — SELECT * FROM event_outbox_status WHERE error_message = 'Recovered from stuck processing state'.
• Dispatch latency — average time between consecutive rows in event_outbox_status for the same event_id.

recoverStuckEvents() re-claims wedged rows (processing → pending) for re-dispatch. The other plausible recovery mode — force-complete (processing → processed) — is intentionally not exposed: it's only safe when the dispatch is believed to have finished but the bookkeeping commit was lost, and that's a judgement call that belongs in operator tooling rather than a library API. If you need it, the operation is two statements: UPDATE event_outbox SET status = 'processed' WHERE id = ? and an audit insert with a recovery marker.

See ROADMAP.md for known limitations and planned work.

Testing

Or run everything:

License

MIT — see LICENSE.