package serving

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

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

func TestConnectorDigestProfileResolutionIsExactAndClosed(t *testing.T) {
	authoring, err := ResolveConnectorDigestProfileForAuthoring(ConnectorDigestProfileV2ID)
	if err != nil {
		t.Fatal(err)
	}
	retainedV1, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV1ID)
	if err != nil {
		t.Fatal(err)
	}
	retainedV2, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV2ID)
	if err != nil {
		t.Fatal(err)
	}
	if authoring.ID() != ConnectorDigestProfileV2ID ||
		retainedV1.ID() != ConnectorDigestProfileV1ID ||
		retainedV2.ID() != ConnectorDigestProfileV2ID ||
		!authoring.CompatibleWith(retainedV2) ||
		authoring.CompatibleWith(retainedV1) {
		t.Fatalf(
			"resolved profiles = authoring:%q read-v1:%q read-v2:%q",
			authoring.ID(),
			retainedV1.ID(),
			retainedV2.ID(),
		)
	}
	if profile, err := ResolveConnectorDigestProfileForAuthoring(ConnectorDigestProfileV1ID); !errors.Is(err, ErrUnsupportedConnectorDigestProfile) || profile.ID() != "" {
		t.Fatalf("V1 authoring resolver = %q/%v, want empty ErrUnsupportedConnectorDigestProfile", profile.ID(), err)
	}

	unsupported := []ConnectorDigestProfileID{
		"",
		"v1",
		"nexuscrm.connector-digest-profile",
		"NEXUSCRM.CONNECTOR-DIGEST-PROFILE.V1",
		"nexuscrm.connector-digest-profile.v1 ",
		" nexuscrm.connector-digest-profile.v1",
		"nexuscrm.connector-digest-profile.v3",
	}
	for _, id := range unsupported {
		t.Run(string(id), func(t *testing.T) {
			for name, resolve := range map[string]func(ConnectorDigestProfileID) (ConnectorDigestProfile, error){
				"authoring": ResolveConnectorDigestProfileForAuthoring,
				"read":      ResolveConnectorDigestProfileForRead,
			} {
				profile, err := resolve(id)
				if !errors.Is(err, ErrUnsupportedConnectorDigestProfile) {
					t.Fatalf("%s resolver error = %v, want ErrUnsupportedConnectorDigestProfile", name, err)
				}
				if profile.ID() != "" {
					t.Fatalf("%s resolver returned profile %q on failure", name, profile.ID())
				}
			}
		})
	}

	var zero ConnectorDigestProfile
	if zero.CompatibleWith(zero) || zero.CompatibleWith(authoring) {
		t.Fatal("zero-value profile was treated as compatible")
	}
	forged := ConnectorDigestProfile{id: "nexuscrm.connector-digest-profile.v3"}
	if authoring.CompatibleWith(forged) || forged.CompatibleWith(forged) {
		t.Fatal("unknown or cross-profile value was treated as compatible")
	}
	if _, err := zero.computeServingScopeDigest(catalogReference{}); !errors.Is(err, ErrUnsupportedConnectorDigestProfile) {
		t.Fatalf("zero-value digest error = %v, want ErrUnsupportedConnectorDigestProfile", err)
	}
}

func TestConnectorDigestProfileV2CannotReuseV1FieldRecipes(t *testing.T) {
	v1, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV1ID)
	if err != nil {
		t.Fatal(err)
	}
	v2, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV2ID)
	if err != nil {
		t.Fatal(err)
	}
	legacy := servingEpochTestCandidate(t, "epoch-v1-recipe-guard")
	if digest, err := v2.computeL2VerificationReceiptDigest(legacy.l2.verificationReceipt); !errors.Is(err, ErrUnsupportedConnectorDigestProfile) || digest != "" {
		t.Fatalf("V2 L2 receipt recipe = %q/%v, want empty ErrUnsupportedConnectorDigestProfile", digest, err)
	}
	if digest, err := v2.computeEpochDigest(legacy); !errors.Is(err, ErrUnsupportedConnectorDigestProfile) || digest != "" {
		t.Fatalf("V2 legacy epoch recipe = %q/%v, want empty ErrUnsupportedConnectorDigestProfile", digest, err)
	}

	credentialRefs := []string{"credential-a", "credential-b"}
	v1Blob, err := v1.encodeCredentialRefs(credentialRefs)
	if err != nil {
		t.Fatal(err)
	}
	v2Blob, err := v2.encodeCredentialRefs(credentialRefs)
	if err != nil {
		t.Fatal(err)
	}
	if bytes.Equal(v1Blob, v2Blob) {
		t.Fatal("V1 and V2 credential-reference bytes unexpectedly converged")
	}
	if values, err := v1.decodeCredentialRefs(v2Blob); err == nil || values != nil {
		t.Fatalf("V1 decoded V2 credential-reference bytes as %v/%v", values, err)
	}
	if values, err := v2.decodeCredentialRefs(v1Blob); err == nil || values != nil {
		t.Fatalf("V2 decoded V1 credential-reference bytes as %v/%v", values, err)
	}
}

func TestConnectorDigestProfileV1ConformanceGolden(t *testing.T) {
	profile, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV1ID)
	if err != nil {
		t.Fatal(err)
	}
	legacy, err := NewPreparedEpoch(
		"epoch-canonical-1",
		canonicalTestBindingView(),
		canonicalTestL2Snapshot(t),
	)
	if err != nil {
		t.Fatal(err)
	}
	profiled := profiledConformanceEpoch(t, profile, legacy)

	encoding, err := profile.digestEncoding()
	if err != nil {
		t.Fatal(err)
	}
	profiledCredentialRefs := profiled.bindings[0].credentialRefs
	profiledEpochDigest, err := profile.computeEpochDigest(profiled)
	if err != nil {
		t.Fatal(err)
	}
	goldens := map[string][2]string{
		"profile prefix": {
			hex.EncodeToString(encoding.canonicalPrefix(servingEpochDomain)),
			"6e6578757363726d2e636f6e6e6563746f722d73657276696e672d65706f63682e76310000000000000000246e6578757363726d2e636f6e6e6563746f722d6469676573742d70726f66696c652e7631",
		},
		"serving scope": {
			profiled.servingScopeDigest,
			"2f172cea873024a13a0002701278621e0e4cb432e24817b1e8ac5a0fdfb04787",
		},
		"binding identity": {
			profiled.bindings[0].identityDigest,
			"20f952197edc67394d88848751505ff603d055f810de7ab9f0500bf0a39d5717",
		},
		"binding row ID": {
			profiled.bindings[0].id,
			"fdf1e6972a5514b9f67adfd12afc3c69b092ec6860a0784000eb34360bcd9d46",
		},
		"credential refs": {
			hex.EncodeToString(profiledCredentialRefs),
			"6e6578757363726d2e636f6e6e6563746f722d73657276696e672d63726564656e7469616c2d726566732e76310000000000000000246e6578757363726d2e636f6e6e6563746f722d6469676573742d70726f66696c652e76310000000000000002000000000000000c63726564656e7469616c2d61000000000000000c63726564656e7469616c2d62",
		},
		"L2 receipt": {
			profiled.l2.verificationReceiptDigest,
			"sha256:63da14e8697863d207b0a7c11600c586af8ed36d644717e5034dce3e81c522dc",
		},
		"epoch": {
			profiledEpochDigest,
			"sha256:37e811d186653fe4ee40e301eee913e3b38fbe1465751f1625e48480f10b8ddf",
		},
	}
	for name, values := range goldens {
		if values[0] != values[1] {
			t.Errorf("%s golden = %q, want %q", name, values[0], values[1])
		}
	}

	legacyGoldens := []struct {
		name       string
		profiled   string
		unprofiled string
	}{
		{name: "serving scope", profiled: profiled.servingScopeDigest, unprofiled: legacy.servingScopeDigest},
		{name: "binding identity", profiled: profiled.bindings[0].identityDigest, unprofiled: legacy.bindings[0].identityDigest},
		{name: "binding row ID", profiled: profiled.bindings[0].id, unprofiled: legacy.bindings[0].id},
		{name: "credential refs", profiled: string(profiled.bindings[0].credentialRefs), unprofiled: string(legacy.bindings[0].credentialRefs)},
		{name: "L2 receipt", profiled: profiled.l2.verificationReceiptDigest, unprofiled: legacy.l2.verificationReceiptDigest},
		{name: "epoch", profiled: profiledEpochDigest, unprofiled: legacy.epochDigest},
	}
	for _, value := range legacyGoldens {
		if value.profiled == value.unprofiled {
			t.Errorf("%s did not include the explicit profile ID", value.name)
		}
	}

	if _, err := decodeCredentialRefs(profiledCredentialRefs); err == nil {
		t.Fatal("legacy decoder accepted explicitly profiled credential bytes")
	}
	if _, err := profile.decodeCredentialRefs(legacy.bindings[0].credentialRefs); err == nil {
		t.Fatal("profiled decoder inferred v1 for legacy unprofiled credential bytes")
	}
	decoded, err := profile.decodeCredentialRefs(profiledCredentialRefs)
	if err != nil || len(decoded) != 2 || decoded[0] != "credential-a" || decoded[1] != "credential-b" {
		t.Fatalf("profiled credential round trip = %v, err %v", decoded, err)
	}
}

func TestConnectorDigestProfileV1FramingAndOwnerOpacity(t *testing.T) {
	profile, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV1ID)
	if err != nil {
		t.Fatal(err)
	}
	encoding, err := profile.digestEncoding()
	if err != nil {
		t.Fatal(err)
	}

	left := encoding.canonicalPrefix("nexuscrm.connector-conformance.v1")
	left = appendCanonicalString(left, "ab")
	left = appendCanonicalString(left, "c")
	right := encoding.canonicalPrefix("nexuscrm.connector-conformance.v1")
	right = appendCanonicalString(right, "a")
	right = appendCanonicalString(right, "bc")
	if bytes.Equal(left, right) || rawDigest(left) == rawDigest(right) {
		t.Fatal("uint64 byte-length framing allowed a structural collision")
	}

	composed := catalogv1.IntegrationBinding{
		Kind:               catalogv1.BindingConnectorCalloutProfile,
		OwnerPack:          "pack-\u00e9",
		PackInstallationID: "install-Exact",
		BusinessKey:        "Profile-Key",
	}
	decomposed := composed
	decomposed.OwnerPack = "pack-e\u0301"
	composedDigest, err := profile.computeBindingIdentityDigest(composed)
	if err != nil {
		t.Fatal(err)
	}
	decomposedDigest, err := profile.computeBindingIdentityDigest(decomposed)
	if err != nil {
		t.Fatal(err)
	}
	caseChanged := composed
	caseChanged.PackInstallationID = "install-exact"
	caseDigest, err := profile.computeBindingIdentityDigest(caseChanged)
	if err != nil {
		t.Fatal(err)
	}
	spaceChanged := composed
	spaceChanged.BusinessKey += " "
	spaceDigest, err := profile.computeBindingIdentityDigest(spaceChanged)
	if err != nil {
		t.Fatal(err)
	}
	for name, digest := range map[string]string{
		"Unicode normalization": decomposedDigest,
		"case folding":          caseDigest,
		"trimming":              spaceDigest,
	} {
		if digest == composedDigest {
			t.Errorf("%s silently changed an external-owner fact", name)
		}
	}

	for _, receipts := range [][2][]byte{
		{{0, 1, 2}, {0, 1, 2, 0}},
		{[]byte("receipt"), []byte("receipt ")},
		{[]byte("\u00e9"), []byte("e\u0301")},
	} {
		leftDigest, err := profile.computeL2VerificationReceiptDigest(receipts[0])
		if err != nil {
			t.Fatal(err)
		}
		rightDigest, err := profile.computeL2VerificationReceiptDigest(receipts[1])
		if err != nil {
			t.Fatal(err)
		}
		if leftDigest == rightDigest {
			t.Fatalf("opaque binary receipts %x and %x converged", receipts[0], receipts[1])
		}
	}

	legacy := servingEpochTestCandidate(t, "epoch-runtime-marker")
	profiled := profiledConformanceEpoch(t, profile, legacy)
	withoutRuntime := profiled.copy()
	withoutRuntime.bindings[0].runtimeCommandKind = nil
	withoutRuntime.bindings[0].runtimePayloadFormatVersion = nil
	withoutRuntime.bindings[0].runtimeDefinitionRef = nil
	withoutRuntime.bindings[0].runtimeDefinitionVersion = nil
	withRuntime := withoutRuntime.copy()
	withRuntime.bindings[0].runtimeCommandKind = stringPointer("")
	withRuntime.bindings[0].runtimePayloadFormatVersion = uint32Pointer(0)
	withRuntime.bindings[0].runtimeDefinitionRef = stringPointer("")
	withRuntime.bindings[0].runtimeDefinitionVersion = stringPointer("")
	withoutDigest, err := profile.computeEpochDigest(withoutRuntime)
	if err != nil {
		t.Fatal(err)
	}
	withDigest, err := profile.computeEpochDigest(withRuntime)
	if err != nil {
		t.Fatal(err)
	}
	if withoutDigest == withDigest {
		t.Fatal("optional Runtime 0/1 marker collapsed absent and present-empty tuples")
	}
}

func TestConnectorDigestProfileV1DependencyBoundsAreFrozen(t *testing.T) {
	if connectorDigestProfileV1MaxIdentifierBytes != 255 ||
		connectorDigestProfileV1MaxCredentialRefCount != 256 ||
		connectorDigestProfileV2MaxIdentifierBytes != 255 ||
		connectorDigestProfileV2MaxCredentialRefCount != 256 ||
		maxPersistedL2VerificationReceiptBytes != 1<<20 {
		t.Fatal("profile-v1 or profile-v2 limits changed without an explicit profile decision")
	}
	if catalogv1.MaxIdentifierBytes != connectorDigestProfileV1MaxIdentifierBytes {
		t.Fatalf(
			"Catalog MaxIdentifierBytes = %d, profile v1 = %d; choose an explicit compatibility policy",
			catalogv1.MaxIdentifierBytes,
			connectorDigestProfileV1MaxIdentifierBytes,
		)
	}
	if catalogv1.MaxCredentialRefCount != connectorDigestProfileV1MaxCredentialRefCount {
		t.Fatalf(
			"Catalog MaxCredentialRefCount = %d, profile v1 = %d; choose an explicit compatibility policy",
			catalogv1.MaxCredentialRefCount,
			connectorDigestProfileV1MaxCredentialRefCount,
		)
	}
	if catalogv1.MaxIdentifierBytes != connectorDigestProfileV2MaxIdentifierBytes ||
		catalogv1.MaxCredentialRefCount != connectorDigestProfileV2MaxCredentialRefCount {
		t.Fatal("Catalog bounds drifted from the frozen profile-v2 admission grammar")
	}

	profile, err := ResolveConnectorDigestProfileForRead(ConnectorDigestProfileV1ID)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := profile.encodeCredentialRefs([]string{strings.Repeat("x", connectorDigestProfileV1MaxIdentifierBytes)}); err != nil {
		t.Fatalf("exact profile-v1 identifier bound rejected: %v", err)
	}
	if _, err := profile.encodeCredentialRefs([]string{strings.Repeat("x", connectorDigestProfileV1MaxIdentifierBytes+1)}); err == nil {
		t.Fatal("identifier above profile-v1 bound accepted")
	}
}

func profiledConformanceEpoch(
	t *testing.T,
	profile ConnectorDigestProfile,
	legacy PreparedEpoch,
) PreparedEpoch {
	t.Helper()
	profiled := legacy.copy()
	var err error
	profiled.servingScopeDigest, err = profile.computeServingScopeDigest(profiled.catalog)
	if err != nil {
		t.Fatal(err)
	}
	for i := range profiled.bindings {
		bindingForProfile, err := catalogBindingForValidation(profiled.bindings[i])
		if err != nil {
			t.Fatal(err)
		}
		credentialRefs := append([]string(nil), bindingForProfile.CredentialRefs...)
		profiled.bindings[i].identityDigest, err = profile.computeBindingIdentityDigest(bindingForProfile)
		if err != nil {
			t.Fatal(err)
		}
		profiled.bindings[i].id, err = profile.computeBindingRowID(
			profiled.id,
			profiled.bindings[i].ordinal,
			profiled.bindings[i].identityDigest,
		)
		if err != nil {
			t.Fatal(err)
		}
		profiled.bindings[i].credentialRefs, err = profile.encodeCredentialRefs(credentialRefs)
		if err != nil {
			t.Fatal(err)
		}
	}
	profiled.l2.verificationReceiptDigest, err = profile.computeL2VerificationReceiptDigest(
		profiled.l2.verificationReceipt,
	)
	if err != nil {
		t.Fatal(err)
	}
	profiled.epochDigest, err = profile.computeEpochDigest(profiled)
	if err != nil {
		t.Fatal(err)
	}
	return profiled
}