streamline the database interfaces and add pgx adapter

.github/workflows/pr.yml

@@ -1,5 +1,6 @@
 name: pr
 on:
+  pull_request:
   push:
     branches-ignore:
       - master

go.mod

@@ -6,9 +6,10 @@ toolchain go1.23.0
 
 require (
 	github.com/ThreeDotsLabs/watermill v1.4.0-rc.2
+	github.com/ThreeDotsLabs/watermill-sql/v3 v3.1.0
 	github.com/go-sql-driver/mysql v1.4.1
-	github.com/jackc/pgx/v4 v4.18.2
-	github.com/lib/pq v1.10.2
+	github.com/jackc/pgx/v5 v5.7.1
+	github.com/lib/pq v1.10.9
 	github.com/oklog/ulid v1.3.1
 	github.com/stretchr/testify v1.9.0
 )
@@ -19,19 +20,16 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
-	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
-	github.com/jackc/pgconn v1.14.3 // indirect
-	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
-	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
-	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
-	github.com/jackc/pgtype v1.14.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/lithammer/shortuuid/v3 v3.0.7 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/sony/gobreaker v1.0.0 // indirect
-	golang.org/x/crypto v0.20.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/crypto v0.27.0 // indirect
+	golang.org/x/sync v0.8.0 // indirect
+	golang.org/x/text v0.18.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

pkg/pgx/pgx_adapter.go

@@ -0,0 +1,127 @@
+package pgx
+
+import (
+	"context"
+	stdSQL "database/sql"
+	"fmt"
+	"time"
+
+	"github.com/ThreeDotsLabs/watermill-sql/v3/pkg/sql"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+type Conn interface {
+	BeginTx(ctx context.Context, options pgx.TxOptions) (pgx.Tx, error)
+	Exec(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)
+	Query(ctx context.Context, sql string, arguments ...interface{}) (pgx.Rows, error)
+}
+
+type Beginner struct {
+	Conn
+}
+
+type Tx struct {
+	pgx.Tx
+}
+
+type Result struct {
+	pgconn.CommandTag
+}
+
+type Rows struct {
+	pgx.Rows
+}
+
+func (c Beginner) BeginTx(ctx context.Context, options *stdSQL.TxOptions) (sql.Tx, error) {
+	opts := pgx.TxOptions{}
+	if options != nil {
+		iso, err := toPgxIsolationLevel(options.Isolation)
+		if err != nil {
+			return nil, err
+		}
+
+		opts = pgx.TxOptions{
+			IsoLevel:       iso,
+			AccessMode:     toPgxAccessMode(options.ReadOnly),
+			DeferrableMode: "",
+		}
+	}
+
+	tx, err := c.Conn.BeginTx(ctx, opts)
+
+	return Tx{tx}, err
+}
+
+func (c Beginner) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
+	res, err := c.Conn.Exec(ctx, query, args...)
+
+	return Result{res}, err
+}
+
+func (c Beginner) QueryContext(ctx context.Context, query string, args ...interface{}) (sql.Rows, error) {
+	rows, err := c.Conn.Query(ctx, query, args...)
+
+	return Rows{rows}, err
+}
+
+func (t Tx) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+	res, err := t.Tx.Exec(ctx, query, args...)
+
+	return Result{res}, err
+}
+
+func (t Tx) QueryContext(ctx context.Context, query string, args ...any) (sql.Rows, error) {
+	rows, err := t.Tx.Query(ctx, query, args...)
+
+	return Rows{rows}, err
+}
+
+func (t Tx) Rollback() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	return t.Tx.Rollback(ctx)
+}
+
+func (t Tx) Commit() error {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	return t.Tx.Commit(ctx)
+}
+
+func (p Result) RowsAffected() (int64, error) {
+	return p.CommandTag.RowsAffected(), nil
+}
+
+func (p Rows) Close() error {
+	p.Rows.Close()
+
+	return nil
+}
+
+func toPgxIsolationLevel(level stdSQL.IsolationLevel) (pgx.TxIsoLevel, error) {
+	switch level {
+	case stdSQL.LevelReadUncommitted:
+		return pgx.ReadUncommitted, nil
+	case stdSQL.LevelReadCommitted:
+		return pgx.ReadCommitted, nil
+	case stdSQL.LevelRepeatableRead:
+		return pgx.RepeatableRead, nil
+	case stdSQL.LevelSerializable:
+		return pgx.Serializable, nil
+	case stdSQL.LevelSnapshot:
+		return pgx.Serializable, fmt.Errorf("pgx does not support snapshot isolation")
+	default:
+		return pgx.Serializable, nil
+	}
+}
+
+func toPgxAccessMode(readOnly bool) pgx.TxAccessMode {
+	if readOnly {
+		return pgx.ReadOnly
+	}
+
+	return pgx.ReadWrite
+}

pkg/sql/context.go

@@ -2,7 +2,6 @@ package sql
 
 import (
 	"context"
-	"database/sql"
 )
 
 type contextKey string
@@ -11,7 +10,7 @@ const (
 	txContextKey contextKey = "tx"
 )
 
-func setTxToContext(ctx context.Context, tx *sql.Tx) context.Context {
+func setTxToContext(ctx context.Context, tx Tx) context.Context {
 	return context.WithValue(ctx, txContextKey, tx)
 }
 
@@ -21,7 +20,7 @@ func setTxToContext(ctx context.Context, tx *sql.Tx) context.Context {
 //
 // It is useful when you want to ensure that data is updated only when the message is processed.
 // Example usage: https://github.com/ThreeDotsLabs/watermill/tree/master/_examples/real-world-examples/exactly-once-delivery-counter
-func TxFromContext(ctx context.Context) (*sql.Tx, bool) {
-	tx, ok := ctx.Value(txContextKey).(*sql.Tx)
+func TxFromContext(ctx context.Context) (Tx, bool) {
+	tx, ok := ctx.Value(txContextKey).(Tx)
 	return tx, ok
 }