《MMO 服务器开发》8.1 备份与恢复

// snippet: internal/dr/policy.go —— 保留链必须保住增量可以回放的基线
func SelectRetained(backups []BackupMeta, now time.Time, p BackupPolicy) []BackupMeta {
	// 1. 先按保留窗口过滤掉太老的。
	// 2. 从后往前保留 KeepFull 份全量。
	// 3. 保留这些全量里最早那份之后的所有增量，保证基线链完整。
	fullSeen, cutIdx := 0, 0
	for i := len(within) - 1; i >= 0; i-- {
		if within[i].Kind == Full {
			fullSeen++
			if fullSeen == p.KeepFull {
				cutIdx = i
				break
			}
		}
	}
	if fullSeen < p.KeepFull {
		cutIdx = 0 // 全量不够 KeepFull 份，全部保留。
	}
	return within[cutIdx:]
}

// file: internal/dr/backup.go
// tier: L2
// verify: go test ./internal/dr/ -run TestPostgresBackupRestore -v
package dr

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"os"
	"sort"
	"time"

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

// WalletRow 是玩家钱包表的一行，用来演示「逻辑导出 → 恢复 → 一致性校验」。
type WalletRow struct {
	PlayerID  int64     `json:"player_id"`
	Gold      int64     `json:"gold"`
	Version   int64     `json:"version"`
	UpdatedAt time.Time `json:"updated_at"`
}

// Snapshot 是一张表的逻辑快照，附带内容校验和。
type Snapshot struct {
	Schema    string      `json:"schema"`
	Table     string      `json:"table"`
	Rows      []WalletRow `json:"rows"`
	Checksum  string      `json:"-"`
	CreatedAt time.Time   `json:"created_at"`
}

// ChecksumRows 对行内容做稳定校验和（按主键排序后逐行哈希）。
// 恢复后只要重算一遍、与备份时的值相等，就证明「恢复到一致状态」。
func ChecksumRows(rows []WalletRow) string {
	sorted := append([]WalletRow(nil), rows...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].PlayerID < sorted[j].PlayerID })
	h := sha256.New()
	for _, r := range sorted {
		fmt.Fprintf(h, "%d|%d|%d|%d\n", r.PlayerID, r.Gold, r.Version, r.UpdatedAt.UnixNano())
	}
	return hex.EncodeToString(h.Sum(nil))
}

// ExportTable 做一次逻辑导出：把整张表读进内存并算校验和。
func ExportTable(ctx context.Context, pool *pgxpool.Pool, schema, table string) (*Snapshot, error) {
	q := fmt.Sprintf("SELECT player_id, gold, version, updated_at FROM %s.%s ORDER BY player_id", schema, table)
	rows, err := pool.Query(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []WalletRow
	for rows.Next() {
		var r WalletRow
		if err := rows.Scan(&r.PlayerID, &r.Gold, &r.Version, &r.UpdatedAt); err != nil {
			return nil, err
		}
		out = append(out, r)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return &Snapshot{Schema: schema, Table: table, Rows: out,
		Checksum: ChecksumRows(out), CreatedAt: time.Now()}, nil
}

// WriteBackup 把快照落成一份备份文件。
func WriteBackup(path string, snap *Snapshot) error {
	b, err := json.Marshal(snap)
	if err != nil {
		return err
	}
	return os.WriteFile(path, b, 0o644)
}

// ReadBackup 读回备份并复算校验和，用于在恢复前发现备份本身被损坏。
func ReadBackup(path string) (*Snapshot, error) {
	b, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var snap Snapshot
	if err := json.Unmarshal(b, &snap); err != nil {
		return nil, err
	}
	snap.Checksum = ChecksumRows(snap.Rows)
	return &snap, nil
}

// RestoreTable 从快照恢复：先清空，再在一个事务里批量写入。
// 返回恢复后重新导出的校验和，调用方拿它和快照校验和比对。
func RestoreTable(ctx context.Context, pool *pgxpool.Pool, snap *Snapshot) (string, error) {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return "", err
	}
	defer tx.Rollback(ctx)

	if _, err := tx.Exec(ctx, fmt.Sprintf("TRUNCATE %s.%s", snap.Schema, snap.Table)); err != nil {
		return "", err
	}
	_, err = tx.CopyFrom(ctx,
		pgx.Identifier{snap.Schema, snap.Table},
		[]string{"player_id", "gold", "version", "updated_at"},
		pgx.CopyFromSlice(len(snap.Rows), func(i int) ([]any, error) {
			r := snap.Rows[i]
			return []any{r.PlayerID, r.Gold, r.Version, r.UpdatedAt}, nil
		}),
	)
	if err != nil {
		return "", err
	}
	if err := tx.Commit(ctx); err != nil {
		return "", err
	}
	after, err := ExportTable(ctx, pool, snap.Schema, snap.Table)
	if err != nil {
		return "", err
	}
	return after.Checksum, nil
}

=== RUN   TestPostgresBackupRestore
    backup_test.go:133: 备份行数=5000 备份校验和=71a8ba0282d0 恢复校验和=71a8ba0282d0 一致
--- PASS: TestPostgresBackupRestore (0.20s)
=== RUN   TestSelectRetained
    backup_test.go:156: 共 9 份备份，保留 6 份，基线=2026-10-10T20:00:00Z
--- PASS: TestSelectRetained (0.00s)
PASS
ok  	aegis/internal/dr	0.866s