package com.hammurapi.store;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;

import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.Predicate;
import com.hammurapi.store.Store.Handle;

public abstract class AbstractIndex<ST, T extends ST, PK, V, S extends Store<ST,PK,S>> implements OrderedIndex<ST, T, PK, V, S> {

	private Predicate<ST, S> predicate;
	private Extractor<T, V, S> extractor; 
	protected Index.Type type; 
	protected boolean ordered;
	protected Comparator<V> comparator;
	protected AbstractStore<ST, PK, S> store;
	protected AtomicBoolean isCorrupt = new AtomicBoolean(false);
	
	protected AbstractIndex(
			Predicate<ST, S> predicate,
			Extractor<T, V, S> extractor, 
			Index.Type type, 
			boolean ordered, 
			Comparator<V> comparator,
			AbstractStore<ST,PK,S> store) {
		this.predicate = predicate;
		this.extractor = extractor;
		this.type = type;
		this.ordered = ordered;
		this.comparator = comparator;
		this.store = store;
	}
	
	/**
	 * @return Index unique store. If index is ordered, the index store
	 * shall be of type SortedMap.
	 */
	protected abstract Map<V, Handle<ST,PK,S>> getIndexUniqueStore();
	
	/**
	 * @return Index multi-store. If index is ordered, the index store
	 * shall be of type SortedMap.
	 */
	protected abstract Map<V, Collection<Handle<ST,PK,S>>> getIndexMultiStore();
	
	protected abstract Lock readLock();
	
	protected abstract Lock writeLock();
	
	/**
	 * Upgrades index type.
	 * @param type Asynchronous and lazy indices can be 
	 * upgraded to synchronous. All index types can be upgraded
	 * to unique if current index data does not contain duplicates.
	 * @param ordered unordered index can be upgraded to ordered.
	 */
	public abstract void upgrade(Index.Type type, boolean ordered, Comparator<V> comparator);
	
	Collection<Handle<ST,PK,S>> getHandles() {
		checkCorrupt();
		applyUpdates();
		try { 
			Collection<Handle<ST,PK,S>> ret = new LinkedList<Handle<ST,PK,S>>();
			if (isUnique()) {
				for (Handle<ST,PK,S> h: getIndexUniqueStore().values()) {
					if (h.isValid()) {
						ret.add(h);
					}
				}
			} else {
				for (Collection<Handle<ST,PK,S>> ch: getIndexMultiStore().values()) {
					for (Handle<ST,PK,S> h: ch) {
						if (h.isValid()) {
							ret.add(h);
						}
					}
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public Iterator<T> iterator() {
		checkCorrupt();
		getStore().readLock().lock();
		try {
			applyUpdates();
			try { 
				Collection<T> ret = new LinkedList<T>();
				if (isUnique()) {
					for (Handle<ST,PK,S> h: getIndexUniqueStore().values()) {
						if (h.isValid()) {
							ret.add((T) h.get());
						}
					}
				} else {
					for (Collection<Handle<ST,PK,S>> ch: getIndexMultiStore().values()) {
						for (Handle<ST,PK,S> h: ch) {
							if (h.isValid()) {
								ret.add((T) h.get());
							}
						}
					}
				}
				return ret.iterator();
			} finally {
				readLock().unlock();
			}
		} finally {
			getStore().readLock().unlock();
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public S getStore() {
		return (S) store;
	}

	@Override
	public Predicate<ST, S> getPredicate() {
		return predicate;
	}

	@Override
	public Extractor<T, V, S> getExtractor() {
		return extractor;
	}

	@Override
	public boolean isUnique() {
		return Type.UNIQUE.equals(type);
	}

	@Override
	public boolean isOrdered() {
		return ordered;
	}		

	@Override
	public Comparator<V> getComparator() {
		return comparator;
	}
	
	private Comparator<V> naturalComparator = new NaturalComparator<V>();
	
	protected abstract Collection<UpdateEntry> getPendingUpdates();

	/**
	 * Updates index
	 * @param changed Changed/added handle.
	 * @param isNew If true, then handle was added, updated otherwise.
	 */
	public void update(Handle<ST,PK,S> changed, boolean isNew) {
		checkCorrupt();
		switch (type) {
		case UNIQUE:
		case SYNCHRONOUS:
			writeLock().lock();
			try {
				applyUpdate(changed, isNew);
			} finally {
				writeLock().unlock();
			}
			break;
		case ASYNCHRONOUS:
			writeLock().lock();
			try {
				getPendingUpdates().add(new UpdateEntry(changed, isNew));
				getExecutorService().submit(applyUpdatesTask);
			} finally {
				writeLock().unlock();
			}
			break;
		case LAZY:
			writeLock().lock();
			try {
				getPendingUpdates().add(new UpdateEntry(changed, isNew));
			} finally {
				writeLock().unlock();
			}
			break;
		}
	}
	
	protected class UpdateEntry {
		Handle<ST,PK,S> handle;
		boolean isNew;
		
		public UpdateEntry(Handle<ST, PK, S> handle, boolean isNew) {
			super();
			this.handle = handle;
			this.isNew = isNew;
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result	+ ((handle == null) ? 0 : handle.hashCode());
			result = prime * result + (isNew ? 1231 : 1237);
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			UpdateEntry other = (UpdateEntry) obj;
			if (handle == null) {
				if (other.handle != null)
					return false;
			} else if (!handle.equals(other.handle))
				return false;
			if (isNew != other.isNew)
				return false;
			return true;
		}

	}
	
	protected abstract ExecutorService getExecutorService();
	
	protected void checkCorrupt() {
		if (isCorrupt.get()) {
			throw new StoreException("Index is corrupt");
		}
	}
	
	boolean isCorrupt() {
		return isCorrupt.get();
	}
	
	protected Runnable applyUpdatesTask = new Runnable() {
		
		public void run() {
			writeLock().lock();
			try {
				Iterator<UpdateEntry> hit = getPendingUpdates().iterator();
				while (hit.hasNext()) {
					UpdateEntry updateEntry = hit.next();
					applyUpdate(updateEntry.handle, updateEntry.isNew);
					hit.remove();
				}
			} catch (Exception e) {
				isCorrupt.set(true);
				handleAsynchException(e);
			} finally {
				writeLock().unlock();
			}
		};
	};
	
	
	/**
	 * Applies pending updates, if any. This method acquires
	 * index write lock. If update operation is successful,
	 * then this method downgrades the lock to read lock.
	 * In the case of exception, write lock is released and
	 * read lock is not acquired.   
	 */
	private void applyUpdates() {
		writeLock().lock();
		try {
			Iterator<UpdateEntry> hit = getPendingUpdates().iterator();
			while (hit.hasNext()) {
				UpdateEntry updateEntry = hit.next();
				applyUpdate(updateEntry.handle, updateEntry.isNew);
				hit.remove();
			}
			readLock().lock();
		} finally {
			writeLock().unlock();
		}
	}
	
	private AtomicInteger granularity = new AtomicInteger();
	
	/**
	 * This method is invoked within write lock, no need
	 * to acquire own lock.
	 * @param changed
	 */
	@SuppressWarnings("unchecked")
	protected void applyUpdate(Handle<ST,PK,S> changed, boolean isNew) {		
		if (isUnique()) {
			Map<V, Handle<ST, PK, S>> is = getIndexUniqueStore();
			if (!isNew) {
				is.values().remove(changed);
			}
			if (predicate==null || changed.extract(predicate)) {
				V key = extractor==null ? (V) changed.get() : changed.extract((Extractor<ST, V, S>) extractor);
				if (is.containsKey(key)) {
					throw new StoreException("Violation of unique index: "+this);
				}
				is.put(key, changed);				
			}
			granularity.set(is.size());
		} else {
			Map<V, Collection<Handle<ST, PK, S>>> is = getIndexMultiStore();
			if (!isNew) {
				Iterator<Collection<Handle<ST, PK, S>>> vit = is.values().iterator();
				while (vit.hasNext()) {
					Collection<Handle<ST, PK, S>> nextBucket = vit.next();
					if (nextBucket.remove(changed)) {;
						if (nextBucket.isEmpty()) {
							vit.remove();
						}
					}
				}
			}
			
			if (predicate==null || changed.extract(predicate)) {
				V key = extractor==null ? (V) changed.get() : changed.extract((Extractor<ST, V, S>) extractor);
				Collection<Handle<ST, PK, S>> bucket = is.get(key);
				if (bucket == null) {
					bucket = new LinkedList<Store.Handle<ST,PK,S>>();
					is.put(key, bucket);
				}
				bucket.add(changed);				
			}			
			
			granularity.set(getIndexMultiStore().size());			
		}
		
	}
	
	public int getGranularity() {
		return granularity.get();
	}
	
	/**
	 * Handles exceptions thrown in asynchronous applyUpdates().
	 * @param e
	 */
	protected abstract void handleAsynchException(Exception e);
	
	/**
	 * Removes invalid handles from index store. This method
	 * operates under store's write lock, so it is not necessary
	 * to acquire own lock. 
	 */
	// TODO - invoke from store cleanup asynchronously.
	public void cleanup() {
		if (isUnique()) {
			Iterator<Handle<ST, PK, S>> hit = getIndexUniqueStore().values().iterator();
			while (hit.hasNext()) {
				if (!hit.next().isValid()) {
					hit.remove();
				}
			}
		} else {
			Iterator<Collection<Handle<ST, PK, S>>> chit = getIndexMultiStore().values().iterator();
			while (chit.hasNext()) {
				Collection<Handle<ST, PK, S>> ch = chit.next();
				Iterator<Handle<ST, PK, S>> hit = ch.iterator();
				while (hit.hasNext()) {
					if (!hit.next().isValid()) {
						hit.remove();
					}
				}
				if (ch.isEmpty()) {
					chit.remove();
				}
			}
		}
	}
	
	private Iterable<ST> findMulti(V value) {
		checkCorrupt();
		applyUpdates();
		try {
			if (isUnique()) { // For upgraded
				Handle<ST, PK, S> ret = getIndexUniqueStore().get(value);
				if (ret==null || !ret.isValid()) {
					return Collections.emptyList();
				}
				return Collections.singleton(ret.get());
			}
			
			Collection<Handle<ST, PK, S>> bucket = getIndexMultiStore().get(value);
			if (bucket==null) {
				return Collections.emptyList();
			}
			Collection<ST> ret = new ArrayList<ST>();
			for (Handle<ST,PK,S> h: bucket) {
				if (h.isValid()) {
					ret.add(h.get());
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}
	
	private ST findUnique(V value) {
		checkCorrupt();
		applyUpdates();
		try {
			Handle<ST, PK, S> ret = getIndexUniqueStore().get(value);
			return ret!=null && ret.isValid() ? ret.get() : null;
		} finally {
			readLock().unlock();
		}
	}
	
	@SuppressWarnings("unchecked")
	public Iterable<T> find(V from, V to, boolean fromInclusive, boolean toInclusive) {
		checkCorrupt();
		applyUpdates();
		try {
			Collection<T> ret = new LinkedList<T>();
			// Step 1 - key set to array.
			Collection<V> keys = isUnique() ? getIndexUniqueStore().keySet() : getIndexMultiStore().keySet();
			List<V> keyList = keys instanceof List ? (List<V>) keys : new ArrayList<V>(keys);
			Comparator<V> theComparator = getComparator()==null ? naturalComparator : getComparator();
			int fromIdx = Collections.binarySearch(keyList, from, theComparator);
			if (fromIdx<0) {
				// -Insertion point - 1
				// Inclusiveness doesn't matter
				fromIdx = -fromIdx - 1; 
			} else {
				if (!fromInclusive) {
					++fromIdx;
				}
			}
			
			int toIdx = Collections.binarySearch(keyList, to, theComparator);
			if (toIdx<0) {
				// -Insertion point - 1
				// Inclusiveness doesn't matter
				toIdx = -toIdx - 1; 				
			} else {
				if (!toInclusive) {
					--toIdx;
				}
			}
			
			for (int i=fromIdx; i<=toIdx; ++i) {
				V currentKey = keyList.get(i);
				
				if (isUnique()) {
					Handle<ST, PK, S> handle = getIndexUniqueStore().get(currentKey);
					if (handle.isValid()) {
						ret.add((T) handle.get());
					}
				} else {
					for (Handle<ST,PK,S> h: getIndexMultiStore().get(currentKey)) {
						if (h.isValid()) {
							ret.add((T) h.get());
						}
					}
				}				
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}
	
	private InvocationHandler invocationHandler = new InvocationHandler() {
		
		@SuppressWarnings("unchecked")
		@Override
		public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
			if (method.getName().equals("find") && method.getParameterTypes().length==1) {
				if (UniqueIndex.class.equals(method.getDeclaringClass())) {
					return findUnique((V) args[0]);
				} 
				return findMulti((V) args[0]);
			}
			return method.invoke(AbstractIndex.this, args);
		}
	};
	
	/**
	 * @return proxy which implements interfaces appropriate for the index type.
	 */
	@SuppressWarnings("unchecked")
	public Index<ST, T,PK,V,S> createProxy() {
		if (isUnique()) {
			if (isOrdered()) {
				return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {OrderedIndex.class, UniqueIndex.class}, invocationHandler);
			} else {
				return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {UniqueIndex.class}, invocationHandler);				
			}
		} else {
			if (isOrdered()) {
				return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {OrderedIndex.class, MultiIndex.class}, invocationHandler);
			} else {
				return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {MultiIndex.class}, invocationHandler);				
			}
		}
	}

	@Override
	public String toString() {
		return "AbstractIndex [predicate=" + predicate + ", extractor="
				+ extractor + ", type=" + type + ", ordered=" + ordered
				+ ", comparator=" + comparator + ", isCorrupt=" + isCorrupt
				+ "]";
	}


	
}
