defmodule Oban.Pro.Engines.Smart do
  @moduledoc """
  The `Smart` engine provides advanced concurrency features, enhanced observability, lighter
  weight bulk processing, and provides a foundation for accurate job lifecycle management. As an
  `Oban.Engine`, it is responsible for all non-plugin database interaction, from inserting through
  executing jobs.

  Major features include:

  * [Global Concurrency](#module-global-concurrency) — limit the number of concurrent jobs that
    run across _all_ nodes
  * [Rate Limiting](#module-rate-limiting) — limit the number of jobs that execute within a
    window of time
  * [Queue Partitioning](#module-queue-partitioning) — segment a queue so concurrency or rate
    limits apply separately to each partition
  * [Async Tracking](#module-async-tracking) — bundle job acks (completed, cancelled, etc.) to
    minimize transactions and reduce load on the database
  * [Enhanced Unique](#module-enhanced-unique) — enforce job uniqueness with a custom index to
    accelerate inserting unique jobs safely between processes and nodes
  * [Bulk Inserts](#module-bulk-inserts) — automatically batch inserts to prevent hitting
    database limits, reduce data sent over the wire, and respect `unique` options when using
    `Oban.insert_all/2`
  * [Accurate Snooze](#module-accurate-snooze) — differentiate between attempts with errors and
    intentional snoozes

  ## Installation

  See the [Smart Engine](adoption.md#1-smart-engine) section in the [adoption guide](adoption.md)
  to get started.

  ## Global Concurrency

  Global concurrency limits the number of concurrent jobs that run across all nodes.

  Typically the global concurrency limit is `local_limit * num_nodes`. For example, with three
  nodes and a local limit of 10, you'll have a global limit of 30. If a `global_limit` is present,
  and the `local_limit` is omitted, then the `local_limit` falls back to the `global_limit`.

  The only way to guarantee that all connected nodes will run _exactly one job_ concurrently is to
  set `global_limit: 1`.

  Here are some examples:

  ```elixir
  # Execute 10 jobs concurrently across all nodes, with up to 10 on a single node
  my_queue: [global_limit: 10]

  # Execute 10 jobs concurrently, but only 3 jobs on a single node
  my_queue: [local_limit: 3, global_limit: 10]

  # Execute at most 1 job concurrently
  my_queue: [global_limit: 1]
  ```

  ## Rate Limiting

  Rate limiting controls the number of jobs that execute within a period of time.

  Rate limiting uses counts for the same queue from all other nodes in the cluster (with or
  without Distributed Erlang). The limiter uses a sliding window over the configured period to
  accurately approximate a limit.

  Every job execution counts toward the rate limit, regardless of whether the job completes,
  errors, snoozes, etc.

  Without a modifier, the `rate_limit` period is defined in seconds. However, you can provide a
  `:second`, `:minute`, `:hour` or `:day` modifier to use more intuitive values.

  * `period: 30` — 30 seconds
  * `period: {1, :minute}` — 60 seconds
  * `period: {2, :minutes}` — 120 seconds
  * `period: {1, :hour}` — 3,600 seconds
  * `period: {1, :day}` —86,400 seconds

  Here are a few examples:

  ```elixir
  # Execute at most 60 jobs per minute (1 job per second equivalent)
  my_queue: [rate_limit: [allowed: 60, period: {1, :minute}]]

  # Execute at most 10 jobs per 30 seconds
  my_queue: [rate_limit: [allowed: 10, period: 30]]

  # Execute at most 10 jobs per minute
  my_queue: [rate_limit: [allowed: 10, period: {1, :minute}]]

  # Execute at most 1000 jobs per hour
  my_queue: [rate_limit: [allowed: 1000, period: {1, :hour}]]
  ```

  Using larger time periods allows for smoother tracking of rate limits. For example, expressing
  "1 job per second" as "60 jobs per minute" provides the same throughput but reduces the
  granularity of tracking, resulting in more consistent job execution patterns.

  > #### Understanding Concurrency Limits {: .info}
  >
  > The local, global, or rate limit with the **lowest value** determines how many jobs are executed
  > concurrently. For example, with a `local_limit` of 10 and a `global_limit` of 20, a single node
  > will run 10 jobs concurrently. If that same queue had a `rate_limit` that allowed 5 jobs within
  > a period, then a single node is limited to 5 jobs.

  ## Queue Partitioning

  In addition to global and rate limits at the queue level, you can partition a queue so that it's
  treated as multiple queues where concurrency or rate limits apply separately to each partition.

  Partitions are specified with a list of fields like `:worker`, `:args`, or `:meta`. When
  partitioning by `:args`, choosing specific keys is highly recommended to keep partitioning
  meaningful. Focused partitioning minimizes the amount of data a queue needs to track and
  simplifies job-fetching queries.

  ### Configuring Partitions

  The partition syntax is identical for global and rate limits (note that you can partition by
  _global or rate_, but not both.)

  Here are a few examples of viable partitioning schemes:

  ```elixir
  # Partition by worker alone
  partition: :worker

  # Partition by the `id` and `account_id` from args, ignoring the worker
  partition: [args: [:id, :account_id]]

  # Partition by worker and the `account_id` key from args
  partition: [:worker, args: :account_id]
  ```

  Remember, take care to minimize partition cardinality by using a few `keys` whenever possible.
  Partitioning based on _every permutation_ of your `args` makes concurrency or rate limits hard to
  reason about and can negatively impact queue performance.

  ### Global Partitioning

  Global partitioning changes global concurency behavior. Rather than applying a fixed number for
  the queue, it applies to every partition within the queue.

  Consider the following example:

  ```elixir
  local_limit: 10, global_limit: [allowed: 1, partition: :worker]
  ```

  The queue is configured to run one job per-worker across every node, but only 10 concurrently on a
  single node. That is in contrast to the standard behaviour of `global_limit`, which would override
  the `local_limit` and only allow 1 concurrent job across every node.

  Alternatively, you could partition by a single key:

  ```elixir
  local_limit: 10, global_limit: [allowed: 1, partition: [args: :tenant_id]]
  ```

  That configures the queue to run one job concurrently across the entire cluster per `tenant_id`.

  ### Partitioning with Burst Mode

  Global partitioning includes an advanced feature called "burst mode" for global limits
  with partitioning. This feature allows you to maximize throughput by temporarily exceeding
  per-partition global limits when there are available resources.

  Each global partition is typically restricted to the configured `allowed` value. However, with
  burst mode enabled, the system can intelligently allocate more jobs to each active partition,
  potentially exceeding the per-partition limit while still respecting the overall queue
  concurrency.

  This is particularly useful when:

  1. You have many potential partitions but only a few are active at any given time
  2. You want to maximize throughput while maintaining some level of fairness between partitions
  3. You need to ensure your queues aren't sitting idle when capacity is available

  Here's an example of a queue that will 5 jobs from a single partition concurrently under load,
  but can burst up to 100 for a single partition when there is available capacity:

  ```elixir
  queues: [
    exports: [
      global_limit: [
        allowed: 5,
        burst: true,
        partition: [args: :tenant_id]
      ],
      local_limit: 100
    ]
  ]
  ```

  Bursting _still respects the overall global concurrency limit_. Using the example above, the
  queue can only execute 100 of a given partition's jobs concurrently across all nodes, even if
  there are more available resources.

  ### Rate Limit Partitioning

  Rate limit partitions operate similarly to global partitions. Rather than limiting all jobs within
  the queue, they limit each partition within the queue.

  For example, to allow one job per-worker, every ten seconds, across every instance of the `alpha`
  queue in your cluster:

  ```elixir
  local_limit: 10, rate_limit: [allowed: 1, period: 10, partition: :worker]
  ```

  ### Partition Keys Cache

  The Smart engine maintains a cache of available partition keys to optimize performance when
  fetching jobs. The cache has a configurable time-to-live (TTL) which controls how long partition
  keys are remembered before needing to be refreshed from the database.

  The default TTL is set to `3_000ms`, which is suitable for most applications. You can adjust it
  by adding configuration in your application's config:

  ```elixir
  # In config/config.exs
  config :oban_pro, Oban.Pro.Partition, keys_cache_ttl: 5_000  # 5 seconds
  ```

  A longer TTL reduces database load but might cause the system to be slower to recognize newly
  created partition keys. A shorter TTL ensures more up-to-date partition information at the cost
  of more frequent database queries.

  The number of partition keys retrieved is based on the queue's `local_limit` multiplied by a
  configurable factor (default `3`). This ensures the query scales appropriately with queue
  capacity while keeping result sets manageable:

  ```elixir
  # In config/config.exs
  config :oban_pro, Oban.Pro.Partition, keys_limit_multiplier: 5
  ```

  Partitions are selected fairly by prioritizing those with the earliest scheduled jobs, ensuring
  work is distributed across active partitions rather than favoring any single partition.

  ## Async Tracking

  The results of job execution, e.g. `completed`, `cancelled`, etc., are bundled together into a
  single transaction to minimize load on an app's Ecto pool and the database.

  Bundling updates and reporting them asynchronously dramatically reduces the number of
  transactions per second. However, async bundling introduces a slight lag (up to 5ms) between job
  execution finishing and recording the outcome in the database.

  Async tracking can be disabled for specific queues with the `ack_async` option:

  ```elixir
  queues: [
    standard: 30,
    critical: [ack_async: false, local_limit: 10]
  ]
  ```

  ## Enhanced Unique

  The `Smart` engine uses an alternative mechanism for unique jobs that's designed for speed,
  correctness, scalability, and simplicity. Uniqueness is enforced through a unique index that
  makes insertion entirely safe between processes and nodes, without the use of advisory locks or
  multiple queries.

  Unlike standard uniqueness, which is only checked as jobs are inserted, the index-backed version
  applies for the job's entire lifetime. That prevents race conditions where a job changes states
  and inadvertently causes a conflict.

  When conflicts are detected, the conflicted job, i.e. the one already in the database, is
  annotated with `uniq_conflict: true`.

  > #### Safe Hash for Uniqueness {: .info}
  >
  > To avoid potential hash collisions when using unique jobs with sub-fields in `args`, enable
  > "safe" hashing:
  >
  > ```elixir
  > config :oban_pro, Oban.Pro.Utils, safe_hash: true
  > ```
  >
  > This applies to `uniq_key`, `chain_key`, and `partition_key` values stored in job meta.
  > Note that generated values will not match previous values for configurations using
  > sub-fields in `args`.

  > #### Period-based Uniqueness Uses Buckets {: .info}
  >
  > To leverage unique indexes, period-based uniqueness snaps timestamps to fixed time buckets
  > rather than using a sliding window. The current time is rounded down to the nearest multiple
  > of the period, and uniqueness is enforced within that bucket.
  >
  > For example, with a period of 15 minutes (`unique: [period: {15, :minutes}]`) and a job
  > inserted at 10:21:00, uniqueness applies from 10:15:00 to 10:30:00. A duplicate job
  > inserted at 10:28:00 would be rejected, but one inserted at 10:31:00 would be allowed
  > because it falls into the next bucket.

  ## Bulk Inserts

  Where the `Basic` engine requires you to insert unique jobs individually, the `Smart` engine adds
  unique job support to `Oban.insert_all/2`. No additional configuration is necessary—simply use
  `insert_all` instead for unique jobs.

  ```elixir
  Oban.insert_all(lots_of_unique_jobs)
  ```

  Bulk insert also features automatic batching to support inserting an arbitrary number of jobs
  without hitting database limits (PostgreSQL's binary protocol has a limit of 65,535 parameters
  that may be sent in a single call. That presents an upper limit on the number of rows that may be
  inserted at one time.)

  ```elixir
  list_of_args
  |> Enum.map(&MyApp.MyWorker.new/1)
  |> Oban.insert_all()
  ```

  The default batch size for unique jobs is `250`, and `1_000` for non-unique jobs. Regardless, you
  can override with `batch_size`:

  ```elixir
  Oban.insert_all(lots_of_jobs, batch_size: 1500)
  ```

  It's also possible to set a custom timeout for batch inserts:

  ```elixir
  Oban.insert_all(lots_of_jobs, timeout: :infinity)
  ```

  ## Accurate Snooze

  Unlike the `Basic` engine which increments `attempts` and `max_attempts`, the Smart engine rolls
  back the `attempt` on snooze. This approach preserves the original `max_attempts` and records a
  `snoozed` count in `meta`. As a result, it's simple to differentiate between "real" attempts and
  snoozes, and backoff calculation remains accurate regardless of snoozing.

  The following `process/1` function demonstrates checking a job's `meta` for a `snoozed` count:

  ```elixir
  def process(job) do
    case job.meta do
      %{"orig_scheduled_at" => unix_microseconds, "snoozed" => snoozed} ->
        IO.inspect({snoozed, unix_microseconds}, label: "Times snoozed since")

      _ ->
        # This job has never snoozed before
    end
  end
  ```
  """

  @behaviour Oban.Engine
  @behaviour Oban.Pro.Handler

  import Ecto.Query
  import DateTime, only: [utc_now: 0]

  alias Ecto.{Changeset, Multi}
  alias Oban.{Backoff, Config, Engine, Job, Repo}
  alias Oban.Pro.Limiters.{Global, Local, Rate}
  alias Oban.Pro.{Flusher, Handler, Partition, Producer, Unique, Utils}
  alias Oban.Pro.Stages.Chain

  require Logger

  @type partition ::
          :worker
          | {:args, atom()}
          | {:meta, atom()}
          | [:worker | {:args, atom()} | {:meta, atom()}]
          | [fields: [:worker | :args | :meta], keys: [atom()]]

  @type period :: pos_integer() | {pos_integer(), unit()}

  @type global_limit :: pos_integer() | [allowed: pos_integer(), partition: partition()]

  @type local_limit :: pos_integer()

  @type rate_limit :: [allowed: pos_integer(), period: period(), partition: partition()]

  @type unit :: :second | :seconds | :minute | :minutes | :hour | :hours | :day | :days

  # Module Attributes

  @ack_tabs Map.new(0..7, &{&1, :"pro_ack_tab_#{&1}"})

  @drain_uuid "00000000-0000-0000-0000-000000000000"

  @registry Oban.Registry

  @virtual_opts ~w(ack_async queue refresh_interval xact_delay xact_retry xact_timeout updated_at)a

  @smart_opts Application.compile_env(:oban_pro, __MODULE__, [])

  @xact_expected_delay Access.get(@smart_opts, :xact_expected_delay, 20)
  @base_batch_size Access.get(@smart_opts, :base_batch_size, 1_000)
  @max_uniq_retries 10

  # Macros

  defguardp is_global(producer) when is_map(producer.meta) and is_map(producer.meta.global_limit)

  defmacrop merge_jsonb(column, map) do
    quote do
      fragment("? || ?", unquote(column), unquote(map))
    end
  end

  defmacrop xact_lock(pref_key, lock_key) do
    quote do
      fragment("pg_try_advisory_xact_lock(?::int, ?::int)", unquote(pref_key), unquote(lock_key))
    end
  end

  defmacrop state_in_string(column, string, prefix) do
    quote do
      fragment(
        "? = ANY(regexp_split_to_array(?, ',')::?.oban_job_state[])",
        unquote(column),
        unquote(string),
        unquote(prefix)
      )
    end
  end

  # Handler

  @impl Handler
  def on_start do
    for {_idx, name} <- @ack_tabs do
      :ets.new(name, [:public, :named_table, read_concurrency: true, write_concurrency: true])
    end
  end

  @impl Handler
  def on_stop, do: :ok

  # Engine

  @impl Engine
  def init(%Config{} = conf, [_ | _] = opts) do
    {validate?, opts} = Keyword.pop(opts, :validate, false)

    {prod_opts, meta_opts} =
      opts
      |> Keyword.put_new(:ack_async, conf.testing == :disabled)
      |> Keyword.split(@virtual_opts)

    changeset =
      prod_opts
      |> Keyword.put(:name, conf.name)
      |> Keyword.put(:node, conf.node)
      |> Keyword.put(:meta, meta_opts)
      |> Keyword.put_new(:queue, :default)
      |> Keyword.put_new(:started_at, utc_now())
      |> Keyword.put_new(:updated_at, utc_now())
      |> Producer.new()

    case Changeset.apply_action(changeset, :insert) do
      {:ok, producer} ->
        if validate? do
          {:ok, producer}
        else
          ack_tab =
            producer.queue
            |> :erlang.phash2(map_size(@ack_tabs))
            |> then(&Map.fetch!(@ack_tabs, &1))

          virt_opts =
            prod_opts
            |> Keyword.take(~w(ack_async refresh_interval xact_delay xact_retry)a)
            |> Map.new()

          fun =
            fn ->
              conf
              |> Repo.insert!(changeset)
              |> Map.put(:ack_tab, ack_tab)
              |> Map.merge(virt_opts)
              |> put_producer(conf)
            end

          transaction(conf, fun, virt_opts)
        end

      {:error, changeset} ->
        {:error, Utils.to_exception(changeset)}
    end
  end

  @impl Engine
  def refresh(_conf, %Producer{} = producer) do
    # The `Refresher` module handles updating the database rows independently of this callback
    # being triggered. When the queue's producer is stuck in a transaction retry loop it isn't
    # able to handle new messages, including periodic refresh. That would allow the producer
    # record to become outdated, at which point it is subject to erroneous cleanup.
    %{producer | updated_at: utc_now()}
  end

  @impl Engine
  def shutdown(%Config{} = conf, %Producer{} = producer) do
    monitor_unregister(conf, producer)

    put_meta(conf, producer, :shutdown, true)
  catch
    _kind, _reason ->
      producer
      |> Producer.update_meta(:paused, true)
      |> Changeset.apply_action!(:update)
  end

  defp monitor_unregister(conf, producer) do
    parent = self()

    Task.start(fn ->
      ref = Process.monitor(parent)

      receive do
        {:DOWN, ^ref, :process, _pid, _reason} ->
          Process.demonitor(ref, [:flush])

          Registry.delete_meta(@registry, reg_key(conf, producer))
      end
    end)
  end

  @impl Engine
  def put_meta(%Config{} = conf, %Producer{} = producer, :flush, _any) do
    producer = run_acks(conf, producer)

    if is_global(producer) do
      Producer
      |> where(uuid: ^producer.uuid)
      |> then(&Repo.update_all(conf, &1, set: [meta: producer.meta, updated_at: utc_now()]))
    end

    producer
  end

  def put_meta(%Config{} = conf, %Producer{} = producer, :paused, true) do
    changeset =
      conf
      |> run_acks(producer)
      |> Producer.update_meta(:paused, true)

    conf
    |> Repo.update!(changeset)
    |> put_producer(conf)
  end

  def put_meta(%Config{} = conf, %Producer{} = producer, :shutdown, true) do
    changeset =
      conf
      |> run_acks(producer)
      |> Producer.update_meta(%{paused: true, shutdown_started_at: utc_now()})

    conf
    |> Repo.update!(changeset)
    |> put_producer(conf)
  end

  def put_meta(%Config{} = conf, %Producer{} = producer, key, value) do
    changeset = Producer.update_meta(producer, key, value)

    conf
    |> Repo.update!(changeset)
    |> put_producer(conf)
  end

  @impl Engine
  def check_meta(_conf, %Producer{} = producer, running) do
    jids = for {_, {_, exec}} <- running, do: exec.job.id

    meta =
      producer.meta
      |> Map.from_struct()
      |> flatten_windows()

    producer
    |> Map.take(~w(name node queue uuid started_at updated_at)a)
    |> Map.put(:running, jids)
    |> Map.merge(meta)
  end

  defp flatten_windows(%{rate_limit: %{windows: windows}} = meta) do
    {pacc, cacc} =
      Enum.reduce(windows, {0, 0}, fn {_key, map}, {pacc, cacc} ->
        %{"prev_count" => prev, "curr_count" => curr} = map

        {pacc + prev, cacc + curr}
      end)

    put_in(meta.rate_limit.windows, [%{"curr_count" => cacc, "prev_count" => pacc}])
  end

  defp flatten_windows(meta), do: meta

  @impl Engine
  def fetch_jobs(_conf, %{meta: %{paused: true}} = prod, _running) do
    {:ok, {prod, []}}
  end

  def fetch_jobs(_conf, %{meta: %{local_limit: limit}} = prod, running)
      when map_size(running) >= limit do
    {:ok, {prod, []}}
  end

  def fetch_jobs(%Config{} = conf, %Producer{} = producer, running) do
    acks = get_acks(producer)

    multi =
      Multi.new()
      |> Multi.put(:acks, acks)
      |> Multi.put(:conf, conf)
      |> Multi.put(:running, running)
      |> Multi.put(:producer, track_acks(acks, producer))
      |> Multi.run(:all_producers, &all_producers/2)
      |> Multi.run(:ack_ids, &ack_jobs/2)
      |> Multi.run(:afh_ids, &run_flush_handlers/2)
      |> Multi.run(:local_demand, &Local.check/2)
      |> Multi.run(:global_demand, &Global.check/2)
      |> Multi.run(:rate_demand, &Rate.check/2)
      |> Multi.run(:jobs, &fetch_jobs/2)
      |> Multi.run(:tracked, &track_jobs(&1, &2, running))

    case transaction(conf, multi, producer) do
      {:ok, %{ack_ids: ack_ids, jobs: jobs, tracked: producer}} ->
        del_acks(ack_ids, producer)

        {:ok, {producer, jobs}}

      {:error, _op, %{postgres: %{code: :unique_violation, detail: detail}}, changes} ->
        clear_uniq_violation(changes.conf, detail)

        retry_unique_violation(:fetch_jobs, [conf, producer, running])

      {:error, _operation, :xact_lock, _changes} ->
        jittery_sleep()

        fetch_jobs(conf, producer, running)

      {:error, _operation, error, _changes} ->
        raise error
    end
  catch
    :error, %Postgrex.Error{postgres: %{code: :unique_violation, detail: detail}} ->
      clear_uniq_violation(conf, detail, Enum.map(Job.states(), &to_string/1))

      retry_unique_violation(:fetch_jobs, [conf, producer, running])
  end

  @doc false
  def fetch_for_drain(conf, %{queue: queue, with_limit: limit} = opts) do
    subset_query =
      if queue == :__all__ do
        where(Job, state: "available")
      else
        where(Job, state: "available", queue: ^to_string(queue))
      end

    subset_query =
      case opts do
        %{batch_ids: ids} -> where(subset_query, [j], j.meta["batch_id"] in ^ids)
        %{workflow_ids: ids} -> where(subset_query, [j], j.meta["workflow_id"] in ^ids)
        _ -> subset_query
      end

    subset_query =
      subset_query
      |> order_by(asc: :priority, asc: :scheduled_at, asc: :id)
      |> lock("FOR UPDATE SKIP LOCKED")
      |> limit(^limit)

    query =
      Job
      |> with_cte("subset", as: ^subset_query)
      |> join(:inner, [j], x in fragment(~s("subset")), on: true)
      |> where([j, x], j.id == x.id)
      |> select([j, _], j)

    updates = [
      set: [state: "executing", attempted_at: utc_now(), attempted_by: [conf.node, @drain_uuid]],
      inc: [attempt: 1]
    ]

    {_count, jobs} = Repo.update_all(conf, query, updates, [])

    jobs
  end

  @impl Engine
  def stage_jobs(%Config{} = conf, queryable, opts) do
    limit = Keyword.fetch!(opts, :limit)

    subquery =
      queryable
      |> select([:id, :state])
      |> where([j], j.state in ~w(scheduled retryable))
      |> where([j], not is_nil(j.queue))
      |> where([j], j.scheduled_at <= ^DateTime.utc_now())
      |> limit(^limit)

    query =
      Job
      |> join(:inner, [j], x in subquery(subquery), on: j.id == x.id)
      |> select([j, x], %{id: j.id, queue: j.queue, state: x.state})

    {_count, staged} = Repo.update_all(conf, query, set: [state: "available"])

    {:ok, staged}
  catch
    :error, %Postgrex.Error{postgres: %{code: :unique_violation, detail: detail}} ->
      clear_uniq_violation(conf, detail, ["scheduled", "retryable"])

      retry_unique_violation(:stage_jobs, [conf, queryable, opts])
  end

  defp retry_unique_violation(fun, args) do
    attempt = Process.get(fun, 0)

    if attempt >= @max_uniq_retries do
      Process.delete(fun)

      Logger.error(fn ->
        "[Oban.Pro.Engines.Smart] Unique violation retry limit (#{@max_uniq_retries}) exceeded " <>
          "in #{fun}. This may indicate a persistent unique constraint conflict that cannot " <>
          "be automatically resolved."
      end)

      uniq_violation_failure(fun, args)
    else
      Process.put(fun, attempt + 1)

      apply(__MODULE__, fun, args)
    end
  end

  defp uniq_violation_failure(:fetch_jobs, [_conf, producer, _running]) do
    {:ok, {producer, []}}
  end

  defp uniq_violation_failure(:stage_jobs, _args) do
    {:ok, []}
  end

  @impl Engine
  def check_available(%Config{} = conf) do
    exists =
      Job
      |> where([j], j.state == "available" and parent_as(:producers).queue == j.queue)
      |> select(1)

    query =
      from(p in Producer, as: :producers)
      |> where([p], exists(subquery(exists)))
      |> select([p], p.queue)

    {:ok, Repo.all(conf, query)}
  end

  @impl Engine
  def complete_job(%Config{} = conf, %Job{} = job) do
    set = [state: "completed", completed_at: utc_now()]

    case Process.get(:oban_recorded) do
      nil -> put_ack(conf, %{job | state: "completed"}, set)
      map -> put_ack(conf, %{job | state: "completed"}, set ++ [meta: map])
    end

    :ok
  end

  @impl Engine
  def discard_job(%Config{} = conf, %Job{} = job) do
    put_ack(conf, %{job | state: "discarded"},
      state: "discarded",
      discarded_at: utc_now(),
      error: Job.format_attempt(job)
    )

    :ok
  end

  @impl Engine
  def error_job(%Config{} = conf, %Job{} = job, seconds) do
    orig_at = Map.get(job.meta, "orig_scheduled_at", to_unix(job.scheduled_at))

    set =
      if job.attempt >= job.max_attempts do
        [state: "discarded", discarded_at: utc_now()]
      else
        [state: "retryable", scheduled_at: seconds_from_now(seconds)]
      end

    put_ack(
      conf,
      %{job | state: set[:state]},
      set ++ [error: Job.format_attempt(job), meta: %{orig_scheduled_at: orig_at}]
    )

    :ok
  end

  @impl Engine
  def snooze_job(%Config{} = conf, %Job{} = job, seconds) do
    snoozed = Map.get(job.meta, "snoozed", 0)
    orig_at = Map.get(job.meta, "orig_scheduled_at", to_unix(job.scheduled_at))

    put_ack(conf, %{job | state: "scheduled"},
      attempt_change: -1,
      state: "scheduled",
      scheduled_at: seconds_from_now(seconds),
      meta: %{orig_scheduled_at: orig_at, snoozed: snoozed + 1}
    )

    :ok
  end

  @impl Engine
  def cancel_job(%Config{} = conf, %Job{unsaved_error: %{}} = job) do
    put_ack(conf, %{job | state: "cancelled"},
      state: "cancelled",
      cancelled_at: utc_now(),
      error: Job.format_attempt(job)
    )
  end

  def cancel_job(%Config{} = conf, %Job{} = job) do
    cancel_all_jobs(conf, where(Job, id: ^job.id))

    :ok
  end

  @impl Engine
  def cancel_all_jobs(%Config{} = conf, queryable) do
    subquery = where(queryable, [j], j.state not in ["cancelled", "completed", "discarded"])

    query =
      Job
      |> join(:inner, [j], x in subquery(subquery), on: j.id == x.id)
      |> update(set: [state: "cancelled", cancelled_at: ^utc_now()])
      |> select([_, x], map(x, [:id, :args, :attempted_by, :meta, :queue, :state, :worker]))

    {:ok, %{jobs: {_count, jobs}}} =
      Multi.new()
      |> Multi.put(:conf, conf)
      |> Multi.update_all(:jobs, query, [], Repo.default_options(conf))
      |> Multi.run(:ack, &track_cancelled_jobs/2)
      |> then(&Repo.transaction(conf, &1))

    {:ok, jobs}
  end

  @doc false
  # This is used to unblock a uniqueness violation by clearing out the uniq_bmp field. The
  # generated `uniq_key` column uses the `uniq_key` from meta, but is only present when the job's
  # state maps to a value in `uniq_bmp`.
  #
  # We find conflicting jobs by matching the unique key in meta and filtering by the expected
  # states. Clearing their `uniq_bmp` disables the unique constraint, allowing the transition.
  def clear_uniq_violation(%Config{} = conf, detail, states \\ ["executing"]) do
    with %{"key" => key} <- Regex.named_captures(~r/\(uniq_key\)=\((?<key>.+)\)/, detail) do
      query =
        Job
        |> where([j], j.state in ^states)
        |> where([j], fragment("? @> ?", j.meta, ^%{uniq_key: key}))
        |> update([j], set: [meta: merge_jsonb(j.meta, ^%{uniq_bmp: []})])

      Repo.update_all(conf, query, [])

      :telemetry.execute(
        [:oban, :engine, :uniq_violation_repaired],
        %{count: 1},
        %{conf: conf, uniq_key: key, states: states}
      )

      log_key = {__MODULE__, :uniq_violation_logged, key}

      unless :persistent_term.get(log_key, false) do
        :persistent_term.put(log_key, true)

        Logger.info(fn ->
          """
          [Oban.Pro.Engines.Smart] Unique constraint violation repaired.

          This may indicate a unique job misconfiguration where jobs are being inserted
          with conflicting unique keys across different states. The engine repaired the
          violation for uniq_key: #{key}

          Check your unique job configuration to ensure the :states option matches the
          expected job lifecycle for this worker.
          """
        end)
      end
    end
  end

  @impl Engine
  def insert_job(conf, changeset, opts) do
    if changeset.valid? do
      case insert_all_jobs(conf, [changeset], opts) do
        [job] ->
          {:ok, job}

        _ ->
          {:error, changeset}
      end
    else
      {:error, changeset}
    end
  end

  @impl Engine
  def insert_all_jobs(conf, changesets, opts) do
    fun = fn -> insert_all_batches(conf, changesets, opts) end

    {:ok, jobs} = Repo.transaction(conf, fun, opts)

    jobs
  end

  defp insert_all_batches(conf, changesets, opts) do
    batch_size = Keyword.get(opts, :batch_size, @base_batch_size)

    changesets
    |> Stream.map(&Unique.with_uniq_meta/1)
    |> Stream.map(&Partition.with_partition_meta(&1, conf))
    |> Stream.uniq_by(&Unique.get_key(&1, System.unique_integer()))
    |> Stream.chunk_every(batch_size)
    |> Enum.flat_map(fn changesets ->
      {uniq_map, link_map, replace?} =
        Enum.reduce(changesets, {%{}, %{}, false}, fn changeset, {uniq_map, link_map, replace?} ->
          uniq_map =
            case Unique.get_key(changeset) do
              uniq_key when is_binary(uniq_key) ->
                Map.put(uniq_map, uniq_key, changeset)

              _ ->
                uniq_map
            end

          link_map =
            case Chain.get_key(changeset) do
              chain_id when is_binary(chain_id) ->
                Map.update(link_map, chain_id, [changeset], &[changeset | &1])

              _ ->
                link_map
            end

          {uniq_map, link_map, replace? or Unique.replace?(changeset)}
        end)

      {:ok, %{all_jobs: jobs}} =
        Multi.new()
        |> Multi.put(:conf, conf)
        |> Multi.put(:opts, opts)
        |> Multi.put(:changesets, changesets)
        |> Multi.put(:uniq_map, uniq_map)
        |> Multi.put(:link_map, link_map)
        |> Multi.put(:replacements?, replace?)
        |> Multi.run(:uniq_index?, &uniq_index?/2)
        |> Multi.run(:chain_changesets, &prepare_chains/2)
        |> Multi.run(:dupe_map, &find_dupes/2)
        |> Multi.run(:new_jobs, &insert_entries/2)
        |> Multi.run(:rep_jobs, &apply_replacements/2)
        |> Multi.run(:all_jobs, &apply_conflicts/2)
        |> then(&Repo.transaction(conf, &1, opts))

      jobs
    end)
  end

  defp uniq_index?(_repo, %{conf: conf}) do
    %{name: name, prefix: prefix} = conf

    Utils.persistent_cache({__MODULE__, :uniq_index?, name}, fn ->
      query =
        from("columns")
        |> put_query_prefix("information_schema")
        |> where(table_schema: ^prefix, table_name: "oban_jobs", column_name: "uniq_key")
        |> select(true)

      {:ok, Repo.one(conf, query) == true}
    end)
  end

  defp prepare_chains(_repo, %{link_map: link_map, conf: conf}) when map_size(link_map) > 0 do
    chain_ids = Map.keys(link_map)

    if has_advisory_locks?(conf) do
      lock_keys =
        chain_ids
        |> Enum.map(&:erlang.phash2({conf.prefix, &1}))
        |> Enum.sort()

      Repo.query!(conf, "SELECT pg_advisory_xact_lock(unnest($1::bigint[]))", [lock_keys])
    end

    query =
      from(
        f in fragment("json_array_elements_text(?)", ^chain_ids),
        as: :list,
        inner_lateral_join:
          j in subquery(
            Job
            |> select([:state])
            |> where([j], j.state != "completed")
            |> where([j], fragment("? \\? 'chain_id'", j.meta))
            |> where([j], fragment("?->>'chain_id'", j.meta) == parent_as(:list).value)
            |> where([j], fragment("?->>'on_hold'", j.meta) in ~w(true false))
            |> order_by(desc: :id)
            |> lock("FOR UPDATE")
            |> limit(1)
          ),
        on: true,
        select: {f.value, j.state}
      )

    found_map =
      conf
      |> Repo.all(query)
      |> Map.new()

    linked =
      Enum.flat_map(link_map, fn {chain_id, changesets} ->
        changesets
        |> Enum.reverse()
        |> Enum.with_index()
        |> Enum.map(fn {changeset, index} ->
          state = Map.get(found_map, chain_id)
          meta = Changeset.get_field(changeset, :meta)

          cond do
            index == 0 and is_nil(state) -> changeset
            index == 0 and Chain.continuable?(state, meta) -> changeset
            true -> Chain.to_postponed(changeset)
          end
        end)
      end)

    {:ok, linked}
  end

  defp prepare_chains(_repo, _changes), do: {:ok, []}

  @impl Engine
  def retry_job(%Config{} = conf, %Job{id: id}) do
    retry_all_jobs(conf, where(Job, [j], j.id == ^id))

    :ok
  end

  @impl Engine
  def retry_all_jobs(%Config{} = conf, queryable) do
    subquery =
      queryable
      |> where([j], j.state not in ["available", "executing"])
      |> where([j], not fragment("? @> ?", j.meta, ^%{on_hold: true}))

    query =
      Job
      |> join(:inner, [j], x in subquery(subquery), on: j.id == x.id)
      |> select([_, x], map(x, [:id, :queue, :state]))
      |> update([j],
        set: [
          state: "available",
          max_attempts: fragment("GREATEST(?, ? + 1)", j.max_attempts, j.attempt),
          scheduled_at: ^utc_now(),
          completed_at: nil,
          cancelled_at: nil,
          discarded_at: nil
        ]
      )

    {_, jobs} = Repo.update_all(conf, query, [])

    {:ok, jobs}
  end

  @impl Engine
  def update_job(%Config{} = conf, %Job{id: id}, changes) when is_map(changes) do
    updater = fn ->
      query =
        Job
        |> where([j], j.id == ^id)
        |> lock("FOR UPDATE SKIP LOCKED")

      case Repo.one(conf, query) do
        nil ->
          {:error, :locked_or_not_found}

        job when is_map_key(job.meta, "structured") and is_map_key(changes, :args) ->
          {:ok, worker} = Oban.Worker.from_string(job.worker)

          changeset = worker.new(changes.args)

          if changeset.valid? do
            job
            |> Job.update(changes)
            |> then(&Repo.update(conf, &1))
          else
            {:error, changeset}
          end

        job ->
          job
          |> Job.update(changes)
          |> then(&Repo.update(conf, &1))
      end
    end

    case Repo.transaction(conf, updater) do
      {:ok, result} -> result
      {:error, reason} -> {:error, reason}
    end
  end

  # Producer Fetching Helpers

  defp get_producer(conf, job_or_producer) do
    case Registry.meta(@registry, reg_key(conf, job_or_producer)) do
      {:ok, producer} -> producer
      :error -> %Producer{ack_async: false}
    end
  end

  defp put_producer(producer, conf) do
    Registry.put_meta(@registry, reg_key(conf, producer), producer)

    producer
  end

  defp reg_key(%{name: name}, %{queue: queue}), do: {name, {:producer, queue}}

  # Acking Helpers

  defp put_ack(conf, job, updates) do
    producer = get_producer(conf, job)
    global_key = Partition.get_key(job, "*")
    flush_handlers = Flusher.get_flush_handlers(job, conf)

    ack_entry = {{:ack, producer.name, job.queue, job.id}, global_key, flush_handlers, updates}

    if producer.ack_tab, do: :ets.insert(producer.ack_tab, ack_entry)

    cond do
      Process.get(:oban_draining) ->
        Process.delete(:oban_recorded)

        {:ok, jids} = ack_jobs([ack_entry], conf)

        run_flush_handlers([ack_entry])

        if producer.ack_tab, do: del_acks(jids, producer)

      not producer.ack_async or producer.meta.paused ->
        pid = Oban.Registry.whereis(conf.name, {:producer, producer.queue})

        if is_pid(pid) and Process.alive?(pid) do
          GenServer.call(pid, {:put_meta, :flush, true})
        end

      true ->
        :ok
    end
  end

  defp get_acks(%{ack_tab: tab, name: name, queue: queue}) do
    :ets.select(tab, [{{{:ack, name, queue, :_}, :_, :_, :_}, [], [:"$_"]}])
  end

  defp del_acks(ids, %{ack_tab: tab, name: name, queue: queue}) do
    Enum.each(ids, &:ets.delete(tab, {:ack, name, queue, &1}))
  end

  defp run_acks(conf, producer) do
    acks = get_acks(producer)

    {:ok, jids} = ack_jobs(acks, conf)

    run_flush_handlers(acks)

    del_acks(jids, producer)

    track_acks(acks, producer)
  end

  defp run_flush_handlers(_repo, %{acks: acks}) do
    {:ok, run_flush_handlers(acks)}
  end

  defp run_flush_handlers(acks) do
    mfas = for {_, _, all, _} <- acks, mfa <- all, is_tuple(mfa), uniq: true, do: mfa

    Enum.each(mfas, fn {mod, fun, arg} -> apply(mod, fun, arg) end)
  end

  # Fetch Helpers

  defp all_producers(_repo, %{conf: conf, producer: producer}) do
    %{ack_tab: tab, meta: meta, queue: queue} = producer

    needs_lock? = is_map(meta.global_limit) or is_map(meta.rate_limit) or any_flush_handlers?(tab)
    query = where(Producer, queue: ^queue)

    cond do
      not needs_lock? ->
        {:ok, []}

      not has_advisory_locks?(conf) ->
        {:ok, Repo.all(conf, lock(query, "FOR UPDATE NOWAIT"))}

      take_advisory_lock?(conf, queue) ->
        {:ok, Repo.all(conf, query)}

      true ->
        {:error, :xact_lock}
    end
  end

  defp any_flush_handlers?(tab) do
    match = {:_, :_, :"$1", :_}
    guard = [{:"/=", :"$1", []}]

    :ets.select_count(tab, [{match, guard, [true]}]) > 0
  end

  defp has_advisory_locks?(conf) do
    Utils.persistent_cache({__MODULE__, :has_advisory_locks?}, fn ->
      query =
        from("pg_proc")
        |> put_query_prefix("pg_catalog")
        |> where(proname: "pg_try_advisory_xact_lock", pronargs: 2)
        |> select(true)

      Repo.one(conf, query) == true
    end)
  end

  defp take_advisory_lock?(conf, queue) do
    pre = :erlang.phash2(conf.prefix)
    key = :erlang.phash2(queue)

    query = from(f in xact_lock(^pre, ^key), select: f.f0)

    Repo.one(conf, query)
  end

  defp fetch_jobs(_repo, %{conf: conf, producer: producer} = changes) do
    subset_query = fetch_subquery(changes)

    query =
      Job
      |> with_cte("subset", as: ^subset_query)
      |> join(:inner, [j], x in fragment(~s("subset")), on: true)
      |> where([j, x], j.id == x.id and j.state == "available" and j.attempt < j.max_attempts)
      |> select([j, _], j)

    updates = [
      set: [state: "executing", attempted_at: utc_now(), attempted_by: [conf.node, producer.uuid]],
      inc: [attempt: 1]
    ]

    {_count, jobs} = Repo.update_all(conf, query, updates)

    {:ok, jobs}
  end

  defp fetch_subquery(%{local_demand: local, producer: producer} = changes) do
    case changes do
      %{global_demand: nil, rate_demand: nil} ->
        fetch_subquery(producer, local)

      %{global_demand: global, rate_demand: nil} when is_integer(global) ->
        fetch_subquery(producer, min(local, global))

      %{global_demand: nil, rate_demand: rated} when is_integer(rated) ->
        fetch_subquery(producer, min(local, rated))

      %{global_demand: global, rate_demand: %{} = demands} ->
        fetch_subquery(producer, demands, min(local, global || local))

      %{global_demand: %{} = demands, rate_demand: rated} ->
        fetch_subquery(producer, demands, min(local, rated || local))

      %{global_demand: global, rate_demand: rated} ->
        limit = rated |> min(local) |> min(global)

        fetch_subquery(producer, limit)
    end
  end

  defp fetch_subquery(producer, limit) do
    Job
    |> select([:id])
    |> where(state: "available", queue: ^producer.queue)
    |> order_by([:priority, :scheduled_at, :id])
    |> limit(^max(limit, 0))
    |> lock("FOR UPDATE SKIP LOCKED")
  end

  defp fetch_subquery(producer, demands, limit) do
    {keys, lims} = Enum.unzip(demands)

    part_query =
      from(
        p in fragment("SELECT unnest(?::text[]) as key, unnest(?::int[]) AS lim", ^keys, ^lims),
        as: :part,
        inner_lateral_join:
          j in subquery(
            Job
            |> select([:id, :priority, :scheduled_at])
            |> where([j], j.state == "available" and j.queue == ^producer.queue)
            |> where([j], fragment("partition_key") == parent_as(:part).key)
            |> order_by([:priority, :scheduled_at, :id])
            |> limit(parent_as(:part).lim)
          ),
        on: true,
        select: %{id: j.id, priority: j.priority, scheduled_at: j.scheduled_at}
      )

    from(j in subquery(part_query),
      select: j.id,
      limit: ^limit,
      order_by: [:priority, :scheduled_at, :id]
    )
  end

  # Tracking Helpers

  @ack_fields ~w(state cancelled_at completed_at discarded_at scheduled_at attempt_change error meta)a

  defp ack_jobs(_repo, %{acks: acks, conf: conf}), do: ack_jobs(acks, conf)
  defp ack_jobs([], _conf), do: {:ok, []}

  defp ack_jobs(acks, conf) do
    [ids | params] =
      acks
      |> Enum.map(fn {{:ack, _, _, id}, _, _, set} -> [id | Enum.map(@ack_fields, &set[&1])] end)
      |> Enum.zip_with(&Function.identity/1)

    case Repo.query(conf, ack_query(conf), [ids | params]) do
      {:ok, %{rows: rows}} -> {:ok, List.flatten(rows)}
      error -> error
    end
  end

  defp ack_query(conf) do
    Utils.persistent_cache({__MODULE__, :ack_query, conf.prefix}, fn ->
      """
      WITH params AS (
        SELECT unnest($1::bigint[]) AS id,
               unnest($2::"#{conf.prefix}".oban_job_state[]) AS state,
               unnest($3::timestamp[]) AS cancelled_at,
               unnest($4::timestamp[]) AS completed_at,
               unnest($5::timestamp[]) AS discarded_at,
               unnest($6::timestamp[]) AS scheduled_at,
               unnest($7::integer[]) AS attempt_change,
               unnest($8::jsonb[]) AS error,
               unnest($9::jsonb[]) AS meta
      ),
      locked AS (
        SELECT oj.id
        FROM "#{conf.prefix}"."oban_jobs" oj
        INNER JOIN params tmp ON oj.id = tmp.id
        FOR UPDATE OF oj
      )
      UPDATE "#{conf.prefix}"."oban_jobs" oj
      SET state = tmp.state,
          cancelled_at = COALESCE(tmp.cancelled_at, oj.cancelled_at),
          completed_at = COALESCE(tmp.completed_at, oj.completed_at),
          discarded_at = COALESCE(tmp.discarded_at, oj.discarded_at),
          scheduled_at = COALESCE(tmp.scheduled_at, oj.scheduled_at),
          attempt = COALESCE(tmp.attempt_change + oj.attempt, oj.attempt),
          errors = CASE WHEN tmp.error IS NULL THEN oj.errors ELSE oj.errors || tmp.error END,
          meta = CASE WHEN tmp.meta IS NULL THEN oj.meta ELSE oj.meta || tmp.meta END
      FROM params tmp
      INNER JOIN locked l ON tmp.id = l.id
      WHERE oj.id = tmp.id
      RETURNING oj.id
      """
    end)
  end

  defp track_jobs(_repo, %{conf: conf, jobs: jobs, producer: producer}, running) do
    old_jobs = for {_ref, {_pid, %{job: job}}} <- running, do: job
    all_jobs = jobs ++ old_jobs

    meta =
      producer.meta
      |> Global.track(all_jobs)
      |> Local.track(jobs)
      |> Rate.track(jobs)

    if meta == producer.meta do
      {:ok, producer}
    else
      now = utc_now()
      query = where(Producer, uuid: ^producer.uuid)

      case Repo.update_all(conf, query, set: [meta: meta, updated_at: now]) do
        {1, _} ->
          {:ok, %{producer | meta: meta, updated_at: now}}

        # In this case the producer was erroneously deleted, possibly due to a connection error,
        # downtime, or in development after waking from sleep.
        {0, _} ->
          %{producer | meta: meta, updated_at: now}
          |> Changeset.change()
          |> then(&Repo.insert(conf, &1))
      end
    end
  end

  defp track_acks(acks, producer) when is_global(producer) do
    keys = Enum.map(acks, &elem(&1, 1))
    meta = Global.prepare_tracked(producer.meta, keys)

    %{producer | meta: meta}
  end

  defp track_acks(_acks, producer), do: producer

  defp track_cancelled_jobs(_repo, %{conf: conf, jobs: {_count, jobs}}) do
    jobs
    |> Enum.filter(&(&1.state == "executing"))
    |> Enum.group_by(& &1.attempted_by)
    |> Enum.each(fn {[_node, uuid | _], jobs} ->
      query =
        Producer
        |> where([p], p.uuid == ^uuid)
        |> where([p], fragment("?->'global_limit' \\? 'tracked'", p.meta))

      with %Producer{meta: meta} = producer <- Repo.one(conf, query) do
        keys = Enum.map(jobs, &Partition.get_key(&1, "*"))
        meta = Global.prepare_tracked(meta, keys)

        %{producer | meta: meta, updated_at: utc_now()}
        |> Changeset.change()
        |> then(&Repo.update(conf, &1))
      end
    end)

    {:ok, nil}
  end

  # Insert Helpers

  defp find_dupes(_repo, %{conf: conf, uniq_index?: false, uniq_map: uniq_map})
       when map_size(uniq_map) > 0 do
    {uniq_keys, uniq_states} =
      Enum.reduce(uniq_map, {[], []}, fn {key, changeset}, {key_acc, sta_acc} ->
        states = changeset |> Unique.get_states() |> Enum.join(",")

        {[key | key_acc], [states | sta_acc]}
      end)

    query =
      from(
        t in fragment(
          "SELECT unnest(?::text[]) AS uk, unnest(?::text[]) AS us",
          ^uniq_keys,
          ^uniq_states
        ),
        join: j in Job,
        on:
          fragment("? \\? 'uniq_key'", j.meta) and
            fragment("?->>'uniq_key'", j.meta) == t.uk and
            state_in_string(j.state, t.us, literal(^conf.prefix)),
        select: {t.uk, j.id, j.state}
      )

    dupes =
      conf
      |> Repo.all(query)
      |> Map.new(fn {key, id, state} -> {key, {id, state, Map.get(uniq_map, key)}} end)

    {:ok, dupes}
  end

  defp find_dupes(_repo, _changes), do: {:ok, %{}}

  defp insert_entries(_repo, %{changesets: []}), do: {:ok, []}

  defp insert_entries(_repo, changes) do
    %{conf: conf, changesets: changesets, chain_changesets: chains} = changes
    %{dupe_map: dupe_map, opts: opts} = changes

    changesets =
      if chains == [] do
        changesets
      else
        Enum.reject(changesets, &Chain.chain?/1) ++ chains
      end

    {entries, placeholders} =
      for changeset <- changesets,
          not is_map_key(dupe_map, Unique.get_key(changeset)),
          reduce: {[], nil} do
        {entries, acc} ->
          map = Job.to_map(changeset)

          if is_nil(acc) do
            {[map | entries], map}
          else
            {[map | entries], Map.filter(acc, fn {key, val} -> map[key] == val end)}
          end
      end

    # Ensure placeholders aren't nil before use in is_map_key.
    placeholders = placeholders || %{}

    entries =
      for entry <- Enum.reverse(entries) do
        Map.new(entry, fn {key, val} ->
          if is_map_key(placeholders, key) do
            {key, {:placeholder, key}}
          else
            {key, val}
          end
        end)
      end

    opts =
      opts
      |> Keyword.merge(placeholders: placeholders, returning: true)
      |> Keyword.merge(conflict_opts(changes))

    {_count, jobs} = Repo.insert_all(conf, Job, entries, opts)

    {:ok, jobs}
  end

  defp conflict_opts(%{uniq_map: uniq_map, uniq_index?: uniq_index?}) do
    if uniq_map == %{} or not uniq_index? do
      []
    else
      [
        conflict_target: {:unsafe_fragment, "(uniq_key) WHERE uniq_key IS NOT NULL"},
        on_conflict: update(Job, [j], set: [meta: merge_jsonb(j.meta, ^%{uniq_conflict: true})])
      ]
    end
  end

  defp apply_replacements(_repo, %{replacements?: false}), do: {:ok, []}

  defp apply_replacements(_repo, %{conf: conf, opts: opts} = changes) do
    updates =
      for {_key, {id, state, %{changes: %{replace: replace} = changes}}} <- dupe_map(changes),
          reduce: %{} do
        acc ->
          state = String.to_existing_atom(state)
          rep_keys = Keyword.get(replace, state, [])

          changes
          |> Map.take(rep_keys)
          |> Enum.reduce(acc, fn {key, val}, sub_acc ->
            Map.update(sub_acc, {key, val}, [id], &[id | &1])
          end)
      end

    Enum.each(updates, fn {val, ids} ->
      Repo.update_all(conf, where(Job, [j], j.id in ^ids), [set: [val]], opts)
    end)

    {:ok, []}
  end

  defp dupe_map(%{dupe_map: dupe_map, uniq_index?: false}), do: dupe_map

  defp dupe_map(%{new_jobs: new_jobs, uniq_map: uniq_map}) do
    Enum.reduce(new_jobs, %{}, fn job, acc ->
      case job do
        %{meta: %{"uniq_conflict" => true, "uniq_key" => uniq_key}} ->
          Map.put(acc, uniq_key, {job.id, job.state, Map.get(uniq_map, uniq_key)})

        _ ->
          acc
      end
    end)
  end

  defp apply_conflicts(_repo, %{uniq_index?: false, dupe_map: dupe_map, new_jobs: new_jobs}) do
    old_jobs =
      dupe_map
      |> Map.values()
      |> Enum.map(fn {_id, _state, changeset} ->
        changeset
        |> Changeset.apply_action!(:insert)
        |> Map.replace!(:conflict?, true)
      end)

    {:ok, old_jobs ++ new_jobs}
  end

  defp apply_conflicts(_repo, %{new_jobs: new_jobs}) do
    new_jobs =
      Enum.map(new_jobs, fn
        %{meta: %{"uniq_conflict" => true}} = job -> %{job | conflict?: true}
        job -> job
      end)

    {:ok, new_jobs}
  end

  # Time Helpers

  defp seconds_from_now(seconds), do: DateTime.add(utc_now(), seconds, :second)

  defp to_unix(datetime), do: DateTime.to_unix(datetime, :microsecond)

  # Xact Helpers

  defp jittery_sleep do
    @xact_expected_delay
    |> Backoff.jitter()
    |> Process.sleep()
  end

  defp transaction(conf, fun_or_multi, prod_opts) do
    opts = [
      delay: Map.get(prod_opts, :xact_delay, 1000),
      retry: Map.get(prod_opts, :xact_retry, 5),
      timeout: Map.get(prod_opts, :xact_timeout, :timer.seconds(30)),
      expected_delay: @xact_expected_delay,
      expected_retry: 500
    ]

    Repo.transaction(conf, fun_or_multi, opts)
  end
end