package automation

import (
	"context"
	"errors"
	"slices"
	"strings"
	"testing"

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

type initialHistoryPrelockCapturePersistence struct {
	*stubPersistenceCapture
	initial            []string
	planned            []string
	coordinatorActive  bool
	lockedParentRows   []recordlocks.ParentKey
	bookkeepingInitial [][]string
	bookkeepingPlanned [][]string
}

func (p *initialHistoryPrelockCapturePersistence) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	p.initial = ports.InitialHistorySequenceLocks(ctx)
	return fn(ctx)
}

func (p *initialHistoryPrelockCapturePersistence) Insert(ctx context.Context, objectName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
	if objectName == "FlowInterview" {
		p.bookkeepingInitial = append(p.bookkeepingInitial, ports.InitialHistorySequenceLocks(ctx))
		p.bookkeepingPlanned = append(p.bookkeepingPlanned, ports.PlannedHistorySequenceLocks(ctx))
	}
	return p.stubPersistenceCapture.Insert(ctx, objectName, record, user)
}

func (p *initialHistoryPrelockCapturePersistence) Update(ctx context.Context, objectName string, id string, updates models.SObject, user *models.UserSession) error {
	if objectName == "FlowInterview" {
		p.bookkeepingInitial = append(p.bookkeepingInitial, ports.InitialHistorySequenceLocks(ctx))
		p.bookkeepingPlanned = append(p.bookkeepingPlanned, ports.PlannedHistorySequenceLocks(ctx))
	}
	return p.stubPersistenceCapture.Update(ctx, objectName, id, updates, user)
}

func (p *initialHistoryPrelockCapturePersistence) RunWithParentLockCoordinator(ctx context.Context, fn func(context.Context) error) error {
	p.coordinatorActive = true
	defer func() { p.coordinatorActive = false }()
	return fn(ctx)
}

func (p *initialHistoryPrelockCapturePersistence) LockParentRows(ctx context.Context, keys []recordlocks.ParentKey) ([]recordlocks.ParentKey, error) {
	if !p.coordinatorActive {
		return nil, errors.New("missing parent-lock coordinator")
	}
	p.initial = ports.InitialHistorySequenceLocks(ctx)
	p.planned = ports.PlannedHistorySequenceLocks(ctx)
	p.lockedParentRows = append([]recordlocks.ParentKey(nil), keys...)
	return p.lockedParentRows, nil
}

func TestExecuteFlowRejectsInitialHistoryPlanWithoutAtomicBoundary(t *testing.T) {
	persistence := &stubPersistenceCapture{}
	executor := NewFlowExecutor(nil, nil, nil, persistence, nil, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	executor.fieldMetaLookup = func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{ID: "fld-account-name", APIName: "Name"}
	}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start:                 &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "Done"}},
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
			AtomicHistoryPrelocks: []models.FlowAtomicHistoryPrelock{{Object: "Account", Fields: []string{"Name"}}},
		},
	}

	result, err := executor.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()),
		version,
		"NonAtomicHistoryPlan",
		nil,
		&models.UserSession{ID: "user-1"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion(): %v", err)
	}
	if result == nil || result.IsSuccess {
		t.Fatalf("executeFlowVersion() result = %#v, want invalid flow", result)
	}
}

func TestExecuteBeforeSaveFlowRejectsInitialHistoryPlanInstalledInsideOuterTransaction(t *testing.T) {
	persistence := &initialHistoryPrelockCapturePersistence{stubPersistenceCapture: &stubPersistenceCapture{}}
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, query, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	executor.fieldMetaLookup = func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{ID: "fld-account-name", APIName: "Name"}
	}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			RunAtomically: true,
			Start:         &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "Done"}},
			Assignments:   []models.FlowAssignment{{Name: "Done"}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:         "AccountRows",
				ParentObject: "Account",
				RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "account-1"}},
			}},
			AtomicHistoryPrelocks: []models.FlowAtomicHistoryPrelock{{Object: "Account", Fields: []string{"Name"}}},
		},
	}

	result, err := executor.executeFlowVersion(
		WithBeforeSaveFlow(WithTransientFlowInterviews(context.Background())),
		version,
		"BeforeSaveHistoryPlan",
		nil,
		&models.UserSession{ID: "user-1"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion(): %v", err)
	}
	if result == nil || result.IsSuccess || len(result.Errors) == 0 || !strings.Contains(result.Errors[0].Message, "before-save") {
		t.Fatalf("executeFlowVersion() result = %#v, want before-save history-plan rejection", result)
	}
}

func TestExecuteNestedAtomicFlowRejectsInitialHistoryPlanInstalledInsideOuterTransaction(t *testing.T) {
	persistence := &initialHistoryPrelockCapturePersistence{stubPersistenceCapture: &stubPersistenceCapture{}}
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, query, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	executor.fieldMetaLookup = func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{ID: "fld-account-name", APIName: "Name"}
	}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			RunAtomically: true,
			Start:         &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "Done"}},
			Assignments:   []models.FlowAssignment{{Name: "Done"}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:         "AccountRows",
				ParentObject: "Account",
				RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "account-1"}},
			}},
			AtomicHistoryPrelocks: []models.FlowAtomicHistoryPrelock{{Object: "Account", Fields: []string{"Name"}}},
		},
	}

	result, err := executor.executeFlowVersion(
		withAtomicFlowBoundary(WithTransientFlowInterviews(context.Background())),
		version,
		"NestedHistoryPlan",
		nil,
		&models.UserSession{ID: "user-1"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion(): %v", err)
	}
	if result == nil || result.IsSuccess || len(result.Errors) == 0 || !strings.Contains(result.Errors[0].Message, "root atomic flow") {
		t.Fatalf("executeFlowVersion() result = %#v, want nested history-plan rejection", result)
	}
}

func TestExecutePreparedFlowClosureRootPlansHistoryLocksWithinOwnedTransaction(t *testing.T) {
	persistence := &initialHistoryPrelockCapturePersistence{stubPersistenceCapture: &stubPersistenceCapture{}}
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, query, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	executor.fieldMetaLookup = func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{ID: "fld-account-name", APIName: "Name"}
	}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		DefinitionRuntimeLoaded: true,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			RunAtomically: true,
			Start:         &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "Done"}},
			Assignments:   []models.FlowAssignment{{Name: "Done"}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:         "AccountRows",
				ParentObject: "Account",
				RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "account-1"}},
			}},
			AtomicHistoryPrelocks: []models.FlowAtomicHistoryPrelock{{Object: "Account", Fields: []string{"Name"}}},
		},
	}

	outcome, err := executor.ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Background(),
		"PreparedHistoryPlan",
		map[string]*models.FlowVersion{"PreparedHistoryPlan": version},
		nil,
		&models.UserSession{ID: "user-1"},
		10,
	)
	if err != nil {
		t.Fatalf("ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(): %v", err)
	}
	if outcome == nil || outcome.Result == nil || !outcome.Result.IsSuccess {
		t.Fatalf("prepared closure outcome = %#v, want success", outcome)
	}
	if !slices.Equal(persistence.initial, []string{"fld-account-name"}) {
		t.Fatalf("initial history prelocks = %v, want fld-account-name", persistence.initial)
	}
	if !slices.Equal(persistence.planned, []string{"fld-account-name"}) {
		t.Fatalf("first-attempt history prelocks = %v, want fld-account-name", persistence.planned)
	}
	if !slices.Equal(persistence.lockedParentRows, []recordlocks.ParentKey{{ObjectName: "Account", RecordID: "account-1"}}) {
		t.Fatalf("business-row prelocks = %v, want Account/account-1", persistence.lockedParentRows)
	}
}

func TestPersistentAtomicFlowDoesNotLeakHistoryPlanIntoInterviewBookkeeping(t *testing.T) {
	persistence := &initialHistoryPrelockCapturePersistence{stubPersistenceCapture: &stubPersistenceCapture{
		insertResult: models.SObject{"Id": "interview-1"},
	}}
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, query, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	executor.fieldMetaLookup = func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{ID: "fld-account-name", APIName: "Name"}
	}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			RunAtomically: true,
			Start:         &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "Done"}},
			Assignments:   []models.FlowAssignment{{Name: "Done"}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name:         "AccountRows",
				ParentObject: "Account",
				RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "account-1"}},
			}},
			AtomicHistoryPrelocks: []models.FlowAtomicHistoryPrelock{{Object: "Account", Fields: []string{"Name"}}},
		},
	}

	result, err := executor.executeFlowVersion(
		context.Background(),
		version,
		"PersistentHistoryPlan",
		nil,
		&models.UserSession{ID: "user-1"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion(): %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("executeFlowVersion() result = %#v, want success", result)
	}
	if len(persistence.bookkeepingInitial) != 2 || len(persistence.bookkeepingPlanned) != 2 {
		t.Fatalf("FlowInterview bookkeeping calls: initial=%d planned=%d, want two each", len(persistence.bookkeepingInitial), len(persistence.bookkeepingPlanned))
	}
	for i := range persistence.bookkeepingInitial {
		if len(persistence.bookkeepingInitial[i]) != 0 || len(persistence.bookkeepingPlanned[i]) != 0 {
			t.Fatalf("FlowInterview bookkeeping call %d inherited history plans: initial=%v planned=%v", i+1, persistence.bookkeepingInitial[i], persistence.bookkeepingPlanned[i])
		}
	}
}

func TestExecuteAtomicFlowPlansDeclaredHistoryLocksBeforeFirstTransactionAttempt(t *testing.T) {
	persistence := &initialHistoryPrelockCapturePersistence{stubPersistenceCapture: &stubPersistenceCapture{}}
	query := &flowParentPrelockQuery{countingQuery: &countingQuery{}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, query, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	executor.fieldMetaLookup = func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
		ids := map[string]string{
			"Account.Name":              "fld-account-name",
			"Campaign.NumberOfContacts": "fld-campaign-number-of-contacts",
		}
		if id := ids[objectAPIName+"."+fieldAPIName]; id != "" {
			return &models.FieldMetadata{ID: id, APIName: fieldAPIName}
		}
		return nil
	}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			RunAtomically: true,
			Start:         &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "Done"}},
			Assignments:   []models.FlowAssignment{{Name: "Done"}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{
				{
					Name:         "CampaignRows",
					ParentObject: "Campaign",
					RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "campaign-1"}},
				},
				{
					Name:         "AccountRows",
					ParentObject: "Account",
					RecordIDs:    []interface{}{map[string]interface{}{"stringValue": "account-1"}},
				},
			},
			AtomicHistoryPrelocks: []models.FlowAtomicHistoryPrelock{
				{Object: "Campaign", Fields: []string{"NumberOfContacts"}},
				{Object: "Account", Fields: []string{"Name"}},
			},
		},
	}

	ctx := recordlocks.WithForcedParentPrelocks(
		WithTransientFlowInterviews(context.Background()),
		[]recordlocks.ParentKey{{ObjectName: "Account", RecordID: "account-retry"}},
	)
	result, err := executor.executeFlowVersion(
		ctx,
		version,
		"AtomicHistoryPlan",
		nil,
		&models.UserSession{ID: "user-1"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion(): %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("executeFlowVersion() result = %#v, want success", result)
	}
	want := []string{"fld-account-name", "fld-campaign-number-of-contacts"}
	if !slices.Equal(persistence.initial, want) {
		t.Fatalf("initial history prelocks = %v, want %v", persistence.initial, want)
	}
	wantRows := []recordlocks.ParentKey{
		{ObjectName: "Account", RecordID: "account-1"},
		{ObjectName: "Account", RecordID: "account-retry"},
		{ObjectName: "Campaign", RecordID: "campaign-1"},
	}
	if !slices.Equal(persistence.lockedParentRows, wantRows) {
		t.Fatalf("business-row prelocks = %v, want %v", persistence.lockedParentRows, wantRows)
	}
}