defmodule QuackDB.Source do
  @moduledoc """
  Safe SQL fragment builders for DuckDB table-producing data sources.

  DuckDB can query files, object stores, and lakehouse tables directly through
  table functions such as `read_parquet/2`, `read_csv/2`, `read_json/2`,
  `read_xlsx/2`, `delta_scan/2`, and `iceberg_scan/2`. This module builds those
  fragments with QuackDB's SQL literal formatting so callers do not need to
  manually interpolate paths or options.

  These helpers return SQL fragments, not executable queries. Use them in raw SQL
  that is sent to `QuackDB.query/4` or `Ecto.Adapters.SQL.query/4`:

      source = QuackDB.Source.parquet("s3://bucket/events/*.parquet", hive_partitioning: true)
      QuackDB.query!(conn, ["SELECT count(*) FROM ", source])

  Options are emitted as DuckDB named parameters:

      QuackDB.Source.csv("events.csv", header: true, columns: %{id: "INTEGER", name: "VARCHAR"})

  Plain Elixir maps are formatted as DuckDB struct literals, which is the shape
  used by options such as `columns`. Use `{:map, map}` when a DuckDB `MAP {...}`
  literal is required.
  """

  alias QuackDB.Error
  alias QuackDB.SQL

  @type path_or_paths :: String.t() | [String.t()]
  @type option_value ::
          SQL.parameter()
          | atom()
          | %{optional(atom() | String.t() | integer()) => option_value()}
          | {:struct, keyword(option_value()) | map()}
          | {:map, map()}

  @doc "Builds a `read_parquet(...)` table function fragment."
  @spec parquet(path_or_paths(), keyword(option_value())) :: String.t()
  def parquet(path_or_paths, options \\ []),
    do: table_function("read_parquet", path_or_paths, options)

  @doc "Builds a `read_csv(...)` table function fragment."
  @spec csv(path_or_paths(), keyword(option_value())) :: String.t()
  def csv(path_or_paths, options \\ []), do: table_function("read_csv", path_or_paths, options)

  @doc "Builds a `read_json(...)` table function fragment."
  @spec json(path_or_paths(), keyword(option_value())) :: String.t()
  def json(path_or_paths, options \\ []), do: table_function("read_json", path_or_paths, options)

  @doc "Builds a `read_xlsx(...)` table function fragment."
  @spec xlsx(String.t(), keyword(option_value())) :: String.t()
  def xlsx(path, options \\ []) when is_binary(path),
    do: table_function("read_xlsx", path, options)

  @doc "Builds a `delta_scan(...)` table function fragment."
  @spec delta(path_or_paths(), keyword(option_value())) :: String.t()
  def delta(path_or_paths, options \\ []),
    do: table_function("delta_scan", path_or_paths, options)

  @doc "Builds an `iceberg_scan(...)` table function fragment."
  @spec iceberg(path_or_paths(), keyword(option_value())) :: String.t()
  def iceberg(path_or_paths, options \\ []),
    do: table_function("iceberg_scan", path_or_paths, options)

  @doc "Builds a `histogram_values(...)` table function fragment."
  @spec histogram_values(String.t(), String.t() | atom(), keyword(option_value())) :: String.t()
  def histogram_values(source, column, options \\ [])
      when is_binary(source) and is_list(options) do
    column = option_name(column)

    ["histogram_values(", source, ", ", column, options(:duckdb_named, options), ")"]
    |> IO.iodata_to_binary()
  end

  @doc "Builds a DuckDB table-function fragment for a validated function name."
  @spec table_function(String.t(), path_or_paths(), keyword(option_value())) :: String.t()
  def table_function(function_name, path_or_paths, options \\ [])
      when is_binary(function_name) and is_list(options) do
    validate_identifier!(function_name, :function)

    [function_name, "(", literal!(path_or_paths), options(:equals, options), ")"]
    |> IO.iodata_to_binary()
  end

  @doc "Wraps a source in a DuckDB `USING SAMPLE` subquery."
  @spec sample(String.t(), keyword()) :: String.t()
  def sample(source, options) when is_binary(source) and is_list(options) do
    ["(SELECT * FROM ", source, " USING SAMPLE ", sample_clause(options), ")"]
    |> IO.iodata_to_binary()
  end

  @doc "Returns true when a value looks like a QuackDB source table-function fragment."
  @spec source?(term()) :: boolean()
  def source?(value) when is_binary(value) do
    String.starts_with?(value, "(SELECT ") or source_table_function?(value)
  end

  def source?(_value), do: false

  defp source_table_function?(value) do
    case table_function_name(value) do
      "read_parquet" -> true
      "read_csv" -> true
      "read_json" -> true
      "read_xlsx" -> true
      "delta_scan" -> true
      "iceberg_scan" -> true
      "histogram_values" -> true
      "generate_series" -> true
      _other -> false
    end
  end

  defp sample_clause(options) do
    cond do
      rows = options[:rows] -> [literal!(rows), " ROWS"]
      percent = options[:percent] -> [literal!(percent), " PERCENT"]
      true -> raise ArgumentError, "expected :rows or :percent sample option"
    end
  end

  defp options(_style, []), do: []

  defp options(style, options) do
    Enum.map(options, fn {name, value} ->
      [", ", option_name(name), option_separator(style), literal!(value)]
    end)
  end

  defp option_separator(:equals), do: " = "
  defp option_separator(:duckdb_named), do: " := "

  defp option_name(name) when is_atom(name), do: name |> Atom.to_string() |> option_name()

  defp option_name(name) when is_binary(name) do
    validate_identifier!(name, :option)
    name
  end

  defp option_name(name) do
    raise ArgumentError,
          "expected DuckDB option name to be an atom or string, got: #{inspect(name)}"
  end

  defp literal!({:struct, values}) when is_list(values) or is_map(values),
    do: struct_literal(values)

  defp literal!({:map, values}) when is_map(values), do: ["MAP ", map_entries(values)]
  defp literal!(values) when is_map(values), do: struct_literal(values)

  defp literal!(value) when is_atom(value) and not is_boolean(value) and not is_nil(value),
    do: literal!(Atom.to_string(value))

  defp literal!(value) do
    case SQL.literal(value) do
      {:ok, literal} -> literal
      {:error, %Error{} = error} -> raise error
    end
  end

  defp struct_literal(values), do: map_entries(values)

  defp map_entries(values) do
    entries =
      values
      |> Enum.map(fn {key, value} -> [literal_key(key), ": ", literal!(value)] end)
      |> Enum.intersperse(", ")

    ["{", entries, "}"]
  end

  defp literal_key(key) when is_atom(key), do: key |> Atom.to_string() |> literal_key()
  defp literal_key(key) when is_binary(key), do: ["'", String.replace(key, "'", "''"), "'"]
  defp literal_key(key) when is_integer(key), do: Integer.to_string(key)

  defp literal_key(key) do
    raise ArgumentError, "unsupported DuckDB literal key: #{inspect(key)}"
  end

  defp table_function_name(value) do
    case :binary.match(value, "(") do
      {index, 1} when index > 0 -> binary_part(value, 0, index)
      _not_found -> nil
    end
  end

  defp validate_identifier!(value, kind) do
    unless QuackDB.Identifier.valid?(value) do
      raise ArgumentError, "invalid DuckDB #{kind} identifier: #{inspect(value)}"
    end

    :ok
  end
end