package metadata

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/nexuscrm/automation/flowauthoring"
	"github.com/nexuscrm/automation/triggerprojection"
	"github.com/nexuscrm/automation/triggersub"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/metadata/programchecker"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
)

const flowSourceCustom = "custom"

// GetFlowDefinitions returns all flow definitions from cache
func (ms *MetadataService) GetFlowDefinitions(ctx context.Context) []*models.FlowDefinition {
	if err := ms.ensureCacheInitialized(ctx); err != nil {
		log.Printf("\u26a0\ufe0f Failed to initialize cache in GetFlowDefinitions: %v", err)
		return []*models.FlowDefinition{}
	}
	snap := ms.loadCache()
	if snap == nil {
		return []*models.FlowDefinition{}
	}
	return snap.flowDefinitions
}

// GetFlowDefinition returns a specific flow definition by ID
func (ms *MetadataService) GetFlowDefinition(ctx context.Context, id string) *models.FlowDefinition {
	if ms.UsesRequestBoundRuntimeView() {
		view, err := ms.requireRequestBoundRuntimeMutationView(ctx, "Flow definition execution metadata")
		if err != nil {
			return nil
		}
		definition, _ := view.FlowDefinition(id)
		return definition
	}
	if err := ms.ensureCacheInitialized(ctx); err != nil {
		log.Printf("\u26a0\ufe0f Failed to initialize cache in GetFlowDefinition: %v", err)
		return nil
	}
	snap := ms.loadCache()
	if snap == nil {
		return nil
	}
	return snap.flowDefinitionMap[id]
}

// GetFlowDefinitionByName returns a flow definition by developer name
func (ms *MetadataService) GetFlowDefinitionByName(ctx context.Context, developerName string) *models.FlowDefinition {
	if err := ms.ensureCacheInitialized(ctx); err != nil {
		log.Printf("\u26a0\ufe0f Failed to initialize cache in GetFlowDefinitionByName: %v", err)
		return nil
	}
	snap := ms.loadCache()
	if snap == nil {
		return nil
	}
	for _, def := range snap.flowDefinitions {
		if def.DeveloperName == developerName {
			return def
		}
	}
	return nil
}

// GetScheduledFlowDefinitions returns all flow definitions with trigger_type = "Scheduled"
func (ms *MetadataService) GetScheduledFlowDefinitions(ctx context.Context) []*models.FlowDefinition {
	if ms.UsesRequestBoundRuntimeView() {
		definitions, err := ms.GetRuntimeScheduledFlowDefinitions(ctx)
		if err != nil {
			return []*models.FlowDefinition{}
		}
		return definitions
	}
	if err := ms.ensureCacheInitialized(ctx); err != nil {
		log.Printf("\u26a0\ufe0f Failed to initialize cache in GetScheduledFlowDefinitions: %v", err)
		return []*models.FlowDefinition{}
	}

	snap := ms.loadCache()
	if snap == nil {
		return []*models.FlowDefinition{}
	}

	// Filter for scheduled flows from cache
	var scheduled []*models.FlowDefinition
	for _, def := range snap.flowDefinitions {
		if def.TriggerType == "Scheduled" {
			scheduled = append(scheduled, def)
		}
	}
	return scheduled
}

// GetRuntimeScheduledFlowDefinitions enumerates scheduled roots from the exact
// request-bound immutable Runtime view. Unlike the legacy slice-only API, it
// preserves missing or wrong-profile authority as an error so a Worker cannot
// advance its discovery cursor as though the Catalog contained no schedules.
func (ms *MetadataService) GetRuntimeScheduledFlowDefinitions(
	ctx context.Context,
) ([]*models.FlowDefinition, error) {
	view, err := ms.requireRequestBoundRuntimeView(
		ctx,
		catalogsnapshot.ProjectionRecordMutationV1,
		"scheduled Flow discovery metadata",
	)
	if err != nil {
		return nil, err
	}
	return view.ScheduledFlowDefinitions(), nil
}

// GetFlowVersionsByIDs retrieves multiple flow versions by IDs in a single
// query. Runtime projection maintenance must see repository errors so
// install/apply gates fail closed instead of silently losing subscriptions.
func (ms *MetadataService) GetFlowVersionsByIDs(ctx context.Context, ids []string) (map[string]*models.FlowVersion, error) {
	if len(ids) == 0 {
		return nil, nil
	}
	if ms.UsesRequestBoundRuntimeView() {
		view, err := ms.requireRequestBoundRuntimeMutationView(ctx, "Flow version execution metadata")
		if err != nil {
			return nil, err
		}
		result := view.FlowVersions(ids)
		if len(result) != len(ids) {
			return nil, fmt.Errorf("request-bound Runtime metadata view is missing one or more flow versions")
		}
		return result, nil
	}
	result, err := ms.repo.GetFlowVersionsByIDs(ctx, ids)
	if err != nil {
		return nil, err
	}
	return result, nil
}

// GetFlowVersionById retrieves a flow version by ID from the database
// Note: This is not cached, as flow versions contain large Metadata JSON
func (ms *MetadataService) GetFlowVersionById(ctx context.Context, versionId string) *models.FlowVersion {
	if versionId == "" {
		return nil
	}
	if ms.UsesRequestBoundRuntimeView() {
		view, ok := ms.runtimeView(ctx)
		if !ok || view == nil {
			return nil
		}
		version, _ := view.FlowVersion(versionId)
		return version
	}

	version, err := ms.repo.GetFlowVersion(ctx, versionId)
	if err != nil {
		log.Printf("\u26a0\ufe0f Failed to get flow version %s: %v", versionId, err)
		return nil
	}
	return version
}

// GetRuntimeActiveFlowVersion resolves subflow execution from the same
// request-bound revision as its parent mutation. Independent Runtime callers
// never query FlowDefinition/Flow rows behind the view.
func (ms *MetadataService) GetRuntimeActiveFlowVersion(ctx context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
	view, err := ms.requireRequestBoundRuntimeMutationView(ctx, "active Flow execution metadata")
	if err != nil {
		return nil, err
	}
	version, _ := view.ActiveFlowVersion(apiName)
	return version, nil
}

// Flow Definition CRUD (Tooling API)

// GetFlowDefinitionById returns a flow definition by ID (alias for GetFlowDefinition)
func (ms *MetadataService) GetFlowDefinitionById(ctx context.Context, id string) *models.FlowDefinition {
	return ms.GetFlowDefinition(ctx, id)
}

// CreateFlowDefinition creates a new flow definition.
func (ms *MetadataService) CreateFlowDefinition(ctx context.Context, def *models.FlowDefinition) error {
	if def.Id == "" {
		def.Id = ms.generateStandardSObjectID(ctx, "FlowDefinition")
	}
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		if err := validateFlowDefinitionCreateShadowAuthority(def); err != nil {
			return err
		}
		admitAdminOwnedFlowDefinition(def)
		if err := ms.repo.CreateFlowDefinition(ctx, def); err != nil {
			return err
		}
		return nil
	})
}

// CreateFlowDefinitionForLifecycle persists FlowDefinition projection fields for
// the Flow-version lifecycle path without reopening direct Tooling CRUD writes.
func (ms *MetadataService) CreateFlowDefinitionForLifecycle(ctx context.Context, def *models.FlowDefinition) error {
	if def == nil {
		return nil
	}
	if def.Id == "" {
		def.Id = ms.generateStandardSObjectID(ctx, "FlowDefinition")
	}
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		admitAdminOwnedFlowDefinition(def)
		return ms.repo.CreateFlowDefinition(ctx, def)
	})
}

// UpdateFlowDefinition applies a sparse Tooling PATCH to an existing
// FlowDefinition and marks the row customized in the same transaction so
// future apply passes preserve admin ownership of the definition.
func (ms *MetadataService) UpdateFlowDefinition(ctx context.Context, id string, patch *models.FlowDefinitionPatch) error {
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		if patch == nil || patch.Empty() {
			return nil
		}
		if err := validateFlowDefinitionPatchShadowAuthority(patch); err != nil {
			return err
		}
		existing, err := ms.repo.GetFlowDefinition(ctx, id)
		if err != nil {
			return err
		}
		if existing == nil {
			return apperrors.NewNotFoundError("FlowDefinition", id)
		}
		if err := admitExistingAdminFlowDefinitionWrite(existing); err != nil {
			return err
		}
		if err := ms.repo.PatchFlowDefinition(ctx, id, patch); err != nil {
			return err
		}
		return ms.markFlowDefinitionCustomized(ctx, id)
	})
}

// ApplyFlowDefinitionPatch executes a dialect-resolved sparse patch intent
// against the authoritative FlowDefinition row in the tooling write transaction.
func (ms *MetadataService) ApplyFlowDefinitionPatch(ctx context.Context, id string, patch contract.FlowAdminDefinitionPatchIntent) error {
	if err := validateFlowDefinitionPatchIntent(patch); err != nil {
		return err
	}
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		existing, err := ms.repo.GetFlowDefinition(ctx, id)
		if err != nil {
			return err
		}
		if existing == nil {
			return apperrors.NewNotFoundError("FlowDefinition", id)
		}
		if err := admitExistingAdminFlowDefinitionWrite(existing); err != nil {
			return err
		}

		switch patch.Action {
		case contract.FlowAdminDefinitionPatchActionActivateVersion:
			versions, err := ms.repo.GetFlowVersions(ctx, id)
			if err != nil {
				return err
			}
			for _, version := range versions {
				if version.VersionNumber == patch.ActiveVersionNumber {
					return ms.ActivateFlowVersion(ctx, version.Id)
				}
			}
			return apperrors.NewNotFoundWireError(
				"Flow version not found for the specified version number")
		case contract.FlowAdminDefinitionPatchActionDeactivate:
			if strings.TrimSpace(existing.ActiveVersionId) == "" {
				return nil
			}
			return ms.DeactivateFlowVersion(ctx, existing.ActiveVersionId)
		case contract.FlowAdminDefinitionPatchActionUpdate:
			defPatch := &models.FlowDefinitionPatch{
				DeveloperName: patch.Name,
				MasterLabel:   patch.Label,
				Description:   patch.Description,
			}
			if defPatch.Empty() {
				return apperrors.NewInvalidFieldError("No supported FlowDefinition fields to update", "FlowDefinition")
			}
			if err := ms.repo.PatchFlowDefinition(ctx, id, defPatch); err != nil {
				return err
			}
			return ms.markFlowDefinitionCustomized(ctx, id)
		default:
			return apperrors.NewInvalidFieldError("Unsupported FlowDefinition patch action", "Action")
		}
	})
}

func validateFlowDefinitionPatchIntent(patch contract.FlowAdminDefinitionPatchIntent) error {
	carriesUpdate := patch.Name != nil || patch.Label != nil || patch.Description != nil
	carriesActiveVersion := patch.ActiveVersionNumber != 0

	switch patch.Action {
	case contract.FlowAdminDefinitionPatchActionUpdate:
		if carriesActiveVersion {
			return apperrors.NewInvalidFieldError("activeVersionNumber cannot be mixed with FlowDefinition field updates", "ActiveVersionNumber")
		}
		if !carriesUpdate {
			return apperrors.NewInvalidFieldError("No supported FlowDefinition fields to update", "FlowDefinition")
		}
	case contract.FlowAdminDefinitionPatchActionActivateVersion:
		if carriesUpdate {
			return apperrors.NewInvalidFieldError("activate_version cannot be mixed with FlowDefinition field updates", "FlowDefinition")
		}
		if patch.ActiveVersionNumber <= 0 {
			return apperrors.NewInvalidFieldError("activeVersionNumber must be positive", "ActiveVersionNumber")
		}
	case contract.FlowAdminDefinitionPatchActionDeactivate:
		if carriesUpdate || carriesActiveVersion {
			return apperrors.NewInvalidFieldError("deactivate cannot be mixed with FlowDefinition field updates or activeVersionNumber", "FlowDefinition")
		}
	default:
		return apperrors.NewInvalidFieldError("Unsupported FlowDefinition patch action", "Action")
	}
	return nil
}

// WithToolingWriteTx exposes the tooling write transaction boundary so handlers
// can wrap multi-step user intents in one transaction.
func (ms *MetadataService) WithToolingWriteTx(ctx context.Context, fn func(ctx context.Context) error) error {
	return ms.inToolingWriteTx(ctx, fn)
}

func (ms *MetadataService) inToolingWriteTx(ctx context.Context, fn func(ctx context.Context) error) error {
	if ms.txManager == nil {
		err := fn(ctx)
		if err == nil {
			ms.InvalidateCache()
		}
		return err
	}
	if ms.txManager.IsTransactionActive(ctx) {
		return fn(ctx)
	}
	err := ms.txManager.WithPessimisticTransaction(ctx, fn)
	if err == nil {
		ms.InvalidateCache()
	}
	return err
}

func (ms *MetadataService) markFlowDefinitionCustomized(ctx context.Context, definitionID string) error {
	if definitionID == "" {
		return nil
	}
	return ms.repo.MarkFlowDefinitionCustomized(ctx, definitionID)
}

func admitAdminOwnedFlowDefinition(def *models.FlowDefinition) {
	if def == nil {
		return
	}
	def.OwnerPack = ""
	def.IsCustomized = true
}

func admitExistingAdminFlowDefinitionWrite(def *models.FlowDefinition) error {
	if def == nil {
		return nil
	}
	ownerPack := strings.TrimSpace(def.OwnerPack)
	if def.IsCustomized {
		return nil
	}
	name := strings.TrimSpace(def.DeveloperName)
	if name == "" {
		name = strings.TrimSpace(def.Id)
	}
	if ownerPack == "" {
		return sharedErrors.NewInvalidOperationError(
			fmt.Sprintf("FlowDefinition %s has no admitted owner tuple; ordinary tooling writes must not mutate ambiguous flow metadata", name),
		)
	}
	return sharedErrors.NewInvalidOperationError(
		fmt.Sprintf("FlowDefinition %s is pack-owned by %q; ordinary tooling writes must not mutate pack-owned flow metadata", name, ownerPack),
	)
}

func validateFlowDefinitionPatchShadowAuthority(patch *models.FlowDefinitionPatch) error {
	if patch == nil {
		return nil
	}
	var blocked []string
	if patch.ProcessType != nil {
		blocked = append(blocked, "ProcessType")
	}
	if patch.TriggerType != nil {
		blocked = append(blocked, "TriggerType")
	}
	if patch.TriggerObjectOrEvent != nil {
		blocked = append(blocked, "TriggerObjectOrEvent")
	}
	if patch.ActiveVersionId != nil {
		blocked = append(blocked, "ActiveVersionId")
	}
	if patch.LatestVersionId != nil {
		blocked = append(blocked, "LatestVersionId")
	}
	if len(blocked) == 0 {
		return nil
	}
	return apperrors.NewFLSDeniedErrorForFields(blocked)
}

func validateFlowDefinitionCreateShadowAuthority(def *models.FlowDefinition) error {
	if def == nil {
		return nil
	}
	var blocked []string
	if strings.TrimSpace(def.ProcessType) != "" {
		blocked = append(blocked, "ProcessType")
	}
	if strings.TrimSpace(def.TriggerType) != "" {
		blocked = append(blocked, "TriggerType")
	}
	if strings.TrimSpace(def.TriggerObjectOrEvent) != "" {
		blocked = append(blocked, "TriggerObjectOrEvent")
	}
	if strings.TrimSpace(def.ActiveVersionId) != "" {
		blocked = append(blocked, "ActiveVersionId")
	}
	if strings.TrimSpace(def.LatestVersionId) != "" {
		blocked = append(blocked, "LatestVersionId")
	}
	if len(blocked) == 0 {
		return nil
	}
	return apperrors.NewFLSDeniedErrorForFields(blocked)
}

// DeleteFlowDefinition deletes a flow definition.
func (ms *MetadataService) DeleteFlowDefinition(ctx context.Context, id string) error {
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		existing, err := ms.repo.GetFlowDefinition(ctx, id)
		if err != nil {
			return err
		}
		if existing == nil {
			return apperrors.NewNotFoundError("FlowDefinition", id)
		}
		if err := admitExistingAdminFlowDefinitionWrite(existing); err != nil {
			return err
		}
		if err := ms.repo.DeleteFlowDefinition(ctx, id); err != nil {
			return err
		}
		return ms.deleteTriggerSubscriptions(ctx, id)
	})
}

// Flow Version CRUD (Tooling API)

// GetFlowVersions returns all versions for a flow definition
func (ms *MetadataService) GetFlowVersions(ctx context.Context, definitionId string) ([]*models.FlowVersion, error) {
	if definitionId == "" {
		// Return all versions if no definition ID specified
		return []*models.FlowVersion{}, nil
	}
	return ms.repo.GetFlowVersions(ctx, definitionId)
}

// CreateFlowVersion creates a new flow version, updates the parent
// FlowDefinition projection fields, and marks the definition customized in one
// transaction.
func (ms *MetadataService) CreateFlowVersion(ctx context.Context, version *models.FlowVersion) error {
	if version.Id == "" {
		version.Id = ms.generateStandardSObjectID(ctx, "Flow")
	}
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		var definition *models.FlowDefinition
		if version.DefinitionId != "" {
			if err := ms.repo.LockFlowDefinitionForUpdate(ctx, version.DefinitionId); err != nil {
				return err
			}
			var err error
			definition, err = ms.repo.GetFlowDefinition(ctx, version.DefinitionId)
			if err != nil {
				return err
			}
			if definition == nil {
				return apperrors.NewNotFoundError("FlowDefinition", version.DefinitionId)
			}
			if err := admitExistingAdminFlowDefinitionWrite(definition); err != nil {
				return err
			}
		}
		if err := ms.validateSupportedAtomicFlowAuthoring(ctx, version, definition); err != nil {
			return err
		}
		if version.VersionNumber <= 0 && version.DefinitionId != "" {
			existing, err := ms.repo.GetFlowVersionsForUpdate(ctx, version.DefinitionId)
			if err != nil {
				return fmt.Errorf("failed to query existing versions for auto-increment: %w", err)
			}
			maxVer := 0
			for _, v := range existing {
				if v.VersionNumber > maxVer {
					maxVer = v.VersionNumber
				}
			}
			version.VersionNumber = maxVer + 1
		}
		if version.VersionNumber <= 0 {
			version.VersionNumber = 1
		}

		if err := ms.repo.CreateFlowVersion(ctx, version); err != nil {
			return err
		}
		if version.DefinitionId != "" {
			if err := ms.repo.SetLatestVersionId(ctx, version.DefinitionId, version.Id); err != nil {
				return fmt.Errorf("update LatestVersionId: %w", err)
			}
			if version.Status == models.FlowStatusActive {
				if err := ms.repo.ActivateFlowVersion(ctx, version.Id); err != nil {
					return fmt.Errorf("activate new FlowVersion: %w", err)
				}
			}
			if err := ms.syncTriggerInfo(ctx, version.DefinitionId); err != nil {
				return fmt.Errorf("sync trigger info: %w", err)
			}
			if version.Status == models.FlowStatusActive {
				if err := ms.reconcileTriggerSubscriptions(ctx, version.DefinitionId); err != nil {
					return fmt.Errorf("sync trigger subscriptions: %w", err)
				}
			}
			return ms.markFlowDefinitionCustomized(ctx, version.DefinitionId)
		}
		return nil
	})
}

// UpdateFlowVersion updates a flow version and marks the parent definition
// customized in the same transaction.
func (ms *MetadataService) UpdateFlowVersion(ctx context.Context, id string, version *models.FlowVersion) error {
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		existing, err := ms.repo.GetFlowVersion(ctx, id)
		if err != nil {
			return err
		}
		definitionID := ""
		if version != nil {
			definitionID = version.DefinitionId
		}
		if definitionID == "" && existing != nil {
			definitionID = existing.DefinitionId
		}
		var definition *models.FlowDefinition
		if definitionID != "" {
			definition, err = ms.repo.GetFlowDefinition(ctx, definitionID)
			if err != nil {
				return err
			}
			if definition == nil {
				return apperrors.NewNotFoundError("FlowDefinition", definitionID)
			}
			if err := admitExistingAdminFlowDefinitionWrite(definition); err != nil {
				return err
			}
		}
		if flowVersionPatchCarriesMetadata(version) {
			if err := ms.validateSupportedAtomicFlowAuthoring(ctx, version, definition); err != nil {
				return err
			}
		}
		if err := ms.repo.UpdateFlowVersion(ctx, id, version); err != nil {
			return err
		}
		if definitionID != "" {
			if flowVersionPatchCarriesMetadata(version) {
				if err := ms.syncTriggerInfo(ctx, definitionID); err != nil {
					return fmt.Errorf("sync trigger info: %w", err)
				}
				if err := ms.reconcileTriggerSubscriptions(ctx, definitionID); err != nil {
					return fmt.Errorf("sync trigger subscriptions: %w", err)
				}
			}
			return ms.markFlowDefinitionCustomized(ctx, definitionID)
		}
		return nil
	})
}

// ApplyFlowVersionPatch executes a dialect-resolved sparse patch intent against
// the authoritative stored Flow row, preserving the transaction-local metadata
// merge and trigger validation path.
func (ms *MetadataService) ApplyFlowVersionPatch(ctx context.Context, id string, patch contract.FlowAdminVersionPatchIntent) error {
	if err := validateFlowVersionPatchIntent(patch); err != nil {
		return err
	}
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		existing, err := ms.repo.GetFlowVersion(ctx, id)
		if err != nil {
			return err
		}
		if existing == nil {
			return apperrors.NewNotFoundError("Flow", id)
		}
		var definition *models.FlowDefinition
		if strings.TrimSpace(existing.DefinitionId) != "" {
			definition, err = ms.repo.GetFlowDefinition(ctx, existing.DefinitionId)
			if err != nil {
				return err
			}
			if definition == nil {
				return apperrors.NewNotFoundError("FlowDefinition", existing.DefinitionId)
			}
			if err := admitExistingAdminFlowDefinitionWrite(definition); err != nil {
				return err
			}
		}

		switch patch.Action {
		case contract.FlowAdminVersionPatchActionActivate:
			return ms.ActivateFlowVersion(ctx, id)
		case contract.FlowAdminVersionPatchActionDeactivate:
			return ms.DeactivateFlowVersion(ctx, id)
		case contract.FlowAdminVersionPatchActionUpdate:
			metadataChanged := len(bytes.TrimSpace(patch.MetadataPatchRaw)) > 0
			if !patch.DescriptionProvided && !metadataChanged {
				return apperrors.NewInvalidFieldError("No supported Flow fields to update", "Flow")
			}
			version := cloneFlowVersion(existing)
			if patch.DescriptionProvided {
				version.Description = patch.Description
			}
			if metadataChanged {
				merged, err := mergeFlowMetadataRaw(version.MetadataRaw, patch.MetadataPatchRaw)
				if err != nil {
					return err
				}
				version.MetadataRaw = merged
				version.MetadataJSON = string(merged)
				if err := json.Unmarshal(merged, &version.Metadata); err != nil {
					return apperrors.NewInvalidFieldError("Metadata contains malformed Flow payload", "Metadata")
				}
				if err := ms.validateSupportedAtomicFlowAuthoring(ctx, version, definition); err != nil {
					return err
				}
				if err := validateFlowTriggerProjectionPatch(definition, version); err != nil {
					return err
				}
			}
			if err := ms.repo.UpdateFlowVersion(ctx, id, version); err != nil {
				return err
			}
			definitionID := existing.DefinitionId
			if definitionID == "" && version != nil {
				definitionID = version.DefinitionId
			}
			if definitionID == "" {
				return nil
			}
			if metadataChanged {
				if err := ms.syncTriggerInfo(ctx, definitionID); err != nil {
					if errors.Is(err, triggerprojection.ErrMalformed) {
						return apperrors.NewInvalidFieldError(err.Error(), "Metadata")
					}
					return fmt.Errorf("sync trigger info: %w", err)
				}
				if err := ms.reconcileTriggerSubscriptions(ctx, definitionID); err != nil {
					if errors.Is(err, triggerprojection.ErrMalformed) {
						return apperrors.NewInvalidFieldError(err.Error(), "Metadata")
					}
					return fmt.Errorf("sync trigger subscriptions: %w", err)
				}
			}
			return ms.markFlowDefinitionCustomized(ctx, definitionID)
		default:
			return apperrors.NewInvalidFieldError("Unsupported Flow patch action", "Action")
		}
	})
}

func validateFlowVersionPatchIntent(patch contract.FlowAdminVersionPatchIntent) error {
	carriesMetadata := len(bytes.TrimSpace(patch.MetadataPatchRaw)) > 0
	carriesUpdate := patch.DescriptionProvided || carriesMetadata

	switch patch.Action {
	case contract.FlowAdminVersionPatchActionUpdate:
		if !carriesUpdate {
			return apperrors.NewInvalidFieldError("No supported Flow fields to update", "Flow")
		}
	case contract.FlowAdminVersionPatchActionActivate:
		if carriesUpdate {
			return apperrors.NewInvalidFieldError("activate cannot be mixed with Flow field updates", "Flow")
		}
	case contract.FlowAdminVersionPatchActionDeactivate:
		if carriesUpdate {
			return apperrors.NewInvalidFieldError("deactivate cannot be mixed with Flow field updates", "Flow")
		}
	default:
		return apperrors.NewInvalidFieldError("Unsupported Flow patch action", "Action")
	}
	return nil
}

// DeleteFlowVersion deletes a flow version.
// Salesforce behavior: when the last version of a FlowDefinition is deleted,
// the FlowDefinition itself is automatically removed.
func (ms *MetadataService) DeleteFlowVersion(ctx context.Context, id string) error {
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		version, err := ms.repo.GetFlowVersion(ctx, id)
		if err != nil {
			return err
		}
		if version != nil && strings.TrimSpace(version.DefinitionId) != "" {
			definition, err := ms.repo.GetFlowDefinition(ctx, version.DefinitionId)
			if err != nil {
				return err
			}
			if definition == nil {
				return apperrors.NewNotFoundError("FlowDefinition", version.DefinitionId)
			}
			if err := admitExistingAdminFlowDefinitionWrite(definition); err != nil {
				return err
			}
		}
		if err := ms.repo.DeleteFlowVersion(ctx, id); err != nil {
			return err
		}
		if version == nil || version.DefinitionId == "" {
			return nil
		}
		remaining, err := ms.repo.GetFlowVersions(ctx, version.DefinitionId)
		if err != nil {
			return fmt.Errorf("check remaining versions before auto-cleanup: %w", err)
		}
		if err := ms.reconcileTriggerSubscriptions(ctx, version.DefinitionId); err != nil {
			return fmt.Errorf("sync trigger subscriptions after flow delete: %w", err)
		}
		if len(remaining) > 0 {
			return ms.markFlowDefinitionCustomized(ctx, version.DefinitionId)
		}
		if err := ms.deleteTriggerSubscriptions(ctx, version.DefinitionId); err != nil {
			return fmt.Errorf("delete trigger subscriptions for empty flow definition: %w", err)
		}
		if err := ms.repo.DeleteFlowDefinition(ctx, version.DefinitionId); err != nil {
			return fmt.Errorf("auto-delete empty FlowDefinition: %w", err)
		}
		return nil
	})
}

// ActivateFlowVersion activates a flow version and only marks the parent
// definition customized when the activated version is admin-owned
// (Source='custom').
func (ms *MetadataService) ActivateFlowVersion(ctx context.Context, id string) error {
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		version, err := ms.repo.GetFlowVersion(ctx, id)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return apperrors.NewNotFoundError("Flow", id)
			}
			return err
		}
		if version == nil {
			return apperrors.NewNotFoundError("Flow", id)
		}
		var definition *models.FlowDefinition
		if strings.TrimSpace(version.DefinitionId) != "" {
			definition, err = ms.repo.GetFlowDefinition(ctx, version.DefinitionId)
			if err != nil {
				return err
			}
			if definition == nil {
				return apperrors.NewNotFoundError("FlowDefinition", version.DefinitionId)
			}
			if err := admitExistingAdminFlowDefinitionWrite(definition); err != nil {
				return err
			}
		}
		if err := ms.validateSupportedAtomicFlowAuthoring(ctx, version, definition); err != nil {
			return err
		}
		if err := ms.repo.ActivateFlowVersion(ctx, id); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return apperrors.NewNotFoundError("Flow", id)
			}
			return err
		}
		if version != nil && version.DefinitionId != "" {
			if err := ms.syncTriggerInfo(ctx, version.DefinitionId); err != nil {
				return fmt.Errorf("sync trigger info: %w", err)
			}
			if err := ms.reconcileTriggerSubscriptions(ctx, version.DefinitionId); err != nil {
				return fmt.Errorf("sync trigger subscriptions: %w", err)
			}
			if version.Source == flowSourceCustom {
				return ms.markFlowDefinitionCustomized(ctx, version.DefinitionId)
			}
		}
		return nil
	})
}

// DeactivateFlowVersion deactivates a flow version and marks the parent
// definition customized so bootstrap preserves the intentionally inactive
// state.
func (ms *MetadataService) DeactivateFlowVersion(ctx context.Context, id string) error {
	return ms.inToolingWriteTx(ctx, func(ctx context.Context) error {
		version, err := ms.repo.GetFlowVersion(ctx, id)
		if err != nil {
			return err
		}
		if version != nil && strings.TrimSpace(version.DefinitionId) != "" {
			definition, err := ms.repo.GetFlowDefinition(ctx, version.DefinitionId)
			if err != nil {
				return err
			}
			if definition == nil {
				return apperrors.NewNotFoundError("FlowDefinition", version.DefinitionId)
			}
			if err := admitExistingAdminFlowDefinitionWrite(definition); err != nil {
				return err
			}
		}
		if err := ms.repo.DeactivateFlowVersion(ctx, id); err != nil {
			return err
		}
		if version != nil && version.DefinitionId != "" {
			if err := ms.reconcileTriggerSubscriptions(ctx, version.DefinitionId); err != nil {
				return fmt.Errorf("sync trigger subscriptions after deactivate: %w", err)
			}
			return ms.markFlowDefinitionCustomized(ctx, version.DefinitionId)
		}
		return nil
	})
}

func flowVersionPatchCarriesMetadata(version *models.FlowVersion) bool {
	return version != nil &&
		(len(version.MetadataRaw) > 0 ||
			version.Metadata.Start != nil ||
			version.Metadata.RunAtomically ||
			len(version.Metadata.Variables) > 0 ||
			len(version.Metadata.Formulas) > 0 ||
			len(version.Metadata.AtomicParentPrelocks) > 0 ||
			len(version.Metadata.AtomicHistoryPrelocks) > 0)
}

func (ms *MetadataService) ValidateFlowVersionAuthoring(ctx context.Context, version *models.FlowVersion) error {
	return ms.validateSupportedAtomicFlowAuthoring(ctx, version, nil)
}

func validateFlowTriggerProjectionPatch(definition *models.FlowDefinition, version *models.FlowVersion) error {
	if _, err := triggerprojection.Resolve(definition, version); err != nil {
		if errors.Is(err, triggerprojection.ErrMalformed) {
			return apperrors.NewInvalidFieldError(err.Error(), "Metadata")
		}
		return err
	}
	return nil
}

func (ms *MetadataService) validateSupportedAtomicFlowAuthoring(ctx context.Context, version *models.FlowVersion, definition *models.FlowDefinition) error {
	if version == nil {
		return nil
	}
	version.SerializeMetadata()
	if err := models.ValidateFlowAtomicParentPrelocks(version.Metadata); err != nil {
		return apperrors.NewInvalidFieldError(
			fmt.Sprintf("Flow Metadata: %s", err.Error()),
			"Metadata",
		)
	}
	if err := models.ValidateFlowAtomicHistoryPrelocks(version.Metadata); err != nil {
		return apperrors.NewInvalidFieldError(
			fmt.Sprintf("Flow Metadata: %s", err.Error()),
			"Metadata",
		)
	}
	if len(version.Metadata.AtomicHistoryPrelocks) > 0 && version.Metadata.Start != nil &&
		strings.EqualFold(strings.TrimSpace(version.Metadata.Start.TriggerType), "RecordBeforeSave") {
		return apperrors.NewInvalidFieldError(
			"Flow Metadata: atomicHistoryPrelocks is not supported for before-save flow execution",
			"Metadata",
		)
	}
	if len(version.Metadata.AtomicParentPrelocks) > 0 || len(version.Metadata.AtomicHistoryPrelocks) > 0 {
		diagnostics := programchecker.CheckFlowMetadata(ctx, version.Metadata, programchecker.FlowCheckOptions{
			Metadata: NewProgramMetadataReader(ms),
		})
		for _, diagnostic := range diagnostics {
			if diagnostic.IsError() && (strings.HasPrefix(diagnostic.JSONPointer, "/atomicParentPrelocks/") ||
				strings.HasPrefix(diagnostic.JSONPointer, "/atomicHistoryPrelocks/")) {
				return apperrors.NewInvalidFieldError(
					fmt.Sprintf("Flow Metadata: %s", diagnostic.Message),
					"Metadata",
				)
			}
		}
	}
	if err := flowauthoring.ValidateNoLegacyEmailAlertActionCalls(version.Metadata); err != nil {
		return apperrors.NewInvalidFieldError(
			fmt.Sprintf("Flow Metadata: %s", err.Error()),
			"Metadata",
		)
	}
	if definition == nil && strings.TrimSpace(version.DefinitionId) != "" {
		var err error
		definition, err = ms.repo.GetFlowDefinition(ctx, version.DefinitionId)
		if err != nil {
			return err
		}
	}
	if err := triggerprojection.ValidateAtomicTriggerSupport(definition, version); err != nil {
		return apperrors.NewInvalidFieldError(
			fmt.Sprintf("Flow Metadata: %s", err.Error()),
			"Metadata",
		)
	}
	formulaCatalog, err := programchecker.NewActiveFormulaCatalogReader()
	if err != nil {
		return fmt.Errorf("build Flow authoring formula catalog: %w", err)
	}
	diagnostics := programchecker.CheckFlowFormulaResources(ctx, version.Metadata, programchecker.FlowCheckOptions{
		Formula:  formulaCatalog,
		Metadata: NewProgramMetadataReader(ms),
	})
	if diagnostics.HasErrors() {
		return apperrors.NewInvalidFieldError(
			fmt.Sprintf("Flow Metadata: %s", diagnostics.Error()),
			"Metadata",
		)
	}

	// Validate filterFormula on the authoring path using the cross-object-aware
	// validator (ValidateFilterFormulaExpressionWithSchemas):
	//   Tier 1: syntax + registered-function check (does NOT reject $-globals).
	//   Tier 2: global-allowlist check (rejects $Flow, $Setup, $Label; permits
	//           $Permission, $User, $Profile, $UserRole, $Record, $Record__Old).
	//   Tier 3: schema-aware boolean + field-ref check against live object schema.
	//           Skipped when schema is nil (object not found) or ms.repo is nil.
	//           __c custom fields and __r relationship paths are tolerated (#1018, #1724).
	//   extraSchemas: full object index so cross-object paths ($Record.Owner.Name)
	//           are validated at authoring time (HIGH-2 fix, #2047 round-4).
	if version.Metadata.Start != nil {
		if expr := strings.TrimSpace(version.Metadata.Start.FilterFormula); expr != "" {
			eng := formula.NewEngine()
			// Resolve trigger schema: nil when object unknown or no repo (unit tests without DB).
			var objSchema *models.ObjectMetadata
			objectName := strings.TrimSpace(version.Metadata.Start.Object)
			if objectName != "" && ms.repo != nil {
				objSchema = ms.GetSchema(ctx, objectName)
			}
			// Build a cross-object index so __r path validation covers related objects.
			// Use GetSchemas (decorated, served from coreCache) NOT GetAllObjects
			// (raw repo): the trigger schema above comes from GetSchema (decorated),
			// so the cross-object index must be decorated too — otherwise it lacks
			// decorator-injected virtual fields such as User.Name (compound), and
			// $Record.Owner.Name would false-reject here while the trigger path
			// accepts it (the same schema-source inconsistency behind earlier rounds).
			// GetSchemas is also cache-served, avoiding a full-table scan per save.
			var extraSchemas map[string]*models.ObjectMetadata
			if ms.repo != nil {
				allObjs := ms.GetSchemas(ctx)
				extraSchemas = make(map[string]*models.ObjectMetadata, len(allObjs))
				for _, obj := range allObjs {
					key := strings.ToLower(strings.TrimSpace(obj.APIName))
					if key != "" {
						extraSchemas[key] = obj
					}
				}
			}
			if err := eng.ValidateFilterFormulaExpressionWithSchemas(expr, objSchema, extraSchemas); err != nil {
				return apperrors.NewInvalidFieldError(
					fmt.Sprintf("Flow Metadata: start.filterFormula: %s", err.Error()),
					"Metadata.Start.FilterFormula",
				)
			}
		}
	}
	if version.Metadata.Start != nil {
		if err := models.ValidateConnectorSource(version.Metadata.Start); err != nil {
			return apperrors.NewInvalidFieldError(fmt.Sprintf("Flow Metadata: %s", err.Error()), "Metadata")
		}
	}
	return nil
}

// syncTriggerInfo resolves trigger projection from the active FlowDefinition /
// FlowVersion pair and updates FlowDefinition only when the active version
// explicitly encodes a record or non-record trigger shape.
func (ms *MetadataService) syncTriggerInfo(ctx context.Context, definitionId string) error {
	def, err := ms.repo.GetFlowDefinition(ctx, definitionId)
	if err != nil {
		return fmt.Errorf("load flow definition: %w", err)
	}
	if def == nil || def.ActiveVersionId == "" {
		return nil
	}
	version, err := ms.repo.GetFlowVersion(ctx, def.ActiveVersionId)
	if err != nil {
		return fmt.Errorf("load active flow version: %w", err)
	}
	if version == nil {
		return nil
	}
	projection, err := triggerprojection.Resolve(def, version)
	if err != nil {
		return fmt.Errorf("resolve trigger projection: %w", err)
	}
	if projection.Kind == triggerprojection.KindUnknown {
		return nil
	}
	return ms.repo.SetTriggerInfo(ctx, definitionId, projection.TriggerType, projection.ObjectOrEvent)
}

// BackfillTriggerInfo is the public version of syncTriggerInfo, used during bootstrap
// to fix FlowDefinitions that are missing trigger metadata.
func (ms *MetadataService) BackfillTriggerInfo(ctx context.Context, definitionId string, triggerType string, objectOrEvent string) error {
	if err := ms.repo.SetTriggerInfo(ctx, definitionId, triggerType, objectOrEvent); err != nil {
		return err
	}
	ms.InvalidateCache()
	return nil
}

// BackfillTriggerSubscriptions rebuilds the TriggerSubscription shadow table
// from authoritative FlowDefinition + active Flow metadata.
func (ms *MetadataService) BackfillTriggerSubscriptions(ctx context.Context) error {
	exec, err := ms.triggerSubscriptionExecutor(ctx)
	if err != nil {
		return err
	}
	return triggersub.Backfill(ctx, exec)
}

func (ms *MetadataService) reconcileTriggerSubscriptions(ctx context.Context, definitionID string) error {
	exec, err := ms.triggerSubscriptionExecutor(ctx)
	if err != nil {
		return err
	}
	return triggersub.ReconcileDefinition(ctx, exec, definitionID)
}

func (ms *MetadataService) deleteTriggerSubscriptions(ctx context.Context, definitionID string) error {
	exec, err := ms.triggerSubscriptionExecutor(ctx)
	if err != nil {
		return err
	}
	return triggersub.Delete(ctx, exec, definitionID)
}

func (ms *MetadataService) triggerSubscriptionExecutor(ctx context.Context) (triggersub.Executor, error) {
	type executorProvider interface {
		GetExecutor(ctx context.Context) persistence.Executor
	}
	provider, ok := ms.repo.(executorProvider)
	if !ok {
		return nil, fmt.Errorf("metadata repository does not expose SQL executor for trigger subscriptions")
	}
	return provider.GetExecutor(ctx), nil
}

// generateStandardSObjectID generates a Salesforce-compatible 18-char record ID
// using the key prefix cache rebuilt from _System_Object.KeyPrefix metadata.
func (ms *MetadataService) generateStandardSObjectID(ctx context.Context, objectAPIName string) string {
	prefix := ""
	if resolved, ok := ms.KeyPrefixForObject(ctx, objectAPIName); ok {
		prefix = resolved
	}
	if prefix == "" && ms != nil && ms.repo != nil {
		if obj, err := ms.repo.GetSchemaByAPIName(ctx, objectAPIName); err == nil && obj != nil && obj.KeyPrefix != nil {
			prefix = *obj.KeyPrefix
		}
	}
	if prefix == "" {
		panic(fmt.Sprintf("generateStandardSObjectID: no key prefix metadata for %q", objectAPIName))
	}
	return fmt.Sprintf("%s%015d", prefix, time.Now().UnixNano()%1000000000000000)
}