@molecule/api-queue-sqs

Auto-generated, AI-first package reference for the molecule.dev ecosystem. It is written to be read by coding agents as much as by people, and is generated from this package's source — edit src/index.ts JSDoc, not this file.

AWS SQS queue provider for molecule.dev.

Uses AWS_REGION (default us-east-1) with the standard AWS credential chain; set SQS_ENDPOINT to target LocalStack. Queue URLs are resolved lazily on first operation — the queue must already exist (create it with createQueue() or in AWS) or the first send/receive rejects with the AWS QueueDoesNotExist error. Pass { autoCreateQueues: true } to createProvider() to auto-create a standard queue on first use instead (opt-in — unlike the memory/redis bonds, silently creating AWS resources has cost/IAM implications, so it is never the default).

Quick Start

import { setProvider, send, subscribe } from '@molecule/api-queue'
import { provider } from '@molecule/api-queue-sqs'

setProvider(provider)

subscribe<{ userId: string }>('emails', async (message) => {
  await deliver(message.body) // returning normally acks (deletes) the message
})

await send('emails', { body: { userId: 'u1' } })

Type

provider

Installation

npm install @molecule/api-queue-sqs @aws-sdk/client-sqs @molecule/api-bond @molecule/api-proxy-agent @molecule/api-queue @molecule/api-secrets

API

Interfaces

Queue

Handle for a named queue, providing send, receive, and subscribe operations.

interface Queue {
  /**
   * Queue name.
   */
  name: string
  /**
   * Sends a message to the queue.
   */
  send<T = unknown>(message: QueueMessage<T>): Promise<string>
  /**
   * Sends multiple messages to the queue.
   */
  sendBatch?<T = unknown>(messages: QueueMessage<T>[]): Promise<string[]>
  /**
   * Receives messages from the queue.
   */
  receive<T = unknown>(options?: ReceiveOptions): Promise<ReceivedMessage<T>[]>
  /**
   * Subscribes to messages from the queue.
   * Returns a function to unsubscribe.
   */
  subscribe<T = unknown>(handler: MessageHandler<T>, options?: ReceiveOptions): () => void
  /**
   * Gets the approximate number of messages in the queue.
   */
  size?(): Promise<number>
  /**
   * Purges all messages from the queue.
   */
  purge?(): Promise<void>
}

QueueCreateOptions

Options for creating a new queue, including FIFO mode, timeouts, retention periods, and dead-letter queue configuration.

interface QueueCreateOptions {
  /**
   * Whether this is a FIFO queue.
   */
  fifo?: boolean
  /**
   * Default visibility timeout in seconds.
   */
  visibilityTimeout?: number
  /**
   * Message retention period in seconds.
   */
  messageRetentionSeconds?: number
  /**
   * Maximum message size in bytes.
   */
  maxMessageSize?: number
  /**
   * Dead letter queue configuration.
   */
  deadLetterQueue?: {
    name: string
    maxReceiveCount: number
  }
}

QueueMessage

Message to be sent to a queue.

interface QueueMessage<T = unknown> {
  /**
   * Message payload.
   */
  body: T
  /**
   * Message ID (auto-generated if not provided).
   */
  id?: string
  /**
   * Delay in seconds before the message becomes visible.
   */
  delaySeconds?: number
  /**
   * Message attributes/headers.
   */
  attributes?: Record<string, string | number | boolean>
  /**
   * Message group ID (for FIFO queues).
   */
  groupId?: string
  /**
   * Deduplication ID (for FIFO queues).
   */
  deduplicationId?: string
}

QueueProvider

Queue provider interface that all queue bond packages must implement. Provides queue handle creation and optional queue management operations.

interface QueueProvider {
  /**
   * Gets or creates a queue by name.
   */
  queue(name: string): Queue
  /**
   * Lists all available queues.
   */
  listQueues?(): Promise<string[]>
  /**
   * Creates a new queue.
   */
  createQueue?(name: string, options?: QueueCreateOptions): Promise<Queue>
  /**
   * Deletes a queue.
   */
  deleteQueue?(name: string): Promise<void>
  /**
   * Closes all connections.
   */
  close?(): Promise<void>
}

ReceivedMessage

Received message from a queue.

interface ReceivedMessage<T = unknown> {
  /**
   * Message ID.
   */
  id: string
  /**
   * Message payload.
   */
  body: T
  /**
   * Receipt handle for acknowledging the message.
   */
  receiptHandle: string
  /**
   * Message attributes/headers.
   */
  attributes?: Record<string, string | number | boolean>
  /**
   * Number of times this message has been received.
   */
  receiveCount?: number
  /**
   * Timestamp when the message was sent.
   */
  sentTimestamp?: Date
  /**
   * Acknowledges (deletes) the message from the queue.
   */
  ack(): Promise<void>
  /**
   * Rejects the message (returns it to the queue).
   */
  nack?(): Promise<void>
}

ReceiveOptions

Options for receiving messages.

interface ReceiveOptions {
  /**
   * Maximum number of messages to receive.
   */
  maxMessages?: number
  /**
   * Visibility timeout in seconds.
   */
  visibilityTimeout?: number
  /**
   * Wait time in seconds for long polling.
   */
  waitTimeSeconds?: number
}

SQSOptions

Options for creating an SQS queue provider.

interface SQSOptions {
  region?: string
  accessKeyId?: string
  secretAccessKey?: string
  endpoint?: string
  accountId?: string

  /**
   * When `true`, resolving a queue name that does not yet exist in AWS
   * auto-creates a standard (non-FIFO) queue with default settings instead
   * of rejecting with `QueueDoesNotExist` — matching the memory/redis
   * bonds' "just works" first-send behavior. Off by default: unlike an
   * in-process or self-hosted broker, silently creating AWS resources has
   * cost and IAM-permission implications, so opting in is a deliberate
   * choice. When off (the default), create the queue via `createQueue()`,
   * the AWS console, or infrastructure-as-code before sending to it.
   */
  autoCreateQueues?: boolean
}

Types

MessageHandler

Async callback invoked for each message received from a queue subscription.

type MessageHandler<T = unknown> = (message: ReceivedMessage<T>) => Promise<void>

Functions

createProvider(options)

Creates an AWS SQS queue provider. Connects using AWS_REGION env var (default 'us-east-1'), optional explicit credentials, and optional custom endpoint (e.g. for LocalStack).

function createProvider(options?: SQSOptions): QueueProvider

• options — Optional AWS region, credentials, endpoint, and auto-create configuration. Falls back to environment variables.

Returns: A QueueProvider that manages SQS queues. Queue URLs are resolved lazily on first operation.

Constants

provider

Lazily-initialized SQS queue provider proxy that creates the provider on first access.

const provider: QueueProvider

queueSqsSecretDefinitions

Secret definitions required by the SQS queue bond.

const queueSqsSecretDefinitions: SecretDefinition[]

Core Interface

Implements @molecule/api-queue interface.

Bond Wiring

Setup function to register this provider with the core interface:

import { setProvider } from '@molecule/api-queue'
import { provider } from '@molecule/api-queue-sqs'

export function setupQueueSqs(): void {
  setProvider(provider)
}

Injection Notes

Requirements

Peer dependencies:

• @molecule/api-bond ^1.0.1
• @molecule/api-queue ^1.0.1
• @molecule/api-secrets ^1.0.1

Environment Variables

• AWS_REGION (required) — AWS region — default: us-east-1
  • Setup: The AWS region your resources live in.
  • Example: us-east-1
• AWS_ACCESS_KEY_ID (required) — AWS access key ID
  • Setup: Create an IAM user with the needed policy (SES/S3/SQS) and create an access key under Security credentials.
  • Get it here: https://console.aws.amazon.com/iam/
  • Example: AKIA...
• AWS_SECRET_ACCESS_KEY (required) — AWS secret access key
  • Setup: Shown once when creating the IAM access key — store it immediately.
  • Get it here: https://console.aws.amazon.com/iam/
• SQS_ENDPOINT (optional) — SQS endpoint override
  • Setup: Only for LocalStack or a custom SQS-compatible endpoint. Leave unset for real AWS — queues are addressed by NAME (the queue must exist, or pass { autoCreateQueues: true } to createProvider()).
  • Example: http://localhost:4566

Runtime Dependencies

• @aws-sdk/client-sqs
• @molecule/api-bond
• @molecule/api-proxy-agent
• @molecule/api-queue
• @molecule/api-secrets

Delivery semantics (at-least-once — handlers must be idempotent):

• Handler success acks (deletes) automatically; a handler throw leaves the message leased, and it returns to the queue when the visibility timeout expires (throw = retry). Bound poison messages with a redrive policy: createQueue(name, { deadLetterQueue: { name, maxReceiveCount } }).
• ack()/nack() settle at most once; nack() returns the message to the queue immediately (visibility timeout 0) instead of waiting out the lease.
• FIFO queues need the .fifo suffix (createQueue(name, { fifo: true }) appends it) and a groupId per message; a deduplicationId is derived from the message id when not provided.
• delaySeconds is capped at 900 (15 minutes) by SQS itself.
• subscribe() retries a failed queue-URL resolution (bad region, credentials not yet propagated, a QueueDoesNotExist race) with bounded exponential backoff (1s → 30s) instead of logging once and leaving the subscription permanently dead — it self-heals once the queue/credentials become valid.
• Runs behind an outbound proxy when HTTPS_PROXY is set. The AWS SDK v3 builds its own agent and reads no proxy variable, so on a host whose only egress path is a proxy every queue operation used to fail with a bare connection error. The client now gets a CONNECT-capable agent through its own requestHandler hook (@molecule/api-proxy-agent, resolved against SQS_ENDPOINT when set and the regional endpoint otherwise, so a LocalStack endpoint in NO_PROXY keeps connecting directly). With no proxy configured nothing is passed. Allowlist *.amazonaws.com on the proxy.

E2E Tests

Integration checklist — exercise the REAL behavior end-to-end (drive the app action that enqueues/consumes work in the live preview, no mocks), adapt each item to this app's actual screens/flows, and check every box off one by one. A box you can't check is an integration bug to fix — not a skip:

• The action that enqueues work responds fast — send() returns a message id immediately and the request/response does NOT block on the job. The #1 trap: the executor awaits the heavy work inline (defeating the queue); confirm the triggering UI action returns quickly and the work happens in the background.
• The enqueued job actually RUNS — a subscribe() consumer (a MessageHandler) is wired and running, so the message's real side effect (email sent, file processed, notification delivered — whatever the app does) actually appears in the UI/data. A message enqueued with no worker wired is the silent failure.
• Payload round-trips intact — the ReceivedMessage.body the handler sees equals the QueueMessage.body that was sent, with no dropped or renamed fields.
• Failure is handled — a handler that throws is redelivered (up to QueueCreateOptions.deadLetterQueue.maxReceiveCount, tracked via receiveCount) or dead-lettered, never silently lost. Delivery is at-least-once, so the handler is idempotent (dedupe on the job/record id) — a redelivery must not double-charge or double-send.
• Ordering/concurrency is not assumed — the app does not rely on strict FIFO (QueueMessage.groupId/fifo) or exactly-once delivery unless the bonded provider actually guarantees it.
• Least-authority payloads — the body carries only the ids/refs the job needs (never a secret or stale authority); the consumer re-loads and re-scopes on the CURRENT data (owner id from body, re-checked server-side) so one user's job cannot act on another user's resource.