@@ -2,49 +2,51 @@ package pgfsm
22
33import (
44 "context"
5- "database/sql"
65 _ "embed"
76 "errors"
87 "log/slog"
98 "strings"
9+
10+ "github.com/jackc/pgx/v5"
11+ "github.com/jackc/pgx/v5/pgxpool"
1012)
1113
12- func transaction (ctx context.Context , db * sql. DB , fn func (ctx context.Context , tx * sql .Tx ) error ) error {
13- tx , err := db .BeginTx (ctx , & sql. TxOptions {} )
14+ func transaction (ctx context.Context , db * pgxpool. Pool , fn func (ctx context.Context , tx pgx .Tx ) error ) error {
15+ tx , err := db .Begin (ctx )
1416 if err != nil {
1517 return err
1618 }
1719
1820 if err = fn (ctx , tx ); err != nil {
19- txErr := tx .Rollback ()
20- if errors .Is (txErr , sql . ErrTxDone ) {
21+ txErr := tx .Rollback (ctx )
22+ if errors .Is (txErr , pgx . ErrTxClosed ) {
2123 return err
2224 }
2325
2426 return errors .Join (err , txErr )
2527 }
2628
27- err = tx .Commit ()
28- if errors .Is (err , sql . ErrTxDone ) {
29+ err = tx .Commit (ctx )
30+ if errors .Is (err , pgx . ErrTxClosed ) {
2931 return nil
3032 }
3133
3234 return err
3335}
3436
35- func insert (ctx context.Context , tx * sql .Tx , encoder Encoding , cmd Command ) error {
37+ func insert (ctx context.Context , tx pgx .Tx , encoder Encoding , cmd Command ) error {
3638 data , err := encoder .Encode (cmd )
3739 if err != nil {
3840 return err
3941 }
4042
4143 const q = `INSERT INTO pgfsm.command (kind, data) VALUES ($1, $2)`
4244
43- _ , err = tx .ExecContext (ctx , q , cmd .Kind (), data )
45+ _ , err = tx .Exec (ctx , q , cmd .Kind (), data )
4446 return err
4547}
4648
47- func next (ctx context.Context , tx * sql .Tx ) (int64 , string , []byte , error ) {
49+ func next (ctx context.Context , tx pgx .Tx ) (int64 , string , []byte , error ) {
4850 const q = `
4951 SELECT id, kind, data FROM pgfsm.command
5052 ORDER BY id ASC
@@ -58,24 +60,24 @@ func next(ctx context.Context, tx *sql.Tx) (int64, string, []byte, error) {
5860 data []byte
5961 )
6062
61- if err := tx .QueryRowContext (ctx , q ).Scan (& id , & kind , & data ); err != nil {
63+ if err := tx .QueryRow (ctx , q ).Scan (& id , & kind , & data ); err != nil {
6264 return 0 , "" , []byte {}, err
6365 }
6466
6567 return id , kind , data , nil
6668}
6769
68- func remove (ctx context.Context , tx * sql .Tx , id int64 ) error {
70+ func remove (ctx context.Context , tx pgx .Tx , id int64 ) error {
6971 const q = `DELETE FROM pgfsm.command WHERE id = $1`
7072
71- _ , err := tx .ExecContext (ctx , q , id )
73+ _ , err := tx .Exec (ctx , q , id )
7274 return err
7375}
7476
7577//go:embed migrate.sql
7678var migration string
7779
78- func migrateUp (ctx context.Context , db * sql. DB , logger * slog.Logger ) error {
80+ func migrateUp (ctx context.Context , db * pgxpool. Pool , logger * slog.Logger ) error {
7981 logger .DebugContext (ctx , "performing migrations" )
8082
8183 statements := strings .Split (migration , ";" )
@@ -89,7 +91,7 @@ func migrateUp(ctx context.Context, db *sql.DB, logger *slog.Logger) error {
8991 With (slog .String ("statement" , statement )).
9092 DebugContext (ctx , "executing statement" )
9193
92- if _ , err := db .ExecContext (ctx , statement ); err != nil {
94+ if _ , err := db .Exec (ctx , statement ); err != nil {
9395 return err
9496 }
9597 }
0 commit comments