GeoGenius.Runners.PgFlow (geo_genius v0.2.0)

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:

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.
Set config :geo_genius, pgflow_job_timeout_seconds: seconds at compile time, choosing a deadline for the host's bounded workload.
Compile the job into the database with mix pgflow.gen.job_migration GeoGenius.Runners.PgFlow.Job followed by mix ecto.migrate.
Start {PgFlow, repo: MyApp.Repo, jobs: [GeoGenius.Runners.PgFlow.Job]} under its own supervision tree.

Summary

Functions

available?()

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

enqueue(context, run_id, args)

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.

job_context(input, repo)

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

job_input_opts(input)

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

job_outcome(result, run_id)

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

job_timeout_seconds()

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

name()

Returns "pgflow".

Functions

available?()

@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(context, run_id, args)

@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(input, repo)

@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(input)

@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(result, run_id)

@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()

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

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

name()

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

Returns "pgflow".