Implement polling for topics (#388)

Author: kibertoad

README.md

@@ -244,6 +244,206 @@ 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 |
+| `throwOnTimeout` | `boolean` | `true` | When `true`, throws error on timeout. When `false`, reports error via errorReporter, resets timeout counter, and continues polling |
+| `nonBlocking` | `boolean` | `false` | When `true`, `init()` returns immediately if resource is not available, and polling continues in the background. A callback is invoked when the resource becomes available |
+
+### Environment-Specific Configuration
+
+```typescript
+import { NO_TIMEOUT } from '@message-queue-toolkit/core'
+
+// Production - with timeout (throws error when timeout reached)
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000, // 5 minutes
+    }
+  }
+}
+
+// Production - report timeout but keep trying
+// Useful when you want visibility into prolonged unavailability without failing
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000, // Report every 5 minutes
+      throwOnTimeout: false,    // Don't throw, just report and continue
+    }
+  }
+}
+
+// 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
+    }
+  }
+}
+
+// Non-blocking mode - service starts immediately, polling continues in background
+// Useful when you want the service to be available even if dependencies are not ready
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000,
+      nonBlocking: true,  // init() resolves immediately
+    }
+  }
+}
+```
+
+### Non-Blocking Mode
+
+When `nonBlocking: true` is set, `init()` returns immediately if the resource is not available on the first check. Polling continues in the background, and you can use callbacks to be notified when resources become available.
+
+**Important behaviors:**
+- If the resource IS immediately available, `init()` returns normally with the resource ready
+- If the resource is NOT immediately available, `init()` returns with `resourcesReady: false` (for SNS/SQS) or `queueArn: undefined` (for SQS)
+- Background polling continues and invokes the callback when the resource becomes available
+- For SNS+SQS consumers, the callback is only invoked when BOTH topic and queue are available
+
+```typescript
+// SQS non-blocking with callback
+import { initSqs } from '@message-queue-toolkit/sqs'
+
+const result = await initSqs(
+  sqsClient,
+  {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000,
+      nonBlocking: true,
+    },
+  },
+  undefined,
+  undefined,
+  {
+    onQueueReady: ({ queueArn }) => {
+      console.log(`Queue is now available: ${queueArn}`)
+      // Start consuming messages, update health checks, etc.
+    },
+  },
+)
+
+if (result.queueArn) {
+  console.log('Queue was immediately available')
+} else {
+  console.log('Queue not yet available, will be notified via callback')
+}
+
+// SNS+SQS non-blocking with callback
+import { initSnsSqs } from '@message-queue-toolkit/sns'
+
+const result = await initSnsSqs(
+  sqsClient,
+  snsClient,
+  stsClient,
+  {
+    topicArn: '...',
+    queueUrl: '...',
+    subscriptionArn: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000,
+      nonBlocking: true,
+    },
+  },
+  undefined,
+  undefined,
+  {
+    onResourcesReady: ({ topicArn, queueUrl }) => {
+      // Called only when BOTH topic and queue are available
+      console.log(`Resources ready: topic=${topicArn}, queue=${queueUrl}`)
+    },
+  },
+)
+
+if (result.resourcesReady) {
+  console.log('All resources were immediately available')
+} else {
+  console.log('Resources not yet available, will be notified via callback')
+}
+```
+
+### 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/MessageQueueTypes.ts

@@ -1,4 +1,8 @@
-import type { CommonLogger, TransactionObservabilityManager } from '@lokalise/node-core'
+import type {
+  CommonLogger,
+  ErrorReporter,
+  TransactionObservabilityManager,
+} from '@lokalise/node-core'
 import type { ZodSchema } from 'zod/v4'
 
 import type { PublicHandlerSpy } from '../queues/HandlerSpy.ts'
@@ -38,6 +42,7 @@ export type { TransactionObservabilityManager }
 
 export type ExtraParams = {
   logger?: CommonLogger
+  errorReporter?: ErrorReporter
 }
 
 export type SchemaMap<SupportedMessageTypes extends string> = Record<

packages/core/lib/types/queueOptionsTypes.ts

@@ -130,6 +130,126 @@ 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
+
+  /**
+   * Whether to throw an error when timeout is reached.
+   * - `true` (default): Throws `StartupResourcePollingTimeoutError` when timeout is reached
+   * - `false`: Reports the error via errorReporter, resets the timeout counter, and continues polling
+   *
+   * Use `false` when you want to be notified about prolonged unavailability but don't want to fail.
+   * Default: true
+   */
+  throwOnTimeout?: boolean
+
+  /**
+   * Whether to run polling in non-blocking mode.
+   * - `false` (default): init() waits for the resource to become available before resolving
+   * - `true`: If resource is not immediately available, init() resolves immediately and
+   *   polling continues in the background. When the resource becomes available,
+   *   the `onResourceAvailable` callback is invoked.
+   *
+   * Use `true` when you want the service to start quickly without waiting for dependencies.
+   * Default: false
+   */
+  nonBlocking?: boolean
+}
+
+/**
+ * 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
 }