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

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
	_ "github.com/go-sql-driver/mysql"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
	mysqlDSN := flag.String("mysql", os.Getenv("MIGRATE_MYSQL_DSN"), "MySQL DSN (required unless -fixture); also accepts mysql:// URLs")
	postgresURL := flag.String("postgres", os.Getenv("DATABASE_URL"), "Postgres URL")
	dryRun := flag.Bool("dry-run", false, "Count/remap without writing to Postgres")
	mapsDir := flag.String("maps-dir", "maps", "Directory for id-map / validation artifacts (maps/*.json; do not commit)")
	idMapPath := flag.String("id-map", "", "Unified id-map.json path (default: <maps-dir>/id-map.json)")
	reportDir := flag.String("report-dir", "", "JSON report directory (default: <maps-dir>; also copies to docs/migration-reports when present)")
	fixturePath := flag.String("fixture", "", "JSON fixture path for dry-run without MySQL (see testdata/fixture.json)")
	resume := flag.Bool("resume", true, "Reuse existing id-map.json UUIDs for idempotent remaps")
	companyFilter := flag.String("company", "", "Comma-separated legacy company_id filter (empty = all)")
	domains := flag.String("domains", "all", "Comma domains: identity,billing,catalog,feeds,products,files,settings,formulas,tags,woo,usage,jobs (or all)")
	skipPostImport := flag.Bool("skip-post-import", false, "Skip set-password invite hook generation")
	issueSetPassword := flag.Bool("issue-set-password-invites", false, "Create set-password invites for must_set_password users; write password_invites.json and print URLs (works alone with -postgres)")
	setPassword := flag.String("set-password", "", "Local bootstrap: set password for one user as email:password (requires -postgres)")
	ensureDemo := flag.Bool("ensure-demo", false, "Upsert demo user (admin of all companies; rename richest → A1 Slovenija)")
	demoEmail := flag.String("demo-email", "demo@descrybe.local", "Demo user email for -ensure-demo")
	demoPassword := flag.String("demo-password", "DemoPass123!", "Demo user password for -ensure-demo (local only; do not commit)")
	demoName := flag.String("demo-name", "Demo User", "Demo display name")
	localDemoCo := flag.String("local-demo-name", "Platform Demo", "Standalone demo sandbox company when -ensure-demo (never A1)")
	listWithoutPlans := flag.Bool("list-companies-without-plans", false, "List Postgres companies with no active company_plans row (works alone with -postgres)")
	assignMissingPlans := flag.Bool("assign-missing-plans", false, "Assign -plan-name to companies without an active plan (never overwrites existing active plans; works alone with -postgres; requires -dry-run or -confirm)")
	planName := flag.String("plan-name", "Free", "Plan name for -assign-missing-plans (case-insensitive)")
	listMemberMemberships := flag.Bool("list-member-memberships", false, "List active memberships with role=member (Postgres-only; optional -email/-user-id/-company-id filters)")
	promoteCompanyAdmins := flag.Bool("promote-company-admins", false, "Promote matching active member memberships to company admin (requires -email, -user-id, or -company-id; never promotes A1 a1=true; requires -dry-run or -confirm)")
	promoteEmail := flag.String("email", "", "Filter/target user email for -list-member-memberships / -promote-company-admins")
	promoteUserID := flag.String("user-id", "", "Filter/target Postgres user UUID for membership role tooling")
	promoteCompanyID := flag.String("company-id", "", "Filter/target Postgres company UUID for membership role tooling (alone is enough to scope -promote-company-admins; A1 still skipped)")
	confirmAssign := flag.Bool("confirm", false, "Required for live -assign-missing-plans / -promote-company-admins / -patch-emails writes (omit with -dry-run to preview only; no blind live writes)")
	fallbackPlanName := flag.String("fallback-plan-name", "", "During ETL: when legacy company_plans plan_id is missing from plans, assign this Postgres plan name instead of skipping (e.g. Free)")
	listLegacyEmails := flag.Bool("list-legacy-emails", false, "List Postgres users with synthetic @legacy.local emails (works alone with -postgres; read-only)")
	exportLegacyEmails := flag.Bool("export-legacy-emails", false, "Export @legacy.local inventory + emails stub map to -emails-out / maps-dir (works alone with -postgres; read-only)")
	patchEmails := flag.Bool("patch-emails", false, "Patch users.email from -emails-file for rows that still end with @legacy.local (never overwrites real emails; requires -dry-run or -confirm)")
	emailsFile := flag.String("emails-file", "", "Clerk export or emails map (JSON/CSV) for -patch-emails")
	emailsOut := flag.String("emails-out", "", "Output path for -export-legacy-emails (default: <maps-dir>/legacy-emails.json)")
	flag.Parse()

	if *setPassword != "" {
		runSetPasswordOnly(*postgresURL, *setPassword)
		return
	}
	if *issueSetPassword && *fixturePath == "" && *mysqlDSN == "" {
		runIssueSetPasswordInvites(*postgresURL, *mapsDir, *dryRun)
		return
	}
	if *listWithoutPlans || *assignMissingPlans {
		runCompaniesWithoutPlansRepair(*postgresURL, *planName, *listWithoutPlans, *assignMissingPlans, *dryRun, *confirmAssign)
		return
	}
	if *listMemberMemberships || *promoteCompanyAdmins {
		runMembershipRoleRepair(*postgresURL, *promoteEmail, *promoteUserID, *promoteCompanyID, *listMemberMemberships, *promoteCompanyAdmins, *dryRun, *confirmAssign)
		return
	}
	if *listLegacyEmails || *exportLegacyEmails || *patchEmails {
		runLegacyEmailTools(*postgresURL, *mapsDir, *emailsFile, *emailsOut, *listLegacyEmails, *exportLegacyEmails, *patchEmails, *dryRun, *confirmAssign)
		return
	}
	if *fixturePath == "" && *mysqlDSN == "" {
		log.Fatal("BLOCKER: set -mysql / MIGRATE_MYSQL_DSN for real cutover validation, or pass -fixture testdata/fixture.json for offline dry-run; or use -issue-set-password-invites / -set-password / -list-companies-without-plans / -assign-missing-plans / -list-member-memberships / -promote-company-admins / -list-legacy-emails / -export-legacy-emails / -patch-emails with -postgres")
	}
	if *postgresURL == "" && !*dryRun {
		log.Fatal("-postgres / DATABASE_URL is required for live loads (omit only with -dry-run + -fixture)")
	}
	if *dryRun && *fixturePath != "" && *postgresURL == "" {
		runFixtureDryRun(*fixturePath, *mapsDir, *idMapPath)
		return
	}
	if *postgresURL == "" {
		log.Fatal("-postgres / DATABASE_URL is required")
	}

	ctx := context.Background()

	var mysqlDB *sql.DB
	if *fixturePath != "" {
		log.Fatal("fixture mode only supports offline -dry-run without -postgres (omit DATABASE_URL); for live loads use real -mysql")
	}
	normalizedMySQL, err := normalizeMySQLDSN(*mysqlDSN)
	if err != nil {
		log.Fatalf("mysql DSN: %v\n%s", err, mysqlDSNHelp())
	}
	db, err := sql.Open("mysql", normalizedMySQL)
	if err != nil {
		log.Fatalf("mysql open (%s): %v\n%s", maskMySQLDSN(normalizedMySQL), err, mysqlDSNHelp())
	}
	mysqlDB = db
	defer mysqlDB.Close()
	mysqlDB.SetConnMaxLifetime(time.Minute)
	if err := mysqlDB.PingContext(ctx); err != nil {
		log.Fatalf("mysql ping (%s): %v\n%s", maskMySQLDSN(normalizedMySQL), err, mysqlDSNHelp())
	}

	pg, err := pgxpool.New(ctx, *postgresURL)
	if err != nil {
		log.Fatalf("postgres: %v", err)
	}
	defer pg.Close()

	if err := os.MkdirAll(*mapsDir, 0o755); err != nil {
		log.Fatalf("maps dir: %v", err)
	}
	outIDMap := *idMapPath
	if outIDMap == "" {
		outIDMap = filepath.Join(*mapsDir, "id-map.json")
	}
	cfg := MigratorConfig{
		MySQLDSN:       *mysqlDSN,
		PostgresURL:    *postgresURL,
		DryRun:         *dryRun,
		MapsDir:        *mapsDir,
		IDMapPath:      outIDMap,
		ReportDir:      *reportDir,
		Resume:         *resume,
		CompanyFilter:  parseCompanyFilter(*companyFilter),
		Domains:        parseDomains(*domains),
		SkipPostImport: *skipPostImport,
		EnsureDemo:     *ensureDemo,
		DemoEmail:      *demoEmail,
		DemoPassword:   *demoPassword,
		DemoName:       *demoName,
		LocalDemoCo:    *localDemoCo,
	}
	if cfg.ReportDir == "" {
		cfg.ReportDir = *mapsDir
	}
	runReport := newRunReport(cfg, *dryRun)
	started := time.Now()
	log.Printf("migrator domains=%s company_filter=%v resume=%v dry_run=%v", cfg.Domains, cfg.CompanyFilter, cfg.Resume, *dryRun)

	report := map[string]int{}
	companyMap := map[string]string{}
	userMap := map[string]string{}
	resumeCategories := map[string]string{}
	resumeAttributes := map[string]string{}
	resumeFeeds := map[string]string{}
	resumeRaw := map[string]string{}
	resumeFiles := map[string]string{}
	if cfg.Resume {
		if _, err := os.Stat(outIDMap); err == nil {
			if loaded, err := ReadIDMap(outIDMap); err == nil {
				for k, v := range loaded.Companies {
					companyMap[k] = v
				}
				for k, v := range loaded.Users {
					userMap[k] = v
				}
				for k, v := range loaded.Categories {
					resumeCategories[k] = v
				}
				for k, v := range loaded.Attributes {
					resumeAttributes[k] = v
				}
				for k, v := range loaded.Feeds {
					resumeFeeds[k] = v
				}
				for k, v := range loaded.RawProducts {
					resumeRaw[k] = v
				}
				for k, v := range loaded.Files {
					resumeFiles[k] = v
				}
				log.Printf("reusing id map %s for idempotent remaps (companies=%d feeds=%d)", outIDMap, len(companyMap), len(resumeFeeds))
			}
		}
	} else {
		log.Printf("resume disabled: generating fresh UUIDs (still upserts by natural keys)")
	}

	allowCompanies := companyFilterSet(cfg.CompanyFilter)

	companies, err := loadCompanies(ctx, mysqlDB)
	if err != nil {
		log.Fatalf("load companies: %v", err)
	}
	companies = filterCompanies(companies, allowCompanies)
	if len(cfg.CompanyFilter) > 0 && len(companies) == 0 {
		log.Fatalf("company filter matched 0 companies: %v", cfg.CompanyFilter)
	}
	for _, c := range companies {
		newID := uuid.New()
		if existing, ok := companyMap[c.LegacyID]; ok {
			if parsed, err := uuid.Parse(existing); err == nil {
				newID = parsed
			}
		}
		// Always prefer the live Postgres row when present so gap-only domains
		// and drifted id-maps still resolve FKs correctly.
		if !*dryRun {
			var existing uuid.UUID
			if err := pg.QueryRow(ctx, `SELECT id FROM companies WHERE legacy_company_id = $1`, c.LegacyID).Scan(&existing); err == nil && existing != uuid.Nil {
				newID = existing
			}
		}
		companyMap[c.LegacyID] = newID.String()
		if !*dryRun && cfg.Domains.has("identity") {
			_, err = pg.Exec(ctx, `
				INSERT INTO companies (id, name, language, legacy_company_id, created_at, updated_at)
				VALUES ($1, $2, $3, $4, now(), now())
				ON CONFLICT (legacy_company_id) DO UPDATE SET name = EXCLUDED.name
				RETURNING id`, newID, c.Name, c.Language, c.LegacyID)
			if err != nil {
				log.Printf("company insert %s: %v", c.LegacyID, err)
			}
			var existing uuid.UUID
			if err2 := pg.QueryRow(ctx, `SELECT id FROM companies WHERE legacy_company_id = $1`, c.LegacyID).Scan(&existing); err2 == nil && existing != uuid.Nil {
				companyMap[c.LegacyID] = existing.String()
			}
		}
		report["companies"]++
	}

	users, err := loadUsers(ctx, mysqlDB)
	if err != nil {
		log.Fatalf("load users: %v", err)
	}
	for _, u := range users {
		newID := uuid.New()
		if existing, ok := userMap[u.LegacyID]; ok {
			if parsed, err := uuid.Parse(existing); err == nil {
				newID = parsed
			}
		}
		userMap[u.LegacyID] = newID.String()
		if !*dryRun && cfg.Domains.has("identity") {
			// SECURITY: never import legacy password hashes; force must_set_password.
			_, err = pg.Exec(ctx, `
				INSERT INTO users (id, email, name, password_hash, must_set_password, is_active, legacy_user_id, created_at, updated_at)
				VALUES ($1, $2, $3, NULL, true, $4, $5, now(), now())
				ON CONFLICT (email) DO UPDATE SET
					legacy_user_id = COALESCE(users.legacy_user_id, EXCLUDED.legacy_user_id),
					password_hash = NULL,
					must_set_password = true`,
				newID, strings.ToLower(u.Email), nullStr(u.Name), u.Active, u.LegacyID)
			if err != nil {
				log.Printf("user insert %s: %v", u.Email, err)
			}
			var existing uuid.UUID
			if err := pg.QueryRow(ctx, `SELECT id FROM users WHERE legacy_user_id = $1 OR email = $2`, u.LegacyID, strings.ToLower(u.Email)).Scan(&existing); err == nil {
				userMap[u.LegacyID] = existing.String()
			}
		}
		report["users"]++
	}

	if cfg.Domains.has("identity") {
		applyPlatformAdmins(ctx, mysqlDB, pg, userMap, report, *dryRun)
	}

	memberships, err := loadMemberships(ctx, mysqlDB)
	if err != nil {
		log.Fatalf("load memberships: %v", err)
	}
	for _, m := range memberships {
		if allowCompanies != nil && !allowCompanies[m.CompanyLegacy] {
			continue
		}
		cid, okC := companyMap[m.CompanyLegacy]
		uid, okU := userMap[m.UserLegacy]
		if !okC || !okU {
			report["memberships_skipped"]++
			continue
		}
		if !*dryRun && cfg.Domains.has("identity") {
			_, err = pg.Exec(ctx, `
				INSERT INTO memberships (company_id, user_id, role, status)
				VALUES ($1, $2, $3, $4)
				ON CONFLICT (company_id, user_id) DO UPDATE SET role = EXCLUDED.role, status = EXCLUDED.status`,
				cid, uid, m.Role, m.Status)
			if err != nil {
				log.Printf("membership: %v", err)
			}
		}
		report["memberships"]++
	}

	if !cfg.Domains.has("billing") {
		log.Printf("billing domain skipped")
	} else {
		plans, err := loadPlans(ctx, mysqlDB)
		if err != nil {
			log.Printf("plans skipped: %v", err)
		} else {
			planMap := map[int64]int64{}
			for _, pl := range plans {
				if !*dryRun {
					var newID int64
					err = pg.QueryRow(ctx, `
					INSERT INTO plans (name, description, monthly_credits, yearly_credits, max_products, is_custom, term)
					VALUES ($1, $2, $3, $4, $5, $6, $7)
					RETURNING id`, pl.Name, pl.Description, pl.MonthlyCredits, pl.YearlyCredits, pl.MaxProducts, pl.IsCustom, pl.Term).Scan(&newID)
					if err != nil {
						log.Printf("plan: %v", err)
					} else {
						planMap[pl.LegacyID] = newID
					}
				} else {
					// Dry-run: keep legacy ids so company_plans linkage can be counted.
					planMap[pl.LegacyID] = pl.LegacyID
				}
				report["plans"]++
			}

			var fallbackPlanID int64
			if name := strings.TrimSpace(*fallbackPlanName); name != "" {
				if *dryRun {
					// Dry-run: treat fallback as available so linkage can be counted.
					fallbackPlanID = -1
					log.Printf("company_plans: dry-run fallback plan name %q (would resolve on live load)", name)
				} else {
					id, err := resolveFallbackPlanID(ctx, pg, name)
					if err != nil {
						log.Fatalf("fallback-plan-name %q: %v", name, err)
					}
					fallbackPlanID = id
					log.Printf("company_plans: fallback plan %q -> id=%d", name, fallbackPlanID)
				}
			}

			cps, err := loadCompanyPlans(ctx, mysqlDB)
			if err == nil {
				for _, cp := range cps {
					if billing.IsA1CohortCompany(cp.CompanyLegacy, "") {
						report["company_plans_skipped_a1"]++
						log.Printf("company_plans: skip A1 cohort company_legacy=%s (preserve PAYG/Legacy; use EnsureLegacyDefaults)", cp.CompanyLegacy)
						continue
					}
					cid, ok := companyMap[cp.CompanyLegacy]
					pid, okP := planMap[cp.PlanLegacy]
					if !ok {
						report["company_plans_skipped"]++
						log.Printf("company_plans: skip company_legacy=%s plan_id=%d (company not remapped)", cp.CompanyLegacy, cp.PlanLegacy)
						continue
					}
					if !okP {
						if fallbackPlanID != 0 {
							if fallbackPlanID > 0 {
								pid = fallbackPlanID
							}
							okP = true
							report["company_plans_fallback"]++
							log.Printf("company_plans: fallback for company_legacy=%s missing plan_id=%d -> plan %q", cp.CompanyLegacy, cp.PlanLegacy, strings.TrimSpace(*fallbackPlanName))
						} else {
							report["company_plans_skipped"]++
							log.Printf("company_plans: skip company_legacy=%s plan_id=%d (not in plans map; use -fallback-plan-name or -assign-missing-plans)", cp.CompanyLegacy, cp.PlanLegacy)
							continue
						}
					}
					if !*dryRun {
						_, err = pg.Exec(ctx, `
						INSERT INTO company_plans (company_id, plan_id, is_active, billing_cycle_start, next_billing_date, is_trial, trial_credits)
						VALUES ($1, $2, $3, $4, $5, $6, $7)`,
							cid, pid, cp.IsActive, cp.BillingStart, cp.NextBilling, cp.IsTrial, cp.TrialCredits)
						if err != nil {
							log.Printf("company_plan: %v", err)
						}
					}
					report["company_plans"]++
				}
			}
		}

		balances, err := loadCreditBalances(ctx, mysqlDB)
		if err != nil {
			log.Printf("credit_balances skipped: %v", err)
		} else {
			for _, b := range balances {
				cid, ok := companyMap[b.CompanyLegacy]
				if !ok {
					report["credit_balances_skipped"]++
					continue
				}
				if !*dryRun {
					_, err = pg.Exec(ctx, `
					INSERT INTO credit_balances (company_id, total_credits, used_credits, updated_at)
					VALUES ($1, $2, $3, now())
					ON CONFLICT (company_id) DO UPDATE SET
						total_credits = EXCLUDED.total_credits,
						used_credits = EXCLUDED.used_credits,
						updated_at = now()`, cid, b.Total, b.Used)
					if err != nil {
						log.Printf("credit_balance: %v", err)
					}
				}
				report["credit_balances"]++
			}
		}

	} // end billing domain

	categoryMap, attributeMap, feedMap, rawMap, fileMap := migrateCatalogAndFeeds(ctx, mysqlDB, pg, companyMap, userMap, allowCompanies, cfg.Domains, report, *dryRun)
	for k, v := range resumeCategories {
		if _, ok := categoryMap[k]; !ok {
			categoryMap[k] = v
		}
	}
	for k, v := range resumeAttributes {
		if _, ok := attributeMap[k]; !ok {
			attributeMap[k] = v
		}
	}
	for k, v := range resumeFeeds {
		if _, ok := feedMap[k]; !ok {
			feedMap[k] = v
		}
	}
	for k, v := range resumeRaw {
		if _, ok := rawMap[k]; !ok {
			rawMap[k] = v
		}
	}
	for k, v := range resumeFiles {
		if _, ok := fileMap[k]; !ok {
			fileMap[k] = v
		}
	}
	_ = attributeMap

	migrateGapDomains(ctx, mysqlDB, pg, companyMap, feedMap, allowCompanies, cfg.Domains, report, *dryRun)
	migrateJobsDomain(ctx, mysqlDB, pg, companyMap, userMap, rawMap, allowCompanies, cfg.Domains, report, *dryRun)

	if !*skipPostImport || *issueSetPassword {
		hooks, err := prepareSetPasswordHooks(ctx, pg, 7*24*time.Hour, *dryRun, report)
		if err != nil {
			log.Printf("set-password hooks: %v", err)
		} else if err := writeSetPasswordArtifacts(*mapsDir, hooks); err != nil {
			log.Printf("write set-password hooks: %v", err)
		}
	}

	validation := runValidation(ctx, mysqlDB, pg, *dryRun)
	printValidation(validation)
	if err := writeJSON(filepath.Join(*mapsDir, "validation-report.json"), validation); err != nil {
		log.Printf("validation report: %v", err)
	}

	idDoc := NewIDMapDocument(userMap, companyMap, *dryRun)
	idDoc.AttachEntityMaps(categoryMap, attributeMap, feedMap, rawMap, fileMap, report)
	if err := WriteIDMap(outIDMap, idDoc); err != nil {
		log.Fatalf("id-map: %v", err)
	}
	writeEntityMapFiles(*mapsDir, companyMap, userMap, categoryMap, attributeMap, feedMap, rawMap, fileMap)

	if cfg.EnsureDemo {
		demoRep, err := ensureDemoUser(ctx, pg, cfg.DemoEmail, cfg.DemoPassword, cfg.DemoName, cfg.LocalDemoCo, *dryRun, report)
		if err != nil {
			log.Printf("ensure-demo: %v", err)
		} else {
			runReport.Demo = demoRep
			fmt.Printf("demo user: %s (password set locally; see docs/portable-mysql-pg-migration.md)\n", cfg.DemoEmail)
			if demoRep != nil && demoRep.PrimaryName != "" {
				fmt.Printf("demo primary company: %s (%s)\n", demoRep.PrimaryName, demoRep.PrimaryCompany)
			}
		}
	}

	runReport.Counts = report
	runReport.Validation = validation
	runReport.ElapsedMS = time.Since(started).Milliseconds()
	if err := writeMigrationReports(cfg.ReportDir, runReport); err != nil {
		log.Printf("migration report: %v", err)
	}

	printMigrationReport(report, *dryRun)
	fmt.Printf("wrote maps under %s (unified: %s)\n", *mapsDir, outIDMap)
	fmt.Printf("wrote migration report under %s\n", cfg.ReportDir)
	if *fixturePath != "" {
		fmt.Println("BLOCKER: fixture mode is not a substitute for dry-run against production MySQL — set MIGRATE_MYSQL_DSN and re-run before cutover.")
	}
}

func writeMigrationReports(reportDir string, rep *MigrationRunReport) error {
	if reportDir == "" || rep == nil {
		return nil
	}
	if err := os.MkdirAll(reportDir, 0o755); err != nil {
		return err
	}
	stamp := time.Now().UTC().Format("20060102T150405Z")
	primary := filepath.Join(reportDir, "migration-report.json")
	stamped := filepath.Join(reportDir, "migration-report-"+stamp+".json")
	if err := writeJSON(primary, rep); err != nil {
		return err
	}
	_ = writeJSON(stamped, rep)
	// Optional copy under docs/migration-reports (repo-relative from apps/api).
	docsDir := filepath.Clean(filepath.Join("..", "..", "docs", "migration-reports"))
	if st, err := os.Stat(docsDir); err == nil && st.IsDir() {
		_ = writeJSON(filepath.Join(docsDir, "migration-report-latest.json"), rep)
		_ = writeJSON(filepath.Join(docsDir, "migration-report-"+stamp+".json"), rep)
	}
	return nil
}

func writeEntityMapFiles(
	mapsDir string,
	companyMap, userMap, categoryMap, attributeMap, feedMap, rawMap, fileMap map[string]string,
) {
	if companyMap == nil {
		companyMap = map[string]string{}
	}
	if userMap == nil {
		userMap = map[string]string{}
	}
	if categoryMap == nil {
		categoryMap = map[string]string{}
	}
	if attributeMap == nil {
		attributeMap = map[string]string{}
	}
	if feedMap == nil {
		feedMap = map[string]string{}
	}
	if rawMap == nil {
		rawMap = map[string]string{}
	}
	if fileMap == nil {
		fileMap = map[string]string{}
	}
	_ = writeJSON(filepath.Join(mapsDir, "company_map.json"), companyMap)
	_ = writeJSON(filepath.Join(mapsDir, "user_map.json"), userMap)
	_ = writeJSON(filepath.Join(mapsDir, "category_map.json"), categoryMap)
	_ = writeJSON(filepath.Join(mapsDir, "attribute_map.json"), attributeMap)
	_ = writeJSON(filepath.Join(mapsDir, "feed_map.json"), feedMap)
	_ = writeJSON(filepath.Join(mapsDir, "raw_product_map.json"), rawMap)
	_ = writeJSON(filepath.Join(mapsDir, "file_map.json"), fileMap)
}

func printMigrationReport(report map[string]int, dryRun bool) {
	fmt.Println("=== Migration report ===")
	if dryRun {
		fmt.Println("mode: dry-run")
	}
	keys := make([]string, 0, len(report))
	for k := range report {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, k := range keys {
		fmt.Printf("%s: %d\n", k, report[k])
	}
}

func runSetPasswordOnly(postgresURL, emailPass string) {
	if postgresURL == "" {
		log.Fatal("-postgres / DATABASE_URL is required for -set-password")
	}
	ctx := context.Background()
	pg, err := pgxpool.New(ctx, postgresURL)
	if err != nil {
		log.Fatalf("postgres: %v", err)
	}
	defer pg.Close()
	if err := setPasswordByEmail(ctx, pg, emailPass); err != nil {
		log.Fatal(err)
	}
}

func runIssueSetPasswordInvites(postgresURL, mapsDir string, dryRun bool) {
	if postgresURL == "" {
		log.Fatal("-postgres / DATABASE_URL is required for -issue-set-password-invites")
	}
	ctx := context.Background()
	pg, err := pgxpool.New(ctx, postgresURL)
	if err != nil {
		log.Fatalf("postgres: %v", err)
	}
	defer pg.Close()
	if err := os.MkdirAll(mapsDir, 0o755); err != nil {
		log.Fatalf("maps dir: %v", err)
	}
	report := map[string]int{}
	hooks, err := prepareSetPasswordHooks(ctx, pg, 7*24*time.Hour, dryRun, report)
	if err != nil {
		log.Fatalf("set-password invites: %v", err)
	}
	if dryRun {
		fmt.Println("mode: dry-run (no invites written)")
		return
	}
	if len(hooks) == 0 {
		fmt.Println("no active users with must_set_password=true (and a membership)")
		return
	}
	if err := writeSetPasswordArtifacts(mapsDir, hooks); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("set_password_hooks: %d\n", report["set_password_hooks"])
	if skipped := report["set_password_hooks_skipped"]; skipped > 0 {
		fmt.Printf("set_password_hooks_skipped: %d\n", skipped)
	}
}

type companyRow struct {
	LegacyID, Name, Language string
}

type userRow struct {
	LegacyID, Email, Name string
	Active                bool
}

type membershipRow struct {
	CompanyLegacy, UserLegacy, Role, Status string
}

type planRow struct {
	LegacyID                      int64
	Name, Description, Term       string
	MonthlyCredits, YearlyCredits int
	MaxProducts                   *int
	IsCustom                      bool
}

type companyPlanRow struct {
	CompanyLegacy string
	PlanLegacy    int64
	IsActive      bool
	BillingStart  time.Time
	NextBilling   time.Time
	IsTrial       bool
	TrialCredits  int
}

type balanceRow struct {
	CompanyLegacy string
	Total, Used   int
}

func loadCompanies(ctx context.Context, db *sql.DB) ([]companyRow, error) {
	// Legacy companies has no language column (lives on company_settings when present).
	var queries []string
	if mysqlTableExists(ctx, db, "companies") {
		nameExpr := mysqlCoalesce(ctx, db, "companies", "name", "id")
		if mysqlTableExists(ctx, db, "company_settings") && mysqlColumnExists(ctx, db, "company_settings", "language") {
			queries = append(queries, fmt.Sprintf(`
				SELECT c.id, %s, COALESCE(cs.language, 'en')
				FROM companies c
				LEFT JOIN company_settings cs ON cs.company_id = c.id`, nameExpr))
		}
		queries = append(queries, fmt.Sprintf(`SELECT id, %s, 'en' FROM companies`, nameExpr))
		if mysqlColumnExists(ctx, db, "companies", "language") {
			queries = append([]string{fmt.Sprintf(
				`SELECT id, %s, %s FROM companies`,
				nameExpr, mysqlCoalesce(ctx, db, "companies", "language", "'en'"),
			)}, queries...)
		}
	}
	queries = append(queries,
		`SELECT DISTINCT company_id, COALESCE(MAX(company_name), company_id), 'en'
		 FROM profiles WHERE company_id IS NOT NULL AND company_id <> ''
		 GROUP BY company_id`,
		`SELECT DISTINCT company_id, company_id, 'en' FROM profiles
		 WHERE company_id IS NOT NULL AND company_id <> ''`,
	)
	rows, err := queryFirstOK(ctx, db, queries...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []companyRow
	for rows.Next() {
		var c companyRow
		if err := rows.Scan(&c.LegacyID, &c.Name, &c.Language); err != nil {
			return nil, err
		}
		out = append(out, c)
	}
	return out, rows.Err()
}

func loadUsers(ctx context.Context, db *sql.DB) ([]userRow, error) {
	// Prefer users.email (joined with profiles for Clerk legacy id); fall back to profiles.
	// Many legacy dumps have no users table and profiles without email/name (Clerk-only identity).
	if mysqlTableExists(ctx, db, "users") && mysqlColumnExists(ctx, db, "users", "email") {
		q := `
			SELECT
				COALESCE(NULLIF(p.user_id, ''), u.id) AS legacy_id,
				u.email,
				COALESCE(u.name, ''),
				CASE WHEN COALESCE(u.is_active, 1) = 0 THEN 0 ELSE 1 END
			FROM users u
			LEFT JOIN profiles p ON p.user_id = u.id
			WHERE u.email IS NOT NULL AND TRIM(u.email) <> ''`
		rows, err := db.QueryContext(ctx, q)
		if err != nil {
			log.Printf("loadUsers users+profiles failed (%v); trying users alone", err)
			rows, err = db.QueryContext(ctx, `
				SELECT id, email, COALESCE(name, ''),
					CASE WHEN COALESCE(is_active, 1) = 0 THEN 0 ELSE 1 END
				FROM users
				WHERE email IS NOT NULL AND TRIM(email) <> ''`)
		}
		if err == nil {
			out, err2 := scanUserRows(rows)
			rows.Close()
			if err2 != nil {
				return nil, err2
			}
			if len(out) > 0 {
				return enrichUserEmailsFromAdminUsers(ctx, db, out), nil
			}
		} else {
			log.Printf("loadUsers users table failed (%v); falling back to profiles", err)
		}
	}

	if !mysqlTableExists(ctx, db, "profiles") {
		return nil, fmt.Errorf("neither users nor profiles table found")
	}

	emailExpr := `CONCAT(user_id, '@legacy.local')`
	if mysqlColumnExists(ctx, db, "profiles", "email") {
		emailExpr = `COALESCE(NULLIF(TRIM(email), ''), CONCAT(user_id, '@legacy.local'))`
	} else {
		log.Printf("profiles.email missing; using synthetic @legacy.local addresses (enrich from admin_users when present)")
	}
	nameExpr := `''`
	if mysqlColumnExists(ctx, db, "profiles", "name") {
		nameExpr = `COALESCE(name, '')`
	}
	activeExpr := `1`
	if mysqlColumnExists(ctx, db, "profiles", "status") {
		activeExpr = `CASE WHEN COALESCE(status, 'active') = 'inactive' THEN 0 ELSE 1 END`
	}

	q := fmt.Sprintf(`
		SELECT user_id, %s, %s, %s
		FROM profiles
		WHERE user_id IS NOT NULL AND user_id <> ''`, emailExpr, nameExpr, activeExpr)
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out, err := scanUserRows(rows)
	if err != nil {
		return nil, err
	}
	return enrichUserEmailsFromAdminUsers(ctx, db, out), nil
}

// enrichUserEmailsFromAdminUsers overlays real emails from admin_users onto Clerk-id users.
func enrichUserEmailsFromAdminUsers(ctx context.Context, db *sql.DB, users []userRow) []userRow {
	if len(users) == 0 || !mysqlTableExists(ctx, db, "admin_users") {
		return users
	}
	hasUserID := mysqlColumnExists(ctx, db, "admin_users", "user_id")
	hasEmail := mysqlColumnExists(ctx, db, "admin_users", "email")
	if !hasUserID || !hasEmail {
		return users
	}
	rows, err := db.QueryContext(ctx, `
		SELECT user_id, email FROM admin_users
		WHERE user_id IS NOT NULL AND user_id <> ''
		  AND email IS NOT NULL AND TRIM(email) <> ''`)
	if err != nil {
		log.Printf("admin_users email enrich skipped: %v", err)
		return users
	}
	defer rows.Close()
	byLegacy := map[string]string{}
	for rows.Next() {
		var id, email string
		if err := rows.Scan(&id, &email); err != nil {
			continue
		}
		byLegacy[id] = strings.ToLower(strings.TrimSpace(email))
	}
	if len(byLegacy) == 0 {
		return users
	}
	for i := range users {
		if email, ok := byLegacy[users[i].LegacyID]; ok {
			users[i].Email = email
		}
	}
	return users
}

func scanUserRows(rows *sql.Rows) ([]userRow, error) {
	seen := map[string]bool{}
	var out []userRow
	for rows.Next() {
		var u userRow
		var active int
		if err := rows.Scan(&u.LegacyID, &u.Email, &u.Name, &active); err != nil {
			return nil, err
		}
		if seen[u.LegacyID] {
			continue
		}
		seen[u.LegacyID] = true
		u.Active = active == 1
		out = append(out, u)
	}
	return out, rows.Err()
}

func loadMemberships(ctx context.Context, db *sql.DB) ([]membershipRow, error) {
	if !mysqlTableExists(ctx, db, "profiles") {
		return nil, fmt.Errorf("profiles table missing")
	}
	// Legacy profiles often omit role (Clerk held org roles). Default to member.
	roleExpr := `'member'`
	if mysqlColumnExists(ctx, db, "profiles", "role") {
		roleExpr = `CASE WHEN COALESCE(role, 'member') IN ('admin', 'org:admin') THEN 'admin' ELSE 'member' END`
	} else {
		log.Printf("profiles.role missing; defaulting all memberships to role=member (promote admins after cutover)")
	}
	statusExpr := `'active'`
	if mysqlColumnExists(ctx, db, "profiles", "status") {
		statusExpr = `CASE WHEN COALESCE(status, 'active') = 'inactive' THEN 'inactive' ELSE 'active' END`
	}
	q := fmt.Sprintf(`
		SELECT company_id, user_id, %s, %s
		FROM profiles
		WHERE company_id IS NOT NULL AND company_id <> ''
		  AND user_id IS NOT NULL AND user_id <> ''`, roleExpr, statusExpr)
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []membershipRow
	for rows.Next() {
		var m membershipRow
		if err := rows.Scan(&m.CompanyLegacy, &m.UserLegacy, &m.Role, &m.Status); err != nil {
			return nil, err
		}
		out = append(out, m)
	}
	return out, rows.Err()
}

func normalizeMySQLDSN(dsn string) (string, error) {
	dsn = strings.TrimSpace(dsn)
	if dsn == "" {
		return "", fmt.Errorf("empty DSN")
	}
	if !strings.Contains(dsn, "://") {
		return ensureParseTime(dsn), nil
	}
	u, err := url.Parse(dsn)
	if err != nil {
		return "", fmt.Errorf("parse URL DSN: %w", err)
	}
	scheme := strings.ToLower(u.Scheme)
	if scheme != "mysql" && scheme != "mariadb" {
		return "", fmt.Errorf("unsupported scheme %q (want mysql:// or user:pass@tcp(...)/db)", u.Scheme)
	}
	user := ""
	pass := ""
	if u.User != nil {
		user = u.User.Username()
		pass, _ = u.User.Password()
	}
	host := u.Hostname()
	if host == "" {
		host = "127.0.0.1"
	}
	port := u.Port()
	if port == "" {
		port = "3306"
	}
	dbName := strings.TrimPrefix(u.Path, "/")
	if dbName == "" {
		return "", fmt.Errorf("database name missing in DSN path")
	}
	out := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, pass, host, port, dbName)
	q := u.RawQuery
	if q == "" {
		q = "parseTime=true&charset=utf8mb4"
	} else if !strings.Contains(q, "parseTime=") {
		q += "&parseTime=true"
	}
	return out + "?" + q, nil
}

func ensureParseTime(dsn string) string {
	if strings.Contains(dsn, "parseTime=") {
		return dsn
	}
	if strings.Contains(dsn, "?") {
		return dsn + "&parseTime=true"
	}
	return dsn + "?parseTime=true"
}

func maskMySQLDSN(dsn string) string {
	// user:pass@tcp(...) → user:****@tcp(...)
	if at := strings.Index(dsn, "@"); at > 0 {
		cred := dsn[:at]
		if colon := strings.Index(cred, ":"); colon >= 0 {
			return cred[:colon+1] + "****" + dsn[at:]
		}
	}
	// URL form fallback
	if u, err := url.Parse(dsn); err == nil && u.User != nil {
		u.User = url.UserPassword(u.User.Username(), "****")
		return u.String()
	}
	return dsn
}

func mysqlDSNHelp() string {
	return strings.TrimSpace(`
Hints:
  - Start Laragon MySQL (often root@127.0.0.1:3306) and confirm the DB exists.
  - Preferred DSN: user:pass@tcp(127.0.0.1:3306)/dbname?parseTime=true
  - mysql://user:pass@host:3306/dbname URLs are accepted and converted.
  - Set MIGRATE_MYSQL_DSN or pass -mysql; do not commit credentials.`)
}

func loadPlans(ctx context.Context, db *sql.DB) ([]planRow, error) {
	if !mysqlTableExists(ctx, db, "plans") {
		return nil, fmt.Errorf("plans table missing")
	}
	q := mysqlSelectList(
		"id",
		"name",
		mysqlCoalesce(ctx, db, "plans", "description", "NULL"),
		mysqlCoalesce(ctx, db, "plans", "monthly_credits", "0"),
		mysqlCoalesce(ctx, db, "plans", "yearly_credits", "0"),
		mysqlCol(ctx, db, "plans", "max_products", "NULL"),
		mysqlCoalesce(ctx, db, "plans", "is_custom", "0"),
		mysqlCoalesce(ctx, db, "plans", "term", "'monthly'"),
	) + " FROM plans"
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		// Minimal shape fallback.
		rows, err = db.QueryContext(ctx, `SELECT id, name, NULL, monthly_credits, 0, NULL, 0, 'monthly' FROM plans`)
		if err != nil {
			return nil, err
		}
	}
	defer rows.Close()
	var out []planRow
	for rows.Next() {
		var p planRow
		var custom int
		var desc sql.NullString
		if err := rows.Scan(&p.LegacyID, &p.Name, &desc, &p.MonthlyCredits, &p.YearlyCredits, &p.MaxProducts, &custom, &p.Term); err != nil {
			return nil, err
		}
		if desc.Valid {
			p.Description = desc.String
		}
		p.IsCustom = custom == 1
		if p.Term == "" {
			p.Term = "monthly"
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

func loadCompanyPlans(ctx context.Context, db *sql.DB) ([]companyPlanRow, error) {
	if !mysqlTableExists(ctx, db, "company_plans") {
		return nil, fmt.Errorf("company_plans table missing")
	}
	// Legacy plan_id is TEXT storing numeric plan ids — scan as string then parse.
	q := mysqlSelectList(
		"company_id",
		"CAST(plan_id AS CHAR)",
		mysqlCoalesce(ctx, db, "company_plans", "is_active", "1"),
		mysqlCol(ctx, db, "company_plans", "billing_cycle_start", "NOW()"),
		mysqlCol(ctx, db, "company_plans", "next_billing_date", "NOW()"),
		mysqlCoalesce(ctx, db, "company_plans", "is_trial", "0"),
		mysqlCoalesce(ctx, db, "company_plans", "trial_credits", "0"),
	) + " FROM company_plans"
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		rows, err = db.QueryContext(ctx, `
			SELECT company_id, CAST(plan_id AS CHAR), 1, NOW(), NOW(), 0, 0 FROM company_plans`)
		if err != nil {
			return nil, err
		}
	}
	defer rows.Close()
	var out []companyPlanRow
	for rows.Next() {
		var cp companyPlanRow
		var planIDStr string
		var active, trial int
		if err := rows.Scan(&cp.CompanyLegacy, &planIDStr, &active, &cp.BillingStart, &cp.NextBilling, &trial, &cp.TrialCredits); err != nil {
			return nil, err
		}
		pid, err := strconv.ParseInt(strings.TrimSpace(planIDStr), 10, 64)
		if err != nil {
			log.Printf("company_plans: skip non-numeric plan_id %q", planIDStr)
			continue
		}
		cp.PlanLegacy = pid
		cp.IsActive = active == 1
		cp.IsTrial = trial == 1
		out = append(out, cp)
	}
	return out, rows.Err()
}

func loadCreditBalances(ctx context.Context, db *sql.DB) ([]balanceRow, error) {
	if !mysqlTableExists(ctx, db, "credit_balances") {
		return nil, fmt.Errorf("credit_balances table missing")
	}
	// total/used may be DECIMAL — pull as strings then parse.
	q := mysqlSelectList(
		"company_id",
		mysqlCoalesce(ctx, db, "credit_balances", "total_credits", "0"),
		mysqlCoalesce(ctx, db, "credit_balances", "used_credits", "0"),
	) + " FROM credit_balances"
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []balanceRow
	for rows.Next() {
		var b balanceRow
		var totalRaw, usedRaw any
		if err := rows.Scan(&b.CompanyLegacy, &totalRaw, &usedRaw); err != nil {
			return nil, err
		}
		b.Total = scanIntish(totalRaw)
		b.Used = scanIntish(usedRaw)
		out = append(out, b)
	}
	return out, rows.Err()
}

func writeJSON(path string, v any) error {
	b, err := json.MarshalIndent(v, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(path, b, 0o644)
}

func nullStr(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}