agentic-coding-workflow/internal/jobs/sqlc/jobs.sql.go

// Code generated by sqlc. DO NOT EDIT.
// versions:
//   sqlc v1.31.1
// source: jobs.sql

package jobssqlc

import (
	"context"

	"github.com/jackc/pgx/v5/pgtype"
)

const completeJob = `-- name: CompleteJob :execrows
UPDATE jobs
   SET status='completed', completed_at=now(),
       leased_at=NULL, leased_by=NULL, last_error=NULL,
       updated_at=now()
 WHERE id=$1 AND leased_by=$2 AND status='running'
`

type CompleteJobParams struct {
	ID       pgtype.UUID `json:"id"`
	LeasedBy *string     `json:"leased_by"`
}

// 围栏谓词：只有持有当前租约的 worker 才能写回，防止被 reaper 重新租出后旧 worker 覆盖新一轮状态。
func (q *Queries) CompleteJob(ctx context.Context, arg CompleteJobParams) (int64, error) {
	result, err := q.db.Exec(ctx, completeJob, arg.ID, arg.LeasedBy)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

const deadLetterJob = `-- name: DeadLetterJob :execrows
UPDATE jobs
   SET status='dead', completed_at=now(),
       leased_at=NULL, leased_by=NULL, last_error=$2,
       updated_at=now()
 WHERE id=$1 AND leased_by=$3 AND status='running'
`

type DeadLetterJobParams struct {
	ID        pgtype.UUID `json:"id"`
	LastError *string     `json:"last_error"`
	LeasedBy  *string     `json:"leased_by"`
}

// 围栏谓词同上。
func (q *Queries) DeadLetterJob(ctx context.Context, arg DeadLetterJobParams) (int64, error) {
	result, err := q.db.Exec(ctx, deadLetterJob, arg.ID, arg.LastError, arg.LeasedBy)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

const enqueueJob = `-- name: EnqueueJob :one
INSERT INTO jobs (id, type, payload, status, scheduled_at, attempts, max_attempts)
VALUES ($1, $2, $3, 'pending', $4, 0, $5)
RETURNING id, type, payload, status, scheduled_at, attempts, max_attempts, leased_at, leased_by, last_error, completed_at, created_at, updated_at
`

type EnqueueJobParams struct {
	ID          pgtype.UUID        `json:"id"`
	Type        string             `json:"type"`
	Payload     []byte             `json:"payload"`
	ScheduledAt pgtype.Timestamptz `json:"scheduled_at"`
	MaxAttempts int32              `json:"max_attempts"`
}

func (q *Queries) EnqueueJob(ctx context.Context, arg EnqueueJobParams) (Job, error) {
	row := q.db.QueryRow(ctx, enqueueJob,
		arg.ID,
		arg.Type,
		arg.Payload,
		arg.ScheduledAt,
		arg.MaxAttempts,
	)
	var i Job
	err := row.Scan(
		&i.ID,
		&i.Type,
		&i.Payload,
		&i.Status,
		&i.ScheduledAt,
		&i.Attempts,
		&i.MaxAttempts,
		&i.LeasedAt,
		&i.LeasedBy,
		&i.LastError,
		&i.CompletedAt,
		&i.CreatedAt,
		&i.UpdatedAt,
	)
	return i, err
}

const failJobWithRetry = `-- name: FailJobWithRetry :execrows
UPDATE jobs
   SET status='pending',
       scheduled_at=$2,
       leased_at=NULL, leased_by=NULL,
       last_error=$3,
       updated_at=now()
 WHERE id=$1 AND leased_by=$4 AND status='running'
`

type FailJobWithRetryParams struct {
	ID          pgtype.UUID        `json:"id"`
	ScheduledAt pgtype.Timestamptz `json:"scheduled_at"`
	LastError   *string            `json:"last_error"`
	LeasedBy    *string            `json:"leased_by"`
}

// 围栏谓词同上。
func (q *Queries) FailJobWithRetry(ctx context.Context, arg FailJobWithRetryParams) (int64, error) {
	result, err := q.db.Exec(ctx, failJobWithRetry,
		arg.ID,
		arg.ScheduledAt,
		arg.LastError,
		arg.LeasedBy,
	)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

const getJob = `-- name: GetJob :one
SELECT id, type, payload, status, scheduled_at, attempts, max_attempts, leased_at, leased_by, last_error, completed_at, created_at, updated_at FROM jobs WHERE id=$1
`

func (q *Queries) GetJob(ctx context.Context, id pgtype.UUID) (Job, error) {
	row := q.db.QueryRow(ctx, getJob, id)
	var i Job
	err := row.Scan(
		&i.ID,
		&i.Type,
		&i.Payload,
		&i.Status,
		&i.ScheduledAt,
		&i.Attempts,
		&i.MaxAttempts,
		&i.LeasedAt,
		&i.LeasedBy,
		&i.LastError,
		&i.CompletedAt,
		&i.CreatedAt,
		&i.UpdatedAt,
	)
	return i, err
}

const leaseNextJob = `-- name: LeaseNextJob :one
UPDATE jobs
   SET status='running',
       leased_at=now(),
       leased_by=$1,
       attempts=attempts+1,
       updated_at=now()
 WHERE id = (
   SELECT id FROM jobs
    WHERE status='pending' AND scheduled_at <= now()
    ORDER BY scheduled_at ASC
    FOR UPDATE SKIP LOCKED
    LIMIT 1
 )
RETURNING id, type, payload, status, scheduled_at, attempts, max_attempts, leased_at, leased_by, last_error, completed_at, created_at, updated_at
`

// SKIP LOCKED 让多 worker 并行抢不冲突；只返回单行。
func (q *Queries) LeaseNextJob(ctx context.Context, leasedBy *string) (Job, error) {
	row := q.db.QueryRow(ctx, leaseNextJob, leasedBy)
	var i Job
	err := row.Scan(
		&i.ID,
		&i.Type,
		&i.Payload,
		&i.Status,
		&i.ScheduledAt,
		&i.Attempts,
		&i.MaxAttempts,
		&i.LeasedAt,
		&i.LeasedBy,
		&i.LastError,
		&i.CompletedAt,
		&i.CreatedAt,
		&i.UpdatedAt,
	)
	return i, err
}

const listDeadJobs = `-- name: ListDeadJobs :many
SELECT id, type, payload, status, scheduled_at, attempts, max_attempts, leased_at, leased_by, last_error, completed_at, created_at, updated_at FROM jobs
 WHERE status='dead' AND completed_at >= $1
 ORDER BY completed_at DESC
 LIMIT $2
`

type ListDeadJobsParams struct {
	CompletedAt pgtype.Timestamptz `json:"completed_at"`
	Limit       int32              `json:"limit"`
}

func (q *Queries) ListDeadJobs(ctx context.Context, arg ListDeadJobsParams) ([]Job, error) {
	rows, err := q.db.Query(ctx, listDeadJobs, arg.CompletedAt, arg.Limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []Job
	for rows.Next() {
		var i Job
		if err := rows.Scan(
			&i.ID,
			&i.Type,
			&i.Payload,
			&i.Status,
			&i.ScheduledAt,
			&i.Attempts,
			&i.MaxAttempts,
			&i.LeasedAt,
			&i.LeasedBy,
			&i.LastError,
			&i.CompletedAt,
			&i.CreatedAt,
			&i.UpdatedAt,
		); err != nil {
			return nil, err
		}
		items = append(items, i)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}

const purgeCompletedJobs = `-- name: PurgeCompletedJobs :execrows
DELETE FROM jobs
 WHERE status IN ('completed','dead') AND completed_at < $1
`

func (q *Queries) PurgeCompletedJobs(ctx context.Context, completedAt pgtype.Timestamptz) (int64, error) {
	result, err := q.db.Exec(ctx, purgeCompletedJobs, completedAt)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

const reapStuckRunningJobs = `-- name: ReapStuckRunningJobs :many
UPDATE jobs
   SET status='pending', leased_at=NULL, leased_by=NULL, scheduled_at=now(),
       updated_at=now()
 WHERE status='running' AND leased_at < $1
RETURNING id
`

// 复位 worker 崩了卡 running 的任务。
func (q *Queries) ReapStuckRunningJobs(ctx context.Context, leasedAt pgtype.Timestamptz) ([]pgtype.UUID, error) {
	rows, err := q.db.Query(ctx, reapStuckRunningJobs, leasedAt)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var items []pgtype.UUID
	for rows.Next() {
		var id pgtype.UUID
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		items = append(items, id)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return items, nil
}