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

package postgres

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"

	"knowledge-governance/internal/governance"
)

// 整库作业与版本：队列（jobs）负责租约重试，本文件负责业务身份、阶段与结果。
// 互斥与幂等都由数据库唯一索引保证，不依赖前端或单 worker 部署。

func (s *Store) CreateLibraryJob(ctx context.Context, job governance.LibraryJob, stages []string, now time.Time) (governance.LibraryJob, bool, error) {
	request := job.RequestDigest
	if request == "" {
		return governance.LibraryJob{}, false, errors.New("library job requires a request digest")
	}
	rawRequest, err := json.Marshal(map[string]any{"digest": request, "kind": string(job.Kind), "scope": job.Scope})
	if err != nil {
		return governance.LibraryJob{}, false, err
	}
	var id string
	err = s.db.QueryRow(ctx, `
INSERT INTO library_jobs (id, scope, kind, state, stage, idempotency_key, request_digest, request,
    queue_job_id, snapshot_id, target_snapshot_id, actor, error, recovery, created_at, updated_at)
VALUES ($1,$2,$3,'pending','',$4,$5,$6,$7,$8,$9,$10,'','',$11,$11)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id`,
		job.ID, job.Scope, string(job.Kind), job.IdempotencyKey, request, rawRequest,
		job.QueueJobID, job.SnapshotID, job.TargetSnapshotID, job.Actor, now.UTC()).Scan(&id)
	if err != nil {
		var pgErr *pgconn.PgError
		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
			// 同一目标的写作业（sync/restore）共用一个唯一索引，跨 kind 互斥。
			if pgErr.ConstraintName == "library_jobs_write_target" || pgErr.ConstraintName == "library_jobs_active" {
				return governance.LibraryJob{}, false, governance.ErrLibraryJobActive
			}
		}
		if !errors.Is(err, pgx.ErrNoRows) {
			return governance.LibraryJob{}, false, fmt.Errorf("create library job: %w", err)
		}
		// 幂等键已存在：必须返回**已存**作业，并核对请求体是否一致。
		existing, loadErr := s.libraryJobByKey(ctx, job.IdempotencyKey)
		if loadErr != nil {
			return governance.LibraryJob{}, false, loadErr
		}
		if existing.RequestDigest != request {
			return existing, false, governance.ErrLibraryJobIdempotencyConflict
		}
		return existing, true, nil
	}
	for _, stage := range stages {
		if _, err := s.db.Exec(ctx, `
INSERT INTO library_job_stages (job_id, stage, state, detail, updated_at)
VALUES ($1,$2,'pending','',$3)
ON CONFLICT (job_id, stage) DO NOTHING`, id, stage, now.UTC()); err != nil {
			return governance.LibraryJob{}, false, fmt.Errorf("create library job stage: %w", err)
		}
	}
	created, err := s.LibraryJob(ctx, id)
	return created, false, err
}

func (s *Store) libraryJobByKey(ctx context.Context, key string) (governance.LibraryJob, error) {
	var id string
	if err := s.db.QueryRow(ctx, `SELECT id FROM library_jobs WHERE idempotency_key=$1`, key).Scan(&id); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return governance.LibraryJob{}, errors.New("library job idempotency key not found")
		}
		return governance.LibraryJob{}, fmt.Errorf("load library job by key: %w", err)
	}
	return s.LibraryJob(ctx, id)
}

func (s *Store) LibraryJob(ctx context.Context, id string) (governance.LibraryJob, error) {
	var job governance.LibraryJob
	var kind, state string
	var finished *time.Time
	if err := s.db.QueryRow(ctx, `
SELECT id, scope, kind, state, stage, idempotency_key, request_digest, queue_job_id, snapshot_id,
       target_snapshot_id, actor, error, recovery, created_at, updated_at, finished_at
FROM library_jobs WHERE id=$1`, id).Scan(&job.ID, &job.Scope, &kind, &state, &job.Stage, &job.IdempotencyKey,
		&job.RequestDigest, &job.QueueJobID, &job.SnapshotID, &job.TargetSnapshotID, &job.Actor, &job.Error,
		&job.Recovery, &job.CreatedAt, &job.UpdatedAt, &finished); err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return governance.LibraryJob{}, err
		}
		return governance.LibraryJob{}, fmt.Errorf("load library job: %w", err)
	}
	job.Kind, job.State, job.FinishedAt = governance.LibraryJobKind(kind), governance.LibraryJobState(state), finished
	stages, err := s.libraryStages(ctx, id)
	if err != nil {
		return governance.LibraryJob{}, err
	}
	job.Stages = stages
	return job, nil
}

func (s *Store) libraryStages(ctx context.Context, id string) ([]governance.LibraryStage, error) {
	rows, err := s.db.Query(ctx, `
SELECT stage, state, total, completed, bytes_total, bytes_done, detail, updated_at
FROM library_job_stages WHERE job_id=$1 ORDER BY updated_at, stage`, id)
	if err != nil {
		return nil, fmt.Errorf("load library job stages: %w", err)
	}
	defer rows.Close()
	result := make([]governance.LibraryStage, 0, 7)
	for rows.Next() {
		var stage governance.LibraryStage
		if err := rows.Scan(&stage.Stage, &stage.State, &stage.Total, &stage.Completed, &stage.BytesTotal,
			&stage.BytesDone, &stage.Detail, &stage.UpdatedAt); err != nil {
			return nil, fmt.Errorf("scan library job stage: %w", err)
		}
		result = append(result, stage)
	}
	return result, rows.Err()
}

// ListLibraryJobs 返回某目标最近的作业（按创建时间倒序），供页面在重启后恢复展示。
func (s *Store) ListLibraryJobs(ctx context.Context, scope string, limit int) ([]governance.LibraryJob, error) {
	if limit < 1 || limit > 50 {
		return nil, errors.New("library job list limit out of range")
	}
	rows, err := s.db.Query(ctx, `SELECT id FROM library_jobs WHERE scope=$1 ORDER BY created_at DESC, id DESC LIMIT $2`, scope, limit)
	if err != nil {
		return nil, fmt.Errorf("list library jobs: %w", err)
	}
	defer rows.Close()
	ids := make([]string, 0, limit)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("scan library job id: %w", err)
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	result := make([]governance.LibraryJob, 0, len(ids))
	for _, id := range ids {
		job, err := s.LibraryJob(ctx, id)
		if err != nil {
			return nil, err
		}
		result = append(result, job)
	}
	return result, nil
}

// UpdateLibraryStage 只改调用方给了值的维度：nil 表示"这一维还没有可靠数据"，
// 保持 NULL，绝不写成 0。
func (s *Store) UpdateLibraryStage(ctx context.Context, jobID, stage, state string, total, completed, bytesTotal, bytesDone *int64, detail string, now time.Time) error {
	tag, err := s.db.Exec(ctx, `
UPDATE library_job_stages SET
    state = CASE WHEN $3='' THEN state ELSE $3 END,
    total = COALESCE($4, total),
    completed = COALESCE($5, completed),
    bytes_total = COALESCE($6, bytes_total),
    bytes_done = COALESCE($7, bytes_done),
    detail = CASE WHEN $8='' THEN detail ELSE $8 END,
    updated_at = $9
WHERE job_id=$1 AND stage=$2`, jobID, stage, state, total, completed, bytesTotal, bytesDone, detail, now.UTC())
	if err != nil {
		return fmt.Errorf("update library stage: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errors.New("library stage not found")
	}
	return nil
}

// AttachLibraryQueueJob 把队列作业号写到已建的业务作业上（两步创建，避免先入队后才发现互斥冲突）。
func (s *Store) AttachLibraryQueueJob(ctx context.Context, jobID, queueJobID string, now time.Time) error {
	tag, err := s.db.Exec(ctx, `UPDATE library_jobs SET queue_job_id=$2, updated_at=$3 WHERE id=$1`, jobID, queueJobID, now.UTC())
	if err != nil {
		return fmt.Errorf("attach library queue job: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errors.New("library job not found")
	}
	return nil
}

func (s *Store) CancelLibraryJob(ctx context.Context, jobID, reason string, now time.Time) error {
	return s.AdvanceLibraryJob(ctx, jobID, governance.LibraryJobCancelled, "", reason, "", "", "", now)
}

func (s *Store) AdvanceLibraryJob(ctx context.Context, jobID string, state governance.LibraryJobState, stage, message, recovery, snapshotID, targetSnapshotID string, now time.Time) error {
	finished := any(nil)
	if state == governance.LibraryJobSucceeded || state == governance.LibraryJobFailed || state == governance.LibraryJobCancelled {
		finished = now.UTC()
	}
	tag, err := s.db.Exec(ctx, `
UPDATE library_jobs SET
    state=$2,
    stage = CASE WHEN $3='' THEN stage ELSE $3 END,
    error = CASE WHEN $4='' THEN error ELSE $4 END,
    recovery = CASE WHEN $5='' THEN recovery ELSE $5 END,
    snapshot_id = CASE WHEN $6='' THEN snapshot_id ELSE $6 END,
    target_snapshot_id = CASE WHEN $7='' THEN target_snapshot_id ELSE $7 END,
    updated_at=$8,
    finished_at=COALESCE($9, finished_at)
WHERE id=$1`, jobID, string(state), stage, message, recovery, snapshotID, targetSnapshotID, now.UTC(), finished)
	if err != nil {
		return fmt.Errorf("advance library job: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errors.New("library job not found")
	}
	return nil
}

// RecordLibraryVersion 写入一个版本记录。生产生效版本必须先把旧的 supersede 掉，
// 否则唯一索引会拒绝——这正是"不许两个生效版本并存"的保证。
func (s *Store) RecordLibraryVersion(ctx context.Context, version governance.LibraryVersion) error {
	evidence, err := json.Marshal(version.Evidence)
	if err != nil {
		return fmt.Errorf("encode library version evidence: %w", err)
	}
	tx, err := s.db.Begin(ctx)
	if err != nil {
		return fmt.Errorf("begin library version: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if version.Kind == governance.LibraryVersionProductionEffective {
		if _, err := tx.Exec(ctx, `
UPDATE library_versions SET superseded_at=$2
WHERE scope=$1 AND kind='production_effective' AND superseded_at IS NULL`, version.Scope, version.CreatedAt.UTC()); err != nil {
			return fmt.Errorf("supersede library version: %w", err)
		}
	}
	if _, err := tx.Exec(ctx, `
INSERT INTO library_versions (id, scope, kind, snapshot_id, version_label, source_job_id, evidence, created_by, created_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
ON CONFLICT (id) DO UPDATE SET
    -- 恢复/重放同一个快照时必须把它重新"激活"：否则它还是被 supersede 的旧行，
    -- 页面会显示"生产当前版本未知"，而实际生产已经恢复到这个快照。
    superseded_at = NULL,
    source_job_id = EXCLUDED.source_job_id,
    evidence = EXCLUDED.evidence,
    created_at = EXCLUDED.created_at`,
		version.ID, version.Scope, string(version.Kind), version.SnapshotID, version.VersionLabel,
		version.SourceJobID, evidence, version.CreatedBy, version.CreatedAt.UTC()); err != nil {
		return fmt.Errorf("record library version: %w", err)
	}
	return tx.Commit(ctx)
}

// InterruptedLibraryJobs 找出"业务作业 running、但队列里已没有活跃作业"的作业：
// worker 被杀/重启后它们需要被接管，而不是被当成失败丢掉已完成的预备份。
func (s *Store) InterruptedLibraryJobs(ctx context.Context) ([]governance.LibraryJob, error) {
	// lease 过期才算中断：worker 刚被杀但 lease 未到期时，队列里还是 running，
	// 直接判"没中断"会让作业永远悬着；所以这里按 lease_until 判断。
	rows, err := s.db.Query(ctx, `
SELECT lj.id FROM library_jobs lj
LEFT JOIN jobs j ON j.id=lj.queue_job_id
WHERE lj.kind IN ('sync','restore','baseline')
  AND (
    -- 业务作业在跑、队列已经没了/停了/租约过期
    (lj.state='running' AND (j.id IS NULL OR j.state IN ('failed','completed','cancelled','dead')
        OR (j.state IN ('pending','running','retry') AND j.lease_until IS NOT NULL AND j.lease_until < now())))
    -- 业务作业已被前一轮接管置为 pending，但队列作业的租约也没了（worker 又死了）
    OR (lj.state='pending' AND j.state='running' AND j.lease_until IS NOT NULL AND j.lease_until < now())
  )
ORDER BY lj.created_at`)
	if err != nil {
		return nil, fmt.Errorf("list interrupted library jobs: %w", err)
	}
	defer rows.Close()
	ids := make([]string, 0, 4)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		ids = append(ids, id)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	jobs := make([]governance.LibraryJob, 0, len(ids))
	for _, id := range ids {
		job, err := s.LibraryJob(ctx, id)
		if err != nil {
			return nil, err
		}
		jobs = append(jobs, job)
	}
	return jobs, nil
}

// ResetQueueJobForResume 把中断/失败过的队列作业重新排队：attempt 必须归零，
// 否则 attempt<max_attempts 永远不成立，作业再也不会被任何 worker 领取。
func (s *Store) ResetQueueJobForResume(ctx context.Context, queueJobID string, now time.Time) error {
	if _, err := s.db.Exec(ctx, `UPDATE jobs
SET state='pending', attempt=0, lease_owner=NULL, lease_until=NULL, available_at=$2, updated_at=$2
WHERE id=$1 AND state IN ('running','retry','failed','pending')`, queueJobID, now); err != nil {
		return fmt.Errorf("reset queue job for resume: %w", err)
	}
	return nil
}

// ClaimLibraryJobForResume 用一次条件更新把中断作业抢过来：只有把 running 改成 pending
// 的那一个 worker 拿到 true，多个 worker 同时重启也只会有一个执行（fencing）。
func (s *Store) ClaimLibraryJobForResume(ctx context.Context, jobID string, now time.Time) (bool, error) {
	tag, err := s.db.Exec(ctx, `UPDATE library_jobs SET state='pending', updated_at=$2
WHERE id=$1 AND state IN ('running','pending')`, jobID, now)
	if err != nil {
		return false, fmt.Errorf("claim interrupted library job: %w", err)
	}
	return tag.RowsAffected() == 1, nil
}

// ReconcileLibraryJobs 收拾"业务作业还在飞、队列里已经没有对应作业"的孤儿：
// 它们会永久占用 (scope,kind) 唯一索引，让同一目标再也发不出作业。
// 只标失败并写明恢复责任，不猜结果、不重放。
func (s *Store) ReconcileLibraryJobs(ctx context.Context, now time.Time) (int, error) {
	if _, err := s.db.Exec(ctx, `
UPDATE library_jobs AS lj
SET state='failed', error='队列中已无对应作业，业务作业未收到结果', recovery='封存作业由 agent 恢复来源；确认来源 healthy 后可重发', updated_at=$1, finished_at=$1
WHERE lj.state IN ('pending','running')
  AND lj.queue_job_id <> ''
  AND NOT EXISTS (
    SELECT 1 FROM jobs AS j
    WHERE j.id = lj.queue_job_id AND j.state IN ('pending','running','retry')
  )`, now.UTC()); err != nil {
		return 0, fmt.Errorf("reconcile library jobs: %w", err)
	}
	// 建了业务作业但队列投递没完成（queue_job_id 为空）超过 5 分钟，也是孤儿。
	tag, err := s.db.Exec(ctx, `
UPDATE library_jobs
SET state='failed', error='队列投递未完成，作业从未开始', recovery='无需恢复来源；可直接重发', updated_at=$1, finished_at=$1
WHERE state IN ('pending','running') AND queue_job_id='' AND created_at < $2`, now.UTC(), now.Add(-5*time.Minute).UTC())
	if err != nil {
		return 0, fmt.Errorf("reconcile unattached library jobs: %w", err)
	}
	return int(tag.RowsAffected()), nil
}

func (s *Store) LibraryVersions(ctx context.Context, scope string, limit int) ([]governance.LibraryVersion, error) {
	if limit < 1 || limit > 100 {
		return nil, errors.New("library version limit out of range")
	}
	rows, err := s.db.Query(ctx, `
SELECT id, scope, kind, snapshot_id, version_label, source_job_id, evidence, created_by, created_at, superseded_at
FROM library_versions WHERE scope=$1 ORDER BY created_at DESC, id DESC LIMIT $2`, scope, limit)
	if err != nil {
		return nil, fmt.Errorf("list library versions: %w", err)
	}
	defer rows.Close()
	result := make([]governance.LibraryVersion, 0, limit)
	for rows.Next() {
		var item governance.LibraryVersion
		var kind string
		var evidence []byte
		if err := rows.Scan(&item.ID, &item.Scope, &kind, &item.SnapshotID, &item.VersionLabel, &item.SourceJobID,
			&evidence, &item.CreatedBy, &item.CreatedAt, &item.SupersededAt); err != nil {
			return nil, fmt.Errorf("scan library version: %w", err)
		}
		item.Kind = governance.LibraryVersionKind(kind)
		if len(evidence) > 0 {
			_ = json.Unmarshal(evidence, &item.Evidence)
		}
		result = append(result, item)
	}
	return result, rows.Err()
}

// CreateLibraryAuthorization 记录一次生产发布授权；同一 (scope,snapshot,digest) 只保留最新有效。
func (s *Store) CreateLibraryAuthorization(ctx context.Context, item governance.LibraryReleaseAuthorization) error {
	if item.ID == "" || item.Scope == "" || item.SnapshotID == "" || item.ParamsDigest == "" || item.ExpiresAt.IsZero() {
		return errors.New("invalid library release authorization")
	}
	_, err := s.db.Exec(ctx, `
INSERT INTO library_release_authorizations (id, scope, snapshot_id, params_digest, authorized_by, expires_at, created_at)
VALUES ($1,$2,$3,$4,$5,$6,$7)`,
		item.ID, item.Scope, item.SnapshotID, item.ParamsDigest, item.AuthorizedBy, item.ExpiresAt.UTC(), item.CreatedAt.UTC())
	if err != nil {
		return fmt.Errorf("create library release authorization: %w", err)
	}
	return nil
}

// ActiveLibraryAuthorization 只返回"未撤销、未过期、且参数摘要完全一致"的授权。
// 参数一变就查不到，自然拒绝，而不是靠调用方记得复核。
func (s *Store) ActiveLibraryAuthorization(ctx context.Context, scope, snapshotID, paramsDigest string, now time.Time) (governance.LibraryReleaseAuthorization, bool, error) {
	var item governance.LibraryReleaseAuthorization
	err := s.db.QueryRow(ctx, `
SELECT id, scope, snapshot_id, params_digest, authorized_by, expires_at, revoked_at, created_at
FROM library_release_authorizations
WHERE scope=$1 AND snapshot_id=$2 AND params_digest=$3 AND revoked_at IS NULL AND expires_at > $4
ORDER BY expires_at DESC, created_at DESC
LIMIT 1`, scope, snapshotID, paramsDigest, now.UTC()).Scan(&item.ID, &item.Scope, &item.SnapshotID,
		&item.ParamsDigest, &item.AuthorizedBy, &item.ExpiresAt, &item.RevokedAt, &item.CreatedAt)
	if errors.Is(err, pgx.ErrNoRows) {
		return governance.LibraryReleaseAuthorization{}, false, nil
	}
	if err != nil {
		return governance.LibraryReleaseAuthorization{}, false, fmt.Errorf("load library authorization: %w", err)
	}
	return item, true, nil
}

func (s *Store) RevokeLibraryAuthorization(ctx context.Context, id string, now time.Time) error {
	tag, err := s.db.Exec(ctx, `UPDATE library_release_authorizations SET revoked_at=$2 WHERE id=$1 AND revoked_at IS NULL`, id, now.UTC())
	if err != nil {
		return fmt.Errorf("revoke library authorization: %w", err)
	}
	if tag.RowsAffected() != 1 {
		return errors.New("library authorization not found or already revoked")
	}
	return nil
}

func (s *Store) LibraryAuthorizations(ctx context.Context, scope string, limit int) ([]governance.LibraryReleaseAuthorization, error) {
	if limit < 1 || limit > 100 {
		return nil, errors.New("library authorization limit out of range")
	}
	rows, err := s.db.Query(ctx, `
SELECT id, scope, snapshot_id, params_digest, authorized_by, expires_at, revoked_at, created_at
FROM library_release_authorizations WHERE scope=$1 ORDER BY created_at DESC, id DESC LIMIT $2`, scope, limit)
	if err != nil {
		return nil, fmt.Errorf("list library authorizations: %w", err)
	}
	defer rows.Close()
	result := make([]governance.LibraryReleaseAuthorization, 0, limit)
	for rows.Next() {
		var item governance.LibraryReleaseAuthorization
		if err := rows.Scan(&item.ID, &item.Scope, &item.SnapshotID, &item.ParamsDigest, &item.AuthorizedBy,
			&item.ExpiresAt, &item.RevokedAt, &item.CreatedAt); err != nil {
			return nil, fmt.Errorf("scan library authorization: %w", err)
		}
		result = append(result, item)
	}
	return result, rows.Err()
}