river_job.sql.go

// Code generated by sqlc. DO NOT EDIT.
// versions:
//   sqlc v1.29.0
// source: river_job.sql

package dbsqlc

import (
	"context"
	"database/sql"
	"strings"
	"time"
)

const jobCancel = `-- name: JobCancel :one
UPDATE /* TEMPLATE: schema */river_job
SET
    -- If the job is actively running, we want to let its current client and
    -- producer handle the cancellation. Otherwise, immediately cancel it.
    state = CASE WHEN state = 'running' THEN state ELSE 'cancelled' END,
    finalized_at = CASE WHEN state = 'running' THEN finalized_at ELSE coalesce(cast(?1 AS text), datetime('now', 'subsec')) END,
    -- Mark the job as cancelled by query so that the rescuer knows not to
    -- rescue it, even if it gets stuck in the running state:
    metadata = json_set(metadata, '$.cancel_attempted_at', cast(?2 AS text))
WHERE id = ?3
    AND state NOT IN ('cancelled', 'completed', 'discarded')
    AND finalized_at IS NULL
RETURNING 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
`

type JobCancelParams struct {
	Now               *string
	CancelAttemptedAt string
	ID                int64
}

// Differs by necessity from other drivers because SQLite doesn't support
// `UPDATE` inside CTEs so we can't retry if running but select otherwise.
// Instead, the driver uses a transaction to optimisticaly try an update, but
// perform a subsequent fetch on a not found to return the right status.
//
// I had to invert the last 'AND' expression below (was an 'ANT NOT) due to an
// sqlc bug. Something about sqlc's SQLite parser cannot detect a parameter
// inside an `AND NOT`.
func (q *Queries) JobCancel(ctx context.Context, db DBTX, arg *JobCancelParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobCancel, arg.Now, arg.CancelAttemptedAt, arg.ID)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobCountByState = `-- name: JobCountByState :one
SELECT count(*)
FROM /* TEMPLATE: schema */river_job
WHERE state = ?1
`

func (q *Queries) JobCountByState(ctx context.Context, db DBTX, state string) (int64, error) {
	row := db.QueryRowContext(ctx, jobCountByState, state)
	var count int64
	err := row.Scan(&count)
	return count, err
}

const jobDelete = `-- name: JobDelete :one
DELETE
FROM /* TEMPLATE: schema */river_job
WHERE id = ?1
    -- Do not touch running jobs:
    AND river_job.state != 'running'
RETURNING 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
`

// Differs by necessity from other drivers because SQLite doesn't support
// `DELETE` inside CTEs so we can't delete if running but select otherwise.
// Instead, the driver uses a transaction to optimisticaly try a delete, but
// perform a subsequent fetch on a not found to return the right status.
func (q *Queries) JobDelete(ctx context.Context, db DBTX, id int64) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobDelete, id)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobDeleteBefore = `-- name: JobDeleteBefore :execresult
DELETE FROM /* TEMPLATE: schema */river_job
WHERE id IN (
    SELECT id
    FROM /* TEMPLATE: schema */river_job
    WHERE
        (state = 'cancelled' AND finalized_at < cast(?1 AS text)) OR
        (state = 'completed' AND finalized_at < cast(?2 AS text)) OR
        (state = 'discarded' AND finalized_at < cast(?3 AS text))
    ORDER BY id
    LIMIT ?4
)
`

type JobDeleteBeforeParams struct {
	CancelledFinalizedAtHorizon string
	CompletedFinalizedAtHorizon string
	DiscardedFinalizedAtHorizon string
	Max                         int64
}

func (q *Queries) JobDeleteBefore(ctx context.Context, db DBTX, arg *JobDeleteBeforeParams) (sql.Result, error) {
	return db.ExecContext(ctx, jobDeleteBefore,
		arg.CancelledFinalizedAtHorizon,
		arg.CompletedFinalizedAtHorizon,
		arg.DiscardedFinalizedAtHorizon,
		arg.Max,
	)
}

const jobGetAvailable = `-- name: JobGetAvailable :many
UPDATE /* TEMPLATE: schema */river_job
SET
    attempt = river_job.attempt + 1,
    attempted_at = coalesce(cast(?1 AS text), datetime('now', 'subsec')),
    attempted_by = json_insert(coalesce(attempted_by, json('[]')), '$[#]', cast(?2 AS text)),
    state = 'running'
WHERE id IN (
    SELECT id
    FROM /* TEMPLATE: schema */river_job
    WHERE
        priority >= 0
        AND river_job.queue = ?3
        AND scheduled_at <= coalesce(cast(?1 AS text), datetime('now', 'subsec'))
        AND state = 'available'
    ORDER BY
        priority ASC,
        scheduled_at ASC,
        id ASC
    LIMIT ?4
)
RETURNING 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
`

type JobGetAvailableParams struct {
	Now         *string
	AttemptedBy string
	Queue       string
	Max         int64
}

// Differs from the Postgres version in that we don't have `FOR UPDATE SKIP
// LOCKED`. It doesn't exist in SQLite, but more aptly, there's only one writer
// on SQLite at a time, so nothing else has the rows locked.
func (q *Queries) JobGetAvailable(ctx context.Context, db DBTX, arg *JobGetAvailableParams) ([]*RiverJob, error) {
	rows, err := db.QueryContext(ctx, jobGetAvailable,
		arg.Now,
		arg.AttemptedBy,
		arg.Queue,
		arg.Max,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobGetByID = `-- name: JobGetByID :one
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE id = ?1
LIMIT 1
`

func (q *Queries) JobGetByID(ctx context.Context, db DBTX, id int64) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobGetByID, id)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobGetByIDMany = `-- name: JobGetByIDMany :many
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE id IN (/*SLICE:id*/?)
ORDER BY id
`

func (q *Queries) JobGetByIDMany(ctx context.Context, db DBTX, id []int64) ([]*RiverJob, error) {
	query := jobGetByIDMany
	var queryParams []interface{}
	if len(id) > 0 {
		for _, v := range id {
			queryParams = append(queryParams, v)
		}
		query = strings.Replace(query, "/*SLICE:id*/?", strings.Repeat(",?", len(id))[1:], 1)
	} else {
		query = strings.Replace(query, "/*SLICE:id*/?", "NULL", 1)
	}
	rows, err := db.QueryContext(ctx, query, queryParams...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobGetByKindMany = `-- name: JobGetByKindMany :many
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE kind IN (/*SLICE:kind*/?)
ORDER BY id
`

func (q *Queries) JobGetByKindMany(ctx context.Context, db DBTX, kind []string) ([]*RiverJob, error) {
	query := jobGetByKindMany
	var queryParams []interface{}
	if len(kind) > 0 {
		for _, v := range kind {
			queryParams = append(queryParams, v)
		}
		query = strings.Replace(query, "/*SLICE:kind*/?", strings.Repeat(",?", len(kind))[1:], 1)
	} else {
		query = strings.Replace(query, "/*SLICE:kind*/?", "NULL", 1)
	}
	rows, err := db.QueryContext(ctx, query, queryParams...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobGetStuck = `-- name: JobGetStuck :many
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE state = 'running'
    AND attempted_at < cast(?1 AS text)
ORDER BY id
LIMIT ?2
`

type JobGetStuckParams struct {
	StuckHorizon string
	Max          int64
}

func (q *Queries) JobGetStuck(ctx context.Context, db DBTX, arg *JobGetStuckParams) ([]*RiverJob, error) {
	rows, err := db.QueryContext(ctx, jobGetStuck, arg.StuckHorizon, arg.Max)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobInsertFast = `-- name: JobInsertFast :one
INSERT INTO /* TEMPLATE: schema */river_job(
    args,
    created_at,
    kind,
    max_attempts,
    metadata,
    priority,
    queue,
    scheduled_at,
    state,
    tags,
    unique_key,
    unique_states
) VALUES (
    ?1,
    coalesce(cast(?2 AS text), datetime('now', 'subsec')),
    ?3,
    ?4,
    json(cast(?5 AS blob)),
    ?6,
    ?7,
    coalesce(cast(?8 AS text), datetime('now', 'subsec')),
    ?9,
    json(cast(?10 AS blob)),
    CASE WHEN length(cast(?11 AS blob)) = 0 THEN NULL ELSE ?11 END,
    ?12
)
ON CONFLICT (unique_key)
    WHERE unique_key IS NOT NULL
        AND unique_states IS NOT NULL
        AND CASE state
                WHEN 'available' THEN unique_states & (1 << 0)
                WHEN 'cancelled' THEN unique_states & (1 << 1)
                WHEN 'completed' THEN unique_states & (1 << 2)
                WHEN 'discarded' THEN unique_states & (1 << 3)
                WHEN 'pending'   THEN unique_states & (1 << 4)
                WHEN 'retryable' THEN unique_states & (1 << 5)
                WHEN 'running'   THEN unique_states & (1 << 6)
                WHEN 'scheduled' THEN unique_states & (1 << 7)
                ELSE 0
            END >= 1
    -- Something needs to be updated for a row to be returned on a conflict.
    DO UPDATE SET kind = EXCLUDED.kind
RETURNING 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
`

type JobInsertFastParams struct {
	Args         []byte
	CreatedAt    *string
	Kind         string
	MaxAttempts  int64
	Metadata     []byte
	Priority     int64
	Queue        string
	ScheduledAt  *string
	State        string
	Tags         []byte
	UniqueKey    []byte
	UniqueStates *int64
}

// Insert a job.
//
// This is supposed to be a batch insert, but various limitations of the
// combined SQLite + sqlc has left me unable to find a way of injecting many
// arguments en masse (like how we slightly abuse arrays to pull it off for the
// Postgres drivers), so we loop over many insert operations instead, with the
// expectation that this may be fixable in the future. Because SQLite targets
// will often be local and therefore with a very minimal round trip compared to
// a network, looping over operations is probably okay performance-wise.
func (q *Queries) JobInsertFast(ctx context.Context, db DBTX, arg *JobInsertFastParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobInsertFast,
		arg.Args,
		arg.CreatedAt,
		arg.Kind,
		arg.MaxAttempts,
		arg.Metadata,
		arg.Priority,
		arg.Queue,
		arg.ScheduledAt,
		arg.State,
		arg.Tags,
		arg.UniqueKey,
		arg.UniqueStates,
	)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobInsertFastNoReturning = `-- name: JobInsertFastNoReturning :execrows
INSERT INTO /* TEMPLATE: schema */river_job(
    args,
    created_at,
    kind,
    max_attempts,
    metadata,
    priority,
    queue,
    scheduled_at,
    state,
    tags,
    unique_key,
    unique_states
) VALUES (
    ?1,
    coalesce(cast(?2 AS text), datetime('now', 'subsec')),
    ?3,
    ?4,
    json(cast(?5 AS blob)),
    ?6,
    ?7,
    coalesce(cast(?8 AS text), datetime('now', 'subsec')),
    ?9,
    json(cast(?10 AS blob)),
    CASE WHEN length(cast(?11 AS blob)) = 0 THEN NULL ELSE ?11 END,
    ?12
)
ON CONFLICT (unique_key)
    WHERE unique_key IS NOT NULL
        AND unique_states IS NOT NULL
        AND CASE state
                WHEN 'available' THEN unique_states & (1 << 0)
                WHEN 'cancelled' THEN unique_states & (1 << 1)
                WHEN 'completed' THEN unique_states & (1 << 2)
                WHEN 'discarded' THEN unique_states & (1 << 3)
                WHEN 'pending'   THEN unique_states & (1 << 4)
                WHEN 'retryable' THEN unique_states & (1 << 5)
                WHEN 'running'   THEN unique_states & (1 << 6)
                WHEN 'scheduled' THEN unique_states & (1 << 7)
                ELSE 0
            END >= 1
DO NOTHING
`

type JobInsertFastNoReturningParams struct {
	Args         []byte
	CreatedAt    *string
	Kind         string
	MaxAttempts  int64
	Metadata     []byte
	Priority     int64
	Queue        string
	ScheduledAt  *string
	State        string
	Tags         []byte
	UniqueKey    []byte
	UniqueStates *int64
}

func (q *Queries) JobInsertFastNoReturning(ctx context.Context, db DBTX, arg *JobInsertFastNoReturningParams) (int64, error) {
	result, err := db.ExecContext(ctx, jobInsertFastNoReturning,
		arg.Args,
		arg.CreatedAt,
		arg.Kind,
		arg.MaxAttempts,
		arg.Metadata,
		arg.Priority,
		arg.Queue,
		arg.ScheduledAt,
		arg.State,
		arg.Tags,
		arg.UniqueKey,
		arg.UniqueStates,
	)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected()
}

const jobInsertFull = `-- name: JobInsertFull :one
INSERT INTO /* TEMPLATE: schema */river_job(
    args,
    attempt,
    attempted_at,
    attempted_by,
    created_at,
    errors,
    finalized_at,
    kind,
    max_attempts,
    metadata,
    priority,
    queue,
    scheduled_at,
    state,
    tags,
    unique_key,
    unique_states
) VALUES (
    ?1,
    ?2,
    cast(?3 as text),
    CASE WHEN length(cast(?4 AS blob)) = 0 THEN NULL ELSE json(?4) END,
    coalesce(cast(?5 AS text), datetime('now', 'subsec')),
    CASE WHEN length(cast(?6 AS blob)) = 0 THEN NULL ELSE ?6 END,
    cast(?7 as text),
    ?8,
    ?9,
    json(cast(?10 AS blob)),
    ?11,
    ?12,
    coalesce(cast(?13 AS text), datetime('now', 'subsec')),
    ?14,
    json(cast(?15 AS blob)),
    CASE WHEN length(cast(?16 AS blob)) = 0 THEN NULL ELSE ?16 END,
    ?17
) RETURNING 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
`

type JobInsertFullParams struct {
	Args         []byte
	Attempt      int64
	AttemptedAt  *string
	AttemptedBy  []byte
	CreatedAt    *string
	Errors       []byte
	FinalizedAt  *string
	Kind         string
	MaxAttempts  int64
	Metadata     []byte
	Priority     int64
	Queue        string
	ScheduledAt  *string
	State        string
	Tags         []byte
	UniqueKey    []byte
	UniqueStates *int64
}

func (q *Queries) JobInsertFull(ctx context.Context, db DBTX, arg *JobInsertFullParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobInsertFull,
		arg.Args,
		arg.Attempt,
		arg.AttemptedAt,
		arg.AttemptedBy,
		arg.CreatedAt,
		arg.Errors,
		arg.FinalizedAt,
		arg.Kind,
		arg.MaxAttempts,
		arg.Metadata,
		arg.Priority,
		arg.Queue,
		arg.ScheduledAt,
		arg.State,
		arg.Tags,
		arg.UniqueKey,
		arg.UniqueStates,
	)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobList = `-- name: JobList :many
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE /* TEMPLATE_BEGIN: where_clause */ 1 /* TEMPLATE_END */
ORDER BY /* TEMPLATE_BEGIN: order_by_clause */ id /* TEMPLATE_END */
LIMIT ?1
`

func (q *Queries) JobList(ctx context.Context, db DBTX, max int64) ([]*RiverJob, error) {
	rows, err := db.QueryContext(ctx, jobList, max)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobRescue = `-- name: JobRescue :exec
UPDATE /* TEMPLATE: schema */river_job
SET
    errors = json_insert(coalesce(errors, json('[]')), '$[#]', json(cast(?1 AS blob))),
    finalized_at = cast(?2 as text),
    scheduled_at = ?3,
    state = ?4
WHERE id = ?5
`

type JobRescueParams struct {
	Error       []byte
	FinalizedAt *string
	ScheduledAt time.Time
	State       string
	ID          int64
}

// Rescue a job.
//
// This is supposed to rescue jobs in batches, but various limitations of the
// combined SQLite + sqlc has left me unable to find a way of injecting many
// arguments en masse (like how we slightly abuse arrays to pull it off for the
// Postgres drivers), and SQLite doesn't support `UPDATE` in CTEs, so we loop
// over many insert operations instead, with the expectation that this may be
// fixable in the future. Because SQLite targets will often be local and with a
// very minimal round trip compared to a network, looping over operations is
// probably okay performance-wise.
func (q *Queries) JobRescue(ctx context.Context, db DBTX, arg *JobRescueParams) error {
	_, err := db.ExecContext(ctx, jobRescue,
		arg.Error,
		arg.FinalizedAt,
		arg.ScheduledAt,
		arg.State,
		arg.ID,
	)
	return err
}

const jobRetry = `-- name: JobRetry :one
UPDATE /* TEMPLATE: schema */river_job
SET
    state = 'available',
    max_attempts = CASE WHEN attempt = max_attempts THEN max_attempts + 1 ELSE max_attempts END,
    finalized_at = NULL,
    scheduled_at = coalesce(cast(?1 AS text), datetime('now', 'subsec'))
WHERE id = ?2
    -- Do not touch running jobs:
    AND state != 'running'
    -- If the job is already available with a prior scheduled_at, leave it alone.
    --
    -- I had to invert the original 'AND NOT' to 'AND'. Something about
    -- sqlc's SQLite parser cannot detect a parameter inside an ` + "`" + `AND NOT` + "`" + `. An
    -- unfortunate bug that will hopefully be fixed in the future ...
    AND (
        state <> 'available'
        OR scheduled_at > coalesce(cast(?1 AS text), datetime('now', 'subsec'))
    )
RETURNING 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
`

type JobRetryParams struct {
	Now *string
	ID  int64
}

// Differs by necessity from other drivers because SQLite doesn't support
// `UPDATE` inside CTEs so we can't retry if running but select otherwise.
// Instead, the driver uses a transaction to optimisticaly try an update, but
// perform a subsequent fetch on a not found to return the right status.
//
// I had to invert the last 'AND' expression below (was an 'AND NOT') due to an
// sqlc bug. Something about sqlc's SQLite parser cannot detect a parameter
// inside an `AND NOT`. I'll try to get this fixed upstream at some point so we
// can clean this up and keep it more like the Postgres version.
func (q *Queries) JobRetry(ctx context.Context, db DBTX, arg *JobRetryParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobRetry, arg.Now, arg.ID)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobScheduleGetCollision = `-- name: JobScheduleGetCollision :one
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE id <> ?1
    AND unique_key = ?2
    AND unique_states IS NOT NULL
    AND CASE state
            WHEN 'available' THEN unique_states & (1 << 0)
            WHEN 'cancelled' THEN unique_states & (1 << 1)
            WHEN 'completed' THEN unique_states & (1 << 2)
            WHEN 'discarded' THEN unique_states & (1 << 3)
            WHEN 'pending'   THEN unique_states & (1 << 4)
            WHEN 'retryable' THEN unique_states & (1 << 5)
            WHEN 'running'   THEN unique_states & (1 << 6)
            WHEN 'scheduled' THEN unique_states & (1 << 7)
            ELSE 0
        END >= 1
`

type JobScheduleGetCollisionParams struct {
	ID        int64
	UniqueKey []byte
}

func (q *Queries) JobScheduleGetCollision(ctx context.Context, db DBTX, arg *JobScheduleGetCollisionParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobScheduleGetCollision, arg.ID, arg.UniqueKey)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobScheduleGetEligible = `-- name: JobScheduleGetEligible :many
SELECT 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
FROM /* TEMPLATE: schema */river_job
WHERE
    state IN ('retryable', 'scheduled')
    AND scheduled_at <= coalesce(cast(?1 AS text), datetime('now', 'subsec'))
ORDER BY
    priority,
    scheduled_at,
    id
LIMIT ?2
`

type JobScheduleGetEligibleParams struct {
	Now *string
	Max int64
}

func (q *Queries) JobScheduleGetEligible(ctx context.Context, db DBTX, arg *JobScheduleGetEligibleParams) ([]*RiverJob, error) {
	rows, err := db.QueryContext(ctx, jobScheduleGetEligible, arg.Now, arg.Max)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobScheduleSetAvailable = `-- name: JobScheduleSetAvailable :many
UPDATE /* TEMPLATE: schema */river_job
SET
    state = 'available'
WHERE id IN (/*SLICE:id*/?)
RETURNING 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
`

func (q *Queries) JobScheduleSetAvailable(ctx context.Context, db DBTX, id []int64) ([]*RiverJob, error) {
	query := jobScheduleSetAvailable
	var queryParams []interface{}
	if len(id) > 0 {
		for _, v := range id {
			queryParams = append(queryParams, v)
		}
		query = strings.Replace(query, "/*SLICE:id*/?", strings.Repeat(",?", len(id))[1:], 1)
	} else {
		query = strings.Replace(query, "/*SLICE:id*/?", "NULL", 1)
	}
	rows, err := db.QueryContext(ctx, query, queryParams...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobScheduleSetDiscarded = `-- name: JobScheduleSetDiscarded :many
UPDATE /* TEMPLATE: schema */river_job
SET metadata = json_patch(metadata, json('{"unique_key_conflict": "scheduler_discarded"}')),
    finalized_at = coalesce(cast(?1 AS text), datetime('now', 'subsec')),
    state = 'discarded'
WHERE id IN (/*SLICE:id*/?)
RETURNING 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
`

type JobScheduleSetDiscardedParams struct {
	Now *string
	ID  []int64
}

func (q *Queries) JobScheduleSetDiscarded(ctx context.Context, db DBTX, arg *JobScheduleSetDiscardedParams) ([]*RiverJob, error) {
	query := jobScheduleSetDiscarded
	var queryParams []interface{}
	queryParams = append(queryParams, arg.Now)
	if len(arg.ID) > 0 {
		for _, v := range arg.ID {
			queryParams = append(queryParams, v)
		}
		query = strings.Replace(query, "/*SLICE:id*/?", strings.Repeat(",?", len(arg.ID))[1:], 1)
	} else {
		query = strings.Replace(query, "/*SLICE:id*/?", "NULL", 1)
	}
	rows, err := db.QueryContext(ctx, query, queryParams...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobSetCompletedIfRunning = `-- name: JobSetCompletedIfRunning :many
UPDATE /* TEMPLATE: schema */river_job
SET finalized_at = coalesce(cast(?1 AS text), datetime('now', 'subsec')),
    state = 'completed'
WHERE id IN (/*SLICE:id*/?)
    AND state = 'running'
RETURNING 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
`

type JobSetCompletedIfRunningParams struct {
	FinalizedAt *string
	ID          []int64
}

// This doesn't exist under the Postgres driver, but is used for an optimized
// happy path for setting jobs to `complete` where metadata isn't required.
func (q *Queries) JobSetCompletedIfRunning(ctx context.Context, db DBTX, arg *JobSetCompletedIfRunningParams) ([]*RiverJob, error) {
	query := jobSetCompletedIfRunning
	var queryParams []interface{}
	queryParams = append(queryParams, arg.FinalizedAt)
	if len(arg.ID) > 0 {
		for _, v := range arg.ID {
			queryParams = append(queryParams, v)
		}
		query = strings.Replace(query, "/*SLICE:id*/?", strings.Repeat(",?", len(arg.ID))[1:], 1)
	} else {
		query = strings.Replace(query, "/*SLICE:id*/?", "NULL", 1)
	}
	rows, err := db.QueryContext(ctx, query, queryParams...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []*RiverJob
	for rows.Next() {
		var i RiverJob
		if err := rows.Scan(
			&i.ID,
			&i.Args,
			&i.Attempt,
			&i.AttemptedAt,
			&i.AttemptedBy,
			&i.CreatedAt,
			&i.Errors,
			&i.FinalizedAt,
			&i.Kind,
			&i.MaxAttempts,
			&i.Metadata,
			&i.Priority,
			&i.Queue,
			&i.State,
			&i.ScheduledAt,
			&i.Tags,
			&i.UniqueKey,
			&i.UniqueStates,
		); err != nil {
			return nil, err
		}
		items = append(items, &i)
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const jobSetMetadataIfNotRunning = `-- name: JobSetMetadataIfNotRunning :one
UPDATE /* TEMPLATE: schema */river_job
SET metadata = json_patch(metadata, json(cast(?1 AS blob)))
WHERE id = ?2
    AND state != 'running'
RETURNING 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
`

type JobSetMetadataIfNotRunningParams struct {
	MetadataUpdates []byte
	ID              int64
}

// This doesn't exist under the Postgres driver, but needed as an extra query
// for JobSetStateIfRunning to use when falling back to non-running jobs.
func (q *Queries) JobSetMetadataIfNotRunning(ctx context.Context, db DBTX, arg *JobSetMetadataIfNotRunningParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobSetMetadataIfNotRunning, arg.MetadataUpdates, arg.ID)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobSetStateIfRunning = `-- name: JobSetStateIfRunning :one
UPDATE /* TEMPLATE: schema */river_job
SET
    -- should_cancel: (job_input.state IN ('retryable', 'scheduled') AND river_job.metadata ? 'cancel_attempted_at')
    --
    -- or inverted:   (cast(@state AS text) <> 'retryable' AND @state <> 'scheduled' OR NOT (metadata -> 'cancel_attempted_at'))
    attempt      = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?2 AS boolean)
                        THEN ?3
                        ELSE attempt END,
    errors       = CASE WHEN cast(?4 AS boolean)
                        THEN json_insert(coalesce(errors, json('[]')), '$[#]', json(cast(?5 AS blob)))
                        ELSE errors END,
    finalized_at = CASE WHEN /* should_cancel */((?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') iS NOT NULL)
                        THEN coalesce(cast(?6 AS text), datetime('now', 'subsec'))
                        WHEN cast(?7 AS boolean)
                        THEN ?8
                        ELSE finalized_at END,
    metadata     = CASE WHEN cast(?9 AS boolean)
                        THEN json_patch(metadata, json(cast(?10 AS blob)))
                        ELSE metadata END,
    scheduled_at = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?11 AS boolean)
                        THEN ?12
                        ELSE scheduled_at END,
    state        = CASE WHEN /* should_cancel */((?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') IS NOT NULL)
                        THEN 'cancelled'
                        ELSE ?1 END
WHERE id = ?13
    AND state = 'running'
RETURNING 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
`

type JobSetStateIfRunningParams struct {
	State               string
	AttemptDoUpdate     bool
	Attempt             int64
	ErrorsDoUpdate      bool
	Error               []byte
	Now                 *string
	FinalizedAtDoUpdate bool
	FinalizedAt         *time.Time
	MetadataDoMerge     bool
	MetadataUpdates     []byte
	ScheduledAtDoUpdate bool
	ScheduledAt         time.Time
	ID                  int64
}

// Differs significantly from the Postgres version in that it can't do a bulk
// update, and since sqlc doesn't support `UPDATE` in CTEs, we need separate
// queries like JobSetMetadataIfNotRunning to do the fallback work.
func (q *Queries) JobSetStateIfRunning(ctx context.Context, db DBTX, arg *JobSetStateIfRunningParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobSetStateIfRunning,
		arg.State,
		arg.AttemptDoUpdate,
		arg.Attempt,
		arg.ErrorsDoUpdate,
		arg.Error,
		arg.Now,
		arg.FinalizedAtDoUpdate,
		arg.FinalizedAt,
		arg.MetadataDoMerge,
		arg.MetadataUpdates,
		arg.ScheduledAtDoUpdate,
		arg.ScheduledAt,
		arg.ID,
	)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}

const jobUpdate = `-- name: JobUpdate :one
UPDATE /* TEMPLATE: schema */river_job
SET
    attempt = CASE WHEN cast(?1 AS boolean) THEN ?2 ELSE attempt END,
    attempted_at = CASE WHEN cast(?3 AS boolean) THEN ?4 ELSE attempted_at END,
    attempted_by = CASE WHEN cast(?5 AS boolean) THEN ?6 ELSE attempted_by END,
    errors = CASE WHEN cast(?7 AS boolean) THEN ?8 ELSE errors END,
    finalized_at = CASE WHEN cast(?9 AS boolean) THEN ?10 ELSE finalized_at END,
    state = CASE WHEN cast(?11 AS boolean) THEN ?12 ELSE state END
WHERE id = ?13
RETURNING 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
`

type JobUpdateParams struct {
	AttemptDoUpdate     bool
	Attempt             int64
	AttemptedAtDoUpdate bool
	AttemptedAt         *time.Time
	AttemptedByDoUpdate bool
	AttemptedBy         []byte
	ErrorsDoUpdate      bool
	Errors              []byte
	FinalizedAtDoUpdate bool
	FinalizedAt         *time.Time
	StateDoUpdate       bool
	State               string
	ID                  int64
}

// A generalized update for any property on a job. This brings in a large number
// of parameters and therefore may be more suitable for testing than production.
func (q *Queries) JobUpdate(ctx context.Context, db DBTX, arg *JobUpdateParams) (*RiverJob, error) {
	row := db.QueryRowContext(ctx, jobUpdate,
		arg.AttemptDoUpdate,
		arg.Attempt,
		arg.AttemptedAtDoUpdate,
		arg.AttemptedAt,
		arg.AttemptedByDoUpdate,
		arg.AttemptedBy,
		arg.ErrorsDoUpdate,
		arg.Errors,
		arg.FinalizedAtDoUpdate,
		arg.FinalizedAt,
		arg.StateDoUpdate,
		arg.State,
		arg.ID,
	)
	var i RiverJob
	err := row.Scan(
		&i.ID,
		&i.Args,
		&i.Attempt,
		&i.AttemptedAt,
		&i.AttemptedBy,
		&i.CreatedAt,
		&i.Errors,
		&i.FinalizedAt,
		&i.Kind,
		&i.MaxAttempts,
		&i.Metadata,
		&i.Priority,
		&i.Queue,
		&i.State,
		&i.ScheduledAt,
		&i.Tags,
		&i.UniqueKey,
		&i.UniqueStates,
	)
	return &i, err
}