defmodule Electric.Postgres.Inspector do alias Electric.Replication.Eval.Parser import Electric, only: :macros @type relation :: Electric.relation() @type relation_id :: Electric.relation_id() @type relation_kind :: :ordinary_table | :partitioned_table @type type_kind :: :base | :composite | :domain | :enum | :pseudo | :range | :multirange @type column_info :: %{ name: String.t(), type: String.t(), type_mod: integer() | nil, type_kind: type_kind(), formatted_type: String.t(), pk_position: non_neg_integer() | nil, type_id: {typid :: non_neg_integer(), typmod :: integer()}, array_dimensions: non_neg_integer(), not_null: boolean(), array_type: String.t() } @type relation_info :: %{ relation_id: relation_id(), relation: relation(), kind: relation_kind(), parent: nil | relation(), children: nil | [relation(), ...] } @type supported_features :: %{ supports_generated_column_replication: boolean() } @callback load_relation_oid(relation(), opts :: term()) :: {:ok, Electric.oid_relation()} | :table_not_found | {:error, String.t() | :connection_not_available} @callback load_relation_info(relation_id(), opts :: term()) :: {:ok, relation_info()} | :table_not_found | {:error, String.t() | :connection_not_available} @callback load_column_info(relation_id(), opts :: term()) :: {:ok, [column_info()]} | :table_not_found | {:error, String.t() | :connection_not_available} @callback load_supported_features(opts :: term()) :: {:ok, supported_features()} | {:error, String.t() | :connection_not_available} @callback clean(relation_id(), opts :: term()) :: :ok @callback list_relations_with_stale_cache(opts :: term()) :: {:ok, [Electric.oid_relation()]} | :error @type inspector :: {module(), opts :: term()} def for_stack(stack_id) do Electric.StackConfig.lookup!(stack_id, :inspector) end @doc """ Expects the table name provided by the user and validates that the table exists, returning the OID. Table name is expected to have been normalized beforehand """ @spec load_relation_oid(relation(), inspector()) :: {:ok, Electric.oid_relation()} | :table_not_found | {:error, String.t() | :connection_not_available} def load_relation_oid(relation, {module, opts}) when is_relation(relation) do module.load_relation_oid(relation, opts) end @doc """ Load additional information about a given relation. Additional information includes the relation kind, parent/child relationships, and other metadata. """ @spec load_relation_info(relation_id(), inspector()) :: {:ok, relation_info()} | :table_not_found | {:error, String.t() | :connection_not_available} def load_relation_info(relation_id, {module, opts}) when is_relation_id(relation_id) do module.load_relation_info(relation_id, opts) end @doc """ Load column information about a given table using a provided inspector. """ @spec load_column_info(relation_id(), inspector()) :: {:ok, [column_info()]} | :table_not_found | {:error, String.t() | :connection_not_available} def load_column_info(relation_id, {module, opts}) when is_relation_id(relation_id) do module.load_column_info(relation_id, opts) end @doc """ Load the supported features on the target database using a provided inspector. """ @spec load_supported_features(inspector()) :: {:ok, supported_features()} | {:error, String.t() | :connection_not_available} def load_supported_features({module, opts}) do module.load_supported_features(opts) end @doc """ Clean up all information about a given relation using a provided inspector. """ @spec clean(relation_id(), inspector()) :: :ok def clean(relation_id, {module, opts}) when is_relation_id(relation_id) do module.clean(relation_id, opts) end @doc """ Get columns that should be considered a PK for table. If the table has no PK, then we're considering all columns as identifying. """ @spec get_pk_cols([column_info(), ...]) :: [String.t(), ...] def get_pk_cols([_ | _] = columns) do columns |> Enum.reject(&is_nil(&1.pk_position)) |> Enum.sort_by(& &1.pk_position) |> Enum.map(& &1.name) |> case do [] -> Enum.map(columns, & &1.name) results -> results end end @doc """ Convert a column list into something that can be used by `Electric.Replication.Eval.Parser.parse_and_validate_expression/2` """ @spec columns_to_expr([column_info(), ...]) :: Parser.refs_map() def columns_to_expr(columns) when is_list(columns) do Map.new(columns, fn %{name: name, array_type: arr_type} when not is_nil(arr_type) -> {[name], {:array, as_atom(arr_type)}} %{name: name, type_kind: :enum, type: type_name} -> {[name], {:enum, type_name}} %{name: name, type: type} -> {[name], as_atom(type)} end) end @doc """ List relations that have stale cache. Doesn't clean the cache immediately, that's left to the caller. Inspectors without cache will return an `:error`. """ @spec list_relations_with_stale_cache(inspector()) :: {:ok, [Electric.oid_relation()]} | :error def list_relations_with_stale_cache({module, opts}) do module.list_relations_with_stale_cache(opts) end defp as_atom(type) when is_binary(type), do: String.to_atom(type) defp as_atom(type) when is_atom(type), do: type end