package serving

import (
	"bytes"
	"encoding/hex"
	"errors"
	"fmt"
	"strings"
	"testing"

	"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 TestProfiledPreparedEpochV2ConformanceGolden(t *testing.T) {
	fixture := newProfiledCanonicalFixture(t)
	epoch := mustNewProfiledPreparedEpoch(t, fixture)

	if epoch.DigestProfileID() != ConnectorDigestProfileV2ID ||
		epoch.ID() != "epoch-profile-v2" || epoch.State() != preparedState ||
		epoch.BindingCount() != 2 || epoch.CredentialGenerationCount() != 2 {
		t.Fatalf(
			"profiled identity = profile:%q id:%q state:%q bindings:%d credentials:%d",
			epoch.DigestProfileID(),
			epoch.ID(),
			epoch.State(),
			epoch.BindingCount(),
			epoch.CredentialGenerationCount(),
		)
	}
	if _, active := epoch.ActiveScopeDigest(); active {
		t.Fatal("prepared profiled epoch exposed an active scope")
	}
	if epoch.serviceConfiguration.generationID != fixture.configuration.generationID ||
		epoch.serviceConfiguration.generationDigest != fixture.configuration.generationDigest ||
		!bytes.Equal(epoch.serviceConfiguration.ownerReceipt, fixture.configuration.ownerReceipt) ||
		epoch.serviceConfiguration.ownerReceiptDigest != fixture.configuration.ownerReceiptDigest {
		t.Fatal("profiled epoch did not preserve the exact ServiceConfiguration owner proof")
	}
	if epoch.credentials[0].generationID != "Generation-A" || epoch.credentials[0].ordinal != 0 ||
		epoch.credentials[1].generationID != "generation-z" || epoch.credentials[1].ordinal != 1 {
		t.Fatalf("credential proof order = %#v", epoch.credentials)
	}
	for index := range epoch.bindings {
		if epoch.bindings[index].owner.servingEpochBindingID != epoch.bindings[index].binding.id {
			t.Fatalf("binding %d owner proof is not keyed by the derived binding row", index)
		}
	}

	profile, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV2ID)
	if err != nil {
		t.Fatal(err)
	}
	encoding, err := profile.digestEncoding()
	if err != nil {
		t.Fatal(err)
	}
	goldens := map[string][2]string{
		"profile prefix": {
			hex.EncodeToString(encoding.canonicalPrefix(servingEpochDomainV2)),
			"6e6578757363726d2e636f6e6e6563746f722d73657276696e672d65706f63682e76320000000000000000246e6578757363726d2e636f6e6e6563746f722d6469676573742d70726f66696c652e7632",
		},
		"serving scope": {
			epoch.servingScopeDigest,
			"1083c41085a584031b39f9427eb3c5f0e9294babd0a56ecb91012869bd5b68a4",
		},
		"binding 0 identity": {
			epoch.bindings[0].binding.identityDigest,
			"40ce18da83df1f345a2f9f0838b45f72abdad247153c161ecfac644fb89ea053",
		},
		"binding 0 row": {
			epoch.bindings[0].binding.id,
			"2c7983421db47fa22b949329b96fb5aa043284f2381da07176290b85a561277f",
		},
		"binding 0 credential refs": {
			hex.EncodeToString(epoch.bindings[0].binding.credentialRefs),
			"6e6578757363726d2e636f6e6e6563746f722d73657276696e672d63726564656e7469616c2d726566732e76320000000000000000246e6578757363726d2e636f6e6e6563746f722d6469676573742d70726f66696c652e76320000000000000002000000000000000c63726564656e7469616c2d61000000000000000c63726564656e7469616c2d62",
		},
		"binding 1 identity": {
			epoch.bindings[1].binding.identityDigest,
			"14de9f2f73a6cfe7ea7651dc873055a59411427c0fe283ec1b75dd7ef5f84a3d",
		},
		"binding 1 row": {
			epoch.bindings[1].binding.id,
			"32191b43f206df2ada7db83d97df003cee2b907ea17c9842359f08130a679545",
		},
		"epoch": {
			epoch.epochDigest,
			"sha256:cee1bb5e5e7ddb4dee82b6d724687887059ec4769a4fc576d3ef29e318120381",
		},
	}
	for name, values := range goldens {
		if values[0] != values[1] {
			t.Errorf("%s golden = %q, want %q", name, values[0], values[1])
		}
	}
	if err := epoch.validateStored(); err != nil {
		t.Fatalf("validateStored(profiled golden): %v", err)
	}
}

func TestReconstructProfiledPreparedEpochForReadPreservesRetainedCandidate(t *testing.T) {
	fixture := newProfiledCanonicalFixture(t)
	retainedProfile, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV2ID)
	if err != nil {
		t.Fatalf("ResolveConnectorDigestProfileForRead(): %v", err)
	}
	written, err := deriveProfiledPreparedEpoch(
		retainedProfile,
		"epoch-profile-v2",
		fixture.view,
		fixture.configuration,
		fixture.bindingOwners,
		fixture.credentials,
	)
	if err != nil {
		t.Fatalf("deriveProfiledPreparedEpoch(retained profile): %v", err)
	}

	reconstructed, err := reconstructProfiledPreparedEpochForRead(
		written.profileID,
		written.id,
		fixture.view,
		fixture.configuration,
		fixture.bindingOwners,
		fixture.credentials,
	)
	if err != nil {
		t.Fatalf("reconstructProfiledPreparedEpochForRead(): %v", err)
	}
	if !written.equal(reconstructed) {
		t.Fatal("retained-read reconstruction changed the persisted candidate")
	}
}

func TestProfiledEpochDigestCoversEveryOwnerProofAxis(t *testing.T) {
	base := mustNewProfiledPreparedEpoch(t, newProfiledCanonicalFixture(t))
	profile, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV2ID)
	if err != nil {
		t.Fatal(err)
	}
	tests := []struct {
		name   string
		mutate func(*ProfiledPreparedEpoch)
	}{
		{name: "ServiceConfiguration generation protocol", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.serviceConfiguration.generationProtocolVersion = "generation-protocol-mutated"
		}},
		{name: "ServiceConfiguration receipt protocol", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.serviceConfiguration.receiptProtocolVersion = "receipt-protocol-mutated"
		}},
		{name: "ServiceConfiguration generation ID", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.serviceConfiguration.generationID += "-mutated"
		}},
		{name: "ServiceConfiguration generation digest", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.serviceConfiguration.generationDigest = profiledTestDigest('1')
		}},
		{name: "ServiceConfiguration owner receipt", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.serviceConfiguration.ownerReceipt[0] ^= 0xff
		}},
		{name: "ServiceConfiguration owner receipt digest", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.serviceConfiguration.ownerReceiptDigest = profiledTestDigest('2')
		}},
		{name: "binding owner PackGeneration ID", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings[0].owner.packGenerationID += "-mutated"
		}},
		{name: "binding owner envelope version", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings[0].owner.definitionEnvelopeVersion = "definition-envelope-mutated"
		}},
		{name: "binding owner receipt version", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings[0].owner.definitionReceiptVersion = "definition-receipt-mutated"
		}},
		{name: "binding owner canonical definition digest", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings[0].owner.canonicalDefinitionDigest = profiledTestDigest('3')
		}},
		{name: "binding owner receipt", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings[0].owner.definitionOwnerReceipt[0] ^= 0xff
		}},
		{name: "binding owner receipt digest", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings[0].owner.definitionOwnerReceiptDigest = profiledTestDigest('4')
		}},
		{name: "binding proof count", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.bindings = epoch.bindings[:1]
		}},
		{name: "credential ordinal", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials[0].ordinal++
		}},
		{name: "credential generation ID", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials[0].generationID += "-mutated"
		}},
		{name: "credential receipt protocol", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials[0].receiptProtocolVersion = "credential-receipt-mutated"
		}},
		{name: "credential transport digest", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials[0].transportConfigurationDigest = profiledTestDigest('5')
		}},
		{name: "credential owner receipt", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials[0].ownerReceipt[0] ^= 0xff
		}},
		{name: "credential owner receipt digest", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials[0].ownerReceiptDigest = profiledTestDigest('6')
		}},
		{name: "credential proof count", mutate: func(epoch *ProfiledPreparedEpoch) {
			epoch.credentials = epoch.credentials[:1]
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			mutated := base.copy()
			test.mutate(&mutated)
			digest, err := computeProfiledEpochDigest(profile, mutated)
			if err != nil {
				t.Fatalf("computeProfiledEpochDigest(): %v", err)
			}
			if digest == base.epochDigest {
				t.Fatalf("profile-v2 epoch digest omitted %s", test.name)
			}
			if err := mutated.validateStored(); !errors.Is(err, ErrStoredCandidateInvalid) {
				t.Fatalf("validateStored(mutated %s) error = %v, want ErrStoredCandidateInvalid", test.name, err)
			}
		})
	}
}

func TestProfiledPreparedEpochRequiresExactBindingOwnerSet(t *testing.T) {
	tests := []struct {
		name   string
		mutate func([]BindingOwnerProof) []BindingOwnerProof
		want   string
	}{
		{name: "missing", mutate: func(proofs []BindingOwnerProof) []BindingOwnerProof {
			return proofs[:1]
		}, want: "proof count is 1, want 2"},
		{name: "extra", mutate: func(proofs []BindingOwnerProof) []BindingOwnerProof {
			return append(proofs, proofs[0])
		}, want: "proof count is 3, want 2"},
		{name: "duplicate", mutate: func(proofs []BindingOwnerProof) []BindingOwnerProof {
			return []BindingOwnerProof{proofs[0], proofs[0]}
		}, want: "duplicate identity"},
		{name: "wrong exact identity", mutate: func(proofs []BindingOwnerProof) []BindingOwnerProof {
			proofs[1].identity.BusinessKey = "connector-pack/missing"
			return proofs
		}, want: "proof is missing"},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			fixture := newProfiledCanonicalFixture(t)
			fixture.bindingOwners = test.mutate(fixture.bindingOwners)
			_, err := newProfiledPreparedEpoch(
				ConnectorDigestProfileV2ID,
				"epoch-profile-v2",
				fixture.view,
				fixture.configuration,
				fixture.bindingOwners,
				fixture.credentials,
			)
			if !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("newProfiledPreparedEpoch() error = %v, want ErrInvalidCandidate containing %q", err, test.want)
			}
		})
	}
}

func TestProfiledPreparedEpochRequiresExactCredentialGenerationSet(t *testing.T) {
	fixture := newProfiledCanonicalFixture(t)
	epoch := mustNewProfiledPreparedEpoch(t, fixture)
	if got := []string{epoch.credentials[0].generationID, epoch.credentials[1].generationID}; got[0] != "Generation-A" || got[1] != "generation-z" {
		t.Fatalf("bytewise credential order = %#v", got)
	}
	if epoch.credentials[0].ordinal != 0 || epoch.credentials[1].ordinal != 1 {
		t.Fatalf("credential ordinals = %d/%d, want 0/1", epoch.credentials[0].ordinal, epoch.credentials[1].ordinal)
	}

	tests := []struct {
		name   string
		mutate func([]profiledCredentialGenerationInput) []profiledCredentialGenerationInput
		want   string
	}{
		{name: "nil", mutate: func([]profiledCredentialGenerationInput) []profiledCredentialGenerationInput {
			return nil
		}, want: "must be present"},
		{name: "missing", mutate: func(values []profiledCredentialGenerationInput) []profiledCredentialGenerationInput {
			return values[:1]
		}, want: "missing or contains an extra ID"},
		{name: "extra", mutate: func(values []profiledCredentialGenerationInput) []profiledCredentialGenerationInput {
			extra := profiledCredentialInput("generation-extra", "credential-a", "account-1", "provider-1")
			return append(values, extra)
		}, want: "missing or contains an extra ID"},
		{name: "duplicate", mutate: func(values []profiledCredentialGenerationInput) []profiledCredentialGenerationInput {
			return append(values, values[0])
		}, want: "duplicate ID"},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			candidate := newProfiledCanonicalFixture(t)
			candidate.credentials = test.mutate(candidate.credentials)
			_, err := newProfiledPreparedEpoch(
				ConnectorDigestProfileV2ID,
				"epoch-profile-v2",
				candidate.view,
				candidate.configuration,
				candidate.bindingOwners,
				candidate.credentials,
			)
			if !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("newProfiledPreparedEpoch() error = %v, want ErrInvalidCandidate containing %q", err, test.want)
			}
		})
	}
}

func TestProfiledPreparedEpochRejectsServiceConfigurationScopeAndLineageMismatch(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*profiledServiceConfigurationInput)
	}{
		{name: "organization", mutate: func(input *profiledServiceConfigurationInput) { input.orgID += "-other" }},
		{name: "application installation", mutate: func(input *profiledServiceConfigurationInput) {
			input.appInstallationID = "01900000-0000-7000-8000-000000000099"
		}},
		{name: "authorized workload", mutate: func(input *profiledServiceConfigurationInput) { input.authorizedWorkloadID += "-other" }},
		{name: "Catalog revision", mutate: func(input *profiledServiceConfigurationInput) { input.lineage.CatalogRevision++ }},
		{name: "source snapshot", mutate: func(input *profiledServiceConfigurationInput) {
			input.lineage.SourceSnapshotDigest = profiledTestDigest('7')
		}},
		{name: "Catalog receipt", mutate: func(input *profiledServiceConfigurationInput) { input.receipts.SecurityEpoch++ }},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			fixture := newProfiledCanonicalFixture(t)
			test.mutate(&fixture.configuration)
			_, err := newProfiledPreparedEpoch(
				ConnectorDigestProfileV2ID,
				"epoch-profile-v2",
				fixture.view,
				fixture.configuration,
				fixture.bindingOwners,
				fixture.credentials,
			)
			if !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), "Catalog authority disagrees") {
				t.Fatalf("newProfiledPreparedEpoch() error = %v, want exact authority mismatch", err)
			}
		})
	}
}

func TestProfiledOwnerReceiptsAreOpaqueAndDefensivelyCopied(t *testing.T) {
	fixture := newProfiledCanonicalFixture(t)
	serviceReceipt := append([]byte(nil), fixture.configuration.ownerReceipt...)
	bindingReceipt := append([]byte(nil), fixture.bindingOwners[0].definitionOwnerReceipt...)
	credentialReceipt := append([]byte(nil), fixture.credentials[0].ownerReceipt...)
	if rehashed := computeL2VerificationReceiptDigest(serviceReceipt); rehashed == fixture.configuration.ownerReceiptDigest {
		t.Fatalf("test fixture accidentally equates external owner digest with Connector L2 rehash: %q", rehashed)
	}

	epoch := mustNewProfiledPreparedEpoch(t, fixture)
	fixture.configuration.ownerReceipt[0] ^= 0xff
	fixture.bindingOwners[0].definitionOwnerReceipt[0] ^= 0xff
	fixture.credentials[0].ownerReceipt[0] ^= 0xff
	if !bytes.Equal(epoch.serviceConfiguration.ownerReceipt, serviceReceipt) ||
		!bytes.Equal(epoch.bindings[0].owner.definitionOwnerReceipt, bindingReceipt) ||
		!bytes.Equal(epoch.credentials[1].ownerReceipt, credentialReceipt) {
		t.Fatal("caller receipt mutation crossed the profiled candidate boundary")
	}
	if epoch.serviceConfiguration.ownerReceiptDigest != profiledTestDigest('d') {
		t.Fatalf("ServiceConfiguration receipt digest = %q, want exact opaque owner digest", epoch.serviceConfiguration.ownerReceiptDigest)
	}

	copy := epoch.copy()
	copy.serviceConfiguration.ownerReceipt[0] ^= 0xff
	copy.bindings[0].owner.definitionOwnerReceipt[0] ^= 0xff
	copy.credentials[0].ownerReceipt[0] ^= 0xff
	if !bytes.Equal(epoch.serviceConfiguration.ownerReceipt, serviceReceipt) ||
		!bytes.Equal(epoch.bindings[0].owner.definitionOwnerReceipt, bindingReceipt) ||
		bytes.Equal(epoch.credentials[0].ownerReceipt, copy.credentials[0].ownerReceipt) {
		t.Fatal("ProfiledPreparedEpoch.copy aliases retained receipt bytes")
	}

	changed := newProfiledCanonicalFixture(t)
	changed.configuration.ownerReceiptDigest = profiledTestDigest('8')
	changedEpoch := mustNewProfiledPreparedEpoch(t, changed)
	if changedEpoch.serviceConfiguration.ownerReceiptDigest != profiledTestDigest('8') ||
		changedEpoch.epochDigest == epoch.epochDigest {
		t.Fatal("profile v2 rehashed or omitted an opaque external owner digest")
	}
}

func TestNewBindingOwnerProofDetachesEnvelopeReceipt(t *testing.T) {
	view := profiledTestBindingView()
	envelope := profiledBindingEnvelope(t, view.Bindings[0], []byte(`{"definition":"detached"}`))
	want := append([]byte(nil), envelope.OwnerReceipt...)
	proof, err := NewBindingOwnerProof("pack-generation-detached", envelope)
	if err != nil {
		t.Fatal(err)
	}
	envelope.OwnerReceipt[0] ^= 0xff
	if !bytes.Equal(proof.definitionOwnerReceipt, want) {
		t.Fatal("NewBindingOwnerProof retained caller-owned receipt bytes")
	}
}

func TestProfiledPreparedEpochHonorsOwnerLimits(t *testing.T) {
	t.Run("frozen constants", func(t *testing.T) {
		if serviceconfiguration.MaxCanonicalReceiptBytes != 4<<10 ||
			serviceconfiguration.MaxCanonicalGenerationBytes != 8<<20 ||
			credentialgeneration.MaxTransportConfigurationBytes != 1<<20 ||
			credentialgeneration.MaxExactIdentitySetEntries != 4_096 ||
			credentialgeneration.MaxExactIdentitySetIDBytes != 512<<10 ||
			credentialgeneration.MaxExactIdentitySetIdentityBytes != 4<<20 ||
			credentialgeneration.MaxExactIdentitySetPayloadBytes != 16<<20 ||
			connectorbindingownerv1.MaxDefinitionCount != 4_096 ||
			connectorbindingownerv1.MaxCanonicalDefinitionBytes != 1<<20 ||
			connectorbindingownerv1.MaxSerializedEnvelopeBytes != 2<<20 ||
			connectorbindingownerv1.MaxDefinitionSetBytes != 64<<20 {
			t.Fatal("an external owner bound changed without an explicit profile-v2 decision")
		}
	})

	t.Run("ServiceConfiguration receipt", func(t *testing.T) {
		atLimit := newProfiledCanonicalFixture(t)
		atLimit.configuration.ownerReceipt = bytes.Repeat([]byte{'s'}, serviceconfiguration.MaxCanonicalReceiptBytes)
		mustNewProfiledPreparedEpoch(t, atLimit)

		over := newProfiledCanonicalFixture(t)
		over.configuration.ownerReceipt = bytes.Repeat([]byte{'s'}, serviceconfiguration.MaxCanonicalReceiptBytes+1)
		_, err := newProfiledPreparedEpoch(
			ConnectorDigestProfileV2ID,
			"epoch-profile-v2",
			over.view,
			over.configuration,
			over.bindingOwners,
			over.credentials,
		)
		if !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), "exceeds 4096 bytes") {
			t.Fatalf("over-limit ServiceConfiguration receipt error = %v", err)
		}
	})

	t.Run("binding definition envelope", func(t *testing.T) {
		view := profiledTestBindingView()
		atLimit := profiledBindingEnvelope(
			t,
			view.Bindings[0],
			bytes.Repeat([]byte{'d'}, connectorbindingownerv1.MaxCanonicalDefinitionBytes),
		)
		if _, err := NewBindingOwnerProof("pack-generation-limit", atLimit); err != nil {
			t.Fatalf("exact binding-definition limit rejected: %v", err)
		}
		_, err := connectorbindingownerv1.NewDefinitionEnvelope(connectorbindingownerv1.AuthoringInput{
			Identity:            connectorbindingownerv1.IdentityFromBinding(view.Bindings[0]),
			CanonicalDefinition: bytes.Repeat([]byte{'d'}, connectorbindingownerv1.MaxCanonicalDefinitionBytes+1),
		})
		if err == nil {
			t.Fatal("binding definition above the owner limit was accepted")
		}
	})

	t.Run("credential transport", func(t *testing.T) {
		view, configuration, credentials := profiledCredentialBudgetSet(1, 32, credentialgeneration.MaxTransportConfigurationBytes, 1)
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); err != nil {
			t.Fatalf("exact credential transport limit rejected: %v", err)
		}
		credentials[0].transportBytes++
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), "transport shape") {
			t.Fatalf("over-limit credential transport error = %v", err)
		}
	})

	t.Run("exact set entry count", func(t *testing.T) {
		view, configuration, credentials := profiledCredentialBudgetSet(
			credentialgeneration.MaxExactIdentitySetEntries,
			32,
			0,
			1,
		)
		proofs, err := buildProfiledCredentialProofs(view, configuration, credentials)
		if err != nil || len(proofs) != credentialgeneration.MaxExactIdentitySetEntries {
			t.Fatalf("exact credential entry limit = %d/%v", len(proofs), err)
		}
		view, configuration, credentials = profiledCredentialBudgetSet(
			credentialgeneration.MaxExactIdentitySetEntries+1,
			32,
			0,
			1,
		)
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), "exceeds 4096 entries") {
			t.Fatalf("over-limit credential entry count error = %v", err)
		}
	})

	t.Run("exact set ID bytes", func(t *testing.T) {
		view, configuration, credentials := profiledCredentialBudgetSet(4_096, 128, 0, 1)
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); err != nil {
			t.Fatalf("exact credential ID-byte limit rejected: %v", err)
		}
		view, configuration, credentials = profiledCredentialBudgetSet(4_096, 129, 0, 1)
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), "ID bytes exceed") {
			t.Fatalf("over-limit credential ID bytes error = %v", err)
		}
	})

	t.Run("exact set identity bytes", func(t *testing.T) {
		// The independent owner ceiling is intentionally non-binding under the
		// stricter profile-v2 entry, aggregate-ID, and per-identifier bounds.
		// Exercise the largest identity projection those other bounds permit so
		// any future widening must make an explicit budget decision.
		maximumProfileIdentityBytes := uint64(credentialgeneration.MaxExactIdentitySetIDBytes) +
			3*uint64(credentialgeneration.MaxExactIdentitySetEntries*connectorDigestProfileV2MaxIdentifierBytes)
		if maximumProfileIdentityBytes > credentialgeneration.MaxExactIdentitySetIdentityBytes {
			t.Fatalf(
				"profile-v2 maximum identity projection = %d, owner ceiling = %d",
				maximumProfileIdentityBytes,
				credentialgeneration.MaxExactIdentitySetIdentityBytes,
			)
		}
		view, configuration, credentials := profiledCredentialBudgetSet(4_096, 128, 0, 1)
		stableIdentity := strings.Repeat("c", connectorDigestProfileV2MaxIdentifierBytes)
		accountID := strings.Repeat("a", connectorDigestProfileV2MaxIdentifierBytes)
		providerScopeID := strings.Repeat("p", connectorDigestProfileV2MaxIdentifierBytes)
		view.Bindings[0].CredentialRefs = []string{stableIdentity}
		for index := range credentials {
			configuration.selections[index].credentialRef = stableIdentity
			configuration.selections[index].connectorAccountID = accountID
			configuration.selections[index].providerScopeID = providerScopeID
			credentials[index].stableCredentialIdentity = stableIdentity
			credentials[index].connectorAccountID = accountID
			credentials[index].providerScopeID = providerScopeID
		}
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); err != nil {
			t.Fatalf("maximum profile-v2 identity projection rejected: %v", err)
		}
	})

	t.Run("exact set payload", func(t *testing.T) {
		const count = 16
		view, configuration, credentials := profiledCredentialBudgetSet(
			count,
			32,
			credentialgeneration.MaxTransportConfigurationBytes-1,
			1,
		)
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); err != nil {
			t.Fatalf("exact credential payload limit rejected: %v", err)
		}
		credentials[0].ownerReceipt = []byte{'r', 'x'}
		if _, err := buildProfiledCredentialProofs(view, configuration, credentials); !errors.Is(err, ErrInvalidCandidate) || !strings.Contains(err.Error(), "payload set exceeds") {
			t.Fatalf("over-limit credential payload error = %v", err)
		}
	})
}

type profiledCanonicalFixture struct {
	view          catalogv1.IntegrationBindingView
	configuration profiledServiceConfigurationInput
	bindingOwners []BindingOwnerProof
	credentials   []profiledCredentialGenerationInput
}

func newProfiledCanonicalFixture(t *testing.T) profiledCanonicalFixture {
	t.Helper()
	view := profiledTestBindingView()
	configuration := profiledServiceConfigurationInput{
		storedServiceConfigurationProof: storedServiceConfigurationProof{
			generationProtocolVersion: serviceconfiguration.GenerationProtocolV1,
			receiptProtocolVersion:    serviceconfiguration.ReceiptProtocolV1,
			generationID:              "service-configuration-generation-1",
			generationDigest:          profiledTestDigest('c'),
			ownerReceipt:              []byte("service-configuration-owner\x00receipt"),
			ownerReceiptDigest:        profiledTestDigest('d'),
		},
		orgID:                    view.OrgID,
		appInstallationID:        view.AppInstallationID,
		authorizedWorkloadID:     view.AuthorizedWorkloadID,
		lineage:                  view.Lineage,
		receipts:                 view.Receipts,
		accountCount:             1,
		bindingCount:             2,
		credentialSelectionCount: 3,
		selections: []profiledCredentialSelection{
			{catalogBindingOrdinal: 0, credentialRef: "credential-a", generationID: "generation-z", connectorAccountID: "account-1", providerScopeID: "provider-1"},
			{catalogBindingOrdinal: 0, credentialRef: "credential-b", generationID: "Generation-A", connectorAccountID: "account-1", providerScopeID: "provider-1"},
			{catalogBindingOrdinal: 1, credentialRef: "credential-a", generationID: "generation-z", connectorAccountID: "account-1", providerScopeID: "provider-1"},
		},
	}
	bindingOwners := make([]BindingOwnerProof, len(view.Bindings))
	for index, binding := range view.Bindings {
		envelope := profiledBindingEnvelope(
			t,
			binding,
			[]byte(fmt.Sprintf(`{"binding":%d}`, index)),
		)
		proof, err := NewBindingOwnerProof(fmt.Sprintf("pack-generation-%d", index), envelope)
		if err != nil {
			t.Fatalf("NewBindingOwnerProof(%d): %v", index, err)
		}
		bindingOwners[index] = proof
	}
	return profiledCanonicalFixture{
		view:          view,
		configuration: configuration,
		bindingOwners: bindingOwners,
		credentials: []profiledCredentialGenerationInput{
			profiledCredentialInput("generation-z", "credential-a", "account-1", "provider-1"),
			profiledCredentialInput("Generation-A", "credential-b", "account-1", "provider-1"),
		},
	}
}

func mustNewProfiledPreparedEpoch(
	t *testing.T,
	fixture profiledCanonicalFixture,
) ProfiledPreparedEpoch {
	t.Helper()
	epoch, err := newProfiledPreparedEpoch(
		ConnectorDigestProfileV2ID,
		"epoch-profile-v2",
		fixture.view,
		fixture.configuration,
		fixture.bindingOwners,
		fixture.credentials,
	)
	if err != nil {
		t.Fatalf("newProfiledPreparedEpoch(): %v", err)
	}
	return epoch
}

func profiledTestBindingView() catalogv1.IntegrationBindingView {
	view := copyIntegrationBindingView(canonicalTestBindingView())
	// The definition-set owner contract fixes every non-route definition at v1.
	view.Bindings[0].DefinitionVersion = 1
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	return view
}

func profiledBindingEnvelope(
	t *testing.T,
	binding catalogv1.IntegrationBinding,
	definition []byte,
) connectorbindingownerv1.DefinitionEnvelope {
	t.Helper()
	envelope, err := connectorbindingownerv1.NewDefinitionEnvelope(
		connectorbindingownerv1.AuthoringInput{
			Identity:            connectorbindingownerv1.IdentityFromBinding(binding),
			CanonicalDefinition: definition,
		},
	)
	if err != nil {
		t.Fatalf("NewDefinitionEnvelope(%s/%s): %v", binding.Kind, binding.BusinessKey, err)
	}
	return envelope
}

func profiledCredentialInput(
	id string,
	stableIdentity string,
	accountID string,
	providerScopeID string,
) profiledCredentialGenerationInput {
	return profiledCredentialGenerationInput{
		storedCredentialGenerationProof: storedCredentialGenerationProof{
			generationID:                 id,
			receiptProtocolVersion:       credentialgeneration.ReceiptProtocolV1,
			transportConfigurationDigest: profiledTestDigest('a'),
			ownerReceipt:                 []byte("credential-owner\x00" + id),
			ownerReceiptDigest:           profiledTestDigest('b'),
		},
		stableCredentialIdentity: stableIdentity,
		connectorAccountID:       accountID,
		providerScopeID:          providerScopeID,
		transportBytes:           uint64(len("transport:" + id)),
	}
}

func profiledCredentialBudgetSet(
	count int,
	idBytes int,
	transportBytes uint64,
	receiptBytes int,
) (
	catalogv1.IntegrationBindingView,
	profiledServiceConfigurationInput,
	[]profiledCredentialGenerationInput,
) {
	view := profiledTestBindingView()
	configuration := profiledServiceConfigurationInput{
		selections:               make([]profiledCredentialSelection, count),
		credentialSelectionCount: uint64(count),
	}
	credentials := make([]profiledCredentialGenerationInput, count)
	for index := 0; index < count; index++ {
		id := profiledBudgetCredentialID(index, idBytes)
		accountID := fmt.Sprintf("account-%04d", index)
		configuration.selections[index] = profiledCredentialSelection{
			catalogBindingOrdinal: 0,
			credentialRef:         "credential-a",
			generationID:          id,
			connectorAccountID:    accountID,
			providerScopeID:       "provider-budget",
		}
		credentials[index] = profiledCredentialInput(
			id,
			"credential-a",
			accountID,
			"provider-budget",
		)
		credentials[index].transportBytes = transportBytes
		credentials[index].ownerReceipt = bytes.Repeat([]byte{'r'}, receiptBytes)
	}
	return view, configuration, credentials
}

func profiledBudgetCredentialID(index, length int) string {
	suffix := fmt.Sprintf("%08d", index)
	if length <= len(suffix) {
		return suffix[len(suffix)-length:]
	}
	return strings.Repeat("g", length-len(suffix)) + suffix
}

func profiledTestDigest(character byte) string {
	return "sha256:" + strings.Repeat(string(character), 64)
}