package connector_inbound_handler_config

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"

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

// MissingHandlerConfigError reports a live installed connector capability whose
// source declares the capability but whose current pack incarnation has no
// persisted handler-config row. ApplyAll may repair this by re-applying the
// current source pack; verify/startup still fail closed and never write rows.
type MissingHandlerConfigError struct {
	Namespace         string
	CapabilityKind    connectorcapabilities.CapabilityKind
	InstalledPackName string
	PackInstallID     string
}

func (e *MissingHandlerConfigError) Error() string {
	if e == nil {
		return "connector inbound handler config missing"
	}
	return fmt.Sprintf(
		"connector inbound handler config missing for connector capability %s:%s in active connector pack %s (packInstallId=%s)",
		e.Namespace,
		e.CapabilityKind,
		e.InstalledPackName,
		e.PackInstallID,
	)
}

// IsMissingHandlerConfig reports whether err wraps MissingHandlerConfigError.
func IsMissingHandlerConfig(err error) bool {
	var target *MissingHandlerConfigError
	return errors.As(err, &target)
}

// BuildCapabilityHandler validates one handler config blob and returns the
// process-local handler plus a canonical JSON representation suitable for DB
// storage and drift comparison.
func BuildCapabilityHandler(namespace string, kind connectorcapabilities.CapabilityKind, rawConfig []byte) (any, string, error) {
	if strings.TrimSpace(namespace) == "" {
		return nil, "", fmt.Errorf("connector inbound handler config: namespace is required")
	}
	if !kind.IsKnown() {
		return nil, "", fmt.Errorf("%w: kind=%q", connectorcapabilities.ErrUnknownCapabilityKind, kind)
	}
	if len(bytes.TrimSpace(rawConfig)) == 0 {
		return nil, "", fmt.Errorf("connector inbound handler config: config JSON is required for %s/%s", namespace, kind)
	}
	switch kind {
	case connectorcapabilities.SignatureVerifier:
		cfg, canonical, err := decodeSignatureSchemeConfigSet(rawConfig)
		if err != nil {
			return nil, "", err
		}
		return cfg.Build(namespace, canonical)
	case connectorcapabilities.EnvelopeParser:
		cfg, canonical, err := decodeEnvelopeMappingConfigSet(namespace, rawConfig)
		if err != nil {
			return nil, "", err
		}
		return cfg.Build(namespace, canonical)
	default:
		return nil, "", fmt.Errorf("%w: kind=%q", connectorcapabilities.ErrUnknownCapabilityKind, kind)
	}
}

type signatureSchemeConfigSet struct {
	Default         *connector_inbound.SignatureSchemeConfig
	EndpointConfigs map[string]connector_inbound.SignatureSchemeConfig
}

func (s signatureSchemeConfigSet) Build(namespace, canonical string) (any, string, error) {
	if s.Default != nil {
		handler, err := connector_inbound.NewSignatureVerifier(*s.Default)
		if err != nil {
			return nil, "", fmt.Errorf("build signature verifier for %s: %w", namespace, err)
		}
		return handler, canonical, nil
	}
	handlers := make(map[string]connectorcapabilities.SignatureVerifierHandler, len(s.EndpointConfigs))
	for endpoint, cfg := range s.EndpointConfigs {
		handler, err := connector_inbound.NewSignatureVerifier(cfg)
		if err != nil {
			return nil, "", fmt.Errorf("build signature verifier for %s endpoint %q: %w", namespace, endpoint, err)
		}
		handlers[endpoint] = handler
	}
	return &endpointSignatureVerifier{namespace: namespace, handlers: handlers}, canonical, nil
}

type envelopeMappingConfigSet struct {
	Default                   *connector_inbound.EnvelopeMappingConfig
	DefaultComposites         []*connector_inbound.IdempotencyKeyComposite
	EndpointConfigs           map[string]connector_inbound.EnvelopeMappingConfig
	EndpointCompositeSidecars map[string][]*connector_inbound.IdempotencyKeyComposite
}

func (s envelopeMappingConfigSet) Build(namespace, canonical string) (any, string, error) {
	if s.Default != nil {
		handler, err := connector_inbound.NewEnvelopeParserWithIdempotencyComposites(namespace, *s.Default, s.DefaultComposites)
		if err != nil {
			return nil, "", fmt.Errorf("build envelope parser for %s: %w", namespace, err)
		}
		return handler, canonical, nil
	}
	handlers := make(map[string]connectorcapabilities.EnvelopeParserHandler, len(s.EndpointConfigs))
	for endpoint, cfg := range s.EndpointConfigs {
		handler, err := connector_inbound.NewEnvelopeParserWithIdempotencyComposites(namespace, cfg, s.EndpointCompositeSidecars[endpoint])
		if err != nil {
			return nil, "", fmt.Errorf("build envelope parser for %s endpoint %q: %w", namespace, endpoint, err)
		}
		handlers[endpoint] = handler
	}
	return &endpointEnvelopeParser{namespace: namespace, handlers: handlers}, canonical, nil
}

type endpointSignatureVerifier struct {
	namespace string
	handlers  map[string]connectorcapabilities.SignatureVerifierHandler
}

func (v *endpointSignatureVerifier) Verify(ctx context.Context, input *connectorcapabilities.VerifierInput) error {
	if input == nil {
		return fmt.Errorf("signature verifier input is nil for namespace=%s", v.namespace)
	}
	endpoint := canonicalEndpointName(input.SecretName)
	if endpoint == "" {
		return fmt.Errorf("signature verifier config missing endpoint identity for namespace=%s", v.namespace)
	}
	handler, ok := v.handlers[endpoint]
	if !ok {
		return fmt.Errorf("signature verifier config missing for namespace=%s endpoint=%q", v.namespace, endpoint)
	}
	return handler.Verify(ctx, input)
}

type endpointEnvelopeParser struct {
	namespace string
	handlers  map[string]connectorcapabilities.EnvelopeParserHandler
}

func (p *endpointEnvelopeParser) Parse(ctx context.Context, input *connectorcapabilities.ParserInput) ([]connectorcapabilities.ConnectorEvent, error) {
	if input == nil {
		return nil, fmt.Errorf("envelope parser input is nil for namespace=%s: %w", p.namespace, connectorcapabilities.ErrParserConfigDrift)
	}
	endpoint := canonicalEndpointName(input.SecretName)
	if endpoint == "" {
		return nil, fmt.Errorf("envelope parser config missing endpoint identity for namespace=%s: %w", p.namespace, connectorcapabilities.ErrParserConfigDrift)
	}
	handler, ok := p.handlers[endpoint]
	if !ok {
		return nil, fmt.Errorf("envelope parser config missing for namespace=%s endpoint=%q: %w", p.namespace, endpoint, connectorcapabilities.ErrParserConfigDrift)
	}
	return handler.Parse(ctx, input)
}

func canonicalEndpointName(name string) string {
	return strings.ToLower(strings.TrimSpace(name))
}

func decodeSignatureSchemeConfigSet(raw []byte) (signatureSchemeConfigSet, string, error) {
	cfgSet, canonical, err := configwire.DecodeSignatureSchemeConfigSet(raw)
	if err != nil {
		return signatureSchemeConfigSet{}, "", err
	}
	convert := func(cfg configwire.SignatureSchemeConfig) connector_inbound.SignatureSchemeConfig {
		return connector_inbound.SignatureSchemeConfig{
			Scheme:                             connector_inbound.SchemeName(cfg.Scheme),
			SigningTemplate:                    cfg.SigningTemplate,
			TimestampHeader:                    cfg.TimestampHeader,
			SignatureHeader:                    cfg.SignatureHeader,
			SignatureEncoding:                  connector_inbound.SignatureEncoding(cfg.SignatureEncoding),
			DeliveryIdentityBodyPath:           cfg.DeliveryIdentityBodyPath,
			DeliveryIdentityCollectionPath:     cfg.DeliveryIdentityCollectionPath,
			DeliveryIdentityCollectionKeyPaths: append([]string(nil), cfg.DeliveryIdentityCollectionKeyPaths...),
			ReceiptHeaderAllowlist:             append([]string(nil), cfg.ReceiptHeaderAllowlist...),
			SkewWindow:                         cfg.SkewWindow,
			SecretName:                         cfg.SecretName,
			MethodHeader:                       cfg.MethodHeader,
			URIHeader:                          cfg.URIHeader,
		}
	}
	if cfgSet.Default != nil {
		defaultCfg := convert(*cfgSet.Default)
		return signatureSchemeConfigSet{Default: &defaultCfg}, canonical, nil
	}
	endpointConfigs := make(map[string]connector_inbound.SignatureSchemeConfig, len(cfgSet.EndpointConfigs))
	for endpoint, cfg := range cfgSet.EndpointConfigs {
		endpointConfigs[endpoint] = convert(cfg)
	}
	return signatureSchemeConfigSet{EndpointConfigs: endpointConfigs}, canonical, nil
}

func decodeEnvelopeMappingConfigSet(namespace string, raw []byte) (envelopeMappingConfigSet, string, error) {
	cfgSet, compositeSidecars, canonical, err := configwire.DecodeEnvelopeMappingConfigSetWithComposites(namespace, raw)
	if err != nil {
		return envelopeMappingConfigSet{}, "", err
	}
	convert := func(cfg configwire.EnvelopeMappingConfig, composites []*configwire.IdempotencyKeyComposite) (connector_inbound.EnvelopeMappingConfig, []*connector_inbound.IdempotencyKeyComposite) {
		out := connector_inbound.EnvelopeMappingConfig{
			EventArrayPath:        cfg.EventArrayPath,
			EventObjectPath:       cfg.EventObjectPath,
			InboundProcessingMode: connector_inbound.InboundProcessingMode(cfg.InboundProcessingMode),
			Rules:                 make([]connector_inbound.EnvelopeMappingRule, 0, len(cfg.Rules)),
		}
		var convertedComposites []*connector_inbound.IdempotencyKeyComposite
		if len(composites) != 0 {
			convertedComposites = make([]*connector_inbound.IdempotencyKeyComposite, len(composites))
			for i, composite := range composites {
				if composite == nil {
					continue
				}
				convertedComposite := &connector_inbound.IdempotencyKeyComposite{
					Version:  composite.Version,
					Encoding: composite.Encoding,
					Parts:    make([]connector_inbound.IdempotencyKeyPart, 0, len(composite.Parts)),
				}
				for _, part := range composite.Parts {
					convertedComposite.Parts = append(convertedComposite.Parts, connector_inbound.IdempotencyKeyPart{Path: part.Path})
				}
				convertedComposites[i] = convertedComposite
			}
		}
		for _, rule := range cfg.Rules {
			converted := connector_inbound.EnvelopeMappingRule{
				MatchPath:            rule.MatchPath,
				MatchEquals:          rule.MatchEquals,
				EventTypeFromPath:    rule.EventTypeFromPath,
				EventTypeLiteral:     rule.EventTypeLiteral,
				NamespaceFromCtx:     rule.NamespaceFromCtx,
				ObjectKind:           rule.ObjectKind,
				SourceIDPath:         rule.SourceIDPath,
				SourceLaneObjectKind: rule.SourceLaneObjectKind,
				SourceLaneIDPath:     rule.SourceLaneIDPath,
				IdempotencyKeyPath:   rule.IdempotencyKeyPath,
				PayloadProjection:    make([]connector_inbound.PayloadField, 0, len(rule.PayloadProjection)),
			}
			if rule.MatchAll != nil {
				converted.MatchAll = make([]connector_inbound.MatchCondition, 0, len(rule.MatchAll))
				for _, cond := range rule.MatchAll {
					converted.MatchAll = append(converted.MatchAll, connector_inbound.MatchCondition{
						Path:   cond.Path,
						Equals: cond.Equals,
					})
				}
			}
			for _, field := range rule.PayloadProjection {
				converted.PayloadProjection = append(converted.PayloadProjection, connector_inbound.PayloadField{
					Name: field.Name,
					Path: field.Path,
					Mode: connector_inbound.PayloadFieldMode(field.Mode),
				})
			}
			out.Rules = append(out.Rules, converted)
		}
		return out, convertedComposites
	}
	if cfgSet.Default != nil {
		defaultCfg, defaultComposites := convert(*cfgSet.Default, compositeSidecars.Default)
		return envelopeMappingConfigSet{Default: &defaultCfg, DefaultComposites: defaultComposites}, canonical, nil
	}
	endpointConfigs := make(map[string]connector_inbound.EnvelopeMappingConfig, len(cfgSet.EndpointConfigs))
	endpointComposites := make(map[string][]*connector_inbound.IdempotencyKeyComposite, len(cfgSet.EndpointConfigs))
	for endpoint, cfg := range cfgSet.EndpointConfigs {
		endpointConfigs[endpoint], endpointComposites[endpoint] = convert(cfg, compositeSidecars.EndpointConfigs[endpoint])
	}
	return envelopeMappingConfigSet{EndpointConfigs: endpointConfigs, EndpointCompositeSidecars: endpointComposites}, canonical, nil
}

// RegisterActiveHandlerConfigs rebuilds the in-memory registry projection from
// active, hash-stable connector pack handler config rows. Startup must call this
// before binding the default capability gate and accepting webhook traffic. Active
// installed rows from pre-handler-config manifests are grandfathered only when a
// real compiled legacy handler can be bound to the current pack incarnation.
// Without a config row or fallback handler, startup fails closed instead of
// accepting a pack that cannot dispatch inbound webhook traffic.
func RegisterActiveHandlerConfigs(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry) error {
	return registerHandlerConfigs(ctx, db, registry, "", "", []string{"active"})
}

// RegisterHandlerConfigsForPack rebuilds handler projections for one connector
// pack while it is being installed or after it is active. Controlplane calls
// this before marking a pack active so semantic config errors fail the install
// before runtime dispatch can pass the DB gate. New installs are still guarded
// by the forward-authoring preflight; already-existing manifests without config
// rows require a real compiled legacy fallback in the active registry.
func RegisterHandlerConfigsForPack(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry, packName string) error {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return fmt.Errorf("connector inbound handler config registrar: packName is required")
	}
	return registerHandlerConfigs(ctx, db, registry, packName, "", []string{"installing", "active"})
}

// RegisterActiveHandlerConfigsForPackInstall registers handler projections for
// one caller-selected active connector-pack generation. Independent serving
// compositions use this exact entry point instead of enumerating active or
// installing generations.
func RegisterActiveHandlerConfigsForPackInstall(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry, packName, packInstallID string) error {
	packName = strings.TrimSpace(packName)
	packInstallID = strings.TrimSpace(packInstallID)
	if packName == "" || packInstallID == "" {
		return fmt.Errorf("connector inbound handler config registrar: packName and packInstallID are required")
	}
	return registerHandlerConfigs(ctx, db, registry, packName, packInstallID, []string{"active"})
}

// VerifyHandlerConfigsForPack scans active, hash-stable live rows for one
// connector pack and runs the same semantic BuildCapabilityHandler path used by
// registry registration so setup verify fails closed before startup when the
// persisted config cannot build. Missing config rows on already-installed legacy
// manifests pass only when a real compiled legacy handler can be bound to the
// current pack incarnation.
func VerifyHandlerConfigsForPack(ctx context.Context, db *sql.DB, packName string) error {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return fmt.Errorf("connector inbound handler config verifier: packName is required")
	}
	return visitHandlerConfigs(ctx, db, nil, packName, "", []string{"active"}, func(namespace string, kind connectorcapabilities.CapabilityKind, configJSON string, _ string) error {
		if _, _, err := BuildCapabilityHandler(namespace, kind, []byte(configJSON)); err != nil {
			return fmt.Errorf("build connector inbound handler config %s/%s: %w", namespace, kind, err)
		}
		return nil
	})
}

func registerHandlerConfigs(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry, packName, packInstallID string, statuses []string) error {
	if registry == nil {
		registry = connectorcapabilities.Default()
	}
	return visitHandlerConfigs(ctx, db, registry, packName, packInstallID, statuses, func(namespace string, kind connectorcapabilities.CapabilityKind, configJSON string, packInstallID string) error {
		handler, _, err := BuildCapabilityHandler(namespace, kind, []byte(configJSON))
		if err != nil {
			return fmt.Errorf("build connector inbound handler config %s/%s: %w", namespace, kind, err)
		}
		if err := registry.UpsertCapabilityForPackInstall(namespace, kind, packInstallID, handler); err != nil {
			return fmt.Errorf("register connector inbound handler config %s/%s: %w", namespace, kind, err)
		}
		return nil
	})
}

func visitHandlerConfigs(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry, packName, packInstallID string, statuses []string, visit func(namespace string, kind connectorcapabilities.CapabilityKind, configJSON string, packInstallID string) error) error {
	if db == nil {
		return fmt.Errorf("database is required")
	}
	statusPredicate, hashPredicate := handlerConfigPackPredicates(statuses)
	if err := ensureDeclaredCapabilitiesHaveHandlerConfigs(ctx, db, registry, packName, packInstallID, statusPredicate, hashPredicate); err != nil {
		return err
	}
	packPredicate := ""
	args := []any{}
	if packName != "" {
		packPredicate = "   AND p.Name = ?\n"
		args = append(args, packName)
	}
	if packInstallID != "" {
		packPredicate += "   AND p.PackInstallId = ?\n"
		args = append(args, packInstallID)
	}
	rows, err := db.QueryContext(ctx, `
SELECT c.Namespace, c.CapabilityKind, c.ConfigJSON, c.InstalledPackId
  FROM _System_ConnectorInboundHandlerConfig c
  JOIN _System_InstalledPack p
    ON p.PackInstallId = c.InstalledPackId
   AND p.Name = c.Namespace
  JOIN _System_ConnectorCapabilityDeclaration d
    ON d.InstalledPackName = p.Name
   AND d.Namespace = c.Namespace
   AND d.Kind = c.CapabilityKind
 WHERE p.Category = 'connector-pack'
   AND `+statusPredicate+`
   AND COALESCE(p.IsDeleted, 0) = 0
   AND COALESCE(p.PackInstallId, '') <> ''
`+packPredicate+`
`+hashPredicate+`
   AND COALESCE(c.IsDeleted, 0) = 0
   AND COALESCE(d.IsDeleted, 0) = 0
 ORDER BY c.Namespace, c.CapabilityKind`, args...)
	if err != nil {
		return fmt.Errorf("scan connector inbound handler configs: %w", err)
	}
	defer rows.Close()

	for rows.Next() {
		var namespace string
		var kindRaw string
		var configJSON string
		var packInstallID string
		if err := rows.Scan(&namespace, &kindRaw, &configJSON, &packInstallID); err != nil {
			return fmt.Errorf("scan connector inbound handler config row: %w", err)
		}
		if err := visit(namespace, connectorcapabilities.CapabilityKind(strings.TrimSpace(kindRaw)), configJSON, packInstallID); err != nil {
			return err
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate connector inbound handler configs: %w", err)
	}
	return nil
}

func handlerConfigPackPredicates(statuses []string) (string, string) {
	statusPredicate := "p.Status = 'active'"
	hashPredicate := `   AND p.AppliedPayloadHash IS NOT NULL
   AND p.AppliedPayloadHash <> ''
   AND p.DesiredPayloadHash IS NOT NULL
   AND p.DesiredPayloadHash <> ''
   AND p.AppliedPayloadHash = p.DesiredPayloadHash`
	if len(statuses) == 2 && statuses[0] == "installing" && statuses[1] == "active" {
		statusPredicate = "p.Status IN ('installing', 'active')"
		hashPredicate = `   AND p.DesiredPayloadHash IS NOT NULL
   AND p.DesiredPayloadHash <> ''
   AND (
       p.Status = 'installing'
       OR (
           p.AppliedPayloadHash IS NOT NULL
           AND p.AppliedPayloadHash <> ''
           AND p.AppliedPayloadHash = p.DesiredPayloadHash
       )
   )`
	}
	return statusPredicate, hashPredicate
}

func ensureDeclaredCapabilitiesHaveHandlerConfigs(ctx context.Context, db *sql.DB, registry *connectorcapabilities.Registry, packName, packInstallID, statusPredicate, hashPredicate string) error {
	packPredicate := ""
	args := []any{}
	if packName != "" {
		packPredicate = "   AND p.Name = ?\n"
		args = append(args, packName)
	}
	if packInstallID != "" {
		packPredicate += "   AND p.PackInstallId = ?\n"
		args = append(args, packInstallID)
	}
	rows, err := db.QueryContext(ctx, `
SELECT p.Name, d.Namespace, d.Kind, p.PackInstallId
  FROM _System_ConnectorCapabilityDeclaration d
  JOIN _System_InstalledPack p
    ON p.Name = d.InstalledPackName
  LEFT JOIN _System_ConnectorInboundHandlerConfig c
    ON c.InstalledPackId = p.PackInstallId
   AND c.Namespace = d.Namespace
   AND c.CapabilityKind = d.Kind
   AND COALESCE(c.IsDeleted, 0) = 0
 WHERE p.Category = 'connector-pack'
   AND `+statusPredicate+`
   AND COALESCE(p.IsDeleted, 0) = 0
   AND COALESCE(p.PackInstallId, '') <> ''
`+packPredicate+`
`+hashPredicate+`
   AND COALESCE(d.IsDeleted, 0) = 0
   AND c.Id IS NULL
 ORDER BY p.Name, d.Namespace, d.Kind`, args...)
	if err != nil {
		return fmt.Errorf("check connector inbound handler config coverage: %w", err)
	}
	defer rows.Close()

	for rows.Next() {
		var installedPackName string
		var namespace string
		var kindRaw string
		var packInstallID string
		if err := rows.Scan(&installedPackName, &namespace, &kindRaw, &packInstallID); err != nil {
			return fmt.Errorf("scan connector inbound handler config coverage row: %w", err)
		}
		kind := connectorcapabilities.CapabilityKind(strings.TrimSpace(kindRaw))
		if effectiveRegistry(registry).BindLegacyHandlerToPackInstall(namespace, kind, packInstallID) {
			continue
		}
		return &MissingHandlerConfigError{
			Namespace:         namespace,
			CapabilityKind:    kind,
			InstalledPackName: installedPackName,
			PackInstallID:     packInstallID,
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate connector inbound handler config coverage rows: %w", err)
	}
	return nil
}

func effectiveRegistry(registry *connectorcapabilities.Registry) *connectorcapabilities.Registry {
	if registry != nil {
		return registry
	}
	return connectorcapabilities.Default()
}