Download the PHP package babelqueue/php-sdk without Composer

BabelQueue PHP SDK (core)

The framework-agnostic core of BabelQueue for PHP — the canonical polyglot queue envelope codec, contracts and dead-letter helpers. Framework adapters (babelqueue/laravel, babelqueue/symfony, …) are built on top of this.

You usually don't install this directly — you install an adapter:

…and the adapter pulls this core in. Install it directly only for a framework-less PHP app, or to build a new adapter.

What's in here

The contract this core implements — the canonical envelope, URN scheme, broker bindings and versioning policy — is documented at babelqueue.com. The golden conformance fixtures live in tests/fixtures/ — every PHP package must round-trip them.

Framework-less use

Produce the canonical envelope from a plain PHP app and let any other SDK consume it. The reference transports keep the core dependency-free — install only the broker client you use:

phpredis (ext-redis) users can implement the one-method Transport directly — it is just an rpush.

Extension keys, meta and strict encode() (ADR-0031)

• decode() keeps unknown top-level and meta keys, so a consumer that re-emits a message does not lose them; forbidden keys are warned about and dropped.
• encode() rejects forbidden keys (timestamp, urn, meta.max_retries, meta.attempts, meta.source, meta.ts) with InvalidEnvelopeException instead of stripping them.
• URN identity on decode(): a blank job (empty or ASCII whitespace only) falls back to urn; with both present job wins and urn is dropped with a warning naming /urn.

Idempotency — dedupe on meta.id (ADR-0022)

BabelQueue is at-least-once, so handlers should be idempotent — dedupe on the envelope's meta.id. Idempotent::wrap($store, $handler) makes that a one-liner: a previously-succeeded id is skipped + acked; a throw leaves it unmarked so retry/DLQ still apply. The reference InMemoryStore is single-process; for a fleet of consumers two persistent stores share one dedupe record and add an atomic claim so two workers handed the same id never both run (GR-7 — nothing added to require):

A claim's TTL bounds a crash between claim and commit, after which a redelivery may re-run — still at-least-once, not exactly-once (the dual-write window; the outbox below narrows the produce side). The frozen IdempotencyStore interface is untouched; the claim contract is the opt-in ClaimingStore extension.

Safe DLQ replay + replay-bypass (ADR-0026 / ADR-0027)

Redrive replays dead-lettered messages back to their source (or a sandbox) — reset for reprocessing, with dry-run and select, driven through a RedriveIO you bind to your broker. Replaying into the real queue re-runs the handler, so its external effects (charge, email) would re-fire. Replay-bypass closes that: a redrive can stamp the out-of-band bq-replay-bypass transport header (beside the frozen envelope, never in it — GR-1), and a handler wraps its external, non-idempotent side in ReplayBypass::bypassExternalEffects() to skip it on a replay while the idempotent core still runs.

The marker bq-replay-bypass is identical across SDKs (Go's HeaderReplayBypass), so a Go-produced replay is recognised by a PHP consumer. It rides the same HeaderPublisher / HasHeaders seam as the OTel traceparent; schema_version stays 1 and trace_id is preserved.

Transactional outbox (ADR-0029)

A plain producer makes a dual write — commit the business row and publish to the broker — two systems that can disagree on a crash. The outbox removes it: the message is written into the same database, in the same transaction as the business data (so they commit or roll back atomically), and a separate relay publishes it afterwards. No distributed transaction; exactly-once handoff into the broker (then at-least-once on the wire, deduped on meta.id by the consumer-side Idempotent::wrap, ADR-0022).

The helper is dependency-free (GR-7): the core defines OutboxStore and you bind it to your DB. The transaction boundary is yours — OutboxStore::save() runs inside the transaction you opened around your business write. The envelope is stored verbatim (GR-1) and relayed byte-for-byte, so trace_id is preserved end-to-end (GR-4).

OutboxStore is four methods — save(), fetchUnpublished(), markPublished(), markFailed(). A reference InMemoryOutboxStore ships for tests; a runnable InitORM-backed adapter + the outbox-table DDL live in babelqueue-examples/outbox-initorm/, keeping this core DB-free.

GDPR field encryption (ADR-0030)

When a schema marks a data field x-gdpr-sensitive, BabelQueue can encrypt that field's value end-to-end: the producer encrypts it before publish, the consumer decrypts it after decode. It is opt-in and standalone — a producer/consumer that never calls it behaves exactly as before.

The envelope stays frozen (GR-1): only the sensitive value changes — it becomes a ciphertext string, so data stays pure JSON (GR-3), meta.schema_version stays 1, and trace_id is untouched (GR-4). An SDK or hop without the key still carries the envelope byte-compatibly; it just can't read the protected fields.

The crypto is a caller-provided Cipher (GR-7 — the core pulls no crypto dependency, it stays ext-json). Bind it to your KMS / Vault / HSM / tokenisation service, or use the bundled reference OpenSslCipher (AES-256-GCM) — which needs ext-openssl, an optional suggest:

Mark a field in the data schema with the x-gdpr-sensitive keyword — the boolean true, or a free-form category string for documentation. It is validation-neutral (it never makes a value valid or invalid), so annotating a schema is never a breaking change:

Gdpr::protect() / unprotect() walk every marked leaf — nested objects (profile.full_name), array items (addresses[].line), scalar array items, and a root mark — and rewrite each in place. Each value is canonically JSON-encoded then replaced by the ciphertext string (and the exact inverse on the way back), so the round-trip restores the original value byte-for-byte. Behaviour at the edges:

• Validate cleartext — run schema validation before protect() and after unprotect(); a constrained sensitive field (minLength, enum, …) would reject the ciphertext string otherwise.
• Absent field — a marked path missing from a message is skipped (not an error); schemas evolve.
• Idempotent decrypt — a non-string leaf is left untouched, so unprotect() is safe to re-run on already-cleartext data.
• Wrong key / tampered — unprotect() throws the typed BabelQueue\Gdpr\DecryptException, so the consumer fails the message (retry / dead-letter) rather than process unreadable PII.

The sensitive marks are read from the same per-URN data schema the validation path already loads (BabelQueue\Schema\SensitivePaths exposes them, mirroring babelqueue-registry's inventory). The registry declares and audits sensitivity; this is the SDK that enforces it on the wire.

OpenTelemetry tracing (ADR-0025 / ADR-0028)

BabelQueue\Otel\Tracing adds optional OpenTelemetry spans to a producer or consumer. It is opt-in and dependency-light — open-telemetry/api is only a suggest, so the core stays ext-json (GR-7). Cross-hop trace propagation layers two levels:

• trace_id ↔ TraceId (v0.1): the envelope's trace_id maps 1:1 to an OTel trace id, so every hop that shares a trace_id shares one trace — correlation + per-hop timing with zero wire/transport change. Preserved end-to-end (GR-4).
• W3C traceparent (v0.2): the producer also injects the active span context as a traceparent transport header — beside the frozen envelope, never in it (GR-1) — so the consumer starts its span as a true child of the producer span (real cross-hop parent-child linkage). With no traceparent present it falls back to the v0.1 behaviour — a strict, backward-compatible upgrade.

The traceparent rides the out-of-band HeaderPublisher / HasHeaders seam (the same seam as the replay-bypass marker, ADR-0027). Which reference transports carry it:

A plain publish() (no tracer) is unchanged on every transport.

Consumer schema-version gate (broker-bindings §5.7 / §6.7)

KafkaConsumer and PulsarConsumer read the bq-schema-version header / property before decoding the body. A missing or blank value (empty, or only ASCII whitespace; not trim(), so NBSP, U+3000 and similar are not blank) takes the normal decode path. Any other value is compared as-is (no trimming) to "1"; anything else is rejected without decoding. A rejected record reaches the Dispatcher as a message routed on babelqueue:unsupported-schema-version:<value> (KafkaConsumer::UNSUPPORTED_SCHEMA_URN_PREFIX), so it takes the unknown-URN path (never a handler) and its dead-letter copy keeps the original bytes.

Design

This core is the contract runtime, not a worker. It does not own a broker loop or retry — adapters bind to each framework's native queue (Laravel's drop-in driver, Symfony Messenger) and reuse that framework's worker/retry.

Testing

License

MIT © Muhammet Şafak.