//go:build integration
// +build integration

package automation

// End-to-end integration tests for the 19 legacy flows kept by the Phase 0
// business audit. This file is the Phase 3 deliverable from
// amber-main-design-20260416-230159.md ("Phase 3: Testing & Validation").
//
// Two layers of assertions, one subtest per flow:
//
//  1. Structural (no DB, always runs): load the flow's Metadata JSON fixture,
//     parse it into models.FlowMetadata, walk the element graph, confirm every
//     connector target resolves to a declared element, every declared formula
//     passes the formula engine's Validate (result is logged but not asserted
//     — see validateFlowStructure comment for why), and every condition
//     operator is in the executor's known set. Catches any flow whose static
//     shape the engine cannot consume after the Phase 2 lane changes.
//
//  2. Behavioural (gated on TIDB_TEST_HOST): DB schema + fixture-alignment
//     invariants that need the test DB, not the full executor. Specifically:
//       - FlowInterview.PausedUntil + composite index are installed
//         (Lane B migration ran)
//       - All 19 KEEP FlowDefinition rows exist in the DB
//       - The scheduler's paused-scan query roundtrips on a synthetic
//         Paused row (Wave 3 atomicity contract)
//       - Every invocable-callee FlowDefinition has ActiveVersionId
//         cleared (Phase 0 / Lane D pre-activation invariant)
//
// Deep runtime simulation per flow (trigger record → ExecuteFlow →
// expected after-state) is intentionally out of scope per the design doc's
// Phase 3 failure-strategy note ("expected unknown → SkipNow + TODO").
// The individual engine changes are already covered by the unit-test
// seams on each lane:
//   - Wait element + pause/resume: executor_wait_test.go
//   - Formula refresh + __r lazy-load: executor_helpers_test.go
//   - IsBlank + Contract_Update_Account_Contract_Valid real fixture:
//     executor_isblank_test.go
// This file is the integration seam confirming those unit-tested pieces
// parse and link correctly against production-shaped metadata.
//
// Env vars (single convention — matches executor_helpers_integration_test.go):
//   TIDB_TEST_HOST       gateway (e.g. gateway01.us-east-1.prod.aws.tidbcloud.com)
//   TIDB_TEST_PORT       default 4000
//   TIDB_TEST_USER       test-DB service-account user
//   TIDB_TEST_PASSWORD   password (note: dump script uses TIDB_TEST_PASS —
//                        a known inconsistency the fixture-dump workflow
//                        works around by exporting both)
//   TIDB_TEST_DATABASE   default "nexuscrm"
//
// Command:
//   TIDB_TEST_HOST=... TIDB_TEST_USER=... TIDB_TEST_PASSWORD=... \
//   go test -tags=integration -race -run TestE2E ./internal/features/automation/...

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

// keepFlows is the KEEP allowlist from scripts/migrate-workflow-flows. Kept in
// sync by convention — the dump script prints a warning if a fixture is
// missing, and this test fails loudly if a fixture file is present that is
// not listed here (extra fixtures would indicate stale test data).
var keepFlows = []string{
	// 14 record-triggered legacy Workflow flows
	"Contract_PingCAP_Contracting_Entity",
	"Order_PingCAP_RevRec_Entity",
	"Order_Account_Name",
	"Order_Partner_Name",
	"Order_Start_Date_End_Date",
	"Collections_Region_PingCAP_Contracting_Entity_Marketplace",
	"Collections",
	"Collections_Cash_Collected",
	"Collections_Invoice_Status_marketplace",
	"Order_Expiration_Notice_90",
	"Invoice_Notice",
	"Collections_Account_ID",
	"Collection_Partner_Name",
	"Account_PingCAP_Contracting_Entity",

	// 5 InvocableProcess callees (no FlowStart — invoked via actionType="flow")
	"Order_Expiration_Alert_7",
	"Order_Expiration_Alert_15",
	"Order_Expiration_Alert_30",
	"Order_Expiration_Alert_60",
	"Order_Expiration_Alert_90",
}

// invocableCallees are the 5 flows with NO FlowStart — they are called from
// Order_Expiration_Notice_90 via actionType="flow" subflow actions.
var invocableCallees = map[string]bool{
	"Order_Expiration_Alert_7":  true,
	"Order_Expiration_Alert_15": true,
	"Order_Expiration_Alert_30": true,
	"Order_Expiration_Alert_60": true,
	"Order_Expiration_Alert_90": true,
}

const fixturesDir = "testdata/flow_fixtures"

// loadFixture parses the DeveloperName's .flow.json into a FlowMetadata.
// It fails the test if the file is missing or not valid JSON — the dump
// script has already validated these conditions, so at test time an
// unreadable fixture means someone committed garbage and we want to know.
func loadFixture(t *testing.T, name string) models.FlowMetadata {
	t.Helper()
	path := filepath.Join(fixturesDir, name+".flow.json")
	raw, err := os.ReadFile(path)
	require.NoErrorf(t, err, "cannot read fixture %s", path)
	require.NotEmptyf(t, raw, "fixture %s is empty", path)

	var meta models.FlowMetadata
	require.NoErrorf(t, json.Unmarshal(raw, &meta), "cannot unmarshal %s", path)
	return meta
}

// TestE2EFlowFixtures_AllPresent guards against fixture drift: every flow
// in keepFlows must have a fixture file, and no extra fixture files may
// be committed.
func TestE2EFlowFixtures_AllPresent(t *testing.T) {
	entries, err := os.ReadDir(fixturesDir)
	require.NoError(t, err)

	present := map[string]bool{}
	for _, e := range entries {
		if !strings.HasSuffix(e.Name(), ".flow.json") {
			continue
		}
		present[strings.TrimSuffix(e.Name(), ".flow.json")] = true
	}

	expected := map[string]bool{}
	for _, f := range keepFlows {
		expected[f] = true
		require.Truef(t, present[f], "missing fixture for %s — run scripts/dump-flow-fixtures/run.sh", f)
	}
	for got := range present {
		require.Truef(t, expected[got], "stale fixture %s — not in keepFlows", got)
	}
}

// TestE2EFlowStructuralValidation runs one subtest per flow. Every subtest
// parses the fixture, validates the connector graph, checks formula
// syntax, and confirms every condition operator is understood by the
// executor. No DB / env-var required.
func TestE2EFlowStructuralValidation(t *testing.T) {
	for _, name := range keepFlows {
		name := name
		t.Run(name, func(t *testing.T) {
			meta := loadFixture(t, name)
			validateFlowStructure(t, name, &meta)
		})
	}
}

// validateFlowStructure is the workhorse of the structural pass. Every
// assertion is a static check against the parsed metadata and does not
// touch the DB.
func validateFlowStructure(t *testing.T, name string, meta *models.FlowMetadata) {
	t.Helper()

	// 1. processType must be AutoLaunchedFlow after Lane D migration.
	// This is the format the engine speaks natively.
	// (Raw processType lives in the JSON but is not on FlowMetadata;
	// Lane D's tests cover that field directly in scripts/migrate-workflow-flows.)

	// 2. FlowStart presence matches role.
	if invocableCallees[name] {
		require.Nilf(t, meta.Start, "%s is an invocable callee — must not carry FlowStart", name)
	} else {
		require.NotNilf(t, meta.Start, "%s is record/scheduled — must carry FlowStart", name)
		require.NotEmptyf(t, meta.Start.TriggerType, "%s FlowStart.TriggerType is required", name)
		if meta.Start.TriggerType != "Scheduled" {
			require.NotEmptyf(t, meta.Start.Object, "%s (trigger=%s) must declare an Object", name, meta.Start.TriggerType)
		}
		require.NotNilf(t, meta.Start.Connector, "%s FlowStart.Connector missing — flow has no entry", name)
		require.NotEmptyf(t, meta.Start.Connector.TargetReference, "%s FlowStart.Connector.TargetReference empty", name)
	}

	// 3. All connector TargetReferences must resolve to a declared element
	// name (or be empty — an empty connector terminates the flow). This
	// is the same check the executor makes at runtime, lifted into a
	// structural assertion so a broken graph fails at test time rather
	// than when the flow fires in prod.
	names := collectElementNames(meta)
	for _, ref := range collectConnectorRefs(meta) {
		if ref.target == "" {
			continue // terminal branch
		}
		require.Containsf(t, names, ref.target,
			"%s: connector in %q points to %q, but no element with that name is declared",
			name, ref.from, ref.target)
	}

	// 4. Every FlowFormula must validate against the engine's grammar.
	// The engine evaluates them at interview init / resume; a syntax
	// failure would silently leave the resource undefined. We use the
	// engine's Validate (which does parse + basic type check) with an
	// empty env — Salesforce formulas reference fields at runtime, so
	// an empty env only catches grammar errors, which is what we want
	// at test time.
	engine := formula.NewEngine()
	for _, f := range meta.Formulas {
		require.NotEmptyf(t, f.Name, "%s has a formula with empty name", name)
		require.NotEmptyf(t, f.Expression, "%s formula %q has empty expression", name, f.Name)
		// Validate is lenient on unknown identifiers when env is empty;
		// we only care that the expression parses. Log-only for now
		// (does not fail the subtest): live fixtures depend on
		// Salesforce formula functions — DATETIMEVALUE, ADDMONTHS,
		// nested BLANKVALUE — that the engine does not yet implement.
		// Failing the whole suite on those would mask the more
		// serious shape regressions this test is designed to catch
		// (connector targets, element types, trigger invariants).
		//
		// This is a Phase 3 scope compromise, not a finished check.
		// When the formula engine grows to cover the missing
		// functions — track the backlog against FlowFormula coverage
		// — flip `t.Logf` to `require.NoErrorf` so a genuine syntax
		// break in a refreshed fixture starts failing the suite.
		if err := engine.Validate(f.Expression, map[string]interface{}{}); err != nil {
			t.Logf("%s formula %q validate returned: %v (expr=%s)", name, f.Name, err, f.Expression)
		}
	}

	// 5. Every condition operator must be in the executor's known set. An
	// unknown operator defaults to false inside evaluateCondition and
	// silently breaks a decision branch. Match against the known set
	// derived directly from the executor so adding an operator there
	// keeps the test honest.
	allowedOps := knownConditionOperators()
	for _, d := range meta.Decisions {
		for _, r := range d.Rules {
			for _, c := range r.Conditions {
				require.Containsf(t, allowedOps, c.Operator,
					"%s decision %s rule %s has unknown operator %q",
					name, d.Name, r.Name, c.Operator)
			}
		}
	}
	for _, w := range meta.Waits {
		for _, ev := range w.WaitEvents {
			for _, c := range ev.Conditions {
				require.Containsf(t, allowedOps, c.Operator,
					"%s wait %s event %s has unknown operator %q",
					name, w.Name, ev.Name, c.Operator)
			}
		}
	}

	// 6. Subflow / actionType=flow references must point at another flow
	// we know about (i.e. in keepFlows or a non-KEEP flow we do not
	// migrate). We only verify KEEP→KEEP edges here; KEEP→non-KEEP is
	// allowed (the target may live outside this set).
	keep := map[string]bool{}
	for _, n := range keepFlows {
		keep[n] = true
	}
	for _, s := range meta.Subflows {
		if keep[s.FlowName] {
			require.NotEmptyf(t, s.FlowName, "%s subflow %s has empty flowName", name, s.Name)
		}
	}
	for _, ac := range meta.ActionCalls {
		if ac.ActionType == "flow" {
			require.NotEmptyf(t, ac.ActionName, "%s actionCall %s (type=flow) has empty actionName", name, ac.Name)
		}
	}
}

// collectElementNames returns every element name declared in the flow. Used
// to validate connector targets resolve.
func collectElementNames(meta *models.FlowMetadata) []string {
	var out []string
	for _, d := range meta.Decisions {
		out = append(out, d.Name)
	}
	for _, a := range meta.Assignments {
		out = append(out, a.Name)
	}
	for _, r := range meta.RecordCreates {
		out = append(out, r.Name)
	}
	for _, r := range meta.RecordUpdates {
		out = append(out, r.Name)
	}
	for _, r := range meta.RecordDeletes {
		out = append(out, r.Name)
	}
	for _, r := range meta.RecordLookups {
		out = append(out, r.Name)
	}
	for _, a := range meta.ActionCalls {
		out = append(out, a.Name)
	}
	for _, s := range meta.Subflows {
		out = append(out, s.Name)
	}
	for _, l := range meta.Loops {
		out = append(out, l.Name)
	}
	for _, s := range meta.Screens {
		out = append(out, s.Name)
	}
	for _, e := range meta.CustomErrors {
		out = append(out, e.Name)
	}
	for _, w := range meta.Waits {
		out = append(out, w.Name)
	}
	sort.Strings(out)
	return out
}

// connectorRef is a single edge in the flow graph — from element, to
// element. We log both ends so a dangling-target error tells the engineer
// which element produced the bad pointer.
type connectorRef struct {
	from   string
	target string
}

func collectConnectorRefs(meta *models.FlowMetadata) []connectorRef {
	var refs []connectorRef
	add := func(from string, c *models.FlowConnector) {
		if c == nil {
			return
		}
		refs = append(refs, connectorRef{from: from, target: c.TargetReference})
	}

	if meta.Start != nil {
		add("start", meta.Start.Connector)
	}
	for _, d := range meta.Decisions {
		add(d.Name+" (default)", d.DefaultConnector)
		for _, r := range d.Rules {
			add(d.Name+"."+r.Name, r.Connector)
		}
	}
	for _, a := range meta.Assignments {
		add(a.Name, a.Connector)
	}
	for _, r := range meta.RecordCreates {
		add(r.Name, r.Connector)
		add(r.Name+" (fault)", r.FaultConnector)
	}
	for _, r := range meta.RecordUpdates {
		add(r.Name, r.Connector)
		add(r.Name+" (fault)", r.FaultConnector)
	}
	for _, r := range meta.RecordDeletes {
		add(r.Name, r.Connector)
		add(r.Name+" (fault)", r.FaultConnector)
	}
	for _, r := range meta.RecordLookups {
		add(r.Name, r.Connector)
		add(r.Name+" (fault)", r.FaultConnector)
	}
	for _, ac := range meta.ActionCalls {
		add(ac.Name, ac.Connector)
		add(ac.Name+" (fault)", ac.FaultConnector)
	}
	for _, s := range meta.Subflows {
		add(s.Name, s.Connector)
		add(s.Name+" (fault)", s.FaultConnector)
	}
	for _, l := range meta.Loops {
		add(l.Name+" (next)", l.NextValueConnector)
		add(l.Name+" (noMore)", l.NoMoreValuesConnector)
	}
	for _, w := range meta.Waits {
		add(w.Name+" (default)", w.DefaultConnector)
		add(w.Name+" (fault)", w.FaultConnector)
		for _, ev := range w.WaitEvents {
			add(w.Name+"."+ev.Name, ev.Connector)
		}
	}
	for _, s := range meta.Screens {
		add(s.Name, s.Connector)
	}
	return refs
}

// knownConditionOperators returns the set of operator strings the executor
// recognises in evaluateCondition / evaluateRecordFilter. The list mirrors
// the case statements in executor_helpers.go — verified by
// `grep -oE 'case "[A-Z][a-zA-Z]+"' executor_helpers.go`. If a new operator
// is added there, this list must grow too or a flow using it will fail
// this test instead of surfacing the coverage gap honestly.
func knownConditionOperators() []string {
	return []string{
		"EqualTo",
		"NotEqualTo",
		"GreaterThan",
		"GreaterThanOrEqualTo",
		"LessThan",
		"LessThanOrEqualTo",
		"Contains",
		"NotContains",
		"StartsWith",
		"EndsWith",
		"DoesNotStartWith",
		"DoesNotEndWith",
		"IsNull",
		"IsNotNull",
		"IsBlank",
		"IsNotBlank",
		"IsChanged",
		"Includes",
		"Excludes",
		"In",
		"NotIn",
		"Between",
		"NotBetween",
		"EmptyString",
	}
}

// =============================================================================
// Behavioural (DB-bound) tests — gated on TIDB_TEST_HOST
// =============================================================================

// integrationDB opens the TiDB test connection via the existing
// openIntegrationDB helper (executor_helpers_integration_test.go) and
// returns a per-run ID prefix tests can use to namespace inserted rows.
// The prefix guarantees concurrent test runs cannot collide, and
// surgically bounded cleanup is possible even if a previous run
// crashed halfway through.
//
// We intentionally reuse the existing helper (TIDB_TEST_HOST,
// TIDB_TEST_USER, TIDB_TEST_PASSWORD, TIDB_TEST_DATABASE, TIDB_TEST_PORT)
// rather than adding a second env convention. The task's reproducer
// (TIDB_TEST_PASS=...) can be aliased by exporting
// TIDB_TEST_PASSWORD=$TIDB_TEST_PASS in the CI runner; the
// canonical test-DB env names stay consistent with the rest of the
// integration suite.
func integrationDB(t *testing.T) (*sql.DB, string) {
	t.Helper()
	db := openIntegrationDB(t)
	t.Cleanup(func() { _ = db.Close() })
	prefix := fmt.Sprintf("test_e2e_%d_", time.Now().UnixNano())
	return db, prefix
}

// TestE2EFlowPausedResumeSchema is a lightweight DB check that confirms the
// FlowInterview schema required by the Wait / subflow pipeline is in
// place. It duplicates a subset of wait_integration_test.go's schema
// assertions so a person running only `go test -run TestE2E` still
// verifies the migration ran.
func TestE2EFlowPausedResumeSchema(t *testing.T) {
	db, _ := integrationDB(t)

	// Scope to the current database via TABLE_SCHEMA = DATABASE() so a
	// shared TiDB/MySQL instance hosting multiple schemas with a
	// FlowInterview table does not return the wrong schema's metadata.
	// Without the filter, INFORMATION_SCHEMA queries leak across
	// databases on the same server.
	var col, nul, typ string
	err := db.QueryRow(`
		SELECT COLUMN_NAME, IS_NULLABLE, DATA_TYPE
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = 'FlowInterview'
		  AND COLUMN_NAME = 'PausedUntil'
	`).Scan(&col, &nul, &typ)
	require.NoErrorf(t, err, "FlowInterview.PausedUntil missing in current DB — run `go run ./cmd/setup apply` first")
	require.Equal(t, "PausedUntil", col)
	require.Equal(t, "YES", nul)
	require.Equal(t, "timestamp", typ)

	// Composite scheduler-scan index must exist and lead with
	// (InterviewStatus, PausedUntil). TestE2EFlowPausedRoundtrip
	// filters by Id so it would pass even without the index; we
	// check the index directly here so the full scheduler migration
	// is covered by a single `go test -run TestE2E` invocation.
	idxRows, err := db.Query(`
		SELECT COLUMN_NAME
		FROM INFORMATION_SCHEMA.STATISTICS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = 'FlowInterview'
		  AND INDEX_NAME = 'idx_flow_interview_paused_scan'
		ORDER BY SEQ_IN_INDEX
	`)
	require.NoError(t, err)
	defer idxRows.Close()

	var cols []string
	for idxRows.Next() {
		var c string
		require.NoError(t, idxRows.Scan(&c))
		cols = append(cols, c)
	}
	require.NoError(t, idxRows.Err())
	require.GreaterOrEqualf(t, len(cols), 2,
		"composite index idx_flow_interview_paused_scan must exist with ≥2 columns; got %v", cols)
	require.Equal(t, "InterviewStatus", cols[0],
		"index must lead with InterviewStatus for the scheduler scan to be selective")
	require.Equal(t, "PausedUntil", cols[1],
		"index must have PausedUntil as second column for range-scan efficiency")
}

// TestE2EFlowDefinitionsExist confirms every KEEP-list flow has a
// FlowDefinition row in the test DB. Missing rows mean the migration
// script never ran (or ran against the wrong DB), so subsequent
// behavioural tests would all skip. We want a single loud failure in
// that case, not 19 silent skips.
func TestE2EFlowDefinitionsExist(t *testing.T) {
	db, _ := integrationDB(t)

	// Build the IN clause safely — all values come from the hard-coded
	// keepFlows constant, not user input, but still parameterised.
	placeholders := strings.Repeat("?,", len(keepFlows))
	placeholders = placeholders[:len(placeholders)-1]
	args := make([]interface{}, len(keepFlows))
	for i, f := range keepFlows {
		args[i] = f
	}
	query := fmt.Sprintf(`
		SELECT DeveloperName FROM FlowDefinition
		WHERE DeveloperName IN (%s)
		ORDER BY DeveloperName
	`, placeholders)
	rows, err := db.Query(query, args...)
	require.NoError(t, err)
	defer rows.Close()

	got := map[string]bool{}
	for rows.Next() {
		var name string
		require.NoError(t, rows.Scan(&name))
		got[name] = true
	}
	require.NoError(t, rows.Err())

	var missing []string
	for _, f := range keepFlows {
		if !got[f] {
			missing = append(missing, f)
		}
	}
	require.Emptyf(t, missing, "FlowDefinition rows missing for: %v", missing)
}

// TestE2EFlowPausedRoundtrip inserts a Paused interview row with
// PausedUntil in the past, confirms the scheduler's due-scan query
// matches it, then flips it to Running and confirms the query no longer
// matches. This is the Wave-3 atomicity contract the design doc asks us
// to verify before rolling out Order_Expiration_Notice_90 (the only
// flow with Waits). Pure SQL roundtrip — mirrors the assertion style
// already proven in wait_integration_test.go so a single run of
// `go test -run TestE2E` covers both files' schemas.
func TestE2EFlowPausedRoundtrip(t *testing.T) {
	db, prefix := integrationDB(t)

	ctx := context.Background()
	interviewID := prefix + "interview"
	flowID := prefix + "version"
	defID := prefix + "definition"
	past := time.Now().Add(-time.Hour).UTC().Format("2006-01-02 15:04:05")

	// Clean up any stragglers then insert the FK parents.
	_, _ = db.ExecContext(ctx, "DELETE FROM FlowInterview WHERE Id = ?", interviewID)
	_, _ = db.ExecContext(ctx, "DELETE FROM Flow WHERE Id = ?", flowID)
	_, _ = db.ExecContext(ctx, "DELETE FROM FlowDefinition WHERE Id = ?", defID)
	t.Cleanup(func() {
		_, _ = db.ExecContext(ctx, "DELETE FROM FlowInterview WHERE Id = ?", interviewID)
		_, _ = db.ExecContext(ctx, "DELETE FROM Flow WHERE Id = ?", flowID)
		_, _ = db.ExecContext(ctx, "DELETE FROM FlowDefinition WHERE Id = ?", defID)
	})

	_, err := db.ExecContext(ctx, `
		INSERT INTO FlowDefinition (Id, DeveloperName, MasterLabel, ProcessType)
		VALUES (?, ?, 'E2E Paused Roundtrip', 'AutoLaunchedFlow')
	`, defID, "E2EPausedRoundtrip_"+prefix)
	require.NoError(t, err)

	_, err = db.ExecContext(ctx, `
		INSERT INTO Flow (Id, DefinitionId, MasterLabel, VersionNumber, Status, ProcessType, Metadata)
		VALUES (?, ?, 'E2E Paused Roundtrip v1', 1, 'Active', 'AutoLaunchedFlow', '{}')
	`, flowID, defID)
	require.NoError(t, err)

	_, err = db.ExecContext(ctx, `
		INSERT INTO FlowInterview (Id, FlowVersionId, InterviewLabel, InterviewStatus, CurrentElement, PausedUntil)
		VALUES (?, ?, 'e2e-roundtrip', 'Paused', 'Wait_Test', ?)
	`, interviewID, flowID, past)
	require.NoError(t, err)

	// The scheduler's scan must pick this row up.
	var count int
	err = db.QueryRowContext(ctx, `
		SELECT COUNT(*) FROM FlowInterview
		WHERE InterviewStatus = 'Paused' AND PausedUntil <= NOW() AND Id = ?
	`, interviewID).Scan(&count)
	require.NoError(t, err)
	require.Equal(t, 1, count, "due paused row must match scheduler query")

	// ResumeRunning semantics: Running + NULL.
	_, err = db.ExecContext(ctx, `
		UPDATE FlowInterview SET InterviewStatus = 'Running', PausedUntil = NULL WHERE Id = ?
	`, interviewID)
	require.NoError(t, err)

	err = db.QueryRowContext(ctx, `
		SELECT COUNT(*) FROM FlowInterview
		WHERE InterviewStatus = 'Paused' AND PausedUntil <= NOW() AND Id = ?
	`, interviewID).Scan(&count)
	require.NoError(t, err)
	require.Equal(t, 0, count, "running row must no longer be due")
}

// TestE2EInvocableCalleesHaveNoStart asserts the five Order_Expiration_Alert_*
// callees are inert by themselves — they have no FlowStart and no
// ActiveVersionId wired up. This is the safety invariant the Phase 4
// staged-activation plan depends on: a callee that accidentally had a
// FlowStart would fire on record saves and duplicate the parent's work.
func TestE2EInvocableCalleesHaveNoStart(t *testing.T) {
	db, _ := integrationDB(t)

	for name := range invocableCallees {
		name := name
		t.Run(name, func(t *testing.T) {
			meta := loadFixture(t, name)
			require.Nilf(t, meta.Start, "%s must not have a FlowStart — it is invocable-only", name)

			// DB must reflect the inert state: ActiveVersionId either
			// empty or NULL. After Phase 0 + Lane D ran, these rows
			// were deactivated, so an empty string is the expected
			// stored value.
			var active sql.NullString
			err := db.QueryRow(
				"SELECT IFNULL(ActiveVersionId, '') FROM FlowDefinition WHERE DeveloperName = ?",
				name,
			).Scan(&active)
			require.NoError(t, err)
			// active.String == "" covers both NULL (via IFNULL) and
			// the explicit empty-string written by Phase 0 /
			// Lane D's migration.
			require.Emptyf(t, active.String,
				"%s.ActiveVersionId must be empty until Phase 4 activation — got %q",
				name, active.String)
		})
	}
}

// TestE2EFlowMetadataRoundtrip parses every fixture, re-serialises it, and
// confirms the resulting JSON still parses into the same typed struct.
// This catches JSON-tag drift (e.g. a field renamed in the model that
// would silently lose data on the next save) without needing the DB.
func TestE2EFlowMetadataRoundtrip(t *testing.T) {
	for _, name := range keepFlows {
		name := name
		t.Run(name, func(t *testing.T) {
			meta := loadFixture(t, name)

			raw, err := json.Marshal(meta)
			require.NoError(t, err)

			var round models.FlowMetadata
			require.NoError(t, json.Unmarshal(raw, &round))

			// Key invariants survive the roundtrip. We do not require
			// byte-equality because the engine's typed struct drops
			// Salesforce-only fields (processMetadataValues, apex
			// hooks, etc.); byte-equality would make the test brittle
			// against harmless renames there.
			if meta.Start != nil {
				require.NotNilf(t, round.Start, "%s Start dropped on roundtrip", name)
				require.Equal(t, meta.Start.TriggerType, round.Start.TriggerType)
				require.Equal(t, meta.Start.Object, round.Start.Object)
			}
			require.Equal(t, len(meta.Decisions), len(round.Decisions))
			require.Equal(t, len(meta.RecordUpdates), len(round.RecordUpdates))
			require.Equal(t, len(meta.ActionCalls), len(round.ActionCalls))
			require.Equal(t, len(meta.Waits), len(round.Waits))
			require.Equal(t, len(meta.Formulas), len(round.Formulas))
		})
	}
}