package org.apache.wicket.protocol.http.pagestore;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import javax.servlet.http.HttpSession;

import org.apache.wicket.IClusterable;
import org.apache.wicket.Page;
import org.apache.wicket.RequestCycle;
import org.apache.wicket.WicketRuntimeException;
import org.apache.wicket.protocol.http.WebRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OurTerracottaPageStore extends AbstractPageStore {

	private static final Logger logger = LoggerFactory.getLogger(OurTerracottaPageStore.class);
	private static final String TERRACOTTA_SESSION_KEY = "TERRACOTTA_SESSION_KEY";
	private final int MAX_PAGES_PER_MAP;

	private static final int NO_MAX_PAGES_PER_MAP = -99;

	public OurTerracottaPageStore() {
		if (logger.isDebugEnabled()) {
			logger.debug("New Terracotta Page Store, MAX_PAGES_PER_MAP is Unlimited");
		}
		MAX_PAGES_PER_MAP = OurTerracottaPageStore.NO_MAX_PAGES_PER_MAP;
	}

	public OurTerracottaPageStore(final int maxPagesPerMap) {
		if (logger.isDebugEnabled()) {
			logger.debug("New Terracotta Page Store, MAX_PAGES_PER_MAP is " + maxPagesPerMap);
		}
		MAX_PAGES_PER_MAP = maxPagesPerMap;
	}

	public boolean containsPage(final String sessionId, final String pageMapName, final int pageId, final int pageVersion) {
		return getPageStore(sessionId).getPageMap(pageMapName).containsPage(pageId, pageVersion);
	}

	public void destroy() {
		// nothing to do - PageSores will be destroyed with their sessions
	}

	public Page getPage(final String sessionId, final String pagemap, final int id, final int versionNumber, final int ajaxVersionNumber) {
		final SerializedPage sPage = getPageStore(sessionId).getPageMap(pagemap)
				.getPage(id, versionNumber, ajaxVersionNumber);
		return sPage != null ? deserializePage(sPage.getData(), versionNumber) : null;
	}

	public void pageAccessed(final String sessionId, final Page page) {
		// do nothing
	}

	public void removePage(final String sessionId, final String pagemap, final int id) {
		if (id == -1) {
			if (logger.isDebugEnabled()) {
				logger.debug("Remove page map: " + pagemap);
			}
			getPageStore(sessionId).removePageMap(pagemap);
		} else {
			if (logger.isDebugEnabled()) {
				logger.debug("Remove page: " + id + " from page map " + pagemap);
			}
			getPageStore(sessionId).getPageMap(pagemap).removePage(id);
		}
	}

	@SuppressWarnings("unchecked")
	public void storePage(final String sessionId, final Page page) {
		getPageStore(sessionId).getPageMap(page.getPageMapName()).storePages(serializePage(page));
		if (logger.isDebugEnabled()) {
			logger.debug("Store page: " + page.toString());
			logger.debug(getPageStore(sessionId).getPageMap(page.getPageMapName()).toString());

		}
	}

	public void unbind(final String sessionId) {
		// do nothing: this is only called on session invalidation which will
		// take care anyway
		// getHttpSession(sessionId).removeAttribute(TERRACOTTA_SESSION_KEY);
	}

	protected PageMapStore getPageStore(final String sessionId) {
		final HttpSession session = getHttpSession(sessionId);
		PageMapStore store = (PageMapStore) session.getAttribute(TERRACOTTA_SESSION_KEY);
		if (store == null) {
			session
					.setAttribute(TERRACOTTA_SESSION_KEY, store = new PageMapStore(MAX_PAGES_PER_MAP));
		}
		return store;
	}

	protected HttpSession getHttpSession(final String sessionId) {
		final WebRequest request = (WebRequest) RequestCycle.get().getRequest();
		final HttpSession session = request.getHttpServletRequest().getSession();
		if (sessionId != null && !sessionId.equals(session.getId())) {
			// TODO will this ever happen? is there a way to get other session
			// ids?
			throw new WicketRuntimeException("wrong sessionId " + sessionId + ", expected "
					+ session.getId());
		}
		return session;
	}

	protected static class PageMapStore implements IClusterable {
		private static final long serialVersionUID = 1L;

		private final Map<String, PageStore> _pageMaps = new HashMap<String, PageStore>();
		private final int MAX_PAGES_PER_MAP;

		public PageMapStore(final int maxNumPagesPerMap) {
			MAX_PAGES_PER_MAP = maxNumPagesPerMap;
		}

		public PageStore getPageMap(final String pageMapName) {
			PageStore toReturn;
			if (_pageMaps.containsKey(pageMapName)) {
				toReturn = _pageMaps.get(pageMapName);
			} else {
				_pageMaps.put(pageMapName, toReturn = new PageStore(MAX_PAGES_PER_MAP));
			}
			return toReturn;
		}

		public void removePageMap(final String pagemap) {
			_pageMaps.remove(pagemap);
		}

		@Override
		public String toString() {
			final StringBuilder sb = new StringBuilder();
			for (final Entry<String, PageStore> entry : _pageMaps.entrySet()) {
				sb.append("PageMap: ").append(entry.getKey()).append("\n");
				sb.append(entry.getValue().toString());
			}
			return sb.toString();
		}
	}

	protected static class PageStore implements IClusterable {
		private static final long serialVersionUID = 1L;

		private final ReentrantReadWriteLock _pagesLock = new ReentrantReadWriteLock();
		private final LinkedHashMap<PageKey, SerializedPage> _pages = new LinkedHashMap<PageKey, SerializedPage>();
		private final TreeMap<PageKey, Integer> _pageKeys = new TreeMap<PageKey, Integer>();
		// if we have an overflow, we probably had a 100.000 years uptime,
		// hooray! :)
		private Integer _id = Integer.MIN_VALUE;

		private final int MAX_SIZE;

		public PageStore(final int maxSize) {
			MAX_SIZE = maxSize;
		}

		public void storePages(final List<SerializedPage> pagesToAdd) {
			_pagesLock.writeLock().lock();
			try {
				// reduce size of page store to within set size if required
				if (MAX_SIZE != NO_MAX_PAGES_PER_MAP) {
					int numToRemove = _pages.size() + pagesToAdd.size() - MAX_SIZE;
					if (numToRemove > 0) {
						final Iterator<Entry<PageKey, SerializedPage>> iter = _pages.entrySet()
								.iterator();
						while (iter.hasNext() && numToRemove > 0) {
							final Entry<PageKey, SerializedPage> entry = iter.next();
							iter.remove();
							_pageKeys.remove(entry.getKey());
							numToRemove--;
						}
					}
				}
				for (final SerializedPage sPage : pagesToAdd) {
					final PageKey pageKey = new PageKey(sPage.getPageId(), sPage.getVersionNumber(), sPage
							.getAjaxVersionNumber());
					// remove to preserve access order
					_pages.remove(pageKey);
					_pages.put(pageKey, sPage);
					_pageKeys.put(pageKey, _id++);
				}

			} finally {
				_pagesLock.writeLock().unlock();
			}

		}

		public boolean containsPage(final int pageId, final int pageVersion) {
			_pagesLock.readLock().lock();
			try {
				// make PageKeys for below and above this version
				// this id and version but -1 for ajax
				final PageKey below = new PageKey(pageId, pageVersion, -1);
				// this id and version +1, -1 for ajax
				final PageKey above = new PageKey(pageId, pageVersion + 1, -1);

				final SortedMap<PageKey, Integer> thisPageAndVersion = _pageKeys
						.subMap(below, above);

				return !thisPageAndVersion.isEmpty();

			} finally {
				_pagesLock.readLock().unlock();
			}
		}

		public SerializedPage getPage(final int id, final int versionNumber, final int ajaxVersionNumber) {
			_pagesLock.readLock().lock();
			try {
				SerializedPage sPage = null;
				// just find the exact page version
				if (versionNumber != -1 && ajaxVersionNumber != -1) {
					sPage = _pages.get(new PageKey(id, versionNumber, ajaxVersionNumber));
				}
				// we need to find last recently stored page window - that is
				// page
				// at the end of the list
				else if (versionNumber == -1) {
					final PageKey fromKey = new PageKey(id, -1, -1);
					final PageKey toKey = new PageKey(id + 1, -1, -1);

					final Iterator<Entry<PageKey, Integer>> iter = _pageKeys.subMap(fromKey, toKey)
							.entrySet().iterator();
					int max = -1;
					PageKey maxPageKey = null;
					while (iter.hasNext()) {
						final Entry<PageKey, Integer> entry = iter.next();
						if (entry.getKey().getId() == id) {
							if (entry.getValue() > max) {
								max = entry.getValue();
								maxPageKey = entry.getKey();
							}
						}
					}
					if (maxPageKey != null) {
						sPage = _pages.get(maxPageKey);
					}
				}
				// we need to find index with highest ajax version
				else if (ajaxVersionNumber == -1) {
					// make a page key which will be straight after wanted
					// PageKey, ie, version number is one after this one, ajax
					// version
					// number is -1, page id is the same
					final PageKey toElement = new PageKey(id, versionNumber + 1, -1);
					final SortedMap<PageKey, Integer> posiblePageKeys = _pageKeys
							.headMap(toElement);
					sPage = _pages.get(posiblePageKeys.lastKey());

				}
				return sPage;
			} finally {
				_pagesLock.readLock().unlock();
			}
		}

		public void removePage(final int id) {
			_pagesLock.writeLock().lock();
			try {
				final Iterator<Entry<PageKey, SerializedPage>> iter = _pages.entrySet().iterator();
				while (iter.hasNext()) {
					final PageKey pKey = iter.next().getKey();
					if (id == pKey.getId()) {
						iter.remove();
						_pageKeys.remove(pKey);
					}
				}
			} finally {
				_pagesLock.writeLock().unlock();
			}
		}

		@Override
		public String toString() {
			final StringBuilder sb = new StringBuilder();
			final Iterator<Entry<PageKey, SerializedPage>> iter = _pages.entrySet().iterator();
			while (iter.hasNext()) {
				final Entry<PageKey, SerializedPage> entry = iter.next();
				sb.append("\t").append(entry.getKey().toString()).append("\n");
			}
			if (logger.isTraceEnabled()) {
				sb.append("\tPageKeys TreeSet: ").append(_pageKeys.toString());
			}
			return sb.toString();
		}

	}

	protected static class PageKey implements IClusterable, Comparable<PageKey> {

		private static final long serialVersionUID = 1L;

		private final int _id;
		private final int _versionNumber;
		private final int _ajaxVersionNumber;

		public PageKey(final int id, final int versionNumber, final int ajaxVersionNumber) {
			_id = id;
			_versionNumber = versionNumber;
			_ajaxVersionNumber = ajaxVersionNumber;
		}

		public int getId() {
			return _id;
		}

		public int getVersionNumber() {
			return _versionNumber;
		}

		public int getAjaxVersionNumber() {
			return _ajaxVersionNumber;
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = prime + _ajaxVersionNumber;
			result = prime * result + _id;
			result = prime * result + _versionNumber;
			return result;
		}

		@Override
		public boolean equals(final Object obj) {
			if (this == obj) {
				return true;
			}
			if (obj == null) {
				return false;
			}
			if (obj instanceof PageKey == false) {
				return false;
			}
			final PageKey other = (PageKey) obj;
			if (_ajaxVersionNumber != other._ajaxVersionNumber) {
				return false;
			}
			if (_id != other._id) {
				return false;
			}
			if (_versionNumber != other._versionNumber) {
				return false;
			}
			return true;
		}

		@Override
		public String toString() {
			return "PageID: " + _id + " \tVersion: " + _versionNumber + " \tAjax: "
					+ _ajaxVersionNumber;
		}

		// if this is less return - number
		public int compareTo(final PageKey o) {
			if (_id != o.getId()) {
				return _id - o.getId();
			} else if (_versionNumber != o.getVersionNumber()) {
				return _versionNumber - o.getVersionNumber();
			} else if (_ajaxVersionNumber != o.getAjaxVersionNumber()) {
				return _ajaxVersionNumber - o.getAjaxVersionNumber();
			}
			return 0;
		}

	}
}