feat(AP-6075): Add publisher & consumer level message deduplication

README.md

@@ -389,3 +389,71 @@ It needs to implement the following methods:
   - `messageProcessingMilliseconds` - message processing time in milliseconds
 
 See [@message-queue-toolkit/metrics](packages/metrics/README.md) for concrete implementations
+
+## Producer-level deduplication
+
+Producer-level message deduplication is a mechanism that prevents the same message from being sent to the queue multiple times.
+It is useful when you want to ensure that a message is published only once, regardless of how many times it is sent.
+
+Note that in case of some queuing systems, such as standard SQS, producer-level deduplication is not sufficient to guarantee that a message is **processed** only once.
+This is because standard SQS has an at-least-once delivery guarantee, which means that a message can be delivered more than once.
+In such cases, producer-level deduplication should be combined with consumer-level one.
+
+### Configuration
+
+1. **Install a deduplication store implementation (Redis in example)**:
+    ```bash
+    npm install @message-queue-toolkit/redis-deduplication-store
+    ```
+
+2. **Configure your setup:**
+    ```typescript
+    import { Redis } from 'ioredis'
+    import { RedisDeduplicationStore } from '@message-queue-toolkit/redis-deduplication-store'
+    import { MessageDeduplicationKeyGenerator } from '@message-queue-toolkit/core'
+
+    const redisClient = new Redis({
+        // your redis configuration
+    })
+
+    // Create a new instance of RedisMessageDeduplicationStore
+    messageDeduplicationStore = new RedisMessageDeduplicationStore(
+      { redis: diContainer.cradle.redis },
+      { keyPrefix: 'optional-key-prefix' }, // used to prefix deduplication keys
+    )
+
+    // Producer-level deduplication allows you to provide custom strategies of deduplication key generation for each message type
+    // In this example we'll provide just one strategy for one message type - 'dummy'
+    class DummyMessageDeduplicationKeyGenerator implements MessageDeduplicationKeyGenerator<DummyEvent> {
+      generateKey(message: DummyEvent): string {
+        return message.id
+      }
+    }
+
+    const dummyMessageDeduplicationKeyGenerator = new DummyMessageDeduplicationKeyGenerator()
+
+    export class MyPublisher extends AbstractSqsPublisher<> {
+        constructor(
+            // dependencies and options
+        ) {
+            super(dependencies, {
+                // rest of the configuration
+                messageDeduplicationConfig: {
+                  deduplicationStore: messageDeduplicationStore,
+                  messageTypeToConfigMap: {
+                    dummy: {
+                      deduplicationWindowSeconds: 10,
+                      deduplicationKeyGenerator: dummyMessageDeduplicationKeyGenerator,
+                    },
+                    // In case there are other event types available, you can provide their deduplication strategies here
+                    // If strategy for certain message type is not provided, deduplication will not be performed for this message type
+                  },
+                },
+            })
+        }
+    }
+    ```
+
+
+
+

docker-compose.yml

@@ -25,6 +25,17 @@ services:
       - '${TMPDIR:-/tmp}/localstack:/var/log/localstack'
       - '/var/run/docker.sock:/var/run/docker.sock'
 
+  redis:
+    image: redis:6.2.7-alpine
+    command: redis-server --requirepass sOmE_sEcUrE_pAsS
+    ports:
+      - '6379:6379'
+    volumes:
+      - redis_data:/redis/data:cached
+    restart: on-failure
+
 volumes:
   rabbit_data:
     driver: local
+  redis_data:
+    driver: local

packages/core/index.ts

@@ -78,3 +78,9 @@ export {
   OFFLOADED_PAYLOAD_POINTER_PAYLOAD_SCHEMA,
   isOffloadedPayloadPointerPayload,
 } from './lib/payload-store/offloadedPayloadMessageSchemas'
+export type {
+  MessageDeduplicationConfig,
+  MessageDeduplicationMessageTypeConfig,
+  MessageDeduplicationStore,
+  MessageDeduplicationKeyGenerator,
+} from './lib/message-deduplication/messageDeduplicationTypes'

packages/core/lib/message-deduplication/messageDeduplicationSchemas.ts

@@ -0,0 +1,5 @@
+import { z } from 'zod'
+
+export const MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA = z.object({
+  deduplicationWindowSeconds: z.number().int().gt(0),
+})

packages/core/lib/message-deduplication/messageDeduplicationTypes.ts

@@ -0,0 +1,24 @@
+export interface MessageDeduplicationKeyGenerator<Message extends object = object> {
+  generate(message: Message): string
+}
+
+export interface MessageDeduplicationStore {
+  storeKey(key: string, value: string, ttlSeconds: number): Promise<void>
+  retrieveKey(key: string): Promise<string | null>
+}
+
+export type MessageDeduplicationMessageTypeConfig = {
+  /** How many seconds to keep the deduplication key in the store for a particular message type */
+  deduplicationWindowSeconds: number
+
+  /** The generator to use for generating deduplication keys for a particular message type */
+  deduplicationKeyGenerator: MessageDeduplicationKeyGenerator
+}
+
+export type MessageDeduplicationConfig = {
+  /** The store to use for storage and retrieval of deduplication keys */
+  deduplicationStore: MessageDeduplicationStore
+
+  /** The configuration for deduplication for each message type */
+  messageTypeToConfigMap: Record<string, MessageDeduplicationMessageTypeConfig>
+}

packages/core/lib/queues/AbstractPublisherManager.ts

@@ -131,6 +131,7 @@ export abstract class AbstractPublisherManager<
       this.targetToEventMap[eventTarget].push(supportedEvent)
     }
   }
+
   private registerPublishers() {
     for (const eventTarget in this.targetToEventMap) {
       if (this.targetToPublisherMap[eventTarget]) {

packages/core/lib/queues/AbstractQueueService.ts

@@ -7,8 +7,8 @@ import type { ZodSchema, ZodType } from 'zod'
 
 import type { MessageInvalidFormatError, MessageValidationError } from '../errors/Errors'
 import { jsonStreamStringifySerializer } from '../payload-store/JsonStreamStringifySerializer'
-import { OFFLOADED_PAYLOAD_POINTER_PAYLOAD_SCHEMA } from '../payload-store/offloadedPayloadMessageSchemas'
 import type { OffloadedPayloadPointerPayload } from '../payload-store/offloadedPayloadMessageSchemas'
+import { OFFLOADED_PAYLOAD_POINTER_PAYLOAD_SCHEMA } from '../payload-store/offloadedPayloadMessageSchemas'
 import type { PayloadStoreConfig } from '../payload-store/payloadStoreTypes'
 import { isDestroyable } from '../payload-store/payloadStoreTypes'
 import type { MessageProcessingResult } from '../types/MessageQueueTypes'
@@ -23,6 +23,8 @@ import { isRetryDateExceeded } from '../utils/dateUtils'
 import { streamWithKnownSizeToString } from '../utils/streamUtils'
 import { toDatePreprocessor } from '../utils/toDateProcessor'
 
+import { MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA } from '../message-deduplication/messageDeduplicationSchemas'
+import type { MessageDeduplicationConfig } from '../message-deduplication/messageDeduplicationTypes'
 import type {
   BarrierCallback,
   BarrierResult,
@@ -84,8 +86,9 @@ export abstract class AbstractQueueService<
   protected readonly deletionConfig?: DeletionConfig
   protected readonly payloadStoreConfig?: Omit<PayloadStoreConfig, 'serializer'> &
     Required<Pick<PayloadStoreConfig, 'serializer'>>
-  protected readonly _handlerSpy?: HandlerSpy<MessagePayloadSchemas>
+  protected readonly messageDeduplicationConfig?: MessageDeduplicationConfig
   protected readonly messageMetricsManager?: MessageMetricsManager<MessagePayloadSchemas>
+  protected readonly _handlerSpy?: HandlerSpy<MessagePayloadSchemas>
 
   protected isInitted: boolean
 
@@ -118,6 +121,9 @@ export abstract class AbstractQueueService<
           ...options.payloadStoreConfig,
         }
       : undefined
+    this.messageDeduplicationConfig = this.getValidateMessageDeduplicationConfig(
+      options.messageDeduplicationConfig,
+    )
 
     this.logMessages = options.logMessages ?? false
     this._handlerSpy = resolveHandlerSpy<MessagePayloadSchemas>(options)
@@ -518,4 +524,74 @@ export abstract class AbstractQueueService<
       return { error: new Error('Failed to parse serialized offloaded payload', { cause: e }) }
     }
   }
+
+  /** Checks for an existence of deduplication key in deduplication store */
+  protected async isMessageDuplicated(message: MessagePayloadSchemas): Promise<boolean> {
+    if (!this.messageDeduplicationConfig) {
+      return false
+    }
+
+    // @ts-ignore
+    const messageType = message[this.messageTypeField] as string
+
+    if (!this.messageDeduplicationConfig.messageTypeToConfigMap[messageType]) {
+      return false
+    }
+
+    const deduplicationKey =
+      this.messageDeduplicationConfig.messageTypeToConfigMap[
+        messageType
+      ].deduplicationKeyGenerator.generate(message)
+    const deduplicationValue =
+      await this.messageDeduplicationConfig.deduplicationStore.retrieveKey(deduplicationKey)
+
+    return deduplicationValue !== null
+  }
+
+  /** Stores deduplication key in deduplication store */
+  protected async deduplicateMessage(message: MessagePayloadSchemas): Promise<void> {
+    if (!this.messageDeduplicationConfig) {
+      return
+    }
+
+    // @ts-ignore
+    const messageType = message[this.messageTypeField] as string
+
+    if (!this.messageDeduplicationConfig.messageTypeToConfigMap[messageType]) {
+      return
+    }
+
+    const deduplicationKey =
+      this.messageDeduplicationConfig.messageTypeToConfigMap[
+        messageType
+      ].deduplicationKeyGenerator.generate(message)
+
+    await this.messageDeduplicationConfig.deduplicationStore.storeKey(
+      deduplicationKey,
+      new Date().toISOString(),
+      this.messageDeduplicationConfig.messageTypeToConfigMap[messageType]
+        .deduplicationWindowSeconds,
+    )
+  }
+
+  private getValidateMessageDeduplicationConfig(
+    messageDeduplicationConfig?: MessageDeduplicationConfig,
+  ): MessageDeduplicationConfig | undefined {
+    if (!messageDeduplicationConfig) {
+      return undefined
+    }
+
+    for (const messageConfig of Object.values(messageDeduplicationConfig.messageTypeToConfigMap)) {
+      const messageTypeToConfigMapParseResult =
+        MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA.safeParse(messageConfig)
+
+      if (messageTypeToConfigMapParseResult.error) {
+        throw new Error(
+          `Invalid message deduplication config provided: ${messageTypeToConfigMapParseResult.error.toString()}`,
+        )
+      }
+    }
+
+    return messageDeduplicationConfig
+  }
 }

packages/core/lib/types/queueOptionsTypes.ts

@@ -5,6 +5,7 @@ import type { PayloadStoreConfig } from '../payload-store/payloadStoreTypes'
 import type { MessageHandlerConfig } from '../queues/HandlerContainer'
 import type { HandlerSpy, HandlerSpyParams } from '../queues/HandlerSpy'
 
+import type { MessageDeduplicationConfig } from '../message-deduplication/messageDeduplicationTypes'
 import type { MessageProcessingResult, TransactionObservabilityManager } from './MessageQueueTypes'
 
 export type QueueDependencies = {
@@ -60,6 +61,7 @@ export type CommonQueueOptions = {
   logMessages?: boolean
   deletionConfig?: DeletionConfig
   payloadStoreConfig?: PayloadStoreConfig
+  messageDeduplicationConfig?: MessageDeduplicationConfig
 }
 
 export type CommonCreationConfigType = {

packages/redis-message-deduplication-store/docker-compose.yml

@@ -0,0 +1,13 @@
+services:
+  redis:
+    image: redis:6.2.7-alpine
+    command: redis-server --requirepass sOmE_sEcUrE_pAsS
+    ports:
+      - '6379:6379'
+    volumes:
+      - redis_data:/redis/data:cached
+    restart: on-failure
+
+volumes:
+  redis_data:
+    driver: local

packages/redis-message-deduplication-store/index.ts

@@ -0,0 +1 @@
+export * from './lib/RedisMessageDeduplicationStore'

packages/redis-message-deduplication-store/lib/RedisMessageDeduplicationStore.ts

@@ -0,0 +1,39 @@
+import type { MessageDeduplicationStore } from '@message-queue-toolkit/core'
+import type { Redis } from 'ioredis'
+
+export type RedisMessageDeduplicationStoreDependencies = {
+  redis: Redis
+}
+
+export type RedisMessageDeduplicationStoreConfig = {
+  keyPrefix?: string
+}
+
+export class RedisMessageDeduplicationStore implements MessageDeduplicationStore {
+  private readonly redis: Redis
+  private readonly config: RedisMessageDeduplicationStoreConfig
+
+  constructor(
+    dependencies: RedisMessageDeduplicationStoreDependencies,
+    config: RedisMessageDeduplicationStoreConfig,
+  ) {
+    this.redis = dependencies.redis
+    this.config = config
+  }
+
+  async storeKey(key: string, value: string, ttlSeconds: number): Promise<void> {
+    const keyWithPrefix = this.getKeyWithOptionalPrefix(key)
+
+    await this.redis.set(keyWithPrefix, value, 'EX', ttlSeconds)
+  }
+
+  retrieveKey(key: string): Promise<string | null> {
+    const keyWithPrefix = this.getKeyWithOptionalPrefix(key)
+
+    return this.redis.get(keyWithPrefix)
+  }
+
+  private getKeyWithOptionalPrefix(key: string): string {
+    return this.config?.keyPrefix?.length ? `${this.config.keyPrefix}:${key}` : key
+  }
+}

packages/redis-message-deduplication-store/package.json

@@ -0,0 +1,47 @@
+{
+  "name": "@message-queue-toolkit/redis-message-deduplication-store",
+  "version": "1.0.0",
+  "private": false,
+  "license": "MIT",
+  "description": "Redis-based message deduplication store for message-queue-toolkit",
+  "maintainers": [
+    {
+      "name": "Igor Savin",
+      "email": "[email protected]"
+    }
+  ],
+  "main": "dist/index.js",
+  "types": "dist/index.d.ts",
+  "scripts": {
+    "build": "del-cli dist && tsc",
+    "build:release": "del-cli dist && del-cli coverage && npm run lint && tsc --project tsconfig.release.json",
+    "test": "vitest",
+    "test:coverage": "npm test -- --coverage",
+    "test:ci": "npm run docker:start:dev && npm run test:coverage && npm run docker:stop:dev",
+    "lint": "biome check . && tsc --project tsconfig.json --noEmit",
+    "lint:fix": "biome check --write .",
+    "docker:start:dev": "docker compose up -d",
+    "docker:stop:dev": "docker compose down",
+    "prepublishOnly": "npm run build:release"
+  },
+  "dependencies": {},
+  "peerDependencies": {
+    "ioredis": "^5.3.2"
+  },
+  "devDependencies": {
+    "@biomejs/biome": "1.9.4",
+    "@kibertoad/biome-config": "^1.2.1",
+    "@types/node": "^22.0.0",
+    "@vitest/coverage-v8": "^2.0.4",
+    "del-cli": "^6.0.0",
+    "typescript": "^5.5.3",
+    "vitest": "^2.0.4"
+  },
+  "homepage": "https://github.com/kibertoad/message-queue-toolkit",
+  "repository": {
+    "type": "git",
+    "url": "git://github.com/kibertoad/message-queue-toolkit.git"
+  },
+  "keywords": ["message", "queue", "queues", "utils", "redis", "ioredis", "deduplication", "store"],
+  "files": ["README.md", "LICENSE", "dist/*"]
+}