feat(server): legacy 码表 → codes 库新表的幂等回填(Go 双方言)(#codes-lib)

server/internal/codes/backfill.go

@@ -0,0 +1,189 @@
+package codes
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	dbx "github.com/wangjia/pangolin/server/internal/db"
+)
+
+// BackfillLegacy copies legacy_codes / legacy_code_batches rows (renamed away
+// by migration 000020) into the shared library's codes / codes_batches
+// tables. Idempotent: batches keyed by preserved id, codes keyed by the
+// unique code_hash; rows already present are skipped. Returns the number of
+// code rows migrated. If the legacy tables don't exist (fresh install after
+// the eventual legacy cleanup), it is a no-op.
+func BackfillLegacy(ctx context.Context, database *sql.DB) (int, error) {
+	exists, err := legacyTablesExist(ctx, database)
+	if err != nil {
+		return 0, fmt.Errorf("codes.BackfillLegacy: probe legacy tables: %w", err)
+	}
+	if !exists {
+		return 0, nil
+	}
+
+	type legacyBatch struct {
+		id        int64
+		channel   string
+		createdBy string
+		note      sql.NullString
+		createdAt time.Time
+	}
+	type legacyCode struct {
+		id         int64
+		hash       string
+		planCode   string
+		days       int
+		batchID    int64
+		status     string
+		redeemedBy sql.NullInt64
+		redeemedAt sql.NullTime
+	}
+
+	// ── 读阶段:全量取出、关闭游标(不能在同事务连接上边迭代边写)──
+	var batches []legacyBatch
+	rows, err := database.QueryContext(ctx,
+		`SELECT id, channel, created_by, note, created_at FROM legacy_code_batches ORDER BY id`)
+	if err != nil {
+		return 0, fmt.Errorf("codes.BackfillLegacy: read batches: %w", err)
+	}
+	for rows.Next() {
+		var b legacyBatch
+		if err := rows.Scan(&b.id, &b.channel, &b.createdBy, &b.note, &b.createdAt); err != nil {
+			rows.Close()
+			return 0, fmt.Errorf("codes.BackfillLegacy: scan batch: %w", err)
+		}
+		batches = append(batches, b)
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		return 0, err
+	}
+	rows.Close()
+
+	var lcodes []legacyCode
+	rows, err = database.QueryContext(ctx,
+		`SELECT c.id, c.code_hash, p.code, c.duration_days, c.batch_id, c.status, c.redeemed_by, c.redeemed_at
+		   FROM legacy_codes c JOIN plans p ON p.id = c.plan_id ORDER BY c.id`)
+	if err != nil {
+		return 0, fmt.Errorf("codes.BackfillLegacy: read codes: %w", err)
+	}
+	for rows.Next() {
+		var c legacyCode
+		if err := rows.Scan(&c.id, &c.hash, &c.planCode, &c.days, &c.batchID, &c.status, &c.redeemedBy, &c.redeemedAt); err != nil {
+			rows.Close()
+			return 0, fmt.Errorf("codes.BackfillLegacy: scan code: %w", err)
+		}
+		lcodes = append(lcodes, c)
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		return 0, err
+	}
+	rows.Close()
+
+	// batch id → (payload, createdAt):batch 权益取其第一条 code 合成。
+	batchAt := make(map[int64]time.Time, len(batches))
+	batchPayload := make(map[int64]string, len(batches))
+	for _, b := range batches {
+		batchAt[b.id] = b.createdAt
+	}
+	for _, c := range lcodes {
+		if _, ok := batchPayload[c.batchID]; !ok {
+			batchPayload[c.batchID] = durationPayloadJSON(c.planCode, c.days)
+		}
+	}
+
+	// ── 写阶段:单事务,全成或全无;逐行判重保证幂等。──
+	tx, err := database.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, fmt.Errorf("codes.BackfillLegacy: begin: %w", err)
+	}
+	defer tx.Rollback()
+
+	for _, b := range batches {
+		var n int
+		if err := tx.QueryRowContext(ctx,
+			`SELECT COUNT(*) FROM codes_batches WHERE id = ?`, b.id).Scan(&n); err != nil {
+			return 0, fmt.Errorf("codes.BackfillLegacy: probe batch %d: %w", b.id, err)
+		}
+		if n > 0 {
+			continue
+		}
+		payload, ok := batchPayload[b.id]
+		if !ok {
+			payload = durationPayloadJSON("unknown", 0) // 空批次:信息性占位
+		}
+		var note any
+		if b.note.Valid {
+			note = b.note.String
+		}
+		if _, err := tx.ExecContext(ctx,
+			`INSERT INTO codes_batches (id, channel, entitlement_kind, entitlement_payload, created_by, note, created_at)
+			 VALUES (?, ?, 'duration', ?, ?, ?, ?)`,
+			b.id, b.channel, payload, b.createdBy, note, b.createdAt.UTC()); err != nil {
+			return 0, fmt.Errorf("codes.BackfillLegacy: insert batch %d: %w", b.id, err)
+		}
+	}
+
+	migrated := 0
+	for _, c := range lcodes {
+		var n int
+		if err := tx.QueryRowContext(ctx,
+			`SELECT COUNT(*) FROM codes WHERE code_hash = ?`, c.hash).Scan(&n); err != nil {
+			return 0, fmt.Errorf("codes.BackfillLegacy: probe code %d: %w", c.id, err)
+		}
+		if n > 0 {
+			continue
+		}
+		var redeemedBy, redeemedAt any
+		if c.redeemedBy.Valid {
+			redeemedBy = fmt.Sprintf("user:%d", c.redeemedBy.Int64)
+		}
+		if c.redeemedAt.Valid {
+			redeemedAt = c.redeemedAt.Time.UTC()
+		}
+		createdAt, ok := batchAt[c.batchID]
+		if !ok {
+			createdAt = time.Now()
+		}
+		if _, err := tx.ExecContext(ctx,
+			`INSERT INTO codes (id, code_hash, batch_id, entitlement_kind, entitlement_payload, status, redeemed_by, redeemed_at, void_reason, created_at)
+			 VALUES (?, ?, ?, 'duration', ?, ?, ?, ?, NULL, ?)`,
+			c.id, c.hash, c.batchID, durationPayloadJSON(c.planCode, c.days),
+			c.status, redeemedBy, redeemedAt, createdAt.UTC()); err != nil {
+			return 0, fmt.Errorf("codes.BackfillLegacy: insert code %d: %w", c.id, err)
+		}
+		migrated++
+	}
+
+	if err := tx.Commit(); err != nil {
+		return 0, fmt.Errorf("codes.BackfillLegacy: commit: %w", err)
+	}
+	return migrated, nil
+}
+
+func durationPayloadJSON(plan string, days int) string {
+	b, _ := json.Marshal(struct {
+		Plan string `json:"plan"`
+		Days int    `json:"days"`
+	}{plan, days})
+	return string(b)
+}
+
+// legacyTablesExist probes for legacy_codes per dialect (sqlite_master vs
+// information_schema) — no error-swallowing "SELECT and see" hacks.
+func legacyTablesExist(ctx context.Context, database *sql.DB) (bool, error) {
+	var n int
+	if dbx.DialectForDB(database).Name() == "sqlite" {
+		err := database.QueryRowContext(ctx,
+			`SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='legacy_codes'`).Scan(&n)
+		return n > 0, err
+	}
+	err := database.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'legacy_codes'`).Scan(&n)
+	return n > 0, err
+}