Start with new transaction approach

Author: lpil

gleam.toml

@@ -20,7 +20,7 @@ gleam_erlang = ">= 1.2.0 and < 2.0.0"
 gleam_otp = ">= 1.0.0 and < 2.0.0"
 gleam_stdlib = ">= 0.51.0 and < 2.0.0"
 gleam_time = ">= 1.0.0 and < 2.0.0"
-pgo = ">= 0.12.0 and < 2.0.0"
+pgo = ">= 0.14.0 and < 1.0.0"
 
 [dev-dependencies]
 gleeunit = ">= 1.0.0 and < 2.0.0"

manifest.toml

@@ -21,4 +21,4 @@ gleam_otp = { version = ">= 1.0.0 and < 2.0.0" }
 gleam_stdlib = { version = ">= 0.51.0 and < 2.0.0" }
 gleam_time = { version = ">= 1.0.0 and < 2.0.0" }
 gleeunit = { version = ">= 1.0.0 and < 2.0.0" }
-pgo = { version = ">= 0.12.0 and < 2.0.0" }
+pgo = { version = ">= 0.14.0 and < 1.0.0" }

src/pog.gleam

@@ -2,6 +2,9 @@
 ////
 //// Gleam wrapper around pgo library
 
+// TODO: Connection is a union of the name and the checked out connection
+// TODO: Move handling of this into Gleam to make it easier to write
+
 // TODO: add time and timestamp with zone once pgo supports them
 
 import gleam/dynamic.{type Dynamic}
@@ -406,12 +409,17 @@ pub type TransactionError {
 ///
 /// If the function returns an `Error` or panics then the transaction is rolled
 /// back.
-@external(erlang, "pog_ffi", "transaction")
 pub fn transaction(
   pool: Subject(Message),
   callback: fn(Subject(Message)) -> Result(t, String),
 ) -> Result(t, TransactionError)
 
+@external(erlang, "pog_ffi", "transaction")
+fn run_transaction(
+  pool: Name(Message),
+  callback: fn(Subject(Message)) -> Result(t, String),
+) -> Result(t, TransactionError)
+
 pub fn nullable(inner_type: fn(a) -> Value, value: Option(a)) -> Value {
   case value {
     Some(term) -> inner_type(term)

src/pog_ffi.erl

@@ -80,7 +80,7 @@ start(Config) ->
     end,
     pgo_pool:start_link(PoolName, Options2).
 
-transaction(Pool, Callback) when is_atom(Pool) ->
+old_transaction(Pool, Callback) when is_atom(Pool) ->
     F = fun() ->
         case Callback(Pool) of
             {ok, T} -> {ok, T};
@@ -95,6 +95,35 @@ transaction(Pool, Callback) when is_atom(Pool) ->
     end.
 
 
+transaction(Pool, Fun) when is_atom(Pool) andalso is_function(Fun, 1) ->
+    Exec = fun(Conn, Sql) ->
+        pgo_handler:extended_query(Conn, Sql, [], #{queue_time => undefined})
+    end,
+    case pgo:checkout(Pool) of
+        {ok, Ref, Conn} ->
+            try
+               #{command := 'begin'} = Exec(Conn, "BEGIN"),
+               Result = Fun(),
+               case Exec(Conn, "COMMIT") of
+                   #{command := commit} -> Result;
+                   #{command := rollback} -> Result
+               end
+            catch
+                Type:Reason:Stacktrace ->
+                Exec(Conn, "ROLLBACK"),
+                erlang:raise(Type, Reason, Stacktrace)
+            after
+                pgo:checkin(Ref, Conn)
+            end;
+        {error, _} = Error ->
+            Error
+    end;
+% TODO: remove
+transaction(A, B) ->
+    erlang:display(A),
+    erlang:display(B),
+    erlang:raise(badarg).
+
 query(Pool, Sql, Arguments, Timeout) when is_atom(Pool) ->
     Options = #{
         pool => Pool,