Packages
oban
2.17.7
2.23.0
2.22.1
2.22.0
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.4
2.19.3
2.19.2
2.19.1
2.19.0
2.18.3
2.18.2
2.18.1
2.18.0
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.3
2.16.2
2.16.1
2.16.0
2.15.4
2.15.3
2.15.2
2.15.1
2.15.0
2.14.2
2.14.1
2.14.0
2.13.6
2.13.5
2.13.4
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.3
2.11.2
2.11.1
2.11.0
2.10.1
2.10.0
retired
2.9.2
2.9.1
2.9.0
2.8.0
2.7.2
2.7.1
2.7.0
2.6.1
2.6.0
2.5.0
2.4.3
2.4.2
2.4.1
2.4.0
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
1.2.0
1.1.0
1.0.0
1.0.0-rc.2
1.0.0-rc.1
0.12.1
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.0
0.8.1
0.8.0
0.7.1
0.7.0
0.6.0
0.5.0
0.4.0
0.3.0
0.2.0
0.1.0
Robust job processing, backed by modern PostgreSQL, SQLite3, and MySQL.
Current section
Files
Jump to
Current section
Files
lib/oban/migrations/postgres.ex
defmodule Oban.Migrations.Postgres do
@moduledoc false
@behaviour Oban.Migration
use Ecto.Migration
@initial_version 1
@current_version 12
@default_prefix "public"
@doc false
def initial_version, do: @initial_version
@doc false
def current_version, do: @current_version
@impl Oban.Migration
def up(opts) do
opts = with_defaults(opts, @current_version)
initial = migrated_version(opts)
cond do
initial == 0 ->
change(@initial_version..opts.version, :up, opts)
initial < opts.version ->
change((initial + 1)..opts.version, :up, opts)
true ->
:ok
end
end
@impl Oban.Migration
def down(opts) do
opts = with_defaults(opts, @initial_version)
initial = max(migrated_version(opts), @initial_version)
if initial >= opts.version do
change(initial..opts.version, :down, opts)
end
end
@impl Oban.Migration
def migrated_version(opts) do
opts = with_defaults(opts, @initial_version)
repo = Map.get_lazy(opts, :repo, fn -> repo() end)
escaped_prefix = Map.fetch!(opts, :escaped_prefix)
query = """
SELECT description
FROM pg_class
LEFT JOIN pg_description ON pg_description.objoid = pg_class.oid
LEFT JOIN pg_namespace ON pg_namespace.oid = pg_class.relnamespace
WHERE pg_class.relname = 'oban_jobs'
AND pg_namespace.nspname = '#{escaped_prefix}'
"""
case repo.query(query, [], log: false) do
{:ok, %{rows: [[version]]}} when is_binary(version) -> String.to_integer(version)
_ -> 0
end
end
defp change(range, direction, opts) do
for index <- range do
pad_idx = String.pad_leading(to_string(index), 2, "0")
[__MODULE__, "V#{pad_idx}"]
|> Module.concat()
|> apply(direction, [opts])
end
case direction do
:up -> record_version(opts, Enum.max(range))
:down -> record_version(opts, Enum.min(range) - 1)
end
end
defp record_version(_opts, 0), do: :ok
defp record_version(%{prefix: prefix}, version) do
execute "COMMENT ON TABLE #{inspect(prefix)}.oban_jobs IS '#{version}'"
end
defp with_defaults(opts, version) do
opts = Enum.into(opts, %{prefix: @default_prefix, version: version})
opts
|> Map.put_new(:create_schema, opts.prefix != @default_prefix)
|> Map.put_new(:quoted_prefix, inspect(opts.prefix))
|> Map.put_new(:escaped_prefix, String.replace(opts.prefix, "'", "\\'"))
end
end