package gudusoft.gsqlparser.catalog.runtime;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.catalog.input.model.IdentifierConfig;
import gudusoft.gsqlparser.sqlenv.CollatorProvider;
import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
import gudusoft.gsqlparser.sqlenv.IdentifierProfile;
import gudusoft.gsqlparser.sqlenv.IdentifierService;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
import java.util.Objects;

/**
 * Thin wrapper over {@link IdentifierService} that keeps the {@code catalog/**} dependency
 * direction one-way: the package depends on {@code IdentifierService} and
 * {@link IdentifierConfig} but on no other {@code sqlenv} type.
 *
 * <p>Plan §9.2 / §9.5. Every case-folding, normalization, qualified-name comparison, and
 * map key inside {@code catalog/**} routes through this class (or directly through
 * {@code IdentifierService}); the forbidden-apis Maven plugin enforces this mechanically.</p>
 *
 * <p>{@link IdentifierConfig} fields that map to {@code IdentifierProfile.VendorFlags}
 * (currently {@code mysqlLowerCaseTableNames} and {@code mssqlCollation}) are honored by
 * constructing a per-call {@code IdentifierService} when they differ from defaults.</p>
 *
 * <p><strong>Round 1 scope cuts (tracked for P1B Round 2):</strong>
 * <ul>
 *   <li>{@code foldUnquotedToUpper} / {@code foldUnquotedToLower} on
 *       {@link IdentifierConfig} are recorded but not yet projected onto
 *       {@code IdentifierProfile} — the per-call profile is still constructed via
 *       {@code IdentifierProfile.forVendor(vendor, flags)} which uses vendor-default fold
 *       rules. Callers who declare a non-default fold get vendor-default behavior.</li>
 *   <li>{@code tableCaseSensitive} / {@code columnCaseSensitive} (BigQuery split policy)
 *       are accepted but not honored: extending {@code IdentifierProfile} with per-object-
 *       group sensitivity flags is part of Round 2.</li>
 *   <li>{@link #areEqual} Tier-2 fallback for COLLATION_BASED dialects routes through
 *       {@link IdentifierService#areEqualStatic} (cached vendor defaults) and therefore
 *       does not honor a custom {@code mssqlCollation} or MySQL
 *       {@code lower_case_table_names=2}; Tier 1 (cfg-aware normalized comparison)
 *       handles the lctn=1 / lctn=2 case-insensitive cases since their normalize folds
 *       to a single canonical form.</li>
 *   <li>{@link #keyForMap} can produce different keys for two names that
 *       {@link #areEqual} considers equal in COLLATION_BASED dialects — same contract as
 *       {@code IdentifierService.canUseCompositeKey}. Snapshot lookups must fall back to
 *       {@code areEqual} on a key miss.</li>
 * </ul>
 *
 * <p>The defaults-driven path (the primary use of this class in P1B/P1C) is fully covered
 * by tests and is the one used by every Phase 1 reader and provider.</p>
 */
public final class CatalogIdentifierPolicy {

    private static final char KEY_SEPARATOR = '\u0001';

    private CatalogIdentifierPolicy() {
        // Static utility — no instances.
    }

    /**
     * Parse a possibly-quoted, possibly-multi-segment qualified name into a
     * {@link CatalogQualifiedName} with both raw and normalized segments tracked.
     *
     * <p>Each segment is normalized using the {@link ESQLDataObjectType} appropriate for its
     * position in the qualified name (catalog → schema → table → column), so vendor-specific
     * dual policies like BigQuery's case-sensitive tables vs case-insensitive columns are
     * applied per-segment rather than en bloc.</p>
     */
    public static CatalogQualifiedName parse(String raw, CatalogObjectKind kind,
                                             IdentifierConfig cfg, EDbVendor vendor) {
        if (raw == null || raw.isEmpty()) {
            throw new IllegalArgumentException("CatalogIdentifierPolicy.parse: raw must be non-empty");
        }
        if (kind == null) {
            throw new IllegalArgumentException("CatalogIdentifierPolicy.parse: kind is required");
        }
        if (vendor == null) {
            throw new IllegalArgumentException("CatalogIdentifierPolicy.parse: vendor is required");
        }
        IdentifierService service = (cfg != null) ? serviceFor(cfg, vendor) : null;
        SegmentSplit split = splitSegments(raw, vendor);
        BitSet quoted = split.quoted;
        List<String> rawSegments = split.segments;
        List<String> normalized = new ArrayList<String>(rawSegments.size());
        for (int i = 0; i < rawSegments.size(); i++) {
            String segment = rawSegments.get(i);
            ESQLDataObjectType type = kindForSegment(kind, i, rawSegments.size());
            if (service != null) {
                normalized.add(service.normalize(segment, type));
            } else {
                normalized.add(IdentifierService.normalizeStatic(vendor, type, segment));
            }
        }
        return new CatalogQualifiedName(rawSegments, normalized, quoted, kind, vendor);
    }

    /**
     * Build a {@link CatalogQualifiedName} from segments that have already been normalized
     * by {@link IdentifierService}. Used by the lazy {@code TSQLEnv} bridge, where
     * {@code TSQLEnv.doSearchSchemaObject} hands over already-normalized catalog/schema/
     * object segments and rejoining + re-parsing would mis-handle quoted-embedded-dot
     * names like Oracle {@code "a.b"."c"}.
     *
     * <p>Each segment is taken verbatim — no further normalization is applied — and the
     * resulting name carries identical raw and normalized segments. Quoted-flag bits are
     * empty by default; callers that have lost quoting information at this layer
     * (the legacy bridge path) accept the limitation that quoted-vs-unquoted distinction
     * is no longer recoverable. Snapshot lookups on this name still match because the
     * snapshot keys also derive from the same {@link IdentifierService} normalization.</p>
     */
    public static CatalogQualifiedName fromAlreadyNormalizedSegments(List<String> segments,
                                                                     CatalogObjectKind kind,
                                                                     EDbVendor vendor) {
        if (segments == null || segments.isEmpty()) {
            throw new IllegalArgumentException(
                "CatalogIdentifierPolicy.fromAlreadyNormalizedSegments: segments must be non-empty");
        }
        if (kind == null) {
            throw new IllegalArgumentException(
                "CatalogIdentifierPolicy.fromAlreadyNormalizedSegments: kind is required");
        }
        if (vendor == null) {
            throw new IllegalArgumentException(
                "CatalogIdentifierPolicy.fromAlreadyNormalizedSegments: vendor is required");
        }
        // Defensive copy — don't trust the caller to keep their list stable.
        List<String> normalized = new ArrayList<String>(segments);
        for (int i = 0; i < normalized.size(); i++) {
            String s = normalized.get(i);
            if (s == null) {
                throw new IllegalArgumentException(
                    "CatalogIdentifierPolicy.fromAlreadyNormalizedSegments: segment " + i + " is null");
            }
        }
        return new CatalogQualifiedName(normalized, normalized, new BitSet(), kind, vendor);
    }

    /**
     * Map-friendly stable key for a {@link CatalogQualifiedName}.
     *
     * <p>Length-prefixes each normalized segment so quoted identifiers that contain dots —
     * e.g., Oracle {@code "a.b"."c"} vs {@code "a"."b.c"} — produce distinct keys despite
     * having the same flat join.</p>
     *
     * <p><strong>COLLATION_BASED limitation:</strong> for SQL Server / Azure SQL under a
     * case-insensitive collation, the legacy {@code IdentifierService.normalize} preserves
     * case (the legacy code-path uses Collator-bucketed lookup for those dialects).
     * Consequently {@code keyForMap("dbo.Orders")} and {@code keyForMap("dbo.orders")} can
     * differ even though {@link #areEqual} considers them equal. Snapshot/index lookups
     * that key by {@code keyForMap} on those dialects must therefore fall back to
     * {@link #areEqual} on a key miss; this is the same contract that
     * {@code IdentifierService.canUseCompositeKey()} exposes for the legacy code-path.
     * P1B/P1C may add a Collator-aware key bucket; tracked as a separate task.</p>
     */
    public static String keyForMap(CatalogQualifiedName name) {
        if (name == null) {
            throw new IllegalArgumentException("CatalogIdentifierPolicy.keyForMap: name is required");
        }
        StringBuilder sb = new StringBuilder();
        sb.append(name.kind().name()).append(KEY_SEPARATOR)
          .append(name.vendor().name()).append(KEY_SEPARATOR);
        List<String> segments = name.normalized();
        for (int i = 0; i < segments.size(); i++) {
            String segment = segments.get(i);
            sb.append(segment.length()).append(':').append(segment).append(KEY_SEPARATOR);
        }
        return sb.toString();
    }

    /**
     * Vendor-aware equality test for two qualified names. Per-segment comparisons use the
     * {@link ESQLDataObjectType} that matches each segment's role (catalog/schema/table/
     * column) so dual-policy dialects compare correctly.
     *
     * <p>Comparison is two-tier: first the cfg-aware {@code normalized} segments are
     * compared directly (which handles every dialect whose {@code normalize} captures
     * equality, including non-default {@code IdentifierConfig}s like MySQL
     * {@code lower_case_table_names=1/2}); on a mismatch we fall through to
     * {@link IdentifierService#areEqualStatic} so Collator-based dialects (SQL Server / Azure
     * SQL with case-insensitive collations) still report equality correctly.</p>
     */
    public static boolean areEqual(CatalogQualifiedName a, CatalogQualifiedName b) {
        if (a == null || b == null) {
            return a == b;
        }
        if (a.kind() != b.kind() || a.vendor() != b.vendor() || a.size() != b.size()) {
            return false;
        }
        // Tier 1: compare normalized forms (honors IdentifierConfig used at parse time).
        if (a.normalized().equals(b.normalized())) {
            return true;
        }
        // Tier 2: route through the legacy Collator-aware static path for COLLATION_BASED dialects.
        for (int i = 0; i < a.size(); i++) {
            ESQLDataObjectType type = kindForSegment(a.kind(), i, a.size());
            String aSeg = a.raw().get(i);
            String bSeg = b.raw().get(i);
            if (!IdentifierService.areEqualStatic(a.vendor(), type, aSeg, bSeg)) {
                return false;
            }
        }
        return true;
    }

    /** Normalize a single identifier segment per {@link IdentifierConfig}'s vendor. */
    public static String normalize(String raw, IdentifierConfig cfg) {
        if (raw == null) return null;
        if (cfg == null) {
            throw new IllegalArgumentException("CatalogIdentifierPolicy.normalize: cfg is required");
        }
        return serviceFor(cfg, cfg.vendor()).normalize(raw, ESQLDataObjectType.dotUnknown);
    }

    /** Normalize a segment with explicit quoted-ness (advisory). */
    public static String normalizeSegment(String raw, boolean wasQuoted, IdentifierConfig cfg) {
        Objects.requireNonNull(cfg, "cfg");
        // The IdentifierService.normalize(...) method inspects raw for quote characters
        // itself, so wasQuoted is treated as advisory metadata. Retained for symmetry with
        // the plan §9.2 API and to surface intent at call-sites.
        return serviceFor(cfg, cfg.vendor()).normalize(raw, ESQLDataObjectType.dotUnknown);
    }

    /**
     * Public façade over {@link #serviceFor(IdentifierConfig, EDbVendor)} for callers
     * outside this package (validator, readers, providers) that need an
     * {@link IdentifierService} matching a {@link IdentifierConfig}. Vendor defaults
     * to {@code cfg.vendor()} when {@code vendor} is {@code null}.
     */
    public static IdentifierService identifierServiceFor(IdentifierConfig cfg, EDbVendor vendor) {
        if (cfg == null) {
            throw new IllegalArgumentException(
                "CatalogIdentifierPolicy.identifierServiceFor: cfg is required");
        }
        return serviceFor(cfg, vendor != null ? vendor : cfg.vendor());
    }

    // ---------- internals ----------

    /**
     * Build an {@link IdentifierService} that honors the supplied {@link IdentifierConfig}.
     * Falls back to the cached default service when the config matches vendor defaults
     * (avoids re-allocating on every parse for the common case).
     */
    static IdentifierService serviceFor(IdentifierConfig cfg, EDbVendor vendorOverride) {
        EDbVendor vendor = vendorOverride != null ? vendorOverride : cfg.vendor();
        IdentifierProfile.VendorFlags flags = vendorFlagsFrom(cfg);
        return IdentifierService.forProfile(IdentifierProfile.forVendor(vendor, flags));
    }

    static IdentifierProfile.VendorFlags vendorFlagsFrom(IdentifierConfig cfg) {
        // Unconfigured values adopt the shared static defaults (P0d.2: MySQL mode 1) so
        // this path cannot drift from VendorFlags.defaults(); explicit config still wins.
        IdentifierProfile.VendorFlags d = IdentifierProfile.VendorFlags.defaults();
        int mysqlLctn = cfg.mysqlLowerCaseTableNames() != null
            ? cfg.mysqlLowerCaseTableNames() : d.mysqlLowerCaseTableNames;
        String collation = cfg.mssqlCollation() != null
            ? cfg.mssqlCollation() : d.defaultCollation;
        return new IdentifierProfile.VendorFlags(mysqlLctn, collation,
            d.redshiftEnableCaseSensitive, d.snowflakeQuotedIdentifiersIgnoreCase);
    }

    /**
     * For a qualified name with overall {@code kind}, return the {@link ESQLDataObjectType}
     * appropriate for the segment at {@code segmentIndex} within a name of {@code total}
     * segments.
     *
     * <p>Layouts assumed:</p>
     * <ul>
     *   <li>4 segments → server.catalog.schema.object</li>
     *   <li>3 segments → catalog.schema.object</li>
     *   <li>2 segments → schema.object (or catalog.object on dialects without schemas)</li>
     *   <li>1 segment  → object</li>
     * </ul>
     */
    static ESQLDataObjectType kindForSegment(CatalogObjectKind kind, int segmentIndex, int total) {
        int distanceFromEnd = total - 1 - segmentIndex;
        if (distanceFromEnd == 0) {
            return mapKind(kind);
        }
        if (distanceFromEnd == 1) {
            // Parent of the leaf:
            // * for COLUMN / INDEX / CONSTRAINT, the parent is a TABLE
            // * for TABLE / VIEW / MATERIALIZED_VIEW / ROUTINE / FUNCTION / PROCEDURE /
            //   PACKAGE / SYNONYM / SEQUENCE / TRIGGER / TYPE, the parent is a SCHEMA
            switch (kind) {
                case COLUMN:
                case INDEX:
                case CONSTRAINT:
                    return ESQLDataObjectType.dotTable;
                default:
                    return ESQLDataObjectType.dotSchema;
            }
        }
        if (distanceFromEnd == 2) {
            // Grandparent:
            // * for COLUMN, grandparent is SCHEMA
            // * else CATALOG
            switch (kind) {
                case COLUMN:
                case INDEX:
                case CONSTRAINT:
                    return ESQLDataObjectType.dotSchema;
                default:
                    return ESQLDataObjectType.dotCatalog;
            }
        }
        // distanceFromEnd >= 3: server / catalog level
        if (distanceFromEnd == 3 && (kind == CatalogObjectKind.COLUMN
                || kind == CatalogObjectKind.INDEX || kind == CatalogObjectKind.CONSTRAINT)) {
            return ESQLDataObjectType.dotCatalog;
        }
        return ESQLDataObjectType.dotUnknown;
    }

    /**
     * Map a {@link CatalogObjectKind} to its leaf {@link ESQLDataObjectType}. Internal —
     * use {@link #kindForSegment} when handling multi-segment names.
     */
    /**
     * The comparison/quoting {@link ESQLDataObjectType} for an object of this entity kind — the
     * SINGLE owner of this mapping (stage S0R of
     * docs/refactor/identifier_identity_display_separation_plan.md).
     *
     * <p>It lives here, not in {@code sqlenv}, because {@code CatalogObjectKind} is a catalog
     * concept and {@code catalog.runtime} already depends on {@code sqlenv}. Putting the mapping
     * in {@code sqlenv} would make the low-level identifier layer depend on the high-level catalog
     * layer and close a package cycle. {@code sqlenv.IdentifierRolePolicy} therefore maps the
     * ROLES it owns and refuses {@code OBJECT}, whose comparison type only a catalog caller knows.
     *
     * <p>Several kinds deliberately share a type: a view and a materialized view fold like a table
     * because no dialect gives them separate case rules. Their IDENTITY stays separate because the
     * key carries the kind itself. The kind discriminates the entity; the type decides how its
     * spelling folds.
     *
     * @throws IllegalStateException for an unmapped kind. Falling through to {@code dotUnknown} is
     *         how a whole entity class acquires the wrong case semantics unnoticed, which is
     *         exactly what happened to SEQUENCE, INDEX and CONSTRAINT before the comparison types
     *         for them existed.
     */
    public static ESQLDataObjectType comparisonTypeOfKind(CatalogObjectKind kind) {
        if (kind == null) {
            throw new IllegalArgumentException("catalog object kind is null");
        }
        switch (kind) {
            case CATALOG: return ESQLDataObjectType.dotCatalog;
            case SCHEMA: return ESQLDataObjectType.dotSchema;
            case TABLE:
            case VIEW:
            case MATERIALIZED_VIEW: return ESQLDataObjectType.dotTable;
            case COLUMN: return ESQLDataObjectType.dotColumn;
            case ROUTINE: return ESQLDataObjectType.dotRoutine;
            case FUNCTION: return ESQLDataObjectType.dotFunction;
            case PROCEDURE: return ESQLDataObjectType.dotProcedure;
            case PACKAGE: return ESQLDataObjectType.dotOraclePackage;
            case SYNONYM: return ESQLDataObjectType.dotSynonyms;
            case TYPE: return ESQLDataObjectType.dotDataType;
            case TRIGGER: return ESQLDataObjectType.dotTrigger;
            case SEQUENCE: return ESQLDataObjectType.dotSequence;
            case INDEX: return ESQLDataObjectType.dotIndex;
            case CONSTRAINT: return ESQLDataObjectType.dotConstraint;
            default:
                throw new IllegalStateException("unmapped catalog object kind: " + kind
                        + "; add it to CatalogIdentifierPolicy.comparisonTypeOfKind rather than "
                        + "letting it fall through to dotUnknown");
        }
    }

    /** @deprecated use {@link #comparisonTypeOfKind}; kept so existing callers keep compiling. */
    @Deprecated
    static ESQLDataObjectType mapKind(CatalogObjectKind kind) {
        return comparisonTypeOfKind(kind);
    }

    /** Output of {@link #splitSegments}: parallel raw-segment list plus quoted-flag bits. */
    static final class SegmentSplit {
        final List<String> segments;
        final BitSet quoted;

        SegmentSplit(List<String> segments, BitSet quoted) {
            this.segments = segments;
            this.quoted = quoted;
        }
    }

    /**
     * Split a qualified name into raw segments preserving any surrounding quote characters
     * so {@link CatalogQualifiedName} can record per-segment quoted flags faithfully.
     *
     * <p>BigQuery special-case: a single backtick-delimited path that contains dots
     * (e.g., {@code `project.dataset.table`}) is split internally, mirroring the
     * convention also used by {@code SQLUtil.parseNames}.</p>
     */
    static SegmentSplit splitSegments(String raw, EDbVendor vendor) {
        // BigQuery wraps the whole path in backticks: split internally, mark unquoted.
        if (vendor == EDbVendor.dbvbigquery
                && raw.length() >= 2
                && raw.charAt(0) == '`'
                && raw.charAt(raw.length() - 1) == '`'
                && raw.indexOf('.') > 0
                && countUnescaped(raw, '`') == 2) {
            String inner = raw.substring(1, raw.length() - 1);
            return splitOnDot(inner, false);
        }
        List<String> out = new ArrayList<String>();
        BitSet quotedFlags = new BitSet();
        StringBuilder current = new StringBuilder();
        boolean currentQuoted = false;
        char quoteChar = 0;
        boolean inBracket = false;
        for (int i = 0; i < raw.length(); i++) {
            char c = raw.charAt(i);
            if (quoteChar != 0) {
                current.append(c);
                if (c == quoteChar) {
                    if (i + 1 < raw.length() && raw.charAt(i + 1) == quoteChar) {
                        current.append(raw.charAt(i + 1));
                        i++;
                    } else {
                        quoteChar = 0;
                    }
                }
            } else if (inBracket) {
                current.append(c);
                if (c == ']') {
                    inBracket = false;
                }
            } else if (c == '"' || c == '`') {
                current.append(c);
                quoteChar = c;
                currentQuoted = true;
            } else if (c == '[' && (vendor == EDbVendor.dbvmssql || vendor == EDbVendor.dbvazuresql)) {
                current.append(c);
                inBracket = true;
                currentQuoted = true;
            } else if (c == '.') {
                if (currentQuoted) quotedFlags.set(out.size());
                out.add(current.toString());
                current.setLength(0);
                currentQuoted = false;
            } else {
                current.append(c);
            }
        }
        if (currentQuoted) quotedFlags.set(out.size());
        out.add(current.toString());
        return new SegmentSplit(out, quotedFlags);
    }

    private static SegmentSplit splitOnDot(String inner, boolean markQuoted) {
        List<String> out = new ArrayList<String>();
        BitSet quoted = new BitSet();
        int start = 0;
        for (int i = 0; i < inner.length(); i++) {
            if (inner.charAt(i) == '.') {
                String seg = inner.substring(start, i);
                if (markQuoted) quoted.set(out.size());
                out.add(seg);
                start = i + 1;
            }
        }
        String last = inner.substring(start);
        if (markQuoted) quoted.set(out.size());
        out.add(last);
        return new SegmentSplit(out, quoted);
    }

    private static int countUnescaped(String s, char ch) {
        int n = 0;
        for (int i = 0; i < s.length(); i++) {
            if (s.charAt(i) == ch) {
                if (i + 1 < s.length() && s.charAt(i + 1) == ch) {
                    i++;
                } else {
                    n++;
                }
            }
        }
        return n;
    }
}