package automation

import (
	"context"
	"fmt"
	"testing"

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

func identityDependencyAction(recordID string) *models.FlowActionCall {
	return &models.FlowActionCall{
		Name:       "WaitForLeadCommit",
		ActionType: "setIdentityConvergenceDependency",
		InputParameters: []models.FlowActionParameter{
			{Name: "objectApiName", Value: "Lead"},
			{Name: "recordId", Value: recordID},
			{Name: "destinationSlotId", Value: map[string]interface{}{"stringValue": "person"}},
		},
	}
}

func identityDependencyAuthorizedContext(flowName string) context.Context {
	const owner = "generic-connector"
	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, owner)
	ctx = withFlowRuntimeContext(ctx, flowRuntimeContextFromCtx(ctx).WithCurrentFlow(FlowExecutionIdentity{
		APIName:       flowName,
		DefinitionID:  flowName + "-definition",
		VersionID:     flowName + "-version",
		OwnerPack:     owner,
		VersionSource: flowSourceStandard,
	}))
	return constants.WithConnectorInboundFlowExecution(ctx, flowName, owner)
}

func identityDependencyOwnedAttemptContext(ctx context.Context) (context.Context, *flowAttemptDependencyCollector) {
	ctx = context.WithValue(ctx, fanoutTxKey{}, &fanoutTx{})
	ctx = withAtomicFlowBoundary(ctx)
	ctx = withAtomicFlowPessimisticRetryOwner(ctx)
	return withFlowAttemptDependencyCollector(ctx)
}

func TestSetIdentityConvergenceDependencyIsAuthorizedTerminalControlFlow(t *testing.T) {
	ctx, collector := identityDependencyOwnedAttemptContext(identityDependencyAuthorizedContext("Inbound"))
	next, err := (&FlowExecutor{persistence: &transactionalFanoutPersistence{}}).
		executeSetIdentityConvergenceDependency(ctx, &models.FlowInterview{ContextData: map[string]interface{}{}}, &models.FlowMetadata{RunAtomically: true}, identityDependencyAction("00Q000000000001AAA"), nil)
	if !isFlowDependencySuspension(err) {
		t.Fatalf("error = %v, want dependency suspension", err)
	}
	if next != "" {
		t.Fatalf("next = %q, want terminal action", next)
	}
	if collector.dependency == nil ||
		collector.dependency.Kind != contractmodels.FlowDependencyKindIdentityConvergence ||
		collector.dependency.ObjectAPIName != "Lead" ||
		collector.dependency.RecordID != "00Q000000000001AAA" ||
		collector.dependency.ConnectorOwnerKey != "generic-connector" ||
		collector.dependency.DestinationSlotID != "person" {
		t.Fatalf("dependency = %#v, want exact neutral selector", collector.dependency)
	}

	unauthorized, unauthorizedCollector := withFlowAttemptDependencyCollector(context.Background())
	if _, err := (&FlowExecutor{}).executeSetIdentityConvergenceDependency(
		unauthorized,
		&models.FlowInterview{ContextData: map[string]interface{}{}},
		&models.FlowMetadata{RunAtomically: true},
		identityDependencyAction("00Q000000000001AAA"),
		nil,
	); err == nil {
		t.Fatal("unauthorized action error = nil")
	}
	if unauthorizedCollector.dependency != nil {
		t.Fatalf("unauthorized dependency = %#v", unauthorizedCollector.dependency)
	}

	nonAtomicRoot := identityDependencyAuthorizedContext("Inbound")
	nonAtomicRoot = context.WithValue(nonAtomicRoot, fanoutTxKey{}, &fanoutTx{})
	nonAtomicRoot = withAtomicFlowBoundary(nonAtomicRoot)
	nonAtomicRoot = withAtomicFlowPessimisticRetryOwner(nonAtomicRoot)
	nonAtomicRoot, nonAtomicCollector := withFlowAttemptDependencyCollector(nonAtomicRoot)
	if _, err := (&FlowExecutor{persistence: &transactionalFanoutPersistence{}}).
		executeSetIdentityConvergenceDependency(nonAtomicRoot, &models.FlowInterview{ContextData: map[string]interface{}{}}, &models.FlowMetadata{}, identityDependencyAction("00Q000000000001AAA"), nil); err == nil || isFlowDependencySuspension(err) {
		t.Fatalf("non-atomic root error = %v, want fail-closed rejection", err)
	}
	if nonAtomicCollector.dependency != nil {
		t.Fatalf("non-atomic root dependency = %#v", nonAtomicCollector.dependency)
	}

	child := identityDependencyAuthorizedContext("Parent")
	child = withConnectorInboundCompanionSubflowDelegation(child, "Child")
	child = withFlowRuntimeContext(child, flowRuntimeContextFromCtx(child).WithCurrentFlow(FlowExecutionIdentity{
		APIName:       "Child",
		DefinitionID:  "Child-definition",
		VersionID:     "Child-version",
		OwnerPack:     "generic-connector",
		VersionSource: flowSourceStandard,
	}))
	child, childCollector := identityDependencyOwnedAttemptContext(child)
	if _, err := (&FlowExecutor{persistence: &transactionalFanoutPersistence{}}).
		executeSetIdentityConvergenceDependency(child, &models.FlowInterview{ContextData: map[string]interface{}{}}, &models.FlowMetadata{RunAtomically: true}, identityDependencyAction("00Q000000000001AAA"), nil); err == nil || isFlowDependencySuspension(err) {
		t.Fatalf("child Flow error = %v, want direct-root rejection", err)
	}
	if childCollector.dependency != nil {
		t.Fatalf("child Flow dependency = %#v", childCollector.dependency)
	}
}

type identityDependencyOutcomePersistence struct {
	*transactionalFanoutPersistence
	forcedOutcome ports.TransactionCompletionOutcome
}

func (p *identityDependencyOutcomePersistence) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	err := p.transactionalFanoutPersistence.RunInPessimisticTransaction(ctx, fn)
	if err == nil {
		return ports.TransactionCompletionCommitted, nil
	}
	if p.forcedOutcome != "" {
		return p.forcedOutcome, fmt.Errorf("synthetic transaction completion failure: %w", err)
	}
	return ports.TransactionCompletionRolledBack, err
}

func identityDependencyExecutor(persistence PersistenceProvider) *FlowExecutor {
	action := identityDependencyAction("00Q000000000002AAA")
	action.Name = "SuspendUntilLeadExists"
	metadata := syntheticFanoutFlow(true)
	metadata.RecordCreates = metadata.RecordCreates[:1]
	metadata.RecordCreates[0].Connector = &models.FlowConnector{TargetReference: action.Name}
	metadata.ActionCalls = []models.FlowActionCall{*action}

	executor := newFlowExecutorWithPlatformActionCalls(
		nil,
		nil,
		NewFlowInterviewService(persistence, nil),
		persistence,
		nil,
		nil,
		nil,
	)
	executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:                      "Inbound-version",
			DefinitionId:            "Inbound-definition",
			ProcessType:             "AutoLaunchedFlow",
			Source:                  flowSourceStandard,
			DefinitionOwnerPack:     "generic-connector",
			DefinitionRuntimeLoaded: true,
			Metadata:                metadata,
		}, nil
	}
	return executor
}

func TestIdentityDependencyPublishesOnlyAfterOwnedRollback(t *testing.T) {
	store := &identityDependencyOutcomePersistence{transactionalFanoutPersistence: &transactionalFanoutPersistence{}}
	outcome, err := identityDependencyExecutor(store).ExecuteFlowOutcome(
		WithTransientFlowInterviews(identityDependencyAuthorizedContext("Inbound")),
		"Inbound",
		nil,
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome: %v", err)
	}
	if outcome == nil || outcome.Result == nil || !outcome.Result.IsSuccess || !outcome.Result.IsRetriable || outcome.Dependency == nil {
		t.Fatalf("outcome = %#v, want successful retriable dependency", outcome)
	}
	if len(store.committed) != 0 {
		t.Fatalf("committed effects = %#v, want rollback before replayable wait", store.committed)
	}
}

func TestIdentityDependencyFailsClosedWithoutRollbackProof(t *testing.T) {
	store := &identityDependencyOutcomePersistence{
		transactionalFanoutPersistence: &transactionalFanoutPersistence{},
		forcedOutcome:                  ports.TransactionCompletionRollbackError,
	}
	outcome, err := identityDependencyExecutor(store).ExecuteFlowOutcome(
		WithTransientFlowInterviews(identityDependencyAuthorizedContext("Inbound")),
		"Inbound",
		nil,
		newPreviewUser(),
	)
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome transport error = %v", err)
	}
	if outcome == nil || outcome.Result == nil || outcome.Result.IsSuccess || outcome.Dependency != nil {
		t.Fatalf("outcome = %#v, want dependency-free failure", outcome)
	}
}