package connectors

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"sort"
	"strings"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/callout"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	configwire "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundhandlerconfig/configwire"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/packstate"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
)

type connectorInboundBatchContract struct {
	requiredObjectKinds map[string]struct{}
	projectionFields    map[string]map[string]struct{}
}

const activeInstalledPackStatus = "active"

type connectorInboundBatchContractResolver func(context.Context, *sql.DB, string, string, []*connector_callout.ConnectorCalloutProfile, bool) (connectorInboundBatchContract, error)
type namedCredentialAuthTypeResolver func(context.Context, string) (string, error)
type namedCredentialRowAuthTypeResolver func(context.Context, packstate.ConnectorInboundWorkerPackRow, string) (string, error)
type namedCredentialDrainOnlyEligibilityResolver func(context.Context, string, string) (bool, error)

func loadProfilesForInboundWorkerStartup(ctx context.Context, db *sql.DB, ownerPack, packInstallID string, drainOnly bool) ([]*connector_callout.ConnectorCalloutProfile, error) {
	if drainOnly {
		profiles, err := connector_callout.LoadProfilesForInboundWorkerDrainOnly(ctx, db, ownerPack, packInstallID)
		if err != nil {
			return nil, err
		}
		for _, profile := range profiles {
			if profile != nil && legacyRequestOrderBatchProfile(*profile) {
				connector_callout.AllowLegacyRequestOrderBatchDemux(profile)
			}
		}
		return profiles, nil
	}
	return connector_callout.LoadProfilesForInboundWorker(ctx, db, ownerPack, packInstallID, connector_callout.BatchProfileReadActiveOnly)
}

func resolveBatchContractForInboundWorkerStartup(ctx context.Context, db *sql.DB, ownerPack, packInstallID string, profiles []*connector_callout.ConnectorCalloutProfile, drainOnly bool) (connectorInboundBatchContract, error) {
	if drainOnly {
		return resolveBatchContractFromRuntimeHandlerConfigs(ctx, db, ownerPack, packInstallID, profiles, true)
	}
	return resolveBatchContractFromRuntimeHandlerConfigs(ctx, db, ownerPack, packInstallID, profiles, false)
}

func resolveBatchContractFromRuntimeHandlerConfigs(ctx context.Context, db *sql.DB, ownerPack, packInstallID string, profiles []*connector_callout.ConnectorCalloutProfile, drainOnly bool) (connectorInboundBatchContract, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	packInstallID = strings.TrimSpace(packInstallID)
	statusPredicate := "p.Status = ?"
	handlerDeletedPredicate := "AND COALESCE(c.IsDeleted, 0) = 0"
	statusArg := activeInstalledPackStatus
	if drainOnly {
		statusPredicate = "p.Status <> ?"
		handlerDeletedPredicate = ""
	}
	rows, err := db.QueryContext(ctx, `
	SELECT c.Namespace, c.CapabilityKind, c.ConfigJSON
	  FROM _System_ConnectorInboundHandlerConfig c
  JOIN _System_InstalledPack p
    ON p.PackInstallId = c.InstalledPackId
   AND p.Name = c.Namespace
	 WHERE p.Name = ?
	   AND p.PackInstallId = ?
		   AND p.Category = 'connector-pack'
		   AND `+statusPredicate+`
		   AND TRIM(COALESCE(p.AppliedPayloadHash, '')) <> ''
		   AND TRIM(COALESCE(p.DesiredPayloadHash, '')) <> ''
		   AND p.AppliedPayloadHash = p.DesiredPayloadHash
		   AND COALESCE(p.IsDeleted, 0) = 0
		   `+handlerDeletedPredicate+`
	 ORDER BY c.Namespace, c.CapabilityKind`, ownerPack, packInstallID, statusArg)
	if err != nil {
		return connectorInboundBatchContract{}, fmt.Errorf("load connector inbound handler configs for %s (%s): %w", ownerPack, packInstallID, err)
	}
	defer rows.Close()

	declarations := make([]packstate.ConnectorInboundHandlerConfigComponentPayload, 0)
	for rows.Next() {
		var (
			decl      packstate.ConnectorInboundHandlerConfigComponentPayload
			configRaw string
		)
		if err := rows.Scan(&decl.Namespace, &decl.CapabilityKind, &configRaw); err != nil {
			return connectorInboundBatchContract{}, fmt.Errorf("scan connector inbound handler config for %s (%s): %w", ownerPack, packInstallID, err)
		}
		decl.Config = json.RawMessage(configRaw)
		declarations = append(declarations, decl)
	}
	if err := rows.Err(); err != nil {
		return connectorInboundBatchContract{}, fmt.Errorf("iterate connector inbound handler configs for %s (%s): %w", ownerPack, packInstallID, err)
	}
	status := activeInstalledPackStatus
	if drainOnly {
		status = "removing"
	}
	contract, err := buildBatchContractFromHandlerConfigDeclarations(ownerPack, packInstallID, status, declarations, profiles)
	if err != nil {
		return connectorInboundBatchContract{}, err
	}
	if profilesDeclareBatchProjectionBodyField(profiles) {
		projectionFields, err := resolveInboundMappingProjectionFields(ctx, db, ownerPack, packInstallID)
		if err != nil {
			return connectorInboundBatchContract{}, fmt.Errorf("load connector inbound mapping projection fields for %s (%s): %w", ownerPack, packInstallID, err)
		}
		contract.projectionFields = projectionFields
	}
	return contract, nil
}

func legacyRequestOrderBatchProfile(profile connector_callout.ConnectorCalloutProfile) bool {
	spec := profile.BatchSpec
	if strings.TrimSpace(spec.IDResponseField) != "" {
		return false
	}
	return connector_callout.ValidateBatchSpecForLegacyRequestOrderDemux(spec) == nil
}

func buildBatchContractFromHandlerConfigDeclarations(ownerPack, packInstallID, status string, declarations []packstate.ConnectorInboundHandlerConfigComponentPayload, profiles []*connector_callout.ConnectorCalloutProfile) (connectorInboundBatchContract, error) {
	eligibleObjectKinds := collectEligibleBatchObjectKinds(profiles)
	requiredObjectKinds := make(map[string]struct{})
	requirementByObjectKind := make(map[string]bool)
	foundEnvelopeParser := false

	for _, decl := range declarations {
		if strings.TrimSpace(decl.Namespace) != strings.TrimSpace(ownerPack) {
			continue
		}
		if strings.TrimSpace(decl.CapabilityKind) != configwire.EnvelopeParserKind {
			continue
		}
		foundEnvelopeParser = true

		cfgSet, _, _, err := configwire.DecodeEnvelopeMappingConfigSetWithComposites(ownerPack, decl.Config)
		if err != nil {
			return connectorInboundBatchContract{}, fmt.Errorf("decode envelope_parser handler config for %s (%s): %w", ownerPack, packInstallID, err)
		}
		if cfgSet.Default != nil {
			if err := accumulateRequiredObjectKindRequirements(requiredObjectKinds, requirementByObjectKind, *cfgSet.Default, eligibleObjectKinds); err != nil {
				return connectorInboundBatchContract{}, fmt.Errorf("decode envelope_parser handler config for %s (%s): %w", ownerPack, packInstallID, err)
			}
		}
		for _, cfg := range cfgSet.EndpointConfigs {
			if err := accumulateRequiredObjectKindRequirements(requiredObjectKinds, requirementByObjectKind, cfg, eligibleObjectKinds); err != nil {
				return connectorInboundBatchContract{}, fmt.Errorf("decode envelope_parser handler config for %s (%s): %w", ownerPack, packInstallID, err)
			}
		}
	}

	if !foundEnvelopeParser {
		if strings.EqualFold(strings.TrimSpace(status), activeInstalledPackStatus) ||
			connectorcapabilities.HasLegacyCompiledHandler(ownerPack, connectorcapabilities.EnvelopeParser) {
			return connectorInboundBatchContract{requiredObjectKinds: requiredObjectKinds}, nil
		}
		return connectorInboundBatchContract{}, fmt.Errorf("connector inbound startup contract missing envelope_parser declaration for %s (%s)", ownerPack, packInstallID)
	}

	return connectorInboundBatchContract{requiredObjectKinds: requiredObjectKinds}, nil
}

func accumulateRequiredObjectKindRequirements(required map[string]struct{}, requirementByObjectKind map[string]bool, cfg configwire.EnvelopeMappingConfig, eligible map[string]struct{}) error {
	mode := normalizeInboundProcessingMode(cfg.InboundProcessingMode)
	for _, rule := range cfg.Rules {
		objectKind := strings.TrimSpace(rule.ObjectKind)
		if objectKind == "" {
			continue
		}
		requiredForObjectKind := false
		switch mode {
		case configwire.InboundProcessingModePayloadOnlyAllowed:
			requiredForObjectKind = false
		case configwire.InboundProcessingModeEnrichmentRequired:
			requiredForObjectKind = true
		default:
			if _, ok := eligible[objectKind]; ok {
				requiredForObjectKind = true
			}
		}
		if requirementByObjectKind != nil {
			if existing, ok := requirementByObjectKind[objectKind]; ok && existing != requiredForObjectKind {
				return fmt.Errorf("envelope_parser endpointConfigs have conflicting batch enrichment requirements for objectKind %q; connector inbound queue envelopes do not carry endpoint identity", objectKind)
			}
			requirementByObjectKind[objectKind] = requiredForObjectKind
		}
		if requiredForObjectKind {
			required[objectKind] = struct{}{}
		}
	}
	return nil
}

func normalizeInboundProcessingMode(mode configwire.InboundProcessingMode) configwire.InboundProcessingMode {
	return configwire.InboundProcessingMode(strings.ToLower(strings.TrimSpace(string(mode))))
}

func collectEligibleBatchObjectKinds(profiles []*connector_callout.ConnectorCalloutProfile) map[string]struct{} {
	out := make(map[string]struct{})
	for _, profile := range profiles {
		if !profileBatchSpecConfigured(profile) {
			continue
		}
		for objectKind, objectPath := range profile.ObjectPathMap {
			objectKind = strings.TrimSpace(objectKind)
			if objectKind == "" || strings.TrimSpace(objectPath) == "" {
				continue
			}
			out[objectKind] = struct{}{}
		}
	}
	return out
}

func profileHasUsableBatchObjectCoverage(profile *connector_callout.ConnectorCalloutProfile) bool {
	if profile == nil {
		return false
	}
	for objectKind, objectPath := range profile.ObjectPathMap {
		if strings.TrimSpace(objectKind) == "" || strings.TrimSpace(objectPath) == "" {
			continue
		}
		return true
	}
	return false
}

func hasDuplicateExecutableBatchCoverage(profiles []*connector_callout.ConnectorCalloutProfile) bool {
	seen := make(map[string]struct{})
	for _, profile := range profiles {
		if !profileBatchSpecConfigured(profile) {
			continue
		}
		for objectKind, objectPath := range profile.ObjectPathMap {
			objectKind = strings.TrimSpace(objectKind)
			if objectKind == "" || strings.TrimSpace(objectPath) == "" {
				continue
			}
			if _, ok := seen[objectKind]; ok {
				return true
			}
			seen[objectKind] = struct{}{}
		}
	}
	return false
}

func profileBatchSpecConfigured(profile *connector_callout.ConnectorCalloutProfile) bool {
	if profile == nil {
		return false
	}
	if connector_callout.UsesLegacyRequestOrderBatchDemux(profile) {
		return connector_callout.ValidateBatchSpecForLegacyRequestOrderDemux(profile.BatchSpec) == nil
	}
	return connector_callout.ValidateBatchSpec(profile.BatchSpec) == nil
}

func sortedMissingObjectKinds(required, eligible map[string]struct{}) []string {
	out := make([]string, 0, len(required))
	for objectKind := range required {
		if _, ok := eligible[objectKind]; ok {
			continue
		}
		out = append(out, objectKind)
	}
	sort.Strings(out)
	return out
}

func profilesDeclareBatchProjectionBodyField(profiles []*connector_callout.ConnectorCalloutProfile) bool {
	for _, profile := range profiles {
		if profile == nil {
			continue
		}
		if strings.TrimSpace(profile.BatchSpec.ProjectionBodyField) != "" {
			return true
		}
	}
	return false
}

func resolveInboundMappingProjectionFields(ctx context.Context, db *sql.DB, ownerPack, packInstallID string) (map[string]map[string]struct{}, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	packInstallID = strings.TrimSpace(packInstallID)
	if ownerPack == "" || packInstallID == "" {
		return nil, nil
	}
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT s.ObjectKind, s.SourceField
  FROM _System_ConnectorInboundProjectionSnapshot s
 WHERE s.OwnerPackName = ?
   AND s.OwnerPackInstallId = ?
   AND COALESCE(s.IsDeleted, 0) = 0
   AND (
        EXISTS (
            SELECT 1
              FROM _System_InstalledPackDependency d
             WHERE d.DependsOnPackName = ?
               AND COALESCE(d.IsDeleted, 0) = 0
               AND d.DependentPackName = s.CompanionPackName
               AND d.DependentPackInstallId = s.CompanionPackInstallId
               AND (
                    EXISTS (
                        SELECT 1
                          FROM _System_InstalledPack p
                         WHERE p.Name = d.DependentPackName
                           AND p.PackInstallId = d.DependentPackInstallId
                           AND p.Category = 'app-pack'
                           AND p.Status = 'active'
                           AND TRIM(COALESCE(p.AppliedPayloadHash, '')) <> ''
                           AND TRIM(COALESCE(p.DesiredPayloadHash, '')) <> ''
                           AND p.AppliedPayloadHash = p.DesiredPayloadHash
                           AND COALESCE(p.IsDeleted, 0) = 0
                    )
                    OR EXISTS (
                        SELECT 1
                          FROM _System_InstalledPack p
                         WHERE p.Name = d.DependentPackName
                           AND p.Category = 'app-pack'
                           AND p.Status = 'installing'
                           AND COALESCE(p.IsDeleted, 0) = 0
                    )
               )
        )
        OR EXISTS (
            SELECT 1
              FROM _System_ConnectorEvent e
             WHERE e.InstalledPackId = ?
               AND COALESCE(e.CompanionPackInstallId, '') = s.CompanionPackInstallId
               AND e.Status IN ('pending', 'failed', 'processing')
        )
   )
 ORDER BY s.ObjectKind, s.SourceField`, ownerPack, packInstallID, ownerPack, packInstallID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	out := make(map[string]map[string]struct{})
	for rows.Next() {
		var objectKind, field string
		if err := rows.Scan(&objectKind, &field); err != nil {
			return nil, err
		}
		objectKind = strings.TrimSpace(objectKind)
		field = strings.TrimSpace(field)
		if objectKind == "" || field == "" {
			continue
		}
		if out[objectKind] == nil {
			out[objectKind] = make(map[string]struct{})
		}
		out[objectKind][field] = struct{}{}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func resolveInboundMappingProjectionFingerprint(ctx context.Context, db *sql.DB, ownerPack, packInstallID string, drainOnly bool) (string, error) {
	ownerPack = strings.TrimSpace(ownerPack)
	packInstallID = strings.TrimSpace(packInstallID)
	if ownerPack == "" || packInstallID == "" || drainOnly {
		return "", nil
	}
	rows, err := db.QueryContext(ctx, `
SELECT DISTINCT s.CompanionPackName, s.CompanionPackInstallId, s.ObjectKind, s.SourceField
  FROM _System_ConnectorInboundProjectionSnapshot s
 WHERE s.OwnerPackName = ?
   AND s.OwnerPackInstallId = ?
   AND COALESCE(s.IsDeleted, 0) = 0
   AND (
        EXISTS (
            SELECT 1
              FROM _System_InstalledPackDependency d
             WHERE d.DependsOnPackName = ?
               AND COALESCE(d.IsDeleted, 0) = 0
               AND d.DependentPackName = s.CompanionPackName
               AND d.DependentPackInstallId = s.CompanionPackInstallId
               AND (
                    EXISTS (
                        SELECT 1
                          FROM _System_InstalledPack p
                         WHERE p.Name = d.DependentPackName
                           AND p.PackInstallId = d.DependentPackInstallId
                           AND p.Category = 'app-pack'
                           AND p.Status = 'active'
                           AND TRIM(COALESCE(p.AppliedPayloadHash, '')) <> ''
                           AND TRIM(COALESCE(p.DesiredPayloadHash, '')) <> ''
                           AND p.AppliedPayloadHash = p.DesiredPayloadHash
                           AND COALESCE(p.IsDeleted, 0) = 0
                    )
                    OR EXISTS (
                        SELECT 1
                          FROM _System_InstalledPack p
                         WHERE p.Name = d.DependentPackName
                           AND p.Category = 'app-pack'
                           AND p.Status = 'installing'
                           AND COALESCE(p.IsDeleted, 0) = 0
                    )
               )
        )
        OR EXISTS (
            SELECT 1
              FROM _System_ConnectorEvent e
             WHERE e.InstalledPackId = ?
               AND COALESCE(e.CompanionPackInstallId, '') = s.CompanionPackInstallId
               AND e.Status IN ('pending', 'failed', 'processing')
        )
   )
 ORDER BY s.CompanionPackName, s.CompanionPackInstallId, s.ObjectKind, s.SourceField`, ownerPack, packInstallID, ownerPack, packInstallID)
	if err != nil {
		return "", err
	}
	defer rows.Close()

	var parts []string
	for rows.Next() {
		var dependentPack, dependentInstallID, objectKind, field string
		if err := rows.Scan(&dependentPack, &dependentInstallID, &objectKind, &field); err != nil {
			return "", err
		}
		parts = append(parts, strings.Join([]string{
			strings.TrimSpace(dependentPack),
			strings.TrimSpace(dependentInstallID),
			strings.TrimSpace(objectKind),
			strings.TrimSpace(field),
		}, "\x1f"))
	}
	if err := rows.Err(); err != nil {
		return "", err
	}
	if len(parts) == 0 {
		return "", nil
	}
	sum := sha256.Sum256([]byte(strings.Join(parts, "\x1e")))
	return hex.EncodeToString(sum[:]), nil
}

func applyBatchProjectionFields(profiles []*connector_callout.ConnectorCalloutProfile, projectionFields map[string]map[string]struct{}) ([]*connector_callout.ConnectorCalloutProfile, error) {
	out := make([]*connector_callout.ConnectorCalloutProfile, 0, len(profiles))
	for _, profile := range profiles {
		if profile == nil {
			out = append(out, profile)
			continue
		}
		bodyField := strings.TrimSpace(profile.BatchSpec.ProjectionBodyField)
		if bodyField == "" {
			out = append(out, profile)
			continue
		}
		if strings.TrimSpace(profile.BatchSpec.IDsRequestField) == bodyField {
			return nil, fmt.Errorf("connector inbound: projectionBodyField %q collides with idsRequestField for %s/%s", bodyField, profile.OwnerPack, profile.ProfileName)
		}
		objectPaths := executableObjectPaths(profile.ObjectPathMap)
		if len(objectPaths) == 0 {
			cp, err := cloneProfileWithProjectionFields(profile, bodyField, nil)
			if err != nil {
				return nil, err
			}
			out = append(out, cp)
			continue
		}
		if len(objectPaths) > 1 {
			return nil, fmt.Errorf("connector inbound: projectionBodyField %q for %s/%s requires metadata-owned per-object profile partitioning; multi-object projection profiles are out of scope", bodyField, profile.OwnerPack, profile.ProfileName)
		}
		objectKind := firstExecutableObjectKind(objectPaths)
		cp, err := cloneProfileWithProjectionFields(profile, bodyField, projectionFieldsForObjectKind(objectKind, projectionFields))
		if err != nil {
			return nil, err
		}
		out = append(out, cp)
	}
	return out, nil
}

func executableObjectPaths(objectPathMap map[string]string) map[string]string {
	out := make(map[string]string)
	for objectKind, objectPath := range objectPathMap {
		objectKind = strings.TrimSpace(objectKind)
		objectPath = strings.TrimSpace(objectPath)
		if objectKind == "" || strings.TrimSpace(objectPath) == "" {
			continue
		}
		out[objectKind] = objectPath
	}
	return out
}

func firstExecutableObjectKind(values map[string]string) string {
	for key := range values {
		return key
	}
	return ""
}

func projectionFieldsForObjectKind(objectKind string, projectionFields map[string]map[string]struct{}) map[string]struct{} {
	fields := make(map[string]struct{})
	for field := range projectionFields[strings.TrimSpace(objectKind)] {
		if strings.TrimSpace(field) != "" {
			fields[field] = struct{}{}
		}
	}
	return fields
}

func cloneProfileWithProjectionFields(profile *connector_callout.ConnectorCalloutProfile, bodyField string, fields map[string]struct{}) (*connector_callout.ConnectorCalloutProfile, error) {
	cp := *profile
	cp.BatchSpec.ExtraBodyFields = cloneExtraBodyFields(profile.BatchSpec.ExtraBodyFields)
	merged, err := mergeProjectionBodyField(cp.BatchSpec.ExtraBodyFields[bodyField], fields)
	if err != nil {
		return nil, fmt.Errorf("connector inbound: merge projectionBodyField %q for %s/%s: %w", bodyField, profile.OwnerPack, profile.ProfileName, err)
	}
	cp.BatchSpec.ExtraBodyFields[bodyField] = merged
	return &cp, nil
}

func cloneExtraBodyFields(in map[string]json.RawMessage) map[string]json.RawMessage {
	out := make(map[string]json.RawMessage, len(in)+1)
	for key, value := range in {
		out[key] = append(json.RawMessage(nil), value...)
	}
	return out
}

func mergeProjectionBodyField(raw json.RawMessage, fields map[string]struct{}) (json.RawMessage, error) {
	merged := make(map[string]struct{}, len(fields))
	if len(raw) > 0 {
		var existing []string
		if err := json.Unmarshal(raw, &existing); err != nil {
			return nil, fmt.Errorf("existing projection body field must be a JSON string array: %w", err)
		}
		for _, field := range existing {
			field = strings.TrimSpace(field)
			if field != "" {
				return nil, fmt.Errorf("existing projection body field must be empty when projectionBodyField is set; field inventory is runtime-owned by connector inbound projection snapshots")
			}
		}
	}
	for field := range fields {
		field = strings.TrimSpace(field)
		if field != "" {
			merged[field] = struct{}{}
		}
	}
	ordered := make([]string, 0, len(merged))
	for field := range merged {
		ordered = append(ordered, field)
	}
	sort.Strings(ordered)
	return json.Marshal(ordered)
}

// ActiveInboundWorkersNeedNamedCredentialCipher reports whether any inbound
// worker on the current startup surface exposes an executable batch profile
// that uses a secret-backed named credential auth type. Payload-only workers,
// custom-auth batch profiles, and malformed/non-executable batch specs do not
// require the master key at startup.
func ActiveInboundWorkersNeedNamedCredentialCipher(ctx context.Context, db *sql.DB, rows []packstate.ConnectorInboundWorkerPackRow, resolveAuthType namedCredentialAuthTypeResolver, resolveDrainEligible ...namedCredentialDrainOnlyEligibilityResolver) (bool, error) {
	if resolveAuthType == nil {
		return false, fmt.Errorf("connector inbound named credential preflight: resolveAuthType is required")
	}
	return ActiveInboundWorkersNeedNamedCredentialCipherWithRowResolver(ctx, db, rows, func(ctx context.Context, _ packstate.ConnectorInboundWorkerPackRow, credentialRef string) (string, error) {
		return resolveAuthType(ctx, credentialRef)
	}, firstDrainEligibilityResolver(resolveDrainEligible))
}

func ActiveInboundWorkersNeedNamedCredentialCipherWithRowResolver(ctx context.Context, db *sql.DB, rows []packstate.ConnectorInboundWorkerPackRow, resolveAuthType namedCredentialRowAuthTypeResolver, resolveDrainEligible ...namedCredentialDrainOnlyEligibilityResolver) (bool, error) {
	return activeInboundWorkersNeedNamedCredentialCipherWithRowResolver(ctx, db, rows, loadProfilesForInboundWorkerStartup, resolveAuthType, firstDrainEligibilityResolver(resolveDrainEligible))
}

func activeInboundWorkersNeedNamedCredentialCipher(
	ctx context.Context,
	db *sql.DB,
	rows []packstate.ConnectorInboundWorkerPackRow,
	loadProfiles connectorInboundProfileLoader,
	resolveAuthType namedCredentialAuthTypeResolver,
	resolveDrainEligible ...namedCredentialDrainOnlyEligibilityResolver,
) (bool, error) {
	if resolveAuthType == nil {
		return false, fmt.Errorf("connector inbound named credential preflight: resolveAuthType is required")
	}
	return activeInboundWorkersNeedNamedCredentialCipherWithRowResolver(ctx, db, rows, loadProfiles, func(ctx context.Context, _ packstate.ConnectorInboundWorkerPackRow, credentialRef string) (string, error) {
		return resolveAuthType(ctx, credentialRef)
	}, firstDrainEligibilityResolver(resolveDrainEligible))
}

func activeInboundWorkersNeedNamedCredentialCipherWithRowResolver(
	ctx context.Context,
	db *sql.DB,
	rows []packstate.ConnectorInboundWorkerPackRow,
	loadProfiles connectorInboundProfileLoader,
	resolveAuthType namedCredentialRowAuthTypeResolver,
	resolveDrainEligible ...namedCredentialDrainOnlyEligibilityResolver,
) (bool, error) {
	if loadProfiles == nil {
		return false, fmt.Errorf("connector inbound named credential preflight: loadProfiles is required")
	}
	if resolveAuthType == nil {
		return false, fmt.Errorf("connector inbound named credential preflight: resolveAuthType is required")
	}

	authTypeCache := make(map[string]string)
	for _, row := range rows {
		eligible, err := drainOnlyRowStillEligible(ctx, row, firstDrainEligibilityResolver(resolveDrainEligible))
		if err != nil {
			return false, fmt.Errorf("verify drain eligibility for %s (%s): %w", row.Namespace, row.InstallID, err)
		}
		if !eligible {
			continue
		}

		profiles, err := loadProfiles(ctx, db, row.Namespace, row.InstallID, row.DrainOnly)
		if err != nil {
			if row.DrainOnly {
				continue
			}
			return false, fmt.Errorf("load profiles for %s (%s): %w", row.Namespace, row.InstallID, err)
		}
		contract := connectorInboundBatchContract{}
		if db != nil {
			var err error
			contract, err = resolveBatchContractForInboundWorkerStartup(ctx, db, row.Namespace, row.InstallID, profiles, row.DrainOnly)
			if err != nil {
				if row.DrainOnly {
					continue
				}
				return false, fmt.Errorf("resolve startup contract for %s (%s): %w", row.Namespace, row.InstallID, err)
			}
		}
		effectiveProfiles := profiles
		if len(contract.requiredObjectKinds) > 0 {
			effectiveProfiles = filterProfilesForRequiredObjectKinds(profiles, contract.requiredObjectKinds)
		}
		if len(contract.requiredObjectKinds) == 0 && hasDuplicateExecutableBatchCoverage(effectiveProfiles) {
			continue
		}

		rowNeedsCipher := false
		rowCtx := ctx
		if row.DrainOnly {
			rowCtx = namedcredential.WithDrainOnlyInstalledPack(rowCtx, row.Namespace, row.InstallID)
		}
		for _, profile := range effectiveProfiles {
			if !profileBatchSpecConfigured(profile) {
				continue
			}
			if !profileHasUsableBatchObjectCoverage(profile) {
				continue
			}

			credentialRef := strings.TrimSpace(profile.BatchSpec.CredentialRef)
			cacheKey := namedCredentialAuthTypeCacheKey(row, credentialRef)
			authType, ok := authTypeCache[cacheKey]
			if !ok {
				var err error
				authType, err = resolveAuthType(rowCtx, row, credentialRef)
				if err != nil {
					return false, fmt.Errorf("resolve auth type for %s/%s credential %s: %w", row.Namespace, row.InstallID, credentialRef, err)
				}
				authTypeCache[cacheKey] = authType
			}
			if authTypeUsesNamedCredentialSecret(authType) {
				rowNeedsCipher = true
			}
		}
		if rowNeedsCipher {
			return true, nil
		}
	}

	return false, nil
}

func drainOnlyRowStillEligible(ctx context.Context, row packstate.ConnectorInboundWorkerPackRow, resolveDrainEligible namedCredentialDrainOnlyEligibilityResolver) (bool, error) {
	if !row.DrainOnly || resolveDrainEligible == nil {
		return true, nil
	}
	return resolveDrainEligible(ctx, row.Namespace, row.InstallID)
}

func firstDrainEligibilityResolver(resolvers []namedCredentialDrainOnlyEligibilityResolver) namedCredentialDrainOnlyEligibilityResolver {
	if len(resolvers) == 0 {
		return nil
	}
	return resolvers[0]
}

func namedCredentialAuthTypeCacheKey(row packstate.ConnectorInboundWorkerPackRow, credentialRef string) string {
	return fmt.Sprintf("%t\x00%s\x00%s\x00%s", row.DrainOnly, strings.TrimSpace(row.Namespace), strings.TrimSpace(row.InstallID), strings.TrimSpace(credentialRef))
}

func authTypeUsesNamedCredentialSecret(authType string) bool {
	switch strings.TrimSpace(authType) {
	case namedcredential.AuthTypeCustom:
		return false
	default:
		return true
	}
}

type requiredEnrichmentBatchReadClient struct {
	inner               connector_inbound.BatchReadClient
	requiredObjectKinds map[string]struct{}
}

func (c requiredEnrichmentBatchReadClient) BatchRead(ctx context.Context, namespace, objectKind string, sourceIDs []string) (map[string]json.RawMessage, error) {
	outcome, err := c.BatchReadOutcome(ctx, namespace, objectKind, sourceIDs)
	if err != nil {
		return nil, err
	}
	return connector_inbound.CollapseBatchReadOutcome(outcome, sourceIDs)
}

func (c requiredEnrichmentBatchReadClient) BatchReadOutcome(ctx context.Context, namespace, objectKind string, sourceIDs []string) (connector_inbound.BatchReadOutcome, error) {
	outcome, err := connector_inbound.ReadBatchOutcome(ctx, c.inner, namespace, objectKind, sourceIDs)
	if err != nil {
		return outcome, err
	}
	if _, required := c.requiredObjectKinds[strings.TrimSpace(objectKind)]; !required {
		return outcome, nil
	}
	for _, sourceID := range sourceIDs {
		if _, ok := outcome.Details[sourceID]; ok {
			continue
		}
		if outcome.SourceErrors[sourceID] != nil {
			continue
		}
		if outcome.SourceErrors == nil {
			outcome.SourceErrors = make(map[string]error)
		}
		outcome.SourceErrors[sourceID] = fmt.Errorf(
			"connector inbound: required enrichment missing for %s/%s sourceID %s",
			namespace, objectKind, sourceID,
		)
	}
	return outcome, nil
}