step_flow

defmodule StepFlow.WebController.WorkflowEvents do
  use StepFlow, :controller
  use OpenApiSpex.ControllerSpecs
  require Logger

  import StepFlow.WebController.Helpers

  alias StepFlow.{
    Amqp.CommonEmitter,
    Jobs,
    Repo,
    Step.Live,
    Updates,
    WebController.OpenApiSchemas,
    Workflows
  }

  @moduledoc false

  tags ["Workflows"]
  security [%{"authorization" => %OpenApiSpex.SecurityScheme{type: "http", scheme: "bearer"}}]

  action_fallback(StepFlow.WebController.Fallback)

  operation :handle,
    summary: "Trigger workflows event",
    description: "Trigger workflows event",
    type: :string,
    parameters: [
      workflow_id: [
        in: :path,
        description: "Workflow ID",
        type: :integer,
        example: 1
      ]
    ],
    request_body: {"Eventbody", "application/json", OpenApiSchemas.Workflows.EventBody},
    responses: [
      ok: "OK",
      forbidden: "Forbidden"
    ]

  def handle(
        %Plug.Conn{assigns: %{current_user: user}} = conn,
        %{"workflow_id" => workflow_id} = params
      ) do
    workflow = Workflows.get_workflow!(workflow_id)

    case {params, workflow.is_live} do
      {%{"event" => "abort"}, false} ->
        abort(conn, workflow, user)

      {%{"event" => "pause", "post_action" => action, "trigger_at" => trigger_date_time}, _} ->
        pause(conn, workflow, user, action, trigger_date_time)

      {%{"event" => "resume"}, _} ->
        resume(conn, workflow, user)

      {%{"event" => "update", "job_id" => job_id, "parameters" => parameters}, _} ->
        update(conn, workflow, user, job_id, parameters)

      {%{"event" => "retry", "job_id" => job_id}, _} ->
        retry(conn, workflow, user, job_id)

      {%{"event" => "stop"}, true} ->
        stop(conn, workflow, user)

      {%{"event" => "delete"}, _} ->
        delete(conn, workflow, user)

      {_, _} ->
        send_resp(conn, 422, "event is not supported")
    end
  end

  def handle(conn, _) do
    conn
    |> put_status(:forbidden)
    |> json(%{status: "error", message: "Forbidden to handle workflow with this identifier"})
  end

  defp internal_retry_handle(conn, workflow, job, _job_name, last_status_state, user_uuid)
       when last_status_state in [:error] do
    internal_retry_statuses(job, workflow, user_uuid)

    params = %{
      job_id: job.id,
      parameters: job.parameters
    }

    case CommonEmitter.publish_json(job.name, job.step_id, params) do
      :ok ->
        StepFlow.Notification.send("retry_job", %{workflow_id: workflow.id, body: params})

        Jobs.Status.set_job_status(job.id, :queued)

        conn
        |> put_status(:ok)
        |> json(%{status: "ok"})

      _ ->
        conn
        |> put_status(:ok)
        |> json(%{status: "error", message: "unable to publish message"})
    end
  end

  defp internal_retry_handle(conn, _workflow, _job, _job_name, _last_status_state, _user_uuid) do
    send_resp(conn, :forbidden, "illegal operation")
  end

  defp internal_retry_statuses(job, workflow, user_uuid) do
    {:ok, job_status} = Jobs.Status.set_job_status(job.id, :retrying, %{user_uuid: user_uuid})

    {:ok, _status} =
      Workflows.Status.define_workflow_status(workflow.id, :job_retrying, job_status)

    if workflow.parent_id != nil do
      job = Jobs.get_job!(workflow.parent_id) |> Repo.preload([:child_workflow])
      {:ok, job_status} = Jobs.Status.set_job_status(job.id, :retrying, %{user_uuid: user_uuid})

      {:ok, _status} =
        Workflows.Status.define_workflow_status(job.workflow_id, :job_retrying, job_status)

      {:ok, _status} = Jobs.Status.set_job_status(job.id, :queued)
      {:ok, datetime} = DateTime.now("Etc/UTC")

      CommonEmitter.publish_json(
        "job_progression",
        0,
        %{
          job_id: job.id,
          datetime: datetime,
          docker_container_id: "workflow",
          progression: 0
        },
        "job_response"
      )
    end
  end

  defp update(conn, workflow, user, job_id, parameters) do
    if has_right?("workflow::" <> workflow.identifier, user, "update") &&
         workflow.parent_id == nil do
      Logger.warn("update job #{job_id}")

      job = Jobs.get_job_with_status!(job_id)

      if job.is_updatable do
        Updates.update_parameters(job, parameters)

        conn
        |> put_status(:ok)
        |> json(%{status: "ok"})
      else
        Logger.error("Job #{job_id} cannot be updated !")

        conn
        |> put_status(:error)
        |> json(%{status: "error", message: "Forbidden to update this job"})
      end
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to update this workflow"})
    end
  end

  defp abort(conn, workflow, user) do
    if has_right?("workflow::" <> workflow.identifier, user, "abort") && workflow.parent_id == nil do
      Workflows.abort(workflow)

      conn
      |> put_status(:ok)
      |> json(%{status: "ok"})
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to abort this workflow"})
    end
  end

  defp pause(conn, workflow, user, action, trigger_date_time) do
    if has_right?("workflow::" <> workflow.identifier, user, "pause") && workflow.parent_id == nil do
      case Workflows.pause(workflow, action, trigger_date_time) do
        {:ok, _} ->
          conn
          |> put_status(:ok)
          |> json(%{status: "ok"})

        {:error, message} when is_bitstring(message) ->
          Logger.error(
            "Something went wrong pausing workflow #{workflow.id} #{workflow.identifier}: #{inspect(message)}"
          )

          conn
          |> put_status(:not_found)
          |> json(%{status: "error", message: message})

        {:error, message} ->
          Logger.error(
            "Something went wrong pausing workflow #{workflow.id} #{workflow.identifier}: #{inspect(message)}"
          )

          conn
          |> put_status(:internal_server_error)
          |> json(%{status: "error", message: "Internal server error"})

        _ ->
          conn
          |> put_status(:not_found)
          |> json(%{status: "error", message: "Unknown error"})
      end
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to pause this workflow"})
    end
  end

  defp resume(conn, workflow, user) do
    if has_right?("workflow::" <> workflow.identifier, user, "resume") &&
         workflow.parent_id == nil do
      case Workflows.resume(workflow) do
        {:ok, _} ->
          conn
          |> put_status(:ok)
          |> json(%{status: "ok"})

        {:error, message} ->
          Logger.error(
            "Something went wrong resuming workflow #{workflow.id} #{workflow.identifier}: #{inspect(message)}"
          )

          conn
          |> put_status(:internal_server_error)
          |> json(%{status: "error", message: "Internal server error"})
      end
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to resume this workflow"})
    end
  end

  defp retry(conn, workflow, user, job_id) do
    job = Jobs.get_job_with_status!(job_id)

    if has_right?("workflow::" <> workflow.identifier, user, "retry") && Map.has_key?(user, :uuid) &&
         job.child_workflow == nil do
      Logger.warn("User #{user.uuid} is retrying job #{job_id}")

      last_status = StepFlow.Controllers.Jobs.get_last_status(job.status)

      internal_retry_handle(conn, workflow, job, job.name, last_status.state, user.uuid)
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to retry this workflow"})
    end
  end

  defp stop(conn, workflow, user) do
    if has_right?("workflow::" <> workflow.identifier, user, "abort") do
      workflow_jobs = Repo.preload(workflow, [:jobs]).jobs

      workflow_jobs
      |> Live.stop_jobs()

      {:ok, _status} = Workflows.Status.set_workflow_status(workflow.id, :stopped)

      topic = "update_workflow_" <> Integer.to_string(workflow.id)
      StepFlow.Notification.send(topic, %{workflow_id: workflow.id})

      conn
      |> put_status(:ok)
      |> json(%{status: "ok"})
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to stop this workflow"})
    end
  end

  defp delete(conn, workflow, user) do
    if has_right?("workflow::" <> workflow.identifier, user, "delete") do
      for job <- workflow.jobs do
        if job.child_workflow != nil do
          child_workflow = Workflows.get_workflow!(job.child_workflow.id)
          delete(conn, child_workflow, user)
        end

        Jobs.delete_job(job)
      end

      Workflows.delete_workflow(workflow)
      StepFlow.Notification.send("delete_workflow", %{workflow_id: workflow.id})

      conn
      |> put_status(:ok)
      |> json(%{status: "ok"})
    else
      conn
      |> put_status(:forbidden)
      |> json(%{status: "error", message: "Forbidden to delete this workflow"})
    end
  end
end