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 }