knowledge-governance/internal/platform/postgres/queue_sql.go

package postgres

const enqueueJobSQL = `
INSERT INTO jobs (
    id, type, payload, state, priority, available_at,
    max_attempts, idempotency_key, created_at, updated_at
) VALUES ($1, $2, $3, 'pending', $4, $5, $6, $7, $8, $8)
ON CONFLICT (idempotency_key)
DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
RETURNING id, type, payload, state, priority, available_at,
          lease_owner, lease_until, attempt, max_attempts,
          last_error, idempotency_key, created_at, updated_at
`

const claimJobSQL = `
WITH budget AS (
    SELECT COALESCE((
        SELECT ingestion_concurrency
        FROM ui_setting_revisions
        WHERE state = 'active'
        LIMIT 1
    ), 2) AS ingestion_slots
), candidate AS (
    SELECT jobs.id
    FROM jobs, budget
    WHERE (
        (jobs.state IN ('pending', 'retry') AND jobs.available_at <= $2)
        OR (jobs.state = 'running' AND jobs.lease_until < $2)
    )
      AND jobs.attempt < jobs.max_attempts
      AND jobs.type = ANY($4::text[])
      AND (
        jobs.type <> 'test.upload'
        OR (SELECT COUNT(*) FROM jobs active
            WHERE active.type = 'test.upload' AND active.state = 'running') < budget.ingestion_slots
      )
    ORDER BY jobs.priority DESC, jobs.available_at ASC, jobs.created_at ASC, jobs.id ASC
    FOR UPDATE OF jobs SKIP LOCKED
    LIMIT 1
)
UPDATE jobs AS j
SET state = 'running',
    started_at = $2,
    lease_owner = $1,
    lease_until = $3,
    attempt = j.attempt + 1,
    updated_at = $2
FROM candidate
WHERE j.id = candidate.id
RETURNING j.id, j.type, j.payload, j.state, j.priority, j.available_at,
          j.lease_owner, j.lease_until, j.attempt, j.max_attempts,
          j.last_error, j.idempotency_key, j.created_at, j.updated_at
`

const completeJobSQL = `
WITH done AS (
    UPDATE jobs
    SET state = 'completed',
        completed_at = $3,
        lease_owner = NULL,
        lease_until = NULL,
        updated_at = $3
    WHERE id = $1
      AND lease_owner = $2
      AND state = 'running'
    RETURNING id
), retired AS (
    UPDATE job_failures
    SET state = 'requeued', resolved_at = $3, resolution_actor_ref = 'system:queue',
        resolution_reason = '任务后续重试成功'
    WHERE job_id IN (SELECT id FROM done) AND state = 'open'
    RETURNING 1
)
SELECT EXISTS (SELECT 1 FROM done)
`

const heartbeatJobSQL = `
UPDATE jobs
SET lease_until = $4,
    updated_at = $3
WHERE id = $1
  AND lease_owner = $2
  AND state = 'running'
`

const failJobSQL = `
WITH updated AS (
    UPDATE jobs
    SET state = CASE WHEN attempt >= max_attempts THEN 'failed' ELSE 'retry' END,
        available_at = $6,
        last_error = $3,
        last_error_code = CASE WHEN $5 <> '' THEN $5 ELSE last_error_code END,
        lease_owner = NULL,
        lease_until = NULL,
        updated_at = $4
    WHERE id = $1
      AND lease_owner = $2
      AND state = 'running'
    RETURNING id, type, payload, last_error,
              CASE WHEN last_error_code = '' THEN 'handler_error' ELSE last_error_code END AS error_code,
              attempt, max_attempts,
              COALESCE(business_change_set_id,'') AS bc, COALESCE(business_version_id,'') AS bv,
              COALESCE(business_action,'') AS ba, COALESCE(business_target_kb_id,'') AS bt,
              state
), projected_failure AS (
INSERT INTO job_failures (
    job_id, job_type, publication_id, error_code, last_error, attempt, max_attempts_at_failure,
    business_change_set_id, business_version_id, business_action, business_target_kb_id,
    created_at
)
SELECT updated.id, updated.type, COALESCE(updated.payload->>'publication_id',''), updated.error_code,
       updated.last_error, updated.attempt, updated.max_attempts,
       updated.bc, updated.bv, updated.ba, updated.bt, $4
FROM updated
WHERE updated.state = 'failed' AND updated.attempt > 0
ON CONFLICT (job_id, attempt) DO UPDATE
SET publication_id = EXCLUDED.publication_id,
    error_code = EXCLUDED.error_code,
    last_error = EXCLUDED.last_error,
    max_attempts_at_failure = EXCLUDED.max_attempts_at_failure,
    business_change_set_id = EXCLUDED.business_change_set_id,
    business_version_id = EXCLUDED.business_version_id,
    business_action = EXCLUDED.business_action,
    business_target_kb_id = EXCLUDED.business_target_kb_id,
    state = 'open',
    resolved_at = NULL,
    resolution_actor_ref = '',
    resolution_reason = '',
    created_at = EXCLUDED.created_at
RETURNING job_id
)
SELECT id FROM updated
`

// One job type, newest first: a caller which needs "the recent jobs of MY type"
// must not page the global queue (other traffic would hide its own rows).
const listJobsByTypeSQL = `
SELECT id, type, payload, state, priority, available_at,
       lease_owner, lease_until, attempt, max_attempts,
       last_error, idempotency_key, created_at, updated_at
FROM jobs
WHERE type = $1
ORDER BY created_at DESC, id DESC
LIMIT $2
`

const listJobsSQL = `
SELECT id, type, payload, state, priority, available_at,
       lease_owner, lease_until, attempt, max_attempts,
       last_error, idempotency_key, created_at, updated_at
FROM jobs
ORDER BY created_at DESC, id DESC
LIMIT $1
`

// P6: a permanent failure parks the job in 'failed' immediately and projects
// the terminal attempt into the dead-letter table.
const failTerminalJobSQL = `
WITH parked AS (
    UPDATE jobs
    SET state = 'failed',
        available_at = $5,
        last_error = $3,
        last_error_code = CASE WHEN $4 = '' THEN 'handler_error' ELSE $4 END,
        lease_owner = NULL,
        lease_until = NULL,
        updated_at = $5
    WHERE id = $1
      AND lease_owner = $2
      AND lease_owner IS NOT NULL
      AND state = 'running'
    RETURNING id, type, payload, last_error,
              CASE WHEN $4 = '' THEN 'handler_error' ELSE $4 END AS error_code,
              attempt, max_attempts,
              COALESCE(business_change_set_id,'') AS bc, COALESCE(business_version_id,'') AS bv,
              COALESCE(business_action,'') AS ba, COALESCE(business_target_kb_id,'') AS bt
)
INSERT INTO job_failures (
    job_id, job_type, publication_id, error_code, last_error, attempt, max_attempts_at_failure,
    business_change_set_id, business_version_id, business_action, business_target_kb_id,
    created_at
)
SELECT parked.id, parked.type, COALESCE(parked.payload->>'publication_id',''), parked.error_code,
       parked.last_error, GREATEST(parked.attempt,1), parked.max_attempts,
       parked.bc, parked.bv, parked.ba, parked.bt, $5
FROM parked
ON CONFLICT (job_id, attempt) DO UPDATE
SET publication_id = EXCLUDED.publication_id,
    error_code = EXCLUDED.error_code,
    last_error = EXCLUDED.last_error,
    max_attempts_at_failure = EXCLUDED.max_attempts_at_failure,
    state = 'open',
    resolved_at = NULL,
    resolution_actor_ref = '',
    resolution_reason = '',
    created_at = EXCLUDED.created_at
`

// P10 liveness: every claim attempt refreshes the worker row so the API can
// distinguish "worker loop alive" from "process dead" without an HTTP port.
const workerHeartbeatSQL = `
INSERT INTO worker_heartbeats(worker_id, job_types, started_at, last_seen_at)
VALUES($1, $2, $3, $3)
ON CONFLICT (worker_id) DO UPDATE
SET job_types = EXCLUDED.job_types, last_seen_at = EXCLUDED.last_seen_at
`