feat(sqs): queue lifecycle helpers and Redis-first guidance

Adds opt-in queue lifecycle management for the SNS/SQS adapter so users
who pick the per-instance queue model have first-class tools to keep
queues from accumulating across restarts:

- consumer.lifecycle.deleteQueueOnClose / unsubscribeOnClose run on
  close() for graceful-shutdown cleanup
- consumer.lifecycle.heartbeat writes a layered-loader:heartbeat tag on
  the consumer's own queue on a timer
- reapStaleQueues({ ... }) helper scans queues by prefix and deletes
  ones whose heartbeat is older than the threshold, optionally also
  unsubscribing orphan SNS subscriptions

All opt-in and default off; existing users see no behaviour change.

Documentation rewrites the Update notifications and Flexible
invalidation triggers sections to strongly recommend Redis pub/sub
whenever feasible, and to lead with the Redis-publisher-plus-SQS-trigger
hybrid pattern (shape C) for the upstream-events use case. That hybrid
uses a single shared SQS queue across all instances, avoiding the
per-instance queue churn problem entirely.

The package README now opens with a "Picking your shape" table making
the three deployment shapes (pure Redis, pure SNS/SQS, hybrid) explicit
and recommending shape C whenever both Redis and AWS upstream events
are in play.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: kibertoad

README.md

@@ -27,6 +27,19 @@ export {
   createGroupNotificationPair,
   type SqsGroupNotificationConfig,
 } from './lib/SqsGroupNotificationFactory.js'
+export {
+  DEFAULT_HEARTBEAT_INTERVAL_MS,
+  DEFAULT_REAPER_IDLE_THRESHOLD_MS,
+  HEARTBEAT_TAG_KEY,
+  reapStaleQueues,
+  resolveQueueUrl,
+  startQueueHeartbeat,
+  type HeartbeatOptions,
+  type HeartbeatRunner,
+  type QueueLifecycleOptions,
+  type ReapStaleQueuesParams,
+  type ReapStaleQueuesResult,
+} from './lib/queueLifecycle.js'
 export {
   CLEAR_COMMAND,
   DELETE_COMMAND,

packages/sqs/README.md

@@ -1,3 +1,5 @@
+import { UnsubscribeCommand } from '@aws-sdk/client-sns'
+import { DeleteQueueCommand } from '@aws-sdk/client-sqs'
 import { MessageHandlerConfigBuilder } from '@message-queue-toolkit/core'
 import {
   AbstractSnsSqsConsumer,
@@ -8,6 +10,11 @@ import {
 } from '@message-queue-toolkit/sns'
 import type { ConsumerErrorHandler, SynchronousGroupCache } from 'layered-loader'
 import { AbstractNotificationConsumer } from 'layered-loader'
+import {
+  type HeartbeatRunner,
+  type QueueLifecycleOptions,
+  startQueueHeartbeat,
+} from './queueLifecycle.js'
 import type { SqsSubscriptionOptions } from './SqsNotificationConsumer.js'
 import {
   CLEAR_GROUP_NOTIFICATION_SCHEMA,
@@ -28,6 +35,12 @@ export type SqsGroupNotificationConsumerParams = {
   errorHandler?: ConsumerErrorHandler
   dependencies: SNSSQSConsumerDependencies
   subscriptionConfig?: SqsSubscriptionOptions
+  /**
+   * Optional queue-lifecycle behaviour. When set, controls automatic queue
+   * cleanup on close and/or periodic heartbeat tagging used by
+   * `reapStaleQueues`. See {@link QueueLifecycleOptions}.
+   */
+  lifecycle?: QueueLifecycleOptions
 } & SqsGroupNotificationConsumerConfig
 
 type ConsumerContext<LoadedValue> = {
@@ -71,6 +84,7 @@ export class SqsGroupNotificationConsumer<LoadedValue> extends AbstractNotificat
   private readonly params: SqsGroupNotificationConsumerParams
   private internalConsumer?: SnsSqsGroupInvalidationConsumer<LoadedValue>
   private subscribePromise?: Promise<SnsSqsGroupInvalidationConsumer<LoadedValue>>
+  private heartbeat?: HeartbeatRunner
 
   constructor(params: SqsGroupNotificationConsumerParams) {
     super(params.serverUuid, params.errorHandler)
@@ -143,6 +157,14 @@ export class SqsGroupNotificationConsumer<LoadedValue> extends AbstractNotificat
         await consumer.init()
         await consumer.start()
         this.internalConsumer = consumer
+        if (this.params.lifecycle?.heartbeat) {
+          this.heartbeat = startQueueHeartbeat({
+            sqsClient: this.params.dependencies.sqsClient,
+            queueUrl: consumer.publicQueueUrl,
+            intervalMs: this.params.lifecycle.heartbeat.intervalMs,
+            errorHandler: this.params.lifecycle.heartbeat.errorHandler,
+          })
+        }
         return consumer
       } catch (err) {
         await consumer.close().catch(() => undefined)
@@ -161,7 +183,35 @@ export class SqsGroupNotificationConsumer<LoadedValue> extends AbstractNotificat
     if (!this.internalConsumer) return
     const consumer = this.internalConsumer
     this.internalConsumer = undefined
+
+    this.heartbeat?.stop()
+    this.heartbeat = undefined
+
+    const queueUrl = consumer.publicQueueUrl
+    const subscriptionArn = consumer.publicSubscriptionArn
+    const unsubscribeOnClose = this.params.lifecycle?.unsubscribeOnClose ?? false
+    const deleteQueueOnClose = this.params.lifecycle?.deleteQueueOnClose ?? false
+
     await consumer.close()
+
+    if (unsubscribeOnClose && subscriptionArn) {
+      try {
+        await this.params.dependencies.snsClient.send(
+          new UnsubscribeCommand({ SubscriptionArn: subscriptionArn }),
+        )
+      } catch (err) {
+        this.params.lifecycle?.onCleanupError?.(err as Error, 'unsubscribe')
+      }
+    }
+    if (deleteQueueOnClose && queueUrl) {
+      try {
+        await this.params.dependencies.sqsClient.send(
+          new DeleteQueueCommand({ QueueUrl: queueUrl }),
+        )
+      } catch (err) {
+        this.params.lifecycle?.onCleanupError?.(err as Error, 'deleteQueue')
+      }
+    }
   }
 
   get topicArn(): string | undefined {

packages/sqs/index.ts

@@ -1,6 +1,7 @@
 import { randomUUID } from 'node:crypto'
 import type { SNSDependencies, SNSSQSConsumerDependencies } from '@message-queue-toolkit/sns'
 import type { ConsumerErrorHandler, PublisherErrorHandler } from 'layered-loader'
+import type { QueueLifecycleOptions } from './queueLifecycle.js'
 import {
   SqsGroupNotificationConsumer,
   type SqsGroupNotificationConsumerConfig,
@@ -22,6 +23,13 @@ export type SqsGroupNotificationConfig = {
   consumer: {
     dependencies: SNSSQSConsumerDependencies
     subscriptionConfig?: SqsSubscriptionOptions
+    /**
+     * Optional queue-lifecycle behaviour. Only relevant when each instance owns
+     * its own SQS queue (the default shape for full SNS/SQS fanout). Setups
+     * that pair this consumer with a Redis publisher and a shared queue do not
+     * need any of these knobs.
+     */
+    lifecycle?: QueueLifecycleOptions
   } & SqsGroupNotificationConsumerConfig
 }
 
@@ -59,13 +67,15 @@ export function createGroupNotificationPair<LoadedValue>(config: SqsGroupNotific
           errorHandler: config.consumerErrorHandler,
           dependencies: consumerConfig.dependencies,
           subscriptionConfig: consumerConfig.subscriptionConfig,
+          lifecycle: consumerConfig.lifecycle,
           creationConfig: consumerConfig.creationConfig,
         }
       : {
           serverUuid,
           errorHandler: config.consumerErrorHandler,
           dependencies: consumerConfig.dependencies,
           subscriptionConfig: consumerConfig.subscriptionConfig,
+          lifecycle: consumerConfig.lifecycle,
           locatorConfig: consumerConfig.locatorConfig,
         },
   )

packages/sqs/lib/SqsGroupNotificationConsumer.ts

@@ -1,4 +1,5 @@
-import type { SubscribeCommandInput } from '@aws-sdk/client-sns'
+import { type SubscribeCommandInput, UnsubscribeCommand } from '@aws-sdk/client-sns'
+import { DeleteQueueCommand } from '@aws-sdk/client-sqs'
 import { MessageHandlerConfigBuilder } from '@message-queue-toolkit/core'
 import {
   AbstractSnsSqsConsumer,
@@ -9,6 +10,11 @@ import {
 } from '@message-queue-toolkit/sns'
 import type { ConsumerErrorHandler, SynchronousCache } from 'layered-loader'
 import { AbstractNotificationConsumer } from 'layered-loader'
+import {
+  type HeartbeatRunner,
+  type QueueLifecycleOptions,
+  startQueueHeartbeat,
+} from './queueLifecycle.js'
 import {
   CLEAR_NOTIFICATION_SCHEMA,
   DELETE_MANY_NOTIFICATION_SCHEMA,
@@ -40,6 +46,12 @@ export type SqsNotificationConsumerParams = {
   errorHandler?: ConsumerErrorHandler
   dependencies: SNSSQSConsumerDependencies
   subscriptionConfig?: SqsSubscriptionOptions
+  /**
+   * Optional queue-lifecycle behaviour. When set, controls automatic queue
+   * cleanup on close and/or periodic heartbeat tagging used by
+   * {@link reapStaleQueues}. See {@link QueueLifecycleOptions}.
+   */
+  lifecycle?: QueueLifecycleOptions
 } & SqsNotificationConsumerConfig
 
 type ConsumerContext<LoadedValue> = {
@@ -76,6 +88,7 @@ export class SqsNotificationConsumer<LoadedValue> extends AbstractNotificationCo
   private readonly params: SqsNotificationConsumerParams
   private internalConsumer?: SnsSqsInvalidationConsumer<LoadedValue>
   private subscribePromise?: Promise<SnsSqsInvalidationConsumer<LoadedValue>>
+  private heartbeat?: HeartbeatRunner
 
   constructor(params: SqsNotificationConsumerParams) {
     super(params.serverUuid, params.errorHandler)
@@ -151,6 +164,14 @@ export class SqsNotificationConsumer<LoadedValue> extends AbstractNotificationCo
         await consumer.init()
         await consumer.start()
         this.internalConsumer = consumer
+        if (this.params.lifecycle?.heartbeat) {
+          this.heartbeat = startQueueHeartbeat({
+            sqsClient: this.params.dependencies.sqsClient,
+            queueUrl: consumer.publicQueueUrl,
+            intervalMs: this.params.lifecycle.heartbeat.intervalMs,
+            errorHandler: this.params.lifecycle.heartbeat.errorHandler,
+          })
+        }
         return consumer
       } catch (err) {
         await consumer.close().catch(() => undefined)
@@ -169,7 +190,38 @@ export class SqsNotificationConsumer<LoadedValue> extends AbstractNotificationCo
     if (!this.internalConsumer) return
     const consumer = this.internalConsumer
     this.internalConsumer = undefined
+
+    this.heartbeat?.stop()
+    this.heartbeat = undefined
+
+    const queueUrl = consumer.publicQueueUrl
+    const subscriptionArn = consumer.publicSubscriptionArn
+    const unsubscribeOnClose = this.params.lifecycle?.unsubscribeOnClose ?? false
+    const deleteQueueOnClose = this.params.lifecycle?.deleteQueueOnClose ?? false
+
     await consumer.close()
+
+    // Cleanup is best-effort: missing resources (already deleted, never created)
+    // must not break shutdown. The user opts in to this behaviour, so swallowing
+    // errors is the right default — they can re-run reapStaleQueues if needed.
+    if (unsubscribeOnClose && subscriptionArn) {
+      try {
+        await this.params.dependencies.snsClient.send(
+          new UnsubscribeCommand({ SubscriptionArn: subscriptionArn }),
+        )
+      } catch (err) {
+        this.params.lifecycle?.onCleanupError?.(err as Error, 'unsubscribe')
+      }
+    }
+    if (deleteQueueOnClose && queueUrl) {
+      try {
+        await this.params.dependencies.sqsClient.send(
+          new DeleteQueueCommand({ QueueUrl: queueUrl }),
+        )
+      } catch (err) {
+        this.params.lifecycle?.onCleanupError?.(err as Error, 'deleteQueue')
+      }
+    }
   }
 
   /**

packages/sqs/lib/SqsGroupNotificationFactory.ts

@@ -1,6 +1,7 @@
 import { randomUUID } from 'node:crypto'
 import type { SNSDependencies, SNSSQSConsumerDependencies } from '@message-queue-toolkit/sns'
 import type { ConsumerErrorHandler, PublisherErrorHandler } from 'layered-loader'
+import type { QueueLifecycleOptions } from './queueLifecycle.js'
 import {
   SqsNotificationConsumer,
   type SqsNotificationConsumerConfig,
@@ -30,6 +31,13 @@ export type SqsNotificationConfig = {
   consumer: {
     dependencies: SNSSQSConsumerDependencies
     subscriptionConfig?: SqsSubscriptionOptions
+    /**
+     * Optional queue-lifecycle behaviour. Only relevant when each instance owns
+     * its own SQS queue (the default shape for full SNS/SQS fanout). Setups
+     * that pair this consumer with a Redis publisher and a shared queue do not
+     * need any of these knobs.
+     */
+    lifecycle?: QueueLifecycleOptions
   } & SqsNotificationConsumerConfig
 }
 
@@ -67,13 +75,15 @@ export function createNotificationPair<LoadedValue>(config: SqsNotificationConfi
           errorHandler: config.consumerErrorHandler,
           dependencies: consumerConfig.dependencies,
           subscriptionConfig: consumerConfig.subscriptionConfig,
+          lifecycle: consumerConfig.lifecycle,
           creationConfig: consumerConfig.creationConfig,
         }
       : {
           serverUuid,
           errorHandler: config.consumerErrorHandler,
           dependencies: consumerConfig.dependencies,
           subscriptionConfig: consumerConfig.subscriptionConfig,
+          lifecycle: consumerConfig.lifecycle,
           locatorConfig: consumerConfig.locatorConfig,
         },
   )