ankusa.sdk.sources — io.github.jamescarr/ankusa-clj 0.1.0

The tenant-scoped source-management API: the admin listener (admin.port, default 4002) serves list/get/create/update/delete for a tenant's ingest sources.

A source is addressed as <tenant>.<name>; this client speaks in those terms and builds the paths for you. Only [A-Za-z0-9_-]{1,64} tenants and names are accepted, and both are checked before any path is built, so a caller-supplied name cannot escape its tenant through URL normalization.

A spec is {:sinks [...] :verify {...} :on-verify-failure "reject"}. :sinks is required by the server and must be non-empty. :verify is optional: absent means the server's default ({"type" "none"}), because omitting it is not the same as re-sending a secret the API never returns. nil fields are omitted from the body.

A source is {:tenant :name :source-id :ingest-path :verify :on-verify-failure :sinks}. Every value is redacted by the server (secrets, passwords, tokens, header values), so a source read back is never useful for editing: resending its :verify is not the same as resending the stored secret.

The version latch

:expected-version is an optional safety latch: when set, every API call first makes sure GET /health reported that version, and a mismatch is a VersionMismatchError before the request is sent.

The client is immutable, so the fetched version is returned rather than remembered. Run verify-version once at startup and keep the returned client; every later call then re-checks the cached version without another /health request:

(def sources (sources/verify-version (sources/client url {:expected-version "0.3.0"})))
(sources/list-sources sources "acme")

Failures: SourceNotFoundError (404), SourceConflictError (409), SourceStoreReadOnlyError (409 source_store_read_only: the deployment's source store is a static seed), SourceInvalidError (400, or an invalid tenant or name caught before any request), VersionMismatchError, and SourcesUnavailableError (unreachable, timed out, or 5xx). Every one carries :status and :body (nil when no response was involved).

client clj

(client base-url)

(client base-url opts)

Build a client for the admin listener at base-url (an absolute http(s) URL).

opts may hold:

• :headers: a map of header name (string or keyword) to string value, sent on every request.
• :timeout-ms: a positive integer, default 10000. It bounds the whole call, from connecting through the last body byte.
• :expected-version: a version string to latch against, or nil for no latch.
• :transport: a function that performs one request, to replace the JDK HTTP client (a different HTTP library, an in-process fake). It receives {:method "GET" :url "http://host/path?q" :headers {"name" "value"} :body <byte[] or nil> :timeout-ms 10000} and returns {:status 200 :headers {...} :body <byte[] or nil>}. Throwing any Exception is a transport failure. It must not follow redirects.

Bad input throws IllegalArgumentException.

create-source clj

(create-source client tenant name spec)

Create a source: POST /v1/tenants/{tenant}/sources. The name travels in the body alongside the spec; the tenant comes from the URL.

delete-source clj

(delete-source client tenant name)

Delete a source: DELETE /v1/tenants/{tenant}/sources/{name}. Returns nil on 2xx (204 with an empty body). A source that still has undelivered hooks on the server's node raises SourceConflictError whose body's error is source_has_deliveries (with pending/inflight counts); this never sends the admin API's ?deliveries=dead_letter.

get-source clj

(get-source client tenant name)

One source: GET /v1/tenants/{tenant}/sources/{name}.

list-sources clj

(list-sources client tenant)

A tenant's sources, as a vector: GET /v1/tenants/{tenant}/sources.

server-version clj

(server-version client)

The deployment's Ankusa version: verify-version, then the cached "version" string.

update-source clj

(update-source client tenant name spec)

Replace a source: PUT /v1/tenants/{tenant}/sources/{name}. The name comes from the URL; the spec carries none.

verify-version clj

(verify-version client)

Fetch GET /health when the version is not cached yet, enforce :expected-version, and return the client with the version cached.

Run it once at startup and keep the returned client; later calls then skip the probe.