/*
 * This file is part of Nuts Framework.
 * Copyright(C) 2009-2012 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.aws.dao;

import nuts.core.dao.DaoClient;
import nuts.core.dao.DaoException;
import nuts.core.dao.DaoSession;
import nuts.core.dao.DataHandler;
import nuts.core.dao.ModelDAO;
import nuts.core.dao.ModelMetaData;
import nuts.core.dao.sql.SqlUtils;
import nuts.core.dao.sql.criteria.SqlQueryParameter;
import nuts.core.lang.Strings;
import nuts.core.lang.Types;
import nuts.core.lang.codec.binary.Base64;
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.util.ArrayList;
import java.util.Calendar;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import com.amazonaws.services.simpledb.AmazonSimpleDB;
import com.amazonaws.services.simpledb.model.Attribute;
import com.amazonaws.services.simpledb.model.DeleteAttributesRequest;
import com.amazonaws.services.simpledb.model.Item;
import com.amazonaws.services.simpledb.model.PutAttributesRequest;
import com.amazonaws.services.simpledb.model.ReplaceableAttribute;
import com.amazonaws.services.simpledb.model.SelectRequest;
import com.amazonaws.services.simpledb.model.SelectResult;

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

	private AwsDataAccessClient client;
	private AmazonSimpleDB simpleDB;
	
	/**
	 * @param simpleDB AmazonSimpleDB
	 * @param client DataAccessClient
	 */
	public AwsDataAccessSession(AwsDataAccessClient client, AmazonSimpleDB simpleDB, boolean autoCommit) {
		super();
		this.client = client;
		this.simpleDB = simpleDB;
	}
	
	/**
	 * @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 simpleDB
	 */
	public AmazonSimpleDB getAmazonSimpleDB() {
		return simpleDB;
	}

	/**
	 * @param simpleDB the simpleDB to set
	 */
	public void setAmazonSimpleDB(AmazonSimpleDB simpleDB) {
		this.simpleDB = simpleDB;
	}
	
	/**
	 * 
	 */
	public void commit() throws DaoException {
	}

	/**
	 * 
	 */
	public void rollback() throws DaoException {
	}

	/**
	 * close
	 */
	public void close() {
	}

	/**
	 * @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 {
	}
	
	/**
	 * auto commit transaction
	 * @throws DaoException if a data access error occurs
	 */
	private void autoCommitTransaction() throws DaoException {
	}

	/**
	 * 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 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, String id) {
		String in = mmd.getIdentityName();
		if (Strings.isEmpty(in)) {
			throw new IllegalArgumentException(data.getClass().getName() + " has no identity");
		}
		
		Type toType = mmd.getPropertyType(in);
		if (Types.isAssignable(toType, String.class)) {
			mmd.setPropertyValue(data, in, id);
		}
		else {
			Object cv = convertValueFromAws(mmd, data, in, id);
			mmd.setPropertyValue(data, in, cv);
		}
	}


	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 convertValueFromAws(ModelMetaData mmd, Object data,
			String toProperty, String value) throws DaoException {

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

		return value;
	}

	private String convertValueToAws(Object value) throws DaoException {
		String cv = "";
		if (value instanceof UploadFile) {
			try {
				byte[] b = ((UploadFile)value).getData();
				if (b != null && b.length > 0) {
					cv = Base64.encodeBase64String(b);
				}
			}
			catch (IOException e) {
				throw new DaoException(e);
			}
		}
		else if (value instanceof byte[]) {
			cv = Base64.encodeBase64String((byte[])value);
		}
		else if (value instanceof Date) {
			cv = String.valueOf(((Date)value).getTime());
		}
		else if (value instanceof Calendar) {
			cv = String.valueOf(((Calendar)value).getTime());
		}
		else if (value != null) {
			cv = value.toString();
		}
		return cv;
	}

	private Map<String, String> getItemProperties(Item item) {
		Map<String, String> ps = new HashMap<String, String>();
		List<Attribute> as = item.getAttributes();
		for (Attribute a : as) {
			ps.put(a.getName(), a.getValue());
		}
		return ps;
	}
	
	private void convertItemToData(ModelMetaData mmd, Item item,
			Object data, Collection<String> excludes)
			throws DaoException {
		List<Attribute> as = item.getAttributes();
		Map<String, Object> dps = new HashMap<String, Object>(as.size());

		for (Attribute a : as) {
			String fn = a.getName();
			if (excludes == null || !excludes.contains(fn)) {
				String p = mmd.getPropertyName(fn);
				Object v = convertValueFromAws(mmd, data, p, a.getValue());
				dps.put(p, v);
			}
		}

		mmd.setDataProperties(data, dps);

		setDataIdentity(mmd, data, item.getName());
	}
	
	private List<ReplaceableAttribute> convertDataToAttributes(ModelMetaData mmd, Object data,
			Item item, Collection<String> excludes)
			throws DaoException {

//		//reserve exclude attributes
//		List<Attribute> as = item.getAttributes();
//		for (int i = as.size() - 1; i >= 0; i--) {
//			Attribute a = as.get(i);
//			if (excludes == null || !excludes.contains(a.getName())) {
//				item.getAttributes().remove(a);
//			}
//		}


		List<ReplaceableAttribute> as = new ArrayList<ReplaceableAttribute>();
		
		Map<String, Object> dps = mmd.getDataProperties(data);

		//set attributes 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();
				String s = convertValueToAws(v);
				as.add(new ReplaceableAttribute(fn, s, true));
			}
		}
		
		return as;
	}

	private void saveItem(ModelMetaData mmd, String id, List<ReplaceableAttribute> as) throws DaoException {
		if (log.isDebugEnabled()) {
			log.debug("PUT: <" + id + "> - " + as);
		}
		
		PutAttributesRequest pareq = new PutAttributesRequest(mmd.getTableName(), id, as);

		simpleDB.putAttributes(pareq);
	}

	private Item getItem(ModelMetaData mmd, String id) throws DaoException {
		Item item = null;

		try {
			String sql = "SELECT * " 
					+ " FROM " + mmd.getTableName()
					+ " WHERE " 
					+ mmd.getFieldName(mmd.getIdentityName()) + " = '" + SqlUtils.escapeSql(id) + "'";
			SelectRequest sreq = new SelectRequest(sql);
			SelectResult sres = simpleDB.select(sreq);
			List<Item> is = sres.getItems();
			if (is.size() == 1) {
				item = is.get(0);
			}
			else if (is.size() > 1) {
				throw new DaoException("too mush result (" + is.size() + ")");
			}
		}
		finally {
			if (log.isDebugEnabled()) {
				log.debug("GET: <" + id + "> - " + item);
			}
		}
		
		return item;
	}

	private void deleteItem(ModelMetaData mmd, String id) throws DaoException {
		if (log.isDebugEnabled()) {
			log.debug("DELETE: <" + id + ">");
		}

		DeleteAttributesRequest dreq = new DeleteAttributesRequest(mmd.getTableName(), id);
		simpleDB.deleteAttributes(dreq);
	}

	private void deleteItems(List<String> ids) throws DaoException {
		if (log.isDebugEnabled()) {
			log.debug("DELETE: <" + ids + ">");
		}
		//TODO
	}

	private String prepareWhereClause(SqlQueryParameter qp) {
		//TODO
		return null;
	}

	private String prepareFetchOptions(SqlQueryParameter qp) {
		//TODO
		return null;
	}
	
	/**
	 * 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)) {
				Item item = getItem(mmd, kid.toString());
				return item != null;
			}
			return false;
		}
		finally {
			autoEndTransaction();
		}
	}
	
	/**
	 * countByExample
	 * 
	 * @param mid modelId
	 * @param gqp SqlQueryParameter
	 * @return count
	 * @throws DaoException if a data access error occurs
	 */ 
	public int countByExample(String mid, SqlQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			String where = prepareWhereClause(gqp);
			String fo = prepareFetchOptions(gqp);
			//TODO
			int cnt = 0;
			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)) {
				Item item = getItem(mmd, kid.toString());
				if (item != null) {
					Object data = createModelData(mmd);
					convertItemToData(mmd, item, data, null);
					return data;
				}
			}
			return null;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * selectByExampleWithDataHandler
	 * 
	 * @param mid modelId
	 * @param gqp SqlQueryParameter
	 * @param dataHandler data handler
	 * @return data count
	 * @throws DaoException if a data access error occurs
	 */ 
	public int selectByExampleWithDataHandler(String mid, SqlQueryParameter gqp, DataHandler dataHandler) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			String pq = prepareWhereClause(gqp);
			String fo = prepareFetchOptions(gqp);
			//TODO
			int count = 0;
//			while (it.hasNext()) {
//				Item en = it.next();
//				Object data = createModelData(mmd);
//				convertItemToData(mmd, en, data, gqp.getExcludes().keySet());
//				try {
//					boolean next = dataHandler.handleData(data);
//					count++;
//					if (!next) {
//						break;
//					}
//				}
//				catch (Exception ex) {
//					throw new DataAccessException("Data Handle Error: " + data, ex);
//				}
//			}
			return count;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * selectByExample
	 * 
	 * @param mid modelId
	 * @param gqp SqlQueryParameter
	 * @return list of Object 
	 * @throws DaoException if a data access error occurs
	 */ 
	public List selectByExample(String mid, SqlQueryParameter gqp) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);
			String pq = prepareWhereClause(gqp);
			String fo = prepareFetchOptions(gqp);
			List list = new ArrayList();
			//TODO
//			Iterator<Item> it = pq.asIterator(fo);
//			while (it.hasNext()) {
//				Item en = it.next();
//				Object data = createModelData(mmd);
//				convertItemToData(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 {
		//TODO
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(mid);

			Item item;
			Object kid = getDataIdentity(mmd, data);
			if (isValidIdentity(kid)) {
				item = getItem(mmd, kid.toString());
				if (item != null) {
					saveItem(mmd, kid.toString(), convertDataToAttributes(mmd, data, item, null));
				}
				else {
					item = new Item();
					saveItem(mmd, kid.toString(), convertDataToAttributes(mmd, data, item, null));
				}
			}
			else {
				item = new Item();

				setDataIdentity(mmd, data, item.getName());
				saveItem(mmd, kid.toString(), convertDataToAttributes(mmd, data, item, null));
			}
			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 {
		//TODO
//		try {
//			autoStartTransaction();
//			ModelMetaData mmd = client.getMetaData(id);
//
//			Item item;
//			Object kid = getDataIdentity(mmd, data);
//			if (isValidIdentity(kid)) {
//				item = new Item(key); 
//			}
//			else {
//				item = new Item(mmd.getTableName(), getRootKey());
//				saveItem(item);
//				setDataIdentity(mmd, data, item.getKey().getId());
//			}
//
//			convertDataToAttributes(mmd, data, item, null);
//			saveItem(item);
//			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)) {
				deleteItem(mmd, kid.toString());
				cnt = 1;
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * deleteByExample
	 * 
	 * @param mid modelId
	 * @param gqp SqlQueryParameter
	 * @return count of deleted records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int deleteByExample(String mid, SqlQueryParameter gqp) throws DaoException {
		//TODO
		try {
			autoStartTransaction();
//			Query q = gqp.getQuery();
//			q.setKeysOnly();
//			PreparedQuery pq = prepareWhereClause(q);
//			FetchOptions fo = prepareFetchOptions(gqp);
			
			int cnt = 0;
//			Iterator<Item> it = pq.asIterator(fo);
//			while (it.hasNext()) {
//				deleteItem(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)) {
				//TODO
//				Key k = createKey(mmd, kid);
//				Item item = getItem(k);
//				if (item != null) {
//					convertDataToAttributes(mmd, data, item, null);
//					saveItem(item);
//					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 id, Object data) throws DaoException {
		try {
			autoStartTransaction();
			ModelMetaData mmd = client.getMetaData(id);
			Object kid = getDataIdentity(mmd, data);
			int cnt = 0;
			if (isValidIdentity(kid)) {
				//TODO
//				Key k = createKey(mmd, kid);
//				Item item = getItem(k);
//				Set<String> excludes = new HashSet<String>();
//				addNullExcludes(mmd, data, excludes);
//				convertDataToAttributes(mmd, data, item, excludes);
//				saveItem(item);
				cnt = 1;
			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * updateByExample
	 * 
	 * @param mid modelId
	 * @param data data
	 * @param gqp SqlQueryParameter
	 * @return count of updated records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int updateByExample(String mid, Object data, SqlQueryParameter 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 = prepareWhereClause(q);
//			FetchOptions fo = prepareFetchOptions(gqp);
			
//			Iterator<Item> it = pq.asIterator(fo);
			int cnt = 0;
//			while (it.hasNext()) {
//				Item item = it.next();
//
//				convertDataToAttributes(mmd, data, item, excludes);
//
//				saveItem(item);
//				
//				cnt++;
//			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}

	/**
	 * updateByExampleSelective (ignore null properties)
	 * 
	 * @param mid modelId
	 * @param data data
	 * @param gqp SqlQueryParameter
	 * @return count of updated records
	 * @throws DaoException if a data access error occurs
	 */ 
	public int updateByExampleSelective(String mid, Object data, SqlQueryParameter 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 = prepareWhereClause(q);
//			FetchOptions fo = prepareFetchOptions(gqp);
			
//			Iterator<Item> it = pq.asIterator(fo);
			int cnt = 0;
//			while (it.hasNext()) {
//				Item item = it.next();
//
//				convertDataToAttributes(mmd, data, item, excludes);
//
//				saveItem(item);
//				
//				cnt++;
//			}
			autoCommitTransaction();
			return cnt;
		}
		finally {
			autoEndTransaction();
		}
	}
}
