package automation

import (
	"context"
	"errors"
	"testing"

	"github.com/nexuscrm/shared/pkg/models"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// atomicSpyPersistence embeds MockPersistence to count RunInTransaction
// invocations and optionally force fn errors back out. The base
// MockPersistence.RunInTransaction returns fn(ctx) verbatim; the spy keeps
// that behavior so fn return values propagate like the real persistence
// layer would after a rollback.
type atomicSpyPersistence struct {
	MockPersistence
	runInTxCalls int
	lastFnErr    error
}

var _ PersistenceProvider = (*atomicSpyPersistence)(nil)

func (p *atomicSpyPersistence) RunInTransaction(ctx context.Context, fn func(context.Context) error) error {
	p.runInTxCalls++
	err := fn(ctx)
	p.lastFnErr = err
	return err
}

func TestRunPossiblyAtomic_NonAtomicSkipsTransaction(t *testing.T) {
	spy := &atomicSpyPersistence{}
	e := &FlowExecutor{persistence: spy}

	called := 0
	err := e.runPossiblyAtomic(context.Background(), false, func(ctx context.Context) error {
		called++
		return nil
	})

	assert.NoError(t, err)
	assert.Equal(t, 1, called)
	assert.Equal(t, 0, spy.runInTxCalls)
}

func TestRunPossiblyAtomic_AtomicWrapsInTransaction(t *testing.T) {
	spy := &atomicSpyPersistence{}
	e := &FlowExecutor{persistence: spy}

	called := 0
	err := e.runPossiblyAtomic(context.Background(), true, func(ctx context.Context) error {
		called++
		return nil
	})

	assert.NoError(t, err)
	assert.Equal(t, 1, called)
	assert.Equal(t, 1, spy.runInTxCalls)
}

func TestRunPossiblyAtomic_AtomicPropagatesErrorForRollback(t *testing.T) {
	// Load-bearing invariant for Lead Convert: fn errors must propagate out
	// unchanged so the persistence layer rolls back.
	spy := &atomicSpyPersistence{}
	e := &FlowExecutor{persistence: spy}
	boom := errors.New("opportunity insert failed")

	err := e.runPossiblyAtomic(context.Background(), true, func(ctx context.Context) error {
		return boom
	})

	assert.ErrorIs(t, err, boom)
	assert.Equal(t, 1, spy.runInTxCalls)
	assert.ErrorIs(t, spy.lastFnErr, boom)
}

func TestValidateAtomicFlow_RejectsAtomicWithWaits(t *testing.T) {
	meta := &models.FlowMetadata{
		RunAtomically: true,
		Waits:         []models.FlowWait{{Name: "Wait30d"}},
	}
	err := validateAtomicFlow(meta)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "Wait")
	assert.Contains(t, err.Error(), "transaction")
}

func TestValidateAtomicFlow_PermitsAtomicWithoutWaits(t *testing.T) {
	meta := &models.FlowMetadata{RunAtomically: true}
	assert.NoError(t, validateAtomicFlow(meta))
}

func TestValidateAtomicFlow_PermitsNonAtomicWithWaits(t *testing.T) {
	// Non-atomic is the existing path — Waits remain fully supported.
	meta := &models.FlowMetadata{
		RunAtomically: false,
		Waits:         []models.FlowWait{{Name: "Wait30d"}},
	}
	assert.NoError(t, validateAtomicFlow(meta))
}

func TestValidateAtomicFlow_NilSafe(t *testing.T) {
	assert.NoError(t, validateAtomicFlow(nil))
}

func TestValidateAtomicFlow_RejectsAtomicWithEmailAlertActionCall(t *testing.T) {
	// Synchronous SMTP send inside an open tx would hold DB locks across
	// external I/O — same class of hazard as Waits.
	meta := &models.FlowMetadata{
		RunAtomically: true,
		ActionCalls: []models.FlowActionCall{
			{Name: "SendWelcome", ActionType: "emailAlert"},
		},
	}
	err := validateAtomicFlow(meta)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "emailAlert")
	assert.Contains(t, err.Error(), "transaction")
}

func TestValidateAtomicFlow_PermitsAtomicWithNonEmailActionCalls(t *testing.T) {
	meta := &models.FlowMetadata{
		RunAtomically: true,
		ActionCalls: []models.FlowActionCall{
			{Name: "InvokeSubflow", ActionType: "flow"},
		},
	}
	assert.NoError(t, validateAtomicFlow(meta))
}

func TestExecuteFlowVersion_AtomicWithWaitsRejectedBeforeInterview(t *testing.T) {
	// Rejection must happen before interview creation and before any tx
	// opens, and must return a non-success FlowExecutionResult (not a Go
	// error) so the action-API response shape is preserved.
	spy := &atomicSpyPersistence{}
	e := &FlowExecutor{
		persistence: spy,
		interviews:  &FlowInterviewService{},
	}

	flowVersion := &models.FlowVersion{
		Id: "fv-atomic-wait",
		Metadata: models.FlowMetadata{
			RunAtomically: true,
			Waits:         []models.FlowWait{{Name: "PauseForever"}},
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "shouldNotRun"},
			},
		},
	}

	// Use a transient ctx so the executor doesn't try to persist the
	// interview row. The rejection must happen before interview creation
	// regardless, but this keeps the test independent of the persistence
	// port for the interview service.
	ctx := WithTransientFlowInterviews(context.Background())

	result, err := e.executeFlowVersion(ctx, flowVersion, "AtomicWaitFlow", nil, nil)
	require.NoError(t, err)
	require.NotNil(t, result)
	assert.False(t, result.IsSuccess)
	require.Len(t, result.Errors, 1)
	assert.Equal(t, statusCodeInvalidFlow, result.Errors[0].StatusCode)
	assert.Contains(t, result.Errors[0].Message, "Wait")
	assert.Equal(t, 0, spy.runInTxCalls, "rejection must happen before the transaction opens")
}

func TestExecuteFlowVersion_InheritedAtomicContextRejectsNonAtomicWaits(t *testing.T) {
	// A child flow invoked under an atomic parent still runs inside the
	// parent's open transaction. Even if the child metadata itself omits
	// RunAtomically, Wait/emailAlert must still be rejected at execution time.
	spy := &atomicSpyPersistence{}
	e := &FlowExecutor{
		persistence: spy,
		interviews:  &FlowInterviewService{},
	}

	flowVersion := &models.FlowVersion{
		Id: "fv-child-wait",
		Metadata: models.FlowMetadata{
			RunAtomically: false,
			Waits:         []models.FlowWait{{Name: "PauseChild"}},
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "shouldNotRun"},
			},
		},
	}

	ctx := withAtomicFlowBoundary(WithTransientFlowInterviews(context.Background()))
	result, err := e.executeFlowVersion(ctx, flowVersion, "ChildWaitFlow", nil, nil)
	require.NoError(t, err)
	require.NotNil(t, result)
	assert.False(t, result.IsSuccess)
	require.Len(t, result.Errors, 1)
	assert.Equal(t, statusCodeInvalidFlow, result.Errors[0].StatusCode)
	assert.Contains(t, result.Errors[0].Message, "Wait")
	assert.Equal(t, 0, spy.runInTxCalls, "effective-atomic rejection must happen before opening any child transaction")
}