recall

defmodule Ecto.Adapters.Recall.Join do
  @moduledoc false
  # Mnesia has no join operator, so multi-source queries are evaluated as a
  # nested-loop join: the `from` side is scanned once, then each `join` expands
  # every partial row by its matching right rows, paired with a `for`/`flat_map`
  # and filtered by the `on`/`where` conditions in Elixir.
  #
  # A query with joins carries one *binding* per source — binding 0 is the
  # `from`, binding `i` is the i-th `join`. Field references in `on`/`where`/
  # `select`/`order_by` name their binding explicitly: `{:&, [], [ix]}`. A
  # *combined row* is a map `%{ix => value_list | nil}` — one value list (the
  # stored record minus its tag, in `Table` storage order) per binding, or `nil`
  # for a binding an outer join left unmatched.
  #
  # ## Tiered right-side fetch
  #
  # Rather than scan every right table in full (O(n×m)), the planner inspects each
  # join's `on` and, when it pins a right field to a value the left side already
  # knows, fetches only the matching rows per left row — cheapest first:
  #
  #   * `{:read, _}`  — `on` pins the right *primary key* (the belongs_to
  #     direction, `p.id == c.post_id`): `:mnesia.read/2`, an O(1) key lookup.
  #   * `{:index, _}` — `on` pins a non-key right field (the has_many direction,
  #     `c.post_id == p.id`): `:mnesia.index_read/3`, after lazily adding a
  #     secondary index to that attribute. O(matches), not O(m).
  #   * `:scan`       — no usable equality (cross/`true`/non-equi `on`), a
  #     `:right` join (must see unmatched right rows), or forced off via
  #     `config :recall, force_join_scan: true`: the full-table loop.
  #
  # Whatever the fetch, the candidates are re-checked against the full `on` in
  # Elixir (`on_match?/6`), so a fetch only has to *narrow* soundly — extra `on`
  # conjuncts and edge cases still get filtered precisely.
  #
  # ## `where` pushdown
  #
  # A scanned binding (always the `from`, plus any `:scan` join) pushes the
  # `where` conjuncts that reference *only* it down into its match spec, so the
  # scan returns fewer rows to pair. This only ever narrows: `filter_wheres/4`
  # still re-checks every `where` afterward, so a conjunct that isn't pushable
  # (references another binding, or isn't a guard) just stays there.

  alias Ecto.Adapters.Recall.Table
  alias Ecto.Adapters.Recall.Transaction

  @doc """
  Runs a multi-source `:all` query and returns the `{count, rows}` tuple Ecto
  expects — the same contract as `Result.project_all/4`.
  """
  def execute(meta, query, params) do
    bindings = bindings(query)

    meta
    |> combined_rows(query, params, bindings)
    |> order(query, bindings)
    |> paginate(offset_value(query, params), limit_value(query, params))
    |> project(query, bindings, params)
  end

  @doc """
  The distinct `from`-binding records (full Mnesia tuples) a joined `update_all`
  /`delete_all` should affect: the join + `where` decide which binding-0 rows
  qualify; the mutation itself only ever touches those rows. A from-row that
  matches several join rows is returned once (deduped by primary key).
  """
  def affected_from_records(meta, query, params) do
    bindings = bindings(query)
    {_, from_schema} = query.from.source
    tag = Table.name(from_schema)

    meta
    |> combined_rows(query, params, bindings)
    |> Enum.map(&Map.fetch!(&1, 0))
    |> Enum.uniq_by(&hd/1)
    |> Enum.map(&List.to_tuple([tag | &1]))
  end

  @doc "True when the query has at least one join (i.e. is multi-source)."
  def joined?(%Ecto.Query{joins: []}), do: false
  def joined?(%Ecto.Query{joins: [_ | _]}), do: true

  # Ensures tables/indices, runs the nested loop in one transaction, applies
  # `where`. Shared by `execute/3` (which then orders/projects) and
  # `affected_from_records/3` (which then extracts binding-0 records).
  defp combined_rows(meta, query, params, bindings) do
    Enum.each(bindings, fn schema -> Table.ensure(schema, meta) end)

    plans = plan_joins(query, bindings)

    fn -> combine(query, bindings, params, plans) end
    |> Transaction.run()
    |> filter_wheres(query, bindings, params)
  end

  # --- bindings ---------------------------------------------------------------

  # The schema per binding, positionally: element `ix` is binding `ix`'s schema —
  # `[from_schema, join1_schema, ...]`. Binding 0 is the `from`; binding `i` is
  # the join whose `ix` is `i`. Sorted by `ix` so list position equals binding
  # index regardless of the order joins appear in `query.joins`.
  defp bindings(query) do
    {_, from_schema} = query.from.source

    query.joins
    |> Enum.map(&{&1.ix, join_schema(&1.source)})
    |> List.keysort(0)
    |> Enum.map(&elem(&1, 1))
    |> then(&[from_schema | &1])
  end

  defp join_schema({_source, schema}) when is_atom(schema) and not is_nil(schema), do: schema

  defp join_schema(other) do
    raise ArgumentError,
          "Ecto.Adapters.Recall supports joins on schema sources only; got: #{inspect(other)}"
  end

  # --- planning: pick a fetch strategy per join -------------------------------

  defp plan_joins(query, bindings) do
    Map.new(query.joins, fn join -> {join.ix, plan_join(join, Enum.at(bindings, join.ix))} end)
  end

  # `:right` must surface unmatched right rows and `:cross` has no equality, so
  # both scan. Everything else looks for an equality conjunct that pins a right
  # field to a left-known value.
  defp plan_join(%{qual: qual}, _schema) when qual in [:right, :cross], do: :scan

  defp plan_join(%{ix: ix, on: on}, schema) do
    if Application.get_env(:recall, :force_join_scan, false) do
      :scan
    else
      on
      |> on_expr()
      |> split_and()
      |> Enum.find_value(:scan, &equi_strategy(&1, ix, schema))
    end
  end

  # `right.field == <left-known value>` (or flipped) → a read/index strategy.
  defp equi_strategy({:==, _, [a, b]}, ix, schema) do
    cond do
      right_field?(a, ix) and not refs_binding?(b, ix) -> strategy_for(field_of(a), b, schema)
      right_field?(b, ix) and not refs_binding?(a, ix) -> strategy_for(field_of(b), a, schema)
      true -> nil
    end
  end

  defp equi_strategy(_other, _ix, _schema), do: nil

  # The primary key is always readable; a non-key field only backs an index read
  # when the table physically has an index on it (created by a migration). An
  # unindexed field yields no strategy, so the join falls back to a scan rather
  # than growing an index off the join shape.
  defp strategy_for(field, value_expr, schema) do
    cond do
      field in primary_key_sources(schema) -> {:read, value_expr}
      Table.indexable?(schema, field) -> {:index, field, value_expr}
      true -> nil
    end
  end

  defp primary_key_sources(schema) do
    Enum.map(schema.__schema__(:primary_key), &Table.field_source(schema, &1))
  end

  # A field reference on join binding `ix`: `{{:., _, [{:&, _, [ix]}, field]}, _, []}`.
  # The repeated `ix` requires the binding index to equal the join's.
  defp right_field?({{:., _, [{:&, _, [ix]}, _field]}, _, []}, ix), do: true
  defp right_field?(_expr, _ix), do: false

  defp field_of({{:., _, [{:&, _, [_ix]}, field]}, _, []}), do: field

  defp split_and({:and, _, [l, r]}), do: split_and(l) ++ split_and(r)
  defp split_and(expr), do: [expr]

  # Does `expr` reference join binding `ix` anywhere in its tree?
  defp refs_binding?({:&, _, [ix]}, ix), do: true
  defp refs_binding?({:&, _, [_]}, _ix), do: false

  defp refs_binding?(tuple, ix) when is_tuple(tuple), do: tuple |> Tuple.to_list() |> Enum.any?(&refs_binding?(&1, ix))

  defp refs_binding?(list, ix) when is_list(list), do: Enum.any?(list, &refs_binding?(&1, ix))
  defp refs_binding?(_leaf, _ix), do: false

  # --- the nested loop --------------------------------------------------------

  defp combine(query, bindings, params, plans) do
    initial = bindings |> hd() |> scan(0, query, params) |> Enum.map(&%{0 => &1})

    Enum.reduce(query.joins, initial, fn join, acc ->
      expand(acc, join, bindings, params, plans[join.ix], query)
    end)
  end

  defp expand(acc, %{ix: ix, qual: :right, on: on}, bindings, params, _strategy, query) do
    rights = scan(Enum.at(bindings, ix), ix, query, params)
    right_join(acc, ix, rights, on_expr(on), bindings, params)
  end

  defp expand(acc, %{ix: ix, qual: qual, on: on}, bindings, params, strategy, query) do
    on_expr = on_expr(on)
    fetch = fetcher(strategy, Enum.at(bindings, ix), ix, query, params)

    Enum.flat_map(acc, fn partial ->
      matches =
        partial
        |> fetch.(bindings, params)
        |> Enum.filter(&on_match?(on_expr, partial, ix, &1, bindings, params))

      case {qual, matches} do
        {:left, []} -> [Map.put(partial, ix, nil)]
        _ -> Enum.map(matches, &Map.put(partial, ix, &1))
      end
    end)
  end

  # A right join keeps every right row: matched ones pair with their left rows,
  # unmatched ones surface with all earlier bindings nil. We pair left×right
  # once and derive both the combined rows and the set of matched right rows from
  # that single pass, rather than evaluating `on_match?` over the product twice.
  defp right_join(acc, ix, rights, on_expr, bindings, params) do
    pairs =
      for partial <- acc,
          right <- rights,
          on_match?(on_expr, partial, ix, right, bindings, params),
          do: {partial, right}

    combos = Enum.map(pairs, fn {partial, right} -> Map.put(partial, ix, right) end)
    matched = MapSet.new(pairs, fn {_partial, right} -> right end)

    nils = Map.new(0..(ix - 1), &{&1, nil})

    unmatched =
      for right <- rights, not MapSet.member?(matched, right), do: Map.put(nils, ix, right)

    combos ++ unmatched
  end

  # --- right-side fetchers (one per strategy) ---------------------------------

  # Returns `fn(partial, bindings, params) -> [value_list]`.
  defp fetcher(:scan, schema, ix, query, params) do
    rights = scan(schema, ix, query, params)
    fn _partial, _bindings, _params -> rights end
  end

  defp fetcher({:read, value_expr}, schema, _ix, _query, _params) do
    tag = Table.name(schema)

    fn partial, bindings, params ->
      value = eval(value_expr, partial, bindings, params)
      tag |> :mnesia.read(value) |> Enum.map(&record_values/1)
    end
  end

  defp fetcher({:index, field, value_expr}, schema, _ix, _query, _params) do
    tag = Table.name(schema)

    fn partial, bindings, params ->
      value = eval(value_expr, partial, bindings, params)
      tag |> :mnesia.index_read(value, field) |> Enum.map(&record_values/1)
    end
  end

  # Whole-table scan reduced to value lists (the stored tuple minus its tag),
  # with binding-local `where` conjuncts pushed into the match spec.
  defp scan(schema, ix, query, params) do
    tag = Table.name(schema)
    conditions = pushdown_conditions(query, ix, schema, params)
    spec = {Table.match_head(schema), conditions, [Table.match_body(schema)]}
    Transaction.run(fn -> :mnesia.select(tag, [spec]) end)
  end

  defp record_values(record), do: record |> Tuple.to_list() |> tl()

  defp on_match?(on_expr, partial, ix, right, bindings, params) do
    truthy(eval(on_expr, Map.put(partial, ix, right), bindings, params))
  end

  # A join's `on` defaults to the literal `true` (e.g. a cross join).
  defp on_expr(%Ecto.Query.QueryExpr{expr: expr}), do: expr
  defp on_expr(true), do: true
  defp on_expr(nil), do: true

  # --- where pushdown (match-spec guards for a single binding) ----------------

  # Guards for the `where` conjuncts that reference only binding `ix` and are
  # expressible as a match-spec guard. Mnesia ANDs the condition list, matching
  # the implicit AND between `where` clauses. Non-pushable conjuncts are dropped
  # here and left to `filter_wheres/4`.
  defp pushdown_conditions(%Ecto.Query{wheres: wheres}, ix, schema, params) do
    for %{expr: expr} <- wheres,
        conjunct <- split_and(expr),
        not refs_other_binding?(conjunct, ix),
        {:ok, guard} <- [guard(conjunct, ix, schema, params)] do
      guard
    end
  end

  # Does `expr` reference any binding other than `ix`?
  defp refs_other_binding?({:&, _, [b]}, ix), do: b != ix

  defp refs_other_binding?(tuple, ix) when is_tuple(tuple),
    do: tuple |> Tuple.to_list() |> Enum.any?(&refs_other_binding?(&1, ix))

  defp refs_other_binding?(list, ix) when is_list(list), do: Enum.any?(list, &refs_other_binding?(&1, ix))

  defp refs_other_binding?(_leaf, _ix), do: false

  @guard_ops %{==: :==, !=: :"/=", <: :<, >: :>, <=: :"=<", >=: :>=, and: :andalso, or: :orelse}

  defp guard({op, _, [l, r]}, ix, schema, params) when is_map_key(@guard_ops, op) do
    with {:ok, lg} <- guard(l, ix, schema, params),
         {:ok, rg} <- guard(r, ix, schema, params) do
      {:ok, {@guard_ops[op], lg, rg}}
    end
  end

  defp guard({:not, _, [c]}, ix, schema, params) do
    with {:ok, g} <- guard(c, ix, schema, params), do: {:ok, {:not, g}}
  end

  defp guard({:is_nil, _, [c]}, ix, schema, params) do
    with {:ok, g} <- guard(c, ix, schema, params), do: {:ok, {:==, g, nil}}
  end

  defp guard({:in, _, [field, values]}, ix, schema, params) do
    with {:ok, fg} <- guard(field, ix, schema, params) do
      case eval_list(values, %{}, %{}, params) do
        # `x in []` is unsatisfiable; leave it to filter_wheres rather than emit
        # an empty-disjunction guard.
        [] -> :error
        vals -> {:ok, List.to_tuple([:orelse | Enum.map(vals, &{:==, fg, &1})])}
      end
    end
  end

  defp guard({:^, _, [index]}, _ix, _schema, params), do: {:ok, Enum.at(params, index)}

  defp guard({{:., _, [{:&, _, [ix]}, field]}, _, []}, ix, schema, _params),
    do: {:ok, :"$#{Table.field_index(schema, field) + 1}"}

  defp guard(literal, _ix, _schema, _params) when is_atom(literal) or is_number(literal) or is_binary(literal),
    do: {:ok, literal}

  defp guard(_other, _ix, _schema, _params), do: :error

  # --- where / order / pagination / projection --------------------------------

  defp filter_wheres(rows, %Ecto.Query{wheres: []}, _bindings, _params), do: rows

  defp filter_wheres(rows, %Ecto.Query{wheres: wheres}, bindings, params) do
    Enum.filter(rows, fn row ->
      Enum.all?(wheres, fn %{expr: expr} -> truthy(eval(expr, row, bindings, params)) end)
    end)
  end

  defp order(rows, %Ecto.Query{order_bys: []}, _bindings), do: rows

  defp order(rows, %Ecto.Query{order_bys: order_bys}, bindings) do
    predicates =
      for %Ecto.Query.ByExpr{expr: list} <- order_bys, {direction, expr} <- list do
        {direction, expr}
      end

    Enum.sort(rows, &compare(&1, &2, predicates, bindings))
  end

  defp compare(_left, _right, [], _bindings), do: true

  defp compare(left, right, [{direction, expr} | rest], bindings) do
    l = field_value(expr, left, bindings)
    r = field_value(expr, right, bindings)

    cond do
      l == r -> compare(left, right, rest, bindings)
      direction == :desc -> l > r
      true -> l < r
    end
  end

  defp paginate(rows, offset, limit) do
    rows = if offset, do: Enum.drop(rows, offset), else: rows
    if limit, do: Enum.take(rows, limit), else: rows
  end

  defp project(rows, %Ecto.Query{select: nil}, _bindings, _params) do
    {length(rows), Enum.map(rows, &Map.fetch!(&1, 0))}
  end

  defp project(rows, %Ecto.Query{select: %{fields: fields}}, bindings, params) do
    {projected, count} =
      Enum.map_reduce(rows, 0, fn row, count ->
        {Enum.map(fields, &field_extractor(&1, row, bindings, params)), count + 1}
      end)

    {count, projected}
  end

  defp field_extractor({{:., _, [{:&, _, [_ix]}, _field]}, _, []} = expr, row, bindings, _params) do
    field_value(expr, row, bindings)
  end

  defp field_extractor({:^, [], [index]}, _row, _bindings, params), do: Enum.at(params, index)
  defp field_extractor(literal, _row, _bindings, _params), do: literal

  # --- expression evaluation over a combined row ------------------------------

  defp eval({:and, _, [l, r]}, row, b, p), do: eval(l, row, b, p) and eval(r, row, b, p)
  defp eval({:or, _, [l, r]}, row, b, p), do: eval(l, row, b, p) or eval(r, row, b, p)
  defp eval({:not, _, [c]}, row, b, p), do: not eval(c, row, b, p)

  defp eval({:is_nil, _, [c]}, row, b, p), do: is_nil(eval(c, row, b, p))

  defp eval({:in, _, [l, r]}, row, b, p), do: eval(l, row, b, p) in eval_list(r, row, b, p)

  for {op, fun} <- [==: :==, !=: :!=, <: :<, >: :>, <=: :<=, >=: :>=] do
    defp eval({unquote(op), _, [l, r]}, row, b, p) do
      apply(Kernel, unquote(fun), [eval(l, row, b, p), eval(r, row, b, p)])
    end
  end

  defp eval({:^, _, [index]}, _row, _b, params), do: Enum.at(params, index)

  defp eval({{:., _, [{:&, _, [_ix]}, _field]}, _, []} = expr, row, bindings, _params) do
    field_value(expr, row, bindings)
  end

  defp eval(literal, _row, _b, _p), do: literal

  # A list literal, or an interpolated `^values` / `^values, count` slice.
  defp eval_list({:^, _, [index, count]}, _row, _b, params) do
    for i <- index..(index + count - 1)//1, do: Enum.at(params, i)
  end

  defp eval_list({:^, _, [index]}, _row, _b, params), do: Enum.at(params, index)
  defp eval_list(list, _row, _b, _params) when is_list(list), do: list

  # Resolves `(binding ix, field)` to its value in the combined row. A binding
  # left unmatched by an outer join is `nil`, and every field of it reads `nil`.
  defp field_value({{:., _, [{:&, _, [ix]}, field]}, _, []}, row, bindings) do
    case Map.get(row, ix) do
      nil -> nil
      values -> Enum.at(values, Table.field_index(Enum.at(bindings, ix), field))
    end
  end

  # SQL treats a NULL guard result as false; mirror that so an `on`/`where` that
  # evaluates to `nil` (rather than a boolean) drops the row.
  defp truthy(true), do: true
  defp truthy(_), do: false

  # --- limit / offset (resolved like Queryable's single-source path) ----------

  defp limit_value(%Ecto.Query{limit: limit}, params), do: int_expr(limit, params)
  defp offset_value(%Ecto.Query{offset: offset}, params), do: int_expr(offset, params)

  defp int_expr(nil, _params), do: nil
  defp int_expr(%{expr: expr}, params), do: int_expr(expr, params)
  defp int_expr({:^, [], [index]}, params), do: Enum.at(params, index)
  defp int_expr(int, _params) when is_integer(int), do: int
end