Message Queue Processor

Build a TypeScript message queue processor that demonstrates understanding of at-least-once delivery semantics and message deduplication patterns.

Background

In distributed messaging systems, at-least-once delivery guarantees that messages are delivered one or more times but never zero times. This requires proper message tracking, acknowledgment handling, and deduplication to prevent duplicate processing.

Requirements

Implement a message queue processor with the following features:

Queue Management

Your implementation should maintain a queue that:

• Processes messages sequentially (FIFO order)
• Tracks whether the queue is actively processing (states: IDLE, ACTIVE)
• Ensures messages are not lost even if processing fails

Message Processing

Each message should be:

• Assigned a unique identifier
• Tracked until successfully acknowledged
• Requeued if processing fails or times out
• Deduplicated to prevent redundant processing of already-handled messages

Deduplication

The system must:

• Track which messages have been successfully processed
• Reject attempts to process duplicate messages
• Provide a way to check if a message has already been processed

Core Operations

Your queue processor should support:

• Enqueue: Add messages to the queue for processing
• Process: Execute the next pending message with a provided handler function
• Acknowledge: Mark a message as successfully processed
• GetState: Return current queue state (IDLE or ACTIVE)
• IsProcessed: Check if a message ID has been processed

Test Cases

• When a message is enqueued and processed successfully, it should be marked as processed and not reprocessed @test
• When a message processing fails without acknowledgment, it should remain in the queue for retry @test
• When a duplicate message ID is enqueued after successful processing, it should be rejected @test
• The queue state should be IDLE when no messages are being processed and ACTIVE during processing @test

Implementation

@generates

API

export type QueueState = 'IDLE' | 'ACTIVE';

export interface Message<T = any> {
  id: string;
  payload: T;
}

export interface MessageHandler<T = any> {
  (message: Message<T>): Promise<void>;
}

export class QueueProcessor<T = any> {
  /**
   * Get the current state of the queue processor
   */
  getState(): QueueState;

  /**
   * Add a message to the queue
   * Returns false if the message ID has already been processed
   */
  enqueue(message: Message<T>): boolean;

  /**
   * Process the next message in the queue using the provided handler
   * Returns true if a message was processed, false if queue is empty
   */
  process(handler: MessageHandler<T>): Promise<boolean>;

  /**
   * Acknowledge successful processing of a message
   * This marks the message as processed and enables deduplication
   */
  acknowledge(messageId: string): void;

  /**
   * Check if a message has been successfully processed
   */
  isProcessed(messageId: string): boolean;
}

Dependencies { .dependencies }

@walletconnect/types { .dependency }

Provides type definitions for WalletConnect protocol, including queue management patterns and at-least-once delivery semantics.