package bootstrap

import (
	"database/sql"
	"embed"
	"encoding/json"
	"fmt"
	"io/fs"
	"log"
	"strings"
	"time"

	apperrors "github.com/nexuscrm/backend/pkg/errors"
)

//go:embed standard_flows/*.flow.json
var standardFlowsFS embed.FS

// standardFlowPackDir is the embed directory root. Kept distinct from the
// directory name on disk so future reshaping of the pack layout does not
// leak into seeder logic.
const standardFlowPackDir = "standard_flows"

// standardFlowSeed describes one shipped Flow (parent FlowDefinition +
// active Flow version). DeveloperName is derived from the filename
// (<DeveloperName>.flow.json) and MUST match the `flowDeveloperName`
// referenced by any row in _System_StandardAction that targets this flow.
type standardFlowSeed struct {
	DeveloperName string
	Label         string
	Description   string
	ProcessType   string
	MetadataRaw   []byte
}

// InitializeStandardFlows upserts every .flow.json file under the
// standard_flows pack as an active Flow + FlowDefinition row. Idempotent
// with IsCustomized reconciliation at the FlowDefinition level (tenants
// that override a standard flow keep ownership of the row).
//
// Runs BEFORE InitializeStandardActions (see PlatformConfigPack.Apply).
// Ordering matters: an action row's FlowDeveloperName references a flow
// that must already exist when the dispatcher's first request resolves
// the pair, so flows are seeded first. Changing this order requires
// updating both the call site and this comment together.
func InitializeStandardFlows(rawDB *sql.DB) error {
	log.Println("🔧 Initializing standard flows...")
	start := time.Now()

	seeds, err := loadStandardFlowSeeds()
	if err != nil {
		return err
	}
	if len(seeds) == 0 {
		log.Println("   ↩ No standard flows to seed")
		return nil
	}

	now := time.Now().Format("2006-01-02 15:04:05")

	// One-shot Source-column backfills for upgraded clusters. Deliberately
	// run in bootstrap Go (not the data-spec pure SQL path) because the
	// discriminator between "pack-owned, untouched" and "admin PATCHed in
	// place before the Source column existed" lives in the shipped pack
	// bytes / hashes embedded in the binary.
	if err := migrateFlowSourceBackfill(rawDB, seeds, now); err != nil {
		return err
	}
	if err := migrateHistoricalFlowSourceBackfill(rawDB, seeds, now); err != nil {
		return err
	}
	if err := migrateTenantFlowSourceNormalize(rawDB, now); err != nil {
		return err
	}

	for _, seed := range seeds {
		if err := upsertStandardFlow(rawDB, seed, now); err != nil {
			return err
		}
	}

	log.Printf("✅ Standard flows initialized (%d seeded, %v)",
		len(seeds), time.Since(start).Round(time.Millisecond))
	return nil
}

func loadStandardFlowSeeds() ([]standardFlowSeed, error) {
	entries, err := fs.ReadDir(standardFlowsFS, standardFlowPackDir)
	if err != nil {
		return nil, apperrors.NewInternalError("read standard flows pack", err)
	}
	seeds := make([]standardFlowSeed, 0, len(entries))
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".flow.json") {
			continue
		}
		developer := strings.TrimSuffix(entry.Name(), ".flow.json")
		raw, err := fs.ReadFile(standardFlowsFS, standardFlowPackDir+"/"+entry.Name())
		if err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("read %s", entry.Name()), err)
		}

		var header struct {
			Label       string `json:"label"`
			Description string `json:"description"`
			ProcessType string `json:"processType"`
		}
		if err := json.Unmarshal(raw, &header); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s header", entry.Name()), err)
		}
		if header.ProcessType == "" {
			header.ProcessType = "AutoLaunchedFlow"
		}
		if header.Label == "" {
			header.Label = developer
		}

		seeds = append(seeds, standardFlowSeed{
			DeveloperName: developer,
			Label:         header.Label,
			Description:   header.Description,
			ProcessType:   header.ProcessType,
			MetadataRaw:   raw,
		})
	}
	return seeds, nil
}

func standardFlowDefinitionID(developerName string) string {
	return "flowdef_std_" + strings.ToLower(developerName)
}

func standardFlowVersionID(developerName string, version int) string {
	return fmt.Sprintf("flowver_std_%s_v%d", strings.ToLower(developerName), version)
}

// sqlQuerier is the subset of database/sql's read+write surface shared by
// *sql.DB and *sql.Tx. Seeder helpers take this interface so upsertStandardFlow
// can open one transaction per seed and thread it through every read + write
// for that seed — no helper-side plumbing needed.
type sqlQuerier interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
	Query(query string, args ...interface{}) (*sql.Rows, error)
	QueryRow(query string, args ...interface{}) *sql.Row
}

// upsertStandardFlow seeds a standard flow with Salesforce-shaped version
// semantics under a per-seed transaction:
//
//   - First install (no FlowDefinition row): create the definition and
//     activate v1 as the standard starter content.
//   - Subsequent boots with an existing definition: leave the definition
//     alone. Never overwrite Metadata on an already-seeded version, and
//     never touch ActiveVersionId unless the current Active row is still
//     Source='standard' (see publishStandardFlowDraftIfNew).
//
// All reads (definition existence, byte-equal check, next-version lookup,
// Source probe) and writes for a single seed run in one sql.Tx. This
// closes the upgrade-path atomicity gap: if the FlowDefinition pointer
// UPDATE fails after an Obsolete UPDATE on the old active row, rollback
// restores v1.Status=Active and nothing in the definition points at a
// partial state. The seeder is therefore insert-if-absent, not
// overwrite-on-every-run, AND atomic per seed.
func upsertStandardFlow(db *sql.DB, seed standardFlowSeed, now string) error {
	tx, err := db.Begin()
	if err != nil {
		return apperrors.NewInternalError(
			fmt.Sprintf("begin tx for standard flow %q", seed.DeveloperName), err)
	}
	// Rollback is a no-op after Commit (returns sql.ErrTxDone, swallowed).
	// Any return-before-Commit below triggers a real rollback.
	defer func() {
		if rbErr := tx.Rollback(); rbErr != nil && rbErr != sql.ErrTxDone {
			log.Printf("standard_flows: rollback failed for %q: %v", seed.DeveloperName, rbErr)
		}
	}()

	defState, err := getStandardFlowDefinitionState(tx, seed.DeveloperName)
	if err != nil {
		return err
	}

	if defState == nil {
		if err := installStandardFlowInitial(tx, standardFlowDefinitionID(seed.DeveloperName), seed, now); err != nil {
			return err
		}
	} else {
		if defState.IsCustomized {
			log.Printf("   ⚠️  %s: preserved customized FlowDefinition; skipping standard reconcile", seed.DeveloperName)
			return nil
		}
		if err := publishStandardFlowDraftIfNew(tx, defState, seed, now); err != nil {
			return err
		}
	}

	if err := tx.Commit(); err != nil {
		return apperrors.NewInternalError(
			fmt.Sprintf("commit tx for standard flow %q", seed.DeveloperName), err)
	}
	return nil
}

type standardFlowDefinitionState struct {
	ID              string
	ActiveVersionID string
	LatestVersionID string
	IsDeleted       bool
	IsCustomized    bool
}

func getStandardFlowDefinitionState(q sqlQuerier, developerName string) (*standardFlowDefinitionState, error) {
	var state standardFlowDefinitionState
	var activeVersionID, latestVersionID sql.NullString
	var isDeleted, isCustomized sql.NullBool
	err := q.QueryRow(
		`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized
		 FROM FlowDefinition WHERE DeveloperName = ? LIMIT 1`,
		developerName,
	).Scan(&state.ID, &activeVersionID, &latestVersionID, &isDeleted, &isCustomized)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, apperrors.NewInternalError(fmt.Sprintf("lookup FlowDefinition %q", developerName), err)
	}
	if activeVersionID.Valid {
		state.ActiveVersionID = activeVersionID.String
	}
	if latestVersionID.Valid {
		state.LatestVersionID = latestVersionID.String
	}
	if isDeleted.Valid {
		state.IsDeleted = isDeleted.Bool
	}
	if isCustomized.Valid {
		state.IsCustomized = isCustomized.Bool
	}
	return &state, nil
}

// Flow.Source values. Exported so the tooling repository writes the
// matching 'custom' literal — the two writers are the only paths allowed
// to set this column, and they share this single vocabulary.
const (
	FlowSourceStandard = "standard"
	FlowSourceCustom   = "custom"
)

func installStandardFlowInitial(q sqlQuerier, defID string, seed standardFlowSeed, now string) error {
	versionID := standardFlowVersionID(seed.DeveloperName, 1)

	defSQL := `INSERT INTO FlowDefinition
		(Id, DeveloperName, MasterLabel, Description, ProcessType,
		 TriggerType, TriggerObjectOrEvent, IsDeleted, IsCustomized,
		 CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, NULL, NULL, 0, 0, ?, ?)`
	if _, err := q.Exec(defSQL,
		defID, seed.DeveloperName, seed.Label, nullableString(seed.Description),
		seed.ProcessType, now, now,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to seed FlowDefinition %q", seed.DeveloperName), err)
	}

	flowSQL := `INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description,
		 ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, 1, 'Active', ?, ?, ?, ?, 0, ?, ?)`
	if _, err := q.Exec(flowSQL,
		versionID, defID, seed.Label,
		nullableString(seed.Description), seed.ProcessType, string(seed.MetadataRaw),
		FlowSourceStandard, now, now,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to seed Flow v1 %q", seed.DeveloperName), err)
	}

	pointerSQL := `UPDATE FlowDefinition
		SET ActiveVersionId = ?, LatestVersionId = ?, LastModifiedDate = ?
		WHERE Id = ?`
	if _, err := q.Exec(pointerSQL, versionID, versionID, now, defID); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to activate standard flow %q", seed.DeveloperName), err)
	}
	return nil
}

// publishStandardFlowDraftIfNew handles the upgrade path when a
// FlowDefinition already exists.
//
// If the pack ships byte-identical metadata, it is a no-op. Otherwise it
// inserts a new version. The critical branch is: do we auto-advance
// ActiveVersionId, or land as a Draft and wait?
//
//   - If the currently-active Flow row is Source='standard' (pack-owned
//     and unmodified — any admin in-place edit flips it to 'custom' via
//     UpdateFlowVersion), we auto-activate the new version and mark the
//     old Active row 'Obsolete'. Tenants get pack fixes automatically.
//   - Otherwise we stay conservative: new version lands as Draft, and we
//     leave ActiveVersionId alone. "Otherwise" covers:
//   - admin cloned/authored the active version (Source='custom'),
//   - admin edited the standard row in place (flipped to 'custom'),
//   - ActiveVersionId is NULL or references a missing row (never
//     happens in the normal bootstrap path, but stay defensive).
//
// Admins who want to pin v1 through a pack upgrade clone it via Flow
// Builder (v2 lands Source='custom') and activate v2 — which is the same
// escape hatch Salesforce expects.
func publishStandardFlowDraftIfNew(q sqlQuerier, defState *standardFlowDefinitionState, seed standardFlowSeed, now string) error {
	defID := defState.ID
	// If the standard metadata is byte-identical to an existing row for
	// this definition, there's no new content to publish.
	matchingVersionID, err := existingStandardFlowVersionID(q, defID, seed.MetadataRaw)
	if err != nil {
		return err
	}
	if matchingVersionID != "" {
		activeVersionID := defState.ActiveVersionID
		latestVersionID := defState.LatestVersionID
		if defState.IsDeleted {
			activeVersionID = matchingVersionID
			latestVersionID = matchingVersionID
		}
		return updateManagedFlowDefinition(q, defID, seed, activeVersionID, latestVersionID, now)
	}

	nextVersion, err := nextStandardFlowVersion(q, defID)
	if err != nil {
		return err
	}
	nextVersionID := standardFlowVersionID(seed.DeveloperName, nextVersion)

	autoActivate := defState.IsDeleted
	if !autoActivate {
		autoActivate, err = activeVersionIsPackOwned(q, defID)
		if err != nil {
			return err
		}
	}

	// Status literal is interpolated (not a bound parameter) so the SQL
	// text itself records which branch ran — helps tests assert the
	// auto-advance decision via SQL-text regex instead of positional
	// argument inspection. Only two well-formed string constants are
	// ever produced, so no injection vector.
	status := "Draft"
	if autoActivate {
		status = "Active"
	}

	flowSQL := fmt.Sprintf(`INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description,
		 ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, '%s', ?, ?, ?, ?, 0, ?, ?)`, status)
	if _, err := q.Exec(flowSQL,
		nextVersionID, defID, seed.Label, nextVersion,
		nullableString(seed.Description), seed.ProcessType, string(seed.MetadataRaw),
		FlowSourceStandard, now, now,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to publish standard flow v%d %q", nextVersion, seed.DeveloperName), err)
	}

	if autoActivate {
		// Mark the previously-active (pack-owned) version Obsolete — same
		// convention as MetadataRepository.ActivateFlowVersion. Bound by
		// `Id != nextVersionID` so the newly-activated row we just
		// inserted is never the one marked Obsolete (it already got
		// Status='Active' in the INSERT above).
		if _, err := q.Exec(
			`UPDATE Flow SET Status = 'Obsolete', LastModifiedDate = ?
			 WHERE DefinitionId = ? AND Status = 'Active' AND Id != ?`,
			now, defID, nextVersionID,
		); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to obsolete prior active version of %q", seed.DeveloperName), err)
		}
		if _, err := q.Exec(
			`UPDATE FlowDefinition
			 SET MasterLabel = ?, Description = ?, ProcessType = ?, ActiveVersionId = ?, LatestVersionId = ?, IsDeleted = 0, LastModifiedDate = ?
			 WHERE Id = ?`,
			seed.Label, nullableString(seed.Description), seed.ProcessType,
			nextVersionID, nextVersionID, now, defID,
		); err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to advance ActiveVersionId for %q", seed.DeveloperName), err)
		}
		log.Printf("   ⚡ %s: auto-advanced to v%d (prior active row was pack-owned and unmodified)",
			seed.DeveloperName, nextVersion)
		return nil
	}

	// Conservative path: admin owns the active version (or pointer is
	// missing). Advance LatestVersionId so Flow Builder lists the new
	// draft, leave ActiveVersionId alone.
	if err := updateManagedFlowDefinition(q, defID, seed, defState.ActiveVersionID, nextVersionID, now); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to advance LatestVersionId for %q", seed.DeveloperName), err)
	}
	return nil
}

func updateManagedFlowDefinition(q sqlQuerier, defID string, seed standardFlowSeed, activeVersionID, latestVersionID, now string) error {
	_, err := q.Exec(
		`UPDATE FlowDefinition
		 SET MasterLabel = ?, Description = ?, ProcessType = ?, ActiveVersionId = ?, LatestVersionId = ?, IsDeleted = 0, LastModifiedDate = ?
		 WHERE Id = ?`,
		seed.Label, nullableString(seed.Description), seed.ProcessType,
		nullableString(activeVersionID), nullableString(latestVersionID), now, defID,
	)
	return err
}

// activeVersionIsPackOwned returns true iff FlowDefinition.ActiveVersionId
// points at a Flow row whose Source is 'standard'. LEFT JOIN over
// ActiveVersionId handles both "no active version" (NULL) and "dangling
// pointer" (ActiveVersionId references a deleted row) — in both cases
// the Source returned is NULL/invalid, which maps to "not pack-owned"
// → conservative branch (do not auto-advance).
func activeVersionIsPackOwned(q sqlQuerier, defID string) (bool, error) {
	var src sql.NullString
	err := q.QueryRow(
		`SELECT f.Source
		 FROM FlowDefinition fd
		 LEFT JOIN Flow f ON f.Id = fd.ActiveVersionId
		 WHERE fd.Id = ?`,
		defID,
	).Scan(&src)
	switch {
	case err == sql.ErrNoRows:
		// FlowDefinition missing — shouldn't happen in the seeder path
		// (standardFlowDefinitionExists guards), but return false so any
		// future caller that slips in here stays on the safe branch.
		return false, nil
	case err != nil:
		return false, apperrors.NewInternalError(fmt.Sprintf("lookup active Flow source for %q", defID), err)
	}
	if !src.Valid {
		return false, nil
	}
	return src.String == FlowSourceStandard, nil
}

// StandardFlowActiveVersionIsPackOwned is the runtime counterpart of the
// seeder's active-version guard. It returns true only when the named
// FlowDefinition is still pack-owned and its active Flow version is a
// Source='standard' row. Callers use this before applying L1-only privileges
// that were reviewed against the shipped standard flow graph.
func StandardFlowActiveVersionIsPackOwned(db *sql.DB, developerName string) (bool, error) {
	if db == nil || strings.TrimSpace(developerName) == "" {
		return false, nil
	}
	var isCustomized sql.NullBool
	var src sql.NullString
	err := db.QueryRow(
		`SELECT fd.IsCustomized, f.Source
		   FROM FlowDefinition fd
		   LEFT JOIN Flow f ON f.Id = fd.ActiveVersionId AND f.IsDeleted = 0
		  WHERE fd.DeveloperName = ? AND fd.IsDeleted = 0
		  LIMIT 1`,
		developerName,
	).Scan(&isCustomized, &src)
	switch {
	case err == sql.ErrNoRows:
		return false, nil
	case err != nil:
		return false, apperrors.NewInternalError(fmt.Sprintf("lookup active Flow source for %q", developerName), err)
	}
	if isCustomized.Valid && isCustomized.Bool {
		return false, nil
	}
	if !src.Valid {
		return false, nil
	}
	return src.String == FlowSourceStandard, nil
}

func nextStandardFlowVersion(q sqlQuerier, defID string) (int, error) {
	var max sql.NullInt64
	if err := q.QueryRow(`SELECT MAX(VersionNumber) FROM Flow WHERE DefinitionId = ?`, defID).Scan(&max); err != nil {
		return 0, apperrors.NewInternalError(fmt.Sprintf("read max Flow version for %q", defID), err)
	}
	if !max.Valid {
		return 1, nil
	}
	return int(max.Int64) + 1, nil
}

func existingStandardFlowVersionID(q sqlQuerier, defID string, metadata []byte) (string, error) {
	rows, err := q.Query(`SELECT Id, Metadata FROM Flow WHERE DefinitionId = ?`, defID)
	if err != nil {
		return "", apperrors.NewInternalError(fmt.Sprintf("scan Flow versions for %q", defID), err)
	}
	defer rows.Close()
	target := string(metadata)
	for rows.Next() {
		var versionID, existing string
		if err := rows.Scan(&versionID, &existing); err != nil {
			return "", apperrors.NewInternalError(fmt.Sprintf("scan Flow.Metadata for %q", defID), err)
		}
		if existing == target {
			return versionID, nil
		}
	}
	return "", rows.Err()
}

// flowSourceBackfillGateKey is the _System_BootstrapState StateKey that
// records whether migrateFlowSourceBackfill has finished successfully on
// this deployment. Scoped "v1" so a future backfill variant (e.g. one
// that learns to consult a pack-version history) can land as a separate
// gate without re-running the current one.
const flowSourceBackfillGateKey = "flow_source_backfill_v1"

// flowSourceHistoricalBackfillGateKey records the second Source migration.
// v1 could only recognize rows byte-equal to the currently-shipped pack.
// v2 adds a tiny historical manifest for standard flow bytes that shipped
// before this PR changed Lead Convert flow metadata. It remains a one-time
// bootstrap repair; runtime still reads only Flow.Source.
const flowSourceHistoricalBackfillGateKey = "flow_source_backfill_v2"

var historicalStandardFlowMetadataSHA256 = map[string][]string{
	// origin/main before PR #2032. These are L1 standard pack bytes, not
	// tenant/org metadata. Matching rows are safe to reclaim only when the
	// FlowDefinition is still uncustomized and the Flow row has the seeder's
	// deterministic standard Id prefix.
	"Std_ConvertLead": {
		"4f56241ed20201696d4da6ea9bbc4024daf37b025e7bfc6772f42060594de2dd",
	},
	"Std_RelinkLeadChildren": {
		"ca2bc533506e68432a68a65ee6b973d0ade0968336545723c6926a522974aba6",
	},
}

// migrateFlowSourceBackfill is a one-shot Go migration that flips
// Flow.Source from 'custom' (the DDL column default applied on upgrade)
// to 'standard' for pack-seeded rows whose stored Metadata is
// byte-identical to the currently-shipped pack.
//
// Why this lives in Go, not a pure SQL data-spec: we need the pack bytes
// embedded in the binary to tell "pack-owned, untouched" from "admin
// PATCHed in place before the Source column existed". Pure SQL can only
// see the Id prefix + current column value — that's not enough to avoid
// re-claiming admin metadata as platform metadata. See
// data-specs/2026-04-23e-backfill-flow-source.md for the audit-side doc.
//
// Conservative posture: byte-mismatch leaves Source='custom'. Worst case
// is a pack-owned row whose content came from an older pack shipment that
// differs from today's bytes — it stays 'custom', and the next pack
// upgrade lands a Draft instead of auto-advancing. That is the pre-fix
// behavior shipped by past versions of the seeder; no regression.
// Admin-edit protection is the higher-priority invariant.
//
// Gate stamping runs ONCE at the end, only after every seed's UPDATE
// succeeded. A partial-success (one seed's UPDATE errors mid-loop) leaves
// the gate row untouched, so the next boot retries from scratch —
// idempotent because the UPDATE's WHERE clause filters rows that have
// already been flipped.
func migrateFlowSourceBackfill(db *sql.DB, seeds []standardFlowSeed, now string) error {
	var status string
	err := db.QueryRow(
		fmt.Sprintf("SELECT Status FROM `%s` WHERE `StateKey` = ?", bootstrapStateTable),
		flowSourceBackfillGateKey,
	).Scan(&status)
	switch {
	case err == sql.ErrNoRows:
		// first run — fall through
	case err != nil:
		return apperrors.NewInternalError(
			fmt.Sprintf("check %s gate", flowSourceBackfillGateKey), err)
	case status == "applied":
		return nil
	}

	totalFlipped := 0
	for _, seed := range seeds {
		defID := standardFlowDefinitionID(seed.DeveloperName)
		idPrefix := fmt.Sprintf("flowver_std_%s_v%%", strings.ToLower(seed.DeveloperName))
		result, err := db.Exec(
			`UPDATE Flow
			 SET Source = ?, LastModifiedDate = ?
			 WHERE DefinitionId = ?
			   AND Id LIKE ?
			   AND Source = ?
			   AND Metadata = ?`,
			FlowSourceStandard, now, defID, idPrefix, FlowSourceCustom, string(seed.MetadataRaw),
		)
		if err != nil {
			return apperrors.NewInternalError(
				fmt.Sprintf("backfill Flow.Source for %q", seed.DeveloperName), err)
		}
		if n, err := result.RowsAffected(); err == nil {
			totalFlipped += int(n)
		}
	}

	// Stamp the gate ONLY after every seed's UPDATE succeeded. Per review:
	// do not stamp mid-loop — a partial-success restart must re-run from
	// scratch (safe because the byte-equal UPDATE is idempotent).
	if _, err := db.Exec(
		fmt.Sprintf(
			"INSERT INTO `%s` (`StateKey`,`RunID`,`Phase`,`Status`,`LastError`,`CreatedDate`,`LastModifiedDate`) "+
				"VALUES (?,?,?,?,?,?,?) "+
				"ON DUPLICATE KEY UPDATE `Status`=VALUES(`Status`),`LastModifiedDate`=VALUES(`LastModifiedDate`)",
			bootstrapStateTable),
		flowSourceBackfillGateKey, now, "flow-source-backfill", "applied", "", now, now,
	); err != nil {
		return apperrors.NewInternalError(
			fmt.Sprintf("stamp %s gate", flowSourceBackfillGateKey), err)
	}

	if totalFlipped > 0 {
		log.Printf("   📝 Flow.Source backfill: flipped %d pack-seeded row(s) to 'standard' (byte-match against shipped pack)", totalFlipped)
	}
	return nil
}

func migrateHistoricalFlowSourceBackfill(db *sql.DB, seeds []standardFlowSeed, now string) error {
	var status string
	err := db.QueryRow(
		fmt.Sprintf("SELECT Status FROM `%s` WHERE `StateKey` = ?", bootstrapStateTable),
		flowSourceHistoricalBackfillGateKey,
	).Scan(&status)
	switch {
	case err == sql.ErrNoRows:
		// first run — fall through
	case err != nil:
		return apperrors.NewInternalError(
			fmt.Sprintf("check %s gate", flowSourceHistoricalBackfillGateKey), err)
	case status == "applied":
		return nil
	}

	totalFlipped := 0
	for _, seed := range seeds {
		hashes := historicalStandardFlowMetadataSHA256[seed.DeveloperName]
		if len(hashes) == 0 {
			continue
		}
		defID := standardFlowDefinitionID(seed.DeveloperName)
		idPrefix := fmt.Sprintf("flowver_std_%s_v%%", strings.ToLower(seed.DeveloperName))
		placeholders := strings.TrimRight(strings.Repeat("?,", len(hashes)), ",")
		args := []interface{}{FlowSourceStandard, now, defID, idPrefix, FlowSourceCustom}
		for _, hash := range hashes {
			args = append(args, hash)
		}
		result, err := db.Exec(
			fmt.Sprintf(
				`UPDATE Flow f
				   JOIN FlowDefinition fd ON fd.Id = f.DefinitionId
				   SET f.Source = ?, f.LastModifiedDate = ?
				 WHERE f.DefinitionId = ?
				   AND f.Id LIKE ?
				   AND f.Source = ?
				   AND (fd.IsCustomized = 0 OR fd.IsCustomized IS NULL)
				   AND SHA2(f.Metadata, 256) IN (%s)`,
				placeholders),
			args...,
		)
		if err != nil {
			return apperrors.NewInternalError(
				fmt.Sprintf("backfill historical Flow.Source for %q", seed.DeveloperName), err)
		}
		if n, err := result.RowsAffected(); err == nil {
			totalFlipped += int(n)
		}
	}

	if _, err := db.Exec(
		fmt.Sprintf(
			"INSERT INTO `%s` (`StateKey`,`RunID`,`Phase`,`Status`,`LastError`,`CreatedDate`,`LastModifiedDate`) "+
				"VALUES (?,?,?,?,?,?,?) "+
				"ON DUPLICATE KEY UPDATE `Status`=VALUES(`Status`),`LastModifiedDate`=VALUES(`LastModifiedDate`)",
			bootstrapStateTable),
		flowSourceHistoricalBackfillGateKey, now, "flow-source-backfill-v2", "applied", "", now, now,
	); err != nil {
		return apperrors.NewInternalError(
			fmt.Sprintf("stamp %s gate", flowSourceHistoricalBackfillGateKey), err)
	}

	if totalFlipped > 0 {
		log.Printf("   📝 Flow.Source historical backfill: flipped %d pack-seeded row(s) to 'standard' (known historical pack hash)", totalFlipped)
	}
	return nil
}

// tenantFlowSourceNormalizeGateKey records whether the legacy
// Flow.Source='tenant' normalization has run on this deployment. The
// 'tenant' value was a short-lived branch-local discriminator introduced
// by #2135 and removed by #2173 — it is not a valid Salesforce Flow.Source
// enum (only 'standard' and 'custom' are valid per SF API v63.0). Once
// normalized, no Flow row in any cluster should ever carry it again.
const tenantFlowSourceNormalizeGateKey = "tenant_flow_source_normalize_v1"

// migrateTenantFlowSourceNormalize is a one-shot Go migration that
// rewrites every existing Flow row with Source='tenant' to Source='custom'.
//
// Why this is unconditional (no metadata-byte gate, no prefix gate, no
// IsCustomized check) unlike migrateFlowSourceBackfill: 'tenant' was never
// a valid public Flow.Source enum value, never an admin-set value, and
// never produced by any external loader. Its only producer was a brief
// window of #2135's tenant-overlay write path, which #2173 has already
// removed. There is no failure mode where a 'tenant' row should be
// preserved — every such row violates the SF v63.0 contract and must be
// repaired.
//
// Why this lives in bootstrap, not as a SQL data-spec or only inside
// `cmd/setup apply-tenant`: the SF-contract violation exists on every
// upgraded cluster regardless of whether the tenant-overlay command is
// re-run, so the repair must execute on the normal upgrade path. It is
// audit-documented as a data-spec at
// data-specs/2026-04-28-normalize-tenant-flow-source.md.
//
// Idempotency: gate stamping runs ONLY after the UPDATE succeeds. If the
// stamp fails the next boot retries from scratch — safe because the
// Source='tenant' WHERE filter already filters out rows the previous run
// flipped.
func migrateTenantFlowSourceNormalize(db *sql.DB, now string) error {
	var status string
	err := db.QueryRow(
		fmt.Sprintf("SELECT Status FROM `%s` WHERE `StateKey` = ?", bootstrapStateTable),
		tenantFlowSourceNormalizeGateKey,
	).Scan(&status)
	switch {
	case err == sql.ErrNoRows:
		// first run — fall through
	case err != nil:
		return apperrors.NewInternalError(
			fmt.Sprintf("check %s gate", tenantFlowSourceNormalizeGateKey), err)
	case status == "applied":
		return nil
	}

	result, err := db.Exec(
		`UPDATE Flow SET Source = ?, LastModifiedDate = ? WHERE Source = 'tenant'`,
		FlowSourceCustom, now,
	)
	if err != nil {
		return apperrors.NewInternalError("normalize Flow.Source='tenant' to 'custom'", err)
	}
	var flipped int64
	if n, err := result.RowsAffected(); err == nil {
		flipped = n
	}

	if _, err := db.Exec(
		fmt.Sprintf(
			"INSERT INTO `%s` (`StateKey`,`RunID`,`Phase`,`Status`,`LastError`,`CreatedDate`,`LastModifiedDate`) "+
				"VALUES (?,?,?,?,?,?,?) "+
				"ON DUPLICATE KEY UPDATE `Status`=VALUES(`Status`),`LastModifiedDate`=VALUES(`LastModifiedDate`)",
			bootstrapStateTable),
		tenantFlowSourceNormalizeGateKey, now, "tenant-flow-source-normalize", "applied", "", now, now,
	); err != nil {
		return apperrors.NewInternalError(
			fmt.Sprintf("stamp %s gate", tenantFlowSourceNormalizeGateKey), err)
	}

	if flipped > 0 {
		log.Printf("   📝 Flow.Source normalize: flipped %d historical 'tenant' row(s) to 'custom' (SF v63.0 contract repair)", flipped)
	}
	return nil
}