package gudusoft.gsqlparser.catalog.input;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.catalog.diagnostic.CatalogDiagnostic;
import gudusoft.gsqlparser.catalog.diagnostic.CatalogDiagnosticCode;
import gudusoft.gsqlparser.catalog.diagnostic.CatalogDiagnosticSeverity;
import gudusoft.gsqlparser.catalog.input.model.CatalogModel;
import gudusoft.gsqlparser.catalog.input.model.ColumnModel;
import gudusoft.gsqlparser.catalog.input.model.ConstraintModel;
import gudusoft.gsqlparser.catalog.input.model.IdentifierConfig;
import gudusoft.gsqlparser.catalog.input.model.IndexModel;
import gudusoft.gsqlparser.catalog.input.model.RoutineModel;
import gudusoft.gsqlparser.catalog.input.model.SchemaModel;
import gudusoft.gsqlparser.catalog.input.model.SequenceModel;
import gudusoft.gsqlparser.catalog.input.model.SynonymModel;
import gudusoft.gsqlparser.catalog.input.model.TableModel;
import gudusoft.gsqlparser.catalog.input.model.UnifiedCatalogModel;
import gudusoft.gsqlparser.catalog.input.model.ViewModel;
import gudusoft.gsqlparser.catalog.runtime.CatalogIdentifierPolicy;
import gudusoft.gsqlparser.catalog.runtime.CatalogObjectKind;
import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
import gudusoft.gsqlparser.sqlenv.IdentifierService;
import gudusoft.gsqlparser.sqlenv.TSQLEnv;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * Validates a {@link UnifiedCatalogModel} against required-field rules, duplicate-name
 * checks, identifier-bypass detection, and table↔column referential integrity.
 *
 * <p>Plan §7.1 / §9.6. Identifier-bypass detection rejects models whose names disagree
 * with {@code IdentifierService.normalize(...)} — catching adapters that sneak in a
 * third-party canonicalization step. Strict-mode escalation is the caller's
 * responsibility (see {@code CatalogLoadOptions.strict()} and §15).</p>
 *
 * <p>Validation is best-effort: it never throws; every problem is reported as a
 * {@link CatalogDiagnostic} on the returned {@link CatalogValidationResult}. ERROR
 * diagnostics fail validation; WARN/INFO do not.</p>
 *
 * <p>Duplicate detection uses {@link IdentifierService#areEqual} rather than a
 * normalize-then-{@code Set} short-circuit: in dialects where the compare rule is
 * case-insensitive but the canonical normalize preserves the input spelling
 * (e.g., BigQuery columns, MySQL columns under {@code lower_case_table_names=2},
 * MSSQL with a case-insensitive collation), the keyset alone misses real
 * duplicates. The cost is O(n²) per scope, which is acceptable since each scope
 * is a single schema or table column list.</p>
 */
public final class CatalogModelValidator {

    public CatalogModelValidator() {
        // No state.
    }

    public CatalogValidationResult validate(UnifiedCatalogModel model, CatalogLoadOptions options) {
        if (model == null) {
            throw new IllegalArgumentException("CatalogModelValidator.validate: model is required");
        }
        List<CatalogDiagnostic> diagnostics = new ArrayList<CatalogDiagnostic>();
        EDbVendor vendor = model.vendor();
        IdentifierConfig cfg = effectiveConfig(model, options, diagnostics);
        IdentifierService service = CatalogIdentifierPolicy.identifierServiceFor(cfg, vendor);
        boolean normalizeOnLoad = options == null || options.normalizeOnLoad();

        IdentifierBucket catalogBucket = new IdentifierBucket(service);
        for (CatalogModel c : model.catalogs()) {
            validateCatalog(c, vendor, service, normalizeOnLoad, catalogBucket, diagnostics);
        }
        return CatalogValidationResult.of(diagnostics);
    }

    // ---------- catalog / schema / table / column traversal ----------

    private void validateCatalog(CatalogModel c,
                                 EDbVendor vendor,
                                 IdentifierService service,
                                 boolean normalizeOnLoad,
                                 IdentifierBucket catalogBucket,
                                 List<CatalogDiagnostic> diagnostics) {
        String name = c.name();
        checkRequired("catalog", name, diagnostics);
        checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotCatalog,
            "catalog '" + name + "'", diagnostics);
        flagIfDuplicate(catalogBucket, name, ESQLDataObjectType.dotCatalog, CatalogObjectKind.CATALOG,
            "catalog '" + name + "'", diagnostics);
        // Build a catalog-wide table index so FK constraints can look up their target
        // table across schemas. Keys are normalized via IdentifierService so the lookup
        // honors per-vendor identifier folding (Oracle uppercase, PG lowercase, etc.).
        CatalogTableIndex tableIndex = CatalogTableIndex.build(c, service);
        IdentifierBucket schemaBucket = new IdentifierBucket(service);
        for (SchemaModel s : c.schemas()) {
            validateSchema(s, vendor, service, normalizeOnLoad, c.name(), schemaBucket,
                tableIndex, diagnostics);
        }
    }

    private void validateSchema(SchemaModel s,
                                EDbVendor vendor,
                                IdentifierService service,
                                boolean normalizeOnLoad,
                                String catalogName,
                                IdentifierBucket schemaBucket,
                                CatalogTableIndex tableIndex,
                                List<CatalogDiagnostic> diagnostics) {
        String name = s.name();
        if (name == null) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_MISSING_DEFAULT,
                "schema in catalog '" + catalogName + "' has null name"));
            return;
        }
        if (name.isEmpty()) {
            // Empty schema names are only legal for dialects without a schema layer
            // (per TSQLEnv.supportSchema): MySQL, Teradata, Hive, Impala. For every
            // other dialect — Oracle, PostgreSQL, MSSQL, BigQuery, etc. — an empty
            // schema name silently re-qualifies child objects as catalog.object,
            // which is a manifest bug worth surfacing.
            if (TSQLEnv.supportSchema(vendor)) {
                diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_MISSING_DEFAULT,
                    "schema in catalog '" + catalogName
                        + "' has empty name; vendor " + vendor + " requires a schema layer"));
            }
        } else {
            checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotSchema,
                "schema '" + catalogName + "." + name + "'", diagnostics);
        }
        flagIfDuplicate(schemaBucket, name, ESQLDataObjectType.dotSchema, CatalogObjectKind.SCHEMA,
            "schema '" + catalogName + "." + name + "'", diagnostics);

        // The existing catalog storage (TSQLCatalog) keeps table / view / routine /
        // synonym / sequence indexes type-specific, so a schema may legitimately hold
        // a table named 'Foo' and a function named 'Foo' that resolve by requested
        // type. We honor that by using one bucket per object-type namespace. Tables
        // and views share a bucket because in every dialect we cover (Oracle, MSSQL,
        // PostgreSQL, MySQL, BigQuery) a CREATE TABLE and a CREATE VIEW with the same
        // unqualified name in the same schema is a real collision.
        IdentifierBucket tableViewBucket = new IdentifierBucket(service);
        for (TableModel t : s.tables()) {
            validateTable(t, vendor, service, normalizeOnLoad, catalogName, name,
                tableViewBucket, tableIndex, diagnostics);
        }
        for (ViewModel v : s.views()) {
            validateView(v, vendor, service, normalizeOnLoad, catalogName, name,
                tableViewBucket, diagnostics);
        }
        IdentifierBucket routineBucket = new IdentifierBucket(service);
        for (RoutineModel r : s.routines()) {
            validateRoutine(r, vendor, service, normalizeOnLoad, catalogName, name,
                routineBucket, diagnostics);
        }
        IdentifierBucket synonymBucket = new IdentifierBucket(service);
        for (SynonymModel sy : s.synonyms()) {
            validateSynonym(sy, vendor, service, normalizeOnLoad, catalogName, name,
                synonymBucket, diagnostics);
        }
        IdentifierBucket sequenceBucket = new IdentifierBucket(service);
        for (SequenceModel sq : s.sequences()) {
            validateSequence(sq, vendor, service, normalizeOnLoad, catalogName, name,
                sequenceBucket, diagnostics);
        }
    }

    private void validateTable(TableModel t,
                               EDbVendor vendor,
                               IdentifierService service,
                               boolean normalizeOnLoad,
                               String catalogName,
                               String schemaName,
                               IdentifierBucket objectBucket,
                               CatalogTableIndex tableIndex,
                               List<CatalogDiagnostic> diagnostics) {
        String name = t.name();
        String location = qualified(catalogName, schemaName, name);
        checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotTable,
            "table '" + location + "'", diagnostics);
        flagIfDuplicate(objectBucket, name, ESQLDataObjectType.dotTable, CatalogObjectKind.TABLE,
            "table '" + location + "'", diagnostics);

        IdentifierBucket columnBucket = new IdentifierBucket(service);
        for (ColumnModel c : t.columns()) {
            validateColumn(c, vendor, service, normalizeOnLoad, location, columnBucket, diagnostics);
        }

        IdentifierBucket constraintBucket = new IdentifierBucket(service);
        for (ConstraintModel cs : t.constraints()) {
            String csName = cs.name();
            if (csName != null && !csName.isEmpty()) {
                flagIfDuplicate(constraintBucket, csName, ESQLDataObjectType.dotUnknown,
                    CatalogObjectKind.CONSTRAINT,
                    "constraint '" + csName + "' on table '" + location + "'", diagnostics);
            }
            for (String col : cs.columns()) {
                if (!columnBucket.contains(col, ESQLDataObjectType.dotColumn)) {
                    diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_DUPLICATE_NAME,
                        "constraint '" + (csName == null ? "<unnamed>" : csName)
                            + "' on table '" + location + "' references unknown column '" + col + "'"));
                }
            }
            if (isForeignKeyType(cs.type())) {
                validateForeignKey(cs, location, schemaName, vendor, service, tableIndex, diagnostics);
            }
        }

        IdentifierBucket indexBucket = new IdentifierBucket(service);
        for (IndexModel ix : t.indexes()) {
            flagIfDuplicate(indexBucket, ix.name(), ESQLDataObjectType.dotUnknown,
                CatalogObjectKind.INDEX,
                "index '" + ix.name() + "' on table '" + location + "'", diagnostics);
            for (String col : ix.columns()) {
                if (!columnBucket.contains(col, ESQLDataObjectType.dotColumn)) {
                    diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_DUPLICATE_NAME,
                        "index '" + ix.name() + "' on table '" + location
                            + "' references unknown column '" + col + "'"));
                }
            }
        }
    }

    private void validateView(ViewModel v,
                              EDbVendor vendor,
                              IdentifierService service,
                              boolean normalizeOnLoad,
                              String catalogName,
                              String schemaName,
                              IdentifierBucket objectBucket,
                              List<CatalogDiagnostic> diagnostics) {
        String name = v.name();
        String location = qualified(catalogName, schemaName, name);
        checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotTable,
            "view '" + location + "'", diagnostics);
        flagIfDuplicate(objectBucket, name, ESQLDataObjectType.dotTable,
            v.materialized() ? CatalogObjectKind.MATERIALIZED_VIEW : CatalogObjectKind.VIEW,
            "view '" + location + "'", diagnostics);
        IdentifierBucket columnBucket = new IdentifierBucket(service);
        for (ColumnModel c : v.columns()) {
            validateColumn(c, vendor, service, normalizeOnLoad, location, columnBucket, diagnostics);
        }
    }

    private void validateRoutine(RoutineModel r,
                                 EDbVendor vendor,
                                 IdentifierService service,
                                 boolean normalizeOnLoad,
                                 String catalogName,
                                 String schemaName,
                                 IdentifierBucket objectBucket,
                                 List<CatalogDiagnostic> diagnostics) {
        String name = r.name();
        String location = qualified(catalogName, schemaName, name);
        // For diagnostic-message accuracy use the kind-specific leaf type when running
        // the bypass check; for duplicate detection always compare with dotFunction
        // because IdentifierProfile only routes dotFunction through ROUTINE_GROUP —
        // dotProcedure / dotOraclePackage / dotRoutine fall back to NAME_GROUP. Using
        // dotFunction here means MySQL procedures and packages share the routine
        // case-insensitivity rule, which matches the actual MySQL behavior (stored
        // routine names are case-insensitive regardless of routine kind).
        ESQLDataObjectType bypassType = leafTypeFor(r.kind());
        ESQLDataObjectType compareType = ESQLDataObjectType.dotFunction;
        String label = labelFor(r.kind());
        checkBypass(vendor, service, normalizeOnLoad, name, bypassType,
            label + " '" + location + "'", diagnostics);
        flagIfDuplicate(objectBucket, name, compareType, r.kind(),
            label + " '" + location + "'", diagnostics);
    }

    private void validateSynonym(SynonymModel sy,
                                 EDbVendor vendor,
                                 IdentifierService service,
                                 boolean normalizeOnLoad,
                                 String catalogName,
                                 String schemaName,
                                 IdentifierBucket objectBucket,
                                 List<CatalogDiagnostic> diagnostics) {
        String name = sy.name();
        String location = qualified(catalogName, schemaName, name);
        checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotSynonyms,
            "synonym '" + location + "'", diagnostics);
        flagIfDuplicate(objectBucket, name, ESQLDataObjectType.dotSynonyms, CatalogObjectKind.SYNONYM,
            "synonym '" + location + "'", diagnostics);
    }

    private void validateSequence(SequenceModel sq,
                                  EDbVendor vendor,
                                  IdentifierService service,
                                  boolean normalizeOnLoad,
                                  String catalogName,
                                  String schemaName,
                                  IdentifierBucket objectBucket,
                                  List<CatalogDiagnostic> diagnostics) {
        String name = sq.name();
        String location = qualified(catalogName, schemaName, name);
        checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotUnknown,
            "sequence '" + location + "'", diagnostics);
        flagIfDuplicate(objectBucket, name, ESQLDataObjectType.dotUnknown, CatalogObjectKind.SEQUENCE,
            "sequence '" + location + "'", diagnostics);
    }

    private void validateColumn(ColumnModel c,
                                EDbVendor vendor,
                                IdentifierService service,
                                boolean normalizeOnLoad,
                                String parentLocation,
                                IdentifierBucket columnBucket,
                                List<CatalogDiagnostic> diagnostics) {
        String name = c.name();
        checkBypass(vendor, service, normalizeOnLoad, name, ESQLDataObjectType.dotColumn,
            "column '" + parentLocation + "." + name + "'", diagnostics);
        flagIfDuplicate(columnBucket, name, ESQLDataObjectType.dotColumn, CatalogObjectKind.COLUMN,
            "column '" + parentLocation + "." + name + "'", diagnostics);
    }

    // ---------- helpers ----------

    private void checkRequired(String label, String value, List<CatalogDiagnostic> diagnostics) {
        if (value == null || value.isEmpty()) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_MISSING_DEFAULT,
                label + ".name is required"));
        }
    }

    /**
     * Identifier-bypass detection. When {@code normalizeOnLoad} is true, the raw name
     * must already equal {@code IdentifierService.normalize(name)}; otherwise an adapter
     * either pre-normalized through a non-{@code IdentifierService} path or shipped a
     * raw form that the resolver will fail to match. We emit a WARN, not an ERROR, so
     * downstream code remains tolerant — strict-mode callers escalate at the load
     * entry point.
     *
     * <p>For COLLATION_BASED dialects (MSSQL/Azure SQL) where {@code normalize} preserves
     * the input spelling, this check is a no-op: the resolver will route compares
     * through {@code IdentifierService.areEqual} regardless of stored case.</p>
     */
    private void checkBypass(EDbVendor vendor,
                             IdentifierService service,
                             boolean normalizeOnLoad,
                             String name,
                             ESQLDataObjectType type,
                             String location,
                             List<CatalogDiagnostic> diagnostics) {
        if (!normalizeOnLoad || name == null || name.isEmpty()) {
            return;
        }
        String normalized = service.normalize(name, type);
        if (normalized == null || normalized.equals(name)) {
            return;
        }
        // For COLLATION_BASED dialects (MSSQL/Azure SQL with case-insensitive collation,
        // MySQL lower_case_table_names=2) IdentifierService.normalize preserves the input
        // spelling, so we never reach this point for them — no false positives. The
        // condition above (normalized != name) is enough to identify a bypass for the
        // case-folding dialects (Oracle/DB2 → UPPER, Postgres/Greenplum/Redshift → lower)
        // we care about.
        diagnostics.add(CatalogDiagnostic.builder()
            .severity(CatalogDiagnosticSeverity.WARN)
            .code(CatalogDiagnosticCode.CATALOG_VALIDATION_IDENTIFIER_BYPASS)
            .message(location + ": raw name '" + name + "' is not in canonical form for "
                + vendor + " (expected '" + normalized + "'). The reader bypassed "
                + "IdentifierService normalization or normalizeOnLoad must be disabled.")
            .repairHint("Run input through CatalogIdentifierPolicy.normalize / "
                + "IdentifierService.normalize before constructing the model, or set "
                + "CatalogLoadOptions.normalizeOnLoad(false) when the source intentionally "
                + "preserves a non-canonical form.")
            .build());
    }

    private void flagIfDuplicate(IdentifierBucket bucket,
                                 String name,
                                 ESQLDataObjectType type,
                                 CatalogObjectKind kind,
                                 String location,
                                 List<CatalogDiagnostic> diagnostics) {
        if (name == null) return;
        if (!bucket.add(name, type)) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_DUPLICATE_NAME,
                "duplicate " + labelFor(kind) + ": " + location));
        }
    }

    private static CatalogDiagnostic error(CatalogDiagnosticCode code, String message) {
        return CatalogDiagnostic.builder()
            .severity(CatalogDiagnosticSeverity.ERROR)
            .code(code)
            .message(message)
            .build();
    }

    private static String qualified(String catalog, String schema, String name) {
        StringBuilder sb = new StringBuilder();
        if (catalog != null && !catalog.isEmpty()) sb.append(catalog).append('.');
        if (schema != null && !schema.isEmpty()) sb.append(schema).append('.');
        sb.append(name);
        return sb.toString();
    }

    /**
     * Choose the {@link IdentifierConfig} the validator should compare against.
     *
     * <p>Plan §9.2: the model carries its own config (frequently the vendor default,
     * sometimes adapter-supplied to model MySQL {@code lower_case_table_names} or
     * MSSQL collation). The caller may pass a {@link CatalogLoadOptions} with an
     * explicit override — in that case the override wins, after a vendor-consistency
     * check. When the caller did not pass an explicit override, the model's config is
     * authoritative.</p>
     */
    private static IdentifierConfig effectiveConfig(UnifiedCatalogModel model,
                                                    CatalogLoadOptions options,
                                                    List<CatalogDiagnostic> diagnostics) {
        IdentifierConfig modelCfg = model.identifierConfig();
        // Always check the model's own identifierConfig vendor matches the model vendor —
        // this is independent of the options path and catches a manifest that declared a
        // mismatched IdentifierConfig at the top level (rare but possible when readers
        // forward a config supplied by a different layer of code).
        if (modelCfg != null && modelCfg.vendor() != model.vendor()) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_IDENTIFIER_BYPASS,
                "UnifiedCatalogModel.identifierConfig.vendor=" + modelCfg.vendor()
                    + " does not match model.vendor=" + model.vendor()));
        }
        if (options == null) {
            return modelCfg;
        }
        // Always check options.vendor against model.vendor — a downstream loader / runtime
        // is going to be driven by options.vendor (CatalogQuery, CatalogLoaders), so a
        // mismatch is a real configuration bug regardless of whether identifierConfig was
        // explicit.
        if (options.vendor() != model.vendor()) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_IDENTIFIER_BYPASS,
                "CatalogLoadOptions.vendor=" + options.vendor()
                    + " does not match model.vendor=" + model.vendor()));
        }
        if (!options.hasExplicitIdentifierConfig()) {
            return modelCfg;
        }
        IdentifierConfig optsCfg = options.identifierConfig();
        if (optsCfg.vendor() != model.vendor()) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_IDENTIFIER_BYPASS,
                "CatalogLoadOptions.identifierConfig.vendor=" + optsCfg.vendor()
                    + " does not match model.vendor=" + model.vendor()));
            return modelCfg;
        }
        return optsCfg;
    }

    private static ESQLDataObjectType leafTypeFor(CatalogObjectKind kind) {
        switch (kind) {
            case FUNCTION:  return ESQLDataObjectType.dotFunction;
            case PROCEDURE: return ESQLDataObjectType.dotProcedure;
            case PACKAGE:   return ESQLDataObjectType.dotOraclePackage;
            case ROUTINE:   return ESQLDataObjectType.dotRoutine;
            default:        return ESQLDataObjectType.dotUnknown;
        }
    }

    /**
     * Diagnostic-friendly lower-case label for a kind. Hand-coded to avoid
     * {@link String#toLowerCase()}, which the forbidden-apis plugin bans inside
     * {@code catalog/**} (plan §9.5).
     */
    private static String labelFor(CatalogObjectKind kind) {
        switch (kind) {
            case CATALOG:           return "catalog";
            case SCHEMA:            return "schema";
            case TABLE:             return "table";
            case VIEW:              return "view";
            case MATERIALIZED_VIEW: return "materialized view";
            case COLUMN:            return "column";
            case ROUTINE:           return "routine";
            case FUNCTION:          return "function";
            case PROCEDURE:         return "procedure";
            case PACKAGE:           return "package";
            case SYNONYM:           return "synonym";
            case SEQUENCE:          return "sequence";
            case TYPE:              return "type";
            case TRIGGER:           return "trigger";
            case INDEX:             return "index";
            case CONSTRAINT:        return "constraint";
            default:                return "object";
        }
    }

    /**
     * Per-scope identifier deduplication bucket. Calls {@link IdentifierService#areEqual}
     * for every membership test so dialects whose compare rule is case-insensitive but
     * whose {@code normalize} preserves spelling (BigQuery columns, MSSQL with case-
     * insensitive collation, MySQL {@code lower_case_table_names=2}) still catch
     * collisions that a {@code Set<String>} of normalize keys would miss.
     *
     * <p>The {@link ESQLDataObjectType} used for comparison is supplied per-{@link #add(String, ESQLDataObjectType)}
     * call rather than fixed at construction. This matters when one logical namespace
     * mixes kinds that have different per-vendor case-sensitivity rules — e.g., MySQL's
     * default schema namespace where tables compare case-sensitively
     * ({@link ESQLDataObjectType#dotTable}) but routines compare case-insensitively
     * ({@link ESQLDataObjectType#dotRoutine}/{@code dotFunction}/{@code dotProcedure}).
     * Each entry is checked against existing entries using the type registered on
     * insert; duplicate detection is symmetric (existing → new and new → existing).</p>
     */
    static final class IdentifierBucket {
        private final IdentifierService service;
        private final List<String> seenNames = new ArrayList<String>();
        private final List<ESQLDataObjectType> seenTypes = new ArrayList<ESQLDataObjectType>();

        IdentifierBucket(IdentifierService service) {
            this.service = service;
        }

        /** Add a name with its kind-specific compare type. */
        boolean add(String raw, ESQLDataObjectType type) {
            if (raw == null) return true;
            for (int i = 0; i < seenNames.size(); i++) {
                String existing = seenNames.get(i);
                ESQLDataObjectType existingType = seenTypes.get(i);
                // Cross-kind compare uses the more permissive (incoming) type so the bucket
                // catches "table Fn vs function fn" collisions even when only one of the
                // two compare rules is case-insensitive.
                if (service.areEqual(existing, raw, type)
                    || service.areEqual(existing, raw, existingType)) {
                    return false;
                }
            }
            seenNames.add(raw);
            seenTypes.add(type);
            return true;
        }

        /**
         * Membership test against entries already added with their per-call type.
         * Used by constraint/index references against column names — the caller passes
         * {@link ESQLDataObjectType#dotColumn} (which is what those entries were added
         * with anyway, but the API stays explicit).
         */
        boolean contains(String raw, ESQLDataObjectType type) {
            if (raw == null) return false;
            for (int i = 0; i < seenNames.size(); i++) {
                if (service.areEqual(seenNames.get(i), raw, type)
                    || service.areEqual(seenNames.get(i), raw, seenTypes.get(i))) {
                    return true;
                }
            }
            return false;
        }
    }

    // ---------- FK validation ----------

    /**
     * Lenient FK type recognition. Matches {@code FK}, {@code FOREIGN KEY},
     * {@code FOREIGN_KEY} case-insensitively against the constraint's
     * {@link ConstraintModel#type()} string. Hand-coded to avoid
     * {@link String#equalsIgnoreCase} which the forbidden-apis plugin bans
     * inside {@code catalog/**}.
     */
    private static boolean isForeignKeyType(String type) {
        if (type == null) return false;
        return asciiEqualsIgnoreCase(type, "FK")
            || asciiEqualsIgnoreCase(type, "FOREIGN KEY")
            || asciiEqualsIgnoreCase(type, "FOREIGN_KEY");
    }

    private static boolean asciiEqualsIgnoreCase(String a, String b) {
        if (a == null || b == null) return a == b;
        int len = a.length();
        if (b.length() != len) return false;
        for (int i = 0; i < len; i++) {
            char ca = a.charAt(i);
            char cb = b.charAt(i);
            if (ca == cb) continue;
            if (ca >= 'A' && ca <= 'Z') ca = (char) (ca + 32);
            if (cb >= 'A' && cb <= 'Z') cb = (char) (cb + 32);
            if (ca != cb) return false;
        }
        return true;
    }

    private void validateForeignKey(ConstraintModel cs,
                                    String fkLocation,
                                    String currentSchema,
                                    EDbVendor vendor,
                                    IdentifierService service,
                                    CatalogTableIndex tableIndex,
                                    List<CatalogDiagnostic> diagnostics) {
        String csLabel = "constraint '" + (cs.name() == null ? "<unnamed>" : cs.name())
            + "' on table '" + fkLocation + "'";
        String referencedTable = cs.referencedTable();
        if (referencedTable == null || referencedTable.isEmpty()) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_INVALID_REFERENCE,
                csLabel + " is a FOREIGN KEY but has no referencedTable"));
            return;
        }
        List<String> refCols = cs.referencedColumns();
        if (refCols.size() != cs.columns().size()) {
            diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_INVALID_REFERENCE,
                csLabel + " has " + cs.columns().size() + " column(s) but "
                    + refCols.size() + " referencedColumn(s); FK column counts must match"));
            // Continue to surface other issues (target table resolution) — don't bail.
        }

        CatalogTableIndex.Lookup resolved = tableIndex.resolve(referencedTable, currentSchema);
        switch (resolved.kind) {
            case NOT_FOUND:
                diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_INVALID_REFERENCE,
                    csLabel + " references unknown table '" + referencedTable + "'"));
                return;
            case AMBIGUOUS:
                diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_INVALID_REFERENCE,
                    csLabel + " references ambiguous bare table name '" + referencedTable
                        + "'; qualify with schema"));
                return;
            case FOUND:
            default:
                break;
        }

        // Verify each referencedColumn exists on the target table.
        TableModel target = resolved.table;
        for (String col : refCols) {
            if (!columnExists(target, col, service)) {
                diagnostics.add(error(CatalogDiagnosticCode.CATALOG_VALIDATION_INVALID_REFERENCE,
                    csLabel + " references unknown column '" + col
                        + "' on referenced table '" + referencedTable + "'"));
            }
        }
    }

    private static boolean columnExists(TableModel t, String columnName, IdentifierService service) {
        if (columnName == null) return false;
        for (ColumnModel c : t.columns()) {
            if (service.areEqual(c.name(), columnName, ESQLDataObjectType.dotColumn)) {
                return true;
            }
        }
        return false;
    }

    /**
     * Per-catalog table lookup index, built once at {@link #validateCatalog} entry and
     * threaded down so each FK constraint can resolve its {@code referencedTable}
     * against the whole catalog (not just the current schema).
     *
     * <p>Keys are normalized via {@link IdentifierService#normalize} so the lookup
     * honors the vendor's identifier-folding rule (Oracle uppercase, PG lowercase,
     * etc.). Three key shapes are stored:</p>
     * <ul>
     *   <li><b>Catalog-qualified</b> ({@code "catalog.schema.table"}) — populated
     *       for every table. Lets an FK target an explicit
     *       {@code ConstraintModel#referencedTable} of the form
     *       {@code "ORCL.HR.DEPT"} (the form the Javadoc documents).</li>
     *   <li><b>Schema-qualified</b> ({@code "schema.table"}) — populated for every
     *       table. The primary lookup for FK references of the form
     *       {@code "schema.table"}.</li>
     *   <li><b>Bare</b> ({@code "table"}) — fallback when the FK's
     *       {@code referencedTable} has no dot. If a bare name appears in 2+
     *       schemas, it is marked ambiguous and resolution then requires the
     *       caller to qualify.</li>
     * </ul>
     *
     * <p><b>Resolution order for bare {@code referencedTable}:</b></p>
     * <ol>
     *   <li>First try {@code currentSchema + "." + ref} — current-schema-first
     *       precedence, matching CREATE TABLE / FK semantics in real databases.
     *       A local same-named table wins over a same-named table in another
     *       schema even when the bare name is ambiguous catalog-wide.</li>
     *   <li>Otherwise fall back to a catalog-wide bare lookup; ambiguity is only
     *       reported when the bare name doesn't resolve in the current schema.</li>
     * </ol>
     */
    private static final class CatalogTableIndex {

        enum Kind { FOUND, NOT_FOUND, AMBIGUOUS }

        static final class Lookup {
            final Kind kind;
            final TableModel table;
            Lookup(Kind k, TableModel t) { this.kind = k; this.table = t; }
        }

        private final IdentifierService service;
        // Catalog-qualified ("catalog.schema.table") and schema-qualified ("schema.table") keys
        // share this map. Both forms are pre-normalized once at build time.
        private final Map<String, TableModel> byQualifiedKey;
        private final Map<String, TableModel> byBareKey;
        private final Map<String, Boolean> bareNameAmbiguous;

        private CatalogTableIndex(IdentifierService service) {
            this.service = service;
            this.byQualifiedKey = new LinkedHashMap<String, TableModel>();
            this.byBareKey = new HashMap<String, TableModel>();
            this.bareNameAmbiguous = new HashMap<String, Boolean>();
        }

        static CatalogTableIndex build(CatalogModel c, IdentifierService service) {
            CatalogTableIndex idx = new CatalogTableIndex(service);
            String catalogName = c.name();
            for (SchemaModel s : c.schemas()) {
                String schemaName = s.name();
                for (TableModel t : s.tables()) {
                    String bare = t.name();
                    if (bare == null || bare.isEmpty()) continue;
                    if (schemaName != null && !schemaName.isEmpty()) {
                        String schemaQual = service.normalize(
                            schemaName + "." + bare, ESQLDataObjectType.dotTable);
                        idx.byQualifiedKey.put(schemaQual, t);
                        if (catalogName != null && !catalogName.isEmpty()) {
                            String fullyQual = service.normalize(
                                catalogName + "." + schemaName + "." + bare,
                                ESQLDataObjectType.dotTable);
                            idx.byQualifiedKey.put(fullyQual, t);
                        }
                    }
                    String bareKey = service.normalize(bare, ESQLDataObjectType.dotTable);
                    if (idx.byBareKey.containsKey(bareKey)) {
                        idx.bareNameAmbiguous.put(bareKey, Boolean.TRUE);
                    } else {
                        idx.byBareKey.put(bareKey, t);
                    }
                }
            }
            return idx;
        }

        Lookup resolve(String referencedTable, String currentSchema) {
            if (referencedTable == null || referencedTable.isEmpty()) {
                return new Lookup(Kind.NOT_FOUND, null);
            }
            // Case 1: ref has a dot — treat as qualified. Tries both shapes the index
            // stores (catalog.schema.table and schema.table) without further parsing.
            if (referencedTable.indexOf('.') >= 0) {
                String qual = service.normalize(referencedTable, ESQLDataObjectType.dotTable);
                TableModel hit = byQualifiedKey.get(qual);
                if (hit != null) return new Lookup(Kind.FOUND, hit);
                return new Lookup(Kind.NOT_FOUND, null);
            }
            // Case 2: ref is bare. Try current schema first (current-schema-first
            // precedence — matches FK / CREATE TABLE semantics in real databases).
            if (currentSchema != null && !currentSchema.isEmpty()) {
                String qual = service.normalize(
                    currentSchema + "." + referencedTable, ESQLDataObjectType.dotTable);
                TableModel hit = byQualifiedKey.get(qual);
                if (hit != null) return new Lookup(Kind.FOUND, hit);
            }
            // Case 3: ref is bare and not in current schema. Fall back to catalog-wide
            // bare lookup; ambiguity gate fires only when the local-schema shortcut missed.
            String bareKey = service.normalize(referencedTable, ESQLDataObjectType.dotTable);
            if (Boolean.TRUE.equals(bareNameAmbiguous.get(bareKey))) {
                return new Lookup(Kind.AMBIGUOUS, null);
            }
            TableModel hit = byBareKey.get(bareKey);
            if (hit != null) return new Lookup(Kind.FOUND, hit);
            return new Lookup(Kind.NOT_FOUND, null);
        }
    }
}