database/sql driver: Quote named parameter strings

Fixes a problem reported in #478 in which during a job list, strings aren't properly quoted before being sent to Postgres. This is a bit of an unfortunate problem that stems from the driver being unable to take advantage of Pgx's named parameter system, nor `database/sql`'s `sql.Named` system (because neither Pgx nor `lib/pq` implement it), which required a rough implementation of a custom named parameter system that uses string find/replace. Here, handle a number of possible argument types that `JobList` might support and make sure that the driver sends them to Postgres in an appropriate format, making sure to quote strings and escape subquotes they may have contained. This is all still a little rough and something more robust would definitely be nice, but this should all be input from River's job module, so it doesn't need to have the most robust implementation ever. Fixes #481.
riverqueue · Jul 30, 2024 · 6ea55ae · 6ea55ae
1 parent 280896e
commit 6ea55ae
Show file tree

Hide file tree

Showing 4 changed files with 125 additions and 50 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
-- Include `pending` state in `JobListParams` by default so pending jobs are included in `JobList` / `JobListTx` results.
+- Include `pending` state in `JobListParams` by default so pending jobs are included in `JobList` / `JobListTx` results. [PR #477](https://github.com/riverqueue/river/pull/477).
+- Quote strings when using `Client.JobList` functions with the `database/sql` driver. [PR #XXX](https://github.com/riverqueue/river/pull/XXX).
 
 ## [0.10.1] - 2024-07-23
 

diff --git a/internal/riverinternaltest/riverdrivertest/riverdrivertest.go b/internal/riverinternaltest/riverdrivertest/riverdrivertest.go
@@ -1206,47 +1206,67 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 	t.Run("JobList", func(t *testing.T) {
 		t.Parallel()
 
-		exec, _ := setup(ctx, t)
+		t.Run("ListsJobs", func(t *testing.T) {
+			exec, _ := setup(ctx, t)
 
-		now := time.Now().UTC()
+			now := time.Now().UTC()
 
-		job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
-			Attempt:     ptrutil.Ptr(3),
-			AttemptedAt: &now,
-			CreatedAt:   &now,
-			EncodedArgs: []byte(`{"encoded": "args"}`),
-			Errors:      [][]byte{[]byte(`{"error": "message1"}`), []byte(`{"error": "message2"}`)},
-			FinalizedAt: &now,
-			Metadata:    []byte(`{"meta": "data"}`),
-			ScheduledAt: &now,
-			State:       ptrutil.Ptr(rivertype.JobStateCompleted),
-			Tags:        []string{"tag"},
-		})
-
-		fetchedJobs, err := exec.JobList(
-			ctx,
-			fmt.Sprintf("SELECT %s FROM river_job WHERE id = @job_id_123", exec.JobListFields()),
-			map[string]any{"job_id_123": job.ID},
-		)
-		require.NoError(t, err)
-		require.Len(t, fetchedJobs, 1)
-
-		fetchedJob := fetchedJobs[0]
-		require.Equal(t, job.Attempt, fetchedJob.Attempt)
-		require.Equal(t, job.AttemptedAt, fetchedJob.AttemptedAt)
-		require.Equal(t, job.CreatedAt, fetchedJob.CreatedAt)
-		require.Equal(t, job.EncodedArgs, fetchedJob.EncodedArgs)
-		require.Equal(t, "message1", fetchedJob.Errors[0].Error)
-		require.Equal(t, "message2", fetchedJob.Errors[1].Error)
-		require.Equal(t, job.FinalizedAt, fetchedJob.FinalizedAt)
-		require.Equal(t, job.Kind, fetchedJob.Kind)
-		require.Equal(t, job.MaxAttempts, fetchedJob.MaxAttempts)
-		require.Equal(t, job.Metadata, fetchedJob.Metadata)
-		require.Equal(t, job.Priority, fetchedJob.Priority)
-		require.Equal(t, job.Queue, fetchedJob.Queue)
-		require.Equal(t, job.ScheduledAt, fetchedJob.ScheduledAt)
-		require.Equal(t, job.State, fetchedJob.State)
-		require.Equal(t, job.Tags, fetchedJob.Tags)
+			job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
+				Attempt:     ptrutil.Ptr(3),
+				AttemptedAt: &now,
+				CreatedAt:   &now,
+				EncodedArgs: []byte(`{"encoded": "args"}`),
+				Errors:      [][]byte{[]byte(`{"error": "message1"}`), []byte(`{"error": "message2"}`)},
+				FinalizedAt: &now,
+				Metadata:    []byte(`{"meta": "data"}`),
+				ScheduledAt: &now,
+				State:       ptrutil.Ptr(rivertype.JobStateCompleted),
+				Tags:        []string{"tag"},
+			})
+
+			fetchedJobs, err := exec.JobList(
+				ctx,
+				fmt.Sprintf("SELECT %s FROM river_job WHERE id = @job_id_123", exec.JobListFields()),
+				map[string]any{"job_id_123": job.ID},
+			)
+			require.NoError(t, err)
+			require.Len(t, fetchedJobs, 1)
+
+			fetchedJob := fetchedJobs[0]
+			require.Equal(t, job.Attempt, fetchedJob.Attempt)
+			require.Equal(t, job.AttemptedAt, fetchedJob.AttemptedAt)
+			require.Equal(t, job.CreatedAt, fetchedJob.CreatedAt)
+			require.Equal(t, job.EncodedArgs, fetchedJob.EncodedArgs)
+			require.Equal(t, "message1", fetchedJob.Errors[0].Error)
+			require.Equal(t, "message2", fetchedJob.Errors[1].Error)
+			require.Equal(t, job.FinalizedAt, fetchedJob.FinalizedAt)
+			require.Equal(t, job.Kind, fetchedJob.Kind)
+			require.Equal(t, job.MaxAttempts, fetchedJob.MaxAttempts)
+			require.Equal(t, job.Metadata, fetchedJob.Metadata)
+			require.Equal(t, job.Priority, fetchedJob.Priority)
+			require.Equal(t, job.Queue, fetchedJob.Queue)
+			require.Equal(t, job.ScheduledAt, fetchedJob.ScheduledAt)
+			require.Equal(t, job.State, fetchedJob.State)
+			require.Equal(t, job.Tags, fetchedJob.Tags)
+		})
+
+		t.Run("QuotesStrigngs", func(t *testing.T) {
+			exec, _ := setup(ctx, t)
+
+			job1 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("test_kind1")})
+			_ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("test_kind2")}) // not returned
+
+			fetchedJobs, err := exec.JobList(
+				ctx,
+				fmt.Sprintf("SELECT %s FROM river_job WHERE kind = @kind", exec.JobListFields()),
+				map[string]any{"kind": job1.Kind},
+			)
+			require.NoError(t, err)
+			require.Len(t, fetchedJobs, 1)
+
+			fetchedJob1 := fetchedJobs[0]
+			require.Equal(t, job1.Kind, fetchedJob1.Kind)
+		})
 	})
 
 	t.Run("JobListFields", func(t *testing.T) {

diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver.go b/riverdriver/riverdatabasesql/river_database_sql_driver.go
@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"io/fs"
 	"math"
+	"strconv"
 	"strings"
 	"time"
 
@@ -314,16 +315,9 @@ func (e *Executor) JobInsertUnique(ctx context.Context, params *riverdriver.JobI
 }
 
 func (e *Executor) JobList(ctx context.Context, query string, namedArgs map[string]any) ([]*rivertype.JobRow, error) {
-	// `database/sql` has an `sql.Named` system that should theoretically work
-	// for named parameters, but neither Pgx or lib/pq implement it, so just use
-	// dumb string replacement given we're only injecting a very basic value
-	// anyway.
-	for name, value := range namedArgs {
-		newQuery := strings.Replace(query, "@"+name, fmt.Sprintf("%v", value), 1)
-		if newQuery == query {
-			return nil, fmt.Errorf("named query parameter @%s not found in query", name)
-		}
-		query = newQuery
+	query, err := replaceNamed(query, namedArgs)
+	if err != nil {
+		return nil, err
 	}
 
 	rows, err := e.dbtx.QueryContext(ctx, query)
@@ -364,6 +358,36 @@ func (e *Executor) JobList(ctx context.Context, query string, namedArgs map[stri
 	return mapSliceError(items, jobRowFromInternal)
 }
 
+// `database/sql` has an `sql.Named` system that should theoretically work for
+// named parameters, but neither Pgx or lib/pq implement it, so just use dumb
+// string replacement given we're only injecting a very basic value anyway.
+func replaceNamed(query string, namedArgs map[string]any) (string, error) {
+	for name, value := range namedArgs {
+		var escapedValue string
+
+		switch typedValue := value.(type) {
+		case bool:
+			escapedValue = strconv.FormatBool(typedValue)
+		case float32, float64:
+			escapedValue = fmt.Sprintf("%f", value)
+		case int, int16, int32, int64, uint, uint16, uint32, uint64:
+			escapedValue = fmt.Sprintf("%d", value)
+		case string:
+			escapedValue = "'" + strings.ReplaceAll(typedValue, "'", "''") + "'"
+		default:
+			return "", fmt.Errorf("named query parameter @%s is not a supported type", name)
+		}
+
+		newQuery := strings.Replace(query, "@"+name, escapedValue, 1)
+		if newQuery == query {
+			return "", fmt.Errorf("named query parameter @%s not found in query", name)
+		}
+		query = newQuery
+	}
+
+	return query, nil
+}
+
 func (e *Executor) JobListFields() string {
 	return "id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags"
 }

diff --git a/riverdriver/riverdatabasesql/river_database_sql_driver_test.go b/riverdriver/riverdatabasesql/river_database_sql_driver_test.go
@@ -40,3 +40,33 @@ func TestInterpretError(t *testing.T) {
 	require.ErrorIs(t, interpretError(sql.ErrNoRows), rivertype.ErrNotFound)
 	require.NoError(t, interpretError(nil))
 }
+
+func TestReplacedNamed(t *testing.T) {
+	t.Parallel()
+
+	mustReplaceNamed := func(query string, namedArgs map[string]any) string {
+		query, err := replaceNamed(query, namedArgs)
+		require.NoError(t, err)
+		return query
+	}
+
+	require.Equal(t,
+		"SELECT true",
+		mustReplaceNamed("SELECT @bool", map[string]any{"bool": true}),
+	)
+
+	require.Equal(t,
+		"SELECT 123",
+		mustReplaceNamed("SELECT @int", map[string]any{"int": 123}),
+	)
+
+	require.Equal(t,
+		"SELECT 'string value'",
+		mustReplaceNamed("SELECT @string", map[string]any{"string": "string value"}),
+	)
+
+	require.Equal(t,
+		"SELECT 'string value with '' quote'",
+		mustReplaceNamed("SELECT @string", map[string]any{"string": "string value with ' quote"}),
+	)
+}