package gudusoft.gsqlparser.catalog.input.readers;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.catalog.input.CatalogInputException;
import gudusoft.gsqlparser.catalog.input.CatalogInputKind;
import gudusoft.gsqlparser.catalog.input.CatalogInputReader;
import gudusoft.gsqlparser.catalog.input.CatalogInputReaderFactory;
import gudusoft.gsqlparser.catalog.input.CatalogInputSource;
import gudusoft.gsqlparser.catalog.input.CatalogLoadOptions;
import gudusoft.gsqlparser.catalog.input.model.CatalogModel;
import gudusoft.gsqlparser.catalog.input.model.CatalogSourceInfo;
import gudusoft.gsqlparser.catalog.input.model.ColumnModel;
import gudusoft.gsqlparser.catalog.input.model.DefaultsConfig;
import gudusoft.gsqlparser.catalog.input.model.RoutineModel;
import gudusoft.gsqlparser.catalog.input.model.SchemaModel;
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.CatalogObjectKind;
import gudusoft.gsqlparser.util.json.JSON;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.util.List;
import java.util.Map;

/**
 * Reader for {@link CatalogInputKind#SQLFLOW_JSON} sources — the JSON shape the legacy
 * {@code SqlflowSQLEnv} consumes (plan §11 T2.A.3 / second Phase 2.A static-file adapter).
 *
 * <p>Per plan §6 / §11.2 this reader funnels the SqlFlow export shape into the same
 * {@link UnifiedCatalogModel} the JSON-manifest and SQLDep readers use; the eager bridge
 * then materializes a {@code TSQLEnv} byte-for-byte equivalent (for the SELECT smoke
 * surface) to what the legacy {@code SqlflowSQLEnv} would build (parity test in
 * {@code SqlflowLegacyParityTest}). The legacy path remains untouched per plan §8.5.</p>
 *
 * <h3>SqlFlow export schema</h3>
 *
 * <p>The root JSON is the SqlFlow "server" object. Three layouts depending on dialect:</p>
 *
 * <h4>Layout A — catalogs and schemas (Oracle / MSSQL / PostgreSQL / BigQuery / …)</h4>
 * <pre>{@code
 * {
 *   "name": "<server label>",
 *   "dbVendor": "dbvoracle",                  // optional vendor override (informational)
 *   "databases": [
 *     {
 *       "name": "ORCL",
 *       "schemas": [
 *         {
 *           "name": "HR",
 *           "tables":     [{"name": "EMPLOYEES", "type": "TABLE",
 *                            "columns": [{"name": "EMPLOYEE_ID"}, ...]}, ...],
 *           "views":      [{"name": "EMP_DETAILS_VIEW", "type": "VIEW",
 *                            "columns": [...]}, ...],
 *           "packages":   [{"name": "HR_PKG",
 *                            "procedures": [...], "functions": [...], "triggers": [...]}],
 *           "procedures": [{"name": "ADD_EMP"}],
 *           "functions":  [{"name": "GET_SAL"}],
 *           "triggers":   [{"name": "AUDIT_TRG"}]
 *         }
 *       ]
 *     }
 *   ]
 * }
 * }</pre>
 *
 * <h4>Layout B — catalogs only (MySQL / Teradata / Hive / Impala — {@code supportSchema=false})</h4>
 * <pre>{@code
 * {
 *   "databases": [
 *     {
 *       "name": "mydb",
 *       "tables":   [...],
 *       "views":    [...],
 *       "packages": [...],
 *       "procedures": [...],
 *       "functions": [...],
 *       "triggers": [...]
 *     }
 *   ]
 * }
 * }</pre>
 *
 * <h4>Layout C — schemas only (no catalogs)</h4>
 *
 * <p>The legacy {@code SqlflowSQLEnv} guards Layout C with {@code supportCatalog(vendor)},
 * which currently returns {@code true} for every dialect. This branch is therefore dead
 * in production, but a SqlFlow export pinned to that shape is still tolerated here for
 * defensive reasons — the model is built under the {@code TSQLEnv.DEFAULT_DB_NAME}
 * catalog the way the legacy code would.</p>
 *
 * <h3>Mapping rules</h3>
 *
 * <ul>
 *   <li>{@code tables[]} and {@code views[]} are walked in <strong>insertion order</strong>
 *       (tables first, views second), matching the legacy {@code appendTables} method.
 *       For each entry the {@code type} field is examined: if it contains the substring
 *       {@code "view"} (case-insensitive ASCII) the entry is recorded as a
 *       {@link ViewModel}; otherwise as a {@link TableModel}. Array membership is
 *       informational only — the {@code type} field wins.</li>
 *   <li>{@code procedures[]} → {@link RoutineModel} with {@link CatalogObjectKind#PROCEDURE}.</li>
 *   <li>{@code functions[]} → {@link RoutineModel} with {@link CatalogObjectKind#FUNCTION}.</li>
 *   <li>{@code packages[]} → {@link RoutineModel} with {@link CatalogObjectKind#PACKAGE}.
 *       The package's nested {@code procedures[]} / {@code functions[]} / {@code triggers[]}
 *       have no Phase 1 SPI representation and are dropped (the legacy loader binds them
 *       inside the {@code TSQLEnv}'s package node — that nesting belongs in a follow-up
 *       extension to {@link RoutineModel}). The empty package shell is still present so
 *       callers that resolve {@code SCHEMA.PKG} succeed.</li>
 *   <li>{@code triggers[]} (top-level or package-nested) — there is no Phase 1
 *       {@code RoutineModel.kind} mapping for triggers and the {@code SQLEnvCatalogLoader}
 *       does not synthesize them. They are dropped silently. SELECT-only smoke tests are
 *       unaffected (triggers do not participate in {@code TSQLResolver2} formatter
 *       output for {@code SELECT}/{@code JOIN} queries); future Phase 2 work that needs
 *       trigger metadata should extend the SPI.</li>
 *   <li>The {@code dbVendor} JSON field is informational. The reader honors
 *       {@link CatalogLoadOptions#vendor()} as the single source of truth (defaulting to
 *       {@link EDbVendor#dbvoracle} when the option is unset, matching SqlFlow's historical
 *       Oracle-leaning default). A mismatch between the JSON's {@code dbVendor} and the
 *       option's vendor is not flagged here — {@code CatalogModelValidator} catches the
 *       harmful identifier-bypass cases through its normalize round-trip.</li>
 * </ul>
 *
 * <h3>Dispatch</h3>
 *
 * <p>The reader is dispatch-by-{@link CatalogInputKind#SQLFLOW_JSON} only — callers must
 * tag the source explicitly. There is no auto-claim heuristic on JSON structure (the
 * SqlFlow shape and the SQLDep shape both top-level objects with a {@code databases}
 * array would collide); the explicit kind tag is the contract.</p>
 *
 * <h3>Dotted-identifier defense</h3>
 *
 * <p>Identifier names with embedded dots are wrapped with the vendor's quote characters
 * before storage. Without the wrap, {@code ModelBackedCatalogProvider} later concatenates
 * segments with {@code .} into qualified strings (e.g. {@code catalog.schema.A.B}) that
 * {@code CatalogIdentifierPolicy.parse} would split into four segments instead of three,
 * and the runtime would never resolve the SQL reference {@code "A.B"}. Mirrors the
 * {@code SQLDepSQLEnv} defense and the same wrapping policy as {@link SqldepCatalogInputReader}.
 * The vendor → delimiter-pair table is duplicated from {@code TSQLEnv.delimitedChar} to
 * keep {@code catalog/**} free of a compile-time dependency on the {@code sqlenv} runtime
 * layer.</p>
 */
public final class SqlflowCatalogInputReader implements CatalogInputReader {

    public SqlflowCatalogInputReader() {
    }

    @Override
    public CatalogInputKind kind() {
        return CatalogInputKind.SQLFLOW_JSON;
    }

    @Override
    public boolean supports(CatalogInputSource source, CatalogLoadOptions options) {
        if (source == null || source.inMemoryModel() != null) {
            return false;
        }
        return source.declaredKind() == CatalogInputKind.SQLFLOW_JSON;
    }

    @Override
    public UnifiedCatalogModel read(CatalogInputSource source, CatalogLoadOptions options)
            throws CatalogInputException {
        if (source == null) {
            throw new CatalogInputException("SqlflowCatalogInputReader: source is required");
        }
        final EDbVendor vendor = (options != null && options.vendor() != null)
            ? options.vendor()
            : EDbVendor.dbvoracle;  // SqlFlow exports historically default to Oracle.

        long start = System.currentTimeMillis();
        String text = readAll(source);
        Object parsed;
        try {
            parsed = JSON.parseObject(text);
        } catch (RuntimeException ex) {
            throw new CatalogInputException(
                "Failed to parse SqlFlow JSON from " + source.name() + ": " + ex.getMessage(),
                ex);
        }
        if (!(parsed instanceof Map)) {
            throw new CatalogInputException(
                "SqlFlow root must be an object (got " + typeOf(parsed) + ")");
        }
        @SuppressWarnings("unchecked")
        Map<Object, Object> root = (Map<Object, Object>) parsed;

        try {
            UnifiedCatalogModel.Builder mb = UnifiedCatalogModel.builder().vendor(vendor);
            mb.sourceInfo(buildSourceInfo(source, start));

            // Track defaults in one place so we can layer (options → JSON server name)
            // without reading the model builder back. Writes to mb.defaults(...) happen
            // exactly once at the end.
            DefaultsConfig.Builder defaultsBuilder = DefaultsConfig.builder();
            boolean anyDefault = seedOptionDefaults(defaultsBuilder, options);
            boolean serverNameSeeded = options != null && options.defaultServer() != null
                && !options.defaultServer().isEmpty();

            // The canonical SqlFlow ingester export wraps one or more server objects in
            // a {createdBy, servers:[…]} envelope (see TJSONSQLEnvParser.getJSONSQLEnv).
            // Each entry of `servers[]` is exactly the shape SqlflowSQLEnv directly
            // accepts. When the envelope is present, walk every server and merge its
            // catalogs into a single UnifiedCatalogModel — the legacy parser produces
            // one TSQLEnv per server, but the SPI's single-model contract folds them
            // together. The bare server-object-as-root form (what SqlflowSQLEnv was
            // historically constructed with directly) is also tolerated so callers that
            // already unwrapped the envelope themselves keep working.
            //
            // Catalog-name dedup (codex P2 round 2): two servers that ship a database
            // with the same name produce duplicate top-level CatalogModels, which the
            // CatalogModelValidator flags as an ERROR. Track seen names and reject the
            // ambiguous case explicitly with a clear message — the legacy parser builds
            // one TSQLEnv per server so the conflict is naturally avoided; here it must
            // be surfaced for the single-model contract.
            java.util.HashSet<String> catalogNames = new java.util.HashSet<String>();
            Object servers = root.get("servers");
            if (servers != null) {
                if (!(servers instanceof List)) {
                    throw new CatalogInputException(
                        "SqlFlow field 'servers' must be an array (got "
                            + typeOf(servers) + ")");
                }
                for (Object sobj : (List<?>) servers) {
                    if (!(sobj instanceof Map)) {
                        throw new CatalogInputException(
                            "SqlFlow servers[i] must be an object (got "
                                + typeOf(sobj) + ")");
                    }
                    @SuppressWarnings("unchecked")
                    Map<Object, Object> server = (Map<Object, Object>) sobj;
                    // Codex P2: preserve the server's `name` as DefaultsConfig.defaultServer
                    // when the caller did not supply one, mirroring the legacy
                    // SqlflowSQLEnv.initSQLEnv() call to setDefaultServerName(server.get("name")).
                    // First server wins to avoid clobbering when multiple servers ship in
                    // one envelope (the legacy code makes one TSQLEnv per server, so the
                    // analogue here is "first server's name represents this model").
                    if (!serverNameSeeded) {
                        String serverName = asString(server, "name", false);
                        if (serverName != null && !serverName.isEmpty()) {
                            defaultsBuilder.defaultServer(serverName);
                            anyDefault = true;
                            serverNameSeeded = true;
                        }
                    }
                    // Codex P2 round 2: layout decision is per-server. The legacy parser
                    // constructs each SqlflowSQLEnv with the server's own dbVendor; if the
                    // JSON has dbVendor=dbvmysql but options.vendor=dbvoracle, the legacy
                    // walks tables from databases[] directly while our previous single
                    // precomputed layout would look for schemas[] and silently drop tables.
                    // Honor the per-server vendor for supportsSchema only — the model's own
                    // vendor stays options.vendor (the SPI single-model contract precludes
                    // mixed-vendor models).
                    EDbVendor serverVendor = readServerVendor(server, vendor);
                    boolean serverSupportsSchema = supportSchema(serverVendor);
                    parseServerBody(server, vendor, serverSupportsSchema, mb, catalogNames);
                }
            } else {
                // No envelope — root IS the server body (legacy direct-construction shape).
                // Try the same bare-root server-name capture as the envelope path so a
                // SqlFlow JSON that was hand-edited down to a single server also preserves
                // its `name` field.
                if (!serverNameSeeded) {
                    String serverName = asString(root, "name", false);
                    if (serverName != null && !serverName.isEmpty()) {
                        defaultsBuilder.defaultServer(serverName);
                        anyDefault = true;
                    }
                }
                EDbVendor serverVendor = readServerVendor(root, vendor);
                boolean serverSupportsSchema = supportSchema(serverVendor);
                parseServerBody(root, vendor, serverSupportsSchema, mb, catalogNames);
            }

            if (anyDefault) {
                mb.defaults(defaultsBuilder.build());
            }
            return mb.build();
        } catch (IllegalArgumentException ex) {
            // Model builders enforce structural invariants (non-empty names, etc.) by
            // throwing IllegalArgumentException. Surface those through the reader's
            // checked-exception channel so callers using try/catch on
            // CatalogInputException don't get caught off guard. Same posture as the
            // SQLDep reader.
            throw new CatalogInputException(
                "Malformed SqlFlow export from " + source.name() + ": " + ex.getMessage(), ex);
        }
    }

    /**
     * Walk one SqlFlow server body — the JSON object that {@code SqlflowSQLEnv} would be
     * constructed with directly. Same shape regardless of whether it came from the outer
     * {@code servers[]} envelope or from a bare server-as-root JSON.
     *
     * <p>{@code seenCatalogNames} is the running set across server walks so a multi-server
     * envelope can reject the ambiguous case (two servers ship a database with the same
     * name) with a clear {@link CatalogInputException}, rather than producing a model
     * that fails the validator's duplicate-catalog ERROR check downstream. Pass an empty
     * set when called for a single-server body (validator still catches in-server
     * duplicates).</p>
     */
    private void parseServerBody(Map<Object, Object> body, EDbVendor vendor,
                                 boolean supportsSchema, UnifiedCatalogModel.Builder mb,
                                 java.util.Set<String> seenCatalogNames)
            throws CatalogInputException {
        Object databases = body.get("databases");
        if (databases != null) {
            if (!(databases instanceof List)) {
                throw new CatalogInputException(
                    "SqlFlow field 'databases' must be an array (got "
                        + typeOf(databases) + ")");
            }
            for (Object dobj : (List<?>) databases) {
                if (!(dobj instanceof Map)) {
                    throw new CatalogInputException(
                        "SqlFlow databases[i] must be an object (got "
                            + typeOf(dobj) + ")");
                }
                @SuppressWarnings("unchecked")
                Map<Object, Object> dmap = (Map<Object, Object>) dobj;
                CatalogModel catalog = parseDatabase(dmap, vendor, supportsSchema);
                if (catalog != null) {
                    if (!seenCatalogNames.add(catalog.name())) {
                        throw new CatalogInputException(
                            "SqlFlow envelope contains duplicate catalog name '"
                                + catalog.name() + "' across servers; the SPI's single-model "
                                + "contract cannot fold ambiguous catalogs. Split the export "
                                + "into one file per server, or rename the duplicate "
                                + "database before reading.");
                    }
                    mb.addCatalog(catalog);
                }
            }
            return;
        }
        // Layout C — schemas-at-top. Defensive only; supportCatalog(vendor) returns
        // true for every dialect today, so this branch is unreachable in production
        // but the legacy SqlflowSQLEnv carries it.
        Object schemas = body.get("schemas");
        if (schemas != null) {
            if (!(schemas instanceof List)) {
                throw new CatalogInputException(
                    "SqlFlow field 'schemas' must be an array (got "
                        + typeOf(schemas) + ")");
            }
            CatalogModel catalog = parseSchemasAtTop((List<?>) schemas, vendor);
            if (catalog != null) {
                if (!seenCatalogNames.add(catalog.name())) {
                    throw new CatalogInputException(
                        "SqlFlow envelope contains duplicate catalog name '"
                            + catalog.name() + "' across servers");
                }
                mb.addCatalog(catalog);
            }
        }
    }

    /**
     * Resolve the per-server vendor used for the layout decision. The legacy parser
     * constructs each {@code SqlflowSQLEnv} with the server's own {@code dbVendor} when
     * present; we honor that for {@link #supportSchema} so a MySQL/Hive server in an
     * envelope routed under an Oracle/PostgreSQL options.vendor doesn't silently drop
     * tables that hang directly off {@code databases[i]}. Returns the supplied
     * {@code defaultVendor} when the JSON omits or supplies an unparsable {@code dbVendor}.
     */
    private static EDbVendor readServerVendor(Map<Object, Object> server,
                                              EDbVendor defaultVendor) {
        Object v = server.get("dbVendor");
        if (!(v instanceof String)) {
            return defaultVendor;
        }
        String s = ((String) v).trim();
        if (s.isEmpty()) {
            return defaultVendor;
        }
        try {
            return EDbVendor.valueOf(s);
        } catch (IllegalArgumentException ex) {
            // Unrecognized vendor name in the JSON — fall back to the option vendor;
            // CatalogModelValidator's identifier-bypass check will surface any
            // identifier-fold mismatch downstream. Don't throw: the legacy parser
            // would have done the same EDbVendor.valueOf() and let the caller see the
            // resulting IllegalArgumentException; we choose tolerance over failure.
            return defaultVendor;
        }
    }


    // ---------- catalog / schema / table parsing ----------

    /**
     * Walk one {@code databases[i]} entry into a {@link CatalogModel}. Two sub-shapes:
     *
     * <ul>
     *   <li>{@code supportsSchema}: {@code databases[i].schemas[j]} arrays carry the
     *       per-schema object listings.</li>
     *   <li>{@code !supportsSchema}: object listings (tables / views / etc.) hang off
     *       {@code databases[i]} directly; one synthetic schema named
     *       {@link gudusoft.gsqlparser.sqlenv.TSQLEnv#DEFAULT_SCHEMA_NAME} is produced so
     *       the loader's catalog-tree walk has somewhere to plant them. The legacy
     *       {@code SqlflowSQLEnv} does the same routing in
     *       {@code sqlCatalog.getSchema(TSQLEnv.DEFAULT_SCHEMA_NAME, true)}.</li>
     * </ul>
     */
    private CatalogModel parseDatabase(Map<Object, Object> dmap, EDbVendor vendor,
                                       boolean supportsSchema)
            throws CatalogInputException {
        String dbName = wrapDottedName(asString(dmap, "name", true), vendor);
        CatalogModel.Builder cb = CatalogModel.builder().name(dbName);
        if (supportsSchema) {
            Object schemas = dmap.get("schemas");
            if (schemas != null) {
                if (!(schemas instanceof List)) {
                    throw new CatalogInputException(
                        "SqlFlow database '" + dbName + "' field 'schemas' must be an array (got "
                            + typeOf(schemas) + ")");
                }
                for (Object sobj : (List<?>) schemas) {
                    if (!(sobj instanceof Map)) {
                        throw new CatalogInputException(
                            "SqlFlow database '" + dbName + "' schemas[i] must be an object (got "
                                + typeOf(sobj) + ")");
                    }
                    @SuppressWarnings("unchecked")
                    Map<Object, Object> smap = (Map<Object, Object>) sobj;
                    String schemaName = wrapDottedName(asString(smap, "name", true), vendor);
                    cb.addSchema(parseSchemaBody(smap, schemaName, vendor));
                }
            }
        } else {
            // Catalogs-only dialect: synthesize one default-named schema bucket.
            cb.addSchema(parseSchemaBody(dmap, "", vendor));
        }
        return cb.build();
    }

    /**
     * Defensive Layout-C fallback. Builds a single synthetic catalog whose name is
     * {@code TSQLEnv.DEFAULT_DB_NAME} ({@code "$_-_$"}) — matching the legacy
     * {@code sqlCatalog = getSQLCatalog(TSQLEnv.DEFAULT_DB_NAME, true)} call. Each
     * top-level schema becomes a {@link SchemaModel} under it.
     */
    private CatalogModel parseSchemasAtTop(List<?> schemas, EDbVendor vendor)
            throws CatalogInputException {
        CatalogModel.Builder cb = CatalogModel.builder()
            .name(gudusoft.gsqlparser.sqlenv.TSQLEnv.DEFAULT_DB_NAME);
        for (Object sobj : schemas) {
            if (!(sobj instanceof Map)) {
                throw new CatalogInputException(
                    "SqlFlow top-level schemas[i] must be an object (got " + typeOf(sobj) + ")");
            }
            @SuppressWarnings("unchecked")
            Map<Object, Object> smap = (Map<Object, Object>) sobj;
            String schemaName = wrapDottedName(asString(smap, "name", true), vendor);
            cb.addSchema(parseSchemaBody(smap, schemaName, vendor));
        }
        return cb.build();
    }

    /**
     * Walk a schema-body map (either a {@code schemas[i]} entry, or a {@code databases[i]}
     * entry in the schema-less case) into a {@link SchemaModel}. Handles tables, views,
     * top-level packages, top-level procedures/functions, and silently skips triggers and
     * package-nested children (see class Javadoc for the rationale).
     *
     * <p>The {@code schemaName} is wrapped upstream so the dotted-identifier defense covers
     * the schema layer too.</p>
     */
    private SchemaModel parseSchemaBody(Map<Object, Object> body, String schemaName,
                                        EDbVendor vendor)
            throws CatalogInputException {
        SchemaModel.Builder sb = SchemaModel.builder().name(schemaName);

        // Tables and views are walked in legacy order: tables first, then views. The
        // SqlflowSQLEnv code combines them into a single list via addAll(), so the
        // produced TSQLEnv ordering is tables-then-views — preserve that here so the
        // parity test sees the same insertion order.
        Object tables = body.get("tables");
        if (tables != null) {
            if (!(tables instanceof List)) {
                throw new CatalogInputException(
                    "SqlFlow field 'tables' must be an array (got " + typeOf(tables) + ")");
            }
            for (Object tobj : (List<?>) tables) {
                if (!(tobj instanceof Map)) {
                    throw new CatalogInputException(
                        "SqlFlow tables[i] must be an object (got " + typeOf(tobj) + ")");
                }
                @SuppressWarnings("unchecked")
                Map<Object, Object> tmap = (Map<Object, Object>) tobj;
                addTableOrView(sb, tmap, vendor);
            }
        }
        Object views = body.get("views");
        if (views != null) {
            if (!(views instanceof List)) {
                throw new CatalogInputException(
                    "SqlFlow field 'views' must be an array (got " + typeOf(views) + ")");
            }
            for (Object vobj : (List<?>) views) {
                if (!(vobj instanceof Map)) {
                    throw new CatalogInputException(
                        "SqlFlow views[i] must be an object (got " + typeOf(vobj) + ")");
                }
                @SuppressWarnings("unchecked")
                Map<Object, Object> vmap = (Map<Object, Object>) vobj;
                addTableOrView(sb, vmap, vendor);
            }
        }

        // Top-level routines.
        Object procedures = body.get("procedures");
        if (procedures != null) {
            for (Map<Object, Object> rmap : asMapList(procedures, "procedures")) {
                String name = wrapDottedName(asString(rmap, "name", true), vendor);
                sb.addRoutine(RoutineModel.builder()
                    .name(name)
                    .kind(CatalogObjectKind.PROCEDURE)
                    .build());
            }
        }
        Object functions = body.get("functions");
        if (functions != null) {
            for (Map<Object, Object> rmap : asMapList(functions, "functions")) {
                String name = wrapDottedName(asString(rmap, "name", true), vendor);
                sb.addRoutine(RoutineModel.builder()
                    .name(name)
                    .kind(CatalogObjectKind.FUNCTION)
                    .build());
            }
        }
        Object packages = body.get("packages");
        if (packages != null) {
            for (Map<Object, Object> pmap : asMapList(packages, "packages")) {
                // Package-nested procedures / functions / triggers are intentionally
                // dropped: RoutineModel does not yet model package-nested children, and
                // the SQLEnvCatalogLoader has no path to bind them. The empty package
                // shell still resolves SCHEMA.PKG references.
                String name = wrapDottedName(asString(pmap, "name", true), vendor);
                sb.addRoutine(RoutineModel.builder()
                    .name(name)
                    .kind(CatalogObjectKind.PACKAGE)
                    .build());
            }
        }

        // Top-level triggers — see class Javadoc; silently dropped because no Phase 1
        // RoutineModel.kind exists for triggers and the loader does not synthesize them.
        // Walk the array shape just to fail cleanly on a malformed export.
        Object triggers = body.get("triggers");
        if (triggers != null && !(triggers instanceof List)) {
            throw new CatalogInputException(
                "SqlFlow field 'triggers' must be an array (got " + typeOf(triggers) + ")");
        }

        return sb.build();
    }

    private void addTableOrView(SchemaModel.Builder sb, Map<Object, Object> entry,
                                EDbVendor vendor)
            throws CatalogInputException {
        String objName = wrapDottedName(asString(entry, "name", true), vendor);
        String type = asString(entry, "type", false);
        boolean isView = typeIndicatesView(type);
        // Column names are NOT wrapped: ModelBackedCatalogProvider builds the qualified
        // column form as `<tableQ>.<column>` where <tableQ> already has any required
        // quoting. A column name with an embedded dot would still be ambiguous downstream
        // — but the legacy loader has the same gap, so parity is preserved without the
        // extra wrapping here. (Same posture as SqldepCatalogInputReader.)
        List<ColumnModel> cols = parseColumns(entry, objName);
        if (isView) {
            ViewModel.Builder vb = ViewModel.builder().name(objName);
            for (ColumnModel c : cols) {
                vb.addColumn(c);
            }
            sb.addView(vb.build());
        } else {
            TableModel.Builder tb = TableModel.builder().name(objName);
            for (ColumnModel c : cols) {
                tb.addColumn(c);
            }
            sb.addTable(tb.build());
        }
    }

    /**
     * Case-insensitive ASCII substring check for {@code "view"} in the supplied {@code type}
     * field — matches the legacy {@code SqlflowSQLEnv.appendTables()}'s
     * {@code type.toLowerCase().indexOf("view") != -1} test without using
     * {@link String#toLowerCase()}, which the {@code forbidden-apis} plugin bans inside
     * {@code catalog/**}. {@link Character#toLowerCase(char)} is locale-insensitive for
     * ASCII characters and is permitted.
     */
    private static boolean typeIndicatesView(String type) {
        if (type == null) {
            return false;
        }
        final int n = type.length();
        for (int i = 0; i + 4 <= n; i++) {
            if (Character.toLowerCase(type.charAt(i)) == 'v'
                && Character.toLowerCase(type.charAt(i + 1)) == 'i'
                && Character.toLowerCase(type.charAt(i + 2)) == 'e'
                && Character.toLowerCase(type.charAt(i + 3)) == 'w') {
                return true;
            }
        }
        return false;
    }

    private static List<ColumnModel> parseColumns(Map<Object, Object> tmap, String tableName)
            throws CatalogInputException {
        Object columns = tmap.get("columns");
        java.util.ArrayList<ColumnModel> out = new java.util.ArrayList<ColumnModel>();
        if (columns == null) {
            return out;
        }
        if (!(columns instanceof List)) {
            throw new CatalogInputException(
                "SqlFlow object '" + tableName + "' field 'columns' must be an array (got "
                    + typeOf(columns) + ")");
        }
        for (Object cobj : (List<?>) columns) {
            if (!(cobj instanceof Map)) {
                throw new CatalogInputException(
                    "SqlFlow object '" + tableName + "' columns[i] must be an object (got "
                        + typeOf(cobj) + ")");
            }
            @SuppressWarnings("unchecked")
            Map<Object, Object> cmap = (Map<Object, Object>) cobj;
            String colName = asString(cmap, "name", true);
            ColumnModel.Builder cb = ColumnModel.builder().name(colName);
            String dt = asString(cmap, "dataType", false);
            if (dt == null) dt = asString(cmap, "type", false);
            if (dt != null) cb.dataType(dt);
            out.add(cb.build());
        }
        return out;
    }

    /**
     * Cast and walk a JSON value as a list of object maps. Throws {@link CatalogInputException}
     * with a labelled error when the structure is wrong. Used for {@code procedures} /
     * {@code functions} / {@code packages}.
     */
    @SuppressWarnings("unchecked")
    private static List<Map<Object, Object>> asMapList(Object value, String fieldLabel)
            throws CatalogInputException {
        if (!(value instanceof List)) {
            throw new CatalogInputException(
                "SqlFlow field '" + fieldLabel + "' must be an array (got " + typeOf(value) + ")");
        }
        List<?> raw = (List<?>) value;
        java.util.ArrayList<Map<Object, Object>> out = new java.util.ArrayList<Map<Object, Object>>(raw.size());
        for (Object o : raw) {
            if (!(o instanceof Map)) {
                throw new CatalogInputException(
                    "SqlFlow field '" + fieldLabel + "[i]' must be an object (got "
                        + typeOf(o) + ")");
            }
            out.add((Map<Object, Object>) o);
        }
        return out;
    }

    // ---------- dotted-identifier wrapping ----------

    /**
     * Wrap a raw identifier with the vendor's open/close quote chars when it contains a
     * dot. Same defense as {@link SqldepCatalogInputReader#wrapDottedName} — see that
     * method for the rationale. Vendor → delimiter pair table is duplicated here to keep
     * {@code catalog/**} free of a compile-time dependency on {@code sqlenv/**}; if a
     * dialect adds a new {@code TSQLEnv.delimitedChar} entry, mirror it in both places.
     */
    private static String wrapDottedName(String raw, EDbVendor vendor) {
        if (raw == null || raw.isEmpty() || raw.indexOf('.') == -1) {
            return raw;
        }
        char open = vendorOpenDelimiter(vendor);
        char close = vendorCloseDelimiter(vendor);
        if (raw.length() >= 2
            && raw.charAt(0) == open
            && raw.charAt(raw.length() - 1) == close) {
            return raw;
        }
        return open + raw + close;
    }

    private static char vendorOpenDelimiter(EDbVendor v) {
        if (v == null) return '"';
        switch (v) {
            case dbvmssql:
            case dbvazuresql:
                return '[';
            case dbvathena:
            case dbvmysql:
            case dbvbigquery:
            case dbvcouchbase:
            case dbvhive:
            case dbvimpala:
            case dbvdatabricks:
                return '`';
            case dbvdax:
                return '\'';
            default:
                return '"';
        }
    }

    private static char vendorCloseDelimiter(EDbVendor v) {
        if (v == null) return '"';
        switch (v) {
            case dbvmssql:
            case dbvazuresql:
                return ']';
            default:
                return vendorOpenDelimiter(v);
        }
    }

    // ---------- vendor schema-support ----------

    /**
     * Mirror of {@link gudusoft.gsqlparser.sqlenv.TSQLEnv#supportSchema(EDbVendor)}. Inlined
     * here so {@code catalog/**} does not import from {@code sqlenv/**} on the read path
     * (the loader path already uses {@code TSQLEnv.supportSchema} for the symmetric
     * decision). Kept in sync — if a new schema-less dialect lands in the legacy method
     * (unlikely; the list has been stable for years), reflect it here too.
     */
    private static boolean supportSchema(EDbVendor vendor) {
        return !(vendor == EDbVendor.dbvmysql
            || vendor == EDbVendor.dbvteradata
            || vendor == EDbVendor.dbvhive
            || vendor == EDbVendor.dbvimpala);
    }

    // ---------- defaults / source info ----------

    /**
     * Seed the supplied {@link DefaultsConfig.Builder} from {@code options} non-empty
     * fields. Returns {@code true} if at least one default was contributed, so the caller
     * knows whether to call {@link UnifiedCatalogModel.Builder#defaults} at the end. The
     * server-name capture from JSON layers on top of this seed in the read path.
     */
    private static boolean seedOptionDefaults(DefaultsConfig.Builder db,
                                              CatalogLoadOptions options) {
        if (options == null) return false;
        boolean any = false;
        if (options.defaultCatalog() != null && !options.defaultCatalog().isEmpty()) {
            db.defaultCatalog(options.defaultCatalog());
            any = true;
        }
        if (options.defaultSchema() != null && !options.defaultSchema().isEmpty()) {
            db.defaultSchema(options.defaultSchema());
            any = true;
        }
        if (options.defaultServer() != null && !options.defaultServer().isEmpty()) {
            db.defaultServer(options.defaultServer());
            any = true;
        }
        return any;
    }

    private static CatalogSourceInfo buildSourceInfo(CatalogInputSource source, long startMillis) {
        return CatalogSourceInfo.builder()
            .kind(CatalogInputKind.SQLFLOW_JSON)
            .name(source.name() != null ? source.name() : "<sqlflow>")
            .readMillis(System.currentTimeMillis() - startMillis)
            .build();
    }

    // ---------- input → string ----------

    private static String readAll(CatalogInputSource source) throws CatalogInputException {
        try {
            if (source.inMemoryModel() != null) {
                throw new CatalogInputException(
                    "SqlflowCatalogInputReader cannot read in-memory model sources");
            }
            if (source.path() != null) {
                byte[] b = Files.readAllBytes(source.path());
                return new String(b, StandardCharsets.UTF_8);
            }
            byte[] sourceBytes = source.bytes();
            if (sourceBytes != null) {
                return new String(sourceBytes, StandardCharsets.UTF_8);
            }
            if (source.url() != null) {
                try (InputStream in = source.url().openStream();
                     Reader r = new InputStreamReader(in, StandardCharsets.UTF_8)) {
                    return drain(r);
                }
            }
            if (source.reader() != null) {
                return drain(source.reader());
            }
            throw new CatalogInputException(
                "SqlflowCatalogInputReader: source has no readable backing");
        } catch (IOException io) {
            throw new CatalogInputException(
                "Failed to read SqlFlow export from " + source.name() + ": " + io.getMessage(),
                io);
        }
    }

    private static String drain(Reader r) throws IOException {
        BufferedReader br = (r instanceof BufferedReader) ? (BufferedReader) r : new BufferedReader(r);
        StringBuilder sb = new StringBuilder();
        char[] buf = new char[4096];
        int n;
        while ((n = br.read(buf)) > 0) {
            sb.append(buf, 0, n);
        }
        return sb.toString();
    }

    // ---------- shared field accessors ----------

    private static String asString(Map<Object, Object> obj, String key, boolean required)
            throws CatalogInputException {
        Object v = obj.get(key);
        if (v == null) {
            if (required) {
                throw new CatalogInputException(
                    "SqlFlow entry missing required field '" + key + "'");
            }
            return null;
        }
        return v instanceof String ? (String) v : v.toString();
    }

    private static String typeOf(Object o) {
        return o == null ? "null" : o.getClass().getSimpleName();
    }

    /** ServiceLoader-discoverable factory. Plan §13.1. */
    public static final class Factory implements CatalogInputReaderFactory {

        public Factory() {
            // Required no-arg constructor for ServiceLoader.
        }

        @Override
        public CatalogInputKind kind() {
            return CatalogInputKind.SQLFLOW_JSON;
        }

        @Override
        public CatalogInputReader create() {
            return new SqlflowCatalogInputReader();
        }
    }
}