weichen
86e9d3706b
Persist agent/secrets settings with masked secrets and runner hot-reload; add maintenance windows, maintpage binary, and zero-downtime listener handoff.
235 lines
7.9 KiB
Go
235 lines
7.9 KiB
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)
|
||
// UpsertWithAudit 写入明文 value，审计表写入 auditAfter（secret 键应为指纹 JSON）。
|
||
UpsertWithAudit(ctx context.Context, key string, value, auditAfter json.RawMessage, updatedBy int64, requestID string) (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) {
|
||
return s.UpsertWithAudit(ctx, key, value, value, updatedBy, "")
|
||
}
|
||
|
||
func (s *pgSettingsStore) UpsertWithAudit(ctx context.Context, key string, value, auditAfter json.RawMessage, updatedBy 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
|
||
_ = tx.QueryRow(ctx, `SELECT value FROM system_settings WHERE key = $1`, key).Scan(&beforeRaw)
|
||
|
||
var updatedAt time.Time
|
||
var revision int64
|
||
err = tx.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(),
|
||
revision = system_settings.revision + 1
|
||
RETURNING updated_at, revision
|
||
`, key, []byte(value), updatedBy).Scan(&updatedAt, &revision)
|
||
if err != nil {
|
||
return time.Time{}, 0, err
|
||
}
|
||
|
||
var reqID any
|
||
if requestID != "" {
|
||
reqID = requestID
|
||
}
|
||
_, err = tx.Exec(ctx, `
|
||
INSERT INTO system_settings_audits (setting_key, before_value, after_value, actor_user_id, request_id)
|
||
VALUES ($1, $2::jsonb, $3::jsonb, $4, $5)
|
||
`, key, beforeRaw, []byte(auditAfter), updatedBy, reqID)
|
||
if err != nil {
|
||
return time.Time{}, 0, err
|
||
}
|
||
if err := tx.Commit(ctx); err != nil {
|
||
return time.Time{}, 0, err
|
||
}
|
||
return updatedAt, revision, nil
|
||
}
|
||
|
||
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()
|
||
}
|