electric

defmodule Electric.Shapes.Filter do
  @moduledoc """
  Responsible for knowing which shapes are affected by a change.

  `affected_shapes(filter, change)` will return a set of IDs for the shapes that are affected by the change
  considering all the shapes that have been added to the filter using `add_shape/3`.


  The `Filter` module keeps track of what tables are referenced by the shapes and changes and delegates
  the table specific logic to the `Filter.WhereCondition` module.

  Data is stored in ETS tables (outside the process heap) to avoid GC pressure with large numbers of shapes.
  """

  alias Electric.Replication.Changes
  alias Electric.Replication.Changes.DeletedRecord
  alias Electric.Replication.Changes.NewRecord
  alias Electric.Replication.Changes.Relation
  alias Electric.Replication.Changes.TruncatedRelation
  alias Electric.Replication.Changes.UpdatedRecord
  alias Electric.Shapes.Filter
  alias Electric.Shapes.Filter.WhereCondition
  alias Electric.Shapes.Shape
  alias Electric.Telemetry.OpenTelemetry

  require Logger

  defstruct [
    :shapes_table,
    :tables_table,
    :where_cond_table,
    :eq_index_table,
    :incl_index_table,
    :refs_fun
  ]

  @type t :: %Filter{}
  @type shape_id :: any()

  @spec new(keyword()) :: Filter.t()
  def new(opts \\ []) do
    %Filter{
      shapes_table: :ets.new(:filter_shapes, [:set, :private]),
      tables_table: :ets.new(:filter_tables, [:set, :private]),
      where_cond_table: :ets.new(:filter_where, [:set, :private]),
      eq_index_table: :ets.new(:filter_eq, [:set, :private]),
      incl_index_table: :ets.new(:filter_incl, [:set, :private]),
      refs_fun: Keyword.get(opts, :refs_fun, fn _shape -> %{} end)
    }
  end

  @spec has_shape?(t(), shape_id()) :: boolean()
  def has_shape?(%Filter{shapes_table: table}, shape_handle) do
    :ets.member(table, shape_handle)
  end

  @spec active_shapes(t()) :: [shape_id()]
  def active_shapes(%Filter{shapes_table: table}) do
    :ets.select(table, [{{:"$1", :_}, [], [:"$1"]}])
  end

  @doc """
  Add a shape for the filter to track.

  The `shape_id` can be any term you like to identify the shape. Whatever you use will be returned
  by `affected_shapes/2` when the shape is affected by a change.
  """
  @spec add_shape(Filter.t(), shape_id(), Shape.t()) :: Filter.t()
  def add_shape(%Filter{} = filter, shape_id, shape) do
    if has_shape?(filter, shape_id), do: raise("duplicate shape #{shape_id}")

    :ets.insert(filter.shapes_table, {shape_id, shape})

    where_cond_id = get_or_create_table_condition(filter, shape.root_table)

    WhereCondition.add_shape(filter, where_cond_id, shape_id, shape.where)

    filter
  end

  defp get_or_create_table_condition(filter, table_name) do
    case :ets.lookup(filter.tables_table, table_name) do
      [] ->
        where_cond_id = make_ref()
        WhereCondition.init(filter, where_cond_id)
        :ets.insert(filter.tables_table, {table_name, where_cond_id})
        where_cond_id

      [{_, where_cond_id}] ->
        where_cond_id
    end
  end

  @doc """
  Remove a shape from the filter.
  """
  @spec remove_shape(Filter.t(), shape_id()) :: Filter.t()
  def remove_shape(%Filter{} = filter, shape_id) do
    [{_, shape}] = :ets.lookup(filter.shapes_table, shape_id)
    table_name = shape.root_table

    [{_, where_cond_id}] = :ets.lookup(filter.tables_table, table_name)

    case WhereCondition.remove_shape(filter, where_cond_id, shape_id, shape.where) do
      :deleted -> :ets.delete(filter.tables_table, table_name)
      :ok -> :ok
    end

    :ets.delete(filter.shapes_table, shape_id)

    filter
  end

  @doc """
  Returns the shape IDs for all shapes that have been added to the filter
  that are affected by the given change.
  """
  @spec affected_shapes(Filter.t(), Changes.change() | Relation.t()) ::
          MapSet.t(shape_id())
  def affected_shapes(%Filter{} = filter, change) do
    OpenTelemetry.timed_fun("filter.affected_shapes.duration_µs", fn ->
      try do
        shapes_affected_by_change(filter, change)
      catch
        kind, error ->
          Logger.error("""
          Unexpected error in Filter.affected_shapes:
          #{Exception.format(kind, error, __STACKTRACE__)}
          """)

          OpenTelemetry.record_exception(kind, error, __STACKTRACE__)

          # We can't tell which shapes are affected, the safest thing to do is return all shapes
          all_shape_ids(filter)
      end
    end)
  end

  defp shapes_affected_by_change(%Filter{} = filter, %Relation{} = relation) do
    # Check all shapes is all tables because the table may have been renamed
    for shape_id <- all_shape_ids(filter),
        [{_, shape}] = :ets.lookup(filter.shapes_table, shape_id),
        Shape.is_affected_by_relation_change?(shape, relation),
        into: MapSet.new() do
      shape_id
    end
  end

  defp shapes_affected_by_change(%Filter{} = filter, %NewRecord{
         relation: relation,
         record: record
       }) do
    shapes_affected_by_record(filter, relation, record)
  end

  defp shapes_affected_by_change(%Filter{} = filter, %DeletedRecord{
         relation: relation,
         old_record: record
       }) do
    shapes_affected_by_record(filter, relation, record)
  end

  defp shapes_affected_by_change(%Filter{} = filter, %UpdatedRecord{relation: relation} = change) do
    MapSet.union(
      shapes_affected_by_record(filter, relation, change.record),
      shapes_affected_by_record(filter, relation, change.old_record)
    )
  end

  defp shapes_affected_by_change(%Filter{} = filter, %TruncatedRelation{relation: table_name}) do
    shape_ids_for_table(filter, table_name)
  end

  defp shapes_affected_by_record(filter, table_name, record) do
    case :ets.lookup(filter.tables_table, table_name) do
      [] ->
        MapSet.new()

      [{_, where_cond_id}] ->
        WhereCondition.affected_shapes(filter, where_cond_id, record)
    end
  end

  defp all_shape_ids(%Filter{} = filter) do
    :ets.foldl(
      fn {_table_name, where_cond_id}, acc ->
        MapSet.union(acc, WhereCondition.all_shape_ids(filter, where_cond_id))
      end,
      MapSet.new(),
      filter.tables_table
    )
  end

  defp shape_ids_for_table(%Filter{} = filter, table_name) do
    case :ets.lookup(filter.tables_table, table_name) do
      [] -> MapSet.new()
      [{_, where_cond_id}] -> WhereCondition.all_shape_ids(filter, where_cond_id)
    end
  end

  @doc """
  Get a shape by its ID. Used internally for where clause evaluation.
  """
  def get_shape(%Filter{shapes_table: table}, shape_id) do
    case :ets.lookup(table, shape_id) do
      [{_, shape}] -> shape
      [] -> nil
    end
  end
end