package serving

import (
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"fmt"
	"hash"
	"sort"

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

func newProfiledPreparedEpoch(
	profileID ConnectorDigestProfileID,
	epochID string,
	view catalogv1.IntegrationBindingView,
	configuration profiledServiceConfigurationInput,
	bindingOwners []BindingOwnerProof,
	credentials []profiledCredentialGenerationInput,
) (ProfiledPreparedEpoch, error) {
	profile, err := ResolveConnectorDigestProfileForAuthoring(profileID)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	return deriveProfiledPreparedEpoch(
		profile,
		epochID,
		view,
		configuration,
		bindingOwners,
		credentials,
	)
}

// reconstructProfiledPreparedEpochForRead rederives an already persisted
// candidate with its retained-read recipe. Authoring support may be retired
// while durable candidates still require assessment, duplicate resolution,
// recovery, or audit.
func reconstructProfiledPreparedEpochForRead(
	profileID ConnectorDigestProfileID,
	epochID string,
	view catalogv1.IntegrationBindingView,
	configuration profiledServiceConfigurationInput,
	bindingOwners []BindingOwnerProof,
	credentials []profiledCredentialGenerationInput,
) (ProfiledPreparedEpoch, error) {
	profile, err := ResolveConnectorDigestProfileForRead(profileID)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	return deriveProfiledPreparedEpoch(
		profile,
		epochID,
		view,
		configuration,
		bindingOwners,
		credentials,
	)
}

func deriveProfiledPreparedEpoch(
	profile ConnectorDigestProfile,
	epochID string,
	view catalogv1.IntegrationBindingView,
	configuration profiledServiceConfigurationInput,
	bindingOwners []BindingOwnerProof,
	credentials []profiledCredentialGenerationInput,
) (ProfiledPreparedEpoch, error) {
	if profile.ID() != ConnectorDigestProfileV2ID {
		return ProfiledPreparedEpoch{}, unsupportedConnectorDigestProfile(profile.ID())
	}
	if !validOpaqueIdentifierWithin(epochID, connectorDigestProfileV2MaxIdentifierBytes) {
		return ProfiledPreparedEpoch{}, profiledInvalid("epoch ID is not canonical")
	}
	if err := catalogv1.ValidateIntegrationBindingView(view); err != nil {
		return ProfiledPreparedEpoch{}, profiledInvalid("Catalog IntegrationBindingView: %v", err)
	}
	if view.Lineage.ViewDigest != catalogv1.ComputeIntegrationBindingViewDigest(view) {
		return ProfiledPreparedEpoch{}, profiledInvalid("Catalog IntegrationBindingView digest disagrees")
	}
	if err := validateProfiledServiceConfiguration(configuration, view); err != nil {
		return ProfiledPreparedEpoch{}, err
	}

	baseBindings, err := buildProfiledStoredBindings(profile, epochID, view.Bindings)
	if err != nil {
		return ProfiledPreparedEpoch{}, profiledInvalid("Catalog bindings: %v", err)
	}
	bindings, err := bindProfiledOwnerProofs(baseBindings, view.Bindings, bindingOwners)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	credentialProofs, err := buildProfiledCredentialProofs(view, configuration, credentials)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}

	catalog := catalogReferenceFromView(view)
	servingScopeDigest, err := profile.computeServingScopeDigest(catalog)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	epoch := ProfiledPreparedEpoch{
		id:                   epochID,
		servingScopeDigest:   servingScopeDigest,
		state:                preparedState,
		profileID:            profile.ID(),
		catalog:              catalog,
		serviceConfiguration: configuration.storedServiceConfigurationProof.copy(),
		bindings:             bindings,
		credentials:          credentialProofs,
	}
	epoch.epochDigest, err = computeProfiledEpochDigest(profile, epoch)
	if err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	if err := epoch.validateStored(); err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	return epoch.copy(), nil
}

func validateProfiledServiceConfiguration(
	configuration profiledServiceConfigurationInput,
	view catalogv1.IntegrationBindingView,
) error {
	if _, err := serviceconfiguration.ResolveGenerationProtocolForRead(
		configuration.generationProtocolVersion,
	); err != nil {
		return profiledInvalid("ServiceConfiguration generation protocol: %v", err)
	}
	if _, err := serviceconfiguration.ResolveReceiptProtocolForRead(
		configuration.receiptProtocolVersion,
	); err != nil {
		return profiledInvalid("ServiceConfiguration receipt protocol: %v", err)
	}
	if !validOpaqueIdentifier(configuration.generationID) {
		return profiledInvalid("ServiceConfiguration generation ID is not canonical")
	}
	if !prefixedSHA256Pattern.MatchString(configuration.generationDigest) ||
		!prefixedSHA256Pattern.MatchString(configuration.ownerReceiptDigest) {
		return profiledInvalid("ServiceConfiguration digest shape is invalid")
	}
	if len(configuration.ownerReceipt) == 0 ||
		len(configuration.ownerReceipt) > serviceconfiguration.MaxCanonicalReceiptBytes {
		return profiledInvalid(
			"ServiceConfiguration owner receipt is empty or exceeds %d bytes",
			serviceconfiguration.MaxCanonicalReceiptBytes,
		)
	}
	if configuration.orgID != view.OrgID ||
		configuration.appInstallationID != view.AppInstallationID ||
		configuration.authorizedWorkloadID != view.AuthorizedWorkloadID ||
		configuration.lineage != view.Lineage || configuration.receipts != view.Receipts {
		return profiledInvalid("ServiceConfiguration Catalog authority disagrees with the candidate view")
	}
	if configuration.credentialSelectionCount != uint64(len(configuration.selections)) {
		return profiledInvalid("ServiceConfiguration credential-selection count disagrees")
	}
	if configuration.accountCount > serviceconfiguration.MaxAccounts ||
		configuration.bindingCount > serviceconfiguration.MaxBindingConfigurations ||
		configuration.credentialSelectionCount > serviceconfiguration.MaxCredentialSelections {
		return profiledInvalid("ServiceConfiguration declared counts exceed owner bounds")
	}
	return nil
}

func buildProfiledStoredBindings(
	profile ConnectorDigestProfile,
	epochID string,
	bindings []catalogv1.IntegrationBinding,
) ([]storedBinding, error) {
	if profile.ID() != ConnectorDigestProfileV2ID {
		return nil, unsupportedConnectorDigestProfile(profile.ID())
	}
	result := make([]storedBinding, len(bindings))
	for index, binding := range bindings {
		credentialRefs, err := profile.encodeCredentialRefs(binding.CredentialRefs)
		if err != nil {
			return nil, fmt.Errorf("binding %d credential refs: %w", index, err)
		}
		identityDigest, err := profile.computeBindingIdentityDigest(binding)
		if err != nil {
			return nil, err
		}
		stored := storedBinding{
			servingEpochID:     epochID,
			ordinal:            uint64(index),
			identityDigest:     identityDigest,
			kind:               binding.Kind,
			ownerPack:          binding.OwnerPack,
			packInstallationID: binding.PackInstallationID,
			packPayloadDigest:  binding.PackPayloadDigest,
			businessKey:        binding.BusinessKey,
			definitionVersion:  binding.DefinitionVersion,
			credentialRefs:     credentialRefs,
		}
		stored.id, err = profile.computeBindingRowID(epochID, stored.ordinal, identityDigest)
		if err != nil {
			return nil, err
		}
		if binding.RuntimeCommand != nil {
			stored.runtimeCommandKind = stringPointer(binding.RuntimeCommand.CommandKind)
			stored.runtimePayloadFormatVersion = uint32Pointer(binding.RuntimeCommand.PayloadFormatVersion)
			stored.runtimeDefinitionRef = stringPointer(binding.RuntimeCommand.DefinitionRef)
			stored.runtimeDefinitionVersion = stringPointer(binding.RuntimeCommand.DefinitionVersion)
		}
		result[index] = stored
	}
	return result, nil
}

func bindProfiledOwnerProofs(
	bindings []storedBinding,
	catalogBindings []catalogv1.IntegrationBinding,
	proofs []BindingOwnerProof,
) ([]profiledStoredBinding, error) {
	if len(proofs) != len(catalogBindings) {
		return nil, profiledInvalid(
			"binding-owner proof count is %d, want %d",
			len(proofs),
			len(catalogBindings),
		)
	}
	byIdentity := make(map[connectorbindingownerv1.Identity]BindingOwnerProof, len(proofs))
	var receiptBytes uint64
	for index, proof := range proofs {
		if err := validateBindingOwnerProof(proof); err != nil {
			return nil, profiledInvalid("binding-owner proof %d: %v", index, err)
		}
		var ok bool
		receiptBytes, ok = addWithinLimit(
			receiptBytes,
			uint64(len(proof.definitionOwnerReceipt)),
			connectorbindingownerv1.MaxDefinitionSetBytes,
		)
		if !ok {
			return nil, profiledInvalid("binding-owner receipt set exceeds its owner bound")
		}
		if _, duplicate := byIdentity[proof.identity]; duplicate {
			return nil, profiledInvalid("binding-owner proofs contain a duplicate identity")
		}
		byIdentity[proof.identity] = proof.copy()
	}
	result := make([]profiledStoredBinding, len(bindings))
	for index, catalogBinding := range catalogBindings {
		identity := connectorbindingownerv1.IdentityFromBinding(catalogBinding)
		proof, ok := byIdentity[identity]
		if !ok {
			return nil, profiledInvalid(
				"binding-owner proof is missing for %s/%s",
				catalogBinding.Kind,
				catalogBinding.BusinessKey,
			)
		}
		delete(byIdentity, identity)
		result[index] = profiledStoredBinding{
			binding: bindings[index],
			owner: storedBindingOwnerProof{
				servingEpochBindingID:        bindings[index].id,
				packGenerationID:             proof.packGenerationID,
				definitionEnvelopeVersion:    proof.definitionEnvelopeVersion,
				definitionReceiptVersion:     proof.definitionReceiptVersion,
				canonicalDefinitionDigest:    proof.canonicalDefinitionDigest,
				definitionOwnerReceipt:       append([]byte(nil), proof.definitionOwnerReceipt...),
				definitionOwnerReceiptDigest: proof.definitionOwnerReceiptDigest,
			},
		}
	}
	if len(byIdentity) != 0 {
		return nil, profiledInvalid("binding-owner proof set contains an extra identity")
	}
	return result, nil
}

func validateBindingOwnerProof(proof BindingOwnerProof) error {
	if !validOpaqueIdentifierWithin(proof.packGenerationID, connectorDigestProfileV2MaxIdentifierBytes) {
		return fmt.Errorf("PackGeneration ID is not canonical")
	}
	if err := connectorbindingownerv1.ValidateIdentity(proof.identity); err != nil {
		return fmt.Errorf("identity: %w", err)
	}
	if proof.definitionEnvelopeVersion != connectorbindingownerv1.EnvelopeVersion ||
		proof.definitionReceiptVersion != connectorbindingownerv1.ReceiptVersion {
		return fmt.Errorf("definition protocol is unsupported")
	}
	if !prefixedSHA256Pattern.MatchString(proof.canonicalDefinitionDigest) ||
		!prefixedSHA256Pattern.MatchString(proof.definitionOwnerReceiptDigest) ||
		len(proof.definitionOwnerReceipt) == 0 {
		return fmt.Errorf("definition digest or receipt shape is invalid")
	}
	return nil
}

type requiredCredentialIdentity struct {
	stableCredentialIdentity string
	connectorAccountID       string
	providerScopeID          string
}

func buildProfiledCredentialProofs(
	view catalogv1.IntegrationBindingView,
	configuration profiledServiceConfigurationInput,
	credentials []profiledCredentialGenerationInput,
) ([]storedCredentialGenerationProof, error) {
	if credentials == nil {
		return nil, profiledInvalid("exact credential generation set must be present")
	}
	required, err := requiredProfiledCredentialIdentities(view, configuration)
	if err != nil {
		return nil, err
	}
	if len(credentials) > credentialgeneration.MaxExactIdentitySetEntries {
		return nil, profiledInvalid(
			"credential generation set exceeds %d entries",
			credentialgeneration.MaxExactIdentitySetEntries,
		)
	}

	byID := make(map[string]profiledCredentialGenerationInput, len(credentials))
	var totalIDBytes, totalIdentityBytes, totalPayloadBytes uint64
	for index, generation := range credentials {
		if err := validateProfiledCredentialGeneration(generation); err != nil {
			return nil, profiledInvalid("credential generation %d: %v", index, err)
		}
		if _, duplicate := byID[generation.generationID]; duplicate {
			return nil, profiledInvalid("exact credential generation set contains a duplicate ID")
		}
		var ok bool
		totalIDBytes, ok = addWithinLimit(
			totalIDBytes,
			uint64(len(generation.generationID)),
			credentialgeneration.MaxExactIdentitySetIDBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation ID bytes exceed their owner bound")
		}
		identityBytes := uint64(len(generation.generationID) +
			len(generation.stableCredentialIdentity) + len(generation.connectorAccountID) +
			len(generation.providerScopeID))
		totalIdentityBytes, ok = addWithinLimit(
			totalIdentityBytes,
			identityBytes,
			credentialgeneration.MaxExactIdentitySetIdentityBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation identity bytes exceed their owner bound")
		}
		payloadBytes, ok := addWithinLimit(
			generation.transportBytes,
			uint64(len(generation.ownerReceipt)),
			credentialgeneration.MaxExactIdentitySetPayloadBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation payload exceeds its owner bound")
		}
		totalPayloadBytes, ok = addWithinLimit(
			totalPayloadBytes,
			payloadBytes,
			credentialgeneration.MaxExactIdentitySetPayloadBytes,
		)
		if !ok {
			return nil, profiledInvalid("credential generation payload set exceeds its owner bound")
		}
		byID[generation.generationID] = generation
	}
	if len(byID) != len(required) {
		return nil, profiledInvalid("credential generation proof set is missing or contains an extra ID")
	}
	ids := make([]string, 0, len(required))
	for id, identity := range required {
		generation, ok := byID[id]
		if !ok {
			return nil, profiledInvalid("credential generation proof %q is missing", id)
		}
		if generation.stableCredentialIdentity != identity.stableCredentialIdentity ||
			generation.connectorAccountID != identity.connectorAccountID ||
			generation.providerScopeID != identity.providerScopeID {
			return nil, profiledInvalid(
				"credential generation %q disagrees with its CredentialRef, account, or provider scope",
				id,
			)
		}
		ids = append(ids, id)
	}
	sort.Strings(ids)
	result := make([]storedCredentialGenerationProof, len(ids))
	for index, id := range ids {
		proof := byID[id].storedCredentialGenerationProof.copy()
		proof.ordinal = uint64(index)
		result[index] = proof
	}
	return result, nil
}

func requiredProfiledCredentialIdentities(
	view catalogv1.IntegrationBindingView,
	configuration profiledServiceConfigurationInput,
) (map[string]requiredCredentialIdentity, error) {
	required := make(map[string]requiredCredentialIdentity)
	for index, selection := range configuration.selections {
		if !validOpaqueIdentifier(selection.generationID) ||
			!validOpaqueIdentifier(selection.credentialRef) ||
			!validOpaqueIdentifier(selection.connectorAccountID) ||
			!validOpaqueIdentifier(selection.providerScopeID) {
			return nil, profiledInvalid("ServiceConfiguration credential selection %d is malformed", index)
		}
		if uint64(selection.catalogBindingOrdinal) >= uint64(len(view.Bindings)) ||
			!containsExactString(view.Bindings[selection.catalogBindingOrdinal].CredentialRefs, selection.credentialRef) {
			return nil, profiledInvalid("ServiceConfiguration credential selection %d disagrees with Catalog", index)
		}
		identity := requiredCredentialIdentity{
			stableCredentialIdentity: selection.credentialRef,
			connectorAccountID:       selection.connectorAccountID,
			providerScopeID:          selection.providerScopeID,
		}
		if previous, exists := required[selection.generationID]; exists && previous != identity {
			return nil, profiledInvalid(
				"credential generation %q is selected under conflicting identities",
				selection.generationID,
			)
		}
		required[selection.generationID] = identity
	}
	if len(required) > credentialgeneration.MaxExactIdentitySetEntries {
		return nil, profiledInvalid(
			"credential generation set exceeds %d entries",
			credentialgeneration.MaxExactIdentitySetEntries,
		)
	}
	return required, nil
}

func validateProfiledCredentialGeneration(generation profiledCredentialGenerationInput) error {
	fields := []string{
		generation.generationID,
		generation.stableCredentialIdentity,
		generation.connectorAccountID,
		generation.providerScopeID,
	}
	for _, field := range fields {
		if !validOpaqueIdentifier(field) {
			return fmt.Errorf("identity projection is not canonical")
		}
	}
	if _, err := credentialgeneration.ResolveReceiptProtocolForRead(
		generation.receiptProtocolVersion,
	); err != nil {
		return fmt.Errorf("receipt protocol: %w", err)
	}
	if !prefixedSHA256Pattern.MatchString(generation.transportConfigurationDigest) ||
		!prefixedSHA256Pattern.MatchString(generation.ownerReceiptDigest) ||
		len(generation.ownerReceipt) == 0 ||
		generation.transportBytes > credentialgeneration.MaxTransportConfigurationBytes {
		return fmt.Errorf("digest, receipt, or transport shape is invalid")
	}
	return nil
}

func containsExactString(values []string, expected string) bool {
	for _, value := range values {
		if value == expected {
			return true
		}
	}
	return false
}

func addWithinLimit(total, next, limit uint64) (uint64, bool) {
	if total > limit || next > limit-total {
		return 0, false
	}
	return total + next, true
}

func (epoch ProfiledPreparedEpoch) validateStored() error {
	if epoch.profileID != ConnectorDigestProfileV2ID || epoch.state != preparedState ||
		epoch.activeScopeDigest != nil {
		return storedCandidateInvalid("profiled candidate state or profile is invalid")
	}
	profile, err := ResolveConnectorDigestProfileForRead(epoch.profileID)
	if err != nil {
		return storedCandidateInvalid("digest profile: %v", err)
	}
	if !prefixedSHA256Pattern.MatchString(epoch.epochDigest) ||
		!rawSHA256Pattern.MatchString(epoch.servingScopeDigest) ||
		!validOpaqueIdentifierWithin(epoch.id, connectorDigestProfileV2MaxIdentifierBytes) {
		return storedCandidateInvalid("profiled epoch identity shape is invalid")
	}
	view := catalogv1.IntegrationBindingView{
		OrgID:                epoch.catalog.orgID,
		AppInstallationID:    epoch.catalog.appInstallationID,
		AuthorizedWorkloadID: epoch.catalog.authorizedWorkloadID,
		Lineage:              epoch.catalog.lineage,
		Bindings:             make([]catalogv1.IntegrationBinding, len(epoch.bindings)),
		Receipts:             epoch.catalog.receipts,
	}
	var bindingReceiptBytes uint64
	for index := range epoch.bindings {
		binding := epoch.bindings[index]
		if binding.binding.ordinal != uint64(index) ||
			binding.binding.servingEpochID != epoch.id ||
			binding.owner.servingEpochBindingID != binding.binding.id ||
			!rawSHA256Pattern.MatchString(binding.binding.id) ||
			!rawSHA256Pattern.MatchString(binding.binding.identityDigest) {
			return storedCandidateInvalid("profiled binding %d identity is invalid", index)
		}
		viewBinding, err := catalogBindingForProfiledValidation(profile, binding.binding)
		if err != nil {
			return storedCandidateInvalid("profiled binding %d: %v", index, err)
		}
		view.Bindings[index] = viewBinding
		proof := BindingOwnerProof{
			identity:                     connectorbindingownerv1.IdentityFromBinding(viewBinding),
			packGenerationID:             binding.owner.packGenerationID,
			definitionEnvelopeVersion:    binding.owner.definitionEnvelopeVersion,
			definitionReceiptVersion:     binding.owner.definitionReceiptVersion,
			canonicalDefinitionDigest:    binding.owner.canonicalDefinitionDigest,
			definitionOwnerReceipt:       binding.owner.definitionOwnerReceipt,
			definitionOwnerReceiptDigest: binding.owner.definitionOwnerReceiptDigest,
		}
		if err := validateBindingOwnerProof(proof); err != nil {
			return storedCandidateInvalid("profiled binding %d owner proof: %v", index, err)
		}
		var ok bool
		bindingReceiptBytes, ok = addWithinLimit(
			bindingReceiptBytes,
			uint64(len(binding.owner.definitionOwnerReceipt)),
			connectorbindingownerv1.MaxDefinitionSetBytes,
		)
		if !ok {
			return storedCandidateInvalid("profiled binding-owner receipts exceed their owner bound")
		}
		expectedIdentity, err := profile.computeBindingIdentityDigest(viewBinding)
		if err != nil || expectedIdentity != binding.binding.identityDigest {
			return storedCandidateInvalid("profiled binding %d identity digest disagrees", index)
		}
		expectedRowID, err := profile.computeBindingRowID(epoch.id, uint64(index), expectedIdentity)
		if err != nil || expectedRowID != binding.binding.id {
			return storedCandidateInvalid("profiled binding %d row ID disagrees", index)
		}
	}
	if err := catalogv1.ValidateIntegrationBindingView(view); err != nil ||
		view.Lineage.ViewDigest != catalogv1.ComputeIntegrationBindingViewDigest(view) {
		return storedCandidateInvalid("profiled Catalog view is invalid")
	}
	expectedScope, err := profile.computeServingScopeDigest(epoch.catalog)
	if err != nil || expectedScope != epoch.servingScopeDigest {
		return storedCandidateInvalid("profiled serving scope digest disagrees")
	}
	if err := validateStoredServiceConfigurationProof(epoch.serviceConfiguration); err != nil {
		return storedCandidateInvalid("profiled ServiceConfiguration proof: %v", err)
	}
	if err := validateStoredCredentialProofs(epoch.credentials); err != nil {
		return storedCandidateInvalid("profiled credential proofs: %v", err)
	}
	expectedDigest, err := computeProfiledEpochDigest(profile, epoch)
	if err != nil || expectedDigest != epoch.epochDigest {
		return storedCandidateInvalid("profiled epoch digest disagrees")
	}
	return nil
}

func catalogBindingForProfiledValidation(
	profile ConnectorDigestProfile,
	binding storedBinding,
) (catalogv1.IntegrationBinding, error) {
	credentialRefs, err := profile.decodeCredentialRefs(binding.credentialRefs)
	if err != nil {
		return catalogv1.IntegrationBinding{}, fmt.Errorf("credential refs: %w", err)
	}
	result := catalogv1.IntegrationBinding{
		Kind:               binding.kind,
		OwnerPack:          binding.ownerPack,
		PackInstallationID: binding.packInstallationID,
		PackPayloadDigest:  binding.packPayloadDigest,
		BusinessKey:        binding.businessKey,
		DefinitionVersion:  binding.definitionVersion,
		CredentialRefs:     credentialRefs,
	}
	runtimeFieldCount := 0
	if binding.runtimeCommandKind != nil {
		runtimeFieldCount++
	}
	if binding.runtimePayloadFormatVersion != nil {
		runtimeFieldCount++
	}
	if binding.runtimeDefinitionRef != nil {
		runtimeFieldCount++
	}
	if binding.runtimeDefinitionVersion != nil {
		runtimeFieldCount++
	}
	if runtimeFieldCount != 0 && runtimeFieldCount != 4 {
		return catalogv1.IntegrationBinding{}, fmt.Errorf("runtime reference is partial")
	}
	if runtimeFieldCount == 4 {
		result.RuntimeCommand = &catalogv1.RuntimeCommandBinding{
			CommandKind:          *binding.runtimeCommandKind,
			PayloadFormatVersion: *binding.runtimePayloadFormatVersion,
			DefinitionRef:        *binding.runtimeDefinitionRef,
			DefinitionVersion:    *binding.runtimeDefinitionVersion,
		}
	}
	return result, nil
}

func validateStoredServiceConfigurationProof(proof storedServiceConfigurationProof) error {
	if _, err := serviceconfiguration.ResolveGenerationProtocolForRead(proof.generationProtocolVersion); err != nil {
		return err
	}
	if _, err := serviceconfiguration.ResolveReceiptProtocolForRead(proof.receiptProtocolVersion); err != nil {
		return err
	}
	if !validOpaqueIdentifier(proof.generationID) ||
		!prefixedSHA256Pattern.MatchString(proof.generationDigest) ||
		!prefixedSHA256Pattern.MatchString(proof.ownerReceiptDigest) ||
		len(proof.ownerReceipt) == 0 || len(proof.ownerReceipt) > serviceconfiguration.MaxCanonicalReceiptBytes {
		return fmt.Errorf("shape is invalid")
	}
	return nil
}

func validateStoredCredentialProofs(proofs []storedCredentialGenerationProof) error {
	if proofs == nil || len(proofs) > credentialgeneration.MaxExactIdentitySetEntries {
		return fmt.Errorf("set is absent or exceeds its owner bound")
	}
	var idBytes, receiptBytes uint64
	for index, proof := range proofs {
		if proof.ordinal != uint64(index) || !validOpaqueIdentifier(proof.generationID) ||
			(index > 0 && proof.generationID <= proofs[index-1].generationID) {
			return fmt.Errorf("entry %d identity or order is invalid", index)
		}
		if _, err := credentialgeneration.ResolveReceiptProtocolForRead(proof.receiptProtocolVersion); err != nil {
			return fmt.Errorf("entry %d receipt protocol: %w", index, err)
		}
		if !prefixedSHA256Pattern.MatchString(proof.transportConfigurationDigest) ||
			!prefixedSHA256Pattern.MatchString(proof.ownerReceiptDigest) || len(proof.ownerReceipt) == 0 {
			return fmt.Errorf("entry %d digest or receipt shape is invalid", index)
		}
		var ok bool
		idBytes, ok = addWithinLimit(
			idBytes,
			uint64(len(proof.generationID)),
			credentialgeneration.MaxExactIdentitySetIDBytes,
		)
		if !ok {
			return fmt.Errorf("generation ID bytes exceed their owner bound")
		}
		receiptBytes, ok = addWithinLimit(
			receiptBytes,
			uint64(len(proof.ownerReceipt)),
			credentialgeneration.MaxExactIdentitySetPayloadBytes,
		)
		if !ok {
			return fmt.Errorf("receipt bytes exceed their owner bound")
		}
	}
	return nil
}

func computeProfiledEpochDigest(
	profile ConnectorDigestProfile,
	epoch ProfiledPreparedEpoch,
) (string, error) {
	if profile.ID() != ConnectorDigestProfileV2ID || epoch.profileID != profile.ID() {
		return "", unsupportedConnectorDigestProfile(profile.ID())
	}
	encoding, err := profile.digestEncoding()
	if err != nil {
		return "", err
	}
	writer := newProfiledDigestWriter(encoding.canonicalPrefix(servingEpochDomainV2))
	writer.string(epoch.id)
	writer.string(epoch.servingScopeDigest)
	catalog := epoch.catalog
	writer.string(catalog.orgID)
	writer.string(catalog.appInstallationID)
	writer.string(catalog.authorizedWorkloadID)
	writer.uint64(catalog.lineage.CatalogRevision)
	writer.uint64(catalog.lineage.SchemaGeneration)
	writer.uint64(catalog.lineage.EngineGeneration)
	writer.string(catalog.lineage.SourceSnapshotDigest)
	writer.string(catalog.lineage.ViewDigest)
	writer.string(catalog.receipts.ProtocolVersion)
	writer.uint64(catalog.receipts.CatalogRevision)
	writer.uint64(catalog.receipts.SchemaGeneration)
	writer.uint64(catalog.receipts.EngineGeneration)
	writer.uint64(catalog.receipts.SecurityEpoch)
	writer.string(catalog.receipts.CapabilityVersion)
	configuration := epoch.serviceConfiguration
	writer.string(string(configuration.generationProtocolVersion))
	writer.string(string(configuration.receiptProtocolVersion))
	writer.string(configuration.generationID)
	writer.string(configuration.generationDigest)
	writer.bytes(configuration.ownerReceipt)
	writer.string(configuration.ownerReceiptDigest)
	writer.uint64(uint64(len(epoch.bindings)))
	for _, profiledBinding := range epoch.bindings {
		binding := profiledBinding.binding
		writer.uint64(binding.ordinal)
		writer.string(binding.id)
		writer.string(binding.identityDigest)
		writer.string(string(binding.kind))
		writer.string(binding.ownerPack)
		writer.string(binding.packInstallationID)
		writer.string(binding.packPayloadDigest)
		writer.string(binding.businessKey)
		writer.uint64(binding.definitionVersion)
		writer.bytes(binding.credentialRefs)
		if binding.runtimeCommandKind == nil {
			writer.byte(0)
		} else {
			writer.byte(1)
			writer.string(*binding.runtimeCommandKind)
			writer.uint32(*binding.runtimePayloadFormatVersion)
			writer.string(*binding.runtimeDefinitionRef)
			writer.string(*binding.runtimeDefinitionVersion)
		}
		owner := profiledBinding.owner
		writer.string(owner.packGenerationID)
		writer.string(owner.definitionEnvelopeVersion)
		writer.string(owner.definitionReceiptVersion)
		writer.string(owner.canonicalDefinitionDigest)
		writer.bytes(owner.definitionOwnerReceipt)
		writer.string(owner.definitionOwnerReceiptDigest)
	}
	writer.uint64(uint64(len(epoch.credentials)))
	for _, credential := range epoch.credentials {
		writer.uint64(credential.ordinal)
		writer.string(credential.generationID)
		writer.string(string(credential.receiptProtocolVersion))
		writer.string(credential.transportConfigurationDigest)
		writer.bytes(credential.ownerReceipt)
		writer.string(credential.ownerReceiptDigest)
	}
	return writer.prefixedDigest(), nil
}

type profiledDigestWriter struct{ hash hash.Hash }

func newProfiledDigestWriter(prefix []byte) *profiledDigestWriter {
	writer := &profiledDigestWriter{hash: sha256.New()}
	_, _ = writer.hash.Write(prefix)
	return writer
}

func (writer *profiledDigestWriter) byte(value byte) { _, _ = writer.hash.Write([]byte{value}) }

func (writer *profiledDigestWriter) uint32(value uint32) {
	var encoded [4]byte
	binary.BigEndian.PutUint32(encoded[:], value)
	_, _ = writer.hash.Write(encoded[:])
}

func (writer *profiledDigestWriter) uint64(value uint64) {
	var encoded [8]byte
	binary.BigEndian.PutUint64(encoded[:], value)
	_, _ = writer.hash.Write(encoded[:])
}

func (writer *profiledDigestWriter) string(value string) { writer.bytes([]byte(value)) }

func (writer *profiledDigestWriter) bytes(value []byte) {
	writer.uint64(uint64(len(value)))
	_, _ = writer.hash.Write(value)
}

func (writer *profiledDigestWriter) prefixedDigest() string {
	return "sha256:" + hex.EncodeToString(writer.hash.Sum(nil))
}