package automation

import (
	"context"
	"fmt"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

type flowParentPrelockQuery struct {
	*countingQuery
	specs []queryspec.QuerySpec
}

func (q *flowParentPrelockQuery) QueryWithSpec(_ context.Context, spec queryspec.QuerySpec, _ *models.UserSession) (*models.QueryResult, error) {
	q.specs = append(q.specs, spec)
	if spec.Offset == 0 {
		records := make([]models.SObject, flowAtomicParentPrelockPageSize)
		for i := range records {
			records[i] = models.SObject{"CampaignId": fmt.Sprintf("campaign-%03d", i)}
		}
		return &models.QueryResult{Records: records}, nil
	}
	return &models.QueryResult{Records: []models.SObject{
		{"CampaignId": "campaign-199"},
		{"CampaignId": "campaign-final"},
	}}, nil
}

type leadConvertParentPrelockQuery struct {
	*countingQuery
}

type boundedParentPrelockQuery struct {
	*countingQuery
	records []models.SObject
	specs   []queryspec.QuerySpec
}

func (q *boundedParentPrelockQuery) QueryWithSpec(_ context.Context, spec queryspec.QuerySpec, _ *models.UserSession) (*models.QueryResult, error) {
	q.specs = append(q.specs, spec)
	if spec.Offset >= len(q.records) {
		return &models.QueryResult{}, nil
	}
	end := spec.Offset + spec.Limit
	if end > len(q.records) {
		end = len(q.records)
	}
	return &models.QueryResult{Records: q.records[spec.Offset:end]}, nil
}

func (q *leadConvertParentPrelockQuery) QueryWithSpec(_ context.Context, spec queryspec.QuerySpec, _ *models.UserSession) (*models.QueryResult, error) {
	switch spec.ObjectAPIName {
	case "Account":
		return &models.QueryResult{Records: []models.SObject{{"Id": "account-1"}}}, nil
	case "CampaignMember":
		return &models.QueryResult{Records: []models.SObject{
			{"CampaignId": "campaign-a"},
			{"CampaignId": "campaign-b"},
		}}, nil
	case "Lead":
		return &models.QueryResult{Records: []models.SObject{{"Last_Campaign__c": "campaign-b"}}}, nil
	default:
		return nil, fmt.Errorf("unexpected prelock source object %q", spec.ObjectAPIName)
	}
}

func TestDiscoverFlowAtomicParentPrelocksCompletesPagedDiscoveryBeforeLockPlanning(t *testing.T) {
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := &FlowExecutor{query: query}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"leadId":    "lead-1",
		"accountId": "account-1",
	}}
	meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{
		{
			Name:         "ExistingAccount",
			ParentObject: "Account",
			RecordIDs: []interface{}{
				map[string]interface{}{"elementReference": "accountId"},
			},
		},
		{
			Name:                 "LeadCampaigns",
			ParentObject:         "Campaign",
			SourceObject:         "CampaignMember",
			ParentReferenceField: "CampaignId",
			Filters: []models.FlowRecordFilter{{
				Field:    "LeadId",
				Operator: "EqualTo",
				Value:    map[string]interface{}{"elementReference": "leadId"},
			}},
		},
	}}

	keys, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), interview, meta, &models.UserSession{})
	if err != nil {
		t.Fatalf("discoverFlowAtomicParentPrelocks() error = %v", err)
	}
	if got, want := len(query.specs), 2; got != want {
		t.Fatalf("query calls = %d, want %d", got, want)
	}
	if query.specs[0].Offset != 0 || query.specs[1].Offset != flowAtomicParentPrelockPageSize {
		t.Fatalf("query offsets = [%d, %d], want [0, %d]", query.specs[0].Offset, query.specs[1].Offset, flowAtomicParentPrelockPageSize)
	}
	if got, want := len(keys), flowAtomicParentPrelockPageSize+2; got != want {
		t.Fatalf("discovered keys = %d, want %d", got, want)
	}
	if keys[0] != (recordlocks.ParentKey{ObjectName: "Account", RecordID: "account-1"}) {
		t.Fatalf("first key = %#v, want existing Account before Campaign keys", keys[0])
	}
	if got := keys[len(keys)-1]; got != (recordlocks.ParentKey{ObjectName: "Campaign", RecordID: "campaign-final"}) {
		t.Fatalf("last key = %#v, want final page Campaign", got)
	}
}

func TestResourceBoundedPreparedFlowCapsParentPrelockDiscoveryBeforeLockPlanning(t *testing.T) {
	query := &boundedParentPrelockQuery{
		countingQuery: &countingQuery{},
		records: []models.SObject{
			{"CampaignId": "campaign-a"},
			{"CampaignId": "campaign-b"},
			{"CampaignId": "campaign-c"},
		},
	}
	executor := &FlowExecutor{query: query}
	meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
		Name: "LeadCampaigns", ParentObject: "Campaign",
		SourceObject: "CampaignMember", ParentReferenceField: "CampaignId",
		Filters: []models.FlowRecordFilter{{
			Field: "LeadId", Operator: "EqualTo",
			Value: map[string]interface{}{"elementReference": "leadId"},
		}},
	}}}
	ctx, err := withPreparedFlowDirectRecordOperationLimit(context.Background(), 2)
	if err != nil {
		t.Fatal(err)
	}
	_, err = executor.discoverFlowAtomicParentPrelocks(
		ctx,
		&models.FlowInterview{ContextData: map[string]interface{}{"leadId": "lead-1"}},
		meta,
		&models.UserSession{},
	)
	if err == nil || !strings.Contains(err.Error(), "parent prelock discovery exceeds 2 rows") {
		t.Fatalf("discoverFlowAtomicParentPrelocks() error = %v, want bound rejection", err)
	}
	if len(query.specs) != 1 || query.specs[0].Limit != 3 {
		t.Fatalf("query specs = %#v, want one limit+1 proof query", query.specs)
	}
}

func TestResourceBoundedPreparedFlowAcceptsParentPrelockDiscoveryAtLimit(t *testing.T) {
	query := &boundedParentPrelockQuery{
		countingQuery: &countingQuery{},
		records: []models.SObject{
			{"CampaignId": "campaign-a"},
			{"CampaignId": "campaign-b"},
		},
	}
	executor := &FlowExecutor{query: query}
	meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
		Name: "LeadCampaigns", ParentObject: "Campaign",
		SourceObject: "CampaignMember", ParentReferenceField: "CampaignId",
		Filters: []models.FlowRecordFilter{{
			Field: "LeadId", Operator: "EqualTo",
			Value: map[string]interface{}{"elementReference": "leadId"},
		}},
	}}}
	ctx, err := withPreparedFlowDirectRecordOperationLimit(context.Background(), 2)
	if err != nil {
		t.Fatal(err)
	}
	keys, err := executor.discoverFlowAtomicParentPrelocks(
		ctx,
		&models.FlowInterview{ContextData: map[string]interface{}{"leadId": "lead-1"}},
		meta,
		&models.UserSession{},
	)
	if err != nil {
		t.Fatalf("discoverFlowAtomicParentPrelocks() error = %v", err)
	}
	if len(keys) != 2 {
		t.Fatalf("discovered keys = %d, want 2", len(keys))
	}
}

func TestWithFlowAtomicParentPrelocksUnionsRetryDiscoveredKeys(t *testing.T) {
	retryKey := recordlocks.ParentKey{ObjectName: "Campaign", RecordID: "campaign-retry"}
	declaredKey := recordlocks.ParentKey{ObjectName: "Campaign", RecordID: "campaign-declared"}
	ctx := recordlocks.WithForcedParentPrelocks(context.Background(), []recordlocks.ParentKey{retryKey})

	got := recordlocks.ForcedParentPrelocksFromContext(withFlowAtomicParentPrelocks(ctx, []recordlocks.ParentKey{declaredKey}))
	want := recordlocks.OrderedParentKeys([]recordlocks.ParentKey{retryKey, declaredKey})
	if fmt.Sprint(got) != fmt.Sprint(want) {
		t.Fatalf("forced prelocks = %#v, want union %#v", got, want)
	}
}

func TestDiscoverFlowAtomicParentPrelocksUnwrapsTypedInputRecordIDs(t *testing.T) {
	executor := &FlowExecutor{query: &flowParentPrelockQuery{countingQuery: &countingQuery{}}}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"accountId": map[string]interface{}{"stringValue": "account-typed"},
	}}
	meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
		Name:         "ExistingAccount",
		ParentObject: "Account",
		RecordIDs: []interface{}{
			map[string]interface{}{"elementReference": "accountId"},
		},
	}}}

	keys, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), interview, meta, &models.UserSession{})
	if err != nil {
		t.Fatalf("discoverFlowAtomicParentPrelocks() error = %v", err)
	}
	want := []recordlocks.ParentKey{{ObjectName: "Account", RecordID: "account-typed"}}
	if fmt.Sprint(keys) != fmt.Sprint(want) {
		t.Fatalf("discovered keys = %#v, want typed input normalized to %#v", keys, want)
	}
}

func TestDiscoverFlowAtomicParentPrelocksBuildsFiltersFromNormalizedInputs(t *testing.T) {
	tests := []struct {
		name      string
		operator  string
		input     interface{}
		wantParts []string
	}{
		{
			name:      "starts with",
			operator:  "StartsWith",
			input:     map[string]interface{}{"stringValue": "lead_%"},
			wantParts: []string{"STARTS_WITH(SourceValue, 'lead\\_\\%')"},
		},
		{
			name:      "contains",
			operator:  "Contains",
			input:     map[string]interface{}{"stringValue": "campaign-value"},
			wantParts: []string{"CONTAINS(SourceValue, 'campaign-value')"},
		},
		{
			name:     "between",
			operator: "Between",
			input: []interface{}{
				map[string]interface{}{"dateValue": "2026-07-01"},
				map[string]interface{}{"dateValue": "2026-07-31"},
			},
			wantParts: []string{"SourceValue BETWEEN '2026-07-01' AND '2026-07-31'"},
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
			executor := &FlowExecutor{query: query}
			interview := &models.FlowInterview{ContextData: map[string]interface{}{"filterInput": tc.input}}
			meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:                 "TypedFilter",
				ParentObject:         "Campaign",
				SourceObject:         "CampaignMember",
				ParentReferenceField: "CampaignId",
				Filters: []models.FlowRecordFilter{{
					Field:    "SourceValue",
					Operator: tc.operator,
					Value:    map[string]interface{}{"elementReference": "filterInput"},
				}},
			}}}

			if _, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), interview, meta, &models.UserSession{}); err != nil {
				t.Fatalf("discoverFlowAtomicParentPrelocks() error = %v", err)
			}
			if len(query.specs) == 0 {
				t.Fatal("discoverFlowAtomicParentPrelocks() issued no query")
			}
			filter := query.specs[0].Filter.Operand.Raw
			for _, want := range tc.wantParts {
				if !strings.Contains(filter, want) {
					t.Fatalf("query filter = %q, want normalized literal %q", filter, want)
				}
			}
			if strings.Contains(filter, "map[") {
				t.Fatalf("query filter = %q, typed wrapper leaked into predicate", filter)
			}
		})
	}
}

func TestAtomicFlowInstallsCompleteDiscoveredPrelocksBeforeFirstMutation(t *testing.T) {
	persistence := &transactionalFanoutPersistence{}
	meta := models.FlowMetadata{
		RunAtomically: true,
		Variables: []models.FlowVariable{
			{Name: "leadId", DataType: "String", IsInput: true},
			{Name: "accountId", DataType: "String", IsInput: true},
		},
		AtomicParentPrelocks: []models.FlowAtomicParentPrelock{
			{
				Name:                 "ExistingAccount",
				ParentObject:         "Account",
				SourceObject:         "Account",
				ParentReferenceField: "Id",
				Filters: []models.FlowRecordFilter{{
					Field:    "Id",
					Operator: "EqualTo",
					Value:    map[string]interface{}{"elementReference": "accountId"},
				}},
			},
			{
				Name:                 "LeadCampaigns",
				ParentObject:         "Campaign",
				SourceObject:         "CampaignMember",
				ParentReferenceField: "CampaignId",
				Filters: []models.FlowRecordFilter{{
					Field:    "LeadId",
					Operator: "EqualTo",
					Value:    map[string]interface{}{"elementReference": "leadId"},
				}},
			},
			{
				Name:                 "LeadLastCampaign",
				ParentObject:         "Campaign",
				SourceObject:         "Lead",
				ParentReferenceField: "Last_Campaign__c",
				Filters: []models.FlowRecordFilter{{
					Field:    "Id",
					Operator: "EqualTo",
					Value:    map[string]interface{}{"elementReference": "leadId"},
				}},
			},
		},
		Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CreateResult"}},
		RecordCreates: []models.FlowRecordCreate{{
			Name:   "CreateResult",
			Object: "SyntheticResult",
		}},
	}
	executor := fanoutExecutor(meta, persistence)
	executor.query = &leadConvertParentPrelockQuery{countingQuery: &countingQuery{}}

	result, err := executor.ExecuteFlow(
		WithTransientFlowInterviews(context.Background()),
		"SyntheticAtomicPrelock",
		map[string]interface{}{"leadId": "lead-1", "accountId": "account-1"},
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlow() error = %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("ExecuteFlow() result = %#v, want success", result)
	}
	if got, want := len(persistence.insertPrelocks), 1; got != want {
		t.Fatalf("insert prelock snapshots = %d, want %d", got, want)
	}
	if got, want := len(persistence.insertPrelocks[0]), 3; got != want {
		t.Fatalf("first mutation prelocks = %d, want complete deduplicated set %d", got, want)
	}
	want := []recordlocks.ParentKey{
		{ObjectName: "Account", RecordID: "account-1"},
		{ObjectName: "Campaign", RecordID: "campaign-a"},
		{ObjectName: "Campaign", RecordID: "campaign-b"},
	}
	if fmt.Sprint(persistence.insertPrelocks[0]) != fmt.Sprint(want) {
		t.Fatalf("first mutation prelocks = %#v, want %#v", persistence.insertPrelocks[0], want)
	}
}

func TestValidateFlowAtomicParentPrelocksRejectsUnboundedSourceQuery(t *testing.T) {
	err := validateFlowAtomicParentPrelocks(&models.FlowMetadata{
		RunAtomically: true,
		AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
			Name:                 "Unbounded",
			ParentObject:         "Campaign",
			SourceObject:         "CampaignMember",
			ParentReferenceField: "CampaignId",
		}},
	})
	if err == nil {
		t.Fatal("validateFlowAtomicParentPrelocks() accepted an unbounded source query")
	}
}

func TestDiscoverFlowAtomicParentPrelocksRejectsAlwaysTrueFilter(t *testing.T) {
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := &FlowExecutor{query: query}
	meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
		Name:                 "Malformed",
		ParentObject:         "Campaign",
		SourceObject:         "CampaignMember",
		ParentReferenceField: "CampaignId",
		Filters: []models.FlowRecordFilter{{
			Field:    "LeadId",
			Operator: "NotAQueryOperator",
			Value:    map[string]interface{}{"elementReference": "leadId"},
		}},
	}}}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{"leadId": "lead-1"}}

	_, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), interview, meta, &models.UserSession{})
	if err == nil {
		t.Fatal("discoverFlowAtomicParentPrelocks() accepted a filter that resolved to 1=1")
	}
	if len(query.specs) != 0 {
		t.Fatalf("query calls = %d, want zero for unbounded filter", len(query.specs))
	}
}

func TestDiscoverFlowAtomicParentPrelocksRejectsDroppedBetweenClause(t *testing.T) {
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := &FlowExecutor{query: query}
	meta := &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
		Name:                 "MalformedBounds",
		ParentObject:         "Campaign",
		SourceObject:         "CampaignMember",
		ParentReferenceField: "CampaignId",
		Filters: []models.FlowRecordFilter{
			{Field: "CreatedDate", Operator: "Between", Value: []interface{}{"2026-07-15"}},
			{Field: "LeadId", Operator: "EqualTo", Value: map[string]interface{}{"elementReference": "leadId"}},
		},
	}}}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{"leadId": "lead-1"}}

	_, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), interview, meta, &models.UserSession{})
	if err == nil {
		t.Fatal("discoverFlowAtomicParentPrelocks() silently dropped an invalid Between clause")
	}
	if len(query.specs) != 0 {
		t.Fatalf("query calls = %d, want zero for malformed bounds", len(query.specs))
	}
}

func TestDiscoverFlowAtomicParentPrelocksRejectsUnresolvedInputReferences(t *testing.T) {
	for _, tc := range []struct {
		name string
		meta *models.FlowMetadata
	}{
		{
			name: "record id",
			meta: &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:         "MissingAccount",
				ParentObject: "Account",
				RecordIDs: []interface{}{
					map[string]interface{}{"elementReference": "$Record.Accountd"},
				},
			}}},
		},
		{
			name: "filter value",
			meta: &models.FlowMetadata{AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:                 "MissingLeadFilter",
				ParentObject:         "Campaign",
				SourceObject:         "CampaignMember",
				ParentReferenceField: "CampaignId",
				Filters: []models.FlowRecordFilter{{
					Field:    "LeadId",
					Operator: "EqualTo",
					Value:    map[string]interface{}{"elementReference": "$Record.LeadIdd"},
				}},
			}}},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
			executor := &FlowExecutor{query: query}
			interview := &models.FlowInterview{ContextData: map[string]interface{}{
				"$Record": models.SObject{"AccountId": "account-1", "LeadId": "lead-1"},
			}}

			_, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), interview, tc.meta, &models.UserSession{})
			if err == nil {
				t.Fatal("discoverFlowAtomicParentPrelocks() silently accepted an unresolved input reference")
			}
			if len(query.specs) != 0 {
				t.Fatalf("query calls = %d, want zero before unresolved input rejection", len(query.specs))
			}
		})
	}
}

func TestFlowAtomicParentPrelocksRejectAmbiguousBareStrings(t *testing.T) {
	for _, tc := range []struct {
		name string
		meta models.FlowMetadata
	}{
		{
			name: "input-shaped direct id",
			meta: models.FlowMetadata{
				RunAtomically: true,
				Variables:     []models.FlowVariable{{Name: "accountId", DataType: "String", IsInput: true}},
				AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
					Name: "BareInput", ParentObject: "Account", RecordIDs: []interface{}{"accountId"},
				}},
			},
		},
		{
			name: "missing filter variable",
			meta: models.FlowMetadata{
				RunAtomically: true,
				AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
					Name: "BareMissing", ParentObject: "Campaign", SourceObject: "CampaignMember", ParentReferenceField: "CampaignId",
					Filters: []models.FlowRecordFilter{{Field: "LeadId", Operator: "EqualTo", Value: "missingVar"}},
				}},
			},
		},
		{
			name: "post-start name",
			meta: models.FlowMetadata{
				RunAtomically: true,
				Formulas:      []models.FlowFormula{{Name: "ComputedLeadId", DataType: "String", Expression: "leadId"}},
				AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
					Name: "BareFormula", ParentObject: "Campaign", RecordIDs: []interface{}{"ComputedLeadId"},
				}},
			},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if err := validateFlowAtomicParentPrelocks(&tc.meta); err == nil {
				t.Fatal("validateFlowAtomicParentPrelocks() accepted an ambiguous bare string")
			}

			query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
			executor := &FlowExecutor{query: query}
			_, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), &models.FlowInterview{ContextData: map[string]interface{}{
				"accountId":      "account-real",
				"ComputedLeadId": "lead-late",
			}}, &tc.meta, &models.UserSession{})
			if err == nil {
				t.Fatal("discoverFlowAtomicParentPrelocks() accepted an ambiguous bare string")
			}
			if len(query.specs) != 0 {
				t.Fatalf("query calls = %d, want zero before bare-string rejection", len(query.specs))
			}
		})
	}
}

func TestFlowAtomicParentPrelocksRejectUnsupportedMapsAndPriorRecordRoots(t *testing.T) {
	tests := []struct {
		name         string
		value        interface{}
		sourceFilter bool
	}{
		{
			name:  "arbitrary nested map",
			value: map[string]interface{}{"rhs": map[string]interface{}{"elementReference": "leadId"}},
		},
		{
			name:  "extra wrapper key",
			value: map[string]interface{}{"elementReference": "leadId", "stringValue": "lead-literal"},
		},
		{
			name:  "nested map in typed record id",
			value: map[string]interface{}{"stringValue": map[string]interface{}{"rhs": "account-1"}},
		},
		{
			name:         "slice in typed source filter",
			value:        map[string]interface{}{"stringValue": []interface{}{"lead-1"}},
			sourceFilter: true,
		},
		{
			name:  "prior record on create-capable flow",
			value: map[string]interface{}{"elementReference": "$Record__Old.AccountId"},
		},
		{
			name:         "prior record source filter on create-capable flow",
			value:        map[string]interface{}{"elementReference": "$Record__Old.Id"},
			sourceFilter: true,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			prelock := models.FlowAtomicParentPrelock{
				Name: "InvalidValue", ParentObject: "Account", RecordIDs: []interface{}{tc.value},
			}
			if tc.sourceFilter {
				prelock.RecordIDs = nil
				prelock.SourceObject = "Account"
				prelock.ParentReferenceField = "Id"
				prelock.Filters = []models.FlowRecordFilter{{Field: "Id", Operator: "EqualTo", Value: tc.value}}
			}
			meta := models.FlowMetadata{
				RunAtomically:        true,
				Start:                &models.FlowStart{Object: "Lead", TriggerType: "RecordBeforeSave"},
				Variables:            []models.FlowVariable{{Name: "leadId", DataType: "String", IsInput: true}},
				AtomicParentPrelocks: []models.FlowAtomicParentPrelock{prelock},
			}
			if err := validateFlowAtomicParentPrelocks(&meta); err == nil {
				t.Fatal("validateFlowAtomicParentPrelocks() accepted an unsupported value shape")
			}

			query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
			executor := &FlowExecutor{query: query}
			_, err := executor.discoverFlowAtomicParentPrelocks(context.Background(), &models.FlowInterview{ContextData: map[string]interface{}{
				"leadId":       "lead-1",
				"$Record__Old": nil,
			}}, &meta, &models.UserSession{})
			if err == nil {
				t.Fatal("discoverFlowAtomicParentPrelocks() accepted an unsupported value shape")
			}
			if len(query.specs) != 0 {
				t.Fatalf("query calls = %d, want zero before value-shape rejection", len(query.specs))
			}
		})
	}
}

func TestFlowAtomicParentPrelockResolvedValueContainsMapUsesNeutralRecordCarrier(t *testing.T) {
	if !flowAtomicParentPrelockResolvedValueContainsMap(models.Record{"Id": "record-1"}) {
		t.Fatal("neutral record carrier was not rejected as a map value")
	}
	if flowAtomicParentPrelockResolvedValueContainsMap("record-1") {
		t.Fatal("scalar record identifier was rejected as a map value")
	}
}

func TestValidateFlowAtomicParentPrelocksRejectsPostStartReference(t *testing.T) {
	base := models.FlowMetadata{
		RunAtomically: true,
		Formulas:      []models.FlowFormula{{Name: "ComputedLeadId", DataType: "String", Expression: "leadId"}},
		Variables:     []models.FlowVariable{{Name: "leadId", DataType: "String", IsInput: true}},
	}
	filterMeta := base
	filterMeta.AtomicParentPrelocks = []models.FlowAtomicParentPrelock{{
		Name:                 "LateFormulaFilter",
		ParentObject:         "Campaign",
		SourceObject:         "CampaignMember",
		ParentReferenceField: "CampaignId",
		Filters: []models.FlowRecordFilter{{
			Field:    "LeadId",
			Operator: "EqualTo",
			Value:    map[string]interface{}{"elementReference": "ComputedLeadId"},
		}},
	}}
	if err := validateFlowAtomicParentPrelocks(&filterMeta); err == nil {
		t.Fatal("validateFlowAtomicParentPrelocks() accepted a formula-backed filter reference")
	}

	recordIDMeta := base
	recordIDMeta.AtomicParentPrelocks = []models.FlowAtomicParentPrelock{{
		Name:         "LateFormulaRecordID",
		ParentObject: "Campaign",
		RecordIDs: []interface{}{
			map[string]interface{}{"elementReference": "ComputedLeadId"},
		},
	}}
	if err := validateFlowAtomicParentPrelocks(&recordIDMeta); err == nil {
		t.Fatal("validateFlowAtomicParentPrelocks() accepted a formula-backed recordIds reference")
	}
}

func TestValidateFlowAtomicParentPrelocksChecksCompleteFilterLogic(t *testing.T) {
	base := models.FlowMetadata{
		RunAtomically: true,
		Variables:     []models.FlowVariable{{Name: "leadId", DataType: "String", IsInput: true}},
	}
	filters := []models.FlowRecordFilter{
		{Field: "LeadId", Operator: "EqualTo", Value: map[string]interface{}{"elementReference": "leadId"}},
		{Field: "Status", Operator: "EqualTo", Value: map[string]interface{}{"stringValue": "Sent"}},
		{Field: "IsDeleted", Operator: "EqualTo", Value: map[string]interface{}{"booleanValue": false}},
	}
	for _, logic := range []string{"1 99", "1 AND99", "1 AND (2 OR 3", "1 AND 2"} {
		meta := base
		meta.AtomicParentPrelocks = []models.FlowAtomicParentPrelock{{
			Name:                 "InvalidLogic",
			ParentObject:         "Campaign",
			SourceObject:         "CampaignMember",
			ParentReferenceField: "CampaignId",
			FilterLogic:          logic,
			Filters:              filters,
		}}
		if err := validateFlowAtomicParentPrelocks(&meta); err == nil {
			t.Fatalf("validateFlowAtomicParentPrelocks() accepted filterLogic %q", logic)
		}
	}

	valid := base
	valid.AtomicParentPrelocks = []models.FlowAtomicParentPrelock{{
		Name:                 "ValidLogic",
		ParentObject:         "Campaign",
		SourceObject:         "CampaignMember",
		ParentReferenceField: "CampaignId",
		FilterLogic:          "1 AND (2 OR 3)",
		Filters:              filters,
	}}
	if err := validateFlowAtomicParentPrelocks(&valid); err != nil {
		t.Fatalf("validateFlowAtomicParentPrelocks() rejected valid filterLogic: %v", err)
	}
}

func TestValidateFlowAtomicParentPrelocksRejectsUnsupportedMembershipOperators(t *testing.T) {
	for _, operator := range []string{"In", "NotIn", "NOT IN"} {
		meta := models.FlowMetadata{
			RunAtomically: true,
			Variables:     []models.FlowVariable{{Name: "leadIds", DataType: "String", IsCollection: true, IsInput: true}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:                 "UnsupportedMembership",
				ParentObject:         "Campaign",
				SourceObject:         "CampaignMember",
				ParentReferenceField: "CampaignId",
				Filters: []models.FlowRecordFilter{{
					Field:    "LeadId",
					Operator: operator,
					Value:    map[string]interface{}{"elementReference": "leadIds"},
				}},
			}},
		}

		if err := validateFlowAtomicParentPrelocks(&meta); err == nil {
			t.Fatalf("validateFlowAtomicParentPrelocks() accepted unsupported operator %q", operator)
		}
	}
}