package gudusoft.gsqlparser.dlineage.util;

import gudusoft.gsqlparser.EDbVendor;
import gudusoft.gsqlparser.ETokenType;
import gudusoft.gsqlparser.TGSqlParser;
import gudusoft.gsqlparser.TSourceToken;
import gudusoft.gsqlparser.TSourceTokenList;
import gudusoft.gsqlparser.sqlenv.CanonKey;
import gudusoft.gsqlparser.sqlenv.ESQLDataObjectType;
import gudusoft.gsqlparser.sqlenv.IdentifierCodec;
import gudusoft.gsqlparser.sqlenv.IdentifierInputForm;
import gudusoft.gsqlparser.sqlenv.IdentifierProfile;
import gudusoft.gsqlparser.sqlenv.IdentifierService;
import gudusoft.gsqlparser.util.SQLUtil;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;

/**
 * Keeps two different names from sharing one lineage normal name.
 *
 * <p>Lineage keys its models by {@link SQLUtil#normalizeIdentifier}, and the legacy fold behind it
 * can give two different names one string: PostgreSQL {@code "Amount"} and {@code amount} both
 * become {@code AMOUNT}, ClickHouse {@code userId} and {@code userid} both become {@code USERID}.
 * Lineage then treats the two as one column, and every reference to either picks up both as its
 * source. The same string is also displayed in places, so it cannot change for every name.</p>
 *
 * <p>So before the first statement is analyzed, {@link #scan} reads every identifier token of the
 * analysis's SQL and finds the legacy strings that more than one identity shares. Only a name whose
 * legacy string is contested changes ({@link #normalName}): it keeps the legacy string when that
 * string, read back as SQL, denotes it ({@code amount} keeps {@code AMOUNT} on PostgreSQL), and
 * otherwise gets a spelling that denotes it ({@code "Amount"}). Lineage compares these strings as
 * SQL names, so each one must name its own identity; the first name to arrive keeping the legacy
 * string would not, because {@code AMOUNT} read back as SQL is {@code amount} whoever holds it.
 * Column names are contested without case, because lineage matches column normal names
 * case-insensitively even where the legacy fold keeps case (Oracle, whose quoted {@code "Amount"}
 * gives {@code Amount} next to a bare {@code amount}'s {@code AMOUNT}). Names are told apart under
 * the analysis's own rules, so a configured profile's stricter ones count.</p>
 *
 * <p>The result is a pure function of the contested set, so all spellings of a legacy string change
 * together. A name the scan did not see (dynamic SQL, a name only metadata holds) follows the same
 * rule: it changes if its string is contested, and otherwise keeps the merge the legacy fold always
 * made. No string is ever split for some of its names and merged for the rest.</p>
 */
public final class NormalNameCollisions {

    /** Tokens that can spell a name: identifiers, delimited identifiers, and keywords used as names. */
    private static final Set<ETokenType> NAME_TOKENS = Collections.unmodifiableSet(EnumSet.of(
            ETokenType.ttidentifier, ETokenType.ttkeyword, ETokenType.ttnonreservedkeyword,
            ETokenType.ttdqstring, ETokenType.ttdbstring, ETokenType.ttbrstring));

    /** Characters that only delimit a name; dropping them never separates two spellings of one name. */
    private static final String DELIMITERS = "\"`[]{}'";

    private NormalNameCollisions() {
    }

    /**
     * The legacy normal names that two different identities share in {@code sqlTexts}, keyed for
     * {@link #normalName}. Empty when no two names collide, which is the common case.
     *
     * @param vendor the analysis's vendor; names are compared under its default identifier rules
     *        (see {@link #scan(IdentifierService, Collection)} for a configured profile)
     * @param sqlTexts every unit of SQL the analysis will read
     * @return the contested keys
     */
    public static Set<String> scan(EDbVendor vendor, Collection<String> sqlTexts) {
        return scan(IdentifierService.defaultServiceFor(vendor), sqlTexts);
    }

    /**
     * {@link #scan(EDbVendor, Collection)} under the analysis's own identifier rules, which a
     * configured profile may make stricter than the vendor's (SQL Server under a case-sensitive
     * collation tells {@code Amount} from {@code AMOUNT}).
     */
    public static Set<String> scan(IdentifierService service, Collection<String> sqlTexts) {
        EDbVendor vendor = service.getProfile().getVendor();
        List<TSourceTokenList> tokens = new ArrayList<TSourceTokenList>();
        for (String text : sqlTexts) {
            if (text == null || text.isEmpty()) {
                continue;
            }
            TGSqlParser lexer = new TGSqlParser(vendor);
            lexer.sqltext = text;
            lexer.tokenizeSqltext();
            tokens.add(lexer.sourcetokenlist);
        }
        return scanTokens(service, tokens);
    }

    /**
     * {@link #scan} over tokens already read, such as a parser's token list after it parsed the
     * analysis's only unit, which spares lexing the text a second time.
     */
    public static Set<String> scanTokens(IdentifierService service, Collection<TSourceTokenList> tokenLists) {
        EDbVendor vendor = service.getProfile().getVendor();
        // Names can only share a legacy string when they are the same text apart from case and
        // delimiters, so only such groups are checked in full.
        Map<String, Set<String>> byLooseText = new HashMap<String, Set<String>>();
        for (TSourceTokenList tokens : tokenLists) {
            for (int i = 0; tokens != null && i < tokens.size(); i++) {
                TSourceToken token = tokens.get(i);
                if (!NAME_TOKENS.contains(token.tokentype)) {
                    continue;
                }
                String spelling = token.toString();
                String loose = looseText(spelling);
                Set<String> spellings = byLooseText.get(loose);
                if (spellings == null) {
                    spellings = new LinkedHashSet<String>();
                    byLooseText.put(loose, spellings);
                }
                spellings.add(spelling);
            }
        }

        Map<String, ESQLDataObjectType> nameClasses = nameClasses(service.getProfile());
        Map<String, CanonKey> owners = new HashMap<String, CanonKey>();
        Set<String> contested = new HashSet<String>();
        for (Set<String> spellings : byLooseText.values()) {
            if (spellings.size() < 2) {
                continue;
            }
            for (String spelling : spellings) {
                for (Map.Entry<String, ESQLDataObjectType> nameClass : nameClasses.entrySet()) {
                    CanonKey identity = identity(service, nameClass.getValue(), spelling);
                    String legacy = identity == null ? null
                            : SQLUtil.legacyNormalSegment(vendor, nameClass.getValue(), spelling);
                    if (legacy == null) {
                        continue;
                    }
                    String key = key(nameClass.getKey(), nameClass.getValue(), legacy);
                    CanonKey owner = owners.get(key);
                    if (owner == null) {
                        owners.put(key, identity);
                    } else if (!owner.equals(identity)) {
                        contested.add(key);
                    }
                }
            }
        }
        return contested;
    }

    /**
     * The legacy strings of {@code contested}, whatever their class and upper-cased: a cheap first
     * test, which {@link #normalName} then makes exact.
     */
    public static Set<String> legacyNames(Set<String> contested) {
        Set<String> names = new HashSet<String>();
        for (String key : contested) {
            names.add(key.substring(key.indexOf('\u0000') + 1).toUpperCase(Locale.ROOT));
        }
        return names;
    }

    /**
     * The normal name of one identifier segment, given the contested keys of its analysis.
     *
     * @param contested keys from {@link #scan}
     * @param vendor vendor the legacy name was folded for
     * @param type role of the segment
     * @param origin the segment as written
     * @param legacy its legacy normal name
     * @return {@code legacy}, unless another name shares it and it does not denote {@code origin}
     */
    public static String normalName(Set<String> contested, EDbVendor vendor, ESQLDataObjectType type,
            String origin, String legacy) {
        return normalName(contested, IdentifierService.defaultServiceFor(vendor), type, origin, legacy);
    }

    /**
     * {@link #normalName(Set, EDbVendor, ESQLDataObjectType, String, String)} under the analysis's
     * own identifier rules; pass the service {@link #scan(IdentifierService, Collection)} used.
     */
    public static String normalName(Set<String> contested, IdentifierService service, ESQLDataObjectType type,
            String origin, String legacy) {
        IdentifierProfile profile = service.getProfile();
        if (!contested.contains(key(classOf(profile, type), type, legacy))) {
            return legacy;
        }
        CanonKey identity = identity(service, type, origin);
        if (identity == null || identity.equals(identity(service, type, legacy))) {
            return legacy;
        }
        String stored;
        try {
            stored = IdentifierCodec.isQuoted(profile, type, origin)
                    ? IdentifierCodec.decodeLexical(profile, type, origin)
                    : service.normalizeSegment(origin, type);
        } catch (RuntimeException e) {
            return legacy;
        }
        if (stored == null || stored.isEmpty()) {
            return legacy;
        }
        // The stored name as written when it reads back as this name, as lineage displays names
        // (ClickHouse userId); else delimited (PostgreSQL "Amount"). Either reads back as SQL.
        if (!IdentifierCodec.isQuoted(profile, type, stored)
                && IdentifierCodec.isVendorValidSpelling(profile, type, stored)
                && identity.equals(identity(service, type, stored))) {
            return stored;
        }
        if (IdentifierCodec.hasQuotedForm(profile, type)) {
            String delimited = IdentifierCodec.encodeStored(profile, type, stored);
            if (identity.equals(identity(service, type, delimited))) {
                return delimited;
            }
        }
        return legacy;
    }

    /**
     * The contested key of the legacy name {@code legacy} of a name of role {@code type}, as {@link #scan} returns
     * them: for a caller that finds two names sharing one legacy name outside the SQL text a scan reads.
     */
    public static String contestedKey(IdentifierService service, ESQLDataObjectType type, String legacy) {
        return key(classOf(service.getProfile(), type), type, legacy);
    }

    /** Names in one class get the same legacy string for the same text and follow the same rules. */
    private static String classOf(IdentifierProfile profile, ESQLDataObjectType type) {
        return SQLUtil.legacyFoldClass(type) + '/' + profile.groupOf(type);
    }

    /**
     * The name classes lineage keeps apart: columns, and tables with the objects that share their
     * rules. Routines, variables and schemas keep their legacy names, since code compares those
     * across classes whose legacy folds differ (DB2 folds a routine name keeping case, and the same
     * name folded as a variable to upper case), and one class changing alone would split them.
     */
    private static Map<String, ESQLDataObjectType> nameClasses(IdentifierProfile profile) {
        Map<String, ESQLDataObjectType> classes = new LinkedHashMap<String, ESQLDataObjectType>();
        for (ESQLDataObjectType type : new ESQLDataObjectType[] {ESQLDataObjectType.dotColumn,
                ESQLDataObjectType.dotTable}) {
            classes.put(classOf(profile, type), type);
        }
        return classes;
    }

    /**
     * Column names are contested without case, since lineage matches column normal names
     * case-insensitively; Oracle's fold keeps the case of a quoted name, so {@code "Amount"} gives
     * {@code Amount} and a bare {@code amount} gives {@code AMOUNT}, two strings that lineage still
     * matched. Table names are contested exactly.
     */
    private static String key(String nameClass, ESQLDataObjectType type, String legacy) {
        boolean caseless = "column".equals(SQLUtil.legacyFoldClass(type));
        return nameClass + '\u0000' + (caseless ? legacy.toUpperCase(Locale.ROOT) : legacy);
    }

    private static CanonKey identity(IdentifierService service, ESQLDataObjectType type, String spelling) {
        try {
            return service.canonKeyOf(spelling, type, IdentifierInputForm.SQL_LEXICAL);
        } catch (RuntimeException e) {
            // Not a name the rules can key (a malformed delimited spelling).
            return null;
        }
    }

    private static String looseText(String spelling) {
        StringBuilder loose = new StringBuilder(spelling.length());
        for (int i = 0; i < spelling.length(); i++) {
            char c = spelling.charAt(i);
            if (DELIMITERS.indexOf(c) < 0) {
                loose.append(c);
            }
        }
        return loose.toString().toUpperCase(Locale.ROOT);
    }
}