Implement polling for topics

README.md

@@ -244,6 +244,104 @@ 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.
 
+## Startup Resource 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 `startupResourcePolling` option within `locatorConfig` 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 `startupResourcePolling`, both services can start simultaneously and wait for the other's resources to appear.
+
+### Configuration
+
+```typescript
+import { NO_TIMEOUT } from '@message-queue-toolkit/core'
+
+const consumer = new MySnsSqsConsumer(dependencies, {
+  locatorConfig: {
+    topicArn: 'arn:aws:sns:...',
+    queueUrl: 'https://sqs...',
+    subscriptionArn: '...',
+    // Enable eventual consistency mode
+    startupResourcePolling: {
+      enabled: true,           // Enable polling for resource availability
+      pollingIntervalMs: 5000, // Check every 5 seconds (default: 5000)
+      timeoutMs: 300000,       // Fail after 5 minutes (required)
+    }
+  }
+})
+```
+
+### Configuration Options
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `enabled` | `boolean` | - | Must be set to `true` to enable polling |
+| `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) |
+| `timeoutMs` | `number \| NO_TIMEOUT` | - (required) | Maximum wait time before throwing `StartupResourcePollingTimeoutError`. Use `NO_TIMEOUT` to poll indefinitely |
+
+### Environment-Specific Configuration
+
+```typescript
+import { NO_TIMEOUT } from '@message-queue-toolkit/core'
+
+// Production - with timeout
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000, // 5 minutes
+    }
+  }
+}
+
+// Development/Staging - poll indefinitely
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: NO_TIMEOUT,
+    }
+  }
+}
+
+// Custom timeout and interval
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      pollingIntervalMs: 10000,
+      timeoutMs: 10 * 60 * 1000, // 10 minutes
+    }
+  }
+}
+```
+
+### Error Handling
+
+When `timeoutMs` is configured and the resource doesn't become available within the specified time, a `StartupResourcePollingTimeoutError` is thrown:
+
+```typescript
+import { StartupResourcePollingTimeoutError } from '@message-queue-toolkit/core'
+
+try {
+  await consumer.init()
+} catch (error) {
+  if (error instanceof StartupResourcePollingTimeoutError) {
+    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 {
+  isStartupResourcePollingEnabled,
+  type StartupResourcePollingCheckResult,
+  StartupResourcePollingTimeoutError,
+  type WaitForResourceOptions,
+  waitForResource,
+} from './utils/startupResourcePollingUtils.ts'
 export { toDatePreprocessor } from './utils/toDateProcessor.ts'
 export { waitAndRetry } from './utils/waitUtils.ts'

packages/core/lib/types/queueOptionsTypes.ts

@@ -130,6 +130,104 @@ export type DeletionConfig = {
   forceDeleteInProduction?: boolean
 }
 
+/**
+ * Symbol to indicate no timeout - polling will continue indefinitely.
+ * Use this for dev/staging environments where you want to wait indefinitely for resources.
+ *
+ * @example
+ * {
+ *   locatorConfig: {
+ *     queueUrl: '...',
+ *     startupResourcePolling: {
+ *       enabled: true,
+ *       timeoutMs: NO_TIMEOUT,
+ *     }
+ *   }
+ * }
+ */
+export const NO_TIMEOUT = Symbol('NO_TIMEOUT')
+
+/**
+ * Configuration for startup resource polling 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
+ *
+ * When `startupResourcePolling` is provided with `enabled: true` inside `locatorConfig`,
+ * the consumer will poll for the topic/queue to become available instead of failing immediately.
+ *
+ * @example
+ * // Enable with 5 minute timeout
+ * {
+ *   locatorConfig: {
+ *     queueUrl: '...',
+ *     startupResourcePolling: {
+ *       enabled: true,
+ *       timeoutMs: 5 * 60 * 1000,
+ *     }
+ *   }
+ * }
+ *
+ * @example
+ * // Poll indefinitely (useful for dev/staging environments)
+ * {
+ *   locatorConfig: {
+ *     queueUrl: '...',
+ *     startupResourcePolling: {
+ *       enabled: true,
+ *       timeoutMs: NO_TIMEOUT,
+ *     }
+ *   }
+ * }
+ *
+ * @example
+ * // Custom timeout and interval
+ * {
+ *   locatorConfig: {
+ *     queueUrl: '...',
+ *     startupResourcePolling: {
+ *       enabled: true,
+ *       timeoutMs: 10 * 60 * 1000, // 10 minutes
+ *       pollingIntervalMs: 10000,
+ *     }
+ *   }
+ * }
+ */
+export type StartupResourcePollingConfig = {
+  /**
+   * Controls whether polling is enabled.
+   * Must be set to true to enable polling.
+   */
+  enabled?: boolean
+
+  /**
+   * Maximum time in milliseconds to wait for the resource to become available.
+   * Use `NO_TIMEOUT` to disable timeout and poll indefinitely.
+   */
+  timeoutMs: number | typeof NO_TIMEOUT
+
+  /**
+   * Interval in milliseconds between polling attempts.
+   * Default: 5000 (5 seconds)
+   */
+  pollingIntervalMs?: number
+}
+
+/**
+ * Base type for queue locator configurations that includes startup resource polling.
+ * Protocol-specific locator types should extend this.
+ */
+export type BaseQueueLocatorType = {
+  /**
+   * Configuration for startup resource polling mode.
+   * When enabled, the consumer will poll for the resource to become available
+   * instead of failing immediately.
+   */
+  startupResourcePolling?: StartupResourcePollingConfig
+}
+
 type NewQueueOptions<CreationConfigType extends CommonCreationConfigType> = {
   creationConfig?: CreationConfigType
 }

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

@@ -0,0 +1,172 @@
+import { describe, expect, it, vi } from 'vitest'
+import { NO_TIMEOUT } from '../types/queueOptionsTypes.ts'
+import {
+  isStartupResourcePollingEnabled,
+  StartupResourcePollingTimeoutError,
+  waitForResource,
+} from './startupResourcePollingUtils.ts'
+
+describe('startupResourcePollingUtils', () => {
+  describe('isStartupResourcePollingEnabled', () => {
+    it('returns true when enabled is true', () => {
+      expect(isStartupResourcePollingEnabled({ enabled: true, timeoutMs: 5000 })).toBe(true)
+    })
+
+    it('returns false when enabled is not specified', () => {
+      expect(isStartupResourcePollingEnabled({ timeoutMs: 5000 })).toBe(false)
+      expect(isStartupResourcePollingEnabled({ pollingIntervalMs: 1000, timeoutMs: 5000 })).toBe(
+        false,
+      )
+    })
+
+    it('returns false when enabled is false', () => {
+      expect(isStartupResourcePollingEnabled({ enabled: false, timeoutMs: 5000 })).toBe(false)
+    })
+
+    it('returns false when config is undefined', () => {
+      expect(isStartupResourcePollingEnabled(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, timeoutMs: 5000 },
+        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, timeoutMs: 5000 },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+      expect(checkFn).toHaveBeenCalledTimes(3)
+    })
+
+    it('throws StartupResourcePollingTimeoutError 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(StartupResourcePollingTimeoutError)
+
+      // 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, timeoutMs: 5000 },
+          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, timeoutMs: 5000 },
+        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, timeoutMs: 5000 },
+        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 NO_TIMEOUT is used', 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: NO_TIMEOUT,
+        },
+        checkFn,
+        resourceName: 'test-resource',
+      })
+
+      expect(result).toBe('test-result')
+      expect(checkFn).toHaveBeenCalledTimes(5)
+    })
+  })
+
+  describe('StartupResourcePollingTimeoutError', () => {
+    it('includes resource name and timeout in message', () => {
+      const error = new StartupResourcePollingTimeoutError('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('StartupResourcePollingTimeoutError')
+    })
+  })
+})