package serving

import (
	"bytes"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"errors"
	"fmt"
	"sort"

	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
)

// ConnectorDigestProfileID identifies one immutable Connector-owned digest
// recipe. IDs are exact and case-sensitive; an empty or unknown ID never means
// the current profile.
type ConnectorDigestProfileID string

const (
	// ConnectorDigestProfileV1ID is the first closed Connector digest profile.
	ConnectorDigestProfileV1ID ConnectorDigestProfileID = "nexuscrm.connector-digest-profile.v1"
	// ConnectorDigestProfileV2ID is the first authoring profile for complete
	// reproducible ServingEpoch candidates. It freezes exact external-owner
	// generation and receipt proofs instead of deriving an L2 receipt digest.
	ConnectorDigestProfileV2ID ConnectorDigestProfileID = "nexuscrm.connector-digest-profile.v2"

	servingScopeDomain          = "nexuscrm.connector-serving-scope.v1"
	bindingIdentityDomain       = "nexuscrm.connector-serving-binding-identity.v1"
	bindingRowDomain            = "nexuscrm.connector-serving-binding-row.v1"
	credentialRefsDomain        = "nexuscrm.connector-serving-credential-refs.v1"
	l2VerificationReceiptDomain = "nexuscrm.connector-serving-l2-verification-receipt.v1"
	servingEpochDomain          = "nexuscrm.connector-serving-epoch.v1"

	servingScopeDomainV2    = "nexuscrm.connector-serving-scope.v2"
	bindingIdentityDomainV2 = "nexuscrm.connector-serving-binding-identity.v2"
	bindingRowDomainV2      = "nexuscrm.connector-serving-binding-row.v2"
	credentialRefsDomainV2  = "nexuscrm.connector-serving-credential-refs.v2"
	servingEpochDomainV2    = "nexuscrm.connector-serving-epoch.v2"

	// These limits are profile-v1 facts. The dependency-drift test requires an
	// explicit new-profile decision if Catalog changes its current admission
	// grammar instead of silently redefining persisted Connector bytes.
	connectorDigestProfileV1MaxIdentifierBytes    = 255
	connectorDigestProfileV1MaxCredentialRefCount = 256
	connectorDigestProfileV2MaxIdentifierBytes    = 255
	connectorDigestProfileV2MaxCredentialRefCount = 256

	// maxPersistedL2VerificationReceiptBytes is a Connector-owned admission
	// contract, not the MEDIUMBLOB physical ceiling. One MiB leaves five MiB of
	// stock TiDB's six-MiB single-entry budget for row-encoding overhead, while
	// matching Catalog views' own one-MiB total semantic budget.
	maxPersistedL2VerificationReceiptBytes = 1 << 20
)

var (
	// ErrUnsupportedConnectorDigestProfile reports an omitted or exact profile
	// ID for which the requested authoring or retained-read recipe is absent.
	ErrUnsupportedConnectorDigestProfile = errors.New("connector serving: unsupported digest profile")
)

// ConnectorDigestProfile is a closed, immutable Connector-owned canonical
// recipe. Valid values can only be obtained through an exact resolver.
//
// Profiled recipes use purpose-domain bytes followed by NUL and the uint64-BE
// length-framed profile ID. They frame text and binary values with uint64-BE
// byte lengths, counts and ordinals with uint64-BE integers, Runtime payload
// versions with uint32-BE integers, and optional Runtime tuples with exact 0/1
// presence bytes. Text is exact UTF-8 with no trimming, case folding, or
// Unicode normalization; binary owner receipts are copied verbatim. SHA-256
// output is lowercase raw hex for scope/identity/row keys and lowercase
// `sha256:`-prefixed hex for epoch digests. Each profile has distinct purpose
// domains and an immutable included-field order.
//
// External-owner identities, digests, and receipts are opaque inputs. This
// profile never reinterprets Catalog, pack, L2, or credential canonicalization.
type ConnectorDigestProfile struct {
	id ConnectorDigestProfileID
}

// ResolveConnectorDigestProfileForAuthoring selects an exact profile permitted
// to author new facts. There is deliberately no default, alias, fallback, or
// runtime registration path.
func ResolveConnectorDigestProfileForAuthoring(id ConnectorDigestProfileID) (ConnectorDigestProfile, error) {
	switch id {
	case ConnectorDigestProfileV2ID:
		return ConnectorDigestProfile{id: id}, nil
	default:
		return ConnectorDigestProfile{}, unsupportedConnectorDigestProfile(id)
	}
}

// ResolveConnectorDigestProfileForRead selects an exact retained reader.
// Authoring support may be retired independently, but this switch must retain a
// profile while any active, nonterminal, resumable, duplicate-resolution,
// replay, reconciliation, recovery, audit, or otherwise retained fact can
// reference it. Removal requires a complete durable-reference inventory; age
// or a timeout is not evidence that removal is safe.
func ResolveConnectorDigestProfileForRead(id ConnectorDigestProfileID) (ConnectorDigestProfile, error) {
	switch id {
	case ConnectorDigestProfileV1ID, ConnectorDigestProfileV2ID:
		return ConnectorDigestProfile{id: id}, nil
	default:
		return ConnectorDigestProfile{}, unsupportedConnectorDigestProfile(id)
	}
}

// ID returns the exact immutable profile ID. The zero value returns an empty ID
// and is unsupported by both resolvers.
func (profile ConnectorDigestProfile) ID() ConnectorDigestProfileID { return profile.id }

// CompatibleWith reports exact readable-profile equality. Digest text is never
// compared across profiles, even when two strings happen to be identical.
func (profile ConnectorDigestProfile) CompatibleWith(other ConnectorDigestProfile) bool {
	if profile.id != other.id {
		return false
	}
	_, err := ResolveConnectorDigestProfileForRead(profile.id)
	return err == nil
}

func unsupportedConnectorDigestProfile(id ConnectorDigestProfileID) error {
	return fmt.Errorf("%w: %q", ErrUnsupportedConnectorDigestProfile, id)
}

// connectorDigestEncoding carries the envelope fact that changes canonical
// bytes. The empty encoding is deliberately internal and preserves only the
// legacy, unprofiled prepared-candidate bytes introduced before profile
// persistence. It is not resolvable and must never be inferred as profile v1.
type connectorDigestEncoding struct {
	profileID ConnectorDigestProfileID
}

var legacyUnprofiledConnectorDigestEncoding connectorDigestEncoding

func (profile ConnectorDigestProfile) digestEncoding() (connectorDigestEncoding, error) {
	resolved, err := ResolveConnectorDigestProfileForRead(profile.id)
	if err != nil {
		return connectorDigestEncoding{}, err
	}
	return connectorDigestEncoding{profileID: resolved.id}, nil
}

func (encoding connectorDigestEncoding) canonicalPrefix(domain string) []byte {
	capacity := len(domain) + 1
	if encoding.profileID != "" {
		capacity += 8 + len(encoding.profileID)
	}
	result := make([]byte, 0, capacity)
	result = append(result, domain...)
	result = append(result, 0)
	if encoding.profileID != "" {
		result = appendCanonicalString(result, string(encoding.profileID))
	}
	return result
}

func (encoding connectorDigestEncoding) servingScopeDomain() string {
	if encoding.profileID == ConnectorDigestProfileV2ID {
		return servingScopeDomainV2
	}
	return servingScopeDomain
}

func (encoding connectorDigestEncoding) bindingIdentityDomain() string {
	if encoding.profileID == ConnectorDigestProfileV2ID {
		return bindingIdentityDomainV2
	}
	return bindingIdentityDomain
}

func (encoding connectorDigestEncoding) bindingRowDomain() string {
	if encoding.profileID == ConnectorDigestProfileV2ID {
		return bindingRowDomainV2
	}
	return bindingRowDomain
}

func (encoding connectorDigestEncoding) credentialRefsDomain() string {
	if encoding.profileID == ConnectorDigestProfileV2ID {
		return credentialRefsDomainV2
	}
	return credentialRefsDomain
}

func (encoding connectorDigestEncoding) maxCredentialRefCount() int {
	if encoding.profileID == ConnectorDigestProfileV2ID {
		return connectorDigestProfileV2MaxCredentialRefCount
	}
	return connectorDigestProfileV1MaxCredentialRefCount
}

func (encoding connectorDigestEncoding) maxIdentifierBytes() int {
	if encoding.profileID == ConnectorDigestProfileV2ID {
		return connectorDigestProfileV2MaxIdentifierBytes
	}
	return connectorDigestProfileV1MaxIdentifierBytes
}

func canonicalPrefix(domain string) []byte {
	return legacyUnprofiledConnectorDigestEncoding.canonicalPrefix(domain)
}

func computeServingScopeDigest(reference catalogReference) string {
	return computeServingScopeDigestWithEncoding(legacyUnprofiledConnectorDigestEncoding, reference)
}

func (profile ConnectorDigestProfile) computeServingScopeDigest(reference catalogReference) (string, error) {
	encoding, err := profile.digestEncoding()
	if err != nil {
		return "", err
	}
	return computeServingScopeDigestWithEncoding(encoding, reference), nil
}

func computeServingScopeDigestWithEncoding(
	encoding connectorDigestEncoding,
	reference catalogReference,
) string {
	preimage := encoding.canonicalPrefix(encoding.servingScopeDomain())
	preimage = appendCanonicalString(preimage, reference.orgID)
	preimage = appendCanonicalString(preimage, reference.appInstallationID)
	preimage = appendCanonicalString(preimage, reference.authorizedWorkloadID)
	return rawDigest(preimage)
}

func computeBindingIdentityDigest(binding catalogv1.IntegrationBinding) string {
	return computeBindingIdentityDigestWithEncoding(legacyUnprofiledConnectorDigestEncoding, binding)
}

func (profile ConnectorDigestProfile) computeBindingIdentityDigest(
	binding catalogv1.IntegrationBinding,
) (string, error) {
	encoding, err := profile.digestEncoding()
	if err != nil {
		return "", err
	}
	return computeBindingIdentityDigestWithEncoding(encoding, binding), nil
}

func computeBindingIdentityDigestWithEncoding(
	encoding connectorDigestEncoding,
	binding catalogv1.IntegrationBinding,
) string {
	preimage := encoding.canonicalPrefix(encoding.bindingIdentityDomain())
	preimage = appendCanonicalString(preimage, string(binding.Kind))
	preimage = appendCanonicalString(preimage, binding.OwnerPack)
	preimage = appendCanonicalString(preimage, binding.PackInstallationID)
	preimage = appendCanonicalString(preimage, binding.BusinessKey)
	return rawDigest(preimage)
}

func computeBindingRowID(epochID string, ordinal uint64, bindingIdentityDigest string) string {
	return computeBindingRowIDWithEncoding(
		legacyUnprofiledConnectorDigestEncoding,
		epochID,
		ordinal,
		bindingIdentityDigest,
	)
}

func (profile ConnectorDigestProfile) computeBindingRowID(
	epochID string,
	ordinal uint64,
	bindingIdentityDigest string,
) (string, error) {
	encoding, err := profile.digestEncoding()
	if err != nil {
		return "", err
	}
	return computeBindingRowIDWithEncoding(encoding, epochID, ordinal, bindingIdentityDigest), nil
}

func computeBindingRowIDWithEncoding(
	encoding connectorDigestEncoding,
	epochID string,
	ordinal uint64,
	bindingIdentityDigest string,
) string {
	preimage := encoding.canonicalPrefix(encoding.bindingRowDomain())
	preimage = appendCanonicalString(preimage, epochID)
	preimage = binary.BigEndian.AppendUint64(preimage, ordinal)
	preimage = appendCanonicalString(preimage, bindingIdentityDigest)
	return rawDigest(preimage)
}

func computeL2VerificationReceiptDigest(receipt []byte) string {
	return computeL2VerificationReceiptDigestWithEncoding(legacyUnprofiledConnectorDigestEncoding, receipt)
}

func (profile ConnectorDigestProfile) computeL2VerificationReceiptDigest(receipt []byte) (string, error) {
	if profile.id != ConnectorDigestProfileV1ID {
		return "", unsupportedConnectorDigestProfile(profile.id)
	}
	encoding, err := profile.digestEncoding()
	if err != nil {
		return "", err
	}
	return computeL2VerificationReceiptDigestWithEncoding(encoding, receipt), nil
}

func computeL2VerificationReceiptDigestWithEncoding(
	encoding connectorDigestEncoding,
	receipt []byte,
) string {
	preimage := encoding.canonicalPrefix(l2VerificationReceiptDomain)
	preimage = appendCanonicalBytes(preimage, receipt)
	return prefixedDigest(preimage)
}

func computeEpochDigest(epoch PreparedEpoch) string {
	return computeEpochDigestWithEncoding(legacyUnprofiledConnectorDigestEncoding, epoch)
}

func (profile ConnectorDigestProfile) computeEpochDigest(epoch PreparedEpoch) (string, error) {
	if profile.id != ConnectorDigestProfileV1ID {
		return "", unsupportedConnectorDigestProfile(profile.id)
	}
	encoding, err := profile.digestEncoding()
	if err != nil {
		return "", err
	}
	return computeEpochDigestWithEncoding(encoding, epoch), nil
}

func computeEpochDigestWithEncoding(encoding connectorDigestEncoding, epoch PreparedEpoch) string {
	catalog := epoch.catalog
	preimage := encoding.canonicalPrefix(servingEpochDomain)
	preimage = appendCanonicalString(preimage, epoch.id)
	preimage = appendCanonicalString(preimage, epoch.servingScopeDigest)
	preimage = appendCanonicalString(preimage, catalog.orgID)
	preimage = appendCanonicalString(preimage, catalog.appInstallationID)
	preimage = appendCanonicalString(preimage, catalog.authorizedWorkloadID)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.lineage.CatalogRevision)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.lineage.SchemaGeneration)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.lineage.EngineGeneration)
	preimage = appendCanonicalString(preimage, catalog.lineage.SourceSnapshotDigest)
	preimage = appendCanonicalString(preimage, catalog.lineage.ViewDigest)
	preimage = appendCanonicalString(preimage, catalog.receipts.ProtocolVersion)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.receipts.CatalogRevision)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.receipts.SchemaGeneration)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.receipts.EngineGeneration)
	preimage = binary.BigEndian.AppendUint64(preimage, catalog.receipts.SecurityEpoch)
	preimage = appendCanonicalString(preimage, catalog.receipts.CapabilityVersion)
	preimage = binary.BigEndian.AppendUint64(preimage, uint64(len(epoch.bindings)))
	for i := range epoch.bindings {
		binding := epoch.bindings[i]
		preimage = binary.BigEndian.AppendUint64(preimage, binding.ordinal)
		preimage = appendCanonicalString(preimage, binding.id)
		preimage = appendCanonicalString(preimage, binding.identityDigest)
		preimage = appendCanonicalString(preimage, string(binding.kind))
		preimage = appendCanonicalString(preimage, binding.ownerPack)
		preimage = appendCanonicalString(preimage, binding.packInstallationID)
		preimage = appendCanonicalString(preimage, binding.packPayloadDigest)
		preimage = appendCanonicalString(preimage, binding.businessKey)
		preimage = binary.BigEndian.AppendUint64(preimage, binding.definitionVersion)
		preimage = appendCanonicalBytes(preimage, binding.credentialRefs)
		if binding.runtimeCommandKind == nil {
			preimage = append(preimage, 0)
			continue
		}
		preimage = append(preimage, 1)
		preimage = appendCanonicalString(preimage, *binding.runtimeCommandKind)
		preimage = binary.BigEndian.AppendUint32(preimage, *binding.runtimePayloadFormatVersion)
		preimage = appendCanonicalString(preimage, *binding.runtimeDefinitionRef)
		preimage = appendCanonicalString(preimage, *binding.runtimeDefinitionVersion)
	}
	preimage = appendCanonicalString(preimage, epoch.l2.configurationGenerationRef)
	preimage = appendCanonicalString(preimage, epoch.l2.configurationDigest)
	preimage = appendCanonicalBytes(preimage, epoch.l2.verificationReceipt)
	preimage = appendCanonicalString(preimage, epoch.l2.verificationReceiptDigest)
	return prefixedDigest(preimage)
}

// encodeCredentialRefs emits a self-identifying legacy canonical blob:
// versioned domain, zero delimiter, uint64-BE count, then uint64-BE byte-length
// frames. The profiled form additionally frames its exact profile ID after the
// delimiter. The legacy form remains only for inert pre-extension candidates.
func encodeCredentialRefs(values []string) ([]byte, error) {
	return encodeCredentialRefsWithEncoding(legacyUnprofiledConnectorDigestEncoding, values)
}

func (profile ConnectorDigestProfile) encodeCredentialRefs(values []string) ([]byte, error) {
	encoding, err := profile.digestEncoding()
	if err != nil {
		return nil, err
	}
	return encodeCredentialRefsWithEncoding(encoding, values)
}

func encodeCredentialRefsWithEncoding(
	encoding connectorDigestEncoding,
	values []string,
) ([]byte, error) {
	maxCredentialRefCount := encoding.maxCredentialRefCount()
	if len(values) > maxCredentialRefCount {
		return nil, fmt.Errorf(
			"credential reference count exceeds %d",
			maxCredentialRefCount,
		)
	}
	if !sort.StringsAreSorted(values) {
		return nil, fmt.Errorf("credential references are not ordered")
	}
	preimage := encoding.canonicalPrefix(encoding.credentialRefsDomain())
	preimage = binary.BigEndian.AppendUint64(preimage, uint64(len(values)))
	for i, value := range values {
		if !validOpaqueIdentifierWithin(value, encoding.maxIdentifierBytes()) {
			return nil, fmt.Errorf("credential reference %d is not canonical", i)
		}
		if i > 0 && value == values[i-1] {
			return nil, fmt.Errorf("credential reference %d is duplicated", i)
		}
		preimage = appendCanonicalString(preimage, value)
	}
	return preimage, nil
}

func decodeCredentialRefs(encoded []byte) ([]string, error) {
	return decodeCredentialRefsWithEncoding(legacyUnprofiledConnectorDigestEncoding, encoded)
}

func (profile ConnectorDigestProfile) decodeCredentialRefs(encoded []byte) ([]string, error) {
	encoding, err := profile.digestEncoding()
	if err != nil {
		return nil, err
	}
	return decodeCredentialRefsWithEncoding(encoding, encoded)
}

func decodeCredentialRefsWithEncoding(
	encoding connectorDigestEncoding,
	encoded []byte,
) ([]string, error) {
	prefix := encoding.canonicalPrefix(encoding.credentialRefsDomain())
	if !bytes.HasPrefix(encoded, prefix) {
		return nil, fmt.Errorf("credential reference domain or profile disagrees")
	}
	offset := len(prefix)
	count, ok := consumeUint64(encoded, &offset)
	if !ok || count > uint64(encoding.maxCredentialRefCount()) {
		return nil, fmt.Errorf("credential reference count is invalid")
	}
	values := make([]string, 0, int(count))
	for i := uint64(0); i < count; i++ {
		valueBytes, ok := consumeCanonicalBytes(encoded, &offset)
		if !ok {
			return nil, fmt.Errorf("credential reference %d frame is invalid", i)
		}
		value := string(valueBytes)
		if !validOpaqueIdentifierWithin(value, encoding.maxIdentifierBytes()) {
			return nil, fmt.Errorf("credential reference %d is not canonical", i)
		}
		if len(values) > 0 && value <= values[len(values)-1] {
			return nil, fmt.Errorf("credential references are not strictly ordered")
		}
		values = append(values, value)
	}
	if offset != len(encoded) {
		return nil, fmt.Errorf("credential reference blob has trailing bytes")
	}
	return values, nil
}

func appendCanonicalString(destination []byte, value string) []byte {
	return appendCanonicalBytes(destination, []byte(value))
}

func appendCanonicalBytes(destination, value []byte) []byte {
	destination = binary.BigEndian.AppendUint64(destination, uint64(len(value)))
	return append(destination, value...)
}

func consumeUint64(encoded []byte, offset *int) (uint64, bool) {
	if offset == nil || *offset < 0 || len(encoded)-*offset < 8 {
		return 0, false
	}
	value := binary.BigEndian.Uint64(encoded[*offset : *offset+8])
	*offset += 8
	return value, true
}

func consumeCanonicalBytes(encoded []byte, offset *int) ([]byte, bool) {
	length, ok := consumeUint64(encoded, offset)
	if !ok || length > uint64(len(encoded)-*offset) {
		return nil, false
	}
	end := *offset + int(length)
	value := encoded[*offset:end]
	*offset = end
	return value, true
}

func rawDigest(preimage []byte) string {
	sum := sha256.Sum256(preimage)
	return hex.EncodeToString(sum[:])
}

func prefixedDigest(preimage []byte) string { return "sha256:" + rawDigest(preimage) }