Messenger Dedupe Bundle Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require bytespin/messenger-dedupe-bundle
   

2. Register Bundle: Add to config/bundles.php:

   ByteSpin\MessengerDedupeBundle\MessengerDedupeBundle::class => ['all' => true],
   

3. Update Schema:

   php bin/console doctrine:schema:update --force
   

4. Configure Deduplication: Add to config/packages/messenger.yaml:

   framework:
       messenger:
           transports:
               async: '%env(MESSENGER_TRANSPORT_DSN)%'
           routing:
               'App\Message\YourMessage': async
           buses:
               messenger.bus.default:
                   middleware:
                       - 'ByteSpin\MessengerDedupeBundle\Middleware\DedupeMiddleware'
   

First Use Case

Enable deduplication for a specific message class by:

1. Adding a unique identifier (e.g., messageId, payloadHash) to your message class.
2. Dispatching the message:

   $this->messageBus->dispatch(new YourMessage($uniqueIdentifier));
   

   The bundle will automatically check for duplicates before enqueuing.

Implementation Patterns

Core Workflow

1. Middleware Integration:

   • The DedupeMiddleware checks for existing messages in the messenger_messages table before enqueuing.
   • Configure it globally or per-bus in messenger.yaml.

2. Message Deduplication Logic:

   • Uses a hash of the message payload + a TTL (time-to-live) to avoid false positives.
   • Default TTL is 1 hour (configurable via config/packages/messenger_dedupe.yaml):

     messenger_dedupe:
         ttl: 3600  # 1 hour in seconds
     

3. Handling Duplicates:

   • If a duplicate is found, the message is skipped (no retry).
   • Log duplicates with:

     messenger_dedupe:
         log_duplicates: true
     

Advanced Patterns

1. Custom Dedupe Keys: Override the default deduplication key (hash of payload) by implementing DedupeInterface in your message:

   use ByteSpin\MessengerDedupeBundle\Contract\DedupeInterface;
   
   class YourMessage implements DedupeInterface {
       public function getDedupeKey(): string {
           return $this->uniqueField . '_' . $this->timestamp;
       }
   }
   

2. Bulk Processing: For ETL-like workflows, batch messages with the same deduplication key to avoid redundant processing:

   $this->messageBus->dispatch(new BatchMessage([$msg1, $msg2]));
   

3. Async Transport Only: The bundle only works with Doctrine transport. For other transports (e.g., symfony://), deduplication is manual.

Gotchas and Tips

Pitfalls

1. Schema Mismatch:

   • The bundle adds a dedupe_key and dedupe_ttl column to messenger_messages.
   • Fix: Run doctrine:schema:update after installation. If already in production, use migrations.

2. TTL Too Short:

   • Default 3600 (1 hour) may cause legitimate retries to be dropped.
   • Tip: Adjust ttl in config or implement dynamic TTL per message.

3. No Retry for Duplicates:

   • The middleware silently skips duplicates. Add logging or a custom handler:

     messenger_dedupe:
         on_duplicate: 'App\Handler\DuplicateMessageHandler'
     

4. Memory Usage:

   • Hashing large payloads may impact performance.
   • Optimization: Use serialize() or json_encode() for complex objects.

Debugging

1. Check for Duplicates: Query the database:

   SELECT * FROM messenger_messages WHERE dedupe_key = 'your_hash_here';
   

2. Enable Logging:

   messenger_dedupe:
       log_duplicates: true
       log_level: debug
   

   Check logs for skipped messages.

3. Middleware Order: Ensure DedupeMiddleware runs before DoctrineTransportMiddleware in messenger.yaml:

   middleware:
       - 'ByteSpin\MessengerDedupeBundle\Middleware\DedupeMiddleware'
       - 'doctrine'
   

Extension Points

1. Custom Storage: Override the default Doctrine storage by implementing DedupeStorageInterface:

   class RedisDedupeStorage implements DedupeStorageInterface {
       // Custom logic
   }
   

   Register it in services.yaml:

   ByteSpin\MessengerDedupeBundle\Middleware\DedupeMiddleware:
       arguments:
           $storage: '@redis_dedupe_storage'
   

2. Dynamic TTL: Extend the middleware to set TTL per message:

   class CustomDedupeMiddleware extends DedupeMiddleware {
       protected function getTtl(MessageInterface $message): int {
           return $message instanceof DynamicTtlMessage ? $message->getTtl() : parent::getTtl();
       }
   }
   

3. Testing: Mock the storage layer in tests:

   $storage = $this->createMock(DedupeStorageInterface::class);
   $storage->method('isDuplicate')->willReturn(true);
   $middleware = new DedupeMiddleware($storage, $bus);