arke_postgres

# Copyright 2023 Arkemis S.r.l.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

defmodule ArkePostgres.Query do
  import Ecto.Query

  @record_fields [:id, :arke_id, :data, :metadata, :inserted_at, :updated_at]

  def generate_query(
        %{filters: filters, orders: orders, offset: offset, limit: limit} = arke_query,
        action
      ) do
    base_query(arke_query, action)
    |> handle_filters(filters)
    |> handle_orders(orders)
    |> handle_offset(offset)
    |> handle_limit(limit)
  end

  def execute(query, :raw),
    do: Ecto.Adapters.SQL.to_sql(:all, ArkePostgres.Repo, generate_query(query, :raw))

  def execute(query, :all) do
    generate_query(query, :all)
    |> ArkePostgres.Repo.all(prefix: query.project)
    |> generate_units(query.arke, query.project)
  end

  def execute(query, :one) do
    record = generate_query(query, :one) |> ArkePostgres.Repo.one(prefix: query.project)
    init_unit(record, query.arke, query.project)
  end

  def execute(query, :count) do
    generate_query(query, :count) |> ArkePostgres.Repo.one(prefix: query.project)
  end

  def execute(query, :pseudo_query), do: generate_query(query, :pseudo_query)

  def get_column(%{data: %{persistence: "arke_parameter"}} = parameter),
    do: get_arke_column(parameter)

  def get_column(%{data: %{persistence: "table_column"}} = parameter),
    do: get_table_column(parameter)

  def get_manager_units(project_id) do
    arke_link = Arke.Boundary.ArkeManager.get(:arke_link, :arke_system)

    links =
      from(q in table_query(arke_link, nil), where: q.type in ["parameter", "group"])
      |> ArkePostgres.Repo.all(prefix: project_id)

    parameter_links = Enum.filter(links, fn x -> x.type == "parameter" end)
    group_links = Enum.filter(links, fn x -> x.type == "group" end)

    parameters_id = [
      "boolean",
      "dict",
      "list",
      "float",
      "integer",
      "string",
      "unit",
      "link",
      "date",
      "datetime",
      "time"
    ]

    list_arke_id = parameters_id ++ ["arke", "group"]

    unit_list =
      from(q in base_query(), where: q.arke_id in ^list_arke_id)
      |> ArkePostgres.Repo.all(prefix: project_id)
      |> generate_units(nil, project_id)

    parameters = Enum.filter(unit_list, fn u -> Atom.to_string(u.arke_id) in parameters_id end)

    arke_list =
      parse_arke_list(
        Enum.filter(unit_list, fn u -> Atom.to_string(u.arke_id) == "arke" end),
        parameter_links
      )

    groups =
      parse_groups(
        Enum.filter(unit_list, fn u -> Atom.to_string(u.arke_id) == "group" end),
        group_links
      )

    {parameters, arke_list, groups}
  end

  defp parse_arke_list(arke_list, parameter_links) do
    Enum.reduce(arke_list, [], fn %{id: id} = unit, new_arke_list ->
      params =
        Enum.reduce(
          Enum.filter(parameter_links, fn x -> x.parent_id == Atom.to_string(id) end),
          [],
          fn p, new_params ->
            [%{id: String.to_existing_atom(p.child_id), metadata: p.metadata} | new_params]
          end
        )

      [Arke.Core.Unit.update(unit, parameters: params) | new_arke_list]
    end)
  end

  defp parse_groups(groups, group_links) do
    Enum.reduce(groups, [], fn %{id: id} = unit, new_groups ->
      arke_list =
        Enum.reduce(
          Enum.filter(group_links, fn x -> x.parent_id == Atom.to_string(id) end),
          [],
          fn p, new_params ->
            [%{id: String.to_existing_atom(p.child_id), metadata: p.metadata} | new_params]
          end
        )

      [Arke.Core.Unit.update(unit, arke_list: arke_list) | new_groups]
    end)
  end

  #  def get_parameters(project_id) do
  #    query =  base_query()
  #    parameters_id = ["boolean", "dict", "float", "integer", "string", "unit", "link", "date", "datetime", "time"]
  #    (from q in query, where: q.arke_id in ^parameters_id)
  #                            |>  ArkePostgres.Repo.all(prefix: project_id) |> generate_units(nil, project_id)
  #  end
  #
  #  def get_arke_list(project_id) do
  #    arke_list = (from q in base_query(), where: q.arke_id == "arke")
  #      |>  ArkePostgres.Repo.all(prefix: project_id) |> generate_units("arke", project_id)
  #
  #    arke_link = Arke.Boundary.ArkeManager.get(:arke_link, :arke_system)
  #    links = (from q in table_query(arke_link, nil), where: q.type in ["parameter", "group"])
  #      |>  ArkePostgres.Repo.all(prefix: project_id)
  #    parameter_links = Enum.filter(parameter_links, fn x -> x.type == "parameter" end)
  #    group_links = Enum.filter(parameter_links, fn x -> x.type == "group" end)
  #
  #    Enum.reduce(arke_list, [], fn %{id: id}=unit, new_arke_list ->
  #      params = Enum.reduce(Enum.filter(parameter_links, fn x -> x.parent_id == Atom.to_string(id) end), [], fn p, new_params ->
  #        [%{id: String.to_existing_atom(p.child_id), type: nil, metadata: p.metadata} | new_params]
  #      end)
  #      [Arke.Core.Unit.update(unit, parameters: params) | new_arke_list]
  #    end)
  #  end

  ######################################################################################################################
  # PRIVATE FUNCTIONS ##################################################################################################
  ######################################################################################################################
  defp base_query(%{arke: %{data: %{type: "table"}, id: id} = arke} = _arke_query, action),
    do: table_query(arke, action)

  defp base_query(%{link: nil} = _arke_query, action), do: arke_query(action)

  defp base_query(%{link: link, project: project} = _arke_query, action),
    do:
      get_nodes(
        project,
        action,
        Atom.to_string(link.unit.id),
        link.depth,
        link.direction,
        link.type
      )

  defp base_query(), do: from("arke_unit", select: ^@record_fields)

  defp arke_query(:count), do: from("arke_unit", select: count("*"))
  defp arke_query(_action), do: from("arke_unit", select: ^@record_fields)

  defp table_query(%{id: id, data: data} = arke, action) do
    table_name = Atom.to_string(id)

    fields =
      Enum.reduce(data.parameters, [], fn %{id: parameter_id}, new_fields ->
        [parameter_id | new_fields]
      end)

    from(table_name, select: ^fields)
  end

  defp get_arke(project, %{arke_id: arke_id}, nil) when is_binary(arke_id),
    do: Arke.Boundary.ArkeManager.get(String.to_existing_atom(arke_id), project)

  defp get_arke(project, %{arke_id: arke_id}, nil) when is_atom(arke_id),
    do: Arke.Boundary.ArkeManager.get(arke_id, project)

  defp get_arke(project, %{arke_id: arke_id}, nil),
    do: Arke.Boundary.ArkeManager.get(String.to_existing_atom(arke_id), project)

  defp get_arke(project, %{arke_id: arke_id}, arke) when is_atom(arke),
    do: Arke.Boundary.ArkeManager.get(arke, project)

  defp get_arke(project, %{arke_id: arke_id}, arke) when is_binary(arke),
    do: Arke.Boundary.ArkeManager.get(String.to_existing_atom(arke), project)

  defp get_arke(_, %{arke_id: arke_id}, arke), do: arke
  defp get_arke(_, _data, arke), do: arke

  defp generate_units(data, arke, project) do
    Enum.reduce(data, [], fn d, units ->
      units ++ [init_unit(d, arke, project)]
    end)
  end

  defp init_unit(nil, _, _), do: nil

  defp init_unit(record, arke, project) do
    arke = get_arke(project, record, arke)
    {metadata, record} = Map.pop(record, :metadata)
    {record_data, record} = Map.pop(record, :data, %{})

    record_data =
      Enum.map(arke.data.parameters, fn p ->
        {p.id, Map.get(record_data, Atom.to_string(p.id), nil)}
      end)
      |> Map.new()

    #    record_data = Enum.map(record_data, fn {k, v} -> {String.to_existing_atom(k), v} end) |> Map.new()
    record = Map.put(record, :metadata, Map.merge(metadata, %{project: project}))
    record = Map.merge(record_data, record)
    Arke.Core.Unit.load(arke, record)
  end

  defp handle_filters(query, filters) do
    Enum.reduce(filters, query, fn %{logic: logic, negate: negate, base_filters: base_filters},
                                   new_query ->
      clause = handle_condition(logic, base_filters) |> handle_negate_condition(negate)
      from(q in new_query, where: ^clause)
    end)
  end

  defp handle_condition(logic, base_filters) do
    Enum.reduce(base_filters, nil, fn %{
                                        parameter: parameter,
                                        operator: operator,
                                        value: value,
                                        negate: negate
                                      },
                                      clause ->
      column = get_column(parameter)
      value = get_value(parameter, value)

      condition =
        filter_query_by_operator(column, value, operator) |> handle_negate_condition(negate)

      add_condition_to_clause(condition, clause, logic)
    end)
  end

  defp handle_negate_condition(condition, true), do: dynamic([q], not (^condition))
  defp handle_negate_condition(condition, false), do: condition

  defp add_condition_to_clause(condition, nil, _), do: dynamic([q], ^condition)
  defp add_condition_to_clause(condition, clause, :and), do: dynamic([q], ^clause and ^condition)
  defp add_condition_to_clause(condition, clause, :or), do: dynamic([q], ^clause or ^condition)

  defp handle_orders(query, orders) do
    order_by =
      Enum.reduce(orders, [], fn %{parameter: parameter, direction: direction}, new_order_by ->
        column = get_column(parameter)
        [{direction, column} | new_order_by]
      end)

    from(q in query, order_by: ^order_by)
  end

  defp handle_offset(query, offset) when is_nil(offset), do: query
  defp handle_offset(query, offset), do: from(q in query, offset: ^offset)

  defp handle_limit(query, limit) when is_nil(limit), do: query
  defp handle_limit(query, limit), do: from(q in query, limit: ^limit)

  defp get_table_column(%{id: id} = _parameter), do: dynamic([q], fragment("?", field(q, ^id)))

  defp get_arke_column(%{id: id, arke_id: :string} = _parameter),
    do: dynamic([q], fragment("(? -> ? ->> 'value')::text", field(q, :data), ^Atom.to_string(id)))

  defp get_arke_column(%{id: id, arke_id: :atom} = _parameter),
    do: dynamic([q], fragment("(? -> ? ->> 'value')::text", field(q, :data), ^Atom.to_string(id)))

  defp get_arke_column(%{id: id, arke_id: :boolean} = _parameter),
    do:
      dynamic(
        [q],
        fragment("(? -> ? ->> 'value')::boolean", field(q, :data), ^Atom.to_string(id))
      )

  defp get_arke_column(%{id: id, arke_id: :datetime} = _parameter),
    do:
      dynamic(
        [q],
        fragment("(? -> ? ->> 'value')::datetime", field(q, :data), ^Atom.to_string(id))
      )

  defp get_arke_column(%{id: id, arke_id: :time} = _parameter),
    do: dynamic([q], fragment("(? -> ? ->> 'value')::time", field(q, :data), ^Atom.to_string(id)))

  defp get_arke_column(%{id: id, arke_id: :integer} = _parameter),
    do:
      dynamic(
        [q],
        fragment("(? -> ? ->> 'value')::integer", field(q, :data), ^Atom.to_string(id))
      )

  defp get_arke_column(%{id: id, arke_id: :float} = _parameter),
    do:
      dynamic([q], fragment("(? -> ? ->> 'value')::float", field(q, :data), ^Atom.to_string(id)))

  defp get_arke_column(%{id: id, arke_id: :dict} = _parameter),
    do: dynamic([q], fragment("(? -> ? ->> 'value')::JSON", field(q, :data), ^Atom.to_string(id)))

  defp get_arke_column(%{id: id, arke_id: :list} = _parameter),
    do: dynamic([q], fragment("(? -> ? ->> 'value')::JSON", field(q, :data), ^Atom.to_string(id)))

  defp get_value(_parameter, value) when is_nil(value), do: value
  defp get_value(_parameter, value) when is_list(value), do: value
  defp get_value(_parameter, value) when is_map(value), do: value

  defp get_value(parameter, value) when is_atom(value) and not is_boolean(value),
    do: get_value(parameter, Atom.to_string(value))

  defp get_value(%{id: id, arke_id: :string} = _parameter, value) when is_binary(value), do: value
  defp get_value(%{id: id, arke_id: :string} = _parameter, value), do: Kernel.inspect(value)

  defp get_value(%{id: id, arke_id: :integer} = _parameter, value) when is_number(value),
    do: value

  defp get_value(%{id: id, arke_id: :integer} = _parameter, value) when is_binary(value),
    do: String.to_integer(value)

  defp get_value(%{id: id, arke_id: :float} = _parameter, value) when is_number(value), do: value

  defp get_value(%{id: id, arke_id: :float} = _parameter, value) when is_binary(value),
    do: String.to_float(value)

  defp get_value(%{id: id, arke_id: :boolean} = _parameter, true), do: true
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, "true"), do: true
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, "True"), do: true
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, 1), do: true
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, "1"), do: true
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, false), do: false
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, "false"), do: false
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, "False"), do: false
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, 0), do: false
  defp get_value(%{id: id, arke_id: :boolean} = _parameter, "0"), do: false

  defp get_value(%{id: id, arke_id: :boolean} = _parameter, _),
    do: raise("Parameter(#{id}) value not valid")

  defp get_value(%{id: id, arke_id: :datetime} = _parameter, value), do: value
  defp get_value(%{id: id, arke_id: :time} = _parameter, value), do: value
  defp get_value(%{id: id, arke_id: :dict} = _parameter, value), do: value
  defp get_value(%{id: id, arke_id: :list} = _parameter, value), do: value
  defp get_value(%{id: id}, value), do: raise("Parameter(#{id}) value not valid")

  defp filter_query_by_operator(column, value, :eq), do: dynamic([q], ^column == ^value)

  defp filter_query_by_operator(column, value, :contains),
    do: dynamic([q], like(^column, fragment("?", ^("%" <> value <> "%"))))

  defp filter_query_by_operator(column, value, :icontains),
    do: dynamic([q], ilike(^column, fragment("?", ^("%" <> value <> "%"))))

  defp filter_query_by_operator(column, value, :startswith),
    do: dynamic([q], like(^column, fragment("?", ^("%" <> value))))

  defp filter_query_by_operator(column, value, :istartswith),
    do: dynamic([q], ilike(^column, fragment("?", ^("%" <> value))))

  defp filter_query_by_operator(column, value, :endswith),
    do: dynamic([q], like(^column, fragment("?", ^(value <> "%"))))

  defp filter_query_by_operator(column, value, :iendswith),
    do: dynamic([q], ilike(^column, fragment("?", ^(value <> "%"))))

  defp filter_query_by_operator(column, value, :lte), do: dynamic([q], ^column <= ^value)
  defp filter_query_by_operator(column, value, :lt), do: dynamic([q], ^column < ^value)
  defp filter_query_by_operator(column, value, :gt), do: dynamic([q], ^column > ^value)
  defp filter_query_by_operator(column, value, :gte), do: dynamic([q], ^column >= ^value)
  defp filter_query_by_operator(column, value, :in), do: dynamic([q], ^column in ^value)
  defp filter_query_by_operator(column, value, _), do: dynamic([q], ^column == ^value)

  # defp filter_query_by_operator(query, key, value, "between"), do: from q in query, where: column_table(q, ^key) == ^value

  ######################################################################################################################
  # ARKE LINK ##########################################################################################################
  ######################################################################################################################

  @raw_cte_child_query """
  (
    WITH RECURSIVE tree(depth, parent_id, type, child_id, metadata) AS (
      SELECT 0, parent_id, type, child_id, metadata FROM ?.arke_link WHERE ? = ?
      UNION SELECT
        depth + 1,
        ?.arke_link.parent_id,
        ?.arke_link.type,
        ?.arke_link.child_id,
        ?.arke_link.metadata
      FROM
        ?.arke_link JOIN tree
        ON ?.arke_link.? = tree.?
      WHERE
       depth < ?
    )
    SELECT * FROM tree ORDER BY depth
  )
  """
  @raw_cte_parent_query """
  (
    WITH RECURSIVE tree(depth, parent_id, type, child_id, metadata) AS (
      SELECT 0, parent_id, type, child_id, metadata FROM arke_link WHERE child_id = ?
      UNION SELECT
        depth + 1,
        arke_link.parent_id,
        arke_link.type,
        arke_link.child_id,
        arke_link.metadata
      FROM
        arke_link JOIN tree
        ON arke_link.child_id = tree.parent_id
      WHERE
       depth < ?
    )
    SELECT * FROM tree ORDER BY depth
  )
  """

  #  def get_nodes(project, :count, unit_id, depth, direction) do
  #      from a in "arke_unit",
  #           left_join: cte in fragment(@raw_cte_child_query, literal(^Atom.to_string(project)), literal(^"parent_id"), ^unit_id, literal(^"parent_id"), literal(^"child_id"), ^depth),
  #           where: a.id == cte.child_id,
  #           select: count("*")
  #  end
  def get_nodes(project, action, unit_id, depth, direction, type) do
    project = get_project(project)
    {link_field, tree_field} = get_fields_by_direction(direction)
    where_field = get_where_field_by_direction(direction) |> get_where_condition_by_type(type)
    get_link_query(action, project, unit_id, link_field, tree_field, depth, where_field)
  end

  defp get_project(project) when is_atom(project), do: Atom.to_string(project)
  defp get_project(project), do: project
  defp get_fields_by_direction(:child), do: {"parent_id", "child_id"}
  defp get_fields_by_direction(:parent), do: {"child_id", "parent_id"}

  defp get_where_field_by_direction(:child),
    do: dynamic([a, cte], a.id == fragment("?", field(cte, ^:child_id)))

  defp get_where_field_by_direction(:parent),
    do: dynamic([a, cte], a.id == fragment("?", field(cte, ^:parent_id)))

  defp get_where_condition_by_type(condition, nil), do: condition

  defp get_where_condition_by_type(condition, type),
    do: dynamic([a, cte], ^condition and cte.type == ^type)

  defp get_where_field_by_direction(:parent),
    do: dynamic([a, cte], a.id == fragment("?", field(cte, ^:parent_id)))

  defp get_link_query(:count, project, unit_id, link_field, tree_field, depth, where_field) do
    from(a in "arke_unit",
      left_join:
        cte in fragment(
          @raw_cte_child_query,
          literal(^project),
          literal(^link_field),
          ^unit_id,
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^link_field),
          literal(^tree_field),
          ^depth
        ),
      where: ^where_field,
      select: count("*")
    )
  end

  defp get_link_query(_action, project, unit_id, link_field, tree_field, depth, where_field) do
    from(a in "arke_unit",
      left_join:
        cte in fragment(
          @raw_cte_child_query,
          literal(^project),
          literal(^link_field),
          ^unit_id,
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^project),
          literal(^link_field),
          literal(^tree_field),
          ^depth
        ),
      where: ^where_field,
      select: %{
        id: a.id,
        arke_id: a.arke_id,
        data: a.data,
        metadata: a.metadata,
        inserted_at: a.inserted_at,
        updated_at: a.updated_at,
        depth: cte.depth,
        link_metadata: cte.metadata,
        link_type: cte.type
      }
    )
  end

  #  defp get_select_by_action(_action), do: dynamic([a, cte], merge(map(a, [:id, :arke_id, :data, :metadata, :inserted_at, :updated_at]), map(cte, [:depth, :metadata, :type])))
end