package connectors

import (
	"context"
	"testing"

	connector_inbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/constants"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

// captureExecutor is a test stub that records the context passed to ExecuteFlow.
type captureExecutor struct {
	capturedCtx context.Context
}

func (f *captureExecutor) ExecuteFlow(ctx context.Context, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
	f.capturedCtx = ctx
	return &models.FlowExecutionResult{IsSuccess: true}, nil
}

// ExecuteFlowVersionById satisfies the versioned executor method and captures the
// context identically, so flowAdapter.ExecuteFlowVersion is exercisable in tests.
func (f *captureExecutor) ExecuteFlowVersionById(ctx context.Context, _, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
	f.capturedCtx = ctx
	return &models.FlowExecutionResult{IsSuccess: true}, nil
}

type captureOutcomeExecutor struct {
	captureExecutor
	request models.FlowExecutionDependencyRequest
}

func (f *captureOutcomeExecutor) typedOutcome() *models.FlowExecutionTypedOutcome {
	return &models.FlowExecutionTypedOutcome{
		Result:      &models.FlowExecutionResult{IsSuccess: true},
		Dependency:  &f.request,
		ExecutionID: "flow-execution-123",
		Cause: &models.FlowExecutionCause{
			Code:     "DB_LOCK_WAIT_TIMEOUT",
			Category: models.FlowExecutionCauseCategoryConcurrency,
		},
	}
}

func (f *captureOutcomeExecutor) ExecuteFlowOutcome(ctx context.Context, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionTypedOutcome, error) {
	f.capturedCtx = ctx
	return f.typedOutcome(), nil
}

func (f *captureOutcomeExecutor) ExecuteFlowVersionByIdOutcome(ctx context.Context, _, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionTypedOutcome, error) {
	f.capturedCtx = ctx
	return f.typedOutcome(), nil
}
func TestFlowAdapter_ExecuteFlow_StampsSystemContextAndConnectorInbound(t *testing.T) {
	const flowName = "Std_HubSpot_LeadEnrichment"
	const namespace = "hubspot-crm"

	stub := &captureExecutor{}
	adapter := &flowAdapter{executor: stub}

	// Simulate the worker stamping ContextKeyConnectorSyncOrigin with the pack namespace.
	inCtx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, namespace)

	_, err := adapter.ExecuteFlow(inCtx, flowName, nil, nil)
	if err != nil {
		t.Fatalf("ExecuteFlow returned unexpected error: %v", err)
	}

	// 1. SecurityContext must be SystemContext so RLS is bypassed for
	//    pack-installed FlowDefinition rows whose OwnerId is NULL.
	got := constants.GetSecurityContext(stub.capturedCtx)
	if got != constants.SystemContext {
		t.Errorf("GetSecurityContext = %v, want constants.SystemContext (%v)", got, constants.SystemContext)
	}

	// 2. ConnectorInboundFlowExecution must be stamped for the EXACT routed flow
	//    name AND the same connector namespace so authorizeProtectedStandardFlow
	//    permits execution of that one same-pack protected standard flow via the
	//    narrow inbound path.
	if !constants.ConnectorInboundExecutionAuthorizesFlow(stub.capturedCtx, flowName, namespace) {
		t.Errorf("ConnectorInboundExecutionAuthorizesFlow(%q, %q) = false, want true", flowName, namespace)
	}
	// 2a. SECURITY (round-2 regression): a DIFFERENT flow name must NOT be
	//     authorized — the marker must not grant blanket access to all
	//     pack-owned protected standard flows, only the routed root flow.
	const differentFlow = "Std_HubSpot_ContactEnrichment"
	if constants.ConnectorInboundExecutionAuthorizesFlow(stub.capturedCtx, differentFlow, namespace) {
		t.Errorf("ConnectorInboundExecutionAuthorizesFlow(%q, %q) = true for a different flow name, want false (must not authorize arbitrary descendants)", differentFlow, namespace)
	}
	// 2b. SECURITY (round-3): a DIFFERENT pack namespace must NOT be authorized —
	//     cross-pack routes (pack A routes to pack B's protected flow) must be
	//     rejected even when the flow name matches exactly.
	const otherPack = "salesforce-crm"
	if constants.ConnectorInboundExecutionAuthorizesFlow(stub.capturedCtx, flowName, otherPack) {
		t.Errorf("ConnectorInboundExecutionAuthorizesFlow(%q, %q) = true for a different namespace, want false (cross-pack isolation)", flowName, otherPack)
	}

	// 3. SECURITY: standard-action execution context is intentionally not part
	//    of the connector contract. Connector-inbound authorization is carried
	//    only by the same-pack ConnectorInboundFlowExecution marker asserted
	//    above; standard-action-only actions remain owned by the backend.
}

// TestFlowAdapter_ExecuteFlow_NoNamespace_FailSafe verifies that when the
// incoming ctx carries no ContextKeyConnectorSyncOrigin (e.g. a unit test that
// calls ExecuteFlow directly without worker setup), the connector-inbound marker
// is NOT stamped and ConnectorInboundExecutionAuthorizesFlow returns false for
// any flow — fail-safe behaviour.
func TestFlowAdapter_ExecuteFlow_NoNamespace_FailSafe(t *testing.T) {
	const flowName = "Std_HubSpot_LeadEnrichment"

	stub := &captureExecutor{}
	adapter := &flowAdapter{executor: stub}

	_, err := adapter.ExecuteFlow(context.Background(), flowName, nil, nil)
	if err != nil {
		t.Fatalf("ExecuteFlow returned unexpected error: %v", err)
	}

	// No namespace on ctx → marker not stamped → authorization returns false.
	if constants.ConnectorInboundExecutionAuthorizesFlow(stub.capturedCtx, flowName, "any-pack") {
		t.Errorf("ConnectorInboundExecutionAuthorizesFlow = true when no namespace was stamped; want false (fail-safe)")
	}
}
func TestFlowAdapter_OutcomePreservesTypedEvidenceAndDependencyAuthorizationContext(t *testing.T) {
	const flowName = "Std_HubSpot_LeadEnrichment"
	const namespace = "hubspot-crm"
	request, err := models.NewFlowIdentityConvergenceDependencyRequest("Lead", "00Q000000000001AAA", namespace, "person")
	if err != nil {
		t.Fatal(err)
	}

	for _, tc := range []struct {
		name string
		run  func(*flowAdapter, context.Context) (*connector_inbound.FlowDependencyOutcome, error)
	}{
		{
			name: "active flow",
			run: func(adapter *flowAdapter, ctx context.Context) (*connector_inbound.FlowDependencyOutcome, error) {
				return adapter.ExecuteFlowDependencyOutcome(ctx, flowName, nil, nil)
			},
		},
		{
			name: "exact companion version",
			run: func(adapter *flowAdapter, ctx context.Context) (*connector_inbound.FlowDependencyOutcome, error) {
				return adapter.ExecuteFlowVersionDependencyOutcome(ctx, flowName, "flow-version", nil, nil)
			},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			stub := &captureOutcomeExecutor{request: request}
			adapter := &flowAdapter{executor: stub}
			ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, namespace)
			outcome, err := tc.run(adapter, ctx)
			if err != nil {
				t.Fatalf("outcome call: %v", err)
			}
			if outcome == nil || outcome.Result == nil || !outcome.Result.IsSuccess || outcome.Dependency == nil || *outcome.Dependency != request {
				t.Fatalf("outcome = %#v, want exact request", outcome)
			}
			if outcome.ExecutionID != "flow-execution-123" || outcome.Cause == nil || outcome.Cause.Code != "DB_LOCK_WAIT_TIMEOUT" {
				t.Fatalf("outcome = %#v, want execution id and typed cause", outcome)
			}
			if constants.GetSecurityContext(stub.capturedCtx) != constants.SystemContext ||
				!constants.ConnectorInboundExecutionAuthorizesFlow(stub.capturedCtx, flowName, namespace) {
				t.Fatal("outcome path lost connector authorization context")
			}
		})
	}
}

func TestFlowAdapter_OutcomeFallsBackToLegacyExecutor(t *testing.T) {
	stub := &captureExecutor{}
	adapter := &flowAdapter{executor: stub}
	outcome, err := adapter.ExecuteFlowDependencyOutcome(context.Background(), "LegacyFlow", nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	if outcome == nil || outcome.Result == nil || !outcome.Result.IsSuccess || outcome.Dependency != nil || outcome.ExecutionID != "" || outcome.Cause != nil {
		t.Fatalf("legacy outcome = %#v, want successful connector-evidence-free projection", outcome)
	}
}