pangolin/server/internal/codes/backfill.go

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
}