package catalogassignmentidentity

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"unicode/utf8"

	catalogassignmentidentityv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalogassignmentidentityv1"
)

const maxOrgIDBytes = 255

type preparedArtifact struct {
	view           catalogassignmentidentityv1.View
	digest         string
	canonicalBytes []byte
}

type storedEnvelope struct {
	view                catalogassignmentidentityv1.View
	digest              string
	canonicalBytes      []byte
	canonicalByteLength int
	objectCount         int
	recordTypeCount     int
	productAppCount     int
	navigationAppCount  int
}

type storedArtifact struct {
	view           catalogassignmentidentityv1.View
	digest         string
	canonicalBytes []byte
}

type rowQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

type rowsQuerier interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
}

// PutExact validates, canonicalizes, and atomically inserts one immutable
// view. An exact replay recovers the existing artifact; the same digest can
// never overwrite different durable bytes.
func (s *Store) PutExact(
	ctx context.Context,
	view catalogassignmentidentityv1.View,
) (string, error) {
	prepared, err := prepareArtifact(view)
	if err != nil {
		return "", err
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return "", err
	}
	defer rollbackTx(tx)

	existing, err := readStoredArtifact(
		ctx,
		tx,
		prepared.view.SourceCatalogSnapshot.OrgID,
		prepared.digest,
		true,
	)
	if err == nil {
		if !bytes.Equal(existing.canonicalBytes, prepared.canonicalBytes) {
			return "", fmt.Errorf(
				"%w: digest %q already names different canonical bytes",
				ErrCorrupt,
				prepared.digest,
			)
		}
		if err := tx.Commit(); err != nil {
			return s.recoverExact(ctx, prepared, err)
		}
		return prepared.digest, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return "", err
	}

	if err := requireNoOrphanRows(ctx, tx, prepared); err != nil {
		return "", err
	}
	if err := insertEnvelope(ctx, tx, prepared); err != nil {
		// A concurrent exact insert can win after the absent-row read. Recover
		// only an exact, independently revalidated artifact; every other insert
		// error remains the caller-visible failure.
		_ = tx.Rollback()
		return s.recoverExact(ctx, prepared, err)
	}
	if err := insertObjects(ctx, tx, prepared); err != nil {
		return "", err
	}
	if err := insertRecordTypes(ctx, tx, prepared); err != nil {
		return "", err
	}
	if err := insertProductApps(ctx, tx, prepared); err != nil {
		return "", err
	}
	if err := insertNavigationApps(ctx, tx, prepared); err != nil {
		return "", err
	}
	inserted, err := readStoredArtifact(
		ctx,
		tx,
		prepared.view.SourceCatalogSnapshot.OrgID,
		prepared.digest,
		true,
	)
	if err != nil {
		return "", err
	}
	if !bytes.Equal(inserted.canonicalBytes, prepared.canonicalBytes) {
		return "", fmt.Errorf(
			"%w: inserted digest %q does not reproduce its canonical bytes",
			ErrCorrupt,
			prepared.digest,
		)
	}
	if err := tx.Commit(); err != nil {
		return s.recoverExact(ctx, prepared, err)
	}
	return prepared.digest, nil
}

func (s *Store) recoverExact(
	ctx context.Context,
	prepared preparedArtifact,
	originalErr error,
) (string, error) {
	recovered, recoverErr := s.ReadExact(
		ctx,
		prepared.view.SourceCatalogSnapshot.OrgID,
		prepared.digest,
	)
	if recoverErr == nil {
		recoveredBytes, bytesErr := catalogassignmentidentityv1.CanonicalViewBytes(recovered)
		if bytesErr != nil {
			return "", fmt.Errorf("%w: recovered view is invalid: %v", ErrCorrupt, bytesErr)
		}
		if !bytes.Equal(recoveredBytes, prepared.canonicalBytes) {
			return "", fmt.Errorf(
				"%w: digest %q recovered different canonical bytes",
				ErrCorrupt,
				prepared.digest,
			)
		}
		return prepared.digest, nil
	}
	if !errors.Is(recoverErr, ErrNotFound) {
		return "", recoverErr
	}
	return "", originalErr
}

// ReadExact loads one immutable view by its explicit org and content digest.
// Every normalized row, canonical byte, source binding, and digest is
// revalidated before return.
func (s *Store) ReadExact(
	ctx context.Context,
	orgID string,
	digest string,
) (catalogassignmentidentityv1.View, error) {
	if err := validateLookupOrgID(orgID); err != nil {
		return catalogassignmentidentityv1.View{}, err
	}
	if err := validateDigest(digest); err != nil {
		return catalogassignmentidentityv1.View{}, err
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return catalogassignmentidentityv1.View{}, err
	}
	defer rollbackTx(tx)
	artifact, err := readStoredArtifact(ctx, tx, orgID, digest, false)
	if errors.Is(err, sql.ErrNoRows) {
		return catalogassignmentidentityv1.View{}, fmt.Errorf(
			"%w: view %q for org %q",
			ErrNotFound,
			digest,
			orgID,
		)
	}
	if err != nil {
		return catalogassignmentidentityv1.View{}, err
	}
	if err := tx.Commit(); err != nil {
		return catalogassignmentidentityv1.View{}, err
	}
	return artifact.view, nil
}

func prepareArtifact(view catalogassignmentidentityv1.View) (preparedArtifact, error) {
	canonical, err := catalogassignmentidentityv1.CanonicalizeView(view)
	if err != nil {
		return preparedArtifact{}, fmt.Errorf(
			"catalog assignment identity store: invalid view: %w",
			err,
		)
	}
	canonicalBytes, err := catalogassignmentidentityv1.CanonicalViewBytes(canonical)
	if err != nil {
		return preparedArtifact{}, fmt.Errorf(
			"catalog assignment identity store: canonicalize view: %w",
			err,
		)
	}
	digest, err := catalogassignmentidentityv1.ViewDigest(canonical)
	if err != nil {
		return preparedArtifact{}, fmt.Errorf(
			"catalog assignment identity store: digest view: %w",
			err,
		)
	}
	return preparedArtifact{
		view:           canonical,
		digest:         digest,
		canonicalBytes: canonicalBytes,
	}, nil
}

func requireNoOrphanRows(
	ctx context.Context,
	query rowQuerier,
	artifact preparedArtifact,
) error {
	tables := []string{
		"_System_CatalogAssignmentIdentityViewObject",
		"_System_CatalogAssignmentIdentityViewRecordType",
		"_System_CatalogAssignmentIdentityViewProductApp",
		"_System_CatalogAssignmentIdentityViewNavigationApp",
	}
	for _, table := range tables {
		statement := `SELECT 1 FROM ` + table + `
WHERE OrgId = ? AND ViewDigest = ?
LIMIT 1 FOR UPDATE`
		var marker int
		err := query.QueryRowContext(
			ctx,
			statement,
			artifact.view.SourceCatalogSnapshot.OrgID,
			artifact.digest,
		).Scan(&marker)
		switch {
		case err == nil:
			return fmt.Errorf(
				"%w: child rows exist without an assignment-view envelope in %s",
				ErrCorrupt,
				table,
			)
		case errors.Is(err, sql.ErrNoRows):
			continue
		default:
			return err
		}
	}
	return nil
}

func readStoredArtifact(
	ctx context.Context,
	query interface {
		rowQuerier
		rowsQuerier
	},
	orgID string,
	digest string,
	forUpdate bool,
) (storedArtifact, error) {
	envelope, err := readEnvelope(ctx, query, orgID, digest, forUpdate)
	if err != nil {
		return storedArtifact{}, err
	}
	envelope.view.Objects, err = readObjects(
		ctx, query, orgID, digest, envelope.objectCount, forUpdate,
	)
	if err != nil {
		return storedArtifact{}, err
	}
	envelope.view.RecordTypes, err = readRecordTypes(
		ctx, query, orgID, digest, envelope.recordTypeCount, forUpdate,
	)
	if err != nil {
		return storedArtifact{}, err
	}
	envelope.view.ProductApps, err = readProductApps(
		ctx, query, orgID, digest, envelope.productAppCount, forUpdate,
	)
	if err != nil {
		return storedArtifact{}, err
	}
	envelope.view.NavigationApps, err = readNavigationApps(
		ctx, query, orgID, digest, envelope.navigationAppCount, forUpdate,
	)
	if err != nil {
		return storedArtifact{}, err
	}

	if err := catalogassignmentidentityv1.ValidateView(envelope.view); err != nil {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view %q is invalid: %v",
			ErrCorrupt,
			digest,
			err,
		)
	}
	canonical, err := catalogassignmentidentityv1.CanonicalizeView(envelope.view)
	if err != nil {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view %q cannot be canonicalized: %v",
			ErrCorrupt,
			digest,
			err,
		)
	}
	if !sameEntryOrder(envelope.view, canonical) {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view %q rows are not in canonical ordinal order",
			ErrCorrupt,
			digest,
		)
	}
	canonicalBytes, err := catalogassignmentidentityv1.CanonicalViewBytes(envelope.view)
	if err != nil {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view %q bytes are invalid: %v",
			ErrCorrupt,
			digest,
			err,
		)
	}
	if !bytes.Equal(canonicalBytes, envelope.canonicalBytes) {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view %q canonical bytes differ from its envelope",
			ErrCorrupt,
			digest,
		)
	}
	computedDigest, err := catalogassignmentidentityv1.ViewDigest(envelope.view)
	if err != nil {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view %q digest cannot be computed: %v",
			ErrCorrupt,
			digest,
			err,
		)
	}
	if computedDigest != envelope.digest || computedDigest != digest {
		return storedArtifact{}, fmt.Errorf(
			"%w: stored view digest is %q, envelope %q, requested %q",
			ErrCorrupt,
			computedDigest,
			envelope.digest,
			digest,
		)
	}
	return storedArtifact{
		view:           envelope.view,
		digest:         envelope.digest,
		canonicalBytes: canonicalBytes,
	}, nil
}

func readEnvelope(
	ctx context.Context,
	query rowQuerier,
	orgID string,
	digest string,
	forUpdate bool,
) (storedEnvelope, error) {
	statement := `SELECT OrgId, ViewDigest, CatalogRevision, CatalogSnapshotDigest,
SchemaGeneration, EngineGeneration, CanonicalizationVersion, KeyGrammarVersion,
ObjectKeyNormalizationVersion, CanonicalByteLength, OCTET_LENGTH(CanonicalBytes),
ObjectCount, RecordTypeCount, ProductAppCount, NavigationAppCount
FROM _System_CatalogAssignmentIdentityView
WHERE OrgId = ? AND ViewDigest = ?` + lockSuffix(forUpdate)
	var storedOrg, storedDigest, snapshotDigest []byte
	var canonicalizationVersion, keyGrammarVersion, normalizationVersion []byte
	var rawCatalogRevision, rawSchemaGeneration, rawEngineGeneration []byte
	var rawCanonicalByteLength, rawActualCanonicalByteLength []byte
	var rawObjectCount, rawRecordTypeCount []byte
	var rawProductAppCount, rawNavigationAppCount []byte
	err := query.QueryRowContext(ctx, statement, orgID, digest).Scan(
		&storedOrg,
		&storedDigest,
		&rawCatalogRevision,
		&snapshotDigest,
		&rawSchemaGeneration,
		&rawEngineGeneration,
		&canonicalizationVersion,
		&keyGrammarVersion,
		&normalizationVersion,
		&rawCanonicalByteLength,
		&rawActualCanonicalByteLength,
		&rawObjectCount,
		&rawRecordTypeCount,
		&rawProductAppCount,
		&rawNavigationAppCount,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	if string(storedOrg) != orgID || string(storedDigest) != digest {
		return storedEnvelope{}, fmt.Errorf(
			"%w: assignment-view envelope identity differs from its exact lookup",
			ErrCorrupt,
		)
	}
	envelope := storedEnvelope{
		digest: string(storedDigest),
	}
	envelope.view.SourceCatalogSnapshot.OrgID = string(storedOrg)
	envelope.view.SourceCatalogSnapshot.CatalogSnapshotDigest = string(snapshotDigest)
	envelope.view.CanonicalizationVersion = string(canonicalizationVersion)
	envelope.view.KeyGrammarVersion = string(keyGrammarVersion)
	envelope.view.ObjectKeyNormalizationVersion = string(normalizationVersion)
	envelope.view.SourceCatalogSnapshot.CatalogRevision, err = parseDurableUint64(
		"CatalogRevision", rawCatalogRevision,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	envelope.view.SourceCatalogSnapshot.SchemaGeneration, err = parseDurableUint64(
		"SchemaGeneration", rawSchemaGeneration,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	envelope.view.SourceCatalogSnapshot.EngineGeneration, err = parseDurableUint64(
		"EngineGeneration", rawEngineGeneration,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	canonicalByteLength, err := parseDurableUint64(
		"CanonicalByteLength", rawCanonicalByteLength,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	actualCanonicalByteLength, err := parseDurableUint64(
		"OCTET_LENGTH(CanonicalBytes)", rawActualCanonicalByteLength,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	if canonicalByteLength == 0 ||
		canonicalByteLength > catalogassignmentidentityv1.MaxCanonicalViewBytes ||
		actualCanonicalByteLength != canonicalByteLength {
		return storedEnvelope{}, fmt.Errorf(
			"%w: CanonicalByteLength does not match the bounded envelope bytes",
			ErrCorrupt,
		)
	}
	envelope.canonicalByteLength = int(canonicalByteLength)
	envelope.objectCount, err = parseEntryCount("ObjectCount", rawObjectCount)
	if err != nil {
		return storedEnvelope{}, err
	}
	envelope.recordTypeCount, err = parseEntryCount("RecordTypeCount", rawRecordTypeCount)
	if err != nil {
		return storedEnvelope{}, err
	}
	envelope.productAppCount, err = parseEntryCount("ProductAppCount", rawProductAppCount)
	if err != nil {
		return storedEnvelope{}, err
	}
	envelope.navigationAppCount, err = parseEntryCount(
		"NavigationAppCount", rawNavigationAppCount,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	envelope.canonicalBytes, err = readCanonicalBytes(
		ctx,
		query,
		orgID,
		digest,
		envelope.canonicalByteLength,
		forUpdate,
	)
	if err != nil {
		return storedEnvelope{}, err
	}
	return envelope, nil
}

func readCanonicalBytes(
	ctx context.Context,
	query rowQuerier,
	orgID string,
	digest string,
	expectedLength int,
	forUpdate bool,
) ([]byte, error) {
	statement := `SELECT CanonicalBytes
FROM _System_CatalogAssignmentIdentityView
WHERE OrgId = ? AND ViewDigest = ?` + lockSuffix(forUpdate)
	var canonicalBytes []byte
	if err := query.QueryRowContext(ctx, statement, orgID, digest).Scan(
		&canonicalBytes,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, fmt.Errorf(
				"%w: assignment-view envelope disappeared during exact load",
				ErrCorrupt,
			)
		}
		return nil, err
	}
	if len(canonicalBytes) != expectedLength {
		return nil, fmt.Errorf(
			"%w: canonical bytes changed during exact load",
			ErrCorrupt,
		)
	}
	return bytes.Clone(canonicalBytes), nil
}

func readObjects(
	ctx context.Context,
	query rowsQuerier,
	orgID string,
	digest string,
	expected int,
	forUpdate bool,
) ([]catalogassignmentidentityv1.ObjectIdentity, error) {
	rows, err := query.QueryContext(ctx, `SELECT EntryOrdinal, NormalizedAPIName,
ResolvedObjectId, State
FROM _System_CatalogAssignmentIdentityViewObject
WHERE OrgId = ? AND ViewDigest = ?
ORDER BY EntryOrdinal ASC`+lockSuffix(forUpdate), orgID, digest)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]catalogassignmentidentityv1.ObjectIdentity, 0, expected)
	for rows.Next() {
		if len(result) >= expected {
			return nil, fmt.Errorf("%w: object rows exceed ObjectCount", ErrCorrupt)
		}
		var rawOrdinal, key, resolvedID, state []byte
		if err := rows.Scan(&rawOrdinal, &key, &resolvedID, &state); err != nil {
			return nil, err
		}
		if err := requireOrdinal(rawOrdinal, len(result), "object"); err != nil {
			return nil, err
		}
		result = append(result, catalogassignmentidentityv1.ObjectIdentity{
			Key: catalogassignmentidentityv1.ObjectKey{
				NormalizedAPIName: string(key),
			},
			ResolvedObjectID: string(resolvedID),
			State:            catalogassignmentidentityv1.IdentityState(state),
		})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(result) != expected {
		return nil, fmt.Errorf(
			"%w: object row count is %d, envelope says %d",
			ErrCorrupt,
			len(result),
			expected,
		)
	}
	return result, nil
}

func readRecordTypes(
	ctx context.Context,
	query rowsQuerier,
	orgID string,
	digest string,
	expected int,
	forUpdate bool,
) ([]catalogassignmentidentityv1.RecordTypeIdentity, error) {
	rows, err := query.QueryContext(ctx, `SELECT EntryOrdinal, ObjectNormalizedAPIName,
NamespaceId, LocalKey, ResolvedRecordTypeId, State
FROM _System_CatalogAssignmentIdentityViewRecordType
WHERE OrgId = ? AND ViewDigest = ?
ORDER BY EntryOrdinal ASC`+lockSuffix(forUpdate), orgID, digest)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]catalogassignmentidentityv1.RecordTypeIdentity, 0, expected)
	for rows.Next() {
		if len(result) >= expected {
			return nil, fmt.Errorf("%w: record-type rows exceed RecordTypeCount", ErrCorrupt)
		}
		var rawOrdinal, objectKey, namespaceID, localKey, resolvedID, state []byte
		if err := rows.Scan(
			&rawOrdinal,
			&objectKey,
			&namespaceID,
			&localKey,
			&resolvedID,
			&state,
		); err != nil {
			return nil, err
		}
		if err := requireOrdinal(rawOrdinal, len(result), "record-type"); err != nil {
			return nil, err
		}
		result = append(result, catalogassignmentidentityv1.RecordTypeIdentity{
			ObjectKey: catalogassignmentidentityv1.ObjectKey{
				NormalizedAPIName: string(objectKey),
			},
			Key: catalogassignmentidentityv1.RecordTypeKey{
				NamespaceID: string(namespaceID),
				LocalKey:    string(localKey),
			},
			ResolvedRecordTypeID: string(resolvedID),
			State:                catalogassignmentidentityv1.IdentityState(state),
		})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(result) != expected {
		return nil, fmt.Errorf(
			"%w: record-type row count is %d, envelope says %d",
			ErrCorrupt,
			len(result),
			expected,
		)
	}
	return result, nil
}

func readProductApps(
	ctx context.Context,
	query rowsQuerier,
	orgID string,
	digest string,
	expected int,
	forUpdate bool,
) ([]catalogassignmentidentityv1.ProductAppIdentity, error) {
	rows, err := query.QueryContext(ctx, `SELECT EntryOrdinal, AppId, State,
ResolvedTargetKind, ResolvedAppInstallationId
FROM _System_CatalogAssignmentIdentityViewProductApp
WHERE OrgId = ? AND ViewDigest = ?
ORDER BY EntryOrdinal ASC`+lockSuffix(forUpdate), orgID, digest)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]catalogassignmentidentityv1.ProductAppIdentity, 0, expected)
	for rows.Next() {
		if len(result) >= expected {
			return nil, fmt.Errorf("%w: Product-App rows exceed ProductAppCount", ErrCorrupt)
		}
		var rawOrdinal, appID, state, rawTargetKind []byte
		var resolvedID sql.NullString
		if err := rows.Scan(
			&rawOrdinal,
			&appID,
			&state,
			&rawTargetKind,
			&resolvedID,
		); err != nil {
			return nil, err
		}
		if err := requireOrdinal(rawOrdinal, len(result), "Product-App"); err != nil {
			return nil, err
		}
		targetKind, err := parseDurableUint64("ResolvedTargetKind", rawTargetKind)
		if err != nil || targetKind > 0xff {
			return nil, fmt.Errorf("%w: invalid Product-App target kind", ErrCorrupt)
		}
		target := catalogassignmentidentityv1.ProductAppResolvedTarget{
			Kind: catalogassignmentidentityv1.ProductAppResolvedTargetKind(targetKind),
		}
		if resolvedID.Valid {
			target.ResolvedAppInstallationID = resolvedID.String
		}
		result = append(result, catalogassignmentidentityv1.ProductAppIdentity{
			Key: catalogassignmentidentityv1.ProductAppKey{
				AppID: string(appID),
			},
			State:          catalogassignmentidentityv1.IdentityState(state),
			ResolvedTarget: target,
		})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(result) != expected {
		return nil, fmt.Errorf(
			"%w: Product-App row count is %d, envelope says %d",
			ErrCorrupt,
			len(result),
			expected,
		)
	}
	return result, nil
}

func readNavigationApps(
	ctx context.Context,
	query rowsQuerier,
	orgID string,
	digest string,
	expected int,
	forUpdate bool,
) ([]catalogassignmentidentityv1.NavigationAppIdentity, error) {
	rows, err := query.QueryContext(ctx, `SELECT EntryOrdinal, NamespaceId,
LocalKey, ResolvedAppConfigId, State
FROM _System_CatalogAssignmentIdentityViewNavigationApp
WHERE OrgId = ? AND ViewDigest = ?
ORDER BY EntryOrdinal ASC`+lockSuffix(forUpdate), orgID, digest)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]catalogassignmentidentityv1.NavigationAppIdentity, 0, expected)
	for rows.Next() {
		if len(result) >= expected {
			return nil, fmt.Errorf(
				"%w: navigation-App rows exceed NavigationAppCount",
				ErrCorrupt,
			)
		}
		var rawOrdinal, namespaceID, localKey, resolvedID, state []byte
		if err := rows.Scan(
			&rawOrdinal,
			&namespaceID,
			&localKey,
			&resolvedID,
			&state,
		); err != nil {
			return nil, err
		}
		if err := requireOrdinal(rawOrdinal, len(result), "navigation-App"); err != nil {
			return nil, err
		}
		result = append(result, catalogassignmentidentityv1.NavigationAppIdentity{
			Key: catalogassignmentidentityv1.NavigationAppKey{
				NamespaceID: string(namespaceID),
				LocalKey:    string(localKey),
			},
			ResolvedAppConfigID: string(resolvedID),
			State:               catalogassignmentidentityv1.IdentityState(state),
		})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if len(result) != expected {
		return nil, fmt.Errorf(
			"%w: navigation-App row count is %d, envelope says %d",
			ErrCorrupt,
			len(result),
			expected,
		)
	}
	return result, nil
}

func insertEnvelope(ctx context.Context, tx *sql.Tx, artifact preparedArtifact) error {
	view := artifact.view
	result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogAssignmentIdentityView
(OrgId, ViewDigest, CatalogRevision, CatalogSnapshotDigest, SchemaGeneration,
EngineGeneration, CanonicalizationVersion, KeyGrammarVersion,
ObjectKeyNormalizationVersion, CanonicalBytes, CanonicalByteLength,
ObjectCount, RecordTypeCount, ProductAppCount, NavigationAppCount, CreatedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6))`,
		view.SourceCatalogSnapshot.OrgID,
		artifact.digest,
		strconv.FormatUint(view.SourceCatalogSnapshot.CatalogRevision, 10),
		view.SourceCatalogSnapshot.CatalogSnapshotDigest,
		strconv.FormatUint(view.SourceCatalogSnapshot.SchemaGeneration, 10),
		strconv.FormatUint(view.SourceCatalogSnapshot.EngineGeneration, 10),
		view.CanonicalizationVersion,
		view.KeyGrammarVersion,
		view.ObjectKeyNormalizationVersion,
		artifact.canonicalBytes,
		strconv.Itoa(len(artifact.canonicalBytes)),
		strconv.Itoa(len(view.Objects)),
		strconv.Itoa(len(view.RecordTypes)),
		strconv.Itoa(len(view.ProductApps)),
		strconv.Itoa(len(view.NavigationApps)),
	)
	if err != nil {
		return err
	}
	return requireOneInsert(result)
}

func insertObjects(ctx context.Context, tx *sql.Tx, artifact preparedArtifact) error {
	for ordinal, identity := range artifact.view.Objects {
		result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogAssignmentIdentityViewObject
(OrgId, ViewDigest, EntryOrdinal, NormalizedAPIName, ResolvedObjectId, State)
VALUES (?, ?, ?, ?, ?, ?)`,
			artifact.view.SourceCatalogSnapshot.OrgID,
			artifact.digest,
			strconv.Itoa(ordinal),
			identity.Key.NormalizedAPIName,
			identity.ResolvedObjectID,
			string(identity.State),
		)
		if err != nil {
			return err
		}
		if err := requireOneInsert(result); err != nil {
			return err
		}
	}
	return nil
}

func insertRecordTypes(ctx context.Context, tx *sql.Tx, artifact preparedArtifact) error {
	for ordinal, identity := range artifact.view.RecordTypes {
		result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogAssignmentIdentityViewRecordType
(OrgId, ViewDigest, EntryOrdinal, ObjectNormalizedAPIName, NamespaceId, LocalKey,
ResolvedRecordTypeId, State)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
			artifact.view.SourceCatalogSnapshot.OrgID,
			artifact.digest,
			strconv.Itoa(ordinal),
			identity.ObjectKey.NormalizedAPIName,
			identity.Key.NamespaceID,
			identity.Key.LocalKey,
			identity.ResolvedRecordTypeID,
			string(identity.State),
		)
		if err != nil {
			return err
		}
		if err := requireOneInsert(result); err != nil {
			return err
		}
	}
	return nil
}

func insertProductApps(ctx context.Context, tx *sql.Tx, artifact preparedArtifact) error {
	for ordinal, identity := range artifact.view.ProductApps {
		var resolvedID any
		if identity.ResolvedTarget.Kind ==
			catalogassignmentidentityv1.ProductAppResolvedTargetKindAppInstallationID {
			resolvedID = identity.ResolvedTarget.ResolvedAppInstallationID
		}
		result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogAssignmentIdentityViewProductApp
(OrgId, ViewDigest, EntryOrdinal, AppId, State, ResolvedTargetKind,
ResolvedAppInstallationId)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
			artifact.view.SourceCatalogSnapshot.OrgID,
			artifact.digest,
			strconv.Itoa(ordinal),
			identity.Key.AppID,
			string(identity.State),
			uint8(identity.ResolvedTarget.Kind),
			resolvedID,
		)
		if err != nil {
			return err
		}
		if err := requireOneInsert(result); err != nil {
			return err
		}
	}
	return nil
}

func insertNavigationApps(ctx context.Context, tx *sql.Tx, artifact preparedArtifact) error {
	for ordinal, identity := range artifact.view.NavigationApps {
		result, err := tx.ExecContext(ctx, `INSERT INTO _System_CatalogAssignmentIdentityViewNavigationApp
(OrgId, ViewDigest, EntryOrdinal, NamespaceId, LocalKey, ResolvedAppConfigId,
State)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
			artifact.view.SourceCatalogSnapshot.OrgID,
			artifact.digest,
			strconv.Itoa(ordinal),
			identity.Key.NamespaceID,
			identity.Key.LocalKey,
			identity.ResolvedAppConfigID,
			string(identity.State),
		)
		if err != nil {
			return err
		}
		if err := requireOneInsert(result); err != nil {
			return err
		}
	}
	return nil
}

func requireOneInsert(result sql.Result) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return fmt.Errorf(
			"%w: immutable artifact insert affected %d rows, want 1",
			ErrCorrupt,
			affected,
		)
	}
	return nil
}

func sameEntryOrder(
	left catalogassignmentidentityv1.View,
	right catalogassignmentidentityv1.View,
) bool {
	return slices.Equal(left.Objects, right.Objects) &&
		slices.Equal(left.RecordTypes, right.RecordTypes) &&
		slices.Equal(left.ProductApps, right.ProductApps) &&
		slices.Equal(left.NavigationApps, right.NavigationApps)
}

func requireOrdinal(raw []byte, wanted int, dimension string) error {
	ordinal, err := parseDurableUint64(dimension+" EntryOrdinal", raw)
	if err != nil || ordinal != uint64(wanted) {
		return fmt.Errorf(
			"%w: %s ordinals are not exact and contiguous",
			ErrCorrupt,
			dimension,
		)
	}
	return nil
}

func parseEntryCount(label string, raw []byte) (int, error) {
	value, err := parseDurableUint64(label, raw)
	if err != nil || value > catalogassignmentidentityv1.MaxEntriesPerDimension {
		return 0, fmt.Errorf("%w: invalid %s", ErrCorrupt, label)
	}
	return int(value), nil
}

func parseDurableUint64(label string, raw []byte) (uint64, error) {
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("%w: invalid %s", ErrCorrupt, label)
	}
	return value, nil
}

func validateLookupOrgID(value string) error {
	if value == "" || len(value) > maxOrgIDBytes ||
		!utf8.ValidString(value) {
		return fmt.Errorf(
			"catalog assignment identity store: OrgID must be well-formed UTF-8 with 1..%d bytes",
			maxOrgIDBytes,
		)
	}
	if value[0] == ' ' || value[len(value)-1] == ' ' {
		return fmt.Errorf("catalog assignment identity store: OrgID must not have surrounding ASCII space")
	}
	for _, char := range []byte(value) {
		if char < 0x20 || char == 0x7f {
			return fmt.Errorf("catalog assignment identity store: OrgID must not contain ASCII control or DEL bytes")
		}
	}
	return nil
}

func validateDigest(value string) error {
	const prefix = "sha256:"
	if len(value) != len(prefix)+64 || !strings.HasPrefix(value, prefix) {
		return fmt.Errorf("catalog assignment identity store: digest must be sha256:<64 lowercase hex>")
	}
	hexValue := value[len(prefix):]
	if strings.ToLower(hexValue) != hexValue {
		return fmt.Errorf("catalog assignment identity store: digest hex must be lowercase")
	}
	if _, err := hex.DecodeString(hexValue); err != nil {
		return fmt.Errorf("catalog assignment identity store: digest hex is invalid")
	}
	return nil
}

func lockSuffix(forUpdate bool) string {
	if forUpdate {
		return " FOR UPDATE"
	}
	return ""
}

func rollbackTx(tx *sql.Tx) {
	_ = tx.Rollback()
}