package sfcrmappservice

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"testing"

	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
)

func flowActionsTestComponent(
	t *testing.T,
	kind string,
	name string,
	behavior string,
) catalogsemantic.AppMetadataComponent {
	t.Helper()
	component := catalogsemantic.AppMetadataComponent{
		ComponentKind: kind, ComponentFormatVersion: 1,
		NamespaceID: AppID, BusinessKey: name, DefinitionVersion: "7",
		CanonicalBehavior: []byte(behavior),
	}
	component.ComponentID = appauthority.ComponentID(
		component.NamespaceID, component.ComponentKind, component.BusinessKey,
	)
	var err error
	component.ContentDigest, err = appauthority.ComponentDigest(
		component.ComponentKind,
		component.ComponentFormatVersion,
		component.CanonicalBehavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	return component
}

func flowActionsTestView(
	t *testing.T,
	components ...catalogsemantic.AppMetadataComponent,
) catalogsemantic.AppMetadataView {
	t.Helper()
	view := validSearchScopeView(t, 17, 3)
	view.Components = append(view.Components, components...)
	slices.SortFunc(
		view.Components,
		func(left, right catalogsemantic.AppMetadataComponent) int {
			return strings.Compare(left.ComponentID, right.ComponentID)
		},
	)
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewDigest(view)
	if err := catalogsemantic.ValidateAppMetadataView(view); err != nil {
		t.Fatalf("Flow AppMetadataView is invalid: %v", err)
	}
	return view
}

func flowActionsTestV3View(
	t *testing.T,
	components ...catalogsemantic.AppMetadataComponent,
) catalogsemantic.AppMetadataViewV3 {
	t.Helper()
	appView := flowActionsTestView(t, components...)
	view := catalogsemantic.AppMetadataViewV3{
		OrgID: appView.OrgID, AppInstallationID: appView.AppInstallationID,
		AppID: appView.AppID, AppReleaseDigest: appView.AppReleaseDigest,
		Lineage: appView.Lineage, AppComponents: appView.Components,
		Receipts: appView.Receipts,
	}
	for _, component := range components {
		if component.ComponentKind != flowComponentKind {
			continue
		}
		behavior, err := catalogsemantic.EncodeEffectiveStandardFlowDocument(
			component.DefinitionVersion,
			component.CanonicalBehavior,
		)
		if err != nil {
			t.Fatal(err)
		}
		view.EffectiveComponents = append(
			view.EffectiveComponents,
			catalogsemantic.EffectiveMetadataComponent{
				ComponentKind:          catalogsemantic.EffectiveStandardFlowComponentKind,
				ComponentFormatVersion: 1, BusinessKey: component.BusinessKey,
				ContentDigest:     catalogsemantic.ComputeEffectiveMetadataComponentDigest(behavior),
				CanonicalBehavior: behavior,
			},
		)
	}
	slices.SortFunc(view.EffectiveComponents, func(left, right catalogsemantic.EffectiveMetadataComponent) int {
		if order := strings.Compare(left.ComponentKind, right.ComponentKind); order != 0 {
			return order
		}
		return strings.Compare(left.BusinessKey, right.BusinessKey)
	})
	view.Receipts.CapabilityVersion = catalogsemantic.AppMetadataV3CapabilityName
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(view)
	if err := catalogsemantic.ValidateAppMetadataViewV3(view); err != nil {
		t.Fatalf("Flow AppMetadataViewV3 is invalid: %v", err)
	}
	return view
}

func flowActionsPermissions(granted bool) permissionsemantic.Response {
	response := validPermissionResponse(17, 3)
	response.SystemPermissions = append(
		response.SystemPermissions,
		permissionsemantic.PermissionGrant{
			PermissionID: flowRunPermission,
			Granted:      granted,
		},
	)
	slices.SortFunc(
		response.SystemPermissions,
		func(left, right permissionsemantic.PermissionGrant) int {
			return strings.Compare(left.PermissionID, right.PermissionID)
		},
	)
	return response
}

func flowActionsTestService(
	t *testing.T,
	view catalogsemantic.AppMetadataViewV3,
	permissions permissionsemantic.Response,
	execute RuntimeExecuteFlowFunc,
) *Service {
	t.Helper()
	legacyView := catalogsemantic.AppMetadataView{
		OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
		AppID: view.AppID, AppReleaseDigest: view.AppReleaseDigest,
		Lineage: view.Lineage, Components: view.AppComponents,
		Receipts: view.Receipts,
	}
	legacyView.Receipts.CapabilityVersion = catalogsemantic.AppMetadataCapabilityName
	legacyView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewDigest(legacyView)
	config := permissionMetadataServiceConfig(
		func(
			_ context.Context,
			_ string,
			request permissionsemantic.Request,
		) (permissionsemantic.Response, error) {
			if request.MinCatalogRevision == nil ||
				*request.MinCatalogRevision != view.Receipts.CatalogRevision ||
				request.Selection == nil ||
				!slices.Equal(request.Selection.Sections, []permissionsemantic.Section{
					permissionsemantic.SectionSystemPermissions,
				}) || len(request.Selection.ResourceTypes) != 0 {
				t.Fatalf("Flow Actions permission request=%#v", request)
			}
			return permissions, nil
		},
		func(
			context.Context,
			catalogsemantic.GetRequest,
		) (catalogsemantic.AppMetadataView, error) {
			return legacyView, nil
		},
	)
	config.RuntimeGetAppMetadataViewV3 = func(
		context.Context,
		catalogsemantic.GetRequest,
	) (catalogsemantic.AppMetadataViewV3, error) {
		return view, nil
	}
	config.RuntimeExecuteFlow = execute
	service, err := NewService(config)
	if err != nil {
		t.Fatal(err)
	}
	return service
}

func flowActionsTestRequest(method, target, body string) *http.Request {
	request := httptest.NewRequest(method, target, strings.NewReader(body))
	request.Header.Set("Authorization", "Bearer user-secret")
	return request
}

func flowBehavior(
	label string,
	runAtomically bool,
	trigger string,
	subflow string,
) string {
	document := map[string]any{
		"label": label, "description": "Description for " + label,
		"status": "Active", "runAtomically": runAtomically,
		"variables": []map[string]any{
			{"name": "recordId", "dataType": "String", "isInput": true},
			{"name": "result", "dataType": "String", "isOutput": true},
		},
		"start": map[string]any{},
	}
	if trigger != "" {
		document["start"] = map[string]any{"triggerType": trigger}
	}
	if subflow != "" {
		document["subflows"] = []map[string]any{{"flowName": subflow}}
	}
	encoded, err := json.Marshal(document)
	if err != nil {
		panic(err)
	}
	return string(encoded)
}

func TestFlowActionsListUsesOnlyAuthorizedBoundedManualDefinitions(t *testing.T) {
	view := flowActionsTestV3View(
		t,
		flowActionsTestComponent(t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", "HelperFlow")),
		flowActionsTestComponent(t, flowComponentKind, "HelperFlow", flowBehavior("Helper", true, "", "")),
		flowActionsTestComponent(t, flowComponentKind, "TimedFlow", flowBehavior("Timed", true, "Scheduled", "")),
		flowActionsTestComponent(t, flowComponentKind, "NonAtomicFlow", flowBehavior("Non Atomic", false, "", "")),
		flowActionsTestComponent(t, flowComponentKind, "ActionFlow", flowBehavior("Action", true, "", "")),
		flowActionsTestComponent(t, invocableActionKind, "actions", `[{"flowDeveloperName":"ActionFlow"}]`),
	)
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			t.Fatal("list reached ExecuteFlow")
			return flowsemantic.ResponseV2{}, nil
		},
	)
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodGet,
		"/services/data/v63.0/actions/custom/flow",
		"",
	))
	if recorder.Code != http.StatusOK {
		t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
	}
	var response struct {
		Actions []flowActionWireDefinition `json:"actions"`
	}
	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
		t.Fatal(err)
	}
	if len(response.Actions) != 1 || response.Actions[0].Name != "ManualFlow" ||
		response.Actions[0].URL != "/services/data/v63.0/actions/custom/flow/ManualFlow" {
		t.Fatalf("actions = %#v", response.Actions)
	}
}

func TestFlowActionGetReturnsExactVersionSchema(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			t.Fatal("get reached ExecuteFlow")
			return flowsemantic.ResponseV2{}, nil
		},
	)
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodGet,
		"/services/data/v62.0/actions/custom/flow/ManualFlow",
		"",
	))
	if recorder.Code != http.StatusOK ||
		!strings.Contains(recorder.Body.String(), `"name":"recordId"`) ||
		!strings.Contains(recorder.Body.String(), `"name":"result"`) ||
		!strings.Contains(recorder.Body.String(), `/services/data/v62.0/actions/custom/flow/ManualFlow`) {
		t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
	}
}

func TestFlowActionExecutePinsEffectiveDefinitionAuthorityAndReplayIdentity(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	effectiveBehavior, err := catalogsemantic.EncodeEffectiveStandardFlowDocument(
		"9",
		[]byte(flowBehavior("Effective Manual", true, "", "")),
	)
	if err != nil {
		t.Fatal(err)
	}
	view.EffectiveComponents[0].CanonicalBehavior = effectiveBehavior
	view.EffectiveComponents[0].ContentDigest =
		catalogsemantic.ComputeEffectiveMetadataComponentDigest(effectiveBehavior)
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(view)
	if err := catalogsemantic.ValidateAppMetadataViewV3(view); err != nil {
		t.Fatal(err)
	}
	var (
		mu       sync.Mutex
		requests []flowsemantic.Request
	)
	execute := func(
		_ context.Context,
		bearer string,
		request flowsemantic.Request,
	) (flowsemantic.ResponseV2, error) {
		if bearer != "user-secret" {
			t.Fatalf("bearer = %q", bearer)
		}
		mu.Lock()
		requests = append(requests, request)
		mu.Unlock()
		return flowsemantic.ResponseV2{
			ExecutionID:      "execution-1",
			CanonicalOutputs: []byte(`{"big":9007199254740993,"result":"ok"}`),
			Receipts: commonsemantic.Receipts{
				ProtocolVersion: "1.0", CapabilityVersion: flowsemantic.CapabilityName,
				CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9,
				SecurityEpoch: 3,
			},
		}, nil
	}
	for range 2 {
		service := flowActionsTestService(t, view, flowActionsPermissions(true), execute)
		request := flowActionsTestRequest(
			http.MethodPost,
			"/services/data/v63.0/actions/custom/flow/ManualFlow",
			`{"inputs":[{"recordId":"record-1","big":9007199254740993}]}`,
		)
		request.Header.Set(crmMutationIdempotencyHeader, "stable-flow-invocation")
		recorder := httptest.NewRecorder()
		service.ServeHTTP(recorder, request)
		if recorder.Code != http.StatusOK ||
			!strings.Contains(recorder.Body.String(), `"big":9007199254740993`) ||
			!strings.Contains(recorder.Body.String(), `"isSuccess":true`) {
			t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
		}
	}
	mu.Lock()
	defer mu.Unlock()
	if len(requests) != 2 || requests[0].DefinitionRef != "ManualFlow" ||
		requests[0].DefinitionVersion != "9" ||
		requests[0].MinCatalogRevision == nil || *requests[0].MinCatalogRevision != 17 ||
		string(requests[0].CanonicalInputs) != `{"big":9007199254740993,"recordId":"record-1"}` ||
		requests[0].IdempotencyKey != requests[1].IdempotencyKey {
		t.Fatalf("requests = %#v", requests)
	}
}

func TestFlowActionExecuteCanonicalizesHTMLInputs(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	var got flowsemantic.Request
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(_ context.Context, _ string, request flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			got = request
			return flowsemantic.ResponseV2{
				ExecutionID:      "html-input-execution",
				CanonicalOutputs: []byte(`{"result":"ok"}`),
				Receipts: commonsemantic.Receipts{
					ProtocolVersion: "1.0", CapabilityVersion: flowsemantic.CapabilityName,
					CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9,
					SecurityEpoch: 3,
				},
			}, nil
		},
	)
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodPost,
		"/services/data/v63.0/actions/custom/flow/ManualFlow",
		`{"inputs":[{"htmlBody":"<p>Hello &amp; team</p>","recordId":"record-1"}]}`,
	))
	if recorder.Code != http.StatusOK ||
		!strings.Contains(recorder.Body.String(), `"isSuccess":true`) ||
		string(got.CanonicalInputs) != `{"htmlBody":"<p>Hello &amp; team</p>","recordId":"record-1"}` {
		t.Fatalf("status=%d request=%#v body=%s", recorder.Code, got, recorder.Body.String())
	}
}

func TestFlowActionPresentsRuntimeApprovedErrors(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			return flowsemantic.ResponseV2{
				ExecutionID: "manual-validation",
				Errors: []flowsemantic.ExecutionError{
					{
						Code:    "FIELD_CUSTOM_VALIDATION_EXCEPTION",
						Message: "The record does not satisfy the policy.",
						Fields:  []string{"Type"},
					},
					{
						Code:    "FIELD_WRITE_DENIED",
						Message: "The source field is not writable.",
						Fields:  []string{"AccountSource"},
					},
				},
				Receipts: commonsemantic.Receipts{
					ProtocolVersion: "1.0", CapabilityVersion: flowsemantic.CapabilityName,
					CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9,
					SecurityEpoch: 3,
				},
			}, nil
		},
	)
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodPost,
		"/services/data/v63.0/actions/custom/flow/ManualFlow",
		`{"inputs":[{"recordId":"record-1"}]}`,
	))
	var results []flowActionExecutionResult
	if recorder.Code != http.StatusOK || json.Unmarshal(recorder.Body.Bytes(), &results) != nil ||
		len(results) != 1 || results[0].IsSuccess || results[0].ActionName != "ManualFlow" ||
		len(results[0].Errors) != 2 ||
		results[0].Errors[0].StatusCode != "FIELD_CUSTOM_VALIDATION_EXCEPTION" ||
		results[0].Errors[0].Message != "The record does not satisfy the policy." ||
		!slices.Equal(results[0].Errors[0].Fields, []string{"Type"}) ||
		results[0].Errors[1].StatusCode != "INVALID_FIELD_FOR_INSERT_UPDATE" ||
		results[0].Errors[1].Message != "The source field is not writable." ||
		!slices.Equal(results[0].Errors[1].Fields, []string{"AccountSource"}) {
		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
	}
}

func TestFlowActionStatusCodeProjectsOnlySharedNeutralRuntimeCodes(t *testing.T) {
	tests := map[string]string{
		"FIELD_WRITE_DENIED":                "INVALID_FIELD_FOR_INSERT_UPDATE",
		"invalid_request":                   "invalid_request",
		"field_write_denied":                "field_write_denied",
		"VALIDATION_ERROR":                  "VALIDATION_ERROR",
		"FIELD_CUSTOM_VALIDATION_EXCEPTION": "FIELD_CUSTOM_VALIDATION_EXCEPTION",
		"PACK_AUTHORED_POLICY_ERROR":        "PACK_AUTHORED_POLICY_ERROR",
	}
	for input, want := range tests {
		if got := flowActionStatusCode(input); got != want {
			t.Errorf("flowActionStatusCode(%q) = %q, want %q", input, got, want)
		}
	}
}

func TestStandardActionExecutionStatusCodeProjectsNeutralReasons(t *testing.T) {
	tests := map[string]string{
		"invalid_request":            "INVALID_REQUEST",
		"required_field_missing":     "REQUIRED_FIELD_MISSING",
		"invalid_field":              "INVALID_FIELD",
		"insufficient_permissions":   "INSUFFICIENT_PERMISSIONS",
		"unknown_exception":          "UNKNOWN_EXCEPTION",
		"FIELD_WRITE_DENIED":         "INVALID_FIELD_FOR_INSERT_UPDATE",
		"field_write_denied":         "field_write_denied",
		"PACK_AUTHORED_POLICY_ERROR": "PACK_AUTHORED_POLICY_ERROR",
	}
	for input, want := range tests {
		if got := standardActionExecutionStatusCode(input); got != want {
			t.Errorf("standardActionExecutionStatusCode(%q) = %q, want %q", input, got, want)
		}
	}
}

func TestFlowActionFailsClosedOnPermissionAndMixedRevision(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	t.Run("permission", func(t *testing.T) {
		service := flowActionsTestService(
			t, view, flowActionsPermissions(false),
			func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
				t.Fatal("denied request reached ExecuteFlow")
				return flowsemantic.ResponseV2{}, nil
			},
		)
		recorder := httptest.NewRecorder()
		service.ServeHTTP(recorder, flowActionsTestRequest(
			http.MethodGet,
			"/services/data/v63.0/actions/custom/flow",
			"",
		))
		if recorder.Code != http.StatusForbidden {
			t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
		}
	})
	t.Run("mixed revision", func(t *testing.T) {
		service := flowActionsTestService(
			t, view, flowActionsPermissions(true),
			func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
				return flowsemantic.ResponseV2{
					ExecutionID: "execution-1", CanonicalOutputs: []byte(`{"result":"wrong"}`),
					Receipts: commonsemantic.Receipts{
						ProtocolVersion: "1.0", CapabilityVersion: flowsemantic.CapabilityName,
						CatalogRevision: 18, SchemaGeneration: 8, EngineGeneration: 9,
						SecurityEpoch: 3,
					},
				}, nil
			},
		)
		request := flowActionsTestRequest(
			http.MethodPost,
			"/services/data/v63.0/actions/custom/flow/ManualFlow",
			`{"inputs":[{"recordId":"record-1"}]}`,
		)
		request.Header.Set(crmMutationIdempotencyHeader, "mixed-revision")
		recorder := httptest.NewRecorder()
		service.ServeHTTP(recorder, request)
		if recorder.Code != http.StatusOK ||
			!strings.Contains(recorder.Body.String(), `"isSuccess":false`) ||
			strings.Contains(recorder.Body.String(), `"wrong"`) {
			t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
		}
	})
}

func TestFlowActionRejectsNonAtomicDefinitionBeforeRuntime(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "NonAtomicFlow", flowBehavior("Non Atomic", false, "", ""),
	))
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			t.Fatal("non-atomic request reached ExecuteFlow")
			return flowsemantic.ResponseV2{}, nil
		},
	)
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodPost,
		"/services/data/v63.0/actions/custom/flow/NonAtomicFlow",
		`{"inputs":[{}]}`,
	))
	if recorder.Code != http.StatusNotFound {
		t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
	}
}

func TestFlowActionsFailClosedOnNullInvocableActionMetadata(t *testing.T) {
	view := flowActionsTestV3View(
		t,
		flowActionsTestComponent(
			t, flowComponentKind, "ProtectedFlow", flowBehavior("Protected", true, "", ""),
		),
		flowActionsTestComponent(t, invocableActionKind, "actions", `null`),
	)
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			t.Fatal("malformed protected-action metadata reached ExecuteFlow")
			return flowsemantic.ResponseV2{}, nil
		},
	)
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodGet,
		"/services/data/v63.0/actions/custom/flow",
		"",
	))
	if recorder.Code != http.StatusServiceUnavailable ||
		strings.Contains(recorder.Body.String(), "ProtectedFlow") {
		t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
	}
}

func TestFlowActionsDecodePublishedStandardFlowSource(t *testing.T) {
	raw, err := os.ReadFile(filepath.Join(
		"..", "..", "salesforce-crm-standard", "flows", "Std_ConvertLead.flow.json",
	))
	if err != nil {
		t.Fatal(err)
	}
	component := flowActionsTestComponent(
		t, flowComponentKind, "Std_ConvertLead", string(raw),
	)
	definitions, err := flowDefinitionsFromView(flowActionsTestV3View(t, component))
	if err != nil {
		t.Fatal(err)
	}
	definition, ok := definitions["Std_ConvertLead"]
	if !ok || definition.component.DefinitionVersion != "7" ||
		definition.label != "Standard Convert Lead" || len(definition.variables) != 13 {
		t.Fatalf("published Flow definition = %#v", definition)
	}
}

func TestFlowActionRejectsOversizedInputBatchBeforeRuntime(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(context.Context, string, flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			t.Fatal("oversized input batch reached ExecuteFlow")
			return flowsemantic.ResponseV2{}, nil
		},
	)
	inputs := make([]map[string]any, flowActionMaxInputs+1)
	for index := range inputs {
		inputs[index] = map[string]any{"recordId": index}
	}
	body, err := json.Marshal(flowActionExecutionRequest{Inputs: inputs})
	if err != nil {
		t.Fatal(err)
	}
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, flowActionsTestRequest(
		http.MethodPost,
		"/services/data/v63.0/actions/custom/flow/ManualFlow",
		string(body),
	))
	if recorder.Code != http.StatusBadRequest ||
		!strings.Contains(recorder.Body.String(), `"errorCode":"LIMIT_EXCEEDED"`) ||
		!strings.Contains(recorder.Body.String(), "Maximum 200 Flow inputs allowed") {
		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
	}
}

func TestFlowActionPropagatesCanceledRequestContext(t *testing.T) {
	view := flowActionsTestV3View(t, flowActionsTestComponent(
		t, flowComponentKind, "ManualFlow", flowBehavior("Manual", true, "", ""),
	))
	service := flowActionsTestService(
		t, view, flowActionsPermissions(true),
		func(ctx context.Context, _ string, _ flowsemantic.Request) (flowsemantic.ResponseV2, error) {
			t.Fatal("canceled request reached ExecuteFlow")
			return flowsemantic.ResponseV2{}, nil
		},
	)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	request := flowActionsTestRequest(
		http.MethodPost,
		"/services/data/v63.0/actions/custom/flow/ManualFlow",
		`{"inputs":[{"recordId":"record-1"}]}`,
	).WithContext(ctx)
	request.Header.Set(crmMutationIdempotencyHeader, "canceled-flow")
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, request)
	if recorder.Code != http.StatusOK ||
		!strings.Contains(recorder.Body.String(), `"isSuccess":false`) ||
		!strings.Contains(recorder.Body.String(), "Flow execution canceled") {
		t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String())
	}
}