package catalogassignmentidentity

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"reflect"
	"strconv"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	catalogassignmentidentityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogassignmentidentityv1"
	catalogpublicationv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
)

const (
	envelopeMetadataQueryPattern = `(?s)^SELECT OrgId, ViewDigest, CatalogRevision, CatalogSnapshotDigest,.*CanonicalByteLength, OCTET_LENGTH\(CanonicalBytes\),.*FROM _System_CatalogAssignmentIdentityView.*WHERE OrgId = \? AND ViewDigest = \?`
	envelopeBlobQueryPattern     = `(?s)^SELECT CanonicalBytes.*FROM _System_CatalogAssignmentIdentityView.*WHERE OrgId = \? AND ViewDigest = \?`
	objectQueryPattern           = `(?s)^SELECT EntryOrdinal, NormalizedAPIName, ResolvedObjectId, State.*FROM _System_CatalogAssignmentIdentityViewObject.*WHERE OrgId = \? AND ViewDigest = \?.*ORDER BY EntryOrdinal ASC`
	recordTypeQueryPattern       = `(?s)^SELECT EntryOrdinal, ObjectNormalizedAPIName, NamespaceId, LocalKey, ResolvedRecordTypeId, State.*FROM _System_CatalogAssignmentIdentityViewRecordType.*WHERE OrgId = \? AND ViewDigest = \?.*ORDER BY EntryOrdinal ASC`
	productAppQueryPattern       = `(?s)^SELECT EntryOrdinal, AppId, State,.*ResolvedTargetKind, ResolvedAppInstallationId.*FROM _System_CatalogAssignmentIdentityViewProductApp.*WHERE OrgId = \? AND ViewDigest = \?.*ORDER BY EntryOrdinal ASC`
	navigationAppQueryPattern    = `(?s)^SELECT EntryOrdinal, NamespaceId,.*LocalKey, ResolvedAppConfigId, State.*FROM _System_CatalogAssignmentIdentityViewNavigationApp.*WHERE OrgId = \? AND ViewDigest = \?.*ORDER BY EntryOrdinal ASC`
)

func TestPutAndReadExactRoundTripMaterializedContractVectors(t *testing.T) {
	document := decodeAssignmentIdentityVectors(t)
	if len(document.ViewVectors) != 3 {
		t.Fatalf("materialized view vector count = %d, want 3", len(document.ViewVectors))
	}

	for _, vector := range document.ViewVectors {
		vector := vector
		t.Run(vector.Name, func(t *testing.T) {
			input := vector.View.value(t)
			canonical, canonicalBytes, digest := assertVectorWitness(t, vector, input)
			store, mock, closeDB := newMockStore(t)
			defer closeDB()

			mock.ExpectBegin()
			expectEnvelopeMissing(mock, canonical, digest, true)
			expectNoOrphanRows(mock, canonical.SourceCatalogSnapshot.OrgID, digest)
			expectArtifactInsert(mock, canonical, canonicalBytes, digest)
			expectArtifactRead(mock, storedReadFixture{
				view:           canonical,
				canonicalBytes: canonicalBytes,
			}, digest, true)
			mock.ExpectCommit()

			gotDigest, err := store.PutExact(context.Background(), input)
			if err != nil {
				t.Fatalf("PutExact: %v", err)
			}
			if gotDigest != digest {
				t.Fatalf("PutExact digest = %q, want %q", gotDigest, digest)
			}

			mock.ExpectBegin()
			expectArtifactRead(mock, storedReadFixture{
				view:           canonical,
				canonicalBytes: canonicalBytes,
			}, digest, false)
			mock.ExpectCommit()

			got, err := store.ReadExact(
				context.Background(),
				canonical.SourceCatalogSnapshot.OrgID,
				digest,
			)
			if err != nil {
				t.Fatalf("ReadExact: %v", err)
			}
			if !reflect.DeepEqual(got, canonical) {
				t.Fatalf("ReadExact view = %#v, want %#v", got, canonical)
			}
			if got.Objects == nil || got.RecordTypes == nil ||
				got.ProductApps == nil || got.NavigationApps == nil {
				t.Fatal("ReadExact collapsed a present empty dimension to nil")
			}
			gotBytes, err := catalogassignmentidentityv1.CanonicalViewBytes(got)
			if err != nil {
				t.Fatal(err)
			}
			if !bytes.Equal(gotBytes, canonicalBytes) {
				t.Fatalf("ReadExact canonical bytes differ for vector %q", vector.Name)
			}
			assertMockExpectations(t, mock)
		})
	}
}

func TestPutExactReplayReadsAndReturnsTheExactImmutableArtifact(t *testing.T) {
	vector := materializedVectorByName(t, "all-dimensions-permuted")
	input := vector.View.value(t)
	canonical, canonicalBytes, digest := assertVectorWitness(t, vector, input)
	store, mock, closeDB := newMockStore(t)
	defer closeDB()

	mock.ExpectBegin()
	expectArtifactRead(mock, storedReadFixture{
		view:           canonical,
		canonicalBytes: canonicalBytes,
	}, digest, true)
	mock.ExpectCommit()

	gotDigest, err := store.PutExact(context.Background(), input)
	if err != nil {
		t.Fatalf("PutExact replay: %v", err)
	}
	if gotDigest != digest {
		t.Fatalf("PutExact replay digest = %q, want %q", gotDigest, digest)
	}
	assertMockExpectations(t, mock)
}

func TestPutExactRecoversAnExactConcurrentInsert(t *testing.T) {
	vector := materializedVectorByName(t, "all-dimensions-permuted")
	input := vector.View.value(t)
	canonical, canonicalBytes, digest := assertVectorWitness(t, vector, input)
	store, mock, closeDB := newMockStore(t)
	defer closeDB()

	mock.ExpectBegin()
	expectEnvelopeMissing(mock, canonical, digest, true)
	expectNoOrphanRows(mock, canonical.SourceCatalogSnapshot.OrgID, digest)
	mock.ExpectExec(`^INSERT INTO _System_CatalogAssignmentIdentityView`).
		WillReturnError(errors.New("concurrent duplicate"))
	mock.ExpectRollback()
	mock.ExpectBegin()
	expectArtifactRead(mock, storedReadFixture{
		view:           canonical,
		canonicalBytes: canonicalBytes,
	}, digest, false)
	mock.ExpectCommit()

	gotDigest, err := store.PutExact(context.Background(), input)
	if err != nil {
		t.Fatalf("PutExact concurrent recovery: %v", err)
	}
	if gotDigest != digest {
		t.Fatalf("recovered digest = %q, want %q", gotDigest, digest)
	}
	assertMockExpectations(t, mock)
}

func TestPutExactRejectsChildRowsWithoutAnEnvelope(t *testing.T) {
	vector := materializedVectorByName(t, "all-dimensions-permuted")
	input := vector.View.value(t)
	canonical, _, digest := assertVectorWitness(t, vector, input)
	store, mock, closeDB := newMockStore(t)
	defer closeDB()

	mock.ExpectBegin()
	expectEnvelopeMissing(mock, canonical, digest, true)
	mock.ExpectQuery(
		`(?s)^SELECT 1 FROM _System_CatalogAssignmentIdentityViewObject`+
			`.*WHERE OrgId = \? AND ViewDigest = \?.*LIMIT 1 FOR UPDATE$`,
	).
		WithArgs(canonical.SourceCatalogSnapshot.OrgID, digest).
		WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
	mock.ExpectRollback()

	_, err := store.PutExact(context.Background(), input)
	if !errors.Is(err, ErrCorrupt) {
		t.Fatalf("PutExact orphan-row error = %v, want ErrCorrupt", err)
	}
	assertMockExpectations(t, mock)
}

func TestPutExactRecoversTheExactArtifactAfterAmbiguousCommit(t *testing.T) {
	vector := materializedVectorByName(t, "all-dimensions-permuted")
	input := vector.View.value(t)
	canonical, canonicalBytes, digest := assertVectorWitness(t, vector, input)
	store, mock, closeDB := newMockStore(t)
	defer closeDB()

	mock.ExpectBegin()
	expectEnvelopeMissing(mock, canonical, digest, true)
	expectNoOrphanRows(mock, canonical.SourceCatalogSnapshot.OrgID, digest)
	expectArtifactInsert(mock, canonical, canonicalBytes, digest)
	expectArtifactRead(mock, storedReadFixture{
		view:           canonical,
		canonicalBytes: canonicalBytes,
	}, digest, true)
	mock.ExpectCommit().WillReturnError(errors.New("ambiguous commit result"))

	mock.ExpectBegin()
	expectArtifactRead(mock, storedReadFixture{
		view:           canonical,
		canonicalBytes: canonicalBytes,
	}, digest, false)
	mock.ExpectCommit()

	gotDigest, err := store.PutExact(context.Background(), input)
	if err != nil {
		t.Fatalf("PutExact ambiguous-commit recovery: %v", err)
	}
	if gotDigest != digest {
		t.Fatalf("recovered digest = %q, want %q", gotDigest, digest)
	}
	assertMockExpectations(t, mock)
}

func TestReadExactRejectsReorderedOrdinalsAndCanonicalBlobMismatch(t *testing.T) {
	vector := materializedVectorByName(t, "all-dimensions-permuted")
	canonical, canonicalBytes, digest := assertVectorWitness(
		t,
		vector,
		vector.View.value(t),
	)

	tests := []struct {
		name    string
		fixture storedReadFixture
	}{
		{
			name: "missing object row",
			fixture: storedReadFixture{
				view:              canonical,
				canonicalBytes:    canonicalBytes,
				objectRows:        canonical.Objects[:len(canonical.Objects)-1],
				hasObjectOverride: true,
			},
		},
		{
			name: "extra object row",
			fixture: func() storedReadFixture {
				extra := append(
					[]catalogassignmentidentityv1.ObjectIdentity(nil),
					canonical.Objects...,
				)
				extra = append(extra, catalogassignmentidentityv1.ObjectIdentity{
					Key: catalogassignmentidentityv1.ObjectKey{
						NormalizedAPIName: "extra__c",
					},
					ResolvedObjectID: "object-extra",
					State:            catalogassignmentidentityv1.IdentityStateActive,
				})
				return storedReadFixture{
					view:              canonical,
					canonicalBytes:    canonicalBytes,
					objectRows:        extra,
					hasObjectOverride: true,
				}
			}(),
		},
		{
			name: "duplicate record-type row",
			fixture: func() storedReadFixture {
				duplicate := append(
					[]catalogassignmentidentityv1.RecordTypeIdentity(nil),
					canonical.RecordTypes...,
				)
				duplicate[1] = duplicate[0]
				return storedReadFixture{
					view:                  canonical,
					canonicalBytes:        canonicalBytes,
					recordTypeRows:        duplicate,
					hasRecordTypeOverride: true,
				}
			}(),
		},
		{
			name: "reordered navigation ordinals",
			fixture: func() storedReadFixture {
				reordered := append(
					[]catalogassignmentidentityv1.NavigationAppIdentity(nil),
					canonical.NavigationApps...,
				)
				reordered[0], reordered[1] = reordered[1], reordered[0]
				return storedReadFixture{
					view:                     canonical,
					canonicalBytes:           canonicalBytes,
					navigationAppRows:        reordered,
					hasNavigationAppOverride: true,
				}
			}(),
		},
		{
			name: "mutated Product-App row",
			fixture: func() storedReadFixture {
				mutated := append(
					[]catalogassignmentidentityv1.ProductAppIdentity(nil),
					canonical.ProductApps...,
				)
				mutated[0].State = catalogassignmentidentityv1.IdentityState("MUTATED")
				return storedReadFixture{
					view:                  canonical,
					canonicalBytes:        canonicalBytes,
					productAppRows:        mutated,
					hasProductAppOverride: true,
				}
			}(),
		},
		{
			name: "canonical blob mismatch",
			fixture: func() storedReadFixture {
				mutated := bytes.Clone(canonicalBytes)
				mutated[len(mutated)-1] ^= 0x01
				return storedReadFixture{
					view:           canonical,
					canonicalBytes: mutated,
				}
			}(),
		},
	}

	for _, test := range tests {
		test := test
		t.Run(test.name, func(t *testing.T) {
			store, mock, closeDB := newMockStore(t)
			defer closeDB()
			mock.ExpectBegin()
			expectArtifactRead(mock, test.fixture, digest, false)
			mock.ExpectRollback()

			_, err := store.ReadExact(
				context.Background(),
				canonical.SourceCatalogSnapshot.OrgID,
				digest,
			)
			if !errors.Is(err, ErrCorrupt) {
				t.Fatalf("ReadExact error = %v, want ErrCorrupt", err)
			}
			assertMockExpectations(t, mock)
		})
	}
}

func TestReadExactRejectsEveryChildCorruptionClassInEveryDimension(t *testing.T) {
	vector := materializedVectorByName(t, "all-dimensions-permuted")
	canonical, canonicalBytes, digest := assertVectorWitness(
		t,
		vector,
		vector.View.value(t),
	)
	for _, dimension := range []string{
		"object",
		"record-type",
		"Product-App",
		"navigation-App",
	} {
		for _, corruption := range []string{
			"missing",
			"extra",
			"reordered",
			"duplicate",
			"mutated",
		} {
			dimension := dimension
			corruption := corruption
			t.Run(dimension+"/"+corruption, func(t *testing.T) {
				fixture := corruptStoredRows(
					t,
					canonical,
					canonicalBytes,
					dimension,
					corruption,
				)
				store, mock, closeDB := newMockStore(t)
				defer closeDB()
				mock.ExpectBegin()
				expectArtifactRead(mock, fixture, digest, false)
				mock.ExpectRollback()

				_, err := store.ReadExact(
					context.Background(),
					canonical.SourceCatalogSnapshot.OrgID,
					digest,
				)
				if !errors.Is(err, ErrCorrupt) {
					t.Fatalf("ReadExact error = %v, want ErrCorrupt", err)
				}
				assertMockExpectations(t, mock)
			})
		}
	}
}

type storedReadFixture struct {
	view                     catalogassignmentidentityv1.View
	canonicalBytes           []byte
	objectRows               []catalogassignmentidentityv1.ObjectIdentity
	hasObjectOverride        bool
	recordTypeRows           []catalogassignmentidentityv1.RecordTypeIdentity
	hasRecordTypeOverride    bool
	productAppRows           []catalogassignmentidentityv1.ProductAppIdentity
	hasProductAppOverride    bool
	navigationAppRows        []catalogassignmentidentityv1.NavigationAppIdentity
	hasNavigationAppOverride bool
}

func corruptStoredRows(
	t *testing.T,
	view catalogassignmentidentityv1.View,
	canonicalBytes []byte,
	dimension string,
	corruption string,
) storedReadFixture {
	t.Helper()
	fixture := storedReadFixture{
		view:           view,
		canonicalBytes: canonicalBytes,
	}
	switch dimension {
	case "object":
		rows := append([]catalogassignmentidentityv1.ObjectIdentity(nil), view.Objects...)
		switch corruption {
		case "missing":
			rows = rows[:len(rows)-1]
		case "extra":
			rows = append(rows, catalogassignmentidentityv1.ObjectIdentity{
				Key: catalogassignmentidentityv1.ObjectKey{
					NormalizedAPIName: "zz_extra__c",
				},
				ResolvedObjectID: "object-extra",
				State:            catalogassignmentidentityv1.IdentityStateActive,
			})
		case "reordered":
			rows[0], rows[1] = rows[1], rows[0]
		case "duplicate":
			rows[len(rows)-1] = rows[0]
		case "mutated":
			rows[0].ResolvedObjectID += "-mutated"
		default:
			t.Fatalf("unknown corruption %q", corruption)
		}
		fixture.objectRows = rows
		fixture.hasObjectOverride = true
	case "record-type":
		rows := append(
			[]catalogassignmentidentityv1.RecordTypeIdentity(nil),
			view.RecordTypes...,
		)
		switch corruption {
		case "missing":
			rows = rows[:len(rows)-1]
		case "extra":
			rows = append(rows, rows[0])
		case "reordered":
			rows[0], rows[1] = rows[1], rows[0]
		case "duplicate":
			rows[len(rows)-1] = rows[0]
		case "mutated":
			rows[0].ResolvedRecordTypeID += "-mutated"
		default:
			t.Fatalf("unknown corruption %q", corruption)
		}
		fixture.recordTypeRows = rows
		fixture.hasRecordTypeOverride = true
	case "Product-App":
		rows := append(
			[]catalogassignmentidentityv1.ProductAppIdentity(nil),
			view.ProductApps...,
		)
		switch corruption {
		case "missing":
			rows = rows[:len(rows)-1]
		case "extra":
			rows = append(rows, rows[0])
		case "reordered":
			rows[0], rows[1] = rows[1], rows[0]
		case "duplicate":
			rows[len(rows)-1] = rows[0]
		case "mutated":
			mutated := false
			for index := range rows {
				if rows[index].ResolvedTarget.Kind !=
					catalogassignmentidentityv1.ProductAppResolvedTargetKindAppInstallationID {
					continue
				}
				rows[index].ResolvedTarget.ResolvedAppInstallationID =
					mutateUUIDTail(rows[index].ResolvedTarget.ResolvedAppInstallationID)
				mutated = true
				break
			}
			if !mutated {
				t.Fatal("Product-App vector has no installation target to mutate")
			}
		default:
			t.Fatalf("unknown corruption %q", corruption)
		}
		fixture.productAppRows = rows
		fixture.hasProductAppOverride = true
	case "navigation-App":
		rows := append(
			[]catalogassignmentidentityv1.NavigationAppIdentity(nil),
			view.NavigationApps...,
		)
		switch corruption {
		case "missing":
			rows = rows[:len(rows)-1]
		case "extra":
			rows = append(rows, rows[0])
		case "reordered":
			rows[0], rows[1] = rows[1], rows[0]
		case "duplicate":
			rows[len(rows)-1] = rows[0]
		case "mutated":
			rows[0].ResolvedAppConfigID += "-mutated"
		default:
			t.Fatalf("unknown corruption %q", corruption)
		}
		fixture.navigationAppRows = rows
		fixture.hasNavigationAppOverride = true
	default:
		t.Fatalf("unknown dimension %q", dimension)
	}
	return fixture
}

func mutateUUIDTail(value string) string {
	if len(value) == 0 {
		return value
	}
	tail := byte('0')
	if value[len(value)-1] == '0' {
		tail = '1'
	}
	return value[:len(value)-1] + string(tail)
}

func newMockStore(t *testing.T) (*Store, sqlmock.Sqlmock, func()) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	store, err := New(db)
	if err != nil {
		_ = db.Close()
		t.Fatal(err)
	}
	return store, mock, func() { _ = db.Close() }
}

func expectEnvelopeMissing(
	mock sqlmock.Sqlmock,
	view catalogassignmentidentityv1.View,
	digest string,
	forUpdate bool,
) {
	mock.ExpectQuery(lockQuery(envelopeMetadataQueryPattern, forUpdate)).
		WithArgs(view.SourceCatalogSnapshot.OrgID, digest).
		WillReturnError(sql.ErrNoRows)
}

func expectNoOrphanRows(mock sqlmock.Sqlmock, orgID, digest string) {
	for _, table := range []string{
		"_System_CatalogAssignmentIdentityViewObject",
		"_System_CatalogAssignmentIdentityViewRecordType",
		"_System_CatalogAssignmentIdentityViewProductApp",
		"_System_CatalogAssignmentIdentityViewNavigationApp",
	} {
		mock.ExpectQuery(
			`(?s)^SELECT 1 FROM `+table+
				`.*WHERE OrgId = \? AND ViewDigest = \?.*LIMIT 1 FOR UPDATE$`,
		).
			WithArgs(orgID, digest).
			WillReturnRows(sqlmock.NewRows([]string{"1"}))
	}
}

func expectArtifactInsert(
	mock sqlmock.Sqlmock,
	view catalogassignmentidentityv1.View,
	canonicalBytes []byte,
	digest string,
) {
	orgID := view.SourceCatalogSnapshot.OrgID
	mock.ExpectExec(`^INSERT INTO _System_CatalogAssignmentIdentityView`).
		WithArgs(
			orgID,
			digest,
			decimalText(view.SourceCatalogSnapshot.CatalogRevision),
			view.SourceCatalogSnapshot.CatalogSnapshotDigest,
			decimalText(view.SourceCatalogSnapshot.SchemaGeneration),
			decimalText(view.SourceCatalogSnapshot.EngineGeneration),
			view.CanonicalizationVersion,
			view.KeyGrammarVersion,
			view.ObjectKeyNormalizationVersion,
			canonicalBytes,
			strconv.Itoa(len(canonicalBytes)),
			strconv.Itoa(len(view.Objects)),
			strconv.Itoa(len(view.RecordTypes)),
			strconv.Itoa(len(view.ProductApps)),
			strconv.Itoa(len(view.NavigationApps)),
		).
		WillReturnResult(sqlmock.NewResult(0, 1))

	for ordinal, identity := range view.Objects {
		mock.ExpectExec(`^INSERT INTO _System_CatalogAssignmentIdentityViewObject`).
			WithArgs(
				orgID,
				digest,
				strconv.Itoa(ordinal),
				identity.Key.NormalizedAPIName,
				identity.ResolvedObjectID,
				string(identity.State),
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
	for ordinal, identity := range view.RecordTypes {
		mock.ExpectExec(`^INSERT INTO _System_CatalogAssignmentIdentityViewRecordType`).
			WithArgs(
				orgID,
				digest,
				strconv.Itoa(ordinal),
				identity.ObjectKey.NormalizedAPIName,
				identity.Key.NamespaceID,
				identity.Key.LocalKey,
				identity.ResolvedRecordTypeID,
				string(identity.State),
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
	for ordinal, identity := range view.ProductApps {
		var resolvedID any
		if identity.ResolvedTarget.Kind ==
			catalogassignmentidentityv1.ProductAppResolvedTargetKindAppInstallationID {
			resolvedID = identity.ResolvedTarget.ResolvedAppInstallationID
		}
		mock.ExpectExec(`^INSERT INTO _System_CatalogAssignmentIdentityViewProductApp`).
			WithArgs(
				orgID,
				digest,
				strconv.Itoa(ordinal),
				identity.Key.AppID,
				string(identity.State),
				uint8(identity.ResolvedTarget.Kind),
				resolvedID,
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
	for ordinal, identity := range view.NavigationApps {
		mock.ExpectExec(`^INSERT INTO _System_CatalogAssignmentIdentityViewNavigationApp`).
			WithArgs(
				orgID,
				digest,
				strconv.Itoa(ordinal),
				identity.Key.NamespaceID,
				identity.Key.LocalKey,
				identity.ResolvedAppConfigID,
				string(identity.State),
			).
			WillReturnResult(sqlmock.NewResult(0, 1))
	}
}

func expectArtifactRead(
	mock sqlmock.Sqlmock,
	fixture storedReadFixture,
	digest string,
	forUpdate bool,
) {
	view := fixture.view
	orgID := view.SourceCatalogSnapshot.OrgID
	byteLength := strconv.Itoa(len(fixture.canonicalBytes))
	mock.ExpectQuery(lockQuery(envelopeMetadataQueryPattern, forUpdate)).
		WithArgs(orgID, digest).
		WillReturnRows(sqlmock.NewRows([]string{
			"OrgId",
			"ViewDigest",
			"CatalogRevision",
			"CatalogSnapshotDigest",
			"SchemaGeneration",
			"EngineGeneration",
			"CanonicalizationVersion",
			"KeyGrammarVersion",
			"ObjectKeyNormalizationVersion",
			"CanonicalByteLength",
			"CanonicalBlobLength",
			"ObjectCount",
			"RecordTypeCount",
			"ProductAppCount",
			"NavigationAppCount",
		}).AddRow(
			[]byte(orgID),
			[]byte(digest),
			[]byte(decimalText(view.SourceCatalogSnapshot.CatalogRevision)),
			[]byte(view.SourceCatalogSnapshot.CatalogSnapshotDigest),
			[]byte(decimalText(view.SourceCatalogSnapshot.SchemaGeneration)),
			[]byte(decimalText(view.SourceCatalogSnapshot.EngineGeneration)),
			[]byte(view.CanonicalizationVersion),
			[]byte(view.KeyGrammarVersion),
			[]byte(view.ObjectKeyNormalizationVersion),
			[]byte(byteLength),
			[]byte(byteLength),
			[]byte(strconv.Itoa(len(view.Objects))),
			[]byte(strconv.Itoa(len(view.RecordTypes))),
			[]byte(strconv.Itoa(len(view.ProductApps))),
			[]byte(strconv.Itoa(len(view.NavigationApps))),
		))
	mock.ExpectQuery(lockQuery(envelopeBlobQueryPattern, forUpdate)).
		WithArgs(orgID, digest).
		WillReturnRows(
			sqlmock.NewRows([]string{"CanonicalBytes"}).
				AddRow(bytes.Clone(fixture.canonicalBytes)),
		)

	objectRows := sqlmock.NewRows([]string{
		"EntryOrdinal",
		"NormalizedAPIName",
		"ResolvedObjectId",
		"State",
	})
	objects := view.Objects
	if fixture.hasObjectOverride {
		objects = fixture.objectRows
	}
	for ordinal, identity := range objects {
		objectRows.AddRow(
			[]byte(strconv.Itoa(ordinal)),
			[]byte(identity.Key.NormalizedAPIName),
			[]byte(identity.ResolvedObjectID),
			[]byte(identity.State),
		)
	}
	mock.ExpectQuery(lockQuery(objectQueryPattern, forUpdate)).
		WithArgs(orgID, digest).
		WillReturnRows(objectRows)
	if len(objects) != len(view.Objects) {
		return
	}

	recordTypeRows := sqlmock.NewRows([]string{
		"EntryOrdinal",
		"ObjectNormalizedAPIName",
		"NamespaceId",
		"LocalKey",
		"ResolvedRecordTypeId",
		"State",
	})
	recordTypes := view.RecordTypes
	if fixture.hasRecordTypeOverride {
		recordTypes = fixture.recordTypeRows
	}
	for ordinal, identity := range recordTypes {
		recordTypeRows.AddRow(
			[]byte(strconv.Itoa(ordinal)),
			[]byte(identity.ObjectKey.NormalizedAPIName),
			[]byte(identity.Key.NamespaceID),
			[]byte(identity.Key.LocalKey),
			[]byte(identity.ResolvedRecordTypeID),
			[]byte(identity.State),
		)
	}
	mock.ExpectQuery(lockQuery(recordTypeQueryPattern, forUpdate)).
		WithArgs(orgID, digest).
		WillReturnRows(recordTypeRows)
	if len(recordTypes) != len(view.RecordTypes) {
		return
	}

	productAppRows := sqlmock.NewRows([]string{
		"EntryOrdinal",
		"AppId",
		"State",
		"ResolvedTargetKind",
		"ResolvedAppInstallationId",
	})
	productApps := view.ProductApps
	if fixture.hasProductAppOverride {
		productApps = fixture.productAppRows
	}
	for ordinal, identity := range productApps {
		var resolvedID any
		if identity.ResolvedTarget.Kind ==
			catalogassignmentidentityv1.ProductAppResolvedTargetKindAppInstallationID {
			resolvedID = identity.ResolvedTarget.ResolvedAppInstallationID
		}
		productAppRows.AddRow(
			[]byte(strconv.Itoa(ordinal)),
			[]byte(identity.Key.AppID),
			[]byte(identity.State),
			[]byte(strconv.Itoa(int(identity.ResolvedTarget.Kind))),
			resolvedID,
		)
	}
	mock.ExpectQuery(lockQuery(productAppQueryPattern, forUpdate)).
		WithArgs(orgID, digest).
		WillReturnRows(productAppRows)
	if len(productApps) != len(view.ProductApps) {
		return
	}

	navigationRows := fixture.navigationAppRows
	if !fixture.hasNavigationAppOverride {
		navigationRows = view.NavigationApps
	}
	navigationAppRows := sqlmock.NewRows([]string{
		"EntryOrdinal",
		"NamespaceId",
		"LocalKey",
		"ResolvedAppConfigId",
		"State",
	})
	for ordinal, identity := range navigationRows {
		navigationAppRows.AddRow(
			[]byte(strconv.Itoa(ordinal)),
			[]byte(identity.Key.NamespaceID),
			[]byte(identity.Key.LocalKey),
			[]byte(identity.ResolvedAppConfigID),
			[]byte(identity.State),
		)
	}
	mock.ExpectQuery(lockQuery(navigationAppQueryPattern, forUpdate)).
		WithArgs(orgID, digest).
		WillReturnRows(navigationAppRows)
}

func lockQuery(pattern string, forUpdate bool) string {
	if forUpdate {
		return pattern + `.*FOR UPDATE$`
	}
	return pattern + `$`
}

func assertMockExpectations(t *testing.T, mock sqlmock.Sqlmock) {
	t.Helper()
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func decimalText(value uint64) string {
	return strconv.FormatUint(value, 10)
}

func assertVectorWitness(
	t *testing.T,
	vector assignmentIdentityViewVector,
	input catalogassignmentidentityv1.View,
) (catalogassignmentidentityv1.View, []byte, string) {
	t.Helper()
	canonical, err := catalogassignmentidentityv1.CanonicalizeView(input)
	if err != nil {
		t.Fatal(err)
	}
	canonicalBytes, err := catalogassignmentidentityv1.CanonicalViewBytes(input)
	if err != nil {
		t.Fatal(err)
	}
	wantLength := parseVectorUint(t, "canonical_bytes_length", vector.CanonicalBytesLength)
	if uint64(len(canonicalBytes)) != wantLength {
		t.Fatalf("canonical byte length = %d, want %d", len(canonicalBytes), wantLength)
	}
	if vector.CanonicalBytesHex != "" {
		wantBytes, err := hex.DecodeString(vector.CanonicalBytesHex)
		if err != nil {
			t.Fatalf("decode canonical vector bytes: %v", err)
		}
		if !bytes.Equal(canonicalBytes, wantBytes) {
			t.Fatalf("canonical bytes differ for vector %q", vector.Name)
		}
	}
	digest, err := catalogassignmentidentityv1.ViewDigest(input)
	if err != nil {
		t.Fatal(err)
	}
	if digest != vector.ViewDigest {
		t.Fatalf("view digest = %q, want %q", digest, vector.ViewDigest)
	}
	return canonical, canonicalBytes, digest
}

func materializedVectorByName(
	t *testing.T,
	name string,
) assignmentIdentityViewVector {
	t.Helper()
	for _, vector := range decodeAssignmentIdentityVectors(t).ViewVectors {
		if vector.Name == name {
			return vector
		}
	}
	t.Fatalf("materialized vector %q is missing", name)
	return assignmentIdentityViewVector{}
}

func decodeAssignmentIdentityVectors(t *testing.T) assignmentIdentityVectorDocument {
	t.Helper()
	var document assignmentIdentityVectorDocument
	decoder := json.NewDecoder(bytes.NewReader(
		catalogassignmentidentityv1.ConformanceVectorsJSON(),
	))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&document); err != nil {
		t.Fatal(err)
	}
	if err := decoder.Decode(&struct{}{}); err != io.EOF {
		t.Fatalf("unexpected trailing conformance JSON: %v", err)
	}
	if document.SchemaVersion !=
		"nexuscrm.catalog-assignment-identity-view-conformance/v1" {
		t.Fatalf("conformance schema = %q", document.SchemaVersion)
	}
	return document
}

type assignmentIdentityVectorDocument struct {
	SchemaVersion        string                         `json:"schema_version"`
	NormalizationVectors []json.RawMessage              `json:"normalization_vectors"`
	ViewVectors          []assignmentIdentityViewVector `json:"view_vectors"`
	RecipeVectors        []json.RawMessage              `json:"recipe_vectors"`
}

type assignmentIdentityViewVector struct {
	Name                 string                       `json:"name"`
	View                 assignmentIdentityVectorView `json:"view"`
	CanonicalBytesLength string                       `json:"canonical_bytes_length"`
	CanonicalBytesHex    string                       `json:"canonical_bytes_hex"`
	ViewDigest           string                       `json:"view_digest"`
}

type assignmentIdentityVectorView struct {
	SourceCatalogSnapshot         assignmentIdentityVectorSnapshot        `json:"source_catalog_snapshot"`
	CanonicalizationVersion       string                                  `json:"canonicalization_version"`
	KeyGrammarVersion             string                                  `json:"key_grammar_version"`
	ObjectKeyNormalizationVersion string                                  `json:"object_key_normalization_version"`
	Objects                       []assignmentIdentityVectorObject        `json:"objects"`
	RecordTypes                   []assignmentIdentityVectorRecordType    `json:"record_types"`
	ProductApps                   []assignmentIdentityVectorProductApp    `json:"product_apps"`
	NavigationApps                []assignmentIdentityVectorNavigationApp `json:"navigation_apps"`
}

type assignmentIdentityVectorSnapshot struct {
	OrgID                 string `json:"org_id"`
	CatalogRevision       string `json:"catalog_revision"`
	CatalogSnapshotDigest string `json:"catalog_snapshot_digest"`
	SchemaGeneration      string `json:"schema_generation"`
	EngineGeneration      string `json:"engine_generation"`
}

type assignmentIdentityVectorObject struct {
	NormalizedAPIName string                                    `json:"normalized_api_name"`
	ResolvedObjectID  string                                    `json:"resolved_object_id"`
	State             catalogassignmentidentityv1.IdentityState `json:"state"`
}

type assignmentIdentityVectorRecordType struct {
	ObjectNormalizedAPIName string                                    `json:"object_normalized_api_name"`
	NamespaceID             string                                    `json:"namespace_id"`
	LocalKey                string                                    `json:"local_key"`
	ResolvedRecordTypeID    string                                    `json:"resolved_record_type_id"`
	State                   catalogassignmentidentityv1.IdentityState `json:"state"`
}

type assignmentIdentityVectorProductApp struct {
	AppID          string                                           `json:"app_id"`
	State          catalogassignmentidentityv1.IdentityState        `json:"state"`
	ResolvedTarget assignmentIdentityVectorProductAppResolvedTarget `json:"resolved_target"`
}

type assignmentIdentityVectorProductAppResolvedTarget struct {
	Kind                      string  `json:"kind"`
	ResolvedAppInstallationID *string `json:"resolved_app_installation_id"`
}

type assignmentIdentityVectorNavigationApp struct {
	NamespaceID         string                                    `json:"namespace_id"`
	LocalKey            string                                    `json:"local_key"`
	ResolvedAppConfigID string                                    `json:"resolved_app_config_id"`
	State               catalogassignmentidentityv1.IdentityState `json:"state"`
}

func (value assignmentIdentityVectorView) value(
	t *testing.T,
) catalogassignmentidentityv1.View {
	t.Helper()
	view := catalogassignmentidentityv1.View{
		SourceCatalogSnapshot:         value.SourceCatalogSnapshot.value(t),
		CanonicalizationVersion:       value.CanonicalizationVersion,
		KeyGrammarVersion:             value.KeyGrammarVersion,
		ObjectKeyNormalizationVersion: value.ObjectKeyNormalizationVersion,
		Objects: make(
			[]catalogassignmentidentityv1.ObjectIdentity,
			len(value.Objects),
		),
		RecordTypes: make(
			[]catalogassignmentidentityv1.RecordTypeIdentity,
			len(value.RecordTypes),
		),
		ProductApps: make(
			[]catalogassignmentidentityv1.ProductAppIdentity,
			len(value.ProductApps),
		),
		NavigationApps: make(
			[]catalogassignmentidentityv1.NavigationAppIdentity,
			len(value.NavigationApps),
		),
	}
	for index, identity := range value.Objects {
		view.Objects[index] = catalogassignmentidentityv1.ObjectIdentity{
			Key: catalogassignmentidentityv1.ObjectKey{
				NormalizedAPIName: identity.NormalizedAPIName,
			},
			ResolvedObjectID: identity.ResolvedObjectID,
			State:            identity.State,
		}
	}
	for index, identity := range value.RecordTypes {
		view.RecordTypes[index] = catalogassignmentidentityv1.RecordTypeIdentity{
			ObjectKey: catalogassignmentidentityv1.ObjectKey{
				NormalizedAPIName: identity.ObjectNormalizedAPIName,
			},
			Key: catalogassignmentidentityv1.RecordTypeKey{
				NamespaceID: identity.NamespaceID,
				LocalKey:    identity.LocalKey,
			},
			ResolvedRecordTypeID: identity.ResolvedRecordTypeID,
			State:                identity.State,
		}
	}
	for index, identity := range value.ProductApps {
		view.ProductApps[index] = catalogassignmentidentityv1.ProductAppIdentity{
			Key: catalogassignmentidentityv1.ProductAppKey{
				AppID: identity.AppID,
			},
			State:          identity.State,
			ResolvedTarget: identity.ResolvedTarget.value(t),
		}
	}
	for index, identity := range value.NavigationApps {
		view.NavigationApps[index] =
			catalogassignmentidentityv1.NavigationAppIdentity{
				Key: catalogassignmentidentityv1.NavigationAppKey{
					NamespaceID: identity.NamespaceID,
					LocalKey:    identity.LocalKey,
				},
				ResolvedAppConfigID: identity.ResolvedAppConfigID,
				State:               identity.State,
			}
	}
	return view
}

func (value assignmentIdentityVectorSnapshot) value(
	t *testing.T,
) catalogpublicationv1.CatalogSnapshotBinding {
	t.Helper()
	return catalogpublicationv1.CatalogSnapshotBinding{
		OrgID:                 value.OrgID,
		CatalogRevision:       parseVectorUint(t, "catalog_revision", value.CatalogRevision),
		CatalogSnapshotDigest: value.CatalogSnapshotDigest,
		SchemaGeneration:      parseVectorUint(t, "schema_generation", value.SchemaGeneration),
		EngineGeneration:      parseVectorUint(t, "engine_generation", value.EngineGeneration),
	}
}

func (value assignmentIdentityVectorProductAppResolvedTarget) value(
	t *testing.T,
) catalogassignmentidentityv1.ProductAppResolvedTarget {
	t.Helper()
	switch value.Kind {
	case "ABSENT":
		if value.ResolvedAppInstallationID != nil {
			t.Fatal("ABSENT target carries resolved_app_installation_id")
		}
		return catalogassignmentidentityv1.ProductAppResolvedTarget{
			Kind: catalogassignmentidentityv1.ProductAppResolvedTargetKindAbsent,
		}
	case "APP_INSTALLATION_ID":
		if value.ResolvedAppInstallationID == nil {
			t.Fatal("APP_INSTALLATION_ID target omits resolved_app_installation_id")
		}
		return catalogassignmentidentityv1.ProductAppResolvedTarget{
			Kind: catalogassignmentidentityv1.
				ProductAppResolvedTargetKindAppInstallationID,
			ResolvedAppInstallationID: *value.ResolvedAppInstallationID,
		}
	default:
		t.Fatalf("unknown Product-App resolved target kind %q", value.Kind)
		return catalogassignmentidentityv1.ProductAppResolvedTarget{}
	}
}

func parseVectorUint(t *testing.T, label, raw string) uint64 {
	t.Helper()
	value, err := strconv.ParseUint(raw, 10, 64)
	if err != nil || strconv.FormatUint(value, 10) != raw {
		t.Fatalf("%s = %q is not a canonical uint64: %v", label, raw, err)
	}
	return value
}