feat(store): add MaintenancePolicy CRUD backend
Типы maintenance_policy, методы store.Backend и реализации для PostgreSQL и in-memory. Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Denozordec
@@ -0,0 +1,304 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"evobgp/internal/store"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
const maintenancePolicySelect = `
|
||||
SELECT id, name, table_name, condition_sql, retention_period_sec, max_rows,
|
||||
vacuum_strategy, schedule_cron, enabled, dry_run_enabled,
|
||||
last_run_at, COALESCE(last_status, ''), COALESCE(last_error, ''),
|
||||
created_at, updated_at
|
||||
FROM maintenance_policy`
|
||||
|
||||
func scanMaintenancePolicy(row pgx.Row) (*store.MaintenancePolicy, error) {
|
||||
var p store.MaintenancePolicy
|
||||
var retention, maxRows *int32
|
||||
var lastRun *time.Time
|
||||
err := row.Scan(
|
||||
&p.ID, &p.Name, &p.TableName, &p.Condition, &retention, &maxRows,
|
||||
&p.VacuumStrategy, &p.Schedule, &p.Enabled, &p.DryRunEnabled,
|
||||
&lastRun, &p.LastStatus, &p.LastError, &p.CreatedAt, &p.UpdatedAt,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if retention != nil {
|
||||
v := int(*retention)
|
||||
p.RetentionPeriodSec = &v
|
||||
}
|
||||
if maxRows != nil {
|
||||
v := int(*maxRows)
|
||||
p.MaxRows = &v
|
||||
}
|
||||
if lastRun != nil {
|
||||
t := lastRun.UTC()
|
||||
p.LastRunAt = &t
|
||||
}
|
||||
return &p, nil
|
||||
}
|
||||
|
||||
func (p *Postgres) ListMaintenancePolicies(cursor string, limit int) ([]*store.MaintenancePolicy, string, bool, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
off := 0
|
||||
if cursor != "" {
|
||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
||||
off = n
|
||||
}
|
||||
}
|
||||
ctx := context.Background()
|
||||
rows, err := p.pool.Query(ctx, maintenancePolicySelect+`
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT $1 OFFSET $2`, limit+1, off)
|
||||
if err != nil {
|
||||
return nil, "", false, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*store.MaintenancePolicy
|
||||
for rows.Next() {
|
||||
pol, err := scanMaintenancePolicy(rows)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
out = append(out, pol)
|
||||
}
|
||||
more := len(out) > limit
|
||||
if more {
|
||||
out = out[:limit]
|
||||
}
|
||||
next := ""
|
||||
if more {
|
||||
next = strconv.Itoa(off + limit)
|
||||
}
|
||||
return out, next, more, rows.Err()
|
||||
}
|
||||
|
||||
func (p *Postgres) GetMaintenancePolicy(id string) (*store.MaintenancePolicy, error) {
|
||||
ctx := context.Background()
|
||||
row := p.pool.QueryRow(ctx, maintenancePolicySelect+` WHERE id=$1`, id)
|
||||
pol, err := scanMaintenancePolicy(row)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, store.ErrNotFound
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
return pol, nil
|
||||
}
|
||||
|
||||
func (p *Postgres) CreateMaintenancePolicy(in *store.MaintenancePolicy) (*store.MaintenancePolicy, error) {
|
||||
if in == nil {
|
||||
return nil, store.ErrInvalidInput
|
||||
}
|
||||
vacuum := in.VacuumStrategy
|
||||
if vacuum == "" {
|
||||
vacuum = store.VacuumStrategyNone
|
||||
}
|
||||
if err := store.ValidateMaintenancePolicyInput(in.Name, in.TableName, vacuum, in.Schedule); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx := context.Background()
|
||||
id := uuid.NewString()
|
||||
now := time.Now().UTC()
|
||||
condition := store.NormalizeMaintenancePolicyCondition(in.Condition)
|
||||
var retention, maxRows *int32
|
||||
if in.RetentionPeriodSec != nil {
|
||||
v := int32(*in.RetentionPeriodSec)
|
||||
retention = &v
|
||||
}
|
||||
if in.MaxRows != nil {
|
||||
v := int32(*in.MaxRows)
|
||||
maxRows = &v
|
||||
}
|
||||
_, err := p.pool.Exec(ctx, `
|
||||
INSERT INTO maintenance_policy (
|
||||
id, name, table_name, condition_sql, retention_period_sec, max_rows,
|
||||
vacuum_strategy, schedule_cron, enabled, dry_run_enabled, created_at, updated_at
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$11)`,
|
||||
id, strings.TrimSpace(in.Name), strings.TrimSpace(in.TableName), condition,
|
||||
retention, maxRows, vacuum, strings.TrimSpace(in.Schedule),
|
||||
in.Enabled, in.DryRunEnabled, now)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p.GetMaintenancePolicy(id)
|
||||
}
|
||||
|
||||
func (p *Postgres) UpdateMaintenancePolicy(id string, patch *store.MaintenancePolicyPatch) (*store.MaintenancePolicy, error) {
|
||||
if patch == nil {
|
||||
return nil, store.ErrInvalidInput
|
||||
}
|
||||
cur, err := p.GetMaintenancePolicy(id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if patch.Name != nil {
|
||||
cur.Name = strings.TrimSpace(*patch.Name)
|
||||
}
|
||||
if patch.TableName != nil {
|
||||
cur.TableName = strings.TrimSpace(*patch.TableName)
|
||||
}
|
||||
if patch.Condition != nil {
|
||||
cur.Condition = store.NormalizeMaintenancePolicyCondition(*patch.Condition)
|
||||
}
|
||||
if patch.RetentionPeriodSec != nil {
|
||||
cur.RetentionPeriodSec = patch.RetentionPeriodSec
|
||||
}
|
||||
if patch.MaxRows != nil {
|
||||
cur.MaxRows = patch.MaxRows
|
||||
}
|
||||
if patch.VacuumStrategy != nil {
|
||||
if !store.ValidVacuumStrategy(*patch.VacuumStrategy) {
|
||||
return nil, store.ErrInvalidInput
|
||||
}
|
||||
cur.VacuumStrategy = strings.TrimSpace(*patch.VacuumStrategy)
|
||||
}
|
||||
if patch.Schedule != nil {
|
||||
cur.Schedule = strings.TrimSpace(*patch.Schedule)
|
||||
}
|
||||
if patch.Enabled != nil {
|
||||
cur.Enabled = *patch.Enabled
|
||||
}
|
||||
if patch.DryRunEnabled != nil {
|
||||
cur.DryRunEnabled = *patch.DryRunEnabled
|
||||
}
|
||||
if err := store.ValidateMaintenancePolicyInput(cur.Name, cur.TableName, cur.VacuumStrategy, cur.Schedule); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var retention, maxRows *int32
|
||||
if cur.RetentionPeriodSec != nil {
|
||||
v := int32(*cur.RetentionPeriodSec)
|
||||
retention = &v
|
||||
}
|
||||
if cur.MaxRows != nil {
|
||||
v := int32(*cur.MaxRows)
|
||||
maxRows = &v
|
||||
}
|
||||
ctx := context.Background()
|
||||
tag, err := p.pool.Exec(ctx, `
|
||||
UPDATE maintenance_policy SET
|
||||
name=$2, table_name=$3, condition_sql=$4, retention_period_sec=$5, max_rows=$6,
|
||||
vacuum_strategy=$7, schedule_cron=$8, enabled=$9, dry_run_enabled=$10, updated_at=now()
|
||||
WHERE id=$1`,
|
||||
id, cur.Name, cur.TableName, cur.Condition, retention, maxRows,
|
||||
cur.VacuumStrategy, cur.Schedule, cur.Enabled, cur.DryRunEnabled)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return nil, store.ErrNotFound
|
||||
}
|
||||
return p.GetMaintenancePolicy(id)
|
||||
}
|
||||
|
||||
func (p *Postgres) DeleteMaintenancePolicy(id string) error {
|
||||
ctx := context.Background()
|
||||
tag, err := p.pool.Exec(ctx, `DELETE FROM maintenance_policy WHERE id=$1`, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return store.ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Postgres) TouchMaintenancePolicyRun(id, status, errMsg string) error {
|
||||
ctx := context.Background()
|
||||
tag, err := p.pool.Exec(ctx, `
|
||||
UPDATE maintenance_policy SET
|
||||
last_run_at=now(), last_status=$2, last_error=NULLIF($3,''), updated_at=now()
|
||||
WHERE id=$1`, id, status, errMsg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return store.ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Postgres) AppendMaintenancePolicyConfigAudit(actor, policyID, action string, before, after map[string]any) error {
|
||||
ctx := context.Background()
|
||||
id := uuid.NewString()
|
||||
var beforeJSON, afterJSON []byte
|
||||
if before != nil {
|
||||
beforeJSON, _ = json.Marshal(before)
|
||||
}
|
||||
if after != nil {
|
||||
afterJSON, _ = json.Marshal(after)
|
||||
}
|
||||
_, err := p.pool.Exec(ctx, `
|
||||
INSERT INTO maintenance_policy_config_audit
|
||||
(id, policy_id, actor_prefix, action, before_json, after_json, created_at)
|
||||
VALUES ($1, NULLIF($2,''), $3, $4, $5::jsonb, $6::jsonb, now())`,
|
||||
id, policyID, strings.TrimSpace(actor), action,
|
||||
nullJSONBytes(beforeJSON), nullJSONBytes(afterJSON))
|
||||
return err
|
||||
}
|
||||
|
||||
func nullJSONBytes(b []byte) any {
|
||||
if len(b) == 0 {
|
||||
return nil
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func (p *Postgres) ListMaintenancePolicyConfigAudit(cursor string, limit int) ([]*store.MaintenancePolicyConfigAudit, string, bool, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
off := 0
|
||||
if cursor != "" {
|
||||
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
|
||||
off = n
|
||||
}
|
||||
}
|
||||
ctx := context.Background()
|
||||
rows, err := p.pool.Query(ctx, `
|
||||
SELECT id, COALESCE(policy_id,''), actor_prefix, action,
|
||||
before_json, after_json, created_at
|
||||
FROM maintenance_policy_config_audit
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT $1 OFFSET $2`, limit+1, off)
|
||||
if err != nil {
|
||||
return nil, "", false, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []*store.MaintenancePolicyConfigAudit
|
||||
for rows.Next() {
|
||||
var r store.MaintenancePolicyConfigAudit
|
||||
var beforeRaw, afterRaw []byte
|
||||
if err := rows.Scan(&r.ID, &r.PolicyID, &r.ActorPrefix, &r.Action, &beforeRaw, &afterRaw, &r.CreatedAt); err != nil {
|
||||
continue
|
||||
}
|
||||
if len(beforeRaw) > 0 {
|
||||
_ = json.Unmarshal(beforeRaw, &r.Before)
|
||||
}
|
||||
if len(afterRaw) > 0 {
|
||||
_ = json.Unmarshal(afterRaw, &r.After)
|
||||
}
|
||||
out = append(out, &r)
|
||||
}
|
||||
more := len(out) > limit
|
||||
if more {
|
||||
out = out[:limit]
|
||||
}
|
||||
next := ""
|
||||
if more {
|
||||
next = strconv.Itoa(off + limit)
|
||||
}
|
||||
return out, next, more, rows.Err()
|
||||
}
|
||||
Reference in New Issue
Block a user