JgitStorageSchemaMigrationConfig.java

package com.taxonomy.dsl.storage;

import io.github.carstenartur.jgit.storage.hibernate.schema.CoreSchemaMigrations;
import io.github.carstenartur.jgit.storage.hibernate.schema.LegacyCoreSchemaAdoption;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;

import javax.sql.DataSource;

import org.flywaydb.core.Flyway;
import org.flywaydb.core.api.configuration.Configuration;
import org.flywaydb.core.api.configuration.FluentConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.flyway.autoconfigure.FlywayConfigurationCustomizer;
import org.springframework.boot.flyway.autoconfigure.FlywayMigrationStrategy;
import org.springframework.context.annotation.Bean;

/**
 * Runs the versioned JGit Core schema before Hibernate initializes its persistence unit.
 *
 * <p>Fresh installation and history establishment are automatic only after the existing
 * schema has been classified. Adoption of Taxonomy's pre-library schema, including the
 * 0.1.8-to-0.1.9 adoption correction, is fail-closed and requires an explicit operator flag
 * after backup and the released read-only preflight.</p>
 */
@org.springframework.context.annotation.Configuration(proxyBeanMethods = false)
@ConditionalOnProperty(name = "spring.flyway.enabled", havingValue = "true")
public class JgitStorageSchemaMigrationConfig {

    private static final Logger log =
            LoggerFactory.getLogger(JgitStorageSchemaMigrationConfig.class);

    private static final int REQUIRED_PACK_EXTENSION_LENGTH = 32;
    private static final int LEGACY_TAXONOMY_PACK_EXTENSION_LENGTH = 255;
    private static final int REQUIRED_REF_NAME_LENGTH = 1024;
    private static final int LEGACY_TAXONOMY_REF_NAME_LENGTH = 255;
    private static final String PRE_PACK_DESCRIPTION_SCHEMA_VERSION = "0.1.14.2";
    private static final String PRE_PACK_DESCRIPTION_BASELINE_DESCRIPTION =
            "jgit-storage-hibernate-core 0.1.14";
    private static final String PRE_REFLOG_KEY_SCHEMA_VERSION = "0.1.18";
    private static final String PRE_REFLOG_KEY_BASELINE_DESCRIPTION =
            "jgit-storage-hibernate-core 0.1.18";
    private static final String LATEST_CORE_SCHEMA_VERSION = "0.9.1";
    private static final String LATEST_CORE_BASELINE_DESCRIPTION =
            "jgit-storage-hibernate-core 0.9.1";
    private static final String DELIVERY_ID_SCHEMA_VERSION = "0.9.2";
    private static final String DELIVERY_ID_BASELINE_DESCRIPTION =
            "jgit-storage-hibernate-core 0.9.2";

    private static final Set<String> LEGACY_PACK_COLUMNS = Set.of(
            "ID",
            "REPOSITORY_NAME",
            "PACK_NAME",
            "PACK_EXTENSION",
            "DATA",
            "FILE_SIZE",
            "CREATED_AT");

    /** Exact physical pack shape released from 0.1.4 through 0.1.14.1. */
    private static final Set<String> PRE_WRITE_LEASE_PACK_COLUMNS = Set.of(
            "ID",
            "REPOSITORY_NAME",
            "PACK_NAME",
            "PACK_EXTENSION",
            "DATA",
            "FILE_SIZE",
            "COMMITTED",
            "CREATED_AT",
            "COMMITTED_AT");

    /** Exact physical pack shape released from 0.1.14.2 through 0.1.17. */
    private static final Set<String> PRE_PACK_DESCRIPTION_PACK_COLUMNS = Set.of(
            "ID",
            "REPOSITORY_NAME",
            "PACK_NAME",
            "PACK_EXTENSION",
            "DATA",
            "FILE_SIZE",
            "COMMITTED",
            "CREATED_AT",
            "COMMITTED_AT",
            "WRITE_TOKEN",
            "WRITE_LEASE_UNTIL");

    /** Exact physical pack shape released by jgit-storage-hibernate 0.1.18. */
    private static final Set<String> CURRENT_PACK_COLUMNS = Set.of(
            "ID",
            "REPOSITORY_NAME",
            "PACK_NAME",
            "PACK_EXTENSION",
            "DATA",
            "FILE_SIZE",
            "COMMITTED",
            "CREATED_AT",
            "COMMITTED_AT",
            "WRITE_TOKEN",
            "WRITE_LEASE_UNTIL",
            "PACK_SOURCE",
            "LAST_MODIFIED",
            "OBJECT_COUNT",
            "DELTA_COUNT",
            "INDEX_VERSION",
            "MIN_UPDATE_INDEX",
            "MAX_UPDATE_INDEX");

    /** Exact physical reflog shape released through Core 0.1.18. */
    private static final Set<String> PRE_REFLOG_KEY_COLUMNS = Set.of(
            "ID",
            "VERSION",
            "REPOSITORY_NAME",
            "REF_NAME",
            "OLD_ID",
            "NEW_ID",
            "WHO_NAME",
            "WHO_EMAIL",
            "WHO_WHEN",
            "MESSAGE");

    /** Exact physical reflog shape released by Core 0.9.1. */
    private static final Set<String> REFLOG_KEY_COLUMNS = Set.of(
            "ID",
            "VERSION",
            "REPOSITORY_NAME",
            "REF_NAME",
            "REF_NAME_KEY",
            "OLD_ID",
            "NEW_ID",
            "WHO_NAME",
            "WHO_EMAIL",
            "WHO_WHEN",
            "MESSAGE");

    /** Exact physical reflog shape released by Core 0.9.2. */
    private static final Set<String> CURRENT_REFLOG_COLUMNS = Set.of(
            "ID",
            "VERSION",
            "REPOSITORY_NAME",
            "REF_NAME",
            "REF_NAME_KEY",
            "OLD_ID",
            "NEW_ID",
            "WHO_NAME",
            "WHO_EMAIL",
            "WHO_WHEN",
            "MESSAGE",
            "DELIVERY_ID");

    /** Restrict the Boot-managed Flyway instance to the released Core migration stream. */
    @Bean
    public FlywayConfigurationCustomizer jgitStorageFlywayConfigurationCustomizer() {
        return configuration -> {
            DatabaseFamily family = DatabaseFamily.detect(configuration.getDataSource());
            configureNormalMigration(configuration, family);
        };
    }

    /**
     * Classify the existing schema and select the only safe migration path.
     *
     * @param legacyAdoptionEnabled explicit one-time operator opt-in for legacy adoption DDL
     */
    @Bean
    public FlywayMigrationStrategy jgitStorageFlywayMigrationStrategy(
            @Value("${taxonomy.jgit-storage.legacy-adoption:false}")
            boolean legacyAdoptionEnabled) {
        return flyway -> migrateCoreSchema(flyway, legacyAdoptionEnabled);
    }

    static void migrateCoreSchema(Flyway flyway, boolean legacyAdoptionEnabled) {
        Configuration configuration = flyway.getConfiguration();
        DataSource dataSource = configuration.getDataSource();
        DatabaseFamily family = DatabaseFamily.detect(dataSource);
        SchemaSnapshot schema = SchemaSnapshot.inspect(dataSource);

        boolean hasCoreHistory = schema.hasTable(CoreSchemaMigrations.SCHEMA_HISTORY_TABLE);
        boolean hasAdoptionHistory =
                schema.hasTable(CoreSchemaMigrations.LEGACY_ADOPTION_SCHEMA_HISTORY_TABLE);

        if (hasCoreHistory) {
            if (hasAdoptionHistory) {
                migratePendingAdoptionCorrections(
                        configuration,
                        family,
                        dataSource,
                        schema,
                        legacyAdoptionEnabled);
                schema = SchemaSnapshot.inspect(dataSource);
            }

            requireMigratableCoreShape(schema);
            requireManagedReflogHistoryConsistency(flyway, schema);
            log.info("Migrating managed JGit Core schema for {}", family.displayName());
            flyway.migrate();
            SchemaSnapshot migratedSchema = SchemaSnapshot.inspect(dataSource);
            requireCurrentCoreShape(migratedSchema);
            requireManagedReflogHistoryConsistency(flyway, migratedSchema);
            return;
        }

        if (hasAdoptionHistory) {
            throw unsafeSchema(
                    "The legacy-adoption history exists without the normal Core history. "
                            + "Restore the pre-migration backup or complete the documented "
                            + "history-establishment step before startup.");
        }

        boolean hasPacks = schema.hasTable("git_packs");
        boolean hasReflog = schema.hasTable("git_reflog");
        if (!hasPacks && !hasReflog) {
            if (schema.tables().isEmpty()) {
                log.info("Installing JGit Core schema into empty {} database", family.displayName());
                flyway.migrate();
            } else {
                log.info(
                        "Installing JGit Core schema into shared {} schema with pre-migration baseline",
                        family.displayName());
                migrateNormalWithBaseline(
                        configuration,
                        family,
                        CoreSchemaMigrations.PRE_MIGRATION_BASELINE_VERSION,
                        CoreSchemaMigrations.PRE_MIGRATION_BASELINE_DESCRIPTION);
                flyway.migrate();
            }
            requireCurrentCoreShape(SchemaSnapshot.inspect(dataSource));
            return;
        }

        if (hasPacks != hasReflog) {
            throw unsafeSchema(
                    "Only one of git_packs and git_reflog exists. Refusing to repair a partial "
                            + "JGit storage schema automatically.");
        }

        if (schema.packColumns().equals(LEGACY_PACK_COLUMNS)) {
            requirePreReflogKeyColumns(schema);
            migrateLegacyTaxonomySchema(
                    configuration, family, dataSource, schema, legacyAdoptionEnabled);
            requireCurrentCoreShape(SchemaSnapshot.inspect(dataSource));
            return;
        }

        if (schema.packColumns().equals(PRE_WRITE_LEASE_PACK_COLUMNS)) {
            requirePreReflogKeyColumns(schema);
            establishUnversionedPreWriteLeaseSchema(configuration, family, dataSource, schema);
            return;
        }

        if (schema.packColumns().equals(PRE_PACK_DESCRIPTION_PACK_COLUMNS)) {
            requirePreReflogKeyColumns(schema);
            establishUnversionedPrePackDescriptionSchema(
                    configuration, family, dataSource, schema);
            return;
        }

        if (schema.packColumns().equals(CURRENT_PACK_COLUMNS)) {
            establishUnversionedCurrentSchema(configuration, family, dataSource, schema);
            return;
        }

        throw unsupportedPackColumns(schema.packColumns());
    }

    private static void establishUnversionedPreWriteLeaseSchema(
            Configuration configuration,
            DatabaseFamily family,
            DataSource dataSource,
            SchemaSnapshot schema) {
        requireSafePackRows(dataSource, false);
        requireCurrentCoreColumnLengths(schema);
        requirePreWriteLeaseCoreIndexes(schema);
        log.info(
                "Establishing Flyway history for an exact pre-0.1.14.2 JGit Core schema on {}",
                family.displayName());
        migrateNormalWithBaseline(
                configuration,
                family,
                CoreSchemaMigrations.CURRENT_SCHEMA_VERSION,
                "verified pre-0.1.14.2 JGit Core schema");
        requireCurrentCoreShape(SchemaSnapshot.inspect(dataSource));
    }

    private static void establishUnversionedPrePackDescriptionSchema(
            Configuration configuration,
            DatabaseFamily family,
            DataSource dataSource,
            SchemaSnapshot schema) {
        requireSafePackRows(dataSource, false);
        requireCurrentCoreColumnLengths(schema);
        requireRepositoryLockTable(schema);
        requireMigratableCurrentCoreIndexes(schema);
        log.info(
                "Establishing Flyway history for an exact unversioned 0.1.14.2-0.1.17 schema on {}",
                family.displayName());
        migrateNormalWithBaseline(
                configuration,
                family,
                PRE_PACK_DESCRIPTION_SCHEMA_VERSION,
                PRE_PACK_DESCRIPTION_BASELINE_DESCRIPTION);
        requireCurrentCoreShape(SchemaSnapshot.inspect(dataSource));
    }

    private static void establishUnversionedCurrentSchema(
            Configuration configuration,
            DatabaseFamily family,
            DataSource dataSource,
            SchemaSnapshot schema) {
        requireSafePackRows(dataSource, false);
        requireCurrentCoreColumnLengths(schema);
        requireCurrentCoreTables(schema);
        requireSupportedReflogColumns(schema);
        requireCurrentCoreIndexes(schema);
        String baselineVersion;
        String baselineDescription;
        if (hasReflogDeliveryId(schema)) {
            baselineVersion = DELIVERY_ID_SCHEMA_VERSION;
            baselineDescription = DELIVERY_ID_BASELINE_DESCRIPTION;
        } else if (hasReflogReferenceKey(schema)) {
            baselineVersion = LATEST_CORE_SCHEMA_VERSION;
            baselineDescription = LATEST_CORE_BASELINE_DESCRIPTION;
        } else {
            baselineVersion = PRE_REFLOG_KEY_SCHEMA_VERSION;
            baselineDescription = PRE_REFLOG_KEY_BASELINE_DESCRIPTION;
        }
        log.info(
                "Establishing Flyway history for an exact unversioned {} schema on {}",
                baselineVersion,
                family.displayName());
        migrateNormalWithBaseline(
                configuration,
                family,
                baselineVersion,
                baselineDescription);
        requireCurrentCoreShape(SchemaSnapshot.inspect(dataSource));
    }

    private static void migrateLegacyTaxonomySchema(
            Configuration configuration,
            DatabaseFamily family,
            DataSource dataSource,
            SchemaSnapshot schema,
            boolean legacyAdoptionEnabled) {
        requireLegacyCoreIndexes(schema);
        requireLegacyColumnLengths(schema);

        if (!legacyAdoptionEnabled) {
            throw unsafeSchema(
                    "Detected the pre-library Taxonomy JGit schema. Stop all writers, take a "
                            + "restorable backup, verify pack BLOB checksums and reflog rows, then "
                            + "restart once with TAXONOMY_JGIT_STORAGE_LEGACY_ADOPTION=true.");
        }

        LegacyCoreSchemaAdoption.LegacySchemaReport report = requireSafePackRows(dataSource, true);
        log.warn(
                "Adopting {} pre-library JGit pack rows on {} through the released migration stream",
                report.packRows(),
                family.displayName());

        migrateAdoptionWithBaseline(configuration, family);
        migrateNormalWithBaseline(
                configuration,
                family,
                CoreSchemaMigrations.CURRENT_SCHEMA_VERSION,
                "adopted pre-library core schema");
    }

    private static void migratePendingAdoptionCorrections(
            Configuration configuration,
            DatabaseFamily family,
            DataSource dataSource,
            SchemaSnapshot schema,
            boolean legacyAdoptionEnabled) {
        requireMigratableCoreContract(schema);

        boolean currentLengths = hasCurrentCoreColumnLengths(schema);
        if (!currentLengths) {
            requireLegacyColumnLengths(schema);
            if (!legacyAdoptionEnabled) {
                throw unsafeSchema(
                        "Detected an adoption history that still has the 0.1.8 Taxonomy column "
                                + "lengths. Stop all writers, take a new backup, verify pack BLOB "
                                + "checksums and reflog rows, then restart once with "
                                + "TAXONOMY_JGIT_STORAGE_LEGACY_ADOPTION=true so released adoption "
                                + "migration V2 can run.");
            }
        }

        LegacyCoreSchemaAdoption.LegacySchemaReport report = requireSafePackRows(dataSource, false);
        if (!currentLengths) {
            log.warn(
                    "Applying released adoption correction through version {} to {} existing pack rows on {}",
                    CoreSchemaMigrations.LEGACY_ADOPTION_VERSION,
                    report.packRows(),
                    family.displayName());
        }

        migrateExistingAdoptionStream(configuration, family);
        requireMigratableCoreShape(SchemaSnapshot.inspect(dataSource));
    }

    private static LegacyCoreSchemaAdoption.LegacySchemaReport requireSafePackRows(
            DataSource dataSource, boolean requireAdoption) {
        try (Connection connection = dataSource.getConnection()) {
            LegacyCoreSchemaAdoption.LegacySchemaReport report =
                    LegacyCoreSchemaAdoption.requireSafeToAdopt(connection);
            if (report.requiresAdoption() != requireAdoption) {
                throw unsafeSchema(requireAdoption
                        ? "The schema changed while the legacy-adoption preflight was running."
                        : "The existing adopted Core schema unexpectedly requires adoption V1.");
            }
            return report;
        } catch (SQLException exception) {
            throw new IllegalStateException(
                    "Could not open the database for JGit Core schema preflight", exception);
        }
    }

    private static void migrateAdoptionWithBaseline(
            Configuration source, DatabaseFamily family) {
        newMigrationConfiguration(source)
                .locations(family.legacyAdoptionLocation())
                .table(CoreSchemaMigrations.LEGACY_ADOPTION_SCHEMA_HISTORY_TABLE)
                .baselineOnMigrate(true)
                .baselineVersion(CoreSchemaMigrations.PRE_MIGRATION_BASELINE_VERSION)
                .baselineDescription("before pre-library core adoption")
                .load()
                .migrate();
    }

    private static void migrateExistingAdoptionStream(
            Configuration source, DatabaseFamily family) {
        newMigrationConfiguration(source)
                .locations(family.legacyAdoptionLocation())
                .table(CoreSchemaMigrations.LEGACY_ADOPTION_SCHEMA_HISTORY_TABLE)
                .load()
                .migrate();
    }

    private static void migrateNormalWithBaseline(
            Configuration source,
            DatabaseFamily family,
            String baselineVersion,
            String baselineDescription) {
        newMigrationConfiguration(source)
                .locations(family.normalLocation())
                .table(CoreSchemaMigrations.SCHEMA_HISTORY_TABLE)
                .baselineOnMigrate(true)
                .baselineVersion(baselineVersion)
                .baselineDescription(baselineDescription)
                .load()
                .migrate();
    }

    private static FluentConfiguration newMigrationConfiguration(Configuration source) {
        return new FluentConfiguration(source.getClassLoader())
                .dataSource(source.getDataSource());
    }

    private static void configureNormalMigration(
            FluentConfiguration configuration, DatabaseFamily family) {
        configuration
                .locations(family.normalLocation())
                .table(CoreSchemaMigrations.SCHEMA_HISTORY_TABLE);
    }

    private static void requireManagedReflogHistoryConsistency(
            Flyway flyway, SchemaSnapshot schema) {
        boolean deliveryIdMigrationApplied =
                isCoreMigrationApplied(flyway, DELIVERY_ID_SCHEMA_VERSION);
        boolean referenceKeyMigrationApplied =
                isCoreMigrationApplied(flyway, LATEST_CORE_SCHEMA_VERSION)
                        || deliveryIdMigrationApplied;
        requireManagedReflogHistoryConsistency(
                schema,
                LATEST_CORE_SCHEMA_VERSION + " or later",
                "REF_NAME_KEY",
                referenceKeyMigrationApplied,
                hasReflogReferenceKey(schema));
        requireManagedReflogHistoryConsistency(
                schema,
                DELIVERY_ID_SCHEMA_VERSION,
                "DELIVERY_ID",
                deliveryIdMigrationApplied,
                hasReflogDeliveryId(schema));
    }

    private static void requireManagedReflogHistoryConsistency(
            SchemaSnapshot schema,
            String migrationVersion,
            String column,
            boolean migrationApplied,
            boolean columnPresent) {
        if (migrationApplied == columnPresent) {
            return;
        }
        throw unsafeSchema(
                "Core migration " + migrationVersion
                        + " and git_reflog." + column + " must appear together; "
                        + "migration applied=" + migrationApplied
                        + ", reflog columns=" + schema.reflogColumns()
                        + ", reflog indexes=" + schema.reflogIndexes());
    }

    private static boolean isCoreMigrationApplied(Flyway flyway, String version) {
        for (var migration : flyway.info().applied()) {
            if (migration.getVersion() != null
                    && version.equals(migration.getVersion().toString())) {
                return true;
            }
        }
        return false;
    }

    /** Accept only exact released shapes that Flyway can safely advance to the pinned release. */
    private static void requireMigratableCoreShape(SchemaSnapshot schema) {
        requireMigratableCoreContract(schema);
        requireCurrentCoreColumnLengths(schema);
    }

    private static void requireMigratableCoreContract(SchemaSnapshot schema) {
        requireCoreTables(schema);
        requireSupportedReflogColumns(schema);
        if (schema.packColumns().equals(PRE_WRITE_LEASE_PACK_COLUMNS)) {
            requirePreWriteLeaseCoreIndexes(schema);
            return;
        }
        if (schema.packColumns().equals(PRE_PACK_DESCRIPTION_PACK_COLUMNS)) {
            requireRepositoryLockTable(schema);
            requireMigratableCurrentCoreIndexes(schema);
            return;
        }
        if (schema.packColumns().equals(CURRENT_PACK_COLUMNS)) {
            requireCurrentCoreTables(schema);
            requireCurrentCoreIndexes(schema);
            return;
        }
        throw unsupportedPackColumns(schema.packColumns());
    }

    private static void requireCurrentCoreShape(SchemaSnapshot schema) {
        requireCoreTables(schema);
        requireCurrentCoreTables(schema);
        requireExactColumns("git_packs", schema.packColumns(), CURRENT_PACK_COLUMNS);
        requireSupportedReflogColumns(schema);
        requireCurrentCoreColumnLengths(schema);
        requireCurrentCoreIndexes(schema);
    }

    private static void requireCoreTables(SchemaSnapshot schema) {
        if (!schema.hasTable("git_packs") || !schema.hasTable("git_reflog")) {
            throw unsafeSchema(
                    "The Core Flyway history exists but one or both owned tables are missing.");
        }
    }

    private static void requireRepositoryLockTable(SchemaSnapshot schema) {
        if (!schema.hasTable("git_repository_lock")) {
            throw unsafeSchema(
                    "The released Core schema requires the git_repository_lock table.");
        }
    }

    private static void requireCurrentCoreTables(SchemaSnapshot schema) {
        requireRepositoryLockTable(schema);
        if (!schema.hasTable("git_repository_lifecycle")) {
            throw unsafeSchema(
                    "The released Core schema requires the git_repository_lifecycle table.");
        }
    }

    private static boolean hasCurrentCoreColumnLengths(SchemaSnapshot schema) {
        return schema.packExtensionLength() == REQUIRED_PACK_EXTENSION_LENGTH
                && schema.reflogRefNameLength() >= REQUIRED_REF_NAME_LENGTH;
    }

    private static void requireCurrentCoreColumnLengths(SchemaSnapshot schema) {
        if (!hasCurrentCoreColumnLengths(schema)) {
            throw unsafeSchema(
                    "The released Core contract requires git_packs.pack_extension length "
                            + REQUIRED_PACK_EXTENSION_LENGTH
                            + " and git_reflog.ref_name capacity at least "
                            + REQUIRED_REF_NAME_LENGTH
                            + "; actual lengths are "
                            + schema.packExtensionLength()
                            + " and "
                            + schema.reflogRefNameLength()
                            + ".");
        }
    }

    private static void requireLegacyColumnLengths(SchemaSnapshot schema) {
        if (schema.packExtensionLength() != LEGACY_TAXONOMY_PACK_EXTENSION_LENGTH
                || schema.reflogRefNameLength() != LEGACY_TAXONOMY_REF_NAME_LENGTH) {
            throw unsafeSchema(
                    "Legacy Taxonomy adoption accepts only the exact pre-library column lengths "
                            + LEGACY_TAXONOMY_PACK_EXTENSION_LENGTH
                            + " for git_packs.pack_extension and "
                            + LEGACY_TAXONOMY_REF_NAME_LENGTH
                            + " for git_reflog.ref_name; actual lengths are "
                            + schema.packExtensionLength()
                            + " and "
                            + schema.reflogRefNameLength()
                            + ".");
        }
    }

    private static void requireLegacyCoreIndexes(SchemaSnapshot schema) {
        requireIndex(
                "git_packs",
                schema.packIndexes(),
                false,
                List.of("REPOSITORY_NAME"));
        requireIndex(
                "git_packs",
                schema.packIndexes(),
                false,
                List.of("REPOSITORY_NAME", "PACK_NAME"));
        requireIndex(
                "git_reflog",
                schema.reflogIndexes(),
                false,
                List.of("REPOSITORY_NAME"));
        requireIndex(
                "git_reflog",
                schema.reflogIndexes(),
                false,
                List.of("REPOSITORY_NAME", "REF_NAME"));
    }

    /** Validate the pack lookup paths shared by every managed post-0.1.5 schema. */
    private static void requireManagedPackIndexes(SchemaSnapshot schema) {
        requireIndexPrefix(
                "git_packs",
                schema.packIndexes(),
                List.of("REPOSITORY_NAME", "PACK_NAME"));
        requireIndex(
                "git_packs",
                schema.packIndexes(),
                true,
                List.of("REPOSITORY_NAME", "PACK_NAME", "PACK_EXTENSION"));
        requireIndex(
                "git_packs",
                schema.packIndexes(),
                false,
                List.of("REPOSITORY_NAME", "COMMITTED"));
    }

    /**
     * Accept released historical reflog indexes before Flyway advances the schema.
     * A schema already carrying the 0.9.1 reference key must already carry its final index.
     */
    private static void requireMigratableReflogIndex(SchemaSnapshot schema) {
        if (hasReflogReferenceKey(schema)) {
            requireIndexPrefix(
                    "git_reflog",
                    schema.reflogIndexes(),
                    List.of("REPOSITORY_NAME", "REF_NAME_KEY", "ID"));
            return;
        }
        requireIndexPrefix(
                "git_reflog",
                schema.reflogIndexes(),
                List.of("REPOSITORY_NAME", "REF_NAME"));
    }

    /** Require the lookup path produced by the released migration stream. */
    private static void requireCurrentReflogIndex(SchemaSnapshot schema) {
        if (hasReflogReferenceKey(schema)) {
            requireIndexPrefix(
                    "git_reflog",
                    schema.reflogIndexes(),
                    List.of("REPOSITORY_NAME", "REF_NAME_KEY", "ID"));
            return;
        }
        requireIndexPrefix(
                "git_reflog",
                schema.reflogIndexes(),
                List.of("REPOSITORY_NAME", "REF_NAME", "ID"));
    }

    private static void requirePreWriteLeaseCoreIndexes(SchemaSnapshot schema) {
        requireManagedPackIndexes(schema);
        requireMigratableReflogIndex(schema);
    }

    private static void requireMigratableCurrentCoreIndexes(SchemaSnapshot schema) {
        requirePreWriteLeaseCoreIndexes(schema);
        requireIndex(
                "git_packs",
                schema.packIndexes(),
                false,
                List.of("REPOSITORY_NAME", "COMMITTED", "WRITE_LEASE_UNTIL"));
    }

    private static void requireCurrentCoreIndexes(SchemaSnapshot schema) {
        requireManagedPackIndexes(schema);
        requireCurrentReflogIndex(schema);
        requireIndex(
                "git_packs",
                schema.packIndexes(),
                false,
                List.of("REPOSITORY_NAME", "COMMITTED", "WRITE_LEASE_UNTIL"));
    }

    private static void requireIndexPrefix(
            String table,
            Map<String, IndexSignature> indexes,
            List<String> columns) {
        boolean covered = indexes.values().stream()
                .map(IndexSignature::columns)
                .anyMatch(indexColumns -> indexColumns.size() >= columns.size()
                        && indexColumns.subList(0, columns.size()).equals(columns));
        if (covered) {
            return;
        }
        throw unsafeSchema(
                table + " is missing an index with leading columns " + columns
                        + "; actual indexes=" + indexes);
    }

    private static void requireIndex(
            String table,
            Map<String, IndexSignature> indexes,
            boolean unique,
            List<String> columns) {
        IndexSignature required = new IndexSignature(unique, columns);
        if (indexes.containsValue(required)) {
            return;
        }
        throw unsafeSchema(
                table + " is missing the required "
                        + (unique ? "unique " : "")
                        + "index on " + columns + "; actual indexes=" + indexes);
    }

    private static boolean hasReflogReferenceKey(SchemaSnapshot schema) {
        return schema.reflogColumns().equals(REFLOG_KEY_COLUMNS)
                || schema.reflogColumns().equals(CURRENT_REFLOG_COLUMNS);
    }

    private static boolean hasReflogDeliveryId(SchemaSnapshot schema) {
        return schema.reflogColumns().equals(CURRENT_REFLOG_COLUMNS);
    }

    private static void requirePreReflogKeyColumns(SchemaSnapshot schema) {
        requireExactColumns(
                "git_reflog", schema.reflogColumns(), PRE_REFLOG_KEY_COLUMNS);
    }

    private static void requireSupportedReflogColumns(SchemaSnapshot schema) {
        if (schema.reflogColumns().equals(PRE_REFLOG_KEY_COLUMNS)) {
            return;
        }
        if (schema.reflogColumns().equals(REFLOG_KEY_COLUMNS)
                || schema.reflogColumns().equals(CURRENT_REFLOG_COLUMNS)) {
            if (!schema.packColumns().equals(CURRENT_PACK_COLUMNS)) {
                throw unsafeSchema(
                        "git_reflog carries a released REF_NAME_KEY shape, but git_packs "
                                + "does not match the current Core pack schema; expected "
                                + "git_packs columns=" + CURRENT_PACK_COLUMNS
                                + ", actual git_packs columns=" + schema.packColumns());
            }
            return;
        }
        throw unsafeSchema(
                "git_reflog is neither the exact pre-0.9.1, 0.9.1 nor 0.9.2 shape; "
                        + "actual=" + schema.reflogColumns());
    }

    private static void requireExactColumns(
            String table, Set<String> actual, Set<String> expected) {
        if (!actual.equals(expected)) {
            Set<String> missing = new LinkedHashSet<>(expected);
            missing.removeAll(actual);
            Set<String> unexpected = new LinkedHashSet<>(actual);
            unexpected.removeAll(expected);
            throw unsafeSchema(
                    table + " does not match the supported schema; missing=" + missing
                            + ", unexpected=" + unexpected);
        }
    }

    private static IllegalStateException unsupportedPackColumns(Set<String> actual) {
        Set<String> missing = new LinkedHashSet<>(CURRENT_PACK_COLUMNS);
        missing.removeAll(actual);
        Set<String> unexpected = new LinkedHashSet<>(actual);
        unexpected.removeAll(CURRENT_PACK_COLUMNS);
        return unsafeSchema(
                "git_packs is neither the exact pre-library Taxonomy shape nor an exact released "
                        + "Core shape; missing from current=" + missing
                        + ", unexpected=" + unexpected
                        + ", actual=" + actual);
    }

    private static IllegalStateException unsafeSchema(String message) {
        return new IllegalStateException("Unsafe JGit Core schema state: " + message);
    }

    private record IndexSignature(boolean unique, List<String> columns) {

        private IndexSignature {
            columns = List.copyOf(columns);
        }
    }

    enum DatabaseFamily {
        HSQLDB(
                "HSQLDB",
                CoreSchemaMigrations.HSQLDB_LOCATION,
                CoreSchemaMigrations.HSQLDB_LEGACY_ADOPTION_LOCATION),
        POSTGRESQL(
                "PostgreSQL",
                CoreSchemaMigrations.POSTGRESQL_LOCATION,
                CoreSchemaMigrations.POSTGRESQL_LEGACY_ADOPTION_LOCATION);

        private final String displayName;
        private final String normalLocation;
        private final String legacyAdoptionLocation;

        DatabaseFamily(
                String displayName,
                String normalLocation,
                String legacyAdoptionLocation) {
            this.displayName = displayName;
            this.normalLocation = normalLocation;
            this.legacyAdoptionLocation = legacyAdoptionLocation;
        }

        String displayName() {
            return displayName;
        }

        String normalLocation() {
            return normalLocation;
        }

        String legacyAdoptionLocation() {
            return legacyAdoptionLocation;
        }

        static DatabaseFamily detect(DataSource dataSource) {
            try (Connection connection = dataSource.getConnection()) {
                String product = connection.getMetaData().getDatabaseProductName()
                        .toLowerCase(Locale.ROOT);
                if (product.contains("hsql")) {
                    return HSQLDB;
                }
                if (product.contains("postgresql")) {
                    return POSTGRESQL;
                }
                throw new IllegalStateException(
                        "JGit Core Flyway migrations are supported only for HSQLDB and PostgreSQL; "
                                + "detected " + connection.getMetaData().getDatabaseProductName());
            } catch (SQLException exception) {
                throw new IllegalStateException(
                        "Could not identify the database for JGit Core migrations", exception);
            }
        }
    }

    record SchemaSnapshot(
            Map<String, String> tables,
            Set<String> packColumns,
            Set<String> reflogColumns,
            int packExtensionLength,
            int reflogRefNameLength,
            Map<String, IndexSignature> packIndexes,
            Map<String, IndexSignature> reflogIndexes) {

        boolean hasTable(String table) {
            return tables.containsKey(normalize(table));
        }

        static SchemaSnapshot inspect(DataSource dataSource) {
            try (Connection connection = dataSource.getConnection()) {
                String schema = connection.getSchema();
                DatabaseMetaData metadata = connection.getMetaData();
                Map<String, String> tables = readTables(metadata, schema);
                String packTable = tables.get("GIT_PACKS");
                String reflogTable = tables.get("GIT_REFLOG");
                Set<String> packColumns = readColumns(metadata, schema, packTable);
                Set<String> reflogColumns = readColumns(metadata, schema, reflogTable);
                int packExtensionLength = readColumnSize(
                        metadata, schema, packTable, "PACK_EXTENSION");
                int refNameLength = readColumnSize(
                        metadata, schema, reflogTable, "REF_NAME");
                Map<String, IndexSignature> packIndexes =
                        readIndexes(metadata, schema, packTable);
                Map<String, IndexSignature> reflogIndexes =
                        readIndexes(metadata, schema, reflogTable);
                return new SchemaSnapshot(
                        Collections.unmodifiableMap(tables),
                        Collections.unmodifiableSet(packColumns),
                        Collections.unmodifiableSet(reflogColumns),
                        packExtensionLength,
                        refNameLength,
                        Collections.unmodifiableMap(packIndexes),
                        Collections.unmodifiableMap(reflogIndexes));
            } catch (SQLException exception) {
                throw new IllegalStateException(
                        "Could not inspect the existing JGit Core schema", exception);
            }
        }

        private static Map<String, String> readTables(
                DatabaseMetaData metadata, String schema) throws SQLException {
            Map<String, String> tables = new LinkedHashMap<>();
            try (ResultSet resultSet =
                         metadata.getTables(null, schema, "%", new String[] {"TABLE"})) {
                while (resultSet.next()) {
                    String table = resultSet.getString("TABLE_NAME");
                    tables.put(normalize(table), table);
                }
            }
            return tables;
        }

        private static Set<String> readColumns(
                DatabaseMetaData metadata, String schema, String table) throws SQLException {
            if (table == null) {
                return new LinkedHashSet<>();
            }
            Set<String> columns = new LinkedHashSet<>();
            try (ResultSet resultSet = metadata.getColumns(null, schema, table, "%")) {
                while (resultSet.next()) {
                    columns.add(normalize(resultSet.getString("COLUMN_NAME")));
                }
            }
            return columns;
        }

        private static int readColumnSize(
                DatabaseMetaData metadata, String schema, String table, String column)
                throws SQLException {
            if (table == null) {
                return -1;
            }
            try (ResultSet resultSet = metadata.getColumns(null, schema, table, "%")) {
                while (resultSet.next()) {
                    if (normalize(resultSet.getString("COLUMN_NAME")).equals(column)) {
                        return resultSet.getInt("COLUMN_SIZE");
                    }
                }
            }
            return -1;
        }

        private static Map<String, IndexSignature> readIndexes(
                DatabaseMetaData metadata, String schema, String table) throws SQLException {
            if (table == null) {
                return new LinkedHashMap<>();
            }

            Map<String, Boolean> uniqueness = new LinkedHashMap<>();
            Map<String, Map<Integer, String>> columnsByIndex = new LinkedHashMap<>();
            try (ResultSet resultSet =
                         metadata.getIndexInfo(null, schema, table, false, false)) {
                while (resultSet.next()) {
                    String indexName = resultSet.getString("INDEX_NAME");
                    String columnName = resultSet.getString("COLUMN_NAME");
                    int position = resultSet.getInt("ORDINAL_POSITION");
                    if (indexName == null || columnName == null || position <= 0) {
                        continue;
                    }
                    String normalizedName = normalize(indexName);
                    uniqueness.putIfAbsent(
                            normalizedName, !resultSet.getBoolean("NON_UNIQUE"));
                    columnsByIndex
                            .computeIfAbsent(normalizedName, ignored -> new TreeMap<>())
                            .put(position, normalize(columnName));
                }
            }

            Map<String, IndexSignature> indexes = new LinkedHashMap<>();
            for (Map.Entry<String, Map<Integer, String>> entry :
                    columnsByIndex.entrySet()) {
                List<String> columns = new ArrayList<>(entry.getValue().values());
                indexes.put(
                        entry.getKey(),
                        new IndexSignature(
                                Boolean.TRUE.equals(uniqueness.get(entry.getKey())),
                                columns));
            }
            return indexes;
        }

        private static String normalize(String identifier) {
            return JgitStorageSchemaMigrationConfig.normalize(identifier);
        }
    }

    private static String normalize(String identifier) {
        return identifier.toUpperCase(Locale.ROOT);
    }
}