// Command activate-flow-wave re-activates migrated FlowDefinitions in waves,
// per the Phase 4 staged-activation plan in
// docs/design/flow/flow-engine-gap-analysis.md §Phase 4.
//
// Phase 1 (scripts/migrate-workflow-flows) migrated the 35 KEEP-list legacy
// flows (19 original + 16 added 2026-04-21) into native AutoLaunchedFlow
// format but cleared FlowDefinition.ActiveVersionId so the trigger listener
// stops dispatching. This script sets ActiveVersionId back to the latest
// Flow version Id (typically the just-migrated version) and marks that
// version Status='Active', putting the flow live under the new engine.
//
// Usage:
//
//	# Dry-run one flow:
//	go run ./scripts/activate-flow-wave --flow=Order_Account_Name
//
//	# Activate one flow (staging/prod require --i-know-this-is-not-test):
//	go run ./scripts/activate-flow-wave --flow=Order_Account_Name --apply --i-know-this-is-not-test
//
//	# Dry-run a whole wave (9 flows for Wave 1, 4 for Wave 2, 6 for Wave 3,
//	# 16 for Wave 4):
//	go run ./scripts/activate-flow-wave --wave=1
//
// Safety model mirrors deactivate-obsolete-flows and migrate-workflow-flows:
//   - --dry-run by default. No DB writes without --apply.
//   - --apply refuses to run unless TIDB_USER matches the hard-coded test-DB
//     prefix, or --i-know-this-is-not-test is passed.
//   - Each per-flow activation runs inside a single transaction with
//     SELECT ... FOR UPDATE on the FlowDefinition row.
//   - Pre-change state (ActiveVersionId + every version's Id/Status/Number)
//     is emitted to stdout as SQL comments so capturing the apply log doubles
//     as a rollback reference.
//   - Idempotent: re-applying when ActiveVersionId already points at the
//     latest version results in 0 affected rows.
//   - Preflight aborts if any requested DeveloperName is missing or if the
//     flow has zero non-obsolete versions (nothing to activate).
package main

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

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

// waves maps wave number to DeveloperNames, sourced from
// docs/design/flow/flow-engine-gap-analysis.md §Phase 4.
//
// Wave 1 (simple Workflow field-updates, no Wait): activate one at a time,
// monitor for duplicate field updates / recursion with existing flows.
// Wave 2 (moderate complexity): after Wave 1 is stable.
// Wave 3 (the hard one: Order_Expiration_Notice_90 Scheduled + 5 callees):
// only after Wave 1+2 stable for days AND Wave 3 start.filters pre-check
// completed (see scripts/migrate-workflow-flows/README.md §Wave 3 note).
// Wave 4 (the 16 additional flows added to keepFlows on 2026-04-21):
// activate after Wave 1-3 are stable. These are the auto-discovered extras
// that migrate-workflow-flows rewrites but the original wave lists did not
// cover; without Wave 4 they stay inert (issue #1744).
var waves = map[int][]string{
	1: {
		"Order_Account_Name",
		"Order_Partner_Name",
		"Order_PingCAP_RevRec_Entity",
		"Order_Start_Date_End_Date",
		"Contract_PingCAP_Contracting_Entity",
		"Account_PingCAP_Contracting_Entity",
		"Collection_Partner_Name",
		"Collections_Account_ID",
		"Collections_Region_PingCAP_Contracting_Entity_Marketplace",
	},
	2: {
		"Collections",
		"Collections_Cash_Collected",
		"Collections_Invoice_Status_marketplace",
		"Invoice_Notice",
	},
	3: {
		"Order_Expiration_Notice_90",
		"Order_Expiration_Alert_7",
		"Order_Expiration_Alert_15",
		"Order_Expiration_Alert_30",
		"Order_Expiration_Alert_60",
		"Order_Expiration_Alert_90",
	},
	4: {
		"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",
	},
	// Wave 5 (the 4 active Workflow flows discovered during issue #1817
	// 04-25 staging runtime re-verification). Same shape as Wave 4 — they
	// were already running on staging as legacy Workflow + ActiveVersionId
	// set, but silently no-oped because matchesRecordTriggerType couldn't
	// read the Process Builder JSON. Activate after Wave 4 is stable.
	5: {
		"Account_PoC_Status",
		"Partner_Account_CM_Approval",
		"Quote_Account_Opportunity_field",
		"RevRec_Account",
	},
}

// testDBUserPrefix mirrors deactivate-obsolete-flows. See that file for why
// this is hard-coded rather than env-driven.
const testDBUserPrefix = "4Qbn3F3nyaU9DLk"

const runTimeout = 5 * time.Minute

func main() {
	dryRun := flag.Bool("dry-run", true, "Print the activation plan 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.")
	wave := flag.Int("wave", 0, "Activate a whole wave (1, 2, 3, 4, or 5). Mutually exclusive with --flow.")
	flowFilter := flag.String("flow", "", "Activate a single DeveloperName. Mutually exclusive with --wave.")
	flag.Parse()

	execute := *apply
	if !*dryRun && !*apply {
		execute = true
	}

	targets, err := resolveTargets(*wave, *flowFilter)
	if err != nil {
		fmt.Fprintf(os.Stderr, "error: %v\n", err)
		os.Exit(2)
	}

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

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

// resolveTargets picks the DeveloperName list based on --wave / --flow flags.
// Requires exactly one to be set; an empty selection is a usage error rather
// than a default-to-all to avoid accidental "activate everything" runs.
func resolveTargets(wave int, flow string) ([]string, error) {
	hasWave := wave != 0
	hasFlow := flow != ""
	if hasWave && hasFlow {
		return nil, errors.New("--wave and --flow are mutually exclusive")
	}
	if !hasWave && !hasFlow {
		return nil, errors.New("must pass either --wave=<1|2|3|4|5> or --flow=<DeveloperName>")
	}
	if hasFlow {
		if !knownFlow(flow) {
			return nil, fmt.Errorf("--flow=%q is not in any wave list; check docs/design/flow/flow-engine-gap-analysis.md §Phase 4", flow)
		}
		return []string{flow}, nil
	}
	flows, ok := waves[wave]
	if !ok {
		return nil, fmt.Errorf("--wave=%d invalid; expected 1, 2, 3, 4, or 5", wave)
	}
	return flows, nil
}

func knownFlow(name string) bool {
	for _, flows := range waves {
		for _, f := range flows {
			if f == name {
				return true
			}
		}
	}
	return false
}

func run(ctx context.Context, apply, allowNonTest bool, out io.Writer, flows []string) error {
	if !apply {
		fmt.Fprintln(out, "-- DRY RUN (no DB writes). Re-run with --apply to execute.")
		fmt.Fprintf(out, "-- Target flows (%d):\n", len(flows))
		for _, f := range flows {
			fmt.Fprintf(out, "--   %s\n", f)
		}
		fmt.Fprintln(out, "-- Planned per-flow transaction:")
		fmt.Fprintln(out, "--   SELECT Id, ActiveVersionId FROM FlowDefinition WHERE DeveloperName = ? FOR UPDATE;")
		fmt.Fprintln(out, "--   SELECT Id, VersionNumber, Status FROM flow WHERE DefinitionId = ? ORDER BY VersionNumber DESC;")
		fmt.Fprintln(out, "--   UPDATE flow SET Status = 'Obsolete' WHERE DefinitionId = ? AND Status = 'Active' AND Id <> <target>;")
		fmt.Fprintln(out, "--   UPDATE flow SET Status = 'Active'   WHERE Id = <target> AND Status <> 'Active';")
		fmt.Fprintln(out, "--   UPDATE FlowDefinition SET ActiveVersionId = <target> WHERE Id = ? AND COALESCE(ActiveVersionId,'') <> <target>;")
		return nil
	}

	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()

	if err := preflight(ctx, db, flows); err != nil {
		return fmt.Errorf("preflight: %w", err)
	}

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

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. "+
			"If you are absolutely sure, pass --i-know-this-is-not-test.",
		user, testDBUserPrefix)
}

// preflight verifies every DeveloperName exists exactly once in FlowDefinition
// and has at least one non-obsolete Flow version to activate. A zero-version
// result is a hard fail, not a silent skip, so a stale wave list doesn't
// collapse into a no-op that masks a missing migration.
func preflight(ctx context.Context, db *sql.DB, flows []string) error {
	placeholders := strings.Repeat("?,", len(flows))
	placeholders = placeholders[:len(placeholders)-1]
	args := make([]any, len(flows))
	for i, f := range flows {
		args[i] = f
	}

	q := "SELECT DeveloperName, COUNT(*), MAX(ProcessType) FROM FlowDefinition WHERE DeveloperName IN (" + placeholders + ") GROUP BY DeveloperName"
	rows, err := db.QueryContext(ctx, q, args...)
	if err != nil {
		return fmt.Errorf("query FlowDefinition: %w", err)
	}
	type row struct {
		count       int
		processType string
	}
	found := make(map[string]row, len(flows))
	for rows.Next() {
		var name, processType string
		var n int
		if err := rows.Scan(&name, &n, &processType); err != nil {
			rows.Close()
			return fmt.Errorf("scan: %w", err)
		}
		found[name] = row{count: n, processType: processType}
	}
	rows.Close()

	var missing, dup, unmigrated []string
	for _, f := range flows {
		r, ok := found[f]
		if !ok {
			missing = append(missing, f)
			continue
		}
		if r.count > 1 {
			dup = append(dup, fmt.Sprintf("%s (%d)", f, r.count))
		}
		// Activation must follow migration. Activating a row whose ProcessType
		// is still "Workflow" or "InvocableProcess" puts a flow live whose
		// stored Metadata is legacy Process Builder JSON — the engine's
		// matchesRecordTriggerType reads flowMeta.Start.RecordTriggerType,
		// which the legacy JSON does not carry, so the flow is silently
		// inert. Hard-fail here so an out-of-order operator (`activate`
		// before `migrate --apply`) gets a loud error instead of a
		// silently broken activation.
		if r.processType != "AutoLaunchedFlow" {
			unmigrated = append(unmigrated, fmt.Sprintf("%s (ProcessType=%s)", f, r.processType))
		}
	}
	if len(missing) > 0 || len(dup) > 0 || len(unmigrated) > 0 {
		return fmt.Errorf("preflight: missing=%v duplicates=%v unmigrated=%v (run scripts/migrate-workflow-flows --apply first)",
			missing, dup, unmigrated)
	}
	return nil
}

// applyFlows runs one transaction per DeveloperName.
func applyFlows(ctx context.Context, db *sql.DB, out io.Writer, flows []string) error {
	var totalFD, totalFV int64
	for _, name := range flows {
		fd, fv, err := applyOneFlow(ctx, db, out, name)
		if err != nil {
			return fmt.Errorf("activate %s: %w", name, err)
		}
		totalFD += fd
		totalFV += fv
	}
	fmt.Fprintf(out, "total rows affected: FlowDefinition=%d, Flow=%d\n", totalFD, totalFV)
	return nil
}

func applyOneFlow(ctx context.Context, db *sql.DB, out io.Writer, name string) (fdAffected, fvAffected int64, err error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, 0, fmt.Errorf("begin tx: %w", err)
	}
	defer func() {
		if err != nil {
			_ = tx.Rollback()
		}
	}()

	var fdID, activeVersionID sql.NullString
	err = tx.QueryRowContext(ctx,
		"SELECT Id, ActiveVersionId FROM FlowDefinition WHERE DeveloperName = ? FOR UPDATE",
		name,
	).Scan(&fdID, &activeVersionID)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, 0, fmt.Errorf("FlowDefinition not found (preflight should have caught this)")
		}
		return 0, 0, fmt.Errorf("select+lock FlowDefinition: %w", err)
	}

	fmt.Fprintf(out, "-- %s: FlowDefinition.Id=%s pre-change ActiveVersionId=%q\n",
		name, fdID.String, activeVersionID.String)

	// Pick the highest VersionNumber Flow row. Phase 1 migration wrote its
	// rewritten Metadata onto that row. Status = 'Obsolete' is fine —
	// activation will flip it to 'Active' below.
	var targetID string
	var targetVersion int
	var targetStatus string
	err = tx.QueryRowContext(ctx,
		"SELECT Id, VersionNumber, Status FROM flow WHERE DefinitionId = ? ORDER BY VersionNumber DESC LIMIT 1",
		fdID.String,
	).Scan(&targetID, &targetVersion, &targetStatus)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, 0, fmt.Errorf("no flow versions exist for DefinitionId=%s", fdID.String)
		}
		return 0, 0, fmt.Errorf("select latest flow version: %w", err)
	}
	fmt.Fprintf(out, "--   target version flow.Id=%s VersionNumber=%d pre-change Status=%q\n",
		targetID, targetVersion, targetStatus)

	// Dump all versions for rollback audit.
	fvRows, err := tx.QueryContext(ctx,
		"SELECT Id, VersionNumber, Status FROM flow WHERE DefinitionId = ? ORDER BY VersionNumber",
		fdID.String)
	if err != nil {
		return 0, 0, fmt.Errorf("select flow versions: %w", err)
	}
	for fvRows.Next() {
		var id, status string
		var vn int
		if scanErr := fvRows.Scan(&id, &vn, &status); scanErr != nil {
			fvRows.Close()
			return 0, 0, fmt.Errorf("scan flow version: %w", scanErr)
		}
		fmt.Fprintf(out, "--   flow.Id=%s VersionNumber=%d pre-change Status=%q\n", id, vn, status)
	}
	if iterErr := fvRows.Err(); iterErr != nil {
		fvRows.Close()
		return 0, 0, fmt.Errorf("flow version iter: %w", iterErr)
	}
	fvRows.Close()

	// Guarded: demote any non-target Active version to Obsolete. In practice
	// Phase 1 left ActiveVersionId empty and all versions Obsolete/Draft,
	// but this belt-and-braces guard keeps the invariant "at most one
	// Active version per FlowDefinition" intact even if something else
	// flipped a version Active between migration and activation.
	res, err := tx.ExecContext(ctx,
		"UPDATE flow SET Status = 'Obsolete', LastModifiedDate = NOW() "+
			"WHERE DefinitionId = ? AND Status = 'Active' AND Id <> ?",
		fdID.String, targetID)
	if err != nil {
		return 0, 0, fmt.Errorf("demote other Active versions: %w", err)
	}
	demoted, _ := res.RowsAffected()
	fvAffected += demoted

	// Promote target version to Active (guarded — no-op if already Active).
	res, err = tx.ExecContext(ctx,
		"UPDATE flow SET Status = 'Active', LastModifiedDate = NOW() "+
			"WHERE Id = ? AND Status <> 'Active'",
		targetID)
	if err != nil {
		return 0, 0, fmt.Errorf("promote target version to Active: %w", err)
	}
	promoted, _ := res.RowsAffected()
	fvAffected += promoted

	// Point FlowDefinition at the target version (guarded — no-op if already
	// pointing there).
	res, err = tx.ExecContext(ctx,
		"UPDATE FlowDefinition SET ActiveVersionId = ?, LastModifiedDate = NOW() "+
			"WHERE Id = ? AND COALESCE(ActiveVersionId,'') <> ?",
		targetID, fdID.String, targetID)
	if err != nil {
		return 0, 0, fmt.Errorf("set FlowDefinition.ActiveVersionId: %w", err)
	}
	fdAffected, _ = res.RowsAffected()

	if err = tx.Commit(); err != nil {
		return 0, 0, fmt.Errorf("commit: %w", err)
	}

	fmt.Fprintf(out,
		"%s: activated flow.Id=%s (VersionNumber=%d); FlowDefinition rows=%d, Flow rows=%d (committed)\n",
		name, targetID, targetVersion, fdAffected, fvAffected)
	return fdAffected, fvAffected, nil
}

// openDB mirrors the DSN construction used by the other migration scripts.
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("activate-flow-wave", &tls.Config{
			MinVersion: tls.VersionTLS12,
			ServerName: host,
		}); err != nil {
			return nil, fmt.Errorf("register TLS: %w", err)
		}
		tlsParam = "&tls=activate-flow-wave"
	}

	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
}