Add sqlight-compatible query functions

Author: Resonious

rust/src/main.rs

@@ -32,6 +32,18 @@ pub struct Execute {
     pub params: Vec<Value>,
 }
 
+#[derive(Debug, SchemaRead, SchemaWrite)]
+pub struct Run {
+    pub db: String,
+    pub sql: String,
+}
+
+#[derive(Debug, SchemaRead, SchemaWrite)]
+pub enum RunResult {
+    Ok,
+    Error(String),
+}
+
 #[derive(Debug, SchemaRead, SchemaWrite)]
 pub enum Updated {
     Ok(u64),
@@ -96,6 +108,7 @@ define_requests! {
     Crap => BadRequest,
     Select => RowsResult,
     Execute => Updated,
+    Run => RunResult,
 }
 
 #[tokio::main]
@@ -180,6 +193,14 @@ async fn handle_request(cache: &DbCache, req: Requests) -> Responses {
             };
             response_enum(p, resp)
         }
+
+        Requests::Run(run, p) => {
+            let resp = match run_batch(cache, run).await {
+                Ok(()) => RunResult::Ok,
+                Err(e) => RunResult::Error(e),
+            };
+            response_enum(p, resp)
+        }
     }
 }
 
@@ -281,6 +302,12 @@ async fn execute_statement(cache: &DbCache, execute: Execute) -> Result<u64, Str
     stmt.execute(params).await.map_err(|e| e.to_string())
 }
 
+async fn run_batch(cache: &DbCache, run: Run) -> Result<(), String> {
+    let cached = get_cached_db(cache, &run.db).await?;
+    let cached = cached.lock().await;
+    cached.conn.execute_batch(&run.sql).await.map_err(|e| e.to_string())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/pturso.gleam

@@ -1,3 +1,6 @@
+import gleam/list
+import gleam/result.{try}
+import gleam/dynamic/decode
 import gleam/dynamic.{type Dynamic}
 
 /// Parameter values for queries (needed for bincode encoding)
@@ -9,29 +12,103 @@ pub type Param {
   Blob(BitArray)
 }
 
-pub type Row =
-  List(Dynamic)
+pub type Port
 
-pub type Connection
+pub type Connection {
+  Connection(port: Port, db: String)
+}
+
+pub type Error {
+  DatabaseError(message: String)
+  DecodeError(errors: List(decode.DecodeError))
+}
 
 @external(erlang, "pturso_ffi", "start")
-pub fn start(binary_path: String) -> Result(Connection, String)
+pub fn start(binary_path: String) -> Result(Port, String)
 
 @external(erlang, "pturso_ffi", "stop")
-pub fn stop(conn: Connection) -> Nil
+pub fn stop(conn: Port) -> Nil
 
+/// Raw select function for queries that return results.
+/// Considered "low level" as it takes in Port and DB separately
+/// and is implemented in Erlang.
 @external(erlang, "pturso_ffi", "select")
 pub fn select(
-  conn: Connection,
+  conn: Port,
   db: String,
   query: String,
   params: List(Param),
-) -> Result(List(Row), String)
+) -> Result(List(Dynamic), String)
 
+/// Raw execute function for queries that do not return anything.
+/// Considered "low level" as it takes in Port and DB separately
+/// and is implemented in Erlang.
 @external(erlang, "pturso_ffi", "execute")
 pub fn execute(
-  conn: Connection,
+  conn: Port,
   db: String,
   query: String,
   params: List(Param),
 ) -> Result(Int, String)
+
+/// Run one or more SQL statements without using prepared statements.
+/// Supports multiple statements separated by semicolons.
+/// Returns nothing on success.
+@external(erlang, "pturso_ffi", "run")
+pub fn run(conn: Port, db: String, sql: String) -> Result(Nil, String)
+
+/// Creates a Connection object.
+/// Note that this doesn't actually cause a real "connection" to
+/// be created. Actual DB connections are created lazily as
+/// queries come in.
+/// This is useful if you want to use the sqlight-compatible
+/// query and exec functions.
+pub fn connect(port: Port, to db: String) -> Connection {
+  Connection(port:, db:)
+}
+
+/// sqlight-compatible query function.
+pub fn query(
+  sql: String,
+  on conn: Connection,
+  with params: List(Param),
+  expecting decoder: decode.Decoder(a),
+) -> Result(List(a), Error) {
+  use rows <- try(select(
+    conn.port,
+    conn.db,
+    sql,
+    params,
+  ) |> result.map_error(DatabaseError))
+
+  let decoded = list.map(rows, decode.run(_, decoder))
+
+  let successes = list.filter_map(decoded, fn(x) {
+    case x {
+      Ok(i) -> Ok(i)
+      Error(_) -> Error(Nil)
+    }
+  })
+
+  let errors = list.filter_map(decoded, fn(x) {
+    case x {
+      Ok(_) -> Error(Nil)
+      Error(e) -> Ok(e)
+    }
+  })
+
+  case successes, errors {
+    s, [] -> Ok(s)
+    _, [_, ..] as e -> Error(DecodeError(list.flatten(e)))
+  }
+}
+
+/// sqlight-compatible exec function.
+/// Runs one or more SQL statements without prepared statements.
+pub fn exec(
+  sql: String,
+  on conn: Connection,
+) -> Result(Nil, Error) {
+  run(conn.port, conn.db, sql)
+  |> result.map_error(DatabaseError)
+}

src/pturso_bincode.erl

@@ -149,13 +149,14 @@ decode_value(<<4:32/little-unsigned, Rest/binary>>) ->
 
 %%====================================================================
 %% Request Encoding
-%% Rust: enum Requests { Crap=0, Select=1, Insert=2 }
+%% Rust: enum Requests { Crap=0, Select=1, Execute=2, Run=3 }
 %%====================================================================
 
 -type crap_request() :: {crap, binary()}.
 -type select_request() :: {select, binary(), binary(), [value()]}.
 -type insert_request() :: {insert, binary(), binary(), [value()]}.
--type request() :: crap_request() | select_request() | insert_request().
+-type run_request() :: {run, binary(), binary()}.
+-type request() :: crap_request() | select_request() | insert_request() | run_request().
 
 -spec encode_request(request()) -> binary().
 encode_request({crap, Reason}) ->
@@ -171,22 +172,29 @@ encode_request({select, Db, Query, Params}) ->
       ParamsBin/binary>>;
 
 encode_request({insert, Db, Query, Params}) ->
-    %% Requests::Insert(Insert { db, query, params })
+    %% Requests::Execute(Execute { db, query, params })
     ParamsBin = encode_vec(fun encode_value/1, Params),
     <<(encode_u32(2))/binary,
       (encode_string(Db))/binary,
       (encode_string(Query))/binary,
-      ParamsBin/binary>>.
+      ParamsBin/binary>>;
+
+encode_request({run, Db, Sql}) ->
+    %% Requests::Run(Run { db, sql })
+    <<(encode_u32(3))/binary,
+      (encode_string(Db))/binary,
+      (encode_string(Sql))/binary>>.
 
 %%====================================================================
 %% Response Decoding
-%% Rust: enum Responses { BadRequest=0, RowsResult=1, Updated=2 }
+%% Rust: enum Responses { BadRequest=0, RowsResult=1, Updated=2, RunResult=3 }
 %%====================================================================
 
 -type bad_request_response() :: {bad_request, binary()}.
 -type rows_result_response() :: {rows_result, {ok, [[value()]]} | {error, binary()}}.
 -type updated_response() :: {updated, {ok, non_neg_integer()} | {error, binary()}}.
--type response() :: bad_request_response() | rows_result_response() | updated_response().
+-type run_result_response() :: {run_result, ok | {error, binary()}}.
+-type response() :: bad_request_response() | rows_result_response() | updated_response() | run_result_response().
 
 -spec decode_response(binary()) -> response().
 decode_response(<<0:32/little-unsigned, Rest/binary>>) ->
@@ -200,7 +208,11 @@ decode_response(<<1:32/little-unsigned, Rest/binary>>) ->
 
 decode_response(<<2:32/little-unsigned, Rest/binary>>) ->
     %% Updated: enum { Ok(u64)=0, Error(String)=1 }
-    decode_updated(Rest).
+    decode_updated(Rest);
+
+decode_response(<<3:32/little-unsigned, Rest/binary>>) ->
+    %% RunResult: enum { Ok=0, Error(String)=1 }
+    decode_run_result(Rest).
 
 decode_rows_result(<<0:32/little-unsigned, Rest/binary>>) ->
     %% RowsResult::Ok(Vec<Vec<Value>>)
@@ -221,3 +233,11 @@ decode_updated(<<1:32/little-unsigned, Rest/binary>>) ->
     %% Updated::Error(String)
     {Reason, <<>>} = decode_string(Rest),
     {updated, {error, Reason}}.
+
+decode_run_result(<<0:32/little-unsigned, _Rest/binary>>) ->
+    %% RunResult::Ok
+    {run_result, ok};
+decode_run_result(<<1:32/little-unsigned, Rest/binary>>) ->
+    %% RunResult::Error(String)
+    {Reason, <<>>} = decode_string(Rest),
+    {run_result, {error, Reason}}.

src/pturso_ffi.erl

@@ -5,7 +5,8 @@
     start/1,
     stop/1,
     select/4,
-    execute/4
+    execute/4,
+    run/3
 ]).
 
 %%====================================================================
@@ -33,7 +34,8 @@ select({connection, Pid}, Db, Query, Params) ->
     Request = {select, Db, Query, ErlParams},
     case pturso_port:call(Pid, Request) of
         {rows_result, {ok, Rows}} ->
-            DynamicRows = [[erl_value_to_dynamic(V) || V <- Row] || Row <- Rows],
+            %% Return rows as tuples so decode.field(index, decoder) works
+            DynamicRows = [list_to_tuple([erl_value_to_dynamic(V) || V <- Row]) || Row <- Rows],
             {ok, DynamicRows};
         {rows_result, {error, Reason}} ->
             {error, Reason};
@@ -55,6 +57,18 @@ execute({connection, Pid}, Db, Query, Params) ->
             {error, Reason}
     end.
 
+-spec run({connection, pid()}, binary(), binary()) -> {ok, nil} | {error, binary()}.
+run({connection, Pid}, Db, Sql) ->
+    Request = {run, Db, Sql},
+    case pturso_port:call(Pid, Request) of
+        {run_result, ok} ->
+            {ok, nil};
+        {run_result, {error, Reason}} ->
+            {error, Reason};
+        {bad_request, Reason} ->
+            {error, Reason}
+    end.
+
 %%====================================================================
 %% Internal functions
 %%====================================================================