step_flow

defmodule StepFlow.Jobs do
  @moduledoc """
  The Jobs context.
  """

  require Logger

  import Ecto.Query, warn: false
  alias StepFlow.Repo

  alias StepFlow.Amqp.CommonEmitter
  alias StepFlow.Controllers.BlackList
  alias StepFlow.Controllers.Jobs
  alias StepFlow.Jobs.Job
  alias StepFlow.Jobs.Status
  alias StepFlow.Progressions.Progression
  alias StepFlow.QueryFilter
  alias StepFlow.Workers.WorkerStatuses
  alias StepFlow.Workflows

  @doc """
  Returns the list of jobs.

  ## Examples

      iex> list_jobs()
      [%Job{}, ...]

  """
  def list_jobs(params \\ %{}) do
    query =
      from(job in Job)
      |> apply_rights(params)
      |> apply_default_query_filters(params)

    internal_query_jobs(query, params)
  end

  def internal_list_jobs(params \\ %{}) do
    query =
      from(job in Job)
      |> apply_default_query_filters(params)

    internal_query_jobs(query, params)
  end

  defp internal_query_jobs(query, params) do
    page =
      Map.get(params, "page", 0)
      |> StepFlow.Integer.force()

    size =
      Map.get(params, "size", 10)
      |> StepFlow.Integer.force()

    offset = page * size

    total_query = from(item in query, select: count(item.id))

    total =
      Repo.all(total_query)
      |> List.first()

    query =
      from(
        job in query,
        order_by: [desc: :inserted_at],
        offset: ^offset,
        limit: ^size
      )

    jobs =
      Repo.all(query)
      |> Repo.preload([
        :status,
        :progressions,
        :updates,
        child_workflow: [:jobs, :artifacts, :status]
      ])

    %{
      data: jobs,
      total: total,
      page: page,
      size: size
    }
  end

  def apply_rights(query, params) do
    allowed_workflows = Workflows.check_rights(params)

    case Enum.member?(allowed_workflows, "*") do
      true ->
        query

      false ->
        allowed_job_ids = Repo.all(query) |> check_rights(allowed_workflows)

        from(
          job in query,
          where: job.id in ^allowed_job_ids
        )
    end
  end

  def apply_default_query_filters(query, params) do
    query =
      case StepFlow.Map.get_by_key_or_atom(params, "workflow_id") do
        nil ->
          query

        str_workflow_id ->
          workflow_id = StepFlow.Integer.force(str_workflow_id)
          from(job in query, where: job.workflow_id == ^workflow_id)
      end

    query =
      query
      |> QueryFilter.filter_query(params, :job_type, :name)
      |> QueryFilter.filter_query(params, :step_id)
      |> filter_status(params)
      |> filter_worker_label(params)
      |> filter_worker_version(params)
      |> filter_worker_instance_id(params)

    query =
      case Map.get(params, "direct_messaging_queue_name") do
        nil ->
          query

        direct_messaging_queue_name ->
          direct_messaging_queue_name =
            String.replace(direct_messaging_queue_name, "direct_messaging_", "")

          expected =
            %{
              id: "direct_messaging_queue_name",
              type: "string",
              value: direct_messaging_queue_name
            }
            |> Jason.encode!()

          from(
            job in query,
            where: fragment("? @> array[?::text]::jsonb[]", job.parameters, ^expected)
          )
      end

    query
    |> QueryFilter.apply_end_date_filter(params, :end_date)
    |> QueryFilter.apply_end_date_filter(params, :before_date)
    |> QueryFilter.apply_start_date_filter(params, :start_date)
    |> QueryFilter.apply_start_date_filter(params, :after_date)
  end

  @doc """
  Returns the list of jobs pending"
  """
  def get_pending_jobs_by_type do
    ids_with_progress =
      from(p in Progression,
        select: %{job_id: p.job_id}
      )

    ids_with_status =
      from(s in Status,
        select: %{job_id: s.job_id}
      )

    query =
      from(job in Job,
        left_join: progress in subquery(ids_with_progress),
        on: job.id == progress.job_id,
        left_join: status in subquery(ids_with_status),
        on: job.id == status.job_id,
        where: is_nil(progress.job_id) and is_nil(status.job_id),
        select: %{name: job.name, value: count(job.id)},
        group_by: job.name
      )

    Repo.all(query)
  end

  @doc """
  Returns the list of jobs pending"
  """
  def get_processing_jobs_by_type do
    sub_query_status =
      from(
        status in Status,
        select: %{job_id: status.job_id, inserted_at: max(status.inserted_at)},
        group_by: [status.job_id]
      )

    sub_query =
      from(
        status in Status,
        join: st in subquery(sub_query_status),
        on: status.job_id == st.job_id and status.inserted_at == st.inserted_at,
        where: status.state == :processing
      )

    query =
      from(job in Job,
        right_join: st in subquery(sub_query),
        on: st.job_id == job.id,
        select: %{name: job.name, value: count(job.id)},
        group_by: job.name
      )

    Repo.all(query)
  end

  defp check_rights(jobs, allowed_workflows) do
    Enum.map(
      jobs,
      fn job ->
        if Enum.member?(
             allowed_workflows,
             StepFlow.Map.get_by_key_or_atom(
               Workflows.get_workflow_for_job!(job.id),
               :identifier
             )
           ) do
          job.id
        end
      end
    )
  end

  def filter_status(query, params) do
    QueryFilter.filter_query_with_related_entry(
      query,
      params,
      :states,
      :id,
      Status,
      :state,
      :job_id
    )
  end

  def filter_worker_version(query, params) do
    QueryFilter.filter_query_with_related_entry(
      query,
      params,
      :versions,
      :last_worker_instance_id,
      StepFlow.Workers.WorkerStatus,
      :version,
      :instance_id
    )
  end

  def filter_worker_label(query, params) do
    QueryFilter.filter_query_with_related_entry(
      query,
      params,
      :labels,
      :last_worker_instance_id,
      StepFlow.Workers.WorkerStatus,
      :label,
      :instance_id
    )
  end

  def filter_worker_instance_id(query, params) do
    QueryFilter.filter_query_with_related_entry(
      query,
      params,
      :instance_ids,
      :last_worker_instance_id,
      StepFlow.Workers.WorkerStatus,
      :instance_id,
      :instance_id
    )
  end

  @doc """
  Gets a single job.

  Raises `Ecto.NoResultsError` if the Job does not exist.

  ## Examples

      iex> get_job!(123)
      %Job{}

      iex> get_job!(456)
      ** (Ecto.NoResultsError)

  """
  def get_job!(id), do: Repo.get!(Job, id)

  @doc """
  Gets a single job.

  ## Examples

      iex> get_job(123)
      %Job{}

      iex> get_job(456)
      nil

  """
  def get_job(id), do: Repo.get(Job, id)

  @doc """
  Gets a single job by workflow ID and step ID

  ## Examples

      iex> get_job(123)
      %Job{}

      iex> get_job(456)
      nil

  """
  def get_by!(%{"workflow_id" => workflow_id, "step_id" => step_id}) do
    Repo.get_by!(Job, workflow_id: workflow_id, step_id: step_id)
  end

  @doc """
  Gets a single job by workflow ID and step ID

  ## Examples

      iex> get_job(123)
      %Job{}

      iex> get_job(456)
      nil

  """
  def get_by(%{"workflow_id" => workflow_id, "step_id" => step_id}) do
    Repo.get_by(Job, workflow_id: workflow_id, step_id: step_id)
  end

  @doc """
  Gets a single job with its related status.

  Raises `Ecto.NoResultsError` if the Job does not exist.

  ## Examples

      iex> get_job_with_status!(123)
      %Job{}

      iex> get_job!(456)
      ** (Ecto.NoResultsError)

  """
  def get_job_with_status!(id) do
    get_job!(id)
    |> Repo.preload([
      :status,
      :progressions,
      :updates,
      child_workflow: [:jobs, :artifacts, :status]
    ])
  end

  @doc """
  Creates a job.

  ## Examples

      iex> create_job(%{field: value})
      {:ok, %Job{}}

      iex> create_job(%{field: bad_value})
      {:error, %Ecto.Changeset{}}

  """
  def create_job(attrs \\ %{}) do
    %Job{}
    |> Job.changeset(attrs)
    |> Repo.insert()
  end

  def copy_job(%Job{} = job) do
    %{
      name: job.name,
      step_id: job.step_id,
      parameters: job.parameters,
      workflow_id: job.workflow_id,
      is_live: job.is_live,
      is_updatable: job.is_updatable,
      allow_failure: job.allow_failure
    }
    |> create_job()
  end

  @doc """
  Creates a job with a paused status.

  ## Examples

      iex> create_paused_job(workflow, 1, "download_http")
      {:ok, "paused"}

  """
  def create_paused_job(workflow_id, step_id, step_name, post_action) do
    job_params = %{
      name: step_name,
      step_id: step_id,
      workflow_id: workflow_id,
      parameters: []
    }

    {:ok, job} = create_job(job_params)
    Status.set_job_status(job.id, :queued)
    Status.set_job_status(job.id, :processing)
    Status.set_job_status(job.id, :paused, post_action)
  end

  @doc """
  Creates a job with a skipped status.

  ## Examples

      iex> create_skipped_job(workflow, 1, "download_http")
      {:ok, "skipped"}

  """
  def create_skipped_job(workflow, step_id, action) do
    job_params = %{
      name: action,
      step_id: step_id,
      workflow_id: workflow.id,
      parameters: []
    }

    {:ok, job} = create_job(job_params)
    Status.set_job_status(job.id, :skipped)
    {:ok, "skipped"}
  end

  @doc """
  Creates a job with an error status.

  ## Examples

      iex> create_error_job(workflow, step_id, "download_http", "unsupported step")
      {:ok, "created"}

  """
  def create_error_job(workflow, step_id, action, description) do
    job_params = %{
      name: action,
      step_id: step_id,
      workflow_id: workflow.id,
      parameters: []
    }

    {:ok, job} = create_job(job_params)
    Status.set_job_status(job.id, :queued)
    Status.set_job_status(job.id, :processing)
    Status.set_job_status(job.id, :error, %{message: description})
    {:ok, "created"}
  end

  @doc """
  Creates a job with a completed status.

  ## Examples

      iex> create_completed_job(workflow, step_id, "webhook_notification")
      {:ok, "completed"}

  """
  def create_completed_job(workflow, step_id, action) do
    job_params = %{
      name: action,
      step_id: step_id,
      workflow_id: workflow.id,
      parameters: []
    }

    {:ok, job} = create_job(job_params)
    Status.set_job_status(job.id, :queued)
    Status.set_job_status(job.id, :processing)
    Status.set_job_status(job.id, :completed)
    {:ok, "completed"}
  end

  defp abort_job(job) do
    worker_status = WorkerStatuses.get_worker_status_for_job(job.id)

    case worker_status do
      nil ->
        Logger.warning("Cannot abort job #{job.id}, no associated running worker found.")

      worker_status ->
        message =
          Jobs.get_message(job)
          |> Map.put(:type, "stop_process")

        Logger.info(
          "Send stop_process message for step #{job.id} to worker #{worker_status.instance_id}"
        )

        publish_result =
          case CommonEmitter.publish_json(
                 worker_status.direct_messaging_queue_name,
                 job.id,
                 message,
                 "direct_messaging",
                 headers: [{"instance_id", :longstr, worker_status.instance_id}]
               ) do
            :ok ->
              {:ok, "stopped"}

            _ ->
              {:error, "unable to publish message"}
          end

        publish_result
    end
  end

  def abort_jobs(workflow, step_id, action) do
    jobs =
      internal_list_jobs(%{
        name: action,
        step_id: step_id,
        workflow_id: workflow.id,
        # assuming that a workflow cannot have more than 1000 jobs
        size: 1000
      })

      # Create dedicated method
      |> Map.get(:data)

    # Black list queued jobs
    jobs
    |> Enum.filter(fn job ->
      last_status = Jobs.get_last_status(job)
      last_status.state == :queued
    end)
    |> Enum.each(fn job ->
      BlackList.add_and_notify!(job.id)
    end)

    # Abort running jobs
    jobs
    |> Enum.filter(fn job ->
      length(job.progressions) > 0
    end)
    |> Enum.each(fn job ->
      abort_job(job)
    end)
  end

  @doc """
  Set skipped status to all queued jobs.

  ## Examples

      iex> skip_jobs(workflow, step_id, "download_http")
      :ok

  """
  def skip_jobs(workflow, step_id, action) do
    internal_list_jobs(%{
      name: action,
      step_id: step_id,
      workflow_id: workflow.id
    })

    # Create dedicated method
    |> Map.get(:data)
    |> Enum.filter(fn job ->
      case job.status do
        [%{state: state}] -> state != "queued"
        _ -> false
      end
    end)
    |> Enum.each(fn job ->
      Status.set_job_status(job.id, :skipped)
    end)
  end

  @doc """
  Updates a job.

  ## Examples

      iex> update_job(job, %{field: new_value})
      {:ok, %Job{}}

      iex> update_job(job, %{field: bad_value})
      {:error, %Ecto.Changeset{}}

  """
  def update_job(%Job{} = job, attrs) do
    job
    |> Job.changeset(attrs)
    |> Repo.update()
  end

  @doc """
  Deletes a Job.

  ## Examples

      iex> delete_job(job)
      {:ok, %Job{}}

      iex> delete_job(job)
      {:error, %Ecto.Changeset{}}

  """
  def delete_job(%Job{} = job) do
    job
    |> Job.changeset(%{deleted: true})
    |> Repo.update()
  end

  @doc """
  Returns an `%Ecto.Changeset{}` for tracking job changes.

  ## Examples

      iex> change_job(job)
      %Ecto.Changeset{source: %Job{}}

  """
  def change_job(%Job{} = job) do
    Job.changeset(job, %{})
  end
end