package de.bwaldvogel.mongo.backend;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import de.bwaldvogel.mongo.MongoCollection;
import de.bwaldvogel.mongo.bson.Document;
import de.bwaldvogel.mongo.exception.CannotIndexParallelArraysError;
import de.bwaldvogel.mongo.exception.KeyConstraintError;

public abstract class Index<P> {

    private final String name;
    private final List<IndexKey> keys;
    private final boolean sparse;

    protected Index(String name, List<IndexKey> keys, boolean sparse) {
        this.name = name;
        this.keys = keys;
        this.sparse = sparse;
    }

    protected boolean isSparse() {
        return sparse;
    }

    protected List<IndexKey> getKeys() {
        return keys;
    }

    public String getName() {
        return name;
    }

    protected List<String> keys() {
        return keys.stream()
            .map(IndexKey::getKey)
            .collect(Collectors.toList());
    }

    protected Set<String> keySet() {
        return keys.stream()
            .map(IndexKey::getKey)
            .collect(Collectors.toCollection(LinkedHashSet::new));
    }

    Set<KeyValue> getKeyValues(Document document) {
        return getKeyValues(document, true);
    }

    Set<KeyValue> getKeyValues(Document document, boolean normalize) {
        Map<String, Object> valuesPerKey = new LinkedHashMap<>();
        for (String key : keys()) {
            Object value = Utils.getSubdocumentValueCollectionAware(document, key);
            if (normalize) {
                value = Utils.normalizeValue(value);
            }
            valuesPerKey.put(key, value);
        }

        Map<String, Object> collectionValues = valuesPerKey.entrySet().stream()
            .filter(entry -> entry.getValue() instanceof Collection)
            .collect(StreamUtils.toLinkedHashMap());

        if (collectionValues.size() == 1) {
            @SuppressWarnings("unchecked")
            Collection<Object> collectionValue = (Collection<Object>) CollectionUtils.getSingleElement(collectionValues.values());
            return CollectionUtils.multiplyWithOtherElements(valuesPerKey.values(), collectionValue).stream()
                .map(KeyValue::new)
                .collect(StreamUtils.toLinkedHashSet());
        } else if (collectionValues.size() > 1) {
            throw new CannotIndexParallelArraysError(collectionValues.keySet());
        } else {
            return Collections.singleton(new KeyValue(valuesPerKey.values()));
        }
    }

    public abstract P getPosition(Document document);

    public abstract void checkAdd(Document document, MongoCollection<P> collection);

    public abstract void add(Document document, P position, MongoCollection<P> collection);

    public abstract P remove(Document document);

    public abstract boolean canHandle(Document query);

    public abstract Iterable<P> getPositions(Document query);

    public abstract long getCount();

    public boolean isEmpty() {
        return getCount() == 0;
    }

    public abstract long getDataSize();

    public abstract void checkUpdate(Document oldDocument, Document newDocument, MongoCollection<P> collection);

    public abstract void updateInPlace(Document oldDocument, Document newDocument, P position, MongoCollection<P> collection) throws KeyConstraintError;

    protected boolean isCompoundIndex() {
        return keys().size() > 1;
    }

    protected boolean nullAwareEqualsKeys(Document oldDocument, Document newDocument) {
        Set<KeyValue> oldKeyValues = getKeyValues(oldDocument);
        Set<KeyValue> newKeyValues = getKeyValues(newDocument);
        return Utils.nullAwareEquals(oldKeyValues, newKeyValues);
    }

    public abstract void drop();

    @Override
    public String toString() {
        return getClass().getSimpleName() + "[name=" + getName() + "]";
    }

}

@@ -1510,155 +1510,39 @@
 y = 
-new LinkedHashMap%3C%3E();%0A        for (String key : keys()) %7B%0A            Object value = Utils.getSubdocumentValueCollectionAware(document, key);%0A    
+collectValuesPerKey(document);%0A
     
@@ -1578,19 +1578,47 @@
     
-   
+valuesPerKey.replaceAll((key,
  value
- =
+) -%3E
  Uti
@@ -1637,30 +1637,27 @@
 Value(value)
+)
 ;%0A
-    
         %7D%0A  
@@ -1654,26 +1654,25 @@
       %7D%0A
-  
+%0A
         
   values
@@ -1667,77 +1667,26 @@
     
-  valuesPerKey.put(key, value);%0A        %7D%0A%0A        Map%3CString, Object
+List%3CCollection%3C?%3E
 %3E co
@@ -1719,16 +1719,14 @@
 Key.
-entrySet
+values
 ().s
@@ -1757,33 +1757,22 @@
 ter(
-entry -%3E entry.getV
+value -%3E v
 alue
-()
  ins
@@ -1808,43 +1808,82 @@
    .
-collect(StreamUtils.toLinkedHashMap
+map(value -%3E (Collection%3C?%3E) value)%0A            .collect(Collectors.toList
 ());
@@ -2090,25 +2090,16 @@
 onValues
-.values()
 );%0A     
@@ -2403,31 +2403,12 @@
 ror(
-collectionValues.keySet
+keys
 ());
@@ -2513,32 +2513,379 @@
        %7D%0A    %7D%0A%0A
+    private Map%3CString, Object%3E collectValuesPerKey(Document document) %7B%0A        Map%3CString, Object%3E valuesPerKey = new LinkedHashMap%3C%3E();%0A        for (String key : keys()) %7B%0A            Object value = Utils.getSubdocumentValueCollectionAware(document, key);%0A            valuesPerKey.put(key, value);%0A        %7D%0A        return valuesPerKey;%0A    %7D%0A%0A
     public abstr

Refactor Index