DynamoDB to SNS and SNS to DynamoDB

Send DynamoDB changes to SNS with a DynamoDB Streams consumer: either a Lambda function that publishes messages or an EventBridge pipe targeting a standard SNS topic. In the reverse direction, subscribe Lambda to a standard SNS topic and write validated messages to DynamoDB. Neither route is an atomic transaction across services, so preserve event IDs and handle retries safely.

SNS and DynamoDB serve different purposes

SNS publishes messages to subscribers, such as queues, functions and supported notification endpoints. DynamoDB stores items that applications can retrieve, query and update. SNS is useful for fan-out; DynamoDB is useful for persistent application state. One does not replace the other.

For example, an order update can produce an event that independent subscribers use to send a receipt, update an analytics system and schedule fulfillment. Each subscriber needs its own delivery and failure policy. A successful publish does not prove all subscribers completed their work.

DynamoDB Streams supplies the change records. It captures inserts, updates and deletes after the database write, retaining records for 24 hours. Choose the images your publisher needs and keep the outgoing message contract independent of the raw DynamoDB representation.

Standard versus FIFO topics

FIFO does not mean your business operation executes exactly once. A downstream worker can finish a side effect and fail before acknowledging its message. A later retry must still be safe. Throughput also depends on current topic settings and quotas, so a single old messages-per-second figure is not a general FIFO limit.

Lambda does not subscribe directly to SNS FIFO topics. Use SNS FIFO → SQS → Lambda when you need that route. SNS FIFO also does not support email, SMS or HTTP endpoints. See AWS's FIFO delivery documentation.

DynamoDB Streams to SNS

The following setup uses an existing Orders table with string key id. Its publisher sends a compact event containing a stable event ID, operation, item key and selected status field. It handles every record rather than only Records[0].

1. Enable the stream

In the DynamoDB console, open the table's Exports and streams tab and turn on the stream. Select New and old images so the publisher can represent both updates and deletes. The AWS CLI v2 equivalent, for a table with no active stream, is:

aws dynamodb update-table \
  --table-name Orders \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

STREAM_ARN=$(aws dynamodb describe-table --table-name Orders \
  --query 'Table.LatestStreamArn' --output text)
aws dynamodb update-table \
  --table-name Orders \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

STREAM_ARN=$(aws dynamodb describe-table --table-name Orders \
  --query 'Table.LatestStreamArn' --output text)

Keys are always present. Inserts have no old image, and deletes have no new image. If the stream is replaced, its ARN changes and the consumer must be reconnected.

2. Create a standard SNS topic

Create a standard topic in the same Region as the publisher:

TOPIC_ARN=$(aws sns create-topic --name order-changes \
  --query TopicArn --output text)
TOPIC_ARN=$(aws sns create-topic --name order-changes \
  --query TopicArn --output text)

Create the desired subscriptions. Email subscriptions require confirmation before messages arrive. For a production worker, an SQS subscription can provide buffering and independent retries; the DynamoDB and SQS guide covers queue consumers.

3. Deploy the publisher

Use the nodejs24.x runtime, save the code as index.mjs, and set handler index.handler. Bundle @aws-sdk/client-sns and @aws-sdk/util-dynamodb. Set TOPIC_ARN in the function environment.

import { SNSClient, PublishCommand } from "@aws-sdk/client-sns";
import { unmarshall } from "@aws-sdk/util-dynamodb";

const sns = new SNSClient({});

export const handler = async (event) => {
  for (const record of event.Records) {
    try {
      const data = record.dynamodb;
      const image = data.NewImage ?? data.OldImage;
      const item = image ? unmarshall(image) : {};
      const message = {
        eventId: `${record.eventSourceARN}#${record.eventID}`,
        operation: record.eventName,
        key: unmarshall(data.Keys),
        status: item.status ?? null,
      };
      await sns.send(new PublishCommand({
        TopicArn: process.env.TOPIC_ARN,
        Message: JSON.stringify(message),
        MessageAttributes: {
          operation: { DataType: "String", StringValue: record.eventName },
        },
      }));
    } catch (error) {
      console.error("Publish failed", record.eventID, error.name);
      return {
        batchItemFailures: [{ itemIdentifier: record.dynamodb.SequenceNumber }],
      };
    }
  }
  return { batchItemFailures: [] };
};
import { SNSClient, PublishCommand } from "@aws-sdk/client-sns";
import { unmarshall } from "@aws-sdk/util-dynamodb";

const sns = new SNSClient({});

export const handler = async (event) => {
  for (const record of event.Records) {
    try {
      const data = record.dynamodb;
      const image = data.NewImage ?? data.OldImage;
      const item = image ? unmarshall(image) : {};
      const message = {
        eventId: `${record.eventSourceARN}#${record.eventID}`,
        operation: record.eventName,
        key: unmarshall(data.Keys),
        status: item.status ?? null,
      };
      await sns.send(new PublishCommand({
        TopicArn: process.env.TOPIC_ARN,
        Message: JSON.stringify(message),
        MessageAttributes: {
          operation: { DataType: "String", StringValue: record.eventName },
        },
      }));
    } catch (error) {
      console.error("Publish failed", record.eventID, error.name);
      return {
        batchItemFailures: [{ itemIdentifier: record.dynamodb.SequenceNumber }],
      };
    }
  }
  return { batchItemFailures: [] };
};

The execution role needs stream-read permissions and sns:Publish on this topic, plus CloudWatch Logs permissions. Restrict the topic ARN rather than granting publish access everywhere. Configure any customer managed encryption keys to permit the required service operations.

Selecting attributes limits accidental data exposure and payload size. SNS topics default to a 256 KiB maximum message size and can now be configured up to 1 MiB; the combined body and attributes must fit the topic's configured maximum. Do not assume a raw DynamoDB before/after record will fit the default. See the SNS Publish API.

4. Connect the publisher to the stream

Create an event source mapping for a deployed function named orders-to-sns:

aws lambda create-event-source-mapping \
  --function-name orders-to-sns \
  --event-source-arn "$STREAM_ARN" \
  --starting-position TRIM_HORIZON \
  --batch-size 100 \
  --function-response-types ReportBatchItemFailures \
  --bisect-batch-on-function-error \
  --maximum-retry-attempts 5 \
  --maximum-record-age-in-seconds 3600
aws lambda create-event-source-mapping \
  --function-name orders-to-sns \
  --event-source-arn "$STREAM_ARN" \
  --starting-position TRIM_HORIZON \
  --batch-size 100 \
  --function-response-types ReportBatchItemFailures \
  --bisect-batch-on-function-error \
  --maximum-retry-attempts 5 \
  --maximum-record-age-in-seconds 3600

Add a failure destination and choose retry limits suitable for the application before production use. The Lambda trigger guide explains the destination setup and recovery implications. The partial response in the code only works when ReportBatchItemFailures is enabled.

If publish succeeds but Lambda loses the response, a retry can publish another message. SNS assigns a new SNS message ID to a new publish, so downstream consumers should deduplicate using the stable eventId in the body when they need to recognize that case.

5. Test the full route

Insert an order, update its status, and delete it. Verify that subscribers receive all three operation types and that a batch containing several records publishes every record. Confirm that a delete carries its key without attempting to dereference NewImage.

Then replay the same test event to the handler and confirm downstream processing is safe. Deliberately remove publish permission in a test environment to verify retries and failure handling. Check iterator age and publish errors rather than relying only on a single successful email.

Alternative: EventBridge Pipes to SNS

For straightforward routing, create a pipe with the DynamoDB stream ARN as source and a standard SNS topic as target. Give the pipe execution role read access to the stream and publish access to the topic. Configure filtering, an input transformation, retry behavior and a dead-letter destination.

Pipes can remove the custom Lambda publisher when the built-in transformation is sufficient. It does not automatically convert every DynamoDB AttributeValue into your desired application schema; inspect a delivered sample. FIFO SNS topics are not supported pipe targets. See Pipes targets.

Use Lambda when you need conditional business logic, SDK calls or a transformation beyond the pipe's supported expressions. In either case, the downstream delivery semantics still matter.

SNS to DynamoDB

Use a separate destination table, ReceivedEvents, with string partition key eventId. Keeping it separate from the source table avoids an accidental publish-and-write feedback loop.

1. Prepare the function and permissions

Create a function named sns-to-dynamodb on nodejs24.x with TABLE_NAME=ReceivedEvents. Grant its execution role dynamodb:PutItem on that table and log access. SNS also needs permission to invoke the function, supplied through the function's resource-based policy.

2. Validate and write every message

Bundle @aws-sdk/client-dynamodb and @aws-sdk/lib-dynamodb. This handler stores the domain event ID and uses a conditional put so a retried notification does not create a second record:

import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export const handler = async (event) => {
  for (const record of event.Records) {
    const message = JSON.parse(record.Sns.Message);
    if (typeof message.eventId !== "string" || message.eventId.length === 0) {
      throw new Error("Message requires a non-empty eventId");
    }
    try {
      await ddb.send(new PutCommand({
        TableName: process.env.TABLE_NAME,
        Item: {
          eventId: message.eventId,
          payload: message,
          receivedAt: record.Sns.Timestamp,
        },
        ConditionExpression: "attribute_not_exists(eventId)",
      }));
    } catch (error) {
      if (error.name !== "ConditionalCheckFailedException") throw error;
      // The immutable event was already stored under this eventId.
    }
  }
};
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient({}));

export const handler = async (event) => {
  for (const record of event.Records) {
    const message = JSON.parse(record.Sns.Message);
    if (typeof message.eventId !== "string" || message.eventId.length === 0) {
      throw new Error("Message requires a non-empty eventId");
    }
    try {
      await ddb.send(new PutCommand({
        TableName: process.env.TABLE_NAME,
        Item: {
          eventId: message.eventId,
          payload: message,
          receivedAt: record.Sns.Timestamp,
        },
        ConditionExpression: "attribute_not_exists(eventId)",
      }));
    } catch (error) {
      if (error.name !== "ConditionalCheckFailedException") throw error;
      // The immutable event was already stored under this eventId.
    }
  }
};

This assumes publishers never reuse an event ID for different content. Validate the full message schema and size for your application. A DynamoDB item cannot exceed 400 KB including attribute names; an SNS topic accepting a larger payload does not enlarge that limit.

This is an SNS-triggered function, not a DynamoDB stream mapping. It throws on failure so Lambda's asynchronous invocation handling can retry. Returning a DynamoDB-style partial-batch object would not give SNS individual-record retry semantics.

3. Subscribe Lambda and test

In the Lambda console, add an SNS trigger and select the standard topic. The console configures the subscription and invocation permission. For CLI setup, configure both explicitly, restricting invocation to the intended topic and account. AWS documents the steps in SNS-triggered Lambda functions.

Publish a small test message:

aws sns publish \
  --topic-arn "$TOPIC_ARN" \
  --message '{"eventId":"test-order-001","operation":"INSERT","key":{"id":"order-001"},"status":"PENDING"}'

aws dynamodb get-item \
  --table-name ReceivedEvents \
  --key '{"eventId":{"S":"test-order-001"}}' \
  --consistent-read
aws sns publish \
  --topic-arn "$TOPIC_ARN" \
  --message '{"eventId":"test-order-001","operation":"INSERT","key":{"id":"order-001"},"status":"PENDING"}'

aws dynamodb get-item \
  --table-name ReceivedEvents \
  --key '{"eventId":{"S":"test-order-001"}}' \
  --consistent-read

Publishing the same domain event again should leave one stored item. If your desired operation is an update to business state instead of immutable event storage, use a version condition and an appropriate update expression; an unconditional put can overwrite a newer state with a delayed older message.

Failure handling, filters and costs

SNS delivery to Lambda and Lambda's processing of an accepted asynchronous invocation are separate failure stages. Configure subscription delivery failure handling where appropriate, and configure Lambda's asynchronous failure destination for handler errors. The SNS retry policy documentation explains what happens when delivery attempts are exhausted.

Use SNS subscription filters to route only the events each consumer needs. Filtering is not an authorization boundary: keep sensitive attributes out of messages that subscribers must not see. If SNS delivers through SQS, decide whether to enable raw message delivery; otherwise the worker receives an SNS envelope around the message body.

Budget for SNS publishes and deliveries, Lambda execution, logs and destination writes. Standard Lambda trigger reads from DynamoDB Streams are free; Lambda Managed Instances are an exception. EventBridge Pipes adds its own processing charge. Neither a successful publish nor a low stream-read bill proves the complete pipeline is working or inexpensive.