StatifierPersistence.Storage.Ecto (StatifierPersistence v0.1.0)

Copy Markdown View Source

The Ecto StatifierPersistence.Storage.Adapter: the storage contract over the schemas a host generates with use StatifierPersistence.Ecto (ADR-0002), against the tables the versioned migrations helper creates. Requires the optional ecto_sql dependency (ADR-0005).

defmodule MyApp.Persistence do
  use StatifierPersistence.Ecto, repo: MyApp.Repo
end

{:ok, store} =
  StatifierPersistence.Storage.new(
    StatifierPersistence.Storage.Ecto,
    persistence: MyApp.Persistence
  )

Options init/1 accepts:

  • :persistence - required, a module that called use StatifierPersistence.Ecto. The repo, the schema modules, and the table names all come from its resolved configuration, so this adapter adds no knobs of its own (ADR-0002 decision 3).
  • :sandbox - when true, isolate/1 checks out an Ecto.Adapters.SQL.Sandbox connection: the hook a test suite (this package's conformance suite included) uses to wrap each test in its own transaction. Default false, and isolate/1 is then a no-op.

Engine identities (content_hash, session_id, run_id) are stored verbatim in text columns and blobs in bytea columns, so both round-trip byte-identically (ADR-0002 decision 1, ADR-0003 decision 1). The identity guard lives in StatifierPersistence.Storage, above this adapter like above every other one (ADR-0003 decision 2); nothing here decodes a blob.

insert_run/2's :run_exists refusal rides the V01 unique index on run_id - one atomic insert, never a check-then-insert. A backend failure a callback cannot observe as a value (the database down, a timeout) raises the driver's own exception rather than being flattened into a default (this package's errors-are-events rule).

Summary

Functions

fetch_chart(opts, content_hash)

Fetches the chart stored under content_hash, or :chart_not_found.

fetch_position(opts, session_id)

Fetches the position stored for session_id, or :position_not_found.

fetch_run(opts, run_id)

Fetches the run stored under run_id, or :run_not_found.

init(opts)

Resolves the :persistence host module into the handle every other callback takes: the host's repo, its three generated schema modules, and its runs table name (for the unique-constraint mapping).

insert_run(opts, run_record)

Inserts run_record, refusing a duplicate run_id with {:error, :run_exists}.

isolate(opts)

Per-test isolation (the optional StatifierPersistence.Storage.Adapter.isolate/1): checks out an Ecto.Adapters.SQL.Sandbox connection when this handle was built with sandbox: true, and is a no-op otherwise.

lock_run(opts, run_id, fun)

Runs fun under per-run mutual exclusion for run_id (the optional StatifierPersistence.Storage.Adapter.lock_run/3, ADR-0004 decision 5 as amended 2026-08-22).

save_chart(opts, chart_record)

Stores chart_record, idempotent on its content_hash: an insert with on_conflict: :nothing against the unique index, so a repeated save of the same hash neither duplicates the row nor rewrites it.

save_position(opts, position_record)

Stores position_record under its session_id, overwriting any position already stored for that session: an upsert replacing the record columns (and updated_at) on the unique index.

update_run(opts, run_record)

Overwrites the run stored under run_record's run_id with the full record, or refuses with :run_not_found.

Functions

fetch_chart(opts, content_hash)

@spec fetch_chart(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.content_hash()
) ::
  {:ok, StatifierPersistence.Storage.Adapter.chart_record()}
  | {:error, StatifierPersistence.Storage.Adapter.error()}

Fetches the chart stored under content_hash, or :chart_not_found.

fetch_position(opts, session_id)

@spec fetch_position(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.session_id()
) ::
  {:ok, StatifierPersistence.Storage.Adapter.position_record()}
  | {:error, StatifierPersistence.Storage.Adapter.error()}

Fetches the position stored for session_id, or :position_not_found.

fetch_run(opts, run_id)

@spec fetch_run(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.run_id()
) ::
  {:ok, StatifierPersistence.Storage.Adapter.run_record()}
  | {:error, StatifierPersistence.Storage.Adapter.error()}

Fetches the run stored under run_id, or :run_not_found.

init(opts)

@spec init(StatifierPersistence.Storage.Adapter.opts()) ::
  {:ok, StatifierPersistence.Storage.Adapter.opts()}
  | {:error, StatifierPersistence.Storage.Adapter.error()}

Resolves the :persistence host module into the handle every other callback takes: the host's repo, its three generated schema modules, and its runs table name (for the unique-constraint mapping).

Refuses a module that never called use StatifierPersistence.Ecto with {:error, {:adapter, {:not_a_persistence_host, module}}}. Makes no database call: reachability surfaces on first use, per call site.

insert_run(opts, run_record)

@spec insert_run(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.run_record()
) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}

Inserts run_record, refusing a duplicate run_id with {:error, :run_exists}.

The refusal is the V01 unique index on run_id speaking: the insert carries a unique_constraint/3 on that index's name, so two concurrent inserts of one run_id cannot both return :ok and no separate existence check ever runs.

isolate(opts)

@spec isolate(StatifierPersistence.Storage.Adapter.opts()) ::
  :ok | {:error, StatifierPersistence.Storage.Adapter.error()}

Per-test isolation (the optional StatifierPersistence.Storage.Adapter.isolate/1): checks out an Ecto.Adapters.SQL.Sandbox connection when this handle was built with sandbox: true, and is a no-op otherwise.

lock_run(opts, run_id, fun)

@spec lock_run(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.run_id(),
  (-> result)
) :: {:ok, result} | {:error, StatifierPersistence.Storage.Adapter.error()}
when result: term()

Runs fun under per-run mutual exclusion for run_id (the optional StatifierPersistence.Storage.Adapter.lock_run/3, ADR-0004 decision 5 as amended 2026-08-22).

Everything happens inside one transaction that spans fun. It takes pg_advisory_xact_lock(hashtextextended(run_id, 0)) first - unconditional per-run exclusion whether or not the run row exists yet - and then SELECT ... FOR UPDATE on the run row when it does, keeping the row itself locked against every other writer for the rest of the transaction. Both locks are transaction-scoped, so any exit from fun releases them: a normal return commits, and a raise rolls back and propagates to the caller with nothing leaked.

save_chart(opts, chart_record)

@spec save_chart(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.chart_record()
) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}

Stores chart_record, idempotent on its content_hash: an insert with on_conflict: :nothing against the unique index, so a repeated save of the same hash neither duplicates the row nor rewrites it.

save_position(opts, position_record)

@spec save_position(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.position_record()
) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}

Stores position_record under its session_id, overwriting any position already stored for that session: an upsert replacing the record columns (and updated_at) on the unique index.

update_run(opts, run_record)

@spec update_run(
  StatifierPersistence.Storage.Adapter.opts(),
  StatifierPersistence.Storage.Adapter.run_record()
) :: :ok | {:error, StatifierPersistence.Storage.Adapter.error()}

Overwrites the run stored under run_record's run_id with the full record, or refuses with :run_not_found.

One update_all/3 keyed on run_id: the match count is the existence check, so refusal and overwrite are a single statement.