/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.jackrabbit.oak.index.merge;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.TreeSet;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import org.apache.jackrabbit.oak.commons.json.JsonObject;
import org.apache.jackrabbit.oak.commons.json.JsopBuilder;
import org.apache.jackrabbit.oak.commons.json.JsopReader;
import org.apache.jackrabbit.oak.commons.json.JsopTokenizer;
import org.apache.jackrabbit.oak.plugins.index.IndexName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import jline.internal.Log;

/**
 * Utility that allows to merge index definitions.
 */
public class IndexDefMergerUtils {

    private final static Logger LOG = LoggerFactory.getLogger(IndexDefMergerUtils.class);

    private static HashSet<String> IGNORE_LEVEL_0 = new HashSet<>(Arrays.asList(
            "reindex",
            "refresh",
            "seed",
            "reindexCount"));
    private static HashSet<String> USE_PRODUCT_PROPERTY = new HashSet<>(Arrays.asList(
            "async",
            "includedPaths",
            "jcr:created",
            "jcr:createdBy",
            "jcr:lastModified",
            "jcr:lastModifiedBy",
            "jcr:uuid",
            "type"
            ));
    private static HashSet<String> USE_PRODUCT_CHILD_LEVEL_0 = new HashSet<>(Arrays.asList(
            "tika"));

    /**
     * Merge index definition changes.
     *
     * @param path the path of the change itself (e.g.  /oak:index/lucene-1/indexRules/acme)
     * @param ancestor the common ancestor (the old product index, e.g. lucene)
     * @param customName the name of the node of the customized index (e.g. /oak:index/lucene-1-custom-1)
     * @param custom the latest customized version (e.g. lucene-1-custom-1)
     * @param product the latest product index (e.g. lucene-2)
     * @param productName the name of the node of the latest product index (e.g. /oak:index/lucene-2)
     * @return the merged index definition (e.g. lucene-2-custom-1)
     */
    public static JsonObject merge(String path, JsonObject ancestor, String customName, JsonObject custom, JsonObject product, String productName) {
        ArrayList<String> conflicts = new ArrayList<>();
        JsonObject merged = merge(path, 0, ancestor, custom, product, conflicts);
        if (!conflicts.isEmpty()) {
            throw new UnsupportedOperationException("Conflicts detected: " + conflicts);
        }
        merged.getProperties().put("merges", "[" +
                JsopBuilder.encode(productName) + ", " +
                JsopBuilder.encode(customName) + "]");
        return merged;
    }

    private static JsonObject merge(String path, int level, JsonObject ancestor, JsonObject custom, JsonObject product,
            ArrayList<String> conflicts) {
        Objects.requireNonNull(conflicts);
        return mergeNoNull(path, level,
                        ancestor == null ? new JsonObject() : ancestor,
                        custom == null ? new JsonObject() : custom,
                        product == null ? new JsonObject() : product,
                        conflicts);
    }

    private static JsonObject mergeNoNull(String path, int level, JsonObject ancestor, JsonObject custom, JsonObject product,
            ArrayList<String> conflicts) {
        Objects.requireNonNull(ancestor);
        Objects.requireNonNull(custom);
        Objects.requireNonNull(product);
        Objects.requireNonNull(conflicts);
        JsonObject merged = new JsonObject(true);

        LinkedHashMap<String, Boolean> properties = new LinkedHashMap<>();
        addAllProperties(ancestor, properties);
        addAllProperties(custom, properties);
        addAllProperties(product, properties);

        for (String k : properties.keySet()) {
            if (level == 0 && IGNORE_LEVEL_0.contains(k)) {
                // ignore some properties
                continue;
            }
            if (k.startsWith(":")) {
                // ignore hidden properties
                continue;
            }
            String result = mergeProperty(path, k, ancestor, custom, product, conflicts);
            if (result != null) {
                merged.getProperties().put(k, result);
            }
        }
        LinkedHashMap<String, Boolean> children = new LinkedHashMap<>();
        // first the (new) product index - to ensure the order of children matches the new product index
        addAllChildren(product, children);
        addAllChildren(ancestor, children);
        addAllChildren(custom, children);
        for (String k : children.keySet()) {
            if (k.startsWith(":")) {
                // ignore hidden nodes
                continue;
            }
            JsonObject result = mergeChild(path + "/" + k, k, level, ancestor, custom, product, conflicts);
            if (result != null) {
                merged.getChildren().put(k, result);
            }

        }
        return merged;
    }

    private static void addAllProperties(JsonObject source, LinkedHashMap<String, Boolean> target) {
        for(String k : source.getProperties().keySet()) {
            target.put(k,  true);
        }
    }

    private static void addAllChildren(JsonObject source, LinkedHashMap<String, Boolean> target) {
        for(String k : source.getChildren().keySet()) {
            target.put(k,  true);
        }
    }

    private static String mergeProperty(String path, String property, JsonObject ancestor, JsonObject custom, JsonObject product,
            ArrayList<String> conflicts) {
        if (USE_PRODUCT_PROPERTY.contains(property)) {
            return product.getProperties().get(property);
        }
        String ap = ancestor.getProperties().get(property);
        String cp = custom.getProperties().get(property);
        String pp = product.getProperties().get(property);
        if (Objects.equals(ap, pp) || Objects.equals(cp, pp)) {
            return cp;
        } else if (Objects.equals(ap, cp)) {
            return pp;
        } else if (equalStringValues(ap, cp)) {
            return pp;
        } else {
            String mergedSets = tryMergeSets(ap, cp, pp);
            if (mergedSets != null) {
                return mergedSets;
            }
            conflicts.add("Could not merge value; path=" + path + " property=" + property + "; ancestor=" + ap + "; custom=" + cp
                    + "; product=" + pp);
            return ap;
        }
    }

    public static String tryMergeSets(String ancestor, String custom, String product) {
        if (ancestor == null || custom == null || product == null) {
            return null;
        }
        if (!ancestor.startsWith("[") && !custom.startsWith("[") && !product.startsWith("[")) {
            // none of the values is an array
            return null;
        }
        TreeSet<String> ancestorSet = getStringSet(ancestor);
        TreeSet<String> customSet = getStringSet(custom);
        TreeSet<String> productSet = getStringSet(product);
        if (ancestorSet == null || customSet == null || productSet == null) {
            return null;
        }
        // combine them all
        TreeSet<String> resultSet = new TreeSet<>();
        resultSet.addAll(ancestorSet);
        resultSet.addAll(customSet);
        resultSet.addAll(productSet);
        // build JSON
        JsopBuilder buff = new JsopBuilder();
        buff.array();
        for (String s : resultSet) {
            buff.value(s);
        }
        buff.endArray();
        return buff.toString();
    }

    public static boolean equalStringValues(String a, String b) {
        String aa = getStringOrStringFromSingleValueArray(a);
        String bb = getStringOrStringFromSingleValueArray(b);
        if (aa == null || bb == null) {
            // a or b are not strings or single-valued array of strings
            return false;
        }
        return Objects.equals(aa, bb);
    }

    public static TreeSet<String> getStringSet(String value) {
        if (value == null) {
            return null;
        }
        try {
            JsopTokenizer tokenizer = new JsopTokenizer(value);
            TreeSet<String> result = new TreeSet<>();
            if (tokenizer.matches(JsopReader.STRING)) {
                result.add(tokenizer.getEscapedToken());
                return result;
            }
            if (!tokenizer.matches('[')) {
                return null;
            }
            if (!tokenizer.matches(']')) {
                do {
                    if (!tokenizer.matches(JsopReader.STRING)) {
                        // not a string
                        return null;
                    }
                    result.add(tokenizer.getEscapedToken());
                } while (tokenizer.matches(','));
                tokenizer.read(']');
            }
            tokenizer.read(JsopReader.END);
            return result;
        } catch (IllegalArgumentException e) {
            return null;
        }
    }

    public static String getStringOrStringFromSingleValueArray(String value) {
        if (value == null) {
            return null;
        }
        JsopTokenizer tokenizer = new JsopTokenizer(value);
        if (tokenizer.matches(JsopReader.STRING)) {
            return tokenizer.getEscapedToken();
        }
        if (!tokenizer.matches('[')) {
            return null;
        }
        if (!tokenizer.matches(JsopReader.STRING)) {
            // not a string
            return null;
        }
        String jsonString = tokenizer.getEscapedToken();
        if (!tokenizer.matches(']')) {
            // not a single-element array
            return null;
        }
        return jsonString;
    }

    private static JsonObject mergeChild(String path, String child, int level, JsonObject ancestor, JsonObject custom, JsonObject product,
            ArrayList<String> conflicts) {
        if (level == 1 && path.indexOf("/aggregates/") >= 0) {
            return mergeAggregates(path, child, level, ancestor, custom, product, conflicts);
        }
        JsonObject a = ancestor.getChildren().get(child);
        JsonObject c = custom.getChildren().get(child);
        JsonObject p = product.getChildren().get(child);
        if (level == 0 && USE_PRODUCT_CHILD_LEVEL_0.contains(child)) {
            return p;
        }
        if (c == null && p != null) {
            return p; // restore child nodes in product index if removed in custom index
        } else if (isSameJson(a, p) || isSameJson(c, p)) {
            return c;
        } else if (isSameJson(a, c)) {
            return p;
        } else {
            return merge(path, level + 1, a, c, p, conflicts);
        }
    }

    private static JsonObject mergeAggregates(String path, String child, int level, JsonObject ancestor, JsonObject custom, JsonObject product,
            ArrayList<String> conflicts) {

        // merge, with level + 1 so that we don't recurse into this function again
        // conflicts are redirected to a new, temporary list
        ArrayList<String> aggregateConflicts = new ArrayList<>();
        JsonObject merged = mergeChild(path, child, level + 1, ancestor, custom, product, aggregateConflicts);

        // if there were conflicts, resolve them
        if (!aggregateConflicts.isEmpty()) {

            // list of "include" elements to move to the end
            ArrayList<JsonObject> elementToMove = new ArrayList<>();

            // which is the next id for "include" (eg. 12)
            long nextIncludeId = getNextIncludeId(ancestor.getChildren().get(child));
            nextIncludeId = Math.max(nextIncludeId, getNextIncludeId(custom.getChildren().get(child)));
            nextIncludeId = Math.max(nextIncludeId, getNextIncludeId(product.getChildren().get(child)));

            // loop over conflicts, and find + remove these
            // the aggregateConflicts will contain entries that look like this:
            // "Could not merge value; path=/oak:index/assets-11/aggregates/asset/include11
            // property=path; ancestor=null; custom=...; product=..."
            // and we need to extract the path
            for (String n : aggregateConflicts) {
                String regex = "path=([^\\s]+)\\sproperty=";
                Pattern pattern = Pattern.compile(regex);
                Matcher matcher = pattern.matcher(n);
                if (matcher.find()) {
                    // the path of the conflicting aggregation node
                    String extractedPath = matcher.group(1);
                    String[] elements = extractedPath.split("/");
                    String conflictElement = elements[elements.length - 1];

                    // remove from the custom list
                    JsonObject conflict = custom.getChildren().get(child).getChildren().remove(conflictElement);

                    // remember the element, to put it back later
                    elementToMove.add(conflict);
                }
            }

            // merge again, with conflicts resolved now
            // (if there are other conflicts unrelated to aggregation,
            // those will not be resolved)
            merged = mergeChild(path, child, level + 1, ancestor, custom, product, conflicts);

            // add the aggregation conflict at the end, with new ids
            // first we need to clone the merged object,
            // because it might be the same object as the product currently
            merged = JsonObject.fromJson(merged.toString(), true);
            for (JsonObject json : elementToMove) {
                merged.getChildren().put("include" + nextIncludeId, json);
                nextIncludeId++;
            }
        }
        return merged;
    }

    private static long getNextIncludeId(JsonObject json) {
        long max = 0;
        for(String n : json.getChildren().keySet()) {
            if (n.startsWith("include")) {
                n = n.substring("include".length());
                try {
                    long id = Long.parseLong(n);
                    max = Math.max(max, id);
                } catch (NumberFormatException e) {
                    LOG.warn("Expected 'include' + number, got " + n);
                    // ignore: it will probably not result in a conflict
                }
            }
        }
        return max + 1;
    }

    private static boolean isSameJson(JsonObject a, JsonObject b) {
        if (a == null || b == null) {
            return a == null && b == null;
        }
        return a.toString().equals(b.toString());
    }

    /**
     * For indexes that were modified both by the customer and in the product, merge
     * the changes, and create a new index.
     *
     * The new index (if any) is stored in the "newIndexes" object.
     *
     * @param newIndexes the new indexes
     * @param allIndexes all index definitions (including the new ones)
     */
    public static void merge(JsonObject newIndexes, JsonObject allIndexes) {

        // TODO when merging, we keep the product index, so two indexes are created.
        // e.g. lucene, lucene-custom-1, lucene-2, lucene-2-custom-1
        // but if we don't have lucene-2, then we can't merge lucene-3.

        // TODO when merging, e.g. lucene-2-custom-1 is created. but
        // it is only imported in the read-write repo, not in the read-only
        // repository currently. so this new index won't be used;
        // instead, lucene-2 will be used

        List<IndexName> newNames = newIndexes.getChildren().keySet().stream().map(s -> IndexName.parse(s))
                .collect(Collectors.toList());
        Collections.sort(newNames);
        List<IndexName> allNames = allIndexes.getChildren().keySet().stream().map(s -> IndexName.parse(s))
                .collect(Collectors.toList());
        Collections.sort(allNames);
        HashMap<String, JsonObject> mergedMap = new HashMap<>();
        for (IndexName n : newNames) {
            if (n.getCustomerVersion() == 0) {
                IndexName latest = n.getLatestCustomized(allNames);
                IndexName ancestor = n.getLatestProduct(allNames);
                if (latest != null && ancestor != null) {
                    if (n.isVersioned() && (n.compareTo(latest) <= 0 || n.compareTo(ancestor) <= 0)) {
                        // ignore older versions of indexes (not applicable to unversioned indexes)
                        continue;
                    }
                    JsonObject latestCustomized = allIndexes.getChildren().get(latest.getNodeName());
                    JsonObject latestAncestor = allIndexes.getChildren().get(ancestor.getNodeName());
                    JsonObject newProduct = newIndexes.getChildren().get(n.getNodeName());
                    try {
                        JsonObject merged = merge(
                                "", latestAncestor,
                                latest.getNodeName(), latestCustomized,
                                newProduct, n.getNodeName());
                        mergedMap.put(n.nextCustomizedName(), merged);
                    } catch (UnsupportedOperationException e) {
                        throw new UnsupportedOperationException("Index: " + n.getNodeName() + ": " + e.getMessage(), e);
                    }
                }
            }
        }
        for (Entry<String, JsonObject> e : mergedMap.entrySet()) {
            newIndexes.getChildren().put(e.getKey(), e.getValue());
        }
    }

}
