package com.hammurapi.store.local;

import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.WeakHashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.hammurapi.common.ObservableConverter;
import com.hammurapi.common.concurrent.CallerThreadExecutorService;
import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.Predicate;
import com.hammurapi.store.AbstractIndex;
import com.hammurapi.store.AbstractStore;
import com.hammurapi.store.DeputyReadWriteLock;
import com.hammurapi.store.Index.Type;
import com.hammurapi.store.Store;
import com.hammurapi.store.StoreException;
import com.hammurapi.store.TrackingLock;
import com.hammurapi.store.local.LocalHandle.HandleStrength;

/**
 * In-memory object store.
 * @author Pavel Vlasov
 *
 * @param <T>
 */
public abstract class LocalStoreBase<T, PK, S extends Store<T,PK,S>> extends AbstractStore<T, PK, S> {
	
	/**
	 * Store configuration.
	 * @author Pavel Vlasov
	 *
	 */
	public static class Config<T,PK,S extends Store<T,PK,S>> implements Cloneable {
		
		private final HandleFactory<T,PK,S,LocalHandle<T,PK,S>> DEFAULT_HANDLE_FACTORY = new HandleFactory<T, PK, S, LocalHandle<T,PK,S>>() {

			@Override
			public LocalHandle<T, PK, S> createHandle(
					S store,
					T obj,
					PK primaryKey,
					Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache,
					Predicate<T, S>[] validators,
					HandleStrength handleStrength,
					boolean cacheExtracted) {
				return new LocalHandle<T,PK,S>((LocalStoreBase<T, PK, S>) store, obj, primaryKey, cache, validators, handleStrength, cacheExtracted);
			}
			
		};
		
		private long cleanupTaskThreshold = CLEANUP_TASK_THRESHOLD;		
		private ExecutorService executorService;
		private Extractor<T, PK, S> primaryKeyExtractor;
		private boolean cacheExtracted = true;
		private HandleStrength handleStrength = HandleStrength.STRONG;
		private boolean weakPrimaryKey;
		private Map<PK, Handle<T,PK,S>> pkStore;
		private Collection<Handle<T,PK,S>> noPkStore;
		private HandleFactory<T, PK, S, LocalHandle<T,PK,S>> handleFactory = DEFAULT_HANDLE_FACTORY;
		private ObservableConverter<T> observableConverter;
//		private TimeUnit statsTimeUnit;
//		private boolean collectStats;
		
		// TODO Asynch index exception listener.
//		BackingStore<T, PK> backingStore; 
//		int writeBehindBuffer;
//		long writeBehindDelay;
//		TimeUnit writeBehindTimeUnit;
		
//		/**
//		 * If time unit is set, store collects execution time statistics.
//		 */
//		public void setStatsTimeUnit(TimeUnit statsTimeUnit) {
//			this.statsTimeUnit = statsTimeUnit;
//		}
//		
//		public TimeUnit getStatsTimeUnit() {
//			return statsTimeUnit;
//		}
//		
//		/**
//		 * If set to true, store collects execution statistics.
//		 * @param collectStats
//		 */
//		public void setCollectStats(boolean collectStats) {
//			this.collectStats = collectStats;
//		}
//		
//		public boolean isCollectStats() {
//			return collectStats;
//		}
		
		/**
		 * @param cleanupTaskThreshold When number of invalidated (removed) handles exceeds this
		 * threshold, the store performs cleanup of internal collections.
		 */
		public void setCleanupTaskThreshold(long cleanupTaskThreshold) {
			this.cleanupTaskThreshold = cleanupTaskThreshold;
		}
		
		/**
		 * @return Cleanup task threshold.
		 */
		public long getCleanupTaskThreshold() {
			return cleanupTaskThreshold;
		}
		
		@Override
		public Object clone() throws CloneNotSupportedException {
			return super.clone();
		}

		/**
		 * @return Executor service. If null, all operations are performed in the caller thread.
		 */
		public ExecutorService getExecutorService() {
			return executorService;
		}

		/**
		 * 
		 * @param executorService Executor service. If null, all operations are performed in the caller thread.
		 */
		public void setExecutorService(ExecutorService executorService) {
			this.executorService = executorService;
		}

		
		public Extractor<T, PK, S> getPrimaryKeyExtractor() {
			return primaryKeyExtractor;
		}

		/**
		 * @param primaryKeyExtractor Primary key extractor. Can be null.
		 */
		public void setPrimaryKeyExtractor(
				Extractor<T, PK, S> primaryKeyExtractor) {
			this.primaryKeyExtractor = primaryKeyExtractor;
		}

		public boolean isCacheExtracted() {
			return cacheExtracted;
		}

		/**
		 * @param cacheExtracted If true, values extracted from stored objects are cached until object is updated 
		 * through update() or put() methods. Default value is true.
		 */
		public void setCacheExtracted(boolean cacheExtracted) {
			this.cacheExtracted = cacheExtracted;
		}
		
		public HandleStrength getHandleStrength() {
			return handleStrength;
		}

		/**
		 * @param handleStrength Handle strength. Default handle strength is HARD.
		 */
		public void setHandleStrength(HandleStrength handleStrength) {
			this.handleStrength = handleStrength;
		}

		public boolean isWeakPrimaryKey() {
			return weakPrimaryKey;
		}

		/**
		 * @param weakStore If true, the store uses WeakHashMap if primary key is set and
		 * pkStore is not set explicitly.
		 */
		public void setWeakPrimaryKey(boolean weakPrimaryKey) {
			this.weakPrimaryKey = weakPrimaryKey;
		}

		public Map<PK, Handle<T, PK, S>> getPkStore() {
			return pkStore;
		}

		/**
		 * @param pkStore Explicitly provided pkStore instance.
		 */
		public void setPkStore(Map<PK, Handle<T, PK, S>> pkStore) {
			this.pkStore = pkStore;
		}

		public Collection<Handle<T, PK, S>> getNoPkStore() {
			return noPkStore;
		}

		/**
		 * @param noPkStore Explicitly provided noPkStore instance.
		 */
		public void setNoPkStore(Collection<Handle<T, PK, S>> noPkStore) {
			this.noPkStore = noPkStore;
		}
		
		public void setHandleFactory(HandleFactory<T, PK, S, LocalHandle<T, PK, S>> handleFactory) {
			this.handleFactory = handleFactory;
		}
		
		public HandleFactory<T, PK, S, LocalHandle<T, PK, S>> getHandleFactory() {
			return handleFactory;
		}
		
		public void setObservableConverter(ObservableConverter<T> observableConverter) {
			this.observableConverter = observableConverter;
		}
		
		public ObservableConverter<T> getObservableConverter() {
			return observableConverter;
		}
	}

	private static final long CLEANUP_TASK_THRESHOLD = 1000;
	
	protected Config<T,PK,S> config;
	
	@SuppressWarnings("unchecked")
	protected LocalStoreBase(Config<T,PK,S> config) {
		try {
			this.config = (Config<T, PK, S>) config.clone();
		} catch (CloneNotSupportedException e) {
			throw new StoreException(e);
		}
		
		if (this.config.getExecutorService()==null) {
			this.config.setExecutorService(CallerThreadExecutorService.INSTANCE);
		}

		if (this.config.getPrimaryKeyExtractor()==null) {
			if (this.config.getNoPkStore()==null) {
				this.config.setNoPkStore(new LinkedList<Handle<T,PK,S>>() { // TODO - Address uniqueness
					
					// Synchronized for situations when updaters perform inserts.
					public synchronized boolean add(Handle<T,PK,S> e) {
						return super.add(e);
					}
				});				
			}
		} else {
			if (this.config.getPkStore()==null) {
				if (this.config.isWeakPrimaryKey()) {
					this.config.setPkStore(new WeakHashMap<PK, Handle<T,PK,S>>() {
						
						// Synchronized for situations when updaters perform inserts.
						public synchronized Handle<T,PK,S> put(PK key, Handle<T,PK,S> value) {
							return super.put(key, value);
						};
					
					});
				} else {
					this.config.setPkStore(new ConcurrentHashMap<PK, Handle<T,PK,S>>());
				}
			}
		}
		
	}
	
	private ReentrantReadWriteLock storeLock = new ReentrantReadWriteLock();
	private TrackingLock readLock = new LocalTrackingLock(storeLock.readLock());
	private TrackingLock writeLock = new LocalTrackingLock(storeLock.writeLock());

	@Override
	public Lock readLock() {
		return readLock;
	}

	@Override
	public Lock writeLock() {
		return writeLock;
	}

	@Override
	protected Map<PK, Handle<T, PK, S>> getPkStore() {
		return config.getPkStore();
	}
	
	@Override
	protected Collection<Handle<T, PK, S>> getNoPkStore() {
		return getPrimaryKeyExtractor()==null ? config.getNoPkStore() : getPkStore().values();
	}

	@Override
	protected ExecutorService getExecutorService() {
		return config.getExecutorService();
	}
	

	@SuppressWarnings("unchecked")
	@Override
	protected Handle<T, PK, S> createHandle(
			T obj,
			PK primaryKey,
			Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache, 
			Predicate<T, S>[] validators) {
		return config.getHandleFactory().createHandle((S) this, obj, primaryKey, cache, validators, config.getHandleStrength(), config.isCacheExtracted());
	}

	@Override
	protected Collection<Handle<T, PK, S>> createIndexCollection(
			boolean unique, 
			boolean ordered, 
			Comparator<T> comparator) {
		
		throw new UnsupportedOperationException();
	}

	@Override
	protected Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> createCache() {
		if (config.isCacheExtracted()) {
			return new ConcurrentHashMap<S, Map<Extractor<T, ? super PK, S>, ? super PK>>();
		}
		return null;
	}

	@Override
	protected ReadWriteLock createMasterLock() {		
		ReadWriteLock master = new ReadWriteLock() {
			
			@Override
			public Lock writeLock() {
				return writeLock;
			}
			
			@Override
			public Lock readLock() {
				return readLock;
			}
		};
		return new DeputyReadWriteLock(master, readLock.isLocked() || writeLock.isLocked(), writeLock.isLocked());
	}
	
	private AtomicLong removedCounter = new AtomicLong(0);

	/**
	 * If there is a lot of cleared handles this method posts a task to compact store collections..
	 */
	@Override
	protected void onRemoved(int removed) {
		if (removedCounter.addAndGet(removed)>config.getCleanupTaskThreshold()) {
			getExecutorService().submit(new Runnable() {

				@Override
				public void run() {
					writeLock().lock();
					int cleanedUp = 0;
					try {
						Iterator<Handle<T, PK, S>> it = getNoPkStore().iterator(); // Takes care of both map and collection.
						while (it.hasNext()) {
							Handle<T, PK, S> h = it.next();
							if (!h.isValid()) {
								it.remove();
								++cleanedUp;
							}
						}
						
						for (AbstractIndex<T,? extends T,PK,?,S> index: getIndices()) {
							index.cleanup();
						}
					} finally {
						writeLock().unlock();
						if (cleanedUp>0) {
							removedCounter.addAndGet(-cleanedUp);
						}
					}
				}
				
			});
		}
	}

	@Override
	public Extractor<T, PK, S> getPrimaryKeyExtractor() {
		return config.getPrimaryKeyExtractor();
	}
	
	private Collection<AbstractIndex<T, ? extends T, PK, ?, S>> indices = new LinkedList<AbstractIndex<T,? extends T,PK,?,S>>();

	@Override
	protected Collection<AbstractIndex<T, ? extends T, PK, ?, S>> getIndices() {
		return indices;
	}
	
	@Override
	protected <V, ST extends T> AbstractIndex<T, ST, PK, V, S> createIndex(
			Predicate<T, S> predicate,
			Extractor<ST, V, S> extractor, 
			Type type,
			boolean ordered, 
			Comparator<V> comparator) {
		
		return new LocalIndex<T, ST, PK, V, S>(predicate, extractor, type, ordered, comparator, this);
	}
 
	protected void handleIndexAsynchException(Exception e) {
		e.printStackTrace();		
	}
	
	/**
	 * To make it visible to LocalHandle.
	 */
	@Override
	protected void updateIndices(Handle<T, PK, S> handle, boolean isNew) {
		super.updateIndices(handle, isNew);
	}
	
	@Override
	protected abstract S createDeputy();
}
