package serving

import (
	"context"
	"errors"
	"fmt"
	"sort"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/bindingowner"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/credentialgeneration"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/serviceconfiguration"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
)

var (
	// ErrProfiledPreparedEpochNotFound reports that the exact local candidate
	// selected for assessment does not exist. Absence is not an external-owner
	// rejection and therefore never creates a rejection decision.
	ErrProfiledPreparedEpochNotFound = errors.New("connector serving: profiled prepared epoch not found")
	// ErrExternalAuthorityUnavailable reports that an external owner could not
	// provide a definitive fact. The underlying dependency error is retained so
	// retry policy can classify it without treating the candidate as rejected.
	ErrExternalAuthorityUnavailable = errors.New("connector serving: external authority unavailable")
)

// ProfiledExternalAuthorityDisposition is the closed result of one read-only
// fresh-owner assessment. It deliberately does not call an exact match
// "verified": this slice writes no ServingEpoch lifecycle state.
type ProfiledExternalAuthorityDisposition string

const (
	ProfiledExternalAuthorityExactMatch ProfiledExternalAuthorityDisposition = "exact_match"
	ProfiledExternalAuthorityRejected   ProfiledExternalAuthorityDisposition = "rejected"
)

// ProfiledExternalAuthorityRejectionReason identifies the authority boundary
// that supplied a definitive mismatch. Dependency failures never become a
// rejection reason.
type ProfiledExternalAuthorityRejectionReason string

const (
	ProfiledExternalAuthorityRejectedByCatalog              ProfiledExternalAuthorityRejectionReason = "catalog"
	ProfiledExternalAuthorityRejectedByBindingOwner         ProfiledExternalAuthorityRejectionReason = "binding_owner"
	ProfiledExternalAuthorityRejectedByServiceConfiguration ProfiledExternalAuthorityRejectionReason = "service_configuration"
	ProfiledExternalAuthorityRejectedByCredentialGeneration ProfiledExternalAuthorityRejectionReason = "credential_generation"
	ProfiledExternalAuthorityRejectedByReconstruction       ProfiledExternalAuthorityRejectionReason = "reconstruction"
)

// ProfiledExternalAuthorityAssessment exposes only non-authorizing scalar
// evidence. It contains no candidate graph and is intentionally unsuitable as
// input to a later state transition.
type ProfiledExternalAuthorityAssessment struct {
	epochID         string
	epochDigest     string
	disposition     ProfiledExternalAuthorityDisposition
	rejectionReason ProfiledExternalAuthorityRejectionReason
}

func (assessment ProfiledExternalAuthorityAssessment) EpochID() string {
	return assessment.epochID
}

func (assessment ProfiledExternalAuthorityAssessment) EpochDigest() string {
	return assessment.epochDigest
}

func (assessment ProfiledExternalAuthorityAssessment) Disposition() ProfiledExternalAuthorityDisposition {
	return assessment.disposition
}

func (assessment ProfiledExternalAuthorityAssessment) RejectionReason() (
	ProfiledExternalAuthorityRejectionReason,
	bool,
) {
	if assessment.disposition != ProfiledExternalAuthorityRejected {
		return "", false
	}
	return assessment.rejectionReason, true
}

// CurrentIntegrationBindingViewReader returns the current complete Catalog
// IntegrationBinding view at or beyond the supplied revision lower bound.
// found=false with nil error is the only authoritative absence signal; all
// errors remain indeterminate dependency failures.
type CurrentIntegrationBindingViewReader interface {
	ReadCurrentIntegrationBindingView(
		context.Context,
		uint64,
	) (catalogv1.IntegrationBindingView, bool, error)
}

// ExactBindingOwnerReader is the bounded multi-generation binding-definition
// read path. One assessment calls it at most once.
type ExactBindingOwnerReader interface {
	ResolveExactBindingSelections(
		context.Context,
		[]bindingowner.ExactBindingSelection,
	) ([][]connectorbindingownerv1.DefinitionEnvelope, error)
}

// ExactServiceConfigurationReader resolves one immutable complete service
// configuration by exact generation ID.
type ExactServiceConfigurationReader interface {
	ResolveExact(
		context.Context,
		string,
	) (serviceconfiguration.Generation, bool, error)
}

// ExactCredentialGenerationReader resolves one bounded exact generation set
// in caller order.
type ExactCredentialGenerationReader interface {
	ResolveExactSet(
		context.Context,
		[]string,
	) ([]credentialgeneration.Generation, []bool, error)
}

var (
	_ profiledPreparedEpochExactReader = (*Repository)(nil)
	_ ExactBindingOwnerReader          = (*bindingowner.Repository)(nil)
	_ ExactServiceConfigurationReader  = (*serviceconfiguration.Repository)(nil)
	_ ExactCredentialGenerationReader  = (*credentialgeneration.Repository)(nil)
)

type profiledPreparedEpochExactReader interface {
	LoadProfiledPrepared(context.Context, string) (ProfiledPreparedEpoch, bool, error)
}

type profiledBindingOwnerSelection struct {
	coordinate bindingowner.PackGenerationCoordinate
	bindings   []catalogv1.IntegrationBinding
}

type profiledBindingOwnerAuthorityReader interface {
	ResolveExactProfiledBindings(
		context.Context,
		[]profiledBindingOwnerSelection,
	) ([][]connectorbindingownerv1.DefinitionEnvelope, error)
}

type profiledServiceConfigurationAuthorityReader interface {
	ResolveExactProfiledServiceConfiguration(
		context.Context,
		string,
	) (profiledServiceConfigurationInput, bool, error)
}

type profiledCredentialGenerationAuthorityReader interface {
	ResolveExactProfiledCredentialGenerations(
		context.Context,
		[]string,
	) ([]profiledCredentialGenerationInput, []bool, error)
}

type profiledPreparedEpochReconstructor func(
	ConnectorDigestProfileID,
	string,
	catalogv1.IntegrationBindingView,
	profiledServiceConfigurationInput,
	[]BindingOwnerProof,
	[]profiledCredentialGenerationInput,
) (ProfiledPreparedEpoch, error)

type exactBindingOwnerReaderAdapter struct {
	reader ExactBindingOwnerReader
}

func (adapter exactBindingOwnerReaderAdapter) ResolveExactProfiledBindings(
	ctx context.Context,
	selections []profiledBindingOwnerSelection,
) ([][]connectorbindingownerv1.DefinitionEnvelope, error) {
	exact := make([]bindingowner.ExactBindingSelection, len(selections))
	for index := range selections {
		selection, err := bindingowner.NewExactBindingSelection(
			selections[index].coordinate,
			selections[index].bindings,
		)
		if err != nil {
			return nil, err
		}
		exact[index] = selection
	}
	return adapter.reader.ResolveExactBindingSelections(ctx, exact)
}

type exactServiceConfigurationReaderAdapter struct {
	reader ExactServiceConfigurationReader
}

func (adapter exactServiceConfigurationReaderAdapter) ResolveExactProfiledServiceConfiguration(
	ctx context.Context,
	id string,
) (profiledServiceConfigurationInput, bool, error) {
	generation, found, err := adapter.reader.ResolveExact(ctx, id)
	if err != nil || !found {
		return profiledServiceConfigurationInput{}, found, err
	}
	return serviceConfigurationInput(generation), true, nil
}

type exactCredentialGenerationReaderAdapter struct {
	reader ExactCredentialGenerationReader
}

func (adapter exactCredentialGenerationReaderAdapter) ResolveExactProfiledCredentialGenerations(
	ctx context.Context,
	ids []string,
) ([]profiledCredentialGenerationInput, []bool, error) {
	generations, found, err := adapter.reader.ResolveExactSet(ctx, ids)
	if err != nil {
		return nil, nil, err
	}
	inputs, err := credentialGenerationInputs(generations)
	if err != nil {
		return nil, nil, err
	}
	return inputs, found, nil
}

// ProfiledExternalAuthorityAssessor fresh-reads every authority referenced by
// one persisted Profile V2 candidate and reconstructs the candidate from those
// facts. It has no state-writer or activation dependency.
type ProfiledExternalAuthorityAssessor struct {
	epochs                profiledPreparedEpochExactReader
	catalog               CurrentIntegrationBindingViewReader
	bindingOwners         profiledBindingOwnerAuthorityReader
	serviceConfigurations profiledServiceConfigurationAuthorityReader
	credentialGenerations profiledCredentialGenerationAuthorityReader
	reconstruct           profiledPreparedEpochReconstructor
}

// NewProfiledExternalAuthorityAssessor binds assessment to the concrete local
// ServingEpoch repository and the four exact external-owner read paths.
func NewProfiledExternalAuthorityAssessor(
	epochs *Repository,
	catalog CurrentIntegrationBindingViewReader,
	bindingOwners ExactBindingOwnerReader,
	serviceConfigurations ExactServiceConfigurationReader,
	credentialGenerations ExactCredentialGenerationReader,
) (*ProfiledExternalAuthorityAssessor, error) {
	if epochs == nil || catalog == nil || bindingOwners == nil ||
		serviceConfigurations == nil || credentialGenerations == nil {
		return nil, fmt.Errorf("%w: external authority assessor dependency is nil", ErrDependencyMissing)
	}
	return newProfiledExternalAuthorityAssessor(
		epochs,
		catalog,
		exactBindingOwnerReaderAdapter{reader: bindingOwners},
		exactServiceConfigurationReaderAdapter{reader: serviceConfigurations},
		exactCredentialGenerationReaderAdapter{reader: credentialGenerations},
	)
}

func newProfiledExternalAuthorityAssessor(
	epochs profiledPreparedEpochExactReader,
	catalog CurrentIntegrationBindingViewReader,
	bindingOwners profiledBindingOwnerAuthorityReader,
	serviceConfigurations profiledServiceConfigurationAuthorityReader,
	credentialGenerations profiledCredentialGenerationAuthorityReader,
) (*ProfiledExternalAuthorityAssessor, error) {
	if epochs == nil || catalog == nil || bindingOwners == nil ||
		serviceConfigurations == nil || credentialGenerations == nil {
		return nil, fmt.Errorf("%w: external authority assessor dependency is nil", ErrDependencyMissing)
	}
	return &ProfiledExternalAuthorityAssessor{
		epochs:                epochs,
		catalog:               catalog,
		bindingOwners:         bindingOwners,
		serviceConfigurations: serviceConfigurations,
		credentialGenerations: credentialGenerations,
		reconstruct:           reconstructProfiledPreparedEpochForRead,
	}, nil
}

// Assess returns a read-only exact-match or authoritative-rejection decision
// for the exact locally persisted candidate. It always loads that candidate
// itself; callers cannot supply or echo a candidate snapshot.
func (assessor *ProfiledExternalAuthorityAssessor) Assess(
	ctx context.Context,
	exactEpochID string,
) (ProfiledExternalAuthorityAssessment, error) {
	evidence, err := assessor.assessEvidence(ctx, exactEpochID)
	return evidence.assessment, err
}

type profiledExternalAuthorityEvidence struct {
	assessment ProfiledExternalAuthorityAssessment
	candidate  ProfiledPreparedEpoch
}

func (assessor *ProfiledExternalAuthorityAssessor) assessEvidence(
	ctx context.Context,
	exactEpochID string,
) (profiledExternalAuthorityEvidence, error) {
	if err := assessor.validate(ctx); err != nil {
		return profiledExternalAuthorityEvidence{}, err
	}

	candidate, found, err := assessor.epochs.LoadProfiledPrepared(ctx, exactEpochID)
	if err != nil {
		return profiledExternalAuthorityEvidence{}, err
	}
	if !found {
		return profiledExternalAuthorityEvidence{}, fmt.Errorf(
			"%w: %q",
			ErrProfiledPreparedEpochNotFound,
			exactEpochID,
		)
	}
	if candidate.id != exactEpochID {
		return profiledExternalAuthorityEvidence{}, storedCandidateInvalid(
			"loaded profiled candidate ID is not byte-exact",
		)
	}
	return assessor.assessCandidateEvidence(ctx, candidate)
}

func (assessor *ProfiledExternalAuthorityAssessor) assessCandidateEvidence(
	ctx context.Context,
	candidate ProfiledPreparedEpoch,
) (profiledExternalAuthorityEvidence, error) {
	if err := assessor.validate(ctx); err != nil {
		return profiledExternalAuthorityEvidence{}, err
	}
	candidate = candidate.copy()
	if err := candidate.validateStored(); err != nil {
		return profiledExternalAuthorityEvidence{}, err
	}

	evidence := profiledExternalAuthorityEvidence{
		candidate: candidate.copy(),
		assessment: ProfiledExternalAuthorityAssessment{
			epochID:     candidate.id,
			epochDigest: candidate.epochDigest,
		},
	}
	reject := func(reason ProfiledExternalAuthorityRejectionReason) (
		profiledExternalAuthorityEvidence,
		error,
	) {
		evidence.assessment.disposition = ProfiledExternalAuthorityRejected
		evidence.assessment.rejectionReason = reason
		return evidence, nil
	}

	view, present, err := assessor.catalog.ReadCurrentIntegrationBindingView(
		ctx,
		candidate.catalog.lineage.CatalogRevision,
	)
	if err != nil {
		return profiledExternalAuthorityEvidence{}, externalAuthorityUnavailable("read Catalog", err)
	}
	if !present || !profiledCandidateMatchesCatalog(candidate, view) {
		return reject(ProfiledExternalAuthorityRejectedByCatalog)
	}

	bindingProofs, definitive, err := assessor.readFreshBindingOwnerProofs(ctx, candidate, view)
	if err != nil {
		return profiledExternalAuthorityEvidence{}, err
	}
	if !definitive {
		return reject(ProfiledExternalAuthorityRejectedByBindingOwner)
	}

	configuration, present, err := assessor.serviceConfigurations.
		ResolveExactProfiledServiceConfiguration(ctx, candidate.serviceConfiguration.generationID)
	if err != nil {
		if isServiceConfigurationRejection(err) {
			return reject(ProfiledExternalAuthorityRejectedByServiceConfiguration)
		}
		return profiledExternalAuthorityEvidence{}, externalAuthorityUnavailable(
			"read ServiceConfiguration",
			err,
		)
	}
	if !present {
		return reject(ProfiledExternalAuthorityRejectedByServiceConfiguration)
	}
	if err := validateProfiledServiceConfiguration(configuration, view); err != nil {
		return reject(ProfiledExternalAuthorityRejectedByServiceConfiguration)
	}

	credentialIDs, err := requiredProfiledCredentialGenerationIDs(view, configuration)
	if err != nil {
		return reject(ProfiledExternalAuthorityRejectedByServiceConfiguration)
	}
	credentials, foundCredentials, err := assessor.credentialGenerations.
		ResolveExactProfiledCredentialGenerations(ctx, credentialIDs)
	if err != nil {
		if isCredentialGenerationRejection(err) {
			return reject(ProfiledExternalAuthorityRejectedByCredentialGeneration)
		}
		return profiledExternalAuthorityEvidence{}, externalAuthorityUnavailable(
			"read CredentialGeneration set",
			err,
		)
	}
	if len(credentials) != len(credentialIDs) || len(foundCredentials) != len(credentialIDs) {
		return reject(ProfiledExternalAuthorityRejectedByCredentialGeneration)
	}
	for index := range credentialIDs {
		if !foundCredentials[index] || credentials[index].generationID != credentialIDs[index] {
			return reject(ProfiledExternalAuthorityRejectedByCredentialGeneration)
		}
		if err := validateProfiledCredentialGeneration(credentials[index]); err != nil {
			return reject(ProfiledExternalAuthorityRejectedByCredentialGeneration)
		}
	}

	reconstructed, err := assessor.reconstruct(
		candidate.profileID,
		candidate.id,
		view,
		configuration,
		bindingProofs,
		credentials,
	)
	if err != nil {
		return profiledExternalAuthorityEvidence{}, externalAuthorityUnavailable(
			"reconstruct retained profiled candidate",
			err,
		)
	}
	if !candidate.equal(reconstructed) {
		return reject(ProfiledExternalAuthorityRejectedByReconstruction)
	}
	evidence.assessment.disposition = ProfiledExternalAuthorityExactMatch
	return evidence, nil
}

func (assessor *ProfiledExternalAuthorityAssessor) validate(ctx context.Context) error {
	if assessor == nil || assessor.epochs == nil || assessor.catalog == nil ||
		assessor.bindingOwners == nil || assessor.serviceConfigurations == nil ||
		assessor.credentialGenerations == nil || assessor.reconstruct == nil {
		return fmt.Errorf(
			"%w: external authority assessor is unavailable",
			ErrDependencyMissing,
		)
	}
	if ctx == nil {
		return fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	return nil
}

func profiledCandidateMatchesCatalog(
	candidate ProfiledPreparedEpoch,
	view catalogv1.IntegrationBindingView,
) bool {
	if err := catalogv1.ValidateIntegrationBindingView(view); err != nil ||
		catalogReferenceFromView(view) != candidate.catalog {
		return false
	}
	profile, err := ResolveConnectorDigestProfileForRead(candidate.profileID)
	if err != nil {
		return false
	}
	bindings, err := buildProfiledStoredBindings(profile, candidate.id, view.Bindings)
	if err != nil || len(bindings) != len(candidate.bindings) {
		return false
	}
	for index := range bindings {
		if !bindings[index].equal(candidate.bindings[index].binding) {
			return false
		}
	}
	return true
}

type profiledBindingOwnerGroup struct {
	selection profiledBindingOwnerSelection
	ordinals  []int
}

type portableBindingOwnerCoordinate struct {
	ownerPack          string
	packInstallationID string
	packPayloadDigest  string
}

func (assessor *ProfiledExternalAuthorityAssessor) readFreshBindingOwnerProofs(
	ctx context.Context,
	candidate ProfiledPreparedEpoch,
	view catalogv1.IntegrationBindingView,
) ([]BindingOwnerProof, bool, error) {
	groups, ok := groupProfiledBindingOwnerSelections(candidate, view)
	if !ok {
		return nil, false, nil
	}
	selections := make([]profiledBindingOwnerSelection, len(groups))
	for index := range groups {
		selections[index] = groups[index].selection
	}
	resolved, err := assessor.bindingOwners.ResolveExactProfiledBindings(ctx, selections)
	if err != nil {
		if isBindingOwnerRejection(err) {
			return nil, false, nil
		}
		return nil, false, externalAuthorityUnavailable("read binding owners", err)
	}
	if len(resolved) != len(groups) {
		return nil, false, nil
	}
	proofs := make([]BindingOwnerProof, len(candidate.bindings))
	for groupIndex := range groups {
		group := groups[groupIndex]
		if len(resolved[groupIndex]) != len(group.ordinals) {
			return nil, false, nil
		}
		for bindingIndex, ordinal := range group.ordinals {
			envelope := resolved[groupIndex][bindingIndex]
			if envelope.Identity != connectorbindingownerv1.IdentityFromBinding(view.Bindings[ordinal]) {
				return nil, false, nil
			}
			proof, err := NewBindingOwnerProof(group.selection.coordinate.StorageID, envelope)
			if err != nil {
				return nil, false, nil
			}
			proofs[ordinal] = proof
		}
	}
	return proofs, true, nil
}

func groupProfiledBindingOwnerSelections(
	candidate ProfiledPreparedEpoch,
	view catalogv1.IntegrationBindingView,
) ([]profiledBindingOwnerGroup, bool) {
	if len(candidate.bindings) != len(view.Bindings) {
		return nil, false
	}
	groups := make([]profiledBindingOwnerGroup, 0, len(view.Bindings))
	byCoordinate := make(map[bindingowner.PackGenerationCoordinate]int, len(view.Bindings))
	portableByStorageID := make(map[string]portableBindingOwnerCoordinate, len(view.Bindings))
	storageIDByPortable := make(map[portableBindingOwnerCoordinate]string, len(view.Bindings))
	for ordinal, binding := range view.Bindings {
		coordinate := bindingowner.PackGenerationCoordinate{
			StorageID:          candidate.bindings[ordinal].owner.packGenerationID,
			OwnerPack:          binding.OwnerPack,
			PackInstallationID: binding.PackInstallationID,
			PackPayloadDigest:  binding.PackPayloadDigest,
		}
		portable := portableBindingOwnerCoordinate{
			ownerPack:          coordinate.OwnerPack,
			packInstallationID: coordinate.PackInstallationID,
			packPayloadDigest:  coordinate.PackPayloadDigest,
		}
		if previous, exists := portableByStorageID[coordinate.StorageID]; exists && previous != portable {
			return nil, false
		}
		portableByStorageID[coordinate.StorageID] = portable
		if previous, exists := storageIDByPortable[portable]; exists && previous != coordinate.StorageID {
			return nil, false
		}
		storageIDByPortable[portable] = coordinate.StorageID
		groupIndex, exists := byCoordinate[coordinate]
		if !exists {
			groupIndex = len(groups)
			byCoordinate[coordinate] = groupIndex
			groups = append(groups, profiledBindingOwnerGroup{
				selection: profiledBindingOwnerSelection{coordinate: coordinate},
			})
		}
		groups[groupIndex].selection.bindings = append(
			groups[groupIndex].selection.bindings,
			binding,
		)
		groups[groupIndex].ordinals = append(groups[groupIndex].ordinals, ordinal)
	}
	return groups, true
}

func requiredProfiledCredentialGenerationIDs(
	view catalogv1.IntegrationBindingView,
	configuration profiledServiceConfigurationInput,
) ([]string, error) {
	required, err := requiredProfiledCredentialIdentities(view, configuration)
	if err != nil {
		return nil, err
	}
	ids := make([]string, 0, len(required))
	for id := range required {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	return ids, nil
}

func isBindingOwnerRejection(err error) bool {
	return errors.Is(err, bindingowner.ErrAuthorityConflict) ||
		errors.Is(err, bindingowner.ErrInvalidRequest) ||
		errors.Is(err, bindingowner.ErrExactReadBudgetExceeded)
}

func isServiceConfigurationRejection(err error) bool {
	return errors.Is(err, serviceconfiguration.ErrStoredGenerationInvalid) ||
		errors.Is(err, serviceconfiguration.ErrUnsupportedGenerationProtocol) ||
		errors.Is(err, serviceconfiguration.ErrUnsupportedReceiptProtocol)
}

func isCredentialGenerationRejection(err error) bool {
	return errors.Is(err, credentialgeneration.ErrStoredGenerationInvalid) ||
		errors.Is(err, credentialgeneration.ErrUnsupportedReceiptProtocol) ||
		errors.Is(err, credentialgeneration.ErrExactSetBudgetExceeded) ||
		errors.Is(err, ErrInvalidCandidate)
}

func externalAuthorityUnavailable(operation string, cause error) error {
	return errors.Join(
		ErrExternalAuthorityUnavailable,
		fmt.Errorf("%s: %w", operation, cause),
	)
}