// Command deactivate-obsolete-flows marks the Phase 0 DROP-list flows inert by
// clearing FlowDefinition.ActiveVersionId and setting Flow.Status = 'Obsolete'.
//
// Source of truth for the DROP list: the Phase 0 section of the design doc
// amber-main-design-20260416-230159.md ("Scope After Business Audit").
//
// Safety model:
//   - Default run is --dry-run: emits the SQL it *would* execute and exits 0.
//     No DB writes happen without --apply.
//   - --apply refuses to run against anything that does not match the
//     allow-listed test-DB user prefix unless --i-know-this-is-not-test is
//     also passed. This prevents an operator from accidentally pointing at
//     prod or staging credentials. The allow-list is hard-coded, not env.
//   - Before mutating, the script runs a preflight that verifies every
//     DeveloperName in the DROP list resolves to exactly one FlowDefinition
//     row. Missing or duplicate targets abort the run.
//   - Every per-flow pair of UPDATEs is wrapped in a single transaction with
//     SELECT ... FOR UPDATE on the FlowDefinition row. A concurrent writer
//     cannot resurrect the flow between the two updates, and any statement
//     failure rolls back the whole pair.
//   - Before each mutation, the script prints a SQL comment with the
//     pre-change state of the target rows (ActiveVersionId and Flow version
//     Id+Status). Capturing stdout gives the operator a per-flow rollback
//     record without requiring a separate manual SELECT.
//   - Idempotent: re-running --apply after everything is already Obsolete
//     results in 0 additional row changes (verified by integration run #2).
package main

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

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

// obsoleteFlows is the hard-coded DROP list from the Phase 0 design doc.
//
// These DeveloperName values were resolved by cross-referencing the design
// doc's Phase 0 bullet list with `SELECT DeveloperName FROM FlowDefinition`
// on the test DB. The design doc lists "BN_* flows (3)" and
// "Account_Related_Leads_Based_on_Website_Domain_Create/Update" (shorthand
// for two distinct flows); this array expands those shorthands to the exact
// DeveloperName values stored in the database.
//
// If the design doc is updated, update this list. It is the SSoT for the
// script.
var obsoleteFlows = []string{
	// Account
	"Account_PoC_Status",
	"Account_Partner_Sales_Owner_Set",
	"Account_status_Flow",
	"Partner_Account_CM_Approval",
	"RevRec_Account",
	"Quote_Account_Opportunity_field",
	"Sync_Account_Status_to_Lead",
	"Account_And_Lead_Sum_Approval_Poc",
	"Account_Auto_Create_Org",
	"Account_Related_Leads_Based_on_Website_Domain_Create",
	"Account_Related_Leads_Based_on_Website_Domain_Update",

	// Lead / PoC / Org
	"BN_Account_Lead_Relate_Leads_to_Accounts",
	"BN_Account_Related_Leads_Based_on_Website_Domain",
	"BN_Lead_Relate_to_Account_by_Email_Domain_Flow",
	"Lead_Owner_Is_Change_By_Account_Send_Email",
	"LeadSetOwnerByAccount",
	"Org_Set_Account",
	"Auto_Set_Poc_Account_Contact_Lead_opportunity",
	"update_account_latest_month_end_arr",
}

// testDBUserPrefix is the known prefix of the Flow-engine test DB user. The
// script aborts if TIDB_USER does not start with this prefix, unless the
// --i-know-this-is-not-test escape hatch is set. This gate exists to make
// accidentally pointing at prod or staging credentials a hard failure rather
// than a silent destructive write. The design doc explicitly restricts
// Phase 0 to the test DB.
const testDBUserPrefix = "4Qbn3F3nyaU9DLk"

// runTimeout bounds the whole --apply run. Longer than a normal run takes
// against this workload (38 UPDATE statements + preflight), short enough that
// a stalled network connection fails loudly instead of hanging forever.
const runTimeout = 5 * time.Minute

func main() {
	dryRun := flag.Bool("dry-run", true, "Emit SQL without touching the DB. Default true; set false or pass --apply to execute.")
	apply := flag.Bool("apply", false, "Execute the 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.")
	flag.Parse()

	// --apply wins over --dry-run. Default behavior without either flag is
	// dry-run=true.
	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()

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

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

	stmts := buildStatements(flows)

	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 _, s := range stmts {
			fmt.Fprintf(out, "-- %s\n", s.label)
			fmt.Fprintln(out, s.sql)
		}
		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)
}

// checkTestDBGate refuses to proceed unless TIDB_USER matches the known test
// prefix, unless the operator has explicitly opted in with the escape hatch.
// The script's contract is "test DB only" — enforce that mechanically so a
// stale shell or copy-pasted env block cannot silently mutate prod.
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)
}

// statement is a single SQL to execute, with a human-readable label.
type statement struct {
	label string
	sql   string
	args  []any
}

// buildStatements generates the idempotent UPDATE statements for each flow.
// For every DeveloperName we emit two statements:
//  1. Clear FlowDefinition.ActiveVersionId so the trigger listener stops
//     dispatching to the flow.
//  2. Mark every related Flow version row as 'Obsolete' unless already
//     'Obsolete' (the WHERE clause is what makes this idempotent — re-running
//     produces 0 affected rows once everything is settled).
//
// Both statements are guarded so that already-deactivated flows are skipped
// and re-apply runs are no-ops.
func buildStatements(flows []string) []statement {
	stmts := make([]statement, 0, len(flows)*2)
	for _, name := range flows {
		stmts = append(stmts, statement{
			label: fmt.Sprintf("clear ActiveVersionId for %s", name),
			sql: "UPDATE FlowDefinition " +
				"SET ActiveVersionId = '', LastModifiedDate = NOW() " +
				"WHERE DeveloperName = ? AND ActiveVersionId IS NOT NULL AND ActiveVersionId <> ''",
			args: []any{name},
		})
		stmts = append(stmts, statement{
			label: fmt.Sprintf("set Flow.Status = 'Obsolete' for %s", name),
			sql: "UPDATE flow f " +
				"JOIN FlowDefinition fd ON f.DefinitionId = fd.Id " +
				"SET f.Status = 'Obsolete', f.LastModifiedDate = NOW() " +
				"WHERE fd.DeveloperName = ? AND f.Status <> 'Obsolete'",
			args: []any{name},
		})
	}
	return stmts
}

// preflight verifies every DeveloperName in the DROP list resolves to exactly
// one FlowDefinition row. Missing targets mean the list is stale (the doc and
// DB have drifted); duplicates mean the schema's uniqueness contract was
// violated (unlikely — FlowDefinition.DeveloperName has a UNIQUE index — but
// worth asserting before we mutate).
//
// Without this check, a typo in the DROP list silently collapses to
// `0 rows affected`, which is indistinguishable from the idempotent
// re-run case and could leave a live flow undeactivated.
func preflight(ctx context.Context, db *sql.DB, flows []string) error {
	placeholders := strings.Repeat("?,", len(flows))
	placeholders = placeholders[:len(placeholders)-1]
	q := "SELECT DeveloperName, COUNT(*) FROM FlowDefinition WHERE DeveloperName IN (" + placeholders + ") GROUP BY DeveloperName"

	args := make([]any, len(flows))
	for i, f := range flows {
		args[i] = f
	}

	rows, err := db.QueryContext(ctx, q, args...)
	if err != nil {
		return fmt.Errorf("query FlowDefinition: %w", err)
	}
	defer rows.Close()

	found := make(map[string]int, len(flows))
	for rows.Next() {
		var name string
		var n int
		if err := rows.Scan(&name, &n); err != nil {
			return fmt.Errorf("scan: %w", err)
		}
		found[name] = n
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("rows iter: %w", err)
	}

	var missing, dup []string
	for _, f := range flows {
		n, ok := found[f]
		if !ok {
			missing = append(missing, f)
			continue
		}
		if n > 1 {
			dup = append(dup, fmt.Sprintf("%s (%d)", f, n))
		}
	}

	if len(missing) > 0 || len(dup) > 0 {
		return fmt.Errorf("preflight failed: missing=%v duplicates=%v", missing, dup)
	}
	return nil
}

// applyFlows runs one transaction per DeveloperName. Each transaction
// locks the FlowDefinition row with SELECT ... FOR UPDATE so a concurrent
// writer cannot resurrect the flow between the two UPDATEs, then runs both
// UPDATEs, then commits. Any statement failure rolls back the pair.
//
// Before each transaction, the current state is emitted as a SQL comment on
// stdout so the operator's capture of the apply log doubles as the rollback
// reference (pre-change ActiveVersionId + per-version Status).
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("apply %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()
		}
	}()

	// Lock the FlowDefinition row. Also captures the pre-change state for
	// the rollback-on-stdout protocol.
	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)

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

	// Clear ActiveVersionId (guarded — no-op if already empty).
	res, err := tx.ExecContext(ctx,
		"UPDATE FlowDefinition SET ActiveVersionId = '', LastModifiedDate = NOW() "+
			"WHERE Id = ? AND ActiveVersionId IS NOT NULL AND ActiveVersionId <> ''",
		fdID.String)
	if err != nil {
		return 0, 0, fmt.Errorf("clear ActiveVersionId: %w", err)
	}
	fdAffected, _ = res.RowsAffected()

	// Obsolete all non-obsolete flow versions (guarded — no-op if already
	// Obsolete).
	res, err = tx.ExecContext(ctx,
		"UPDATE flow SET Status = 'Obsolete', LastModifiedDate = NOW() "+
			"WHERE DefinitionId = ? AND Status <> 'Obsolete'",
		fdID.String)
	if err != nil {
		return 0, 0, fmt.Errorf("set flow.Status = Obsolete: %w", err)
	}
	fvAffected, _ = res.RowsAffected()

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

	fmt.Fprintf(out, "%s: FlowDefinition rows=%d, Flow rows=%d (committed)\n",
		name, fdAffected, fvAffected)
	return fdAffected, fvAffected, nil
}

// openDB reads TiDB_* env vars and opens a connection. Mirrors
// backend/internal/infrastructure/database/tidb.go so behavior matches the
// rest of the codebase (TLS for remote hosts, plaintext for localhost).
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")
	}
	if password == "" {
		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("deactivate-obsolete-flows", &tls.Config{
			MinVersion: tls.VersionTLS12,
			ServerName: host,
		}); err != nil {
			return nil, fmt.Errorf("register TLS: %w", err)
		}
		tlsParam = "&tls=deactivate-obsolete-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
}