package serving

import (
	"bytes"
	"fmt"

	"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"
)

// BindingOwnerProof is a detached exact owner proof supplied while preparing a
// profiled epoch. It deliberately retains neither CanonicalDefinition nor the
// serialized owner envelope; those bytes remain owned by the exact
// PackGeneration definition set.
type BindingOwnerProof struct {
	identity                     connectorbindingownerv1.Identity
	packGenerationID             string
	definitionEnvelopeVersion    string
	definitionReceiptVersion     string
	canonicalDefinitionDigest    string
	definitionOwnerReceipt       []byte
	definitionOwnerReceiptDigest string
}

// NewBindingOwnerProof validates one complete owner envelope and projects only
// the immutable proof fields persisted by a profiled ServingEpoch. The opaque
// PackGeneration storage ID is copied exactly and is never derived by Connector.
func NewBindingOwnerProof(
	packGenerationID string,
	envelope connectorbindingownerv1.DefinitionEnvelope,
) (BindingOwnerProof, error) {
	if !validOpaqueIdentifierWithin(packGenerationID, connectorDigestProfileV2MaxIdentifierBytes) {
		return BindingOwnerProof{}, invalidCandidate("binding owner PackGeneration ID is not canonical")
	}
	if err := connectorbindingownerv1.ValidateDefinitionEnvelope(envelope); err != nil {
		return BindingOwnerProof{}, invalidCandidate("binding owner envelope: %v", err)
	}
	return BindingOwnerProof{
		identity:                     envelope.Identity,
		packGenerationID:             packGenerationID,
		definitionEnvelopeVersion:    envelope.EnvelopeVersion,
		definitionReceiptVersion:     envelope.ReceiptVersion,
		canonicalDefinitionDigest:    envelope.CanonicalDefinitionDigest,
		definitionOwnerReceipt:       append([]byte(nil), envelope.OwnerReceipt...),
		definitionOwnerReceiptDigest: envelope.OwnerReceiptDigest,
	}, nil
}

func (proof BindingOwnerProof) copy() BindingOwnerProof {
	proof.definitionOwnerReceipt = append([]byte(nil), proof.definitionOwnerReceipt...)
	return proof
}

type storedBindingOwnerProof struct {
	servingEpochBindingID        string
	packGenerationID             string
	definitionEnvelopeVersion    string
	definitionReceiptVersion     string
	canonicalDefinitionDigest    string
	definitionOwnerReceipt       []byte
	definitionOwnerReceiptDigest string
}

func (proof storedBindingOwnerProof) copy() storedBindingOwnerProof {
	proof.definitionOwnerReceipt = append([]byte(nil), proof.definitionOwnerReceipt...)
	return proof
}

func (proof storedBindingOwnerProof) equal(other storedBindingOwnerProof) bool {
	return proof.servingEpochBindingID == other.servingEpochBindingID &&
		proof.packGenerationID == other.packGenerationID &&
		proof.definitionEnvelopeVersion == other.definitionEnvelopeVersion &&
		proof.definitionReceiptVersion == other.definitionReceiptVersion &&
		proof.canonicalDefinitionDigest == other.canonicalDefinitionDigest &&
		proof.definitionOwnerReceiptDigest == other.definitionOwnerReceiptDigest &&
		bytes.Equal(proof.definitionOwnerReceipt, other.definitionOwnerReceipt)
}

type profiledStoredBinding struct {
	binding storedBinding
	owner   storedBindingOwnerProof
}

func (binding profiledStoredBinding) copy() profiledStoredBinding {
	binding.binding = binding.binding.copy()
	binding.owner = binding.owner.copy()
	return binding
}

func (binding profiledStoredBinding) equal(other profiledStoredBinding) bool {
	return binding.binding.equal(other.binding) && binding.owner.equal(other.owner)
}

type storedServiceConfigurationProof struct {
	generationProtocolVersion serviceconfiguration.GenerationProtocolVersion
	receiptProtocolVersion    serviceconfiguration.ReceiptProtocolVersion
	generationID              string
	generationDigest          string
	ownerReceipt              []byte
	ownerReceiptDigest        string
}

func (proof storedServiceConfigurationProof) copy() storedServiceConfigurationProof {
	proof.ownerReceipt = append([]byte(nil), proof.ownerReceipt...)
	return proof
}

func (proof storedServiceConfigurationProof) equal(other storedServiceConfigurationProof) bool {
	return proof.generationProtocolVersion == other.generationProtocolVersion &&
		proof.receiptProtocolVersion == other.receiptProtocolVersion &&
		proof.generationID == other.generationID &&
		proof.generationDigest == other.generationDigest &&
		proof.ownerReceiptDigest == other.ownerReceiptDigest &&
		bytes.Equal(proof.ownerReceipt, other.ownerReceipt)
}

type storedCredentialGenerationProof struct {
	ordinal                      uint64
	generationID                 string
	receiptProtocolVersion       credentialgeneration.ReceiptProtocolVersion
	transportConfigurationDigest string
	ownerReceipt                 []byte
	ownerReceiptDigest           string
}

func (proof storedCredentialGenerationProof) copy() storedCredentialGenerationProof {
	proof.ownerReceipt = append([]byte(nil), proof.ownerReceipt...)
	return proof
}

func (proof storedCredentialGenerationProof) equal(other storedCredentialGenerationProof) bool {
	return proof.ordinal == other.ordinal && proof.generationID == other.generationID &&
		proof.receiptProtocolVersion == other.receiptProtocolVersion &&
		proof.transportConfigurationDigest == other.transportConfigurationDigest &&
		proof.ownerReceiptDigest == other.ownerReceiptDigest &&
		bytes.Equal(proof.ownerReceipt, other.ownerReceipt)
}

// ProfiledPreparedEpoch is one complete inert V2 candidate. It is intentionally
// a different Go type from legacy PreparedEpoch, so the legacy Repository cannot
// accept and truncate its reproducibility proof graph. It exposes no state
// transition, persistence conversion, definition bytes, configuration payload,
// credential secret, or mutable owner selector.
type ProfiledPreparedEpoch struct {
	id                   string
	epochDigest          string
	servingScopeDigest   string
	state                string
	activeScopeDigest    *string
	profileID            ConnectorDigestProfileID
	catalog              catalogReference
	serviceConfiguration storedServiceConfigurationProof
	bindings             []profiledStoredBinding
	credentials          []storedCredentialGenerationProof
}

func (epoch ProfiledPreparedEpoch) ID() string                                { return epoch.id }
func (epoch ProfiledPreparedEpoch) EpochDigest() string                       { return epoch.epochDigest }
func (epoch ProfiledPreparedEpoch) ServingScopeDigest() string                { return epoch.servingScopeDigest }
func (epoch ProfiledPreparedEpoch) State() string                             { return epoch.state }
func (epoch ProfiledPreparedEpoch) DigestProfileID() ConnectorDigestProfileID { return epoch.profileID }
func (epoch ProfiledPreparedEpoch) BindingCount() int                         { return len(epoch.bindings) }
func (epoch ProfiledPreparedEpoch) CredentialGenerationCount() int            { return len(epoch.credentials) }

func (epoch ProfiledPreparedEpoch) ActiveScopeDigest() (string, bool) {
	if epoch.activeScopeDigest == nil {
		return "", false
	}
	return *epoch.activeScopeDigest, true
}

func (epoch ProfiledPreparedEpoch) copy() ProfiledPreparedEpoch {
	if epoch.activeScopeDigest != nil {
		value := *epoch.activeScopeDigest
		epoch.activeScopeDigest = &value
	}
	epoch.serviceConfiguration = epoch.serviceConfiguration.copy()
	if epoch.bindings != nil {
		bindings := make([]profiledStoredBinding, len(epoch.bindings))
		for index := range epoch.bindings {
			bindings[index] = epoch.bindings[index].copy()
		}
		epoch.bindings = bindings
	}
	if epoch.credentials != nil {
		credentials := make([]storedCredentialGenerationProof, len(epoch.credentials))
		for index := range epoch.credentials {
			credentials[index] = epoch.credentials[index].copy()
		}
		epoch.credentials = credentials
	}
	return epoch
}

func (epoch ProfiledPreparedEpoch) equal(other ProfiledPreparedEpoch) bool {
	return epoch.state == other.state &&
		equalOptionalString(epoch.activeScopeDigest, other.activeScopeDigest) &&
		epoch.equalImmutable(other)
}

func (epoch ProfiledPreparedEpoch) equalImmutable(other ProfiledPreparedEpoch) bool {
	if epoch.id != other.id || epoch.epochDigest != other.epochDigest ||
		epoch.servingScopeDigest != other.servingScopeDigest ||
		epoch.profileID != other.profileID || epoch.catalog != other.catalog ||
		!epoch.serviceConfiguration.equal(other.serviceConfiguration) ||
		len(epoch.bindings) != len(other.bindings) || len(epoch.credentials) != len(other.credentials) {
		return false
	}
	for index := range epoch.bindings {
		if !epoch.bindings[index].equal(other.bindings[index]) {
			return false
		}
	}
	for index := range epoch.credentials {
		if !epoch.credentials[index].equal(other.credentials[index]) {
			return false
		}
	}
	return true
}

type profiledCredentialSelection struct {
	catalogBindingOrdinal uint32
	credentialRef         string
	generationID          string
	connectorAccountID    string
	providerScopeID       string
}

type profiledServiceConfigurationInput struct {
	storedServiceConfigurationProof
	orgID                    string
	appInstallationID        string
	authorizedWorkloadID     string
	lineage                  catalogv1.ViewLineage
	receipts                 catalogv1.Receipts
	accountCount             uint64
	bindingCount             uint64
	credentialSelectionCount uint64
	selections               []profiledCredentialSelection
}

type profiledCredentialGenerationInput struct {
	storedCredentialGenerationProof
	stableCredentialIdentity string
	connectorAccountID       string
	providerScopeID          string
	transportBytes           uint64
}

func serviceConfigurationInput(
	generation serviceconfiguration.Generation,
) profiledServiceConfigurationInput {
	accounts := generation.Accounts()
	input := profiledServiceConfigurationInput{
		storedServiceConfigurationProof: storedServiceConfigurationProof{
			generationProtocolVersion: generation.GenerationProtocolVersion(),
			receiptProtocolVersion:    generation.ReceiptProtocolVersion(),
			generationID:              generation.ID(),
			generationDigest:          generation.GenerationDigest(),
			ownerReceipt:              generation.OwnerReceipt(),
			ownerReceiptDigest:        generation.OwnerReceiptDigest(),
		},
		orgID:                    generation.OrgID(),
		appInstallationID:        generation.AppInstallationID(),
		authorizedWorkloadID:     generation.AuthorizedWorkloadID(),
		lineage:                  generation.CatalogLineage(),
		receipts:                 generation.CatalogReceipts(),
		accountCount:             generation.AccountCount(),
		bindingCount:             generation.BindingCount(),
		credentialSelectionCount: generation.CredentialSelectionCount(),
	}
	for _, account := range accounts {
		for _, binding := range account.Bindings() {
			for _, selection := range binding.CredentialSelections() {
				input.selections = append(input.selections, profiledCredentialSelection{
					catalogBindingOrdinal: binding.CatalogBindingOrdinal(),
					credentialRef:         selection.CredentialRef(),
					generationID:          selection.CredentialGenerationID(),
					connectorAccountID:    account.ConnectorAccountID(),
					providerScopeID:       account.ProviderScopeID(),
				})
			}
		}
	}
	return input
}

func credentialGenerationInputs(
	generations []credentialgeneration.Generation,
) ([]profiledCredentialGenerationInput, error) {
	if generations == nil {
		return nil, nil
	}
	if len(generations) > credentialgeneration.MaxExactIdentitySetEntries {
		return nil, profiledInvalid(
			"credential generation set exceeds %d entries",
			credentialgeneration.MaxExactIdentitySetEntries,
		)
	}
	var totalIDBytes, totalIdentityBytes, totalPayloadBytes uint64
	for index, generation := range generations {
		var ok bool
		totalIDBytes, ok = addWithinLimit(
			totalIDBytes,
			uint64(len(generation.ID())),
			credentialgeneration.MaxExactIdentitySetIDBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation %d exceeds the exact-set ID budget", index)
		}
		identityBytes := uint64(len(generation.ID()) + len(generation.StableCredentialIdentity()) +
			len(generation.ConnectorAccountID()) + len(generation.ProviderScopeID()))
		totalIdentityBytes, ok = addWithinLimit(
			totalIdentityBytes,
			identityBytes,
			credentialgeneration.MaxExactIdentitySetIdentityBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation %d exceeds the exact-set identity budget", index)
		}
		payloadBytes, ok := addWithinLimit(
			generation.TransportConfigurationByteCount(),
			generation.OwnerReceiptByteCount(),
			credentialgeneration.MaxExactIdentitySetPayloadBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation %d exceeds the exact-set payload budget", index)
		}
		totalPayloadBytes, ok = addWithinLimit(
			totalPayloadBytes,
			payloadBytes,
			credentialgeneration.MaxExactIdentitySetPayloadBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation set exceeds the exact-set payload budget")
		}
	}
	inputs := make([]profiledCredentialGenerationInput, len(generations))
	for index, generation := range generations {
		inputs[index] = profiledCredentialGenerationInput{
			storedCredentialGenerationProof: storedCredentialGenerationProof{
				generationID:                 generation.ID(),
				receiptProtocolVersion:       generation.ReceiptProtocolVersion(),
				transportConfigurationDigest: generation.TransportConfigurationDigest(),
				ownerReceipt:                 generation.OwnerReceipt(),
				ownerReceiptDigest:           generation.OwnerReceiptDigest(),
			},
			stableCredentialIdentity: generation.StableCredentialIdentity(),
			connectorAccountID:       generation.ConnectorAccountID(),
			providerScopeID:          generation.ProviderScopeID(),
			transportBytes:           generation.TransportConfigurationByteCount(),
		}
	}
	return inputs, nil
}

// NewProfiledPreparedEpoch derives one complete inert V2 candidate from exact
// owner values already obtained by the caller. Construction proves canonical
// closure only; it is not the later fresh-read external verification transition.
func NewProfiledPreparedEpoch(
	profileID ConnectorDigestProfileID,
	epochID string,
	view catalogv1.IntegrationBindingView,
	configuration serviceconfiguration.Generation,
	bindingOwners []BindingOwnerProof,
	credentials []credentialgeneration.Generation,
) (ProfiledPreparedEpoch, error) {
	if _, err := ResolveConnectorDigestProfileForAuthoring(profileID); err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	credentialInputs, err := credentialGenerationInputs(credentials)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	return newProfiledPreparedEpoch(
		profileID,
		epochID,
		view,
		serviceConfigurationInput(configuration),
		bindingOwners,
		credentialInputs,
	)
}

func profiledInvalid(format string, args ...any) error {
	return invalidCandidate("profiled candidate: %s", fmt.Sprintf(format, args...))
}