package gudusoft.gsqlparser.sqlenv.parser.sharded;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
import gudusoft.gsqlparser.sqlenv.IdentifierInputForm;
import gudusoft.gsqlparser.sqlenv.IdentifierProfile;
import gudusoft.gsqlparser.sqlenv.TSQLEnv;
import gudusoft.gsqlparser.sqlenv.TSQLCatalog;
import gudusoft.gsqlparser.sqlenv.TSQLSchema;
import gudusoft.gsqlparser.sqlenv.TSQLTable;
import gudusoft.gsqlparser.util.SQLUtil;
import gudusoft.gsqlparser.util.json.JSON;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * SQLEnv implementation for sqlflow-sharded format.
 * <p>
 * Parses catalog.json files from the sharded export format, loading metadata
 * including tables, views, columns, and synonyms. Supports different database
 * topologies based on the manifest configuration.
 * <p>
 * The sharded format structure:
 * <pre>
 * manifest.json
 * catalog/&lt;shardId&gt;.catalog.json
 * source/&lt;shardId&gt;.source.jsonl
 * index/&lt;shardId&gt;.source.idx
 * </pre>
 */
@SuppressWarnings("rawtypes")
public class SqlflowShardedSQLEnv extends TSQLEnv {

    private Map server;
    private String baseDir;
    private Boolean init = false;
    private IdentifierInputForm nameForm = IdentifierInputForm.SQL_LEXICAL;

    /**
     * Constructor with explicit base directory for catalog files.
     *
     * @param defaultServer   default server name
     * @param defaultDatabase default database name
     * @param defaultSchema   default schema name
     * @param dbVendor        database vendor
     * @param server          server configuration from manifest
     * @param baseDir         base directory containing catalog/, source/, index/ subdirectories
     */
    public SqlflowShardedSQLEnv(String defaultServer, String defaultDatabase, String defaultSchema,
                                 EDbVendor dbVendor, Map server, String baseDir) {
        this(defaultServer, defaultDatabase, defaultSchema, dbVendor, server, baseDir,
                IdentifierInputForm.SQL_LEXICAL);
    }

    /**
     * @param nameForm how the manifest and its catalog shards write object names: as SQL text
     *                 ({@code "Amount"}, the exporter's form and the default) or as the names the
     *                 database catalog holds ({@code Amount}). Declared once, at the manifest
     *                 root, as {@code nameForm}. See {@link TSQLEnv#toLexical}.
     */
    public SqlflowShardedSQLEnv(String defaultServer, String defaultDatabase, String defaultSchema,
                                 EDbVendor dbVendor, Map server, String baseDir, IdentifierInputForm nameForm) {
        this(defaultServer, defaultDatabase, defaultSchema, dbVendor, server, baseDir, nameForm, null);
    }

    /**
     * Same input, analyzed under an explicit identifier configuration.
     *
     * <p>The metadata this environment registers is indexed with {@code identifierProfile}, so the
     * analysis that reads it back asks the catalog the same question it was filled with. A profile
     * belonging to another vendor is not applied ({@link IdentifierProfile#forVendorOr}); null means
     * the vendor default.
     */
    public SqlflowShardedSQLEnv(String defaultServer, String defaultDatabase, String defaultSchema,
                                 EDbVendor dbVendor, Map server, String baseDir, IdentifierInputForm nameForm,
                                 IdentifierProfile identifierProfile) {
        super(dbVendor, IdentifierProfile.forVendorOr(identifierProfile, dbVendor));
        initDefaults(defaultServer, defaultDatabase, defaultSchema);
        this.server = server;
        this.baseDir = baseDir;
        this.nameForm = nameForm == null ? IdentifierInputForm.SQL_LEXICAL : nameForm;
        initSQLEnv();
    }

    /** The spelling to register for a name from this input, per its declared {@link #nameForm}. */
    private String lex(ESQLDataObjectType type, String name) {
        return TSQLEnv.toLexical(getDBVendor(), type, name, nameForm);
    }

    private void initDefaults(String defaultServer, String defaultDatabase, String defaultSchema) {
        if (!SQLUtil.isEmpty(defaultServer) && !defaultServer.equals(TSQLEnv.DEFAULT_SERVER_NAME)) {
            setDefaultServerName(defaultServer);
        }
        if (!SQLUtil.isEmpty(defaultDatabase) && !defaultDatabase.equals(TSQLEnv.DEFAULT_DB_NAME)) {
            setDefaultCatalogName(defaultDatabase);
        }
        if (!SQLUtil.isEmpty(defaultSchema) && !defaultSchema.equals(TSQLEnv.DEFAULT_SCHEMA_NAME)) {
            setDefaultSchemaName(defaultSchema);
        }
    }

    @Override
    public void initSQLEnv() {
        synchronized (init) {
            if (init)
                return;

            if (server == null) {
                init = true;
                return;
            }

            try {
                // Parse dbVendor from server config
                String dbVendor = (String) server.get("dbVendor");
                EDbVendor vendor = getDBVendor();
                if (dbVendor != null) {
                    vendor = EDbVendor.valueOf(dbVendor);
                }

                // Set server name from config
                if (!SQLUtil.isEmpty((String) server.get("name"))) {
                    this.setDefaultServerName((String) server.get("name"));
                }

                boolean supportsCatalogs = TSQLEnv.supportCatalog(vendor);
                boolean supportsSchemas = TSQLEnv.supportSchema(vendor);

                // The manifest is authoritative about how THIS export was sharded,
                // which may differ from the vendor's GSP defaults. Oracle defaults
                // to catalog+schema, but its natural export is schema-only shards
                // (servers[].schemas[]). Honor the declared flags so the correct
                // process* branch is chosen; otherwise a schema-topology manifest
                // routes to processDatabases(), finds no "databases", and loads
                // nothing (silent empty env -> empty lineage). Mirror the same
                // override already done in SqlflowShardedMetadataAnalyzer.
                if (server.containsKey("supportsCatalogs")) {
                    supportsCatalogs = Boolean.TRUE.equals(server.get("supportsCatalogs"));
                }
                if (server.containsKey("supportsSchemas")) {
                    supportsSchemas = Boolean.TRUE.equals(server.get("supportsSchemas"));
                }

                // Get nameLevels for topology determination
                List<String> nameLevels = getNameLevels(server);

                // Resolve base directory
                String resolvedBaseDir = resolveBaseDir();

                // List for pending synonyms (resolved after all tables are loaded)
                List<Object[]> pendingSynonyms = new ArrayList<>();

                // Route on the PHYSICAL shard container first, not the vendor
                // default. Schema-topology exports (Oracle) list shards under
                // server.schemas[]; catalog-topology exports (SQL Server, MySQL)
                // list them under server.databases[]. Only after the container
                // is known do the catalog/schema flags decide how a databases[]
                // shard's catalog file is interpreted. Deciding purely from the
                // vendor default sent Oracle schema shards to processDatabases()
                // (which reads databases[]) and loaded nothing; deciding purely
                // from the flags would send a databases[]-wrapped schema-only
                // manifest to processSchemas() (which reads schemas[]) and load
                // nothing. Treat an empty list as absent.
                List schemaShards = (List) server.get("schemas");
                List databaseShards = (List) server.get("databases");
                boolean hasSchemaShards = schemaShards != null && !schemaShards.isEmpty();
                boolean hasDatabaseShards = databaseShards != null && !databaseShards.isEmpty();

                if (hasSchemaShards) {
                    // Schema-topology shards (e.g., Oracle)
                    processSchemas(server, resolvedBaseDir, pendingSynonyms);
                } else if (hasDatabaseShards) {
                    if (supportsCatalogs && supportsSchemas) {
                        // Catalog + Schema topology (e.g., SQL Server, PostgreSQL)
                        processDatabases(server, resolvedBaseDir, pendingSynonyms);
                    } else if (supportsCatalogs) {
                        // Catalog-only topology (e.g., MySQL)
                        processDatabasesAsCatalog(server, resolvedBaseDir, pendingSynonyms);
                    } else {
                        // Schema-only data wrapped in databases[]: the catalog
                        // file nests a schemas array; processDatabases already
                        // reads catalogJson.schemas (mirrors the metadata
                        // analyzer's processCatalogSchemaOnly).
                        processDatabases(server, resolvedBaseDir, pendingSynonyms);
                    }
                }

                // Resolve pending synonyms
                for (Object[] pending : pendingSynonyms) {
                    CatalogMetadataHelper.appendSynonym((TSQLSchema) pending[0], (Map) pending[1], nameForm);
                }

            } catch (Exception e) {
                Logger.getLogger(SqlflowShardedSQLEnv.class.getName())
                        .log(Level.WARNING, "Parse sharded catalog failed", e);
            }

            init = true;
        }
    }

    /**
     * Get nameLevels from server config, defaulting based on vendor capabilities.
     */
    private List<String> getNameLevels(Map server) {
        List<String> nameLevels = (List<String>) server.get("nameLevels");
        if (nameLevels != null && !nameLevels.isEmpty()) {
            return nameLevels;
        }

        // Default based on supportsCatalogs/supportsSchemas
        Boolean supportsCatalogs = (Boolean) server.get("supportsCatalogs");
        Boolean supportsSchemas = (Boolean) server.get("supportsSchemas");

        List<String> defaultLevels = new ArrayList<>();
        if (supportsCatalogs != null && supportsCatalogs) {
            defaultLevels.add("catalog");
        }
        if (supportsSchemas != null && supportsSchemas) {
            defaultLevels.add("schema");
        }

        return defaultLevels;
    }

    /**
     * Get base directory.
     */
    private String resolveBaseDir() {
        return baseDir;
    }

    /**
     * Process databases in catalog+schema topology.
     */
    private void processDatabases(Map server, String baseDir, List<Object[]> pendingSynonyms) {
        List databases = (List) server.get("databases");
        if (databases == null) {
            return;
        }

        for (int i = 0; i < databases.size(); i++) {
            Map jsonDatabase = (Map) databases.get(i);
            String databaseName = lex(ESQLDataObjectType.dotCatalog, (String) jsonDatabase.get("name"));

            // Get catalog path
            Map catalogInfo = (Map) jsonDatabase.get("catalog");
            if (catalogInfo == null) {
                continue;
            }

            String catalogPath = (String) catalogInfo.get("path");
            if (SQLUtil.isEmpty(catalogPath)) {
                continue;
            }

            // Resolve full path to catalog file
            String fullCatalogPath = resolvePath(baseDir, catalogPath);
            Map catalogJson = loadCatalogJson(fullCatalogPath);
            if (catalogJson == null) {
                continue;
            }

            // Process schemas within catalog
            List schemas = (List) catalogJson.get("schemas");
            if (schemas == null) {
                // Fallback: database might be directly in catalog root (nameLevels = ["database"])
                // In this case, create a default schema to hold tables/views
                TSQLCatalog sqlCatalog = getSQLCatalog(databaseName, true);
                TSQLSchema sqlSchema = sqlCatalog.getSchema(TSQLEnv.DEFAULT_SCHEMA_NAME, true);
                appendTables(catalogJson, sqlSchema);
                appendViews(catalogJson, sqlSchema);
                CatalogMetadataHelper.appendPackages(catalogJson, sqlSchema, nameForm);
                CatalogMetadataHelper.appendProcedures(catalogJson, sqlSchema, null, nameForm);
                collectSynonyms(catalogJson, sqlSchema, pendingSynonyms);
            } else {
                TSQLCatalog sqlCatalog = getSQLCatalog(databaseName, true);
                for (int j = 0; j < schemas.size(); j++) {
                    Map jsonSchema = (Map) schemas.get(j);
                    String schemaName = lex(ESQLDataObjectType.dotSchema, (String) jsonSchema.get("name"));
                    TSQLSchema sqlSchema = sqlCatalog.getSchema(schemaName, true);
                    appendTables(jsonSchema, sqlSchema);
                    appendViews(jsonSchema, sqlSchema);
                    CatalogMetadataHelper.appendPackages(jsonSchema, sqlSchema, nameForm);
                    CatalogMetadataHelper.appendProcedures(jsonSchema, sqlSchema, null, nameForm);
                    collectSynonyms(jsonSchema, sqlSchema, pendingSynonyms);
                }
            }
        }
    }

    /**
     * Process databases in catalog-only topology (no schema level).
     */
    private void processDatabasesAsCatalog(Map server, String baseDir, List<Object[]> pendingSynonyms) {
        List databases = (List) server.get("databases");
        if (databases == null) {
            return;
        }

        for (int i = 0; i < databases.size(); i++) {
            Map jsonDatabase = (Map) databases.get(i);
            String databaseName = lex(ESQLDataObjectType.dotCatalog, (String) jsonDatabase.get("name"));

            Map catalogInfo = (Map) jsonDatabase.get("catalog");
            if (catalogInfo == null) {
                continue;
            }

            String catalogPath = (String) catalogInfo.get("path");
            if (SQLUtil.isEmpty(catalogPath)) {
                continue;
            }

            String fullCatalogPath = resolvePath(baseDir, catalogPath);
            Map catalogJson = loadCatalogJson(fullCatalogPath);
            if (catalogJson == null) {
                continue;
            }

            // In catalog-only mode, each database is treated as a catalog
            // Tables/views are at the catalog level (not in schemas)
            TSQLCatalog sqlCatalog = getSQLCatalog(databaseName, true);
            TSQLSchema sqlSchema = sqlCatalog.getSchema(TSQLEnv.DEFAULT_SCHEMA_NAME, true);

            // Tables/views might be at root level or in schemas
            List tables = (List) catalogJson.get("tables");
            List views = (List) catalogJson.get("views");

            if (tables != null) {
                for (Object table : tables) {
                    appendTableObject((Map) table, sqlSchema, false);
                }
            }
            if (views != null) {
                for (Object view : views) {
                    appendTableObject((Map) view, sqlSchema, true);
                }
            }
            CatalogMetadataHelper.appendPackages(catalogJson, sqlSchema, nameForm);
            CatalogMetadataHelper.appendProcedures(catalogJson, sqlSchema, null, nameForm);
        }
    }

    /**
     * Process schemas in schema-only topology (e.g., Oracle).
     */
    private void processSchemas(Map server, String baseDir, List<Object[]> pendingSynonyms) {
        List schemas = (List) server.get("schemas");
        if (schemas == null) {
            return;
        }

        for (int i = 0; i < schemas.size(); i++) {
            Map jsonSchema = (Map) schemas.get(i);
            String schemaName = lex(ESQLDataObjectType.dotSchema, (String) jsonSchema.get("name"));

            // Get catalog path
            Map catalogInfo = (Map) jsonSchema.get("catalog");
            String fullCatalogPath = null;
            Map catalogJson = null;

            if (catalogInfo != null) {
                String catalogPath = (String) catalogInfo.get("path");
                if (!SQLUtil.isEmpty(catalogPath)) {
                    fullCatalogPath = resolvePath(baseDir, catalogPath);
                    catalogJson = loadCatalogJson(fullCatalogPath);
                }
            }

            // Create synthetic catalog for schema-only vendors
            TSQLCatalog sqlCatalog = getSQLCatalog(TSQLEnv.DEFAULT_DB_NAME, true);
            TSQLSchema sqlSchema = sqlCatalog.getSchema(schemaName, true);

            if (catalogJson != null) {
                appendTables(catalogJson, sqlSchema);
                appendViews(catalogJson, sqlSchema);
                CatalogMetadataHelper.appendPackages(catalogJson, sqlSchema, nameForm);
                CatalogMetadataHelper.appendProcedures(catalogJson, sqlSchema, null, nameForm);
                collectSynonyms(catalogJson, sqlSchema, pendingSynonyms);
            } else {
                // Process directly from schema config
                appendTables(jsonSchema, sqlSchema);
                appendViews(jsonSchema, sqlSchema);
                CatalogMetadataHelper.appendPackages(jsonSchema, sqlSchema, nameForm);
                CatalogMetadataHelper.appendProcedures(jsonSchema, sqlSchema, null, nameForm);
                collectSynonyms(jsonSchema, sqlSchema, pendingSynonyms);
            }
        }
    }

    /**
     * Load and parse a catalog JSON file.
     */
    private Map loadCatalogJson(String path) {
        if (path == null) {
            return null; // unresolvable relative path with no base directory
        }
        try {
            File file = new File(path);
            if (!file.exists()) {
                Logger.getLogger(SqlflowShardedSQLEnv.class.getName())
                        .warning("Catalog file not found: " + path);
                return null;
            }

            String content = new String(Files.readAllBytes(file.toPath()), StandardCharsets.UTF_8);
            Object parsed = JSON.parseObject(content);
            if (parsed instanceof Map) {
                return (Map) parsed;
            }
        } catch (IOException e) {
            Logger.getLogger(SqlflowShardedSQLEnv.class.getName())
                    .log(Level.WARNING, "Failed to read catalog file: " + path, e);
        }
        return null;
    }

    /**
     * Resolve relative path against base directory.
     */
    private String resolvePath(String baseDir, String relativePath) {
        // An absolute path stands on its own regardless of base directory.
        if (new File(relativePath).isAbsolute()) {
            return relativePath;
        }
        if (baseDir == null) {
            // Do NOT resolve a relative shard path against the process working
            // directory: that could read a same-named catalog from CWD and merge
            // a wrong environment. Without a base directory the shard is simply
            // unreachable (no catalog is better than a wrong one).
            return null;
        }

        // Handle both forward slash and backslash
        String normalizedBase = baseDir.replace('\\', '/');
        String normalizedRelative = relativePath.replace('\\', '/');

        return Paths.get(normalizedBase, normalizedRelative).toString();
    }

    /**
     * Append tables from a schema/catalog JSON object.
     */
    private void appendTables(Map jsonSchema, TSQLSchema sqlSchema) {
        List tables = (List) jsonSchema.get("tables");
        if (tables == null) {
            return;
        }

        for (int i = 0; i < tables.size(); i++) {
            Map jsonTable = (Map) tables.get(i);
            appendTableObject(jsonTable, sqlSchema, false);
        }
    }

    /**
     * Append views from a schema/catalog JSON object.
     */
    private void appendViews(Map jsonSchema, TSQLSchema sqlSchema) {
        List views = (List) jsonSchema.get("views");
        if (views == null) {
            return;
        }

        for (int i = 0; i < views.size(); i++) {
            Map jsonTable = (Map) views.get(i);
            appendTableObject(jsonTable, sqlSchema, true);
        }
    }

    /**
     * Append a single table or view object.
     */
    private void appendTableObject(Map jsonTable, TSQLSchema sqlSchema, boolean isView) {
        String tableName = lex(ESQLDataObjectType.dotTable, (String) jsonTable.get("name"));
        if (SQLUtil.isEmpty(tableName)) {
            return;
        }

        TSQLTable sqlTable = sqlSchema.createTable(tableName, isView ? 3 : 0);
        if (isView) {
            sqlTable.setView(true);
        }

        // Set table type if present
        String type = (String) jsonTable.get("type");
        if (type != null && type.toLowerCase(java.util.Locale.ROOT).indexOf("view") != -1) {
            sqlTable.setView(true);
        }

        // Add columns
        List columns = (List) jsonTable.get("columns");
        if (columns != null) {
            for (int j = 0; j < columns.size(); j++) {
                Map jsonColumn = (Map) columns.get(j);
                String columnName = lex(ESQLDataObjectType.dotColumn, (String) jsonColumn.get("name"));
                if (!SQLUtil.isEmpty(columnName)) {
                    sqlTable.addColumn(columnName);
                }
            }
        }
    }

    /**
     * Collect synonyms for later resolution.
     */
    private void collectSynonyms(Map jsonSchema, TSQLSchema sqlSchema, List<Object[]> pendingSynonyms) {
        List synonyms = (List) jsonSchema.get("synonyms");
        if (synonyms == null) {
            return;
        }

        for (int i = 0; i < synonyms.size(); i++) {
            Map jsonSynonym = (Map) synonyms.get(i);
            if (jsonSynonym == null) {
                continue;
            }
            pendingSynonyms.add(new Object[]{sqlSchema, jsonSynonym});
        }
    }
}