feat: use redis-semaphore for lock management, refactor

Author: Drodevbar

README.md

@@ -180,7 +180,7 @@ export abstract class AbstractAmqpConsumer<
           // requeue the message if maxRetryDuration is not exceeded, else ack it to avoid infinite loop
           if (this.shouldBeRetried(originalMessage, this.maxRetryDuration)) {
             // TODO: Add retry delay + republish message updating internal properties
-            this.queueMessageForRetry(originalMessage)
+            this.channel.nack(message as Message, false, true)
             this.handleMessageProcessed({
               message: parsedMessage,
               processingResult: 'retryLater',
@@ -295,10 +295,6 @@ export abstract class AbstractAmqpConsumer<
     )
   }
 
-  protected override queueMessageForRetry(message: MessagePayloadType): void {
-    this.channel.nack(message as Message, false, true)
-  }
-
   private deserializeMessage(
     message: Message,
   ): Either<'abort', ParseMessageResult<MessagePayloadType>> {

packages/amqp/lib/AbstractAmqpConsumer.ts

@@ -157,9 +157,5 @@ export abstract class AbstractAmqpPublisher<
   override processMessage(): Promise<Either<'retryLater', 'success'>> {
     throw new Error('Not implemented for publisher')
   }
-
-  protected override queueMessageForRetry(): Promise<void> {
-    throw new Error('Not implemented for publisher')
-  }
   /* c8 ignore stop */
 }

packages/amqp/lib/AbstractAmqpPublisher.ts

@@ -156,6 +156,7 @@ describe('AmqpPermissionConsumer', () => {
           messageId: '1',
           messageType: 'add',
           messageDeduplicationId: undefined,
+          messageDeduplicationWindowSeconds: undefined,
           processingResult: 'consumed',
           queueName: AmqpPermissionConsumer.QUEUE_NAME,
           messageTimestamp: expect.any(Number),

packages/amqp/test/consumers/AmqpPermissionConsumer.spec.ts

@@ -79,10 +79,10 @@ export {
   isOffloadedPayloadPointerPayload,
 } from './lib/payload-store/offloadedPayloadMessageSchemas'
 export {
-  type PublisherMessageDeduplicationMessageTypeConfig,
-  type PublisherMessageDeduplicationStore,
+  type MessageDeduplicationStore,
   type MessageDeduplicationConfig,
-  type ConsumerMessageDeduplicationMessageTypeConfig,
-  type ConsumerMessageDeduplicationStore,
-  ConsumerMessageDeduplicationKeyStatus,
+  type ReleasableLock,
+  DeduplicationRequester,
+  AcquireLockTimeoutError,
+  noopReleasableLock,
 } from './lib/message-deduplication/messageDeduplicationTypes'

packages/core/index.ts

@@ -1,35 +1,7 @@
 import { z } from 'zod'
 
-// Private interface to provide JSDoc feature
-interface PublisherMessageDeduplicationMessageTypeInternal {
-  /** How many seconds to keep the deduplication key in the store for a particular message type */
-  deduplicationWindowSeconds: number
-}
-
-export const PUBLISHER_MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA = z.object({
-  deduplicationWindowSeconds: z.number().int().gt(0),
-})
-
-export type PublisherMessageDeduplicationMessageType = z.infer<
-  typeof PUBLISHER_MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA
-> &
-  PublisherMessageDeduplicationMessageTypeInternal
-
-// Private interface to provide JSDoc feature
-interface ConsumerMessageDeduplicationMessageTypeInternal {
-  /** How many seconds to keep the deduplication key in the store for a particular message type after message is successfully processed */
-  deduplicationWindowSeconds: number
-
-  /** How many seconds it is expected to take to process a message of a particular type */
-  maximumProcessingTimeSeconds: number
-}
-
-export const CONSUMER_MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA =
-  PUBLISHER_MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA.extend({
-    maximumProcessingTimeSeconds: z.number().int().gt(0),
-  })
-
-export type ConsumerMessageDeduplicationMessageType = z.infer<
-  typeof CONSUMER_MESSAGE_DEDUPLICATION_MESSAGE_TYPE_SCHEMA
-> &
-  ConsumerMessageDeduplicationMessageTypeInternal
+export const MESSAGE_DEDUPLICATION_WINDOW_SECONDS_SCHEMA = z
+  .number()
+  .int()
+  .gt(0)
+  .describe('message deduplication window in seconds')

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

@@ -1,57 +1,48 @@
-import type {
-  ConsumerMessageDeduplicationMessageType,
-  PublisherMessageDeduplicationMessageType,
-} from './messageDeduplicationSchemas'
+import type { Either } from '@lokalise/node-core'
 
-export interface PublisherMessageDeduplicationStore {
+export interface ReleasableLock {
+  release(): Promise<void>
+}
+
+export class AcquireLockTimeoutError extends Error {}
+
+export interface MessageDeduplicationStore {
   /**
    * Stores a deduplication key in case it does not already exist.
    * @param {string} key - deduplication key
    * @param {string} value - value to store
    * @param {number} ttlSeconds - time to live in seconds
-   * @returns {boolean} - true if the key was stored, false if it already existed
+   * @returns {Promise<boolean>} - true if the key was stored, false if it already existed
    */
   setIfNotExists(key: string, value: string, ttlSeconds: number): Promise<boolean>
 
-  /** Retrieves value associated with deduplication key */
-  getByKey(key: string): Promise<string | null>
-}
-
-export type PublisherMessageDeduplicationMessageTypeConfig =
-  PublisherMessageDeduplicationMessageType
-
-export interface ConsumerMessageDeduplicationStore extends PublisherMessageDeduplicationStore {
   /**
-   * Retrieves TTL of the deduplication key
-   *
+   * Acquires locks for a given key
    * @param {string} key - deduplication key
-   * @returns {number|null} - TTL of the deduplication key in seconds or null if the key does not exist
+   * @returns {Promise<Either<AcquireLockTimeoutError | Error, ReleasableLock>>} - a promise that resolves to a ReleasableLock if the lock was acquired, AcquireLockTimeoutError error if the lock could not be acquired due to timeout, or an Error if the lock could not be acquired for another reason
    */
-  getKeyTtl(key: string): Promise<number | null>
-
-  /** Sets a value for the deduplication key or updates it if it already exists */
-  setOrUpdate(key: string, value: string, ttlSeconds: number): Promise<void>
+  acquireLock(key: string): Promise<Either<AcquireLockTimeoutError | Error, ReleasableLock>>
 
-  /** Deletes the deduplication key */
-  deleteKey(key: string): Promise<void>
+  /**
+   * Checks if a deduplication key exists in the store
+   * @param {string} key - deduplication key
+   * @returns {Promise<boolean>} - true if the key exists, false otherwise
+   */
+  keyExists(key: string): Promise<boolean>
 }
 
-export type ConsumerMessageDeduplicationMessageTypeConfig = ConsumerMessageDeduplicationMessageType
-
-export type MessageDeduplicationConfig<
-  TStore extends ConsumerMessageDeduplicationStore | PublisherMessageDeduplicationStore,
-  TConfig extends
-    | ConsumerMessageDeduplicationMessageTypeConfig
-    | PublisherMessageDeduplicationMessageTypeConfig,
-> = {
+export type MessageDeduplicationConfig = {
   /** The store to use for storage and retrieval of deduplication keys */
-  deduplicationStore: TStore
+  deduplicationStore: MessageDeduplicationStore
+}
 
-  /** The configuration for deduplication for each message type */
-  messageTypeToConfigMap: Record<string, TConfig>
+export enum DeduplicationRequester {
+  Consumer = 'consumer',
+  Publisher = 'publisher',
 }
 
-export enum ConsumerMessageDeduplicationKeyStatus {
-  PROCESSING = 'PROCESSING',
-  PROCESSED = 'PROCESSED',
+export const DEFAULT_DEDUPLICATION_WINDOW_SECONDS = 10
+
+export const noopReleasableLock: ReleasableLock = {
+  release: async () => {},
 }

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

@@ -9,7 +9,7 @@ export const OFFLOADED_PAYLOAD_POINTER_PAYLOAD_SCHEMA = z
     offloadedPayloadPointer: z.string().min(1),
     offloadedPayloadSize: z.number().int().positive(),
   })
-  // Pass-through allows to pass message ID, type and timestamp that are using dynamic keys.
+  // Pass-through allows to pass message ID, type, timestamp and message-deduplication-related fields that are using dynamic keys.
   .passthrough()
 
 export type OffloadedPayloadPointerPayload = z.infer<