/*
 * This file is part of Nuts Framework.
 * Copyright (C) 2009 Nuts Develop Team.
 *
 * Nuts Framework is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License any later version.
 * 
 * Nuts Framework is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Nuts Framework. If not, see <http://www.gnu.org/licenses/>.
 */
package nuts.gae.dao;

import nuts.core.dao.DaoException;
import nuts.core.dao.DaoClient;
import nuts.core.dao.DaoSession;
import nuts.core.dao.DataHandler;
import nuts.core.dao.ModelDAO;
import nuts.core.dao.ModelMetaData;
import nuts.core.lang.Strings;
import nuts.core.lang.Types;
import nuts.core.log.Log;
import nuts.core.log.Logs;
import nuts.exts.fileupload.UploadFile;

import java.io.IOException;
import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import com.google.appengine.api.datastore.Blob;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.FetchOptions;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.PreparedQuery;
import com.google.appengine.api.datastore.Query;
import com.google.appengine.api.datastore.Text;
import com.google.appengine.api.datastore.Transaction;

/**
 */
@SuppressWarnings("unchecked")
public class GaeDataAccessSession implements DaoSession, DatastoreServiceAware {
	private static Log log = Logs.getLog(GaeDataAccessSession.class);

	private static final String ROOT = "_nuts_root";
	
	private static final int MAX_SL = 500;

	//private static final int MAX_BL = 1048576; //1M

	//private static final int MAX_TL = 1048576; //1M

	private static Map<String, Entity> roots;
	
	private GaeDataAccessClient client;
	private DatastoreService service;
	private Transaction transaction;
	private Entity rootEntity;
	
	private boolean autoCommit = false;
	@SuppressWarnings("unused")
	private boolean dirty = false;

	/**
	 * @param service DatastoreService
	 * @param client DataAccessClient
	 */
	public GaeDataAccessSession(GaeDataAccessClient client, DatastoreService service, boolean autoCommit) {
		super();
		this.client = client;
		this.service = service;
		this.autoCommit = autoCommit;
	}
	
	/**
	 * @return Data Access Client
	 */
	public DaoClient getDaoClient() {
		return client;
	}
	
	/**
	 * @return model meta data map
	 */
	public Map<String, ModelMetaData> getMetaDataMap() {
		return client.getMetaDataMap();
	}
	
	/**
	 * @param name model name
	 * @return model meta data
	 */
	public ModelMetaData getMetaData(String name) {
		return client.getMetaData(name);
	}

	/**
	 * @param name model name
	 * @return modelDao
	 */
	public ModelDAO getModelDAO(String name) {
		return client.getModelDAO(name, this);
	}
	
	/**
	 * @return the datastoreService
	 */
	public DatastoreService getDatastoreService() {
		return service;
	}

	/**
	 * @param datastoreService the datastoreService to set
	 */
	public void setDatastoreService(DatastoreService datastoreService) {
		this.service = datastoreService;
	}
	
	private Entity getRootEntity() {
		if (rootEntity == null) {
			if (roots == null) {
				synchronized (GaeDataAccessSession.class) {
					roots = new HashMap<String, Entity>();
					Query q = new Query(ROOT);
					PreparedQuery pq = service.prepare(q);
					if (log.isDebugEnabled()) {
						log.debug("QUERY: " + q);
					}
					Iterator<Entity> it = pq.asIterator();
					while (it.hasNext()) {
						Entity en = it.next();
						if (log.isDebugEnabled()) {
							log.debug("GET ROOT: " + en);
						}
						roots.put(en.getKey().getName(), en);
					}
				}
			}

			rootEntity = roots.get(client.getName());
			if (rootEntity == null) {
				synchronized (GaeDataAccessSession.class) {
					Transaction transaction = service.beginTransaction();
					Key key = KeyFactory.createKey(ROOT, client.getName());
					rootEntity = new Entity(key);
					if (log.isDebugEnabled()) {
						log.debug("PUT ROOT: " + rootEntity);
					}
					service.put(transaction, rootEntity);
					transaction.commit();
					roots.put(key.getName(), rootEntity);
				}
			}
		}
		return rootEntity;
	}

	private Key getRootKey() {
		return getRootEntity().getKey();
	}

	/**
	 * 
	 */
	public void commit() throws DaoException {
		try {
			if (transaction != null/* && dirty*/) {
				transaction.commit();
			}
		}
		finally {
			transaction = null;
			dirty = false;
		}
	}

	/**
	 * 
	 */
	public void rollback() throws DaoException {
		try {
			if (transaction != null/* && dirty*/) {
				transaction.rollback();
			}
		}
		finally {
			transaction = null;
			dirty = false;
		}
	}

	/**
	 * close
	 */
	public void close() {
		try {
			commit();
		}
		catch (DaoException e) {
			log.warn(e.getMessage());
		}
	}

	/**
	 * @return true if session is closed
	 */
	public boolean isClosed() {
		return false;
	}

	/**
	 * auto start transaction
	 * @throws DaoException if a data access error occurs
	 */
	private void autoStartTransaction() throws DaoException {
		if (transaction == null) {
			transaction = service.beginTransaction();
		}
	}
	
	/**
	 * auto commit transaction
	 * @throws DaoException if a data access error occurs
	 */
	private void autoCommitTransaction() throws DaoException {
		if (transaction == null) {
			throw new DaoException("No active transaction");
		}

		if (autoCommit) {
			try {
				transaction.commit();
			}
			finally {
				transaction = null;
				dirty = false;
			}
		}
		else {
			dirty = true;
		}
	}

	/**
	 * auto end transaction
	 * @throws DaoException if a data access error occurs
	 */
	private void autoEndTransaction() throws DaoException {
	}

	private Object createModelData(ModelMetaData mmd) throws DaoException {
		try {
			return mmd.getModelBeanClass().newInstance();
		}
		catch (Exception e) {
			throw new DaoException(e);
		}
	}
	
	private Key createKey(ModelMetaData mmd, Object id) {
		if (id instanceof Long) {
			return KeyFactory.createKey(getRootKey(), mmd.getTableName(), (Long)id);
		}
		else {
			return KeyFactory.createKey(getRootKey(), mmd.getTableName(), (String)id);
		}
	}

	private boolean isValidIdentity(Object id) {
		if (id == null) {
			return false;
		}
		else if (id instanceof Number) {
			return ((Number)id).longValue() > 0;
		}
		else {
			return id.toString().length() > 0;
		}
	}
	
	private Object getDataIdentity(ModelMetaData mmd, Object data) {
		String in = mmd.getIdentityName();
		if (Strings.isEmpty(in)) {
			throw new IllegalArgumentException(data.getClass().getName() + " has no identity");
		}
		Object id = mmd.getPropertyValue(data, in);
		return id;
	}

	private void setDataIdentity(ModelMetaData mmd, Object data, Key key) {
		String in = mmd.getIdentityName();
		if (Strings.isEmpty(in)) {
			throw new IllegalArgumentException(data.getClass().getName() + " has no identity");
		}

		Type type = mmd.getPropertyType(data, in);
		if (Types.isAssignable(type, String.class)) {
			mmd.setPropertyValue(data, in, key.getName());
		}
		else {
			Object id = convertValueFromGae(mmd, data, in, key.getId());
			mmd.setPropertyValue(data, in, id);
		}
	}

	private void addNullExcludes(ModelMetaData mmd, Object data, Collection<String> excludes) {
		String[] fns = mmd.getFieldNames();
		for (String fn : fns) {
			String pn = mmd.getPropertyName(fn);
			if (Strings.isNotEmpty(fn)) {
				Object v = mmd.getPropertyValue(data, pn);
				if (v == null) {
					excludes.add(fn);
				}
			}
		}
	}

	private void addPKeyExcludes(ModelMetaData mmd, Object data, Collection<String> excludes) {
		String[] pns = mmd.getPrimaryKeys();
		for (String pn : pns) {
			String fn = mmd.getFieldName(pn);
			excludes.add(fn);
		}
	}

	private Object convertValueFromGae(ModelMetaData mmd, Object data,
			String toProperty, Object value) throws DaoException {
		
		if (value instanceof Text) {
			value = ((Text)value).getValue();
		}
		else if (value instanceof Blob) {
			value = ((Blob)value).getBytes();
		}

		Type toType = mmd.getPropertyType(data, toProperty);
		if (!Types.isInstance(value, toType)) {
			value = client.getCastors().cast(value, toType);
		}

		return value;
	}

	private Object convertValueToGae(Object value) throws DaoException {
		if (value instanceof String) {
			if (((String)value).length() > MAX_SL) {
				value = new Text((String)value);
			}
		}
		else if (value instanceof UploadFile) {
			try {
				byte[] b = ((UploadFile)value).getData();
				if (b != null && b.length > 0) {
					value = new Blob(b);
				}
				else {
					value = null;
				}
			}
			catch (IOException e) {
				throw new DaoException(e);
			}
		}
		else if (value instanceof byte[]) {
			value = new Blob((byte[])value);
		}
		else if (value instanceof BigInteger) {
			value = ((BigInteger)value).longValue();
		}
		else if (value instanceof BigDecimal) {
			value = ((BigDecimal)value).doubleValue();
		}
		return value;
	}

	private void convertEntityToData(ModelMetaData mmd, Entity entity,
			Object data, Collection<String> excludes)
			throws DaoException {

		Map<String, Object> eps = entity.getProperties();
		for (Entry<String, Object> en : eps.entrySet()) {
			String fn = en.getKey();
			if (excludes == null || !excludes.contains(fn)) {
				String p = mmd.getPropertyName(fn);
				if (Strings.isNotEmpty(p)) {
					Object v = en.getValue();
					v = convertValueFromGae(mmd, data, p, v);
					mmd.setPropertyValue(data, p, v);
				}
			}
		}

		setDataIdentity(mmd, data, entity.getKey());
	}
	
	private void convertDataToEntity(ModelMetaData mmd, Object data,
			Entity entity, Collection<String> excludes)
			throws DaoException {

		//reserve exclude properties
		Map<String, Object> eps = entity.getProperties();
		for (String k : eps.keySet()) {
			if (excludes == null || !excludes.contains(k)) {
				entity.removeProperty(k);
			}
		}
		
		Map<String, Object> dps = mmd.getDataProperties(data);

		//set properties except excludes
		for (Entry<String, Object> en : dps.entrySet()) {
			String pn = en.getKey();
			String fn = mmd.getFieldName(pn);
			if (Strings.isNotEmpty(fn) 
					&& (excludes == null || !excludes.contains(fn))) {
				Object v = en.getValue();
				v = convertValueToGae(v);
				entity.setProperty(fn, v);
			}
		}
	}

	private void saveEntity(Entity entity) throws DaoException {
		if (log.isDebugEnabled()) {
			log.debug("PUT: " + entity);
		}
		service.put(transaction, entity);
	}

	private Entity getEntity(Key key) throws DaoException {
		Entity entity = null;

		try {
			entity = service.get(transaction, key);
		}
		catch (EntityNotFoundException e) {
		}
		finally {
			if (log.isDebugEnabled()) {
				log.debug("GET: (" + key + ") - " + entity);
			}
		}
		
		return entity;
	}

	private void deleteEntity(Key key) throws DaoException {
		if (log.isDebugEnabled()) {
			log.debug("DELETE: (" + key + ")");
		}
		service.delete(transaction, key);
	}

	@SuppressWarnings("unused")
	private void deleteEntity(Iterable<Key> keys) throws DaoException {
		if (log.isDebugEnabled()) {
			log.debug("DELETE: [" + keys + "]");
		}
		service.delete(transaction, keys);
	}

	private PreparedQuery prepareQuery(Query query) {
		query.setAncestor(getRootKey());
		if (log.isDebugEnabled()) {
			log.debug("QUERY: " + query);
		}
		return service.prepare(transaction, query);
	}

	private FetchOptions getFetchOptions(GaeQueryParameter gqp) {
		FetchOptions fo = FetchOptions.Builder.withDefaults();
		if (gqp.getStart() != null) {
			fo.offset(gqp.getStart());
		}
		if (gqp.getLimit() != null) {
			fo.limit(gqp.getLimit());
		}
		return fo;
	}
	
	/**
	 * exists
	 * 
	 * @param mid modelId
	 * @param key Object
	 * @return Object
	 * @throws DaoException if a data access error occurs
	 */ 
	public boolean exists(String mid, Object key) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			Object kid = getDataIdentity(mmd, key);
			if (isValidIdentity(kid)) {
				Key k = createKey(mmd, kid);
				Entity entity = getEntity(k);
				return entity != null;
			}
			return false;
		}
		finally {
			autoEndTransaction();
		}
	}
	
	/**
	 * countByExample
	 * 
	 * @param mid modelId
	 * @param gqp GaeQueryParameter
	 * @return count
	 * @throws DaoException if a data access error occurs
	 */ 
	public int countByExample(String mid, GaeQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			PreparedQuery pq = prepareQuery(gqp.getQuery());
			FetchOptions fo = getFetchOptions(gqp);
			int cnt = pq.countEntities(fo);
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * selectByPrimaryKey
	 * 
	 * @param mid modelId
	 * @param key Object
	 * @return Object
	 * @throws DaoException if a data access error occurs
	 */ 
	public Object selectByPrimaryKey(String mid, Object key) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			Object kid = getDataIdentity(mmd, key);
			if (isValidIdentity(kid)) {
				Key k = createKey(mmd, kid);
				Entity entity = getEntity(k);
				if (entity != null) {
					Object data = createModelData(mmd);
					convertEntityToData(mmd, entity, data, null);
					return data;
				}
			}
			return null;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * selectByExampleWithDataHandler
	 * 
	 * @param mid modelId
	 * @param gqp GaeQueryParameter
	 * @param dataHandler data handler
	 * @return data count
	 * @throws DaoException if a data access error occurs
	 */ 
	public int selectByExampleWithDataHandler(String mid, GaeQueryParameter gqp, DataHandler dataHandler) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			PreparedQuery pq = prepareQuery(gqp.getQuery());
			FetchOptions fo = getFetchOptions(gqp);
			Iterator<Entity> it = pq.asIterator(fo);
			int count = 0;
			while (it.hasNext()) {
				Entity en = it.next();
				Object data = createModelData(mmd);
				convertEntityToData(mmd, en, data, gqp.getExcludes().keySet());
				try {
					boolean next = dataHandler.handleData(data);
					count++;
					if (!next) {
						break;
					}
				}
				catch (Exception ex) {
					throw new DaoException("Data Handle Error: " + data, ex);
				}
			}
			return count;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * selectByExample
	 * 
	 * @param mid modelId
	 * @param gqp GaeQueryParameter
	 * @return list of Object 
	 * @throws DaoException if a data access error occurs
	 */ 
	public List selectByExample(String mid, GaeQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			PreparedQuery pq = prepareQuery(gqp.getQuery());
			FetchOptions fo = getFetchOptions(gqp);
			Iterator<Entity> it = pq.asIterator(fo);
			List list = new ArrayList();
			while (it.hasNext()) {
				Entity en = it.next();
				Object data = createModelData(mmd);
				convertEntityToData(mmd, en, data, gqp.getExcludes().keySet());
				list.add(data);
			}
			return list;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * save
	 * 
	 * @param mid modelId
	 * @param data data
	 * @throws DaoException if a data access error occurs
	 */ 
	public void save(String mid, Object data) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);

			Entity entity;
			Object kid = getDataIdentity(mmd, data);
			if (isValidIdentity(kid)) {
				Key key = createKey(mmd, kid);
				entity = getEntity(key);
				if (entity != null) {
					convertDataToEntity(mmd, data, entity, null);
					saveEntity(entity);
				}
				else {
					entity = new Entity(key);
					convertDataToEntity(mmd, data, entity, null);
					saveEntity(entity);
				}
			}
			else {
				entity = new Entity(mmd.getTableName(), getRootKey());

				saveEntity(entity);
				setDataIdentity(mmd, data, entity.getKey());

				convertDataToEntity(mmd, data, entity, null);
				saveEntity(entity);
			}
			autoCommitTransaction();
		}
		finally {
			autoEndTransaction();
		}
	}
	
	/**
	 * insert
	 * 
	 * @param mid modelId
	 * @param data data
	 * @throws DaoException if a data access error occurs
	 */ 
	public void insert(String mid, Object data) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);

			Entity entity;
			Object kid = getDataIdentity(mmd, data);
			if (isValidIdentity(kid)) {
				Key key = createKey(mmd, kid);
				entity = new Entity(key); 
			}
			else {
				entity = new Entity(mmd.getTableName(), getRootKey());
				saveEntity(entity);
				setDataIdentity(mmd, data, entity.getKey());
			}

			convertDataToEntity(mmd, data, entity, null);
			saveEntity(entity);
			autoCommitTransaction();
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * deleteByPrimaryKey
	 * 
	 * @param mid modelId
	 * @param key Object
	 * @return count of deleted records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int deleteByPrimaryKey(String mid, Object key) throws DaoException {
		try {
			int cnt = 0;
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			Object kid = getDataIdentity(mmd, key);
			if (isValidIdentity(kid)) {
				Key k = createKey(mmd, kid);
				Entity entity = getEntity(k);
				if (entity != null) {
					deleteEntity(k);
					cnt = 1;
				}
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * deleteByExample
	 * 
	 * @param mid modelId
	 * @param gqp GaeQueryParameter
	 * @return count of deleted records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int deleteByExample(String mid, GaeQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			Query q = gqp.getQuery();
			q.setKeysOnly();
			PreparedQuery pq = prepareQuery(q);
			FetchOptions fo = getFetchOptions(gqp);
			
			Iterator<Entity> it = pq.asIterator(fo);
			int cnt = 0;
			while (it.hasNext()) {
				deleteEntity(it.next().getKey());
				cnt++;
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * updateByPrimaryKey
	 * 
	 * @param mid modelId
	 * @param data data
	 * @return count of updated records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int updateByPrimaryKey(String mid, Object data) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			Object kid = getDataIdentity(mmd, data);
			int cnt = 0;
			if (isValidIdentity(kid)) {
				Key k = createKey(mmd, kid);
				Entity entity = getEntity(k);
				if (entity != null) {
					convertDataToEntity(mmd, data, entity, null);
					saveEntity(entity);
					cnt = 1;
				}
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * updateByPrimaryKeySelective (ignore null properties)
	 * 
	 * @param mid modelId
	 * @param data data
	 * @return count of updated records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int updateByPrimaryKeySelective(String mid, Object data) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			Object kid = getDataIdentity(mmd, data);
			int cnt = 0;
			if (isValidIdentity(kid)) {
				Key k = createKey(mmd, kid);
				Entity entity = getEntity(k);
				Set<String> excludes = new HashSet<String>();
				addNullExcludes(mmd, data, excludes);
				convertDataToEntity(mmd, data, entity, excludes);
				saveEntity(entity);
				cnt = 1;
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * updateByExample
	 * 
	 * @param mid modelId
	 * @param data data
	 * @param gqp GaeQueryParameter
	 * @return count of updated records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int updateByExample(String mid, Object data, GaeQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);

			Set<String> excludes = new HashSet<String>();
			excludes.addAll(gqp.getExcludes().keySet());
			addPKeyExcludes(mmd, data, excludes);

			Query q = gqp.getQuery();
			PreparedQuery pq = prepareQuery(q);
			FetchOptions fo = getFetchOptions(gqp);
			
			Iterator<Entity> it = pq.asIterator(fo);
			int cnt = 0;
			while (it.hasNext()) {
				Entity entity = it.next();

				convertDataToEntity(mmd, data, entity, excludes);

				saveEntity(entity);
				
				cnt++;
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * updateByExampleSelective (ignore null properties)
	 * 
	 * @param mid modelId
	 * @param data data
	 * @param gqp GaeQueryParameter
	 * @return count of updated records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int updateByExampleSelective(String mid, Object data, GaeQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);

			Set<String> excludes = new HashSet<String>();
			addPKeyExcludes(mmd, data, excludes);
			addNullExcludes(mmd, data, excludes);

			Query q = gqp.getQuery();
			PreparedQuery pq = prepareQuery(q);
			FetchOptions fo = getFetchOptions(gqp);
			
			Iterator<Entity> it = pq.asIterator(fo);
			int cnt = 0;
			while (it.hasNext()) {
				Entity entity = it.next();

				convertDataToEntity(mmd, data, entity, excludes);

				saveEntity(entity);
				
				cnt++;
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}
}
