feat(consumer): opt-in idempotent consumption (I2)

Adds a native idempotency layer to the consumer: an IdempotencyStore contract
(seen/claim/remember/forget) + an in-memory reference, and IdempotencyOptions
(opt-in via a 3rd Dispatcher ctor arg; key = keyResolver -> meta.id -> trace_id).
The Dispatcher claims before the handler (a lost claim -> Drop/ack-discard),
remembers only after success, and forgets on a throw so an at-least-once
redelivery stays retryable. Default (no options) is unchanged. BYO Redis/PDO
backend documented.

Author: muhammetsafak

CHANGELOG.md

@@ -45,3 +45,12 @@ Bugs from `1.x` that no longer exist in the new design:
   and **RabbitMQ** (AMQP), each paired with the SDK's publish side.
 - A thin `Producer` facade over `EnvelopeCodec` + a `Transport`.
 - A `bin/queue` CLI worker entry point.
+- **Opt-in idempotent consumption.** A new `IdempotencyStore` contract
+  (`Contracts/IdempotencyStore.php`) with an in-memory reference implementation
+  (`InMemoryIdempotencyStore`) lets the `Dispatcher` dedupe at-least-once
+  redeliveries, processing a message **at most once**. Enable it by passing an
+  `IdempotencyOptions` to the `Dispatcher`; the dedup key derives from a custom
+  resolver, the envelope's `meta.id`, or its `trace_id`, and is recorded only
+  after a handler succeeds (a throwing handler stays retryable). Disabled by
+  default — existing behaviour is unchanged. Bring a durable, atomic store
+  (Redis/PDO) for fleet-wide deduplication; see the README.

README.md

@@ -157,7 +157,105 @@ new WorkerOptions(
 );
 ```
 
-Delivery is **at-least-once** — make handlers idempotent.
+Delivery is **at-least-once** — make handlers idempotent, or let the worker do
+it for you (see below).
+
+## Idempotent consumption
+
+Because delivery is at-least-once, a transport may hand the same message to a
+worker more than once (after a crash before the ack, a lapsed reservation, or a
+broker hiccup). The dispatcher can dedupe these redeliveries so a message is
+processed **at most once** — opt in by giving it an `IdempotencyOptions`:
+
+```php
+use InitPHP\Queue\Consumer\Dispatcher;
+use InitPHP\Queue\Consumer\IdempotencyOptions;
+use InitPHP\Queue\Consumer\InMemoryIdempotencyStore;
+
+$dispatcher = new Dispatcher(
+    $handlers,
+    idempotency: new IdempotencyOptions(new InMemoryIdempotencyStore()),
+);
+```
+
+With no options (the default) deduplication is off and the dispatcher behaves
+exactly as before — this is fully backward-compatible.
+
+**How it dedupes.** Before running a handler the dispatcher derives a stable
+dedup key from the message and *claims* it; a key that is already recorded means
+the message was processed before, so the handler is skipped and the message is
+acknowledged and discarded. The key is **only recorded after the handler
+succeeds** — a handler that throws leaves the key free, so the at-least-once
+redelivery is retried as normal.
+
+The key is derived, in order, from:
+
+1. a custom `keyResolver` you supply (`fn (ReceivedMessage): ?string`), else
+2. the producer-minted **`meta.id`** (the canonical message identity), else
+3. the **`trace_id`**.
+
+A message with none of these has no stable identity and is processed without
+deduplication.
+
+```php
+new IdempotencyOptions(
+    store: $store,
+    keyPrefix: 'bq:idemp:',   // namespaces keys in a shared store
+    ttl: 86_400,              // seconds a processed key is retained (null = forever)
+    leaseSeconds: 300,        // in-flight claim lease (frees a crashed worker's claim)
+    keyResolver: fn ($m) => $m->getData()['order_id'] ?? null, // optional
+);
+```
+
+### Bring your own store
+
+`InMemoryIdempotencyStore` is correct for a **single long-running worker** but it
+is not shared between processes and does not survive a restart, so it cannot
+dedupe across the whole fleet. For that, back the
+`InitPHP\Queue\Contracts\IdempotencyStore` contract with a durable, atomic store —
+exactly how the InitPHP ecosystem treats Cache and Database backends as
+pluggable seams:
+
+```php
+use InitPHP\Queue\Contracts\IdempotencyStore;
+
+final class RedisIdempotencyStore implements IdempotencyStore
+{
+    public function __construct(private \Redis $redis) {}
+
+    public function seen(string $key): bool
+    {
+        return (bool) $this->redis->exists($key);
+    }
+
+    public function claim(string $key, int $leaseSeconds = 0): bool
+    {
+        // Atomic test-and-set: only the first caller wins the claim.
+        $options = ['NX'];
+        if ($leaseSeconds > 0) {
+            $options['EX'] = $leaseSeconds;
+        }
+
+        return (bool) $this->redis->set($key, '1', $options);
+    }
+
+    public function remember(string $key, ?int $ttl = null): void
+    {
+        $ttl !== null && $ttl > 0
+            ? $this->redis->set($key, '1', ['EX' => $ttl])
+            : $this->redis->set($key, '1');
+    }
+
+    public function forget(string $key): void
+    {
+        $this->redis->del($key);
+    }
+}
+```
+
+A PDO-backed store works the same way: `claim()` is an `INSERT` that fails on a
+unique `key` column (caught and reported as `false`), `remember()` upserts the
+row, and a `processed_at`/`expires_at` column drives the TTL.
 
 ## Unknown-URN strategy
 

src/Consumer/Dispatcher.php

@@ -7,6 +7,7 @@
 use BabelQueue\Exceptions\UnknownUrnException;
 use BabelQueue\Routing\UnknownUrnStrategy;
 use BabelQueue\Validation\EnvelopeValidator;
+use InitPHP\Queue\Contracts\Handler;
 use InitPHP\Queue\Contracts\HandlerResolver;
 use InitPHP\Queue\Exceptions\ConfigurationException;
 use InitPHP\Queue\Message\ReceivedMessage;
@@ -21,18 +22,27 @@
  * resolve the handler by URN, and run it. A handler that returns is an ack; one
  * that throws is a retry; an unmapped URN is resolved by the configured
  * {@see UnknownUrnStrategy} (`fail` / `delete` / `release` / `dead_letter`).
+ *
+ * When an {@see IdempotencyOptions} is supplied the pipeline gains a
+ * deduplication step: a redelivered message whose key is already recorded is
+ * dropped without re-running the handler, and a key is only recorded *after*
+ * the handler succeeds — so a throwing handler stays retryable. With no options
+ * (the default) the dispatcher behaves exactly as before.
  */
 final class Dispatcher
 {
     /**
      * @param  HandlerResolver  $resolver  Maps a message URN to its handler.
      * @param  string  $unknownUrnStrategy  One of the {@see UnknownUrnStrategy} constants.
+     * @param  IdempotencyOptions|null  $idempotency  Opt-in deduplication; null
+     *         (the default) disables it and preserves the prior behaviour.
      *
      * @throws ConfigurationException When `$unknownUrnStrategy` is not a known strategy.
      */
     public function __construct(
         private readonly HandlerResolver $resolver,
         private readonly string $unknownUrnStrategy = UnknownUrnStrategy::FAIL,
+        private readonly ?IdempotencyOptions $idempotency = null,
     ) {
         if (! in_array($unknownUrnStrategy, UnknownUrnStrategy::all(), true)) {
             throw new ConfigurationException(sprintf(
@@ -60,13 +70,44 @@ public function dispatch(ReceivedMessage $message): Outcome
             return $this->onUnknownUrn($message);
         }
 
+        return $this->runHandler($handler, $message);
+    }
+
+    /**
+     * Run the resolved handler, guarded by the idempotency layer when enabled.
+     *
+     * The dedup key is claimed *before* the handler runs (so a concurrent
+     * redelivery is skipped) and only committed with {@see IdempotencyStore::remember()}
+     * *after* the handler returns — a throwing handler leaves the key
+     * uncommitted, so the at-least-once redelivery is retried as normal.
+     */
+    private function runHandler(Handler $handler, ReceivedMessage $message): Outcome
+    {
+        $key = $this->idempotency?->keyFor($message);
+
+        // A duplicate that is already recorded: it was processed before, so
+        // acknowledge and discard rather than running the handler again.
+        if ($key !== null && ! $this->idempotency->store->claim($key, $this->idempotency->leaseSeconds)) {
+            return Outcome::drop('duplicate');
+        }
+
         try {
             $handler->handle($message);
-
-            return Outcome::ack();
         } catch (Throwable $e) {
+            // Release the in-flight claim so the redelivery can re-run; leaving
+            // it set would suppress a legitimate retry.
+            if ($key !== null) {
+                $this->idempotency->store->forget($key);
+            }
+
             return Outcome::retry($e, 'failed');
         }
+
+        if ($key !== null) {
+            $this->idempotency->store->remember($key, $this->idempotency->ttl);
+        }
+
+        return Outcome::ack();
     }
 
     private function onUnknownUrn(ReceivedMessage $message): Outcome

src/Consumer/IdempotencyOptions.php

@@ -0,0 +1,98 @@
+<?php
+
+declare(strict_types=1);
+
+namespace InitPHP\Queue\Consumer;
+
+use Closure;
+use InitPHP\Queue\Contracts\IdempotencyStore;
+use InitPHP\Queue\Exceptions\ConfigurationException;
+use InitPHP\Queue\Message\ReceivedMessage;
+
+/**
+ * Opt-in configuration for the {@see Dispatcher}'s idempotency layer: the
+ * {@see IdempotencyStore} that records seen messages, how a stable dedup key is
+ * derived from each message, and the TTL/lease bookkeeping.
+ *
+ * Constructing one of these and passing it to the {@see Dispatcher} is the
+ * single switch that turns deduplication on; with no instance (the default) the
+ * dispatcher behaves exactly as before. It is immutable, mirroring
+ * {@see WorkerOptions}.
+ *
+ * Key derivation, in order:
+ *  - a caller-supplied `$keyResolver` (`fn (ReceivedMessage): ?string`) wins
+ *    when it returns a non-empty string;
+ *  - otherwise the producer-minted **`meta.id`** (the canonical message
+ *    identity) is used;
+ *  - failing that, the **`trace_id`** is used as a last resort.
+ * A message that yields none of these has no stable identity and is processed
+ * normally (the dispatcher cannot safely dedupe it).
+ */
+final class IdempotencyOptions
+{
+    /** @var (Closure(ReceivedMessage): ?string)|null */
+    private readonly ?Closure $keyResolver;
+
+    /**
+     * @param  IdempotencyStore  $store  Backend recording processed/in-flight keys.
+     * @param  string  $keyPrefix  Namespacing prefix for every key (lets one
+     *                             shared store serve several workers without
+     *                             collisions).
+     * @param  int|null  $ttl  Seconds a processed key is retained (should
+     *                         comfortably outlive the transport's redelivery
+     *                         window); null retains it indefinitely.
+     * @param  int  $leaseSeconds  How long an in-flight claim is held before it
+     *                            may be reclaimed should the worker die mid-handler.
+     * @param  (callable(ReceivedMessage): ?string)|null  $keyResolver  Optional
+     *         custom key derivation; return null/'' to fall back to meta.id/trace_id.
+     *
+     * @throws ConfigurationException When a numeric option is out of range.
+     */
+    public function __construct(
+        public readonly IdempotencyStore $store,
+        public readonly string $keyPrefix = 'bq:idemp:',
+        public readonly ?int $ttl = 86_400,
+        public readonly int $leaseSeconds = 300,
+        ?callable $keyResolver = null,
+    ) {
+        if ($ttl !== null && $ttl < 0) {
+            throw new ConfigurationException('Idempotency ttl must be null or zero/positive.');
+        }
+
+        if ($leaseSeconds < 0) {
+            throw new ConfigurationException('Idempotency leaseSeconds must be zero or positive.');
+        }
+
+        $this->keyResolver = $keyResolver !== null ? $keyResolver(...) : null;
+    }
+
+    /**
+     * Derive the stable dedup key for `$message`, or null when the message has
+     * no stable identity (in which case it is processed without deduplication).
+     */
+    public function keyFor(ReceivedMessage $message): ?string
+    {
+        $identity = $this->resolveIdentity($message);
+
+        return $identity === null ? null : $this->keyPrefix . $identity;
+    }
+
+    private function resolveIdentity(ReceivedMessage $message): ?string
+    {
+        if ($this->keyResolver !== null) {
+            $custom = ($this->keyResolver)($message);
+            if (is_string($custom) && $custom !== '') {
+                return $custom;
+            }
+        }
+
+        $metaId = $message->getMeta()['id'] ?? null;
+        if (is_string($metaId) && $metaId !== '') {
+            return $metaId;
+        }
+
+        $traceId = $message->getTraceId();
+
+        return $traceId !== '' ? $traceId : null;
+    }
+}