From ff3f27c1dc88beded3728354a9c5bd6e7f826680 Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Tue, 5 May 2026 16:59:03 +0800 Subject: [PATCH] feat(jobs): sqlc queries for jobs table --- internal/jobs/queries/jobs.sql | 65 +++++++++ internal/jobs/sqlc/db.go | 32 ++++ internal/jobs/sqlc/jobs.sql.go | 257 +++++++++++++++++++++++++++++++++ internal/jobs/sqlc/models.go | 249 ++++++++++++++++++++++++++++++++ sqlc.yaml | 10 ++ 5 files changed, 613 insertions(+) create mode 100644 internal/jobs/queries/jobs.sql create mode 100644 internal/jobs/sqlc/db.go create mode 100644 internal/jobs/sqlc/jobs.sql.go create mode 100644 internal/jobs/sqlc/models.go diff --git a/internal/jobs/queries/jobs.sql b/internal/jobs/queries/jobs.sql new file mode 100644 index 0000000..b697b8e --- /dev/null +++ b/internal/jobs/queries/jobs.sql @@ -0,0 +1,65 @@ +-- name: EnqueueJob :one +INSERT INTO jobs (id, type, payload, status, scheduled_at, attempts, max_attempts) +VALUES ($1, $2, $3, 'pending', $4, 0, $5) +RETURNING *; + +-- name: LeaseNextJob :one +-- SKIP LOCKED 让多 worker 并行抢不冲突；只返回单行。 +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 *; + +-- name: CompleteJob :exec +UPDATE jobs + SET status='completed', completed_at=now(), + leased_at=NULL, leased_by=NULL, last_error=NULL, + updated_at=now() + WHERE id=$1; + +-- name: FailJobWithRetry :exec +UPDATE jobs + SET status='pending', + scheduled_at=$2, + leased_at=NULL, leased_by=NULL, + last_error=$3, + updated_at=now() + WHERE id=$1; + +-- name: DeadLetterJob :exec +UPDATE jobs + SET status='dead', completed_at=now(), + leased_at=NULL, leased_by=NULL, last_error=$2, + updated_at=now() + WHERE id=$1; + +-- name: GetJob :one +SELECT * FROM jobs WHERE id=$1; + +-- name: ReapStuckRunningJobs :many +-- 复位 worker 崩了卡 running 的任务。 +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; + +-- name: PurgeCompletedJobs :execrows +DELETE FROM jobs + WHERE status IN ('completed','dead') AND completed_at < $1; + +-- name: ListDeadJobs :many +SELECT * FROM jobs + WHERE status='dead' AND completed_at >= $1 + ORDER BY completed_at DESC + LIMIT $2; diff --git a/internal/jobs/sqlc/db.go b/internal/jobs/sqlc/db.go new file mode 100644 index 0000000..f18495f --- /dev/null +++ b/internal/jobs/sqlc/db.go @@ -0,0 +1,32 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package jobssqlc + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +type DBTX interface { + Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error) + Query(context.Context, string, ...interface{}) (pgx.Rows, error) + QueryRow(context.Context, string, ...interface{}) pgx.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +type Queries struct { + db DBTX +} + +func (q *Queries) WithTx(tx pgx.Tx) *Queries { + return &Queries{ + db: tx, + } +} diff --git a/internal/jobs/sqlc/jobs.sql.go b/internal/jobs/sqlc/jobs.sql.go new file mode 100644 index 0000000..0e7c18e --- /dev/null +++ b/internal/jobs/sqlc/jobs.sql.go @@ -0,0 +1,257 @@ +// 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 :exec +UPDATE jobs + SET status='completed', completed_at=now(), + leased_at=NULL, leased_by=NULL, last_error=NULL, + updated_at=now() + WHERE id=$1 +` + +func (q *Queries) CompleteJob(ctx context.Context, id pgtype.UUID) error { + _, err := q.db.Exec(ctx, completeJob, id) + return err +} + +const deadLetterJob = `-- name: DeadLetterJob :exec +UPDATE jobs + SET status='dead', completed_at=now(), + leased_at=NULL, leased_by=NULL, last_error=$2, + updated_at=now() + WHERE id=$1 +` + +type DeadLetterJobParams struct { + ID pgtype.UUID `json:"id"` + LastError *string `json:"last_error"` +} + +func (q *Queries) DeadLetterJob(ctx context.Context, arg DeadLetterJobParams) error { + _, err := q.db.Exec(ctx, deadLetterJob, arg.ID, arg.LastError) + return err +} + +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 :exec +UPDATE jobs + SET status='pending', + scheduled_at=$2, + leased_at=NULL, leased_by=NULL, + last_error=$3, + updated_at=now() + WHERE id=$1 +` + +type FailJobWithRetryParams struct { + ID pgtype.UUID `json:"id"` + ScheduledAt pgtype.Timestamptz `json:"scheduled_at"` + LastError *string `json:"last_error"` +} + +func (q *Queries) FailJobWithRetry(ctx context.Context, arg FailJobWithRetryParams) error { + _, err := q.db.Exec(ctx, failJobWithRetry, arg.ID, arg.ScheduledAt, arg.LastError) + return err +} + +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 +} diff --git a/internal/jobs/sqlc/models.go b/internal/jobs/sqlc/models.go new file mode 100644 index 0000000..4a84f0c --- /dev/null +++ b/internal/jobs/sqlc/models.go @@ -0,0 +1,249 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package jobssqlc + +import ( + "net/netip" + + "github.com/jackc/pgx/v5/pgtype" +) + +type AuditLog struct { + ID int64 `json:"id"` + UserID pgtype.UUID `json:"user_id"` + Action string `json:"action"` + TargetType *string `json:"target_type"` + TargetID *string `json:"target_id"` + Ip *netip.Addr `json:"ip"` + Metadata []byte `json:"metadata"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Conversation struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + ProjectID pgtype.UUID `json:"project_id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + IssueID pgtype.UUID `json:"issue_id"` + ModelID pgtype.UUID `json:"model_id"` + SystemPrompt string `json:"system_prompt"` + Title *string `json:"title"` + TitleGeneratedAt pgtype.Timestamptz `json:"title_generated_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + DeletedAt pgtype.Timestamptz `json:"deleted_at"` +} + +type GitCredential struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Kind string `json:"kind"` + Username *string `json:"username"` + EncryptedSecret []byte `json:"encrypted_secret"` + Fingerprint *string `json:"fingerprint"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type Issue struct { + ID pgtype.UUID `json:"id"` + ProjectID pgtype.UUID `json:"project_id"` + RequirementID pgtype.UUID `json:"requirement_id"` + Number int32 `json:"number"` + Title string `json:"title"` + Description string `json:"description"` + Status string `json:"status"` + AssigneeID pgtype.UUID `json:"assignee_id"` + CreatedBy pgtype.UUID `json:"created_by"` + ClosedAt pgtype.Timestamptz `json:"closed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + WorkspaceID pgtype.UUID `json:"workspace_id"` +} + +type Job struct { + ID pgtype.UUID `json:"id"` + Type string `json:"type"` + Payload []byte `json:"payload"` + Status string `json:"status"` + ScheduledAt pgtype.Timestamptz `json:"scheduled_at"` + Attempts int32 `json:"attempts"` + MaxAttempts int32 `json:"max_attempts"` + LeasedAt pgtype.Timestamptz `json:"leased_at"` + LeasedBy *string `json:"leased_by"` + LastError *string `json:"last_error"` + CompletedAt pgtype.Timestamptz `json:"completed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type LlmEndpoint struct { + ID pgtype.UUID `json:"id"` + Provider string `json:"provider"` + DisplayName string `json:"display_name"` + BaseUrl string `json:"base_url"` + ApiKeyEncrypted []byte `json:"api_key_encrypted"` + CreatedBy pgtype.UUID `json:"created_by"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type LlmModel struct { + ID pgtype.UUID `json:"id"` + EndpointID pgtype.UUID `json:"endpoint_id"` + ModelID string `json:"model_id"` + DisplayName string `json:"display_name"` + Capabilities []byte `json:"capabilities"` + ContextWindow int32 `json:"context_window"` + MaxOutputTokens int32 `json:"max_output_tokens"` + PromptPricePerMillionUsd pgtype.Numeric `json:"prompt_price_per_million_usd"` + CompletionPricePerMillionUsd pgtype.Numeric `json:"completion_price_per_million_usd"` + ThinkingPricePerMillionUsd pgtype.Numeric `json:"thinking_price_per_million_usd"` + IsTitleGenerator bool `json:"is_title_generator"` + Enabled bool `json:"enabled"` + SortOrder int32 `json:"sort_order"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type LlmUsage struct { + ID int64 `json:"id"` + ConversationID pgtype.UUID `json:"conversation_id"` + MessageID int64 `json:"message_id"` + UserID pgtype.UUID `json:"user_id"` + EndpointID pgtype.UUID `json:"endpoint_id"` + ModelID pgtype.UUID `json:"model_id"` + PromptTokens int32 `json:"prompt_tokens"` + CompletionTokens int32 `json:"completion_tokens"` + ThinkingTokens int32 `json:"thinking_tokens"` + CostUsd pgtype.Numeric `json:"cost_usd"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Message struct { + ID int64 `json:"id"` + ConversationID pgtype.UUID `json:"conversation_id"` + Role string `json:"role"` + Content string `json:"content"` + Thinking *string `json:"thinking"` + ToolCalls []byte `json:"tool_calls"` + Status string `json:"status"` + ErrorMessage *string `json:"error_message"` + PromptTokens *int32 `json:"prompt_tokens"` + CompletionTokens *int32 `json:"completion_tokens"` + ThinkingTokens *int32 `json:"thinking_tokens"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type MessageAttachment struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + MessageID *int64 `json:"message_id"` + Filename string `json:"filename"` + MimeType string `json:"mime_type"` + SizeBytes int64 `json:"size_bytes"` + Sha256 string `json:"sha256"` + StoragePath string `json:"storage_path"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Notification struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + Topic string `json:"topic"` + Severity string `json:"severity"` + Title string `json:"title"` + Body string `json:"body"` + Link *string `json:"link"` + Metadata []byte `json:"metadata"` + ReadAt pgtype.Timestamptz `json:"read_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Project struct { + ID pgtype.UUID `json:"id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + Visibility string `json:"visibility"` + OwnerID pgtype.UUID `json:"owner_id"` + ArchivedAt pgtype.Timestamptz `json:"archived_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type PromptTemplate struct { + ID pgtype.UUID `json:"id"` + Name string `json:"name"` + Content string `json:"content"` + Scope string `json:"scope"` + OwnerID pgtype.UUID `json:"owner_id"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type Requirement struct { + ID pgtype.UUID `json:"id"` + ProjectID pgtype.UUID `json:"project_id"` + Number int32 `json:"number"` + Title string `json:"title"` + Description string `json:"description"` + Phase string `json:"phase"` + Status string `json:"status"` + OwnerID pgtype.UUID `json:"owner_id"` + ClosedAt pgtype.Timestamptz `json:"closed_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` + WorkspaceID pgtype.UUID `json:"workspace_id"` +} + +type User struct { + ID pgtype.UUID `json:"id"` + Email string `json:"email"` + PasswordHash string `json:"password_hash"` + DisplayName string `json:"display_name"` + IsAdmin bool `json:"is_admin"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type UserSession struct { + ID pgtype.UUID `json:"id"` + UserID pgtype.UUID `json:"user_id"` + TokenHash []byte `json:"token_hash"` + ExpiresAt pgtype.Timestamptz `json:"expires_at"` + LastSeenAt pgtype.Timestamptz `json:"last_seen_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} + +type Workspace struct { + ID pgtype.UUID `json:"id"` + ProjectID pgtype.UUID `json:"project_id"` + Slug string `json:"slug"` + Name string `json:"name"` + Description string `json:"description"` + GitRemoteUrl string `json:"git_remote_url"` + DefaultBranch string `json:"default_branch"` + MainPath string `json:"main_path"` + SyncStatus string `json:"sync_status"` + LastSyncedAt pgtype.Timestamptz `json:"last_synced_at"` + LastSyncError *string `json:"last_sync_error"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + UpdatedAt pgtype.Timestamptz `json:"updated_at"` +} + +type WorkspaceWorktree struct { + ID pgtype.UUID `json:"id"` + WorkspaceID pgtype.UUID `json:"workspace_id"` + Branch string `json:"branch"` + Path string `json:"path"` + Status string `json:"status"` + ActiveHolder *string `json:"active_holder"` + AcquiredAt pgtype.Timestamptz `json:"acquired_at"` + LastUsedAt pgtype.Timestamptz `json:"last_used_at"` + CreatedAt pgtype.Timestamptz `json:"created_at"` + PruneWarningAt pgtype.Timestamptz `json:"prune_warning_at"` +} diff --git a/sqlc.yaml b/sqlc.yaml index 27609e5..51a5e23 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -60,3 +60,13 @@ sql: sql_package: pgx/v5 emit_pointers_for_null_types: true emit_json_tags: true + - engine: postgresql + queries: internal/jobs/queries + schema: migrations + gen: + go: + package: jobssqlc + out: internal/jobs/sqlc + sql_package: pgx/v5 + emit_pointers_for_null_types: true + emit_json_tags: true