package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"strconv"
	"strings"

	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/componentbehavior"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// connectorBindingDefinitionSourceTransaction attests that route-version
// selection and definition-set publication share one SQL transaction. A plain
// *sql.DB intentionally cannot satisfy this boundary: FOR UPDATE only protects
// the selected owner version until publication when the lock is transaction
// scoped.
type connectorBindingDefinitionSourceTransaction interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	Commit() error
	Rollback() error
}

type canonicalConnectorBindingDefinitionSource struct {
	kind                consumerv1.IntegrationBindingKind
	businessKey         string
	inboundEventKind    string
	canonicalDefinition []byte
}

type connectorBindingDefinitionSourceKey struct {
	kind        consumerv1.IntegrationBindingKind
	businessKey string
}

func sourceBackedPackConnectorBindingDefinitionSet(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	applied appliedAppPack,
	generation sourceBackedPackGeneration,
) (connectorbindingownerv1.DefinitionSet, error) {
	if ctx == nil || tx == nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: connector binding definition publication requires context, transaction, and a source-backed pack",
			ErrPackGenerationPrecondition,
		)
	}
	publicationTx, ok := tx.(connectorBindingDefinitionSourceTransaction)
	if !ok {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: connector binding definition publication requires one SQL transaction",
			ErrPackGenerationPrecondition,
		)
	}
	pack, coordinate, sources, err := canonicalSourceBackedPackConnectorBindingDefinitions(
		applied,
		generation,
	)
	if err != nil {
		return connectorbindingownerv1.DefinitionSet{}, err
	}
	versions := make(map[connectorBindingDefinitionSourceKey]uint64, len(sources))
	for _, source := range sources {
		key := connectorBindingDefinitionSourceKey{
			kind:        source.kind,
			businessKey: source.businessKey,
		}
		definitionVersion := uint64(1)
		if source.kind == consumerv1.BindingConnectorInboundRoute {
			definitionVersion, err = readExactConnectorInboundRouteDefinitionVersion(
				ctx,
				publicationTx,
				pack.Name(),
				generation.packInstallID,
				source.inboundEventKind,
			)
			if err != nil {
				return connectorbindingownerv1.DefinitionSet{}, err
			}
		}
		versions[key] = definitionVersion
	}
	return connectorBindingDefinitionSetFromCanonicalSources(
		pack.Name(),
		coordinate,
		sources,
		versions,
	)
}

// sourceBackedPackConnectorBindingDefinitionSetAtFrozenVersions rebuilds the
// exact source-authored definitions while preserving the immutable versions
// already committed by one validated stored set. Inbound route repair may
// advance the mutable live route version; an exact generation retry must never
// reinterpret that later route row as the historical generation's version.
// Non-route binding kinds remain fixed at their publication protocol version.
func sourceBackedPackConnectorBindingDefinitionSetAtFrozenVersions(
	applied appliedAppPack,
	generation sourceBackedPackGeneration,
	stored connectorbindingownerv1.DefinitionSet,
) (connectorbindingownerv1.DefinitionSet, error) {
	pack, coordinate, sources, err := canonicalSourceBackedPackConnectorBindingDefinitions(
		applied,
		generation,
	)
	if err != nil {
		return connectorbindingownerv1.DefinitionSet{}, err
	}
	if _, err := connectorbindingownerv1.ValidateDefinitionSet(stored); err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q stored connector binding definition set is invalid: %v",
			ErrPackGenerationConflict,
			pack.Name(),
			err,
		)
	}
	if stored.Header.DefinitionSetCoordinate != coordinate {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q stored connector binding definition set is bound to a different generation",
			ErrPackGenerationConflict,
			pack.Name(),
		)
	}
	storedVersions := make(map[connectorBindingDefinitionSourceKey]uint64, len(stored.Entries))
	for _, entry := range stored.Entries {
		storedVersions[connectorBindingDefinitionSourceKey{
			kind:        entry.Kind,
			businessKey: entry.BusinessKey,
		}] = entry.DefinitionVersion
	}

	versions := make(map[connectorBindingDefinitionSourceKey]uint64, len(sources))
	for _, source := range sources {
		key := connectorBindingDefinitionSourceKey{
			kind:        source.kind,
			businessKey: source.businessKey,
		}
		definitionVersion := uint64(1)
		if source.kind == consumerv1.BindingConnectorInboundRoute {
			var found bool
			definitionVersion, found = storedVersions[key]
			if !found {
				return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
					"%w: pack %q stored generation is missing the frozen version for connector binding definition %s/%s",
					ErrPackGenerationConflict,
					pack.Name(),
					source.kind,
					source.businessKey,
				)
			}
		}
		versions[key] = definitionVersion
	}
	return connectorBindingDefinitionSetFromCanonicalSources(
		pack.Name(),
		coordinate,
		sources,
		versions,
	)
}

// canonicalSourceBackedPackConnectorBindingDefinitions is the pure source
// compiler shared by new publication and exact retry verification. Version
// selection is deliberately separate: publication freezes live route state,
// while retries retain versions from immutable generation authority.
func canonicalSourceBackedPackConnectorBindingDefinitions(
	applied appliedAppPack,
	generation sourceBackedPackGeneration,
) (
	*packs.SourceBackedPack,
	connectorbindingownerv1.DefinitionSetCoordinate,
	[]canonicalConnectorBindingDefinitionSource,
	error,
) {
	pack, ok := applied.pack.(*packs.SourceBackedPack)
	if !ok || pack == nil {
		return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, fmt.Errorf(
			"%w: connector binding definition compilation requires a source-backed pack",
			ErrPackGenerationPrecondition,
		)
	}
	if generation.packName != pack.Name() ||
		generation.packInstallID == "" ||
		strings.TrimSpace(generation.packInstallID) != generation.packInstallID ||
		!generation.appliedPayloadHash.Valid ||
		generation.appliedPayloadHash.String != applied.payloadHash ||
		generation.desiredPayloadHash != generation.appliedPayloadHash {
		return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, fmt.Errorf(
			"%w: pack %q connector binding definitions are not bound to the exact generation",
			ErrPackGenerationPrecondition,
			pack.Name(),
		)
	}

	coordinate := connectorbindingownerv1.DefinitionSetCoordinate{
		OwnerPack:          generation.packName,
		PackInstallationID: generation.packInstallID,
		PackPayloadDigest:  generation.appliedPayloadHash.String,
	}
	components := pack.Components()
	sources := make([]canonicalConnectorBindingDefinitionSource, 0, len(components))
	seen := make(map[connectorBindingDefinitionSourceKey]struct{}, len(components))
	for index, component := range components {
		if component == nil {
			return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, fmt.Errorf(
				"%w: pack %q component %d is nil",
				ErrPackGenerationPrecondition,
				pack.Name(),
				index,
			)
		}
		bindingKind, closed := connectorBindingKindForPackComponent(component.Kind())
		if !closed {
			continue
		}
		key := connectorBindingDefinitionSourceKey{
			kind:        bindingKind,
			businessKey: component.BusinessKey(),
		}
		if _, duplicate := seen[key]; duplicate {
			return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, fmt.Errorf(
				"%w: pack %q source repeats connector binding definition %s/%s",
				ErrPackGenerationPrecondition,
				pack.Name(),
				bindingKind,
				component.BusinessKey(),
			)
		}
		seen[key] = struct{}{}
		inboundEventKind := ""
		if bindingKind == consumerv1.BindingConnectorInboundRoute {
			var err error
			inboundEventKind, err = connectorInboundRouteEventKindForDefinitionPublication(
				pack.Name(), component.BusinessKey(),
			)
			if err != nil {
				return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, err
			}
		}

		payload, err := component.ExpectedPayload()
		if err != nil {
			return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, fmt.Errorf(
				"%w: read pack %q connector binding definition %s/%s: %v",
				ErrPackGenerationPrecondition,
				pack.Name(),
				bindingKind,
				component.BusinessKey(),
				err,
			)
		}
		canonicalDefinition, err := componentbehavior.CanonicalJSONV1(payload)
		if err != nil {
			return nil, connectorbindingownerv1.DefinitionSetCoordinate{}, nil, fmt.Errorf(
				"%w: canonicalize pack %q connector binding definition %s/%s: %v",
				ErrPackGenerationPrecondition,
				pack.Name(),
				bindingKind,
				component.BusinessKey(),
				err,
			)
		}
		sources = append(sources, canonicalConnectorBindingDefinitionSource{
			kind:                bindingKind,
			businessKey:         component.BusinessKey(),
			inboundEventKind:    inboundEventKind,
			canonicalDefinition: canonicalDefinition,
		})
	}
	return pack, coordinate, sources, nil
}

func connectorBindingDefinitionSetFromCanonicalSources(
	packName string,
	coordinate connectorbindingownerv1.DefinitionSetCoordinate,
	sources []canonicalConnectorBindingDefinitionSource,
	versions map[connectorBindingDefinitionSourceKey]uint64,
) (connectorbindingownerv1.DefinitionSet, error) {
	if len(versions) != len(sources) {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q connector binding definition version coverage is incomplete",
			ErrPackGenerationPrecondition,
			packName,
		)
	}
	definitions := make([]connectorbindingownerv1.DefinitionEnvelope, 0, len(sources))
	for _, source := range sources {
		definitionVersion, found := versions[connectorBindingDefinitionSourceKey{
			kind:        source.kind,
			businessKey: source.businessKey,
		}]
		if !found {
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition %s/%s has no version",
				ErrPackGenerationPrecondition,
				packName,
				source.kind,
				source.businessKey,
			)
		}
		definition, err := connectorbindingownerv1.NewDefinitionEnvelope(
			connectorbindingownerv1.AuthoringInput{
				Identity: connectorbindingownerv1.Identity{
					Kind:               source.kind,
					OwnerPack:          coordinate.OwnerPack,
					PackInstallationID: coordinate.PackInstallationID,
					PackPayloadDigest:  coordinate.PackPayloadDigest,
					BusinessKey:        source.businessKey,
					DefinitionVersion:  definitionVersion,
				},
				CanonicalDefinition: source.canonicalDefinition,
			},
		)
		if err != nil {
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: build pack %q connector binding definition %s/%s: %v",
				ErrPackGenerationPrecondition,
				packName,
				source.kind,
				source.businessKey,
				err,
			)
		}
		definitions = append(definitions, definition)
	}

	set, err := connectorbindingownerv1.NewDefinitionSet(coordinate, definitions)
	if err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: build pack %q connector binding definition set: %v",
			ErrPackGenerationPrecondition,
			packName,
			err,
		)
	}
	return set, nil
}

func connectorBindingKindForPackComponent(
	kind packs.ComponentKind,
) (consumerv1.IntegrationBindingKind, bool) {
	switch kind {
	case packs.ComponentKindConnectorInboundRoute:
		return consumerv1.BindingConnectorInboundRoute, true
	case packs.ComponentKindConnectorCalloutProfile:
		return consumerv1.BindingConnectorCalloutProfile, true
	case packs.ComponentKindConnectorFieldMapping:
		return consumerv1.BindingConnectorFieldMapping, true
	case packs.ComponentKindConnectorPushContract:
		return consumerv1.BindingConnectorPushContract, true
	default:
		return "", false
	}
}

func connectorInboundRouteEventKindForDefinitionPublication(
	packName string,
	businessKey string,
) (string, error) {
	prefix := packName + "/"
	eventKind := strings.TrimPrefix(businessKey, prefix)
	if packName == "" ||
		strings.TrimSpace(packName) != packName ||
		!strings.HasPrefix(businessKey, prefix) ||
		eventKind == "" ||
		strings.TrimSpace(eventKind) != eventKind ||
		strings.TrimSpace(businessKey) != businessKey {
		return "", fmt.Errorf(
			"%w: pack %q connector inbound route business key %q does not identify one event kind",
			ErrPackGenerationPrecondition,
			packName,
			businessKey,
		)
	}
	return eventKind, nil
}

func readExactConnectorInboundRouteDefinitionVersion(
	ctx context.Context,
	tx connectorBindingDefinitionSourceTransaction,
	namespace string,
	packInstallID string,
	eventKind string,
) (uint64, error) {
	rows, err := tx.QueryContext(ctx, `SELECT Version
FROM _System_ConnectorInboundRoute
WHERE BINARY Namespace = BINARY ?
  AND BINARY InstalledPackId = BINARY ?
  AND BINARY EventKind = BINARY ?
  AND COALESCE(IsCurrent, 0) = 1
  AND COALESCE(IsDeleted, 0) = 0
ORDER BY Version
LIMIT 2
FOR UPDATE`, namespace, packInstallID, eventKind)
	if err != nil {
		return 0, fmt.Errorf(
			"resolve pack %q connector inbound route %q exact definition version: %w",
			namespace,
			eventKind,
			err,
		)
	}
	defer rows.Close()

	var (
		version uint64
		found   bool
	)
	for rows.Next() {
		var rawVersion []byte
		if err := rows.Scan(&rawVersion); err != nil {
			return 0, fmt.Errorf(
				"scan pack %q connector inbound route %q exact definition version: %w",
				namespace,
				eventKind,
				err,
			)
		}
		parsedVersion, err := parseCanonicalDefinitionUint64(rawVersion, false)
		if err != nil {
			return 0, fmt.Errorf(
				"%w: pack %q connector inbound route %q has malformed definition version: %v",
				ErrPackGenerationConflict,
				namespace,
				eventKind,
				err,
			)
		}
		if found {
			return 0, fmt.Errorf(
				"%w: pack %q connector inbound route %q installation %q has more than one exact current definition row",
				ErrPackGenerationConflict,
				namespace,
				eventKind,
				packInstallID,
			)
		}
		version = parsedVersion
		found = true
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf(
			"iterate pack %q connector inbound route %q exact definition versions: %w",
			namespace,
			eventKind,
			err,
		)
	}
	if !found {
		return 0, fmt.Errorf(
			"%w: pack %q connector inbound route %q installation %q has no exact current definition row",
			ErrPackGenerationConflict,
			namespace,
			eventKind,
			packInstallID,
		)
	}
	return version, nil
}

func parseCanonicalDefinitionUint64(raw []byte, allowZero bool) (uint64, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("value is empty")
	}
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil || strconv.FormatUint(value, 10) != string(raw) {
		return 0, fmt.Errorf("value is not a canonical unsigned decimal")
	}
	if !allowZero && value == 0 {
		return 0, fmt.Errorf("value must be positive")
	}
	return value, nil
}