bullmq

defmodule BullMQ.Scripts do
  @moduledoc """
  Manages Lua scripts for BullMQ Redis operations.

  This module loads Lua scripts from priv/scripts at compile time,
  extracting the number of keys from the filename pattern `scriptName-numberOfKeys.lua`.

  All scripts are loaded and cached for efficient execution using Redis EVALSHA.

  ## Script Location

  Scripts are copied from the root `rawScripts/` directory to `priv/scripts/`
  before compilation. Run `mix scripts.copy` to update the scripts, or they
  will be copied automatically during CI builds.
  """

  alias BullMQ.{Keys, RedisConnection}

  # Path to the scripts directory
  # First try priv/scripts (for CI and production), fallback to rawScripts (for local dev)
  @scripts_path (
                  priv_scripts_path = Path.expand("../../priv/scripts", __DIR__)
                  raw_scripts_path = Path.expand("../../../rawScripts", __DIR__)

                  cond do
                    File.dir?(priv_scripts_path) and
                        File.ls!(priv_scripts_path) |> Enum.any?(&String.ends_with?(&1, ".lua")) ->
                      priv_scripts_path

                    File.dir?(raw_scripts_path) ->
                      raw_scripts_path

                    true ->
                      # Will result in empty scripts - error will be raised at runtime
                      priv_scripts_path
                  end
                )

  # Map of Elixir atom names to script file base names (without the -N.lua suffix)
  @script_name_mapping %{
    add_delayed_job: "addDelayedJob",
    add_job_scheduler: "addJobScheduler",
    add_log: "addLog",
    add_parent_job: "addParentJob",
    add_prioritized_job: "addPrioritizedJob",
    add_repeatable_job: "addRepeatableJob",
    add_standard_job: "addStandardJob",
    change_delay: "changeDelay",
    change_priority: "changePriority",
    clean_jobs_in_set: "cleanJobsInSet",
    drain: "drain",
    extend_lock: "extendLock",
    extend_locks: "extendLocks",
    get_counts: "getCounts",
    get_counts_per_priority: "getCountsPerPriority",
    get_dependency_counts: "getDependencyCounts",
    get_job_scheduler: "getJobScheduler",
    get_metrics: "getMetrics",
    get_ranges: "getRanges",
    get_rate_limit_ttl: "getRateLimitTtl",
    get_state: "getState",
    get_state_v2: "getStateV2",
    is_finished: "isFinished",
    is_job_in_list: "isJobInList",
    is_maxed: "isMaxed",
    move_job_from_active_to_wait: "moveJobFromActiveToWait",
    move_jobs_to_wait: "moveJobsToWait",
    move_stalled_jobs_to_wait: "moveStalledJobsToWait",
    move_to_active: "moveToActive",
    move_to_delayed: "moveToDelayed",
    move_to_finished: "moveToFinished",
    move_to_waiting_children: "moveToWaitingChildren",
    obliterate: "obliterate",
    paginate: "paginate",
    pause: "pause",
    promote: "promote",
    release_lock: "releaseLock",
    remove_child_dependency: "removeChildDependency",
    remove_job: "removeJob",
    remove_job_scheduler: "removeJobScheduler",
    remove_repeatable: "removeRepeatable",
    reprocess_job: "reprocessJob",
    retry_job: "retryJob",
    retry_jobs: "retryJobs",
    save_stacked_job: "saveStackedJob",
    update_data: "updateData",
    update_job_option: "updateJobOption",
    update_job_scheduler: "updateJobScheduler",
    update_parent: "updateParent",
    update_progress: "updateProgress"
  }

  # Find all script files and register them as external resources
  @script_files (case File.ls(@scripts_path) do
                   {:ok, files} ->
                     files
                     |> Enum.filter(&String.ends_with?(&1, ".lua"))
                     |> Enum.sort()

                   {:error, _} ->
                     []
                 end)

  # Register each script file as an external resource for recompilation on change
  for file <- @script_files do
    @external_resource Path.join(@scripts_path, file)
  end

  # Parse script files to extract name and key count
  @scripts @script_files
           |> Enum.map(fn filename ->
             # Parse filename pattern: scriptName-numberOfKeys.lua
             case Regex.run(~r/^(.+)-(\d+)\.lua$/, filename) do
               [_, base_name, key_count_str] ->
                 key_count = String.to_integer(key_count_str)
                 file_path = Path.join(@scripts_path, filename)
                 content = File.read!(file_path)
                 {base_name, %{content: content, key_count: key_count, filename: filename}}

               nil ->
                 nil
             end
           end)
           |> Enum.reject(&is_nil/1)
           |> Map.new()

  # Build reverse mapping from Elixir atoms to script data
  @script_by_atom @script_name_mapping
                  |> Enum.map(fn {atom_name, base_name} ->
                    case Map.get(@scripts, base_name) do
                      nil -> nil
                      script_data -> {atom_name, script_data}
                    end
                  end)
                  |> Enum.reject(&is_nil/1)
                  |> Map.new()

  # Precompute SHAs at compile time for all scripts
  @script_shas @script_by_atom
               |> Enum.map(fn {atom_name, %{content: content}} ->
                 {atom_name, :crypto.hash(:sha, content) |> Base.encode16(case: :lower)}
               end)
               |> Map.new()

  @type script_name :: atom()
  @type script_result :: {:ok, any()} | {:error, any()}
  @type queue_context :: Keys.queue_context()

  # ---------------------------------------------------------------------------
  # Public API - Script Access
  # ---------------------------------------------------------------------------

  @doc """
  Returns the script content and number of keys for a given script name.

  ## Parameters

    * `name` - The script name as an atom (e.g., `:extend_lock`, `:move_to_active`)

  ## Returns

    * `{content, key_count}` tuple if script exists
    * `nil` if script not found

  ## Examples

      iex> {content, keys} = BullMQ.Scripts.get(:extend_lock)
      iex> is_binary(content) and is_integer(keys)
      true

  """
  @spec get(script_name()) :: {String.t(), non_neg_integer()} | nil
  def get(name) when is_atom(name) do
    case Map.get(@script_by_atom, name) do
      nil -> nil
      %{content: content, key_count: key_count} -> {content, key_count}
    end
  end

  @doc """
  Returns the script content for a given script name.
  """
  @spec get_content(script_name()) :: String.t() | nil
  def get_content(name) when is_atom(name) do
    case get(name) do
      {content, _} -> content
      nil -> nil
    end
  end

  @doc """
  Returns the number of keys for a given script.
  """
  @spec get_key_count(script_name()) :: non_neg_integer() | nil
  def get_key_count(name) when is_atom(name) do
    case get(name) do
      {_, key_count} -> key_count
      nil -> nil
    end
  end

  @doc """
  Lists all available script names.
  """
  @spec list_scripts() :: [script_name()]
  def list_scripts do
    Map.keys(@script_by_atom)
  end

  @doc """
  Checks if a script exists.
  """
  @spec exists?(script_name()) :: boolean()
  def exists?(name) when is_atom(name) do
    Map.has_key?(@script_by_atom, name)
  end

  # ---------------------------------------------------------------------------
  # Script Execution
  # ---------------------------------------------------------------------------

  @doc """
  Executes a Lua script against Redis by name.

  Uses EVALSHA for efficiency, falling back to EVAL if the script
  is not yet cached in Redis.

  ## Parameters

    * `conn` - The Redis connection pool name
    * `script_name` - The script name as an atom
    * `keys` - List of Redis keys
    * `args` - List of arguments to pass to the script

  ## Returns

    * `{:ok, result}` on success
    * `{:error, reason}` on failure

  """
  @spec execute(atom(), script_name(), [String.t()], [any()]) :: script_result()
  def execute(conn, script_name, keys, args) when is_atom(script_name) do
    case get(script_name) do
      nil ->
        {:error, {:script_not_found, script_name}}

      {script, _expected_key_count} ->
        # Use cached SHA for better performance
        sha = get_sha(script_name)
        execute_with_sha(conn, script, sha, keys, args)
    end
  end

  @doc """
  Executes a raw Lua script against Redis.
  """
  @spec execute_raw(atom(), String.t(), [String.t()], [any()]) :: script_result()
  def execute_raw(conn, script, keys, args) do
    sha = sha1(script)
    execute_with_sha(conn, script, sha, keys, args)
  end

  # Internal: Execute with pre-computed SHA
  defp execute_with_sha(conn, script, sha, keys, args) do
    encoded_args = Enum.map(args, &encode_arg/1)
    num_keys = length(keys)

    # Try EVALSHA first (cached script)
    case RedisConnection.command(conn, ["EVALSHA", sha, num_keys | keys ++ encoded_args]) do
      {:ok, result} ->
        {:ok, decode_result(result)}

      {:error, %Redix.Error{message: "NOSCRIPT" <> _}} ->
        # Script not cached, use EVAL which will also cache it
        case RedisConnection.command(conn, ["EVAL", script, num_keys | keys ++ encoded_args]) do
          {:ok, result} -> {:ok, decode_result(result)}
          {:error, reason} -> {:error, reason}
        end

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc """
  Builds a Redis command for a Lua script without executing it.
  Useful for pipelining multiple script calls.

  Returns `{:ok, command}` where command is a list that can be passed to Redis pipeline,
  or `{:error, reason}` if the script is not found.
  """
  @spec build_command(script_name(), [String.t()], [any()]) ::
          {:ok, [String.t()]} | {:error, term()}
  def build_command(script_name, keys, args) when is_atom(script_name) do
    case get_sha(script_name) do
      nil ->
        {:error, {:script_not_found, script_name}}

      sha ->
        encoded_args = Enum.map(args, &encode_arg/1)
        num_keys = length(keys)
        {:ok, ["EVALSHA", sha, num_keys | keys ++ encoded_args]}
    end
  end

  @doc """
  Gets the precomputed SHA for a script.
  SHAs are computed at compile time for efficiency.
  """
  @spec get_sha(script_name()) :: String.t() | nil
  def get_sha(script_name) do
    Map.get(@script_shas, script_name)
  end

  @doc """
  Builds multiple add_standard_job commands efficiently by precomputing shared data.
  Much faster than calling build_add_standard_job_command multiple times.

  Returns a list of {job, command} tuples.
  """
  @spec build_bulk_add_commands(queue_context(), [{map() | struct(), map()}]) ::
          {:ok, [{map() | struct(), [String.t()]}]} | {:error, term()}
  def build_bulk_add_commands(ctx, jobs_with_opts) do
    case get_sha(:add_standard_job) do
      nil ->
        {:error, {:script_not_found, :add_standard_job}}

      sha ->
        # Precompute keys (same for all jobs in queue)
        keys = [
          Keys.wait(ctx),
          Keys.paused(ctx),
          Keys.meta(ctx),
          Keys.id(ctx),
          Keys.completed(ctx),
          Keys.delayed(ctx),
          Keys.active(ctx),
          Keys.events(ctx),
          Keys.marker(ctx)
        ]

        key_prefix = Keys.key_prefix(ctx)
        num_keys = 9

        # Build commands efficiently
        results =
          Enum.map(jobs_with_opts, fn {job, opts} ->
            timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
            job_id = get_job_id(job)
            job_name = get_job_name(job)
            {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
            repeat_job_key = get_repeat_job_key(job)
            deduplication_key = get_deduplication_key(job, ctx)

            packed_args =
              Msgpax.pack!(
                [
                  key_prefix,
                  job_id || "",
                  job_name,
                  timestamp,
                  parent_key,
                  parent_deps_key,
                  parent,
                  repeat_job_key,
                  deduplication_key
                ],
                iodata: false
              )

            job_data = get_job_data(job) |> Jason.encode!()
            packed_opts = pack_job_opts(job, opts)

            # Build command directly without going through build_command
            cmd = ["EVALSHA", sha, num_keys | keys ++ [packed_args, job_data, packed_opts]]
            {job, cmd}
          end)

        {:ok, results}
    end
  end

  @doc """
  Executes multiple script commands in a pipeline.
  Returns a list of results in the same order as the commands.

  Note: If any script is not cached (NOSCRIPT error), this will fail.
  Use `ensure_scripts_loaded/2` first to cache scripts.
  """
  @spec execute_pipeline(atom(), [[String.t()]]) :: {:ok, [any()]} | {:error, term()}
  def execute_pipeline(conn, commands) when is_list(commands) do
    case RedisConnection.pipeline(conn, commands) do
      {:ok, results} ->
        decoded =
          Enum.map(results, fn
            %Redix.Error{} = error -> {:error, error}
            result -> {:ok, decode_result(result)}
          end)

        {:ok, decoded}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc """
  Executes multiple script commands in a Redis transaction (MULTI/EXEC).
  All commands are executed atomically - either all succeed or none do.

  Returns `{:ok, results}` on success or `{:error, reason}` on failure.

  Note: If any script is not cached (NOSCRIPT error), this will fail.
  Use `ensure_scripts_loaded/2` first to cache scripts.
  """
  @spec execute_transaction(atom(), [[String.t()]]) :: {:ok, [any()]} | {:error, term()}
  def execute_transaction(conn, commands) when is_list(commands) do
    case RedisConnection.transaction(conn, commands) do
      {:ok, results} ->
        decoded =
          Enum.map(results, fn
            %Redix.Error{} = error -> {:error, error}
            result -> {:ok, decode_result(result)}
          end)

        {:ok, decoded}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc """
  Ensures scripts are loaded into Redis cache by executing a dummy SCRIPT LOAD.
  Call this before using pipelined operations to avoid NOSCRIPT errors.
  """
  @spec ensure_scripts_loaded(atom(), [script_name()]) :: :ok | {:error, term()}
  def ensure_scripts_loaded(conn, script_names) do
    Enum.each(script_names, fn script_name ->
      case get(script_name) do
        {script, _} ->
          RedisConnection.command(conn, ["SCRIPT", "LOAD", script])

        nil ->
          :ok
      end
    end)

    :ok
  end

  # ---------------------------------------------------------------------------
  # Job Operations - Used by Queue module
  # ---------------------------------------------------------------------------

  @doc """
  Adds a standard job to the queue.
  """
  @spec add_standard_job(atom(), queue_context(), map() | struct(), map()) :: script_result()
  def add_standard_job(conn, ctx, job, opts) do
    # KEYS in order expected by Lua script
    keys = [
      # KEYS[1] 'wait'
      Keys.wait(ctx),
      # KEYS[2] 'paused'
      Keys.paused(ctx),
      # KEYS[3] 'meta'
      Keys.meta(ctx),
      # KEYS[4] 'id'
      Keys.id(ctx),
      # KEYS[5] 'completed'
      Keys.completed(ctx),
      # KEYS[6] 'delayed'
      Keys.delayed(ctx),
      # KEYS[7] 'active'
      Keys.active(ctx),
      # KEYS[8] events stream key
      Keys.events(ctx),
      # KEYS[9] marker key
      Keys.marker(ctx)
    ]

    timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
    job_id = get_job_id(job)
    job_name = get_job_name(job)
    {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
    repeat_job_key = get_repeat_job_key(job)
    deduplication_key = get_deduplication_key(job, ctx)

    # ARGV[1] msgpacked arguments array
    packed_args =
      Msgpax.pack!(
        [
          # [1] key prefix (with trailing colon for job key building)
          Keys.key_prefix(ctx),
          # [2] custom id (empty = auto generate)
          job_id || "",
          # [3] name
          job_name,
          # [4] timestamp
          timestamp,
          # [5] parentKey?
          parent_key,
          # [6] parent dependencies key
          parent_deps_key,
          # [7] parent? {id, queueKey}
          parent,
          # [8] repeat job key
          repeat_job_key,
          # [9] deduplication key
          deduplication_key
        ],
        iodata: false
      )

    # ARGV[2] Json stringified job data
    job_data = get_job_data(job) |> Jason.encode!()

    # ARGV[3] msgpacked options
    packed_opts = pack_job_opts(job, opts)

    args = [packed_args, job_data, packed_opts]

    execute(conn, :add_standard_job, keys, args)
  end

  @doc """
  Builds a command for adding a standard job without executing it.
  Used for pipelining multiple job additions.
  """
  @spec build_add_standard_job_command(queue_context(), map() | struct(), map()) ::
          {:ok, [String.t()]}
  def build_add_standard_job_command(ctx, job, opts) do
    keys = [
      Keys.wait(ctx),
      Keys.paused(ctx),
      Keys.meta(ctx),
      Keys.id(ctx),
      Keys.completed(ctx),
      Keys.delayed(ctx),
      Keys.active(ctx),
      Keys.events(ctx),
      Keys.marker(ctx)
    ]

    timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
    job_id = get_job_id(job)
    job_name = get_job_name(job)
    {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
    repeat_job_key = get_repeat_job_key(job)
    deduplication_key = get_deduplication_key(job, ctx)

    packed_args =
      Msgpax.pack!(
        [
          Keys.key_prefix(ctx),
          job_id || "",
          job_name,
          timestamp,
          parent_key,
          parent_deps_key,
          parent,
          repeat_job_key,
          deduplication_key
        ],
        iodata: false
      )

    job_data = get_job_data(job) |> Jason.encode!()
    packed_opts = pack_job_opts(job, opts)
    args = [packed_args, job_data, packed_opts]

    build_command(:add_standard_job, keys, args)
  end

  @doc """
  Adds multiple standard jobs atomically in a single transaction (MULTI/EXEC).
  Much more efficient than calling add_standard_job multiple times.

  This operation is atomic - all jobs are added or none are.

  Returns `{:ok, job_ids}` on success or `{:error, reason}` on failure.
  """
  @spec add_standard_jobs_pipelined(atom(), queue_context(), [{map() | struct(), map()}]) ::
          {:ok, [String.t()]} | {:error, term()}
  def add_standard_jobs_pipelined(conn, ctx, jobs_with_opts) do
    # Ensure script is loaded first
    ensure_scripts_loaded(conn, [:add_standard_job])

    # Build all commands
    commands =
      Enum.map(jobs_with_opts, fn {job, opts} ->
        {:ok, cmd} = build_add_standard_job_command(ctx, job, opts)
        cmd
      end)

    # Execute using transaction (MULTI/EXEC) for atomicity
    case execute_transaction(conn, commands) do
      {:ok, results} ->
        # Extract job IDs from results
        job_ids =
          Enum.map(results, fn
            {:ok, job_id} when is_binary(job_id) or is_integer(job_id) -> to_string(job_id)
            {:ok, other} -> other
            {:error, _} = err -> err
          end)

        errors = Enum.filter(job_ids, &match?({:error, _}, &1))

        if Enum.empty?(errors) do
          {:ok, job_ids}
        else
          {:error, {:partial_failure, job_ids}}
        end

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc """
  Adds a delayed job to the queue.
  """
  @spec add_delayed_job(atom(), queue_context(), map() | struct(), map()) :: script_result()
  def add_delayed_job(conn, ctx, job, opts) do
    # KEYS in order expected by Lua script
    keys = [
      # KEYS[1] 'marker'
      Keys.marker(ctx),
      # KEYS[2] 'meta'
      Keys.meta(ctx),
      # KEYS[3] 'id'
      Keys.id(ctx),
      # KEYS[4] 'delayed'
      Keys.delayed(ctx),
      # KEYS[5] 'completed'
      Keys.completed(ctx),
      # KEYS[6] events stream key
      Keys.events(ctx)
    ]

    timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
    job_id = get_job_id(job)
    job_name = get_job_name(job)
    delay = get_job_delay(job)
    {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
    repeat_job_key = get_repeat_job_key(job)
    deduplication_key = get_deduplication_key(job, ctx)

    # For delayed jobs, the timestamp should include the delay
    delayed_timestamp = timestamp + delay

    # ARGV[1] msgpacked arguments array
    packed_args =
      Msgpax.pack!(
        [
          # [1] key prefix (with trailing colon)
          Keys.key_prefix(ctx),
          # [2] custom id (empty = auto generate)
          job_id || "",
          # [3] name
          job_name,
          # [4] delayed timestamp
          delayed_timestamp,
          # [5] parentKey?
          parent_key,
          # [6] parent dependencies key
          parent_deps_key,
          # [7] parent? {id, queueKey}
          parent,
          # [8] repeat job key
          repeat_job_key,
          # [9] deduplication key
          deduplication_key
        ],
        iodata: false
      )

    # ARGV[2] Json stringified job data
    job_data = get_job_data(job) |> Jason.encode!()

    # ARGV[3] msgpacked options
    packed_opts = pack_job_opts(job, opts)

    args = [packed_args, job_data, packed_opts]

    execute(conn, :add_delayed_job, keys, args)
  end

  @doc """
  Adds a prioritized job to the queue.
  """
  @spec add_prioritized_job(atom(), queue_context(), map() | struct(), map()) :: script_result()
  def add_prioritized_job(conn, ctx, job, opts) do
    # KEYS in order expected by Lua script
    keys = [
      # KEYS[1] 'marker'
      Keys.marker(ctx),
      # KEYS[2] 'meta'
      Keys.meta(ctx),
      # KEYS[3] 'id'
      Keys.id(ctx),
      # KEYS[4] 'prioritized'
      Keys.prioritized(ctx),
      # KEYS[5] 'delayed'
      Keys.delayed(ctx),
      # KEYS[6] 'completed'
      Keys.completed(ctx),
      # KEYS[7] 'active'
      Keys.active(ctx),
      # KEYS[8] events stream key
      Keys.events(ctx),
      # KEYS[9] 'pc' priority counter
      Keys.pc(ctx)
    ]

    timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
    job_id = get_job_id(job)
    job_name = get_job_name(job)
    {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
    repeat_job_key = get_repeat_job_key(job)
    deduplication_key = get_deduplication_key(job, ctx)

    # ARGV[1] msgpacked arguments array
    packed_args =
      Msgpax.pack!(
        [
          # [1] key prefix (with trailing colon)
          Keys.key_prefix(ctx),
          # [2] custom id (empty = auto generate)
          job_id || "",
          # [3] name
          job_name,
          # [4] timestamp
          timestamp,
          # [5] parentKey?
          parent_key,
          # [6] parent dependencies key
          parent_deps_key,
          # [7] parent? {id, queueKey}
          parent,
          # [8] repeat job key
          repeat_job_key,
          # [9] deduplication key
          deduplication_key
        ],
        iodata: false
      )

    # ARGV[2] Json stringified job data
    job_data = get_job_data(job) |> Jason.encode!()

    # ARGV[3] msgpacked options (includes priority)
    packed_opts = pack_job_opts(job, opts)

    args = [packed_args, job_data, packed_opts]

    execute(conn, :add_prioritized_job, keys, args)
  end

  @doc """
  Builds a command for adding a parent job without executing it.
  Used for building flow transactions where all jobs are added atomically.
  """
  @spec build_add_parent_job_command(queue_context(), map() | struct(), map()) ::
          {:ok, [String.t()]}
  def build_add_parent_job_command(ctx, job, opts) do
    keys = [
      # KEYS[1] 'meta'
      Keys.meta(ctx),
      # KEYS[2] 'id'
      Keys.id(ctx),
      # KEYS[3] 'delayed'
      Keys.delayed(ctx),
      # KEYS[4] 'waiting-children'
      Keys.waiting_children(ctx),
      # KEYS[5] 'completed'
      Keys.completed(ctx),
      # KEYS[6] events stream key
      Keys.events(ctx)
    ]

    timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
    job_id = get_job_id(job)
    job_name = get_job_name(job)
    {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
    repeat_job_key = get_repeat_job_key(job)
    deduplication_key = get_deduplication_key(job, ctx)

    packed_args =
      Msgpax.pack!(
        [
          Keys.key_prefix(ctx),
          job_id || "",
          job_name,
          timestamp,
          parent_key,
          parent_deps_key,
          parent,
          repeat_job_key,
          deduplication_key
        ],
        iodata: false
      )

    job_data = get_job_data(job) |> Jason.encode!()
    packed_opts = pack_job_opts(job, opts)
    args = [packed_args, job_data, packed_opts]

    build_command(:add_parent_job, keys, args)
  end

  @doc """
  Adds a parent job to the queue (waiting-children state).

  Parent jobs are added in waiting-children state until all children complete.
  Used by FlowProducer to create job hierarchies.
  """
  @spec add_parent_job(atom(), queue_context(), map() | struct(), map()) :: script_result()
  def add_parent_job(conn, ctx, job, opts) do
    # KEYS in order expected by addParentJob-6.lua:
    # KEYS[1] 'meta'
    # KEYS[2] 'id'
    # KEYS[3] 'delayed'
    # KEYS[4] 'waiting-children'
    # KEYS[5] 'completed'
    # KEYS[6] events stream key
    keys = [
      # KEYS[1] 'meta'
      Keys.meta(ctx),
      # KEYS[2] 'id'
      Keys.id(ctx),
      # KEYS[3] 'delayed'
      Keys.delayed(ctx),
      # KEYS[4] 'waiting-children'
      Keys.waiting_children(ctx),
      # KEYS[5] 'completed'
      Keys.completed(ctx),
      # KEYS[6] events stream key
      Keys.events(ctx)
    ]

    timestamp = get_job_timestamp(job) || System.system_time(:millisecond)
    job_id = get_job_id(job)
    job_name = get_job_name(job)
    {parent_key, parent_deps_key, parent} = get_parent_info_full(job)
    repeat_job_key = get_repeat_job_key(job)
    deduplication_key = get_deduplication_key(job, ctx)

    # ARGV[1] msgpacked arguments array
    packed_args =
      Msgpax.pack!(
        [
          # [1] key prefix (with trailing colon)
          Keys.key_prefix(ctx),
          # [2] custom id (empty = auto generate)
          job_id || "",
          # [3] name
          job_name,
          # [4] timestamp
          timestamp,
          # [5] parentKey?
          parent_key,
          # [6] parent dependencies key
          parent_deps_key,
          # [7] parent? {id, queueKey}
          parent,
          # [8] repeat job key
          repeat_job_key,
          # [9] deduplication key
          deduplication_key
        ],
        iodata: false
      )

    # ARGV[2] Json stringified job data
    job_data = get_job_data(job) |> Jason.encode!()

    # ARGV[3] msgpacked options
    packed_opts = pack_job_opts(job, opts)

    args = [packed_args, job_data, packed_opts]

    execute(conn, :add_parent_job, keys, args)
  end

  # ---------------------------------------------------------------------------
  # Worker Operations - Used by Worker module
  # ---------------------------------------------------------------------------

  @doc """
  Moves a job to the active state for processing.
  """
  @spec move_to_active(atom(), queue_context(), String.t(), keyword()) :: script_result()
  def move_to_active(conn, ctx, token, opts \\ []) do
    # Keys in order expected by Lua script
    keys = [
      # KEYS[1] wait key
      Keys.wait(ctx),
      # KEYS[2] active key
      Keys.active(ctx),
      # KEYS[3] prioritized key
      Keys.prioritized(ctx),
      # KEYS[4] stream events key
      Keys.events(ctx),
      # KEYS[5] stalled key
      Keys.stalled(ctx),
      # KEYS[6] rate limiter key
      Keys.limiter(ctx),
      # KEYS[7] delayed key
      Keys.delayed(ctx),
      # KEYS[8] paused key
      Keys.paused(ctx),
      # KEYS[9] meta key
      Keys.meta(ctx),
      # KEYS[10] pc priority counter
      Keys.pc(ctx),
      # KEYS[11] marker key
      Keys.marker(ctx)
    ]

    timestamp = Keyword.get(opts, :timestamp, System.system_time(:millisecond))
    lock_duration = Keyword.get(opts, :lock_duration, 30_000)
    limiter = Keyword.get(opts, :limiter)
    name = Keyword.get(opts, :name)

    # ARGV[3] msgpacked opts
    packed_opts =
      Msgpax.pack!(
        %{
          "token" => token,
          "lockDuration" => lock_duration,
          "limiter" => limiter,
          "name" => name
        },
        iodata: false
      )

    args = [
      # ARGV[1] key prefix (with trailing colon)
      Keys.key_prefix(ctx),
      # ARGV[2] timestamp
      timestamp,
      # ARGV[3] msgpacked opts
      packed_opts
    ]

    execute(conn, :move_to_active, keys, args)
  end

  @doc """
  Moves a job to completed state.
  """
  @spec move_to_completed(atom(), queue_context(), String.t(), String.t(), any(), keyword()) ::
          script_result()
  def move_to_completed(conn, ctx, job_id, token, return_value, opts \\ []) do
    move_to_finished(conn, ctx, job_id, token, return_value, :completed, opts)
  end

  @doc """
  Moves a job to failed state.
  """
  @spec move_to_failed(atom(), queue_context(), String.t(), String.t(), any(), keyword()) ::
          script_result()
  def move_to_failed(conn, ctx, job_id, token, error, opts \\ []) do
    move_to_finished(conn, ctx, job_id, token, error, :failed, opts)
  end

  @doc """
  Moves a job to finished (completed/failed) state.
  """
  @spec move_to_finished(atom(), queue_context(), String.t(), String.t(), any(), atom(), keyword()) ::
          script_result()
  def move_to_finished(conn, ctx, job_id, token, result, target, opts \\ []) do
    target_str = to_string(target)
    metrics_key = Keys.metrics(ctx, target_str)

    # Keys in order expected by Lua script
    keys = [
      # KEYS[1] wait key
      Keys.wait(ctx),
      # KEYS[2] active key
      Keys.active(ctx),
      # KEYS[3] prioritized key
      Keys.prioritized(ctx),
      # KEYS[4] event stream key
      Keys.events(ctx),
      # KEYS[5] stalled key
      Keys.stalled(ctx),
      # KEYS[6] rate limiter key
      Keys.limiter(ctx),
      # KEYS[7] delayed key
      Keys.delayed(ctx),
      # KEYS[8] paused key
      Keys.paused(ctx),
      # KEYS[9] meta key
      Keys.meta(ctx),
      # KEYS[10] pc priority counter
      Keys.pc(ctx),
      # KEYS[11] completed/failed key
      Keys.get(ctx, target_str),
      # KEYS[12] jobId key
      Keys.job(ctx, job_id),
      # KEYS[13] metrics key
      metrics_key,
      # KEYS[14] marker key
      Keys.marker(ctx)
    ]

    timestamp = Keyword.get(opts, :timestamp, System.system_time(:millisecond))
    lock_duration = Keyword.get(opts, :lock_duration, 30_000)
    fetch_next = if Keyword.get(opts, :fetch_next, true), do: 1, else: 0
    name = Keyword.get(opts, :name)
    attempts = Keyword.get(opts, :attempts, 0)
    limiter = Keyword.get(opts, :limiter)

    # Determine keepJobs based on target
    keep_jobs =
      case target do
        :completed -> Keyword.get(opts, :remove_on_complete, %{"count" => -1})
        :failed -> Keyword.get(opts, :remove_on_fail, %{"count" => -1})
        _ -> %{"count" => -1}
      end
      |> normalize_keep_jobs()

    # Determine prop val attribute based on target
    prop_val =
      case target do
        :completed -> "returnvalue"
        :failed -> "failedReason"
        _ -> "returnvalue"
      end

    # ARGV[8] msgpacked opts
    packed_opts =
      Msgpax.pack!(
        %{
          "token" => token,
          "name" => name,
          "keepJobs" => keep_jobs,
          "limiter" => limiter,
          "lockDuration" => lock_duration,
          "attempts" => attempts,
          "maxMetricsSize" => Keyword.get(opts, :max_metrics_size, ""),
          "fpof" => Keyword.get(opts, :fail_parent_on_failure, false),
          "cpof" => Keyword.get(opts, :continue_parent_on_failure, false),
          "idof" => Keyword.get(opts, :ignore_dependency_on_failure, false),
          "rdof" => Keyword.get(opts, :remove_dependency_on_failure, false)
        },
        iodata: false
      )

    # Encode result value
    result_val = if result == nil, do: "null", else: encode_result_value(result)

    # Build fields to update (for stacktrace)
    fields_to_update = build_fields_to_update(opts)

    args = [
      # ARGV[1] jobId
      job_id,
      # ARGV[2] timestamp
      timestamp,
      # ARGV[3] msg property returnvalue / failedReason
      prop_val,
      # ARGV[4] return value / failed reason
      result_val,
      # ARGV[5] target (completed/failed)
      target_str,
      # ARGV[6] fetch next?
      fetch_next,
      # ARGV[7] keys prefix (with trailing colon)
      Keys.key_prefix(ctx),
      # ARGV[8] opts
      packed_opts,
      # ARGV[9] job fields to update (for stacktrace)
      fields_to_update
    ]

    execute(conn, :move_to_finished, keys, args)
  end

  # Normalize keep_jobs to BullMQ format
  defp normalize_keep_jobs(true), do: %{"count" => 0}
  defp normalize_keep_jobs(false), do: %{"count" => -1}
  defp normalize_keep_jobs(n) when is_integer(n), do: %{"count" => n}
  defp normalize_keep_jobs(%{count: n}), do: %{"count" => n}
  defp normalize_keep_jobs(%{"count" => _} = m), do: m
  defp normalize_keep_jobs(%{age: n}), do: %{"age" => n}
  defp normalize_keep_jobs(%{"age" => _} = m), do: m
  defp normalize_keep_jobs(%{limit: n}), do: %{"limit" => n}
  defp normalize_keep_jobs(%{"limit" => _} = m), do: m

  # Handle maps with multiple keys (age + count, age + limit, etc.)
  defp normalize_keep_jobs(m) when is_map(m) do
    # Convert atom keys to string keys and keep the map as-is
    # This handles cases like %{age: 1000, limit: 5}
    for {k, v} <- m, into: %{} do
      key = if is_atom(k), do: Atom.to_string(k), else: k
      {key, v}
    end
  end

  defp normalize_keep_jobs(_), do: %{"count" => -1}

  # Build msgpacked fields to update on the job (for stacktrace)
  # The Lua script expects a flat list: ["field1", "value1", "field2", "value2", ...]
  defp build_fields_to_update(opts) do
    stacktrace = Keyword.get(opts, :stacktrace)

    fields =
      if stacktrace && stacktrace != "" do
        # Stacktrace is stored as JSON array in Redis (like Node.js)
        # Each retry appends to the array, but we just store one for now
        stacktrace_json = Jason.encode!([stacktrace])
        ["stacktrace", stacktrace_json]
      else
        []
      end

    if fields == [] do
      ""
    else
      Msgpax.pack!(fields, iodata: false)
    end
  end

  @doc """
  Moves a job to delayed state (for retry with delay).
  """
  @spec move_to_delayed(
          atom(),
          queue_context(),
          String.t(),
          String.t(),
          non_neg_integer(),
          keyword()
        ) :: script_result()
  def move_to_delayed(conn, ctx, job_id, token, delay, opts \\ []) do
    # Lua script expects:
    # KEYS[1] marker key
    # KEYS[2] active key
    # KEYS[3] prioritized key
    # KEYS[4] delayed key
    # KEYS[5] job key
    # KEYS[6] events stream
    # KEYS[7] meta key
    # KEYS[8] stalled key
    keys = [
      Keys.marker(ctx),
      Keys.active(ctx),
      Keys.pc(ctx),
      Keys.delayed(ctx),
      Keys.job(ctx, job_id),
      Keys.events(ctx),
      Keys.meta(ctx),
      Keys.stalled(ctx)
    ]

    timestamp = Keyword.get(opts, :timestamp, System.system_time(:millisecond))
    skip_attempt = if Keyword.get(opts, :skip_attempt, false), do: "1", else: "0"

    # Build fields to update (for stacktrace on retry)
    fields_to_update = build_fields_to_update(opts)

    # ARGV[1] key prefix
    # ARGV[2] timestamp (current time)
    # ARGV[3] job id
    # ARGV[4] queue token
    # ARGV[5] delay value
    # ARGV[6] skip attempt flag ("0" or "1")
    # ARGV[7] optional job fields to update (for stacktrace)
    args = [
      Keys.key_prefix(ctx),
      timestamp,
      job_id,
      token,
      delay,
      skip_attempt,
      fields_to_update
    ]

    execute(conn, :move_to_delayed, keys, args)
  end

  @doc """
  Moves a job from active state back to wait.

  This is useful when manually processing jobs and you need to release
  a job back to the queue (e.g., due to rate limiting).

  ## Returns

    * `{:ok, pttl}` - The rate limit TTL in milliseconds (0 if no rate limit)
  """
  @spec move_job_from_active_to_wait(atom(), queue_context(), String.t(), String.t()) ::
          script_result()
  def move_job_from_active_to_wait(conn, ctx, job_id, token \\ "0") do
    # Keys expected by moveJobFromActiveToWait-9.lua:
    # KEYS[1]  active key
    # KEYS[2]  wait key
    # KEYS[3]  stalled key
    # KEYS[4]  paused key
    # KEYS[5]  meta key
    # KEYS[6]  limiter key
    # KEYS[7]  prioritized key
    # KEYS[8]  marker key
    # KEYS[9]  event key
    keys = [
      Keys.active(ctx),
      Keys.wait(ctx),
      Keys.stalled(ctx),
      Keys.paused(ctx),
      Keys.meta(ctx),
      Keys.limiter(ctx),
      Keys.pc(ctx),
      Keys.marker(ctx),
      Keys.events(ctx)
    ]

    # ARGV[1] job id
    # ARGV[2] lock token
    # ARGV[3] job id key
    args = [
      job_id,
      token,
      Keys.job(ctx, job_id)
    ]

    case execute(conn, :move_job_from_active_to_wait, keys, args) do
      {:ok, pttl} when is_integer(pttl) -> {:ok, pttl}
      {:ok, _} -> {:ok, 0}
      error -> error
    end
  end

  @doc """
  Moves a job from active to waiting-children state.

  This is used when a job needs to wait for its child jobs to complete
  before continuing. The job will be automatically moved back to waiting
  when all children complete.

  ## Returns

    * `{:ok, 0}` - Successfully moved to waiting-children
    * `{:ok, 1}` - No pending dependencies
    * `{:ok, -1}` - Missing job
    * `{:ok, -2}` - Missing lock
    * `{:ok, -3}` - Job not in active set
    * `{:ok, -9}` - Job has failed children
  """
  @spec move_to_waiting_children(atom(), queue_context(), String.t(), String.t(), keyword()) ::
          script_result()
  def move_to_waiting_children(conn, ctx, job_id, token, opts \\ []) do
    # Keys expected by moveToWaitingChildren-7.lua:
    # KEYS[1] active key
    # KEYS[2] wait-children key
    # KEYS[3] job key
    # KEYS[4] job dependencies key
    # KEYS[5] job unsuccessful key
    # KEYS[6] stalled key
    # KEYS[7] events key
    keys = [
      Keys.active(ctx),
      Keys.waiting_children(ctx),
      Keys.job(ctx, job_id),
      Keys.job_dependencies(ctx, job_id),
      Keys.job_unsuccessful(ctx, job_id),
      Keys.stalled(ctx),
      Keys.events(ctx)
    ]

    child_key = Keyword.get(opts, :child_key, "")
    timestamp = Keyword.get(opts, :timestamp, System.system_time(:millisecond))

    # ARGV[1] token
    # ARGV[2] child key
    # ARGV[3] timestamp
    # ARGV[4] jobId
    # ARGV[5] prefix
    args = [
      token,
      child_key,
      timestamp,
      job_id,
      Keys.key_prefix(ctx)
    ]

    execute(conn, :move_to_waiting_children, keys, args)
  end

  @doc """
  Extends the lock on a job.
  """
  @spec extend_lock(atom(), queue_context(), String.t(), String.t(), non_neg_integer()) ::
          script_result()
  def extend_lock(conn, ctx, job_id, token, duration) do
    keys = [
      Keys.lock(ctx, job_id),
      Keys.stalled(ctx)
    ]

    args = [token, duration, job_id]

    execute(conn, :extend_lock, keys, args)
  end

  @doc """
  Extends locks for multiple jobs in a single call.

  This is more efficient than calling extend_lock multiple times when
  processing many concurrent jobs, as it uses a single Redis call.

  Returns a list of results (1 for success, job_id for failures).
  """
  @spec extend_locks(atom(), queue_context(), [String.t()], [String.t()], non_neg_integer()) ::
          script_result()
  def extend_locks(conn, ctx, job_ids, tokens, duration)
      when is_list(job_ids) and is_list(tokens) do
    keys = [
      Keys.stalled(ctx)
    ]

    # Pack job_ids and tokens using msgpack for the Lua script
    args = [
      Keys.key(ctx),
      Msgpax.pack!(tokens, iodata: false),
      Msgpax.pack!(job_ids, iodata: false),
      duration
    ]

    execute(conn, :extend_locks, keys, args)
  end

  @doc """
  Releases the lock on a job.
  """
  @spec release_lock(atom(), queue_context(), String.t(), String.t()) :: script_result()
  def release_lock(conn, ctx, job_id, token) do
    keys = [Keys.lock(ctx, job_id)]
    args = [token]

    execute(conn, :release_lock, keys, args)
  end

  @doc """
  Moves stalled jobs back to wait.
  """
  @spec move_stalled_jobs_to_wait(atom(), queue_context(), non_neg_integer(), keyword()) ::
          script_result()
  def move_stalled_jobs_to_wait(conn, ctx, max_stalled_count, opts \\ []) do
    keys = [
      Keys.stalled(ctx),
      Keys.wait(ctx),
      Keys.active(ctx),
      Keys.failed(ctx),
      Keys.stalled_check(ctx),
      Keys.meta(ctx),
      Keys.paused(ctx),
      Keys.marker(ctx)
    ]

    timestamp = Keyword.get(opts, :timestamp, System.system_time(:millisecond))

    args = [
      # ARGV[1] prefix:queueName
      Keys.key(ctx),
      # ARGV[2] max stalled count
      max_stalled_count,
      # ARGV[3] timestamp
      timestamp
    ]

    execute(conn, :move_stalled_jobs_to_wait, keys, args)
  end

  @doc """
  Retries a failed job.
  """
  @spec retry_job(atom(), queue_context(), String.t(), boolean(), String.t()) :: script_result()
  def retry_job(conn, ctx, job_id, lifo, token) do
    keys = [
      Keys.failed(ctx),
      Keys.wait(ctx),
      Keys.active(ctx),
      Keys.marker(ctx)
    ]

    args = [
      # ARGV[1] prefix:queueName
      Keys.key(ctx),
      # ARGV[2] job id
      job_id,
      # ARGV[3] push type
      if(lifo, do: "LIFO", else: "FIFO"),
      # ARGV[4] worker token
      token
    ]

    execute(conn, :retry_job, keys, args)
  end

  # ---------------------------------------------------------------------------
  # Queue Operations - Used by Queue module
  # ---------------------------------------------------------------------------

  @doc """
  Pauses or resumes the queue.
  """
  @spec pause(atom(), queue_context(), boolean()) :: script_result()
  def pause(conn, ctx, paused?) do
    keys = [
      Keys.wait(ctx),
      Keys.paused(ctx),
      Keys.meta(ctx),
      Keys.prioritized(ctx),
      Keys.pc(ctx),
      Keys.marker(ctx),
      Keys.events(ctx)
    ]

    args = [if(paused?, do: "paused", else: "resumed")]

    execute(conn, :pause, keys, args)
  end

  @doc """
  Drains the queue (removes all jobs).
  """
  @spec drain(atom(), queue_context(), boolean()) :: script_result()
  def drain(conn, ctx, delayed?) do
    keys = [
      Keys.wait(ctx),
      Keys.paused(ctx),
      Keys.delayed(ctx),
      Keys.prioritized(ctx),
      Keys.pc(ctx)
    ]

    args = [
      # ARGV[1] prefix:queueName
      Keys.key(ctx),
      # ARGV[2] include delayed
      if(delayed?, do: 1, else: 0)
    ]

    execute(conn, :drain, keys, args)
  end

  @doc """
  Obliterates the queue (removes everything including meta).
  """
  @spec obliterate(atom(), queue_context(), non_neg_integer(), boolean()) :: script_result()
  def obliterate(conn, ctx, count, force \\ false) do
    keys = [
      Keys.meta(ctx),
      Keys.key_prefix(ctx)
    ]

    args = [count, if(force, do: "force", else: "")]

    execute(conn, :obliterate, keys, args)
  end

  @doc """
  Removes a job from the queue.
  """
  @spec remove_job(atom(), queue_context(), String.t(), boolean()) :: script_result()
  def remove_job(conn, ctx, job_id, remove_children) do
    keys = [
      # KEYS[1] jobKey
      Keys.job(ctx, job_id),
      # KEYS[2] repeat key
      Keys.repeat(ctx)
    ]

    args = [
      # ARGV[1] jobId
      job_id,
      # ARGV[2] remove children
      if(remove_children, do: "1", else: "0"),
      # ARGV[3] queue prefix (with trailing colon)
      Keys.key(ctx) <> ":"
    ]

    execute(conn, :remove_job, keys, args)
  end

  @doc """
  Gets the state of a job.
  """
  @spec get_state(atom(), queue_context(), String.t()) :: script_result()
  def get_state(conn, ctx, job_id) do
    # Keys expected by getState-8.lua:
    # KEYS[1] 'completed' key
    # KEYS[2] 'failed' key
    # KEYS[3] 'delayed' key
    # KEYS[4] 'active' key
    # KEYS[5] 'wait' key
    # KEYS[6] 'paused' key
    # KEYS[7] 'waiting-children' key
    # KEYS[8] 'prioritized' key
    keys = [
      Keys.completed(ctx),
      Keys.failed(ctx),
      Keys.delayed(ctx),
      Keys.active(ctx),
      Keys.wait(ctx),
      Keys.paused(ctx),
      Keys.waiting_children(ctx),
      Keys.prioritized(ctx)
    ]

    args = [job_id]

    execute(conn, :get_state, keys, args)
  end

  @doc """
  Promotes a delayed job to wait.
  """
  @spec promote(atom(), queue_context(), String.t()) :: script_result()
  def promote(conn, ctx, job_id) do
    keys = [
      Keys.delayed(ctx),
      Keys.wait(ctx),
      Keys.pc(ctx),
      Keys.prioritized(ctx),
      Keys.paused(ctx),
      Keys.marker(ctx),
      Keys.events(ctx),
      Keys.meta(ctx),
      Keys.job(ctx, job_id)
    ]

    args = [
      # ARGV[1] prefix:queueName
      Keys.key(ctx),
      # ARGV[2] job id
      job_id
    ]

    execute(conn, :promote, keys, args)
  end

  @doc """
  Updates a job scheduler and adds the next delayed job.

  Called by the worker after completing a repeatable job to schedule
  the next iteration.

  ## Parameters

    * `conn` - Redis connection
    * `ctx` - Queue context (keys structure)
    * `scheduler_id` - The job scheduler ID (repeat_job_key)
    * `next_millis` - Next execution time in milliseconds
    * `template_data` - JSON-encoded job data
    * `job_opts` - Msgpacked job options
    * `producer_id` - The ID of the job that produced this iteration

  ## Returns

    * `{:ok, job_id}` - The ID of the next scheduled job
    * `{:ok, nil}` - Scheduler doesn't exist or duplicate
    * `{:error, reason}` - Error
  """
  @spec update_job_scheduler(
          atom(),
          queue_context(),
          String.t(),
          non_neg_integer(),
          String.t(),
          binary(),
          String.t()
        ) :: script_result()
  def update_job_scheduler(
        conn,
        ctx,
        scheduler_id,
        next_millis,
        template_data,
        job_opts,
        producer_id
      ) do
    # KEYS in order expected by Lua script (updateJobScheduler-12.lua)
    keys = [
      # KEYS[1] 'repeat' key
      Keys.repeat(ctx),
      # KEYS[2] 'delayed'
      Keys.delayed(ctx),
      # KEYS[3] 'wait' key
      Keys.wait(ctx),
      # KEYS[4] 'paused' key
      Keys.paused(ctx),
      # KEYS[5] 'meta'
      Keys.meta(ctx),
      # KEYS[6] 'prioritized' key
      Keys.prioritized(ctx),
      # KEYS[7] 'marker'
      Keys.marker(ctx),
      # KEYS[8] 'id'
      Keys.id(ctx),
      # KEYS[9] events stream key
      Keys.events(ctx),
      # KEYS[10] 'pc' priority counter
      Keys.pc(ctx),
      # KEYS[11] producer key
      Keys.job(ctx, producer_id),
      # KEYS[12] 'active' key
      Keys.active(ctx)
    ]

    timestamp = System.system_time(:millisecond)

    args = [
      # ARGV[1] next milliseconds
      next_millis,
      # ARGV[2] jobs scheduler id
      scheduler_id,
      # ARGV[3] Json stringified delayed data
      template_data,
      # ARGV[4] msgpacked delayed opts
      job_opts,
      # ARGV[5] timestamp
      timestamp,
      # ARGV[6] prefix key
      Keys.key_prefix(ctx),
      # ARGV[7] producer id
      producer_id
    ]

    execute(conn, :update_job_scheduler, keys, args)
  end

  @doc """
  Adds a log entry to a job.
  """
  @spec add_log(atom(), queue_context(), String.t(), String.t(), non_neg_integer() | nil) ::
          script_result()
  def add_log(conn, ctx, job_id, log_message, keep_logs) do
    keys = [
      Keys.job(ctx, job_id),
      Keys.logs(ctx, job_id)
    ]

    args = [
      job_id,
      log_message,
      keep_logs || ""
    ]

    execute(conn, :add_log, keys, args)
  end

  @doc """
  Updates the progress of a job.
  """
  @spec update_progress(atom(), queue_context(), String.t(), any()) :: script_result()
  def update_progress(conn, ctx, job_id, progress) do
    keys = [
      Keys.job(ctx, job_id),
      Keys.events(ctx),
      Keys.meta(ctx)
    ]

    args = [
      job_id,
      encode_progress(progress)
    ]

    execute(conn, :update_progress, keys, args)
  end

  @doc """
  Updates the data of a job.
  """
  @spec update_data(atom(), queue_context(), String.t(), map()) :: script_result()
  def update_data(conn, ctx, job_id, data) do
    keys = [Keys.job(ctx, job_id)]
    args = [Jason.encode!(data)]

    execute(conn, :update_data, keys, args)
  end

  @doc """
  Gets job counts for the queue.
  """
  @spec get_counts(atom(), queue_context()) :: script_result()
  def get_counts(conn, ctx) do
    keys = [Keys.key(ctx)]
    args = []

    execute(conn, :get_counts, keys, args)
  end

  @doc """
  Checks if the queue is at its max limit.
  """
  @spec is_maxed(atom(), queue_context()) :: script_result()
  def is_maxed(conn, ctx) do
    keys = [
      Keys.limiter(ctx),
      Keys.meta(ctx)
    ]

    args = []

    execute(conn, :is_maxed, keys, args)
  end

  @doc """
  Gets the rate limit TTL.

  ## Options
    * `:max_jobs` - Maximum jobs for rate limit (default: 0, uses meta key)
  """
  @spec get_rate_limit_ttl(atom(), queue_context(), keyword()) :: script_result()
  def get_rate_limit_ttl(conn, ctx, opts \\ []) do
    keys = [
      Keys.limiter(ctx),
      Keys.meta(ctx)
    ]

    max_jobs = Keyword.get(opts, :max_jobs, 0)
    args = [max_jobs]

    execute(conn, :get_rate_limit_ttl, keys, args)
  end

  @doc """
  Reprocesses a job that is in completed or failed state.

  Moves a finished job back to the wait queue for reprocessing.

  ## Parameters

    * `conn` - Redis connection
    * `ctx` - Queue context from Keys.new/2
    * `job_id` - The job ID to reprocess
    * `state` - The expected current state (:failed or :completed)
    * `opts` - Options:
      * `:lifo` - If true, push to front of queue (default: false)
      * `:reset_attempts_made` - Reset attempts counter (default: false)
      * `:reset_attempts_started` - Reset attempts started counter (default: false)

  ## Returns

    * `{:ok, 1}` - Job successfully moved to wait
    * `{:error, reason}` - Error with code indicating failure:
      * -1: Job does not exist
      * -3: Job was not found in the expected state
  """
  @spec reprocess_job(atom(), queue_context(), String.t(), atom(), keyword()) :: script_result()
  def reprocess_job(conn, ctx, job_id, state, opts \\ []) when state in [:failed, :completed] do
    state_str = to_string(state)
    lifo = Keyword.get(opts, :lifo, false)
    reset_attempts_made = Keyword.get(opts, :reset_attempts_made, false)
    reset_attempts_started = Keyword.get(opts, :reset_attempts_started, false)

    keys = [
      Keys.job(ctx, job_id),
      Keys.events(ctx),
      if(state == :failed, do: Keys.failed(ctx), else: Keys.completed(ctx)),
      Keys.wait(ctx),
      Keys.meta(ctx),
      Keys.paused(ctx),
      Keys.active(ctx),
      Keys.marker(ctx)
    ]

    args = [
      job_id,
      if(lifo, do: "RPUSH", else: "LPUSH"),
      if(state == :failed, do: "failedReason", else: "returnvalue"),
      state_str,
      if(reset_attempts_made, do: "1", else: "0"),
      if(reset_attempts_started, do: "1", else: "0")
    ]

    execute(conn, :reprocess_job, keys, args)
  end

  @doc """
  Gets queue metrics for completed or failed jobs.

  Returns metrics data including count, previous timestamp, previous count,
  data points, and total number of points.
  """
  @spec get_metrics(atom(), queue_context(), :completed | :failed, integer(), integer()) ::
          script_result()
  def get_metrics(conn, ctx, type, start_idx \\ 0, end_idx \\ -1) do
    type_str = Atom.to_string(type)

    keys = [
      # KEYS[1] metrics key
      Keys.metrics(ctx, type_str),
      # KEYS[2] metrics data key
      "#{Keys.metrics(ctx, type_str)}:data"
    ]

    args = [
      # ARGV[1] start index
      start_idx,
      # ARGV[2] end index
      end_idx
    ]

    case execute(conn, :get_metrics, keys, args) do
      {:ok, [meta, data, count]} ->
        [count_str, prev_ts_str, prev_count_str] = meta || ["0", "0", "0"]

        {:ok,
         %{
           meta: %{
             count: parse_int(count_str || "0"),
             prev_ts: parse_int(prev_ts_str || "0"),
             prev_count: parse_int(prev_count_str || "0")
           },
           data: Enum.map(data || [], fn point -> parse_int(point || "0") end),
           count: count || 0
         }}

      {:ok, result} ->
        {:ok, result}

      {:error, _} = error ->
        error
    end
  end

  defp parse_int(nil), do: 0
  defp parse_int(val) when is_integer(val), do: val

  defp parse_int(val) when is_binary(val) do
    case Integer.parse(val) do
      {int, _} -> int
      :error -> 0
    end
  end

  # ---------------------------------------------------------------------------
  # Private Helpers
  # ---------------------------------------------------------------------------

  defp get_job_id(job) when is_struct(job), do: Map.get(job, :id)
  defp get_job_id(job) when is_map(job), do: Map.get(job, :id) || Map.get(job, "id")
  defp get_job_id(_), do: nil

  defp get_job_name(job) when is_struct(job), do: Map.get(job, :name) || ""
  defp get_job_name(job) when is_map(job), do: Map.get(job, :name) || Map.get(job, "name") || ""
  defp get_job_name(_), do: ""

  defp get_job_timestamp(job) when is_struct(job), do: Map.get(job, :timestamp)

  defp get_job_timestamp(job) when is_map(job),
    do: Map.get(job, :timestamp) || Map.get(job, "timestamp")

  defp get_job_timestamp(_), do: nil

  defp get_job_data(job) when is_struct(job), do: Map.get(job, :data) || %{}
  defp get_job_data(job) when is_map(job), do: Map.get(job, :data) || Map.get(job, "data") || %{}
  defp get_job_data(_), do: %{}

  defp get_job_delay(job) when is_struct(job), do: Map.get(job, :delay) || 0
  defp get_job_delay(job) when is_map(job), do: Map.get(job, :delay) || Map.get(job, "delay") || 0
  defp get_job_delay(_), do: 0

  defp get_job_priority(job) when is_struct(job), do: Map.get(job, :priority) || 0

  defp get_job_priority(job) when is_map(job),
    do: Map.get(job, :priority) || Map.get(job, "priority") || 0

  defp get_job_priority(_), do: 0

  defp get_repeat_job_key(job) when is_struct(job), do: Map.get(job, :repeat_job_key)

  defp get_repeat_job_key(job) when is_map(job),
    do: Map.get(job, :repeat_job_key) || Map.get(job, "repeatJobKey")

  defp get_repeat_job_key(_), do: nil

  defp get_deduplication_key(job, ctx) do
    dedup_id =
      cond do
        is_struct(job) -> Map.get(job, :deduplication_id)
        is_map(job) -> Map.get(job, :deduplication_id) || Map.get(job, "deduplicationId")
        true -> nil
      end

    if dedup_id do
      # Build deduplication key: prefix:queueName:de:deduplicationId
      "#{Keys.key(ctx)}:de:#{dedup_id}"
    else
      nil
    end
  end

  # Get full parent info: {parentKey, parentDepsKey, parent}
  defp get_parent_info_full(job) when is_struct(job) do
    parent = Map.get(job, :parent)

    if parent do
      parent_key = Map.get(parent, :key) || build_parent_key(parent)
      parent_id = Map.get(parent, :id) || ""
      queue_key = Map.get(parent, :queue_key) || ""
      parent_deps_key = if parent_key != "", do: "#{parent_key}:dependencies", else: nil

      parent_obj =
        if parent_id != "" do
          %{"id" => parent_id, "queueKey" => queue_key}
        else
          nil
        end

      {parent_key, parent_deps_key, parent_obj}
    else
      {nil, nil, nil}
    end
  end

  defp get_parent_info_full(job) when is_map(job) do
    parent = Map.get(job, :parent) || Map.get(job, "parent")

    if parent do
      parent_key = Map.get(parent, :key) || Map.get(parent, "key") || build_parent_key(parent)
      parent_id = Map.get(parent, :id) || Map.get(parent, "id") || ""
      queue_key = Map.get(parent, :queue_key) || Map.get(parent, "queueKey") || ""
      parent_deps_key = if parent_key != "", do: "#{parent_key}:dependencies", else: nil

      parent_obj =
        if parent_id != "" do
          %{"id" => parent_id, "queueKey" => queue_key}
        else
          nil
        end

      {parent_key, parent_deps_key, parent_obj}
    else
      {nil, nil, nil}
    end
  end

  defp get_parent_info_full(_), do: {nil, nil, nil}

  defp build_parent_key(parent) when is_map(parent) do
    queue_key = Map.get(parent, :queue_key) || Map.get(parent, "queueKey") || ""
    id = Map.get(parent, :id) || Map.get(parent, "id") || ""

    if queue_key != "" and id != "" do
      "#{queue_key}:#{id}"
    else
      ""
    end
  end

  defp build_parent_key(_), do: ""

  # Get job options (opts field)
  defp get_job_opts(job) when is_struct(job), do: Map.get(job, :opts) || %{}
  defp get_job_opts(job) when is_map(job), do: Map.get(job, :opts) || Map.get(job, "opts") || %{}
  defp get_job_opts(_), do: %{}

  # Pack job options using msgpack with BullMQ's compressed option names
  # Note: This function is kept for potential future use in add_job operations
  @doc false
  def pack_job_opts(job, additional_opts \\ %{}) do
    job_opts = get_job_opts(job)
    delay = get_job_delay(job)
    priority = get_job_priority(job)

    # Build options map with BullMQ's short key names
    opts =
      %{}
      |> maybe_add_opt("del", Map.get(job_opts, :delay) || Map.get(job_opts, "delay") || delay, 0)
      |> maybe_add_opt(
        "pri",
        Map.get(job_opts, :priority) || Map.get(job_opts, "priority") || priority,
        0
      )
      |> maybe_add_opt("at", Map.get(job_opts, :attempts) || Map.get(job_opts, "attempts"), nil)
      |> maybe_add_opt("bo", get_backoff_opts(job_opts), nil)
      |> maybe_add_opt("lifo", Map.get(job_opts, :lifo) || Map.get(job_opts, "lifo"), nil)
      |> maybe_add_opt(
        "ro",
        Map.get(job_opts, :remove_on_complete) || Map.get(job_opts, "removeOnComplete"),
        nil
      )
      |> maybe_add_opt(
        "rof",
        Map.get(job_opts, :remove_on_fail) || Map.get(job_opts, "removeOnFail"),
        nil
      )
      |> maybe_add_opt(
        "fpof",
        Map.get(job_opts, :fail_parent_on_failure) || Map.get(job_opts, "failParentOnFailure"),
        nil
      )
      |> maybe_add_opt(
        "idof",
        Map.get(job_opts, :ignore_dependency_on_failure) ||
          Map.get(job_opts, "ignoreDependencyOnFailure"),
        nil
      )
      |> maybe_add_opt(
        "rdof",
        Map.get(job_opts, :remove_dependency_on_failure) ||
          Map.get(job_opts, "removeDependencyOnFailure"),
        nil
      )
      |> maybe_add_opt("kl", Map.get(job_opts, :keep_logs) || Map.get(job_opts, "keepLogs"), nil)
      |> maybe_add_opt("rep", get_repeat_opts(job_opts), nil)
      |> maybe_add_opt("de", get_deduplication_opts(job_opts), nil)
      |> Map.merge(additional_opts)

    Msgpax.pack!(opts, iodata: false)
  end

  defp maybe_add_opt(map, _key, nil, _default), do: map
  defp maybe_add_opt(map, _key, value, default) when value == default, do: map
  defp maybe_add_opt(map, key, value, _default), do: Map.put(map, key, value)

  defp get_backoff_opts(opts) do
    backoff = Map.get(opts, :backoff) || Map.get(opts, "backoff")

    case backoff do
      %{type: type, delay: delay} -> %{"type" => type, "delay" => delay}
      %{"type" => _, "delay" => _} = b -> b
      _ -> nil
    end
  end

  defp get_repeat_opts(opts) do
    repeat = Map.get(opts, :repeat) || Map.get(opts, "repeat")

    case repeat do
      nil -> nil
      r when is_map(r) -> r
      _ -> nil
    end
  end

  defp get_deduplication_opts(opts) do
    dedup = Map.get(opts, :deduplication) || Map.get(opts, "deduplication")

    case dedup do
      nil ->
        nil

      d when is_map(d) ->
        # Convert Elixir-style keys to the short keys expected by Lua scripts
        result = %{}

        result =
          case Map.get(d, :id) || Map.get(d, "id") do
            nil -> result
            v -> Map.put(result, "id", v)
          end

        result =
          case Map.get(d, :ttl) || Map.get(d, "ttl") do
            nil -> result
            v -> Map.put(result, "ttl", v)
          end

        result =
          case Map.get(d, :extend) || Map.get(d, "extend") do
            nil -> result
            v -> Map.put(result, "extend", v)
          end

        result =
          case Map.get(d, :replace) || Map.get(d, "replace") do
            nil -> result
            v -> Map.put(result, "replace", v)
          end

        result =
          case Map.get(d, :keep_last_if_active) || Map.get(d, "keepLastIfActive") do
            nil -> result
            v -> Map.put(result, "keepLastIfActive", v)
          end

        if map_size(result) > 0, do: result, else: nil

      _ ->
        nil
    end
  end

  defp encode_result_value(result) when is_map(result), do: Jason.encode!(result)
  defp encode_result_value(result) when is_binary(result), do: result
  defp encode_result_value(result), do: inspect(result)

  defp encode_progress(progress) when is_map(progress), do: Jason.encode!(progress)
  defp encode_progress(progress) when is_integer(progress), do: Integer.to_string(progress)
  defp encode_progress(progress), do: Jason.encode!(progress)

  defp encode_arg(arg) when is_binary(arg), do: arg
  defp encode_arg(arg) when is_integer(arg), do: Integer.to_string(arg)
  defp encode_arg(arg) when is_float(arg), do: Float.to_string(arg)

  defp encode_arg(arg) when is_atom(arg) and not is_nil(arg) and not is_boolean(arg),
    do: Atom.to_string(arg)

  defp encode_arg(true), do: "1"
  defp encode_arg(false), do: "0"
  defp encode_arg(nil), do: ""
  defp encode_arg(arg) when is_map(arg), do: Jason.encode!(arg)
  defp encode_arg(arg) when is_list(arg), do: Jason.encode!(arg)

  defp decode_result(result) when is_list(result), do: Enum.map(result, &decode_result/1)
  defp decode_result(result), do: result

  defp sha1(content) do
    :crypto.hash(:sha, content) |> Base.encode16(case: :lower)
  end
end