# `GeoGenius.Staging`
[🔗](https://github.com/agoodway/geo_genius/blob/main/lib/geo_genius/staging.ex#L1)

The per-run landing area between an artifact and the catalog.

A provider parses one artifact into rows here, and normalization reads them
back out one at a time. The indirection buys separation: normalization reads
one row shape whatever format the artifact arrived in, the catalog is
written to at a pace no parser sets, and a provider's parsing is testable
against a table rather than against the catalog's constraints.

It is not recovery input. Every attempt stages afresh from its immutable
artifact snapshot, so a pass begins by emptying the run's table -- see
`reset/3`.

The table is created and dropped per run and is `UNLOGGED`, because its
content is reproducible from a checksummed artifact and skipping WAL for a
bulk load of hundreds of thousands of rows is the reason staging is its own
phase.

# `count`

```elixir
@spec count(GeoGenius.Context.t(), Ecto.UUID.t()) :: non_neg_integer()
```

The number of rows currently staged for a run.

# `create`

```elixir
@spec create(GeoGenius.Context.t(), Ecto.UUID.t(), Ecto.UUID.t()) :: String.t()
```

Creates a run's staging table, if it does not already exist, returning its name.

# `drop`

```elixir
@spec drop(GeoGenius.Context.t(), Ecto.UUID.t()) :: :ok
```

Drops a terminal or orphaned run's staging table, if it exists.

# `drop`

```elixir
@spec drop(GeoGenius.Context.t(), Ecto.UUID.t(), Ecto.UUID.t()) :: :ok
```

Drops the current executor's active staging table, if it exists.

# `insert`

```elixir
@spec insert(
  GeoGenius.Context.t(),
  Ecto.UUID.t(),
  Ecto.UUID.t(),
  [GeoGenius.Staging.Row.t()],
  keyword()
) :: non_neg_integer()
```

Bulk-inserts `rows`, returning the number of rows written.

Issues one round trip regardless of batch size: the three fields unzip into
parallel arrays and cross as a single `unnest` insert rather than as one
statement per row. `payload` binds as a list of maps, never as pre-encoded
JSON text -- a string would decode back as a jsonb string value holding an
escaped object rather than the object itself, and no constraint catches
that because a JSON string is valid jsonb.

Returns `0` without touching the database for an empty list, but still
validates `run_id` first, so a malformed run id fails the same way it would
for a non-empty batch rather than being masked by the empty-list shortcut.

`opts` reaches `Repo.query/3`. One batch of a national artifact is a single
multi-megabyte `unnest` insert, which is exactly the shape DBConnection's
fifteen-second default cuts short, so a caller raises `:timeout` here rather
than on its whole Repo. The insert always forces `checkout_retries: 0`: after
a disconnect, only the pipeline can safely decide whether to start a new
import attempt rather than repeating an ambiguously completed mutation.

# `leaked`

```elixir
@spec leaked(GeoGenius.Context.t()) :: [{Ecto.UUID.t(), String.t()}]
```

Every staging table no run still needs.

A table is leaked when its run has reached a terminal status, and also when
no `import_run` row carries its id at all. The second case is the one that
cannot be recovered any other way: an earlier version of this function
joined `import_run` to `pg_class`, so a table whose run row was gone had
nothing to join to, never appeared here, and `mix geo_genius.sweep_staging`
could never drop it. The schema down migration refuses to drop a non-empty schema, so
an unreclaimable table eventually blocks an uninstall outright.

The scan therefore starts from `pg_class` and left-joins `import_run`.
Candidate names must match `staging_` followed by 32 hex characters, and
the run id is read back out of the name, so a table with no run row still
yields the id `GeoGenius.Catalog.drop_staging/2` needs. That id is then
round-tripped: a row survives only when `staging_table_name(id)` reproduces
the table's own name exactly, which keeps PostgreSQL the authority on how a
staging table is named and means nothing whose name this library did not
derive is ever offered up for dropping.

Returns `{run_id, table_name}` pairs ordered by table name.

# `reset`

```elixir
@spec reset(GeoGenius.Context.t(), Ecto.UUID.t(), Ecto.UUID.t()) :: String.t()
```

Empties a run's staging table, creating it if it is not there, and returns
its name.

`create/3` alone is `CREATE UNLOGGED TABLE IF NOT EXISTS`, which is what a
staging pass wants on the second call and not what it wants at the start of
execution. A table can survive an abrupt worker or VM death, or be created
manually before the pipeline starts. Neither case authorizes those rows as
input to a future executor.

So a pass starts from a table that is empty by construction. The table is
dropped rather than truncated, which also resets the identity sequence the
keyset pagination in `stream/3` reads through, and leaves nothing of the
previous attempt's shape behind for this one to inherit.

Nothing durable is lost by emptying it: staging is reproducible scratch
space, not attempt evidence or a recovery checkpoint. A corrected retry has
a new run and stages from its own artifact snapshot. What makes that retry
cheap is the artifact cache, which skips the network.

# `stream`

```elixir
@spec stream(GeoGenius.Context.t(), Ecto.UUID.t(), keyword()) :: Enumerable.t()
```

Streams a run's staged rows in insertion order.

Reads through keyset pagination -- `WHERE id > $1 ORDER BY id LIMIT $2` --
rather than `Ecto.Adapters.SQL.stream/4`, which needs an enclosing
transaction. Normalization writes hundreds of thousands of rows through
separate statements after each row is read, and holding one transaction
open for that whole phase would pin a connection and a snapshot for as long
as the slowest phase in the pipeline runs. Each page carries the last row's
`id` forward as the next page's lower bound rather than an offset, since
`OFFSET` re-scans and a concurrent insert would shift it.

`:batch_size` defaults to 1000 and must be a positive
integer -- a batch size of zero would page with `LIMIT 0`, which always
returns no rows and would silently stream nothing rather than fail.

# `table_name`

```elixir
@spec table_name(Ecto.UUID.t()) :: String.t()
```

The staging table's name for a run.

Derived from the run's uuid, so the identifier is `[0-9a-f_]+` by
construction rather than by escaping. `staging_table_name/1` in SQL applies
the same rule; the test asserts the two agree rather than trusting that they
do.

---

*Consult [api-reference.md](api-reference.md) for complete listing*