step_flow

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

  import Ecto.Query, warn: false
  alias StepFlow.Repo
  alias StepFlow.Roles
  alias StepFlow.WorkflowDefinitions.ExternalLoader
  alias StepFlow.WorkflowDefinitions.WorkflowDefinition
  require Logger

  @doc """
  Returns the list of Workflow Definitions.
  """
  def list_workflow_definitions(params \\ %{}) do
    allowed_workflows = check_rights(params)

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

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

    mode = Map.get(params, "mode", "full")

    offset = page * size

    query =
      from(workflow_definition in WorkflowDefinition)
      |> filter_by_rights(allowed_workflows)
      |> filter_by_label_or_identifier(Map.get(params, "search"))
      |> filter_by_versions(Map.get(params, "versions"))
      |> select_by_mode(mode)

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

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

    query =
      from(
        workflow_definition in subquery(query),
        order_by: [
          desc: workflow_definition.version_major,
          desc: workflow_definition.version_minor,
          desc: workflow_definition.version_micro
        ]
      )
      |> paginate(offset, size)

    workflow_definitions = Repo.all(query)

    %{
      data: workflow_definitions,
      total: total,
      page: page,
      size: size,
      mode: mode
    }
  end

  defp check_rights(params) do
    roles =
      StepFlow.Map.get_by_key_or_atom(params, :roles)
      |> Roles.get_roles()

    view_rights =
      case roles do
        nil ->
          []

        _ ->
          Roles.get_rights_for_entity_type_and_action(roles, "workflow", "view")
      end

    for %{entity: entity} <- view_rights,
        do:
          String.split(entity, "::")
          |> List.last()
  end

  defp paginate(query, offset, size) do
    case size do
      -1 ->
        query

      _ ->
        from(
          workflow_definition in subquery(query),
          offset: ^offset,
          limit: ^size
        )
    end
  end

  def filter_by_versions(query, versions) do
    case versions do
      ["latest"] ->
        from(
          workflow_definition in subquery(query),
          order_by: [
            desc: workflow_definition.version_major,
            desc: workflow_definition.version_minor,
            desc: workflow_definition.version_micro
          ],
          distinct: :identifier
        )

      versions when is_list(versions) and length(versions) != 0 ->
        from(
          workflow_definition in subquery(query),
          where:
            fragment(
              "concat(?, '.', ?, '.', ?) = ANY(?)",
              workflow_definition.version_major,
              workflow_definition.version_minor,
              workflow_definition.version_micro,
              ^versions
            )
        )

      _ ->
        query
    end
  end

  defp filter_by_label_or_identifier(query, search) do
    case search do
      nil ->
        query

      search ->
        like = "%#{search}%"

        from(
          workflow_definition in subquery(query),
          where:
            ilike(workflow_definition.label, ^like) or
              ilike(workflow_definition.identifier, ^search)
        )
    end
  end

  defp filter_by_rights(query, allowed) do
    if Enum.member?(allowed, "*") do
      query
    else
      from(
        workflow_definition in subquery(query),
        where: workflow_definition.identifier in ^allowed
      )
    end
  end

  defp select_by_mode(query, mode) do
    case mode do
      "simple" ->
        from(
          workflow_definition in subquery(query),
          select: %{
            id: workflow_definition.id,
            identifier: workflow_definition.identifier,
            is_live: workflow_definition.is_live,
            label: workflow_definition.label,
            version_major: workflow_definition.version_major,
            version_minor: workflow_definition.version_minor,
            version_micro: workflow_definition.version_micro
          }
        )

      "full" ->
        query

      _ ->
        query
    end
  end

  @doc """
  Gets a single WorkflowDefinition (latest version).

  Returns nil if the WorkflowDefinition does not exist.
  """
  def get_workflow_definition(identifier) do
    query =
      from(workflow_definition in WorkflowDefinition,
        where: workflow_definition.identifier == ^identifier,
        order_by: [
          desc: workflow_definition.version_major,
          desc: workflow_definition.version_minor,
          desc: workflow_definition.version_micro
        ],
        limit: 1
      )

    Repo.one(query)
  end

  @doc """
  Gets a single WorkflowDefinition by specifying the version.

  Returns nil if the WorkflowDefinition does not exist.
  """
  def get_workflow_definition(identifier, version_major, version_minor, version_micro) do
    query =
      from(workflow_definition in WorkflowDefinition,
        where:
          workflow_definition.identifier == ^identifier and
            workflow_definition.version_major == ^version_major and
            workflow_definition.version_minor == ^version_minor and
            workflow_definition.version_micro == ^version_micro,
        limit: 1
      )

    Repo.one(query)
  end

  @doc """
  Loads workflows
  """
  def load_workflows do
    get_workflow_definition_directories()
    |> Enum.map(fn directory ->
      list_workflow_definitions_for_a_directory(directory)
    end)
    |> List.flatten()
  end

  @doc """
  Loads workflows in database
  """
  def load_workflows_in_database do
    get_workflow_definition_directories()
    |> Enum.map(fn directory ->
      list_workflow_definitions_for_a_directory(directory)
    end)
    |> List.flatten()
    |> Enum.each(fn workflow_definition ->
      case get_workflow_definition(
             StepFlow.Map.get_by_key_or_atom(workflow_definition, :identifier),
             StepFlow.Map.get_by_key_or_atom(workflow_definition, :version_major),
             StepFlow.Map.get_by_key_or_atom(workflow_definition, :version_minor),
             StepFlow.Map.get_by_key_or_atom(workflow_definition, :version_micro)
           ) do
        nil ->
          %WorkflowDefinition{}
          |> WorkflowDefinition.changeset(workflow_definition)
          |> Repo.insert()

        _ ->
          Logger.warn("Workflow already present in database")
      end
    end)
  end

  @doc """
  Get workflow definitions from directories
  """
  def get_workflow_definition_directories do
    Application.get_env(:step_flow, StepFlow)
    |> Keyword.get(:workflow_definition)
    |> case do
      {:system, key} ->
        System.get_env(key)
        |> String.split(get_separator())

      key when is_list(key) ->
        key

      key when is_bitstring(key) ->
        [key]

      key ->
        Logger.info("unable to use #{inspect(key)} to list directory")
        []
    end
  end

  def valid?(definition) do
    get_schema()
    |> JsonXema.valid?(definition)
  end

  def validate(definition) do
    get_schema()
    |> JsonXema.validate(definition)
  end

  defp get_schema do
    schema =
      Application.get_env(
        StepFlow.WorkflowDefinitions.WorkflowDefinition,
        :workflow_schema_url,
        "https://media-cloud.ai/standard/1.10/workflow-definition.schema.json"
      )
      |> load_content()
      |> Jason.decode!()

    :ok = JsonXema.SchemaValidator.validate("http://json-schema.org/draft-07/schema#", schema)

    JsonXema.new(schema, loader: ExternalLoader)
  end

  defp load_content("http://" <> _ = url) do
    HTTPoison.get!(url)
    |> Map.get(:body)
  end

  defp load_content("https://" <> _ = url) do
    HTTPoison.get!(url)
    |> Map.get(:body)
  end

  defp load_content(source_filename) do
    File.read!(source_filename)
  end

  defp get_separator do
    if :os.type() |> elem(0) == :unix do
      ":"
    else
      ";"
    end
  end

  defp list_workflow_definitions_for_a_directory(directory) do
    File.ls!(directory)
    |> Enum.filter(fn filename ->
      String.ends_with?(filename, ".json")
    end)
    |> Enum.map(fn filename ->
      Path.join(directory, filename)
      |> File.read!()
      |> Jason.decode!()
    end)
    |> Enum.filter(fn workflow_definition ->
      if valid?(workflow_definition) do
        true
      else
        fun = fn _error, path, acc ->
          ["at " <> inspect(path) | acc]
        end

        errors =
          validate(workflow_definition)
          |> JsonXema.ValidationError.travers_errors([], fun)

        Logger.error(
          "Workflow definition #{inspect(Map.get(workflow_definition, "identifier"))} not valid: #{inspect(errors)}"
        )

        false
      end
    end)
  end
end