Implement polling for topics

Author: kibertoad

README.md

@@ -244,6 +244,83 @@ Both publishers and consumers accept a queue name and configuration as parameter
 
 If you do not want to create a new queue/topic, you can set `queueLocator` field for `queueConfiguration`. In that case `message-queue-toolkit` will not attempt to create a new queue or topic, and instead throw an error if they don't already exist.
 
+## Resource Availability Polling (Eventual Consistency Mode)
+
+When using `locatorConfig` to reference existing queues or topics, the default behavior is to fail immediately if the resource doesn't exist. However, in some deployment scenarios (especially with cross-service dependencies), resources may not be available at startup time.
+
+The `resourceAvailabilityConfig` option enables "eventual consistency mode" where consumers poll for resources to become available instead of failing immediately.
+
+### Use Case: Cross-Service Dependencies
+
+Consider two services that need to subscribe to each other's topics:
+- **Service A** needs to subscribe to **Service B's** topic
+- **Service B** needs to subscribe to **Service A's** topic
+
+Without eventual consistency mode, neither service can deploy first because the other's topic doesn't exist yet. With `resourceAvailabilityConfig`, both services can start simultaneously and wait for the other's resources to appear.
+
+### Configuration
+
+```typescript
+const consumer = new MySnsSqsConsumer(dependencies, {
+  locatorConfig: {
+    topicArn: 'arn:aws:sns:...',
+    queueUrl: 'https://sqs...',
+    subscriptionArn: '...'
+  },
+  // Enable eventual consistency mode
+  resourceAvailabilityConfig: {
+    enabled: true,           // Enable polling for resource availability
+    pollingIntervalMs: 5000, // Check every 5 seconds (default: 5000)
+    timeoutMs: 300000,       // Fail after 5 minutes (optional, default: no timeout)
+  }
+})
+```
+
+### Configuration Options
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `enabled` | `boolean` | `false` | Enable eventual consistency mode |
+| `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) |
+| `timeoutMs` | `number` | `undefined` | Maximum wait time before throwing `ResourceAvailabilityTimeoutError`. If not set, polls indefinitely |
+
+### Environment-Specific Configuration
+
+```typescript
+// Development/Staging - poll indefinitely
+{
+  resourceAvailabilityConfig: {
+    enabled: true,
+    pollingIntervalMs: 5000,
+  }
+}
+
+// Production - poll with timeout to catch misconfigurations
+{
+  resourceAvailabilityConfig: {
+    enabled: true,
+    pollingIntervalMs: 10000,
+    timeoutMs: 5 * 60 * 1000, // 5 minutes
+  }
+}
+```
+
+### Error Handling
+
+When `timeoutMs` is configured and the resource doesn't become available within the specified time, a `ResourceAvailabilityTimeoutError` is thrown:
+
+```typescript
+import { ResourceAvailabilityTimeoutError } from '@message-queue-toolkit/core'
+
+try {
+  await consumer.init()
+} catch (error) {
+  if (error instanceof ResourceAvailabilityTimeoutError) {
+    console.error(`Resource ${error.resourceName} not available after ${error.timeoutMs}ms`)
+  }
+}
+```
+
 ## SQS Policy Configuration
 
 SQS queues can be configured with access policies to control who can send messages to and receive messages from the queue. The `policyConfig` parameter allows you to define these policies when creating or updating SQS queues.

packages/core/lib/index.ts

@@ -105,5 +105,12 @@ export { isProduction, reloadConfig } from './utils/envUtils.ts'
 export { isShallowSubset, objectMatches } from './utils/matchUtils.ts'
 export { type ParseMessageResult, parseMessage } from './utils/parseUtils.ts'
 export { objectToBuffer } from './utils/queueUtils.ts'
+export {
+  isResourceAvailabilityWaitingEnabled,
+  type ResourceAvailabilityCheckResult,
+  ResourceAvailabilityTimeoutError,
+  type WaitForResourceOptions,
+  waitForResource,
+} from './utils/resourceAvailabilityUtils.ts'
 export { toDatePreprocessor } from './utils/toDateProcessor.ts'
 export { waitAndRetry } from './utils/waitUtils.ts'

packages/core/lib/queues/AbstractQueueService.ts

@@ -46,6 +46,7 @@ import type {
   ProcessedMessageMetadata,
   QueueDependencies,
   QueueOptions,
+  ResourceAvailabilityConfig,
 } from '../types/queueOptionsTypes.ts'
 import { isRetryDateExceeded } from '../utils/dateUtils.ts'
 import { streamWithKnownSizeToString } from '../utils/streamUtils.ts'
@@ -125,6 +126,7 @@ export abstract class AbstractQueueService<
   protected readonly creationConfig?: QueueConfiguration
   protected readonly locatorConfig?: QueueLocatorType
   protected readonly deletionConfig?: DeletionConfig
+  protected readonly resourceAvailabilityConfig?: ResourceAvailabilityConfig
   protected readonly payloadStoreConfig?:
     | MakeRequired<SinglePayloadStoreConfig, 'serializer'>
     | MakeRequired<MultiPayloadStoreConfig, 'serializer'>
@@ -160,6 +162,7 @@ export abstract class AbstractQueueService<
     this.creationConfig = options.creationConfig
     this.locatorConfig = options.locatorConfig
     this.deletionConfig = options.deletionConfig
+    this.resourceAvailabilityConfig = options.resourceAvailabilityConfig
     this.payloadStoreConfig = options.payloadStoreConfig
       ? {
           serializer: jsonStreamStringifySerializer,

packages/core/lib/types/queueOptionsTypes.ts

@@ -118,6 +118,13 @@ export type CommonQueueOptions = {
   deletionConfig?: DeletionConfig
   payloadStoreConfig?: PayloadStoreConfig
   messageDeduplicationConfig?: MessageDeduplicationConfig
+  /**
+   * Configuration for eventual consistency mode.
+   * When enabled, the consumer will poll for the topic/queue to become available
+   * instead of failing immediately when using locatorConfig.
+   * This is useful for handling cross-service dependencies during deployment.
+   */
+  resourceAvailabilityConfig?: ResourceAvailabilityConfig
 }
 
 export type CommonCreationConfigType = {
@@ -130,6 +137,59 @@ export type DeletionConfig = {
   forceDeleteInProduction?: boolean
 }
 
+/**
+ * Configuration for eventual consistency mode when resources may not exist at startup.
+ *
+ * This is useful in scenarios where services have cross-dependencies:
+ * - Service A needs to subscribe to Service B's topic
+ * - Service B needs to subscribe to Service A's topic
+ * - Neither can deploy first without the other's topic existing
+ *
+ * With `resourceAvailabilityConfig.enabled = true`, the consumer will poll for the
+ * topic/queue to become available instead of failing immediately.
+ *
+ * @example
+ * // Development/staging - poll indefinitely
+ * {
+ *   resourceAvailabilityConfig: {
+ *     enabled: true,
+ *     pollingIntervalMs: 5000,
+ *   }
+ * }
+ *
+ * @example
+ * // Production - poll with timeout to catch misconfigurations
+ * {
+ *   resourceAvailabilityConfig: {
+ *     enabled: true,
+ *     timeoutMs: 5 * 60 * 1000, // 5 minutes
+ *     pollingIntervalMs: 10000,
+ *   }
+ * }
+ */
+export type ResourceAvailabilityConfig = {
+  /**
+   * If true, the consumer will poll for the topic/queue to become available
+   * instead of failing immediately when using locatorConfig.
+   * Default: false (fail immediately for backwards compatibility)
+   */
+  enabled: boolean
+
+  /**
+   * Maximum time in milliseconds to wait for the resource to become available.
+   * If not set or set to 0, will poll indefinitely (useful for dev/staging environments).
+   * For production, it's recommended to set a reasonable timeout (e.g., 5 minutes).
+   * Default: undefined (no timeout - poll indefinitely)
+   */
+  timeoutMs?: number
+
+  /**
+   * Interval in milliseconds between polling attempts.
+   * Default: 5000 (5 seconds)
+   */
+  pollingIntervalMs?: number
+}
+
 type NewQueueOptions<CreationConfigType extends CommonCreationConfigType> = {
   creationConfig?: CreationConfigType
 }

packages/core/lib/utils/resourceAvailabilityUtils.spec.ts

@@ -0,0 +1,180 @@
+import { describe, expect, it, vi } from 'vitest'
+import {
+  isResourceAvailabilityWaitingEnabled,
+  ResourceAvailabilityTimeoutError,
+  waitForResource,
+} from './resourceAvailabilityUtils.ts'
+
+describe('resourceAvailabilityUtils', () => {
+  describe('isResourceAvailabilityWaitingEnabled', () => {
+    it('returns true when enabled is true', () => {
+      expect(isResourceAvailabilityWaitingEnabled({ enabled: true })).toBe(true)
+    })
+
+    it('returns false when enabled is false', () => {
+      expect(isResourceAvailabilityWaitingEnabled({ enabled: false })).toBe(false)
+    })
+
+    it('returns false when config is undefined', () => {
+      expect(isResourceAvailabilityWaitingEnabled(undefined)).toBe(false)
+    })
+  })
+
+  describe('waitForResource', () => {
+    it('returns immediately when resource is available on first check', async () => {
+      const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' })
+
+      const result = await waitForResource({
+        config: { enabled: true, pollingIntervalMs: 100 },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+      expect(checkFn).toHaveBeenCalledTimes(1)
+    })
+
+    it('polls until resource becomes available', async () => {
+      let callCount = 0
+      const checkFn = vi.fn().mockImplementation(() => {
+        callCount++
+        if (callCount < 3) {
+          return Promise.resolve({ isAvailable: false })
+        }
+        return Promise.resolve({ isAvailable: true, result: 'test-result' })
+      })
+
+      const result = await waitForResource({
+        config: { enabled: true, pollingIntervalMs: 10 },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+      expect(checkFn).toHaveBeenCalledTimes(3)
+    })
+
+    it('throws ResourceAvailabilityTimeoutError when timeout is reached', async () => {
+      const checkFn = vi.fn().mockResolvedValue({ isAvailable: false })
+
+      await expect(
+        waitForResource({
+          config: { enabled: true, pollingIntervalMs: 10, timeoutMs: 50 },
+          checkFn,
+          resourceName: 'test-resource',
+        }),
+      ).rejects.toThrow(ResourceAvailabilityTimeoutError)
+
+      // Should have made at least a few attempts
+      expect(checkFn.mock.calls.length).toBeGreaterThan(0)
+    })
+
+    it('throws the original error when checkFn throws', async () => {
+      const checkFn = vi.fn().mockRejectedValue(new Error('Unexpected error'))
+
+      await expect(
+        waitForResource({
+          config: { enabled: true, pollingIntervalMs: 10 },
+          checkFn,
+          resourceName: 'test-resource',
+        }),
+      ).rejects.toThrow('Unexpected error')
+
+      expect(checkFn).toHaveBeenCalledTimes(1)
+    })
+
+    it('uses default polling interval when not specified', async () => {
+      const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' })
+
+      const result = await waitForResource({
+        config: { enabled: true },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+    })
+
+    it('logs progress when logger is provided', async () => {
+      const logger = {
+        info: vi.fn(),
+        debug: vi.fn(),
+        error: vi.fn(),
+      }
+      const checkFn = vi.fn().mockResolvedValue({ isAvailable: true, result: 'test-result' })
+
+      await waitForResource({
+        config: { enabled: true, pollingIntervalMs: 10 },
+        checkFn,
+        resourceName: 'test-resource',
+        // @ts-expect-error - partial logger for testing
+        logger,
+      })
+
+      expect(logger.info).toHaveBeenCalledWith(
+        expect.objectContaining({
+          message: expect.stringContaining('Waiting for resource'),
+          resourceName: 'test-resource',
+        }),
+      )
+      expect(logger.info).toHaveBeenCalledWith(
+        expect.objectContaining({
+          message: expect.stringContaining('is now available'),
+          resourceName: 'test-resource',
+        }),
+      )
+    })
+
+    it('polls indefinitely when timeoutMs is not set', async () => {
+      let callCount = 0
+      const checkFn = vi.fn().mockImplementation(() => {
+        callCount++
+        if (callCount < 10) {
+          return Promise.resolve({ isAvailable: false })
+        }
+        return Promise.resolve({ isAvailable: true, result: 'test-result' })
+      })
+
+      const result = await waitForResource({
+        config: { enabled: true, pollingIntervalMs: 1 },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+      expect(checkFn).toHaveBeenCalledTimes(10)
+    })
+
+    it('polls indefinitely when timeoutMs is 0', async () => {
+      let callCount = 0
+      const checkFn = vi.fn().mockImplementation(() => {
+        callCount++
+        if (callCount < 5) {
+          return Promise.resolve({ isAvailable: false })
+        }
+        return Promise.resolve({ isAvailable: true, result: 'test-result' })
+      })
+
+      const result = await waitForResource({
+        config: { enabled: true, pollingIntervalMs: 1, timeoutMs: 0 },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+      expect(checkFn).toHaveBeenCalledTimes(5)
+    })
+  })
+
+  describe('ResourceAvailabilityTimeoutError', () => {
+    it('includes resource name and timeout in message', () => {
+      const error = new ResourceAvailabilityTimeoutError('my-queue', 5000)
+
+      expect(error.message).toContain('my-queue')
+      expect(error.message).toContain('5000')
+      expect(error.resourceName).toBe('my-queue')
+      expect(error.timeoutMs).toBe(5000)
+      expect(error.name).toBe('ResourceAvailabilityTimeoutError')
+    })
+  })
+})