Merge pull request #163 from vgarvardt/feat/db-adapter-unwrap-tx

feat: added UnwrapTx to all db adapters to get driver-specific tx

adapter/libpq/db.go

@@ -68,6 +68,17 @@ func NewTx(tx *sql.Tx) adapter.Tx {
 	return &aTx{tx: tx}
 }
 
+// UnwrapTx tries to unwrap driver-specific transaction instance from the interface.
+// Returns unwrap success as the second parameter.
+func UnwrapTx(tx adapter.Tx) (*sql.Tx, bool) {
+	driverTx, ok := tx.(*aTx)
+	if !ok {
+		return nil, false
+	}
+
+	return driverTx.tx, ok
+}
+
 // Exec implements adapter.Tx.Exec() using github.com/lib/pq
 func (tx *aTx) Exec(ctx context.Context, query string, args ...any) (adapter.CommandTag, error) {
 	ct, err := tx.tx.ExecContext(ctx, query, args...)

adapter/pgxv4/db.go

@@ -65,6 +65,17 @@ func NewTx(tx pgx.Tx) adapter.Tx {
 	return &aTx{tx: tx}
 }
 
+// UnwrapTx tries to unwrap driver-specific transaction instance from the interface.
+// Returns unwrap success as the second parameter.
+func UnwrapTx(tx adapter.Tx) (pgx.Tx, bool) {
+	driverTx, ok := tx.(*aTx)
+	if !ok {
+		return nil, false
+	}
+
+	return driverTx.tx, ok
+}
+
 // Exec implements adapter.Tx.Exec() using github.com/jackc/pgx/v4
 func (tx *aTx) Exec(ctx context.Context, query string, args ...any) (adapter.CommandTag, error) {
 	ct, err := tx.tx.Exec(ctx, query, args...)

adapter/pgxv5/db.go

@@ -65,6 +65,17 @@ func NewTx(tx pgx.Tx) adapter.Tx {
 	return &aTx{tx: tx}
 }
 
+// UnwrapTx tries to unwrap driver-specific transaction instance from the interface.
+// Returns unwrap success as the second parameter.
+func UnwrapTx(tx adapter.Tx) (pgx.Tx, bool) {
+	driverTx, ok := tx.(*aTx)
+	if !ok {
+		return nil, false
+	}
+
+	return driverTx.tx, ok
+}
+
 // Exec implements adapter.Tx.Exec() using github.com/jackc/pgx/v5
 func (tx *aTx) Exec(ctx context.Context, query string, args ...any) (adapter.CommandTag, error) {
 	ct, err := tx.tx.Exec(ctx, query, args...)

job_test.go

@@ -0,0 +1,83 @@
+package gue
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/vgarvardt/gue/v5/adapter"
+	"github.com/vgarvardt/gue/v5/adapter/libpq"
+	"github.com/vgarvardt/gue/v5/adapter/pgxv4"
+	"github.com/vgarvardt/gue/v5/adapter/pgxv5"
+	adapterTesting "github.com/vgarvardt/gue/v5/adapter/testing"
+)
+
+func TestJob_Tx(t *testing.T) {
+	for name, openFunc := range adapterTesting.AllAdaptersOpenTestPool {
+		t.Run(name, func(t *testing.T) {
+			testJobTxUnwrapTx(t, name, openFunc(t))
+		})
+	}
+}
+
+func testJobTxUnwrapTx(t *testing.T, name string, connPool adapter.ConnPool) {
+	ctx := context.Background()
+
+	c, err := NewClient(connPool)
+	require.NoError(t, err)
+
+	newJob := &Job{Type: "MyJob", Args: []byte(`{}`)}
+	err = c.Enqueue(ctx, newJob)
+	require.NoError(t, err)
+	require.NotEmpty(t, newJob.ID)
+
+	j, err := c.LockJob(ctx, "")
+	require.NoError(t, err)
+	require.NotNil(t, j)
+	require.NotNil(t, j.tx)
+
+	t.Cleanup(func() {
+		err := j.Done(ctx)
+		assert.NoError(t, err)
+	})
+
+	switch name {
+	case "pgx/v4":
+		_, okPgxV5 := pgxv5.UnwrapTx(j.Tx())
+		require.False(t, okPgxV5)
+		_, okLibPQ := libpq.UnwrapTx(j.Tx())
+		require.False(t, okLibPQ)
+
+		tx, okPgxV4 := pgxv4.UnwrapTx(j.Tx())
+		require.True(t, okPgxV4)
+
+		_, err := tx.Exec(ctx, `SELECT COUNT(1) FROM gue_jobs`)
+		require.NoError(t, err)
+
+	case "pgx/v5":
+		_, okPgxV4 := pgxv4.UnwrapTx(j.Tx())
+		require.False(t, okPgxV4)
+		_, okLibPQ := libpq.UnwrapTx(j.Tx())
+		require.False(t, okLibPQ)
+
+		tx, okPgxV5 := pgxv5.UnwrapTx(j.Tx())
+		require.True(t, okPgxV5)
+
+		_, err := tx.Exec(ctx, `SELECT COUNT(1) FROM gue_jobs`)
+		require.NoError(t, err)
+
+	case "lib/pq":
+		_, okPgxV4 := pgxv4.UnwrapTx(j.Tx())
+		require.False(t, okPgxV4)
+		_, okPgxV5 := pgxv5.UnwrapTx(j.Tx())
+		require.False(t, okPgxV5)
+
+		tx, okLibPQ := libpq.UnwrapTx(j.Tx())
+		require.True(t, okLibPQ)
+
+		_, err := tx.ExecContext(ctx, `SELECT COUNT(1) FROM gue_jobs`)
+		require.NoError(t, err)
+	}
+}