AshArcadic tour — Ash on ArcadeDB

# Hex install (the published packages):
Mix.install([
  {:ash_arcadic, "~> 1.0"},
  # `replicant` is the CDC transport (Postgres logical replication → decoded
  # transactions) for the CDC-mirror section near the end. ash_arcadic declares
  # it `optional: true`, so a tour without that section can drop this line.
  {:replicant, "~> 1.2"}
])

# Local dev from a checkout instead: path-install both.
#   Mix.install([
#     {:ash_arcadic, path: Path.join(__DIR__, "..")},
#     {:replicant, path: Path.join(__DIR__, "../../replicant")}
#   ])

What this is

A runnable tour of AshArcadic — the Ash DataLayer for ArcadeDB. It walks the full surface against a real ArcadeDB: CRUD + upserts + atomics, query push-down, aggregates + calculations, keyset streaming, concurrent bulk writes, fail-closed multitenancy, vector search (dense, sparse, hybrid), graph traversal over edges, and the CDC mirror.

You need a running ArcadeDB. One docker command:

docker run --rm -d --name arcadedb-tour -p 2480:2480 \
  -e JAVA_OPTS="-Darcadedb.server.rootPassword=playwithdata" \
  arcadedata/arcadedb:latest

Then set the two environment variables below (Livebook: use the Secrets panel or export them before livebook server). The tour creates a throwaway database and drops it at the end — it never touches existing data.

Connect

url = System.get_env("ARCADEDB_URL", "http://localhost:2480")
password = System.get_env("ARCADEDB_PASSWORD", "playwithdata")

# A throwaway database for this tour (dropped in the last cell).
db = "tour_" <> Base.encode16(:crypto.strong_rand_bytes(3), case: :lower)

admin = Arcadic.connect(url, db, auth: {"root", password})
:ok = Arcadic.Server.create_database!(admin, db)

# The data layer asks the HOST for its connection through one small behaviour —
# the `repo` analog. Stash the tour's database name where the client can read it.
Application.put_env(:ash_arcadic_tour, :database, db)

defmodule Tour.ArcadicClient do
  @behaviour AshArcadic.Client

  @impl true
  def conn do
    url = System.get_env("ARCADEDB_URL", "http://localhost:2480")
    password = System.get_env("ARCADEDB_PASSWORD", "playwithdata")
    db = Application.fetch_env!(:ash_arcadic_tour, :database)
    Arcadic.connect(url, db, auth: {"root", password}, timeout: 15_000)
  end
end

{:ok, true} = Arcadic.Server.health?(admin)

Define resources

One document resource (CRUD, queries, keyset, bulk, vector dense + sparse + hybrid full-text) and one person resource (edges + graph traversal). Both are multitenant — AshArcadic's flagship posture: every read, write, traversal, vector search, and keyset cursor is fail-closed tenant-scoped.

defmodule Tour.Domain do
  use Ash.Domain, validate_config_inclusion?: false

  resources do
    allow_unregistered? true
  end
end

defmodule Tour.Doc do
  use Ash.Resource, domain: Tour.Domain, data_layer: AshArcadic.DataLayer

  arcade do
    client Tour.ArcadicClient
    label :Doc
    vector_index :embedding, dimensions: 3, similarity: :cosine
    sparse_vector_index :sparse_embedding, tokens: :tokens, weights: :weights
  end

  attributes do
    attribute :id, :string, primary_key?: true, allow_nil?: false, public?: true
    attribute :org_id, :string, public?: true
    attribute :title, :string, public?: true
    attribute :body, :string, public?: true
    attribute :score, :integer, public?: true
    attribute :published_at, :utc_datetime, public?: true
    attribute :embedding, {:array, :float}, public?: true
    attribute :tokens, {:array, :integer}, public?: true
    attribute :weights, {:array, :float}, public?: true
  end

  calculations do
    calculate :banner, :string, expr(title <> "!"), public?: true
  end

  multitenancy do
    strategy :attribute
    attribute :org_id
  end

  actions do
    default_accept [:id, :org_id, :title, :body, :score, :published_at, :embedding,
                    :tokens, :weights]
    defaults [:create, :update, :destroy]

    read :read do
      primary? true
      pagination keyset?: true, offset?: true, countable: true, required?: false
    end

    create :upsert do
      upsert? true
    end

    update :bump do
      change atomic_update(:score, expr(score + 5))
    end

    read :semantic_search do
      argument :query_vector, {:array, :float}, allow_nil?: false
      argument :k, :integer, allow_nil?: false
      prepare {AshArcadic.Preparations.VectorSearch, index: :embedding}
    end

    read :sparse_search do
      argument :query_tokens, {:array, :integer}, allow_nil?: false
      argument :query_weights, {:array, :float}, allow_nil?: false
      argument :k, :integer, allow_nil?: false

      prepare {AshArcadic.Preparations.VectorSearch, kind: :sparse, index: :sparse_embedding}
    end

    read :hybrid_search do
      argument :query_vector, {:array, :float}, allow_nil?: false
      argument :query_tokens, {:array, :integer}, allow_nil?: false
      argument :query_weights, {:array, :float}, allow_nil?: false
      argument :text_query, :string, allow_nil?: false
      argument :k, :integer, allow_nil?: false

      prepare {AshArcadic.Preparations.VectorSearch,
               kind: :hybrid,
               arms: [{:dense, :embedding}, {:sparse, :sparse_embedding}, {:fulltext, :body}],
               fusion: :rrf}
    end
  end
end

defmodule Tour.Person do
  use Ash.Resource, domain: Tour.Domain, data_layer: AshArcadic.DataLayer

  arcade do
    client Tour.ArcadicClient
    label :Person

    edge :knows do
      label :KNOWS
      direction :outgoing
      destination Tour.Person
      properties [:since]
    end
  end

  attributes do
    attribute :id, :string, primary_key?: true, allow_nil?: false, public?: true
    attribute :org_id, :string, public?: true
    attribute :name, :string, public?: true
  end

  multitenancy do
    strategy :attribute
    attribute :org_id
  end

  relationships do
    # Graph traversal AS an Ash relationship: everyone reachable over :KNOWS
    # within 1..2 hops — per-hop tenant-scoped (every node AND edge on the path).
    has_many :network, __MODULE__ do
      manual(
        {AshArcadic.ManualRelationships.Traverse,
         edge_label: :KNOWS, direction: :outgoing, min_depth: 1, max_depth: 2}
      )
    end
  end

  actions do
    default_accept [:id, :org_id, :name]
    defaults [:read, :create]

    update :befriend do
      require_atomic? false
      argument :to, {:array, :string}
      argument :since, :string
      change {AshArcadic.Changes.CreateEdge, edge: :knows, to: :to}
    end
  end
end

CRUD & upsert

Plain Ash. The MERGE-based upsert is idempotent — run the cell twice, still one row.

require Ash.Query

Ash.create!(Tour.Doc, %{id: "d1", title: "Graphs", body: "graph graph databases", score: 10}, tenant: "org1")
Ash.create!(Tour.Doc, %{id: "d2", title: "Vectors", body: "embeddings nearest", score: 30}, tenant: "org1")

# Upsert: first call creates, second call (same id) updates in place.
Tour.Doc
|> Ash.Changeset.for_create(:upsert, %{id: "d3", title: "Cypher", body: "match merge", score: 20}, tenant: "org1")
|> Ash.create!()

Tour.Doc
|> Ash.Changeset.for_create(:upsert, %{id: "d3", title: "OpenCypher", body: "match merge", score: 25}, tenant: "org1")
|> Ash.create!()

docs = Tour.Doc |> Ash.Query.sort(score: :asc) |> Ash.read!(tenant: "org1")
Enum.map(docs, &{&1.id, &1.title, &1.score})

Atomic updates

An atomic_update change pushes down to a Cypher SET n.score = n.score + $p — the arithmetic runs in the engine, not in your app (also available on creates and upserts, and in bulk).

doc = Ash.get!(Tour.Doc, "d1", tenant: "org1")

updated =
  doc
  |> Ash.Changeset.for_update(:bump, %{}, tenant: "org1")
  |> Ash.update!()

%{score_after: updated.score, score_before: doc.score}

Query push-down

Filters, sorts, and counts compile to parameterized Cypher — including temporal comparisons (AshArcadic wraps datetime/time params in ArcadeDB's native temporal constructors, so they compare temporal-to-temporal, not string-to-native).

Ash.update!(Ash.get!(Tour.Doc, "d1", tenant: "org1"), %{published_at: ~U[2024-06-15 12:00:00Z]})
Ash.update!(Ash.get!(Tour.Doc, "d2", tenant: "org1"), %{published_at: ~U[2023-12-31 23:59:59Z]})

high_scores =
  Tour.Doc
  |> Ash.Query.filter(score > 15 and title != "nope")
  |> Ash.Query.sort(score: :desc)
  |> Ash.read!(tenant: "org1")

recent =
  Tour.Doc
  |> Ash.Query.filter(published_at > ^~U[2024-01-01 00:00:00Z])
  |> Ash.read!(tenant: "org1")

page = Ash.read!(Tour.Doc, tenant: "org1", page: [limit: 2, count: true])

%{
  high_scores: Enum.map(high_scores, &{&1.id, &1.score}),
  recent: Enum.map(recent, & &1.id),
  total: page.count
}

Aggregates & calculations

Query aggregates run as one tenant-scoped Cypher statement each; expression calculations load in Elixir (or push down into filters/sorts).

count = Ash.count!(Tour.Doc, tenant: "org1")
avg = Ash.avg(Tour.Doc, :score, tenant: "org1")
max = Ash.max(Tour.Doc, :score, tenant: "org1")

# A loaded expression calculation:
banners =
  Tour.Doc
  |> Ash.Query.load(:banner)
  |> Ash.Query.sort(score: :desc)
  |> Ash.read!(tenant: "org1")
  |> Enum.map(& &1.banner)

%{count: count, avg: avg, max: max, banners: banners}

Keyset streaming

can?(:keyset) powers cursor pagination and Ash.stream! — bounded memory over any result size, correct across duplicate sort values (primary-key tiebreaker).

for i <- 1..25 do
  Ash.create!(
    Tour.Doc,
    %{id: "s#{String.pad_leading(to_string(i), 2, "0")}", title: "Doc #{i}", score: rem(i, 5)},
    tenant: "org1"
  )
end

# Cursor pagination: walk a page, then ask for what follows it.
page1 = Tour.Doc |> Ash.Query.sort(score: :asc) |> Ash.read!(tenant: "org1", page: [limit: 5])
cursor = List.last(page1.results).__metadata__.keyset

page2 =
  Tour.Doc
  |> Ash.Query.sort(score: :asc)
  |> Ash.read!(tenant: "org1", page: [limit: 5, after: cursor])

# Or just stream the whole set — keyset batches under the hood.
streamed =
  Tour.Doc
  |> Ash.Query.sort(score: :asc)
  |> Ash.stream!(tenant: "org1", batch_size: 10)
  |> Enum.count()

%{page1: length(page1.results), page2: length(page2.results), streamed_total: streamed}

Concurrent bulk writes

With transaction: false each batch is one autocommit UNWIND statement, and AshArcadic retries optimistic-lock conflicts at two levels (ArcadeDB's server-side statement retry + client jittered backoff) — so concurrent bulk converges, even on a freshly auto-created type.

rows = for i <- 1..200, do: %{id: "b#{i}", title: "Bulk #{i}", score: rem(i, 10)}

result =
  Ash.bulk_create(rows, Tour.Doc, :create,
    tenant: "org1",
    transaction: false,
    max_concurrency: 8,
    batch_size: 25,
    return_errors?: true
  )

count = Tour.Doc |> Ash.Query.filter(contains(id, "b")) |> Ash.count!(tenant: "org1")
%{status: result.status, persisted: count}

Multitenancy is fail-closed

Same database, different tenant — none of org1's rows are reachable. The discriminator is injected by the data layer on every path (reads, writes, bulk, traversal, vector, keyset cursors); a missing tenant on a tenant-strategy resource is an error, never a global read.

org2_sees = Ash.read!(Tour.Doc, tenant: "org2")
org1_count = Ash.count!(Tour.Doc, tenant: "org1")
%{org2_sees: length(org2_sees), org1_count: org1_count}

Vector search — dense, sparse, hybrid

Declare the indexes in the arcade block (compile-checked metadata); the host creates the physical indexes once (no migration machinery). Results come back ranked with distance/score in record metadata — tenant-scoped like everything else.

# One-time host-side schema (pairs with the declarations): the vector properties'
# physical types, the data, then the indexes.
Arcadic.command!(admin, "CREATE PROPERTY Doc.embedding ARRAY_OF_FLOATS", %{}, language: "sql")
Arcadic.command!(admin, "CREATE PROPERTY Doc.tokens ARRAY_OF_INTEGERS", %{}, language: "sql")
Arcadic.command!(admin, "CREATE PROPERTY Doc.weights ARRAY_OF_FLOATS", %{}, language: "sql")
Arcadic.command!(admin, "CREATE PROPERTY Doc.body STRING", %{}, language: "sql")

Ash.update!(Ash.get!(Tour.Doc, "d1", tenant: "org1"), %{
  embedding: [1.0, 0.0, 0.0], tokens: [10, 20], weights: [1.0, 0.5]
})

Ash.update!(Ash.get!(Tour.Doc, "d2", tenant: "org1"), %{
  embedding: [0.0, 1.0, 0.0], tokens: [20, 30], weights: [1.0, 0.5]
})

Ash.update!(Ash.get!(Tour.Doc, "d3", tenant: "org1"), %{
  embedding: [0.9, 0.1, 0.0], tokens: [10, 30], weights: [0.9, 0.1]
})

Arcadic.Vector.create_dense_index!(admin, "Doc", "embedding", 3, similarity: :cosine)
:ok = Arcadic.FullText.create_index(admin, "Doc", "body")
:ok = Arcadic.Vector.create_sparse_index(admin, "Doc", "tokens", "weights")

# Dense kNN — ranked by distance, closest first.
nearest =
  Tour.Doc
  |> Ash.Query.for_read(:semantic_search, %{query_vector: [1.0, 0.0, 0.0], k: 2})
  |> Ash.read!(tenant: "org1", authorize?: false)

dense = Enum.map(nearest, &{&1.id, &1.__metadata__[:vector_distance]})

# Sparse (learned-sparse / BM25-style) kNN — ranked by score, higher = better.
sparse =
  Tour.Doc
  |> Ash.Query.for_read(:sparse_search, %{query_tokens: [10], query_weights: [1.0], k: 2})
  |> Ash.read!(tenant: "org1", authorize?: false)

sparse_hits = Enum.map(sparse, &{&1.id, &1.__metadata__[:vector_score]})

# Hybrid fusion — dense + sparse + full-text arms fused (RRF) into one list.
hybrid =
  Tour.Doc
  |> Ash.Query.for_read(:hybrid_search, %{
    query_vector: [1.0, 0.0, 0.0],
    query_tokens: [10],
    query_weights: [1.0],
    text_query: "graph",
    k: 3
  })
  |> Ash.read!(tenant: "org1", authorize?: false)

hybrid_hits = Enum.map(hybrid, &{&1.id, &1.__metadata__[:vector_score]})

%{dense: dense, sparse: sparse_hits, hybrid: hybrid_hits}

Graph: edges & traversal

Edges are first-class writes (CreateEdge/DestroyEdge changes); traversal is an Ash relationship, so it composes with loading, aggregates, and policies. The traversal predicate scopes every node and edge on the path to the tenant.

for {id, name} <- [{"ann", "Ann"}, {"bo", "Bo"}, {"cy", "Cy"}] do
  Ash.create!(Tour.Person, %{id: id, name: name}, tenant: "org1")
end

# ann -> bo -> cy over :KNOWS edges (edge writes ride an update action).
Tour.Person
|> Ash.get!("ann", tenant: "org1")
|> Ash.Changeset.for_update(:befriend, %{to: ["bo"], since: "2024"}, tenant: "org1")
|> Ash.update!()

Tour.Person
|> Ash.get!("bo", tenant: "org1")
|> Ash.Changeset.for_update(:befriend, %{to: ["cy"], since: "2025"}, tenant: "org1")
|> Ash.update!()

# ann's 1..2-hop network: bo (direct) AND cy (through bo).
ann = Tour.Person |> Ash.get!("ann", tenant: "org1") |> Ash.load!(:network, tenant: "org1")
ann.network |> Enum.map(& &1.name) |> Enum.sort()

CDC mirror — Postgres→ArcadeDB effect-once sink

An optional subsystem (AshArcadic.Replicant.*) mirrors a Postgres logical-replication stream into an ArcadeDB graph projection exactly once — over the sibling replicant CDC transport. In production replicant decodes the WAL into %Replicant.Transaction{} structs and delivers them to the sink; here we hand-build one to show the effect-once contract end-to-end against the same throwaway DB.

The mirror is a normal AshArcadic graph resource carrying the AshArcadic.Replicant extension: replicant do source_table ... tenant_attribute ... end declares which Postgres {schema, table} it mirrors and which source column carries the tenant. It is seam-locked — a forbid_if always() policy forbids ordinary writes, so only the sink writes (bypassing with authorize?: false).

defmodule Tour.MirrorDoc do
  use Ash.Resource,
    domain: Tour.MirrorDomain,
    validate_domain_inclusion?: false,
    data_layer: AshArcadic.DataLayer,
    extensions: [AshArcadic.Replicant],
    authorizers: [Ash.Policy.Authorizer]

  arcade do
    client Tour.ArcadicClient
    label :MirrorDoc
  end

  replicant do
    source_table "docs"
    tenant_attribute :org
  end

  attributes do
    attribute :id, :string, primary_key?: true, allow_nil?: false, public?: true
    attribute :org, :string, allow_nil?: false, public?: true
    attribute :title, :string, public?: true
  end

  multitenancy do
    strategy :attribute
    attribute :org
  end

  actions do
    default_accept [:id, :org, :title]
    defaults [:read, :create, :update, :destroy]
  end

  # Seam-lock: forbid ordinary writes so only the sink writes (authorize?: false).
  policies do
    policy always() do
      forbid_if always()
    end
  end
end

# The sink resolves its mirror targets from a domain's REGISTERED resources, so the mirror
# lives in its own explicitly-registered domain (not the tour's allow_unregistered? one — an
# unregistered domain yields an empty resolver index, which fails closed :empty_index).
defmodule Tour.MirrorDomain do
  use Ash.Domain, validate_config_inclusion?: false

  resources do
    resource Tour.MirrorDoc
  end
end

The checkpoint is an ArcadeDB-resident watermark vertex (one integer LSN per slot); the sink is a Replicant.Sink impl baked with the host's config. The watermark advance rides the SAME Ash.transaction as the mirrored writes — that co-commit is the whole effect-once mechanism. The checkpoint client: must point at the same database as the mirror.

defmodule Tour.ReplicantCheckpoint do
  use AshArcadic.ReplicantCheckpoint,
    domain: Tour.Domain,
    client: Tour.ArcadicClient
end

defmodule Tour.ReplicantSink do
  use AshArcadic.ReplicantSink,
    domains: [Tour.MirrorDomain],
    checkpoint_resource: Tour.ReplicantCheckpoint,
    slot_name: "tour"
end

Feed a hand-built transaction (LSN 10, one insert) through handle_transaction/1. It upserts the mirror vertex AND advances the watermark in one session, returning {:ok, 10}. The source record is string-keyed (Postgres column names), and org — the tenant_attribute — resolves the tenant the sink writes under.

txn = %Replicant.Transaction{
  commit_lsn: 10,
  changes: [
    %Replicant.Change{
      op: :insert,
      schema: "public",
      table: "docs",
      record: %{"id" => "d1", "org" => "acme", "title" => "hello"}
    }
  ]
}

{:ok, 10} = Tour.ReplicantSink.handle_transaction(txn)

# The mirrored vertex (peek past the seam-lock with authorize?: false).
Tour.MirrorDoc
|> Ash.read!(tenant: "acme", authorize?: false)
|> Enum.map(&{&1.id, &1.org, &1.title})

The durable watermark now reads 10.

Tour.ReplicantSink.checkpoint()

Re-deliver the SAME commit_lsn: 10 — a replay. The integer <= gate skips it: it returns the held watermark and applies nothing (dup = 0). The vertex count is unchanged.

before = Tour.MirrorDoc |> Ash.count!(tenant: "acme", authorize?: false)
{:ok, 10} = Tour.ReplicantSink.handle_transaction(txn)
after_replay = Tour.MirrorDoc |> Ash.count!(tenant: "acme", authorize?: false)
%{replay: :no_op, count_before: before, count_after: after_replay}

A higher commit_lsn: 11 update upserts in place and advances the watermark to 11.

txn2 = %Replicant.Transaction{
  commit_lsn: 11,
  changes: [
    %Replicant.Change{
      op: :update,
      schema: "public",
      table: "docs",
      record: %{"id" => "d1", "org" => "acme", "title" => "hello (v2)"},
      old_record: %{"id" => "d1", "org" => "acme"}
    }
  ]
}

{:ok, 11} = Tour.ReplicantSink.handle_transaction(txn2)

titles = Tour.MirrorDoc |> Ash.read!(tenant: "acme", authorize?: false) |> Enum.map(&{&1.id, &1.title})
%{titles: titles, watermark: Tour.ReplicantSink.checkpoint()}

Cleanup

Arcadic.Server.drop_database(admin, db)

Where next

• usage-rules.md — the per-feature fine print: supported operators, sortable types, vector scoping rules, bulk-write concurrency, documented limitations.
• README.md — capability matrix and connection setup.
• usage-rules.md → CDC mirror — the full Postgres→ArcadeDB effect-once sink contract: the eight fail-closed consumer contracts, the checkpoint / sink / pipeline wiring, and the REPLICA IDENTITY FULL precondition for a tenant-scoped mirror.
• Troubleshooting — the error catalog: every fail-closed error and its fix.
• Telemetry — the full event catalog.
• Combinations (union/intersect/except), update_many, traversal aggregates, query-scoped bulk update/destroy — all live; see the test suite for working examples of every capability.