if Code.ensure_loaded?(Ecto) do defmodule Honeydew.EctoSource.SQL.Postgres do @moduledoc false alias Honeydew.EctoSource alias Honeydew.EctoSource.SQL alias Honeydew.EctoSource.State @behaviour SQL @impl true def integer_type do :bigint end @impl true def table_name(schema) do source = schema.__schema__(:source) prefix = schema.__schema__(:prefix) if prefix do prefix <> "." <> source else source end end @impl true def ready do SQL.far_in_the_past() |> timestamp_in_msecs |> msecs_ago end @impl true def delay_ready(state) do "UPDATE #{state.table} SET #{state.lock_field} = (#{ready()} + $1 * 1000), #{state.private_field} = $2 WHERE #{state.key_field} = $3" end @impl true def reserve(state) do "UPDATE #{state.table} SET #{state.lock_field} = #{reserve_at(state)} WHERE id = ( SELECT id FROM #{state.table} WHERE #{state.lock_field} BETWEEN 0 AND #{ready()} #{run_if(state)} ORDER BY #{state.lock_field}, #{state.key_field} LIMIT 1 FOR UPDATE SKIP LOCKED ) RETURNING #{state.key_field}, #{state.private_field}" end defp run_if(%State{run_if: nil}), do: nil defp run_if(%State{run_if: run_if}), do: "AND (#{run_if})" @impl true def cancel(state) do "UPDATE #{state.table} SET #{state.lock_field} = NULL WHERE id = $1 RETURNING #{state.lock_field}" end @impl true def status(state) do "SELECT COUNT(#{state.lock_field}) AS count, COUNT(*) FILTER (WHERE #{state.lock_field} = #{EctoSource.abandoned()}) AS abandoned, COUNT(*) FILTER (WHERE #{state.lock_field} BETWEEN 0 AND #{ready()}) AS ready, COUNT(*) FILTER (WHERE #{ready()} < #{state.lock_field} AND #{state.lock_field} < #{stale_at()}) AS delayed, COUNT(*) FILTER (WHERE #{stale_at()} <= #{state.lock_field} AND #{state.lock_field} < #{now()}) AS stale, COUNT(*) FILTER (WHERE #{now()} < #{state.lock_field} AND #{state.lock_field} <= #{reserve_at(state)}) AS in_progress FROM #{state.table}" end @impl true def reset_stale(state) do "UPDATE #{state.table} SET #{state.lock_field} = DEFAULT, #{state.private_field} = DEFAULT WHERE #{stale_at()} < #{state.lock_field} AND #{state.lock_field} < #{now()}" end @impl true def filter(state, :abandoned) do "SELECT id FROM #{state.table} WHERE #{state.lock_field} = #{EctoSource.abandoned()}" end def reserve_at(state) do "#{now()} + #{state.stale_timeout}" end def stale_at do time_in_msecs("(NOW() - INTERVAL '5 year')") end defp msecs_ago(msecs) do "#{now()} - #{msecs}" end def now do time_in_msecs("NOW()") end defp timestamp_in_msecs(time) do time_in_msecs("timestamp '#{time}'") end defp time_in_msecs(time) do "(CAST(EXTRACT(epoch from #{time}) * 1000 AS BIGINT))" end end end