descrybe/apps/api/cmd/migrator/report.go

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
}