Current section

Files

Jump to
nex_base lib nex_base.ex
Raw

lib/nex_base.ex

defmodule NexBase do
@moduledoc """
A fluent database query builder for Elixir, modeled after the Supabase JS SDK
(postgrest-js). Supports PostgreSQL and SQLite with automatic adapter detection
and multiple simultaneous database connections.
## Quick Start
# Single connection (simplest)
NexBase.init(url: "postgres://localhost/mydb")
NexBase.from("users") |> NexBase.select("*") |> NexBase.run()
# Multiple connections
main = NexBase.init(url: "postgres://localhost/main")
cache = NexBase.init(url: "sqlite::memory:")
main |> NexBase.from("users") |> NexBase.run()
cache |> NexBase.from("sessions") |> NexBase.run()
## Supabase API parity
All PostgrestFilterBuilder, PostgrestTransformBuilder, and
PostgrestQueryBuilder methods are implemented with equivalent semantics.
"""
alias NexBase.{Query, Conn}
require Ecto.Query
# -- Initialization --
@doc """
Initialize a database connection. Returns a `%NexBase.Conn{}` struct.
The adapter is auto-detected from the URL scheme:
- `postgres://` or `postgresql://` → PostgreSQL
- `sqlite://` → SQLite
## Options
- `:url` - Database URL (falls back to DATABASE_URL env var)
- `:ssl` - Enable SSL for cloud databases (default: false, ignored for SQLite)
- `:pool_size` - Connection pool size (default: 10)
- `:start` - Start the Repo immediately (for scripts, default: false)
## Examples
# Single connection (in application.ex)
NexBase.init(url: "postgres://localhost/mydb", ssl: true)
# In a script
conn = NexBase.init(url: "sqlite::memory:", start: true, pool_size: 1)
conn |> NexBase.from("users") |> NexBase.run()
"""
def init(opts \\ []) do
url = opts[:url] || System.get_env("DATABASE_URL")
adapter = detect_adapter(url)
pool_size = opts[:pool_size] || 10
repo_module = repo_module_for(adapter)
repo_config = build_repo_config(adapter, url, pool_size, opts)
is_first = Application.get_env(:nex_base, :default_conn) == nil
name = if is_first, do: repo_module, else: :"nex_base_#{:erlang.unique_integer([:positive, :monotonic])}"
conn = %Conn{
name: name,
adapter: adapter,
repo_module: repo_module,
repo_config: repo_config
}
Application.put_env(:nex_base, :repo_config, repo_config)
Application.put_env(:nex_base, name, repo_config)
if is_first do
Application.put_env(:nex_base, :default_conn, conn)
end
Application.put_env(:nex_base, :adapter, adapter)
if opts[:start] do
start_conn(conn)
end
conn
end
defp detect_adapter(nil), do: :postgres
defp detect_adapter(url) when is_binary(url) do
cond do
String.starts_with?(url, "sqlite") -> :sqlite
true -> :postgres
end
end
defp build_repo_config(:postgres, url, pool_size, opts) do
config = [
url: url,
pool_size: pool_size,
prepare: Keyword.get(opts, :prepare, :unnamed)
]
config = if opts[:ssl] do
config ++ [
ssl: [verify: :verify_none],
queue_target: 10_000,
queue_interval: 20_000
]
else
config
end
extra_keys = [:queue_target, :queue_interval, :timeout, :connect_timeout]
extra = Keyword.take(opts, extra_keys)
Keyword.merge(config, extra)
end
defp build_repo_config(:sqlite, url, pool_size, _opts) do
database = parse_sqlite_url(url)
[database: database, pool_size: pool_size]
end
defp parse_sqlite_url("sqlite::memory:"), do: ":memory:"
defp parse_sqlite_url("sqlite:///" <> path), do: "/" <> path
defp parse_sqlite_url("sqlite://" <> path), do: path
defp parse_sqlite_url(path), do: path
defp repo_module_for(:postgres), do: NexBase.Repo.Postgres
defp repo_module_for(:sqlite), do: NexBase.Repo.SQLite
defp start_conn(%Conn{adapter: adapter, repo_module: repo_module, name: name}) do
case adapter do
:postgres -> Application.ensure_all_started(:postgrex)
:sqlite -> Application.ensure_all_started(:exqlite)
end
Application.ensure_all_started(:ecto_sql)
start_opts = if name == repo_module, do: [], else: [name: name]
case repo_module.start_link(start_opts) do
{:ok, _pid} -> :ok
{:error, {:already_started, _pid}} -> :ok
{:error, reason} -> {:error, reason}
end
end
@doc "Returns the default connection, or raises if none configured."
def default_conn do
Application.get_env(:nex_base, :default_conn) ||
raise "NexBase not initialized. Call NexBase.init/1 first."
end
@doc "Returns the adapter for a connection (`:postgres` or `:sqlite`)."
def adapter(%Conn{adapter: adapter}), do: adapter
def adapter, do: default_conn().adapter
# -- Query Building --
@doc """
Starts a query builder for the given table. Supabase `.from(table)` equivalent.
"""
def from(%Conn{} = conn, table_name) when is_binary(table_name) do
%Query{table: table_name, conn: conn}
end
def from(table_name) when is_binary(table_name) do
%Query{table: table_name}
end
@doc """
Selects specific columns. Supabase `.select(columns)` equivalent.
Accepts either a list of columns (atoms or strings) or a Supabase-style
comma-separated column string, which supports column aliases via `alias:column`
syntax. Calling `.select()` on insert/update/upsert/delete queries enables
`RETURNING`, so the affected rows are returned (Supabase `return=representation`).
## Examples
# List of atoms
NexBase.from("users") |> NexBase.select([:id, :name])
# String with aliases
NexBase.from("users") |> NexBase.select("id,display_name:name,profile(*)")
# Enable RETURNING after insert
NexBase.from("users") |> NexBase.insert(%{name: "Alice"}) |> NexBase.select() |> NexBase.run()
"""
def select(query, columns \\ "*")
def select(%Query{type: type} = query, columns) when type in [:insert, :update, :delete, :upsert] do
%{query | select: normalize_select(columns), returning: true}
end
def select(%Query{} = query, columns) do
%{query | select: normalize_select(columns)}
end
@doc """
Like `select/2` but accepts options `head:` and `count:`.
Supabase `.select("col", { head: true, count: "exact" })` equivalent.
"""
def select(%Query{} = query, columns, opts) when is_list(opts) do
base = select(query, columns)
base = if opts[:head], do: %{base | head: true}, else: base
if opts[:count], do: count(base, opts[:count]), else: base
end
defp normalize_select("*"), do: []
defp normalize_select(columns) when is_list(columns), do: columns
defp normalize_select(columns) when is_binary(columns) do
columns
|> String.split(",", trim: true)
|> Enum.map(&String.trim/1)
|> Enum.reject(&(&1 == ""))
|> Enum.flat_map(&parse_select_col/1)
end
defp parse_select_col(col) do
# Handle "alias:column" → column AS alias
case String.split(col, ":", parts: 2) do
[alias_name, col_name] when alias_name != "" and col_name != "" ->
# Strip foreign-table parens like posts(*)
col_name_clean = String.replace(col_name, ~r/\(.*\)$/, "")
if String.contains?(col_name, "(") do
# Foreign table select: just select the raw column name for now
[col_name_clean]
else
["#{col_name_clean} AS #{alias_name}"]
end
[col] ->
[col]
end
end
# -- Basic Filters --
@doc "Supabase `.eq(column, value)` equivalent."
def eq(%Query{} = query, column, value) do
append_filter(query, {:eq, column, value})
end
@doc "Supabase `.neq(column, value)` equivalent."
def neq(%Query{} = query, column, value) do
append_filter(query, {:neq, column, value})
end
@doc "Supabase `.gt(column, value)` equivalent."
def gt(%Query{} = query, column, value) do
append_filter(query, {:gt, column, value})
end
@doc "Supabase `.lt(column, value)` equivalent."
def lt(%Query{} = query, column, value) do
append_filter(query, {:lt, column, value})
end
@doc "Supabase `.gte(column, value)` equivalent."
def gte(%Query{} = query, column, value) do
append_filter(query, {:gte, column, value})
end
@doc "Supabase `.lte(column, value)` equivalent."
def lte(%Query{} = query, column, value) do
append_filter(query, {:lte, column, value})
end
@doc """
Supabase `.is(column, value)` equivalent. Use for NULL checks and boolean columns.
## Examples
NexBase.from("users") |> NexBase.is(:deleted_at, nil)
NexBase.from("users") |> NexBase.is(:active, true)
"""
def is(%Query{} = query, column, value) do
append_filter(query, {:is, column, value})
end
@doc "Shorthand for `is(column, nil)`."
def is_null(%Query{} = query, column) do
append_filter(query, {:is, column, nil})
end
@doc "Shorthand for `not_filter(column, :is, nil)`."
def is_not_null(%Query{} = query, column) do
append_not_filter(query, {:is, column, nil})
end
@doc """
Supabase `.in(column, values)` equivalent. Named `in_list` to avoid the
reserved `in/2` operator. Use `filter_in/3` for a friendlier alias.
"""
def in_list(%Query{} = query, column, values) when is_list(values) do
append_filter(query, {:in, column, values})
end
@doc "Alias for `in_list/3`."
def filter_in(%Query{} = query, column, values) when is_list(values) do
in_list(query, column, values)
end
@doc "Supabase `.not(column, :in, values)` convenience. Equivalent to `not_filter(column, :in, values)`."
def not_in_list(%Query{} = query, column, values) when is_list(values) do
append_not_filter(query, {:in, column, values})
end
@doc "Supabase `.like(column, pattern)` equivalent."
def like(%Query{} = query, column, pattern) do
append_filter(query, {:like, column, pattern})
end
@doc "Supabase `.not(column, :like, pattern)` convenience."
def nlike(%Query{} = query, column, pattern) do
append_filter(query, {:nlike, column, pattern})
end
@doc "Supabase `.ilike(column, pattern)` equivalent."
def ilike(%Query{} = query, column, pattern) do
append_filter(query, {:ilike, column, pattern})
end
@doc "Supabase `.not(column, :ilike, pattern)` convenience."
def nilike(%Query{} = query, column, pattern) do
append_filter(query, {:nilike, column, pattern})
end
@doc """
Supabase `.likeAllOf(column, patterns)` equivalent. Matches if the column matches
ALL of the given LIKE patterns (ANDed together).
"""
def like_all_of(%Query{} = query, column, patterns) when is_list(patterns) do
filters = Enum.map(patterns, &{:like, column, &1})
%{query | filters: query.filters ++ filters}
end
@doc """
Supabase `.likeAnyOf(column, patterns)` equivalent. Matches if the column matches
ANY of the given LIKE patterns (ORed together).
"""
def like_any_of(%Query{} = query, column, patterns) when is_list(patterns) do
group = Enum.map(patterns, &{:like, column, &1})
%{query | or_filters: query.or_filters ++ [group]}
end
@doc """
Supabase `.ilikeAllOf(column, patterns)` equivalent. Case-insensitive.
"""
def ilike_all_of(%Query{} = query, column, patterns) when is_list(patterns) do
filters = Enum.map(patterns, &{:ilike, column, &1})
%{query | filters: query.filters ++ filters}
end
@doc """
Supabase `.ilikeAnyOf(column, patterns)` equivalent. Case-insensitive.
"""
def ilike_any_of(%Query{} = query, column, patterns) when is_list(patterns) do
group = Enum.map(patterns, &{:ilike, column, &1})
%{query | or_filters: query.or_filters ++ [group]}
end
defp append_filter(%Query{} = query, filter) do
%{query | filters: query.filters ++ [filter]}
end
defp append_not_filter(%Query{} = query, filter) do
%{query | not_filters: query.not_filters ++ [filter]}
end
# -- Pagination / Ordering --
@doc """
Supabase `.limit(count, opts)` equivalent.
## Options
- `:referenced_table` / `:foreign_table` — apply to a referenced/embedded table
"""
def limit(%Query{} = query, count, opts \\ []) when is_integer(count) and count >= 0 do
ref_table = opts[:referenced_table] || opts[:foreign_table]
%{query | limit: count, limit_referenced_table: ref_table}
end
@doc """
Supabase `.offset(value, opts)` equivalent.
"""
def offset(%Query{} = query, offset, opts \\ []) when is_integer(offset) and offset >= 0 do
ref_table = opts[:referenced_table] || opts[:foreign_table]
%{query | offset: offset, offset_referenced_table: ref_table}
end
@doc """
Supabase `.order(column, opts)` equivalent.
## Options
- `:ascending` — boolean, default `true`
- `:nulls_first` — if `true`, NULLs sort first; if `false`, NULLs sort last
- `:nulls_last` — alias for `nulls_first: false`
- `:referenced_table` / `:foreign_table` — apply to a referenced/embedded table
Backward-compatible: `order(query, col, :asc)` or `order(query, col, :desc, nulls_first: true)`
are still supported.
"""
def order(query, column, direction_or_opts \\ :asc)
def order(%Query{} = query, column, direction) when is_atom(direction) and not is_list(direction) do
order(query, column, direction, [])
end
def order(%Query{} = query, column, opts) when is_list(opts) do
dir = if opts[:ascending] == false, do: :desc, else: :asc
order(query, column, dir, opts)
end
def order(%Query{} = query, column, direction, opts) when is_list(opts) do
ref_table = opts[:referenced_table] || opts[:foreign_table]
entry =
case {direction, opts} do
{dir, []} -> {dir, column}
{dir, o} ->
stripped = Enum.reject(o, fn {k, _} -> k in [:referenced_table, :foreign_table, :ascending] end)
if stripped == [], do: {dir, column}, else: {dir, column, stripped}
end
entry_with_ref = if ref_table, do: {entry, referenced_table: ref_table}, else: entry
%{query | order_by: query.order_by ++ [entry_with_ref]}
end
# -- Generic filter dispatch --
@doc """
Adds a generic filter clause. Supabase `filter(column, operator, value)` equivalent.
`operator` can be either an atom (`:eq`, `:gt`, ...) or a string, including the
`"not.eq"` form supported by Supabase.
## Examples
NexBase.from("users") |> NexBase.filter(:status, :eq, "active")
NexBase.from("users") |> NexBase.filter("status", "not.eq", "banned")
"""
def filter(%Query{} = query, column, operator, value) when is_atom(operator) do
case operator do
:or -> or_filter(query, value)
:not -> apply_query_not_filter(query, column, value)
op ->
normalized = normalize_operator(op)
append_filter(query, {normalized, column, value})
end
end
def filter(%Query{} = query, column, "not." <> op_str, value) do
op = String.to_existing_atom(op_str)
normalized = normalize_operator(op)
append_not_filter(query, {normalized, column, value})
end
def filter(%Query{} = query, column, operator, value) when is_binary(operator) do
op = String.to_existing_atom(operator)
filter(query, column, op, value)
end
@doc """
Adds a negated filter. Supabase `.not(column, operator, value)` equivalent.
Named `not_filter/4` in Elixir because `not` is a reserved keyword.
"""
def not_filter(%Query{} = query, column, operator, value) when is_atom(operator) do
normalized = normalize_operator(operator)
append_not_filter(query, {normalized, column, value})
end
def not_filter(%Query{} = query, column, operator, value) when is_binary(operator) do
not_filter(query, column, String.to_existing_atom(operator), value)
end
defp apply_query_not_filter(%Query{} = query, column, {op, value}) when is_atom(op) do
normalized = normalize_operator(op)
append_not_filter(query, {normalized, column, value})
end
defp apply_query_not_filter(%Query{} = query, column, value) do
not_filter(query, column, :eq, value)
end
@doc """
Adds an OR filter group. Supabase `.or(filters, opts)` equivalent.
Accepts a list of filter tuples `{operator, column, value}`. All filters inside
are joined with OR and combined with the query's AND filters.
## Options
- `:referenced_table` / `:foreign_table` — filter on a referenced/embedded table
"""
def or_filter(%Query{} = query, filters, opts \\ []) when is_list(filters) do
if Enum.empty?(filters) do
query
else
normalized =
Enum.map(filters, fn
{op, col, val} -> {normalize_operator(op), col, val}
{col, val} -> {:eq, col, val}
end)
ref_table = opts[:referenced_table] || opts[:foreign_table]
entry = if ref_table, do: {normalized, referenced_table: ref_table}, else: normalized
%{query | or_filters: query.or_filters ++ [entry]}
end
end
# -- Full Text Search --
@doc """
Supabase `.textSearch(column, query, opts)` equivalent.
## Options
- `:config` — text search config (default: `"english"`)
- `:type` — `:plain` (default), `:phrase`, or `:websearch`
"""
def text_search(query, column, query_text, opts \\ [])
def text_search(%Query{} = query, column, query_text, opts) when is_list(opts) do
config = opts[:config] || "english"
type = opts[:type] || :plain
op = case type do
:plain -> :plfts
:phrase -> :phfts
:phfts -> :phfts
:websearch -> :wfts
other when is_atom(other) -> other
end
append_filter(query, {op, {column, config}, query_text})
end
# For backward compat: text_search(query, col, query_text, "english") — 4th arg as string config
def text_search(%Query{} = query, column, query_text, config) when is_binary(config) do
append_filter(query, {:fts, {column, config}, query_text})
end
@doc """
Supabase `.fts(column, query, opts)` — plainto_tsquery full text search.
"""
def fts(%Query{} = query, column, query_text, opts \\ []) do
config = if is_list(opts), do: opts[:config] || "english", else: opts
append_filter(query, {:fts, {column, config}, query_text})
end
@doc """
Supabase `.plfts(column, query, opts)` — plainto_tsquery full text search.
"""
def plfts(%Query{} = query, column, query_text, opts \\ []) do
config = if is_list(opts), do: opts[:config] || "english", else: opts
append_filter(query, {:plfts, {column, config}, query_text})
end
@doc """
Supabase `.phfts(column, query, opts)` — phraseto_tsquery full text search.
"""
def phfts(%Query{} = query, column, query_text, opts \\ []) do
config = if is_list(opts), do: opts[:config] || "english", else: opts
append_filter(query, {:phfts, {column, config}, query_text})
end
@doc """
Supabase `.wfts(column, query, opts)` — websearch_to_tsquery full text search.
"""
def wfts(%Query{} = query, column, query_text, opts \\ []) do
config = if is_list(opts), do: opts[:config] || "english", else: opts
append_filter(query, {:wfts, {column, config}, query_text})
end
# -- Array / Range operators --
@doc "Supabase `.contains(column, value)` equivalent."
def contains(%Query{} = query, column, values) do
append_filter(query, {:cs, column, values})
end
@doc """
Supabase `.containedBy(column, value)` equivalent. Named `contained_in/3` in
Elixir for natural-language phrasing.
"""
def contained_in(%Query{} = query, column, values) do
append_filter(query, {:cd, column, values})
end
@doc "Alias for `contained_in/3`."
def contained_by(%Query{} = query, column, values) do
contained_in(query, column, values)
end
@doc "Supabase `.overlaps(column, value)` equivalent."
def overlaps(%Query{} = query, column, values) do
append_filter(query, {:ov, column, values})
end
@doc "Supabase `.rangeLt(column, range)` — strictly left of."
def range_lt(%Query{} = query, column, value) do
append_filter(query, {:sl, column, value})
end
@doc "Supabase `.rangeGte(column, range)` — does not extend to the right of."
def range_gte(%Query{} = query, column, value) do
append_filter(query, {:nxl, column, value})
end
@doc "Supabase `.rangeGt(column, range)` — strictly right of."
def range_gt(%Query{} = query, column, value) do
append_filter(query, {:sr, column, value})
end
@doc "Supabase `.rangeLte(column, range)` — does not extend to the left of."
def range_lte(%Query{} = query, column, value) do
append_filter(query, {:nxr, column, value})
end
@doc "Supabase `.rangeAdjacent(column, range)` — ranges are adjacent."
def range_adjacent(%Query{} = query, column, value) do
append_filter(query, {:adj, column, value})
end
@doc "Supabase `.match(map)` equivalent — multiple eq filters ANDed together."
def match(%Query{} = query, conditions) when is_map(conditions) do
new_filters = Enum.map(conditions, fn {col, val} -> {:eq, col, val} end)
%{query | filters: query.filters ++ new_filters}
end
# -- Count / Single Results --
@doc """
Supabase `.count(mode)` equivalent. Accepts `:exact`, `:planned`, or `:estimated`.
When `run/1` executes with a count set, it returns `{:ok, data, count}`.
"""
def count(%Query{} = query, mode \\ :exact)
when mode in [:exact, :planned, :estimated] do
%{query | count: mode}
end
@doc """
Supabase `.single()` equivalent. Returns the query as a single unwrapped object
when `.run()/.run!()` executes. Raises if 0 rows or >1 rows are returned.
"""
def single(%Query{} = query) do
%{query | single: true, limit: 1}
end
@doc """
Supabase `.maybeSingle()` equivalent. Returns the query as a single unwrapped
object or `nil` when `.run()/.run!()` executes. Raises only if >1 rows are returned.
"""
def maybe_single(%Query{} = query) do
%{query | maybe_single: true, limit: 1}
end
@doc """
Supabase `.range(from, to, opts)` equivalent. `from` and `to` are 0-based inclusive.
"""
def range(%Query{} = query, from, to, opts \\ []) do
limit = max(0, to - from + 1)
from = max(0, from)
ref_table = opts[:referenced_table] || opts[:foreign_table]
%{query |
limit: limit, offset: from,
limit_referenced_table: ref_table || query.limit_referenced_table,
offset_referenced_table: ref_table || query.offset_referenced_table
}
end
# -- Explain / CSV / Rollback --
@doc """
Supabase `.explain(opts)` equivalent.
Causes the query to return its EXPLAIN plan instead of rows.
## Options
- `:analyze` — execute the query and show actual timing
- `:verbose` — include query identifier and output columns
- `:settings` — include configuration parameters affecting the plan
- `:buffers` — include buffer usage information
- `:wal` — include WAL record generation info
- `:format` — `:text` (default) or `:json`
"""
def explain(%Query{} = query, opts \\ []) do
defaults = [analyze: false, verbose: false, settings: false, buffers: false, wal: false, format: :text]
merged = Keyword.merge(defaults, opts)
%{query | explain_opts: merged}
end
@doc """
Supabase `.csv()` equivalent. Returns query results as a CSV string.
"""
def csv(%Query{} = query) do
%{query | csv: true}
end
@doc """
Supabase `.geojson()` equivalent.
Returns results as a GeoJSON FeatureCollection. Each row becomes a Feature;
columns named `geometry`/`geom` are treated as the Feature geometry. Remaining
columns become Feature properties.
Requires PostgreSQL with PostGIS for true spatial output; the Ecto adapter
converts rows to GeoJSON maps at the client level.
"""
def geojson(%Query{} = query) do
%{query | geojson: true}
end
@doc """
Supabase `.throwOnError()` equivalent. Causes `run/1` to raise instead of
returning `{:error, reason}`.
"""
def throw_on_error(%Query{} = query) do
%{query | throw_on_error: true}
end
@doc """
Supabase `.rollback()` equivalent. Executes the query inside a transaction and
rolls it back (data is not persisted, but query results are returned).
"""
def rollback(%Query{} = query) do
%{query | rollback: true}
end
@doc """
Switch the schema for this query. Supabase `PostgrestClient.schema(name)` equivalent.
When called on a `%Query{}`, all subsequent SQL references the given schema.
When called on a `%Conn{}`, returns a new connection tagged with that schema
(useful for `schema("private") |> from("users")` style).
"""
def schema(%Query{} = query, name) when is_binary(name) do
%{query | schema: name}
end
def schema(%Conn{} = conn, name) when is_binary(name) do
fn table_name ->
conn
|> from(table_name)
|> schema(name)
end
end
@doc """
Set the maximum number of rows that can be affected by an update or delete.
Supabase `.maxAffected(n)` equivalent (PostgREST 13+).
"""
def max_affected(%Query{} = query, n) when is_integer(n) and n >= 0 do
%{query | max_affected: n}
end
# -- Raw SQL --
@doc """
Executes a raw SQL query and returns results as a list of maps. Parameterize with
`$1`, `$2`, ... — placeholders are automatically adapted for SQLite.
"""
def sql(%Conn{} = conn, sql_str, params) when is_binary(sql_str) and is_list(params) do
repo_mod = resolve_repo(conn)
sql_str = normalize_placeholders(sql_str, conn.adapter)
case Ecto.Adapters.SQL.query(repo_mod, sql_str, params) do
{:ok, %{rows: rows, columns: columns}} when is_list(rows) ->
{:ok, Enum.map(rows, fn row -> columns |> Enum.zip(row) |> Map.new() end)}
{:ok, _} ->
{:ok, []}
{:error, _} = err -> err
end
end
def sql(sql_str, params \\ []) when is_binary(sql_str) do
sql(default_conn(), sql_str, params)
end
@doc "Executes a raw SQL query and returns the raw driver result."
def query(%Conn{} = conn, sql_str, params) when is_binary(sql_str) and is_list(params) do
repo_mod = resolve_repo(conn)
Ecto.Adapters.SQL.query(repo_mod, normalize_placeholders(sql_str, conn.adapter), params)
end
def query(sql_str, params \\ []) when is_binary(sql_str) do
query(default_conn(), sql_str, params)
end
@doc "Executes a raw SQL query, raising on error."
def query!(%Conn{} = conn, sql_str, params) when is_binary(sql_str) and is_list(params) do
repo_mod = resolve_repo(conn)
Ecto.Adapters.SQL.query!(repo_mod, normalize_placeholders(sql_str, conn.adapter), params)
end
def query!(sql_str, params \\ []) when is_binary(sql_str) do
query!(default_conn(), sql_str, params)
end
# -- CRUD --
@doc """
Supabase `.insert(values, opts)` equivalent.
## Options
- `:count` — count mode (`:exact`, `:planned`, `:estimated`)
- `:default_to_null` — when `false`, missing fields use column defaults
instead of `NULL` (PostgREST `missing=default`); default is `true`
"""
def insert(%Query{} = query, data, opts \\ [])
when is_map(data) or is_list(data) do
count = opts[:count]
default_to_null = Keyword.get(opts, :default_to_null, true)
q = %{query | type: :insert, data: data, default_to_null: default_to_null}
if count, do: count(q, count), else: q
end
@doc """
Supabase `.update(values, opts)` equivalent.
## Options
- `:count` — count mode
"""
def update(%Query{} = query, data, opts \\ []) when is_map(data) do
q = %{query | type: :update, data: data}
if opts[:count], do: count(q, opts[:count]), else: q
end
@doc """
Supabase `.delete(opts)` equivalent.
## Options
- `:count` — count mode
"""
def delete(%Query{} = query, opts \\ []) do
q = %{query | type: :delete}
if opts[:count], do: count(q, opts[:count]), else: q
end
@doc """
Supabase `.upsert(values, opts)` equivalent.
## Options
- `:on_conflict` — column or list of columns defining the unique constraint
- `:ignore_duplicates` — if `true`, ignore duplicates instead of merging
(Supabase `resolution=ignore-duplicates`)
- `:default_to_null` — default `true`
- `:count` — count mode
"""
def upsert(%Query{} = query, data, opts \\ [])
when is_map(data) or is_list(data) do
default_to_null = Keyword.get(opts, :default_to_null, true)
upsert_opts = [
on_conflict: opts[:on_conflict],
ignore_duplicates: Keyword.get(opts, :ignore_duplicates, false),
default_to_null: default_to_null
]
q = %{query | type: :upsert, data: data, upsert_opts: upsert_opts, default_to_null: default_to_null}
if opts[:count], do: count(q, opts[:count]), else: q
end
# -- RPC --
@doc """
Supabase `.rpc(function_name, args, opts)` equivalent. Executes a stored
procedure / function (PostgreSQL only).
## Options
- `:head` — return count only (no body). Supabase `head: true`.
- `:get` — call via GET instead of POST (read-only mode)
- `:count` — count mode for set-returning functions
- `:conn` — use a specific connection (otherwise uses default)
"""
def rpc(function_name, params \\ %{}, opts \\ []) do
conn = opts[:conn] || default_conn()
if conn.adapter == :sqlite do
raise "NexBase.rpc is not supported with SQLite (stored procedures are a PostgreSQL feature)"
end
repo_mod = resolve_repo(conn)
head = !!opts[:head]
count_mode = opts[:count]
keys = Map.keys(params)
values = Map.values(params)
placeholders = Enum.map(1..map_size(params), fn i -> "$#{i}" end)
# Quote function name and argument names to prevent SQL injection.
quoted_fn = quote_ident(function_name)
args_str =
Enum.zip(keys, placeholders)
|> Enum.map(fn {k, p} -> "#{quote_ident(k)} := #{p}" end)
|> Enum.join(", ")
sql_str = "SELECT * FROM #{quoted_fn}(#{args_str})"
result = Ecto.Adapters.SQL.query(repo_mod, sql_str, values)
case result do
{:ok, %{rows: rows, columns: columns}} ->
data =
if head do
nil
else
Enum.map(rows, fn row -> columns |> Enum.zip(row) |> Map.new() end)
end
if count_mode do
{:ok, data, length(rows)}
else
{:ok, data}
end
{:error, _} = err -> err
end
end
# -- Transaction --
@doc """
Supabase `.rpc` / transaction flow equivalent. Runs `fun` in a transaction;
`fun` receives the connection as argument.
## Options
- `:conn` — connection to use (default: default connection)
- `:timeout` — transaction timeout in ms (default: 60_000)
"""
def transaction(fun, opts \\ []) when is_function(fun, 0) do
conn = Keyword.get(opts, :conn) || default_conn()
repo_mod = resolve_repo(conn)
timeout = Keyword.get(opts, :timeout, 60_000)
repo_mod.transaction(fun, timeout: timeout)
end
# -- Execution helpers --
@doc """
Bang variant of `run/1`. Raises on error instead of returning `{:error, reason}`.
"""
def run!(%Query{} = query) do
case run(query) do
{:ok, result} -> result
{:ok, result, count} -> {result, count}
{:error, error} -> raise "Query failed: #{inspect(error)}"
end
end
@doc """
Supabase `.single()` enforcement: returns exactly one row; raises if 0 or >1.
"""
def one!(%Query{} = query) do
query = if query.type in [:insert, :update, :delete, :upsert], do: %{query | type: :select}, else: query
case run(%{query | limit: 2}) do
{:ok, [row]} -> row
{:ok, [_row, _]} -> raise "Expected exactly one row, got more"
{:ok, []} -> raise "Expected exactly one row, got none"
{:ok, _data, _count} -> one!(%{query | type: :select, count: nil})
{:error, error} -> raise "Query failed: #{inspect(error)}"
end
end
@doc """
Supabase `.maybeSingle()`: returns one row or `nil`. Raises only if >1 rows.
"""
def maybe_one(%Query{} = query) do
query = if query.type in [:insert, :update, :delete, :upsert], do: %{query | type: :select}, else: query
case run(%{query | limit: 2}) do
{:ok, [row]} -> row
{:ok, []} -> nil
{:ok, [_, _ | _]} -> raise "Expected at most one row, got more"
{:ok, _data, _count} -> maybe_one(%{query | type: :select, count: nil})
{:error, error} -> raise "Query failed: #{inspect(error)}"
end
end
@doc """
Returns rows as a list (executes the query eagerly and returns them).
"""
def stream(%Query{} = query, _opts \\ []) do
conn = query.conn || default_conn()
_repo_mod = resolve_repo(conn)
query = Map.put(query, :type, :select)
case run(query) do
{:ok, rows} -> rows
{:ok, rows, _count} -> rows
{:error, error} -> raise "Query failed: #{inspect(error)}"
end
end
# -- Execution (run) --
@doc """
Executes the built query.
Returns:
- `{:ok, data}` for normal select (list of maps, or single unwrapped map when
`.single()`/`.maybe_single()` is set)
- `{:ok, data, count}` when `.count(mode)` is set
- `{:error, exception}` on failure
"""
def run(query)
def run(%Query{throw_on_error: true} = query) do
case run(%{query | throw_on_error: false}) do
{:ok, result} -> result
{:ok, result, count} -> {result, count}
{:error, error} -> raise "Query failed: #{inspect(error)}"
end
end
def run(%Query{rollback: true} = query) do
# Run inside a transaction and capture the result before rolling back.
# We store it via the process dictionary so the rollback doesn't discard it.
conn = query.conn || default_conn()
repo_mod = resolve_repo(conn)
key = {:nexbase_rollback_result, make_ref()}
result =
repo_mod.transaction(fn ->
inner = run(%{query | rollback: false})
Process.put(key, inner)
repo_mod.rollback(:nexbase_rollback)
end)
case result do
{:error, :nexbase_rollback} -> Process.delete(key) || {:ok, []}
{:error, reason} -> {:error, reason}
end
end
def run(%Query{type: :select} = query) do
run_select(query)
rescue
e -> {:error, e}
end
def run(%Query{type: :insert, table: table, data: data, conn: conn,
select: select, returning: returning, count: count_mode,
default_to_null: default_to_null, filters: filters,
single: single, maybe_single: maybe_single} = _query) do
conn = conn || default_conn()
repo_mod = resolve_repo(conn)
data_list = if is_list(data), do: data, else: [data]
data_list =
if default_to_null do
data_list
else
# Remove nil fields so DB defaults apply
Enum.map(data_list, fn row ->
Enum.reject(row, fn {_k, v} -> is_nil(v) end) |> Map.new()
end)
end
opts = build_returning_opts(returning, select)
{count, rows} =
if returning and conn.adapter == :sqlite do
sqlite_insert_with_returning(repo_mod, table, data_list, select)
else
repo_mod.insert_all(table, data_list, opts)
end
build_mutation_result(count, rows, count_mode, filters)
|> handle_single_maybe_single(single, maybe_single)
rescue
e -> {:error, e}
end
def run(%Query{type: :update, table: table, data: data, filters: filters, conn: conn,
select: select, returning: returning, count: count_mode,
or_filters: or_filters, not_filters: not_filters,
max_affected: max_affected, single: single, maybe_single: maybe_single} = _query) do
conn = conn || default_conn()
repo_mod = resolve_repo(conn)
base_query = Ecto.Query.from(t in table)
query_with_filters = Enum.reduce(filters, base_query, fn filter, acc ->
apply_filter(acc, filter, conn.adapter)
end)
query_with_filters = Enum.reduce(not_filters, query_with_filters, fn filter, acc ->
apply_not_filter(acc, filter, conn.adapter)
end)
query_with_filters = Enum.reduce(or_filters, query_with_filters, fn group, acc ->
apply_or_filter_group(acc, group, conn.adapter)
end)
{count, rows} =
cond do
returning and conn.adapter == :sqlite ->
{where_sql, params} = build_where_for_sqlite(filters, or_filters, not_filters)
sqlite_update_with_returning(repo_mod, table, data, where_sql, params, select, max_affected)
max_affected != nil and conn.adapter == :postgres ->
# PostgreSQL doesn't support LIMIT on UPDATE/DELETE. Use raw SQL with a subquery.
{where_sql, params} = build_where(filters, or_filters, not_filters, :postgres)
postgres_update_with_limit(repo_mod, table, data, where_sql, params, select, max_affected, returning)
true ->
updates = [set: Enum.to_list(data)] ++ if returning, do: [returning: normalize_select_for_returning(select)], else: []
repo_mod.update_all(query_with_filters, updates)
end
build_mutation_result(count, rows, count_mode, filters)
|> handle_single_maybe_single(single, maybe_single)
rescue
e -> {:error, e}
end
def run(%Query{type: :delete, table: table, filters: filters, conn: conn,
select: select, returning: returning, count: count_mode,
or_filters: or_filters, not_filters: not_filters,
max_affected: max_affected, single: single, maybe_single: maybe_single} = _query) do
conn = conn || default_conn()
repo_mod = resolve_repo(conn)
base_query = Ecto.Query.from(t in table)
query_with_filters = Enum.reduce(filters, base_query, fn filter, acc ->
apply_filter(acc, filter, conn.adapter)
end)
query_with_filters = Enum.reduce(not_filters, query_with_filters, fn filter, acc ->
apply_not_filter(acc, filter, conn.adapter)
end)
query_with_filters = Enum.reduce(or_filters, query_with_filters, fn group, acc ->
apply_or_filter_group(acc, group, conn.adapter)
end)
{count, rows} =
cond do
returning and conn.adapter == :sqlite ->
{where_sql, params} = build_where_for_sqlite(filters, or_filters, not_filters)
sqlite_delete_with_returning(repo_mod, table, where_sql, params, select, max_affected)
max_affected != nil and conn.adapter == :postgres ->
{where_sql, params} = build_where(filters, or_filters, not_filters, :postgres)
postgres_delete_with_limit(repo_mod, table, where_sql, params, select, max_affected, returning)
true ->
opts = if returning, do: [returning: normalize_select_for_returning(select)], else: []
repo_mod.delete_all(query_with_filters, opts)
end
build_mutation_result(count, rows, count_mode, filters)
|> handle_single_maybe_single(single, maybe_single)
rescue
e -> {:error, e}
end
def run(%Query{type: :upsert, table: table, data: data, conn: conn,
select: select, returning: returning, count: count_mode,
upsert_opts: upsert_opts, filters: filters,
single: single, maybe_single: maybe_single} = _query) do
conn = conn || default_conn()
repo_mod = resolve_repo(conn)
data_list = if is_list(data), do: data, else: [data]
on_conflict = if upsert_opts[:ignore_duplicates], do: :nothing, else: :replace_all
default_to_null = Keyword.get(upsert_opts, :default_to_null, true)
data_list =
if default_to_null do
data_list
else
Enum.map(data_list, fn row ->
Enum.reject(row, fn {_k, v} -> is_nil(v) end) |> Map.new()
end)
end
opts =
if upsert_opts[:on_conflict] do
[
on_conflict: on_conflict,
conflict_target: upsert_opts[:on_conflict]
] ++ build_returning_opts(returning, select)
else
[on_conflict: on_conflict] ++ build_returning_opts(returning, select)
end
{count, rows} =
if conn.adapter == :sqlite do
# SQLite ecto adapter does not support schema-less insert_all with
# :replace_all on_conflict. Always use raw SQL with RETURNING.
sqlite_upsert_with_returning(repo_mod, table, data_list, on_conflict, upsert_opts[:on_conflict], select)
else
repo_mod.insert_all(table, data_list, opts)
end
build_mutation_result(count, rows, count_mode, filters)
|> handle_single_maybe_single(single, maybe_single)
rescue
e -> {:error, e}
end
# SQLite ecto adapter does not support schema-less insert_all/update_all/delete_all
# with the `:returning` option. We fall back to raw SQL with RETURNING clause
# (SQLite 3.35+ supports RETURNING natively).
defp sqlite_insert_with_returning(repo_mod, table, data_list, select) do
if Enum.empty?(data_list) do
{0, nil}
else
# Union of all column keys across all rows (matches Supabase behavior).
column_keys =
data_list
|> Enum.flat_map(&Map.keys/1)
|> Enum.uniq()
columns = Enum.map(column_keys, &to_string/1)
returning_cols = sqlite_returning_cols(select)
returning_column_names =
if select == [] do
sqlite_table_columns(repo_mod, table)
else
Enum.map(select, fn
col when is_atom(col) -> col
col when is_binary(col) ->
case String.split(col, " AS ", parts: 2) do
[_c, alias_name] -> String.to_atom(String.trim(alias_name))
[c] -> String.to_atom(String.trim(c))
end
end)
end
placeholders_per_row =
Enum.map(columns, fn _ -> "?" end) |> Enum.join(", ")
values_sql =
Enum.map(data_list, fn _row -> "(#{placeholders_per_row})" end) |> Enum.join(", ")
params =
Enum.flat_map(data_list, fn row ->
Enum.map(column_keys, fn k -> Map.get(row, k) end)
end)
sql = """
INSERT INTO #{quote_ident(table)} (#{Enum.map(columns, &quote_ident/1) |> Enum.join(", ")})
VALUES #{values_sql}
RETURNING #{returning_cols}
"""
result = Ecto.Adapters.SQL.query!(repo_mod, sql, params)
rows = sqlite_rows_to_maps(result, returning_column_names)
{length(rows), rows}
end
end
defp sqlite_update_with_returning(repo_mod, table, data, where_sql, params, select, max_affected) do
set_clause =
data
|> Enum.to_list()
|> Enum.map(fn {k, _v} -> "#{quote_ident(k)} = ?" end)
|> Enum.join(", ")
data_values = data |> Enum.to_list() |> Enum.map(&elem(&1, 1))
returning_cols = sqlite_returning_cols(select)
returning_column_names =
if select == [] do
sqlite_table_columns(repo_mod, table)
else
Enum.map(select, &sqlite_select_col_name/1)
end
limit_sql = if max_affected, do: " LIMIT #{max_affected}", else: ""
sql = """
UPDATE #{quote_ident(table)} SET #{set_clause}
#{where_sql}
#{limit_sql}
RETURNING #{returning_cols}
"""
result = Ecto.Adapters.SQL.query!(repo_mod, sql, data_values ++ params)
rows = sqlite_rows_to_maps(result, returning_column_names)
{length(rows), rows}
end
defp sqlite_delete_with_returning(repo_mod, table, where_sql, params, select, max_affected) do
returning_cols = sqlite_returning_cols(select)
returning_column_names =
if select == [] do
sqlite_table_columns(repo_mod, table)
else
Enum.map(select, &sqlite_select_col_name/1)
end
limit_sql = if max_affected, do: " LIMIT #{max_affected}", else: ""
sql = """
DELETE FROM #{quote_ident(table)}
#{where_sql}
#{limit_sql}
RETURNING #{returning_cols}
"""
result = Ecto.Adapters.SQL.query!(repo_mod, sql, params)
rows = sqlite_rows_to_maps(result, returning_column_names)
{length(rows), rows}
end
defp sqlite_upsert_with_returning(repo_mod, table, data_list, on_conflict, conflict_target, select) do
if Enum.empty?(data_list) do
{0, nil}
else
# Union of all column keys across all rows (matches Supabase behavior).
column_keys =
data_list
|> Enum.flat_map(&Map.keys/1)
|> Enum.uniq()
columns = Enum.map(column_keys, &to_string/1)
col_refs = Enum.map(columns, &quote_ident/1) |> Enum.join(", ")
placeholders_per_row = Enum.map(columns, fn _ -> "?" end) |> Enum.join(", ")
values_sql = Enum.map(data_list, fn _ -> "(#{placeholders_per_row})" end) |> Enum.join(", ")
params =
Enum.flat_map(data_list, fn row ->
Enum.map(column_keys, fn k -> Map.get(row, k) end)
end)
conflict_cols =
case conflict_target do
nil -> columns # No explicit target — assume all columns or let DB infer
col when is_atom(col) or is_binary(col) -> [to_string(col)]
cols when is_list(cols) -> Enum.map(cols, &to_string/1)
_ -> columns
end
conflict_ref = Enum.map(conflict_cols, &quote_ident/1) |> Enum.join(", ")
on_conflict_sql =
case on_conflict do
:nothing -> "ON CONFLICT (#{conflict_ref}) DO NOTHING"
_ ->
updates =
columns
|> Enum.reject(&(&1 in conflict_cols))
|> Enum.map(fn c -> "#{quote_ident(c)} = excluded.#{quote_ident(c)}" end)
|> Enum.join(", ")
if updates == "" do
"ON CONFLICT (#{conflict_ref}) DO NOTHING"
else
"ON CONFLICT (#{conflict_ref}) DO UPDATE SET #{updates}"
end
end
returning_cols = sqlite_returning_cols(select)
returning_column_names =
if select == [] do
sqlite_table_columns(repo_mod, table)
else
Enum.map(select, &sqlite_select_col_name/1)
end
sql = """
INSERT INTO #{quote_ident(table)} (#{col_refs})
VALUES #{values_sql}
#{on_conflict_sql}
RETURNING #{returning_cols}
"""
result = Ecto.Adapters.SQL.query!(repo_mod, sql, params)
rows = sqlite_rows_to_maps(result, returning_column_names)
{length(rows), rows}
end
end
defp sqlite_returning_cols([]), do: "*"
defp sqlite_returning_cols(select) do
Enum.map(select, fn
col when is_atom(col) -> ~s("#{col}")
col when is_binary(col) ->
case String.split(col, " AS ", parts: 2) do
[c, alias_name] -> ~s("#{String.trim(c)}" AS "#{String.trim(alias_name)}")
[c] -> ~s("#{String.trim(c)}")
end
end) |> Enum.join(", ")
end
defp sqlite_select_col_name(col) when is_atom(col), do: col
defp sqlite_select_col_name(col) when is_binary(col) do
case String.split(col, " AS ", parts: 2) do
[_c, alias_name] -> String.to_atom(String.trim(alias_name))
[c] -> String.to_atom(String.trim(c))
end
end
defp sqlite_table_columns(repo_mod, table) do
case Ecto.Adapters.SQL.query(repo_mod, "PRAGMA table_info(#{quote_ident(table)})", []) do
{:ok, %{rows: rows}} ->
Enum.map(rows, fn [_cid, name, _type, _notnull, _dflt, _pk] -> String.to_atom(name) end)
_ ->
[]
end
end
defp sqlite_rows_to_maps(result, fallback_cols) do
cols =
case result.columns do
[] -> fallback_cols
cs -> Enum.map(cs, &String.to_atom/1)
end
Enum.map(result.rows, fn row ->
Enum.zip(cols, row) |> Map.new()
end)
end
# PostgreSQL UPDATE/DELETE with row limit via ctid subquery (since Postgres doesn't
# support LIMIT on UPDATE/DELETE directly). Uses the table's system ctid column.
defp postgres_update_with_limit(repo_mod, table, data, where_sql, params, select, limit, returning) do
set_clause =
data
|> Enum.to_list()
|> Enum.with_index(1)
|> Enum.map(fn {{k, _v}, i} -> "#{quote_ident(k)} = $#{i}" end)
|> Enum.join(", ")
data_values = data |> Enum.to_list() |> Enum.map(&elem(&1, 1))
params_offset = length(data_values)
# Build WHERE ... LIMIT n via ctid self-join
table_ident = quote_ident(table)
returning_cols = if select == [], do: "*", else: Enum.map_join(select, ", ", &quote_select_field/1)
returning_sql = if returning, do: " RETURNING #{returning_cols}", else: ""
_returning_column_names =
if select == [] do
["*"] # Postgres populates columns properly
else
Enum.map(select, &sqlite_select_col_name/1)
end
# Re-number placeholders in where_sql: the $1..$N in where_sql need to shift past the SET params
shifted_where_sql = shift_placeholders(where_sql, params_offset)
all_params = data_values ++ params
sql = """
UPDATE #{table_ident} SET #{set_clause}
WHERE ctid IN (
SELECT ctid FROM #{table_ident}
#{shifted_where_sql != "" && "WHERE " || ""}#{shifted_where_sql}
LIMIT #{limit}
)
#{returning_sql}
"""
result = Ecto.Adapters.SQL.query!(repo_mod, sql, all_params)
rows =
if returning_sql != "" do
Enum.map(result.rows, fn row ->
cols = Enum.map(result.columns, &String.to_atom/1)
Enum.zip(cols, row) |> Map.new()
end)
else
nil
end
{length(rows || []), rows}
end
defp postgres_delete_with_limit(repo_mod, table, where_sql, params, select, limit, returning) do
table_ident = quote_ident(table)
returning_cols = if select == [], do: "*", else: Enum.map_join(select, ", ", &quote_select_field/1)
returning_sql = if returning, do: " RETURNING #{returning_cols}", else: ""
sql = """
DELETE FROM #{table_ident}
WHERE ctid IN (
SELECT ctid FROM #{table_ident}
#{where_sql != "" && "WHERE " || ""}#{where_sql}
LIMIT #{limit}
)
#{returning_sql}
"""
result = Ecto.Adapters.SQL.query!(repo_mod, sql, params)
rows =
if returning_sql != "" do
Enum.map(result.rows, fn row ->
cols = Enum.map(result.columns, &String.to_atom/1)
Enum.zip(cols, row) |> Map.new()
end)
else
nil
end
{length(rows || []), rows}
end
# Shift $N placeholders in a SQL string by `offset` (e.g. $1 → $4 if offset=3).
defp shift_placeholders(sql, 0), do: sql
defp shift_placeholders(sql, offset) do
Regex.replace(~r/\$(\d+)/, sql, fn _, n ->
"$#{String.to_integer(n) + offset}"
end)
end
# Convert "$1", "$2" placeholders to "?" for SQLite raw queries
defp sqlite_placeholders(sql), do: Regex.replace(~r/\$\d+/, sql, "?")
defp build_where_for_sqlite(filters, or_filters, not_filters) do
{sql, params} = build_where(filters, or_filters, not_filters, :sqlite)
{sqlite_placeholders(sql), params}
end
defp build_returning_opts(false, _select), do: []
defp build_returning_opts(true, select) do
fields = normalize_select_for_returning(select)
[returning: fields]
end
defp normalize_select_for_returning([]), do: true
defp normalize_select_for_returning(select) do
Enum.map(select, fn
col when is_atom(col) -> col
col when is_binary(col) ->
# strip " AS alias" part if present
col |> String.split(" AS ", parts: 2) |> hd() |> String.trim() |> String.to_atom()
end)
end
defp build_mutation_result(count, rows, nil, _filters) do
case rows do
nil -> {:ok, %{count: count}}
[] -> {:ok, %{count: count}}
rows when is_list(rows) ->
row_maps = Enum.map(rows, &normalize_row/1)
{:ok, %{count: count, data: row_maps}}
end
end
defp build_mutation_result(count, rows, mode, _filters) when mode in [:exact, :planned, :estimated] do
case rows do
nil -> {:ok, %{count: count}, count}
[] -> {:ok, %{count: count}, count}
rows when is_list(rows) ->
row_maps = Enum.map(rows, &normalize_row/1)
{:ok, %{count: count, data: row_maps}, count}
end
end
defp normalize_row(%_{} = struct_row), do: Map.from_struct(struct_row)
defp normalize_row(row) when is_map(row), do: row
# -- Internal Helpers --
defp resolve_repo(%Conn{repo_module: repo_mod, name: name}) do
if name != repo_mod do
repo_mod.put_dynamic_repo(name)
end
repo_mod
end
defp run_select(%Query{
table: table,
schema: schema,
select: select_fields,
filters: filters,
or_filters: or_filters,
not_filters: not_filters,
limit: limit,
offset: offset,
order_by: order_by,
count: count_mode,
conn: conn,
single: single,
maybe_single: maybe_single,
explain_opts: explain_opts,
csv: csv,
geojson: geojson,
head: head
}) do
conn = conn || default_conn()
repo_mod = resolve_repo(conn)
columns_sql =
cond do
head -> "NULL"
explain_opts -> "1"
true ->
if select_fields == [] or select_fields == ["*"] or select_fields == nil do
"*"
else
Enum.map_join(select_fields, ", ", &quote_select_field/1)
end
end
table_sql = if schema, do: "#{quote_ident(schema)}.#{quote_ident(table)}", else: quote_ident(table)
{where_clause, params} = build_where(filters, or_filters, not_filters, conn.adapter)
order_clause = build_order(order_by)
limit_clause = if limit && not head, do: " LIMIT #{limit}", else: ""
offset_clause = if offset && not head, do: " OFFSET #{offset}", else: ""
sql_prefix =
cond do
explain_opts ->
format = Keyword.get(explain_opts, :format, :text)
opts = [
if(explain_opts[:analyze], do: "ANALYZE"),
if(explain_opts[:verbose], do: "VERBOSE"),
if(explain_opts[:settings], do: "SETTINGS"),
if(explain_opts[:buffers], do: "BUFFERS"),
if(explain_opts[:wal], do: "WAL"),
if(format == :json, do: "FORMAT JSON")
] |> Enum.reject(&is_nil/1)
"EXPLAIN (#{Enum.join(opts, ", ")}) "
true -> ""
end
sql_str = "#{sql_prefix}SELECT #{columns_sql} FROM #{table_sql}#{where_clause}#{order_clause}#{limit_clause}#{offset_clause}"
sql_str = normalize_placeholders(sql_str, conn.adapter)
case Ecto.Adapters.SQL.query(repo_mod, sql_str, params) do
{:ok, %{rows: rows, columns: cols}} ->
cond do
csv ->
csv_str = rows_to_csv(cols, rows)
{:ok, csv_str}
geojson ->
{:ok, rows_to_geojson(cols, rows)}
explain_opts && Keyword.get(explain_opts, :format) == :json ->
{:ok, hd(hd(rows))}
explain_opts ->
text = rows |> Enum.map(fn [line] -> line end) |> Enum.join("\n")
{:ok, text}
head ->
result = {:ok, []}
maybe_attach_count(result, table, filters, or_filters, not_filters, count_mode, schema, conn, repo_mod)
true ->
data = Enum.map(rows, fn row -> cols |> Enum.zip(row) |> Map.new() end)
result = {:ok, data}
maybe_attach_count(result, table, filters, or_filters, not_filters, count_mode, schema, conn, repo_mod)
|> handle_single_maybe_single(single, maybe_single)
end
{:error, _} = err ->
err
end
end
defp rows_to_csv(cols, rows) do
header = Enum.map_join(cols, ",", &csv_escape/1)
body =
rows
|> Enum.map(fn row ->
row |> Enum.map(&csv_escape/1) |> Enum.join(",")
end)
|> Enum.join("\n")
if body == "", do: header, else: header <> "\n" <> body
end
defp csv_escape(nil), do: ""
defp csv_escape(val) when is_binary(val) do
if String.contains?(val, [",", "\"", "\n", "\r"]) do
"\"" <> String.replace(val, "\"", "\"\"") <> "\""
else
val
end
end
defp csv_escape(val), do: to_string(val)
defp rows_to_geojson(cols, rows) do
col_atoms = Enum.map(cols, &String.to_atom/1)
geom_idx = Enum.find_index(col_atoms, &(&1 in [:geometry, :geom, :geom_geojson]))
geom_key = if geom_idx, do: Enum.at(col_atoms, geom_idx), else: nil
features =
Enum.map(rows, fn row ->
geom = if geom_idx, do: Enum.at(row, geom_idx), else: nil
geom = if is_binary(geom), do: Jason.decode!(geom), else: geom
props =
col_atoms
|> Enum.zip(row)
|> Enum.reject(fn {k, _v} -> k == geom_key end)
|> Enum.into(%{})
%{
"type" => "Feature",
"geometry" => geom,
"properties" => props
}
end)
%{
"type" => "FeatureCollection",
"features" => features
}
end
defp handle_single_maybe_single({:ok, %{count: c, data: rows}}, true, _maybe) do
case rows do
[row] -> {:ok, row}
[] -> {:error, %RuntimeError{message: "JSON object requested, no rows returned"}}
[_ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, %{count: c, data: rows}}
end
end
defp handle_single_maybe_single({:ok, %{count: c, data: rows}}, _s, true) do
case rows do
[row] -> {:ok, row}
[] -> {:ok, nil}
[_, _ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, %{count: c, data: rows}}
end
end
defp handle_single_maybe_single({:ok, %{count: c, data: rows}, total}, true, _) do
case rows do
[row] -> {:ok, row, total}
[] -> {:error, %RuntimeError{message: "JSON object requested, no rows returned"}}
[_ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, %{count: c, data: rows}, total}
end
end
defp handle_single_maybe_single({:ok, %{count: c, data: rows}, total}, _s, true) do
case rows do
[row] -> {:ok, row, total}
[] -> {:ok, nil, total}
[_, _ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, %{count: c, data: rows}, total}
end
end
defp handle_single_maybe_single({:ok, data}, true, _maybe) do
case data do
[row] -> {:ok, row}
[] -> {:error, %RuntimeError{message: "JSON object requested, no rows returned"}}
[_ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, data}
end
end
defp handle_single_maybe_single({:ok, data}, _s, true) do
case data do
[row] -> {:ok, row}
[] -> {:ok, nil}
[_, _ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, data}
end
end
defp handle_single_maybe_single({:ok, data, count}, true, _) do
case data do
[row] -> {:ok, row, count}
[] -> {:error, %RuntimeError{message: "JSON object requested, no rows returned"}}
[_ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, data, count}
end
end
defp handle_single_maybe_single({:ok, data, count}, _s, true) do
case data do
[row] -> {:ok, row, count}
[] -> {:ok, nil, count}
[_, _ | _] -> {:error, %RuntimeError{message: "JSON object requested, multiple rows returned"}}
_other -> {:ok, data, count}
end
end
defp handle_single_maybe_single(result, _, _), do: result
defp maybe_attach_count({:ok, data}, _table, _filters, _or_filters, _not_filters, nil, _schema, _conn, _repo_mod) do
{:ok, data}
end
defp maybe_attach_count({:ok, data}, table, filters, or_filters, not_filters, mode, schema, conn, repo_mod) do
{where_clause, params} = build_where(filters, or_filters, not_filters, conn.adapter)
table_sql = if schema, do: "#{quote_ident(schema)}.#{quote_ident(table)}", else: quote_ident(table)
count_sql =
case mode do
:exact -> "SELECT COUNT(*) FROM #{table_sql}#{where_clause}"
:planned -> "SELECT reltuples::bigint FROM pg_class WHERE relname = $1"
:estimated -> "EXPLAIN SELECT * FROM #{table_sql}#{where_clause}"
end
count_sql = normalize_placeholders(count_sql, conn.adapter)
count_params = if mode == :planned, do: [table], else: params
total =
case Ecto.Adapters.SQL.query(repo_mod, count_sql, count_params) do
{:ok, %{rows: [[count]]}} when is_integer(count) or is_number(count) -> trunc(count)
{:ok, %{rows: [[count_str]]}} when is_binary(count_str) ->
case Integer.parse(count_str) do
{n, _} -> n
_ -> length(data)
end
{:ok, %{rows: rows}} when is_list(rows) and mode == :estimated ->
extract_estimated_rows(rows)
_ -> length(data)
end
{:ok, data, total}
end
defp extract_estimated_rows(rows) do
rows
|> Enum.map(fn [line] -> to_string(line) end)
|> Enum.find_value(0, fn line ->
case Regex.run(~r/rows=(\d+)/, line) do
[_, n] -> String.to_integer(n)
_ -> nil
end
end)
end
# -- WHERE --
defp build_where([], [], [], _adapter), do: {"", []}
defp build_where(filters, or_filters, not_filters, adapter) do
{and_clauses, params, idx} =
Enum.reduce(filters, {[], [], 1}, fn filter, {clauses, params, idx} ->
{clause, new_params, next_idx} = filter_to_sql(filter, idx, adapter)
{clauses ++ [clause], params ++ new_params, next_idx}
end)
{not_clauses, params, idx} =
Enum.reduce(not_filters, {and_clauses, params, idx}, fn filter, {clauses, params, idx} ->
{clause, new_params, next_idx} = filter_to_sql(filter, idx, adapter)
{clauses ++ ["NOT (#{clause})"], params ++ new_params, next_idx}
end)
{or_groups, params, _idx} =
Enum.reduce(or_filters, {not_clauses, params, idx}, fn
{group, referenced_table: _ref_table}, {clauses, params, idx} ->
# PostgREST supports ${referencedTable}.or but in raw SQL we just filter the main table
{group_clauses, group_params, next_idx} =
Enum.reduce(group, {[], params, idx}, fn filter, {gcs, ps, i} ->
{clause, new_ps, ni} = filter_to_sql(filter, i, adapter)
{gcs ++ [clause], ps ++ new_ps, ni}
end)
or_sql = "(#{Enum.join(group_clauses, " OR ")})"
{clauses ++ [or_sql], group_params, next_idx}
group, {clauses, params, idx} ->
{group_clauses, group_params, next_idx} =
Enum.reduce(group, {[], params, idx}, fn filter, {gcs, ps, i} ->
{clause, new_ps, ni} = filter_to_sql(filter, i, adapter)
{gcs ++ [clause], ps ++ new_ps, ni}
end)
or_sql = "(#{Enum.join(group_clauses, " OR ")})"
{clauses ++ [or_sql], group_params, next_idx}
end)
{" WHERE " <> Enum.join(or_groups, " AND "), params}
end
# -- filter_to_sql --
defp filter_to_sql({:eq, col, val}, idx, _), do: {"#{quote_ident(col)} = $#{idx}", [val], idx + 1}
defp filter_to_sql({:neq, col, val}, idx, _), do: {"#{quote_ident(col)} != $#{idx}", [val], idx + 1}
defp filter_to_sql({:gt, col, val}, idx, _), do: {"#{quote_ident(col)} > $#{idx}", [val], idx + 1}
defp filter_to_sql({:lt, col, val}, idx, _), do: {"#{quote_ident(col)} < $#{idx}", [val], idx + 1}
defp filter_to_sql({:gte, col, val}, idx, _), do: {"#{quote_ident(col)} >= $#{idx}", [val], idx + 1}
defp filter_to_sql({:lte, col, val}, idx, _), do: {"#{quote_ident(col)} <= $#{idx}", [val], idx + 1}
defp filter_to_sql({:like, col, val}, idx, _), do: {"#{quote_ident(col)} LIKE $#{idx}", [val], idx + 1}
defp filter_to_sql({:nlike, col, val}, idx, _), do: {"#{quote_ident(col)} NOT LIKE $#{idx}", [val], idx + 1}
defp filter_to_sql({:ilike, col, val}, idx, :sqlite), do: {"#{quote_ident(col)} LIKE $#{idx}", [val], idx + 1}
defp filter_to_sql({:ilike, col, val}, idx, _), do: {"#{quote_ident(col)} ILIKE $#{idx}", [val], idx + 1}
defp filter_to_sql({:nilike, col, val}, idx, :sqlite), do: {"#{quote_ident(col)} NOT LIKE $#{idx}", [val], idx + 1}
defp filter_to_sql({:nilike, col, val}, idx, _), do: {"#{quote_ident(col)} NOT ILIKE $#{idx}", [val], idx + 1}
defp filter_to_sql({:is, col, val}, idx, _) do
cond do
is_nil(val) or val == :null -> {"#{quote_ident(col)} IS NULL", [], idx}
val == true -> {"#{quote_ident(col)} IS TRUE", [], idx}
val == false -> {"#{quote_ident(col)} IS FALSE", [], idx}
val == :unknown -> {"#{quote_ident(col)} IS UNKNOWN", [], idx}
true -> {"#{quote_ident(col)} = $#{idx}", [val], idx + 1}
end
end
defp filter_to_sql({:in, col, values}, idx, _) do
placeholders = Enum.map_join(0..(length(values) - 1), ", ", fn i -> "$#{idx + i}" end)
{"#{quote_ident(col)} IN (#{placeholders})", values, idx + length(values)}
end
# Full text search — parameterize the config value to avoid SQL injection.
defp filter_to_sql({:fts, {col, _config}, query_text}, idx, :sqlite) do
{"#{quote_ident(col)} LIKE $#{idx}", ["%#{query_text}%"], idx + 1}
end
defp filter_to_sql({:fts, {col, config}, query_text}, idx, _) do
{"to_tsvector($#{idx}::regconfig, #{quote_ident(col)}) @@ plainto_tsquery($#{idx}::regconfig, $#{idx + 1})",
[config, query_text], idx + 2}
end
defp filter_to_sql({:plfts, {col, _config}, query_text}, idx, :sqlite) do
{"#{quote_ident(col)} LIKE $#{idx}", ["%#{query_text}%"], idx + 1}
end
defp filter_to_sql({:plfts, {col, config}, query_text}, idx, _) do
{"to_tsvector($#{idx}::regconfig, #{quote_ident(col)}) @@ plainto_tsquery($#{idx}::regconfig, $#{idx + 1})",
[config, query_text], idx + 2}
end
defp filter_to_sql({:phfts, {col, _config}, query_text}, idx, :sqlite) do
{"#{quote_ident(col)} LIKE $#{idx}", ["%#{query_text}%"], idx + 1}
end
defp filter_to_sql({:phfts, {col, config}, query_text}, idx, _) do
{"to_tsvector($#{idx}::regconfig, #{quote_ident(col)}) @@ phraseto_tsquery($#{idx}::regconfig, $#{idx + 1})",
[config, query_text], idx + 2}
end
defp filter_to_sql({:wfts, {col, _config}, query_text}, idx, :sqlite) do
{"#{quote_ident(col)} LIKE $#{idx}", ["%#{query_text}%"], idx + 1}
end
defp filter_to_sql({:wfts, {col, config}, query_text}, idx, _) do
{"to_tsvector($#{idx}::regconfig, #{quote_ident(col)}) @@ websearch_to_tsquery($#{idx}::regconfig, $#{idx + 1})",
[config, query_text], idx + 2}
end
# Range / array operators
defp filter_to_sql({:cs, col, values}, idx, :sqlite) do
{placeholders, params, next_idx} =
Enum.reduce(values, {[], [], idx}, fn v, {phs, ps, i} ->
{phs ++ ["#{quote_ident(col)} LIKE $#{i}"], ps ++ ["%#{inspect(v)}%"], i + 1}
end)
{"(" <> Enum.join(placeholders, " AND ") <> ")", params, next_idx}
end
defp filter_to_sql({:cs, col, values}, idx, _) do
{"#{quote_ident(col)} @> $#{idx}", [values], idx + 1}
end
defp filter_to_sql({:cd, col, values}, idx, :sqlite) do
{"#{quote_ident(col)} = $#{idx}", [values], idx + 1}
end
defp filter_to_sql({:cd, col, values}, idx, _) do
{"#{quote_ident(col)} <@ $#{idx}", [values], idx + 1}
end
defp filter_to_sql({:ov, col, values}, idx, :sqlite) do
{"#{quote_ident(col)} = $#{idx}", [values], idx + 1}
end
defp filter_to_sql({:ov, col, values}, idx, _) do
{"#{quote_ident(col)} && $#{idx}", [values], idx + 1}
end
defp filter_to_sql({:sl, col, value}, idx, _) do
{"#{quote_ident(col)} << $#{idx}", [value], idx + 1}
end
defp filter_to_sql({:sr, col, value}, idx, _) do
{"#{quote_ident(col)} >> $#{idx}", [value], idx + 1}
end
defp filter_to_sql({:nxr, col, value}, idx, _) do
{"#{quote_ident(col)} &> $#{idx}", [value], idx + 1}
end
defp filter_to_sql({:nxl, col, value}, idx, _) do
{"#{quote_ident(col)} &< $#{idx}", [value], idx + 1}
end
defp filter_to_sql({:adj, col, value}, idx, _) do
{"#{quote_ident(col)} -|- $#{idx}", [value], idx + 1}
end
# Fallback
defp filter_to_sql({op, col, val}, idx, _) do
{"#{quote_ident(col)} #{Atom.to_string(op)} $#{idx}", [val], idx + 1}
end
# -- ORDER --
defp normalize_direction(dir) do
case Atom.to_string(dir) |> String.downcase() do
"desc" -> "DESC"
"descending" -> "DESC"
"dsc" -> "DESC"
_ -> "ASC"
end
end
defp build_order([]), do: ""
defp build_order(order_by) do
clauses =
Enum.map(order_by, fn entry ->
{actual, _ref_opts} = case entry do
{clause, referenced_table: _rt} -> {clause, true}
clause -> {clause, false}
end
case actual do
{dir, col} ->
direction = normalize_direction(dir)
"#{quote_ident(col)} #{direction}"
{dir, col, opts} ->
direction = normalize_direction(dir)
base = "#{quote_ident(col)} #{direction}"
cond do
Keyword.has_key?(opts, :nulls_first) ->
if opts[:nulls_first], do: "#{base} NULLS FIRST", else: "#{base} NULLS LAST"
opts[:nulls_last] -> "#{base} NULLS LAST"
true -> base
end
end
end)
" ORDER BY " <> Enum.join(clauses, ", ")
end
defp normalize_placeholders(sql, :sqlite), do: Regex.replace(~r/\$\d+/, sql, "?")
defp normalize_placeholders(sql, _), do: sql
# Quote SQL identifiers safely (table names, column names, schema names).
# Follows PostgreSQL/SQLite double-quote rules: quote ", escape embedded " as "".
defp quote_ident(ident) when is_atom(ident), do: quote_ident(Atom.to_string(ident))
defp quote_ident(ident) when is_binary(ident) do
escaped = String.replace(ident, "\"", "\"\"")
"\"#{escaped}\""
end
# Quote a single select field, which may be in "column AS alias" form or a raw atom.
defp quote_select_field(field) when is_atom(field), do: quote_ident(field)
defp quote_select_field(field) when is_binary(field) do
case String.split(field, " AS ", parts: 2) do
[col, alias_name] ->
"#{quote_ident(String.trim(col))} AS #{quote_ident(String.trim(alias_name))}"
[_] ->
# Could be a raw expression like "count(*)" — if it contains special chars, pass through.
# Otherwise quote as identifier.
if String.match?(field, ~r/^[A-Za-z_][A-Za-z0-9_\.]*$/) do
quote_ident(field)
else
field
end
end
end
# -- Operator normalization (aliases) --
defp normalize_operator(op) do
case op do
:equals -> :eq
:not_equals -> :neq
:greater_than -> :gt
:greater_than_or_equal -> :gte
:less_than -> :lt
:less_than_or_equal -> :lte
:in_list -> :in
:is_null -> :is
:not_like -> :nlike
:not_ilike -> :nilike
:contains -> :cs
:contained_in -> :cd
:containedBy -> :cd
:overlaps -> :ov
:strictly_left -> :sl
:strictly_right -> :sr
:not_extend_right -> :nxr
:not_extend_left -> :nxl
:adjacent -> :adj
:text_search -> :fts
:full_text_search -> :fts
:phrase_full_text_search -> :phfts
:plain_full_text_search -> :plfts
:web_search -> :wfts
other -> other
end
end
# -- Ecto filter application (for update/delete) --
defp apply_filter(query, {:eq, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) == ^val)
end
defp apply_filter(query, {:neq, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) != ^val)
end
defp apply_filter(query, {:gt, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) > ^val)
end
defp apply_filter(query, {:lt, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) < ^val)
end
defp apply_filter(query, {:like, col, pattern}, _) do
Ecto.Query.where(query, [t], like(field(t, ^col), ^pattern))
end
defp apply_filter(query, {:nlike, col, pattern}, _) do
Ecto.Query.where(query, [t], not like(field(t, ^col), ^pattern))
end
defp apply_filter(query, {:ilike, col, pattern}, :sqlite) do
Ecto.Query.where(query, [t], like(field(t, ^col), ^pattern))
end
defp apply_filter(query, {:ilike, col, pattern}, _) do
Ecto.Query.where(query, [t], ilike(field(t, ^col), ^pattern))
end
defp apply_filter(query, {:nilike, col, pattern}, :sqlite) do
Ecto.Query.where(query, [t], not like(field(t, ^col), ^pattern))
end
defp apply_filter(query, {:nilike, col, pattern}, _) do
Ecto.Query.where(query, [t], not ilike(field(t, ^col), ^pattern))
end
defp apply_filter(query, {:gte, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) >= ^val)
end
defp apply_filter(query, {:lte, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) <= ^val)
end
defp apply_filter(query, {:is, col, val}, _) do
if is_nil(val) or val == :null do
Ecto.Query.where(query, [t], is_nil(field(t, ^col)))
else
Ecto.Query.where(query, [t], field(t, ^col) == ^val)
end
end
defp apply_filter(query, {:in, col, values}, _) do
Ecto.Query.where(query, [t], field(t, ^col) in ^values)
end
defp apply_filter(query, {:cs, col, values}, _) do
# PostgreSQL @> operator (contains)
Ecto.Query.where(query, [t], fragment("? @> ?", field(t, ^col), type(^values, {:array, :any})))
end
defp apply_filter(query, {:cd, col, values}, _) do
# PostgreSQL <@ operator (contained by)
Ecto.Query.where(query, [t], fragment("? <@ ?", field(t, ^col), type(^values, {:array, :any})))
end
defp apply_filter(query, {:ov, col, values}, _) do
Ecto.Query.where(query, [t], fragment("? && ?", field(t, ^col), type(^values, {:array, :any})))
end
defp apply_filter(query, {:sl, col, value}, _) do
Ecto.Query.where(query, [t], fragment("? << ?", field(t, ^col), ^value))
end
defp apply_filter(query, {:sr, col, value}, _) do
Ecto.Query.where(query, [t], fragment("? >> ?", field(t, ^col), ^value))
end
defp apply_filter(query, {:nxr, col, value}, _) do
Ecto.Query.where(query, [t], fragment("? &> ?", field(t, ^col), ^value))
end
defp apply_filter(query, {:nxl, col, value}, _) do
Ecto.Query.where(query, [t], fragment("? &< ?", field(t, ^col), ^value))
end
defp apply_filter(query, {:adj, col, value}, _) do
Ecto.Query.where(query, [t], fragment("? -|- ?", field(t, ^col), ^value))
end
defp apply_filter(query, {:fts, {col, config}, query_text}, _) do
Ecto.Query.where(
query,
[t],
fragment("to_tsvector(?, ?) @@ plainto_tsquery(?, ?)", ^config, field(t, ^col), ^config, ^query_text)
)
end
defp apply_filter(query, {:plfts, {col, config}, query_text}, _) do
Ecto.Query.where(
query,
[t],
fragment("to_tsvector(?, ?) @@ plainto_tsquery(?, ?)", ^config, field(t, ^col), ^config, ^query_text)
)
end
defp apply_filter(query, {:phfts, {col, config}, query_text}, _) do
Ecto.Query.where(
query,
[t],
fragment("to_tsvector(?, ?) @@ phraseto_tsquery(?, ?)", ^config, field(t, ^col), ^config, ^query_text)
)
end
defp apply_filter(query, {:wfts, {col, config}, query_text}, _) do
Ecto.Query.where(
query,
[t],
fragment("to_tsvector(?, ?) @@ websearch_to_tsquery(?, ?)", ^config, field(t, ^col), ^config, ^query_text)
)
end
# Catch-all: UNSUPPORTED filter operator for Ecto update/delete.
# Raising here rather than silently dropping the filter prevents accidental mass updates/deletes.
defp apply_filter(_query, filter, _adapter) do
raise "Unsupported filter operator for Ecto-based update/delete: #{inspect(filter)}. " <>
"Supported operators are: eq, neq, gt, lt, gte, lte, like, nlike, ilike, nilike, is, in, " <>
"and array/range operators (cs, cd, ov, sl, sr, nxr, nxl, adj), plus fts/plfts/phfts/wfts."
end
# -- Negated filters for update/delete via Ecto --
defp apply_not_filter(query, {:eq, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) != ^val)
end
defp apply_not_filter(query, {:neq, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) == ^val)
end
defp apply_not_filter(query, {:gt, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) <= ^val)
end
defp apply_not_filter(query, {:lt, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) >= ^val)
end
defp apply_not_filter(query, {:gte, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) < ^val)
end
defp apply_not_filter(query, {:lte, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) > ^val)
end
defp apply_not_filter(query, {:like, col, pattern}, _) do
Ecto.Query.where(query, [t], not like(field(t, ^col), ^pattern))
end
defp apply_not_filter(query, {:nlike, col, pattern}, _) do
Ecto.Query.where(query, [t], like(field(t, ^col), ^pattern))
end
defp apply_not_filter(query, {:ilike, col, pattern}, :sqlite) do
Ecto.Query.where(query, [t], not like(field(t, ^col), ^pattern))
end
defp apply_not_filter(query, {:ilike, col, pattern}, _) do
Ecto.Query.where(query, [t], not ilike(field(t, ^col), ^pattern))
end
defp apply_not_filter(query, {:nilike, col, pattern}, :sqlite) do
Ecto.Query.where(query, [t], like(field(t, ^col), ^pattern))
end
defp apply_not_filter(query, {:nilike, col, pattern}, _) do
Ecto.Query.where(query, [t], ilike(field(t, ^col), ^pattern))
end
defp apply_not_filter(query, {:is, col, nil}, _) do
Ecto.Query.where(query, [t], not is_nil(field(t, ^col)))
end
defp apply_not_filter(query, {:is, col, val}, _) do
Ecto.Query.where(query, [t], field(t, ^col) != ^val)
end
defp apply_not_filter(query, {:in, col, values}, _) do
Ecto.Query.where(query, [t], field(t, ^col) not in ^values)
end
# Catch-all: UNSUPPORTED NOT filter operator for Ecto update/delete.
defp apply_not_filter(_query, filter, _adapter) do
raise "Unsupported NOT filter operator for Ecto-based update/delete: #{inspect(filter)}"
end
# -- OR filter groups for update/delete via Ecto --
defp apply_or_filter_group(query, {group, referenced_table: _ref}, adapter) do
apply_or_filter_group(query, group, adapter)
end
defp apply_or_filter_group(query, group, adapter) when is_list(group) and group != [] do
# Build a dynamic OR from the filter group
dynamic =
Enum.reduce(group, nil, fn
filter, nil ->
filter_to_dynamic(filter, adapter)
filter, acc ->
Ecto.Query.dynamic([t], ^acc or ^filter_to_dynamic(filter, adapter))
end)
Ecto.Query.where(query, [_t], ^dynamic)
end
defp apply_or_filter_group(query, [], _adapter), do: query
defp filter_to_dynamic({:eq, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) == ^val)
defp filter_to_dynamic({:neq, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) != ^val)
defp filter_to_dynamic({:gt, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) > ^val)
defp filter_to_dynamic({:lt, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) < ^val)
defp filter_to_dynamic({:gte, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) >= ^val)
defp filter_to_dynamic({:lte, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) <= ^val)
defp filter_to_dynamic({:like, col, pattern}, _), do: Ecto.Query.dynamic([t], like(field(t, ^col), ^pattern))
defp filter_to_dynamic({:nlike, col, pattern}, _), do: Ecto.Query.dynamic([t], not like(field(t, ^col), ^pattern))
defp filter_to_dynamic({:ilike, col, pattern}, :sqlite), do: Ecto.Query.dynamic([t], like(field(t, ^col), ^pattern))
defp filter_to_dynamic({:ilike, col, pattern}, _), do: Ecto.Query.dynamic([t], ilike(field(t, ^col), ^pattern))
defp filter_to_dynamic({:nilike, col, pattern}, :sqlite), do: Ecto.Query.dynamic([t], not like(field(t, ^col), ^pattern))
defp filter_to_dynamic({:nilike, col, pattern}, _), do: Ecto.Query.dynamic([t], not ilike(field(t, ^col), ^pattern))
defp filter_to_dynamic({:is, col, nil}, _), do: Ecto.Query.dynamic([t], is_nil(field(t, ^col)))
defp filter_to_dynamic({:is, col, val}, _), do: Ecto.Query.dynamic([t], field(t, ^col) == ^val)
defp filter_to_dynamic({:in, col, values}, _), do: Ecto.Query.dynamic([t], field(t, ^col) in ^values)
# Catch-all: UNSUPPORTED filter operator for Ecto-based update/delete OR groups.
# Raise rather than silently returning true (which would disable the filter).
defp filter_to_dynamic(filter, _) do
raise "Unsupported filter operator for Ecto-based update/delete OR groups: #{inspect(filter)}"
end
end