> ## Documentation Index
> Fetch the complete documentation index at: https://siglo.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# AWS SQS Adapter

> AWS SQS batch consumer adapter for Lambda event source mappings and long-polling workers with partial batch failure handling.

# @siglo/sqs Adapter Guide

`@siglo/sqs` provides an AWS SQS batch consumer adapter for executing `Siglo` workflows. It is optimized for AWS Lambda event source mappings and standalone long-polling worker processes, implementing AWS partial batch failure protocols (`ReportBatchItemFailures`), message deduplication, and tenant namespace extraction.

***

## 1. Overview & Architecture

When AWS Lambda is triggered by an SQS queue, it passes a batch of records (`SQSEvent`). If one message in the batch fails, returning an uncaught exception causes the entire batch to be retried by AWS, resulting in duplicate processing of previously successful messages.

Furthermore, when concurrent Lambda invocations receive duplicate messages while execution is already underway, `@siglo/sqs` queries the `IdempotencyStore`. If an in-progress claim is detected, the adapter marks the message as a batch item failure. This causes SQS to retry the message after the visibility timeout expires rather than executing duplicate concurrent workflows.

`@siglo/sqs` handles this using AWS SQS Partial Batch Responses:

```mermaid theme={null}
flowchart TD
    SQS["AWS SQS"] -->|Batch of N records| Lambda["AWS Lambda Trigger"]
    Lambda --> Handler["createSqsBatchHandler<br/>• Iterates event.Records<br/>• Checks IdempotencyStore<br/>• Parses JSON Record Body<br/>• Executes Siglo"]
    Handler --> Split{Record Result}
    Split -->|Success Message| SSuccess["• Complete Idemp Key<br/>• Increment Count"]
    Split -->|In-Progress Claim| SInProg["• Yield to SQS Retry<br/>• Append to batchItemFailures"]
    Split -->|Failed Message| SFail["• Fail Idemp Key<br/>• Append to batchItemFailures"]
    SSuccess --> BatchRes["Return Batch Result<br/>{ batchItemFailures: [...] }"]
    SInProg --> BatchRes
    SFail --> BatchRes
    BatchRes --> Final["SQS deletes succeeded records<br/>SQS retries failed and in-progress items"]
```

***

## 2. Installation & Requirements

Install `@siglo/sqs`:

```bash theme={null}
npm install @siglo/core @siglo/sqs
npm install --save-dev @types/aws-lambda
```

Peer dependency matrix:

* `Siglo`: `*` (v1.0.0+)
* `@aws-sdk/client-sqs`: `^3.0.0` (optional if running in AWS Lambda runtime)

***

## 3. API Reference

### `createSqsBatchHandler(workflow, options)`

Creates an AWS Lambda-compliant batch processor function:

```typescript theme={null}
function createSqsBatchHandler<TInput = any, TOutput = any>(
    workflow: Flow<TInput, TOutput>,
    options?: SigloSqsOptions<TInput, TOutput>,
): (event: SigloSqsEvent) => Promise<SqsBatchResponse>;
```

### TypeScript Interfaces

```typescript theme={null}
export interface SigloSqsRecord {
    messageId: string;
    receiptHandle: string;
    body: string;
    attributes?: Record<string, string>;
    messageAttributes?: Record<string, any>;
}

export interface SigloSqsEvent {
    Records: SigloSqsRecord[];
}

export interface SqsBatchResponse {
    batchItemFailures: Array<{ itemIdentifier: string }>;
    processedCount: number;
}
```

***

## 4. Configuration Options (`SigloSqsOptions`)

| Option | Type | Default | Description |
| :- | :- | :- | :- |
| `input` | `(record: SigloSqsRecord) => Promise<TInput> \| TInput` | `JSON.parse(record.body)` | Custom mapper transforming an SQS record into workflow parameters. |
| `output` | `(result: any, record: SigloSqsRecord) => Promise<TOutput> \| TOutput` | `result` | Custom post-processor for step result. |
| `idempotencyStore` | `IdempotencyStore` | `undefined` | Distributed store to deduplicate across re-deliveries. |
| `idempotencyTtlMs` | `number` | `undefined` | Time-to-live for idempotency lock in milliseconds. Passed through to store (Redis/Postgres default: 24h; InMemory has no default expiration unless specified). |
| `timeout` | `number` | `undefined` | SLA execution timeout per record in milliseconds. |
| `namespace` | `string \| ((record: SigloSqsRecord) => string \| undefined)` | `record.messageAttributes.namespace` | Namespace extractor for multi-tenant isolation. |

***

## 5. AWS Lambda Configuration: `ReportBatchItemFailures`

To enable partial batch responses in AWS Lambda:

1. In the AWS Lambda Console, navigate to your SQS Trigger configuration.
2. Under **Additional settings**, check the option: **Report batch item failures**.
3. In AWS SAM / CloudFormation / Serverless Framework / CDK:
   ```yaml theme={null}
   FunctionEventSource:
       Type: SQS
       Properties:
           Queue: !GetAtt MyQueue.Arn
           FunctionResponseTypes:
               - ReportBatchItemFailures
   ```

When this is enabled, returning `{ batchItemFailures: [{ itemIdentifier: 'msg-01' }] }` tells AWS SQS to delete all successful messages in the batch from the queue, while leaving only `msg-01` in the queue to be retried or routed to the Dead Letter Queue (DLQ).

***

## 6. Complete AWS Lambda Production Example

```typescript theme={null}
import { SQSEvent, SQSBatchResponse, Context } from 'aws-lambda';
import { Step, flow } from '@siglo/core';
import { createSqsBatchHandler } from '@siglo/sqs';
import { PostgresIdempotencyStore } from '@siglo/postgres';
import { Pool } from 'pg';

// 1. Initialize PostgreSQL Connection Pool outside handler
const pool = new Pool({
    connectionString: process.env.DATABASE_URL,
    max: 5,
});

const idempotencyStore = new PostgresIdempotencyStore({ client: pool });

// 2. Define Notification Workflow
interface NotificationPayload {
    userId: string;
    template: string;
    destination: string;
}

const formatNotification = Step.create('formatNotification', async (param: NotificationPayload) => {
    return { formattedBody: `Hello user ${param.userId}, template ${param.template}` };
});

const dispatchEmail = Step.create('dispatchEmail', async (param: any) => {
    // Send email via AWS SES or external provider
    return { dispatched: true, messageRef: `ses_${Date.now()}` };
});

const notificationFlow = flow(formatNotification, dispatchEmail);

// 3. Create SQS Batch Handler
const sqsHandler = createSqsBatchHandler(notificationFlow, {
    idempotencyStore,
    timeout: 10000, // 10-second deadline per message
    namespace: record => record.messageAttributes?.tenantId?.stringValue,
});

// 4. Export AWS Lambda Handler
export const handler = async (event: SQSEvent, context: Context): Promise<SQSBatchResponse> => {
    context.callbackWaitsForEmptyEventLoop = false;

    const response = await sqsHandler(event);
    console.log(`Processed ${response.processedCount} messages. Failures: ${response.batchItemFailures.length}`);

    return {
        batchItemFailures: response.batchItemFailures,
    };
};
```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.