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() }