package me.ivanyu.simplekv.inmemoryimpl;

import java.util.*;

import me.ivanyu.simplekv.IBucket;
import me.ivanyu.simplekv.types.Key;
import me.ivanyu.simplekv.types.NodeId;
import me.ivanyu.simplekv.types.Value;

public class InMemoryBucket implements IBucket {
    private final boolean debug = true;

    private final StorageStrategy storageStrategy;

    private Node root;

    private static final int START_ID = 1;

    private NodeId lastId = null;

    private final Map<NodeId, Node> nodes = new HashMap<NodeId, Node>();
    private final TreeSet<NodeId> freeList = new TreeSet<NodeId>();

    public InMemoryBucket(StorageStrategy storageStrategy) {
        this.storageStrategy = storageStrategy;
        this.root = allocateNode(true, true);
    }

    @Override
    public String get(final String key) {
        checkKey(key);

        final Value valueResult = get0(this.root, Key.ofValue(key));

        if (debug) {
            checkInvariants();
        }

        return valueResult != null ? valueResult.getValue() : null;
    }

    private Value get0(final Node node, final Key key) {
        // See Collections.binarySearch doc on place values
        final int place = Collections.binarySearch(node.keys, key);

        if (node.isLeaf()) {
            if (place >= 0) {
                return ((LeafNode) node).values.get(place);
            } else {
                return null;
            }
        } else {
            final InternalNode internal = (InternalNode) node;
            final NodeId childId = internal.getChildIdForKey(key);
            final Node child = getNode(childId);
            return get0(child, key);
        }
    }

    @Override
    public String put(final String key, final String value) {
        return put(key, value, false);
    }

    @Override
    public String putIfAbsent(final String key, final String value) {
        return put(key, value, true);
    }

    private String put(final String key, final String value, final boolean onlyIfAbsent) {
        checkKey(key);
        checkValue(value);

        final Key oKey = Key.ofValue(key);
        final Value oValue = Value.ofValue(value);
        PutResult putResult = put0(root, null, oKey, oValue, onlyIfAbsent);
        if (putResult.split) {
            final InternalNode newRoot = (InternalNode) allocateNode(true, false);

            final Node oldRoot = root;
            oldRoot.setRoot(false);

            newRoot.keys.add(putResult.key);
            newRoot.children.add(oldRoot.getId());
            newRoot.children.add(putResult.rightPart.getId());

            this.root = newRoot;
        }

        if (debug) {
            checkInvariants();
        }

        return putResult.originalValue != null
                ? putResult.originalValue.getValue()
                : null;
    }

    private PutResult put0(final Node node, final InternalNode parent,
                           final Key key, final Value value,
                           final boolean onlyIfAbsent) {
        if (node.isLeaf()) {
            return put0Leaf((LeafNode) node, parent, key, value, onlyIfAbsent);
        } else {
            return put0Internal((InternalNode) node, parent, key, value, onlyIfAbsent);
        }
    }

    private PutResult put0Leaf(final LeafNode leaf, final InternalNode parent,
                               final Key key, final Value value,
                               final boolean onlyIfAbsent) {
        // See Collections.binarySearch doc on place values
        final int place = Collections.binarySearch(leaf.keys, key);
        if (place >= 0) {
            final Value originalValue = leaf.values.get(place);
            if (originalValue == null || !onlyIfAbsent) {
                leaf.values.set(place, value);
            }
            return new PutResult(originalValue, false, null, null);
        } else {
            if (storageStrategy.canPutWithoutSplit(leaf, key, value)) {
                insertIntoLeaf(leaf, key, value);
                return new PutResult(null, false, null, null);
            } else {
                final int originalCount = leaf.keys.size();

                final int insertionPoint = -place - 1;
                final StorageStrategy.Split optimalSplit =
                        storageStrategy.findOptimalSplit(leaf, parent, key, value);
                final LeafNode rightPart = (LeafNode) allocateNode(false, true);

                final boolean newAtBeginningOfRight =
                        insertionPoint == optimalSplit.getSplitPoint() && !optimalSplit.isInsertionToLeft();
                final boolean newAtEndOfRight =
                        insertionPoint == leaf.keys.size() && !optimalSplit.isInsertionToLeft();
                final boolean newAtBeginningOfLeft =
                        insertionPoint == 0 && optimalSplit.isInsertionToLeft();
                final boolean newAtEndOfLeft =
                        insertionPoint == optimalSplit.getSplitPoint() && optimalSplit.isInsertionToLeft();

                for (int i = optimalSplit.getSplitPoint(); i < leaf.keys.size(); i++) {
                    if (i == insertionPoint && !optimalSplit.isInsertionToLeft()) {
                        rightPart.keys.add(key);
                        rightPart.values.add(value);
                    }

                    Key k = leaf.keys.get(i);
                    Value v = leaf.values.get(i);
                    rightPart.keys.add(k);
                    rightPart.values.add(v);
                }

                for (int i = leaf.keys.size() - 1; i >= optimalSplit.getSplitPoint(); i--) {
                    leaf.keys.remove(i);
                    leaf.values.remove(i);
                }

                if (optimalSplit.isInsertionToLeft() && !newAtBeginningOfLeft && !newAtEndOfLeft) {
                    leaf.keys.add(insertionPoint, key);
                    leaf.values.add(insertionPoint, value);
                }

                if (newAtBeginningOfLeft) {
                    leaf.keys.add(0, key);
                    leaf.values.add(0, value);
                }
                if (newAtEndOfLeft) {
                    leaf.keys.add(key);
                    leaf.values.add(value);
                }
                if (newAtEndOfRight) {
                    rightPart.keys.add(key);
                    rightPart.values.add(value);
                }

                leaf.setNext(rightPart.getId());

                assert leaf.keys.size() + rightPart.keys.size() == originalCount + 1;
                assert leaf.values.size() + rightPart.values.size() == originalCount + 1;

                Key propagatedKey = rightPart.keys.get(0);
                return new PutResult(null, true, propagatedKey, rightPart);
            }
        }
    }

    private PutResult put0Internal(final InternalNode internal, final InternalNode parent,
                                   final Key key, final Value value,
                                   final boolean onlyIfAbsent) {
        final NodeId childId = internal.getChildIdForKey(key);
        final Node child = getNode(childId);

        final PutResult putResult = child.isLeaf()
                ? put0Leaf((LeafNode) child, internal, key, value, onlyIfAbsent)
                : put0Internal((InternalNode) child, internal, key, value, onlyIfAbsent);

        if (putResult.split) {
            if (storageStrategy.canPutWithoutSplit(internal, putResult.key)) {
                final int childIdx = internal.getChildIndex(key);
                internal.keys.add(childIdx, putResult.key);
                internal.children.add(childIdx + 1, putResult.rightPart.getId());
                return new PutResult(putResult.originalValue, false, null, null);
            } else {
                final int originalCount = internal.keys.size();

                // Key propagated from the children split
                final Key propagatedKey = putResult.key;

                // See Collections.binarySearch doc on place values
                final int place = Collections.binarySearch(internal.keys, propagatedKey);
                final int insertionPoint = -place - 1;
                final StorageStrategy.Split optimalSplit =
                        storageStrategy.findOptimalSplit(internal, parent, propagatedKey);
                final InternalNode rightPart = (InternalNode) allocateNode(false, false);

                final boolean newIsMedian = !optimalSplit.isInsertionToLeft()
                        && insertionPoint == optimalSplit.getSplitPoint();

                int moveKeysFrom = optimalSplit.getSplitPoint();
                int moveChildrenFrom = moveKeysFrom + 1;
                if (!newIsMedian) {
                    moveKeysFrom += 1;
                }

                rightPart.keys.addAll(internal.keys.subList(moveKeysFrom, internal.keys.size()));
                for (int i = internal.keys.size() - 1; i >= moveKeysFrom; i--) {
                    internal.keys.remove(i);
                }

                rightPart.children.addAll(internal.children.subList(moveChildrenFrom, internal.children.size()));
                for (int i = internal.children.size() - 1; i >= moveChildrenFrom; i--) {
                    internal.children.remove(i);
                }

                Key median;
                if (newIsMedian) {
                    median = propagatedKey;
                    rightPart.children.add(0, putResult.rightPart.getId());
                } else {
                    median = internal.keys.remove(internal.keys.size() - 1);

                    if (optimalSplit.isInsertionToLeft()) {
                        internal.keys.add(insertionPoint, propagatedKey);
                        internal.children.add(insertionPoint + 1, putResult.rightPart.getId());
                    } else {
                        rightPart.keys.add(insertionPoint - moveKeysFrom, propagatedKey);
                        rightPart.children.add(insertionPoint - moveKeysFrom + 1, putResult.rightPart.getId());
                    }
                }

                assert internal.keys.size() + rightPart.keys.size() == originalCount;
                assert internal.children.size() + rightPart.children.size() == originalCount + 2;

                return new PutResult(putResult.originalValue, true, median, rightPart);
            }
        } else {
            return new PutResult(putResult.originalValue, false, null, null);
        }
    }


    private void insertIntoLeaf(final LeafNode leaf, final Key key, final Value value) {
        // See Collections.binarySearch doc on place values
        final int place = Collections.binarySearch(leaf.keys, key);

        if (place >= 0) {  // the key is already there
            leaf.values.set(place, value);
        } else {
            final int insertionPoint = -place - 1;
            // Expand lists
            leaf.keys.add(null);
            leaf.values.add(null);
            // Move to the right from the insertion point
            for (int i = leaf.keys.size() - 2; i >= insertionPoint; i--) {
                leaf.keys.set(i + 1, leaf.keys.get(i));
                leaf.values.set(i + 1, leaf.values.get(i));
            }
            leaf.keys.set(insertionPoint, key);
            leaf.values.set(insertionPoint, value);
        }
    }

    @Override
    public String delete(final String key) {
        checkKey(key);

        final Key oKey = Key.ofValue(key);
        DeleteResult deleteResult;
        if (root.isLeaf()) {
            deleteResult = deleteFromLeaf((LeafNode) root, null, -1, null, null, oKey);
        } else {
            deleteResult = deleteRecursiveInternal((InternalNode) root, null, -1, null, null, oKey);
        }

        if (debug) {
            checkInvariants();
        }

        return deleteResult.originalValue != null
                ? deleteResult.originalValue.getValue()
                : null;
    }

    private DeleteResult deleteFromLeaf(final LeafNode leaf,
                                        final InternalNode parent, final int leafIndex,
                                        final LeafNode leftSibling, final LeafNode rightSibling,
                                        final Key key) {
        // See Collections.binarySearch doc on place values
        final int place = Collections.binarySearch(leaf.keys, key);
        if (place < 0) {
            return new DeleteResult(null, false);
        } else {
            if (storageStrategy.canDeleteWithoutMerge(leaf, key)) {
                final Value originalValue = deleteFromLeaf(leaf, place);
                return new DeleteResult(originalValue, false);
            } else {
                final Value originalValue = leaf.values.get(place);

                // Try to borrow from left or right if possible
                if (leftSibling != null) {
                    final DeleteResult tryLeft = tryWithBorrowFromSibling(
                            parent, leaf, leafIndex, key, place, leftSibling, false);
                    if (tryLeft != null) {
                        return tryLeft;
                    }
                }

                if (rightSibling != null) {
                    final DeleteResult tryRight = tryWithBorrowFromSibling(
                            parent, leaf, leafIndex, key, place, rightSibling, true);
                    if (tryRight != null) {
                        return tryRight;
                    }
                }

                deleteFromLeaf(leaf, place);
                return new DeleteResult(originalValue, true);
            }
        }
    }

    private DeleteResult deleteFromInternal(final InternalNode intNode,
                                            final InternalNode parent, final int intNodeIndex,
                                            final InternalNode leftSibling, final InternalNode rightSibling,
                                            final int keyIndex, final Value originalValue) {
        final Key key = intNode.keys.get(keyIndex);

        if (storageStrategy.canDeleteWithoutMerge(intNode, key)) {
            deleteFromInternalDirect(intNode, keyIndex);
            return new DeleteResult(originalValue, false);
        } else {
            // Try to borrow from left or right if possible
            if (leftSibling != null) {
                final DeleteResult tryLeft = tryWithBorrowFromSibling(parent, intNode, intNodeIndex,
                        key, keyIndex, originalValue, leftSibling, false);
                if (tryLeft != null) {
                    return tryLeft;
                }
            }

            if (rightSibling != null) {
                final DeleteResult tryRight = tryWithBorrowFromSibling(parent, intNode, intNodeIndex,
                        key, keyIndex, originalValue, rightSibling, true);
                if (tryRight != null) {
                    return tryRight;
                }
            }

            deleteFromInternalDirect(intNode, keyIndex);
            return new DeleteResult(originalValue, true);
        }
    }

    private DeleteResult tryWithBorrowFromSibling(final InternalNode parent,
                                                  final InternalNode intNode,
                                                  final int intNodeIndex,
                                                  final Key key, final int place,
                                                  final Value originalValue,
                                                  final InternalNode sibling,
                                                  final boolean acceptorAtLeft) {
        final Integer keyCountToBorrow = storageStrategy.getKeyCountToBorrowToDelete(
                intNode, key, parent, sibling, acceptorAtLeft);

        if (keyCountToBorrow != null) {
            deleteFromInternalDirect(intNode, place);
            for (int i = 0; i < keyCountToBorrow; i++) {
                // Distribute the keys through the parent.
                if (!acceptorAtLeft) {
                    intNode.keys.add(0, parent.keys.get(intNodeIndex - 1));
                    parent.keys.set(intNodeIndex - 1, sibling.keys.remove(sibling.keys.size() - 1));
                    intNode.children.add(0, sibling.children.remove(sibling.children.size() - 1));
                } else {
                    intNode.keys.add(parent.keys.get(intNodeIndex));
                    parent.keys.set(intNodeIndex, sibling.keys.remove(0));
                    intNode.children.add(sibling.children.remove(0));
                }
            }

            return new DeleteResult(originalValue, false);
        }

        return null;
    }

    private DeleteResult tryWithBorrowFromSibling(final InternalNode parent,
                                                  final LeafNode leaf,
                                                  final int leafIndex,
                                                  final Key key, final int place,
                                                  final LeafNode sibling,
                                                  final boolean acceptorAtLeft) {
        final Value originalValue = leaf.values.get(place);
        final Integer keyCountToBorrow = storageStrategy.getKeyCountToBorrowToDelete(
                leaf, key, originalValue, sibling, acceptorAtLeft);

        if (keyCountToBorrow != null) {
            deleteFromLeaf(leaf, place);
            for (int i = 0; i < keyCountToBorrow; i++) {
                if (!acceptorAtLeft) {
                    leaf.keys.add(0, sibling.keys.remove(sibling.keys.size() - 1));
                    leaf.values.add(0, sibling.values.remove(sibling.values.size() - 1));
                    parent.keys.set(leafIndex - 1, leaf.keys.get(0));
                } else {
                    leaf.keys.add(sibling.keys.remove(0));
                    leaf.values.add(sibling.values.remove(0));
                    parent.keys.set(leafIndex, sibling.keys.get(0));
                }
            }

            return new DeleteResult(originalValue, false);
        }

        return null;
    }

    private DeleteResult deleteRecursiveInternal(final InternalNode intNode,
                                                 final InternalNode parent, final int intNodeIndex,
                                                 final InternalNode leftSibling, final InternalNode rightSibling,
                                                 final Key key) {
        final int childIndex = intNode.getChildIndex(key);
        final NodeId childId = intNode.children.get(childIndex);
        final Node child = getNode(childId);

        final Node childLeftSibling = (childIndex - 1 >= 0)
                ? getNode(intNode.children.get(childIndex - 1))
                : null;
        final Node childRightSibling = (childIndex + 1 < intNode.children.size())
                ? getNode(intNode.children.get(childIndex + 1))
                : null;

        DeleteResult deleteResult;
        if (child.isLeaf()) {
            deleteResult = deleteFromLeaf((LeafNode) child, intNode, childIndex,
                    (LeafNode) childLeftSibling, (LeafNode) childRightSibling, key);
        } else {
            deleteResult = deleteRecursiveInternal((InternalNode) child, intNode, childIndex,
                    (InternalNode) childLeftSibling, (InternalNode) childRightSibling, key);
        }

        if (deleteResult.mergeNeeded) {
            int keyIndexToDelete = childIndex - 1;

            if (childLeftSibling != null || childRightSibling != null) {
                if (child.isLeaf()) {
                    final boolean mergeResultLeft = mergeChildren(intNode, childIndex - 1,
                            (LeafNode) childLeftSibling, (LeafNode) child);
                    if (!mergeResultLeft) {
                        final boolean mergeResultRight = mergeChildren(intNode, childIndex,
                                (LeafNode) child, (LeafNode) childRightSibling);
                        assert mergeResultRight;

                        keyIndexToDelete = childIndex;
                    }

                } else {
                    final boolean mergeResultLeft = mergeChildren(intNode, childIndex - 1,
                            (InternalNode) childLeftSibling, (InternalNode) child);
                    if (!mergeResultLeft) {
                        final boolean mergeResultRight = mergeChildren(intNode, childIndex,
                                (InternalNode) child, (InternalNode) childRightSibling);
                        assert mergeResultRight;

                        keyIndexToDelete = childIndex;
                    }
                }
            }

            if (intNode.keys.isEmpty() && intNode.isRoot()) {
                NodeId oldRootId = this.root.getId();
                releaseNode(oldRootId);

                this.root = child;
                child.setRoot(true);
                return new DeleteResult(deleteResult.originalValue, false);
            }

            final DeleteResult deleteFromInternalResult = deleteFromInternal(intNode, parent, intNodeIndex,
                    leftSibling, rightSibling, keyIndexToDelete, deleteResult.originalValue);
            return deleteFromInternalResult;
        } else {
            return deleteResult;
        }
    }

    private Value deleteFromLeaf(final LeafNode leaf, final int place) {
        assert place >= 0;

        final Value originalValue = leaf.values.get(place);

        leaf.keys.remove(place);
        leaf.values.remove(place);

        return originalValue;
    }

    private void deleteFromInternalDirect(final InternalNode internal, final int place) {
        assert place >= 0;

        internal.keys.remove(place);
        internal.children.remove(place + 1);
    }

    private boolean mergeChildren(final InternalNode parent, final int keyIndex,
                                  final LeafNode left, final LeafNode right) {
        if (left == null || right == null || !storageStrategy.canMerge(left, right)) {
            return false;
        }

        left.keys.addAll(right.keys);
        left.values.addAll(right.values);

        left.setNext(null);
        releaseNode(right.getId());

        return true;
    }

    private boolean mergeChildren(final InternalNode parent, final int keyIndex,
                                  final InternalNode left, final InternalNode right) {
        if (left == null || right == null)
            return false;

        final Key median = parent.keys.get(keyIndex);
        if (!storageStrategy.canMerge(left, right, median)) {
            return false;
        }

        left.keys.add(median);
        left.keys.addAll(right.keys);
        left.children.addAll(right.children);

        // TODO release right

        return true;
    }

    private Node getNode(NodeId node) {
        return nodes.get(node);
    }

    private Node allocateNode(final boolean root, final boolean leaf) {
        Node newNode = null;
        NodeId freeId = freeList.pollFirst();

        if (freeId != null) {
            if (leaf) {
                newNode = new LeafNode(freeId, root);
            } else {
                newNode = new InternalNode(freeId, root);
            }
            nodes.put(freeId, newNode);
        } else {
            if (lastId != null) {
                lastId = lastId.next();
            } else {
                lastId = NodeId.ofValue(START_ID);
            }

            if (leaf) {
                newNode = new LeafNode(lastId, root);
            } else {
                newNode = new InternalNode(lastId, root);
            }
            nodes.put(lastId,newNode);
        }

        return newNode;
    }

    private void releaseNode(final NodeId id) {
        nodes.remove(id);
        freeList.add(id);
    }

    private static void checkKey(final String key) {
        // TODO implement
    }

    private static void checkValue(final String value) {
        // do nothing for now
    }

    private void checkInvariants() {
        checkInvariants0(root, null, null);
    }
    private void checkInvariants0(final Node node, final Key lessThan, final Key greaterOrEqualThan) {
        List<Key> sortedKeys = new ArrayList<Key>(node.keys);
        Collections.sort(sortedKeys);
        for (int i = 0; i < sortedKeys.size(); i++) {
            assert sortedKeys.get(i) == node.keys.get(i);

            if (lessThan != null) {
                assert node.keys.get(i).compareTo(lessThan) < 0;
            }
            if (greaterOrEqualThan != null) {
                assert node.keys.get(i).compareTo(greaterOrEqualThan) >= 0;
            }
        }

        if (node.isLeaf()) {
            final LeafNode leaf = (LeafNode) node;
            assert storageStrategy.checkNodeLayout(leaf);
            assert leaf.keys.size() == leaf.values.size();

            if (leaf.getNext() != null) {
                final LeafNode nextNode = (LeafNode) getNode(leaf.getNext());
                assert leaf.keys.get(leaf.keys.size() - 1).compareTo(nextNode.keys.get(0)) < 0;
            }
        } else {
            final InternalNode internal = (InternalNode) node;
            assert storageStrategy.checkNodeLayout(internal);
            assert internal.keys.size() + 1 == internal.children.size();

            for (int i = 0; i < internal.keys.size(); i++) {
                final Node child = getNode(internal.children.get(i));
                checkInvariants0(child, internal.keys.get(i), null);
            }

            if (internal.keys.size() > 0) {
                final Node lastChild = getNode(internal.children.get(internal.keys.size()));
                checkInvariants0(lastChild, null, internal.keys.get(internal.keys.size() - 1));
            }
        }
    }


    public final void print() {
        print(root, "");
    }

    private void print(Node n, String indent) {
        System.out.print(indent + "[");
        for (Key k : n.keys) {
            System.out.print(k);
            System.out.print(" ");
        }
        System.out.println("]");

        if (!n.isLeaf()) {
            for (NodeId nodeId : ((InternalNode)n).children) {
                print(getNode(nodeId), indent + "\t");
            }
        }
    }
}
