# `GeoGenius.Runners.PgFlow`
[🔗](https://github.com/agoodway/geo_genius/blob/main/lib/geo_genius/runners/pgflow.ex#L1)

Runs an import as a durable PgFlow background job.

`pgflow` `~> 0.7.0` is an optional dependency. It establishes
compile order when a host opts into PgFlow, so `PgFlow.Job` is available
before this file conditionally defines `GeoGenius.Runners.PgFlow.Job`.
Hosts that do not install PgFlow do not receive it or compile the job.

PgFlow's dashboard code also compiles against its optional Phoenix,
Phoenix LiveView, and LiveFilter packages when those packages are present.
GeoGenius declares matching optional edges so its own integration build can
exercise the real job module once that build supplies its job deadline. A
host using a PgFlow release that compiles those dashboard modules
unconditionally must opt into those packages too; they are not made
mandatory by GeoGenius.

The job is a single-step definition (`@job queue: :geo_genius_import`)
whose `perform :execute` block calls `GeoGenius.Pipeline.execute/3` through
`run/2`, `job_context/2`, `job_input_opts/1`, and `job_outcome/2` below.

`available?/0` reports whether the PgFlow engine is present, and `enqueue/3`
verifies the host obligations required to accept this particular job. The
engine probe takes three things, all checked at runtime
rather than assumed from configuration:

  * `pgflow` itself is loaded;
  * `GeoGenius.Runners.PgFlow.Job` -- this project's own submodule, not
    `pgflow`'s -- is loaded too;
  * `PgFlow.Supervisor` -- the process PgFlow's own supervision tree
    registers itself under once `{PgFlow, repo: ..., jobs: [...]}` has
    actually started -- is running.

Before enqueueing, the runner also verifies that the job's flow was compiled
into PgFlow's catalog and that at least one healthy worker is polling its
queue. A half-wired PgFlow host therefore fails immediately with the exact
setup remedy instead of raising a database error or leaving an import queued
until a distant await timeout.

A host that satisfies the first two but never started that supervisor
gets the same "unavailable" answer as a host that never added `pgflow` at
all, exactly as `GeoGenius.Runners.Task` treats a configured but dead
`Task.Supervisor` name.

`enqueue/3` hands `GeoGenius.Runners.PgFlow.Job` a JSON-serializable
input: `run_id`, `publish`, `stale_after_seconds`, and `prefix`. It never
sends `repo`. A durable job outlives the process that enqueued it, so its
handler uses the Repo in the `%PgFlow.Context{}` supplied by the worker
instead of reviving a module atom decoded out of storage or resolving a
second application-global Repo. Before enqueueing, this runner verifies
that the operation's Repo is the same Repo PgFlow is configured to use.
`prefix` carries no code-execution risk -- it is a validated string -- so
it travels through the job input unchanged.

PgFlow's `@job timeout:` is compile-time definition data, not a per-enqueue
option. GeoGenius therefore does not choose a universal deadline for work
whose size it cannot know. A host opting into this runner must set a
positive `config :geo_genius, :pgflow_job_timeout_seconds` before compiling
GeoGenius, then compile and migrate the job definition. Each import must
also carry an explicit `:stale_after_seconds` smaller than that deadline.
That relation is a necessary safety floor, not an estimate of total import
duration; the host remains responsible for choosing a job deadline that
covers its complete workload.

A host that installs `pgflow` wires this up once:

  1. Add `{:pgflow, "~> 0.7.0"}` to `mix.exs`.
     For PgFlow releases whose dashboard compiles unconditionally, also add
     `:phoenix`, `:phoenix_live_view`, and `:livefilter` at PgFlow's
     supported versions.
  2. Set `config :geo_genius, pgflow_job_timeout_seconds: seconds` at compile
     time, choosing a deadline for the host's bounded workload.
  3. Compile the job into the database with `mix pgflow.gen.job_migration
     GeoGenius.Runners.PgFlow.Job` followed by `mix ecto.migrate`.
  4. Start `{PgFlow, repo: MyApp.Repo, jobs: [GeoGenius.Runners.PgFlow.Job]}`
     under its own supervision tree.

# `available?`

```elixir
@spec available?() :: boolean()
```

Reports whether `pgflow`, this project's own `Job` submodule, and a
running `PgFlow.Supervisor` are all present right now.

All three are checked live rather than assumed from configuration. A host
that added the dependency but never chose a compile-time job deadline, or
compiled the job but never started
`{PgFlow, repo: ..., jobs: [...]}` (`Process.whereis/1`) gets the same
`false` a host that never touched `pgflow` at all gets -- the same
distinction `GeoGenius.Runners.Task` draws between a configured `Task.Supervisor`
name and one that is dead.

# `enqueue`

```elixir
@spec enqueue(GeoGenius.Context.t(), Ecto.UUID.t(), GeoGenius.Runner.args()) ::
  :ok | {:error, GeoGenius.Runner.enqueue_error()}
```

Enqueues the import as a PgFlow job and returns once PgFlow has durably
queued it. Before writing the job, verifies that its flow is compiled and a
healthy worker is polling the queue.

Returns `:ok` once `PgFlow.enqueue/2` has written the job's own run row --
the import's outcome will be recorded in the catalog once a worker picks
the job up, so the caller reads it back from there rather than polling
this call. Missing runtime availability, flow registration, or worker
readiness is a definite `:not_enqueued` rejection. A generic error after
calling `PgFlow.enqueue/2` is `:outcome_unknown`, because its durable write
may have committed before the caller lost confirmation.

# `job_context`

```elixir
@spec job_context(map(), module()) :: GeoGenius.Context.t()
```

Rebuilds the `Context` a job runs under, from its decoded input.

Lives here for the same reason `job_input_opts/1` does. `repo` comes from
the `%PgFlow.Context{}` PgFlow builds for the executing worker rather than
from a module atom carried in `input` or a second application-global
lookup. `:prefix` travels through `input` as given, already validated once
when the enqueuing context was built.

# `job_input_opts`

```elixir
@spec job_input_opts(map()) :: keyword()
```

Builds the `Pipeline.execute/3` opts a job runs with, from its decoded input.

Lives here, on the module that always compiles, for the same reason
`job_outcome/2` does: `Job` does not compile in the no-optional-dependencies
build, so that gate would otherwise miss a renamed key or dropped conversion. Reads
both `"publish"` and `"stale_after_seconds"` through `Runner.publish?/1`
and `Runner.stale_after_seconds/1` rather than `Map.get/2` directly, since
`input` has already round-tripped through PgFlow's own JSON storage and
carries string keys, not the atom keys a `Runner.args()` built in-process
would.

# `job_outcome`

```elixir
@spec job_outcome(
  {:ok, GeoGenius.ImportRun.t()}
  | {:noop, GeoGenius.ImportRun.t()}
  | {:error, GeoGenius.ImportRun.t()}
  | {:error, {:unrecorded, :not_started | :outcome_unknown, String.t()}},
  Ecto.UUID.t()
) :: %{required(String.t()) => String.t()}
```

Maps one `GeoGenius.Pipeline.execute/3` result to the plain,
JSON-serializable outcome `GeoGenius.Runners.PgFlow.Job`'s `perform` step
hands back to PgFlow.

Lives here, on the module that always compiles, rather than inside `Job`
itself, so the mapping remains covered even in a no-optional-dependencies
build. Otherwise nothing there would notice if
`GeoGenius.Pipeline.execute/3` gained or renamed a return shape --
`Job.run/2` would keep its stale clauses, and the first host running this
backend under real `pgflow` would get a `CaseClauseError` inside a
worker instead of a clean, tested outcome.

Both a completed run and one that ran and recorded its own failure return
a plain outcome map -- either way the catalog already holds what
happened, and that is where a caller reads it back from. Raises only when
`Pipeline.execute/3` reports nothing was recorded at all, since that is
the one case nothing durable exists for a later reader to find.

# `job_timeout_seconds`

```elixir
@spec job_timeout_seconds() :: pos_integer() | nil
```

Returns the host's compile-time PgFlow job deadline, or `nil` when none was chosen.

# `name`

```elixir
@spec name() :: String.t()
```

Returns `"pgflow"`.

---

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