package packs

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

	"github.com/nexuscrm/automation/triggerprojection"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
)

// standardFlowSeed describes one shipped Flow (parent FlowDefinition + desired
// 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
	Status        string
	MetadataRaw   []byte
}

// InitializeStandardFlowsFromFS upserts every .flow.json file under root as an
// Flow + FlowDefinition row in the authored Active/Draft state. Idempotent
// with IsCustomized reconciliation at the FlowDefinition level (tenants
// that override a standard flow keep ownership of the row).
//
// Runs BEFORE InitializeStandardActions (see the current app-pack apply path).
// 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 InitializeStandardFlowsFromFS(rawDB *sql.DB, flowFS fs.FS, root string) error {
	log.Println("🔧 Initializing standard flows...")
	start := time.Now()

	seeds, err := loadStandardFlowSeedsFromFS(flowFS, root)
	if err != nil {
		return err
	}
	return initializeStandardFlowSeeds(rawDB, seeds, start, FlowSourceStandard)
}

func InitializeStandardFlowsFromPayloads(rawDB *sql.DB, payloads map[string][]byte) error {
	return InitializeStandardFlowsFromPayloadsForOwner(rawDB, payloads, FlowSourceStandard)
}

func InitializeStandardFlowsFromPayloadsForOwner(rawDB *sql.DB, payloads map[string][]byte, ownerPack string) error {
	log.Println("🔧 Initializing standard flows...")
	start := time.Now()

	seeds, err := LoadStandardFlowSeedsFromPayloads(payloads)
	if err != nil {
		return err
	}
	return initializeStandardFlowSeeds(rawDB, seeds, start, normalizeStandardFlowOwnerPack(ownerPack))
}

func initializeStandardFlowSeeds(rawDB *sql.DB, seeds []standardFlowSeed, start time.Time, ownerPack string) error {
	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 backfill 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 source payload
	// bytes passed by the caller.
	if err := migrateFlowSourceBackfill(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, ownerPack); err != nil {
			return err
		}
	}

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

func loadStandardFlowSeedsFromFS(flowFS fs.FS, root string) ([]standardFlowSeed, error) {
	root = strings.Trim(strings.TrimSpace(root), "/")
	if root == "" {
		return nil, apperrors.NewInternalError("standard flow root is required", nil)
	}
	entries, err := fs.ReadDir(flowFS, root)
	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(flowFS, root+"/"+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"`
			Status      string `json:"status"`
		}
		if err := json.Unmarshal(raw, &header); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s header", entry.Name()), err)
		}
		var meta models.FlowMetadata
		if err := json.Unmarshal(raw, &meta); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s metadata", entry.Name()), err)
		}
		if header.ProcessType == "" {
			header.ProcessType = "AutoLaunchedFlow"
		}
		status, err := normalizeStandardFlowDesiredStatus(header.Status)
		if err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s status", entry.Name()), err)
		}
		if err := triggerprojection.ValidateAtomicTriggerSupport(
			&models.FlowDefinition{ProcessType: header.ProcessType},
			&models.FlowVersion{ProcessType: header.ProcessType, Metadata: meta},
		); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("validate %s atomic trigger support", entry.Name()), err)
		}
		if header.Label == "" {
			header.Label = developer
		}

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

func LoadStandardFlowSeedsFromPayloads(payloads map[string][]byte) ([]standardFlowSeed, error) {
	if len(payloads) == 0 {
		return nil, nil
	}
	names := make([]string, 0, len(payloads))
	for name := range payloads {
		names = append(names, name)
	}
	sort.Strings(names)
	seeds := make([]standardFlowSeed, 0, len(names))
	for _, name := range names {
		if !strings.HasSuffix(name, ".flow.json") {
			return nil, apperrors.NewInternalError(fmt.Sprintf("standard flow payload %q must end with .flow.json", name), nil)
		}
		developer := strings.TrimSuffix(name, ".flow.json")
		raw := payloads[name]
		var header struct {
			Label       string `json:"label"`
			Description string `json:"description"`
			ProcessType string `json:"processType"`
			Status      string `json:"status"`
		}
		if err := json.Unmarshal(raw, &header); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s header", name), err)
		}
		var meta models.FlowMetadata
		if err := json.Unmarshal(raw, &meta); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s metadata", name), err)
		}
		if header.ProcessType == "" {
			header.ProcessType = "AutoLaunchedFlow"
		}
		status, err := normalizeStandardFlowDesiredStatus(header.Status)
		if err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("parse %s status", name), err)
		}
		if err := triggerprojection.ValidateAtomicTriggerSupport(
			&models.FlowDefinition{ProcessType: header.ProcessType},
			&models.FlowVersion{ProcessType: header.ProcessType, Metadata: meta},
		); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("validate %s atomic trigger support", name), err)
		}
		if header.Label == "" {
			header.Label = developer
		}
		seeds = append(seeds, standardFlowSeed{
			DeveloperName: developer,
			Label:         header.Label,
			Description:   header.Description,
			ProcessType:   header.ProcessType,
			Status:        status,
			MetadataRaw:   append([]byte(nil), raw...),
		})
	}
	return seeds, nil
}

// StandardFlowsPackHashFromFS returns the shipped standard-flow payload hash.
func StandardFlowsPackHashFromFS(flowFS fs.FS, root string) (string, error) {
	payload, err := StandardFlowsPackPayloadFromFS(flowFS, root)
	if err != nil {
		return "", err
	}
	return sha256Hex(payload), nil
}

func StandardFlowsPackPayloadFromFS(flowFS fs.FS, root string) ([]byte, error) {
	root = strings.Trim(strings.TrimSpace(root), "/")
	if root == "" {
		return nil, apperrors.NewInternalError("standard flow root is required", nil)
	}
	entries, err := fs.ReadDir(flowFS, root)
	if err != nil {
		return nil, apperrors.NewInternalError("read standard flows pack", err)
	}
	var payload []byte
	payload = append(payload, []byte("standard_flows:v1\n")...)
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".flow.json") {
			continue
		}
		raw, err := fs.ReadFile(flowFS, root+"/"+entry.Name())
		if err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("read %s", entry.Name()), err)
		}
		payload = append(payload, []byte(entry.Name())...)
		payload = append(payload, '\n')
		payload = append(payload, raw...)
		payload = append(payload, '\n')
	}
	return payload, nil
}

// VerifyStandardFlows checks that every shipped standard flow has a
// FlowDefinition and at least one live Source='standard' Flow version with the
// expected metadata body. Admin-customized definitions or active custom
// versions can coexist, but the pack-owned version must remain present so the
// installed app-pack registry does not claim a missing payload is healthy.
func VerifyStandardFlowsFromFS(db *sql.DB, flowFS fs.FS, root string) error {
	seeds, err := loadStandardFlowSeedsFromFS(flowFS, root)
	if err != nil {
		return err
	}
	return verifyStandardFlowSeeds(db, seeds, FlowSourceStandard)
}

func VerifyStandardFlowsFromPayloads(db *sql.DB, payloads map[string][]byte) error {
	return VerifyStandardFlowsFromPayloadsForOwner(db, payloads, FlowSourceStandard)
}

func VerifyStandardFlowsFromPayloadsForOwner(db *sql.DB, payloads map[string][]byte, ownerPack string) error {
	seeds, err := LoadStandardFlowSeedsFromPayloads(payloads)
	if err != nil {
		return err
	}
	return verifyStandardFlowSeeds(db, seeds, normalizeStandardFlowOwnerPack(ownerPack))
}

func verifyStandardFlowSeeds(db *sql.DB, seeds []standardFlowSeed, ownerPack string) error {
	var errs []string
	for _, seed := range seeds {
		var definitionID string
		var gotOwnerPack sql.NullString
		var activeVersionID sql.NullString
		var gotProcessType, gotTriggerType, gotTriggerObject sql.NullString
		var isDeleted, isCustomized bool
		err := db.QueryRow(`
SELECT Id, OwnerPack, ActiveVersionId, IsDeleted, IsCustomized, ProcessType, TriggerType, TriggerObjectOrEvent
FROM FlowDefinition
WHERE DeveloperName = ?
LIMIT 1`, seed.DeveloperName).Scan(
			&definitionID, &gotOwnerPack, &activeVersionID, &isDeleted, &isCustomized,
			&gotProcessType, &gotTriggerType, &gotTriggerObject,
		)
		if err == sql.ErrNoRows {
			errs = append(errs, fmt.Sprintf("flow definition %s is missing", seed.DeveloperName))
			continue
		}
		if err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("query FlowDefinition %q", seed.DeveloperName), err)
		}
		if isDeleted {
			errs = append(errs, fmt.Sprintf("flow definition %s is deleted", seed.DeveloperName))
			continue
		}
		if strings.TrimSpace(gotOwnerPack.String) != ownerPack {
			errs = append(errs, fmt.Sprintf("flow definition %s OwnerPack = %q, want %q", seed.DeveloperName, gotOwnerPack.String, ownerPack))
			continue
		}
		if isCustomized {
			continue
		}

		desiredStatus, err := normalizeStandardFlowDesiredStatus(seed.Status)
		if err != nil {
			errs = append(errs, fmt.Sprintf("flow %s has invalid desired status: %v", seed.DeveloperName, err))
			continue
		}
		activeID := strings.TrimSpace(activeVersionID.String)
		if activeID == "" {
			if desiredStatus == models.FlowStatusActive {
				errs = append(errs, fmt.Sprintf("flow %s has no active version", seed.DeveloperName))
				continue
			}
			if err := verifyManagedStandardFlowDraft(db, definitionID, seed); err != nil {
				errs = append(errs, err.Error())
				continue
			}
			projection, err := standardFlowSeedProjection(definitionID, seed)
			if err != nil {
				errs = append(errs, fmt.Sprintf("flow definition %s trigger projection is malformed: %v", seed.DeveloperName, err))
				continue
			}
			if projection.Kind != triggerprojection.KindUnknown &&
				(gotTriggerType.String != projection.TriggerType ||
					gotTriggerObject.String != projection.ObjectOrEvent) {
				errs = append(errs, fmt.Sprintf(
					"flow definition %s trigger projection = (%q, %q), want (%q, %q)",
					seed.DeveloperName,
					gotTriggerType.String,
					gotTriggerObject.String,
					projection.TriggerType,
					projection.ObjectOrEvent,
				))
			}
			continue
		}
		var gotFlowID string
		var gotStatus, gotSource, gotMetadata sql.NullString
		var activeDeleted bool
		err = db.QueryRow(`
SELECT Id, Status, Source, Metadata, IsDeleted
FROM Flow
WHERE Id = ? AND DefinitionId = ?
LIMIT 1`, activeID, definitionID).Scan(&gotFlowID, &gotStatus, &gotSource, &gotMetadata, &activeDeleted)
		if err == sql.ErrNoRows {
			errs = append(errs, fmt.Sprintf("flow %s active version %s is missing", seed.DeveloperName, activeID))
			continue
		}
		if err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("query active standard Flow version %q", seed.DeveloperName), err)
		}
		if activeDeleted {
			errs = append(errs, fmt.Sprintf("flow %s active version %s is deleted", seed.DeveloperName, activeID))
			continue
		}
		if desiredStatus == models.FlowStatusDraft {
			if strings.TrimSpace(gotSource.String) == FlowSourceStandard {
				errs = append(errs, fmt.Sprintf("flow %s is authored Draft but standard version %s is active", seed.DeveloperName, activeID))
				continue
			}
			if err := verifyMatchingStandardFlowDraft(db, definitionID, seed); err != nil {
				errs = append(errs, err.Error())
			}
			// A custom/admin active version owns its own trigger projection.
			continue
		}
		if strings.TrimSpace(gotSource.String) == FlowSourceStandard &&
			strings.TrimSpace(gotStatus.String) != models.FlowStatusActive {
			errs = append(errs, fmt.Sprintf("flow %s active version %s status = %q, want %q", seed.DeveloperName, activeID, gotStatus.String, models.FlowStatusActive))
			continue
		}
		if strings.TrimSpace(gotSource.String) == FlowSourceStandard && gotMetadata.String != string(seed.MetadataRaw) {
			errs = append(errs, fmt.Sprintf("flow %s active version %s metadata differs from shipped metadata", seed.DeveloperName, activeID))
			continue
		}
		if strings.TrimSpace(gotSource.String) != FlowSourceStandard {
			var matchingStandardID string
			err = db.QueryRow(`
SELECT Id
FROM Flow
WHERE DefinitionId = ? AND Source = ? AND IsDeleted = 0 AND Metadata = ?
LIMIT 1`, definitionID, FlowSourceStandard, string(seed.MetadataRaw)).Scan(&matchingStandardID)
			if err == sql.ErrNoRows {
				errs = append(errs, fmt.Sprintf("flow %s is missing shipped standard metadata while active version %s Source = %q", seed.DeveloperName, activeID, gotSource.String))
				continue
			}
			if err != nil {
				return apperrors.NewInternalError(fmt.Sprintf("query shipped standard Flow version %q", seed.DeveloperName), err)
			}
		}
		projection, err := standardFlowSeedProjection(definitionID, seed)
		if err != nil {
			errs = append(errs, fmt.Sprintf("flow definition %s trigger projection is malformed: %v", seed.DeveloperName, err))
			continue
		}
		if strings.TrimSpace(gotSource.String) == FlowSourceStandard &&
			projection.Kind != triggerprojection.KindUnknown &&
			(gotTriggerType.String != projection.TriggerType ||
				gotTriggerObject.String != projection.ObjectOrEvent) {
			errs = append(errs, fmt.Sprintf(
				"flow definition %s trigger projection = (%q, %q), want (%q, %q)",
				seed.DeveloperName,
				gotTriggerType.String,
				gotTriggerObject.String,
				projection.TriggerType,
				projection.ObjectOrEvent,
			))
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("standard flows drift:\n  - %s", strings.Join(errs, "\n  - "))
	}
	return nil
}

func verifyManagedStandardFlowDraft(db *sql.DB, definitionID string, seed standardFlowSeed) error {
	var latestVersionID sql.NullString
	if err := db.QueryRow(`
SELECT LatestVersionId
FROM FlowDefinition
WHERE Id = ?
LIMIT 1`, definitionID).Scan(&latestVersionID); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("query latest standard Flow version %q", seed.DeveloperName), err)
	}
	latestID := strings.TrimSpace(latestVersionID.String)
	if latestID == "" {
		return fmt.Errorf("flow %s is authored Draft but has no latest version", seed.DeveloperName)
	}

	var gotID string
	var gotStatus, gotSource, gotMetadata sql.NullString
	var isDeleted bool
	err := db.QueryRow(`
SELECT Id, Status, Source, Metadata, IsDeleted
FROM Flow
WHERE Id = ? AND DefinitionId = ?
LIMIT 1`, latestID, definitionID).Scan(&gotID, &gotStatus, &gotSource, &gotMetadata, &isDeleted)
	if err == sql.ErrNoRows {
		return fmt.Errorf("flow %s latest draft version %s is missing", seed.DeveloperName, latestID)
	}
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("query latest draft Flow version %q", seed.DeveloperName), err)
	}
	if isDeleted {
		return fmt.Errorf("flow %s latest draft version %s is deleted", seed.DeveloperName, latestID)
	}
	if strings.TrimSpace(gotSource.String) != FlowSourceStandard {
		return fmt.Errorf("flow %s latest draft version %s Source = %q, want %q", seed.DeveloperName, latestID, gotSource.String, FlowSourceStandard)
	}
	if strings.TrimSpace(gotStatus.String) != models.FlowStatusDraft {
		return fmt.Errorf("flow %s latest standard version %s status = %q, want %q", seed.DeveloperName, latestID, gotStatus.String, models.FlowStatusDraft)
	}
	if gotMetadata.String != string(seed.MetadataRaw) {
		return fmt.Errorf("flow %s latest draft version %s metadata differs from shipped metadata", seed.DeveloperName, latestID)
	}
	return nil
}

func verifyMatchingStandardFlowDraft(db *sql.DB, definitionID string, seed standardFlowSeed) error {
	var versionID string
	var status sql.NullString
	err := db.QueryRow(`
SELECT Id, Status
FROM Flow
WHERE DefinitionId = ? AND Source = ? AND IsDeleted = 0 AND Metadata = ?
LIMIT 1`, definitionID, FlowSourceStandard, string(seed.MetadataRaw)).Scan(&versionID, &status)
	if err == sql.ErrNoRows {
		return fmt.Errorf("flow %s is missing shipped Draft metadata while a custom version is active", seed.DeveloperName)
	}
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("query shipped draft Flow version %q", seed.DeveloperName), err)
	}
	if strings.TrimSpace(status.String) != models.FlowStatusDraft {
		return fmt.Errorf("flow %s shipped version %s status = %q, want %q", seed.DeveloperName, versionID, status.String, models.FlowStatusDraft)
	}
	return 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 pessimistic transaction:
//
//   - First install (no FlowDefinition row): stake-out the definition row
//     via INSERT IGNORE, then read state post-lock and activate v1.
//   - Subsequent boots with an existing definition: lock the row, re-read
//     state, and reconcile. Never overwrite Metadata on an already-seeded
//     version, and always converge ActiveVersionId to the shipped standard
//     version while the FlowDefinition remains pack-owned (`IsCustomized=0`).
//
// All reads (definition existence, byte-equal check, next-version lookup,
// Source probe) and writes for a single seed run in one pessimistic tx.
// The FOR UPDATE lock on the FlowDefinition row serializes concurrent
// bootstrap invocations on the same DeveloperName, closing both the
// first-install race and the stale defState write-back gap.
func upsertStandardFlow(db *sql.DB, seed standardFlowSeed, now string, ownerPack string) error {
	tx, cleanup, err := persistence.BeginPessimisticTx(context.Background(), db)
	if err != nil {
		return apperrors.NewInternalError(
			fmt.Sprintf("begin pessimistic tx for standard flow %q", seed.DeveloperName), err)
	}
	defer func() {
		if rbErr := tx.Rollback(); rbErr != nil && rbErr != sql.ErrTxDone {
			log.Printf("standard_flows: rollback failed for %q: %v", seed.DeveloperName, rbErr)
		}
		_ = cleanup()
	}()

	// Stake-out: INSERT IGNORE on FlowDefinition to prevent concurrent
	// first-create races on UNIQUE(DeveloperName).
	defID := standardFlowDefinitionID(seed.DeveloperName)
	if _, err := tx.Exec(`INSERT IGNORE INTO FlowDefinition
		(Id, DeveloperName, MasterLabel, Description, ProcessType,
		 TriggerType, TriggerObjectOrEvent, IsDeleted, IsCustomized,
		 OwnerPack, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, NULL, NULL, 0, 0, ?, ?, ?)`,
		defID, seed.DeveloperName, seed.Label, nullableString(seed.Description),
		seed.ProcessType, ownerPack, now, now,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("stake-out FlowDefinition %q", seed.DeveloperName), err)
	}

	// Lock and re-read state AFTER the stake-out so every version read/write
	// uses the authoritative FlowDefinition row, including legacy rows whose
	// Id differs from the deterministic standard id.
	defState, err := lockStandardFlowDefinitionState(tx, seed.DeveloperName)
	if err != nil {
		return err
	}
	if defState == nil {
		return apperrors.NewInternalError(fmt.Sprintf("FlowDefinition %q disappeared after stake-out", seed.DeveloperName), nil)
	}

	if err := reconcileStandardFlow(tx, defState, seed, now, ownerPack); 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
	OwnerPack       string
	ActiveVersionID string
	LatestVersionID string
	IsDeleted       bool
	IsCustomized    bool
}

type standardFlowVersionState struct {
	ID            string
	Metadata      string
	VersionNumber int
	Status        string
	Source        string
	IsDeleted     bool
}

func lockStandardFlowDefinitionState(q sqlQuerier, developerName string) (*standardFlowDefinitionState, error) {
	var state standardFlowDefinitionState
	var ownerPack, activeVersionID, latestVersionID sql.NullString
	var isDeleted, isCustomized sql.NullBool
	err := q.QueryRow(
		`SELECT Id, OwnerPack, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized
		 FROM FlowDefinition WHERE DeveloperName = ? LIMIT 1 FOR UPDATE`,
		developerName,
	).Scan(&state.ID, &ownerPack, &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 ownerPack.Valid {
		state.OwnerPack = strings.TrimSpace(ownerPack.String)
	}
	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 normalizeStandardFlowOwnerPack(ownerPack string) string {
	ownerPack = strings.TrimSpace(ownerPack)
	if ownerPack == "" {
		return FlowSourceStandard
	}
	return ownerPack
}

func normalizeStandardFlowDesiredStatus(status string) (string, error) {
	switch strings.ToLower(strings.TrimSpace(status)) {
	case "", "active":
		// Missing status is retained as Active compatibility for older packs.
		return models.FlowStatusActive, nil
	case "draft":
		return models.FlowStatusDraft, nil
	default:
		return "", fmt.Errorf("source-backed standard flow status must be Active or Draft")
	}
}

// reconcileStandardFlow handles both first-install and upgrade paths for a
// locked standard FlowDefinition. The pack-authored Active/Draft state is the
// source of truth while the definition remains pack-owned and uncustomized.
func reconcileStandardFlow(q sqlQuerier, defState *standardFlowDefinitionState, seed standardFlowSeed, now string, ownerPack string) error {
	if defState == nil {
		return apperrors.NewInternalError(fmt.Sprintf("missing locked FlowDefinition state for %q", seed.DeveloperName), nil)
	}
	defID := defState.ID
	if defState.OwnerPack != ownerPack {
		return fmt.Errorf("flow definition %s OwnerPack = %q, refusing to install for %q", seed.DeveloperName, defState.OwnerPack, ownerPack)
	}
	if defState.IsCustomized {
		log.Printf("   ⚠️  %s: preserved customized FlowDefinition; skipping standard reconcile", seed.DeveloperName)
		return nil
	}
	versions, err := lockStandardFlowVersions(q, defID)
	if err != nil {
		return err
	}
	// First install on a freshly-staked FlowDefinition row: install the exact
	// authored status. A Draft ships a latest version but no active pointer.
	if defState.ActiveVersionID == "" && defState.LatestVersionID == "" && len(versions) == 0 {
		return installStandardFlowInitial(q, defID, seed, now)
	}
	return publishStandardFlowDraftIfNew(q, defState, versions, defID, seed, now, ownerPack)
}

func installStandardFlowInitial(q sqlQuerier, defID string, seed standardFlowSeed, now string) error {
	versionID := standardFlowVersionID(seed.DeveloperName, 1)
	status, err := normalizeStandardFlowDesiredStatus(seed.Status)
	if err != nil {
		return err
	}

	flowSQL := `INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description,
		 ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, 1, ?, ?, ?, ?, ?, 0, ?, ?)`
	if _, err := q.Exec(flowSQL,
		versionID, defID, seed.Label, status,
		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 = ?`
	activeVersionID := ""
	if status == models.FlowStatusActive {
		activeVersionID = versionID
	}
	if _, err := q.Exec(pointerSQL, nullableString(activeVersionID), versionID, now, defID); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to publish standard flow %q", seed.DeveloperName), err)
	}
	if err := syncManagedFlowDefinitionTriggerProjection(q, defID, seed, now); err != nil {
		return err
	}
	return nil
}

func publishStandardFlowDraftIfNew(q sqlQuerier, defState *standardFlowDefinitionState, versions []standardFlowVersionState, defID string, seed standardFlowSeed, now string, ownerPack string) error {
	desiredStatus, err := normalizeStandardFlowDesiredStatus(seed.Status)
	if err != nil {
		return err
	}
	// Preserve the historical Flow.Source ownership contract: custom/admin
	// active rows may coexist with shipped standard rows. We only reclaim the
	// active pointer when the currently-active row is still pack-owned.
	matchingVersion := existingStandardFlowVersion(versions, seed.MetadataRaw)
	if matchingVersion != nil {
		if desiredStatus == models.FlowStatusDraft &&
			(defState.IsDeleted || defState.ActiveVersionID == "" || activeVersionIsPackOwned(versions, defState.ActiveVersionID)) {
			return deactivateManagedStandardFlowVersion(q, versions, defID, matchingVersion.ID, seed, now, ownerPack)
		}
		autoActivate := desiredStatus == models.FlowStatusActive &&
			(defState.IsDeleted ||
				activeVersionIsPackOwned(versions, defState.ActiveVersionID) ||
				(defState.ActiveVersionID == "" && matchingVersion.Source == FlowSourceStandard))
		if autoActivate &&
			!defState.IsDeleted &&
			defState.ActiveVersionID == matchingVersion.ID &&
			defState.LatestVersionID == matchingVersion.ID &&
			matchingVersion.Status == "Active" {
			if err := updateManagedFlowDefinition(q, defID, seed, matchingVersion.ID, matchingVersion.ID, now, ownerPack); err != nil {
				return err
			}
			return syncManagedFlowDefinitionTriggerProjection(q, defID, seed, now)
		}
		if autoActivate {
			return activateManagedStandardFlowVersion(q, versions, defID, matchingVersion.ID, seed, now, ownerPack)
		}
		return updateManagedFlowDefinition(q, defID, seed, defState.ActiveVersionID, defState.LatestVersionID, now, ownerPack)
	}

	nextVersion := nextStandardFlowVersion(versions)
	nextVersionID := standardFlowVersionID(seed.DeveloperName, nextVersion)
	autoActivate := desiredStatus == models.FlowStatusActive &&
		(defState.IsDeleted || activeVersionIsPackOwned(versions, defState.ActiveVersionID) || defState.ActiveVersionID == "")
	status := models.FlowStatusDraft
	if autoActivate {
		status = models.FlowStatusActive
	}
	flowSQL := `INSERT INTO Flow
		(Id, DefinitionId, MasterLabel, VersionNumber, Status, Description,
		 ProcessType, Metadata, Source, IsDeleted, CreatedDate, LastModifiedDate)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`
	if _, err := q.Exec(flowSQL,
		nextVersionID, defID, seed.Label, nextVersion, status,
		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 {
		if err := activateManagedStandardFlowVersion(q, append(versions, standardFlowVersionState{
			ID:            nextVersionID,
			VersionNumber: nextVersion,
			Status:        status,
			Source:        FlowSourceStandard,
			IsDeleted:     false,
		}), defID, nextVersionID, seed, now, ownerPack); err != nil {
			return err
		}
		log.Printf("   ⚡ %s: converged active standard flow to v%d", seed.DeveloperName, nextVersion)
		return nil
	}
	if desiredStatus == models.FlowStatusDraft && activeVersionIsPackOwned(versions, defState.ActiveVersionID) {
		return deactivateManagedStandardFlowVersion(q, append(versions, standardFlowVersionState{
			ID: nextVersionID, VersionNumber: nextVersion, Status: status,
			Source: FlowSourceStandard,
		}), defID, nextVersionID, seed, now, ownerPack)
	}
	if err := updateManagedFlowDefinition(q, defID, seed, defState.ActiveVersionID, nextVersionID, now, ownerPack); 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, ownerPack string) error {
	_, err := q.Exec(
		`UPDATE FlowDefinition
		 SET MasterLabel = ?, Description = ?, ProcessType = ?, ActiveVersionId = ?, LatestVersionId = ?, OwnerPack = ?, IsDeleted = 0, LastModifiedDate = ?
		 WHERE Id = ?`,
		seed.Label, nullableString(seed.Description), seed.ProcessType,
		nullableString(activeVersionID), nullableString(latestVersionID), ownerPack, now, defID,
	)
	return err
}

// activeVersionIsPackOwned returns true iff the locked version set contains
// FlowDefinition.ActiveVersionId and that row's Source is 'standard'. Missing
// pointers stay conservative and do not auto-advance.
func activeVersionIsPackOwned(versions []standardFlowVersionState, activeVersionID string) bool {
	if strings.TrimSpace(activeVersionID) == "" {
		return false
	}
	for _, v := range versions {
		if v.ID == activeVersionID {
			return v.Source == FlowSourceStandard
		}
	}
	return false
}

func standardFlowSeedProjection(definitionID string, seed standardFlowSeed) (triggerprojection.Projection, error) {
	definition := &models.FlowDefinition{
		Id:            definitionID,
		DeveloperName: seed.DeveloperName,
		ProcessType:   seed.ProcessType,
	}
	version := &models.FlowVersion{
		DefinitionId: definitionID,
		ProcessType:  seed.ProcessType,
		MetadataRaw:  seed.MetadataRaw,
	}
	version.SerializeMetadata()
	return triggerprojection.Resolve(definition, version)
}

func syncManagedFlowDefinitionTriggerProjection(q sqlQuerier, defID string, seed standardFlowSeed, now string) error {
	projection, err := standardFlowSeedProjection(defID, seed)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("resolve trigger projection for %q", seed.DeveloperName), err)
	}
	triggerType := ""
	triggerObjectOrEvent := ""
	if projection.Kind != triggerprojection.KindUnknown {
		triggerType = projection.TriggerType
		triggerObjectOrEvent = projection.ObjectOrEvent
	}
	if _, err := q.Exec(
		`UPDATE FlowDefinition
		 SET TriggerType = ?, TriggerObjectOrEvent = ?, LastModifiedDate = ?
		 WHERE Id = ?`,
		nullableString(triggerType), nullableString(triggerObjectOrEvent), now, defID,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("sync trigger projection for %q", seed.DeveloperName), err)
	}
	return nil
}

func activateManagedStandardFlowVersion(q sqlQuerier, versions []standardFlowVersionState, defID, versionID string, seed standardFlowSeed, now string, ownerPack string) error {
	if _, err := q.Exec(
		`UPDATE Flow SET Status = 'Obsolete', LastModifiedDate = ?
		 WHERE DefinitionId = ? AND Status = 'Active' AND Id != ?`,
		now, defID, versionID,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to obsolete prior active version of %q", seed.DeveloperName), err)
	}
	if _, err := q.Exec(
		`UPDATE Flow SET Status = 'Active', LastModifiedDate = ?
		 WHERE Id = ?`,
		now, versionID,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to activate standard flow version for %q", seed.DeveloperName), err)
	}
	if err := updateManagedFlowDefinition(q, defID, seed, versionID, latestLiveFlowVersionID(versions, versionID), now, ownerPack); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to advance ActiveVersionId for %q", seed.DeveloperName), err)
	}
	if err := syncManagedFlowDefinitionTriggerProjection(q, defID, seed, now); err != nil {
		return err
	}
	return nil
}

func deactivateManagedStandardFlowVersion(q sqlQuerier, versions []standardFlowVersionState, defID, versionID string, seed standardFlowSeed, now string, ownerPack string) error {
	if _, err := q.Exec(
		`UPDATE Flow SET Status = 'Obsolete', LastModifiedDate = ?
		 WHERE DefinitionId = ? AND Status = 'Active' AND Id != ? AND Source = ?`,
		now, defID, versionID, FlowSourceStandard,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to obsolete prior active version of %q", seed.DeveloperName), err)
	}
	if _, err := q.Exec(
		`UPDATE Flow SET Status = 'Draft', LastModifiedDate = ?
		 WHERE Id = ? AND Source = ?`,
		now, versionID, FlowSourceStandard,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to retain draft standard flow version for %q", seed.DeveloperName), err)
	}
	if err := updateManagedFlowDefinition(
		q, defID, seed, "", latestLiveFlowVersionID(versions, versionID), now, ownerPack,
	); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to deactivate standard flow %q", seed.DeveloperName), err)
	}
	if err := syncManagedFlowDefinitionTriggerProjection(q, defID, seed, now); err != nil {
		return err
	}
	return nil
}

func latestLiveFlowVersionID(versions []standardFlowVersionState, fallbackID string) string {
	latestID := strings.TrimSpace(fallbackID)
	latestVersionNumber := -1
	for _, v := range versions {
		if v.IsDeleted || strings.TrimSpace(v.ID) == "" {
			continue
		}
		if v.VersionNumber > latestVersionNumber {
			latestID = v.ID
			latestVersionNumber = v.VersionNumber
		}
	}
	return latestID
}

func existingStandardFlowVersion(versions []standardFlowVersionState, metadata []byte) *standardFlowVersionState {
	target := string(metadata)
	for i := range versions {
		// Soft-deleted versions still reserve their version numbers, but they
		// must not be re-activated as the live shipped standard row.
		if !versions[i].IsDeleted && versions[i].Source == FlowSourceStandard && versions[i].Metadata == target {
			return &versions[i]
		}
	}
	return nil
}

// StandardFlowActiveVersionIsPackOwned (the runtime READ counterpart of this
// seeder) STAYS in internal/bootstrap (standard_flow_active_version.go) because
// it is consulted by the SHARED standard-action internal-access-scope reader
// exposed via the contract runtimeports.StandardActionReader port.

func nextStandardFlowVersion(versions []standardFlowVersionState) int {
	next := 1
	for _, v := range versions {
		if v.VersionNumber >= next {
			next = v.VersionNumber + 1
		}
	}
	return next
}

func lockStandardFlowVersions(q sqlQuerier, defID string) ([]standardFlowVersionState, error) {
	rows, err := q.Query(
		`SELECT Id, Metadata, VersionNumber, Status, Source, IsDeleted
		 FROM Flow WHERE DefinitionId = ? ORDER BY VersionNumber DESC, Id DESC FOR UPDATE`,
		defID,
	)
	if err != nil {
		return nil, apperrors.NewInternalError(fmt.Sprintf("lock Flow versions for %q", defID), err)
	}
	defer rows.Close()
	versions := make([]standardFlowVersionState, 0)
	for rows.Next() {
		var v standardFlowVersionState
		var metadata, status, source sql.NullString
		var isDeleted sql.NullBool
		if err := rows.Scan(&v.ID, &metadata, &v.VersionNumber, &status, &source, &isDeleted); err != nil {
			return nil, apperrors.NewInternalError(fmt.Sprintf("scan locked Flow version for %q", defID), err)
		}
		if metadata.Valid {
			v.Metadata = metadata.String
		}
		if status.Valid {
			v.Status = status.String
		}
		if source.Valid {
			v.Source = source.String
		}
		if isDeleted.Valid {
			v.IsDeleted = isDeleted.Bool
		}
		versions = append(versions, v)
	}
	if err := rows.Err(); err != nil {
		return nil, apperrors.NewInternalError(fmt.Sprintf("scan locked Flow versions for %q", defID), err)
	}
	return versions, nil
}

// 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"

// 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 current source payload.
//
// Why this lives in Go, not a pure SQL data-spec: we need the pack bytes
// passed by the source-backed caller 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 source payload)", 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
}