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.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.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.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * Reader for {@link CatalogInputKind#SQLDEP_JSON} sources — the JSON shape the legacy
 * {@code SQLDepSQLEnv} consumes (plan §11 T2.A.4 / first Phase 2.A static-file adapter).
 *
 * <p>Per plan §6 / §11.2 this reader funnels the SQLDep export shape into the same
 * {@link UnifiedCatalogModel} the JSON-manifest reader uses; the eager bridge then
 * materializes a {@code TSQLEnv} byte-for-byte equivalent to what the legacy
 * {@code SQLDepSQLEnv} would build (parity test in {@code SqldepLegacyParityTest}).
 * The legacy path remains untouched per plan §8.5.</p>
 *
 * <p>SQLDep export schema (key fields):</p>
 * <pre>{@code
 * {
 *   "createdBy": "sqldep ...",            // optional but conventional; not required
 *   "databaseModel": { ... },             // optional outer wrapper
 *   "databases": [
 *     {
 *       "name": "ORCL",
 *       "tables": [
 *         {
 *           "schema": "HR",                  // per-table schema label
 *           "name": "EMPLOYEES",
 *           "isView": "false",               // optional, string boolean
 *           "columns": [{"name": "..."}, ...]
 *         }
 *       ]
 *     }
 *   ]
 * }
 * }</pre>
 *
 * <p>The reader is dispatch-by-{@link CatalogInputKind#SQLDEP_JSON} (callers tag the
 * source explicitly). Auto-claim by {@code createdBy} content marker is intentionally
 * not implemented here — that dispatch already lives in the legacy
 * {@code TJSONSQLEnvParser} and reproducing it inside a {@code supports()} pre-scan
 * would force every SQLDep candidate source to be drained twice.</p>
 *
 * <p>Identifier names with embedded dots are wrapped with the vendor's quote
 * character 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"}. The legacy {@code SQLDepSQLEnv} carries the
 * same defense — the delimiter chars come from {@code TSQLEnv.delimitedChar} —
 * and we duplicate that table here so {@code catalog/**} stays free of a
 * compile-time dependency on the {@code sqlenv} runtime layer.
 * {@code CatalogModelValidator} still surfaces an identifier-bypass WARN when
 * the produced names disagree with {@code IdentifierService.normalize}, exactly
 * as for the JSON-manifest reader.</p>
 */
public final class SqldepCatalogInputReader implements CatalogInputReader {

    public SqldepCatalogInputReader() {
    }

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

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

    @Override
    public UnifiedCatalogModel read(CatalogInputSource source, CatalogLoadOptions options)
            throws CatalogInputException {
        if (source == null) {
            throw new CatalogInputException("SqldepCatalogInputReader: source is required");
        }
        final EDbVendor vendor = (options != null && options.vendor() != null)
            ? options.vendor()
            : EDbVendor.dbvoracle;  // SQLDep 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 SQLDep JSON from " + source.name() + ": " + ex.getMessage(),
                ex);
        }
        if (!(parsed instanceof Map)) {
            throw new CatalogInputException(
                "SQLDep root must be an object (got " + typeOf(parsed) + ")");
        }
        @SuppressWarnings("unchecked")
        Map<Object, Object> root = (Map<Object, Object>) parsed;

        // Optional outer wrapper — mirror SQLDepSQLEnv.initSQLEnv().
        Map<Object, Object> body = root;
        Object dm = root.get("databaseModel");
        if (dm instanceof Map) {
            @SuppressWarnings("unchecked")
            Map<Object, Object> dmMap = (Map<Object, Object>) dm;
            body = dmMap;
        } else if (dm != null) {
            throw new CatalogInputException(
                "SQLDep field 'databaseModel' must be an object (got " + typeOf(dm) + ")");
        }

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

            Object databases = body.get("databases");
            if (databases == null) {
                // Empty SQLDep export — return a vendor-only model. Validator gives a
                // WARN downstream if defaults are missing; structural emptiness here is
                // legal (and matches SQLDepSQLEnv's silent no-op when "databases" is null).
                return mb.build();
            }
            if (!(databases instanceof List)) {
                throw new CatalogInputException(
                    "SQLDep field 'databases' must be an array (got " + typeOf(databases) + ")");
            }
            for (Object dobj : (List<?>) databases) {
                if (!(dobj instanceof Map)) {
                    throw new CatalogInputException(
                        "SQLDep databases[i] must be an object (got " + typeOf(dobj) + ")");
                }
                @SuppressWarnings("unchecked")
                Map<Object, Object> dmap = (Map<Object, Object>) dobj;
                CatalogModel catalog = parseDatabase(dmap, vendor);
                if (catalog != null) {
                    mb.addCatalog(catalog);
                }
            }
            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.
            throw new CatalogInputException(
                "Malformed SQLDep export from " + source.name() + ": " + ex.getMessage(), ex);
        }
    }

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

    /**
     * Walk one {@code databases[i]} entry into a {@link CatalogModel}. SQLDep stores the
     * schema name on each table, so we group tables by their {@code schema} field before
     * building {@link SchemaModel}s — preserves insertion order so the round-trip with
     * {@code SQLEnvCatalogLoader} carries the same ordering as the legacy loader.
     *
     * <p>Catalog / schema / table names are wrapped with the vendor's quote char when
     * they contain a dot — see {@link #wrapDottedName} for the rationale.</p>
     */
    private CatalogModel parseDatabase(Map<Object, Object> dmap, EDbVendor vendor)
            throws CatalogInputException {
        String dbName = wrapDottedName(asString(dmap, "name", true), vendor);
        Object tables = dmap.get("tables");
        Map<String, SchemaGroup> groups = new LinkedHashMap<String, SchemaGroup>();
        if (tables != null) {
            if (!(tables instanceof List)) {
                throw new CatalogInputException(
                    "SQLDep database '" + dbName + "' field 'tables' must be an array (got "
                        + typeOf(tables) + ")");
            }
            for (Object tobj : (List<?>) tables) {
                if (!(tobj instanceof Map)) {
                    throw new CatalogInputException(
                        "SQLDep database '" + dbName + "' tables[i] must be an object (got "
                            + typeOf(tobj) + ")");
                }
                @SuppressWarnings("unchecked")
                Map<Object, Object> tmap = (Map<Object, Object>) tobj;
                String rawSchema = asString(tmap, "schema", false);
                String schemaName = (rawSchema == null) ? "" : wrapDottedName(rawSchema, vendor);
                SchemaGroup g = groups.get(schemaName);
                if (g == null) {
                    g = new SchemaGroup(schemaName);
                    groups.put(schemaName, g);
                }
                g.addTable(tmap, vendor);
            }
        }
        CatalogModel.Builder cb = CatalogModel.builder().name(dbName);
        for (SchemaGroup g : groups.values()) {
            cb.addSchema(g.build());
        }
        return cb.build();
    }

    private static final class SchemaGroup {
        private final SchemaModel.Builder b;

        SchemaGroup(String name) {
            this.b = SchemaModel.builder().name(name);
        }

        void addTable(Map<Object, Object> tmap, EDbVendor vendor)
                throws CatalogInputException {
            String tableName = wrapDottedName(asString(tmap, "name", true), vendor);
            boolean isView = parseStringBoolean(tmap.get("isView"));
            // 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.
            List<ColumnModel> cols = parseColumns(tmap, tableName);
            if (isView) {
                ViewModel.Builder vb = ViewModel.builder().name(tableName);
                for (ColumnModel c : cols) {
                    vb.addColumn(c);
                }
                b.addView(vb.build());
            } else {
                TableModel.Builder tb = TableModel.builder().name(tableName);
                for (ColumnModel c : cols) {
                    tb.addColumn(c);
                }
                b.addTable(tb.build());
            }
        }

        SchemaModel build() {
            return b.build();
        }
    }

    /**
     * Wrap a raw identifier with the vendor's open/close quote chars when it contains
     * a dot. Mirrors the legacy {@code SQLDepSQLEnv} defense: without the wrap, the
     * downstream string concatenation in {@code ModelBackedCatalogProvider.qualified}
     * would emit {@code catalog.schema.A.B} for a name {@code "A.B"}, and
     * {@code CatalogIdentifierPolicy.parse} would split that into four segments —
     * the table would never resolve. Already-wrapped names (delimiter chars on both
     * ends) are returned unchanged so a manifest that already quoted its dotted
     * names round-trips intact.
     *
     * <p>The vendor → delimiter-pair mapping is duplicated from
     * {@code TSQLEnv.delimitedChar}; we keep {@code catalog/**} free of a
     * compile-time dependency on {@code sqlenv} for layering reasons. If a new
     * dialect adds a non-{@code "..."} quote rule there, it must also be added here.</p>
     */
    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 the raw name is already wrapped on both ends with the vendor's pair,
        // do nothing — the caller already handed us a delimited form and round-trip
        // through the parser would just nest the delimiters.
        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 '"';
        }
    }

    /**
     * MSSQL/Azure use {@code [name]} (asymmetric); every other vendor uses the same
     * quote char on both ends. The legacy {@code TSQLEnv.delimitedChar} returns just
     * the open char and the legacy loader uses it on both sides — that produces an
     * invalid SQL form ({@code [foo[}) but happens to work for the legacy lookup
     * because the env stores the literal string as a single key. We use a proper
     * close-bracket here so the wrapped name is also valid SQL when echoed in
     * diagnostics; lookups still succeed because {@code IdentifierConfig.stripQuotedDelimiters}
     * canonicalizes both forms before comparison.
     */
    private static char vendorCloseDelimiter(EDbVendor v) {
        if (v == null) return '"';
        switch (v) {
            case dbvmssql:
            case dbvazuresql:
                return ']';
            default:
                return vendorOpenDelimiter(v);
        }
    }

    /**
     * SQLDep encodes booleans as quoted strings ({@code "true"} / {@code "false"}) — see
     * {@code SQLDepSQLEnv.initSQLEnv} which calls {@code Boolean.parseBoolean(...)}. We
     * mirror that: any non-string value tolerates a real Boolean too, and missing ⇒
     * false (a regular table).
     */
    private static boolean parseStringBoolean(Object v) {
        if (v == null) return false;
        if (v instanceof Boolean) return (Boolean) v;
        return Boolean.parseBoolean(v.toString());
    }

    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(
                "SQLDep table '" + tableName + "' field 'columns' must be an array (got "
                    + typeOf(columns) + ")");
        }
        for (Object cobj : (List<?>) columns) {
            if (!(cobj instanceof Map)) {
                throw new CatalogInputException(
                    "SQLDep table '" + 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;
    }

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

    private static void applyOptionDefaults(UnifiedCatalogModel.Builder mb,
                                            CatalogLoadOptions options) {
        if (options == null) return;
        DefaultsConfig.Builder db = DefaultsConfig.builder();
        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;
        }
        if (any) {
            mb.defaults(db.build());
        }
    }

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

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

    private static String readAll(CatalogInputSource source) throws CatalogInputException {
        try {
            if (source.inMemoryModel() != null) {
                throw new CatalogInputException(
                    "SqldepCatalogInputReader 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(
                "SqldepCatalogInputReader: source has no readable backing");
        } catch (IOException io) {
            throw new CatalogInputException(
                "Failed to read SQLDep 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(
                    "SQLDep 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.SQLDEP_JSON;
        }

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