351 lines
9 KiB
Go
351 lines
9 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: workflows.sql
|
|
|
|
package dbgen
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const completeStepRun = `-- name: CompleteStepRun :one
|
|
UPDATE step_runs
|
|
SET status = 'succeeded',
|
|
output = $3,
|
|
leased_by = NULL,
|
|
leased_until = NULL,
|
|
finished_at = now(),
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND leased_by = $2
|
|
AND status IN ('leased', 'running')
|
|
RETURNING id, project_id, workflow_run_id, step_key, executor_kind, status, attempt, max_attempts, priority, input, output, error, idempotency_key, available_at, leased_by, leased_until, started_at, finished_at, created_at, updated_at
|
|
`
|
|
|
|
type CompleteStepRunParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
LeasedBy pgtype.Text `json:"leased_by"`
|
|
Output []byte `json:"output"`
|
|
}
|
|
|
|
func (q *Queries) CompleteStepRun(ctx context.Context, arg CompleteStepRunParams) (StepRun, error) {
|
|
row := q.db.QueryRow(ctx, completeStepRun, arg.ID, arg.LeasedBy, arg.Output)
|
|
var i StepRun
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ProjectID,
|
|
&i.WorkflowRunID,
|
|
&i.StepKey,
|
|
&i.ExecutorKind,
|
|
&i.Status,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.Priority,
|
|
&i.Input,
|
|
&i.Output,
|
|
&i.Error,
|
|
&i.IdempotencyKey,
|
|
&i.AvailableAt,
|
|
&i.LeasedBy,
|
|
&i.LeasedUntil,
|
|
&i.StartedAt,
|
|
&i.FinishedAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createStepRun = `-- name: CreateStepRun :one
|
|
INSERT INTO step_runs (
|
|
id,
|
|
project_id,
|
|
workflow_run_id,
|
|
step_key,
|
|
executor_kind,
|
|
max_attempts,
|
|
priority,
|
|
input,
|
|
idempotency_key,
|
|
available_at
|
|
)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
|
|
RETURNING id, project_id, workflow_run_id, step_key, executor_kind, status, attempt, max_attempts, priority, input, output, error, idempotency_key, available_at, leased_by, leased_until, started_at, finished_at, created_at, updated_at
|
|
`
|
|
|
|
type CreateStepRunParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
ProjectID pgtype.UUID `json:"project_id"`
|
|
WorkflowRunID pgtype.UUID `json:"workflow_run_id"`
|
|
StepKey string `json:"step_key"`
|
|
ExecutorKind string `json:"executor_kind"`
|
|
MaxAttempts int32 `json:"max_attempts"`
|
|
Priority int32 `json:"priority"`
|
|
Input []byte `json:"input"`
|
|
IdempotencyKey string `json:"idempotency_key"`
|
|
AvailableAt pgtype.Timestamptz `json:"available_at"`
|
|
}
|
|
|
|
func (q *Queries) CreateStepRun(ctx context.Context, arg CreateStepRunParams) (StepRun, error) {
|
|
row := q.db.QueryRow(ctx, createStepRun,
|
|
arg.ID,
|
|
arg.ProjectID,
|
|
arg.WorkflowRunID,
|
|
arg.StepKey,
|
|
arg.ExecutorKind,
|
|
arg.MaxAttempts,
|
|
arg.Priority,
|
|
arg.Input,
|
|
arg.IdempotencyKey,
|
|
arg.AvailableAt,
|
|
)
|
|
var i StepRun
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ProjectID,
|
|
&i.WorkflowRunID,
|
|
&i.StepKey,
|
|
&i.ExecutorKind,
|
|
&i.Status,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.Priority,
|
|
&i.Input,
|
|
&i.Output,
|
|
&i.Error,
|
|
&i.IdempotencyKey,
|
|
&i.AvailableAt,
|
|
&i.LeasedBy,
|
|
&i.LeasedUntil,
|
|
&i.StartedAt,
|
|
&i.FinishedAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createWorkflowRun = `-- name: CreateWorkflowRun :one
|
|
INSERT INTO workflow_runs (
|
|
id,
|
|
project_id,
|
|
workflow_version_id,
|
|
status,
|
|
trigger_type,
|
|
input,
|
|
idempotency_key
|
|
)
|
|
VALUES ($1, $2, $3, 'pending', $4, $5, $6)
|
|
RETURNING id, project_id, workflow_version_id, status, trigger_type, input, output, error, idempotency_key, started_at, finished_at, created_at, updated_at
|
|
`
|
|
|
|
type CreateWorkflowRunParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
ProjectID pgtype.UUID `json:"project_id"`
|
|
WorkflowVersionID pgtype.UUID `json:"workflow_version_id"`
|
|
TriggerType string `json:"trigger_type"`
|
|
Input []byte `json:"input"`
|
|
IdempotencyKey pgtype.Text `json:"idempotency_key"`
|
|
}
|
|
|
|
func (q *Queries) CreateWorkflowRun(ctx context.Context, arg CreateWorkflowRunParams) (WorkflowRun, error) {
|
|
row := q.db.QueryRow(ctx, createWorkflowRun,
|
|
arg.ID,
|
|
arg.ProjectID,
|
|
arg.WorkflowVersionID,
|
|
arg.TriggerType,
|
|
arg.Input,
|
|
arg.IdempotencyKey,
|
|
)
|
|
var i WorkflowRun
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ProjectID,
|
|
&i.WorkflowVersionID,
|
|
&i.Status,
|
|
&i.TriggerType,
|
|
&i.Input,
|
|
&i.Output,
|
|
&i.Error,
|
|
&i.IdempotencyKey,
|
|
&i.StartedAt,
|
|
&i.FinishedAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const failStepRun = `-- name: FailStepRun :one
|
|
UPDATE step_runs
|
|
SET status = CASE
|
|
WHEN attempt < max_attempts AND $1::boolean
|
|
THEN 'retryable_failed'
|
|
ELSE 'terminal_failed'
|
|
END,
|
|
error = $2,
|
|
available_at = CASE
|
|
WHEN attempt < max_attempts AND $1::boolean
|
|
THEN now() + make_interval(secs => $3::integer)
|
|
ELSE available_at
|
|
END,
|
|
leased_by = NULL,
|
|
leased_until = NULL,
|
|
finished_at = CASE
|
|
WHEN attempt < max_attempts AND $1::boolean
|
|
THEN NULL
|
|
ELSE now()
|
|
END,
|
|
updated_at = now()
|
|
WHERE id = $4
|
|
AND leased_by = $5
|
|
AND status IN ('leased', 'running')
|
|
RETURNING id, project_id, workflow_run_id, step_key, executor_kind, status, attempt, max_attempts, priority, input, output, error, idempotency_key, available_at, leased_by, leased_until, started_at, finished_at, created_at, updated_at
|
|
`
|
|
|
|
type FailStepRunParams struct {
|
|
Retryable bool `json:"retryable"`
|
|
ErrorPayload []byte `json:"error_payload"`
|
|
RetryDelaySeconds int32 `json:"retry_delay_seconds"`
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkerID pgtype.Text `json:"worker_id"`
|
|
}
|
|
|
|
func (q *Queries) FailStepRun(ctx context.Context, arg FailStepRunParams) (StepRun, error) {
|
|
row := q.db.QueryRow(ctx, failStepRun,
|
|
arg.Retryable,
|
|
arg.ErrorPayload,
|
|
arg.RetryDelaySeconds,
|
|
arg.ID,
|
|
arg.WorkerID,
|
|
)
|
|
var i StepRun
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ProjectID,
|
|
&i.WorkflowRunID,
|
|
&i.StepKey,
|
|
&i.ExecutorKind,
|
|
&i.Status,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.Priority,
|
|
&i.Input,
|
|
&i.Output,
|
|
&i.Error,
|
|
&i.IdempotencyKey,
|
|
&i.AvailableAt,
|
|
&i.LeasedBy,
|
|
&i.LeasedUntil,
|
|
&i.StartedAt,
|
|
&i.FinishedAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const leaseNextStepRun = `-- name: LeaseNextStepRun :one
|
|
WITH candidate AS (
|
|
SELECT queued.id
|
|
FROM step_runs AS queued
|
|
WHERE queued.executor_kind = $3
|
|
AND queued.status IN ('pending', 'retryable_failed')
|
|
AND queued.available_at <= now()
|
|
AND queued.attempt < queued.max_attempts
|
|
AND (queued.leased_until IS NULL OR queued.leased_until < now())
|
|
ORDER BY queued.priority DESC, queued.created_at
|
|
FOR UPDATE SKIP LOCKED
|
|
LIMIT 1
|
|
)
|
|
UPDATE step_runs AS step
|
|
SET status = 'leased',
|
|
attempt = step.attempt + 1,
|
|
leased_by = $1,
|
|
leased_until = now() + make_interval(secs => $2::integer),
|
|
updated_at = now()
|
|
FROM candidate
|
|
WHERE step.id = candidate.id
|
|
RETURNING step.id, step.project_id, step.workflow_run_id, step.step_key, step.executor_kind, step.status, step.attempt, step.max_attempts, step.priority, step.input, step.output, step.error, step.idempotency_key, step.available_at, step.leased_by, step.leased_until, step.started_at, step.finished_at, step.created_at, step.updated_at
|
|
`
|
|
|
|
type LeaseNextStepRunParams struct {
|
|
WorkerID pgtype.Text `json:"worker_id"`
|
|
LeaseSeconds int32 `json:"lease_seconds"`
|
|
RequestedExecutorKind string `json:"requested_executor_kind"`
|
|
}
|
|
|
|
func (q *Queries) LeaseNextStepRun(ctx context.Context, arg LeaseNextStepRunParams) (StepRun, error) {
|
|
row := q.db.QueryRow(ctx, leaseNextStepRun, arg.WorkerID, arg.LeaseSeconds, arg.RequestedExecutorKind)
|
|
var i StepRun
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ProjectID,
|
|
&i.WorkflowRunID,
|
|
&i.StepKey,
|
|
&i.ExecutorKind,
|
|
&i.Status,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.Priority,
|
|
&i.Input,
|
|
&i.Output,
|
|
&i.Error,
|
|
&i.IdempotencyKey,
|
|
&i.AvailableAt,
|
|
&i.LeasedBy,
|
|
&i.LeasedUntil,
|
|
&i.StartedAt,
|
|
&i.FinishedAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const markStepRunRunning = `-- name: MarkStepRunRunning :one
|
|
UPDATE step_runs
|
|
SET status = 'running',
|
|
started_at = COALESCE(started_at, now()),
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND leased_by = $2
|
|
AND status = 'leased'
|
|
RETURNING id, project_id, workflow_run_id, step_key, executor_kind, status, attempt, max_attempts, priority, input, output, error, idempotency_key, available_at, leased_by, leased_until, started_at, finished_at, created_at, updated_at
|
|
`
|
|
|
|
type MarkStepRunRunningParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
LeasedBy pgtype.Text `json:"leased_by"`
|
|
}
|
|
|
|
func (q *Queries) MarkStepRunRunning(ctx context.Context, arg MarkStepRunRunningParams) (StepRun, error) {
|
|
row := q.db.QueryRow(ctx, markStepRunRunning, arg.ID, arg.LeasedBy)
|
|
var i StepRun
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ProjectID,
|
|
&i.WorkflowRunID,
|
|
&i.StepKey,
|
|
&i.ExecutorKind,
|
|
&i.Status,
|
|
&i.Attempt,
|
|
&i.MaxAttempts,
|
|
&i.Priority,
|
|
&i.Input,
|
|
&i.Output,
|
|
&i.Error,
|
|
&i.IdempotencyKey,
|
|
&i.AvailableAt,
|
|
&i.LeasedBy,
|
|
&i.LeasedUntil,
|
|
&i.StartedAt,
|
|
&i.FinishedAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|