package controlplane

import (
	"fmt"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
)

// assertConnectorCapabilitiesHaveHandlers fails closed if a newly-authored
// connector pack declares a connector_capability without a same-manifest
// connector_inbound_handler_config component for the same (namespace, kind), or
// if it declares orphan handler config without a matching capability.
func assertConnectorCapabilitiesHaveHandlers(pack *packs.SourceBackedPack) error {
	return assertConnectorCapabilitiesHaveHandlersWithInstalledCompat(pack, false)
}

// assertConnectorCapabilitiesHaveHandlersWithInstalledCompat keeps installed
// legacy manifests verifiable only when this process has a real compiled legacy
// handler for the same binding. Without that fallback, readiness must fail
// closed instead of accepting an active pack that will 503 at runtime.
func assertConnectorCapabilitiesHaveHandlersWithInstalledCompat(pack *packs.SourceBackedPack, allowInstalledRowCompat bool) error {
	if pack == nil || pack.Category() != packs.ConnectorPack {
		return nil
	}
	capabilities := pack.ConnectorCapabilityDeclarations()
	configs := pack.ConnectorInboundHandlerConfigDeclarations()

	configByKey := make(map[string]packs.ConnectorInboundHandlerConfigComponentPayload, len(configs))
	for _, config := range configs {
		kind := connectorcapabilities.CapabilityKind(config.CapabilityKind)
		if !kind.IsKnown() {
			return fmt.Errorf("connector pack %q declares connector_inbound_handler_config %s:%s with unknown kind", pack.Name(), config.Namespace, config.CapabilityKind)
		}
		configByKey[connectorCapabilityGuardKey(config.Namespace, config.CapabilityKind)] = config
	}

	capabilityByKey := make(map[string]packs.ConnectorCapabilityComponentPayload, len(capabilities))
	for _, capability := range capabilities {
		kind := connectorcapabilities.CapabilityKind(capability.Kind)
		if !kind.IsKnown() {
			return fmt.Errorf("connector pack %q declares capability %s:%s with unknown kind", pack.Name(), capability.Namespace, capability.Kind)
		}
		key := connectorCapabilityGuardKey(capability.Namespace, capability.Kind)
		capabilityByKey[key] = capability
		if _, ok := configByKey[key]; !ok {
			if allowInstalledRowCompat && connectorcapabilities.HasLegacyCompiledHandler(capability.Namespace, kind) {
				continue
			}
			return fmt.Errorf(
				"connector pack %q declares capability %s:%s but no connector_inbound_handler_config component is declared in the same manifest; pack-data handler config is required",
				pack.Name(), capability.Namespace, capability.Kind,
			)
		}
	}

	for _, config := range configs {
		key := connectorCapabilityGuardKey(config.Namespace, config.CapabilityKind)
		if _, ok := capabilityByKey[key]; !ok {
			return fmt.Errorf(
				"connector pack %q declares orphan connector_inbound_handler_config %s:%s without a matching connector_capability component",
				pack.Name(), config.Namespace, config.CapabilityKind,
			)
		}
	}
	return nil
}

func connectorCapabilityGuardKey(namespace, kind string) string {
	return namespace + "\x00" + kind
}