姚章浩
86732ca987
逐项明细 DTO 漏了 action（前端 parseReleaseItemExecution 必需），点 「查看结果」整个明细区报 invalid release item execution。同时把遷移 000044 的 target_knowledge_base_id 读出来透出，页面逐项结果显示「目标库」， 让'目标知识库可选'在界面上可核对。测试：Go 契约用例 + 前端解析回归用例。
542 lines
25 KiB
Go
542 lines
25 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"knowledge-governance/internal/governance"
|
|
)
|
|
|
|
var (
|
|
ErrInvalidReleaseAuthorization = errors.New("invalid release authorization")
|
|
ErrReleaseAuthorizationGateConflict = errors.New("release authorization gate conflict")
|
|
ErrInvalidReleaseExecutionFilter = errors.New("invalid release execution filter")
|
|
ErrReleaseExecutionNotFound = errors.New("release execution not found")
|
|
ErrReleaseExecutionRetryConflict = errors.New("release execution retry conflict")
|
|
)
|
|
|
|
const authorizeReleaseBatchSQL = `
|
|
WITH ` + releaseGateCTEs + `, ` + releaseDeleteReadyCTEs + `,
|
|
batch_gate AS (
|
|
SELECT batch.id AS release_batch_id,
|
|
batch.item_count,
|
|
batch.release_change_set_id,
|
|
COUNT(item.publication_id)::INTEGER AS total_items,
|
|
COUNT(*) FILTER (WHERE
|
|
(item.action = 'upsert' AND ready.publication_id IS NOT NULL)
|
|
OR (item.action = 'delete' AND delete_ready.publication_id IS NOT NULL)
|
|
)::INTEGER AS ready_items
|
|
FROM release_batches AS batch
|
|
JOIN release_batch_items AS item ON item.release_batch_id = batch.id
|
|
LEFT JOIN release_upsert_ready AS ready
|
|
ON ready.publication_id = item.publication_id
|
|
AND item.action = 'upsert'
|
|
AND ready.review_id = item.review_id
|
|
AND ready.acceptance_run_id = item.acceptance_run_id
|
|
AND ready.publication_updated_at = item.publication_updated_at
|
|
AND ready.decided_visibility = item.decided_visibility
|
|
LEFT JOIN release_delete_ready AS delete_ready
|
|
ON delete_ready.publication_id = item.publication_id
|
|
AND item.action = 'delete'
|
|
AND delete_ready.review_id = item.review_id
|
|
AND delete_ready.acceptance_run_id = item.acceptance_run_id
|
|
AND delete_ready.publication_updated_at = item.publication_updated_at
|
|
AND delete_ready.decided_visibility = item.decided_visibility
|
|
WHERE batch.id = $1::VARCHAR
|
|
AND batch.state = 'sealed'
|
|
AND batch.policy_version = 'release-gate-v1'
|
|
GROUP BY batch.id, batch.item_count, batch.release_change_set_id
|
|
), eligible_batch AS (
|
|
SELECT release_batch_id, item_count, release_change_set_id
|
|
FROM batch_gate
|
|
WHERE total_items > 0
|
|
AND total_items = ready_items
|
|
AND item_count = total_items
|
|
), inserted_job AS (
|
|
INSERT INTO jobs (
|
|
id, type, payload, state, idempotency_key,
|
|
max_attempts, available_at, created_at, updated_at
|
|
)
|
|
SELECT $3::VARCHAR, 'release.publish',
|
|
jsonb_build_object(
|
|
'execution_id', $2::VARCHAR,
|
|
'release_batch_id', eligible_batch.release_batch_id,
|
|
'operation', 'publish'
|
|
),
|
|
'pending', $7::VARCHAR, 3, $6::TIMESTAMPTZ, $6::TIMESTAMPTZ, $6::TIMESTAMPTZ
|
|
FROM eligible_batch
|
|
ON CONFLICT (idempotency_key)
|
|
DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
|
|
RETURNING id
|
|
), inserted_execution AS (
|
|
INSERT INTO release_executions (
|
|
id, release_batch_id, operation, state, target_environment,
|
|
authorized_by, authorization_code, job_id, item_count, created_at, release_change_set_id
|
|
)
|
|
SELECT $2::VARCHAR, eligible_batch.release_batch_id, 'publish', 'pending', 'production',
|
|
$4::VARCHAR, $5::VARCHAR, inserted_job.id, eligible_batch.item_count, $6::TIMESTAMPTZ,
|
|
eligible_batch.release_change_set_id
|
|
FROM eligible_batch
|
|
CROSS JOIN inserted_job
|
|
ON CONFLICT (release_batch_id) WHERE operation = 'publish'
|
|
DO UPDATE SET release_batch_id = EXCLUDED.release_batch_id
|
|
RETURNING id, release_batch_id, operation, COALESCE(prior_execution_id, '') AS prior_execution_id, state, target_environment,
|
|
authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, COALESCE(release_change_set_id::text,'') AS release_change_set_id
|
|
)
|
|
SELECT id, release_batch_id, operation, prior_execution_id, state, target_environment,
|
|
authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, release_change_set_id
|
|
FROM inserted_execution
|
|
`
|
|
|
|
const authorizeReleaseRollbackSQL = `
|
|
WITH eligible_prior AS (
|
|
SELECT execution.id, execution.release_batch_id, execution.release_change_set_id, COUNT(*)::INTEGER AS item_count
|
|
FROM release_executions AS execution
|
|
JOIN release_item_executions AS item ON item.release_execution_id = execution.id
|
|
WHERE execution.id = $1::VARCHAR
|
|
AND execution.operation = 'publish'
|
|
AND execution.state IN ('completed', 'failed')
|
|
AND execution.failure_level <> 'overall'
|
|
AND item.state = 'verified'
|
|
GROUP BY execution.id, execution.release_batch_id, execution.release_change_set_id
|
|
HAVING COUNT(*) > 0
|
|
), inserted_job AS (
|
|
INSERT INTO jobs (
|
|
id, type, payload, state, idempotency_key,
|
|
max_attempts, available_at, created_at, updated_at
|
|
)
|
|
SELECT $3::VARCHAR, 'release.rollback',
|
|
jsonb_build_object(
|
|
'execution_id', $2::VARCHAR,
|
|
'prior_execution_id', eligible_prior.id,
|
|
'release_batch_id', eligible_prior.release_batch_id,
|
|
'operation', 'rollback'
|
|
),
|
|
'pending', $7::VARCHAR, 3, $6::TIMESTAMPTZ, $6::TIMESTAMPTZ, $6::TIMESTAMPTZ
|
|
FROM eligible_prior
|
|
ON CONFLICT (idempotency_key)
|
|
DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key
|
|
RETURNING id
|
|
), inserted_execution AS (
|
|
INSERT INTO release_executions (
|
|
id, release_batch_id, operation, prior_execution_id, state, target_environment,
|
|
authorized_by, authorization_code, job_id, item_count, created_at, release_change_set_id
|
|
)
|
|
SELECT $2::VARCHAR, eligible_prior.release_batch_id, 'rollback', eligible_prior.id,
|
|
'pending', 'production', $4::VARCHAR, $5::VARCHAR,
|
|
inserted_job.id, eligible_prior.item_count, $6::TIMESTAMPTZ,
|
|
eligible_prior.release_change_set_id
|
|
FROM eligible_prior
|
|
CROSS JOIN inserted_job
|
|
ON CONFLICT (prior_execution_id) WHERE operation = 'rollback'
|
|
DO UPDATE SET prior_execution_id = EXCLUDED.prior_execution_id
|
|
RETURNING id, release_batch_id, operation, COALESCE(prior_execution_id, '') AS prior_execution_id, state, target_environment,
|
|
authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, COALESCE(release_change_set_id::text,'') AS release_change_set_id
|
|
)
|
|
SELECT id, release_batch_id, operation, prior_execution_id, state, target_environment,
|
|
authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, release_change_set_id
|
|
FROM inserted_execution
|
|
`
|
|
|
|
const scheduleAutomaticRollbackSQL = `
|
|
WITH eligible_prior AS (
|
|
SELECT execution.id, execution.release_batch_id, execution.release_change_set_id, COUNT(*)::INTEGER AS item_count
|
|
FROM release_executions AS execution
|
|
JOIN release_item_executions AS item ON item.release_execution_id = execution.id
|
|
WHERE execution.id = $1::VARCHAR
|
|
AND execution.operation = 'publish'
|
|
AND execution.state = 'failed'
|
|
AND item.state = 'verified'
|
|
GROUP BY execution.id, execution.release_batch_id, execution.release_change_set_id
|
|
HAVING COUNT(*)>0
|
|
), inserted_job AS (
|
|
INSERT INTO jobs(id,type,payload,state,idempotency_key,max_attempts,available_at,created_at,updated_at)
|
|
SELECT $3::VARCHAR,'release.rollback',jsonb_build_object(
|
|
'execution_id',$2::VARCHAR,
|
|
'prior_execution_id',eligible_prior.id,
|
|
'release_batch_id',eligible_prior.release_batch_id,
|
|
'operation','rollback'),
|
|
'pending',$5::VARCHAR,3,$4::TIMESTAMPTZ,$4::TIMESTAMPTZ,$4::TIMESTAMPTZ
|
|
FROM eligible_prior
|
|
ON CONFLICT(idempotency_key) DO UPDATE SET idempotency_key=EXCLUDED.idempotency_key
|
|
RETURNING id
|
|
), inserted_execution AS (
|
|
INSERT INTO release_executions(
|
|
id,release_batch_id,operation,prior_execution_id,state,target_environment,
|
|
authorized_by,authorization_code,job_id,item_count,created_at,release_change_set_id)
|
|
SELECT $2::VARCHAR,eligible_prior.release_batch_id,'rollback',eligible_prior.id,
|
|
'pending','production','system:auto-rollback','automatic_rollback_after_failure',
|
|
inserted_job.id,eligible_prior.item_count,$4::TIMESTAMPTZ,
|
|
eligible_prior.release_change_set_id
|
|
FROM eligible_prior CROSS JOIN inserted_job
|
|
ON CONFLICT(prior_execution_id) WHERE operation='rollback'
|
|
DO UPDATE SET prior_execution_id=EXCLUDED.prior_execution_id
|
|
RETURNING id,release_batch_id,operation,COALESCE(prior_execution_id,'') AS prior_execution_id,state,target_environment,
|
|
authorized_by,authorization_code,job_id,manifest_sha256,item_count,
|
|
batch_size,current_batch,total_batches,completed_item_count,failed_item_count,error_code,created_at,started_at,completed_at,failure_level,restore_trigger,COALESCE(release_change_set_id::text,'') AS release_change_set_id
|
|
)
|
|
SELECT id,release_batch_id,operation,prior_execution_id,state,target_environment,
|
|
authorized_by,authorization_code,job_id,manifest_sha256,item_count,
|
|
batch_size,current_batch,total_batches,completed_item_count,failed_item_count,error_code,created_at,started_at,completed_at,failure_level,restore_trigger,release_change_set_id
|
|
FROM inserted_execution
|
|
`
|
|
|
|
const retryReleaseExecutionSQL = `
|
|
WITH ` + releaseGateCTEs + `, ` + releaseDeleteReadyCTEs + `,
|
|
candidate AS (
|
|
SELECT execution.id, execution.release_batch_id, execution.operation,
|
|
execution.prior_execution_id, execution.job_id
|
|
FROM release_executions AS execution
|
|
WHERE execution.id = $1
|
|
AND (execution.state = 'failed' OR (execution.operation = 'rollback' AND execution.state IN ('rollback_failed', 'rolling_back')))
|
|
AND (
|
|
execution.operation <> 'publish'
|
|
OR NOT EXISTS (
|
|
SELECT 1 FROM release_executions AS rollback
|
|
WHERE rollback.prior_execution_id = execution.id
|
|
AND rollback.operation = 'rollback'
|
|
)
|
|
)
|
|
AND (
|
|
execution.operation = 'publish'
|
|
OR EXISTS (
|
|
SELECT 1 FROM release_executions AS prior
|
|
WHERE prior.id = execution.prior_execution_id
|
|
AND prior.operation = 'publish'
|
|
AND prior.state IN ('completed', 'failed')
|
|
)
|
|
)
|
|
),
|
|
gate_stats AS (
|
|
SELECT candidate.id, candidate.release_batch_id, candidate.operation,
|
|
candidate.prior_execution_id, candidate.job_id,
|
|
COUNT(item.publication_id)::INTEGER AS total_items,
|
|
COUNT(*) FILTER (
|
|
WHERE (item.action = 'upsert'
|
|
AND ready.publication_id IS NOT NULL
|
|
AND ready.review_id = item.review_id
|
|
AND ready.acceptance_run_id = item.acceptance_run_id
|
|
AND ready.publication_updated_at = item.publication_updated_at
|
|
AND ready.decided_visibility = item.decided_visibility)
|
|
OR (item.action = 'delete'
|
|
AND delete_ready.publication_id IS NOT NULL
|
|
AND delete_ready.review_id = item.review_id
|
|
AND delete_ready.acceptance_run_id = item.acceptance_run_id
|
|
AND delete_ready.publication_updated_at = item.publication_updated_at
|
|
AND delete_ready.decided_visibility = item.decided_visibility)
|
|
)::INTEGER AS ready_items
|
|
FROM candidate
|
|
JOIN release_batch_items AS item ON item.release_batch_id = candidate.release_batch_id
|
|
LEFT JOIN release_upsert_ready AS ready ON ready.publication_id = item.publication_id AND item.action = 'upsert'
|
|
LEFT JOIN release_delete_ready AS delete_ready ON delete_ready.publication_id = item.publication_id AND item.action = 'delete'
|
|
GROUP BY candidate.id, candidate.release_batch_id, candidate.operation,
|
|
candidate.prior_execution_id, candidate.job_id
|
|
),
|
|
eligible AS (
|
|
SELECT * FROM gate_stats
|
|
WHERE operation = 'rollback'
|
|
OR (total_items > 0 AND ready_items = total_items)
|
|
),
|
|
reset_items AS (
|
|
UPDATE release_item_executions AS item
|
|
SET state = 'pending', bundle_sha256 = '', previous_production_revision = '',
|
|
production_revision = '', error_code = '', updated_at = $2, completed_at = NULL
|
|
FROM eligible
|
|
WHERE item.release_execution_id = eligible.id
|
|
AND item.state = 'failed'
|
|
RETURNING item.release_execution_id
|
|
),
|
|
reset_item_count AS (
|
|
SELECT release_execution_id, COUNT(*)::INTEGER AS count
|
|
FROM reset_items
|
|
GROUP BY release_execution_id
|
|
),
|
|
completed_counts AS (
|
|
SELECT item.release_execution_id,
|
|
COUNT(*) FILTER (WHERE item.state IN ('verified', 'rolled_back'))::INTEGER AS completed
|
|
FROM release_item_executions AS item
|
|
JOIN eligible ON eligible.id = item.release_execution_id
|
|
GROUP BY item.release_execution_id
|
|
),
|
|
reset_job AS (
|
|
UPDATE jobs AS job
|
|
SET state = 'pending', attempt = 0, last_error = '', available_at = $2,
|
|
lease_owner = NULL, lease_until = NULL, completed_at = NULL, updated_at = $2
|
|
FROM eligible
|
|
JOIN reset_item_count ON reset_item_count.release_execution_id = eligible.id
|
|
WHERE job.id = eligible.job_id
|
|
AND job.state IN ('completed', 'failed')
|
|
RETURNING job.id
|
|
),
|
|
reset_execution AS (
|
|
UPDATE release_executions AS execution
|
|
SET state = 'pending', completed_item_count = completed_counts.completed,
|
|
failed_item_count = 0, error_code = '', started_at = NULL, completed_at = NULL,
|
|
failure_level = ''
|
|
FROM eligible
|
|
JOIN completed_counts ON completed_counts.release_execution_id = eligible.id
|
|
JOIN reset_job ON reset_job.id = eligible.job_id
|
|
WHERE execution.id = eligible.id
|
|
RETURNING execution.id, execution.release_batch_id, execution.operation,
|
|
COALESCE(execution.prior_execution_id, '') AS prior_execution_id,
|
|
execution.state, execution.target_environment, execution.authorized_by,
|
|
execution.authorization_code, execution.job_id, execution.manifest_sha256,
|
|
execution.item_count, execution.batch_size, execution.current_batch, execution.total_batches,
|
|
execution.completed_item_count, execution.failed_item_count,
|
|
execution.error_code, execution.created_at, execution.started_at, execution.completed_at,
|
|
execution.failure_level, execution.restore_trigger, COALESCE(execution.release_change_set_id::text,'') AS release_change_set_id
|
|
)
|
|
SELECT id, release_batch_id, operation, prior_execution_id, state, target_environment,
|
|
authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, release_change_set_id
|
|
FROM reset_execution
|
|
`
|
|
|
|
const countReleaseExecutionsSQL = `
|
|
SELECT COUNT(*)::INTEGER
|
|
FROM release_executions
|
|
WHERE ($1 = '' OR operation = $1)
|
|
AND ($2 = '' OR state = $2)
|
|
`
|
|
|
|
const listReleaseExecutionsSQL = `
|
|
SELECT id, release_batch_id, operation, COALESCE(prior_execution_id, ''), state,
|
|
target_environment, authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, COALESCE(release_change_set_id::text,'')
|
|
FROM release_executions
|
|
WHERE ($1 = '' OR operation = $1)
|
|
AND ($2 = '' OR state = $2)
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT $3 OFFSET $4
|
|
`
|
|
|
|
const getReleaseExecutionSQL = `
|
|
SELECT id, release_batch_id, operation, COALESCE(prior_execution_id, ''), state,
|
|
target_environment, authorized_by, authorization_code, job_id, manifest_sha256,
|
|
item_count, batch_size, current_batch, total_batches, completed_item_count, failed_item_count, error_code,
|
|
created_at, started_at, completed_at, failure_level, restore_trigger, COALESCE(release_change_set_id::text,'')
|
|
FROM release_executions
|
|
WHERE id = $1
|
|
`
|
|
|
|
const listReleaseExecutionItemsSQL = `
|
|
SELECT item.publication_id, source.title, item.action, item.state,
|
|
item.previous_production_revision, item.production_revision,
|
|
item.error_code, item.created_at, item.completed_at,
|
|
COALESCE(item.target_knowledge_base_id, '')
|
|
FROM release_item_executions AS item
|
|
JOIN test_publications AS publication ON publication.id = item.publication_id
|
|
JOIN source_item_versions AS version ON version.id = publication.source_item_version_id
|
|
JOIN source_items AS source ON source.id = version.source_item_id
|
|
WHERE item.release_execution_id = $1
|
|
ORDER BY item.created_at, item.publication_id
|
|
`
|
|
|
|
func (s *Store) AuthorizeReleaseBatch(ctx context.Context, request governance.AuthorizeReleaseBatchRequest) (governance.ReleaseExecution, error) {
|
|
if !validShortID(request.BatchID) || !validReviewerRef(request.AuthorizedBy) || request.AuthorizationCode != "production_release_authorized" || request.Now.IsZero() {
|
|
return governance.ReleaseExecution{}, ErrInvalidReleaseAuthorization
|
|
}
|
|
executionID, err := newOpaqueID("rx-")
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, err
|
|
}
|
|
jobID, err := newOpaqueID("job-")
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, err
|
|
}
|
|
idempotencyKey := "release:publish:" + request.BatchID
|
|
execution, err := scanReleaseExecution(s.db.QueryRow(ctx, authorizeReleaseBatchSQL,
|
|
request.BatchID, executionID, jobID, request.AuthorizedBy,
|
|
request.AuthorizationCode, request.Now.UTC(), idempotencyKey,
|
|
))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return governance.ReleaseExecution{}, ErrReleaseAuthorizationGateConflict
|
|
}
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, fmt.Errorf("authorize release batch: %w", err)
|
|
}
|
|
return execution, nil
|
|
}
|
|
|
|
func (s *Store) AuthorizeReleaseRollback(ctx context.Context, request governance.AuthorizeReleaseRollbackRequest) (governance.ReleaseExecution, error) {
|
|
if !validShortID(request.PriorExecutionID) || !validReviewerRef(request.AuthorizedBy) || request.AuthorizationCode != "production_rollback_authorized" || request.Now.IsZero() {
|
|
return governance.ReleaseExecution{}, ErrInvalidReleaseAuthorization
|
|
}
|
|
executionID, err := newOpaqueID("rx-")
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, err
|
|
}
|
|
jobID, err := newOpaqueID("job-")
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, err
|
|
}
|
|
idempotencyKey := "release:rollback:" + request.PriorExecutionID
|
|
execution, err := scanReleaseExecution(s.db.QueryRow(ctx, authorizeReleaseRollbackSQL,
|
|
request.PriorExecutionID, executionID, jobID, request.AuthorizedBy,
|
|
request.AuthorizationCode, request.Now.UTC(), idempotencyKey,
|
|
))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return governance.ReleaseExecution{}, ErrReleaseAuthorizationGateConflict
|
|
}
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, fmt.Errorf("authorize release rollback: %w", err)
|
|
}
|
|
return execution, nil
|
|
}
|
|
|
|
func (s *Store) ScheduleAutomaticRollback(ctx context.Context, priorExecutionID string, now time.Time) error {
|
|
if !validShortID(priorExecutionID) || now.IsZero() {
|
|
return ErrInvalidReleaseAuthorization
|
|
}
|
|
executionID, err := newOpaqueID("rx-")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
jobID, err := newOpaqueID("job-")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = scanReleaseExecution(s.db.QueryRow(ctx, scheduleAutomaticRollbackSQL,
|
|
priorExecutionID, executionID, jobID, now.UTC(), "release:auto-rollback:"+priorExecutionID,
|
|
))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("schedule automatic release rollback: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) RetryReleaseExecution(ctx context.Context, request governance.RetryReleaseExecutionRequest) (governance.ReleaseExecution, error) {
|
|
if !validShortID(request.ExecutionID) || request.Now.IsZero() {
|
|
return governance.ReleaseExecution{}, ErrReleaseExecutionRetryConflict
|
|
}
|
|
execution, err := scanReleaseExecution(s.db.QueryRow(ctx, retryReleaseExecutionSQL, request.ExecutionID, request.Now.UTC()))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return governance.ReleaseExecution{}, ErrReleaseExecutionRetryConflict
|
|
}
|
|
if err != nil {
|
|
return governance.ReleaseExecution{}, fmt.Errorf("retry release execution: %w", err)
|
|
}
|
|
return execution, nil
|
|
}
|
|
|
|
func (s *Store) ListReleaseExecutions(ctx context.Context, filter governance.ReleaseExecutionFilter) (governance.ReleaseExecutionPage, error) {
|
|
if !validReleaseExecutionFilter(filter) {
|
|
return governance.ReleaseExecutionPage{}, ErrInvalidReleaseExecutionFilter
|
|
}
|
|
var total int
|
|
if err := s.db.QueryRow(ctx, countReleaseExecutionsSQL, filter.Operation, filter.State).Scan(&total); err != nil {
|
|
return governance.ReleaseExecutionPage{}, fmt.Errorf("count release executions: %w", err)
|
|
}
|
|
rows, err := s.db.Query(ctx, listReleaseExecutionsSQL, filter.Operation, filter.State, filter.Limit, filter.Offset)
|
|
if err != nil {
|
|
return governance.ReleaseExecutionPage{}, fmt.Errorf("list release executions: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
items := make([]governance.ReleaseExecution, 0)
|
|
for rows.Next() {
|
|
item, err := scanReleaseExecution(rows)
|
|
if err != nil {
|
|
return governance.ReleaseExecutionPage{}, fmt.Errorf("scan release execution: %w", err)
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return governance.ReleaseExecutionPage{}, fmt.Errorf("iterate release executions: %w", err)
|
|
}
|
|
return governance.ReleaseExecutionPage{Items: items, Total: total, Limit: filter.Limit, Offset: filter.Offset}, nil
|
|
}
|
|
|
|
func (s *Store) GetReleaseExecution(ctx context.Context, executionID string) (governance.ReleaseExecutionDetail, error) {
|
|
if !validShortID(executionID) {
|
|
return governance.ReleaseExecutionDetail{}, ErrReleaseExecutionNotFound
|
|
}
|
|
execution, err := scanReleaseExecution(s.db.QueryRow(ctx, getReleaseExecutionSQL, executionID))
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return governance.ReleaseExecutionDetail{}, ErrReleaseExecutionNotFound
|
|
}
|
|
if err != nil {
|
|
return governance.ReleaseExecutionDetail{}, fmt.Errorf("get release execution: %w", err)
|
|
}
|
|
rows, err := s.db.Query(ctx, listReleaseExecutionItemsSQL, executionID)
|
|
if err != nil {
|
|
return governance.ReleaseExecutionDetail{}, fmt.Errorf("list release execution items: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
items := make([]governance.ReleaseItemExecutionView, 0)
|
|
for rows.Next() {
|
|
var item governance.ReleaseItemExecutionView
|
|
if err := rows.Scan(&item.PublicationID, &item.Title, &item.Action, &item.State,
|
|
&item.PreviousProductionRevision, &item.ProductionRevision, &item.ErrorCode,
|
|
&item.CreatedAt, &item.CompletedAt, &item.TargetKnowledgeBaseID); err != nil {
|
|
return governance.ReleaseExecutionDetail{}, fmt.Errorf("scan release execution item: %w", err)
|
|
}
|
|
if item.Action != "upsert" && item.Action != "delete" {
|
|
return governance.ReleaseExecutionDetail{}, errors.New("invalid release execution item action")
|
|
}
|
|
if !validReleaseItemState(item.State) {
|
|
return governance.ReleaseExecutionDetail{}, errors.New("invalid release item execution state")
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return governance.ReleaseExecutionDetail{}, fmt.Errorf("iterate release execution items: %w", err)
|
|
}
|
|
return governance.ReleaseExecutionDetail{ReleaseExecution: execution, Items: items}, nil
|
|
}
|
|
|
|
func scanReleaseExecution(row rowScanner) (governance.ReleaseExecution, error) {
|
|
var execution governance.ReleaseExecution
|
|
var operation, state string
|
|
if err := row.Scan(
|
|
&execution.ID, &execution.ReleaseBatchID, &operation, &execution.PriorExecutionID, &state,
|
|
&execution.TargetEnvironment, &execution.AuthorizedBy, &execution.AuthorizationCode,
|
|
&execution.JobID, &execution.ManifestSHA256, &execution.ItemCount,
|
|
&execution.BatchSize, &execution.CurrentBatch, &execution.TotalBatches,
|
|
&execution.CompletedItemCount, &execution.FailedItemCount, &execution.ErrorCode,
|
|
&execution.CreatedAt, &execution.StartedAt, &execution.CompletedAt,
|
|
&execution.FailureLevel, &execution.RestoreTrigger, &execution.ReleaseChangeSetID,
|
|
); err != nil {
|
|
return governance.ReleaseExecution{}, err
|
|
}
|
|
execution.Operation = governance.ReleaseOperation(operation)
|
|
execution.State = governance.ReleaseExecutionState(state)
|
|
if !execution.Operation.Valid() || !execution.State.Valid() || execution.TargetEnvironment != "production" {
|
|
return governance.ReleaseExecution{}, errors.New("invalid release execution row")
|
|
}
|
|
return execution, nil
|
|
}
|
|
|
|
func validReleaseExecutionFilter(filter governance.ReleaseExecutionFilter) bool {
|
|
operation := governance.ReleaseOperation(filter.Operation)
|
|
state := governance.ReleaseExecutionState(filter.State)
|
|
return (filter.Operation == "" || operation.Valid()) &&
|
|
(filter.State == "" || state.Valid()) &&
|
|
filter.Limit > 0 && filter.Limit <= 100 && filter.Offset >= 0 && filter.Offset <= 1000000
|
|
}
|
|
|
|
func validReleaseItemState(state string) bool {
|
|
switch state {
|
|
case "pending", "staged", "activated", "verified", "failed", "rolled_back":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|