package main import ( "context" "database/sql" "fmt" "log" "sort" "strings" "github.com/jackc/pgx/v5/pgxpool" ) // CountPair is one MySQL vs Postgres table count comparison. type CountPair struct { Entity string `json:"entity"` MySQL int64 `json:"mysql"` Postgres int64 `json:"postgres"` Delta int64 `json:"delta"` Note string `json:"note,omitempty"` } // OrphanFinding is a remapped FK that does not resolve in Postgres. type OrphanFinding struct { Check string `json:"check"` Count int64 `json:"count"` Sample string `json:"sample,omitempty"` Pass bool `json:"pass"` } // ValidationReport is written next to the ID map for cutover verification. type ValidationReport struct { Mode string `json:"mode"` Counts []CountPair `json:"counts"` PostgresCounts map[string]int64 `json:"postgres_counts,omitempty"` Orphans []OrphanFinding `json:"orphans"` OrphanSummary OrphanSummary `json:"orphan_summary"` OK bool `json:"ok"` } // OrphanSummary is a compact end-of-run pass/fail tally. type OrphanSummary struct { Passed int `json:"passed"` Failed int `json:"failed"` Total int `json:"total"` } func mysqlCount(ctx context.Context, db *sql.DB, table string) (int64, error) { quoted, err := quoteMySQLIdent(table) if err != nil { return -1, err } if !mysqlTableExists(ctx, db, table) { return -1, fmt.Errorf("missing") } var n int64 err = db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+quoted).Scan(&n) return n, err } func pgCount(ctx context.Context, pg *pgxpool.Pool, table string) (int64, error) { quoted, err := quotePGIdent(table) if err != nil { return -1, err } var n int64 err = pg.QueryRow(ctx, "SELECT COUNT(*) FROM "+quoted).Scan(&n) return n, err } // pgVerificationTables are Postgres targets printed at end-of-run. var pgVerificationTables = []string{ "companies", "users", "memberships", "plans", "company_plans", "credit_balances", "categories", "attributes", "category_attributes", "custom_variables", "input_feeds", "feed_mappings", "export_feeds", "raw_products", "processed_products", "files", } func collectPostgresCounts(ctx context.Context, pg *pgxpool.Pool, dryRun bool) map[string]int64 { out := map[string]int64{} if dryRun || pg == nil { return out } for _, t := range pgVerificationTables { if n, err := pgCount(ctx, pg, t); err == nil { out[t] = n } else { out[t] = -1 } } return out } // buildCountReport compares allowlisted MySQL source tables to Postgres targets. func buildCountReport(ctx context.Context, mysqlDB *sql.DB, pg *pgxpool.Pool, dryRun bool) []CountPair { pairs := []struct{ mysql, postgres, note string }{ {"companies", "companies", ""}, {"profiles", "memberships", "profiles → memberships"}, {"users", "users", "no password_hash imported"}, {"admin_users", "", "folded into users.is_platform_admin"}, {"plans", "plans", ""}, {"company_plans", "company_plans", ""}, {"credit_balances", "credit_balances", ""}, {"categories", "categories", ""}, {"attributes", "attributes", ""}, {"category_attributes", "category_attributes", ""}, {"custom_variables", "custom_variables", "label/example → value"}, {"xml_feeds", "input_feeds", "xml_feeds → input_feeds"}, {"raw_products", "raw_products", ""}, {"processed_products", "processed_products", ""}, {"export_feeds", "export_feeds", ""}, {"files", "files", "metadata only; blobs not copied"}, {"company_settings", "company_settings", "partial: language + merge_products only"}, {"api_keys", "", "not migrated; clients must create new keys"}, {"processing_jobs", "processing_jobs", "migrated when domain jobs enabled (ai_provider_mode=migrated)"}, } out := make([]CountPair, 0, len(pairs)) for _, p := range pairs { cp := CountPair{Entity: p.mysql, Note: p.note, MySQL: -1, Postgres: -1} if n, err := mysqlCount(ctx, mysqlDB, p.mysql); err == nil { cp.MySQL = n } else { cp.Note = strings.TrimSpace(cp.Note + " mysql_missing") } if p.postgres != "" && !dryRun { if n, err := pgCount(ctx, pg, p.postgres); err == nil { cp.Postgres = n if cp.MySQL >= 0 { cp.Delta = cp.Postgres - cp.MySQL } } else { cp.Note = strings.TrimSpace(cp.Note + " pg_error") } } out = append(out, cp) } return out } // checkOrphanFKs runs Postgres-side orphan queries after a live load. // Dry-run skips (no writes to validate). func checkOrphanFKs(ctx context.Context, pg *pgxpool.Pool, dryRun bool) []OrphanFinding { if dryRun { return []OrphanFinding{{ Check: "skipped_dry_run", Pass: true, Sample: "orphan FK checks require a live Postgres load", }} } checks := []struct { name string sql string }{ {"memberships_missing_user", `SELECT COUNT(*) FROM memberships m LEFT JOIN users u ON u.id = m.user_id WHERE u.id IS NULL`}, {"memberships_missing_company", `SELECT COUNT(*) FROM memberships m LEFT JOIN companies c ON c.id = m.company_id WHERE c.id IS NULL`}, {"categories_missing_company", `SELECT COUNT(*) FROM categories x LEFT JOIN companies c ON c.id = x.company_id WHERE c.id IS NULL`}, {"attributes_missing_company", `SELECT COUNT(*) FROM attributes x LEFT JOIN companies c ON c.id = x.company_id WHERE c.id IS NULL`}, {"custom_variables_missing_company", `SELECT COUNT(*) FROM custom_variables x LEFT JOIN companies c ON c.id = x.company_id WHERE c.id IS NULL`}, {"raw_products_missing_company", `SELECT COUNT(*) FROM raw_products r LEFT JOIN companies c ON c.id = r.company_id WHERE c.id IS NULL`}, {"raw_products_missing_feed", `SELECT COUNT(*) FROM raw_products r LEFT JOIN input_feeds f ON f.id = r.feed_id WHERE r.feed_id IS NOT NULL AND f.id IS NULL`}, {"processed_missing_company", `SELECT COUNT(*) FROM processed_products p LEFT JOIN companies c ON c.id = p.company_id WHERE c.id IS NULL`}, {"processed_missing_raw", `SELECT COUNT(*) FROM processed_products p LEFT JOIN raw_products r ON r.id = p.raw_product_id WHERE p.raw_product_id IS NOT NULL AND r.id IS NULL`}, {"processed_missing_feed", `SELECT COUNT(*) FROM processed_products p LEFT JOIN input_feeds f ON f.id = p.feed_id WHERE p.feed_id IS NOT NULL AND f.id IS NULL`}, {"export_feeds_missing_company", `SELECT COUNT(*) FROM export_feeds e LEFT JOIN companies c ON c.id = e.company_id WHERE c.id IS NULL`}, {"export_feeds_missing_source", `SELECT COUNT(*) FROM export_feeds e LEFT JOIN input_feeds f ON f.id = e.source_feed_id WHERE e.source_feed_id IS NOT NULL AND f.id IS NULL`}, {"feed_mappings_missing_feed", `SELECT COUNT(*) FROM feed_mappings m LEFT JOIN input_feeds f ON f.id = m.feed_id WHERE f.id IS NULL`}, {"files_missing_company", `SELECT COUNT(*) FROM files f LEFT JOIN companies c ON c.id = f.company_id WHERE c.id IS NULL`}, {"company_plans_missing_plan", `SELECT COUNT(*) FROM company_plans cp LEFT JOIN plans p ON p.id = cp.plan_id WHERE p.id IS NULL`}, {"companies_without_active_plan", `SELECT COUNT(*) FROM companies c WHERE NOT EXISTS (SELECT 1 FROM company_plans cp WHERE cp.company_id = c.id AND cp.is_active = true)`}, {"platform_admins", `SELECT COUNT(*) FROM users WHERE is_platform_admin = true`}, } out := make([]OrphanFinding, 0, len(checks)) for _, c := range checks { var n int64 err := pg.QueryRow(ctx, c.sql).Scan(&n) f := OrphanFinding{Check: c.name, Count: n, Pass: err == nil} if err != nil { f.Pass = false f.Sample = err.Error() } else if c.name == "platform_admins" { // Informational — not an orphan. f.Pass = true } else if c.name == "companies_without_active_plan" { // Informational cutover signal (use -list-companies-without-plans / -assign-missing-plans). f.Pass = true if n > 0 { f.Sample = fmt.Sprintf("%d companies lack an active plan", n) } } else { f.Pass = n == 0 } out = append(out, f) } return out } func summarizeOrphans(orphans []OrphanFinding) OrphanSummary { s := OrphanSummary{Total: len(orphans)} for _, o := range orphans { if o.Check == "skipped_dry_run" || o.Check == "skipped_fixture" || o.Check == "platform_admins" || o.Check == "companies_without_active_plan" { s.Passed++ continue } if o.Pass { s.Passed++ } else { s.Failed++ } } return s } func printValidation(v ValidationReport) { fmt.Println("=== Validation report ===") fmt.Printf("mode: %s ok=%v\n", v.Mode, v.OK) fmt.Println("-- mysql vs postgres counts --") for _, c := range v.Counts { fmt.Printf("%s mysql=%d postgres=%d delta=%d %s\n", c.Entity, c.MySQL, c.Postgres, c.Delta, c.Note) } if len(v.PostgresCounts) > 0 { fmt.Println("-- postgres table counts --") keys := make([]string, 0, len(v.PostgresCounts)) for k := range v.PostgresCounts { keys = append(keys, k) } sort.Strings(keys) for _, k := range keys { fmt.Printf("%s: %d\n", k, v.PostgresCounts[k]) } } fmt.Println("-- orphans --") names := make([]string, 0, len(v.Orphans)) byName := map[string]OrphanFinding{} for _, o := range v.Orphans { names = append(names, o.Check) byName[o.Check] = o } sort.Strings(names) for _, name := range names { o := byName[name] status := "PASS" if !o.Pass { status = "FAIL" } fmt.Printf("%s %s count=%d %s\n", status, o.Check, o.Count, o.Sample) } fmt.Printf("-- orphan summary -- passed=%d failed=%d total=%d\n", v.OrphanSummary.Passed, v.OrphanSummary.Failed, v.OrphanSummary.Total) } func runValidation( ctx context.Context, mysqlDB *sql.DB, pg *pgxpool.Pool, dryRun bool, ) ValidationReport { mode := "live" if dryRun { mode = "dry-run" } counts := buildCountReport(ctx, mysqlDB, pg, dryRun) pgCounts := collectPostgresCounts(ctx, pg, dryRun) orphans := checkOrphanFKs(ctx, pg, dryRun) summary := summarizeOrphans(orphans) ok := summary.Failed == 0 v := ValidationReport{ Mode: mode, Counts: counts, PostgresCounts: pgCounts, Orphans: orphans, OrphanSummary: summary, OK: ok, } if !ok { log.Printf("validation: orphan FK failures present — inspect report before DNS cutover") } return v }