package gudusoft.gsqlparser.catalog.runtime;

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.diagnostic.CatalogException;
import gudusoft.gsqlparser.catalog.input.CatalogLoadingMode;
import gudusoft.gsqlparser.catalog.input.model.DefaultsConfig;
import gudusoft.gsqlparser.catalog.input.model.IdentifierConfig;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;

/**
 * Top-level runtime: owns the active {@link CatalogProvider}, holds the current
 * {@link CatalogSnapshot}, exposes the {@link CatalogResolver}, and manages the mutable
 * {@link CatalogOverlay}.
 *
 * <p>Plan §5.1 / §5.4 / §7.2. Resolver lookup precedence:</p>
 * <ol>
 *   <li>{@link CatalogOverlay} — DDL discovered mid-batch, temp tables, addTable.</li>
 *   <li>{@link CatalogSnapshot} — immutable snapshot from the last
 *       {@link CatalogProvider#snapshot(CatalogQuery)} call.</li>
 *   <li>Provider on-miss fetch (LAZY/AUTO modes only) — {@code provider.snapshot(...)}
 *       is asked for the missing name and the result merged into the cached snapshot.</li>
 * </ol>
 *
 * <p>Read-only consumers (resolver, future SQLGuard, future Semantic IR) target the
 * snapshot/resolver; live sources implement {@link CatalogProvider}. The runtime is
 * single-threaded by default; concurrent parsers should use one runtime per analysis run.</p>
 */
public final class CatalogRuntime implements AutoCloseable {

    private final CatalogProvider provider;
    private final EDbVendor vendor;
    private final CatalogLoadingMode loadingMode;
    private final long ttlMillis;
    private final int maxFetchesPerAnalysis;
    private final boolean includeColumns;
    private final boolean includeViews;
    private final boolean includeRoutines;
    private final InMemoryCatalogOverlay overlay = new InMemoryCatalogOverlay();
    /** Accumulated lazy fetch hits. Separate from {@link #snapshot} so a partial provider
     *  return from a single name lookup does not discard previously cached lazy results. */
    private final InMemoryCatalogOverlay lazyCache = new InMemoryCatalogOverlay();
    /** Children captured during a lazy provider fetch. Indexed by parent id → kind →
     *  list of child entries. Populated by the lazy resolver path so the bridge can
     *  materialize columns / partitions / etc. when the legacy resolver asks for them. */
    private final java.util.Map<CatalogObjectId,
        java.util.Map<CatalogObjectKind, List<CatalogEntry>>> lazyChildren =
        new java.util.HashMap<CatalogObjectId,
            java.util.Map<CatalogObjectKind, List<CatalogEntry>>>();
    private final ResolverImpl resolver = new ResolverImpl();

    private CatalogSnapshot snapshot;
    private int fetches;

    private CatalogRuntime(Builder b) {
        if (b.provider == null) {
            throw new IllegalArgumentException("CatalogRuntime.provider is required");
        }
        if (b.vendor == null) {
            throw new IllegalArgumentException("CatalogRuntime.vendor is required");
        }
        this.provider = b.provider;
        this.vendor = b.vendor;
        this.loadingMode = b.loadingMode != null ? b.loadingMode : CatalogLoadingMode.EAGER;
        this.ttlMillis = b.ttlMillis;
        this.maxFetchesPerAnalysis = b.maxFetchesPerAnalysis;
        this.includeColumns = b.includeColumns;
        this.includeViews = b.includeViews;
        this.includeRoutines = b.includeRoutines;
        if (b.initialSnapshot != null) {
            this.snapshot = b.initialSnapshot;
        }
    }

    public static Builder builder() {
        return new Builder();
    }

    public CatalogProvider provider() {
        return provider;
    }

    public CatalogOverlay overlay() {
        return overlay;
    }

    public CatalogResolver resolver() {
        return resolver;
    }

    public EDbVendor vendor() {
        return vendor;
    }

    public CatalogLoadingMode loadingMode() {
        return loadingMode;
    }

    /**
     * Current cached snapshot. May be {@code null} until {@link #snapshot(CatalogQuery)}
     * or {@link #refresh(CatalogQuery)} is called.
     */
    public CatalogSnapshot snapshot() {
        return snapshot;
    }

    /**
     * Eagerly request a snapshot for the given query and cache it. EAGER and LAZY modes
     * both hit this method — the difference is whether the query has explicit
     * {@code requestedNames} or not. Clears the lazy fetch cache because previously
     * fetched per-name results may now be stale relative to the new snapshot.
     */
    public CatalogSnapshot snapshot(CatalogQuery query) {
        this.snapshot = provider.snapshot(query);
        this.lazyCache.clear();
        this.lazyChildren.clear();
        return this.snapshot;
    }

    /**
     * Bypass any cached snapshot and force a fresh fetch. Also clears the lazy fetch
     * cache so previously cached per-name lookups do not mask the refreshed data.
     */
    public CatalogSnapshot refresh(CatalogQuery query) {
        this.snapshot = provider.refresh(query);
        this.lazyCache.clear();
        this.lazyChildren.clear();
        return this.snapshot;
    }

    /**
     * Reset per-analysis state: clears the overlay, the lazy fetch cache, and the
     * fetch counter. Snapshot is preserved across batches by default; callers that
     * want a clean slate should also call {@link #refresh(CatalogQuery)}.
     */
    public void resetForNewAnalysis() {
        overlay.clear();
        lazyCache.clear();
        lazyChildren.clear();
        fetches = 0;
    }

    /**
     * Walk overlay → lazy cache → snapshot in resolver-precedence order and return the
     * first matching {@link CatalogEntry}. Used by the legacy {@code TSQLEnv} bridge
     * after a successful {@link CatalogResolver#resolve} to recover the underlying entry
     * (the binding alone does not carry properties+children). Returns {@code null} when
     * no layer holds the entry.
     */
    public CatalogEntry findEntry(CatalogQualifiedName name, CatalogObjectKind kind) {
        if (name == null || kind == null) {
            return null;
        }
        Optional<CatalogEntry> hit = overlay.find(name, kind);
        if (hit.isPresent()) return hit.get();
        hit = lazyCache.find(name, kind);
        if (hit.isPresent()) return hit.get();
        if (snapshot != null) {
            hit = snapshot.find(name, kind);
            if (hit.isPresent()) return hit.get();
        }
        return null;
    }

    /**
     * Children of {@code parent} restricted to {@code kind}, drawing from the lazy
     * children map captured at fetch time and (additively) the cached snapshot. Used by
     * the bridge to materialize columns when the legacy resolver asks for a table/view
     * via {@link gudusoft.gsqlparser.sqlenv.catalog.ICatalogProvider#findObject}; the
     * lazy resolver path preserves children at fetch time so the bridge can attach them
     * to the materialized {@code TSQLTable}.
     */
    public List<CatalogEntry> findChildren(CatalogObjectId parent, CatalogObjectKind kind) {
        if (parent == null || kind == null) {
            return java.util.Collections.emptyList();
        }
        List<CatalogEntry> out = new ArrayList<CatalogEntry>();
        java.util.Map<CatalogObjectKind, List<CatalogEntry>> kids = lazyChildren.get(parent);
        if (kids != null) {
            List<CatalogEntry> fromLazy = kids.get(kind);
            if (fromLazy != null) {
                out.addAll(fromLazy);
            }
        }
        if (snapshot != null) {
            out.addAll(snapshot.children(parent, kind));
        }
        return out;
    }

    @Override
    public void close() {
        try {
            provider.close();
        } catch (CatalogException ignored) {
            // The provider's close() contract permits CatalogException; nothing useful
            // we can do at runtime shutdown beyond swallowing it. Adapters that need
            // teardown diagnostics should surface them via their own channel.
        }
    }

    private final class ResolverImpl implements CatalogResolver {

        @Override
        public CatalogResolutionResult resolve(CatalogContext ctx, CatalogQualifiedName name) {
            if (ctx == null || name == null) {
                return CatalogResolutionResult.miss(Collections.<CatalogDiagnostic>emptyList());
            }
            CatalogObjectKind kind = name.kind();
            // Build candidate fully-qualified names per plan §9.3: defaults fill missing
            // catalog/schema, then the search path widens further. Walking candidates lets
            // overlay/snapshot/lazy/provider all see the same expansion ladder.
            List<CatalogQualifiedName> candidates = expandCandidates(name, ctx);
            // Plan §9.3: a TABLE lookup must also try the snapshot's VIEW / MATERIALIZED_VIEW
            // buckets — the legacy bridge path's ESQLDataObjectType has dotTable but no
            // view type, so a relation reference is always asked as TABLE.
            List<CatalogObjectKind> kinds = widenKinds(kind);

            // Phase 1: overlay.
            for (CatalogQualifiedName candidate : candidates) {
                for (CatalogObjectKind k : kinds) {
                    Optional<CatalogEntry> hit = overlay.find(candidate.withKind(k), k);
                    if (hit.isPresent()) {
                        return CatalogResolutionResult.ok(CatalogBindings.of(hit.get()));
                    }
                }
            }
            // Phase 2: lazy cache (previously fetched on-miss results).
            for (CatalogQualifiedName candidate : candidates) {
                for (CatalogObjectKind k : kinds) {
                    Optional<CatalogEntry> hit = lazyCache.find(candidate.withKind(k), k);
                    if (hit.isPresent()) {
                        return CatalogResolutionResult.ok(CatalogBindings.of(hit.get()));
                    }
                }
            }
            // Phase 3: cached snapshot.
            if (snapshot != null) {
                for (CatalogQualifiedName candidate : candidates) {
                    for (CatalogObjectKind k : kinds) {
                        Optional<CatalogEntry> hit = snapshot.find(candidate.withKind(k), k);
                        if (hit.isPresent()) {
                            return CatalogResolutionResult.ok(CatalogBindings.of(hit.get()));
                        }
                    }
                }
            }
            // Phase 4: provider on-miss fetch (LAZY/AUTO only).
            if (loadingMode != CatalogLoadingMode.EAGER) {
                List<CatalogDiagnostic> diags = new ArrayList<CatalogDiagnostic>();
                if (maxFetchesPerAnalysis > 0 && fetches >= maxFetchesPerAnalysis) {
                    // Count the attempt against the cap so the WARN→ERROR escalation can
                    // actually trigger. Without this increment, fetches stays pinned at
                    // maxFetchesPerAnalysis and the 2× threshold is never reached.
                    fetches++;
                    diags.add(CatalogDiagnostic.builder()
                        .severity(fetches >= 2 * maxFetchesPerAnalysis
                            ? CatalogDiagnosticSeverity.ERROR : CatalogDiagnosticSeverity.WARN)
                        .code(CatalogDiagnosticCode.CATALOG_RUNTIME_FETCH_LIMIT_EXCEEDED)
                        .message("CatalogRuntime fetch cap (" + maxFetchesPerAnalysis
                            + ") exceeded for name=" + name)
                        .name(name)
                        .build());
                    return CatalogResolutionResult.miss(diags);
                }
                CatalogQuery.Builder qb = CatalogQuery.builder()
                    .vendor(vendor)
                    .defaults(DefaultsConfig.builder()
                        .defaultCatalog(ctx.activeCatalog())
                        .defaultSchema(ctx.activeSchema())
                        .build())
                    .searchPath(ctx.searchPath())
                    .addRequestedKind(kind)
                    .includeColumns(includeColumns)
                    .includeViews(includeViews)
                    .includeRoutines(includeRoutines)
                    .loadingMode(loadingMode)
                    .ttlMillis(ttlMillis)
                    .maxFetchesPerAnalysis(maxFetchesPerAnalysis);
                for (CatalogQualifiedName candidate : candidates) {
                    qb.addRequestedName(candidate);
                }
                fetches++;
                CatalogSnapshot fetched;
                try {
                    fetched = provider.snapshot(qb.build());
                } catch (CatalogException ex) {
                    diags.add(CatalogDiagnostic.builder()
                        .severity(CatalogDiagnosticSeverity.WARN)
                        .code(CatalogDiagnosticCode.CATALOG_RUNTIME_FETCH_FAILED)
                        .message("Provider fetch failed for name=" + name + ": " + ex.getMessage())
                        .name(name)
                        .build());
                    return CatalogResolutionResult.miss(diags);
                }
                // Carry over the fetched snapshot's own diagnostics. Provider-level
                // build/load issues (recorded by the snapshot builder, e.g. unsupported-
                // kind INFOs from the static-file readers) are otherwise dropped on the
                // floor before they can reach the bridge's diagnostic sink.
                List<CatalogDiagnostic> fetchedDiags = fetched.diagnostics();
                if (fetchedDiags != null && !fetchedDiags.isEmpty()) {
                    diags.addAll(fetchedDiags);
                }
                // Walk candidates over the fetched snapshot; first hit wins. Any hit is
                // also folded into lazyCache so a subsequent resolve for a different
                // name does not lose this one when the provider returns only its
                // requested entries. Walk widened kinds too so a relation referenced as
                // TABLE matches a VIEW or MATERIALIZED_VIEW entry.
                for (CatalogQualifiedName candidate : candidates) {
                    for (CatalogObjectKind k : kinds) {
                        Optional<CatalogEntry> hit = fetched.find(candidate.withKind(k), k);
                        if (hit.isPresent()) {
                            CatalogEntry entry = hit.get();
                            lazyCache.put(entry);
                            // Preserve children from the fetched snapshot so the bridge can
                            // materialize columns/partitions/etc. when the legacy resolver
                            // asks for the same name. The fetched snapshot itself is
                            // discarded after this resolve, so children must be copied to
                            // an indexed cache that survives.
                            captureLazyChildren(fetched, entry);
                            return CatalogResolutionResult.ok(CatalogBindings.of(entry), diags);
                        }
                    }
                }
                diags.add(CatalogDiagnostic.builder()
                    .severity(CatalogDiagnosticSeverity.INFO)
                    .code(CatalogDiagnosticCode.CATALOG_RUNTIME_PARTIAL_RESULT)
                    .message("Provider returned no entry for name=" + name + ", kind=" + kind)
                    .name(name)
                    .build());
                return CatalogResolutionResult.miss(diags);
            }
            return CatalogResolutionResult.miss(Collections.<CatalogDiagnostic>emptyList());
        }

        /**
         * Build the candidate-name ladder per plan §9.3: original first, then
         * {@code activeSchema.name}, then {@code activeCatalog.activeSchema.name}, then
         * each entry of the search path. Two-segment inputs only get a catalog prefix.
         * Three-or-more segment inputs are treated as already fully qualified.
         */
        private List<CatalogQualifiedName> expandCandidates(CatalogQualifiedName name,
                                                            CatalogContext ctx) {
            List<CatalogQualifiedName> out = new ArrayList<CatalogQualifiedName>();
            out.add(name);
            int segs = name.size();
            if (segs >= 3) {
                return out;
            }
            String catalog = nullIfEmpty(ctx.activeCatalog());
            String schema = nullIfEmpty(ctx.activeSchema());
            IdentifierConfig cfg = ctx.identifierConfig();
            String localRaw = name.raw().get(name.size() - 1);
            CatalogObjectKind kind = name.kind();

            if (segs == 1) {
                if (schema != null) {
                    addCandidate(out, schema + "." + localRaw, kind, cfg);
                    if (catalog != null) {
                        addCandidate(out, catalog + "." + schema + "." + localRaw, kind, cfg);
                    }
                }
                // Schemaless dialects (MySQL, Hive, Teradata, Impala — see
                // TSQLEnv.supportSchema) materialize tables as catalog.table; an
                // unqualified lookup must therefore also try catalog.localRaw with
                // no schema in the middle.
                if (catalog != null && (schema == null || schema.isEmpty())) {
                    addCandidate(out, catalog + "." + localRaw, kind, cfg);
                }
                for (String sp : ctx.searchPath().segments()) {
                    if (sp == null || sp.isEmpty()) continue;
                    addCandidate(out, sp + "." + localRaw, kind, cfg);
                    if (catalog != null) {
                        addCandidate(out, catalog + "." + sp + "." + localRaw, kind, cfg);
                    }
                }
            } else if (segs == 2) {
                String maybeSchema = name.raw().get(0);
                if (catalog != null) {
                    addCandidate(out, catalog + "." + maybeSchema + "." + localRaw, kind, cfg);
                }
            }
            return out;
        }

        private void addCandidate(List<CatalogQualifiedName> out, String raw,
                                  CatalogObjectKind kind, IdentifierConfig cfg) {
            try {
                out.add(CatalogIdentifierPolicy.parse(raw, kind, cfg, vendor));
            } catch (IllegalArgumentException ignored) {
                // Skip malformed candidates — they aren't valid names anyway.
            }
        }

        private String nullIfEmpty(String s) {
            return (s == null || s.isEmpty()) ? null : s;
        }

        /**
         * Plan §9.3 widened kind matrix: a relation reference parsed as
         * {@link CatalogObjectKind#TABLE TABLE} should also see VIEW and
         * MATERIALIZED_VIEW entries, since the legacy bridge path always asks for
         * relations as TABLE. Other kinds are returned as-is.
         */
        private List<CatalogObjectKind> widenKinds(CatalogObjectKind kind) {
            if (kind == CatalogObjectKind.TABLE) {
                List<CatalogObjectKind> out = new ArrayList<CatalogObjectKind>(3);
                out.add(CatalogObjectKind.TABLE);
                out.add(CatalogObjectKind.VIEW);
                out.add(CatalogObjectKind.MATERIALIZED_VIEW);
                return out;
            }
            return Collections.singletonList(kind);
        }
    }

    /**
     * Copy {@code parent}'s children from a transient {@code fetched} snapshot into
     * {@link #lazyChildren} so subsequent {@link #findChildren} calls (and the bridge
     * mapper) can recover columns for a lazy-resolved entry. The bridge currently
     * materializes COLUMN children only; if the runtime grows new child kinds (e.g.
     * parameters on a routine) this list is the place to extend.
     */
    private void captureLazyChildren(CatalogSnapshot fetched, CatalogEntry parent) {
        if (fetched == null || parent == null || parent.id() == null) {
            return;
        }
        CatalogObjectKind[] childKinds = { CatalogObjectKind.COLUMN };
        for (CatalogObjectKind ck : childKinds) {
            List<CatalogEntry> kids = fetched.children(parent.id(), ck);
            if (kids != null && !kids.isEmpty()) {
                java.util.Map<CatalogObjectKind, List<CatalogEntry>> byKind = lazyChildren.get(parent.id());
                if (byKind == null) {
                    byKind = new java.util.HashMap<CatalogObjectKind, List<CatalogEntry>>();
                    lazyChildren.put(parent.id(), byKind);
                }
                byKind.put(ck, new ArrayList<CatalogEntry>(kids));
            }
        }
    }

    public static final class Builder {

        private CatalogProvider provider;
        private EDbVendor vendor;
        private CatalogLoadingMode loadingMode;
        private long ttlMillis;
        private int maxFetchesPerAnalysis;
        private boolean includeColumns = true;
        private boolean includeViews = true;
        private boolean includeRoutines = true;
        private CatalogSnapshot initialSnapshot;

        private Builder() {
        }

        public Builder provider(CatalogProvider v) {
            this.provider = v;
            return this;
        }

        public Builder vendor(EDbVendor v) {
            this.vendor = v;
            return this;
        }

        public Builder loadingMode(CatalogLoadingMode v) {
            this.loadingMode = v;
            return this;
        }

        public Builder ttlMillis(long v) {
            this.ttlMillis = v;
            return this;
        }

        public Builder maxFetchesPerAnalysis(int v) {
            this.maxFetchesPerAnalysis = v;
            return this;
        }

        public Builder includeColumns(boolean v) {
            this.includeColumns = v;
            return this;
        }

        public Builder includeViews(boolean v) {
            this.includeViews = v;
            return this;
        }

        public Builder includeRoutines(boolean v) {
            this.includeRoutines = v;
            return this;
        }

        public Builder initialSnapshot(CatalogSnapshot v) {
            this.initialSnapshot = v;
            return this;
        }

        public CatalogRuntime build() {
            return new CatalogRuntime(this);
        }
    }
}