ankusa_sdk

The Elixir client SDK for Ankusa deployments: everything an Elixir consumer needs on the subscribe side of the delivery, plus clients for the operator APIs.

• Receive Ankusa.Sink.Http deliveries with a Plug (Ankusa.SDK.Receiver) that hands each hook to your handler module.
• Decode queue messages (Ankusa.SDK.Message) published by Ankusa.Sink.RabbitMQ/Kafka/NATS/Redis, and redeem the claim check when the payload was too large to ride inline. The SDK ships no broker client: bring your own Broadway/AMQP/brod/gnat/Redix consumer.
• Redeem claim checks (Ankusa.SDK.ClaimCheck) with the end-to-end sha256 check the gateway itself does not run.
• Drive the routes, admin, and sources APIs (Ankusa.SDK.Routes, Ankusa.SDK.Admin, Ankusa.SDK.Sources).

The SDK never depends on ankusa core and never defines the bare Ankusa module, so an app may load both.

HTTP sink     Ankusa --> Ankusa.SDK.Receiver ---------------------+
                                                                  |
Queue sinks   Ankusa --> your consumer --> Message.decode/1       +--> MyApp.Hooks.handle_hook/2
                                           Message.to_hook/2 -----+
                                           (redeems a claim through Ankusa.SDK.ClaimCheck)

Install

# mix.exs
def deps do
  [
    {:ankusa_sdk, "~> 0.6"}
  ]
end

Everything the SDK exposes lives under Ankusa.SDK..

Claim-check client

Redeem a claim-check ref, verify the bytes it returns against the message's sha256, and classify failures into dead-letter vs. retry — without holding any object-store credentials.

claim_check = Ankusa.SDK.ClaimCheck.new(ENV.fetch("CLAIM_CHECK_URL", "http://localhost:4001"))

# claim and sha256 are the queue message's fields:
#   "claim":  "urn:ankusa:claim:v1:<tenant>:<claim_id>"  (claim_id: uppercase ULID)
#   "sha256": 64-char lowercase hex of the claim's bytes
case Ankusa.SDK.ClaimCheck.redeem(claim_check, claim, sha256) do
  {:ok, body} ->
    handle(body)

  {:error, %{retryable: false} = error} ->
    # bad ref/sha256, 404, or an integrity mismatch: dead-letter, don't requeue
    dead_letter(error)

  {:error, %{retryable: true} = error} ->
    # gateway unreachable or 5xx: safe to retry
    requeue(error)
end

redeem/3 does three things a bare GET /v1/claims/... doesn't:

1. Parses the ref into its tenant, claim id, and gateway path (Ankusa.SDK.ClaimRef.parse/1, also usable on its own).
2. Fetches the bytes.
3. Verifies them against the message's sha256 (the gateway itself does not check this — see "Redeem a claim" in docs/claim-check.md) before returning them.

Every failure carries retryable, so a consumer needs exactly one bit:

health/1 hits GET /health for a liveness probe.

Webhook receiver

Ankusa.SDK.Receiver is a Plug that reads the raw body, parses the x-ankusa-* headers, calls your handler, and answers 202, 503, or 4xx matching what Ankusa.Sink.Http treats as success.

defmodule MyApp.Hooks do
  @behaviour Ankusa.SDK.Handler

  @impl Ankusa.SDK.Handler
  def handle_hook(%Ankusa.SDK.Hook{} = hook, _arg) do
    # Return :ok only once the hook is durably handled; delivery is
    # at-least-once, so dedupe on the idempotency key
    # (Ankusa.SDK.Idempotency.key/2), not on the arrival alone.
    case MyApp.Store.insert(hook.id, hook.body) do
      :inserted -> :ok
      :duplicate -> :ok
      {:error, reason} -> {:error, reason}
    end
  end
end

Standalone (Bandit):

# mix.exs: {:bandit, "~> 1.12"}
children = [
  {Bandit,
   plug: {Ankusa.SDK.Receiver, handler: MyApp.Hooks, secret: System.fetch_env!("ANKUSA_WHSEC")},
   port: 4200}
]

In a Phoenix endpoint, mount the receiver above the body parsers:

defmodule MyAppWeb.Endpoint do
  use Phoenix.Endpoint, otp_app: :my_app

  plug Ankusa.SDK.Receiver,
    path: "/deliveries",
    handler: MyApp.Hooks,
    secret: System.fetch_env!("ANKUSA_WHSEC")

  plug Plug.Parsers, parsers: [:json], json_decoder: JSON

  # … the rest of the endpoint
end

Options: :handler (a module, or {module, arg}), :secret (required unless allow_unsigned: true, see below), :path (only that path is handled; everything else passes through untouched), :max_body_bytes (default 8_000_000, the same cap core's ingest applies to a raw body), :allow_unsigned and :tolerance_seconds.

Signed deliveries (required by default)

Ankusa.Sink.Http signs every delivery the Standard Webhooks way unless the sink is configured allow_unsigned: true, so the receiver is closed by default: give it the sink's :secret and it verifies each delivery before your handler sees it, answering 401 {"error":"invalid_signature","code":...} otherwise. Without :secret it refuses to start (ArgumentError at init/1), so an unconfigured receiver is never open to any host that can reach its URL:

plug Ankusa.SDK.Receiver,
  handler: MyApp.Hooks,
  secret: System.fetch_env!("ANKUSA_WHSEC"),
  tolerance_seconds: 300

allow_unsigned: true (matching a sink configured the same way) accepts deliveries with no :secret, unverified; a configured :secret is never weakened by it. The signature is judged before x-ankusa-id, so a request that is not a signed delivery is a 401 whatever else it lacks.

Secrets are whsec_ + base64, or any other string used as its own bytes; pass a list during a rotation. The check on its own is Ankusa.SDK.Signature.verify/4, returning {:error, %Ankusa.SDK.InvalidSignatureError{code: ..., field: ...}} (never retryable); Ankusa.SDK.Webhook.receive_delivery/3 is the whole decision (signature, then headers) without Plug. The comparison is :crypto.hash_equals/2.

The handler sees one value whichever transport delivered the hook:

Dedupe on the key Ankusa.SDK.Idempotency.key/2 returns (see "Consuming queue messages" below), not on id alone: a provider retry can reach you as several hook ids.

The header parsing on its own is Ankusa.SDK.Webhook.parse_headers/1, for receivers built on something else.

hook.verification (also on parse_headers/1 and Message.to_hook/2 results) says how the node verified the hook at ingest, nil from a node older than the field. hook.verification.flagged == true means the node accepted a failed check (on_verify_failure: accept_flag) — treat it as unverified. Ankusa.SDK.Hook.trusted?/1 is that decision in one call: true only for status: "ok" with flagged: false, so a flagged, "failed" or "skipped" hook, or one from a node older than the field, is untrusted. It takes a Hook, a decoded Message or the parsed Webhook.Headers.

Queue messages

Ankusa.SDK.Message.decode/1 parses the Ankusa.Sink.Message JSON that RabbitMQ, Kafka, NATS, and Redis sinks publish; Message.to_hook/2 turns it into the same Ankusa.SDK.Hook the Plug delivers, redeeming the claim check when the payload lived in the claim store. A ClaimCheck client is required even for inline messages, so a consumer cannot forget it the day a payload crosses the sink's inline_max_bytes (64 KiB by default).

A consumer of any broker looks the same:

# A Broadway/AMQP/Redix callback, whatever your consumer calls it.
def handle_message(message_body, claim_check) do
  with {:ok, message} <- Ankusa.SDK.Message.decode(message_body),
       {:ok, hook} <- Ankusa.SDK.Message.to_hook(message, claim_check) do
    case MyApp.Hooks.handle_hook(hook, []) do
      :ok -> :ack
      {:error, _reason} -> :requeue
    end
  else
    # Undecodable bytes (InvalidMessageError), a bad ref, a 404, an integrity
    # mismatch: every one is `retryable: false` — dead-letter, don't requeue.
    {:error, %{retryable: false} = error} -> {:dlq, error}
    # An unreachable or 5xx gateway: safe to retry.
    {:error, %{retryable: true} = error} -> {:requeue, error}
  end
end

decode/1 returns an Ankusa.SDK.InvalidMessageError with a stable code ("invalid_json", "not_an_object", "unsupported_version", "invalid_field", "ambiguous_body", "missing_body", "invalid_body_base64", "size_mismatch", "integrity", "tenant_mismatch") and, when the code is "invalid_field", the offending field. It also verifies the inline body's length against size, its bytes against sha256, and a claim's tenant against tenant_id. Keys it doesn't know are ignored, so a newer producer adding a field doesn't break an older consumer.

Every one of those errors is retryable: false: the same bytes fail the same way on every retry, so send them to your terminal queue, count them by code, and never requeue them. The full policy (what is permanent, what is worth a retry, and what to count) is Consumer failure policy.

Kafka/NATS headers and the RabbitMQ routing key are not read: the JSON body carries everything.

The message's v is the wire format's major version: decode/1 accepts every v from Message.min_supported_version/0 to Message.max_supported_version/0 (1 and 2) into the one Message struct, with v as the wire said it. A v: 2 message is a v: 1 message plus additive fields, so nothing else changes for a consumer when the producer moves to it; a v outside the range is unsupported_version (permanent: dead-letter it). message.instance and message.region are the Ankusa instance that published the message and the region it is configured with (nil for none).

Lifecycle events

A message from the instance's lifecycle sinks carries a CloudEvents 1.0 event as its body (a source or route was created, updated or deleted, a tenant was erased, the instance was restored from a backup). Ankusa.SDK.LifecycleEvent.decode/1 parses it from hook.body, with the same InvalidMessageError codes ("unsupported_version" is a specversion other than "1.0"), ignoring the keys and the event types it does not know:

{:ok, event} = Ankusa.SDK.LifecycleEvent.decode(hook.body)
# event.type, event.subject, event.revision, event.sequence, event.data

Consuming queue messages

A queue consumer is at-least-once end to end, so the handler must make its effect idempotent. Decode the message, compute the idempotency key, and insert-or-ignore it into a processed-ids table before doing the work:

defmodule MyApp.Consumer do
  alias Ankusa.SDK.{Idempotency, Message}

  def handle(raw, claim_check) do
    with {:ok, message} <- Message.decode(raw),
         {:ok, hook} <- Message.to_hook(message, claim_check),
         key = Idempotency.key(message) do
      case MyApp.Store.claim(key, hook.id) do
        :inserted -> MyApp.Hooks.handle_hook(hook, [])
        :duplicate -> :ok
      end
    else
      # Undecodable bytes and every bad-ref/404/integrity failure are
      # `retryable: false` — dead-letter, don't requeue.
      {:error, %{retryable: false} = error} -> {:dlq, error}
      # An unreachable or 5xx gateway is safe to retry.
      {:error, %{retryable: true} = error} -> {:requeue, error}
    end
  end
end

The key is the table's primary key, so the database enforces the dedupe:

defmodule MyApp.Repo.Migrations.CreateProcessedHooks do
  use Ecto.Migration

  def change do
    create table(:processed_hooks, primary_key: false) do
      add :idempotency_key, :text, primary_key: true
      add :ankusa_id, :text, null: false
      add :processed_at, :utc_datetime_usec
    end

    create index(:processed_hooks, [:ankusa_id])
  end
end

Ankusa.SDK.Idempotency.key/2 accepts the decoded Message, the parsed Ankusa.SDK.Webhook.Headers, or a Hook. Ankusa computes the key once per hook and ships it (the message's idempotency_key, the x-ankusa-idempotency-key header), and key/2 returns that value. It is tenant:source_id:dedupe_key when the source extracted a provider event key (Stripe's id, GitHub's x-github-delivery, ...) and the hook id otherwise, so a provider's own retries collapse even when ingest minted a fresh hook id for each one, and two tenants that share a provider event id stay apart. For a message or delivery from a node that predates the field key/2 computes the same key itself (tenant default when there is none). A replay of an already-processed delivery computes the same key and is dropped; pass include_replay: true when the consumer must reprocess replays:

key = Ankusa.SDK.Idempotency.key(message, include_replay: true)

Routes client

Manage route definitions and the global IP rules on the route-management listener (routes.admin.port, default 4003).

routes = Ankusa.SDK.Routes.new(ENV.fetch("ROUTES_URL", "http://localhost:4003"))

{:ok, page} = Ankusa.SDK.Routes.list_routes(routes, enabled: true, limit: 10)
{:ok, _} = Ankusa.SDK.Routes.create_route(routes, %{"id" => "stripe", "path" => "/webhooks/stripe"})
{:ok, rules} = Ankusa.SDK.Routes.get_ip_rules(routes)

Functions: health/1, list_routes/2, create_route/2, get_route/2, replace_route/3, update_route/3, delete_route/2, get_ip_rules/1, put_ip_rules/2, test_route/2. Route ids are percent-encoded as one path segment, so /, ?, #, % and a space in an id can't reshape the URL; an id that isn't a string, is empty, or is ./.. is refused with Ankusa.SDK.InvalidRouteIdError before any request is sent.

Failures: InvalidRouteIdError, RouteNotFoundError (404), RoutesRejectedError (any other 4xx, carrying :code, :field, :message, :conflicting_id, :max_routes), and RoutesUnavailableError (5xx, an unfollowed redirect, a non-JSON success body, or unreachable; retryable).

Admin client

The operator API on admin.port (default 4002): health, Prometheus metrics, the redacted config, the DLQ, replay jobs, and the quarantine list.

admin = Ankusa.SDK.Admin.new(ENV.fetch("ADMIN_URL", "http://localhost:4002"))

{:ok, _} = Ankusa.SDK.Admin.health(admin)
{:ok, text} = Ankusa.SDK.Admin.metrics(admin)
{:ok, %{"total" => total}} = Ankusa.SDK.Admin.list_dead_letters(admin, limit: 10)
{:ok, _} = Ankusa.SDK.Admin.list_quarantined(admin)
{:ok, %{"entries" => entries}} = Ankusa.SDK.Admin.list_audit(admin, limit: 10)

# Replay jobs: re-send the dead-letter queue ("dlq"), an archived time window
# ("archive"), or re-verify held hooks and release those that pass ("quarantine").
{:ok, replay} = Ankusa.SDK.Admin.create_replay(admin, %{"kind" => "dlq", "rate" => 500})
{:ok, replay} = Ankusa.SDK.Admin.get_replay(admin, replay["id"])
{:ok, %{"replays" => replays}} = Ankusa.SDK.Admin.list_replays(admin)
{:ok, paused} = Ankusa.SDK.Admin.update_replay(admin, replay["id"], %{"state" => "paused"})

Failures: RoleNotEnabledError (409 role_not_enabled, carrying :role), AdminRejectedError (any other 4xx, carrying :code — a missing job is 404 replay_not_found), and AdminUnavailableError (5xx, an unfollowed redirect, a non-JSON success body, or unreachable; retryable).

Sources client

Sources are tenant-scoped: a source is addressed as <tenant>.<name>, and Ankusa.Admin.Router serves their CRUD API on the same admin.port as the operator API.

sources = Ankusa.SDK.Sources.new(ENV.fetch("ADMIN_URL", "http://localhost:4002"))

{:ok, list} = Ankusa.SDK.Sources.list_sources(sources, "acme")
{:ok, source} = Ankusa.SDK.Sources.get_source(sources, "acme", "billing")

spec = %Ankusa.SDK.Sources.Spec{sinks: [%{"type" => "log"}]}
{:ok, _} = Ankusa.SDK.Sources.create_source(sources, "acme", "billing", spec)
{:ok, _} = Ankusa.SDK.Sources.update_source(sources, "acme", "billing", spec)
:ok = Ankusa.SDK.Sources.delete_source(sources, "acme", "billing")

A read returns an Ankusa.SDK.Sources.Source, which is always redacted — resending a read-back verify map is not the same as resending the stored secret, so supply secrets through Ankusa.SDK.Sources.Spec.

expected_version: "0.3.0" is an optional latch: the client is immutable, so run Sources.verify_version/1 once at startup and keep the returned client; every later call then re-checks the cached version without another /health request, and a mismatch is an Ankusa.SDK.VersionMismatchError before any API call.

{:ok, sources} =
  Ankusa.SDK.Sources.new("http://localhost:4002", expected_version: "0.3.0")
  |> Ankusa.SDK.Sources.verify_version()

Failures, all carrying :status and :body: SourceNotFoundError (404), SourceConflictError (409), SourceStoreReadOnlyError (409 — the deployment's source store is a static seed), SourceInvalidError (400, or an invalid tenant/name caught before any request), VersionMismatchError, and SourcesUnavailableError (unreachable, timed out, or 5xx). Tenants and names must match [A-Za-z0-9_-]{1,64}.

Layout

lib/ankusa/sdk/
  http.ex                    # the shared Req layer (@moduledoc false)
  errors.ex                  # every exception the SDK returns
  hook.ex                    # one delivered hook, either transport
  handler.ex                 # the behaviour a handler implements
  webhook.ex                 # x-ankusa-* header parsing
  receiver.ex                # the Plug that receives HTTP-sink deliveries
  message.ex                 # queue wire format: decode + to_hook
  idempotency.ex             # the idempotency key for any delivery
  claim_ref.ex               # urn:ankusa:claim:v1:<tenant>:<claim_id>
  claim_check.ex             # the claim-check gateway client
  routes.ex                  # the route-management client
  admin.ex                   # the operator client
  sources.ex                 # the tenant-scoped source client
  sources/spec.ex            # the writable spec
  sources/source.ex          # a stored, redacted source
test/
  conformance_test.exs       # runs the language-neutral vectors in conformance/
  support/                   # the mock gateway, transports, recorder

Develop

mix deps.get
mix test                      # unit suites + the conformance vectors
mix test test/conformance_test.exs   # just the vectors

The same vectors every other SDK runs live in conformance/; mise run check:conformance runs them all.