// Command migrate-workflow-flows rewrites the 19 legacy SF-Workflow / InvocableProcess
// flows identified by the Phase 0 business audit into native AutoLaunchedFlow
// format so the NexusCRM flow engine can execute them with a single code path.
//
// Source of truth for the KEEP list and migration semantics: the Phase 1
// section of the design doc amber-main-design-20260416-230159.md
// ("Migration Script (unblocks everything)"). In particular:
//
//   - 14 record-triggered legacy Workflow flows get a native FlowStart
//     element synthesized from their root processMetadataValues (ObjectType,
//     TriggerType mapping).
//   - 5 InvocableProcess callees (Order_Expiration_Alert_7/_15/_30/_60/_90)
//     are rewritten as ProcessType=AutoLaunchedFlow with NO FlowStart —
//     they remain callable-only subflows.
//   - Order_Expiration_Notice_90 is special-cased: even though it is a
//     Workflow on record-trigger, its 3 Waits orchestrate scheduled
//     day-boundary notifications. The design doc (Phase 1 + Wave 3) asks
//     that it be migrated as ProcessType=AutoLaunchedFlow with
//     FlowStart.triggerType="Scheduled".
//   - Every string-valued leaf in the metadata tree has myVariable_current /
//     myVariable_old tokens rewritten to $Record / $Record__Old. __r segments
//     are preserved verbatim (resolved at runtime by GAP 6 lazy-load).
//
// Safety model mirrors scripts/deactivate-obsolete-flows (Phase 0):
//   - Default run is --dry-run: writes <output-dir>/<flow>/<flow>.original.json
//     and .migrated.json for every flow, emits a diff summary, no DB writes.
//   - --apply refuses to run unless TIDB_USER begins with the hard-coded
//     test-DB prefix. Escape hatch: --i-know-this-is-not-test.
//   - Per-flow transaction: SELECT ... FOR UPDATE on the Flow row locks
//     FlowDefinition + its active Flow version, runs UPDATE FlowDefinition
//     + UPDATE flow, commits. Any failure rolls back the pair.
//   - Idempotent. Re-running --apply on already-migrated flows is a no-op:
//     detection is presence of "$Record" token AND absence of
//     "myVariable_current" token in the stored Metadata.
//   - Per-flow failure does not halt the run. A flow whose metadata cannot
//     be parsed is appended to a .skipped.log and the script continues.
//   - Every flow goes through roundtrip validation (serialize -> deserialize
//     -> deep-equal) before being offered for apply.
package main

import (
	"context"
	"crypto/tls"
	"database/sql"
	"errors"
	"flag"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
)

// keepFlows is the hard-coded KEEP list from the Phase 0 business audit.
// Source: amber-main-design-20260416-230159.md, "Scope After Business Audit".
//
// The list below is deliberately flat (no per-object grouping) so it lines
// up 1:1 with the migration output directory. Order does not matter.
//
// If the design doc is updated, update this list — it is the SSoT for
// the migration target set.
// isKeepFlow reports whether name is in the audited KEEP list. Used to
// gate --flow.
func isKeepFlow(name string) bool {
	for _, f := range keepFlows {
		if f == name {
			return true
		}
	}
	return false
}

var keepFlows = []string{
	// 14 legacy Workflow flows (record-triggered after migration)
	"Contract_PingCAP_Contracting_Entity",
	"Order_PingCAP_RevRec_Entity",
	"Order_Account_Name",
	"Order_Partner_Name",
	"Order_Start_Date_End_Date",
	"Collections_Region_PingCAP_Contracting_Entity_Marketplace",
	"Collections",
	"Collections_Cash_Collected",
	"Collections_Invoice_Status_marketplace",
	"Order_Expiration_Notice_90",
	"Invoice_Notice",
	"Collections_Account_ID",
	"Collection_Partner_Name",
	"Account_PingCAP_Contracting_Entity",

	// 5 InvocableProcess callees — migrated as AutoLaunchedFlow (no FlowStart)
	"Order_Expiration_Alert_7",
	"Order_Expiration_Alert_15",
	"Order_Expiration_Alert_30",
	"Order_Expiration_Alert_60",
	"Order_Expiration_Alert_90",

	// 16 additional Workflow flows discovered during flow audit (2026-04-21)
	"Campaign_Member_Key_Message",
	"Campaign_Member_Key_Message_from_Contact",
	"Campaign_Member_Key_Message_from_Lead_when_create",
	"Exchange_Rate_in_RevRec",
	"Lead_Internal_User",
	"Opp_New_OR_Lost_Date",
	"Opp_Won_Notice",
	"PoC_Status_Change",
	"Quote_Contact_Update",
	"Quote_Partner_Name",
	"Quote_Product_Name",
	"Quote_Start_Date_End_Date",
	"RevRec_Partner_Name",
	"Sales_Contact_Email_Opt_Out",
	"Servicemandayownerdm",
	"TiDB_Cloud_User",

	// 4 additional active Workflow flows discovered during 04-25 staging
	// runtime re-verification (issue #1817): all four are
	// ProcessType=Workflow + TriggerType=RecordAfterSave + ActiveVersionId
	// set, but were not in the prior KEEP list, so the trigger listener
	// loaded them yet `matchesRecordTriggerType` couldn't read
	// flowMeta.Start.RecordTriggerType (the legacy Process Builder JSON
	// has no Start element) and silently skipped execution. Two of them
	// (Account_PoC_Status, Quote_Account_Opportunity_field) are explicit
	// FAILs in the 04-25 staging runtime report.
	"Account_PoC_Status",
	"Partner_Account_CM_Approval",
	"Quote_Account_Opportunity_field",
	"RevRec_Account",
}

// testDBUserPrefix is the known prefix of the Flow-engine test DB user. The
// script refuses to --apply unless TIDB_USER begins with this prefix. Mirrors
// scripts/deactivate-obsolete-flows for consistency across the lane.
const testDBUserPrefix = "4Qbn3F3nyaU9DLk"

// runTimeout bounds a full --apply run end to end. 19 flows × (read +
// transactional update) fits in well under a minute; 10 minutes is slack
// for TiDB Cloud gateway hiccups without letting a stalled connection hang
// forever.
const runTimeout = 10 * time.Minute

func main() {
	dryRun := flag.Bool("dry-run", true, "Emit migrated JSON files and a diff summary without touching the DB. Default true.")
	apply := flag.Bool("apply", false, "Execute UPDATE statements against the DB. Overrides --dry-run.")
	allowNonTest := flag.Bool("i-know-this-is-not-test", false, "Escape hatch for the test-DB allow-list. DO NOT use without sign-off.")
	flowFilter := flag.String("flow", "", "If non-empty, only migrate this one DeveloperName (retry / single-flow debug).")
	outputDir := flag.String("output-dir", "migration-output", "Directory to write original.json + migrated.json pairs in --dry-run mode.")
	flag.Parse()

	execute := *apply
	if !*dryRun && !*apply {
		// Operator passed --dry-run=false without --apply. Treat as apply.
		execute = true
	}

	ctx, cancel := context.WithTimeout(context.Background(), runTimeout)
	defer cancel()

	targets := keepFlows
	if *flowFilter != "" {
		// --flow must be a member of keepFlows. Otherwise the default
		// classifier would silently rewrite an unrelated flow as a
		// record-triggered AutoLaunchedFlow — a typo or hostile
		// argument could corrupt production metadata. No escape hatch;
		// if a new flow needs migrating, add it to keepFlows so the
		// audit trail makes sense.
		if !isKeepFlow(*flowFilter) {
			fmt.Fprintf(os.Stderr,
				"error: --flow=%q is not in the audited KEEP list. "+
					"Valid names:\n  %s\n",
				*flowFilter, strings.Join(keepFlows, "\n  "))
			os.Exit(1)
		}
		targets = []string{*flowFilter}
	}

	if err := run(ctx, execute, *allowNonTest, *outputDir, os.Stdout, targets); err != nil {
		fmt.Fprintf(os.Stderr, "error: %v\n", err)
		os.Exit(1)
	}
}

func run(ctx context.Context, apply, allowNonTest bool, outputDir string, out io.Writer, flows []string) error {
	if len(flows) == 0 {
		return errors.New("empty target list — nothing to migrate")
	}

	if !apply {
		return runDryRun(ctx, outputDir, out, flows)
	}

	if err := checkTestDBGate(allowNonTest); err != nil {
		return err
	}

	db, err := openDB(ctx)
	if err != nil {
		return fmt.Errorf("open DB: %w", err)
	}
	defer db.Close()

	return runApply(ctx, db, out, flows)
}

// runDryRun reads each target flow from the DB (or from testdata/ if the
// caller has no DB credentials), migrates in-memory, and writes
// <outputDir>/<flow>/<flow>.original.json + .migrated.json. Diff summary
// is emitted to `out`.
//
// Dry-run still requires DB read access — there is no safe way to do a
// useful "would this migration work?" preview without the real stored
// metadata. The test-DB gate does NOT apply to dry-run (read-only), so
// an operator can dry-run against staging/prod with read-only creds.
func runDryRun(ctx context.Context, outputDir string, out io.Writer, flows []string) error {
	db, err := openDB(ctx)
	if err != nil {
		return fmt.Errorf("open DB (read-only for dry-run): %w", err)
	}
	defer db.Close()

	if err := os.MkdirAll(outputDir, 0o700); err != nil {
		return fmt.Errorf("mkdir %s: %w", outputDir, err)
	}

	var (
		skipped []string
		ok      int
	)
	for _, name := range flows {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return fmt.Errorf("aborted during %s: %w", name, ctxErr)
		}
		rec, err := readFlow(ctx, db, name)
		if err != nil {
			fmt.Fprintf(out, "-- SKIP %s: %v\n", name, err)
			skipped = append(skipped, name)
			continue
		}

		result, err := Migrate(rec)
		if err != nil {
			fmt.Fprintf(out, "-- SKIP %s: migrate: %v\n", name, err)
			skipped = append(skipped, name)
			continue
		}

		flowDir := filepath.Join(outputDir, name)
		if err := os.MkdirAll(flowDir, 0o700); err != nil {
			return fmt.Errorf("mkdir %s: %w", flowDir, err)
		}
		if err := writeDumpFile(filepath.Join(flowDir, name+".original.json"), result.OriginalPretty); err != nil {
			return fmt.Errorf("write original: %w", err)
		}
		if err := writeDumpFile(filepath.Join(flowDir, name+".migrated.json"), result.MigratedPretty); err != nil {
			return fmt.Errorf("write migrated: %w", err)
		}

		fmt.Fprintf(out, "%s: %s (orig %dB -> migrated %dB), rewrites=%d, processType %s -> %s, triggerType %s\n",
			name,
			diffSummary(result),
			len(result.OriginalPretty), len(result.MigratedPretty),
			result.RewriteCount,
			rec.ProcessType, result.NewProcessType,
			result.NewTriggerType)
		ok++
	}

	fmt.Fprintf(out, "\n-- DRY RUN SUMMARY: %d migrated, %d skipped\n", ok, len(skipped))
	if len(skipped) > 0 {
		fmt.Fprintf(out, "-- skipped: %s\n", strings.Join(skipped, ", "))
		if err := os.WriteFile(filepath.Join(outputDir, ".skipped.log"), []byte(strings.Join(skipped, "\n")+"\n"), 0o600); err != nil {
			return fmt.Errorf("write skipped log: %w", err)
		}
		return fmt.Errorf("dry-run completed with %d skipped flow(s); see log", len(skipped))
	}
	return nil
}

// runApply reads each flow from the DB, migrates, and writes the migrated
// Metadata back inside a per-flow transaction that also updates
// FlowDefinition.TriggerType + TriggerObjectOrEvent and Flow.ProcessType.
// Already-migrated flows (detected heuristically) are no-ops.
//
// Per-flow failures are logged and the loop continues — a flow with
// malformed metadata should not abort the rest of the run. Context-level
// errors (deadline exceeded, canceled) ARE fatal, since every subsequent
// flow will hit the same error and emit noise into the log.
func runApply(ctx context.Context, db *sql.DB, out io.Writer, flows []string) error {
	var skipped []string
	var totalFDAff, totalFlowAff int64

	for _, name := range flows {
		fdAff, flowAff, err := applyOneFlow(ctx, db, out, name)
		if err != nil {
			// Context errors indicate the whole run is over — don't
			// keep looping producing more deadline-exceeded logs.
			if ctxErr := ctx.Err(); ctxErr != nil {
				fmt.Fprintf(out, "-- ABORT on %s: %v\n", name, err)
				fmt.Fprintf(out, "\ntotal rows affected so far: FlowDefinition=%d, Flow=%d\n", totalFDAff, totalFlowAff)
				return fmt.Errorf("context %w while processing %s", ctxErr, name)
			}
			fmt.Fprintf(out, "-- SKIP %s: %v\n", name, err)
			skipped = append(skipped, fmt.Sprintf("%s: %v", name, err))
			continue
		}
		totalFDAff += fdAff
		totalFlowAff += flowAff
	}

	fmt.Fprintf(out, "\ntotal rows affected: FlowDefinition=%d, Flow=%d\n", totalFDAff, totalFlowAff)
	if len(skipped) > 0 {
		fmt.Fprintf(out, "skipped: %d\n", len(skipped))
		for _, s := range skipped {
			fmt.Fprintf(out, "  %s\n", s)
		}
		// CI / operator automation reads the exit code. A partial
		// migration must not look "green" — an inert set of flows with
		// some unmigrated members is a failure mode someone needs to
		// investigate.
		return fmt.Errorf("apply completed with %d skipped flow(s); see log", len(skipped))
	}
	return nil
}

// checkTestDBGate mirrors scripts/deactivate-obsolete-flows: refuse to
// --apply unless TIDB_USER starts with the known test prefix, unless the
// operator explicitly overrides with --i-know-this-is-not-test.
func checkTestDBGate(allowNonTest bool) error {
	user := os.Getenv("TIDB_USER")
	if strings.HasPrefix(user, testDBUserPrefix) {
		return nil
	}
	if allowNonTest {
		fmt.Fprintf(os.Stderr,
			"WARNING: TIDB_USER=%q does not match test-DB prefix %q; proceeding because --i-know-this-is-not-test was set.\n",
			user, testDBUserPrefix)
		return nil
	}
	return fmt.Errorf(
		"refusing to --apply: TIDB_USER=%q does not match the test-DB prefix %q. "+
			"This script is scoped to the Flow-engine test DB only. "+
			"If you are absolutely sure, pass --i-know-this-is-not-test.",
		user, testDBUserPrefix)
}

// writeDumpFile creates a dry-run dump file with owner-only
// permissions, refusing to follow a pre-existing symlink. The
// O_CREATE|O_EXCL combination would refuse to reuse any existing file
// too, but dry-run often runs multiple times against the same output
// directory — so we remove an existing regular file first (Lstat
// ensures we are not chasing a symlink).
func writeDumpFile(path string, data []byte) error {
	if info, err := os.Lstat(path); err == nil {
		if info.Mode()&os.ModeSymlink != 0 {
			return fmt.Errorf("refusing to write over symlink: %s", path)
		}
		if err := os.Remove(path); err != nil {
			return fmt.Errorf("remove stale %s: %w", path, err)
		}
	}
	f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
	if err != nil {
		return err
	}
	defer f.Close()
	_, err = f.Write(data)
	return err
}

// openDB reads TIDB_* env vars and opens a TLS-enabled connection for
// remote hosts (plaintext for localhost). Mirrors
// scripts/deactivate-obsolete-flows for consistency; a utility package is
// overkill for two call sites.
func openDB(ctx context.Context) (*sql.DB, error) {
	host := os.Getenv("TIDB_HOST")
	port := os.Getenv("TIDB_PORT")
	user := os.Getenv("TIDB_USER")
	password := os.Getenv("TIDB_PASSWORD")
	database := os.Getenv("TIDB_DATABASE")

	missing := make([]string, 0, 5)
	if host == "" {
		missing = append(missing, "TIDB_HOST")
	}
	if user == "" {
		missing = append(missing, "TIDB_USER")
	}
	// Localhost dev TiDB typically runs root with no password. Mirror the
	// TLS skip below: only require TIDB_PASSWORD for remote hosts.
	if password == "" && host != "127.0.0.1" && host != "localhost" {
		missing = append(missing, "TIDB_PASSWORD")
	}
	if database == "" {
		missing = append(missing, "TIDB_DATABASE")
	}
	if len(missing) > 0 {
		return nil, fmt.Errorf("missing required env vars: %s", strings.Join(missing, ", "))
	}
	if port == "" {
		port = "4000"
	}

	tlsParam := ""
	if host != "127.0.0.1" && host != "localhost" {
		if err := mysql.RegisterTLSConfig("migrate-workflow-flows", &tls.Config{
			MinVersion: tls.VersionTLS12,
			ServerName: host,
		}); err != nil {
			return nil, fmt.Errorf("register TLS: %w", err)
		}
		tlsParam = "&tls=migrate-workflow-flows"
	}

	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local%s",
		user, password, host, port, database, tlsParam)

	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, fmt.Errorf("open: %w", err)
	}
	db.SetConnMaxLifetime(5 * time.Minute)
	if err := db.PingContext(ctx); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("ping: %w", err)
	}
	return db, nil
}