Emithook Docs

Appearance

Deliver to Amazon SQS ​

Route delivered events into an Amazon SQS queue. Emithook delivers from its own IAM principal and you grant it access with a queue policy — there are no per-destination access keys to create, rotate, or leak.

How delivery works ​

Emithook signs every delivery with a single, shared IAM principal (its own AWS account). To let it write to your queue, you add one statement to the queue's resource-based access policy granting that principal permission. This is standard AWS cross-account access — the same model SNS→SQS and EventBridge→SQS use.

emithook delivery worker  ──(SendMessage)──▶  your SQS queue
   principal: arn:aws:iam::589958197158:user/emithook
   authorized by: your queue's Access policy

Emithook needs two actions:

  • sqs:SendMessage — deliver each event to the queue.
  • sqs:GetQueueAttributes — the console's Reachable check calls this to confirm the grant took effect before you route traffic.

1. Find your queue URL and ARN ​

In the SQS console, open your queue. The URL looks like:

https://sqs.us-east-1.amazonaws.com/123456789012/my-queue

Its ARN is the same three parts rearranged:

arn:aws:sqs:us-east-1:123456789012:my-queue

(region → account id → queue name). The emithook console derives this ARN from the URL for you.

2. Add the queue policy ​

In the SQS console: your queue → Access policy → Edit, and paste the statement below (merge it into an existing policy's Statement array if you already have one). Replace Resource with your queue's ARN.

json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "EmithookDelivery",
      "Effect": "Allow",
      "Principal": { "AWS": "arn:aws:iam::589958197158:user/emithook" },
      "Action": ["sqs:SendMessage", "sqs:GetQueueAttributes"],
      "Resource": "<your-queue-arn>"
    }
  ]
}

Save the policy.

Region and ARN must match

The queue URL you register in emithook, its ARN in the policy, and the queue itself must all be in the same region. A mismatched region is the most common cause of a Queue not found reachability result.

Self-hosting

The principal ARN above is emithook's managed delivery identity. A self-hosted deployment publishes its own via SQS_DELIVERY_PRINCIPAL_ARN — the console always shows the correct ARN for the deployment you're using, so copy it from there rather than from this page.

3. Create the destination in emithook ​

In the console: Destinations → New destination → Amazon SQS, enter your queue URL, and copy the pre-filled queue policy (it already has the right principal ARN and your derived queue ARN). The access-key fields are gone — cross-account grants replace them.

Prefer the API? Create the destination with just the queue URL:

bash
curl -X POST https://api.emithook.com/v1/destinations \
  -H "Authorization: Bearer $EK_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-queue",
    "type": "sqs",
    "config": { "url": "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue" }
  }'

4. Verify with the Reachable indicator ​

Back in the console, the destination shows a reachability badge. Click Re-check — emithook calls GetQueueAttributes against your queue:

  • Reachable ✓ — the policy is in effect; you can route endpoints to this destination.
  • Not reachable ✗ with Access denied — the queue policy hasn't granted access yet (or is still propagating). Re-check after saving the policy.
  • Not reachable ✗ with Queue not found — the URL or region is wrong; confirm the queue exists in that region.

Once it reads Reachable, add the destination to an endpoint's routes and events start flowing to your queue.

5. Consume the messages ​

Every event arrives as one SQS message whose MessageBody is the delivery envelope as JSON: the request's metadata, its parsed body, headers and query string.

  • No message attributes and no signature. Queue deliveries aren't signed: your queue policy already decides who can write to the queue. (HTTPS deliveries carry webhook-* signature headers; queue messages don't.)
  • At least once. A message can arrive more than once (a retry after a timeout, a redelivery). metadata.event_id is the same every time, so use it to drop duplicates.
  • Replays reuse event_id. A replay re-sends the same event with a new metadata.posted_at and no other marker. If you de-duplicate on event_id, a replayed event is dropped too, so if you want replays reprocessed, make the handler an idempotent upsert instead.
  • FIFO queues (URL ending in .fifo) get MessageGroupId: "default" and a unique MessageDeduplicationId per send, so content-based deduplication has no effect and all events to that destination keep one order.
  • Size. The whole envelope must fit the queue's message size limit. An event whose envelope is too large fails, is retried, and ends in the DLQ.
  • Inbound email arrives as a notice. An email routed to a queue doesn't carry the mail: data.body is an email.received notice (sender, recipients, subject, attachment list) with API URLs to fetch the content. See Receive email into SQS.

What a message looks like ​

A Shiprocket status update received on an endpoint named "Shiprocket orders". Shiprocket sends charge_info as a string containing JSON; Emithook parses it and lists it in expanded_json_paths:

json
{
  "version": 2,
  "metadata": {
    "event_id": "01M2XHCDW3H02X0AZ0998JJYC5",
    "identifier": "ep_01M0Z3B3DB4W93V3VJMMVKTXGZ",
    "name": "Shiprocket orders",
    "source": "https",
    "url": "https://wh.acme.in/ar/shiprocket",
    "method": "POST",
    "path": "/ar/shiprocket",
    "content_type": "application/json",
    "body_encoding": "json",
    "size": 2351,
    "source_ip": "203.0.113.74",
    "verification": { "status": "unverified", "scheme": "none" },
    "transformed": false,
    "captured_at": "2026-09-19T19:13:10.531Z",
    "posted_at": "2026-09-19T19:13:10.804Z",
    "expanded_json_paths": ["/charge_info"]
  },
  "data": {
    "body": {
      "awb": "77957065756",
      "current_status": "IN TRANSIT",
      "current_status_id": 20,
      "order_id": "#380974",
      "sr_order_id": 1590238168,
      "charge_info": {
        "zone_info": { "is_mongo": true, "is_old_zone": false, "on_new_zone": 0 },
        "other_charges": 0
      },
      "scans": [
        { "date": "2026-09-16 17:01:00", "status": "501-S-PU", "activity": "PICKUP EMPLOYEE IS OUT TO P/U SHIPMENT", "sr-status-label": "OUT FOR PICKUP" }
      ]
    },
    "headers": {
      "content-type": "application/json",
      "user-agent": "curl/8.5.0",
      "x-api-key": "<the provider's key, forwarded as sent>"
    },
    "queries": {}
  }
}

data.body depends on metadata.body_encoding. The other three shapes:

json
{
  "metadata": { "content_type": "application/x-www-form-urlencoded", "body_encoding": "form",
                "expanded_json_paths": ["/payload"], "…": "…" },
  "data": {
    "body": {
      "payload": { "type": "block_actions", "user": { "id": "U01ABC" }, "actions": [{ "action_id": "approve" }] }
    },
    "headers": { "content-type": "application/x-www-form-urlencoded", "x-slack-signature": "v0=…" },
    "queries": {}
  }
}
json
{
  "metadata": { "content_type": "text/plain", "body_encoding": "text", "expanded_json_paths": [], "…": "…" },
  "data": { "body": "order 380974 shipped", "headers": { "content-type": "text/plain" }, "queries": {} }
}
json
{
  "metadata": { "body_encoding": "json", "transformed": true, "expanded_json_paths": [], "…": "…" },
  "data": {
    "body": { "orderId": "#380974", "status": "IN_TRANSIT" },
    "headers": { "content-type": "application/json" },
    "queries": {}
  },
  "original": {
    "body": { "order_id": "#380974", "current_status": "IN TRANSIT", "charge_info": "{\"zone_info\": …}" },
    "body_encoding": "json",
    "headers": { "content-type": "application/json" }
  }
}

A base64 body is the raw bytes, base64-encoded (a file or other binary upload). When a transform ran, data.body is exactly what it returned (nothing expanded) and original carries the request as it arrived.

Types ​

Copy these into your consumer:

ts
export type BodyEncoding = 'json' | 'form' | 'text' | 'base64'

export interface EmithookEnvelope<TBody = unknown> {
  version: 2
  metadata: {
    /** Same on every retry/replay of one event — your idempotency key. */
    event_id: string
    /** Endpoint id (ep_…); for an email with no endpoint, the alias id. Route on this. */
    identifier: string | null
    /** Endpoint name (falls back to its URL path); for email, the alias address. */
    name: string | null
    /** 'https' | 'email' | 'queue' | 'send-api' | … */
    source: string
    url: string | null
    method: string | null
    path: string | null
    /** The inbound request's content type. */
    content_type: string | null
    body_encoding: BodyEncoding
    /** Inbound body size in bytes. */
    size: number
    source_ip: string | null
    verification: { status: 'verified' | 'failed' | 'unverified'; scheme: string }
    /** true ⇔ `original` is present. */
    transformed: boolean
    /** When Emithook received the request (ISO-8601). */
    captured_at: string
    /** When Emithook posted THIS delivery (ISO-8601); new on every retry/replay. */
    posted_at: string
    /** JSON Pointers of body strings that held JSON and were parsed in place. */
    expanded_json_paths: string[]
  }
  data: {
    /** json → parsed value · form → Record<string, string | string[] | object> · text → string · base64 → string */
    body: TBody
    /** Inbound headers minus authorization / cookie / set-cookie / proxy-authorization. */
    headers: Record<string, string>
    /** Query string, one value per key; {} when none. */
    queries: Record<string, string>
  }
  /** Only when a transform ran: the request as it arrived. */
  original?: { body: unknown; body_encoding: BodyEncoding; headers: Record<string, string> }
}

Type each provider's payload on top of it, e.g. EmithookEnvelope<ShiprocketStatus>.

Example: a Node.js consumer ​

With the AWS SDK v3. It routes by endpoint, drops duplicates, and deletes a message only after it was handled:

ts
import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs'
import type { EmithookEnvelope } from './emithook-envelope'

const sqs = new SQSClient({ region: 'ap-south-1' })
const QueueUrl = process.env.QUEUE_URL!

interface ShiprocketStatus {
  awb: string
  order_id: string
  current_status: string
  charge_info?: { other_charges: number; zone_info?: Record<string, unknown> }
}

const handlers: Record<string, (env: EmithookEnvelope<any>) => Promise<void>> = {
  // key = metadata.identifier (the endpoint id)
  ep_01M0Z3B3DB4W93V3VJMMVKTXGZ: async (env: EmithookEnvelope<ShiprocketStatus>) => {
    const s = env.data.body
    await updateShipment(s.order_id, s.awb, s.current_status, s.charge_info?.other_charges ?? 0)
  },
}

while (true) {
  const { Messages = [] } = await sqs.send(
    new ReceiveMessageCommand({ QueueUrl, MaxNumberOfMessages: 10, WaitTimeSeconds: 20 }),
  )
  for (const m of Messages) {
    const env = JSON.parse(m.Body!) as EmithookEnvelope
    if (env.version !== 2) throw new Error(`unsupported envelope version ${env.version}`)
    if (env.metadata.body_encoding !== 'json') {
      console.warn('non-JSON body from', env.metadata.name, env.metadata.body_encoding)
    }
    if (!(await seenBefore(env.metadata.event_id))) {
      await (handlers[env.metadata.identifier ?? ''] ?? logUnrouted)(env)
      await markSeen(env.metadata.event_id)
    }
    await sqs.send(new DeleteMessageCommand({ QueueUrl, ReceiptHandle: m.ReceiptHandle! }))
  }
}

Example: a NestJS DTO ​

With class-validator and class-transformer:

ts
import { Type } from 'class-transformer'
import {
  Allow, Equals, IsArray, IsBoolean, IsDateString, IsIn, IsInt, IsObject,
  IsOptional, IsString, Min, ValidateIf, ValidateNested,
} from 'class-validator'

const ENCODINGS = ['json', 'form', 'text', 'base64'] as const
const nullable = ValidateIf((_o, v) => v !== null)

export class VerificationDto {
  @IsIn(['verified', 'failed', 'unverified']) status!: 'verified' | 'failed' | 'unverified'
  @IsString() scheme!: string
}

export class EnvelopeMetadataDto {
  @IsString() event_id!: string
  @nullable @IsString() identifier!: string | null
  @nullable @IsString() name!: string | null
  @IsString() source!: string
  @nullable @IsString() url!: string | null
  @nullable @IsString() method!: string | null
  @nullable @IsString() path!: string | null
  @nullable @IsString() content_type!: string | null
  @IsIn(ENCODINGS) body_encoding!: (typeof ENCODINGS)[number]
  @IsInt() @Min(0) size!: number
  @nullable @IsString() source_ip!: string | null
  @ValidateNested() @Type(() => VerificationDto) verification!: VerificationDto
  @IsBoolean() transformed!: boolean
  @IsDateString() captured_at!: string
  @IsDateString() posted_at!: string
  @IsArray() @IsString({ each: true }) expanded_json_paths!: string[]
}

export class EnvelopeDataDto<TBody = unknown> {
  /** Shape depends on metadata.body_encoding — validate per provider after routing. */
  @Allow() body!: TBody
  @IsObject() headers!: Record<string, string>
  @IsObject() queries!: Record<string, string>
}

export class EnvelopeOriginalDto {
  @Allow() body!: unknown
  @IsIn(ENCODINGS) body_encoding!: (typeof ENCODINGS)[number]
  @IsObject() headers!: Record<string, string>
}

export class EmithookEnvelopeDto<TBody = unknown> {
  @Equals(2) version!: 2
  @ValidateNested() @Type(() => EnvelopeMetadataDto) metadata!: EnvelopeMetadataDto
  @ValidateNested() @Type(() => EnvelopeDataDto) data!: EnvelopeDataDto<TBody>
  @IsOptional() @ValidateNested() @Type(() => EnvelopeOriginalDto) original?: EnvelopeOriginalDto
}
ts
// in your SQS handler
const env = plainToInstance(EmithookEnvelopeDto, JSON.parse(message.Body))
await validateOrReject(env) // NOT { whitelist: true } / forbidNonWhitelisted — see below

Example: a Python consumer ​

python
import json, boto3

sqs = boto3.client("sqs", region_name="ap-south-1")
QUEUE_URL = "https://sqs.ap-south-1.amazonaws.com/123456789012/orders"

while True:
    resp = sqs.receive_message(QueueUrl=QUEUE_URL, MaxNumberOfMessages=10, WaitTimeSeconds=20)
    for m in resp.get("Messages", []):
        env = json.loads(m["Body"])
        assert env["version"] == 2
        meta, data = env["metadata"], env["data"]
        if not seen_before(meta["event_id"]):
            if meta["identifier"] == "ep_01M0Z3B3DB4W93V3VJMMVKTXGZ" and meta["body_encoding"] == "json":
                body = data["body"]
                update_shipment(body["order_id"], body["awb"], body["current_status"],
                                body.get("charge_info", {}).get("other_charges", 0))
            mark_seen(meta["event_id"])
        sqs.delete_message(QueueUrl=QUEUE_URL, ReceiptHandle=m["ReceiptHandle"])

Consumer checklist ​

  • Route on metadata.identifier, not name or url: it stays the same if the endpoint is renamed or moved to another domain.
  • Check metadata.body_encoding before reading data.body. A provider that sends malformed JSON arrives as text, not as an object.
  • Don't reject unknown fields. Version 2 can gain new metadata fields (as posted_at and expanded_json_paths were added). Reject only a different version.
  • Treat data.headers as sensitive. Only authorization, cookie, set-cookie and proxy-authorization are removed; a provider's own key header (such as x-api-key) is forwarded as sent, so don't log headers wholesale.
  • Latency from receipt to your queue is posted_at − captured_at.
  • Expanded strings are values, not text. If you verify a provider signature over a string that Emithook expanded (listed in expanded_json_paths), rebuild it from the original request in the request log instead. See the delivery envelope.

Emithook · a Finnoto product

© 2026 Emithook