GeoGenius.Pipeline.State (geo_genius v0.2.0)

One run's working set, threaded through every phase.

Everything durable about a run lives in PostgreSQL; this carries only what one phase hands the next: the rebuilt manifest and the providers it resolves to, the local path and attributed source release of each artifact the download phase obtained, and the metrics the last phase measured, which the next phase boundary writes onto the run.

A release names a provider per source, so there is no single provider for a run. artifact_providers maps a logical name to the module that stages it, which is how staging dispatches on an artifact and how normalizing and relating dispatch on row.artifact. providers is the same set without the keys, in declaration order, for the release-level question relations/1 asks.

Summary

Types

metrics()

What a phase measured, as the jsonb stage_metrics merge patch it becomes.

result()

What every phase function returns, whichever module it lives in.

t()

Functions

with_metrics(error, fun)

Turns a phase's own accumulator into the phase contract.

Types

metrics()

@type metrics() :: %{optional(String.t()) => non_neg_integer()}

What a phase measured, as the jsonb stage_metrics merge patch it becomes.

result()

@type result() :: {:ok, t()} | {:error, String.t()}

What every phase function returns, whichever module it lives in.

t()

@type t() :: %GeoGenius.Pipeline.State{
  artifact_providers: %{optional(String.t()) => module()},
  batch_size: pos_integer(),
  context: GeoGenius.Context.t(),
  executor_id: Ecto.UUID.t(),
  manifest: GeoGenius.Manifest.t() | nil,
  manifest_artifacts: %{optional(String.t()) => GeoGenius.Manifest.Artifact.t()},
  metrics: metrics(),
  opts: keyword(),
  providers: [module()],
  publish?: boolean(),
  resolved: %{optional(String.t()) => Path.t()},
  run: GeoGenius.ImportRun.t(),
  sources: %{optional(String.t()) => Ecto.UUID.t()},
  timeout: timeout(),
  work_dir: Path.t()
}

Functions

with_metrics(error, fun)

@spec with_metrics(term(), (term() -> t())) :: result()

Turns a phase's own accumulator into the phase contract.

An error passes through untouched; anything else is handed to fun, which returns the state the next phase receives. Every phase reduces over something -- artifacts, staged artifacts, batches of rows -- and each one needs the same two-line tail, which is here once rather than in all three phase modules.