package automation

import (
	"context"
	"errors"
	"fmt"
	"strings"

	"github.com/nexuscrm/automation/triggerprojection"
	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	scheduledFlowMaxClosureDepth      = 10
	scheduledFlowMaxElementExecutions = 1000
)

var ErrScheduledFlowNotEligible = errors.New("scheduled Flow is not eligible for fenced Runtime execution")

// ScheduledFlowMetadataView is the minimum immutable Catalog projection needed
// to prepare one scheduled Flow execution. Implementations must be
// content-addressed views; repository/current-version selectors do not satisfy
// this contract.
type ScheduledFlowMetadataView interface {
	FlowDefinition(string) (*models.FlowDefinition, bool)
	FlowVersion(string) (*models.FlowVersion, bool)
	ActiveFlowVersion(string) (*models.FlowVersion, bool)
}

// ScheduledFlowExecutionPlan is a complete exact closure selected from one
// immutable Runtime metadata view. Versions are keyed by developer name because
// that is the existing engine's subflow reference identity.
type ScheduledFlowExecutionPlan struct {
	Definition   *models.FlowDefinition
	Root         *models.FlowVersion
	Versions     map[string]*models.FlowVersion
	EntryElement string
	// AsyncAfterCommit is true only for the strict record-triggered
	// AsyncAfterCommit entry that was selected from the immutable root. The
	// worker uses it to install a version-bound execution marker for the two
	// connector identity actions it owns; a positive timed path must not gain
	// that authority merely because it also has an entry element.
	AsyncAfterCommit     bool
	MaxElementExecutions uint64
}

// ScheduledFlowExecutionInstaller installs the already-verified immutable
// authority for one durable scheduled execution. Preparing the Flow closure
// and installing its effect metadata are intentionally one envelope: callers
// must not execute an exact Flow plan against a worker-startup/current view.
//
// The installer is prepared before a durable row is claimed and must perform
// no I/O when it is installed immediately before the effect transaction.
type ScheduledFlowExecutionInstaller func(context.Context) (context.Context, error)

// PreparedScheduledFlowExecution keeps the exact Flow closure and the
// immutable metadata authority used by all nested effects inseparable across
// fresh and recovered execution paths.
type PreparedScheduledFlowExecution struct {
	ScheduledFlowExecutionPlan
	installExecution ScheduledFlowExecutionInstaller
}

func NewPreparedScheduledFlowExecution(
	plan ScheduledFlowExecutionPlan,
	installExecution ScheduledFlowExecutionInstaller,
) (PreparedScheduledFlowExecution, error) {
	prepared := PreparedScheduledFlowExecution{
		ScheduledFlowExecutionPlan: plan,
		installExecution:           installExecution,
	}
	if err := prepared.Validate(); err != nil {
		return PreparedScheduledFlowExecution{}, err
	}
	return prepared, nil
}

func (p PreparedScheduledFlowExecution) Validate() error {
	if err := p.ScheduledFlowExecutionPlan.Validate(); err != nil {
		return err
	}
	if p.installExecution == nil {
		return errors.New("scheduled Flow execution authority installer is required")
	}
	return nil
}

func (p PreparedScheduledFlowExecution) InstallExecution(
	ctx context.Context,
) (context.Context, error) {
	if err := p.Validate(); err != nil {
		return nil, err
	}
	executionCtx, err := p.installExecution(ctx)
	if err != nil {
		return nil, fmt.Errorf("install scheduled Flow execution authority: %w", err)
	}
	if executionCtx == nil {
		return nil, errors.New("install scheduled Flow execution authority: installer returned a nil context")
	}
	return executionCtx, nil
}

func (p ScheduledFlowExecutionPlan) Validate() error {
	if p.Definition == nil || p.Root == nil ||
		p.Definition.Id == "" || p.Definition.DeveloperName == "" ||
		p.Definition.ActiveVersionId != p.Root.Id ||
		p.Root.DefinitionId != p.Definition.Id ||
		len(p.Versions) == 0 ||
		p.MaxElementExecutions == 0 ||
		p.MaxElementExecutions > scheduledFlowMaxElementExecutions {
		return ErrScheduledFlowNotEligible
	}
	root, ok := p.Versions[strings.ToLower(p.Definition.DeveloperName)]
	if !ok || root == nil || root.Id != p.Root.Id {
		return ErrScheduledFlowNotEligible
	}
	if p.EntryElement != "" &&
		(p.Root.Metadata.Start == nil ||
			p.Root.Metadata.Start.Connector == nil ||
			p.Root.Metadata.Start.Connector.TargetReference != p.EntryElement) {
		return ErrScheduledFlowNotEligible
	}
	return nil
}

// PrepareScheduledFlowExecutionPlan selects and validates a complete
// transaction-owned scheduled closure from one already verified immutable
// Runtime view. Scheduled/timed work is already durable before execution, so
// this profile must not inherit the synchronous ExecuteFlow admission rule that
// requires authored RunAtomically=true. The worker imposes one pessimistic
// transaction and a hard element budget around the existing interpreter.
//
// The launch profile permits bounded database collection lookup/loop/write
// shapes, while rejecting waits, screens, connector-sourced input,
// unknown/external actions, caller-supplied collections, cycles, and
// revision-unbound children. Collection writes are accepted only because the
// scheduled effect entrypoint installs the engine's hard pre-effect row bound.
func PrepareScheduledFlowExecutionPlan(
	ctx context.Context,
	view ScheduledFlowMetadataView,
	definitionID string,
	versionID string,
) (ScheduledFlowExecutionPlan, error) {
	return PrepareScheduledFlowExecutionPlanForEntry(
		ctx,
		view,
		definitionID,
		versionID,
		"",
	)
}

// PrepareScheduledFlowExecutionPlanForEntry prepares either a schedule-started
// Flow (entryElement empty) or one exact timed record-triggered path. A timed
// plan preserves the immutable Flow version identity but executes a private
// copy whose Start points only at the persisted entry. It cannot re-evaluate
// today's record-start gate or silently select the first scheduled branch.
func PrepareScheduledFlowExecutionPlanForEntry(
	ctx context.Context,
	view ScheduledFlowMetadataView,
	definitionID string,
	versionID string,
	entryElement string,
) (ScheduledFlowExecutionPlan, error) {
	entryElement = strings.TrimSpace(entryElement)
	if ctx == nil || ctx.Err() != nil || view == nil ||
		strings.TrimSpace(definitionID) == "" ||
		strings.TrimSpace(versionID) == "" {
		return ScheduledFlowExecutionPlan{}, ErrScheduledFlowNotEligible
	}
	definition, ok := view.FlowDefinition(definitionID)
	if !ok || definition == nil || definition.IsDeleted ||
		definition.ActiveVersionId != versionID ||
		strings.TrimSpace(definition.DeveloperName) == "" {
		return ScheduledFlowExecutionPlan{}, ErrScheduledFlowNotEligible
	}
	root, ok := view.FlowVersion(versionID)
	if !ok || root == nil || root.IsDeleted ||
		root.DefinitionId != definition.Id ||
		!strings.EqualFold(root.Status, models.FlowStatusActive) ||
		root.Metadata.Start == nil {
		return ScheduledFlowExecutionPlan{}, ErrScheduledFlowNotEligible
	}
	asyncAfterCommit := false
	if entryElement == "" {
		if root.Metadata.Start.Schedule == nil {
			return ScheduledFlowExecutionPlan{}, ErrScheduledFlowNotEligible
		}
	} else {
		asyncAfterCommit = isStrictAsyncAfterCommitEntry(&root.Metadata, entryElement)
		preparedRoot, err := prepareTimedFlowRoot(root, entryElement)
		if err != nil {
			return ScheduledFlowExecutionPlan{}, err
		}
		root = preparedRoot
	}

	versions := make(map[string]*models.FlowVersion)
	visiting := make(map[string]bool)
	var total uint64
	var walk func(string, *models.FlowVersion, int, bool) error
	walk = func(name string, version *models.FlowVersion, depth int, isRoot bool) error {
		key := strings.ToLower(strings.TrimSpace(name))
		if key == "" || version == nil || version.IsDeleted ||
			!strings.EqualFold(version.Status, models.FlowStatusActive) ||
			depth > scheduledFlowMaxClosureDepth {
			return ErrScheduledFlowNotEligible
		}
		if visiting[key] {
			return fmt.Errorf("%w: subflow cycle", ErrScheduledFlowNotEligible)
		}
		if existing, seen := versions[key]; seen {
			if existing.Id != version.Id {
				return fmt.Errorf("%w: mixed exact child version", ErrScheduledFlowNotEligible)
			}
			return nil
		}
		if isRoot {
			if version.Metadata.Start == nil ||
				(entryElement == "" && version.Metadata.Start.Schedule == nil) ||
				(entryElement != "" &&
					(version.Metadata.Start.Connector == nil ||
						version.Metadata.Start.Connector.TargetReference != entryElement)) {
				return ErrScheduledFlowNotEligible
			}
		} else if version.Metadata.Start != nil &&
			(version.Metadata.Start.Schedule != nil ||
				version.Metadata.Start.Object != "" ||
				version.Metadata.Start.TriggerType != "" ||
				version.Metadata.Start.RecordTriggerType != "") {
			return fmt.Errorf("%w: child has an independent trigger", ErrScheduledFlowNotEligible)
		}
		if err := validateAtomicFlowForExecution(
			scheduledFlowValidationContext(ctx),
			version.ProcessType,
			&version.Metadata,
		); err != nil {
			return fmt.Errorf("%w: engine atomic validation: %v", ErrScheduledFlowNotEligible, err)
		}
		count, children, err := validateScheduledFlowLaunchShape(
			&version.Metadata,
			asyncAfterCommit && isRoot,
		)
		if err != nil {
			return err
		}
		if count == 0 || count > scheduledFlowMaxElementExecutions ||
			total > scheduledFlowMaxElementExecutions-count {
			return fmt.Errorf("%w: execution budget", ErrScheduledFlowNotEligible)
		}
		total += count
		versions[key] = version
		visiting[key] = true
		for _, childName := range children {
			child, found := view.ActiveFlowVersion(childName)
			if !found {
				return fmt.Errorf("%w: missing exact child %q", ErrScheduledFlowNotEligible, childName)
			}
			if err := walk(childName, child, depth+1, false); err != nil {
				return err
			}
		}
		visiting[key] = false
		return nil
	}
	if err := walk(definition.DeveloperName, root, 0, true); err != nil {
		return ScheduledFlowExecutionPlan{}, err
	}
	plan := ScheduledFlowExecutionPlan{
		Definition: definition, Root: root, Versions: versions,
		EntryElement: entryElement, AsyncAfterCommit: asyncAfterCommit,
		MaxElementExecutions: scheduledFlowMaxElementExecutions,
	}
	if err := plan.Validate(); err != nil {
		return ScheduledFlowExecutionPlan{}, err
	}
	return plan, nil
}

func isStrictAsyncAfterCommitEntry(metadata *models.FlowMetadata, entryElement string) bool {
	if metadata == nil || metadata.Start == nil || metadata.Start.Connector != nil ||
		strings.TrimSpace(metadata.StartElementReference) != "" ||
		len(metadata.Start.ScheduledPaths) != 1 {
		return false
	}
	for _, path := range metadata.Start.ScheduledPaths {
		if path.Connector == nil || strings.TrimSpace(path.Connector.TargetReference) != entryElement {
			continue
		}
		return strings.EqualFold(strings.TrimSpace(path.PathType), "AsyncAfterCommit") &&
			strings.TrimSpace(path.TimeSource) == "" &&
			path.OffsetNumber == 0 &&
			strings.TrimSpace(path.OffsetUnit) == ""
	}
	return false
}

// scheduledFlowValidationContext preserves caller cancellation while clearing
// record-trigger execution markers. A timed path is being prepared for a
// separate scheduled-worker transaction; validating the prepared root as if it
// were still the current record-trigger root both mixes owners and rejects the
// intentionally trigger-free prepared entry.
func scheduledFlowValidationContext(ctx context.Context) context.Context {
	return withAtomicFlowBoundary(
		withFlowRuntimeContext(ctx, NewFlowRuntimeContext(ctx, nil)),
	)
}

func prepareTimedFlowRoot(
	root *models.FlowVersion,
	entryElement string,
) (*models.FlowVersion, error) {
	if root == nil || root.Metadata.Start == nil {
		return nil, ErrScheduledFlowNotEligible
	}
	triggerType, supported := triggerprojection.CanonicalRecordTriggerType(
		root.Metadata.Start.TriggerType,
	)
	if !supported || triggerType != "RecordAfterSave" {
		return nil, ErrScheduledFlowNotEligible
	}
	matches := 0
	for _, path := range root.Metadata.Start.ScheduledPaths {
		if path.Connector != nil &&
			strings.TrimSpace(path.Connector.TargetReference) == entryElement {
			matches++
		}
	}
	if matches != 1 {
		return nil, fmt.Errorf("%w: timed entry is absent or ambiguous", ErrScheduledFlowNotEligible)
	}

	prepared := *root
	prepared.Metadata = root.Metadata
	start := *root.Metadata.Start
	start.Connector = &models.FlowConnector{TargetReference: entryElement}
	start.TriggerType = ""
	start.RecordTriggerType = ""
	start.FilterLogic = ""
	start.FilterFormula = ""
	start.Filters = nil
	start.ScheduledPaths = nil
	start.Schedule = nil
	start.ConnectorSource = nil
	prepared.Metadata.Start = &start
	prepared.Metadata.StartElementReference = ""
	return &prepared, nil
}

func validateScheduledFlowLaunchShape(
	metadata *models.FlowMetadata,
	allowAsyncAfterCommitIdentityActions bool,
) (uint64, []string, error) {
	if metadata == nil ||
		len(metadata.Waits) != 0 || len(metadata.Screens) != 0 ||
		len(metadata.AtomicParentPrelocks) != 0 ||
		(metadata.Start != nil && metadata.Start.ConnectorSource != nil) {
		return 0, nil, ErrScheduledFlowNotEligible
	}
	singleRecordReferences := make(map[string]struct{}, len(metadata.RecordLookups)*2)
	collectionReferences := make(map[string]struct{}, len(metadata.RecordLookups)*2)
	for _, variable := range metadata.Variables {
		if !variable.IsCollection {
			continue
		}
		name := strings.ToLower(strings.TrimSpace(variable.Name))
		if variable.IsInput || name == "" ||
			strings.TrimSpace(flowVariableObjectTypeFromDeclaration(variable)) == "" {
			return 0, nil, ErrScheduledFlowNotEligible
		}
		collectionReferences[name] = struct{}{}
	}
	for _, lookup := range metadata.RecordLookups {
		references := singleRecordReferences
		if !lookup.GetFirstRecordOnly {
			references = collectionReferences
		}
		references[strings.ToLower(strings.TrimSpace(lookup.Name))] = struct{}{}
		if output := strings.ToLower(strings.TrimSpace(lookup.OutputReference)); output != "" {
			references[output] = struct{}{}
		}
	}
	for _, loop := range metadata.Loops {
		if strings.TrimSpace(loop.Name) == "" {
			return 0, nil, ErrScheduledFlowNotEligible
		}
		if _, ok := collectionReferences[strings.ToLower(strings.TrimSpace(loop.CollectionReference))]; !ok {
			return 0, nil, ErrScheduledFlowNotEligible
		}
	}
	for _, update := range metadata.RecordUpdates {
		if len(update.Filters) != 0 {
			// The engine resolves at most limit+1 identities and rejects the
			// complete element before its first write when more than the hard
			// bound match. Requiring an explicit object keeps this launch proof
			// independent of mutable relationship-shape inference.
			if update.InputReference != "" || strings.TrimSpace(update.Object) == "" {
				return 0, nil, ErrScheduledFlowNotEligible
			}
			continue
		}
		if strings.TrimSpace(update.InputReference) == "" {
			return 0, nil, ErrScheduledFlowNotEligible
		}
		reference := strings.ToLower(strings.TrimSpace(update.InputReference))
		if reference == "$record" {
			continue
		}
		if _, ok := singleRecordReferences[reference]; ok {
			continue
		}
		if _, ok := collectionReferences[reference]; ok {
			continue
		}
		return 0, nil, ErrScheduledFlowNotEligible
	}
	for _, deleteElement := range metadata.RecordDeletes {
		if len(deleteElement.Filters) != 0 {
			if deleteElement.InputReference != "" || strings.TrimSpace(deleteElement.Object) == "" {
				return 0, nil, ErrScheduledFlowNotEligible
			}
			continue
		}
		if strings.TrimSpace(deleteElement.InputReference) == "" {
			return 0, nil, ErrScheduledFlowNotEligible
		}
		reference := strings.ToLower(strings.TrimSpace(deleteElement.InputReference))
		if reference == "$record" {
			continue
		}
		if _, ok := singleRecordReferences[reference]; ok {
			continue
		}
		if _, ok := collectionReferences[reference]; ok {
			continue
		}
		return 0, nil, ErrScheduledFlowNotEligible
	}
	for _, create := range metadata.RecordCreates {
		if create.ConnectorFieldMapping != nil {
			return 0, nil, ErrScheduledFlowNotEligible
		}
	}

	children := make([]string, 0, len(metadata.Subflows)+len(metadata.ActionCalls))
	for _, action := range metadata.ActionCalls {
		switch {
		case strings.EqualFold(strings.TrimSpace(action.ActionType), "flow"):
			if strings.TrimSpace(action.ActionName) == "" {
				return 0, nil, ErrScheduledFlowNotEligible
			}
			children = append(children, action.ActionName)
		case allowAsyncAfterCommitIdentityActions && metadata.RunAtomically &&
			(strings.EqualFold(strings.TrimSpace(action.ActionType), flowactions.ActionTypeResolveConnectorRemoteIdentity) ||
				strings.EqualFold(strings.TrimSpace(action.ActionType), flowactions.ActionTypeTransferConnectorRemoteIdentity)):
			// The scheduled Runtime worker wires precisely these two atomic,
			// no-provider-effect actions. They remain root-only and receive a
			// version-bound context marker at execution time.
		default:
			return 0, nil, ErrScheduledFlowNotEligible
		}
	}
	for _, subflow := range metadata.Subflows {
		if strings.TrimSpace(subflow.FlowName) == "" {
			return 0, nil, ErrScheduledFlowNotEligible
		}
		children = append(children, subflow.FlowName)
	}
	count := uint64(
		len(metadata.Decisions) +
			len(metadata.Assignments) +
			len(metadata.RecordCreates) +
			len(metadata.RecordUpdates) +
			len(metadata.RecordDeletes) +
			len(metadata.RecordLookups) +
			len(metadata.Loops) +
			len(metadata.ActionCalls) +
			len(metadata.Subflows) +
			len(metadata.CustomErrors),
	)
	return count, children, nil
}