Source config, secrets, and policy. This module is both the behaviour and the
instance-scoped facade (resolves config.source_store and delegates).
Reads happen on every request, so adapters should be read-mostly and fast.
Reads and writes
fetch/2 and list/1 are the ingest-facing, read-only callbacks every store
implements. The optional admin callbacks (put/5 or put/6, get/3,
list_tenant/2, delete/3 or delete/4) back the admin API's tenant-scoped
source management; a store that only reads (like Ankusa.SourceStore.Static)
leaves them out and the facade answers {:error, :read_only} / :error /
[]. See Ankusa.SourceStore.Persistent for the writable implementation.
Audit
Every change to a source is audited (Ankusa.Audit). A store that keeps its
sources in this node's Ankusa.Store exports put/6 and delete/4, which
take audit: %Ankusa.Audit.Context{} and commit the audit row in the same
synced batch as the change. A store that keeps them elsewhere exports
put/5 and delete/3; the facade then brackets the call with a pending
audit row (a dual write: see Ankusa.Audit).
Three answers to fetch/2
{:ok, source} and :error are verdicts: the source exists, or it does not
(the edge answers 404, dispatch dead-letters the source's rows as
{:source_gone, id}). A store backed by something that can fail — a
database, a network service — returns {:error, :unavailable} when it could
not tell: the edge answers 503 store_unavailable with Retry-After: 1,
dispatch reschedules the row without spending an attempt, and a replay
stops at that entry and retries it on its next tick. Never answer :error
for "I could not look": that turns an outage into lost hooks.
Writes follow the same rule: a store that could not persist a put or
delete answers {:error, :store_unavailable} (the admin API's
503 store_unavailable), never {:error, :invalid, _}, which tells the
caller its request is wrong.
A store that runs a process exports start_link/1. Ankusa.Instance starts
it with [instance: instance], and it reads its options from
Ankusa.config(instance).source_store. Never put configuration in a child
spec: a supervisor prints its children's start arguments.
Summary
Types
One tenant-scoped source as the store keeps it. spec is the submitted JSON
map with string keys, secrets included; redaction is the caller's job (the
admin router), never the store's.
Functions
Delete one tenant-scoped source.
Fetch one tenant-scoped source, or :error if there is no such source.
Every source owned by tenant, or [] for a read-only store.
Create (:create) or replace (:update) one tenant-scoped source.
Whether the store has loaded what it serves. A store that can start empty and
fill from elsewhere (Redis) says so with the optional ready?/1 callback; one
without it is always ready. Ankusa.Lifecycle.reconcile/2 waits for it before
it reads a source to settle a change nobody saw finish.
Types
@type stored() :: %{ :tenant => String.t(), :name => String.t(), :source_id => String.t(), :spec => map(), optional(:revision) => pos_integer() }
One tenant-scoped source as the store keeps it. spec is the submitted JSON
map with string keys, secrets included; redaction is the caller's job (the
admin router), never the store's.
Callbacks
@callback fetch(instance :: atom(), source_id :: String.t()) :: {:ok, Ankusa.Source.t()} | :error | {:error, :unavailable}
@callback put( instance :: atom(), tenant :: String.t(), name :: String.t(), spec :: map(), mode :: :create | :update, opts :: keyword() ) :: {:ok, stored()} | {:error, :invalid, String.t()} | {:error, :retry_exceeds_claim_expiry, String.t()} | {:error, :exists} | {:error, :not_found} | {:error, :store_unavailable}
Functions
@spec delete(atom(), String.t(), String.t(), keyword()) :: :ok | {:error, :not_found} | {:error, :invalid, String.t()} | {:error, :store_unavailable} | {:error, :read_only}
Delete one tenant-scoped source.
tenant and name are validated here, before the store sees them, so a bad
identity is {:error, :invalid, message} no matter which store is configured.
A store that exports neither delete/4 nor delete/3 is read-only:
{:error, :read_only}. Options: audit:, as for put/6.
@spec fetch(atom(), String.t()) :: {:ok, Ankusa.Source.t()} | :error | {:error, :unavailable}
Fetch one tenant-scoped source, or :error if there is no such source.
Every source owned by tenant, or [] for a read-only store.
@spec put(atom(), String.t(), String.t(), map(), :create | :update, keyword()) :: {:ok, stored()} | {:error, :invalid, String.t()} | {:error, :retry_exceeds_claim_expiry, String.t()} | {:error, :exists} | {:error, :not_found} | {:error, :store_unavailable} | {:error, :read_only}
Create (:create) or replace (:update) one tenant-scoped source.
tenant and name are validated here, before the store sees them, so a bad
identity is {:error, :invalid, message} no matter which store is configured.
A store that exports neither put/6 nor put/5 is read-only:
{:error, :read_only}. A source that routes to a claim-checking sink in an
object store whose retry horizon outlasts claim_check.bucket_expiry_days
(Ankusa.ClaimCheck.check_retry_horizon/2) is
{:error, :retry_exceeds_claim_expiry, message}.
Options: audit: the Ankusa.Audit.Context of the change (default
Ankusa.Audit.Context.system/0).
Whether the store has loaded what it serves. A store that can start empty and
fill from elsewhere (Redis) says so with the optional ready?/1 callback; one
without it is always ready. Ankusa.Lifecycle.reconcile/2 waits for it before
it reads a source to settle a change nobody saw finish.