Reliability & governance helpers

Beyond produce/consume, the PHP core ships a set of optional, opt-in helpers that mechanise the reliability and governance contracts the wire spec defines. None of them touch the frozen envelope (schema_version: 1); each is a tooling layer over what the codec already carries. They are the PHP face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

BabelQueue\Idempotency\Idempotent::wrap($store, $handler) makes a handler run at most once per meta.id, even on at-least-once redelivery — skip if already seen, run then remember on success, fail-open on a missing id.

use BabelQueue\Idempotency\Idempotent;
use BabelQueue\Idempotency\InMemoryStore;

$store = new InMemoryStore(); // tests / single process
$dispatch->on('urn:babel:orders:created', Idempotent::wrap($store,
    fn (\BabelQueue\Contracts\ConsumedMessage $m) => chargeOnce($m)));

For a fleet, swap the in-memory store for a shipped persistent one — both implement the ClaimingStore interface (which extends IdempotencyStore with an atomic claim / release so two workers can’t both run an in-flight id):

use BabelQueue\Idempotency\PdoStore;
use BabelQueue\Idempotency\RedisStore;

$pdo->exec(PdoStore::ddl());          // portable CREATE TABLE (Postgres/MySQL/SQLite)
$store = new PdoStore($pdo);          // or: new RedisStore($predisClient)

See Idempotency and the store deep-dive.

Transactional outbox

BabelQueue\Outbox removes the producer dual write: persist the encoded envelope in the same DB transaction as your business row, then a relay publishes the durable rows.

use BabelQueue\Outbox\Outbox;
use BabelQueue\Outbox\OutboxRelay;
use BabelQueue\Outbox\InMemoryOutboxStore;

$store  = new InMemoryOutboxStore();   // production: a PDO/ORM-backed OutboxStore
$outbox = new Outbox($store);

// inside YOUR $db->transaction(...), beside the business write — no commit of its own:
$id = $outbox->write($envelope);       // encodes via EnvelopeCodec, calls $store->save()

// later, a separate process drains the durable rows to the broker:
$result = (new OutboxRelay($transport, $store))->drain(); // ->published / ->failed

OutboxStore is the four-method contract (save / fetchUnpublished / markPublished / markFailed); the transaction boundary is yours. The relay publishes the stored bytes verbatim, so trace_id is preserved. See Transactional Outbox.

DLQ redrive & replay-bypass

BabelQueue\Redrive\Redrive::run($io, $dlq, $options) moves dead-lettered messages back onto a queue — dead_letter removed, attempts reset to 0, everything else preserved — with dryRun, select, toQueue (sandbox) and bypass options. PHP’s Transport is publish-only, so you bind a RedriveIO (reserve / ack / publish) to your broker.

use BabelQueue\Redrive\Redrive;
use BabelQueue\Redrive\RedriveOptions;
use BabelQueue\Redrive\ReplayBypass;

$result = Redrive::run($io, 'orders.dlq', new RedriveOptions(
    toQueue: 'orders.sandbox',
    bypass: true, // stamp bq-replay-bypass (needs a HeaderRedriveIO)
));

// the handler skips effects that already fired on a deliberate replay:
$handler = ReplayBypass::wrap(function (\BabelQueue\Contracts\ConsumedMessage $m): void {
    saveOrder($m);                                          // idempotent core
    ReplayBypass::bypassExternalEffects($m, fn () => sendEmail($m)); // skipped on replay
});

See DLQ redrive & replay-bypass.

GDPR field encryption

BabelQueue\Gdpr\Gdpr::protect() / ::unprotect() encrypt only the data leaves a schema marks x-gdpr-sensitive, in place, leaving data pure JSON and the envelope frozen. The Cipher is caller-bound (so the core pulls no crypto dependency); OpenSslCipher is the reference over ext-openssl (a suggest, not a require). Validate cleartext — protect after validation on produce, unprotect before validation on consume.

use BabelQueue\Gdpr\Gdpr;
use BabelQueue\Gdpr\OpenSslCipher;

$cipher = new OpenSslCipher($key); // or implement BabelQueue\Gdpr\Cipher over a KMS

Gdpr::protect($data, $schema, $cipher);   // producer: encrypt marked leaves
// ... encode + publish ... decode ...
try {
    Gdpr::unprotect($data, $schema, $cipher); // consumer: inverse
} catch (\BabelQueue\Gdpr\DecryptException $e) {
    // wrong key / tampered → retry / dead-letter
}

See GDPR field encryption.

OpenTelemetry (traceparent)

BabelQueue\Otel\Tracing emits publish <urn> / process <urn> spans and, on a header-carrying transport, injects/extracts the W3C traceparent so a consumer span is a true child of the producer span — degrading to v0.1 trace_id correlation otherwise. The in-repo Redis/AMQP/SQS transports carry the header; Kafka/Pulsar/STOMP producer wiring is a documented follow-up. open-telemetry/api is an optional suggest, so the core stays ext-json. See Observability.

Per-URN schema validation

BabelQueue\Schema\SchemaValidated::assert() (producer guard) / ::wrap() (consumer safety net) validate a message’s data against the JSON Schema registered for its URN, bridged from a babelqueue-registry manifest via DirProvider. See Per-URN schema validation.