package bootstrap

import (
	"encoding/json"
	"errors"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/nexuscrm/shared/pkg/models"
)

// expectFlowSourceBackfillGateApplied short-circuits the backfill in tests
// that care about the seeder path, not the backfill itself. The gate row
// reads "applied" so migrateFlowSourceBackfill returns immediately — no
// per-seed UPDATE, no gate stamp.
func expectFlowSourceBackfillGateApplied(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}).AddRow("applied"))
	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v2").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}).AddRow("applied"))
	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("tenant_flow_source_normalize_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}).AddRow("applied"))
}

func flowDefinitionStateRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{
		"Id", "ActiveVersionId", "LatestVersionId", "IsDeleted", "IsCustomized",
	})
}

func flowVersionMetadataRows() *sqlmock.Rows {
	return sqlmock.NewRows([]string{"Id", "Metadata"})
}

func TestStandardFlowsFS_ShipsAtLeastOneFlow(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	require.NotEmpty(t, seeds, "pack must ship at least Std_ConvertLead")
}

func TestStandardFlowsFS_StdConvertLeadShipsAndIsAtomic(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	var found *standardFlowSeed
	for i := range seeds {
		if seeds[i].DeveloperName == "Std_ConvertLead" {
			found = &seeds[i]
			break
		}
	}
	require.NotNil(t, found, "Std_ConvertLead must be in the pack")

	var meta models.FlowMetadata
	require.NoError(t, json.Unmarshal(found.MetadataRaw, &meta))
	assert.True(t, meta.RunAtomically, "Std_ConvertLead must be atomic so a mid-flow failure rolls back Contact/Opportunity creates and Lead updates")
	assert.Empty(t, meta.Waits, "atomic flows must not have Wait elements")
	assert.NotEmpty(t, meta.RecordCreates, "flow must create records")
	// The terminal step now rides on the generic stampConvertedAnchors
	// invocable action so custom Converted*Id lookups on Lead get picked
	// up without a flow edit.
	hasStamp := false
	for _, ac := range meta.ActionCalls {
		if ac.ActionType == "stampConvertedAnchors" {
			hasStamp = true
			break
		}
	}
	assert.True(t, hasStamp, "flow must terminate with a stampConvertedAnchors action call")
}

func TestStandardFlowsFS_StdConvertLeadCreatesOrReusesAccountDeclaratively(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	var found *standardFlowSeed
	for i := range seeds {
		if seeds[i].DeveloperName == "Std_ConvertLead" {
			found = &seeds[i]
			break
		}
	}
	require.NotNil(t, found, "Std_ConvertLead must be in the pack")

	var meta models.FlowMetadata
	require.NoError(t, json.Unmarshal(found.MetadataRaw, &meta))

	var accountIDVar *models.FlowVariable
	for i := range meta.Variables {
		if meta.Variables[i].Name == "accountId" {
			accountIDVar = &meta.Variables[i]
			break
		}
	}
	require.NotNil(t, accountIDVar, "flow must declare accountId")
	assert.True(t, accountIDVar.IsInput, "accountId input enables Account reuse when supplied")
	assert.True(t, accountIDVar.IsOutput, "accountId must be returned as the converted Account anchor")

	var resolveAccount *models.FlowDecision
	for i := range meta.Decisions {
		if meta.Decisions[i].Name == "ResolveAccount" {
			resolveAccount = &meta.Decisions[i]
			break
		}
	}
	require.NotNil(t, resolveAccount, "flow keeps the declarative Account create-vs-reuse decision")
	require.Len(t, resolveAccount.Rules, 1)
	assert.Equal(t, "GetAccountById", resolveAccount.Rules[0].Connector.TargetReference)
	require.NotNil(t, resolveAccount.DefaultConnector)
	assert.Equal(t, "CreateAccount", resolveAccount.DefaultConnector.TargetReference)

	var accountCreate *models.FlowRecordCreate
	for i := range meta.RecordCreates {
		if meta.RecordCreates[i].Name == "CreateAccount" {
			accountCreate = &meta.RecordCreates[i]
			break
		}
	}
	require.NotNil(t, accountCreate, "CreateAccount must remain declarative metadata")
	assert.Equal(t, "Account", accountCreate.Object)
	require.NotNil(t, accountCreate.FieldMapping)
	assert.Equal(t, "Lead", accountCreate.FieldMapping.SourceObject)
	assert.Equal(t, "Account", accountCreate.FieldMapping.TargetObject)
	assert.Equal(t, "convert", accountCreate.FieldMapping.ActionType)
	assert.Equal(t, "accountId", accountCreate.AssignRecordIdToReference)

	var accountLookup *models.FlowRecordLookup
	for i := range meta.RecordLookups {
		if meta.RecordLookups[i].Name == "GetAccountById" {
			accountLookup = &meta.RecordLookups[i]
			break
		}
	}
	require.NotNil(t, accountLookup, "flow must validate the caller-supplied accountId")
	assert.Equal(t, "Account", accountLookup.Object)
	require.Len(t, accountLookup.Filters, 1)
	assert.Equal(t, "Id", accountLookup.Filters[0].Field)
	assert.Equal(t, "EqualTo", accountLookup.Filters[0].Operator)
	assert.Equal(t, "accountId", elementReferenceValue(accountLookup.Filters[0].Value))
}

func TestStandardFlowsFS_StdConvertLeadReusesAccountWithoutUpdatingIt(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	var found *standardFlowSeed
	for i := range seeds {
		if seeds[i].DeveloperName == "Std_ConvertLead" {
			found = &seeds[i]
			break
		}
	}
	require.NotNil(t, found, "Std_ConvertLead must be in the pack")

	var meta models.FlowMetadata
	require.NoError(t, json.Unmarshal(found.MetadataRaw, &meta))

	assertNoRecordUpdateNamed(t, meta.RecordUpdates, "MergeAccountFields")
	assertFillBlankMappingUpdate(t, meta.RecordUpdates, "MergeContactFields", "Contact", "contactId", "$contact")
}

func TestStandardFlowsFS_StdConvertLeadAssignsPrimaryCampaignDeclaratively(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	var found *standardFlowSeed
	for i := range seeds {
		if seeds[i].DeveloperName == "Std_ConvertLead" {
			found = &seeds[i]
			break
		}
	}
	require.NotNil(t, found, "Std_ConvertLead must be in the pack")

	var meta models.FlowMetadata
	require.NoError(t, json.Unmarshal(found.MetadataRaw, &meta))

	var campaignLookup *models.FlowRecordLookup
	for i := range meta.RecordLookups {
		if meta.RecordLookups[i].Name == "GetPrimaryCampaignMember" {
			campaignLookup = &meta.RecordLookups[i]
			break
		}
	}
	require.NotNil(t, campaignLookup, "new Opportunity path must look up CampaignMember in flow metadata")
	assert.Equal(t, "CampaignMember", campaignLookup.Object)
	assert.True(t, campaignLookup.GetFirstRecordOnly)
	assert.Equal(t, "CreatedDate", campaignLookup.SortField)
	assert.Equal(t, "DESC", campaignLookup.SortOrder)
	require.Len(t, campaignLookup.Filters, 1)
	assert.Equal(t, "LeadId", campaignLookup.Filters[0].Field)
	assert.Equal(t, "leadId", elementReferenceValue(campaignLookup.Filters[0].Value))
	require.NotNil(t, campaignLookup.Connector)
	assert.Equal(t, "CreateOpportunity", campaignLookup.Connector.TargetReference)

	var opportunityNameDecision *models.FlowDecision
	for i := range meta.Decisions {
		if meta.Decisions[i].Name == "ResolveOpportunityName" {
			opportunityNameDecision = &meta.Decisions[i]
			break
		}
	}
	require.NotNil(t, opportunityNameDecision)
	require.NotNil(t, opportunityNameDecision.DefaultConnector)
	assert.Equal(t, "GetPrimaryCampaignMember", opportunityNameDecision.DefaultConnector.TargetReference)

	var createOpportunity *models.FlowRecordCreate
	for i := range meta.RecordCreates {
		if meta.RecordCreates[i].Name == "CreateOpportunity" {
			createOpportunity = &meta.RecordCreates[i]
			break
		}
	}
	require.NotNil(t, createOpportunity)
	assertInputAssignmentRef(t, createOpportunity.InputAssignments, "CampaignId", "$primaryCM.CampaignId")
}

func TestStandardFlowsFS_StdConvertLeadInvokesTenantNeutralOCRSync(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	var convertFlow *standardFlowSeed
	var syncFlow *standardFlowSeed
	for i := range seeds {
		switch seeds[i].DeveloperName {
		case "Std_ConvertLead":
			convertFlow = &seeds[i]
		case "Std_SyncOpportunityContactRole":
			syncFlow = &seeds[i]
		}
	}
	require.NotNil(t, convertFlow, "Std_ConvertLead must be in the pack")
	require.NotNil(t, syncFlow, "Std_SyncOpportunityContactRole must be in the pack")

	var convertMeta models.FlowMetadata
	require.NoError(t, json.Unmarshal(convertFlow.MetadataRaw, &convertMeta))
	var syncCall *models.FlowSubflow
	for i := range convertMeta.Subflows {
		if convertMeta.Subflows[i].Name == "SyncOpportunityContactRole" {
			syncCall = &convertMeta.Subflows[i]
			break
		}
	}
	require.NotNil(t, syncCall, "Std_ConvertLead must invoke OCR sync declaratively")
	assert.Equal(t, "Std_SyncOpportunityContactRole", syncCall.FlowName)
	require.NotNil(t, syncCall.Connector)
	assert.Equal(t, "RelinkLeadChildren", syncCall.Connector.TargetReference)

	var syncMeta models.FlowMetadata
	require.NoError(t, json.Unmarshal(syncFlow.MetadataRaw, &syncMeta))
	for _, create := range syncMeta.RecordCreates {
		if create.Object != "OpportunityContactRole" {
			continue
		}
		for _, assignment := range create.InputAssignments {
			assert.NotEqual(t, "Role", assignment.Field,
				"standard OCR sync must stay tenant-neutral; Role defaults belong in tenant/org metadata")
		}
	}
}

func TestStandardFlowsFS_StdConvertLeadConsumesCreatePanelActionInputs(t *testing.T) {
	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	var found *standardFlowSeed
	for i := range seeds {
		if seeds[i].DeveloperName == "Std_ConvertLead" {
			found = &seeds[i]
			break
		}
	}
	require.NotNil(t, found, "Std_ConvertLead must be in the pack")

	var meta models.FlowMetadata
	require.NoError(t, json.Unmarshal(found.MetadataRaw, &meta))
	inputVars := map[string]bool{}
	for _, variable := range meta.Variables {
		if variable.IsInput {
			inputVars[variable.Name] = true
		}
	}
	for _, name := range []string{"contactFirstName", "contactLastName", "contactTitle", "contactEmail", "opportunityName"} {
		assert.True(t, inputVars[name], "%s must be a Flow input variable before the action schema row is active", name)
	}

	var createContact *models.FlowRecordCreate
	var createOpportunity *models.FlowRecordCreate
	for i := range meta.RecordCreates {
		switch meta.RecordCreates[i].Name {
		case "CreateContact":
			createContact = &meta.RecordCreates[i]
		case "CreateOpportunity":
			createOpportunity = &meta.RecordCreates[i]
		}
	}
	require.NotNil(t, createContact)
	assertInputAssignmentRef(t, createContact.InputAssignments, "FirstName", "contactFirstName")
	assertInputAssignmentRef(t, createContact.InputAssignments, "LastName", "contactLastName")
	assertInputAssignmentRef(t, createContact.InputAssignments, "Title", "contactTitle")
	assertInputAssignmentRef(t, createContact.InputAssignments, "Email", "contactEmail")
	require.NotNil(t, createOpportunity)
	assertInputAssignmentRef(t, createOpportunity.InputAssignments, "Name", "opportunityName")

	var actionDefs []StandardActionDef
	require.NoError(t, json.Unmarshal(standardActionsJSON, &actionDefs))
	require.NoError(t, validateStandardActionFlowBindings(actionDefs, seeds))
}

func assertFillBlankMappingUpdate(t *testing.T, updates []models.FlowRecordUpdate, name, objectName, inputReference, targetReference string) {
	t.Helper()
	for i := range updates {
		update := updates[i]
		if update.Name != name {
			continue
		}
		assert.Equal(t, objectName, update.Object)
		assert.Equal(t, inputReference, update.InputReference)
		require.NotNil(t, update.FieldMapping)
		assert.Equal(t, "Lead", update.FieldMapping.SourceObject)
		assert.Equal(t, "$lead", update.FieldMapping.SourceReference)
		assert.Equal(t, objectName, update.FieldMapping.TargetObject)
		assert.Equal(t, targetReference, update.FieldMapping.TargetReference)
		assert.Equal(t, "convert", update.FieldMapping.ActionType)
		assert.Equal(t, "fillBlankOnly", update.FieldMapping.MergeMode)
		return
	}
	t.Fatalf("missing recordUpdate %s", name)
}

func assertNoRecordUpdateNamed(t *testing.T, updates []models.FlowRecordUpdate, name string) {
	t.Helper()
	for i := range updates {
		if updates[i].Name == name {
			t.Fatalf("recordUpdate %s must not be present", name)
		}
	}
}

func TestStandardFlowSeedIDs_AreDeterministic(t *testing.T) {
	assert.Equal(t, "flowdef_std_std_convertlead", standardFlowDefinitionID("Std_ConvertLead"))
	assert.Equal(t, "flowver_std_std_convertlead_v1", standardFlowVersionID("Std_ConvertLead", 1))
}

func elementReferenceValue(raw interface{}) string {
	m, ok := raw.(map[string]interface{})
	if !ok {
		return ""
	}
	ref, _ := m["elementReference"].(string)
	return ref
}

func assertInputAssignmentRef(t *testing.T, assignments []models.FlowInputAssignment, field, ref string) {
	t.Helper()
	for _, assignment := range assignments {
		if assignment.Field == field {
			assert.Equal(t, ref, elementReferenceValue(assignment.Value))
			return
		}
	}
	t.Fatalf("missing input assignment for field %s", field)
}

func TestInitializeStandardFlows_FirstInstallActivatesV1(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	expectFlowSourceBackfillGateApplied(mock)
	for range seeds {
		// Each seed runs in its own tx: Begin → reads/writes → Commit.
		mock.ExpectBegin()
		// Definition not found → fresh install path.
		mock.ExpectQuery(`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized FROM FlowDefinition WHERE DeveloperName`).
			WillReturnRows(flowDefinitionStateRows())
		mock.ExpectExec(`INSERT INTO FlowDefinition`).WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectExec(`INSERT INTO Flow`).WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectExec(`UPDATE FlowDefinition\s+SET ActiveVersionId`).WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectCommit()
	}

	require.NoError(t, InitializeStandardFlows(db))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestInitializeStandardFlows_ReinstallWithSameBytesIsNoOp(t *testing.T) {
	// On re-boot of the same platform version, the seeder must not
	// touch existing rows — tenant customizations stay intact. The
	// byte-identical check short-circuits BEFORE any Source / version
	// lookups; an incorrect short-circuit order would fire extra
	// queries here and trip sqlmock.
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	expectFlowSourceBackfillGateApplied(mock)

	for _, seed := range seeds {
		defID := standardFlowDefinitionID(seed.DeveloperName)
		v1ID := standardFlowVersionID(seed.DeveloperName, 1)
		// Begin + read-only probe + Commit: the byte-identical short-circuit
		// emits no writes, but Commit still runs (empty-tx is the price of
		// closing reads and writes in the same boundary).
		mock.ExpectBegin()
		mock.ExpectQuery(`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized FROM FlowDefinition WHERE DeveloperName`).
			WillReturnRows(flowDefinitionStateRows().AddRow(defID, v1ID, v1ID, false, false))
		mock.ExpectQuery(`SELECT Id, Metadata FROM Flow`).
			WillReturnRows(flowVersionMetadataRows().AddRow(v1ID, string(seed.MetadataRaw)))
		mock.ExpectExec(`UPDATE FlowDefinition\s+SET MasterLabel = \?, Description = \?, ProcessType = \?, ActiveVersionId = \?, LatestVersionId = \?, IsDeleted = 0, LastModifiedDate = \?\s+WHERE Id = \?`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectCommit()
	}

	require.NoError(t, InitializeStandardFlows(db))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestInitializeStandardFlows_AdminEditedActiveStaysConservative(t *testing.T) {
	// An admin edit via the tooling API flips Flow.Source to 'custom'.
	// A subsequent pack upgrade with new metadata must NOT auto-activate
	// the new version — the admin's currently-active row keeps running.
	// New content lands as Draft; only LatestVersionId advances.
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	expectFlowSourceBackfillGateApplied(mock)

	for _, seed := range seeds {
		defID := standardFlowDefinitionID(seed.DeveloperName)
		v1ID := standardFlowVersionID(seed.DeveloperName, 1)
		mock.ExpectBegin()
		mock.ExpectQuery(`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized FROM FlowDefinition WHERE DeveloperName`).
			WillReturnRows(flowDefinitionStateRows().AddRow(defID, v1ID, v1ID, false, false))
		// Stored metadata differs → fall through the no-op check.
		mock.ExpectQuery(`SELECT Id, Metadata FROM Flow`).
			WillReturnRows(flowVersionMetadataRows().AddRow(v1ID, `{"label":"tenant-edited"}`))
		mock.ExpectQuery(`SELECT MAX\(VersionNumber\) FROM Flow`).
			WillReturnRows(sqlmock.NewRows([]string{"max"}).AddRow(1))
		// Active version Source='custom' (admin edited it) → conservative.
		mock.ExpectQuery(`SELECT f\.Source FROM FlowDefinition fd LEFT JOIN Flow f`).
			WillReturnRows(sqlmock.NewRows([]string{"Source"}).AddRow("custom"))
		mock.ExpectExec(`INSERT INTO Flow[\s\S]+'Draft'`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		// Only LatestVersionId advances; ActiveVersionId must NOT be touched.
		mock.ExpectExec(`UPDATE FlowDefinition\s+SET MasterLabel = \?, Description = \?, ProcessType = \?, ActiveVersionId = \?, LatestVersionId = \?, IsDeleted = 0, LastModifiedDate = \?\s+WHERE Id = \?`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectCommit()
	}

	require.NoError(t, InitializeStandardFlows(db))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestInitializeStandardFlows_AutoAdvancesWhenActiveIsStandard(t *testing.T) {
	// The P1 fix: when the currently-active Flow row is Source='standard'
	// (pack-seeded, never mutated), a pack upgrade with new metadata
	// auto-activates the new version so tenants stop running buggy v1.
	// Mechanism: INSERT new row Status='Active', UPDATE old Active row
	// Status='Obsolete', UPDATE FlowDefinition pointers to new version.
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	expectFlowSourceBackfillGateApplied(mock)

	for _, seed := range seeds {
		defID := standardFlowDefinitionID(seed.DeveloperName)
		v1ID := standardFlowVersionID(seed.DeveloperName, 1)
		mock.ExpectBegin()
		mock.ExpectQuery(`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized FROM FlowDefinition WHERE DeveloperName`).
			WillReturnRows(flowDefinitionStateRows().AddRow(defID, v1ID, v1ID, false, false))
		mock.ExpectQuery(`SELECT Id, Metadata FROM Flow`).
			WillReturnRows(flowVersionMetadataRows().AddRow(v1ID, `{"label":"old-standard"}`))
		mock.ExpectQuery(`SELECT MAX\(VersionNumber\) FROM Flow`).
			WillReturnRows(sqlmock.NewRows([]string{"max"}).AddRow(1))
		mock.ExpectQuery(`SELECT f\.Source FROM FlowDefinition fd LEFT JOIN Flow f`).
			WillReturnRows(sqlmock.NewRows([]string{"Source"}).AddRow("standard"))
		// New version lands as Active (not Draft) because caller may
		// auto-advance.
		mock.ExpectExec(`INSERT INTO Flow[\s\S]+'Active'`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		// Old active row → Obsolete.
		mock.ExpectExec(`UPDATE Flow SET Status = 'Obsolete'`).
			WillReturnResult(sqlmock.NewResult(0, 1))
		// BOTH pointers advance.
		mock.ExpectExec(`UPDATE FlowDefinition\s+SET MasterLabel = \?, Description = \?, ProcessType = \?, ActiveVersionId = \?, LatestVersionId = \?, IsDeleted = 0, LastModifiedDate = \?\s+WHERE Id = \?`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectCommit()
	}

	require.NoError(t, InitializeStandardFlows(db))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestInitializeStandardFlows_NullActiveStaysConservative(t *testing.T) {
	// Defensive branch per reviewer caveat: FlowDefinition with a NULL
	// ActiveVersionId (or a dangling pointer) must NOT trigger auto-
	// advance. LEFT JOIN returns a NULL Source — treat as conservative.
	// This should not happen via the normal bootstrap path (v1 always
	// gets activated on first install), but a test-data import or a
	// manual fixup could leave a FlowDefinition without a valid active
	// pointer.
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	expectFlowSourceBackfillGateApplied(mock)

	for _, seed := range seeds {
		defID := standardFlowDefinitionID(seed.DeveloperName)
		v1ID := standardFlowVersionID(seed.DeveloperName, 1)
		mock.ExpectBegin()
		mock.ExpectQuery(`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized FROM FlowDefinition WHERE DeveloperName`).
			WillReturnRows(flowDefinitionStateRows().AddRow(defID, nil, v1ID, false, false))
		mock.ExpectQuery(`SELECT Id, Metadata FROM Flow`).
			WillReturnRows(flowVersionMetadataRows().AddRow(v1ID, `{"label":"pre-v1"}`))
		mock.ExpectQuery(`SELECT MAX\(VersionNumber\) FROM Flow`).
			WillReturnRows(sqlmock.NewRows([]string{"max"}).AddRow(1))
		// Source column scanned as NULL → sql.NullString.Valid == false
		// → activeVersionIsPackOwned returns (false, nil).
		mock.ExpectQuery(`SELECT f\.Source FROM FlowDefinition fd LEFT JOIN Flow f`).
			WillReturnRows(sqlmock.NewRows([]string{"Source"}).AddRow(nil))
		mock.ExpectExec(`INSERT INTO Flow[\s\S]+'Draft'`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectExec(`UPDATE FlowDefinition\s+SET MasterLabel = \?, Description = \?, ProcessType = \?, ActiveVersionId = \?, LatestVersionId = \?, IsDeleted = 0, LastModifiedDate = \?\s+WHERE Id = \?`).
			WillReturnResult(sqlmock.NewResult(1, 1))
		mock.ExpectCommit()
	}

	require.NoError(t, InitializeStandardFlows(db))
	require.NoError(t, mock.ExpectationsWereMet())
}

// TestInitializeStandardFlows_AutoAdvanceRollsBackOnPointerFailure pins the
// P2 atomicity contract: if the FlowDefinition pointer UPDATE fails AFTER
// the Obsolete UPDATE on the old active row, the whole seeder step rolls
// back. Without the txn wrap, the caller would be left with v1 Obsolete
// but FlowDefinition.ActiveVersionId still pointing at v1 — runtime
// dispatch would then resolve v1 (Obsolete) and refuse to execute, AND
// next-boot `existingFlowHasStandardMetadata` would see v2's metadata
// already present and short-circuit (never re-attempting the pointer
// advance). Rollback restores v1.Status=Active so the old active version
// keeps serving; next boot retries.
func TestInitializeStandardFlows_AutoAdvanceRollsBackOnPointerFailure(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	require.NotEmpty(t, seeds)
	seed := seeds[0]
	defID := standardFlowDefinitionID(seed.DeveloperName)
	v1ID := standardFlowVersionID(seed.DeveloperName, 1)
	expectFlowSourceBackfillGateApplied(mock)

	mock.ExpectBegin()
	mock.ExpectQuery(`SELECT Id, ActiveVersionId, LatestVersionId, IsDeleted, IsCustomized FROM FlowDefinition WHERE DeveloperName`).
		WillReturnRows(flowDefinitionStateRows().AddRow(defID, v1ID, v1ID, false, false))
	mock.ExpectQuery(`SELECT Id, Metadata FROM Flow`).
		WillReturnRows(flowVersionMetadataRows().AddRow(v1ID, `{"label":"old-standard"}`))
	mock.ExpectQuery(`SELECT MAX\(VersionNumber\) FROM Flow`).
		WillReturnRows(sqlmock.NewRows([]string{"max"}).AddRow(1))
	mock.ExpectQuery(`SELECT f\.Source FROM FlowDefinition fd LEFT JOIN Flow f`).
		WillReturnRows(sqlmock.NewRows([]string{"Source"}).AddRow("standard"))
	mock.ExpectExec(`INSERT INTO Flow[\s\S]+'Active'`).
		WillReturnResult(sqlmock.NewResult(1, 1))
	mock.ExpectExec(`UPDATE Flow SET Status = 'Obsolete'`).
		WillReturnResult(sqlmock.NewResult(0, 1))
	// Pointer UPDATE fails → rollback must fire, InitializeStandardFlows
	// returns an error, and NO Commit is expected.
	mock.ExpectExec(`UPDATE FlowDefinition\s+SET MasterLabel = \?, Description = \?, ProcessType = \?, ActiveVersionId = \?, LatestVersionId = \?, IsDeleted = 0, LastModifiedDate = \?\s+WHERE Id = \?`).
		WillReturnError(errors.New("deadlock found"))
	mock.ExpectRollback()

	err = InitializeStandardFlows(db)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "ActiveVersionId")
	require.NoError(t, mock.ExpectationsWereMet())
}

// TestMigrateFlowSourceBackfill_FlipsOnlyByteMatches pins the conservative
// semantics: the backfill flips Source='standard' ONLY for pack-Id rows
// whose stored Metadata byte-equals the currently-shipped pack. Rows
// whose bytes differ (either admin-edited OR seeded from an older pack
// version) stay 'custom'. This is the P1 fix — nothing the backfill
// writes can re-claim admin metadata as platform metadata.
func TestMigrateFlowSourceBackfill_FlipsOnlyByteMatches(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	require.NotEmpty(t, seeds)

	// Gate not applied → full backfill runs.
	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}))

	// Per-seed UPDATE: WHERE clause binds the seed's Metadata bytes as the
	// last argument so byte-mismatched rows are never touched at the DB
	// level. RowsAffected simulates: first seed matches (1 flip), second
	// no match (0 flips), third matches (1 flip). Total = 2 flipped.
	mock.ExpectExec(`UPDATE Flow\s+SET Source = \?, LastModifiedDate = \?\s+WHERE DefinitionId = \?\s+AND Id LIKE \?\s+AND Source = \?\s+AND Metadata = \?`).
		WithArgs("standard", sqlmock.AnyArg(),
			standardFlowDefinitionID(seeds[0].DeveloperName),
			"flowver_std_"+strings.ToLower(seeds[0].DeveloperName)+"_v%",
			"custom",
			string(seeds[0].MetadataRaw)).
		WillReturnResult(sqlmock.NewResult(0, 1))
	if len(seeds) >= 2 {
		mock.ExpectExec(`UPDATE Flow\s+SET Source = \?, LastModifiedDate = \?\s+WHERE DefinitionId = \?\s+AND Id LIKE \?\s+AND Source = \?\s+AND Metadata = \?`).
			WithArgs("standard", sqlmock.AnyArg(),
				standardFlowDefinitionID(seeds[1].DeveloperName),
				"flowver_std_"+strings.ToLower(seeds[1].DeveloperName)+"_v%",
				"custom",
				string(seeds[1].MetadataRaw)).
			WillReturnResult(sqlmock.NewResult(0, 0))
	}
	for i := 2; i < len(seeds); i++ {
		mock.ExpectExec(`UPDATE Flow\s+SET Source = \?, LastModifiedDate = \?\s+WHERE DefinitionId = \?\s+AND Id LIKE \?\s+AND Source = \?\s+AND Metadata = \?`).
			WithArgs("standard", sqlmock.AnyArg(),
				standardFlowDefinitionID(seeds[i].DeveloperName),
				"flowver_std_"+strings.ToLower(seeds[i].DeveloperName)+"_v%",
				"custom",
				string(seeds[i].MetadataRaw)).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}

	// Gate stamp AFTER all seeds succeeded.
	mock.ExpectExec(`INSERT INTO .*BootstrapState.*ON DUPLICATE KEY UPDATE`).
		WithArgs("flow_source_backfill_v1", sqlmock.AnyArg(), "flow-source-backfill",
			"applied", "", sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(1, 1))

	require.NoError(t, migrateFlowSourceBackfill(db, seeds, "2026-04-23 19:00:00"))
	require.NoError(t, mock.ExpectationsWereMet())
}

// TestMigrateFlowSourceBackfill_GateSkipsWhenApplied pins the warm-install
// fast-path: when the gate is already stamped, the backfill short-circuits
// after one SELECT. No per-seed UPDATE runs, no stamp is re-written.
func TestMigrateFlowSourceBackfill_GateSkipsWhenApplied(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}).AddRow("applied"))

	// NO other SQL is expected — sqlmock will fail the test if anything else runs.

	require.NoError(t, migrateFlowSourceBackfill(db, seeds, "2026-04-23 19:00:00"))
	require.NoError(t, mock.ExpectationsWereMet())
}

// TestMigrateFlowSourceBackfill_GateNotStampedOnPartialFailure pins the
// reviewer's guidance: stamp the gate ONLY after every seed's UPDATE
// succeeded. If the second seed's UPDATE errors, the gate stays unstamped
// so the next boot retries from scratch. The UPDATE is idempotent (its
// WHERE clause filters rows already flipped), so the retry is safe.
func TestMigrateFlowSourceBackfill_GateNotStampedOnPartialFailure(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)
	require.GreaterOrEqual(t, len(seeds), 2, "test requires at least 2 seeds to exercise mid-loop failure")

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}))

	// First seed succeeds.
	mock.ExpectExec(`UPDATE Flow\s+SET Source = \?`).
		WithArgs("standard", sqlmock.AnyArg(),
			standardFlowDefinitionID(seeds[0].DeveloperName),
			sqlmock.AnyArg(), "custom", string(seeds[0].MetadataRaw)).
		WillReturnResult(sqlmock.NewResult(0, 1))
	// Second seed fails → entire migration returns error → no gate stamp.
	mock.ExpectExec(`UPDATE Flow\s+SET Source = \?`).
		WithArgs("standard", sqlmock.AnyArg(),
			standardFlowDefinitionID(seeds[1].DeveloperName),
			sqlmock.AnyArg(), "custom", string(seeds[1].MetadataRaw)).
		WillReturnError(errors.New("lock wait timeout"))

	// NO gate stamp is expected — partial success must not stamp.

	err = migrateFlowSourceBackfill(db, seeds, "2026-04-23 19:00:00")
	require.Error(t, err)
	assert.Contains(t, err.Error(), seeds[1].DeveloperName)
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestMigrateHistoricalFlowSourceBackfill_FlipsKnownHistoricalPackHashes(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	seeds := []standardFlowSeed{
		{DeveloperName: "Std_ConvertLead"},
		{DeveloperName: "Std_SyncOpportunityContactRole"},
	}

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v2").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}))

	mock.ExpectExec(`UPDATE Flow f\s+JOIN FlowDefinition fd ON fd\.Id = f\.DefinitionId\s+SET f\.Source = \?, f\.LastModifiedDate = \?\s+WHERE f\.DefinitionId = \?\s+AND f\.Id LIKE \?\s+AND f\.Source = \?\s+AND \(fd\.IsCustomized = 0 OR fd\.IsCustomized IS NULL\)\s+AND SHA2\(f\.Metadata, 256\) IN \(\?\)`).
		WithArgs("standard", sqlmock.AnyArg(),
			standardFlowDefinitionID("Std_ConvertLead"),
			"flowver_std_std_convertlead_v%",
			"custom",
			historicalStandardFlowMetadataSHA256["Std_ConvertLead"][0]).
		WillReturnResult(sqlmock.NewResult(0, 1))

	mock.ExpectExec(`INSERT INTO .*BootstrapState.*ON DUPLICATE KEY UPDATE`).
		WithArgs("flow_source_backfill_v2", sqlmock.AnyArg(), "flow-source-backfill-v2",
			"applied", "", sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(1, 1))

	require.NoError(t, migrateHistoricalFlowSourceBackfill(db, seeds, "2026-04-27 12:00:00"))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestMigrateHistoricalFlowSourceBackfill_GateSkipsWhenApplied(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v2").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}).AddRow("applied"))

	require.NoError(t, migrateHistoricalFlowSourceBackfill(db, []standardFlowSeed{{DeveloperName: "Std_ConvertLead"}}, "2026-04-27 12:00:00"))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestMigrateHistoricalFlowSourceBackfill_GateNotStampedOnFailure(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("flow_source_backfill_v2").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}))

	mock.ExpectExec(`UPDATE Flow f\s+JOIN FlowDefinition fd`).
		WithArgs("standard", sqlmock.AnyArg(),
			standardFlowDefinitionID("Std_ConvertLead"),
			"flowver_std_std_convertlead_v%",
			"custom",
			historicalStandardFlowMetadataSHA256["Std_ConvertLead"][0]).
		WillReturnError(errors.New("lock wait timeout"))

	err = migrateHistoricalFlowSourceBackfill(db, []standardFlowSeed{{DeveloperName: "Std_ConvertLead"}}, "2026-04-27 12:00:00")
	require.Error(t, err)
	assert.Contains(t, err.Error(), "Std_ConvertLead")
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestMigrateTenantFlowSourceNormalize_FlipsTenantToCustom(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("tenant_flow_source_normalize_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}))

	mock.ExpectExec(`UPDATE Flow SET Source = \?, LastModifiedDate = \? WHERE Source = 'tenant'`).
		WithArgs("custom", "2026-04-28 12:00:00").
		WillReturnResult(sqlmock.NewResult(0, 3))

	mock.ExpectExec(`INSERT INTO .*BootstrapState.*ON DUPLICATE KEY UPDATE`).
		WithArgs("tenant_flow_source_normalize_v1", sqlmock.AnyArg(), "tenant-flow-source-normalize",
			"applied", "", sqlmock.AnyArg(), sqlmock.AnyArg()).
		WillReturnResult(sqlmock.NewResult(1, 1))

	require.NoError(t, migrateTenantFlowSourceNormalize(db, "2026-04-28 12:00:00"))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestMigrateTenantFlowSourceNormalize_GateSkipsWhenApplied(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("tenant_flow_source_normalize_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}).AddRow("applied"))

	require.NoError(t, migrateTenantFlowSourceNormalize(db, "2026-04-28 12:00:00"))
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestMigrateTenantFlowSourceNormalize_GateNotStampedOnUpdateFailure(t *testing.T) {
	db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp))
	require.NoError(t, err)
	defer db.Close()

	mock.ExpectQuery(`SELECT Status FROM .*BootstrapState.* WHERE .*StateKey`).
		WithArgs("tenant_flow_source_normalize_v1").
		WillReturnRows(sqlmock.NewRows([]string{"Status"}))

	mock.ExpectExec(`UPDATE Flow SET Source`).
		WithArgs("custom", sqlmock.AnyArg()).
		WillReturnError(errors.New("lock wait timeout"))

	err = migrateTenantFlowSourceNormalize(db, "2026-04-28 12:00:00")
	require.Error(t, err)
	assert.Contains(t, err.Error(), "tenant")
	require.NoError(t, mock.ExpectationsWereMet())
}

func TestStdConvertLead_StandardActionRowMatchesFlowPack(t *testing.T) {
	// Load-bearing: the action row's flowDeveloperName must match a flow
	// file in the pack, otherwise the dispatcher would route to a
	// non-existent flow at runtime.
	var actionDefs []StandardActionDef
	require.NoError(t, json.Unmarshal(standardActionsJSON, &actionDefs))

	flowSeeds, err := loadStandardFlowSeeds()
	require.NoError(t, err)

	flowNames := map[string]struct{}{}
	for _, seed := range flowSeeds {
		flowNames[seed.DeveloperName] = struct{}{}
	}

	for _, a := range actionDefs {
		_, ok := flowNames[a.FlowDeveloperName]
		assert.True(t, ok, "standard action %q references unknown flow %q", a.Name, a.FlowDeveloperName)
	}
}