package meerkat.bulletinboard.sqlserver;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.sql.*;
import java.util.*;

import com.google.protobuf.*;

import com.google.protobuf.Timestamp;
import meerkat.bulletinboard.*;
import meerkat.bulletinboard.sqlserver.mappers.*;

import meerkat.comm.CommunicationException;

import meerkat.comm.MessageInputStream;
import meerkat.comm.MessageOutputStream;
import meerkat.crypto.DigitalSignature;
import meerkat.crypto.concrete.SHA256Digest;

import meerkat.protobuf.BulletinBoardApi.*;
import meerkat.protobuf.Comm;
import meerkat.protobuf.Crypto.Signature;
import meerkat.protobuf.Crypto.SignatureVerificationKey;


import static meerkat.bulletinboard.sqlserver.BulletinBoardSQLServer.SQLQueryProvider.*;

import javax.sql.DataSource;

import meerkat.util.BulletinBoardUtils;
import meerkat.util.TimestampComparator;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.support.GeneratedKeyHolder;
import org.springframework.jdbc.support.KeyHolder;



/**
 * This is a generic SQL implementation of the BulletinBoardServer API.
 */
public class BulletinBoardSQLServer implements DeletableBulletinBoardServer{

	/**
	 * This interface provides the required implementation-specific data to enable an access to an actual SQL server.
	 * It accounts for differences in languages between SQL DB types and for the different addresses needed to access them.
	 */
	public interface SQLQueryProvider {

		/**
		 * Allowed query types.
		 * Note that each query returned has to comply with the parameter names specified ny getParamNames
		 */
		public static enum QueryType {

			FIND_MSG_ID(
					new String[] {"MsgId"},
					new int[] {Types.BLOB}
			),

			FIND_TAG_ID(
					new String[] {"Tag"},
					new int[] {Types.VARCHAR}
			),

			INSERT_MSG(
					new String[] {"MsgId","TimeStamp","Msg"},
					new int[] {Types.BLOB, Types.TIMESTAMP, Types.BLOB}
			),

			DELETE_MSG_BY_ENTRY(
					new String[] {"EntryNum"},
					new int[] {Types.INTEGER}
			),

			DELETE_MSG_BY_ID(
					new String[] {"MsgId"},
					new int[] {Types.BLOB}
			),

			INSERT_NEW_TAG(
					new String[] {"Tag"},
					new int[] {Types.VARCHAR}
			),

			CONNECT_TAG(
					new String[] {"EntryNum","Tag"},
					new int[] {Types.INTEGER, Types.VARCHAR}
			),

			ADD_SIGNATURE(
					new String[] {"EntryNum","SignerId","Signature"},
					new int[] {Types.INTEGER, Types.BLOB, Types.BLOB}
			),

			GET_SIGNATURES(
					new String[] {"EntryNum"},
					new int[] {Types.INTEGER}
			),

			GET_MESSAGES(
					new String[] {},
					new int[] {}
			),

			COUNT_MESSAGES(
					new String[] {},
					new int[] {}
			),

			GET_MESSAGE_STUBS(
					new String[] {},
					new int[] {}
			),

			GET_LAST_MESSAGE_ENTRY(
					new String[] {},
					new int[] {}
			),

			CHECK_BATCH_LENGTH(
					new String[] {"BatchId"},
					new int[] {Types.BLOB, Types.INTEGER}
			),

			CHECK_BATCH_OPEN(
					new String[] {"BatchId"},
					new int[] {Types.BLOB, Types.INTEGER}
			),

			GET_BATCH_MESSAGE_DATA_BY_MSG_ID(
					new String[] {"MsgId", "StartPosition"},
					new int[] {Types.BLOB, Types.INTEGER}
			),

			GET_BATCH_MESSAGE_DATA_BY_BATCH_ID(
					new String[] {"BatchId", "StartPosition"},
					new int[] {Types.INTEGER, Types.INTEGER}
			),

			INSERT_BATCH_DATA(
					new String[] {"BatchId", "SerialNum", "Data"},
					new int[] {Types.INTEGER, Types.INTEGER, Types.BLOB}
			),

			STORE_BATCH_TAGS(
					new String[] {"Tags"},
					new int[] {Types.BLOB}
			),

			GET_BATCH_TAGS(
					new String[] {"BatchId"},
					new int[] {Types.BLOB, Types.INTEGER}
			),

			ADD_ENTRY_NUM_TO_BATCH(
					new String[] {"BatchId", "EntryNum"},
					new int[] {Types.BLOB, Types.INTEGER, Types.INTEGER}
			);

			private String[] paramNames;
			private int[] paramTypes;

			private QueryType(String[] paramNames, int[] paramTypes) {
				this.paramNames = paramNames;
				this.paramTypes = paramTypes;
			}

			public String[] getParamNames() {
				return paramNames;
			}

			public String getParamName(int num) {
				return paramNames[num];
			}

			public int[] getParamTypes() {
				return paramTypes;
			}

			public int getParamType(int num) {
				return paramTypes[num];
			}

		}

		/**
		 * This enum provides the standard translation between a filter type and the corresponding parameter name in the SQL query
		 */
		public static enum FilterTypeParam {

			ENTRY_NUM("EntryNum", Types.INTEGER),
			MSG_ID("MsgId", Types.BLOB),
			SIGNER_ID("SignerId", Types.BLOB),
			TAG("Tag", Types.VARCHAR),
			LIMIT("Limit", Types.INTEGER),
			TIMESTAMP("TimeStamp", Types.TIMESTAMP);

			private FilterTypeParam(String paramName, int paramType) {
				this.paramName = paramName;
				this.paramType = paramType;
			}

			private String paramName;
			private int paramType;

			public static FilterTypeParam getFilterTypeParamName(FilterType filterType) {
				switch (filterType) {

					case MSG_ID:
						return MSG_ID;

					case EXACT_ENTRY:		// Go through
					case MAX_ENTRY:			// Go through
					case MIN_ENTRY:
						return ENTRY_NUM;

					case SIGNER_ID:
						return SIGNER_ID;

					case TAG:
						return TAG;

					case MAX_MESSAGES:
						return LIMIT;

					case BEFORE_TIME:		// Go through
					case AFTER_TIME:
						return TIMESTAMP;

					default:
						return null;
				}
			}

			public String getParamName() {
				return paramName;
			}

			public int getParamType() {
				return paramType;
			}
		}

		/**
		 * This function translates a QueryType into an actual SQL query.
		 * @param queryType is the type of query requested
         * @return a string representation of the query for the specific type of SQL database implemented.
         */
		public String getSQLString(QueryType queryType) throws IllegalArgumentException;

		/**
		 * Used to retrieve a condition to add to an SQL statement that will make the result comply with the filter type
		 * @param filterType is the filter type
		 * @param serialNum is a unique number used to identify the condition variables from other condition instances
		 * @return The SQL string for the condition
		 * @throws IllegalArgumentException if the filter type used is not supported
		 */
		public String getCondition(FilterType filterType, int serialNum) throws IllegalArgumentException;

		public String getConditionParamTypeName(FilterType filterType) throws IllegalArgumentException;

		/**
		 * @return the string needed in order to connect to the DB.
         */
		public DataSource getDataSource();

		/**
		 * This is used to get a list of queries that together create the schema needed for the DB.
		 * Note that these queries should not assume anything about the current state of the DB.
		 * In particular: they should not erase any existing tables and/or entries.
		 * @return The list of queries.
         */
		public List<String> getSchemaCreationCommands();

		/**
		 * This is used to get a list of queries that together delete the schema needed for the DB.
		 * This is useful primarily for tests, in which we want to make sure we start with a clean DB.
		 * @return The list of queries.
		 */
		public List<String> getSchemaDeletionCommands();

	}

	/**
	 * This method returns the value of the parameter specified in a message filter
	 * @param messageFilter is the filter
	 * @return the object parameter for the SQL query embedded in the filter (this depends on the filter type)
     */
	private Object getParam(MessageFilter messageFilter) {

		switch (messageFilter.getType()) {

			case MSG_ID:		// Go through
			case SIGNER_ID:
				return messageFilter.getId().toByteArray();

			case EXACT_ENTRY:	// Go through
			case MAX_ENTRY:		// Go through
			case MIN_ENTRY:
				return messageFilter.getEntry();

			case TAG:
				return messageFilter.getTag();

			case MAX_MESSAGES:
				return messageFilter.getMaxMessages();

			case BEFORE_TIME:	// Go through
			case AFTER_TIME:
				return BulletinBoardUtils.toSQLTimestamp(messageFilter.getTimestamp());

			default:	// Unsupported filter type
				return null;
		}

	}

	/**
	 * This class implements a comparator for the MessageFilter class
	 * The comparison is done solely by comparing the type of the filter
	 * This is used to sort the filters by type
	 */
	public class FilterTypeComparator implements Comparator<MessageFilter> {

		@Override
		public int compare(MessageFilter filter1, MessageFilter filter2) {
			return filter1.getTypeValue() - filter2.getTypeValue();
		}
	}

	protected SQLQueryProvider sqlQueryProvider;

	protected NamedParameterJdbcTemplate jdbcTemplate;
	
	protected BulletinBoardDigest digest;
	protected DigitalSignature signer;
	
	protected List<SignatureVerificationKey> trusteeSignatureVerificationArray;
	protected int minTrusteeSignatures;

	protected List<List<SignatureVerificationKey>> pollingCommitteeSignatureVerificationKeyArray;
	protected int minCommiteeSignatures;

	/**
	 * This constructor sets the type of SQL language in use.
	 * @param sqlQueryProvider is the provider of the SQL query strings required for actual operation of the server.
     */
	public BulletinBoardSQLServer(SQLQueryProvider sqlQueryProvider) {
		this.sqlQueryProvider = sqlQueryProvider;
	}

	/**
	 * This method creates the schema in the given DB to prepare for future transactions
	 * It does not assume anything about the current state of the database
	 * @throws SQLException
     */
	private void createSchema() throws SQLException {

		for (String command : sqlQueryProvider.getSchemaCreationCommands()) {
			jdbcTemplate.update(command,(Map) null);
		}

	}

	/**
	 * This method initializes the signatures, connects to the DB and creates the schema (if required).
	 */
	@Override
	public void init() throws CommunicationException {
		// TODO write signature reading part.
		
		digest = new GenericBulletinBoardDigest(new SHA256Digest());

		jdbcTemplate = new NamedParameterJdbcTemplate(sqlQueryProvider.getDataSource());

		try {
			createSchema();
		} catch (SQLException e) {
			throw new CommunicationException("Couldn't create schema " + e.getMessage());
		}

	}

	/**
	 * This method verifies the authenticity of the received message based on
	 * the stored signatures.
	 * 
	 * @param msg
	 *            is the message to authenticate (containing the signature).
	 * @return TRUE if the message is authenticated and FALSE otherwise.
	 */
	private boolean verifyMessage(BulletinBoardMessage msg) {
		//TODO: Replace with actual verification.
		return true;
	}
	
	/**
	 * This procedure makes sure that all tags in the given list have an entry in the tags table. 
	 * @param tags
	 */
	protected void insertNewTags(String[] tags) throws SQLException {

		String sql;

		sql = sqlQueryProvider.getSQLString(QueryType.INSERT_NEW_TAG);
		Map namedParameters[] = new HashMap[tags.length];

		for (int i = 0 ; i < tags.length ; i++){
			namedParameters[i] = new HashMap();
			namedParameters[i].put(QueryType.INSERT_NEW_TAG.getParamName(0), tags[i]);
		}

		jdbcTemplate.batchUpdate(sql, namedParameters);

	}
	
	/**
	 * This procedure is used to convert a boolean to a BoolValue.
	 * @param b is the boolean to convert.
	 * @return a ProtoBuf message with boolean payload.
	 */
	private BoolValue boolToBoolValue(boolean b){
		return BoolValue.newBuilder()
			.setValue(b)
			.build();
	}


	/**
	 * This method posts a messages to the server
	 * @param msg is the message to post
	 * @param precalculatedMsgID is an optional precalculated message ID
	 *              It is used when the message is the stub of a batch message
	 *              In this case the validity of the signature is not checked either
	 * @return -1 if the message is not verified
	 * 			The entry number of the message if the message is posted
	 * @throws CommunicationException
     */
	private long postMessage(BulletinBoardMessage msg, byte[] precalculatedMsgID) throws CommunicationException{

		if (precalculatedMsgID != null && !verifyMessage(msg)) {
			return -1;
		}

		String sql;
		Map[] namedParameterArray;

		byte[] msgID;
		long entryNum;

		ProtocolStringList tagList;
		String[] tags;

		List<Signature> signatureList;
		Signature[] signatures;


		if (precalculatedMsgID != null){
			msgID = precalculatedMsgID;
		} else{
			// Calculate message ID (depending only on the the unsigned message)
			digest.reset();
			digest.update(msg.getMsg());

			msgID = digest.digest();
		}

		// Add message to table if needed and store entry number of message.

		sql = sqlQueryProvider.getSQLString(QueryType.FIND_MSG_ID);
		Map namedParameters = new HashMap();

		namedParameters.put(QueryType.FIND_MSG_ID.getParamName(0),msgID);

		List<Long> entryNums = jdbcTemplate.query(sql, namedParameters, new LongMapper());

		if (entryNums.size() > 0){

			entryNum = entryNums.get(0);

		} else{

			sql = sqlQueryProvider.getSQLString(QueryType.INSERT_MSG);

			namedParameters.put(QueryType.INSERT_MSG.getParamName(1), BulletinBoardUtils.toSQLTimestamp(msg.getMsg().getTimestamp()));
			namedParameters.put(QueryType.INSERT_MSG.getParamName(2), msg.getMsg().toByteArray());

			KeyHolder keyHolder = new GeneratedKeyHolder();
			jdbcTemplate.update(sql, new MapSqlParameterSource(namedParameters), keyHolder);

			entryNum = keyHolder.getKey().longValue();

		}

		// Retrieve tags and store new ones in tag table.

		try {

			tagList = msg.getMsg().getTagList();
			tags = new String[tagList.size()];
			tags = tagList.toArray(tags);

			insertNewTags(tags);

		} catch (SQLException e) {
			throw new CommunicationException(e.getMessage());
		}

		// Connect message to tags.

		sql = sqlQueryProvider.getSQLString(QueryType.CONNECT_TAG);

		namedParameterArray = new HashMap[tags.length];

		for (int i = 0 ; i < tags.length ; i++) {
			namedParameterArray[i] = new HashMap();
			namedParameterArray[i].put(QueryType.CONNECT_TAG.getParamName(0), entryNum);
			namedParameterArray[i].put(QueryType.CONNECT_TAG.getParamName(1), tags[i]);
		}

		jdbcTemplate.batchUpdate(sql, namedParameterArray);

		// Retrieve signatures.

		signatureList = msg.getSigList();
		signatures = new Signature[signatureList.size()];
		signatures = signatureList.toArray(signatures);

		// Connect message to signatures.

		sql = sqlQueryProvider.getSQLString(QueryType.ADD_SIGNATURE);

		namedParameterArray = new HashMap[signatures.length];

		for (int i = 0 ; i < signatures.length ; i++) {
			namedParameterArray[i] = new HashMap();
			namedParameterArray[i].put(QueryType.ADD_SIGNATURE.getParamName(0), entryNum);
			namedParameterArray[i].put(QueryType.ADD_SIGNATURE.getParamName(1), signatures[i].getSignerId().toByteArray());
			namedParameterArray[i].put(QueryType.ADD_SIGNATURE.getParamName(2), signatures[i].toByteArray());
		}

		jdbcTemplate.batchUpdate(sql,namedParameterArray);

		return entryNum;

	}

	private void checkConnection() throws CommunicationException {
		if (jdbcTemplate == null) {
			throw new CommunicationException("DB connection not initialized");
		}
	}

	@Override
	public BoolValue postMessage(BulletinBoardMessage msg) throws CommunicationException {

		checkConnection();

		// Perform a post, calculate the message ID and check the signature for authenticity
		if (postMessage(msg, null) != -1){
			return BoolValue.newBuilder().setValue(true).build();	// Message was posted
		}

		return BoolValue.newBuilder().setValue(false).build();	// Message was not posted

	}

	@Override
	public BoolValue deleteMessage(MessageID msgID) throws CommunicationException {

		checkConnection();

		String sql = sqlQueryProvider.getSQLString(QueryType.DELETE_MSG_BY_ID);
		Map namedParameters = new HashMap();

		namedParameters.put(QueryType.DELETE_MSG_BY_ID.getParamName(0),msgID);

		int affectedRows = jdbcTemplate.update(sql, namedParameters);

		//TODO: Log

		return BoolValue.newBuilder().setValue(affectedRows > 0).build();

	}

	@Override
	public BoolValue deleteMessage(long entryNum) throws CommunicationException {

		checkConnection();

		String sql = sqlQueryProvider.getSQLString(QueryType.DELETE_MSG_BY_ENTRY);
		Map namedParameters = new HashMap();

		namedParameters.put(QueryType.DELETE_MSG_BY_ENTRY.getParamName(0),entryNum);

		int affectedRows = jdbcTemplate.update(sql, namedParameters);

		//TODO: Log

		return BoolValue.newBuilder().setValue(affectedRows > 0).build();

	}


	/**
	 * This is a container class for and SQL string builder and a MapSqlParameterSource to be used with it
	 */
	class SQLAndParameters {

		public StringBuilder sql;
		public MapSqlParameterSource parameters;

		public SQLAndParameters(int numOfFilters) {
			sql = new StringBuilder(50 * numOfFilters);
			parameters = new MapSqlParameterSource();
		}

	}

	SQLAndParameters getSQLFromFilters(MessageFilterList filterList) {

		SQLAndParameters result = new SQLAndParameters(filterList.getFilterCount());

		List<MessageFilter> filters = new ArrayList<MessageFilter>(filterList.getFilterList());

		Collections.sort(filters, new FilterTypeComparator());

		boolean isFirstFilter = true;

		if (!filters.isEmpty()) {
			result.sql.append(" WHERE ");

			for (int paramNum = 0 ; paramNum < filters.size() ; paramNum++) {

				MessageFilter filter = filters.get(paramNum);

				if (filter.getType().getNumber() != FilterType.MAX_MESSAGES_VALUE) {
					if (isFirstFilter) {
						isFirstFilter = false;
					} else {
						result.sql.append(" AND ");
					}
				}

				result.sql.append(sqlQueryProvider.getCondition(filter.getType(), paramNum));

				FilterTypeParam filterTypeParam = FilterTypeParam.getFilterTypeParamName(filter.getType());

				result.parameters.addValue(
						filterTypeParam.getParamName() + Integer.toString(paramNum),
						getParam(filter),
						filterTypeParam.getParamType(),
						sqlQueryProvider.getConditionParamTypeName(filter.getType()));

			}

		}

		return result;

	}

	/**
	 * Private implementation of the message stub reader for returning result as a list
	 * @param filterList is a filter list that defines which messages the client is interested in
	 * @return the requested list of message stubs
     */
	private List<BulletinBoardMessage> readMessageStubs(MessageFilterList filterList) {

		StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1));

		sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.GET_MESSAGE_STUBS));

		// Get Conditions

		SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList);

		sqlBuilder.append(sqlAndParameters.sql);

		// Run query

		return jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new MessageStubMapper());

	}

	/**
	 * Private implementation of the message reader for returning result as a list
	 * @param filterList is a filter list that defines which messages the client is interested in
	 * @return the requested list of messages
	 */
	private List<BulletinBoardMessage> readMessages(MessageFilterList filterList) throws CommunicationException {

		try {

			ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

			readMessages(filterList, new MessageOutputStream<BulletinBoardMessage>(outputStream));

			MessageInputStream<BulletinBoardMessage> inputStream =
					MessageInputStream.MessageInputStreamFactory.createMessageInputStream(new ByteArrayInputStream(
									outputStream.toByteArray()),
							BulletinBoardMessage.class);

			return inputStream.asList();

		} catch (Exception e) {
			throw new CommunicationException(e.getCause() + " " + e.getMessage());
		}

	}


	@Override
	public void readMessages(MessageFilterList filterList, MessageOutputStream<BulletinBoardMessage> out) throws CommunicationException {

		checkConnection();

		BulletinBoardMessageList.Builder resultListBuilder = BulletinBoardMessageList.newBuilder();

		// SQL length is roughly 50 characters per filter + 50 for the query itself
		StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1));

		// Check if Tag/Signature tables are required for filtering purposes

		sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.GET_MESSAGES));

		// Get conditions

		SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList);
		sqlBuilder.append(sqlAndParameters.sql);

		// Run query and stream the output using a MessageCallbackHandler

		jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new MessageCallbackHandler(jdbcTemplate, sqlQueryProvider, out));

	}

	@Override
	public Int32Value getMessageCount(MessageFilterList filterList) throws CommunicationException {

		checkConnection();

		BulletinBoardMessageList.Builder resultListBuilder = BulletinBoardMessageList.newBuilder();

		// SQL length is roughly 50 characters per filter + 50 for the query itself
		StringBuilder sqlBuilder = new StringBuilder(50 * (filterList.getFilterCount() + 1));

		// Check if Tag/Signature tables are required for filtering purposes

		sqlBuilder.append(sqlQueryProvider.getSQLString(QueryType.COUNT_MESSAGES));

		// Get conditions

		SQLAndParameters sqlAndParameters = getSQLFromFilters(filterList);
		sqlBuilder.append(sqlAndParameters.sql);

		// Run query and stream the output using a MessageCallbackHandler

		List<Long> count = jdbcTemplate.query(sqlBuilder.toString(), sqlAndParameters.parameters, new LongMapper());
		return Int32Value.newBuilder().setValue(count.get(0).intValue()).build();

	}


	/**
	 * This method checks if a specified batch exists and is already closed
	 * @param msgID is the unique ID of the batch message
     * @return TRUE if the batch is closed and FALSE if it is still open or doesn't exist at all
     */
	private boolean isBatchClosed(MessageID msgID) throws CommunicationException {

		MessageFilterList filterList = MessageFilterList.newBuilder()
				.addFilter(MessageFilter.newBuilder()
						.setType(FilterType.MSG_ID)
						.setId(msgID.getID())
						.build())
				.build();


		List<BulletinBoardMessage> messages = readMessages(filterList);

		if (messages.size() <= 0){
			return false;
		}

		return (messages.get(0).getMsg().getDataTypeCase() == UnsignedBulletinBoardMessage.DataTypeCase.MSGID);

	}

	/**
	 * This method checks if a specified batch exists and is still open
	 * @param batchId is the temporary batch ID
	 * @return TRUE if the batch is closed and FALSE if it is still open or doesn't exist at all
	 */
	private boolean isBatchOpen(long batchId) throws CommunicationException {

		String sql = sqlQueryProvider.getSQLString(QueryType.CHECK_BATCH_OPEN);
		MapSqlParameterSource namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.CHECK_BATCH_OPEN.getParamName(0),batchId);

		List<Long> result = jdbcTemplate.query(sql, namedParameters, new LongMapper());

		return (result.size() > 0 && result.get(0) > 0);

	}

	@Override
	public Int64Value beginBatch(BeginBatchMessage message) throws CommunicationException {

		checkConnection();

		// Store tags
		String sql = sqlQueryProvider.getSQLString(QueryType.STORE_BATCH_TAGS);
		MapSqlParameterSource namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.STORE_BATCH_TAGS.getParamName(0),message.toByteArray());

		jdbcTemplate.update(sql,namedParameters);

		KeyHolder keyHolder = new GeneratedKeyHolder();
		jdbcTemplate.update(sql, namedParameters, keyHolder);

		long entryNum = keyHolder.getKey().longValue();

		return Int64Value.newBuilder().setValue(entryNum).build();

	}


	@Override
	public BoolValue postBatchMessage(BatchMessage batchMessage) throws CommunicationException{

		checkConnection();

		// Make sure batch is open
		if (!isBatchOpen(batchMessage.getBatchId())) {
			return BoolValue.newBuilder().setValue(false).build();
		}

		// Add data
		String sql = sqlQueryProvider.getSQLString(QueryType.INSERT_BATCH_DATA);
		MapSqlParameterSource namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(0),batchMessage.getBatchId());
		namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(1),batchMessage.getSerialNum());
		namedParameters.addValue(QueryType.INSERT_BATCH_DATA.getParamName(2),batchMessage.getData().toByteArray());

		jdbcTemplate.update(sql, namedParameters);

		return BoolValue.newBuilder().setValue(true).build();

	}


	@Override
	public BoolValue closeBatch(CloseBatchMessage message) throws CommunicationException {

		checkConnection();

		// Check batch size

		String sql = sqlQueryProvider.getSQLString(QueryType.CHECK_BATCH_LENGTH);
		MapSqlParameterSource namedParameters = new MapSqlParameterSource();


		namedParameters.addValue(QueryType.CHECK_BATCH_LENGTH.getParamName(0),message.getBatchId());

		List<Long> lengthResult = jdbcTemplate.query(sql, namedParameters, new LongMapper());

		if (lengthResult.get(0) != message.getBatchLength()) {
			return BoolValue.newBuilder().setValue(false).build();
		}

		// Get Tags and add them to BulletinBoardMessage

		sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_TAGS);
		namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.GET_BATCH_TAGS.getParamName(0),message.getBatchId());

		List<BeginBatchMessage> beginBatchMessages = jdbcTemplate.query(sql, namedParameters, new BeginBatchMessageMapper());

		if (beginBatchMessages == null || beginBatchMessages.size() <= 0 || beginBatchMessages.get(0) == null) {
			return BoolValue.newBuilder().setValue(false).build();
		}

		UnsignedBulletinBoardMessage unsignedMessage = UnsignedBulletinBoardMessage.newBuilder()
				.addAllTag(beginBatchMessages.get(0).getTagList())
				.setTimestamp(message.getTimestamp())
				.build();

		// Digest the data

		digest.reset();
		digest.update(unsignedMessage);

		sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID);
		namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(0),message.getBatchId());
		namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_BATCH_ID.getParamName(1),0); // Read from the beginning
		jdbcTemplate.query(sql, namedParameters, new BatchDataDigestHandler(digest));

		byte[] msgID = digest.digest();

		//TODO: Signature verification

		// Create Bulletin Board message
		BulletinBoardMessage bulletinBoardMessage = BulletinBoardMessage.newBuilder()
				.setMsg(UnsignedBulletinBoardMessage.newBuilder()
						.mergeFrom(unsignedMessage)
						.setMsgId(ByteString.copyFrom(msgID)))
				.addAllSig(message.getSigList())
				.build();

		// Post message with pre-calculated ID and without checking signature validity
		long entryNum = postMessage(bulletinBoardMessage, msgID);

		// Add entry num to tag data table
		sql = sqlQueryProvider.getSQLString(QueryType.ADD_ENTRY_NUM_TO_BATCH);
		namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(0), message.getBatchId());
		namedParameters.addValue(QueryType.ADD_ENTRY_NUM_TO_BATCH.getParamName(1), entryNum);

		jdbcTemplate.update(sql, namedParameters);

		// Return TRUE

		return BoolValue.newBuilder().setValue(true).build();

	}


	@Override
	public void readBatch(BatchQuery batchQuery, MessageOutputStream<BatchChunk> out) throws CommunicationException, IllegalArgumentException{

		checkConnection();

		// Check that batch is closed
		if (!isBatchClosed(batchQuery.getMsgID())) {
			throw new IllegalArgumentException("No such batch");
		}

		String sql = sqlQueryProvider.getSQLString(QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID);
		MapSqlParameterSource namedParameters = new MapSqlParameterSource();

		namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(0),batchQuery.getMsgID().getID().toByteArray());
		namedParameters.addValue(QueryType.GET_BATCH_MESSAGE_DATA_BY_MSG_ID.getParamName(1),batchQuery.getStartPosition());

		jdbcTemplate.query(sql, namedParameters, new BatchDataCallbackHandler(out));

	}

	/**
	 * Finds the entry number of the last entry in the database
	 * @return the entry number, or -1 if no entries are found
     */
	protected long getLastMessageEntry() {

		String sql = sqlQueryProvider.getSQLString(QueryType.GET_LAST_MESSAGE_ENTRY);

		List<Long> resultList = jdbcTemplate.query(sql, new LongMapper());

		if (resultList.size() <= 0){
			return -1;
		}

		return resultList.get(0);

	}

	@Override
	public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException{

		checkConnection();

		if (generateSyncQueryParams == null
				|| !generateSyncQueryParams.hasFilterList()
				|| generateSyncQueryParams.getFilterList().getFilterCount() <= 0
				|| generateSyncQueryParams.getBreakpointListCount() <= 0){

			return SyncQuery.getDefaultInstance();

		}

		List<BulletinBoardMessage> messages = readMessageStubs(generateSyncQueryParams.getFilterList());

		if (messages.size() <= 0){
			return SyncQuery.newBuilder().build();
		}

		SyncQuery.Builder resultBuilder = SyncQuery.newBuilder();

		Iterator<BulletinBoardMessage> messageIterator = messages.iterator();
		Iterator<Float> breakpointIterator = generateSyncQueryParams.getBreakpointListList().iterator();

		Checksum checksum = new SimpleChecksum();
		checksum.setDigest(new SHA256Digest());

		Timestamp lastTimestamp = Timestamp.getDefaultInstance();
		BulletinBoardMessage message = messageIterator.next();
		long currentMessageNum = 1;

		boolean checksumChanged = true;

		while (breakpointIterator.hasNext()){

			Float breakpoint = breakpointIterator.next();

			// Continue while breakpoint not reached, or it has been reached but no new timestamp has been encountered since
			while ( messageIterator.hasNext()
					&& ((float) currentMessageNum / (float) messages.size() <= breakpoint)
					|| ((float) currentMessageNum / (float) messages.size() > breakpoint
					&& lastTimestamp.equals(message.getMsg().getTimestamp()))){

				checksumChanged = true;

				checksum.update(message.getMsg().getMsgId());

				lastTimestamp = message.getMsg().getTimestamp();
				message = messageIterator.next();

			}

			if (checksumChanged) {

				checksum.update(message.getMsg().getData());
				resultBuilder.addQuery(SingleSyncQuery.newBuilder()
						.setTimeOfSync(message.getMsg().getTimestamp())
						.setChecksum(checksum.getChecksum())
						.build());

			}

			checksumChanged = false;

		}

		return resultBuilder.build();

	}


	/**
	 * Searches for the latest time of sync of the DB relative to a given query and returns the metadata needed to complete the sync
	 * The checksum up to (and including) each given timestamp is calculated using an instance of SimpleChecksum
	 * @param syncQuery contains a succinct representation of states to compare against
	 * @return the current last entry num and latest time of sync if there is one; -1 as last entry and empty timestamp otherwise
	 * @throws CommunicationException
     */
	@Override
	public SyncQueryResponse querySync(SyncQuery syncQuery) throws CommunicationException {

		checkConnection();

		if (syncQuery == null){
			return SyncQueryResponse.newBuilder()
					.setLastEntryNum(-1)
					.setLastTimeOfSync(com.google.protobuf.Timestamp.getDefaultInstance())
					.build();
		}

		com.google.protobuf.Timestamp lastTimeOfSync = null;

		TimestampComparator timestampComparator = new TimestampComparator();

		long lastEntryNum = getLastMessageEntry();

		Iterator<SingleSyncQuery> queryIterator = syncQuery.getQueryList().iterator();

		SingleSyncQuery currentQuery = queryIterator.next();

		List<BulletinBoardMessage> messageStubs = readMessageStubs(syncQuery.getFilterList());

		Checksum checksum = new SimpleChecksum();

		for (BulletinBoardMessage message : messageStubs){

			// Check for end of current query
			if (timestampComparator.compare(message.getMsg().getTimestamp(), currentQuery.getTimeOfSync()) > 0){

				if (checksum.getChecksum() == currentQuery.getChecksum()){
					lastTimeOfSync = currentQuery.getTimeOfSync();
				} else {
					break;
				}

				if (queryIterator.hasNext()){
					currentQuery = queryIterator.next();
				} else{
					break;
				}

			}

			// Advance checksum

			ByteString messageID = message.getMsg().getMsgId();	// The data field contains the message ID

			checksum.update(messageID);

		}

		if (checksum.getChecksum() == currentQuery.getChecksum()){
			lastTimeOfSync = currentQuery.getTimeOfSync();
		}

		if (lastTimeOfSync == null){
			return SyncQueryResponse.newBuilder()
					.setLastEntryNum(-1)
					.setLastTimeOfSync(com.google.protobuf.Timestamp.getDefaultInstance())
					.build();
		} else{
			return SyncQueryResponse.newBuilder()
					.setLastEntryNum(lastEntryNum)
					.setLastTimeOfSync(lastTimeOfSync)
					.build();
		}

	}


	@Override
	public void close() {
		jdbcTemplate = null;
	}

}
