phoenix_kit

defmodule PhoenixKit.Modules.Sync.SchemaInspector do
  @moduledoc """
  Inspects database schema for DB Sync module.

  Discovers available tables, their columns, and metadata.
  Uses PostgreSQL information_schema for introspection.

  ## Security Considerations

  - Only returns tables in the public schema by default
  - System tables (pg_*, information_schema) are excluded
  - Admin can configure allowed/blocked tables (future feature)

  ## Example

      iex> SchemaInspector.list_tables()
      {:ok, [
        %{name: "users", estimated_count: 150},
        %{name: "posts", estimated_count: 1200},
        ...
      ]}

      iex> SchemaInspector.get_schema("users")
      {:ok, %{
        table: "users",
        columns: [
          %{name: "id", type: "bigint", nullable: false, primary_key: true},
          %{name: "email", type: "character varying", nullable: false},
          ...
        ],
        primary_key: ["id"]
      }}
  """

  alias PhoenixKit.Modules.Sync.{ColumnInfo, TableSchema}
  alias PhoenixKit.RepoHelper

  # Tables to always exclude from sync
  @excluded_tables [
    # Ecto/Phoenix internal tables
    "schema_migrations",
    # Oban internal tables
    "oban_jobs",
    "oban_peers",
    "oban_producers",
    # Session/token tables (security sensitive)
    "phoenix_kit_user_tokens"
  ]

  # Prefixes for tables to exclude
  @excluded_prefixes [
    "pg_",
    "oban_"
  ]

  @doc """
  Lists all available tables with row counts.

  Returns tables from the public schema, excluding system tables
  and security-sensitive tables.

  ## Options

  - `:include_phoenix_kit` - Include phoenix_kit_* tables (default: true)
  - `:schema` - Database schema to inspect (default: "public")
  - `:exact_counts` - Use exact COUNT(*) instead of pg_stat estimates (default: true)
  """
  @spec list_tables(keyword()) :: {:ok, [map()]} | {:error, any()}
  def list_tables(opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")
    include_phoenix_kit = Keyword.get(opts, :include_phoenix_kit, true)
    exact_counts = Keyword.get(opts, :exact_counts, true)

    # First get the list of tables
    tables_query = """
    SELECT t.tablename as name
    FROM pg_catalog.pg_tables t
    WHERE t.schemaname = $1
    ORDER BY t.tablename
    """

    case RepoHelper.query(tables_query, [schema]) do
      {:ok, %{rows: rows}} ->
        tables =
          rows
          |> Enum.map(fn [name] -> name end)
          |> Enum.reject(fn name -> excluded_table?(name, include_phoenix_kit) end)
          |> Enum.map(fn name ->
            count =
              if exact_counts do
                get_exact_count(name, schema)
              else
                get_estimated_count(name, schema)
              end

            %{name: name, estimated_count: count}
          end)

        {:ok, tables}

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

  defp get_exact_count(table_name, schema) do
    if valid_identifier?(table_name) and valid_identifier?(schema) do
      query = "SELECT COUNT(*) FROM \"#{schema}\".\"#{table_name}\""

      case RepoHelper.query(query, []) do
        {:ok, %{rows: [[count]]}} -> count
        _ -> 0
      end
    else
      0
    end
  end

  defp get_estimated_count(table_name, schema) do
    query = """
    SELECT COALESCE(n_live_tup, 0)
    FROM pg_catalog.pg_stat_user_tables
    WHERE relname = $1 AND schemaname = $2
    """

    case RepoHelper.query(query, [table_name, schema]) do
      {:ok, %{rows: [[count]]}} -> count || 0
      _ -> 0
    end
  end

  @doc """
  Gets the schema (columns, types, constraints) for a specific table.

  ## Returns

  - `{:ok, schema}` - Map with table info and columns
  - `{:error, :not_found}` - Table doesn't exist
  - `{:error, reason}` - Database error
  """
  @spec get_schema(String.t(), keyword()) :: {:ok, map()} | {:error, any()}
  def get_schema(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    # First check if table exists
    exists_query = """
    SELECT EXISTS (
      SELECT FROM pg_catalog.pg_tables
      WHERE schemaname = $1 AND tablename = $2
    )
    """

    case RepoHelper.query(exists_query, [schema, table_name]) do
      {:ok, %{rows: [[true]]}} ->
        fetch_table_schema(table_name, schema)

      {:ok, %{rows: [[false]]}} ->
        {:error, :not_found}

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

  @doc """
  Gets the primary key columns for a table.
  """
  @spec get_primary_key(String.t(), keyword()) :: {:ok, [String.t()]} | {:error, any()}
  def get_primary_key(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    query = """
    SELECT a.attname
    FROM pg_index i
    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
    JOIN pg_class c ON c.oid = i.indrelid
    JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE i.indisprimary
      AND c.relname = $1
      AND n.nspname = $2
    ORDER BY array_position(i.indkey, a.attnum)
    """

    case RepoHelper.query(query, [table_name, schema]) do
      {:ok, %{rows: rows}} ->
        {:ok, Enum.map(rows, fn [col] -> col end)}

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

  @doc """
  Checks if a table exists.
  """
  @spec table_exists?(String.t(), keyword()) :: boolean()
  def table_exists?(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    query = """
    SELECT EXISTS (
      SELECT FROM pg_catalog.pg_tables
      WHERE schemaname = $1 AND tablename = $2
    )
    """

    case RepoHelper.query(query, [schema, table_name]) do
      {:ok, %{rows: [[exists]]}} -> exists
      _ -> false
    end
  end

  @doc """
  Creates a table based on a schema definition from another database.

  Used by DB Sync to create tables that exist on sender but not on receiver.

  ## Parameters

  - `table_name` - Name of the table to create
  - `schema_def` - Schema definition map with columns and primary_key

  ## Example

      schema_def = %{
        "columns" => [
          %{"name" => "id", "type" => "bigint", "nullable" => false, "primary_key" => true},
          %{"name" => "name", "type" => "character varying", "nullable" => true}
        ],
        "primary_key" => ["id"]
      }
      SchemaInspector.create_table("users", schema_def)
  """
  @spec create_table(String.t(), map(), keyword()) :: :ok | {:error, any()}
  def create_table(table_name, schema_def, opts \\ []) do
    db_schema = Keyword.get(opts, :schema, "public")

    if valid_identifier?(table_name) do
      columns = Map.get(schema_def, "columns") || Map.get(schema_def, :columns) || []
      primary_key = Map.get(schema_def, "primary_key") || Map.get(schema_def, :primary_key) || []

      column_defs = Enum.map_join(columns, ",\n  ", &column_to_sql/1)

      pk_constraint =
        if primary_key != [] do
          pk_cols = Enum.join(primary_key, ", ")
          ",\n  PRIMARY KEY (#{pk_cols})"
        else
          ""
        end

      query = """
      CREATE TABLE "#{db_schema}"."#{table_name}" (
        #{column_defs}#{pk_constraint}
      )
      """

      case RepoHelper.query(query, []) do
        {:ok, _} -> :ok
        {:error, reason} -> {:error, reason}
      end
    else
      {:error, :invalid_table_name}
    end
  end

  defp column_to_sql(column) do
    name = column["name"] || column[:name]
    type = map_column_type(column["type"] || column[:type])
    nullable = column["nullable"] || column[:nullable]

    null_constraint = if nullable, do: "", else: " NOT NULL"

    # Handle auto-increment for bigint primary keys
    type_with_serial =
      if (column["primary_key"] || column[:primary_key]) and type in ["bigint", "integer"] do
        if type == "bigint", do: "bigserial", else: "serial"
      else
        type
      end

    "\"#{name}\" #{type_with_serial}#{null_constraint}"
  end

  defp map_column_type("character varying"), do: "varchar(255)"

  defp map_column_type("character varying(" <> rest),
    do: "varchar(#{String.trim_trailing(rest, ")")}"

  defp map_column_type("timestamp without time zone"), do: "timestamp"
  defp map_column_type("timestamp with time zone"), do: "timestamptz"
  defp map_column_type(type), do: type

  @doc """
  Gets the exact count of records in a local table.

  This is used by the receiver to compare local vs sender record counts.
  """
  @spec get_local_count(String.t(), keyword()) :: {:ok, non_neg_integer()} | {:error, any()}
  def get_local_count(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    # Sanitize table name to prevent SQL injection
    if valid_identifier?(table_name) and valid_identifier?(schema) do
      query = "SELECT COUNT(*) FROM \"#{schema}\".\"#{table_name}\""

      case RepoHelper.query(query, []) do
        {:ok, %{rows: [[count]]}} -> {:ok, count}
        {:error, reason} -> {:error, reason}
      end
    else
      {:error, :invalid_identifier}
    end
  end

  @doc """
  Returns a map of table_name => list of referenced table names (FK dependencies).
  Only includes tables in the public schema.
  """
  def get_all_foreign_keys(opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    query = """
    SELECT DISTINCT
      t.relname AS table_name,
      r.relname AS referenced_table
    FROM pg_constraint c
    JOIN pg_class t ON c.conrelid = t.oid
    JOIN pg_class r ON c.confrelid = r.oid
    JOIN pg_namespace n ON t.relnamespace = n.oid
    WHERE c.contype = 'f'
      AND n.nspname = $1
    ORDER BY t.relname, r.relname
    """

    case RepoHelper.query(query, [schema]) do
      {:ok, %{rows: rows}} ->
        # Query uses DISTINCT so no dedup needed — just group by table
        fk_map =
          Enum.group_by(rows, fn [table, _] -> table end, fn [_, ref] -> ref end)

        {:ok, fk_map}

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

  @doc """
  Returns a content checksum for a table (MD5 hash of all rows cast to text).
  Used to detect actual data differences beyond row count.

  Skips checksumming for tables with more than `max_rows` rows (default: 10_000)
  to avoid expensive full-table scans. Returns `{:ok, :too_large}` in that case.
  """
  @checksum_max_rows 10_000

  def get_table_checksum(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")
    max_rows = Keyword.get(opts, :max_rows, @checksum_max_rows)

    if valid_identifier?(table_name) and valid_identifier?(schema) do
      do_table_checksum(schema, table_name, max_rows)
    else
      {:error, :invalid_identifier}
    end
  end

  defp do_table_checksum(schema, table_name, max_rows) do
    count_query = "SELECT COUNT(*) FROM \"#{schema}\".\"#{table_name}\""

    case RepoHelper.query(count_query, []) do
      {:ok, %{rows: [[0]]}} ->
        {:ok, "empty"}

      {:ok, %{rows: [[count]]}} when count > max_rows ->
        {:ok, :too_large}

      {:ok, %{rows: [[_count]]}} ->
        compute_table_md5(schema, table_name)

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

  defp compute_table_md5(schema, table_name) do
    query = """
    SELECT md5(string_agg(t::text, '' ORDER BY t::text))
    FROM "#{schema}"."#{table_name}" AS t
    """

    case RepoHelper.query(query, []) do
      {:ok, %{rows: [[nil]]}} -> {:ok, "empty"}
      {:ok, %{rows: [[checksum]]}} -> {:ok, checksum}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc """
  Returns FK column details for a table: list of %{column, referenced_table, referenced_column}.
  """
  def get_foreign_key_columns(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    query = """
    SELECT
      a.attname AS column_name,
      r.relname AS referenced_table,
      a2.attname AS referenced_column
    FROM pg_constraint c
    JOIN pg_class t ON c.conrelid = t.oid
    JOIN pg_namespace n ON t.relnamespace = n.oid
    JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY(c.conkey)
    JOIN pg_class r ON c.confrelid = r.oid
    JOIN pg_attribute a2 ON a2.attrelid = r.oid AND a2.attnum = ANY(c.confkey)
    WHERE c.contype = 'f'
      AND n.nspname = $1
      AND t.relname = $2
    """

    case RepoHelper.query(query, [schema, table_name]) do
      {:ok, %{rows: rows}} ->
        fks =
          Enum.map(rows, fn [col, ref_table, ref_col] ->
            %{column: col, referenced_table: ref_table, referenced_column: ref_col}
          end)

        {:ok, fks}

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

  @doc """
  Returns unique constraint columns for a table (excluding primary key).
  Used to match records by unique fields (e.g., match users by email).
  Returns list of lists (each inner list is a set of columns forming a unique constraint).
  """
  def get_unique_columns(table_name, opts \\ []) do
    schema = Keyword.get(opts, :schema, "public")

    query = """
    SELECT
      i.indexrelid,
      a.attname
    FROM pg_index i
    JOIN pg_class t ON i.indrelid = t.oid
    JOIN pg_namespace n ON t.relnamespace = n.oid
    JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY(i.indkey)
    WHERE i.indisunique = true
      AND NOT i.indisprimary
      AND n.nspname = $1
      AND t.relname = $2
    ORDER BY i.indexrelid, a.attnum
    """

    case RepoHelper.query(query, [schema, table_name]) do
      {:ok, %{rows: rows}} ->
        # Group columns by index
        unique_sets =
          rows
          |> Enum.group_by(fn [idx_oid, _col] -> idx_oid end, fn [_idx_oid, col] -> col end)
          |> Map.values()

        {:ok, unique_sets}

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

  defp valid_identifier?(name) when is_binary(name) do
    # Only allow alphanumeric and underscores, must start with letter or underscore
    Regex.match?(~r/^[a-zA-Z_][a-zA-Z0-9_]*$/, name)
  end

  defp valid_identifier?(_), do: false

  # ===========================================
  # PRIVATE FUNCTIONS
  # ===========================================

  defp fetch_table_schema(table_name, schema) do
    columns_query = """
    SELECT
      c.column_name,
      c.data_type,
      c.is_nullable = 'YES' as nullable,
      c.column_default,
      c.character_maximum_length,
      c.numeric_precision,
      c.numeric_scale
    FROM information_schema.columns c
    WHERE c.table_schema = $1
      AND c.table_name = $2
    ORDER BY c.ordinal_position
    """

    with {:ok, %{rows: column_rows}} <- RepoHelper.query(columns_query, [schema, table_name]),
         {:ok, primary_key} <- get_primary_key(table_name, schema: schema) do
      columns =
        Enum.map(column_rows, fn [name, type, nullable, default, max_len, precision, scale] ->
          %ColumnInfo{
            name: name,
            type: type,
            nullable: nullable,
            primary_key: name in primary_key,
            default: default,
            max_length: max_len,
            precision: precision,
            scale: scale
          }
        end)

      {:ok,
       %TableSchema{
         table: table_name,
         schema: schema,
         columns: columns,
         primary_key: primary_key
       }}
    end
  end

  defp excluded_table?(name, include_phoenix_kit) do
    cond do
      name in @excluded_tables ->
        true

      Enum.any?(@excluded_prefixes, &String.starts_with?(name, &1)) ->
        true

      not include_phoenix_kit and String.starts_with?(name, "phoenix_kit_") ->
        true

      true ->
        false
    end
  end
end