package packs

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"sort"
	"strings"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/kernel/pkg/utils"
)

const tableFlowActionCallHandler = "_System_FlowActionCallHandler"

const (
	flowActionCallHandlerAtomicPolicySafe   = "safe"
	flowActionCallHandlerAtomicPolicyReject = "reject"

	flowActionCallHandlerPreviewPolicyReject                      = "reject"
	flowActionCallHandlerPreviewPolicyAllow                       = "allow"
	flowActionCallHandlerPreviewPolicyAllowWithPreviewPersistence = "allow_with_preview_persistence"
	flowActionCallHandlerPreviewPolicyRejectEmailIO               = "reject_email_io"
	flowActionCallHandlerPreviewPolicyInvokeHandler               = "invoke_handler"
	flowActionCallHandlerContextNone                              = "none"
	flowActionCallHandlerContextStandardActionProtected           = "standard_action_protected"
	flowActionCallHandlerDefaultPreviewPolicy                     = flowActionCallHandlerPreviewPolicyReject
	flowActionCallHandlerDefaultContext                           = flowActionCallHandlerContextNone
)

type sourceBackedFlowActionCallHandlerPayload struct {
	ActionType      string `json:"actionType"`
	HandlerRef      string `json:"handlerRef"`
	AtomicPolicy    string `json:"atomicPolicy,omitempty"`
	PreviewPolicy   string `json:"previewPolicy,omitempty"`
	RequiresContext string `json:"requiresContext,omitempty"`
	SendsEmail      bool   `json:"sendsEmail,omitempty"`
}

type sourceBackedFlowActionCallHandlerProjection struct {
	OwnerPack       string `json:"ownerPack"`
	ActionType      string `json:"actionType"`
	HandlerRef      string `json:"handlerRef"`
	AtomicPolicy    string `json:"atomicPolicy"`
	PreviewPolicy   string `json:"previewPolicy"`
	RequiresContext string `json:"requiresContext"`
	SendsEmail      bool   `json:"sendsEmail,omitempty"`
}

type sourceBackedFlowActionCallHandlerComponent struct {
	packName                    string
	manifestComponent           ManifestComponent
	projection                  sourceBackedFlowActionCallHandlerProjection
	payload                     []byte
	flowActionCallRegistrations []automation.FlowActionCallRegistration
}

func newSourceBackedFlowActionCallHandlerComponent(packName string, component ManifestComponent, payload []byte, flowActionCallRegistrations ...automation.FlowActionCallRegistration) (*sourceBackedFlowActionCallHandlerComponent, error) {
	projection, err := sourceBackedFlowActionCallHandlerProjectionFromPayload(packName, component, payload)
	if err != nil {
		return nil, err
	}
	return &sourceBackedFlowActionCallHandlerComponent{
		packName:                    strings.TrimSpace(packName),
		manifestComponent:           component,
		projection:                  projection,
		payload:                     append([]byte(nil), payload...),
		flowActionCallRegistrations: append([]automation.FlowActionCallRegistration(nil), flowActionCallRegistrations...),
	}, nil
}

func (c *sourceBackedFlowActionCallHandlerComponent) Kind() ComponentKind {
	return ComponentKindFlowActionCallHandler
}

func (c *sourceBackedFlowActionCallHandlerComponent) PackName() string {
	return c.packName
}

func (c *sourceBackedFlowActionCallHandlerComponent) BusinessKey() string {
	return c.projection.ActionType
}

func (c *sourceBackedFlowActionCallHandlerComponent) Manageability() Manageability {
	return c.manifestComponent.Manageability
}

func (c *sourceBackedFlowActionCallHandlerComponent) DependsOn() []ComponentReference {
	return append([]ComponentReference(nil), c.manifestComponent.DependsOn...)
}

func (c *sourceBackedFlowActionCallHandlerComponent) ExpectedPayload() ([]byte, error) {
	return append([]byte(nil), c.payload...), nil
}

func (c *sourceBackedFlowActionCallHandlerComponent) ExpectedProjection() (any, error) {
	return c.projection, nil
}

func (c *sourceBackedFlowActionCallHandlerComponent) LiveProjection(ctx context.Context, deps VerifyDeps) (any, error) {
	live, err := liveSourceBackedFlowActionCallHandlerProjection(ctx, deps.DB, c.projection.OwnerPack, c.projection.ActionType)
	if err != nil {
		return nil, err
	}
	if err := validateFlowActionCallHandlerRuntimeRegistration(live, c.flowActionCallRegistrations...); err != nil {
		return nil, err
	}
	return live, nil
}

func (c *sourceBackedFlowActionCallHandlerComponent) Apply(ctx context.Context, deps ApplyDeps) error {
	return upsertSourceBackedFlowActionCallHandler(ctx, deps.DB, c.projection)
}

func sourceBackedFlowActionCallHandlerProjectionFromPayload(packName string, component ManifestComponent, payload []byte) (sourceBackedFlowActionCallHandlerProjection, error) {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler component: pack name is required")
	}
	if component.Manageability != "" && component.Manageability != ManageabilityLocked {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler requires manageability %q, got %q", ManageabilityLocked, component.Manageability)
	}

	var def sourceBackedFlowActionCallHandlerPayload
	dec := json.NewDecoder(bytes.NewReader(payload))
	dec.DisallowUnknownFields()
	if err := dec.Decode(&def); err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("parse %s: %w", componentPayloadLabel(component), err)
	}
	if err := ensureJSONEOF(dec); err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("parse %s: %w", componentPayloadLabel(component), err)
	}

	actionType := normalizeFlowActionCallHandlerActionType(def.ActionType)
	if actionType == "" {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s: actionType is required", componentPayloadLabel(component))
	}
	if strings.Contains(actionType, "/") {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler actionType %q must not contain /", def.ActionType)
	}
	handlerRef := strings.TrimSpace(def.HandlerRef)
	if handlerRef == "" {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s: handlerRef is required", actionType)
	}
	atomicPolicy := strings.ToLower(strings.TrimSpace(def.AtomicPolicy))
	if atomicPolicy == "" {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s: atomicPolicy is required", actionType)
	}
	if !validFlowActionCallHandlerAtomicPolicy(atomicPolicy) {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s atomicPolicy %q is not supported", actionType, strings.TrimSpace(def.AtomicPolicy))
	}
	previewPolicy := normalizeFlowActionCallHandlerPolicy(def.PreviewPolicy, flowActionCallHandlerDefaultPreviewPolicy)
	if !validFlowActionCallHandlerPreviewPolicy(previewPolicy) {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s previewPolicy %q is not supported", actionType, strings.TrimSpace(def.PreviewPolicy))
	}
	requiresContext := normalizeFlowActionCallHandlerPolicy(def.RequiresContext, flowActionCallHandlerDefaultContext)
	if !validFlowActionCallHandlerContext(requiresContext) {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s requiresContext %q is not supported", actionType, strings.TrimSpace(def.RequiresContext))
	}
	if err := validateFlowActionCallHandlerBusinessKey(packName, actionType, component.BusinessKey); err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, err
	}

	return sourceBackedFlowActionCallHandlerProjection{
		OwnerPack:       packName,
		ActionType:      actionType,
		HandlerRef:      handlerRef,
		AtomicPolicy:    atomicPolicy,
		PreviewPolicy:   previewPolicy,
		RequiresContext: requiresContext,
		SendsEmail:      def.SendsEmail,
	}, nil
}

func flowActionCallRegistrationsFromSource(source *ResolvedPackSource, implementations []automation.FlowActionCallHandlerImplementation) ([]automation.FlowActionCallRegistration, error) {
	if source == nil || source.Manifest == nil || len(implementations) == 0 {
		return nil, nil
	}
	descriptors := make([]automation.FlowActionCallRegistrationDescriptor, 0)
	for _, component := range source.Manifest.Components {
		if component.Kind != ComponentKindFlowActionCallHandler {
			continue
		}
		payload, err := sourceBackedComponentPayload(source, component)
		if err != nil {
			continue
		}
		projection, err := sourceBackedFlowActionCallHandlerProjectionFromPayload(source.Manifest.Name, component, payload)
		if err != nil {
			continue
		}
		descriptors = append(descriptors, flowActionCallRegistrationDescriptorFromProjection(projection))
	}
	return automation.FlowActionCallRegistrationsFromDescriptors(descriptors, implementations...)
}

func flowActionCallRegistrationsForProjection(projection sourceBackedFlowActionCallHandlerProjection, registrations []automation.FlowActionCallRegistration, implementations []automation.FlowActionCallHandlerImplementation) ([]automation.FlowActionCallRegistration, error) {
	derived, err := automation.FlowActionCallRegistrationsFromDescriptors(
		[]automation.FlowActionCallRegistrationDescriptor{flowActionCallRegistrationDescriptorFromProjection(projection)},
		implementations...,
	)
	if err != nil {
		return nil, err
	}
	return mergeFlowActionCallRegistrationsByDescriptor(registrations, derived)
}

func mergeFlowActionCallRegistrationsByDescriptor(registrations []automation.FlowActionCallRegistration, additions []automation.FlowActionCallRegistration) ([]automation.FlowActionCallRegistration, error) {
	out := make([]automation.FlowActionCallRegistration, 0, len(registrations)+len(additions))
	indexByActionType := map[string]int{}
	for _, registration := range append(append([]automation.FlowActionCallRegistration(nil), registrations...), additions...) {
		descriptor := registration.Descriptor()
		if descriptor.ActionType == "" {
			out = append(out, registration)
			continue
		}
		if existingIndex, ok := indexByActionType[descriptor.ActionType]; ok {
			existingDescriptor := out[existingIndex].Descriptor()
			if !flowActionCallRegistrationDescriptorsEqual(existingDescriptor, descriptor) {
				return nil, fmt.Errorf("conflicting flow action-call registration for actionType %q", descriptor.ActionType)
			}
			continue
		}
		indexByActionType[descriptor.ActionType] = len(out)
		out = append(out, registration)
	}
	return out, nil
}

func flowActionCallRegistrationDescriptorsEqual(a automation.FlowActionCallRegistrationDescriptor, b automation.FlowActionCallRegistrationDescriptor) bool {
	return a.ActionType == b.ActionType &&
		a.HandlerRef == b.HandlerRef &&
		a.SendsEmail == b.SendsEmail &&
		a.AtomicPolicy == b.AtomicPolicy &&
		a.PreviewPolicy == b.PreviewPolicy &&
		a.RequiresContext == b.RequiresContext
}

func flowActionCallRegistrationDescriptorFromProjection(projection sourceBackedFlowActionCallHandlerProjection) automation.FlowActionCallRegistrationDescriptor {
	return automation.FlowActionCallRegistrationDescriptor{
		ActionType:      projection.ActionType,
		HandlerRef:      projection.HandlerRef,
		SendsEmail:      projection.SendsEmail,
		AtomicPolicy:    projection.AtomicPolicy,
		PreviewPolicy:   projection.PreviewPolicy,
		RequiresContext: projection.RequiresContext,
	}
}

// ActiveFlowActionCallHandlerDescriptors reads live app-pack handler
// declarations. Callers combine these metadata-owned descriptors with
// distribution-supplied handler implementations to build executable runtime
// registrations.
func ActiveFlowActionCallHandlerDescriptors(ctx context.Context, db *sql.DB) ([]automation.FlowActionCallRegistrationDescriptor, error) {
	if db == nil {
		return nil, fmt.Errorf("database is required")
	}
	rows, err := db.QueryContext(ctx, `
SELECT OwnerPack, ActionType, HandlerRef, AtomicPolicy, PreviewPolicy, RequiresContext, COALESCE(SendsEmail, 0)
  FROM _System_FlowActionCallHandler
 WHERE COALESCE(IsActive, 1) = 1
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY OwnerPack, ActionType`)
	if err != nil {
		return nil, fmt.Errorf("query active flow_action_call_handler rows: %w", err)
	}
	defer rows.Close()

	descriptors := make([]automation.FlowActionCallRegistrationDescriptor, 0)
	for rows.Next() {
		var projection sourceBackedFlowActionCallHandlerProjection
		if err := rows.Scan(
			&projection.OwnerPack,
			&projection.ActionType,
			&projection.HandlerRef,
			&projection.AtomicPolicy,
			&projection.PreviewPolicy,
			&projection.RequiresContext,
			&projection.SendsEmail,
		); err != nil {
			return nil, fmt.Errorf("scan active flow_action_call_handler row: %w", err)
		}
		projection, err = normalizeSourceBackedFlowActionCallHandlerProjection(projection)
		if err != nil {
			return nil, fmt.Errorf("normalize active flow_action_call_handler row: %w", err)
		}
		descriptors = append(descriptors, flowActionCallRegistrationDescriptorFromProjection(projection))
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate active flow_action_call_handler rows: %w", err)
	}
	return descriptors, nil
}

func validateFlowActionCallHandlerBusinessKey(_ string, actionType, businessKey string) error {
	rawActionType := strings.TrimSpace(businessKey)
	if rawActionType == "" {
		return fmt.Errorf("flow_action_call_handler businessKey is required and must equal normalized actionType %q", actionType)
	}
	if strings.Contains(rawActionType, "/") {
		return fmt.Errorf("flow_action_call_handler businessKey %q must be the global actionType, not owner/actionType", rawActionType)
	}
	if rawActionType != actionType {
		return fmt.Errorf("flow_action_call_handler businessKey actionType %q must be canonical and match normalized payload actionType %q", rawActionType, actionType)
	}
	return nil
}

func normalizeFlowActionCallHandlerActionType(actionType string) string {
	return strings.ToLower(strings.TrimSpace(actionType))
}

func normalizeFlowActionCallHandlerPolicy(value string, defaultValue string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	if value == "" {
		return defaultValue
	}
	return value
}

func validateFlowActionCallHandlerRuntimeRegistration(projection sourceBackedFlowActionCallHandlerProjection, appRegistrations ...automation.FlowActionCallRegistration) error {
	projection, err := normalizeSourceBackedFlowActionCallHandlerProjection(projection)
	if err != nil {
		return err
	}
	// Transitional guard: until Flow dispatch reads source-backed handler rows
	// directly, pack declarations must be an exact mirror of the executor
	// registry so startup metadata cannot drift from runtime behavior.
	registry := map[string]automation.FlowActionCallRegistrationDescriptor{}
	descriptors, err := automation.EffectiveFlowActionCallRegistrationDescriptors(appRegistrations...)
	if err != nil {
		return fmt.Errorf("flow_action_call_handler runtime registration registry is invalid: %w", err)
	}
	for _, descriptor := range descriptors {
		actionType := normalizeFlowActionCallHandlerActionType(descriptor.ActionType)
		if actionType == "" {
			continue
		}
		descriptor.ActionType = actionType
		descriptor.HandlerRef = strings.TrimSpace(descriptor.HandlerRef)
		descriptor.AtomicPolicy = strings.ToLower(strings.TrimSpace(descriptor.AtomicPolicy))
		descriptor.PreviewPolicy = normalizeFlowActionCallHandlerPolicy(descriptor.PreviewPolicy, flowActionCallHandlerDefaultPreviewPolicy)
		descriptor.RequiresContext = normalizeFlowActionCallHandlerPolicy(descriptor.RequiresContext, flowActionCallHandlerDefaultContext)
		registry[actionType] = descriptor
	}
	expected, ok := registry[projection.ActionType]
	if !ok {
		return fmt.Errorf("flow_action_call_handler %s/%s has no runtime registration", projection.OwnerPack, projection.ActionType)
	}

	var mismatches []string
	if projection.HandlerRef != expected.HandlerRef {
		mismatches = append(mismatches, fmt.Sprintf("handlerRef declared %q runtime %q", projection.HandlerRef, expected.HandlerRef))
	}
	if projection.AtomicPolicy != expected.AtomicPolicy {
		mismatches = append(mismatches, fmt.Sprintf("atomicPolicy declared %q runtime %q", projection.AtomicPolicy, expected.AtomicPolicy))
	}
	if projection.PreviewPolicy != expected.PreviewPolicy {
		mismatches = append(mismatches, fmt.Sprintf("previewPolicy declared %q runtime %q", projection.PreviewPolicy, expected.PreviewPolicy))
	}
	if projection.RequiresContext != expected.RequiresContext {
		mismatches = append(mismatches, fmt.Sprintf("requiresContext declared %q runtime %q", projection.RequiresContext, expected.RequiresContext))
	}
	if projection.SendsEmail != expected.SendsEmail {
		mismatches = append(mismatches, fmt.Sprintf("sendsEmail declared %t runtime %t", projection.SendsEmail, expected.SendsEmail))
	}
	if len(mismatches) > 0 {
		return fmt.Errorf("flow_action_call_handler %s/%s conflicts with runtime registration: %s", projection.OwnerPack, projection.ActionType, strings.Join(mismatches, "; "))
	}
	return nil
}

func validFlowActionCallHandlerAtomicPolicy(value string) bool {
	switch value {
	case flowActionCallHandlerAtomicPolicySafe, flowActionCallHandlerAtomicPolicyReject:
		return true
	default:
		return false
	}
}

func validFlowActionCallHandlerPreviewPolicy(value string) bool {
	switch value {
	case flowActionCallHandlerPreviewPolicyReject,
		flowActionCallHandlerPreviewPolicyAllow,
		flowActionCallHandlerPreviewPolicyAllowWithPreviewPersistence,
		flowActionCallHandlerPreviewPolicyRejectEmailIO,
		flowActionCallHandlerPreviewPolicyInvokeHandler:
		return true
	default:
		return false
	}
}

func validFlowActionCallHandlerContext(value string) bool {
	switch value {
	case flowActionCallHandlerContextNone, flowActionCallHandlerContextStandardActionProtected:
		return true
	default:
		return false
	}
}

func normalizeSourceBackedFlowActionCallHandlerProjection(p sourceBackedFlowActionCallHandlerProjection) (sourceBackedFlowActionCallHandlerProjection, error) {
	p.OwnerPack = strings.TrimSpace(p.OwnerPack)
	p.ActionType = normalizeFlowActionCallHandlerActionType(p.ActionType)
	p.HandlerRef = strings.TrimSpace(p.HandlerRef)
	p.AtomicPolicy = strings.ToLower(strings.TrimSpace(p.AtomicPolicy))
	p.PreviewPolicy = normalizeFlowActionCallHandlerPolicy(p.PreviewPolicy, flowActionCallHandlerDefaultPreviewPolicy)
	p.RequiresContext = normalizeFlowActionCallHandlerPolicy(p.RequiresContext, flowActionCallHandlerDefaultContext)
	switch {
	case p.OwnerPack == "":
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("ownerPack is required")
	case p.ActionType == "":
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("actionType is required")
	case p.HandlerRef == "":
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("handlerRef is required")
	case p.AtomicPolicy == "":
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("atomicPolicy is required")
	case !validFlowActionCallHandlerAtomicPolicy(p.AtomicPolicy):
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("atomicPolicy %q is not supported", p.AtomicPolicy)
	case !validFlowActionCallHandlerPreviewPolicy(p.PreviewPolicy):
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("previewPolicy %q is not supported", p.PreviewPolicy)
	case !validFlowActionCallHandlerContext(p.RequiresContext):
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("requiresContext %q is not supported", p.RequiresContext)
	}
	return p, nil
}

func runtimeStartupShallowFlowActionCallHandlerLookupProjection(row runtimeStartupShallowDeclaredRow, declaredProjection string) (sourceBackedFlowActionCallHandlerProjection, error) {
	var p sourceBackedFlowActionCallHandlerProjection
	if err := json.Unmarshal([]byte(declaredProjection), &p); err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, err
	}
	p, err := normalizeSourceBackedFlowActionCallHandlerProjection(p)
	if err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, err
	}
	if p.ActionType != row.BusinessKey {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("actionType %q does not match business key %q", p.ActionType, row.BusinessKey)
	}
	return p, nil
}

func upsertSourceBackedFlowActionCallHandler(ctx context.Context, db *sql.DB, projection sourceBackedFlowActionCallHandlerProjection) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	projection, err := normalizeSourceBackedFlowActionCallHandlerProjection(projection)
	if err != nil {
		return err
	}
	if err := verifyFlowActionCallHandlerGlobalOwner(ctx, db, projection); err != nil {
		return err
	}
	_, err = db.ExecContext(ctx, `
INSERT INTO _System_FlowActionCallHandler (
  Id, OwnerPack, ActionType, HandlerRef, AtomicPolicy, PreviewPolicy, RequiresContext, SendsEmail,
  IsActive, IsDeleted, CreatedDate, LastModifiedDate
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, 0, NOW(), NOW())
ON DUPLICATE KEY UPDATE
  OwnerPack = VALUES(OwnerPack),
  ActionType = VALUES(ActionType),
  HandlerRef = VALUES(HandlerRef),
  AtomicPolicy = VALUES(AtomicPolicy),
  PreviewPolicy = VALUES(PreviewPolicy),
  RequiresContext = VALUES(RequiresContext),
  SendsEmail = VALUES(SendsEmail),
  IsActive = 1,
  IsDeleted = 0,
  LastModifiedDate = NOW()`,
		utils.GenerateID(),
		projection.OwnerPack,
		projection.ActionType,
		projection.HandlerRef,
		projection.AtomicPolicy,
		projection.PreviewPolicy,
		projection.RequiresContext,
		projection.SendsEmail,
	)
	if err != nil {
		return fmt.Errorf("upsert %s %s/%s: %w", tableFlowActionCallHandler, projection.OwnerPack, projection.ActionType, err)
	}
	return nil
}

func verifyFlowActionCallHandlerGlobalOwner(ctx context.Context, db *sql.DB, projection sourceBackedFlowActionCallHandlerProjection) error {
	rows, err := db.QueryContext(ctx, `
SELECT OwnerPack
  FROM _System_FlowActionCallHandler
 WHERE ActionType = ?
   AND COALESCE(IsActive, 1) = 1
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY OwnerPack`, projection.ActionType)
	if err != nil {
		return fmt.Errorf("query flow_action_call_handler owner for %s: %w", projection.ActionType, err)
	}
	defer rows.Close()

	var conflictingOwners []string
	for rows.Next() {
		var ownerPack string
		if err := rows.Scan(&ownerPack); err != nil {
			return fmt.Errorf("scan flow_action_call_handler owner for %s: %w", projection.ActionType, err)
		}
		ownerPack = strings.TrimSpace(ownerPack)
		if ownerPack != "" && ownerPack != projection.OwnerPack {
			conflictingOwners = append(conflictingOwners, ownerPack)
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate flow_action_call_handler owner for %s: %w", projection.ActionType, err)
	}
	if len(conflictingOwners) > 0 {
		sort.Strings(conflictingOwners)
		return fmt.Errorf("flow_action_call_handler actionType %s already owned by pack %s", projection.ActionType, strings.Join(conflictingOwners, ", "))
	}
	return nil
}

func liveSourceBackedFlowActionCallHandlerProjection(ctx context.Context, db *sql.DB, ownerPack string, actionType string) (sourceBackedFlowActionCallHandlerProjection, error) {
	if db == nil {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("database is required")
	}
	ownerPack = strings.TrimSpace(ownerPack)
	actionType = normalizeFlowActionCallHandlerActionType(actionType)
	rows, err := db.QueryContext(ctx, `
SELECT OwnerPack, ActionType, HandlerRef, AtomicPolicy, PreviewPolicy, RequiresContext, COALESCE(SendsEmail, 0)
  FROM _System_FlowActionCallHandler
 WHERE ActionType = ?
   AND COALESCE(IsActive, 1) = 1
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY OwnerPack`, actionType)
	if err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("query flow_action_call_handler %s/%s: %w", ownerPack, actionType, err)
	}
	defer rows.Close()

	var found []sourceBackedFlowActionCallHandlerProjection
	for rows.Next() {
		var p sourceBackedFlowActionCallHandlerProjection
		if err := rows.Scan(
			&p.OwnerPack,
			&p.ActionType,
			&p.HandlerRef,
			&p.AtomicPolicy,
			&p.PreviewPolicy,
			&p.RequiresContext,
			&p.SendsEmail,
		); err != nil {
			return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("scan flow_action_call_handler %s/%s: %w", ownerPack, actionType, err)
		}
		found = append(found, p)
	}
	if err := rows.Err(); err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("iterate flow_action_call_handler %s/%s: %w", ownerPack, actionType, err)
	}
	if len(found) == 0 {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s/%s: row missing or inactive", ownerPack, actionType)
	}
	if len(found) > 1 {
		var owners []string
		for _, row := range found {
			owners = append(owners, strings.TrimSpace(row.OwnerPack))
		}
		sort.Strings(owners)
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler %s has duplicate live rows owned by: %s", actionType, strings.Join(owners, ", "))
	}
	p := found[0]
	p, err = normalizeSourceBackedFlowActionCallHandlerProjection(p)
	if err != nil {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("normalize live flow_action_call_handler %s/%s: %w", ownerPack, actionType, err)
	}
	if ownerPack != "" && p.OwnerPack != ownerPack {
		return sourceBackedFlowActionCallHandlerProjection{}, fmt.Errorf("flow_action_call_handler actionType %s is already owned by pack %s, want %s", actionType, p.OwnerPack, ownerPack)
	}
	return p, nil
}

type flowActionCallHandlerPackBatch struct {
	declaredActionTypes []string
}

func newFlowActionCallHandlerPackSweeperComponent(packName string, batch *flowActionCallHandlerPackBatch) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindFlowActionCallHandler,
		BusinessKey:   JoinComponentBusinessKey(packName, "@sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applyFlowActionCallHandlerPackSweeper(ctx, deps.DB, batch, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifyFlowActionCallHandlerPackSweeper(ctx, deps.DB, batch, packName)
		},
	}
}

func applyFlowActionCallHandlerPackSweeper(ctx context.Context, db *sql.DB, batch *flowActionCallHandlerPackBatch, ownerPack string) error {
	orphans, err := queryOrphanFlowActionCallHandlerRows(ctx, db, batch, ownerPack)
	if err != nil {
		return err
	}
	for _, orphanID := range orphans {
		if _, err := db.ExecContext(ctx, `
UPDATE _System_FlowActionCallHandler
   SET IsActive = 0,
       IsDeleted = 1,
       LastModifiedDate = NOW()
 WHERE Id = ?
   AND COALESCE(IsDeleted, 0) = 0`, orphanID); err != nil {
			return fmt.Errorf("soft-delete orphan flow_action_call_handler %s for pack %s: %w", orphanID, strings.TrimSpace(ownerPack), err)
		}
	}
	return nil
}

func verifyFlowActionCallHandlerPackSweeper(ctx context.Context, db *sql.DB, batch *flowActionCallHandlerPackBatch, ownerPack string) error {
	orphans, err := queryOrphanFlowActionCallHandlerRows(ctx, db, batch, ownerPack)
	if err != nil {
		return err
	}
	if len(orphans) == 0 {
		return nil
	}
	sort.Strings(orphans)
	return fmt.Errorf("pack %s has %d undeclared flow_action_call_handler row ID(s) still live: %s", strings.TrimSpace(ownerPack), len(orphans), strings.Join(orphans, ", "))
}

func queryOrphanFlowActionCallHandlerRows(ctx context.Context, db *sql.DB, batch *flowActionCallHandlerPackBatch, ownerPack string) ([]string, error) {
	if db == nil {
		return nil, fmt.Errorf("database is required")
	}
	declaredSet := map[string]struct{}{}
	if batch != nil {
		for _, actionType := range batch.declaredActionTypes {
			actionType = normalizeFlowActionCallHandlerActionType(actionType)
			if actionType != "" {
				declaredSet[actionType] = struct{}{}
			}
		}
	}
	ownerPack = strings.TrimSpace(ownerPack)
	rows, err := db.QueryContext(ctx, `
SELECT Id, ActionType
  FROM _System_FlowActionCallHandler
 WHERE COALESCE(OwnerPack, '') = ?
   AND COALESCE(IsActive, 1) = 1
   AND COALESCE(IsDeleted, 0) = 0
 ORDER BY ActionType`, ownerPack)
	if err != nil {
		return nil, fmt.Errorf("query live flow_action_call_handler rows for pack sweep %s: %w", ownerPack, err)
	}
	defer rows.Close()

	var orphanIDs []string
	for rows.Next() {
		var id, actionType string
		if err := rows.Scan(&id, &actionType); err != nil {
			return nil, fmt.Errorf("scan flow_action_call_handler row for pack sweep %s: %w", ownerPack, err)
		}
		if _, ok := declaredSet[normalizeFlowActionCallHandlerActionType(actionType)]; ok {
			continue
		}
		orphanIDs = append(orphanIDs, id)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate flow_action_call_handler rows for pack sweep %s: %w", ownerPack, err)
	}
	return orphanIDs, nil
}