Agent-System/internal/sysadmin/store.go

package sysadmin

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

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

// ErrSettingNotFound DB 中无该 key 的行（回退 env/default）。
var ErrSettingNotFound = errors.New("sysadmin: setting not found")

// ErrSettingRevisionConflict CAS 期望 revision 与当前不一致。
var ErrSettingRevisionConflict = errors.New("sysadmin: setting revision conflict")

// SettingsStore system_settings 表读写（表 owner 为 sysadmin）。
type SettingsStore interface {
	Get(ctx context.Context, key string) (value json.RawMessage, updatedBy *int64, updatedAt *time.Time, revision int64, err error)
	Upsert(ctx context.Context, key string, value json.RawMessage, updatedBy int64) (updatedAt time.Time, revision int64, err error)
	// UpsertCAS 在同一事务内校验 expectedRevision、写入新值并追加策略审计；冲突返回 ErrSettingRevisionConflict。
	UpsertCAS(ctx context.Context, key string, value json.RawMessage, updatedBy, expectedRevision int64, requestID string) (updatedAt time.Time, revision int64, err error)
	// GetForUpdate 在调用方事务内锁定目标行（FOR UPDATE），供激活路径与关闭串行化。
	GetForUpdate(ctx context.Context, tx pgx.Tx, key string) (value json.RawMessage, updatedBy *int64, updatedAt *time.Time, revision int64, err error)
	// ListPolicyAudits 按 setting_key 倒序列出策略审计（最新在前）。
	ListPolicyAudits(ctx context.Context, key string, limit int) ([]PolicyAudit, error)
}

// PolicyAudit system_setting_policy_audits 行视图。
type PolicyAudit struct {
	ID             int64           `json:"id"`
	SettingKey     string          `json:"settingKey"`
	BeforeValue    json.RawMessage `json:"beforeValue,omitempty"`
	AfterValue     json.RawMessage `json:"afterValue"`
	BeforeRevision *int64          `json:"beforeRevision,omitempty"`
	AfterRevision  int64           `json:"afterRevision"`
	ActorUserID    int64           `json:"actorUserId"`
	RequestID      string          `json:"requestId,omitempty"`
	CreatedAt      time.Time       `json:"createdAt"`
}

type pgSettingsStore struct {
	pool *pgxpool.Pool
}

func newPGSettingsStore(pool *pgxpool.Pool) SettingsStore {
	return &pgSettingsStore{pool: pool}
}

func (s *pgSettingsStore) Get(ctx context.Context, key string) (json.RawMessage, *int64, *time.Time, int64, error) {
	var raw []byte
	var updatedBy *int64
	var updatedAt time.Time
	var revision int64
	err := s.pool.QueryRow(ctx, `
		SELECT value, updated_by, updated_at, revision FROM system_settings WHERE key = $1
	`, key).Scan(&raw, &updatedBy, &updatedAt, &revision)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, nil, nil, 0, ErrSettingNotFound
		}
		return nil, nil, nil, 0, err
	}
	return json.RawMessage(raw), updatedBy, &updatedAt, revision, nil
}

func (s *pgSettingsStore) Upsert(ctx context.Context, key string, value json.RawMessage, updatedBy int64) (time.Time, int64, error) {
	var updatedAt time.Time
	var revision int64
	err := s.pool.QueryRow(ctx, `
		INSERT INTO system_settings (key, value, updated_by, updated_at, revision)
		VALUES ($1, $2, $3, now(), 0)
		ON CONFLICT (key) DO UPDATE
		SET value = EXCLUDED.value, updated_by = EXCLUDED.updated_by, updated_at = now()
		RETURNING updated_at, revision
	`, key, []byte(value), updatedBy).Scan(&updatedAt, &revision)
	return updatedAt, revision, err
}

func (s *pgSettingsStore) UpsertCAS(ctx context.Context, key string, value json.RawMessage, updatedBy, expectedRevision int64, requestID string) (time.Time, int64, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return time.Time{}, 0, err
	}
	defer tx.Rollback(ctx)

	var beforeRaw []byte
	var beforeRev int64
	err = tx.QueryRow(ctx, `
		SELECT value, revision FROM system_settings WHERE key = $1 FOR UPDATE
	`, key).Scan(&beforeRaw, &beforeRev)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return time.Time{}, 0, ErrSettingNotFound
		}
		return time.Time{}, 0, err
	}
	if beforeRev != expectedRevision {
		return time.Time{}, 0, ErrSettingRevisionConflict
	}

	var updatedAt time.Time
	var newRev int64
	err = tx.QueryRow(ctx, `
		UPDATE system_settings
		SET value = $2, updated_by = $3, updated_at = now(), revision = revision + 1
		WHERE key = $1 AND revision = $4
		RETURNING updated_at, revision
	`, key, []byte(value), updatedBy, expectedRevision).Scan(&updatedAt, &newRev)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return time.Time{}, 0, ErrSettingRevisionConflict
		}
		return time.Time{}, 0, err
	}

	var reqID any
	if requestID != "" {
		reqID = requestID
	}
	_, err = tx.Exec(ctx, `
		INSERT INTO system_setting_policy_audits (
			setting_key, before_value, after_value, before_revision, after_revision, actor_user_id, request_id
		) VALUES ($1, $2::jsonb, $3::jsonb, $4, $5, $6, $7)
	`, key, beforeRaw, []byte(value), beforeRev, newRev, updatedBy, reqID)
	if err != nil {
		return time.Time{}, 0, err
	}

	if err := tx.Commit(ctx); err != nil {
		return time.Time{}, 0, err
	}
	return updatedAt, newRev, nil
}

func (s *pgSettingsStore) GetForUpdate(ctx context.Context, tx pgx.Tx, key string) (json.RawMessage, *int64, *time.Time, int64, error) {
	if tx == nil {
		return nil, nil, nil, 0, errors.New("sysadmin: GetForUpdate requires transaction")
	}
	var raw []byte
	var updatedBy *int64
	var updatedAt time.Time
	var revision int64
	err := tx.QueryRow(ctx, `
		SELECT value, updated_by, updated_at, revision FROM system_settings WHERE key = $1 FOR UPDATE
	`, key).Scan(&raw, &updatedBy, &updatedAt, &revision)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, nil, nil, 0, ErrSettingNotFound
		}
		return nil, nil, nil, 0, err
	}
	return json.RawMessage(raw), updatedBy, &updatedAt, revision, nil
}

func (s *pgSettingsStore) ListPolicyAudits(ctx context.Context, key string, limit int) ([]PolicyAudit, error) {
	if key == "" {
		return nil, nil
	}
	if limit <= 0 {
		limit = 50
	}
	if limit > 200 {
		limit = 200
	}
	rows, err := s.pool.Query(ctx, `
		SELECT id, setting_key, before_value, after_value, before_revision, after_revision,
		       actor_user_id, COALESCE(request_id, ''), created_at
		FROM system_setting_policy_audits
		WHERE setting_key = $1
		ORDER BY created_at DESC, id DESC
		LIMIT $2
	`, key, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]PolicyAudit, 0)
	for rows.Next() {
		var a PolicyAudit
		var before []byte
		var after []byte
		var beforeRev *int64
		if err := rows.Scan(
			&a.ID, &a.SettingKey, &before, &after, &beforeRev, &a.AfterRevision,
			&a.ActorUserID, &a.RequestID, &a.CreatedAt,
		); err != nil {
			return nil, err
		}
		if before != nil {
			a.BeforeValue = json.RawMessage(before)
		}
		a.AfterValue = json.RawMessage(after)
		a.BeforeRevision = beforeRev
		out = append(out, a)
	}
	return out, rows.Err()
}