Arcadic.Transport behaviour (Arcadic v0.7.0)

The transport seam (charter D2). One implementation ships — Arcadic.Transport.HTTP (Req/Finch). A future Bolt adapter implements the same callbacks. Callbacks are SEMANTIC (mode/session), not HTTP-verb-shaped, so a non-HTTP transport can honor them. This behaviour is also the mock seam consumed by ash_arcadic's tests.

request is the logical statement bundle %{statement: String.t(), params: map(), language: String.t()}; the transport shapes its own wire format from it.

Summary

Types

plan_result()

request()

result()

Callbacks

batch_ingest(t, ndjson, opts)

Bulk-ingest NDJSON vertex/edge records (POST /api/v1/batch/<db>). The body is already-serialized NDJSON iodata; opts carries the query params (:light_edges, :commit_every). Edge endpoints resolve by the vertex @id temp key in the body, not a query param. Optional — HTTP implements it; Bolt has no batch endpoint. Returns the parsed {verticesCreated, edgesCreated, elapsedMs, idMapping} map.

begin(t, opts)

Begin a session transaction; returns the session id.

commit(t)

Commit the session carried by conn.session_id.

create_record(t, type, props, opts)

Create a single record of type from a property map (gRPC CreateRecord). Returns the @rid.

database_exists?(t, name)

Whether a database exists.

database_info(t)

Database-level info for conn.database — %{database, type, records, classes, size_bytes}. Fields a transport can't cheaply provide are nil (HTTP's schema:database gives name/size but not record/class counts; gRPC GetDatabaseInfo gives all). Optional — HTTP and gRPC implement it.

delete_record(t, rid, opts)

Delete a single record by rid (gRPC DeleteRecord).

execute(t, mode, request, opts)

Run a read (:read → idempotent endpoint) or write (:write) statement.

execute_async(t, request, opts)

Fire-and-forget write (server enqueues, does not await). Optional — HTTP implements it.

execute_with_index(t, mode, request, opts)

Like execute/4 for a read/write but also returns the server's HA commit index (X-ArcadeDB-Commit-Index response header) as {:ok, rows, index} — index is nil on a single (non-HA) server where the header is absent. Optional — HTTP implements it; Bolt has no HA header. Used by query_bookmarked/command_bookmarked.

explain(t, request, opts)

Run an EXPLAIN/PROFILE statement and return its plan. The request.statement already carries the EXPLAIN/PROFILE prefix (the facade prepends it). Optional. Returns %{plan: <human string>, plan_tree: <raw, transport-defined map>, rows: <executed rows>}.

health?(t)

Liveness probe (GET /api/v1/health → 204).

insert_rows(t, target_class, rows, opts)

Bulk-insert rows (property maps) into target_class — document/table ingest (gRPC BulkInsert), distinct from batch_ingest's graph vertex/edge batch. opts carries :conflict_mode/:key_columns. Returns the InsertSummary-shaped counts map (per-row errors surfaced value-free as %{row_index, code}). Optional — gRPC implements it; HTTP/Bolt do not.

list_databases(t)

List all database names (GET /api/v1/databases).

Mint a session token from the conn's credentials (POST /api/v1/login).

logout(t)

Revoke the current session (POST /api/v1/logout).

lookup_record(t, rid, opts)

Look up a single record by rid (gRPC LookupByRid). {:ok, map} or {:ok, nil} if absent.

query_stream(t, request, opts)

Lazily stream a large read result as raw row maps (Bolt cursor, or HTTP @rid offset paging). Optional.

ready?(t)

Server readiness.

rollback(t)

Roll back the session carried by conn.session_id.

server_command(t, command)

Run a server-level admin command (create/drop database).

server_get(t, path)

Authenticated admin GET returning a decoded JSON map (HTTP-only admin surface).

transaction(t, function, opts)

Native fun-based transaction (for transports whose sessions are not detachable, e.g. Bolt). Optional.

ts_latest(t, params, opts)

Fetch the newest point (GET /api/v1/ts/<db>/latest). params is a [{String.t(), String.t()}] query-param list (type required; at most one tag — the server applies only the FIRST, probed). Returns %{columns, latest} (atomized). Optional — HTTP-only.

ts_prom_get(t, op, params, opts)

PromQL read family (GET /api/v1/ts/<db>/prom/api/v1/…). op is the allowlisted route atom — :query | :query_range | :labels | :series | {:label_values, label} (the label is facade-validated AND URL-encoded here). Unwraps the Prometheus envelope: status:"success" → {:ok, data}; status:"error" → a typed error. Optional — HTTP-only.

ts_query(t, body, opts)

Run a time-series query (POST /api/v1/ts/<db>/query). body is the already-shaped wire map. Returns the RAW %{columns, rows, count} or AGGREGATED %{aggregations, buckets, count} shape (atomized top-level keys; bucket maps atomized to %{timestamp, values}). Optional — HTTP-only.

ts_write(t, lines, opts)

Write raw InfluxDB line protocol to POST /api/v1/ts/<db>/write (204 on success). lines is already-built line-protocol iodata; opts[:precision] is the validated "ns"|"us"|"ms"|"s" string (the FACADE validates — an invalid value is silently ignored server-side, probed 26.7.2). Append-only, no dedup: a lost response + naive retry duplicates every point. Optional — HTTP-only.

update_record(t, rid, props, opts)

Update a single record by rid (gRPC UpdateRecord; partial merge, or replace: true).

Types

plan_result()

@type plan_result() ::
  {:ok, %{plan: String.t(), plan_tree: map(), rows: [map()]}}
  | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

request()

@type request() :: %{statement: String.t(), params: map(), language: String.t()}

result()

@type result() ::
  {:ok, [map()]} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Callbacks

batch_ingest(t, ndjson, opts)

(optional)

@callback batch_ingest(Arcadic.Conn.t(), ndjson :: iodata(), opts :: keyword()) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

begin(t, opts)

(optional)

@callback begin(Arcadic.Conn.t(), opts :: keyword()) ::
  {:ok, String.t()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Begin a session transaction; returns the session id.

commit(t)

(optional)

@callback commit(Arcadic.Conn.t()) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Commit the session carried by conn.session_id.

create_record(t, type, props, opts)

(optional)

@callback create_record(
  Arcadic.Conn.t(),
  type :: String.t(),
  props :: map(),
  opts :: keyword()
) ::
  {:ok, String.t()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Create a single record of type from a property map (gRPC CreateRecord). Returns the @rid.

database_exists?(t, name)

(optional)

@callback database_exists?(Arcadic.Conn.t(), name :: String.t()) ::
  {:ok, boolean()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Whether a database exists.

database_info(t)

(optional)

@callback database_info(Arcadic.Conn.t()) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

delete_record(t, rid, opts)

(optional)

@callback delete_record(Arcadic.Conn.t(), rid :: String.t(), opts :: keyword()) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Delete a single record by rid (gRPC DeleteRecord).

execute(t, mode, request, opts)

@callback execute(Arcadic.Conn.t(), mode :: :read | :write, request(), opts :: keyword()) ::
  result()

Run a read (:read → idempotent endpoint) or write (:write) statement.

execute_async(t, request, opts)

(optional)

@callback execute_async(Arcadic.Conn.t(), request(), opts :: keyword()) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Fire-and-forget write (server enqueues, does not await). Optional — HTTP implements it.

execute_with_index(t, mode, request, opts)

(optional)

@callback execute_with_index(
  Arcadic.Conn.t(),
  mode :: :read | :write,
  request(),
  opts :: keyword()
) ::
  {:ok, [map()], integer() | nil}
  | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

explain(t, request, opts)

(optional)

@callback explain(Arcadic.Conn.t(), request(), opts :: keyword()) :: plan_result()

health?(t)

(optional)

@callback health?(Arcadic.Conn.t()) ::
  {:ok, boolean()} | {:error, Arcadic.TransportError.t()}

Liveness probe (GET /api/v1/health → 204).

insert_rows(t, target_class, rows, opts)

(optional)

@callback insert_rows(
  Arcadic.Conn.t(),
  target_class :: String.t(),
  rows :: [map()],
  opts :: keyword()
) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

list_databases(t)

(optional)

@callback list_databases(Arcadic.Conn.t()) ::
  {:ok, [String.t()]} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

List all database names (GET /api/v1/databases).

login(t)

(optional)

@callback login(Arcadic.Conn.t()) ::
  {:ok, String.t()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Mint a session token from the conn's credentials (POST /api/v1/login).

logout(t)

(optional)

@callback logout(Arcadic.Conn.t()) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Revoke the current session (POST /api/v1/logout).

lookup_record(t, rid, opts)

(optional)

@callback lookup_record(Arcadic.Conn.t(), rid :: String.t(), opts :: keyword()) ::
  {:ok, map() | nil} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Look up a single record by rid (gRPC LookupByRid). {:ok, map} or {:ok, nil} if absent.

query_stream(t, request, opts)

(optional)

@callback query_stream(Arcadic.Conn.t(), request(), opts :: keyword()) ::
  {:ok, Enumerable.t()}
  | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Lazily stream a large read result as raw row maps (Bolt cursor, or HTTP @rid offset paging). Optional.

ready?(t)

(optional)

@callback ready?(Arcadic.Conn.t()) ::
  {:ok, boolean()} | {:error, Arcadic.TransportError.t()}

Server readiness.

rollback(t)

(optional)

@callback rollback(Arcadic.Conn.t()) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Roll back the session carried by conn.session_id.

server_command(t, command)

(optional)

@callback server_command(Arcadic.Conn.t(), command :: String.t()) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Run a server-level admin command (create/drop database).

server_get(t, path)

(optional)

@callback server_get(Arcadic.Conn.t(), path :: String.t()) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Authenticated admin GET returning a decoded JSON map (HTTP-only admin surface).

transaction(t, function, opts)

(optional)

@callback transaction(Arcadic.Conn.t(), (Arcadic.Conn.t() -> result), opts :: keyword()) ::
  {:ok, result} | {:error, term()}
when result: var

Native fun-based transaction (for transports whose sessions are not detachable, e.g. Bolt). Optional.

ts_latest(t, params, opts)

(optional)

@callback ts_latest(
  Arcadic.Conn.t(),
  params :: [{String.t(), String.t()}],
  opts :: keyword()
) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

ts_prom_get(t, op, params, opts)

(optional)

@callback ts_prom_get(
  Arcadic.Conn.t(),
  op :: atom() | {:label_values, String.t()},
  params :: [{String.t(), String.t()}],
  opts :: keyword()
) ::
  {:ok, map() | list()}
  | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

ts_query(t, body, opts)

(optional)

@callback ts_query(Arcadic.Conn.t(), body :: map(), opts :: keyword()) ::
  {:ok, map()} | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

ts_write(t, lines, opts)

(optional)

@callback ts_write(Arcadic.Conn.t(), lines :: iodata(), opts :: keyword()) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

update_record(t, rid, props, opts)

(optional)

@callback update_record(
  Arcadic.Conn.t(),
  rid :: String.t(),
  props :: map(),
  opts :: keyword()
) ::
  :ok | {:error, Arcadic.Error.t() | Arcadic.TransportError.t()}

Update a single record by rid (gRPC UpdateRecord; partial merge, or replace: true).