README

The client SDK for Ankusa deployments: one Packagist package meant to bundle everything a non-Elixir consumer needs to talk to an Ankusa deployment. Today that's the claim-check gateway client, the route-management and operator (admin) clients, the source-management client, and a webhook-receiving header helper; more clients (ingest) land here as they're built.

This repository is the read-only split mirror of packages/sdk-php in jamescarr/ankusa, published so Packagist — which reads composer.json at a repository root — can serve it. Every push here is generated by the release workflow; development, issues and pull requests happen in the main repository.

Install

composer require jamescarr/ankusa

PHP 8.3 or newer. guzzlehttp/guzzle and its PSR-7/PSR-18 interfaces are installed as dependencies; any other PSR-18 client can be injected instead.

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. Conforms to the framework's own contract, priv/openapi/claim_check.v1.yaml: the spec is the source of truth, this package conforms to it, not the reverse.

use Ankusa\ClaimCheck\ClaimCheckClient;
use Ankusa\ClaimCheck\ClaimCheckError;

$claimCheck = new ClaimCheckClient(getenv('CLAIM_CHECK_URL') ?: 'http://localhost:4001');

// A queue message that carries a claim also carries its sha256:
//   "claim":  "urn:ankusa:claim:v1:<tenant>:<claim_id>"  (claim_id: uppercase ULID)
//   "sha256": 64-char lowercase hex of the claim's bytes
function resolveBody(ClaimCheckClient $claimCheck, array $message): string
{
    try {
        return $claimCheck->redeem($message['claim'], $message['sha256']);
    } catch (ClaimCheckError $err) {
        if (! $err->isRetryable()) {
            // bad ref/sha256, 404, or an integrity mismatch: dead-letter, don't requeue
            throw $err;
        }
        // gateway unreachable or 5xx: safe to retry
        throw $err;
    }
}

redeem() does three things GET /v1/claims/... alone doesn't:

Parses the ref into its tenant id, claim id, and gateway path (GET /v1/claims/{tenant_id}/{claim_id}) — ParsedClaimRef::parse(), which is also public on its own.
Fetches the bytes.
Verifies them against the message's sha256 (the gateway itself does not check this, see "Redeem a claim" in docs/claim-check.md) before ever returning them to you.

Every failure is a ClaimCheckError with an isRetryable(), so a consumer needs exactly one bit to decide dead-letter vs. retry:

Class	`isRetryable()`	Cause
`InvalidClaimRefError`	`false`	`$ref` isn't a well-formed claim-check URN, or `$sha256` isn't 64-char lowercase hex
`ClaimNotFoundError`	`false`	gateway `404`: expired by retention, or never written
`ClaimRejectedError`	`false`	gateway `4xx` other than `404`, `408`, `429` (public `$status`, `$body`)
`ClaimIntegrityError`	`false`	sha256 of the returned bytes doesn't match
`ClaimCheckUnavailableError`	`true`	gateway `5xx`, `408`, `429`, or unreachable

health() hits GET /health for a liveness probe.

Webhook receiver helper

Every receiver of Ankusa's HTTP sink needs the same handful of headers off each request; HookHeaders::fromHeaders() replaces the hand-rolled $headers[...] lookups with one call and a typed result. It takes a PSR-7 message or a plain array (getallheaders(), Symfony's HeaderBag::all()), and looks the names up case-insensitively either way:

use Ankusa\Webhook\HookHeaders;
use Ankusa\Webhook\MissingHookIdError;

try {
    $hook = HookHeaders::fromHeaders(getallheaders());
} catch (MissingHookIdError) {
    http_response_code(400);

    return;
}

$body = file_get_contents('php://input');
// $hook->id, $hook->source, $hook->tenant, $hook->contentType, $hook->idempotencyKey

HookHeaders->id is what a receiver identifies a delivery by: delivery is at-least-once (see "HTTP handoff" in docs/integrations.md), so the same hook can arrive twice after a retry. Dedupe on $hook->idempotencyKey(): it returns the tenant-scoped key Ankusa shipped in x-ankusa-idempotency-key ($hook->idempotencyKey is the parsed header, null for a sender that predates it, in which case the method computes the key).

Verifying signed deliveries

An HTTP sink with a secret signs every delivery the Standard Webhooks way:

use Ankusa\Webhook\InvalidSignatureError;
use Ankusa\Webhook\Signature;

try {
    // the raw request body, exactly as received
    Signature::verify(getallheaders(), file_get_contents('php://input'), [getenv('ANKUSA_WHSEC')]);
} catch (InvalidSignatureError $e) {
    http_response_code(401);
    echo json_encode(['error' => 'invalid_signature', 'code' => $e->errorCode]);
    exit;
}

Secrets are whsec_ + base64, or any other string used as its own bytes; pass several during a rotation. $toleranceSeconds (default 300) bounds the webhook-timestamp window. Signatures are compared with hash_equals; failures carry $errorCode and $field, never retryable.

Receiving a delivery (signed by default)

Ankusa's HTTP sink signs every delivery unless it is configured allow_unsigned: true, so a receiver is closed by default. Receiver::receive() is the one call that decides: with the sink's $secrets it verifies as above; with none, and no $allowUnsigned, it refuses every delivery (InvalidSignatureError, $errorCode invalid_secret) — otherwise any host that can reach your URL could inject hooks. A configured secret is never weakened by $allowUnsigned. The signature is judged before the hook id, so a delivery that fails it is a 401 even when it also lacks x-ankusa-id (MissingHookIdError, 400).

use Ankusa\Webhook\InvalidSignatureError;
use Ankusa\Webhook\MissingHookIdError;
use Ankusa\Webhook\Receiver;

try {
    $received = Receiver::receive(getallheaders(), file_get_contents('php://input'), [getenv('ANKUSA_WHSEC')]);
    // $received->headers is the HookHeaders; $received->verified is false only
    // for an unsigned delivery `allowUnsigned: true` let through.
} catch (InvalidSignatureError $e) {
    http_response_code(401);
    echo json_encode(['error' => 'invalid_signature', 'code' => $e->errorCode]);
    exit;
} catch (MissingHookIdError) {
    http_response_code(400);
    exit;
}

$hook->verification (also on Receiver::parseHeaders() and Receiver::decodeMessage() results) says how the node verified the hook at ingest, null 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. $hook->isTrusted() (on Message, HookHeaders and Verification) 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.

Consuming queue messages

The message's v is the wire format's major version: Message::decode() accepts every v from Message::MIN_SUPPORTED_VERSION to Message::MAX_SUPPORTED_VERSION (1 and 2) into the one Message, with $v as the wire said it. A v: 2 message is a v: 1 message plus additive fields, so a consumer reads it the same way; any other v 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 (null for none). A message from the instance's lifecycle sinks carries a CloudEvents event as its body: LifecycleEvent::decode($body).

A queue consumer (Redis or a broker; the HTTP sink sends the original body plus x-ankusa-* headers, read with HookHeaders) decodes one v1 message, verifies its integrity, and reads the key it dedupes on before it touches the effect. Message::decode() is the whole decoder — it refuses anything that doesn't verify, so a decoded message is safe to act on:

use Ankusa\Message\Message;
use Ankusa\Message\InvalidMessageError;

try {
    $message = Message::decode($payload);
} catch (InvalidMessageError $err) {
    // never retryable: dead-letter it
    error_log("bad message: {$err->errorCode} " . ($err->field ?? ''));
    return;
}

$body = $message->body ?? $claimCheck->redeem($message->claim, $message->sha256);
$key  = $message->idempotencyKey();

idempotencyKey() returns the key Ankusa computed once for the hook and shipped as the message's idempotency_key ($message->idempotencyKey is that field): tenant:source_id:dedupe_key when the provider event key is present, else the id; delivery is at-least-once, so this is the stable handle when a provider retries, and two tenants that share a provider event id keep distinct keys. For a message from a node that predates the field the method computes the same key itself (tenant default when there is none). It ignores replay_id by default — a replayed event you already processed stays deduped. A consumer that must reprocess replays calls idempotencyKey(includeReplay: true), which appends #replay:<replay_id>.

The key is meant to be a row in a processed-ids table, written with the effect:

CREATE TABLE processed_webhooks (
    idempotency_key text PRIMARY KEY,
    ankusa_id       text        NOT NULL,
    processed_at    timestamptz NOT NULL DEFAULT now()
);

$stmt = $pdo->prepare(
    'INSERT INTO processed_webhooks (idempotency_key, ankusa_id) VALUES (:key, :id)
     ON CONFLICT (idempotency_key) DO NOTHING'
);
$stmt->execute(['key' => $key, 'id' => $message->id]);

if ($stmt->rowCount() === 0) {
    return; // already processed this provider event
}

// ... apply the effect, in the same transaction ...

A raw HookHeaders (the HTTP sink's headers) answers the same method, so a receiver and a queue consumer share one dedupe rule:

$key = HookHeaders::fromHeaders(getallheaders())->idempotencyKey();

Routes client

Manage route definitions and the global IP rules on the route-management listener (routes.admin.port, default 4003) — the routes tag of priv/openapi/admin.v1.yaml.

use Ankusa\Routes\RoutesClient;

$routes = new RoutesClient(getenv('ROUTES_URL') ?: 'http://localhost:4003');

$routes->createRoute(['id' => 'stripe', 'path' => '/webhooks/stripe']);
$routes->getIpRules();   // ['default' => 'allow', 'rules' => []]
$routes->testRoute(['method' => 'POST', 'path' => '/webhooks/stripe', 'ip' => '203.0.113.7']);

Methods: health(), listRoutes(), createRoute(), getRoute(), replaceRoute(), updateRoute(), deleteRoute(), getIpRules(), putIpRules(), testRoute(). Route ids are percent-encoded as one path segment, so /, ?, #, % and a space in an id can't reshape the URL.

Failures are RoutesError subclasses: InvalidRouteIdError (an empty id, or exactly ./.. — raised before any request, because a URL parser would otherwise normalize it into the collection endpoint and hand back the list page as if it were a route), RouteNotFoundError (404), RoutesRejectedError (any other 4xx, carrying $errorCode, $field, $detail — the body's message, since PHP's message is taken by Throwable — $conflictingId and $maxRoutes), 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, the quarantine list, and the audit trail — the operations, dlq, replays, quarantine and audit tags of admin.v1.yaml.

use Ankusa\Admin\AdminClient;

$admin = new AdminClient(getenv('ADMIN_URL') ?: 'http://localhost:4002');

$admin->health();                          // ['status' => 'ok', 'instance' => ..., 'roles' => [...]]
$admin->listDeadLetters(['limit' => 10]);  // ['total' => ..., 'entries' => [...]]
$admin->listQuarantined();

// Replay jobs. A job is node-local, so a fleet replay is one job per node:
$job = $admin->createReplay(['kind' => 'dlq', 'source_id' => 'demo', 'rate' => 500]);
$admin->getReplay($job['id']);              // the Replay object
$admin->listReplays();                      // ['replays' => [...]], newest first
$admin->updateReplay($job['id'], ['state' => 'paused']);  // or 'running'/'cancelled'

// The audit trail, newest first, paged by cursor/limit and filtered by
// since/until/actor/action/tenant:
$page = $admin->listAudit(['tenant' => 'acme', 'limit' => 50]);
$page->entries;       // list<AuditEntry>
$page->nextCursor;    // pass as `cursor` for the next page; null on the last

createReplay() returns the created job (HTTP 202) or, when an equivalent job is already running or paused, that job (HTTP 200) — so a proxy retry is idempotent. Methods: health(), metrics() (Prometheus text), config(), listDeadLetters(), createReplay(), getReplay(), listReplays(), updateReplay(), listQuarantined(), listAudit() (returns an AuditPage of AuditEntry). Failures are AdminError subclasses: RoleNotEnabledError (409 role_not_enabled, carrying $role), AdminRejectedError (any other 4xx, carrying $errorCode), and AdminUnavailableError (5xx, an unfollowed redirect, or unreachable; retryable).

Sources client

Ankusa's ingest 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. SourcesClient speaks in those terms and builds the paths for you.

use Ankusa\Sources\SourcesClient;
use Ankusa\Sources\SourceSpec;

$sources = new SourcesClient(getenv('ADMIN_URL') ?: 'http://localhost:4002');

$sources->listSources('acme');                                   // Source[]
$sources->getSource('acme', 'billing');                          // Source
$sources->createSource('acme', 'billing', new SourceSpec(sinks: [['type' => 'log']]));
$sources->updateSource(
    'acme',
    'billing',
    new SourceSpec(sinks: [['type' => 'log']], onVerifyFailure: 'reject'),
);
$sources->deleteSource('acme', 'billing');

Methods: serverVersion(), listSources(), getSource(), createSource(), updateSource(), deleteSource(). A write takes the whole spec (SourceSpec, whose toJson() omits unset fields); a read returns a Source, which is always redacted — resending a read-back verify map is not the same as resending the stored secret, so supply secrets through SourceSpec.

new SourcesClient($url, '0.3.0') is an optional latch: the first API call fetches GET /health, compares its version field, and raises VersionMismatchError on a mismatch (the version is cached afterwards, so no further request checks it). Failures are SourcesError subclasses, each 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; getMessage() carries the server's own message or its error code), VersionMismatchError, and SourcesUnavailableError (unreachable, timed out, or 5xx).

Tenants and source names must match ^[A-Za-z0-9_-]{1,64}$; anything else raises SourceInvalidError before a path is built.

Errors and retries

Every exception this SDK throws implements Ankusa\AnkusaException, so one catch covers everything the package can raise. Each client's errors share a base class as well (ClaimCheckError, RoutesError, AdminError, SourcesError).
The claim-check, routes and admin hierarchies answer isRetryable(): false for a rejected request or an invalid input caught locally, true for an unreachable listener, a timeout, a 5xx, or an unfollowed redirect. Sorting a failure into dead-letter vs. retry needs no status-code knowledge.
Every client takes an optional PSR-18 client as its last constructor argument. The default is Guzzle, which never follows redirects and never throws on a status code; an injected client owns its own timeout and redirect policy, though the SDK still treats any 3xx as a failure. Client headers ride on every request.
Request bodies are JSON objects: createReplay() and updateReplay() send the spec/patch they're given, never []. Nested empty objects are the caller's job — pass (object) [] for a nested {}, or [] for a nested [].

Layout

src/
  AnkusaException.php      # marker implemented by every exception this SDK throws
  Version.php              # the version the release tooling reads
  Internal/HttpTransport.php  # @internal: query/body encoding + PSR-18 call
  Webhook/                 # x-ankusa-* header parsing for HTTP-sink receivers
  Message/                 # v1 queue-message decoding + idempotency key
  ClaimCheck/              # the claim-check gateway client
  Routes/                  # the route-management client (routes.admin.port)
  Admin/                   # the operator client (admin.port)
  Sources/                 # the tenant-scoped source-management client (admin.port)
tests/
  Unit/
  Conformance/            # runs the language-neutral vectors in conformance/
  Support/RecordingHttpClient.php

A future client (say, an ingest helper) gets its own src/<Name>/ directory with the same shape.

Develop

composer install
composer test          # PHPUnit: unit + conformance
composer conformance   # PHPUnit: only the conformance vectors in ../conformance
composer analyse       # PHPStan, level 10
composer lint          # php-cs-fixer check --diff
composer fix           # php-cs-fixer fix

composer test runs the same vectors every other SDK in the monorepo runs (conformance/check.mjs, mise run check:conformance); the runner starts PHP's built-in server on a free port to stand in for a deployment's gateway.

jamescarr / ankusa

Package info

Security

Statistics