Recording to GCS

Store recorded output in a Google Cloud Storage bucket. This is useful for large results and output that should follow the bucket's retention policy rather than remain in Postgres.

This backend uses Req for requests and Goth for GCS's short-lived OAuth tokens. Add both dependencies, then start Goth in your supervision tree with a service account:

def deps do
  [
    {:req, "~> 0.5"},
    {:goth, "~> 1.4"}
  ]
end

credentials =
  "GOOGLE_APPLICATION_CREDENTIALS_JSON"
  |> System.fetch_env!()
  |> JSON.decode!()

{Goth, name: MyApp.Goth, source: {:service_account, credentials}}

The Module

The module keeps the supervised Goth process in its runtime configuration and fetches a valid token for each request. Only non-sensitive values such as the bucket and process name are stored with jobs, service-account credentials remain in the supervised Goth process.

defmodule MyApp.GCS do
  @behaviour Oban.Pro.Storage

  @impl Oban.Pro.Storage
  def init(opts) do
    bucket = Keyword.get(opts, :bucket, "my-bucket")
    goth = Keyword.get(opts, :goth, MyApp.Goth)

    req = Req.new(base_url: "https://#{bucket}.storage.googleapis.com", decode_body: false)

    %{req: req, goth: goth}
  end

  @impl Oban.Pro.Storage
  def put(key, payload, conf) do
    case request(conf, :put, url: "/#{key}", body: payload) do
      {:ok, %{status: status}} when status in 200..299 -> :ok
      {:ok, response} -> {:error, response}
      {:error, reason} -> {:error, reason}
    end
  end

  @impl Oban.Pro.Storage
  def fetch_all(keys, conf) do
    payloads =
      for {:ok, {key, body}} <- Task.async_stream(keys, &fetch_one(&1, conf), ordered: false),
          not is_nil(body),
          into: %{},
          do: {key, body}

    {:ok, payloads}
  end

  @impl Oban.Pro.Storage
  def delete_all(keys, conf) do
    Enum.each(keys, &request(conf, :delete, url: "/#{&1}"))
  end

  defp fetch_one(key, conf) do
    case request(conf, :get, url: "/#{key}") do
      {:ok, %{status: 200, body: body}} -> {key, body}
      _other -> {key, nil}
    end
  end

  defp request(%{req: req, goth: goth}, method, opts) do
    %{token: token} = Goth.fetch!(goth)

    Req.request(req, [method: method, auth: {:bearer, token}] ++ opts)
  end
end

The recipe fetches batches concurrently and omits missing objects. To keep output after its job is pruned, make delete_all/2 a no-op and use bucket lifecycle rules for retention.

Using It

Configure the backend on a recorded worker and pass non-sensitive options such as the bucket:

use Oban.Pro.Worker, recorded: [storage: {MyApp.GCS, bucket: "oban-output"}]

To send every worker's output to GCS without annotating each one, set it as the compile-time default:

config :oban_pro, Oban.Pro.Worker, recorded: [storage: MyApp.GCS]