Add sqlc templates for arbitrary text substitution

client_test.go

@@ -710,7 +710,7 @@ func Test_Client(t *testing.T) {
 					&overridableJobMiddleware{
 						workFunc: func(ctx context.Context, job *rivertype.JobRow, doInner func(ctx context.Context) error) error {
 							middlewareCalled = true
-							require.Equal(t, `{"name": "inserted name"}`, string(job.EncodedArgs))
+							require.JSONEq(t, `{"name": "inserted name"}`, string(job.EncodedArgs))
 							job.EncodedArgs = []byte(`{"name": "middleware name"}`)
 							return doInner(ctx)
 						},

internal/dblist/db_list.go

@@ -3,25 +3,13 @@ package dblist
 import (
 	"context"
 	"errors"
-	"fmt"
 	"strings"
 
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/rivershared/util/sliceutil"
 	"github.com/riverqueue/river/rivertype"
 )
 
-const jobList = `-- name: JobList :many
-SELECT
-  %s
-FROM
-  river_job
-%s
-ORDER BY
-  %s
-LIMIT @count::integer
-`
-
 type SortOrder int
 
 const (
@@ -47,7 +35,7 @@ type JobListParams struct {
 }
 
 func JobList(ctx context.Context, exec riverdriver.Executor, params *JobListParams) ([]*rivertype.JobRow, error) {
-	var conditionsBuilder strings.Builder
+	var whereBuilder strings.Builder
 
 	orderBy := make([]JobListOrderBy, len(params.OrderBy))
 	for i, o := range params.OrderBy {
@@ -62,41 +50,44 @@ func JobList(ctx context.Context, exec riverdriver.Executor, params *JobListPara
 		namedArgs = make(map[string]any)
 	}
 
-	writeWhereOrAnd := func() {
-		if conditionsBuilder.Len() == 0 {
-			conditionsBuilder.WriteString("WHERE\n	")
-		} else {
-			conditionsBuilder.WriteString("\n  AND ")
+	writeAndAfterFirst := func() {
+		if whereBuilder.Len() != 0 {
+			whereBuilder.WriteString("\n  AND ")
 		}
 	}
 
 	if len(params.Kinds) > 0 {
-		writeWhereOrAnd()
-		conditionsBuilder.WriteString("kind = any(@kinds::text[])")
+		writeAndAfterFirst()
+		whereBuilder.WriteString("kind = any(@kinds::text[])")
 		namedArgs["kinds"] = params.Kinds
 	}
 
 	if len(params.Queues) > 0 {
-		writeWhereOrAnd()
-		conditionsBuilder.WriteString("queue = any(@queues::text[])")
+		writeAndAfterFirst()
+		whereBuilder.WriteString("queue = any(@queues::text[])")
 		namedArgs["queues"] = params.Queues
 	}
 
 	if len(params.States) > 0 {
-		writeWhereOrAnd()
-		conditionsBuilder.WriteString("state = any(@states::river_job_state[])")
+		writeAndAfterFirst()
+		whereBuilder.WriteString("state = any(@states::river_job_state[])")
 		namedArgs["states"] = sliceutil.Map(params.States, func(s rivertype.JobState) string { return string(s) })
 	}
 
 	if params.Conditions != "" {
-		writeWhereOrAnd()
-		conditionsBuilder.WriteString(params.Conditions)
+		writeAndAfterFirst()
+		whereBuilder.WriteString(params.Conditions)
+	}
+
+	// A condition of some kind is needed, so given no others write one that'll
+	// always return true.
+	if whereBuilder.Len() < 1 {
+		whereBuilder.WriteString("1")
 	}
 
 	if params.LimitCount < 1 {
 		return nil, errors.New("required parameter 'Count' in JobList must be greater than zero")
 	}
-	namedArgs["count"] = params.LimitCount
 
 	if len(params.OrderBy) == 0 {
 		return nil, errors.New("sort order is required")
@@ -116,7 +107,10 @@ func JobList(ctx context.Context, exec riverdriver.Executor, params *JobListPara
 		}
 	}
 
-	sql := fmt.Sprintf(jobList, exec.JobListFields(), conditionsBuilder.String(), orderByBuilder.String())
-
-	return exec.JobList(ctx, sql, namedArgs)
+	return exec.JobList(ctx, &riverdriver.JobListParams{
+		Max:           params.LimitCount,
+		NamedArgs:     namedArgs,
+		OrderByClause: orderByBuilder.String(),
+		WhereClause:   whereBuilder.String(),
+	})
 }

internal/riverinternaltest/riverdrivertest/riverdrivertest.go

@@ -1281,11 +1281,15 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 				UniqueStates: 0xFF,
 			})
 
-			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},
-			)
+			// Does not match predicate (makes sure where clause is working).
+			_ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{})
+
+			fetchedJobs, err := exec.JobList(ctx, &riverdriver.JobListParams{
+				Max:           100,
+				NamedArgs:     map[string]any{"job_id_123": job.ID},
+				OrderByClause: "id",
+				WhereClause:   "id = @job_id_123",
+			})
 			require.NoError(t, err)
 			require.Len(t, fetchedJobs, 1)
 
@@ -1316,36 +1320,29 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 			job2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("test_kind2")})
 
 			{
-				fetchedJobs, err := exec.JobList(
-					ctx,
-					fmt.Sprintf("SELECT %s FROM river_job WHERE kind = @kind", exec.JobListFields()),
-					map[string]any{"kind": job1.Kind},
-				)
+				fetchedJobs, err := exec.JobList(ctx, &riverdriver.JobListParams{
+					Max:           100,
+					NamedArgs:     map[string]any{"kind": job1.Kind},
+					OrderByClause: "id",
+					WhereClause:   "kind = @kind",
+				})
 				require.NoError(t, err)
 				require.Len(t, fetchedJobs, 1)
 			}
 
 			{
-				fetchedJobs, err := exec.JobList(
-					ctx,
-					fmt.Sprintf("SELECT %s FROM river_job WHERE kind = any(@kind::text[])", exec.JobListFields()),
-					map[string]any{"kind": []string{job1.Kind, job2.Kind}},
-				)
+				fetchedJobs, err := exec.JobList(ctx, &riverdriver.JobListParams{
+					Max:           100,
+					NamedArgs:     map[string]any{"kind": []string{job1.Kind, job2.Kind}},
+					OrderByClause: "id",
+					WhereClause:   "kind = any(@kind::text[])",
+				})
 				require.NoError(t, err)
 				require.Len(t, fetchedJobs, 2)
 			}
 		})
 	})
 
-	t.Run("JobListFields", func(t *testing.T) {
-		t.Parallel()
-
-		exec, _ := setup(ctx, t)
-
-		require.Equal(t, "id, args, attempt, attempted_at, attempted_by, created_at, errors, finalized_at, kind, max_attempts, metadata, priority, queue, state, scheduled_at, tags, unique_key, unique_states",
-			exec.JobListFields())
-	})
-
 	t.Run("JobRescueMany", func(t *testing.T) {
 		t.Parallel()
 

riverdriver/river_driver_interface.go

@@ -118,8 +118,7 @@ type Executor interface {
 	JobInsertFastMany(ctx context.Context, params []*JobInsertFastParams) ([]*JobInsertFastResult, error)
 	JobInsertFastManyNoReturning(ctx context.Context, params []*JobInsertFastParams) (int, error)
 	JobInsertFull(ctx context.Context, params *JobInsertFullParams) (*rivertype.JobRow, error)
-	JobList(ctx context.Context, query string, namedArgs map[string]any) ([]*rivertype.JobRow, error)
-	JobListFields() string
+	JobList(ctx context.Context, params *JobListParams) ([]*rivertype.JobRow, error)
 	JobRescueMany(ctx context.Context, params *JobRescueManyParams) (*struct{}, error)
 	JobRetry(ctx context.Context, id int64) (*rivertype.JobRow, error)
 	JobSchedule(ctx context.Context, params *JobScheduleParams) ([]*JobScheduleResult, error)
@@ -290,6 +289,13 @@ type JobInsertFullParams struct {
 	UniqueStates byte
 }
 
+type JobListParams struct {
+	Max           int32
+	NamedArgs     map[string]any
+	OrderByClause string
+	WhereClause   string
+}
+
 type JobRescueManyParams struct {
 	ID          []int64
 	Error       [][]byte