step_flow

defmodule StepFlow.Workers.WorkerStatuses do
  @moduledoc """
  The WorkerStatuses context.
  """

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

  alias StepFlow.Workers.WorkerStatus

  @doc """
  Returns the list of WorkerStatuses.

  ## Examples

      iex> StepFlow.WorkerStatuses.list_worker_statuses()
      %{data: [], page: 0, size: 10, total: 0}

  """
  def list_worker_statuses(params \\ %{}) do
    page =
      Map.get(params, "page", 0)
      |> StepFlow.Integer.force()

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

    offset = page * size

    query = from(worker_status in WorkerStatus)

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

        instance_id ->
          from(
            worker_status in query,
            where: worker_status.instance_id == ^instance_id
          )
      end

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

        job_id ->
          {job_id, _} =
            job_id
            |> Integer.parse()

          from(
            worker_status in query,
            where:
              not is_nil(worker_status.current_job) and
                fragment("? ->'job_id'", worker_status.current_job) == ^job_id
          )
      end

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

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

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

    worker_statuses = Repo.all(query)

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

  @doc """
  Gets the last inserted WorkerStatus.

  Returns `nil` if the Worker does not exist.
  """
  def get_worker_status(instance_id) do
    query =
      Ecto.Query.from(worker_status in WorkerStatus,
        where: worker_status.instance_id == ^instance_id,
        order_by: [desc: worker_status.inserted_at],
        limit: 1
      )

    Repo.one(query)
  end

  @doc """
  Gets the last inserted WorkerStatus.


  Raises `Ecto.NoResultsError` if the WorkerStatus does not exist.
  """
  def get_worker_status!(instance_id) do
    query =
      Ecto.Query.from(worker_status in WorkerStatus,
        where: worker_status.instance_id == ^instance_id,
        order_by: [desc: worker_status.inserted_at],
        limit: 1
      )

    Repo.one!(query)
  end

  @doc """
  Gets the first WorkerStatus which is running a job which ID is `job_id`.

  Returns `nil` if that WorkerStatus does not exist.
  """
  def get_worker_status_for_job(job_id) do
    query =
      from(
        worker_status in WorkerStatus,
        where:
          not is_nil(worker_status.current_job) and
            fragment("? ->'job_id'", worker_status.current_job) == ^job_id,
        order_by: [desc: worker_status.inserted_at],
        limit: 1
      )

    Repo.one(query)
  end

  @doc """
  Creates a WorkerStatus.

  ## Examples

      iex> result = StepFlow.Workers.WorkerStatuses.create_worker_status!(%{
      ...>    job: nil,
      ...>    worker: %{
      ...>      activity: "Idle",
      ...>      description: "This worker is just an example.",
      ...>      direct_messaging_queue_name: "direct_messaging_e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>      instance_id: "e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>      label: "UnitTestWorker",
      ...>      queue_name: "job_test_worker",
      ...>      sdk_version: "2.3.4",
      ...>      short_description: "A test worker",
      ...>      system_info: %{
      ...>        docker_container_id: "e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>        number_of_processors: 12,
      ...>        total_memory: 16_574_754,
      ...>        total_swap: 2_046_816,
      ...>        used_memory: 8_865_633,
      ...>        used_swap: 0
      ...>      },
      ...>      version: "1.2.3"
      ...>    }
      ...> })
      ...> match?(%StepFlow.Workers.WorkerStatus{}, result)
      true

  Raises error if something went wrong during creation.
  """
  def create_worker_status!(%{} = message) do
    attrs =
      message
      |> StepFlow.Controllers.WorkerStatus.process_worker_status_message()

    %WorkerStatus{}
    |> WorkerStatus.changeset(attrs)
    |> Repo.insert!()
  end

  def create_worker_status!(%StepFlow.Progressions.Progression{} = message) do
    attrs =
      message
      |> StepFlow.Controllers.WorkerStatus.process_worker_status_message()

    %WorkerStatus{}
    |> WorkerStatus.changeset(attrs)
    |> Repo.insert!()
  end

  @doc """
  Creates a WorkerStatus.

  ## Examples

      iex> result = StepFlow.Workers.WorkerStatuses.create_worker_status!(%{
      ...>    job: nil,
      ...>    worker: %{
      ...>      activity: "Idle",
      ...>      description: "This worker is just an example.",
      ...>      direct_messaging_queue_name: "direct_messaging_e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>      instance_id: "e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>      label: "UnitTestWorker",
      ...>      queue_name: "job_test_worker",
      ...>      sdk_version: "2.3.4",
      ...>      short_description: "A test worker",
      ...>      system_info: %{
      ...>        docker_container_id: "e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>        number_of_processors: 12,
      ...>        total_memory: 16_574_754,
      ...>        total_swap: 2_046_816,
      ...>        used_memory: 8_865_633,
      ...>        used_swap: 0
      ...>      },
      ...>      version: "1.2.3"
      ...>    }
      ...> })
      ...> match?(%StepFlow.Workers.WorkerStatus{}, result)
      true

  Returns `{:ok, WorkerStatus}` on success, `{:error, changeset}` otherwise.
  """
  def create_worker_status(%{} = message) do
    attrs =
      message
      |> StepFlow.Controllers.WorkerStatus.process_worker_status_message()

    %WorkerStatus{}
    |> WorkerStatus.changeset(attrs)
    |> Repo.insert()
  end

  @doc """
  Updates a WorkerStatus.

  ## Examples

      iex> result = StepFlow.Workers.WorkerStatuses.update_worker_status!(%{
      ...>    job: %{
      ...>      destination_paths: [],
      ...>      execution_duration: 0.0,
      ...>      job_id: 1234,
      ...>      parameters: [],
      ...>      status: "processing"
      ...>    },
      ...>    worker: %{
      ...>      activity: "Idle",
      ...>      description: "This worker is just an example.",
      ...>      direct_messaging_queue_name: "direct_messaging_e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>      instance_id: "e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>      label: "UnitTestWorker",
      ...>      queue_name: "job_test_worker",
      ...>      sdk_version: "2.3.4",
      ...>      short_description: "A test worker",
      ...>      system_info: %{
      ...>        docker_container_id: "e1297fe6-fe94-49cf-9ef8-1a751cba28f2",
      ...>        number_of_processors: 12,
      ...>        total_memory: 16_574_754,
      ...>        total_swap: 2_046_816,
      ...>        used_memory: 8_865_633,
      ...>        used_swap: 0
      ...>      },
      ...>      version: "1.2.3"
      ...>    }
      ...> })
      ...> match?(%StepFlow.Workers.WorkerStatus{}, result)
      true

  Raises error if something went wrong during update.
  """
  def update_worker_status!(%WorkerStatus{} = worker_status, %{} = message) do
    attrs =
      message
      |> StepFlow.Controllers.WorkerStatus.process_worker_status_message()

    worker_status
    |> WorkerStatus.changeset(attrs)
    |> Repo.update!()
  end
end