Skip to content

Commit 7883961

Browse files
committed
replace exqlite with esqlite
1 parent 8fe0a37 commit 7883961

File tree

8 files changed

+299
-20
lines changed

8 files changed

+299
-20
lines changed

.changeset/modern-apes-complain.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@core/sync-service': patch
3+
---
4+
5+
Replace exqlite SQLite driver with esqlite

packages/sync-service/lib/electric/shape_cache/shape_status/shape_db/connection.ex

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,8 @@
11
defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Connection do
22
@moduledoc false
33

4-
alias Exqlite.Sqlite3
4+
# alias Exqlite.Sqlite3
5+
alias Electric.ShapeCache.ShapeStatus.ShapeDb.Sqlite3
56
alias Electric.ShapeCache.ShapeStatus.ShapeDb.Query
67
alias Electric.ShapeCache.ShapeStatus.ShapeDb.PoolRegistry
78
alias Electric.Telemetry.OpenTelemetry
@@ -63,7 +64,16 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Connection do
6364
"PRAGMA user_version=#{@schema_version}"
6465
]
6566

66-
defguardp is_raw_connection(conn) when is_reference(conn)
67+
# exqlite represents connections and statements as plain references.
68+
# esqlite wraps them in {:esqlite3, ref} and {:esqlite3_stmt, ref} records.
69+
# Both guards accept either form so the module works with either backend.
70+
defguardp is_raw_connection(conn)
71+
when is_reference(conn) or
72+
(is_tuple(conn) and tuple_size(conn) == 2 and elem(conn, 0) == :esqlite3)
73+
74+
defguardp is_prepared_statement(stmt)
75+
when is_reference(stmt) or
76+
(is_tuple(stmt) and tuple_size(stmt) == 2 and elem(stmt, 0) == :esqlite3_stmt)
6777

6878
defstruct [:conn, :stmts]
6979

@@ -305,7 +315,8 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Connection do
305315
end
306316
end
307317

308-
def fetch_one(conn, stmt, binds) when is_raw_connection(conn) and is_reference(stmt) do
318+
def fetch_one(conn, stmt, binds)
319+
when is_raw_connection(conn) and is_prepared_statement(stmt) do
309320
with :ok <- Sqlite3.bind(stmt, binds) do
310321
case Sqlite3.step(conn, stmt) do
311322
{:row, row} ->
@@ -333,7 +344,8 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Connection do
333344
end
334345
end
335346

336-
def fetch_all(conn, stmt, binds) when is_raw_connection(conn) and is_reference(stmt) do
347+
def fetch_all(conn, stmt, binds)
348+
when is_raw_connection(conn) and is_prepared_statement(stmt) do
337349
with :ok <- Sqlite3.bind(stmt, binds),
338350
{:ok, rows} <- Sqlite3.fetch_all(conn, stmt),
339351
:ok <- Sqlite3.reset(stmt) do
@@ -348,13 +360,14 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Connection do
348360
end
349361

350362
def fetch_all(conn, stmt, binds, mapper_fun)
351-
when is_raw_connection(conn) and is_reference(stmt) do
363+
when is_raw_connection(conn) and is_prepared_statement(stmt) do
352364
with {:ok, rows} <- fetch_all(conn, stmt, binds) do
353365
{:ok, Enum.map(rows, mapper_fun)}
354366
end
355367
end
356368

357-
def modify(conn, stmt, binds) when is_raw_connection(conn) and is_reference(stmt) do
369+
def modify(conn, stmt, binds)
370+
when is_raw_connection(conn) and is_prepared_statement(stmt) do
358371
with :ok <- Sqlite3.bind(stmt, binds),
359372
:done <- Sqlite3.step(conn, stmt),
360373
{:ok, changes} <- Sqlite3.changes(conn),
@@ -444,7 +457,7 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Connection do
444457
end
445458

446459
def stream_query(conn, stmt, row_mapper_fun)
447-
when is_raw_connection(conn) and is_reference(stmt) do
460+
when is_raw_connection(conn) and is_prepared_statement(stmt) do
448461
Stream.resource(
449462
fn -> {:cont, conn, stmt} end,
450463
fn

packages/sync-service/lib/electric/shape_cache/shape_status/shape_db/query.ex

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,8 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Query do
5656

5757
alias Electric.ShapeCache.ShapeStatus.ShapeDb.Connection, as: Conn
5858

59-
alias Exqlite.Sqlite3
59+
# alias Exqlite.Sqlite3
60+
alias Electric.ShapeCache.ShapeStatus.ShapeDb.Sqlite3
6061

6162
import Conn,
6263
only: [
Lines changed: 206 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,206 @@
1+
defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Sqlite3 do
2+
@moduledoc """
3+
Drop-in shim over `:esqlite3` that mirrors the subset of the `Exqlite.Sqlite3` API
4+
used by `Connection` and `Query`.
5+
6+
Only `Connection` and `Query` hold an `alias` to this module. All other code
7+
continues to call those two modules unchanged, so swapping the underlying SQLite
8+
NIF is fully contained here.
9+
10+
## API mapping
11+
12+
| Exqlite.Sqlite3 | :esqlite3 / notes |
13+
|-------------------------------------|--------------------------------------------|
14+
| open(path, opts) | open(uri) – opts encoded as URI params |
15+
| close(conn) | close(conn) |
16+
| execute(conn, sql) | exec(conn, sql) |
17+
| prepare(conn, sql) | prepare(conn, sql) |
18+
| release(conn, stmt) | no-op – GC'd by esqlite |
19+
| bind(stmt, binds) | bind(stmt, binds) |
20+
| step(conn, stmt) | step(stmt) – conn arg dropped |
21+
| reset(stmt) | reset(stmt) |
22+
| fetch_all(conn, stmt) | fetchall(stmt) |
23+
| changes(conn) | {:ok, changes(conn)} |
24+
| multi_step(conn, stmt) | step loop; returns {:rows, rows}/{:done, rows} |
25+
| enable_load_extension(conn, bool) | not supported – always returns error |
26+
| bind_parameter_count(stmt) | column_names heuristic (explain only) |
27+
"""
28+
29+
# ── Types ──────────────────────────────────────────────────────────────────
30+
31+
@type connection :: :esqlite3.esqlite3()
32+
@type statement :: :esqlite3.esqlite3_stmt()
33+
34+
# ── Connection lifecycle ───────────────────────────────────────────────────
35+
36+
@doc """
37+
Opens a SQLite database.
38+
39+
`opts` follows the exqlite convention:
40+
- `[mode: [:readonly, :nomutex]]` → opens as `file:<path>?mode=ro`
41+
- `[]` (default) → opens as `file:<path>?mode=rwc`
42+
43+
The `:memory:` path is passed through unchanged.
44+
"""
45+
@spec open(String.t(), keyword()) :: {:ok, connection()} | {:error, term()}
46+
def open(path, opts \\ []) do
47+
uri = build_uri(path, opts)
48+
:esqlite3.open(String.to_charlist(uri))
49+
end
50+
51+
@spec close(connection()) :: :ok | {:error, term()}
52+
def close(conn) do
53+
:esqlite3.close(conn)
54+
end
55+
56+
# ── DDL / raw execution ────────────────────────────────────────────────────
57+
58+
@doc "Execute a raw SQL statement (no results returned)."
59+
@spec execute(connection(), String.t()) :: :ok | {:error, term()}
60+
def execute(conn, sql) do
61+
:esqlite3.exec(conn, sql)
62+
end
63+
64+
# ── Prepared statements ────────────────────────────────────────────────────
65+
66+
@spec prepare(connection(), String.t()) :: {:ok, statement()} | {:error, term()}
67+
def prepare(conn, sql) do
68+
:esqlite3.prepare(conn, sql)
69+
end
70+
71+
@doc "Release a prepared statement. esqlite relies on GC; this is a no-op."
72+
@spec release(connection(), statement()) :: :ok
73+
def release(_conn, _stmt), do: :ok
74+
75+
@doc """
76+
Bind positional or named parameters to a prepared statement.
77+
78+
Accepts the exqlite bind list format including `{:blob, value}` tagged tuples,
79+
plain integers, binaries, and named `{"@name", value}` pairs.
80+
"""
81+
@spec bind(statement(), list()) :: :ok | {:error, term()}
82+
def bind(stmt, binds) do
83+
converted = Enum.map(binds, &convert_bind/1)
84+
:esqlite3.bind(stmt, converted)
85+
end
86+
87+
@doc """
88+
Step a prepared statement once.
89+
90+
Returns `{:row, row}` or `:done` (matching the exqlite contract).
91+
The `conn` argument is accepted for API compatibility but ignored.
92+
"""
93+
@spec step(connection(), statement()) :: {:row, list()} | :done | {:error, term()}
94+
def step(_conn, stmt) do
95+
case :esqlite3.step(stmt) do
96+
:"$done" -> :done
97+
row when is_list(row) -> {:row, row}
98+
{:error, _} = err -> err
99+
end
100+
end
101+
102+
@spec reset(statement()) :: :ok | {:error, term()}
103+
def reset(stmt) do
104+
:esqlite3.reset(stmt)
105+
end
106+
107+
@doc "Fetch all remaining rows from a prepared statement."
108+
@spec fetch_all(connection(), statement()) :: {:ok, list(list())} | {:error, term()}
109+
def fetch_all(_conn, stmt) do
110+
case :esqlite3.fetchall(stmt) do
111+
rows when is_list(rows) -> {:ok, rows}
112+
{:error, _} = err -> err
113+
end
114+
end
115+
116+
@doc "Return `{:ok, n}` for the number of rows changed by the last DML statement."
117+
@spec changes(connection()) :: {:ok, non_neg_integer()}
118+
def changes(conn) do
119+
{:ok, :esqlite3.changes(conn)}
120+
end
121+
122+
@doc """
123+
Step through a prepared statement in chunks.
124+
125+
Returns `{:rows, rows}` when there are more rows to fetch, or
126+
`{:done, rows}` when the cursor is exhausted.
127+
128+
The `conn` argument is accepted for API compatibility but ignored.
129+
The chunk size matches exqlite's default (50 rows per call).
130+
"""
131+
@spec multi_step(connection(), statement()) ::
132+
{:rows, list(list())} | {:done, list(list())} | {:error, term()}
133+
def multi_step(_conn, stmt, chunk_size \\ 50) do
134+
do_multi_step(stmt, chunk_size, [])
135+
end
136+
137+
defp do_multi_step(_stmt, 0, acc) do
138+
{:rows, Enum.reverse(acc)}
139+
end
140+
141+
defp do_multi_step(stmt, remaining, acc) do
142+
case :esqlite3.step(stmt) do
143+
:"$done" ->
144+
{:done, Enum.reverse(acc)}
145+
146+
row when is_list(row) ->
147+
do_multi_step(stmt, remaining - 1, [row | acc])
148+
149+
{:error, _} = err ->
150+
err
151+
end
152+
end
153+
154+
@doc """
155+
Enable or disable SQLite extension loading.
156+
157+
esqlite does not expose `sqlite3_enable_load_extension`.
158+
Returns `{:error, :not_supported}` so callers can handle gracefully.
159+
"""
160+
@spec enable_load_extension(connection(), boolean()) :: :ok | {:error, :not_supported}
161+
def enable_load_extension(_conn, _enable), do: {:error, :not_supported}
162+
163+
@doc """
164+
Return the number of bind parameters in a prepared statement.
165+
166+
Used only by the `explain/2` diagnostic path. esqlite does not expose
167+
`sqlite3_bind_parameter_count` directly, so we derive it from column names
168+
of the statement. For `EXPLAIN QUERY PLAN` usage the count just needs to
169+
be non-negative; we fall back to 0.
170+
"""
171+
@spec bind_parameter_count(statement()) :: non_neg_integer()
172+
def bind_parameter_count(_stmt) do
173+
# esqlite does not expose sqlite3_bind_parameter_count.
174+
# The explain path just needs a list of empty-string binds for EXPLAIN
175+
# QUERY PLAN to succeed; returning 0 is safe for that path.
176+
0
177+
end
178+
179+
# ── Private helpers ────────────────────────────────────────────────────────
180+
181+
# Build a SQLite URI from a file path and exqlite-style opts.
182+
defp build_uri(":memory:", _opts), do: "file:memory?mode=memory&cache=shared"
183+
184+
defp build_uri(path, opts) do
185+
mode =
186+
case Keyword.get(opts, :mode, []) do
187+
modes when is_list(modes) ->
188+
if :readonly in modes, do: "ro", else: "rwc"
189+
190+
:readonly ->
191+
"ro"
192+
193+
_ ->
194+
"rwc"
195+
end
196+
197+
"file:#{URI.encode(path)}?mode=#{mode}"
198+
end
199+
200+
# Convert an exqlite bind value to an esqlite bind value.
201+
# esqlite's bind/2 supports: integers, floats, binaries (text), and
202+
# {:blob, binary} tuples for BLOBs. nil/null map to undefined.
203+
defp convert_bind(nil), do: :undefined
204+
defp convert_bind(:null), do: :undefined
205+
defp convert_bind(value), do: value
206+
end

packages/sync-service/lib/electric/shape_cache/shape_status/shape_db/statistics.ex

Lines changed: 53 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,11 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Statistics do
2121

2222
@measurement_period 60_000
2323

24+
defstruct total_memory: 0,
25+
page_cache_overflow: 0,
26+
disk_size: 0,
27+
data_size: 0
28+
2429
def name(stack_ref) do
2530
Electric.ProcessRegistry.name(stack_ref, __MODULE__)
2631
end
@@ -33,6 +38,23 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Statistics do
3338
GenServer.call(name(stack_id), :statistics)
3439
end
3540

41+
@doc """
42+
Returns a map describing which stat categories are currently operational.
43+
44+
%{disk: true, memory: false}
45+
46+
`disk` is `true` when the `dbstat` virtual table is available in the
47+
SQLite build (used to report `disk_size` / `data_size`).
48+
49+
`memory` is `true` when the `memstat` loadable extension was successfully
50+
loaded (requires `ELECTRIC_SHAPE_DB_ENABLE_MEMORY_STATS=true` *and* the
51+
`ExSqlean` extension being present and loadable).
52+
"""
53+
@spec stats_enabled(term()) :: %{disk: boolean(), memory: boolean()}
54+
def stats_enabled(stack_id) do
55+
GenServer.call(name(stack_id), :stats_enabled)
56+
end
57+
3658
@impl GenServer
3759
def init(args) do
3860
stack_id = Keyword.fetch!(args, :stack_id)
@@ -47,8 +69,9 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Statistics do
4769
%{
4870
stack_id: stack_id,
4971
page_size: 0,
72+
dbstat_available?: true,
5073
memstat_available?: false,
51-
stats: %{}
74+
stats: %__MODULE__{}
5275
}, {:continue, {:initialize_stats, enable_memory_stats?}}}
5376
end
5477

@@ -97,18 +120,42 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Statistics do
97120

98121
@impl GenServer
99122
def handle_call(:statistics, _from, state) do
100-
{:reply, {:ok, state.stats}, state}
123+
{:reply, {:ok, Map.from_struct(state.stats)}, state}
101124
end
102125

103-
defp read_stats(%{stack_id: stack_id, memstat_available?: memstat_available?} = state) do
104-
{:ok, stats} =
126+
@impl GenServer
127+
def handle_call(:stats_enabled, _from, state) do
128+
{:reply, %{disk: state.dbstat_available?, memory: state.memstat_available?}, state}
129+
end
130+
131+
defp read_stats(
132+
%{stack_id: stack_id, dbstat_available?: true, memstat_available?: memstat_available?} =
133+
state
134+
) do
135+
result =
105136
ShapeDb.Connection.checkout_write!(stack_id, :read_stats, fn %{conn: conn} ->
106137
ShapeDb.Connection.fetch_all(conn, stats_query(memstat_available?), [])
107138
end)
108139

109140
Process.send_after(self(), :read_stats, @measurement_period)
110141

111-
%{state | stats: analyze_stats(stats, state.page_size)}
142+
case result do
143+
{:ok, stats} ->
144+
%{state | stats: analyze_stats(stats, state.page_size)}
145+
146+
{:error, reason} ->
147+
Logger.warning(
148+
"Failed to read SQLite db stats: #{inspect(reason)}. " <>
149+
"Disk size statistics will not be available."
150+
)
151+
152+
%{state | dbstat_available?: false}
153+
end
154+
end
155+
156+
defp read_stats(%{dbstat_available?: false} = state) do
157+
Process.send_after(self(), :read_stats, @measurement_period)
158+
state
112159
end
113160

114161
defp stats_query(true) do
@@ -143,7 +190,7 @@ defmodule Electric.ShapeCache.ShapeStatus.ShapeDb.Statistics do
143190
#
144191
# 3. PAGECACHE_OVERFLOW (heap fallback): When the pre-allocated page
145192
# cache is full, overflow goes to heap. This is already in bytes.
146-
%{
193+
%__MODULE__{
147194
total_memory: memory_used + pagecache_used + pagecache_overflow,
148195
page_cache_overflow: pagecache_overflow,
149196
disk_size: disk_size,

0 commit comments

Comments
 (0)