ITWorx-Pulse-Public/internal/backup/manager.go

package backup

import (
	"archive/zip"
	"bufio"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

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

const (
	formatVersion    = 2
	defaultRetention = 5
	manifestName     = "manifest.json"
)

var ErrNotConfigured = errors.New("backup destination is not configured")

type Manager struct {
	Pool      *pgxpool.Pool
	Directory string
	Retention int
	Now       func() time.Time
}

type Manifest struct {
	FormatVersion int             `json:"formatVersion"`
	BackupID      string          `json:"backupId"`
	CreatedAt     time.Time       `json:"createdAt"`
	SchemaVersion int64           `json:"schemaVersion"`
	Tables        []TableManifest `json:"tables"`
	Excluded      []string        `json:"excluded"`
}

type TableManifest struct {
	Name   string `json:"name"`
	File   string `json:"file"`
	Rows   int64  `json:"rows"`
	SHA256 string `json:"sha256"`
}

type Result struct {
	BackupID string    `json:"backupId"`
	Path     string    `json:"path"`
	SHA256   string    `json:"sha256"`
	Bytes    int64     `json:"bytes"`
	Rows     int64     `json:"rows"`
	Created  time.Time `json:"createdAt"`
}

type tableSpec struct {
	name        string
	columns     string
	restoreCols string
	orderBy     string
}

var tableSpecs = []tableSpec{
	{name: "roles", columns: "id,name,created_at", restoreCols: "id,name,created_at", orderBy: "id"},
	{name: "users", columns: "id,external_subject,display_name,email,status,created_at,updated_at,last_login_at", restoreCols: "id,external_subject,display_name,email,status,created_at,updated_at,last_login_at", orderBy: "id"},
	{name: "user_roles", columns: "user_id,role_id,created_at", restoreCols: "user_id,role_id,created_at", orderBy: "user_id,role_id"},
	{name: "data_sources", columns: "id,type,name,enabled,configuration_ref,capability_document,health_state,last_success_at,last_error_code,last_error_message,freshness_policy,created_at,updated_at", restoreCols: "id,type,name,enabled,configuration_ref,capability_document,health_state,last_success_at,last_error_code,last_error_message,freshness_policy,created_at,updated_at", orderBy: "id"},
	{name: "collectors", columns: "id,datasource_id,kind,version,heartbeat,capabilities,status", restoreCols: "id,datasource_id,kind,version,heartbeat,capabilities,status", orderBy: "id"},
	{name: "entities", columns: "id,entity_type,canonical_name,display_name,status,status_reasons,first_seen_at,last_seen_at,tombstoned_at,attributes", restoreCols: "id,entity_type,canonical_name,display_name,status,status_reasons,first_seen_at,last_seen_at,tombstoned_at,attributes", orderBy: "id"},
	{name: "entity_aliases", columns: "entity_id,source_id,external_type,external_id", restoreCols: "entity_id,source_id,external_type,external_id", orderBy: "source_id,external_type,external_id"},
	{name: "container_aliases", columns: "source_id,runtime_id,entity_id,name,project,service,image_digest,observed_state,observed_health,restart_count,intentional_stop,first_seen_at,last_seen_at,tombstoned_at", restoreCols: "source_id,runtime_id,entity_id,name,project,service,image_digest,observed_state,observed_health,restart_count,intentional_stop,first_seen_at,last_seen_at,tombstoned_at", orderBy: "source_id,runtime_id"},
	{name: "entity_facts", columns: "entity_id,field_name,source_id,value,observed_at,confidence,valid_until", restoreCols: "entity_id,field_name,source_id,value,observed_at,confidence,valid_until", orderBy: "entity_id,field_name,source_id"},
	{name: "entity_overrides", columns: "entity_id,field_name,value,user_id,updated_at", restoreCols: "entity_id,field_name,value,user_id,updated_at", orderBy: "entity_id,field_name"},
	{name: "entity_relations", columns: "id,source_entity_id,relation_type,target_entity_id,source_id,confidence,confirmed,first_seen_at,last_seen_at,tombstoned_at", restoreCols: "id,source_entity_id,relation_type,target_entity_id,source_id,confidence,confirmed,first_seen_at,last_seen_at,tombstoned_at", orderBy: "id"},
	{name: "dashboards", columns: "id,slug,name,description,owner_user_id,scope,archived_at,current_version_id,revision,created_at,updated_at", restoreCols: "id,slug,name,description,owner_user_id,scope,archived_at,revision,created_at,updated_at", orderBy: "id"},
	{name: "dashboard_versions", columns: "id,dashboard_id,version_number,schema_version,document,change_summary,created_by,created_at", restoreCols: "id,dashboard_id,version_number,schema_version,document,change_summary,created_by,created_at", orderBy: "dashboard_id,version_number"},
	{name: "events", columns: "id,event_type,severity,entity_id,source_id,occurred_at,received_at,dedup_key,summary,attributes,correlation_id", restoreCols: "id,event_type,severity,entity_id,source_id,occurred_at,received_at,dedup_key,summary,attributes,correlation_id", orderBy: "id"},
	{name: "audit_events", columns: "id,actor,action,resource_type,resource_id,result,occurred_at,correlation_id,before_diff,after_diff", restoreCols: "id,actor,action,resource_type,resource_id,result,occurred_at,correlation_id,before_diff,after_diff", orderBy: "occurred_at,id"},
	{name: "job_runs", columns: "id,job_type,job_key,scheduled_at,started_at,completed_at,status,counts,error_code,correlation_id,lease_owner,lease_until", restoreCols: "id,job_type,job_key,scheduled_at,started_at,completed_at,status,counts,error_code,correlation_id,lease_owner,lease_until", orderBy: "scheduled_at,id"},
	{name: "services", columns: "id,entity_id,source_id,name,description,state,labels,revision,archived_at,created_by,created_at,updated_at", restoreCols: "id,entity_id,source_id,name,description,state,labels,revision,archived_at,created_by,created_at,updated_at", orderBy: "id"},
	{name: "service_endpoints", columns: "id,service_id,source_id,name,endpoint_type,target,enabled,revision,archived_at,created_at,updated_at", restoreCols: "id,service_id,source_id,name,endpoint_type,target,enabled,revision,archived_at,created_at,updated_at", orderBy: "id"},
	{name: "probes", columns: "id,service_id,endpoint_id,source_id,name,probe_type,target,interval_seconds,timeout_seconds,enabled,expected_status_codes,follow_redirects,verify_tls,content_assertion,network_policy_id,revision,archived_at,created_by,created_at,updated_at", restoreCols: "id,service_id,endpoint_id,source_id,name,probe_type,target,interval_seconds,timeout_seconds,enabled,expected_status_codes,follow_redirects,verify_tls,content_assertion,network_policy_id,revision,archived_at,created_by,created_at,updated_at", orderBy: "id"},
	{name: "probe_results", columns: "id,probe_id,source_id,observed_at,completed_at,state,response_time_ms,status_code,error_class,error_message,attributes", restoreCols: "id,probe_id,source_id,observed_at,completed_at,state,response_time_ms,status_code,error_class,error_message,attributes", orderBy: "probe_id,observed_at"},
	{name: "service_certificates", columns: "id,service_id,endpoint_id,source_id,observed_at,expires_at,issuer,subject,hostname_valid,verification_state,attributes", restoreCols: "id,service_id,endpoint_id,source_id,observed_at,expires_at,issuer,subject,hostname_valid,verification_state,attributes", orderBy: "id"},
	{name: "service_dependencies", columns: "id,service_id,depends_on_service_id,source_id,relation_type,confidence,confirmed,first_seen_at,last_seen_at,archived_at", restoreCols: "id,service_id,depends_on_service_id,source_id,relation_type,confidence,confirmed,first_seen_at,last_seen_at,archived_at", orderBy: "id"},
	{name: "service_permissions", columns: "service_id,role_id,permission,created_at", restoreCols: "service_id,role_id,permission,created_at", orderBy: "service_id,role_id,permission"},
	{name: "alert_rules", columns: "id,schema_version,name,enabled,severity,scope,condition,evaluation_interval_seconds,pending_seconds,resolve_seconds,unknown_behavior,group_by,suppress_when,message,current_version_id,revision,created_by,created_at,updated_at,cooldown_seconds", restoreCols: "id,schema_version,name,enabled,severity,scope,condition,evaluation_interval_seconds,pending_seconds,resolve_seconds,unknown_behavior,group_by,suppress_when,message,revision,created_by,created_at,updated_at,cooldown_seconds", orderBy: "id"},
	{name: "alert_rule_versions", columns: "id,rule_id,version_number,document,change_summary,created_by,created_at", restoreCols: "id,rule_id,version_number,document,change_summary,created_by,created_at", orderBy: "rule_id,version_number"},
	{name: "alert_instances", columns: "id,rule_id,rule_version_id,fingerprint,entity_id,current_state,retained_state,active_since,recovery_since,last_evaluated_at,last_known_at,last_value,reason,source_health,acknowledged_by,acknowledged_at,revision,created_at,updated_at,cooldown_until", restoreCols: "id,rule_id,rule_version_id,fingerprint,entity_id,current_state,retained_state,active_since,recovery_since,last_evaluated_at,last_known_at,last_value,reason,source_health,acknowledged_by,acknowledged_at,revision,created_at,updated_at,cooldown_until", orderBy: "id"},
	{name: "alert_occurrences", columns: "id,instance_id,evaluation_key,event_type,from_state,to_state,observed_at,value,reason,source_health,created_at", restoreCols: "id,instance_id,evaluation_key,event_type,from_state,to_state,observed_at,value,reason,source_health,created_at", orderBy: "instance_id,observed_at,id"},
	{name: "alert_silences", columns: "id,name,reason,owner,matchers,starts_at,expires_at,status,created_by,created_at,revoked_by,revoked_at,expired_at,revision", restoreCols: "id,name,reason,owner,matchers,starts_at,expires_at,status,created_by,created_at,revoked_by,revoked_at,expired_at,revision", orderBy: "id"},
	{name: "maintenance_windows", columns: "id,name,reason,selector,starts_at,ends_at,status,created_by,created_at,revoked_by,revoked_at,expired_at,revision", restoreCols: "id,name,reason,selector,starts_at,ends_at,status,created_by,created_at,revoked_by,revoked_at,expired_at,revision", orderBy: "id"},
	{name: "incidents", columns: "id,correlation_key,title,summary,severity,status,started_at,resolved_at,owner_user_id,correlation_method,confidence,revision,created_at,updated_at", restoreCols: "id,correlation_key,title,summary,severity,status,started_at,resolved_at,owner_user_id,correlation_method,confidence,revision,created_at,updated_at", orderBy: "id"},
	{name: "incident_alerts", columns: "incident_id,alert_id,rationale,confidence,correlation_method,is_manual,added_by,created_at", restoreCols: "incident_id,alert_id,rationale,confidence,correlation_method,is_manual,added_by,created_at", orderBy: "incident_id,alert_id"},
	{name: "incident_entities", columns: "incident_id,entity_id,rationale,confidence,created_at", restoreCols: "incident_id,entity_id,rationale,confidence,created_at", orderBy: "incident_id,entity_id"},
	{name: "incident_notes", columns: "id,incident_id,author,body,created_at", restoreCols: "id,incident_id,author,body,created_at", orderBy: "incident_id,created_at,id"},
}

// backupExcludedTables classifies application tables that deliberately do not
// belong in a portable archive. The PostgreSQL integration test requires every
// migrated application table to appear either here or in tableSpecs, so adding a
// migration cannot silently make restore incomplete.
var backupExcludedTables = map[string]string{
	"agent_snapshots":         "bounded runtime telemetry is republished by the agent after restart",
	"capacity_samples":        "bounded forecast telemetry is republished by the agent after restart",
	"notification_channels":   "secret references and channel configuration must be reattached",
	"notification_deliveries": "runtime notification delivery history is intentionally excluded",
	"notification_outbox":     "runtime notification delivery state is intentionally excluded",
	"system_settings":         "runtime configuration and secret-bearing values must be reattached",
}

func backupExclusions() []string {
	names := make([]string, 0, len(backupExcludedTables))
	for name := range backupExcludedTables {
		names = append(names, name)
	}
	sort.Strings(names)
	result := make([]string, 0, len(names))
	for _, name := range names {
		result = append(result, name+" ("+backupExcludedTables[name]+")")
	}
	return result
}

func (m Manager) Create(ctx context.Context) (Result, error) {
	if m.Pool == nil || strings.TrimSpace(m.Directory) == "" {
		return Result{}, ErrNotConfigured
	}
	if err := os.MkdirAll(m.Directory, 0o700); err != nil {
		return Result{}, fmt.Errorf("create backup directory: %w", err)
	}
	now := time.Now().UTC()
	if m.Now != nil {
		now = m.Now().UTC()
	}
	id, err := newID()
	if err != nil {
		return Result{}, fmt.Errorf("generate backup id: %w", err)
	}
	schemaVersion, err := currentSchemaVersion(ctx, m.Pool)
	if err != nil {
		return Result{}, err
	}
	temp, err := os.CreateTemp(m.Directory, ".pulse-backup-*.tmp")
	if err != nil {
		return Result{}, fmt.Errorf("create temporary backup: %w", err)
	}
	tempName := temp.Name()
	defer os.Remove(tempName)
	archive := zip.NewWriter(temp)
	manifest := Manifest{FormatVersion: formatVersion, BackupID: id, CreatedAt: now, SchemaVersion: schemaVersion, Excluded: backupExclusions()}
	for _, spec := range tableSpecs {
		entry, err := archive.Create("data/" + spec.name + ".jsonl")
		if err != nil {
			return Result{}, fmt.Errorf("create archive entry %s: %w", spec.name, err)
		}
		hash := sha256.New()
		writer := io.MultiWriter(entry, hash)
		rows, err := exportTable(ctx, m.Pool, spec, writer)
		if err != nil {
			return Result{}, fmt.Errorf("export %s: %w", spec.name, err)
		}
		manifest.Tables = append(manifest.Tables, TableManifest{Name: spec.name, File: "data/" + spec.name + ".jsonl", Rows: rows, SHA256: hex.EncodeToString(hash.Sum(nil))})
	}
	manifestEntry, err := archive.Create(manifestName)
	if err != nil {
		return Result{}, fmt.Errorf("create manifest: %w", err)
	}
	if err := json.NewEncoder(manifestEntry).Encode(manifest); err != nil {
		return Result{}, fmt.Errorf("write manifest: %w", err)
	}
	if err := archive.Close(); err != nil {
		return Result{}, fmt.Errorf("close backup archive: %w", err)
	}
	if err := temp.Sync(); err != nil {
		return Result{}, fmt.Errorf("sync backup archive: %w", err)
	}
	if err := temp.Close(); err != nil {
		return Result{}, fmt.Errorf("close backup file: %w", err)
	}
	finalPath := filepath.Join(m.Directory, "pulse-backup-"+id+".zip")
	if err := os.Rename(tempName, finalPath); err != nil {
		return Result{}, fmt.Errorf("finalize backup: %w", err)
	}
	checksum, size, err := fileChecksum(finalPath)
	if err != nil {
		return Result{}, err
	}
	if err := os.WriteFile(finalPath+".sha256", []byte(checksum+"  "+filepath.Base(finalPath)+"\n"), 0o600); err != nil {
		return Result{}, fmt.Errorf("write backup checksum: %w", err)
	}
	if err := m.prune(ctx, finalPath); err != nil {
		return Result{}, err
	}
	var rows int64
	for _, table := range manifest.Tables {
		rows += table.Rows
	}
	return Result{BackupID: id, Path: finalPath, SHA256: checksum, Bytes: size, Rows: rows, Created: now}, nil
}

func (m Manager) Verify(ctx context.Context, path string) (Manifest, error) {
	if strings.TrimSpace(path) == "" {
		return Manifest{}, errors.New("backup path is required")
	}
	archive, err := zip.OpenReader(path)
	if err != nil {
		return Manifest{}, fmt.Errorf("open backup: %w", err)
	}
	defer archive.Close()
	entries := make(map[string]*zip.File, len(archive.File))
	for _, entry := range archive.File {
		if _, exists := entries[entry.Name]; exists {
			return Manifest{}, fmt.Errorf("duplicate archive entry %q", entry.Name)
		}
		entries[entry.Name] = entry
	}
	manifestFile, ok := entries[manifestName]
	if !ok {
		return Manifest{}, errors.New("backup manifest is missing")
	}
	manifestReader, err := manifestFile.Open()
	if err != nil {
		return Manifest{}, fmt.Errorf("open manifest: %w", err)
	}
	var manifest Manifest
	err = json.NewDecoder(manifestReader).Decode(&manifest)
	_ = manifestReader.Close()
	if err != nil || manifest.FormatVersion != formatVersion || manifest.BackupID == "" || manifest.SchemaVersion <= 0 {
		return Manifest{}, errors.New("backup manifest is invalid")
	}
	if len(manifest.Tables) != len(tableSpecs) {
		return Manifest{}, fmt.Errorf("backup table set is incomplete: got %d, want %d", len(manifest.Tables), len(tableSpecs))
	}
	expectedFiles := map[string]bool{manifestName: true}
	for _, spec := range tableSpecs {
		expectedFiles["data/"+spec.name+".jsonl"] = true
	}
	for name := range entries {
		if !expectedFiles[name] {
			return Manifest{}, fmt.Errorf("unexpected backup archive entry %q", name)
		}
	}
	seen := make(map[string]bool, len(manifest.Tables))
	for _, table := range manifest.Tables {
		if seen[table.Name] || table.Rows < 0 || table.SHA256 == "" || table.File != "data/"+table.Name+".jsonl" || !expectedFiles[table.File] {
			return Manifest{}, fmt.Errorf("backup table entry %q is invalid", table.Name)
		}
		seen[table.Name] = true
		entry, ok := entries[table.File]
		if !ok {
			return Manifest{}, fmt.Errorf("backup table file %q is missing", table.File)
		}
		if err := verifyTable(entry, table); err != nil {
			return Manifest{}, err
		}
	}
	for _, spec := range tableSpecs {
		if !seen[spec.name] {
			return Manifest{}, fmt.Errorf("backup table %q is missing", spec.name)
		}
	}
	checksum, _, err := fileChecksum(path)
	if err != nil {
		return Manifest{}, err
	}
	sidecar, err := os.ReadFile(path + ".sha256")
	if err != nil {
		return Manifest{}, fmt.Errorf("read backup checksum: %w", err)
	}
	if !strings.HasPrefix(string(sidecar), checksum+"  ") {
		return Manifest{}, errors.New("backup archive checksum does not match sidecar")
	}
	return manifest, nil
}

func (m Manager) List(ctx context.Context) ([]Result, error) {
	if strings.TrimSpace(m.Directory) == "" {
		return nil, ErrNotConfigured
	}
	entries, err := os.ReadDir(m.Directory)
	if err != nil {
		if os.IsNotExist(err) {
			return []Result{}, nil
		}
		return nil, fmt.Errorf("list backups: %w", err)
	}
	results := make([]Result, 0)
	for _, entry := range entries {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		if entry.IsDir() || !strings.HasPrefix(entry.Name(), "pulse-backup-") || !strings.HasSuffix(entry.Name(), ".zip") {
			continue
		}
		path := filepath.Join(m.Directory, entry.Name())
		manifest, err := m.Verify(ctx, path)
		if err != nil {
			return nil, fmt.Errorf("verify listed backup %s: %w", entry.Name(), err)
		}
		checksum, size, err := fileChecksum(path)
		if err != nil {
			return nil, err
		}
		var rows int64
		for _, table := range manifest.Tables {
			rows += table.Rows
		}
		results = append(results, Result{BackupID: manifest.BackupID, Path: path, SHA256: checksum, Bytes: size, Rows: rows, Created: manifest.CreatedAt})
	}
	sort.Slice(results, func(i, j int) bool { return results[i].Created.After(results[j].Created) })
	return results, nil
}
func (m Manager) Restore(ctx context.Context, path string) (Manifest, error) {
	if m.Pool == nil {
		return Manifest{}, ErrNotConfigured
	}
	manifest, err := m.Verify(ctx, path)
	if err != nil {
		return Manifest{}, err
	}
	for _, spec := range tableSpecs {
		var count int64
		if err := m.Pool.QueryRow(ctx, "SELECT count(*) FROM "+spec.name).Scan(&count); err != nil {
			return Manifest{}, fmt.Errorf("check restore target %s: %w", spec.name, err)
		}
		if count != 0 {
			return Manifest{}, fmt.Errorf("restore target is not empty: %s has %d rows", spec.name, count)
		}
	}
	archive, err := zip.OpenReader(path)
	if err != nil {
		return Manifest{}, fmt.Errorf("open restore archive: %w", err)
	}
	defer archive.Close()
	entries := make(map[string]*zip.File, len(archive.File))
	for _, entry := range archive.File {
		entries[entry.Name] = entry
	}
	tx, err := m.Pool.Begin(ctx)
	if err != nil {
		return Manifest{}, fmt.Errorf("begin restore: %w", err)
	}
	defer tx.Rollback(ctx)
	for _, spec := range tableSpecs {
		entry := entries["data/"+spec.name+".jsonl"]
		reader, err := entry.Open()
		if err != nil {
			return Manifest{}, fmt.Errorf("open restore table %s: %w", spec.name, err)
		}
		decoder := json.NewDecoder(bufio.NewReader(reader))
		for {
			var row json.RawMessage
			if err := decoder.Decode(&row); errors.Is(err, io.EOF) {
				break
			} else if err != nil {
				_ = reader.Close()
				return Manifest{}, fmt.Errorf("decode restore table %s: %w", spec.name, err)
			}
			selectCols := spec.restoreCols
			if spec.name == "alert_instances" {
				selectCols = strings.Replace(selectCols, "last_value", "COALESCE(last_value, 'null'::jsonb)", 1)
			} else if spec.name == "alert_occurrences" {
				selectCols = strings.Replace(selectCols, "value", "COALESCE(value, 'null'::jsonb)", 1)
			}
			query := "INSERT INTO " + spec.name + " (" + spec.restoreCols + ") SELECT " + selectCols + " FROM jsonb_populate_record(NULL::" + spec.name + ", $1::jsonb)"
			if _, err := tx.Exec(ctx, query, []byte(row)); err != nil {
				_ = reader.Close()
				return Manifest{}, fmt.Errorf("restore %s: %w", spec.name, err)
			}
		}
		_ = reader.Close()
	}
	for _, deferred := range []struct{ table, id, column string }{{"dashboards", "id", "current_version_id"}, {"alert_rules", "id", "current_version_id"}} {
		entry := entries["data/"+deferred.table+".jsonl"]
		reader, err := entry.Open()
		if err != nil {
			return Manifest{}, fmt.Errorf("open deferred restore %s: %w", deferred.table, err)
		}
		decoder := json.NewDecoder(bufio.NewReader(reader))
		for {
			var row map[string]json.RawMessage
			if err := decoder.Decode(&row); errors.Is(err, io.EOF) {
				break
			} else if err != nil {
				_ = reader.Close()
				return Manifest{}, fmt.Errorf("decode deferred restore %s: %w", deferred.table, err)
			}
			id, ok := row[deferred.id]
			value, valueOK := row[deferred.column]
			if !ok || !valueOK || string(value) == "null" {
				continue
			}
			var valueID, rowID string
			if err := json.Unmarshal(value, &valueID); err != nil {
				_ = reader.Close()
				return Manifest{}, fmt.Errorf("decode deferred %s id: %w", deferred.table, err)
			}
			if err := json.Unmarshal(id, &rowID); err != nil {
				_ = reader.Close()
				return Manifest{}, fmt.Errorf("decode deferred %s row id: %w", deferred.table, err)
			}
			if _, err := tx.Exec(ctx, "UPDATE "+deferred.table+" SET "+deferred.column+"=$1::uuid WHERE "+deferred.id+"=$2::uuid", valueID, rowID); err != nil {
				_ = reader.Close()
				return Manifest{}, fmt.Errorf("restore deferred %s: %w", deferred.table, err)
			}
		}
		_ = reader.Close()
	}
	if err := tx.Commit(ctx); err != nil {
		return Manifest{}, fmt.Errorf("commit restore: %w", err)
	}
	return manifest, nil
}

func exportTable(ctx context.Context, pool *pgxpool.Pool, spec tableSpec, writer io.Writer) (int64, error) {
	rows, err := pool.Query(ctx, "SELECT row_to_json(t) FROM (SELECT "+spec.columns+" FROM "+spec.name+" ORDER BY "+spec.orderBy+") t")
	if err != nil {
		return 0, err
	}
	defer rows.Close()
	var count int64
	for rows.Next() {
		var raw []byte
		if err := rows.Scan(&raw); err != nil {
			return 0, err
		}
		if containsSensitiveKey(raw) {
			return 0, fmt.Errorf("sensitive key detected in %s export", spec.name)
		}
		if _, err := writer.Write(append(raw, '\n')); err != nil {
			return 0, err
		}
		count++
	}
	return count, rows.Err()
}

func verifyTable(entry *zip.File, expected TableManifest) error {
	reader, err := entry.Open()
	if err != nil {
		return fmt.Errorf("open table %s: %w", expected.Name, err)
	}
	defer reader.Close()
	hash := sha256.New()
	decoder := json.NewDecoder(io.TeeReader(reader, hash))
	var rows int64
	for {
		var raw json.RawMessage
		if err := decoder.Decode(&raw); errors.Is(err, io.EOF) {
			break
		} else if err != nil {
			return fmt.Errorf("validate table %s: %w", expected.Name, err)
		}
		if containsSensitiveKey(raw) {
			return fmt.Errorf("sensitive key detected in %s archive", expected.Name)
		}
		rows++
	}
	if rows != expected.Rows || hex.EncodeToString(hash.Sum(nil)) != expected.SHA256 {
		return fmt.Errorf("table %s checksum or row count mismatch", expected.Name)
	}
	return nil
}

func currentSchemaVersion(ctx context.Context, pool *pgxpool.Pool) (int64, error) {
	var version int64
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM schema_migrations`).Scan(&version); err != nil {
		return 0, fmt.Errorf("read schema version: %w", err)
	}
	return version, nil
}

func fileChecksum(path string) (string, int64, error) {
	file, err := os.Open(path)
	if err != nil {
		return "", 0, fmt.Errorf("open backup for checksum: %w", err)
	}
	defer file.Close()
	info, err := file.Stat()
	if err != nil {
		return "", 0, fmt.Errorf("stat backup: %w", err)
	}
	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		return "", 0, fmt.Errorf("checksum backup: %w", err)
	}
	return hex.EncodeToString(hash.Sum(nil)), info.Size(), nil
}

func (m Manager) prune(ctx context.Context, keepPath string) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	retention := m.Retention
	if retention <= 0 {
		retention = defaultRetention
	}
	entries, err := os.ReadDir(m.Directory)
	if err != nil {
		return fmt.Errorf("list backups: %w", err)
	}
	var archives []os.DirEntry
	for _, entry := range entries {
		if !entry.IsDir() && strings.HasPrefix(entry.Name(), "pulse-backup-") && strings.HasSuffix(entry.Name(), ".zip") {
			archives = append(archives, entry)
		}
	}
	sort.Slice(archives, func(i, j int) bool { return archives[i].Name() > archives[j].Name() })
	if len(archives) <= retention {
		return nil
	}
	for _, entry := range archives[retention:] {
		path := filepath.Join(m.Directory, entry.Name())
		if path == keepPath {
			continue
		}
		if err := os.Remove(path); err != nil {
			return fmt.Errorf("prune backup %s: %w", entry.Name(), err)
		}
		_ = os.Remove(path + ".sha256")
	}
	return nil
}

func containsSensitiveKey(raw []byte) bool {
	var value any
	if json.Unmarshal(raw, &value) != nil {
		return true
	}
	return sensitiveValue(value)
}

func sensitiveValue(value any) bool {
	switch typed := value.(type) {
	case map[string]any:
		for key, nested := range typed {
			lower := strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(key, "-", "_"), " ", "_"))
			for _, part := range []string{"authorization", "cookie", "password", "passwd", "secret", "token", "api_key", "apikey", "client_secret"} {
				if strings.Contains(lower, part) {
					return true
				}
			}
			if sensitiveValue(nested) {
				return true
			}
		}
	case []any:
		for _, nested := range typed {
			if sensitiveValue(nested) {
				return true
			}
		}
	}
	return false
}

func newID() (string, error) {
	var bytes [16]byte
	if _, err := rand.Read(bytes[:]); err != nil {
		return "", err
	}
	bytes[6] = (bytes[6] & 0x0f) | 0x40
	bytes[8] = (bytes[8] & 0x3f) | 0x80
	return fmt.Sprintf("%s-%s-%s-%s-%s", hex.EncodeToString(bytes[0:4]), hex.EncodeToString(bytes[4:6]), hex.EncodeToString(bytes[6:8]), hex.EncodeToString(bytes[8:10]), hex.EncodeToString(bytes[10:16])), nil
}