package catalogapp

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"unicode"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
)

const (
	releaseSourceReceiptSchema          = "nexuscrm.catalog-app-release-source-receipt/v1"
	releaseSourceNonPromotable          = "NON_PROMOTABLE"
	releaseAdmissionConstraint          = "chk_catalog_app_release_admission"
	releaseSourceReceiptConstraint      = "chk_catalog_app_release_source_disposition"
	releaseAdmissionColumnMigration     = `ALTER TABLE _System_CatalogAppRelease ADD COLUMN IF NOT EXISTS AdmissionKind VARCHAR(32) NOT NULL DEFAULT 'SIGNED_PUBLISHER'`
	releaseAttestationKeyMigration      = `ALTER TABLE _System_CatalogAppRelease MODIFY COLUMN AttestationKeyId VARCHAR(128) NULL`
	releaseAttestationDigestMigration   = `ALTER TABLE _System_CatalogAppRelease MODIFY COLUMN AttestationDigest VARCHAR(71) NULL`
	releaseAttestationTimeMigration     = `ALTER TABLE _System_CatalogAppRelease MODIFY COLUMN AttestationVerifiedAt DATETIME(6) NULL`
	releaseAdmissionCheckExpression     = `(AdmissionKind = 'SIGNED_PUBLISHER' AND AttestationKeyId IS NOT NULL AND AttestationDigest IS NOT NULL AND AttestationVerifiedAt IS NOT NULL) OR (AdmissionKind = 'STOPPED_SOURCE_NON_PROMOTABLE' AND AttestationKeyId IS NULL AND AttestationDigest IS NULL AND AttestationVerifiedAt IS NULL)`
	releaseSourceReceiptCheckExpression = `PromotionDisposition = 'NON_PROMOTABLE'`
	releaseAdmissionConstraintDDL       = `ALTER TABLE _System_CatalogAppRelease ADD CONSTRAINT chk_catalog_app_release_admission CHECK (` + releaseAdmissionCheckExpression + `)`
	releaseSourceReceiptConstraintDDL   = `ALTER TABLE _System_CatalogAppReleaseSourceReceipt ADD CONSTRAINT chk_catalog_app_release_source_disposition CHECK (` + releaseSourceReceiptCheckExpression + `)`
)

var lowercaseGitCommit = regexp.MustCompile(`^[0-9a-f]{40}$`)

type releaseAdmissionColumnState struct {
	columnType string
	nullable   string
	defaultVal sql.NullString
}

type retainedReleaseAdmission struct {
	appID                 string
	appReleaseDigest      string
	kind                  ReleaseAdmissionKind
	attestationKeyID      sql.NullString
	attestationDigest     sql.NullString
	attestationVerifiedAt sql.NullTime
	hasArtifactReceipt    bool
	hasSourceReceipt      bool
}

type retainedSourceReleaseReceipt struct {
	appID                string
	appReleaseDigest     string
	receiptSchema        string
	sourceRepository     string
	sourceCommit         string
	packageContentDigest string
	promotionDisposition string
	receiptDigest        string
	releaseAdmissionKind sql.NullString
}

type retainedArtifactReleaseReceiptParent struct {
	appID                string
	appReleaseDigest     string
	releaseAdmissionKind sql.NullString
}

// ensureReleaseAdmissionSchema is the stopped, idempotent physical migration
// from the signed-only AppRelease shape. It does not create an unsigned import
// or lifecycle path. Existing signed rows retain their exact behavior; rows
// with an ambiguous provenance tuple stop the apply before readiness.
func ensureReleaseAdmissionSchema(ctx context.Context, db *sql.DB) error {
	if _, err := db.ExecContext(ctx, releaseAdmissionColumnMigration); err != nil {
		return fmt.Errorf("add Catalog AppRelease admission kind: %w", err)
	}
	columns, err := readReleaseAdmissionColumns(ctx, db)
	if err != nil {
		return err
	}
	if err := validateReleaseAdmissionColumnBase(columns); err != nil {
		return err
	}

	migrations := make([]string, 0, 3)
	for _, column := range []struct {
		name string
		ddl  string
	}{
		{name: "AttestationKeyId", ddl: releaseAttestationKeyMigration},
		{name: "AttestationDigest", ddl: releaseAttestationDigestMigration},
		{name: "AttestationVerifiedAt", ddl: releaseAttestationTimeMigration},
	} {
		if columns[column.name].nullable == "NO" {
			migrations = append(migrations, column.ddl)
		}
	}
	for _, migration := range migrations {
		if _, err := db.ExecContext(ctx, migration); err != nil {
			return fmt.Errorf("make Catalog AppRelease attestation tuple nullable: %w", err)
		}
	}
	if len(migrations) != 0 {
		columns, err = readReleaseAdmissionColumns(ctx, db)
		if err != nil {
			return err
		}
	}
	if err := validateReleaseAdmissionColumnReadiness(columns); err != nil {
		return err
	}
	if err := verifyRetainedReleaseAdmissions(ctx, db); err != nil {
		return err
	}
	if err := ensureReleaseAdmissionConstraints(ctx, db); err != nil {
		return err
	}
	return nil
}

func readReleaseAdmissionColumns(
	ctx context.Context,
	db *sql.DB,
) (map[string]releaseAdmissionColumnState, error) {
	rows, err := db.QueryContext(ctx, `SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_DEFAULT
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_CatalogAppRelease'
AND COLUMN_NAME IN ('AdmissionKind', 'AttestationKeyId', 'AttestationDigest', 'AttestationVerifiedAt')
ORDER BY COLUMN_NAME`)
	if err != nil {
		return nil, fmt.Errorf("inspect Catalog AppRelease admission columns: %w", err)
	}
	defer rows.Close()
	columns := make(map[string]releaseAdmissionColumnState, 4)
	for rows.Next() {
		var name string
		var state releaseAdmissionColumnState
		if err := rows.Scan(&name, &state.columnType, &state.nullable, &state.defaultVal); err != nil {
			return nil, fmt.Errorf("scan Catalog AppRelease admission column: %w", err)
		}
		if _, duplicate := columns[name]; duplicate {
			return nil, fmt.Errorf("%w: duplicate Catalog AppRelease column %q", ErrCorrupt, name)
		}
		columns[name] = state
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate Catalog AppRelease admission columns: %w", err)
	}
	return columns, nil
}

func validateReleaseAdmissionColumnBase(columns map[string]releaseAdmissionColumnState) error {
	expectedTypes := map[string]string{
		"AdmissionKind":         "varchar(32)",
		"AttestationKeyId":      "varchar(128)",
		"AttestationDigest":     "varchar(71)",
		"AttestationVerifiedAt": "datetime(6)",
	}
	if len(columns) != len(expectedTypes) {
		return fmt.Errorf("%w: Catalog AppRelease admission columns are incomplete", ErrCorrupt)
	}
	for name, expectedType := range expectedTypes {
		column, found := columns[name]
		if !found || strings.ToLower(column.columnType) != expectedType {
			return fmt.Errorf("%w: Catalog AppRelease.%s must be %s", ErrCorrupt, name, expectedType)
		}
		if column.nullable != "YES" && column.nullable != "NO" {
			return fmt.Errorf("%w: Catalog AppRelease.%s has invalid nullability", ErrCorrupt, name)
		}
	}
	admission := columns["AdmissionKind"]
	if admission.nullable != "NO" || !admission.defaultVal.Valid ||
		admission.defaultVal.String != string(ReleaseAdmissionSignedPublisher) {
		return fmt.Errorf(
			"%w: Catalog AppRelease.AdmissionKind must be NOT NULL with SIGNED_PUBLISHER default",
			ErrCorrupt,
		)
	}
	for _, name := range []string{"AttestationKeyId", "AttestationDigest", "AttestationVerifiedAt"} {
		if columns[name].defaultVal.Valid {
			return fmt.Errorf("%w: Catalog AppRelease.%s must not have a default", ErrCorrupt, name)
		}
	}
	return nil
}

func validateReleaseAdmissionColumnReadiness(columns map[string]releaseAdmissionColumnState) error {
	if err := validateReleaseAdmissionColumnBase(columns); err != nil {
		return err
	}
	for _, name := range []string{"AttestationKeyId", "AttestationDigest", "AttestationVerifiedAt"} {
		if columns[name].nullable != "YES" {
			return fmt.Errorf("%w: Catalog AppRelease.%s must be nullable after admission migration", ErrCorrupt, name)
		}
	}
	return nil
}

func verifyRetainedReleaseAdmissions(ctx context.Context, db *sql.DB) error {
	rows, err := db.QueryContext(ctx, `SELECT r.AppId, r.AppReleaseDigest, r.AdmissionKind,
r.AttestationKeyId, r.AttestationDigest, r.AttestationVerifiedAt,
(a.AppId IS NOT NULL) AS HasArtifactReceipt, (s.AppId IS NOT NULL) AS HasSourceReceipt
FROM _System_CatalogAppRelease AS r
LEFT JOIN _System_CatalogAppReleaseArtifactReceipt AS a
ON a.AppId = r.AppId AND a.AppReleaseDigest = r.AppReleaseDigest
LEFT JOIN _System_CatalogAppReleaseSourceReceipt AS s
ON s.AppId = r.AppId AND s.AppReleaseDigest = r.AppReleaseDigest
ORDER BY r.AppId, r.AppReleaseDigest`)
	if err != nil {
		return fmt.Errorf("read retained Catalog AppRelease admissions: %w", err)
	}
	for rows.Next() {
		var rawKind string
		var retained retainedReleaseAdmission
		if err := rows.Scan(
			&retained.appID,
			&retained.appReleaseDigest,
			&rawKind,
			&retained.attestationKeyID,
			&retained.attestationDigest,
			&retained.attestationVerifiedAt,
			&retained.hasArtifactReceipt,
			&retained.hasSourceReceipt,
		); err != nil {
			_ = rows.Close()
			return fmt.Errorf("scan retained Catalog AppRelease admission: %w", err)
		}
		retained.kind = ReleaseAdmissionKind(rawKind)
		if err := validateRetainedReleaseAdmission(retained); err != nil {
			_ = rows.Close()
			return err
		}
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return fmt.Errorf("iterate retained Catalog AppRelease admissions: %w", err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("close retained Catalog AppRelease admissions: %w", err)
	}

	artifactRows, err := db.QueryContext(ctx, `SELECT a.AppId, a.AppReleaseDigest, r.AdmissionKind
FROM _System_CatalogAppReleaseArtifactReceipt AS a
LEFT JOIN _System_CatalogAppRelease AS r
ON r.AppId = a.AppId AND r.AppReleaseDigest = a.AppReleaseDigest
ORDER BY a.AppId, a.AppReleaseDigest`)
	if err != nil {
		return fmt.Errorf("read retained Catalog AppRelease artifact receipts: %w", err)
	}
	for artifactRows.Next() {
		var receipt retainedArtifactReleaseReceiptParent
		if err := artifactRows.Scan(
			&receipt.appID,
			&receipt.appReleaseDigest,
			&receipt.releaseAdmissionKind,
		); err != nil {
			_ = artifactRows.Close()
			return fmt.Errorf("scan retained Catalog AppRelease artifact receipt: %w", err)
		}
		if err := validateRetainedArtifactReleaseReceiptParent(receipt); err != nil {
			_ = artifactRows.Close()
			return err
		}
	}
	if err := artifactRows.Err(); err != nil {
		_ = artifactRows.Close()
		return fmt.Errorf("iterate retained Catalog AppRelease artifact receipts: %w", err)
	}
	if err := artifactRows.Close(); err != nil {
		return fmt.Errorf("close retained Catalog AppRelease artifact receipts: %w", err)
	}

	receiptRows, err := db.QueryContext(ctx, `SELECT s.AppId, s.AppReleaseDigest, s.ReceiptSchema,
s.SourceRepository, s.SourceCommit, s.PackageContentDigest, s.PromotionDisposition, s.ReceiptDigest,
r.AdmissionKind
FROM _System_CatalogAppReleaseSourceReceipt AS s
LEFT JOIN _System_CatalogAppRelease AS r
ON r.AppId = s.AppId AND r.AppReleaseDigest = s.AppReleaseDigest
ORDER BY s.AppId, s.AppReleaseDigest`)
	if err != nil {
		return fmt.Errorf("read retained Catalog AppRelease source receipts: %w", err)
	}
	defer receiptRows.Close()
	for receiptRows.Next() {
		var receipt retainedSourceReleaseReceipt
		if err := receiptRows.Scan(
			&receipt.appID,
			&receipt.appReleaseDigest,
			&receipt.receiptSchema,
			&receipt.sourceRepository,
			&receipt.sourceCommit,
			&receipt.packageContentDigest,
			&receipt.promotionDisposition,
			&receipt.receiptDigest,
			&receipt.releaseAdmissionKind,
		); err != nil {
			return fmt.Errorf("scan retained Catalog AppRelease source receipt: %w", err)
		}
		if err := validateRetainedSourceReleaseReceipt(receipt); err != nil {
			return err
		}
	}
	if err := receiptRows.Err(); err != nil {
		return fmt.Errorf("iterate retained Catalog AppRelease source receipts: %w", err)
	}
	return nil
}

func validateRetainedArtifactReleaseReceiptParent(receipt retainedArtifactReleaseReceiptParent) error {
	if !receipt.releaseAdmissionKind.Valid {
		return fmt.Errorf("%w: orphan Catalog AppRelease artifact receipt", ErrCorrupt)
	}
	if ReleaseAdmissionKind(receipt.releaseAdmissionKind.String) != ReleaseAdmissionSignedPublisher {
		return fmt.Errorf("%w: artifact receipt parent is not a signed AppRelease", ErrCorrupt)
	}
	if err := appauthority.ValidateAppID(receipt.appID); err != nil {
		return fmt.Errorf("%w: artifact receipt AppID: %v", ErrCorrupt, err)
	}
	if err := appauthority.ValidateDigest(receipt.appReleaseDigest); err != nil {
		return fmt.Errorf("%w: artifact receipt AppReleaseDigest: %v", ErrCorrupt, err)
	}
	return nil
}

func validateRetainedReleaseAdmission(retained retainedReleaseAdmission) error {
	if err := appauthority.ValidateAppID(retained.appID); err != nil {
		return fmt.Errorf("%w: retained AppRelease AppID: %v", ErrCorrupt, err)
	}
	if err := appauthority.ValidateDigest(retained.appReleaseDigest); err != nil {
		return fmt.Errorf("%w: retained AppRelease digest: %v", ErrCorrupt, err)
	}
	attestationComplete := retained.attestationKeyID.Valid &&
		retained.attestationDigest.Valid && retained.attestationVerifiedAt.Valid
	attestationAbsent := !retained.attestationKeyID.Valid &&
		!retained.attestationDigest.Valid && !retained.attestationVerifiedAt.Valid
	switch retained.kind {
	case ReleaseAdmissionSignedPublisher:
		if !attestationComplete {
			return fmt.Errorf("%w: signed AppRelease has an incomplete attestation tuple", ErrCorrupt)
		}
		if retained.attestationKeyID.String == "" || retained.attestationVerifiedAt.Time.IsZero() {
			return fmt.Errorf("%w: signed AppRelease has an invalid attestation tuple", ErrCorrupt)
		}
		if err := appauthority.ValidateDigest(retained.attestationDigest.String); err != nil {
			return fmt.Errorf("%w: signed AppRelease attestation digest: %v", ErrCorrupt, err)
		}
		if retained.hasSourceReceipt {
			return fmt.Errorf("%w: signed AppRelease has a stopped-source receipt", ErrCorrupt)
		}
		// A signed row may predate artifact-receipt retention. Preserve that
		// exact compatibility: sealed reads still reject it with ErrUnproven.
		return nil
	case ReleaseAdmissionStoppedSource:
		if !attestationAbsent {
			return fmt.Errorf("%w: stopped-source AppRelease has a partial or signed attestation tuple", ErrCorrupt)
		}
		if retained.hasArtifactReceipt || !retained.hasSourceReceipt {
			return fmt.Errorf("%w: stopped-source AppRelease has conflicting or missing provenance receipt", ErrCorrupt)
		}
		return nil
	default:
		return fmt.Errorf("%w: AppRelease has invalid AdmissionKind %q", ErrCorrupt, retained.kind)
	}
}

func validateRetainedSourceReleaseReceipt(receipt retainedSourceReleaseReceipt) error {
	if !receipt.releaseAdmissionKind.Valid {
		return fmt.Errorf("%w: orphan Catalog AppRelease source receipt", ErrCorrupt)
	}
	if ReleaseAdmissionKind(receipt.releaseAdmissionKind.String) != ReleaseAdmissionStoppedSource {
		return fmt.Errorf("%w: source receipt parent is not a stopped-source AppRelease", ErrCorrupt)
	}
	if err := appauthority.ValidateAppID(receipt.appID); err != nil {
		return fmt.Errorf("%w: source receipt AppID: %v", ErrCorrupt, err)
	}
	if err := appauthority.ValidateDigest(receipt.appReleaseDigest); err != nil {
		return fmt.Errorf("%w: source receipt AppReleaseDigest: %v", ErrCorrupt, err)
	}
	if receipt.receiptSchema != releaseSourceReceiptSchema {
		return fmt.Errorf("%w: source receipt has invalid schema", ErrCorrupt)
	}
	if strings.TrimSpace(receipt.sourceRepository) == "" ||
		receipt.sourceRepository != strings.TrimSpace(receipt.sourceRepository) ||
		len(receipt.sourceRepository) > 1024 ||
		strings.IndexFunc(receipt.sourceRepository, func(value rune) bool {
			return unicode.IsControl(value) || unicode.IsSpace(value)
		}) >= 0 {
		return fmt.Errorf("%w: source receipt repository is not canonical", ErrCorrupt)
	}
	if !lowercaseGitCommit.MatchString(receipt.sourceCommit) {
		return fmt.Errorf("%w: source receipt commit must be a lowercase 40-character Git object ID", ErrCorrupt)
	}
	if err := appauthority.ValidateDigest(receipt.packageContentDigest); err != nil {
		return fmt.Errorf("%w: source receipt package content digest: %v", ErrCorrupt, err)
	}
	if receipt.promotionDisposition != releaseSourceNonPromotable {
		return fmt.Errorf("%w: source receipt is not explicitly non-promotable", ErrCorrupt)
	}
	if err := appauthority.ValidateDigest(receipt.receiptDigest); err != nil {
		return fmt.Errorf("%w: source receipt digest: %v", ErrCorrupt, err)
	}
	expected, err := retainedSourceReleaseReceiptDigest(receipt)
	if err != nil {
		return fmt.Errorf("%w: identify source receipt: %v", ErrCorrupt, err)
	}
	if receipt.receiptDigest != expected {
		return fmt.Errorf("%w: source receipt digest differs from its exact tuple", ErrCorrupt)
	}
	return nil
}

func retainedSourceReleaseReceiptDigest(receipt retainedSourceReleaseReceipt) (string, error) {
	encoded, err := json.Marshal(struct {
		Schema               string `json:"schema"`
		AppID                string `json:"app_id"`
		AppReleaseDigest     string `json:"app_release_digest"`
		SourceRepository     string `json:"source_repository"`
		SourceCommit         string `json:"source_commit"`
		PackageContentDigest string `json:"package_content_digest"`
		PromotionDisposition string `json:"promotion_disposition"`
	}{
		Schema:               receipt.receiptSchema,
		AppID:                receipt.appID,
		AppReleaseDigest:     receipt.appReleaseDigest,
		SourceRepository:     receipt.sourceRepository,
		SourceCommit:         receipt.sourceCommit,
		PackageContentDigest: receipt.packageContentDigest,
		PromotionDisposition: receipt.promotionDisposition,
	})
	if err != nil {
		return "", err
	}
	return digestParts(string(encoded)), nil
}

func ensureReleaseAdmissionConstraints(ctx context.Context, db *sql.DB) error {
	type expectedConstraint struct {
		tableName  string
		expression string
		ddl        string
	}
	expected := map[string]expectedConstraint{
		releaseAdmissionConstraint: {
			tableName: "_System_CatalogAppRelease", expression: releaseAdmissionCheckExpression,
			ddl: releaseAdmissionConstraintDDL,
		},
		releaseSourceReceiptConstraint: {
			tableName: "_System_CatalogAppReleaseSourceReceipt", expression: releaseSourceReceiptCheckExpression,
			ddl: releaseSourceReceiptConstraintDDL,
		},
	}
	found, err := readReleaseAdmissionConstraints(ctx, db)
	if err != nil {
		return err
	}
	validateExact := func(
		name string,
		wanted expectedConstraint,
		constraints map[string]releaseAdmissionConstraintState,
	) error {
		actual, present := constraints[name]
		if !present {
			return fmt.Errorf("%w: Catalog AppRelease admission constraint %s is absent", ErrCorrupt, name)
		}
		wantedClause, parseErr := canonicalCatalogCheckClause(wanted.expression)
		if parseErr != nil {
			return fmt.Errorf("internal Catalog AppRelease check contract %s: %w", name, parseErr)
		}
		actualClause, parseErr := canonicalCatalogCheckClause(actual.checkClause)
		if parseErr != nil || actual.tableName != wanted.tableName || actualClause != wantedClause {
			return fmt.Errorf(
				"%w: Catalog AppRelease admission constraint %s has a different CHECK clause",
				ErrCorrupt,
				name,
			)
		}
		return nil
	}
	for name, wanted := range expected {
		_, present := found[name]
		if present {
			if err := validateExact(name, wanted, found); err != nil {
				return err
			}
			continue
		}
		if _, addErr := db.ExecContext(ctx, wanted.ddl); addErr != nil {
			refreshed, inspectErr := readReleaseAdmissionConstraints(ctx, db)
			if inspectErr != nil {
				return fmt.Errorf(
					"add Catalog AppRelease admission constraint %s: %v; inspect concurrent outcome: %w",
					name,
					addErr,
					inspectErr,
				)
			}
			if readyErr := validateExact(name, wanted, refreshed); readyErr != nil {
				return fmt.Errorf(
					"add Catalog AppRelease admission constraint %s: %v; concurrent outcome: %w",
					name,
					addErr,
					readyErr,
				)
			}
			// Another EnsureSchema caller completed the same exact DDL. Continue
			// from the observed schema instead of treating the duplicate add as
			// a migration failure.
			found = refreshed
		}
	}
	found, err = readReleaseAdmissionConstraints(ctx, db)
	if err != nil {
		return err
	}
	for name, wanted := range expected {
		if err := validateExact(name, wanted, found); err != nil {
			return err
		}
	}
	return nil
}

type releaseAdmissionConstraintState struct {
	tableName   string
	checkClause string
}

const tidbReleaseAdmissionConstraintsQuery = `SELECT TABLE_NAME, CONSTRAINT_NAME, CHECK_CLAUSE
FROM INFORMATION_SCHEMA.TIDB_CHECK_CONSTRAINTS
WHERE CONSTRAINT_SCHEMA = DATABASE()
AND ((TABLE_NAME = '_System_CatalogAppRelease' AND CONSTRAINT_NAME = 'chk_catalog_app_release_admission')
OR (TABLE_NAME = '_System_CatalogAppReleaseSourceReceipt' AND CONSTRAINT_NAME = 'chk_catalog_app_release_source_disposition'))
ORDER BY TABLE_NAME, CONSTRAINT_NAME`

const standardReleaseAdmissionConstraintsQuery = `SELECT tc.TABLE_NAME, tc.CONSTRAINT_NAME, cc.CHECK_CLAUSE
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS AS tc
JOIN INFORMATION_SCHEMA.CHECK_CONSTRAINTS AS cc
ON cc.CONSTRAINT_SCHEMA = tc.CONSTRAINT_SCHEMA AND cc.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
WHERE tc.CONSTRAINT_SCHEMA = DATABASE() AND tc.CONSTRAINT_TYPE = 'CHECK'
AND ((tc.TABLE_NAME = '_System_CatalogAppRelease' AND tc.CONSTRAINT_NAME = 'chk_catalog_app_release_admission')
OR (tc.TABLE_NAME = '_System_CatalogAppReleaseSourceReceipt' AND tc.CONSTRAINT_NAME = 'chk_catalog_app_release_source_disposition'))
ORDER BY tc.TABLE_NAME, tc.CONSTRAINT_NAME`

func readReleaseAdmissionConstraints(ctx context.Context, db *sql.DB) (map[string]releaseAdmissionConstraintState, error) {
	// TiDB exposes CHECK constraints through TIDB_CHECK_CONSTRAINTS but does not
	// project them into TABLE_CONSTRAINTS. Prefer that table so an already
	// applied constraint cannot be misclassified as absent and re-added. MySQL
	// does not define the TiDB-specific table, so only its exact missing-table
	// errors select the standard INFORMATION_SCHEMA join.
	rows, err := db.QueryContext(ctx, tidbReleaseAdmissionConstraintsQuery)
	if err == nil {
		return scanReleaseAdmissionConstraints(rows, "TiDB")
	}
	if !isMissingInformationSchemaTable(err) {
		return nil, fmt.Errorf("inspect TiDB Catalog AppRelease admission constraints: %w", err)
	}
	rows, err = db.QueryContext(ctx, standardReleaseAdmissionConstraintsQuery)
	if err != nil {
		return nil, fmt.Errorf("inspect Catalog AppRelease admission constraints: %w", err)
	}
	return scanReleaseAdmissionConstraints(rows, "standard")
}

func isMissingInformationSchemaTable(err error) bool {
	var mysqlError *mysql.MySQLError
	return errors.As(err, &mysqlError) && (mysqlError.Number == 1109 || mysqlError.Number == 1146)
}

func scanReleaseAdmissionConstraints(
	rows *sql.Rows,
	source string,
) (map[string]releaseAdmissionConstraintState, error) {
	defer rows.Close()
	found := make(map[string]releaseAdmissionConstraintState, 2)
	for rows.Next() {
		var tableName, name, checkClause string
		if err := rows.Scan(&tableName, &name, &checkClause); err != nil {
			return nil, fmt.Errorf("scan %s Catalog AppRelease admission constraint: %w", source, err)
		}
		if _, expected := map[string]struct{}{
			releaseAdmissionConstraint: {}, releaseSourceReceiptConstraint: {},
		}[name]; !expected {
			return nil, fmt.Errorf("%w: invalid Catalog AppRelease admission constraint %q", ErrCorrupt, name)
		}
		if _, duplicate := found[name]; duplicate {
			return nil, fmt.Errorf("%w: invalid or duplicate Catalog AppRelease admission constraint %q", ErrCorrupt, name)
		}
		found[name] = releaseAdmissionConstraintState{tableName: tableName, checkClause: checkClause}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate %s Catalog AppRelease admission constraints: %w", source, err)
	}
	return found, nil
}