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

package postgres

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

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

var ErrMaintenanceReleased = errors.New("maintenance already released by this job")

// releasedByThisJob 判断"零行更新"是因为本作业已经释放过，还是根本没持有。
func (s *Store) releasedByThisJob(ctx context.Context, table, keyColumn, key, ownerJobID string) bool {
	var owner string
	var released *time.Time
	query := "SELECT owner_job_id, released_at FROM " + table + " WHERE " + keyColumn + "=$1"
	if err := s.db.QueryRow(ctx, query, key).Scan(&owner, &released); err != nil {
		return false
	}
	return owner == ownerJobID && released != nil
}

// LibraryMaintenance 是目标维护窗口的持有记录：一个 scope 只能有一个持有者。
//
// Level 说明这次维护到底动了什么：
//   - "instance"：入口与写入者被真的关掉（实例级），所有库都受影响；
//   - "library"：只锁了治理库的一个 scope，目标入口/后台写入没被挡，不能当成系统维护。
type LibraryMaintenance struct {
	Scope                string     `json:"scope"`
	OwnerJobID           string     `json:"owner_job_id"`
	TargetIdentity       string     `json:"target_identity"`
	Level                string     `json:"level"`
	TargetFingerprint    string     `json:"target_fingerprint,omitempty"`
	Services             []string   `json:"services,omitempty"`
	IngressState         string     `json:"ingress_state"`
	State                string     `json:"state"`
	OriginalServiceState string     `json:"original_service_state"`
	Note                 string     `json:"note"`
	AcquiredAt           time.Time  `json:"acquired_at"`
	UpdatedAt            time.Time  `json:"updated_at"`
	ReleasedAt           *time.Time `json:"released_at,omitempty"`
}

// ErrMaintenanceHeld 表示目标维护窗口被别的作业占着：任何新的写入都必须拒绝。
var ErrMaintenanceHeld = errors.New("目标正在维护中，被其他作业占用")

const maintenanceColumns = `scope, owner_job_id, target_identity, level, target_fingerprint, services,
 ingress_state, state, original_service_state, note, acquired_at, updated_at, released_at`

// AcquireLibraryMaintenance 原子地拿到维护权：
// 同一作业可以重入（接管后继续），别的作业一律拒绝——这不是"尽力而为"的锁。
func (s *Store) AcquireLibraryMaintenance(ctx context.Context, item LibraryMaintenance, now time.Time) (LibraryMaintenance, error) {
	level := item.Level
	if level == "" {
		level = "library"
	}
	ingress := item.IngressState
	if ingress == "" {
		ingress = "unknown"
	}
	services := item.Services
	if services == nil {
		services = []string{}
	}
	raw, err := json.Marshal(services)
	if err != nil {
		return LibraryMaintenance{}, fmt.Errorf("acquire library maintenance: %w", err)
	}
	row := s.db.QueryRow(ctx, `
INSERT INTO library_maintenance (scope, owner_job_id, target_identity, level, target_fingerprint, services,
    ingress_state, state, original_service_state, note, acquired_at, updated_at)
VALUES ($1,$2,$3,$4,$5,$6::jsonb,$7,$8,$9,$10,$11,$11)
ON CONFLICT (scope) DO UPDATE SET
    owner_job_id = EXCLUDED.owner_job_id,
    target_identity = EXCLUDED.target_identity,
    level = EXCLUDED.level,
    target_fingerprint = EXCLUDED.target_fingerprint,
    services = EXCLUDED.services,
    ingress_state = EXCLUDED.ingress_state,
    state = EXCLUDED.state,
    -- 同上：原运行状态只由第一次进入窗口时决定，重入不能覆盖。
    original_service_state = CASE
        WHEN library_maintenance.owner_job_id = EXCLUDED.owner_job_id
         AND library_maintenance.original_service_state <> ''
        THEN library_maintenance.original_service_state
        ELSE EXCLUDED.original_service_state
    END,
    note = EXCLUDED.note,
    updated_at = EXCLUDED.updated_at,
    released_at = NULL
-- 已经释放的窗口允许被新作业接管（否则第一个同步之后谁也再进不去）；
-- 仍在持有的窗口只允许原作业重入。
WHERE library_maintenance.released_at IS NOT NULL
   OR library_maintenance.owner_job_id = EXCLUDED.owner_job_id
RETURNING `+maintenanceColumns,
		item.Scope, item.OwnerJobID, item.TargetIdentity, level, item.TargetFingerprint, string(raw),
		ingress, item.State, item.OriginalServiceState, item.Note, now.UTC())
	acquired, err := scanMaintenance(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return LibraryMaintenance{}, ErrMaintenanceHeld
	}
	if err != nil {
		return LibraryMaintenance{}, fmt.Errorf("acquire library maintenance: %w", err)
	}
	return acquired, nil
}

// UpdateLibraryMaintenanceState 只允许持有者改自己的窗口状态。
func (s *Store) UpdateLibraryMaintenanceState(ctx context.Context, scope, ownerJobID, state, note string, now time.Time) error {
	tag, err := s.db.Exec(ctx, `UPDATE library_maintenance SET state=$3, note=$4, updated_at=$5
WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL`, scope, ownerJobID, state, note, now.UTC())
	if err != nil {
		return fmt.Errorf("update library maintenance: %w", err)
	}
	if tag.RowsAffected() == 0 {
		if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) {
			return ErrMaintenanceReleased
		}
		return errors.New("library maintenance not held by this job")
	}
	return nil
}

// SetLibraryMaintenanceIngress 记录入口的真实状态（closed/open），
// 页面据此区分"入口关了"和"只是容器停了"。
func (s *Store) SetLibraryMaintenanceIngress(ctx context.Context, scope, ownerJobID, state, note string, now time.Time) error {
	if state != "closed" && state != "open" {
		return errors.New("invalid ingress state")
	}
	tag, err := s.db.Exec(ctx, `UPDATE library_maintenance SET ingress_state=$3, note=$4, updated_at=$5
WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL`, scope, ownerJobID, state, note, now.UTC())
	if err != nil {
		return fmt.Errorf("update library maintenance ingress: %w", err)
	}
	if tag.RowsAffected() == 0 {
		if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) {
			return ErrMaintenanceReleased
		}
		return errors.New("library maintenance not held by this job")
	}
	return nil
}

// ReleaseLibraryMaintenance 只在完整验证通过后调用；失败路径必须保持窗口。
func (s *Store) ReleaseLibraryMaintenance(ctx context.Context, scope, ownerJobID string, now time.Time) error {
	tag, err := s.db.Exec(ctx, `UPDATE library_maintenance SET state='released', released_at=$3, updated_at=$3
WHERE scope=$1 AND owner_job_id=$2 AND released_at IS NULL`, scope, ownerJobID, now.UTC())
	if err != nil {
		return fmt.Errorf("release library maintenance: %w", err)
	}
	if tag.RowsAffected() == 0 {
		if s.releasedByThisJob(ctx, "library_maintenance", "scope", scope, ownerJobID) {
			return ErrMaintenanceReleased
		}
		return errors.New("library maintenance not held by this job")
	}
	return nil
}

// LibraryMaintenanceOf 读当前窗口状态（页面与诊断用）。
func (s *Store) LibraryMaintenanceOf(ctx context.Context, scope string) (*LibraryMaintenance, error) {
	row := s.db.QueryRow(ctx, `SELECT `+maintenanceColumns+` FROM library_maintenance WHERE scope=$1`, scope)
	item, err := scanMaintenance(row)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &item, nil
}

func scanMaintenance(row pgx.Row) (LibraryMaintenance, error) {
	var item LibraryMaintenance
	var raw []byte
	err := row.Scan(&item.Scope, &item.OwnerJobID, &item.TargetIdentity, &item.Level, &item.TargetFingerprint,
		&raw, &item.IngressState, &item.State, &item.OriginalServiceState, &item.Note,
		&item.AcquiredAt, &item.UpdatedAt, &item.ReleasedAt)
	if err != nil {
		return LibraryMaintenance{}, err
	}
	if len(raw) > 0 {
		if err := json.Unmarshal(raw, &item.Services); err != nil {
			return LibraryMaintenance{}, fmt.Errorf("maintenance services unreadable: %w", err)
		}
	}
	return item, nil
}