exsql

defmodule ExSQL.Table do
  @moduledoc """
  In-memory table storage.

  Rows live in a map keyed by rowid, mirroring SQLite's model where every
  table is a B-tree keyed by a 64-bit rowid. A single `INTEGER PRIMARY KEY`
  column is detected as the rowid alias, with SQLite's semantics: inserting
  NULL into it auto-assigns the next rowid, and the column's value *is* the
  row's key.

  All functions are pure — they return an updated table or an error tuple.
  Constraint enforcement (NOT NULL, UNIQUE, PRIMARY KEY) happens here, by
  scanning; real indexes can replace the scans later without changing the
  interface.
  """

  alias ExSQL.AST.ColumnDef
  alias ExSQL.Value

  defstruct name: nil,
            schema: nil,
            columns: [],
            rows: %{},
            next_rowid: 1,
            rowid_alias: nil,
            autoincrement: false,
            sequence: 0,
            sequence_row: false,
            without_rowid: false,
            strict: false,
            indexes: [],
            autoindexes: [],
            composite_keys: [],
            composite_uniques: [],
            foreign_keys: [],
            checks: [],
            # Cached `{key, name, affinity, collate}` tuples for the executor's
            # frame template, so building a frame per query skips recomputing
            # `key/1` for every column. `nil` means "not cached" — recompute;
            # column mutations reset it to `nil`. See `frame_columns/1`.
            frame_columns: nil

  @type index :: %{name: String.t(), columns: [String.t()], unique: boolean()}
  @type row :: %{String.t() => Value.t()}

  @typedoc "A composite PK or UNIQUE constraint: {constraint_name | nil, [column_key]}."
  @type composite_constraint :: {String.t() | nil, [String.t()]}

  @typedoc "A CHECK constraint: {constraint_name | nil, expr}."
  @type check_constraint :: {String.t() | nil, term()}

  @typedoc """
  A table-level FK: {child_keys, parent_table, parent_keys, actions}, where
  actions is `%{on_delete: action, on_update: action, deferred: boolean}`.
  """
  @type foreign_key ::
          {[String.t()], String.t(), [String.t()], ExSQL.AST.CreateTable.fk_actions()}

  @type t :: %__MODULE__{
          name: String.t(),
          schema: String.t() | nil,
          columns: [ColumnDef.t()],
          rows: %{integer() => row()},
          next_rowid: pos_integer(),
          rowid_alias: String.t() | nil,
          autoincrement: boolean(),
          sequence: integer(),
          sequence_row: boolean(),
          without_rowid: boolean(),
          strict: boolean(),
          indexes: [index()],
          autoindexes: [index()],
          composite_keys: [composite_constraint()],
          composite_uniques: [composite_constraint()],
          foreign_keys: [foreign_key()],
          checks: [check_constraint()]
        }

  @doc "Creates a table from a name, parsed column definitions, and optional table-level constraints."
  @spec new(String.t(), [ColumnDef.t()], keyword()) :: t()
  def new(name, columns, opts \\ []) do
    composite_keys = Keyword.get(opts, :composite_keys, [])
    composite_uniques = Keyword.get(opts, :composite_uniques, [])
    foreign_keys = Keyword.get(opts, :foreign_keys, [])
    checks = Keyword.get(opts, :checks, [])
    without_rowid = Keyword.get(opts, :without_rowid, false)
    strict = Keyword.get(opts, :strict, false)
    schema = Keyword.get(opts, :schema)
    columns = if strict, do: strict_columns(columns), else: columns

    # A table-level PRIMARY KEY(a) over a single INTEGER column → rowid alias,
    # same as inline INTEGER PRIMARY KEY.
    rowid_alias =
      if without_rowid, do: nil, else: find_rowid_alias(columns, composite_keys)

    autoincrement =
      rowid_alias != nil and
        Enum.any?(columns, &(key(&1.name) == rowid_alias and &1.autoincrement))

    %__MODULE__{
      name: name,
      schema: schema,
      columns: columns,
      rowid_alias: rowid_alias,
      autoincrement: autoincrement,
      without_rowid: without_rowid,
      strict: strict,
      composite_keys: composite_keys,
      composite_uniques: composite_uniques,
      foreign_keys: foreign_keys,
      checks: checks,
      frame_columns: frame_columns_for(columns)
    }
  end

  @doc """
  The `{key, name, affinity, collate}` tuples used to build a query frame.
  Returns the cached value, recomputing if the cache was invalidated by a
  column mutation.
  """
  @spec frame_columns(t()) :: [{String.t(), String.t(), atom(), String.t() | nil}]
  def frame_columns(%__MODULE__{frame_columns: nil} = table), do: frame_columns_for(table.columns)
  def frame_columns(%__MODULE__{frame_columns: cached}), do: cached

  defp frame_columns_for(columns) do
    Enum.map(columns, &{key(&1.name), &1.name, &1.affinity, &1.collate})
  end

  defp strict_columns(columns) do
    Enum.map(columns, fn column ->
      if is_binary(column.declared_type) and String.upcase(column.declared_type) == "ANY" do
        %{column | affinity: :any}
      else
        column
      end
    end)
  end

  # Look for rowid alias: inline INTEGER PRIMARY KEY, or a table-level
  # PRIMARY KEY(a) over a single column that has INTEGER affinity.
  defp find_rowid_alias(columns, composite_keys) do
    inline =
      Enum.find_value(columns, fn column ->
        if column.primary_key and column.affinity == :integer, do: key(column.name)
      end)

    if inline do
      inline
    else
      case composite_keys do
        [{_name, [col_key]}] ->
          col = Enum.find(columns, &(key(&1.name) == col_key))
          if col && col.affinity == :integer, do: col_key

        _ ->
          nil
      end
    end
  end

  @doc "Case-insensitive column lookup key, since SQL identifiers fold case."
  @spec key(String.t()) :: String.t()
  def key(name) when is_binary(name) do
    # SQLite folds identifier case for ASCII only (a-z), not Unicode. So an
    # ASCII fold is both correct and far cheaper than `String.downcase/1`
    # (Unicode-aware). An already-lowercase name (the common case) is returned
    # unchanged with no allocation.
    if any_upper_ascii?(name), do: ascii_downcase(name), else: name
  end

  defp any_upper_ascii?(<<c, _rest::binary>>) when c >= ?A and c <= ?Z, do: true
  defp any_upper_ascii?(<<_c, rest::binary>>), do: any_upper_ascii?(rest)
  defp any_upper_ascii?(<<>>), do: false

  defp ascii_downcase(name), do: for(<<c <- name>>, into: <<>>, do: <<downcase_byte(c)>>)

  defp downcase_byte(c) when c >= ?A and c <= ?Z, do: c + 32
  defp downcase_byte(c), do: c

  @doc "Returns the ColumnDef for `name`, or `nil`."
  @spec column(t(), String.t()) :: ColumnDef.t() | nil
  def column(table, name) do
    lowered = key(name)
    Enum.find(table.columns, &(key(&1.name) == lowered))
  end

  @doc "All rows in rowid order, as `{rowid, row}` pairs."
  @spec scan(t()) :: [{integer(), row()}]
  def scan(table) do
    rows = table.rows
    n = map_size(rows)

    # Rows live in an unordered map but a scan must yield them in rowid order.
    # Tables loaded by sequential inserts have dense rowids `1..n`, so build
    # them in order directly (O(n)) and only pay the O(n log n) sort when there
    # is a gap (deletes / explicit rowids). The dense build aborts to the sort
    # the moment a rowid is missing, so the result is always correct.
    if n == 0, do: [], else: dense_scan(rows, n, [])
  end

  defp dense_scan(_rows, 0, acc), do: acc

  defp dense_scan(rows, i, acc) do
    case rows do
      %{^i => row} -> dense_scan(rows, i - 1, [{i, row} | acc])
      _ -> Enum.sort_by(Map.to_list(rows), &elem(&1, 0))
    end
  end

  @doc """
  Inserts a row given as a map of column key => value (already evaluated and
  affinity-coerced by the executor). Missing columns get their DEFAULT or NULL.

  Options:

    * `:rowid` — an explicit rowid (from inserting into `rowid` by name)
    * `:on_conflict` — `:abort` (default) errors, `:replace` deletes the
      conflicting rows first, `:ignore` skips the insert and returns `:ignore`
  """
  @spec insert(t(), row(), keyword()) :: {:ok, t(), integer()} | :ignore | {:error, String.t()}
  def insert(table, values, opts \\ []) do
    on_conflict = Keyword.get(opts, :on_conflict) || :abort
    explicit_rowid = Keyword.get(opts, :rowid)

    row =
      table.columns
      |> Map.new(fn column ->
        column_key = key(column.name)
        {column_key, Map.get(values, column_key, :missing)}
      end)
      |> then(&resolve_missing(table, &1))

    with {:ok, rowid, row} <- assign_rowid(table, row, explicit_rowid) do
      case violations(table, row, rowid) do
        :ok ->
          {:ok, store(table, rowid, row), rowid}

        {:conflict, _rowids, _message} when on_conflict == :ignore ->
          :ignore

        {:conflict, rowids, _message} when on_conflict == :replace ->
          {:ok, table |> delete_rows(rowids) |> store(rowid, row), rowid}

        {:conflict, _rowids, message} ->
          {:error, message}

        {:error, _message} when on_conflict == :ignore ->
          :ignore

        {:error, message} ->
          {:error, message}
      end
    end
  end

  defp store(table, rowid, row) do
    rows = Map.put(table.rows, rowid, row)
    sequence = if table.autoincrement, do: max(table.sequence, rowid), else: table.sequence
    sequence_row = table.sequence_row or table.autoincrement

    %{
      table
      | rows: rows,
        next_rowid: max(table.next_rowid, rowid + 1),
        sequence: sequence,
        sequence_row: sequence_row
    }
  end

  defp assign_rowid(table, row, explicit) when is_integer(explicit) do
    case table.rowid_alias do
      nil -> {:ok, explicit, row}
      alias_key -> {:ok, explicit, Map.put(row, alias_key, explicit)}
    end
  end

  defp assign_rowid(table, row, nil) do
    case table.rowid_alias do
      nil ->
        {:ok, table.next_rowid, row}

      alias_key ->
        case Map.fetch!(row, alias_key) do
          nil ->
            rowid =
              cond do
                table.autoincrement and table.sequence_row ->
                  max(table.sequence + 1, table.next_rowid)

                table.autoincrement ->
                  next_available_rowid(table)

                true ->
                  table.next_rowid
              end

            {:ok, rowid, Map.put(row, alias_key, rowid)}

          value when is_integer(value) ->
            {:ok, value, row}

          _other ->
            {:error, "datatype mismatch"}
        end
    end
  end

  defp assign_rowid(_table, _row, _explicit), do: {:error, "datatype mismatch"}

  defp next_available_rowid(table) do
    table.rows
    |> Map.keys()
    |> Enum.max(fn -> 0 end)
    |> Kernel.+(1)
  end

  # Checks the row for constraint problems. NOT NULL is a hard error;
  # rowid/PRIMARY KEY/UNIQUE collisions report the conflicting rowids so
  # OR REPLACE can delete them.
  defp violations(table, row, rowid) do
    not_null =
      Enum.find(table.columns, &not_null_violation?(table, row, &1)) ||
        composite_primary_key_not_null_violation(table, row)

    if not_null do
      {:error, "NOT NULL constraint failed: #{table.name}.#{not_null.name}"}
    else
      rowid_conflicts =
        if Map.has_key?(table.rows, rowid) do
          [{rowid, "UNIQUE constraint failed: #{table.name}.#{table.rowid_alias || "rowid"}"}]
        else
          []
        end

      # Single-column UNIQUE/PK on non-rowid columns
      single_unique_conflicts =
        for column <- table.columns,
            needs_uniqueness?(table, column),
            column_key = key(column.name),
            value = Map.fetch!(row, column_key),
            not is_nil(value),
            conflicting <- duplicates(table, column_key, value, rowid) do
          {conflicting, "UNIQUE constraint failed: #{table.name}.#{column.name}"}
        end

      # Composite PRIMARY KEY conflicts. A single-column PK that became the
      # rowid alias is already handled via the rowid itself.
      composite_pk_conflicts =
        for {_cname, col_keys} <- table.composite_keys,
            col_keys != [table.rowid_alias],
            values = Enum.map(col_keys, &Map.fetch!(row, &1)),
            Enum.all?(values, &(not is_nil(&1))),
            conflicting <- composite_duplicates(table, col_keys, values, rowid) do
          {conflicting, "UNIQUE constraint failed: #{column_list(table, col_keys)}"}
        end

      composite_unique_conflicts =
        for {_cname, col_keys} <- table.composite_uniques,
            values = Enum.map(col_keys, &Map.fetch!(row, &1)),
            Enum.all?(values, &(not is_nil(&1))),
            conflicting <- composite_duplicates(table, col_keys, values, rowid) do
          {conflicting, "UNIQUE constraint failed: #{column_list(table, col_keys)}"}
        end

      all_conflicts =
        rowid_conflicts ++
          single_unique_conflicts ++
          composite_pk_conflicts ++
          composite_unique_conflicts

      case all_conflicts do
        [] -> :ok
        conflicts -> {:conflict, Enum.map(conflicts, &elem(&1, 0)), elem(hd(conflicts), 1)}
      end
    end
  end

  defp not_null_violation?(table, row, column) do
    column_key = key(column.name)

    is_nil(Map.fetch!(row, column_key)) and
      (column.not_null or
         (primary_key_columns_are_not_null?(table) and column.primary_key and
            column_key != table.rowid_alias))
  end

  defp composite_primary_key_not_null_violation(%{strict: false, without_rowid: false}, _row),
    do: nil

  defp composite_primary_key_not_null_violation(table, row) do
    table.composite_keys
    |> Enum.flat_map(fn {_name, col_keys} -> col_keys end)
    |> Enum.find_value(fn col_key ->
      if is_nil(Map.fetch!(row, col_key)) and col_key != table.rowid_alias do
        column(table, col_key)
      end
    end)
  end

  defp primary_key_columns_are_not_null?(table), do: table.strict or table.without_rowid

  # "t.a, t.b" as it appears in UNIQUE constraint failure messages.
  defp column_list(table, col_keys),
    do: Enum.map_join(col_keys, ", ", &"#{table.name}.#{display_column_name(table, &1)}")

  # Returns the display name (original case) for a column key
  defp display_column_name(table, col_key) do
    case Enum.find(table.columns, &(key(&1.name) == col_key)) do
      nil -> col_key
      col -> col.name
    end
  end

  defp resolve_missing(table, row) do
    Map.new(row, fn
      {column_key, :missing} ->
        column = column(table, column_key)
        {column_key, default_value(column)}

      pair ->
        pair
    end)
  end

  defp default_value(%ColumnDef{default: nil}), do: nil

  defp default_value(%ColumnDef{default: {:literal, value}, affinity: affinity}),
    do: Value.apply_affinity(value, affinity)

  defp default_value(%ColumnDef{default: {:negate, {:literal, value}}, affinity: affinity})
       when is_number(value),
       do: Value.apply_affinity(-value, affinity)

  # A bare word default (`f3 text default hi`) is the word as a string.
  defp default_value(%ColumnDef{default: {:column, nil, word}, affinity: affinity}),
    do: Value.apply_affinity(word, affinity)

  defp default_value(_column), do: nil

  # The rowid alias is unique by construction (it is the map key).
  defp needs_uniqueness?(table, column),
    do: (column.primary_key or column.unique) and key(column.name) != table.rowid_alias

  defp duplicates(table, column_key, value, excluding_rowid) do
    for {rowid, row} <- table.rows,
        rowid != excluding_rowid,
        Value.compare(Map.fetch!(row, column_key), value) == :eq,
        do: rowid
  end

  defp composite_duplicates(table, col_keys, values, excluding_rowid) do
    for {rowid, row} <- table.rows,
        rowid != excluding_rowid,
        Enum.zip(col_keys, values)
        |> Enum.all?(fn {k, v} -> Value.compare(Map.fetch!(row, k), v) == :eq end) do
      rowid
    end
  end

  @doc """
  Replaces the row at `rowid` with `row` (a full row map), re-checking
  constraints. If the rowid-alias column changed, the row is re-keyed.
  Takes the same `:on_conflict` option as `insert/3`.
  """
  @spec update_row(t(), integer(), row(), keyword()) ::
          {:ok, t()} | :ignore | {:error, String.t()}
  def update_row(table, rowid, row, opts \\ []) do
    on_conflict = Keyword.get(opts, :on_conflict) || :abort

    rowid_result =
      case Keyword.fetch(opts, :rowid) do
        {:ok, explicit_rowid} when is_integer(explicit_rowid) ->
          row =
            case table.rowid_alias do
              nil -> row
              alias_key -> Map.put(row, alias_key, explicit_rowid)
            end

          {:ok, explicit_rowid, row}

        {:ok, _explicit_rowid} ->
          :error

        :error ->
          case table.rowid_alias do
            nil -> {:ok, rowid, row}
            alias_key -> {:ok, Map.fetch!(row, alias_key), row}
          end
      end

    if match?(:error, rowid_result) do
      {:error, "datatype mismatch"}
    else
      {:ok, new_rowid, row} = rowid_result

      if table.rowid_alias != nil and not is_integer(new_rowid) do
        {:error, "datatype mismatch"}
      else
        # The row being updated does not conflict with itself.
        shadow = delete_rows(table, [rowid])

        case violations(shadow, row, new_rowid) do
          :ok ->
            {:ok, store(shadow, new_rowid, row)}

          {:conflict, _rowids, _message} when on_conflict == :ignore ->
            :ignore

          {:conflict, rowids, _message} when on_conflict == :replace ->
            {:ok, shadow |> delete_rows(rowids) |> store(new_rowid, row)}

          {:conflict, _rowids, message} ->
            {:error, message}

          {:error, _message} when on_conflict == :ignore ->
            :ignore

          {:error, message} ->
            {:error, message}
        end
      end
    end
  end

  @doc "Deletes the rows with the given rowids."
  @spec delete_rows(t(), [integer()]) :: t()
  def delete_rows(table, rowids), do: %{table | rows: Map.drop(table.rows, rowids)}
end