river_database_sql.go

// Package riverdatabasesql bundles a River driver for Go's built-in
// database/sql, making it interoperable with ORMs like Bun and GORM. It's
// generally still powered under the hood by Pgx because it's the only
// maintained, fully functional Postgres driver in the Go ecosystem, but it uses
// some lib/pq constructs internally by virtue of being implemented with Sqlc.
package riverdatabasesql

import (
	"context"
	"database/sql"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"math"
	"strings"
	"time"

	"github.com/lib/pq"

	"github.com/riverqueue/river/riverdriver"
	"github.com/riverqueue/river/riverdriver/riverdatabasesql/internal/dbsqlc"
	"github.com/riverqueue/river/rivershared/util/sliceutil"
	"github.com/riverqueue/river/rivershared/util/valutil"
	"github.com/riverqueue/river/rivertype"
)

//go:embed migration/*/*.sql
var migrationFS embed.FS

// Driver is an implementation of riverdriver.Driver for database/sql.
type Driver struct {
	dbPool  *sql.DB
	queries *dbsqlc.Queries
}

// New returns a new database/sql River driver for use with River.
//
// It takes an sql.DB to use for use with River. The pool should already be
// configured to use the schema specified in the client's Schema field. The pool
// must not be closed while associated River objects are running.
func New(dbPool *sql.DB) *Driver {
	return &Driver{dbPool: dbPool, queries: dbsqlc.New()}
}

func (d *Driver) GetExecutor() riverdriver.Executor {
	return &Executor{d.dbPool, d.dbPool, dbsqlc.New()}
}

func (d *Driver) GetListener() riverdriver.Listener { panic(riverdriver.ErrNotImplemented) }
func (d *Driver) GetMigrationFS(line string) fs.FS {
	if line == riverdriver.MigrationLineMain {
		return migrationFS
	}
	panic("migration line does not exist: " + line)
}
func (d *Driver) GetMigrationLines() []string { return []string{riverdriver.MigrationLineMain} }
func (d *Driver) HasPool() bool               { return d.dbPool != nil }
func (d *Driver) SupportsListener() bool      { return false }

func (d *Driver) UnwrapExecutor(tx *sql.Tx) riverdriver.ExecutorTx {
	return &ExecutorTx{Executor: Executor{nil, tx, dbsqlc.New()}, tx: tx}
}

type Executor struct {
	dbPool  *sql.DB
	dbtx    dbsqlc.DBTX
	queries *dbsqlc.Queries
}

func (e *Executor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) {
	tx, err := e.dbPool.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	return &ExecutorTx{Executor: Executor{nil, tx, e.queries}, tx: tx}, nil
}

func (e *Executor) ColumnExists(ctx context.Context, tableName, columnName string) (bool, error) {
	exists, err := e.queries.ColumnExists(ctx, e.dbtx, &dbsqlc.ColumnExistsParams{
		ColumnName: columnName,
		TableName:  tableName,
	})
	return exists, interpretError(err)
}

func (e *Executor) Exec(ctx context.Context, sql string) (struct{}, error) {
	_, err := e.dbtx.ExecContext(ctx, sql)
	return struct{}{}, interpretError(err)
}

func (e *Executor) JobCancel(ctx context.Context, params *riverdriver.JobCancelParams) (*rivertype.JobRow, error) {
	cancelledAt, err := params.CancelAttemptedAt.MarshalJSON()
	if err != nil {
		return nil, err
	}

	job, err := e.queries.JobCancel(ctx, e.dbtx, &dbsqlc.JobCancelParams{
		ID:                params.ID,
		CancelAttemptedAt: string(cancelledAt),
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobCountByState(ctx context.Context, state rivertype.JobState) (int, error) {
	numJobs, err := e.queries.JobCountByState(ctx, e.dbtx, dbsqlc.RiverJobState(state))
	if err != nil {
		return 0, err
	}
	return int(numJobs), nil
}

func (e *Executor) JobDelete(ctx context.Context, id int64) (*rivertype.JobRow, error) {
	job, err := e.queries.JobDelete(ctx, e.dbtx, id)
	if err != nil {
		return nil, interpretError(err)
	}
	if job.State == dbsqlc.RiverJobStateRunning {
		return nil, rivertype.ErrJobRunning
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobDeleteBefore(ctx context.Context, params *riverdriver.JobDeleteBeforeParams) (int, error) {
	numDeleted, err := e.queries.JobDeleteBefore(ctx, e.dbtx, &dbsqlc.JobDeleteBeforeParams{
		CancelledFinalizedAtHorizon: params.CancelledFinalizedAtHorizon,
		CompletedFinalizedAtHorizon: params.CompletedFinalizedAtHorizon,
		DiscardedFinalizedAtHorizon: params.DiscardedFinalizedAtHorizon,
		Max:                         int64(params.Max),
	})
	return int(numDeleted), interpretError(err)
}

func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) {
	jobs, err := e.queries.JobGetAvailable(ctx, e.dbtx, &dbsqlc.JobGetAvailableParams{
		AttemptedBy: params.AttemptedBy,
		Max:         int32(params.Max),
		Queue:       params.Queue,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return mapSliceError(jobs, jobRowFromInternal)
}

func (e *Executor) JobGetByID(ctx context.Context, id int64) (*rivertype.JobRow, error) {
	job, err := e.queries.JobGetByID(ctx, e.dbtx, id)
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobGetByIDMany(ctx context.Context, id []int64) ([]*rivertype.JobRow, error) {
	jobs, err := e.queries.JobGetByIDMany(ctx, e.dbtx, id)
	if err != nil {
		return nil, interpretError(err)
	}
	return mapSliceError(jobs, jobRowFromInternal)
}

func (e *Executor) JobGetByKindAndUniqueProperties(ctx context.Context, params *riverdriver.JobGetByKindAndUniquePropertiesParams) (*rivertype.JobRow, error) {
	job, err := e.queries.JobGetByKindAndUniqueProperties(ctx, e.dbtx, &dbsqlc.JobGetByKindAndUniquePropertiesParams{
		Args:           valutil.ValOrDefault(string(params.Args), "{}"),
		ByArgs:         params.ByArgs,
		ByCreatedAt:    params.ByCreatedAt,
		ByQueue:        params.ByQueue,
		ByState:        params.ByState,
		CreatedAtBegin: params.CreatedAtBegin,
		CreatedAtEnd:   params.CreatedAtEnd,
		Kind:           params.Kind,
		Queue:          params.Queue,
		State:          params.State,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobGetByKindMany(ctx context.Context, kind []string) ([]*rivertype.JobRow, error) {
	jobs, err := e.queries.JobGetByKindMany(ctx, e.dbtx, kind)
	if err != nil {
		return nil, interpretError(err)
	}
	return mapSliceError(jobs, jobRowFromInternal)
}

func (e *Executor) JobGetStuck(ctx context.Context, params *riverdriver.JobGetStuckParams) ([]*rivertype.JobRow, error) {
	jobs, err := e.queries.JobGetStuck(ctx, e.dbtx, &dbsqlc.JobGetStuckParams{Max: int32(params.Max), StuckHorizon: params.StuckHorizon})
	if err != nil {
		return nil, interpretError(err)
	}
	return mapSliceError(jobs, jobRowFromInternal)
}

func (e *Executor) JobInsertFast(ctx context.Context, params *riverdriver.JobInsertFastParams) (*rivertype.JobRow, error) {
	job, err := e.queries.JobInsertFast(ctx, e.dbtx, &dbsqlc.JobInsertFastParams{
		Args:        string(params.EncodedArgs),
		CreatedAt:   params.CreatedAt,
		Kind:        params.Kind,
		MaxAttempts: int16(min(params.MaxAttempts, math.MaxInt16)),
		Metadata:    valutil.ValOrDefault(string(params.Metadata), "{}"),
		Priority:    int16(min(params.Priority, math.MaxInt16)),
		Queue:       params.Queue,
		ScheduledAt: params.ScheduledAt,
		State:       dbsqlc.RiverJobState(params.State),
		Tags:        params.Tags,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobInsertFastMany(ctx context.Context, params []*riverdriver.JobInsertFastParams) (int, error) {
	insertJobsParams := &dbsqlc.JobInsertFastManyParams{
		Args:        make([]string, len(params)),
		Kind:        make([]string, len(params)),
		MaxAttempts: make([]int16, len(params)),
		Metadata:    make([]string, len(params)),
		Priority:    make([]int16, len(params)),
		Queue:       make([]string, len(params)),
		ScheduledAt: make([]time.Time, len(params)),
		State:       make([]dbsqlc.RiverJobState, len(params)),
		Tags:        make([]string, len(params)),
	}

	for i := 0; i < len(params); i++ {
		params := params[i]

		var scheduledAt time.Time
		if params.ScheduledAt != nil {
			scheduledAt = *params.ScheduledAt
		}

		tags := params.Tags
		if tags == nil {
			tags = []string{}
		}

		insertJobsParams.Args[i] = valutil.ValOrDefault(string(params.EncodedArgs), "{}")
		insertJobsParams.Kind[i] = params.Kind
		insertJobsParams.MaxAttempts[i] = int16(min(params.MaxAttempts, math.MaxInt16))
		insertJobsParams.Metadata[i] = valutil.ValOrDefault(string(params.Metadata), "{}")
		insertJobsParams.Priority[i] = int16(min(params.Priority, math.MaxInt16))
		insertJobsParams.Queue[i] = params.Queue
		insertJobsParams.ScheduledAt[i] = scheduledAt
		insertJobsParams.State[i] = dbsqlc.RiverJobState(params.State)
		insertJobsParams.Tags[i] = strings.Join(tags, ",")
	}

	numInserted, err := e.queries.JobInsertFastMany(ctx, e.dbtx, insertJobsParams)
	if err != nil {
		return 0, interpretError(err)
	}

	return int(numInserted), nil
}

func (e *Executor) JobInsertFull(ctx context.Context, params *riverdriver.JobInsertFullParams) (*rivertype.JobRow, error) {
	job, err := e.queries.JobInsertFull(ctx, e.dbtx, &dbsqlc.JobInsertFullParams{
		Attempt:     int16(params.Attempt),
		AttemptedAt: params.AttemptedAt,
		Args:        string(params.EncodedArgs),
		CreatedAt:   params.CreatedAt,
		Errors:      sliceutil.Map(params.Errors, func(e []byte) string { return string(e) }),
		FinalizedAt: params.FinalizedAt,
		Kind:        params.Kind,
		MaxAttempts: int16(min(params.MaxAttempts, math.MaxInt16)),
		Metadata:    valutil.ValOrDefault(string(params.Metadata), "{}"),
		Priority:    int16(min(params.Priority, math.MaxInt16)),
		Queue:       params.Queue,
		ScheduledAt: params.ScheduledAt,
		State:       dbsqlc.RiverJobState(params.State),
		Tags:        params.Tags,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

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
	}

	rows, err := e.dbtx.QueryContext(ctx, query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var items []*dbsqlc.RiverJob
	for rows.Next() {
		var i dbsqlc.RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			pq.Array(&i.AttemptedBy),
			&i.CreatedAt,
			pq.Array(&i.Errors),
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			pq.Array(&i.Tags),
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Err(); err != nil {
		return nil, interpretError(err)
	}

	return mapSliceError(items, jobRowFromInternal)
}

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"
}

func (e *Executor) JobRescueMany(ctx context.Context, params *riverdriver.JobRescueManyParams) (*struct{}, error) {
	err := e.queries.JobRescueMany(ctx, e.dbtx, &dbsqlc.JobRescueManyParams{
		ID:          params.ID,
		Error:       sliceutil.Map(params.Error, func(e []byte) string { return string(e) }),
		FinalizedAt: params.FinalizedAt,
		ScheduledAt: params.ScheduledAt,
		State:       params.State,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return &struct{}{}, nil
}

func (e *Executor) JobRetry(ctx context.Context, id int64) (*rivertype.JobRow, error) {
	job, err := e.queries.JobRetry(ctx, e.dbtx, id)
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobSchedule(ctx context.Context, params *riverdriver.JobScheduleParams) ([]*rivertype.JobRow, error) {
	jobs, err := e.queries.JobSchedule(ctx, e.dbtx, &dbsqlc.JobScheduleParams{
		Max: int64(params.Max),
		Now: params.Now,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return mapSliceError(jobs, jobRowFromInternal)
}

func (e *Executor) JobSetCompleteIfRunningMany(ctx context.Context, params *riverdriver.JobSetCompleteIfRunningManyParams) ([]*rivertype.JobRow, error) {
	jobs, err := e.queries.JobSetCompleteIfRunningMany(ctx, e.dbtx, &dbsqlc.JobSetCompleteIfRunningManyParams{
		ID:          params.ID,
		FinalizedAt: params.FinalizedAt,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return mapSliceError(jobs, jobRowFromInternal)
}

func (e *Executor) JobSetStateIfRunning(ctx context.Context, params *riverdriver.JobSetStateIfRunningParams) (*rivertype.JobRow, error) {
	var maxAttempts int16
	if params.MaxAttempts != nil {
		maxAttempts = int16(*params.MaxAttempts)
	}

	job, err := e.queries.JobSetStateIfRunning(ctx, e.dbtx, &dbsqlc.JobSetStateIfRunningParams{
		ID:                  params.ID,
		ErrorDoUpdate:       params.ErrData != nil,
		Error:               valutil.ValOrDefault(string(params.ErrData), "{}"),
		FinalizedAtDoUpdate: params.FinalizedAt != nil,
		FinalizedAt:         params.FinalizedAt,
		MaxAttemptsUpdate:   params.MaxAttempts != nil,
		MaxAttempts:         maxAttempts,
		ScheduledAtDoUpdate: params.ScheduledAt != nil,
		ScheduledAt:         params.ScheduledAt,
		State:               dbsqlc.RiverJobState(params.State),
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return jobRowFromInternal(job)
}

func (e *Executor) JobUpdate(ctx context.Context, params *riverdriver.JobUpdateParams) (*rivertype.JobRow, error) {
	job, err := e.queries.JobUpdate(ctx, e.dbtx, &dbsqlc.JobUpdateParams{
		ID:                  params.ID,
		AttemptedAtDoUpdate: params.AttemptedAtDoUpdate,
		AttemptedAt:         params.AttemptedAt,
		AttemptDoUpdate:     params.AttemptDoUpdate,
		Attempt:             int16(params.Attempt),
		ErrorsDoUpdate:      params.ErrorsDoUpdate,
		Errors:              sliceutil.Map(params.Errors, func(e []byte) string { return string(e) }),
		FinalizedAtDoUpdate: params.FinalizedAtDoUpdate,
		FinalizedAt:         params.FinalizedAt,
		StateDoUpdate:       params.StateDoUpdate,
		State:               dbsqlc.RiverJobState(params.State),
	})
	if err != nil {
		return nil, interpretError(err)
	}

	return jobRowFromInternal(job)
}

func (e *Executor) LeaderAttemptElect(ctx context.Context, params *riverdriver.LeaderElectParams) (bool, error) {
	numElectionsWon, err := e.queries.LeaderAttemptElect(ctx, e.dbtx, &dbsqlc.LeaderAttemptElectParams{
		LeaderID: params.LeaderID,
		TTL:      params.TTL,
	})
	if err != nil {
		return false, interpretError(err)
	}
	return numElectionsWon > 0, nil
}

func (e *Executor) LeaderAttemptReelect(ctx context.Context, params *riverdriver.LeaderElectParams) (bool, error) {
	numElectionsWon, err := e.queries.LeaderAttemptReelect(ctx, e.dbtx, &dbsqlc.LeaderAttemptReelectParams{
		LeaderID: params.LeaderID,
		TTL:      params.TTL,
	})
	if err != nil {
		return false, interpretError(err)
	}
	return numElectionsWon > 0, nil
}

func (e *Executor) LeaderDeleteExpired(ctx context.Context) (int, error) {
	numDeleted, err := e.queries.LeaderDeleteExpired(ctx, e.dbtx)
	if err != nil {
		return 0, interpretError(err)
	}
	return int(numDeleted), nil
}

func (e *Executor) LeaderGetElectedLeader(ctx context.Context) (*riverdriver.Leader, error) {
	leader, err := e.queries.LeaderGetElectedLeader(ctx, e.dbtx)
	if err != nil {
		return nil, interpretError(err)
	}
	return leaderFromInternal(leader), nil
}

func (e *Executor) LeaderInsert(ctx context.Context, params *riverdriver.LeaderInsertParams) (*riverdriver.Leader, error) {
	leader, err := e.queries.LeaderInsert(ctx, e.dbtx, &dbsqlc.LeaderInsertParams{
		ElectedAt: params.ElectedAt,
		ExpiresAt: params.ExpiresAt,
		LeaderID:  params.LeaderID,
		TTL:       params.TTL,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return leaderFromInternal(leader), nil
}

func (e *Executor) LeaderResign(ctx context.Context, params *riverdriver.LeaderResignParams) (bool, error) {
	numResigned, err := e.queries.LeaderResign(ctx, e.dbtx, &dbsqlc.LeaderResignParams{
		LeaderID:        params.LeaderID,
		LeadershipTopic: params.LeadershipTopic,
	})
	if err != nil {
		return false, interpretError(err)
	}
	return numResigned > 0, nil
}

func (e *Executor) MigrationDeleteAssumingMainMany(ctx context.Context, versions []int) ([]*riverdriver.Migration, error) {
	migrations, err := e.queries.RiverMigrationDeleteAssumingMainMany(ctx, e.dbtx,
		sliceutil.Map(versions, func(v int) int64 { return int64(v) }))
	if err != nil {
		return nil, interpretError(err)
	}
	return sliceutil.Map(migrations, func(internal *dbsqlc.RiverMigrationDeleteAssumingMainManyRow) *riverdriver.Migration {
		return &riverdriver.Migration{
			ID:        int(internal.ID),
			CreatedAt: internal.CreatedAt.UTC(),
			Line:      riverdriver.MigrationLineMain,
			Version:   int(internal.Version),
		}
	}), nil
}

func (e *Executor) MigrationDeleteByLineAndVersionMany(ctx context.Context, line string, versions []int) ([]*riverdriver.Migration, error) {
	migrations, err := e.queries.RiverMigrationDeleteByLineAndVersionMany(ctx, e.dbtx, &dbsqlc.RiverMigrationDeleteByLineAndVersionManyParams{
		Line:    line,
		Version: sliceutil.Map(versions, func(v int) int64 { return int64(v) }),
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return sliceutil.Map(migrations, migrationFromInternal), nil
}

func (e *Executor) MigrationGetAllAssumingMain(ctx context.Context) ([]*riverdriver.Migration, error) {
	migrations, err := e.queries.RiverMigrationGetAllAssumingMain(ctx, e.dbtx)
	if err != nil {
		return nil, interpretError(err)
	}
	return sliceutil.Map(migrations, func(internal *dbsqlc.RiverMigrationGetAllAssumingMainRow) *riverdriver.Migration {
		return &riverdriver.Migration{
			ID:        int(internal.ID),
			CreatedAt: internal.CreatedAt.UTC(),
			Line:      riverdriver.MigrationLineMain,
			Version:   int(internal.Version),
		}
	}), nil
}

func (e *Executor) MigrationGetByLine(ctx context.Context, line string) ([]*riverdriver.Migration, error) {
	migrations, err := e.queries.RiverMigrationGetByLine(ctx, e.dbtx, line)
	if err != nil {
		return nil, interpretError(err)
	}
	return sliceutil.Map(migrations, migrationFromInternal), nil
}

func (e *Executor) MigrationInsertMany(ctx context.Context, line string, versions []int) ([]*riverdriver.Migration, error) {
	migrations, err := e.queries.RiverMigrationInsertMany(ctx, e.dbtx, &dbsqlc.RiverMigrationInsertManyParams{
		Line:    line,
		Version: sliceutil.Map(versions, func(v int) int64 { return int64(v) }),
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return sliceutil.Map(migrations, migrationFromInternal), nil
}

func (e *Executor) MigrationInsertManyAssumingMain(ctx context.Context, versions []int) ([]*riverdriver.Migration, error) {
	migrations, err := e.queries.RiverMigrationInsertManyAssumingMain(ctx, e.dbtx,
		sliceutil.Map(versions, func(v int) int64 { return int64(v) }),
	)
	if err != nil {
		return nil, interpretError(err)
	}
	return sliceutil.Map(migrations, func(internal *dbsqlc.RiverMigrationInsertManyAssumingMainRow) *riverdriver.Migration {
		return &riverdriver.Migration{
			ID:        int(internal.ID),
			CreatedAt: internal.CreatedAt.UTC(),
			Line:      riverdriver.MigrationLineMain,
			Version:   int(internal.Version),
		}
	}), nil
}

func (e *Executor) NotifyMany(ctx context.Context, params *riverdriver.NotifyManyParams) error {
	return e.queries.PGNotifyMany(ctx, e.dbtx, &dbsqlc.PGNotifyManyParams{
		Payload: params.Payload,
		Topic:   params.Topic,
	})
}

func (e *Executor) PGAdvisoryXactLock(ctx context.Context, key int64) (*struct{}, error) {
	err := e.queries.PGAdvisoryXactLock(ctx, e.dbtx, key)
	return &struct{}{}, interpretError(err)
}

func (e *Executor) QueueCreateOrSetUpdatedAt(ctx context.Context, params *riverdriver.QueueCreateOrSetUpdatedAtParams) (*rivertype.Queue, error) {
	queue, err := e.queries.QueueCreateOrSetUpdatedAt(ctx, e.dbtx, &dbsqlc.QueueCreateOrSetUpdatedAtParams{
		Metadata:  valutil.ValOrDefault(string(params.Metadata), "{}"),
		Name:      params.Name,
		PausedAt:  params.PausedAt,
		UpdatedAt: params.UpdatedAt,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	return queueFromInternal(queue), nil
}

func (e *Executor) QueueDeleteExpired(ctx context.Context, params *riverdriver.QueueDeleteExpiredParams) ([]string, error) {
	queues, err := e.queries.QueueDeleteExpired(ctx, e.dbtx, &dbsqlc.QueueDeleteExpiredParams{
		Max:              int64(params.Max),
		UpdatedAtHorizon: params.UpdatedAtHorizon,
	})
	if err != nil {
		return nil, interpretError(err)
	}
	queueNames := make([]string, len(queues))
	for i, q := range queues {
		queueNames[i] = q.Name
	}
	return queueNames, nil
}

func (e *Executor) QueueGet(ctx context.Context, name string) (*rivertype.Queue, error) {
	queue, err := e.queries.QueueGet(ctx, e.dbtx, name)
	if err != nil {
		return nil, interpretError(err)
	}
	return queueFromInternal(queue), nil
}

func (e *Executor) QueueList(ctx context.Context, limit int) ([]*rivertype.Queue, error) {
	internalQueues, err := e.queries.QueueList(ctx, e.dbtx, int32(limit))
	if err != nil {
		return nil, interpretError(err)
	}
	queues := make([]*rivertype.Queue, len(internalQueues))
	for i, q := range internalQueues {
		queues[i] = queueFromInternal(q)
	}
	return queues, nil
}

func (e *Executor) QueuePause(ctx context.Context, name string) error {
	res, err := e.queries.QueuePause(ctx, e.dbtx, name)
	if err != nil {
		return interpretError(err)
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return interpretError(err)
	}
	if rowsAffected == 0 && name != riverdriver.AllQueuesString {
		return rivertype.ErrNotFound
	}
	return nil
}

func (e *Executor) QueueResume(ctx context.Context, name string) error {
	res, err := e.queries.QueueResume(ctx, e.dbtx, name)
	if err != nil {
		return interpretError(err)
	}
	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return interpretError(err)
	}
	if rowsAffected == 0 && name != riverdriver.AllQueuesString {
		return rivertype.ErrNotFound
	}
	return nil
}

func (e *Executor) TableExists(ctx context.Context, tableName string) (bool, error) {
	exists, err := e.queries.TableExists(ctx, e.dbtx, tableName)
	return exists, interpretError(err)
}

type ExecutorTx struct {
	Executor
	tx *sql.Tx
}

func (t *ExecutorTx) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) {
	return (&ExecutorSubTx{Executor: Executor{nil, t.tx, t.queries}, savepointNum: 0, single: &singleTransaction{}, tx: t.tx}).Begin(ctx)
}

func (t *ExecutorTx) Commit(ctx context.Context) error {
	// unfortunately, `database/sql` does not take a context ...
	return t.tx.Commit()
}

func (t *ExecutorTx) Rollback(ctx context.Context) error {
	// unfortunately, `database/sql` does not take a context ...
	return t.tx.Rollback()
}

type ExecutorSubTx struct {
	Executor
	savepointNum int
	single       *singleTransaction
	tx           *sql.Tx
}

const savepointPrefix = "river_savepoint_"

func (t *ExecutorSubTx) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) {
	if err := t.single.begin(); err != nil {
		return nil, err
	}

	nextSavepointNum := t.savepointNum + 1
	_, err := t.Exec(ctx, fmt.Sprintf("SAVEPOINT %s%02d", savepointPrefix, nextSavepointNum))
	if err != nil {
		return nil, err
	}
	return &ExecutorSubTx{Executor: Executor{nil, t.tx, t.queries}, savepointNum: nextSavepointNum, single: &singleTransaction{parent: t.single}, tx: t.tx}, nil
}

func (t *ExecutorSubTx) Commit(ctx context.Context) error {
	defer t.single.setDone()

	if t.single.done {
		return errors.New("tx is closed") // mirrors pgx's behavior for this condition
	}

	// Release destroys a savepoint, keeping all the effects of commands that
	// were run within it (so it's effectively COMMIT for savepoints).
	_, err := t.Exec(ctx, fmt.Sprintf("RELEASE %s%02d", savepointPrefix, t.savepointNum))
	if err != nil {
		return err
	}

	return nil
}

func (t *ExecutorSubTx) Rollback(ctx context.Context) error {
	defer t.single.setDone()

	if t.single.done {
		return errors.New("tx is closed") // mirrors pgx's behavior for this condition
	}

	_, err := t.Exec(ctx, fmt.Sprintf("ROLLBACK TO %s%02d", savepointPrefix, t.savepointNum))
	if err != nil {
		return err
	}

	return nil
}

func interpretError(err error) error {
	if errors.Is(err, sql.ErrNoRows) {
		return rivertype.ErrNotFound
	}
	return err
}

// Not strictly necessary, but a small struct designed to help us route out
// problems where `Begin` might be called multiple times on the same
// subtransaction, which would silently produce the wrong result.
type singleTransaction struct {
	done            bool
	parent          *singleTransaction
	subTxInProgress bool
}

func (t *singleTransaction) begin() error {
	if t.subTxInProgress {
		return errors.New("subtransaction already in progress")
	}
	t.subTxInProgress = true
	return nil
}

func (t *singleTransaction) setDone() {
	t.done = true
	if t.parent != nil {
		t.parent.subTxInProgress = false
	}
}

func jobRowFromInternal(internal *dbsqlc.RiverJob) (*rivertype.JobRow, error) {
	var attemptedAt *time.Time
	if internal.AttemptedAt != nil {
		t := internal.AttemptedAt.UTC()
		attemptedAt = &t
	}

	errors := make([]rivertype.AttemptError, len(internal.Errors))
	for i, rawError := range internal.Errors {
		if err := json.Unmarshal([]byte(rawError), &errors[i]); err != nil {
			return nil, err
		}
	}

	var finalizedAt *time.Time
	if internal.FinalizedAt != nil {
		t := internal.FinalizedAt.UTC()
		finalizedAt = &t
	}

	return &rivertype.JobRow{
		ID:          internal.ID,
		Attempt:     max(int(internal.Attempt), 0),
		AttemptedAt: attemptedAt,
		AttemptedBy: internal.AttemptedBy,
		CreatedAt:   internal.CreatedAt.UTC(),
		EncodedArgs: []byte(internal.Args),
		Errors:      errors,
		FinalizedAt: finalizedAt,
		Kind:        internal.Kind,
		MaxAttempts: max(int(internal.MaxAttempts), 0),
		Metadata:    []byte(internal.Metadata),
		Priority:    max(int(internal.Priority), 0),
		Queue:       internal.Queue,
		ScheduledAt: internal.ScheduledAt.UTC(),
		State:       rivertype.JobState(internal.State),
		Tags:        internal.Tags,
	}, nil
}

func leaderFromInternal(internal *dbsqlc.RiverLeader) *riverdriver.Leader {
	return &riverdriver.Leader{
		ElectedAt: internal.ElectedAt.UTC(),
		ExpiresAt: internal.ExpiresAt.UTC(),
		LeaderID:  internal.LeaderID,
	}
}

// mapSliceError manipulates a slice and transforms it to a slice of another
// type, returning the first error that occurred invoking the map function, if
// there was one.
func mapSliceError[T any, R any](collection []T, mapFunc func(T) (R, error)) ([]R, error) {
	if collection == nil {
		return nil, nil
	}

	result := make([]R, len(collection))

	for i, item := range collection {
		var err error
		result[i], err = mapFunc(item)
		if err != nil {
			return nil, err
		}
	}

	return result, nil
}

func migrationFromInternal(internal *dbsqlc.RiverMigration) *riverdriver.Migration {
	return &riverdriver.Migration{
		ID:        int(internal.ID),
		CreatedAt: internal.CreatedAt.UTC(),
		Line:      internal.Line,
		Version:   int(internal.Version),
	}
}

func queueFromInternal(internal *dbsqlc.RiverQueue) *rivertype.Queue {
	var pausedAt *time.Time
	if internal.PausedAt != nil {
		t := internal.PausedAt.UTC()
		pausedAt = &t
	}
	return &rivertype.Queue{
		CreatedAt: internal.CreatedAt.UTC(),
		Metadata:  []byte(internal.Metadata),
		Name:      internal.Name,
		PausedAt:  pausedAt,
		UpdatedAt: internal.UpdatedAt.UTC(),
	}
}