package integrationservice

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

	appconnectors "github.com/pingcap-inc/nexuscrm-connectors/application/connectors"
	connectorcapabilities "github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	connectorrouting "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundrouting"
	connectorwebhooks "github.com/pingcap-inc/nexuscrm-connectors/interfaces/rest/connector_webhooks"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	flowcommandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	controlplanecatalog "github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
)

type integrationPackAuthority struct {
	ownerPack        string
	packInstallation string
	payloadDigest    string
}

type integrationRouteAuthority struct {
	pack           integrationPackAuthority
	eventKind      string
	routeVersion   uint64
	runtimeCommand catalogv1.RuntimeCommandBinding
}

type integrationProfileAuthority struct {
	pack           integrationPackAuthority
	profileName    string
	credentialRefs map[string]struct{}
}

// integrationServingAuthority is the immutable, process-lifetime projection
// of one authorized IntegrationBindingView. It does not become another
// metadata authority: every fact originates in the Catalog view, while the
// Integration-owned database is allowed only to resolve those exact identities.
type integrationServingAuthority struct {
	bindings    []catalogv1.IntegrationBinding
	packs       map[string]integrationPackAuthority
	routes      map[string]integrationRouteAuthority
	profiles    map[string]integrationProfileAuthority
	workerPacks map[string]struct{}
}

func newIntegrationServingAuthority(view catalogv1.IntegrationBindingView) (*integrationServingAuthority, error) {
	if err := catalogv1.ValidateIntegrationBindingView(view); err != nil {
		return nil, fmt.Errorf("invalid Integration binding authority: %w", err)
	}
	authority := &integrationServingAuthority{
		bindings:    cloneIntegrationBindingView(view).Bindings,
		packs:       make(map[string]integrationPackAuthority),
		routes:      make(map[string]integrationRouteAuthority),
		profiles:    make(map[string]integrationProfileAuthority),
		workerPacks: make(map[string]struct{}),
	}
	packByOwner := make(map[string]integrationPackAuthority)
	for _, binding := range view.Bindings {
		pack := integrationPackAuthority{
			ownerPack:        binding.OwnerPack,
			packInstallation: binding.PackInstallationID,
			payloadDigest:    binding.PackPayloadDigest,
		}
		packKey := integrationPackKey(pack.ownerPack, pack.packInstallation)
		if existing, found := authority.packs[packKey]; found && existing.payloadDigest != pack.payloadDigest {
			return nil, fmt.Errorf("Integration binding authority mixes payload digests for %s", packKey)
		}
		if existing, found := packByOwner[pack.ownerPack]; found &&
			(existing.packInstallation != pack.packInstallation || existing.payloadDigest != pack.payloadDigest) {
			return nil, fmt.Errorf("Integration binding authority mixes pack generations for %s", pack.ownerPack)
		}
		packByOwner[pack.ownerPack] = pack
		authority.packs[packKey] = pack
		switch binding.Kind {
		case catalogv1.BindingConnectorInboundRoute:
			if binding.RuntimeCommand == nil {
				// N-1 views may contain a route without a Runtime command, but the
				// independent M2 process may not serve it through a process-local path.
				continue
			}
			if !supportedIntegrationRuntimeCommand(*binding.RuntimeCommand) {
				return nil, fmt.Errorf(
					"Integration route binding %q selects unsupported Runtime command %q version %d",
					binding.BusinessKey,
					binding.RuntimeCommand.CommandKind,
					binding.RuntimeCommand.PayloadFormatVersion,
				)
			}
			eventKind, ok := integrationBindingLocalName(binding.OwnerPack, binding.BusinessKey)
			if !ok {
				return nil, fmt.Errorf("Integration route binding %q is outside owner %q", binding.BusinessKey, binding.OwnerPack)
			}
			key := integrationRouteKey(binding.OwnerPack, binding.PackInstallationID, eventKind)
			if _, duplicate := authority.routes[key]; duplicate {
				return nil, fmt.Errorf("Integration binding authority contains duplicate route %s", key)
			}
			authority.routes[key] = integrationRouteAuthority{
				pack: pack, eventKind: eventKind, routeVersion: binding.DefinitionVersion,
				runtimeCommand: *binding.RuntimeCommand,
			}
			authority.workerPacks[packKey] = struct{}{}
		case catalogv1.BindingConnectorCalloutProfile:
			profileName, ok := integrationBindingLocalName(binding.OwnerPack, binding.BusinessKey)
			if !ok {
				return nil, fmt.Errorf("Integration callout profile binding %q is outside owner %q", binding.BusinessKey, binding.OwnerPack)
			}
			key := integrationProfileKey(binding.OwnerPack, binding.PackInstallationID, profileName)
			if _, duplicate := authority.profiles[key]; duplicate {
				return nil, fmt.Errorf("Integration binding authority contains duplicate callout profile %s", key)
			}
			credentials := make(map[string]struct{}, len(binding.CredentialRefs))
			for _, credentialRef := range binding.CredentialRefs {
				credentials[credentialRef] = struct{}{}
			}
			authority.profiles[key] = integrationProfileAuthority{
				pack: pack, profileName: profileName, credentialRefs: credentials,
			}
		}
	}
	return authority, nil
}

func supportedIntegrationRuntimeCommand(binding catalogv1.RuntimeCommandBinding) bool {
	return binding.CommandKind == flowcommandv1.CommandKind &&
		(binding.PayloadFormatVersion == flowcommandv1.PayloadFormatVersion ||
			binding.PayloadFormatVersion == flowcommandv2.PayloadFormatVersion)
}

func (a *integrationServingAuthority) VerifyDatabase(ctx context.Context, db *sql.DB) error {
	if a == nil || db == nil {
		return errors.New("Integration serving authority and database are required")
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin Integration serving-authority verification: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	for _, binding := range a.bindings {
		if err := controlplanecatalog.VerifyIntegrationPackAuthority(ctx, tx, binding); err != nil {
			return fmt.Errorf(
				"verify Integration binding authority %s/%s (%s): %w",
				binding.OwnerPack,
				binding.BusinessKey,
				binding.PackInstallationID,
				err,
			)
		}
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit Integration serving-authority verification: %w", err)
	}
	return a.verifyRuntimeCommandRoutes(ctx, db)
}

// verifyRuntimeCommandRoutes closes the startup identity selected by Catalog:
// the exact connector route must resolve to a cross-pack Flow whose active
// version is the Runtime command version. Same-pack routes belong to the
// connectors process-local execution path and therefore cannot be admitted by
// the independent Runtime-command-only composition.
func (a *integrationServingAuthority) verifyRuntimeCommandRoutes(ctx context.Context, db *sql.DB) error {
	resolver, err := a.BindRouteResolver(connectorrouting.NewDBRouteResolver(db))
	if err != nil {
		return err
	}
	for _, route := range a.authorizedRoutes() {
		_, definitionRef, err := resolver.ResolveFlow(
			ctx,
			route.pack.ownerPack,
			route.pack.packInstallation,
			route.eventKind,
		)
		if err != nil {
			return fmt.Errorf(
				"verify Integration Runtime route %s/%s: %w",
				route.pack.ownerPack,
				route.eventKind,
				err,
			)
		}
		incarnation, crossPack, err := connectorwebhooks.ResolveCompanionFlowIncarnation(
			ctx,
			db,
			route.pack.ownerPack,
			definitionRef,
		)
		if err != nil {
			return fmt.Errorf(
				"verify Integration Runtime route %s/%s Flow identity: %w",
				route.pack.ownerPack,
				route.eventKind,
				err,
			)
		}
		if !crossPack {
			return fmt.Errorf(
				"Integration Runtime route %s/%s resolves to a process-local Flow instead of a cross-pack Runtime handoff",
				route.pack.ownerPack,
				route.eventKind,
			)
		}
		if incarnation.CompanionFlowVersionID != route.runtimeCommand.DefinitionVersion {
			return fmt.Errorf(
				"Integration Runtime route %s/%s active Flow version %q does not match Catalog command version %q",
				route.pack.ownerPack,
				route.eventKind,
				incarnation.CompanionFlowVersionID,
				route.runtimeCommand.DefinitionVersion,
			)
		}
	}
	return nil
}

func (a *integrationServingAuthority) BindRouteResolver(base connectorrouting.RouteResolver) (connectorrouting.RouteResolver, error) {
	if a == nil || base == nil {
		return nil, errors.New("Integration route authority and resolver are required")
	}
	return &integrationRouteResolver{authority: a, base: base}, nil
}

func (a *integrationServingAuthority) RuntimeServingSelection(base connectorrouting.RouteResolver) (appconnectors.RuntimeServingSelection, error) {
	resolver, err := a.BindRouteResolver(base)
	if err != nil {
		return appconnectors.RuntimeServingSelection{}, err
	}
	routesByPack := make(map[string][]appconnectors.RuntimeServingInboundRoute)
	for _, route := range a.authorizedRoutes() {
		packKey := integrationPackKey(route.pack.ownerPack, route.pack.packInstallation)
		routesByPack[packKey] = append(routesByPack[packKey], appconnectors.RuntimeServingInboundRoute{
			EventKind:    route.eventKind,
			RouteVersion: route.routeVersion,
		})
	}
	profilesByPack := make(map[string][]appconnectors.RuntimeServingCalloutProfile)
	for _, profile := range a.profiles {
		packKey := integrationPackKey(profile.pack.ownerPack, profile.pack.packInstallation)
		if _, serves := a.workerPacks[packKey]; serves {
			credentialRefs := make([]string, 0, len(profile.credentialRefs))
			for credentialRef := range profile.credentialRefs {
				credentialRefs = append(credentialRefs, credentialRef)
			}
			sort.Strings(credentialRefs)
			profilesByPack[packKey] = append(profilesByPack[packKey], appconnectors.RuntimeServingCalloutProfile{
				Name:           profile.profileName,
				CredentialRefs: credentialRefs,
			})
		}
	}
	for packKey := range profilesByPack {
		sort.Slice(profilesByPack[packKey], func(i, j int) bool {
			return profilesByPack[packKey][i].Name < profilesByPack[packKey][j].Name
		})
	}
	selection := appconnectors.RuntimeServingSelection{RouteResolver: resolver}
	for _, pack := range a.servingPacks() {
		packKey := integrationPackKey(pack.ownerPack, pack.packInstallation)
		selection.Packs = append(selection.Packs, appconnectors.RuntimeServingPack{
			Namespace:       pack.ownerPack,
			InstalledPackID: pack.packInstallation,
			PayloadDigest:   pack.payloadDigest,
			InboundRoutes:   append([]appconnectors.RuntimeServingInboundRoute(nil), routesByPack[packKey]...),
			CalloutProfiles: append([]appconnectors.RuntimeServingCalloutProfile(nil), profilesByPack[packKey]...),
		})
	}
	return selection, nil
}

func (a *integrationServingAuthority) CapabilityGate(base connectorcapabilities.CapabilityGate) connectorcapabilities.CapabilityGate {
	return func(ctx context.Context, namespace string, kind connectorcapabilities.CapabilityKind, packInstallationID string) (bool, error) {
		if a == nil || base == nil {
			return false, errors.New("Integration capability authority is unavailable")
		}
		if _, allowed := a.workerPacks[integrationPackKey(namespace, packInstallationID)]; !allowed {
			return false, nil
		}
		return base(ctx, namespace, kind, packInstallationID)
	}
}

func (a *integrationServingAuthority) authorizedRoutes() []integrationRouteAuthority {
	result := make([]integrationRouteAuthority, 0, len(a.routes))
	for _, route := range a.routes {
		result = append(result, route)
	}
	sort.Slice(result, func(i, j int) bool {
		return integrationRouteKey(result[i].pack.ownerPack, result[i].pack.packInstallation, result[i].eventKind) <
			integrationRouteKey(result[j].pack.ownerPack, result[j].pack.packInstallation, result[j].eventKind)
	})
	return result
}

func (a *integrationServingAuthority) servingPacks() []integrationPackAuthority {
	result := make([]integrationPackAuthority, 0, len(a.workerPacks))
	for key := range a.workerPacks {
		result = append(result, a.packs[key])
	}
	sort.Slice(result, func(i, j int) bool {
		return integrationPackKey(result[i].ownerPack, result[i].packInstallation) <
			integrationPackKey(result[j].ownerPack, result[j].packInstallation)
	})
	return result
}

type integrationRouteResolver struct {
	authority *integrationServingAuthority
	base      connectorrouting.RouteResolver
}

func (r *integrationRouteResolver) ResolveFlow(ctx context.Context, namespace, packInstallationID, eventKind string) (int64, string, error) {
	selected, found := r.authority.routes[integrationRouteKey(namespace, packInstallationID, eventKind)]
	if !found {
		return 0, "", connectorrouting.ErrRouteNotDeclared
	}
	version, definition, err := r.base.ResolveFlow(ctx, namespace, packInstallationID, eventKind)
	if err != nil {
		return 0, "", err
	}
	if version < 1 || uint64(version) != selected.routeVersion || definition != selected.runtimeCommand.DefinitionRef {
		return 0, "", errors.New("Integration route projection drifted from Catalog authority")
	}
	return version, definition, nil
}

func (r *integrationRouteResolver) ResolveFlowByVersion(ctx context.Context, packInstallationID, eventKind string, version int64) (string, error) {
	selected, found := r.routeByInstall(packInstallationID, eventKind, version)
	if !found {
		return "", connectorrouting.ErrRouteNotDeclared
	}
	definition, err := r.base.ResolveFlowByVersion(ctx, packInstallationID, eventKind, version)
	if err != nil {
		return "", err
	}
	if definition != selected.runtimeCommand.DefinitionRef {
		return "", errors.New("Integration route projection drifted from Catalog authority")
	}
	return definition, nil
}

func (r *integrationRouteResolver) ResolveExecutionSpecByVersion(ctx context.Context, packInstallationID, eventKind string, version int64) (connectorrouting.RouteExecutionSpec, error) {
	selected, found := r.routeByInstall(packInstallationID, eventKind, version)
	if !found {
		return connectorrouting.RouteExecutionSpec{}, connectorrouting.ErrRouteNotDeclared
	}
	spec, err := r.base.ResolveExecutionSpecByVersion(ctx, packInstallationID, eventKind, version)
	if err != nil {
		return connectorrouting.RouteExecutionSpec{}, err
	}
	if spec.FlowDeveloperName != selected.runtimeCommand.DefinitionRef {
		return connectorrouting.RouteExecutionSpec{}, errors.New("Integration route projection drifted from Catalog authority")
	}
	return spec, nil
}

func (r *integrationRouteResolver) routeByInstall(packInstallationID, eventKind string, version int64) (integrationRouteAuthority, bool) {
	if version < 1 {
		return integrationRouteAuthority{}, false
	}
	for _, selected := range r.authority.routes {
		if selected.pack.packInstallation == packInstallationID && selected.eventKind == strings.ToLower(strings.TrimSpace(eventKind)) && selected.routeVersion == uint64(version) {
			return selected, true
		}
	}
	return integrationRouteAuthority{}, false
}

func integrationBindingLocalName(ownerPack, businessKey string) (string, bool) {
	prefix := ownerPack + "/"
	if !strings.HasPrefix(businessKey, prefix) {
		return "", false
	}
	name := strings.TrimSpace(strings.TrimPrefix(businessKey, prefix))
	return name, name != "" && !strings.Contains(name, "/")
}

func integrationPackKey(ownerPack, packInstallationID string) string {
	return strings.TrimSpace(ownerPack) + "\x00" + strings.TrimSpace(packInstallationID)
}

func integrationRouteKey(ownerPack, packInstallationID, eventKind string) string {
	return integrationPackKey(ownerPack, packInstallationID) + "\x00" + strings.ToLower(strings.TrimSpace(eventKind))
}

func integrationProfileKey(ownerPack, packInstallationID, profileName string) string {
	return integrationPackKey(ownerPack, packInstallationID) + "\x00" + strings.ToLower(strings.TrimSpace(profileName))
}

var _ connectorrouting.RouteResolver = (*integrationRouteResolver)(nil)