package jungle.test.bbs;

import java.util.Collections;
import java.util.List;
import java.util.UUID;

import org.apache.cassandra.locator.SimpleStrategy;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.serializers.UUIDSerializer;
import me.prettyprint.cassandra.service.CassandraHost;
import me.prettyprint.cassandra.service.template.SuperCfResult;
import me.prettyprint.cassandra.service.template.SuperCfUpdater;
import me.prettyprint.cassandra.service.template.ThriftSuperCfTemplate;
import me.prettyprint.cassandra.utils.TimeUUIDUtils;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.HSuperColumn;
import me.prettyprint.hector.api.beans.OrderedSuperRows;
import me.prettyprint.hector.api.beans.SuperRow;
import me.prettyprint.hector.api.beans.SuperSlice;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnType;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.RangeSuperSlicesQuery;
import me.prettyprint.hector.api.query.SuperSliceQuery;

public class CassandraBulletinBoard implements BulletinBoard
{
	private final String address;
	private final String clusterName;
	private final Cluster cluster;
	private final String keyspace;
	
	private static final String COLUMN_FAMILY_BOARD = "boards";
	
	public CassandraBulletinBoard(String _clusterName,String _address,String _keyspaceName)
	{
		address = _address;
		clusterName = _clusterName;
		keyspace = _keyspaceName;
		cluster = HFactory.getOrCreateCluster(clusterName,address);
		
		initialize();
	}
	
	private void initialize()
	{
		if(cluster.describeKeyspace(keyspace) == null){
			KeyspaceDefinition keyspaceDefinition = HFactory.createKeyspaceDefinition(keyspace,
					SimpleStrategy.class.getName(),1,Collections.<ColumnFamilyDefinition> emptyList());
			cluster.addKeyspace(keyspaceDefinition,false);
			ColumnFamilyDefinition columnFamilyDefinition = HFactory.createColumnFamilyDefinition(keyspace,COLUMN_FAMILY_BOARD,ComparatorType.UUIDTYPE);
			columnFamilyDefinition.setColumnType(ColumnType.SUPER);
			cluster.addColumnFamily(columnFamilyDefinition);
		}
	}
	
	public Iterable<String> getBoards()
	{
		Keyspace ksp = HFactory.createKeyspace(keyspace, cluster);
		RangeSuperSlicesQuery<String,UUID,String,String> query = HFactory.createRangeSuperSlicesQuery(ksp,StringSerializer.get(),
				UUIDSerializer.get(),StringSerializer.get(),StringSerializer.get());
		query.setColumnFamily(COLUMN_FAMILY_BOARD).setKeys(null,null).setRange(null,null,false,0);
		
		QueryResult<OrderedSuperRows<String,UUID,String,String>> result = query.execute();
		OrderedSuperRows<String,UUID,String,String> rows = result.get();
		List<SuperRow<String,UUID,String,String>> list = rows.getList();
		
		IterableConverter.Converter<String,SuperRow<String,UUID,String,String>> converter
			= new IterableConverter.Converter<String, SuperRow<String,UUID,String,String>>(){
				public String conv(SuperRow<String, UUID, String, String> _b) {
					return _b.getKey();
				}
			};
		
		return new IterableConverter<String,SuperRow<String,UUID,String,String>>(list,converter);
	}
	
	private static final String COLUMN_MESSAGE_AUTHOR = "author";
	private static final String COLUMN_MESSAGE_BODY = "message";
	private static final String COLUMN_MESSAGE_EDIT_KEY = "edit";

	public void createBoardMessage(UUID _time,String _name,String _author,String _message,String _editKey)
	{
		Keyspace ksp = HFactory.createKeyspace(keyspace,cluster);
		ThriftSuperCfTemplate<String,UUID,String> template = 
				new ThriftSuperCfTemplate<String,UUID,String>(ksp,COLUMN_FAMILY_BOARD,StringSerializer.get(),
						UUIDSerializer.get(),StringSerializer.get());
		
		SuperCfUpdater<String,UUID,String> updater = template.createUpdater(_name,_time);
		updater.setString(COLUMN_MESSAGE_AUTHOR,_author);
		updater.setString(COLUMN_MESSAGE_BODY,_message);
		updater.setString(COLUMN_MESSAGE_EDIT_KEY,_editKey);
		
		template.update(updater);
	}
	
	public void createBoards(String _name,String _author,String _initMessage,String _editKey)
	{
		UUID time = TimeUUIDUtils.getTimeUUID(0);
		createBoardMessage(time,_name,_author,_initMessage,_editKey);
	}

	public Iterable<BoardMessage> getMessages(String _boardName)
	{
		Keyspace ksp = HFactory.createKeyspace(keyspace,cluster);
		SuperSliceQuery<String, UUID, String, String> query = HFactory.createSuperSliceQuery(ksp, StringSerializer.get(), UUIDSerializer.get(), StringSerializer.get(), StringSerializer.get());
		
		UUID start = TimeUUIDUtils.getTimeUUID(0);
		query.setKey(_boardName).setColumnFamily(COLUMN_FAMILY_BOARD).setRange(start,null,false,100);
		
		QueryResult<SuperSlice<UUID, String, String>> result = query.execute();
		SuperSlice<UUID,String,String> ss = result.get();
		List<HSuperColumn<UUID,String,String>> list = ss.getSuperColumns();
		
		IterableConverter.Converter<BoardMessage,HSuperColumn<UUID,String,String>> converter =
				new IterableConverter.Converter<BoardMessage,HSuperColumn<UUID,String,String>>(){
					public BoardMessage conv(HSuperColumn<UUID, String, String> _b){
						UUID uuid = _b.getName();
						String author = _b.getSubColumnByName(COLUMN_MESSAGE_AUTHOR).getValue();
						String message = _b.getSubColumnByName(COLUMN_MESSAGE_BODY).getValue();
						BoardMessageImpl bm = new BoardMessageImpl(author,message,uuid.toString());
						return bm;
					}
				};
		
		
		return new IterableConverter<BoardMessage,HSuperColumn<UUID,String,String>>(list,converter);
	}
	
	private static class BoardMessageImpl implements BoardMessage
	{
		private final String author;
		private final String message;
		private final String uuid;
		
		public BoardMessageImpl(String _author,String _message,String _uuid)
		{
			author = _author;
			message = _message;
			uuid = _uuid;
		}

		public String getAuthor()
		{
			return author;
		}

		public String getMessage()
		{
			return message;
		}

		public String getUUID()
		{
			return uuid;
		}
	}

	public void createBoardMessage(String _board, String _author, String _message,String _editKey)
	{
		UUID time = TimeUUIDUtils.getUniqueTimeUUIDinMillis();
		createBoardMessage(time,_board,_author,_message,_editKey);
	}

	public void editMessage(String _board,String _uuid,String _author,String _message,String _editKey)
	{
		Keyspace ksp = HFactory.createKeyspace(keyspace, cluster);
		UUID time = UUID.fromString(_uuid);
		ThriftSuperCfTemplate<String,UUID,String> template = 
				new ThriftSuperCfTemplate<String,UUID,String>(ksp,COLUMN_FAMILY_BOARD,StringSerializer.get(),
						UUIDSerializer.get(),StringSerializer.get());
		
		SuperCfResult<String,UUID,String> result = template.querySuperColumn(_board,time);
		String editKey = result.getString(COLUMN_MESSAGE_EDIT_KEY);
		if(!editKey.equals(editKey)){
			return;
		}
		
		SuperCfUpdater<String, UUID, String> updater = template.createUpdater(_board,time);
		updater.setString(COLUMN_MESSAGE_AUTHOR,_author);
		updater.setString(COLUMN_MESSAGE_BODY,_message);
		updater.setString(COLUMN_MESSAGE_EDIT_KEY,_editKey);
		
		template.update(updater);
	}

}