README

Русская версия

Transactional outbox pattern implementation for Yii3. Provides a stateless core for reliably publishing messages with configurable retry policies.

Using an AI coding assistant? llms.txt has a compact API reference you can use. Projects using the llm/skills Composer plugin also get this package's agent skill synced into .agents/skills/ automatically on install.

Requirements

PHP 8.3+
psr/clock ^1.0
psr/log ^3.0

Installation

composer require rasuvaeff/yii3-outbox

Usage

Recording a message

use DateTimeImmutable;
use Psr\Clock\ClockInterface;
use Rasuvaeff\Yii3Outbox\InMemoryStorage;
use Rasuvaeff\Yii3Outbox\Outbox;

$clock = new class implements ClockInterface {
    public function now(): DateTimeImmutable { return new DateTimeImmutable(); }
};

$outbox = new Outbox(storage: $storage, clock: $clock);

$message = $outbox->record(
    type: 'order.created',
    payload: json_encode(['orderId' => 42]),
    aggregateId: 'order-42',
);

Message ids

The message id is the primary key of the outbox table and, when messages are exported to ClickHouse, the deduplication key of the ReplacingMergeTree. Two ways to control it:

Pass the domain event's id — the right choice whenever the message mirrors an event that already has an identifier. Republishing the same event then cannot mint a second id, so the consumer has something stable to deduplicate on:

$outbox->record(
    type: 'order.created',
    payload: $json,
    aggregateId: 'order-42',
    id: $domainEvent->getId(),
);

Bind a generator for messages that have no domain id. The default RandomHexIdGenerator keeps the historical format (32 random hex characters); a time-ordered id makes inserts append instead of scattering across InnoDB pages and gives batches a stable order:

use Rasuvaeff\Yii3Outbox\MessageIdGeneratorInterface;

// symfony/uid
final readonly class Uuid7IdGenerator implements MessageIdGeneratorInterface
{
    public function generate(): string
    {
        return \Symfony\Component\Uid\Uuid::v7()->toRfc4122();
    }
}

// ramsey/uuid — equally monotonic within the same millisecond
final readonly class RamseyUuid7IdGenerator implements MessageIdGeneratorInterface
{
    public function generate(): string
    {
        return \Ramsey\Uuid\Uuid::uuid7()->toString();
    }
}

$outbox = new Outbox(storage: $storage, clock: $clock, idGenerator: new Uuid7IdGenerator());

The package ships no UUID implementation and depends on no UUID library — id is VARCHAR(255) in rasuvaeff/yii3-outbox-db, so any format fits and the choice stays yours.

Implementing storage

use Rasuvaeff\Yii3Outbox\StorageInterface;
use Rasuvaeff\Yii3Outbox\OutboxMessage;

final class DbStorage implements StorageInterface
{
    public function save(OutboxMessage $message): void
    {
        // INSERT INTO outbox ... ON CONFLICT(id) DO UPDATE ...
    }

    public function findPending(array $types = [], int $limit = 1000): array
    {
        // SELECT * FROM outbox WHERE status = 'pending'
        //   [AND type IN (:types)] LIMIT :limit  -- empty $types = all types
        // For retry support, also return status = 'pending' with attempts > 0
    }

    public function markPublished(OutboxMessage $message): void
    {
        // UPDATE outbox SET status = 'published' WHERE id = ?
    }

    public function markFailed(OutboxMessage $message): void
    {
        // UPDATE outbox SET status = 'failed' WHERE id = ?
    }

    public function getById(string $id): ?OutboxMessage
    {
        // SELECT * FROM outbox WHERE id = ?
    }
}

Implementing a publisher

use Rasuvaeff\Yii3Outbox\PublisherInterface;
use Rasuvaeff\Yii3Outbox\OutboxMessage;
use Rasuvaeff\Yii3Outbox\PublishException;

final class RabbitPublisher implements PublisherInterface
{
    public function publish(OutboxMessage $message): void
    {
        try {
            // publish to RabbitMQ, Kafka, etc.
        } catch (\Throwable $e) {
            throw new PublishException(
                message: $e->getMessage(),
                outboxMessage: $message,
                previous: $e,
            );
        }
    }
}

Processing the outbox

use Rasuvaeff\Yii3Outbox\Processor;
use Rasuvaeff\Yii3Outbox\RetryPolicy;

$processor = new Processor(
    storage: $storage,
    publisher: $publisher,
    retryPolicy: new RetryPolicy(maxAttempts: 3, delaySeconds: 60),
    clock: $clock,
    batchSize: 100,
);

$result = $processor->process();
// $result->published — successfully published
// $result->failed   — publish exceptions (message kept Pending if retries remain)
// $result->skipped  — not yet ready for retry

Retry behaviour

When a publish fails:

If attempts < maxAttempts → message stays Pending, will be retried after delaySeconds
If attempts >= maxAttempts → message is marked Failed (terminal)

$policy = new RetryPolicy(maxAttempts: 3, delaySeconds: 60);

$policy->shouldRetry($message);          // bool — attempts remaining?
$policy->isReadyForRetry($message, $now); // bool — delay elapsed?

Using InMemoryStorage for tests

use Rasuvaeff\Yii3Outbox\InMemoryStorage;

$storage = new InMemoryStorage();
$storage->save($message);

$pending = $storage->findPending();
$storage->count();
$storage->clear();

API reference

Outbox

Method	Description
`__construct(storage, clock, idGenerator?)`	Main entry point; default generator = `RandomHexIdGenerator`
`record(type, payload, aggregateId?, id?)`	Create and persist message, returns `OutboxMessage`. `id` = the domain event's id; omitted → generator

OutboxMessage

Method	Description
`create(type, payload, aggregateId?, createdAt?, id?)`	Factory; `id` omitted → 32-char hex
`getId()`	Message ID (32-char hex)
`getType()`	Message type
`getPayload()`	Raw payload string
`getStatus()`	`OutboxStatus` enum
`getCreatedAt()`	`DateTimeImmutable`
`getAttempts()`	Number of publish attempts
`getLastAttemptAt()`	`?DateTimeImmutable`
`getAggregateId()`	`?string`
`withStatus(status)`	Returns new instance with status
`withAttempt(at)`	Returns new instance with incremented attempts and timestamp

MessageIdGeneratorInterface

Implementation	Produces
`RandomHexIdGenerator` (default)	32 hex characters, 128 random bits
your own	anything non-empty; `id` is `VARCHAR(255)` in the DB adapter

OutboxStatus

Case	Value
`Pending`	`'pending'`
`Published`	`'published'`
`Failed`	`'failed'`

RetryPolicy

Method	Description
`__construct(maxAttempts, delaySeconds)`	Default: 3 attempts, 60s delay
`shouldRetry(message)`	Checks attempt count
`isReadyForRetry(message, now)`	Checks attempts + delay elapsed

Processor

Method	Description
`__construct(storage, publisher, retryPolicy, clock, batchSize, logger)`	Default batch: 100
`process()`	Returns `ProcessingResult`

ProcessingResult

Property/Method	Description
`$published`	Count of successfully published messages
`$failed`	Count of publish exceptions this run
`$skipped`	Count of messages not ready for retry
`total()`	Sum of all counters

Serializer

Method	Description
`serialize(message)`	Message to JSON
`deserialize(data)`	JSON to Message

Security

Storage implementations must use parameterized queries for all user values.
Message payload is stored as-is; validate before saving if needed.

Examples

See examples/ for complete usage examples.

Development

make install
make build
make cs-fix
make test
make test-coverage
make mutation
make release-check

make test-coverage and make mutation bootstrap pcov inside the composer:2 container because the base image has no coverage driver.

License

BSD-3-Clause. See LICENSE.md.

rasuvaeff / yii3-outbox

Maintainers

Package info

Statistics

Security

README

Requirements

Installation

Usage

Recording a message

Message ids

Implementing storage

Implementing a publisher

Processing the outbox

Retry behaviour

Using InMemoryStorage for tests

API reference

Outbox

OutboxMessage

MessageIdGeneratorInterface

OutboxStatus

RetryPolicy

Processor

ProcessingResult

Serializer

Security

Examples

Development

License