AWS SNS

Server Source code Package

Server-side event publishing to AWS Simple Notification Service topics. Each event is published as one JSON message via the official @aws-sdk/client-sns SDK, with FIFO ordering and per-event message attributes. The destination needs only the topic ARN and, off AWS, credentials. It never creates topics while it runs: provisioning (subscriptions, KMS encryption, FIFO attributes) is a separate step, walkeros setup.

The destination ships inside @walkeros/server-destination-aws alongside Firehose. Import the named export destinationSNS and reference it via the code field in flow.json.

Where this fits

SNS is a server destination in the walkerOS flow:

Publishes events to an SNS topic for fan-out delivery to SQS queues, Lambda functions, HTTPS endpoints, email, SMS, and custom subscribers.

Quick start

"destinations": {
  "sns": {
    "package": "@walkeros/server-destination-aws",
    "import": "destinationSNS",
    "config": {
      "settings": {
        "topicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events"
      }
    }
  }
}

The region and the topic name come from the ARN. Add credentials when the flow runs outside AWS.

Installation

npm install @walkeros/server-destination-aws

Integrated

import { startFlow } from '@walkeros/collector';
import { destinationSNS } from '@walkeros/server-destination-aws';

await startFlow({
  destinations: {
    sns: {
      code: destinationSNS,
      config: {
        settings: {
          topicArn: 'arn:aws:sns:eu-central-1:123456789012:walkeros-events',
        },
      },
    },
  },
});

Bundled

Add to your flow.json destinations:

"destinations": {
  "sns": {
    "package": "@walkeros/server-destination-aws",
    "import": "destinationSNS",
    "config": {
      "settings": {
        "topicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events"
      },
      "credentials": {
        "accessKeyId": "$secret.AWS_ACCESS_KEY_ID",
        "secretAccessKey": "$secret.AWS_SECRET_ACCESS_KEY"
      }
    }
  }
}

See bundled mode setup | CLI reference

Configuration

This destination uses the standard destination config wrapper (consent, data, env, id, ...). For the shared fields see destination configuration. Package-specific fields live under config.settings and are listed below.

Settings

Property	Type	Description	More
topicArn	string	Topic ARN (like arn:aws:sns:eu-central-1:123456789012:walkeros-events). Region and topic name derive from it. Required unless topicName is set.
topicName	string	Topic name (like walkeros-events). Used by walkeros setup; without topicArn the ARN is completed with the account id at the first publish.
region	string	AWS region (like eu-central-1). Default: the region in topicArn, AWS_REGION or the active profile, else eu-central-1.
client	any	An SNSClient of your own. Used as is and never closed by the destination.
config	any	Raw SNSClient options passed to the SDK. Put keys in config.credentials instead.

Mapping

Per-event rules under config.mapping. For the standard rule fields (consent, condition, data, batch, name, policy) see mapping.

Property	Type	Description	More
messageAttributes	Record<string, any>
messageGroupId	any
messageDeduplicationId	any

Examples

Publish with per-event message attributes

Demonstrates `messageAttributes` as `Mapping.Map`: each attribute value resolves per event. Operators express the SDK shape (`{ DataType, StringValue }`) per entry.

Event

{
  "name": "product view",
  "data": {
    "id": "SKU-1",
    "tenant_id": "acme"
  },
  "context": {
    "shopping": [
      "detail",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [],
  "consent": {
    "functional": true
  },
  "id": "ev-1700001002",
  "trigger": "load",
  "entity": "product",
  "action": "view",
  "timestamp": 1700001002,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "express",
    "platform": "server"
  }
}

Mapping

{
  "settings": {
    "messageAttributes": {
      "schema_version": {
        "value": {
          "DataType": "String",
          "StringValue": "v4"
        }
      },
      "tenant": {
        "value": {
          "DataType": "String",
          "StringValue": "acme"
        }
      }
    }
  }
}

Out

client.send({
  "TopicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events",
  "Message": "{\"name\":\"product view\",\"data\":{\"id\":\"SKU-1\",\"tenant_id\":\"acme\"},\"context\":{\"shopping\":[\"detail\",0]},\"globals\":{\"pagegroup\":\"shop\"},\"custom\":{\"completely\":\"random\"},\"user\":{\"id\":\"us3r\",\"device\":\"c00k13\",\"session\":\"s3ss10n\"},\"nested\":[],\"consent\":{\"functional\":true},\"id\":\"ev-1700001002\",\"trigger\":\"load\",\"entity\":\"product\",\"action\":\"view\",\"timestamp\":1700001002,\"timing\":3.14,\"source\":{\"count\":1,\"trace\":\"0a1b2c3d4e5f60718293a4b5c6d7e8f9\",\"type\":\"express\",\"platform\":\"server\"}}",
  "MessageAttributes": {
    "schema_version": {
      "DataType": "String",
      "StringValue": "v4"
    },
    "tenant": {
      "DataType": "String",
      "StringValue": "acme"
    }
  }
})

Initialization

Init builds one SNSClient in the region of the topic ARN with the credentials from config.credentials and a 2500 ms per-attempt timeout. No topic is created and nothing is sent until the first event.

Event

{
  "settings": {
    "topicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events"
  },
  "credentials": {
    "accessKeyId": "$secret.AWS_ACCESS_KEY_ID",
    "secretAccessKey": "$secret.AWS_SECRET_ACCESS_KEY"
  }
}

Out

new SNSClient({
  "region": "eu-central-1",
  "credentials": {
    "accessKeyId": "$secret.AWS_ACCESS_KEY_ID",
    "secretAccessKey": "$secret.AWS_SECRET_ACCESS_KEY"
  },
  "requestHandler": {
    "requestTimeout": 2500,
    "throwOnRequestTimeout": true
  }
})

Mapped message

A mapping that produces an object publishes that object as the Message instead of the full event. Keep the event id in it for deduplication.

Event

{
  "name": "order complete",
  "data": {
    "id": "ORD-400",
    "total": 99.99,
    "currency": "EUR"
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "usr-789"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "ev-1700001001",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700001001,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "express",
    "platform": "server"
  }
}

Mapping

{
  "data": {
    "map": {
      "id": "id",
      "name": "name",
      "order_id": "data.id",
      "total": "data.total"
    }
  }
}

Out

client.send({
  "TopicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events",
  "Message": "{\"id\":\"ev-1700001001\",\"name\":\"order complete\",\"order_id\":\"ORD-400\",\"total\":99.99}"
})

Order publish with FIFO group (path-resolved)

An order complete event is published to a FIFO topic with a messageGroupId resolved from the event via the mapping path "user.id". The deduplication id defaults to the event id.

Event

{
  "name": "order complete",
  "data": {
    "id": "ORD-400",
    "total": 99.99,
    "currency": "EUR"
  },
  "context": {
    "shopping": [
      "complete",
      0
    ]
  },
  "globals": {
    "pagegroup": "shop"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "usr-789"
  },
  "nested": [
    {
      "entity": "product",
      "data": {
        "id": "ers",
        "name": "Everyday Ruck Snack",
        "color": "black",
        "size": "l",
        "price": 420
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "product",
      "data": {
        "id": "cc",
        "name": "Cool Cap",
        "size": "one size",
        "price": 42
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    },
    {
      "entity": "gift",
      "data": {
        "name": "Surprise"
      },
      "context": {
        "shopping": [
          "complete",
          0
        ]
      },
      "nested": []
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "ev-1700001001",
  "trigger": "load",
  "entity": "order",
  "action": "complete",
  "timestamp": 1700001001,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "express",
    "platform": "server"
  }
}

Mapping

{
  "settings": {
    "messageGroupId": "user.id"
  }
}

Out

client.send({
  "TopicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events.fifo",
  "Message": "{\"name\":\"order complete\",\"data\":{\"id\":\"ORD-400\",\"total\":99.99,\"currency\":\"EUR\"},\"context\":{\"shopping\":[\"complete\",0]},\"globals\":{\"pagegroup\":\"shop\"},\"custom\":{\"completely\":\"random\"},\"user\":{\"id\":\"usr-789\"},\"nested\":[{\"entity\":\"product\",\"data\":{\"id\":\"ers\",\"name\":\"Everyday Ruck Snack\",\"color\":\"black\",\"size\":\"l\",\"price\":420},\"context\":{\"shopping\":[\"complete\",0]},\"nested\":[]},{\"entity\":\"product\",\"data\":{\"id\":\"cc\",\"name\":\"Cool Cap\",\"size\":\"one size\",\"price\":42},\"context\":{\"shopping\":[\"complete\",0]},\"nested\":[]},{\"entity\":\"gift\",\"data\":{\"name\":\"Surprise\"},\"context\":{\"shopping\":[\"complete\",0]},\"nested\":[]}],\"consent\":{\"functional\":true},\"id\":\"ev-1700001001\",\"trigger\":\"load\",\"entity\":\"order\",\"action\":\"complete\",\"timestamp\":1700001001,\"timing\":3.14,\"source\":{\"count\":1,\"trace\":\"0a1b2c3d4e5f60718293a4b5c6d7e8f9\",\"type\":\"express\",\"platform\":\"server\"}}",
  "MessageGroupId": "usr-789",
  "MessageDeduplicationId": "ev-1700001001"
})

Page view publish

A page view event is JSON-stringified and published as the Message of an SNS Publish call.

Event

{
  "name": "page view",
  "data": {
    "title": "Home",
    "url": "https://example.com/"
  },
  "context": {
    "dev": [
      "test",
      1
    ]
  },
  "globals": {
    "pagegroup": "docs"
  },
  "custom": {
    "completely": "random"
  },
  "user": {
    "id": "us3r",
    "device": "c00k13",
    "session": "s3ss10n"
  },
  "nested": [
    {
      "entity": "child",
      "data": {
        "is": "subordinated"
      }
    }
  ],
  "consent": {
    "functional": true
  },
  "id": "ev-1700001000",
  "trigger": "load",
  "entity": "page",
  "action": "view",
  "timestamp": 1700001000,
  "timing": 3.14,
  "source": {
    "count": 1,
    "trace": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
    "type": "express",
    "platform": "server"
  }
}

Out

client.send({
  "TopicArn": "arn:aws:sns:eu-central-1:123456789012:walkeros-events",
  "Message": "{\"name\":\"page view\",\"data\":{\"title\":\"Home\",\"url\":\"https://example.com/\"},\"context\":{\"dev\":[\"test\",1]},\"globals\":{\"pagegroup\":\"docs\"},\"custom\":{\"completely\":\"random\"},\"user\":{\"id\":\"us3r\",\"device\":\"c00k13\",\"session\":\"s3ss10n\"},\"nested\":[{\"entity\":\"child\",\"data\":{\"is\":\"subordinated\"}}],\"consent\":{\"functional\":true},\"id\":\"ev-1700001000\",\"trigger\":\"load\",\"entity\":\"page\",\"action\":\"view\",\"timestamp\":1700001000,\"timing\":3.14,\"source\":{\"count\":1,\"trace\":\"0a1b2c3d4e5f60718293a4b5c6d7e8f9\",\"type\":\"express\",\"platform\":\"server\"}}"
})

Credentials

On AWS, configure nothing. Attach an IAM role to the runtime (ECS task role, Lambda execution role, EC2 instance profile, App Runner instance role). The AWS SDK's default credential chain picks it up, so the flow carries no keys.

Anywhere else, set config.credentials and back both values with $secret references. Never write a literal key into a flow file.

"config": {
  "settings": { "...": "..." },
  "credentials": {
    "accessKeyId": "$secret.AWS_ACCESS_KEY_ID",
    "secretAccessKey": "$secret.AWS_SECRET_ACCESS_KEY"
  }
}

config.credentials takes { accessKeyId, secretAccessKey, sessionToken? }, or the same object as a JSON string. It wins over settings.config.credentials, which wins over the default chain. The shape is checked when the destination starts, without contacting AWS, and error messages name the field that is wrong, never its value. Instead of config.credentials, the standard AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_SESSION_TOKEN environment variables also work, as does AWS_PROFILE with a shared credentials file.

Kubernetes without keys: with IAM roles for service accounts (EKS) or any OIDC web identity setup, set AWS_ROLE_ARN and AWS_WEB_IDENTITY_TOKEN_FILE in the container. The default chain exchanges the token for short-lived credentials, so the flow again carries no keys.

Credentials are loaded on the first delivery, not at start. A missing or wrong key surfaces as the first failed event, with a message that points at config.credentials or the environment.

A client passed as settings.client is used as is and never closed by the destination.

Topic and region

settings.topicArn is the primary setting. Without it, settings.topicName also works: the destination then completes the ARN once, at the first publish, with the account id from STS GetCallerIdentity, and caches it. A failed lookup is not cached, so the next publish asks again.

At runtime the region is resolved once, when the destination starts:

1. settings.region
2. the region in settings.topicArn
3. settings.config.region
4. AWS_REGION, or the region of the active profile (AWS_PROFILE)
5. eu-central-1

When settings.region contradicts the region in settings.topicArn, the destination refuses to start. Remove settings.region; the ARN already names it.

Message payload

Each Message is the full walkerOS event as JSON, or the mapped data when the event's mapping produces an object. Keep the event id in mapped data, so subscribers can deduplicate on it.

SNS publishes one message per event and has no batching. Setting config.batch on this destination logs a warning at start; remove it.

Delivery

Delivery is at least once: a message can arrive twice, for example when SNS stored it but the answer was lost on the way back. Deduplicate downstream on the event id.

Each publish resolves once SNS accepted the message, else it fails with an error carrying code (the AWS error name), status (the HTTP status) and retryable (as classified by the AWS SDK). The AWS SDK's standard retry stays on, with each attempt bounded at a quarter of config.timeout unless you pass your own settings.config.requestHandler.

Messages rejected by AWS are not replayed yet

There is no retry beyond the SDK's own. A message that SNS throttles or rejects goes to the collector's in-memory dead letter queue, which nothing replays, so it is lost. The collector logs every failure with its cause, so the loss is visible. A standard retry for all destinations is planned in the collector.

Setup

Provision the topic, attributes, tags, and declared subscriptions in one shot:

walkeros setup destination.sns -c flow.json

Setup reads the same config.credentials as the runtime, and resolves $secret references from the environment before it runs. An unset secret stops setup with a message that names the secret, never its value.

Setup targets the first region in this order: setup.region, settings.region, the region in settings.topicArn, settings.config.region, AWS_REGION or the active profile, then eu-central-1. Without setup.region it uses the same region as the runtime.

Setup is authoritative-apply: declared state is written to declared resources via a single idempotent CreateTopic call (plus one Subscribe call per declared subscription). Non-declared subscriptions and tags are left untouched, never listed, never logged. Operators may freely manage subscriptions or tags outside walkerOS without interference. Re-running setup against a fully provisioned topic with declared state matching actual produces zero state mutations.

The result is JSON-stringified to stdout:

{
  "topicArn": "arn:aws:sns:eu-central-1:000000000000:walkeros-events",
  "topicCreated": true,
  "tagsApplied": 2,
  "subscriptionsCreated": 1
}

topicCreated reflects whether the topic existed before this setup run (via a GetTopicAttributes probe against an STS-derived candidate ARN). tagsApplied and subscriptionsCreated count declared resources written.

FIFO topics

Set setup.fifoTopic: true. The destination auto-appends .fifo to the topic name when missing and applies FifoTopic/ContentBasedDeduplication attributes at creation. Reverse: setup.fifoTopic: false with a .fifo suffix throws a clear error.

KMS encryption

Set setup.kmsMasterKeyId to apply server-side encryption at creation. The AWS-managed default key alias is alias/aws/sns.

Subscriptions

Each declared subscription is created with one idempotent Subscribe call. Supported protocols: sqs, lambda, https, http, email, sms. Per-subscription attributes (rawMessageDelivery, filterPolicy, deadLetterTargetArn) are applied at the same call.

"setup": {
  "subscriptions": [
    {
      "protocol": "sqs",
      "endpoint": "arn:aws:sqs:eu-central-1:000000000000:walkeros-q",
      "rawMessageDelivery": true
    }
  ]
}

FIFO ordering

messageGroupId and messageDeduplicationId are Mapping.Value fields resolved per event. A string path drives the value from event data; a value-config form supplies a literal:

"mapping": {
  "order": {
    "complete": {
      "settings": {
        "messageGroupId": "user.id",
        "messageDeduplicationId": "id"
      }
    }
  }
}

On a FIFO topic (an ARN ending in .fifo), MessageDeduplicationId defaults to the event id unless a rule maps one, so SNS drops a resent event within its five-minute deduplication window.

Message attributes

messageAttributes is a Mapping.Map. Each value resolves per event to the SDK's { DataType, StringValue } shape. Bare strings are auto-wrapped as { DataType: 'String', StringValue }:

"mapping": {
  "product": {
    "view": {
      "settings": {
        "messageAttributes": {
          "schema_version": { "value": { "DataType": "String", "StringValue": "v4" } },
          "tenant": "data.tenant_id"
        }
      }
    }
  }
}

IAM

Setup role:

• sns:CreateTopic
• sns:GetTopicAttributes (existence probe)
• sns:Subscribe

Setup also calls sts:GetCallerIdentity to resolve the account id. That call needs no IAM permission.

Runtime push role:

• sns:Publish on the topic

When only settings.topicName is set, the runtime also calls sts:GetCallerIdentity once to complete the ARN. That call needs no IAM permission.

The runtime never creates topics, so it needs no sns:CreateTopic.

Recommended: separate roles for setup and runtime. Setup runs from a provisioner identity; runtime push uses a least-privileged role.

Simulate

walkeros push --simulate destination.sns runs offline with the package's mock SNS client and prints each AWS.SNSClient.send call with its command input; nothing reaches AWS. The step's import (destinationSNS) selects the SNS examples and mock, not Firehose's.

Troubleshooting

NotFoundException at push time. The topic does not exist in that region. Run walkeros setup destination.sns, or set settings.topicArn to an existing topic.

AuthorizationErrorException at push time. The credentials lack sns:Publish on the topic.

UnrecognizedClientException, InvalidSignatureException, ExpiredTokenException, or no credentials found. The keys are wrong, expired or missing. Check config.credentials or the AWS credentials in the environment.

The destination refuses to start. A configuration error found offline: neither settings.topicArn nor settings.topicName is set, settings.topicArn is not an SNS topic ARN, settings.region contradicts it, or config.credentials is malformed. The message names the field.

AccessDenied at setup. The setup role is missing one of the three SNS permissions above. SNS errors propagate verbatim.

FIFO suffix error. A topic name ending in .fifo was supplied with setup.fifoTopic: false, or vice versa. Either set fifoTopic: true or rename the topic.

Subscription endpoint rejected. AWS validates the endpoint at Subscribe time. SNS does NOT create the SQS queue, Lambda, or HTTPS endpoint. Provision those resources separately, then declare the subscription here.

Next steps

• AWS SQS source for the consumer side. Standard pattern: SNS topic fans out to one or more SQS queues that walkerOS pulls.
• Mapping configuration
• Flow configuration