electric_client

if Code.ensure_loaded?(Ecto) do
  defmodule Electric.Client.EctoAdapter do
    @moduledoc false

    alias Electric.Client.ShapeDefinition

    @behaviour Electric.Client.ValueMapper

    def shape!(schema, opts \\ [])

    def shape!(schema, opts) when is_atom(schema), do: shape_from_query!(schema, opts)
    def shape!(%Ecto.Query{} = query, opts), do: shape_from_query!(query, opts)
    def shape!(%Ecto.Changeset{} = changeset, opts), do: shape_from_changeset!(changeset, opts)

    def shape!(changeset_fun, opts) when is_function(changeset_fun, 1),
      do: shape_from_changeset!(changeset_fun, opts)

    @doc false
    @spec shape_from_query!(Ecto.Queryable.t()) :: ShapeDefinition.t()
    def shape_from_query!(queryable, opts \\ []) do
      query = Ecto.Queryable.to_query(queryable)

      validate_query!(query)

      {table_name, namespace, struct} = table_name(query)
      # it's possible that the ecto schema does not contain all the columns in
      # the table so, since we know the columns we want, let's specify them
      # explicitly
      columns = query_columns(query)
      where = where(query)

      ShapeDefinition.new!(
        table_name,
        merge_shape_opts(
          [namespace: namespace, where: where, columns: columns, parser: {__MODULE__, struct}],
          opts
        )
      )
    end

    @shape_from_changeset_opts_schema ShapeDefinition.schema_definition()
                                      |> Keyword.drop([:columns])
                                      |> NimbleOptions.new!()

    @type shape_from_changeset_opts() :: [
            unquote(NimbleOptions.option_typespec(@shape_from_changeset_opts_schema))
          ]

    @spec shape_from_changeset!(
            (map() -> Ecto.Changeset.t()) | Ecto.Changeset.t(),
            shape_from_changeset_opts()
          ) :: ShapeDefinition.t()
    def shape_from_changeset!(changeset_or_fun, opts \\ [])

    def shape_from_changeset!(%Ecto.Changeset{} = changeset, opts) do
      generate_shape_from_changeset(changeset, opts)
    end

    def shape_from_changeset!(changeset_fun, opts) when is_function(changeset_fun, 1) do
      case changeset_fun.(%{}) do
        %Ecto.Changeset{} = changeset ->
          generate_shape_from_changeset(changeset, opts)

        invalid ->
          raise ArgumentError,
            message:
              "Changeset function returned #{inspect(invalid)}, was expecting an Ecto.Changeset struct"
      end
    end

    defp generate_shape_from_changeset(%Ecto.Changeset{data: %schema{}} = changeset, opts) do
      if !function_exported?(schema, :__schema__, 1),
        do:
          raise(ArgumentError,
            message:
              "cannot generate a shape from a schema-less changeset. Use #{inspect(ShapeDefinition)}.new/2"
          )

      table_name = schema.__schema__(:source)
      namespace = schema.__schema__(:prefix)
      pks = schema.__schema__(:primary_key)

      columns =
        [pks, changeset.required, Keyword.keys(changeset.validations)]
        |> Enum.concat()
        |> Enum.uniq()
        |> Enum.map(&schema.__schema__(:field_source, &1))
        |> Enum.map(&to_string/1)

      ShapeDefinition.new!(
        table_name,
        merge_shape_opts(
          [columns: columns, parser: {__MODULE__, schema}, namespace: namespace],
          opts
        )
      )
    end

    defp merge_shape_opts(base, overrides) do
      Keyword.merge(base, overrides, fn _key, base, over ->
        if is_nil(over), do: base, else: over
      end)
    end

    defp table_name(%{
           prefix: query_prefix,
           from: %{prefix: source_prefix, source: {table_name, struct}}
         }) do
      {table_name, query_prefix || source_prefix, struct}
    end

    defp query_columns(%{from: %{source: {_table_name, struct}}}) do
      Enum.map(
        struct.__schema__(:fields),
        &to_string(struct.__schema__(:field_source, &1))
      )
    end

    @doc false
    def where(%{wheres: []} = _query) do
      nil
    end

    def where(query) do
      %{from: %{source: {table_name, struct}}} = query

      {query, bindings, _key} =
        Ecto.Query.Planner.plan(query, :all, Ecto.Adapters.Postgres)

      {query, _} = Ecto.Query.Planner.normalize(query, :all, Ecto.Adapters.Postgres, 1)

      query
      |> Electric.Client.EctoAdapter.Postgres.where(
        {{quote_table(table_name), quote_table(table_name), struct}, []},
        bindings
      )
      |> case do
        [] -> nil
        iodata -> IO.iodata_to_binary(iodata)
      end
    end

    defp quote_table(table_name) do
      [34, table_name, 34]
    end

    @impl Electric.Client.ValueMapper
    def for_schema(_schema, module) do
      fields =
        Enum.map(module.__schema__(:fields), fn field ->
          {module.__schema__(:field_source, field) |> to_string(), field,
           cast_to(module.__schema__(:type, field))}
        end)

      &map_values(&1, module, fields)
    end

    defp map_values(values, module, fields) do
      struct(
        module,
        Enum.map(fields, fn {column, field, fun} ->
          {field, fun.(Map.get(values, column))}
        end)
      )
    end

    defp cast_to(:boolean) do
      fn
        nil -> nil
        "t" -> true
        "f" -> false
        value -> Ecto.Type.cast!(:boolean, value)
      end
    end

    defp cast_to({:parameterized, {_, _}} = type) do
      fn
        nil ->
          nil

        value ->
          decoded_value =
            case Jason.decode(value) do
              {:ok, decoded} -> decoded
              {:error, _} -> value
            end

          case Ecto.Type.load(type, decoded_value) do
            {:ok, loaded} -> loaded
            :error -> raise Ecto.CastError, type: type, value: value
          end
      end
    end

    defp cast_to(type) do
      fn
        nil -> nil
        value -> Ecto.Type.cast!(type, value)
      end
    end

    @problematic_clauses [
      joins: "JOIN",
      updates: "UPDATE",
      order_bys: "ORDER BY",
      havings: "HAVING",
      group_bys: "GROUP BY",
      distinct: "DISTINCT"
    ]

    for {key, name} <- @problematic_clauses do
      defp validate_query!(%{unquote(key) => [_ | _]}),
        do:
          raise(ArgumentError,
            message: "Electric does not support streaming queries with #{unquote(name)} clauses"
          )
    end

    defp validate_query!(_query), do: :ok
  end
end