if Code.ensure_loaded?(Ecto.Adapters.SQL) do
  defmodule Ecto.Adapters.QuackDB.AppendInsert do
    @moduledoc false

    import Ecto.Query, only: [from: 2]

    @connection Ecto.Adapters.QuackDB.Connection

    def run(adapter_meta, schema_meta, header, rows, on_conflict, returning, placeholders, opts) do
      with :ok <- assert_supported!(schema_meta, rows, on_conflict, returning, placeholders, opts),
           conn <- ecto_connection(adapter_meta),
           insert_rows <- append_rows(header, rows),
           options <- append_options(schema_meta, header, opts),
           {:ok, %QuackDB.Result{} = result} <-
             insert_all(conn, schema_meta, header, insert_rows, returning, options) do
        {result.num_rows, result.rows}
      else
        {:error, %QuackDB.Error{} = error} -> raise error
      end
    end

    defp insert_all(conn, schema_meta, header, rows, returning, options) do
      if insert_select?(schema_meta, header, returning) do
        insert_select(conn, schema_meta, header, rows, returning, options)
      else
        QuackDB.insert_rows(conn, schema_meta.source, rows, options)
      end
    end

    defp insert_select?(_schema_meta, _header, returning) when returning != [], do: true
    defp insert_select?(%{schema: nil}, _header, _returning), do: false

    defp insert_select?(%{schema: schema}, header, _returning) do
      schema_sources =
        schema.__schema__(:fields) |> Enum.map(&schema.__schema__(:field_source, &1))

      MapSet.new(header) != MapSet.new(schema_sources)
    end

    defp insert_select(conn, schema_meta, header, rows, returning, options) do
      case DBConnection.status(conn, options) do
        :idle ->
          DBConnection.transaction(
            conn,
            fn tx ->
              case do_insert_select(tx, schema_meta, header, rows, returning, options) do
                {:ok, result} -> result
                {:error, error} -> DBConnection.rollback(tx, error)
              end
            end,
            options
          )

        _status ->
          do_insert_select(conn, schema_meta, header, rows, returning, options)
      end
    end

    defp do_insert_select(conn, schema_meta, header, rows, returning, options) do
      columns = append_columns!(schema_meta, header, options)
      temp_table = temp_table_name()
      create_statement = create_temp_table(temp_table, columns)
      insert_statement = insert_from_temp_statement(schema_meta, columns, temp_table, returning)

      try do
        with {:ok, _result} <- QuackDB.query(conn, create_statement, [], options),
             {:ok, _result} <- QuackDB.insert_rows(conn, temp_table, rows, options),
             {:ok, %QuackDB.Result{} = result} <-
               QuackDB.query(conn, insert_statement, [], options) do
          {:ok, result}
        end
      after
        _ = QuackDB.query(conn, QuackDB.DDL.drop_table(temp_table, if_exists: true), [], options)
      end
    end

    defp temp_table_name do
      "quackdb_append_#{System.unique_integer([:positive])}"
    end

    defp append_columns!(schema_meta, header, options) do
      types = Keyword.fetch!(options, :columns)
      sources = schema_sources(schema_meta)

      Enum.map(header, fn source ->
        %{
          source: source,
          query_field: temp_query_field!(source, sources),
          type: column_type!(types, source)
        }
      end)
    end

    defp schema_sources(%{schema: nil}), do: %{}

    defp schema_sources(%{schema: schema}) do
      Map.new(schema.__schema__(:fields), fn field ->
        {schema.__schema__(:field_source, field), field}
      end)
    end

    defp temp_query_field!(source, sources) when is_map_key(sources, source), do: source
    defp temp_query_field!(source, _sources) when is_atom(source), do: source

    defp temp_query_field!(source, _sources) do
      raise ArgumentError,
            "append insert returning requires atom column names for Ecto query generation, got: #{inspect(source)}"
    end

    defp create_temp_table(temp_table, columns) do
      ddl_columns = Enum.map(columns, fn column -> {column.source, column.type} end)
      QuackDB.DDL.create_table(temp_table, ddl_columns, temporary: true)
    end

    defp column_type!(columns, column) do
      Enum.find_value(columns, fn
        {^column, type} ->
          {:ok, type}

        {name, type} when is_atom(name) and is_binary(column) ->
          if Atom.to_string(name) == column, do: {:ok, type}

        {name, type} when is_binary(name) and is_atom(column) ->
          if name == Atom.to_string(column), do: {:ok, type}

        _entry ->
          nil
      end)
      |> case do
        {:ok, type} -> type
        nil -> raise KeyError, key: column, term: columns
      end
    end

    defp insert_from_temp_statement(schema_meta, columns, temp_table, returning) do
      header = Enum.map(columns, & &1.source)

      @connection.insert(
        schema_meta.prefix,
        schema_meta.source,
        header,
        temp_select_query(temp_table, columns),
        {:raise, [], []},
        returning,
        []
      )
    end

    defp temp_select_query(temp_table, columns) do
      fields = Enum.map(columns, & &1.query_field)
      from(row in temp_table, select: ^fields)
    end

    defp assert_supported!(
           _schema_meta,
           {%Ecto.Query{}, _params},
           _on_conflict,
           _returning,
           _placeholders,
           _opts
         ) do
      unsupported!(
        :schema_inserts,
        "insert_method: :append does not support insert_all from queries"
      )
    end

    defp assert_supported!(
           _schema_meta,
           %Ecto.Query{},
           _on_conflict,
           _returning,
           _placeholders,
           _opts
         ) do
      unsupported!(
        :schema_inserts,
        "insert_method: :append does not support insert_all from queries"
      )
    end

    defp assert_supported!(
           _schema_meta,
           _rows,
           {_kind, _params, targets},
           _returning,
           _placeholders,
           _opts
         )
         when targets != [] do
      unsupported!(:schema_inserts, "insert_method: :append does not support conflict targets")
    end

    defp assert_supported!(schema_meta, _rows, {:raise, _params, []}, returning, [], opts)
         when is_list(returning) do
      if returning == [] or schema_meta.schema != nil or Keyword.has_key?(opts, :columns) do
        :ok
      else
        unsupported!(
          :schema_inserts,
          "insert_method: :append with returning requires a schema or explicit append columns"
        )
      end
    end

    defp assert_supported!(_schema_meta, _rows, _on_conflict, _returning, _placeholders, _opts) do
      unsupported!(
        :schema_inserts,
        "insert_method: :append only supports plain insert_all without returning, placeholders, or upserts"
      )
    end

    defp ecto_connection(%{pid: pool} = adapter_meta) do
      case Process.get({Ecto.Adapters.SQL, pool}) do
        :undefined -> ecto_pool(adapter_meta)
        nil -> ecto_pool(adapter_meta)
        conn -> conn
      end
    end

    defp ecto_pool(%{partition_supervisor: {name, _}}),
      do: {:via, PartitionSupervisor, {name, self()}}

    defp ecto_pool(%{pid: pool}), do: pool

    defp append_rows(header, rows) do
      Enum.map(rows, fn row ->
        Enum.map(header, fn field -> {field, Keyword.fetch!(row, field)} end)
      end)
    end

    defp append_options(schema_meta, header, opts) do
      opts =
        opts
        |> base_options()
        |> maybe_put_schema(schema_meta)

      case Keyword.fetch(opts, :columns) do
        {:ok, _columns} -> opts
        :error -> maybe_put_schema_columns(opts, schema_meta, header)
      end
    end

    defp maybe_put_schema(opts, %{prefix: nil}), do: opts
    defp maybe_put_schema(opts, %{prefix: prefix}), do: Keyword.put(opts, :schema, prefix)

    defp maybe_put_schema_columns(opts, %{schema: nil}, _header), do: opts

    defp maybe_put_schema_columns(opts, %{schema: schema}, header) do
      source_types =
        Map.new(schema.__schema__(:fields), fn field ->
          {schema.__schema__(:field_source, field),
           QuackDB.Ecto.Type.column_type!(schema.__schema__(:type, field), :append)}
        end)

      columns = Enum.map(header, fn source -> {source, Map.fetch!(source_types, source)} end)
      Keyword.put(opts, :columns, columns)
    end

    defp base_options(opts) do
      opts
      |> Keyword.take([:timeout])
      |> maybe_put_batch_size(opts)
    end

    defp maybe_put_batch_size(options, opts) do
      case Keyword.fetch(opts, :chunk_every) do
        {:ok, chunk_every} -> Keyword.put(options, :batch_size, chunk_every)
        :error -> options
      end
    end

    defp unsupported!(feature, message) do
      raise QuackDB.Error.new(:ecto_feature_not_supported, message,
              source: :client,
              metadata: %{feature: feature}
            )
    end
  end
end