match job kind and queue name by substring, not just prefix (#1018)

* match job kind and queue name by substring, not just prefix

If you have long segmented job kinds, it's frustrating to have to type
out a long shared prefix to get to the unique part.

* golangci-lint v2.4.0

Author: bgentry

.github/workflows/ci.yaml

@@ -251,7 +251,7 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     env:
-      GOLANGCI_LINT_VERSION: v2.3.0
+      GOLANGCI_LINT_VERSION: v2.4.0
     permissions:
       contents: read
       # allow read access to pull request. Use with `only-new-issues` option.

client.go

@@ -503,7 +503,7 @@ func (c *Config) validate() error {
 			kind := workerInfo.jobArgs.Kind()
 			if !rivercommon.UserSpecifiedIDOrKindRE.MatchString(kind) {
 				if c.SkipJobKindValidation {
-					c.Logger.Warn("job kind should match regex; this will be an error in future versions", //nolint:noctx
+					c.Logger.Warn("job kind should match regex; this will be an error in future versions",
 						slog.String("kind", kind),
 						slog.String("regex", rivercommon.UserSpecifiedIDOrKindRE.String()),
 					)
@@ -829,7 +829,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 				client.services = append(client.services, client.notifier)
 			}
 		} else {
-			config.Logger.Info("Driver does not support listener; entering poll only mode") //nolint:noctx
+			config.Logger.Info("Driver does not support listener; entering poll only mode")
 		}
 
 		client.elector = leadership.NewElector(archetype, driver.GetExecutor(), client.notifier, &leadership.Config{

producer.go

@@ -260,9 +260,9 @@ func (p *producer) Start(ctx context.Context) error {
 }
 
 func (p *producer) Stop() {
-	p.Logger.Debug(p.Name+": Stopping", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) //nolint:noctx
+	p.Logger.Debug(p.Name+": Stopping", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load()))
 	p.BaseStartStop.Stop()
-	p.Logger.Debug(p.Name+": Stop returned", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load())) //nolint:noctx
+	p.Logger.Debug(p.Name+": Stop returned", slog.String("queue", p.config.Queue), slog.Int64("id", p.id.Load()))
 }
 
 // Start starts the producer. It backgrounds a goroutine which is stopped when

riverdriver/river_driver_interface.go

@@ -215,7 +215,7 @@ type Executor interface {
 	JobInsertFastManyNoReturning(ctx context.Context, params *JobInsertFastManyParams) (int, error)
 	JobInsertFull(ctx context.Context, params *JobInsertFullParams) (*rivertype.JobRow, error)
 	JobInsertFullMany(ctx context.Context, jobs *JobInsertFullManyParams) ([]*rivertype.JobRow, error)
-	JobKindListByPrefix(ctx context.Context, params *JobKindListByPrefixParams) ([]string, error)
+	JobKindList(ctx context.Context, params *JobKindListParams) ([]string, error)
 	JobList(ctx context.Context, params *JobListParams) ([]*rivertype.JobRow, error)
 	JobRescueMany(ctx context.Context, params *JobRescueManyParams) (*struct{}, error)
 	JobRetry(ctx context.Context, params *JobRetryParams) (*rivertype.JobRow, error)
@@ -261,7 +261,7 @@ type Executor interface {
 	QueueDeleteExpired(ctx context.Context, params *QueueDeleteExpiredParams) ([]string, error)
 	QueueGet(ctx context.Context, params *QueueGetParams) (*rivertype.Queue, error)
 	QueueList(ctx context.Context, params *QueueListParams) ([]*rivertype.Queue, error)
-	QueueNameListByPrefix(ctx context.Context, params *QueueNameListByPrefixParams) ([]string, error)
+	QueueNameList(ctx context.Context, params *QueueNameListParams) ([]string, error)
 	QueuePause(ctx context.Context, params *QueuePauseParams) error
 	QueueResume(ctx context.Context, params *QueueResumeParams) error
 	QueueUpdate(ctx context.Context, params *QueueUpdateParams) (*rivertype.Queue, error)
@@ -481,11 +481,11 @@ type JobInsertFullManyParams struct {
 	Schema string
 }
 
-type JobKindListByPrefixParams struct {
+type JobKindListParams struct {
 	After   string
 	Exclude []string
+	Match   string
 	Max     int
-	Prefix  string
 	Schema  string
 }
 
@@ -782,11 +782,11 @@ type QueueListParams struct {
 	Schema string
 }
 
-type QueueNameListByPrefixParams struct {
+type QueueNameListParams struct {
 	After   string
 	Exclude []string
+	Match   string
 	Max     int
-	Prefix  string
 	Schema  string
 }
 

riverdriver/riverdatabasesql/internal/dbsqlc/river_job.sql.go

@@ -541,12 +541,12 @@ func (e *Executor) JobInsertFullMany(ctx context.Context, params *riverdriver.Jo
 	return sliceutil.MapError(items, jobRowFromInternal)
 }
 
-func (e *Executor) JobKindListByPrefix(ctx context.Context, params *riverdriver.JobKindListByPrefixParams) ([]string, error) {
-	kinds, err := dbsqlc.New().JobKindListByPrefix(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobKindListByPrefixParams{
+func (e *Executor) JobKindList(ctx context.Context, params *riverdriver.JobKindListParams) ([]string, error) {
+	kinds, err := dbsqlc.New().JobKindList(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.JobKindListParams{
 		After:   params.After,
 		Exclude: params.Exclude,
+		Match:   params.Match,
 		Max:     int32(params.Max), //nolint:gosec
-		Prefix:  params.Prefix,
 	})
 	if err != nil {
 		return nil, interpretError(err)
@@ -883,12 +883,12 @@ func (e *Executor) QueueList(ctx context.Context, params *riverdriver.QueueListP
 	return sliceutil.Map(queues, queueFromInternal), nil
 }
 
-func (e *Executor) QueueNameListByPrefix(ctx context.Context, params *riverdriver.QueueNameListByPrefixParams) ([]string, error) {
-	queueNames, err := dbsqlc.New().QueueNameListByPrefix(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.QueueNameListByPrefixParams{
+func (e *Executor) QueueNameList(ctx context.Context, params *riverdriver.QueueNameListParams) ([]string, error) {
+	queueNames, err := dbsqlc.New().QueueNameList(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.QueueNameListParams{
 		After:   params.After,
 		Exclude: params.Exclude,
+		Match:   params.Match,
 		Max:     int32(min(params.Max, math.MaxInt32)), //nolint:gosec
-		Prefix:  params.Prefix,
 	})
 	if err != nil {
 		return nil, interpretError(err)

riverdriver/riverdatabasesql/internal/dbsqlc/river_queue.sql.go

@@ -2356,10 +2356,10 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		assertJobEqualsInput(t, results[1], jobParams2)
 	})
 
-	t.Run("JobKindListByPrefix", func(t *testing.T) { //nolint:dupl
+	t.Run("JobKindList", func(t *testing.T) {
 		t.Parallel()
 
-		t.Run("ListsJobKindsInOrderWithMaxLimit", func(t *testing.T) {
+		t.Run("ListsJobKindsInOrderWithMaxLimit", func(t *testing.T) { //nolint:dupl
 			t.Parallel()
 
 			exec, _ := setup(ctx, t)
@@ -2369,11 +2369,11 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 			job3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("job_bbb")})
 			_ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("different_prefix_job")})
 
-			jobKinds, err := exec.JobKindListByPrefix(ctx, &riverdriver.JobKindListByPrefixParams{
+			jobKinds, err := exec.JobKindList(ctx, &riverdriver.JobKindListParams{
 				After:   "job2",
 				Exclude: nil,
+				Match:   "job",
 				Max:     2,
-				Prefix:  "job",
 				Schema:  "",
 			})
 			require.NoError(t, err)
@@ -2389,16 +2389,37 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 			job2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("job_aaa")})
 			job3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("job_bbb")})
 
-			jobKinds, err := exec.JobKindListByPrefix(ctx, &riverdriver.JobKindListByPrefixParams{
+			jobKinds, err := exec.JobKindList(ctx, &riverdriver.JobKindListParams{
 				After:   "job2",
 				Exclude: []string{job2.Kind},
 				Max:     2,
-				Prefix:  "job",
+				Match:   "job",
 				Schema:  "",
 			})
 			require.NoError(t, err)
 			require.Equal(t, []string{job3.Kind, job1.Kind}, jobKinds)
 		})
+
+		t.Run("ListsJobKindsWithSubstringMatch", func(t *testing.T) { //nolint:dupl
+			t.Parallel()
+
+			exec, _ := setup(ctx, t)
+
+			_ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("mid_job_kind")})
+			job2 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("prefix_job")})
+			job3 := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("suffix_job")})
+			_ = testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Kind: ptrutil.Ptr("nojobhere")})
+
+			jobKinds, err := exec.JobKindList(ctx, &riverdriver.JobKindListParams{
+				After:   "",
+				Exclude: nil,
+				Match:   "fix",
+				Max:     3,
+				Schema:  "",
+			})
+			require.NoError(t, err)
+			require.Equal(t, []string{job2.Kind, job3.Kind}, jobKinds)
+		})
 	})
 
 	t.Run("JobList", func(t *testing.T) {
@@ -4155,10 +4176,10 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 		requireQueuesEqual(t, queue3, queues[2])
 	})
 
-	t.Run("QueueNameListByPrefix", func(t *testing.T) { //nolint:dupl
+	t.Run("QueueNameList", func(t *testing.T) {
 		t.Parallel()
 
-		t.Run("ListsQueuesInOrderWithMaxLimit", func(t *testing.T) {
+		t.Run("ListsQueuesInOrderWithMaxLimit", func(t *testing.T) { //nolint:dupl
 			t.Parallel()
 
 			exec, _ := setup(ctx, t)
@@ -4168,11 +4189,11 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 			queue3 := testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("queue_bbb")})
 			_ = testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("different_prefix_queue")})
 
-			queueNames, err := exec.QueueNameListByPrefix(ctx, &riverdriver.QueueNameListByPrefixParams{
+			queueNames, err := exec.QueueNameList(ctx, &riverdriver.QueueNameListParams{
 				After:   "queue2",
 				Exclude: nil,
+				Match:   "queue",
 				Max:     2,
-				Prefix:  "queue",
 				Schema:  "",
 			})
 			require.NoError(t, err)
@@ -4188,16 +4209,36 @@ func Exercise[TTx any](ctx context.Context, t *testing.T,
 			queue2 := testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("queue_aaa")})
 			queue3 := testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("queue_bbb")})
 
-			queueNames, err := exec.QueueNameListByPrefix(ctx, &riverdriver.QueueNameListByPrefixParams{
+			queueNames, err := exec.QueueNameList(ctx, &riverdriver.QueueNameListParams{
 				After:   "queue2",
 				Exclude: []string{queue2.Name},
+				Match:   "queue",
 				Max:     2,
-				Prefix:  "queue",
 				Schema:  "",
 			})
 			require.NoError(t, err)
 			require.Equal(t, []string{queue3.Name, queue1.Name}, queueNames)
 		})
+
+		t.Run("ListsQueuesWithSubstringMatch", func(t *testing.T) {
+			t.Parallel()
+
+			exec, _ := setup(ctx, t)
+
+			queue1 := testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("prefix_queue")})
+			_ = testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("another_queue")})
+			queue3 := testfactory.Queue(ctx, t, exec, &testfactory.QueueOpts{Name: ptrutil.Ptr("suffix_queue")})
+
+			queueNames, err := exec.QueueNameList(ctx, &riverdriver.QueueNameListParams{
+				After:   "",
+				Exclude: nil,
+				Match:   "fix",
+				Max:     3,
+				Schema:  "",
+			})
+			require.NoError(t, err)
+			require.Equal(t, []string{queue1.Name, queue3.Name}, queueNames)
+		})
 	})
 
 	t.Run("QueuePause", func(t *testing.T) {

riverdriver/riverdatabasesql/river_database_sql_driver.go

@@ -458,10 +458,10 @@ SELECT
 FROM raw_job_data
 RETURNING *;
 
--- name: JobKindListByPrefix :many
+-- name: JobKindList :many
 SELECT DISTINCT ON (kind) kind
 FROM /* TEMPLATE: schema */river_job
-WHERE (@prefix = '' OR kind ILIKE @prefix || '%')
+WHERE (@match = '' OR kind ILIKE '%' || @match || '%')
     AND (@after = '' OR kind > @after)
     AND (@exclude::text[] IS NULL OR kind != ALL(@exclude))
 ORDER BY kind ASC