package treecms.proto.cassandra;

import java.nio.CharBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.StringTokenizer;

import treecms.proto.api.Browser;
import treecms.proto.api.Node;
import treecms.proto.api.NodeID;

import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;

public class CassBrowser implements Browser
{
	private Cassandra.Client m_cassandra;
	private String m_host,m_port;
	private String m_keySpace,m_colFamily;
	
	private static final String TREEINFO = "TreeInfo";
	private static final String ROOTNODE = "RootNode";
	
	private static final String CLASSNAME = "ClassName";
	private static final String TITLE = "title";
	private static final String CHILDREN = "Children";
	private static final String CHILDREN_DELIM = ",";
	
	private static final String ATTR_PREFIX = "_";
	
	private CassBrowser(String _host,String _port) throws TTransportException
	{
		m_host = _host;
		m_port = _port;
		m_cassandra = null;
		
		connect();
	}
	
	public synchronized void connect() throws TTransportException
	{
		if(m_cassandra != null && m_cassandra.getOutputProtocol().getTransport().isOpen()){
			return;
		}
		
		TTransport tr = new TFramedTransport(new TSocket(m_host,Integer.parseInt(m_port)));
		TProtocol proto = new TBinaryProtocol(tr);
		m_cassandra = new Cassandra.Client(proto);
		
		tr.open();
	}
	
	public static CassBrowser createInstance(String _host,String _port,String _ks,String _cf)
	{
		CassBrowser browser = null;
		try{
			browser = new CassBrowser(_host,_port);
			browser.m_keySpace = _ks;
			browser.m_colFamily = _cf;
		}catch(Exception _err){
			_err.printStackTrace();
		}
		return browser;
	}
	
	public Node addChild(NodeID _id,Node _child)
	{
		LinkedList<Node> tmp = new LinkedList<Node>();
		tmp.add(_child);
		addChildren(_id,tmp);
		return _child;
	}
	
	public void addChildren(NodeID _id,List<Node> _children)
	{
		String children = getChildrenRaw(_id);
		Map<String,Boolean> set = new HashMap<String,Boolean>();
		StringTokenizer st = new StringTokenizer(children,CHILDREN_DELIM);
		while(st.hasMoreElements()){
			set.put(st.nextToken(),Boolean.TRUE);
		}
		
		StringBuffer buf = new StringBuffer();
		
		if(!set.isEmpty()){
			buf.append(CHILDREN_DELIM);
		}
		
		for(int i = 0;i < _children.size() - 1;i ++){
			String key = _children.get(i).getID().toString();
			if(set.get(key) != null){
				return;
			}
			
			buf.append(key).append(CHILDREN_DELIM);
		}
		
		String key = _children.get(_children.size() - 1).getID().toString();
		if(set.get(key) != null){
			return;
		}
		
		buf.append(key);
		
		setColumn(_id.toString(),CHILDREN,buf.toString(),false);
	}
	
	public void clearChildren(NodeID _id)
	{
		setColumn(_id.toString(),CHILDREN,"",false);
	}
	
	public boolean removeChild(NodeID _id,Node _child)
	{
		String raw = getChildrenRaw(_id);
		StringTokenizer token = new StringTokenizer(raw,CHILDREN_DELIM);
		
		CharBuffer buf = CharBuffer.allocate(raw.length());
		
		boolean changed = false;
		
		while(token.hasMoreElements()){
			String child = token.nextToken();
			if(child.equals(_child.getID().toString())){
				changed = true;
				continue;
			}
			
			buf.append(child);
			break;
		}
		
		while(token.hasMoreElements()){
			String child = token.nextToken();
			if(child.equals(_child.getID().toString())){
				changed = true;
				continue;
			}
			
			buf.append(CHILDREN_DELIM+child);
		}
		
		if(changed){
			setColumn(_id.toString(),CHILDREN,buf.toString(),false);
		}
		
		return changed;
	}
	
	public void replace(NodeID _id,Node _child1,Node _child2)
	{
		String children = getChildrenRaw(_id);
		CharBuffer buf = CharBuffer.allocate(children.length());
		StringTokenizer tokens = new StringTokenizer(children,CHILDREN_DELIM);

		int flagSuccess = 0;
		String id1 = _child1.getID().toString();
		String id2 = _child2.getID().toString();
		for(String prefix = "";tokens.hasMoreElements();prefix = CHILDREN_DELIM){
			String token = tokens.nextToken();
			if(token.equals(id1)){
				buf.append(prefix+id2);
				flagSuccess++;
			}else if(token.equals(id2)){
				buf.append(prefix+id1);
				flagSuccess++;
			}else{
				buf.append(prefix+token);
			}
		}
		
		if(flagSuccess == 2){
			setColumn(_id.toString(),CHILDREN,buf.toString(),false);
		}
	}
	
	public void down(NodeID _id,Node _child)
	{
		String children = getChildrenRaw(_id);
		String id = _child.getID().toString();
		CharBuffer buf = CharBuffer.allocate(children.length());
		StringTokenizer tokens = new StringTokenizer(children,CHILDREN_DELIM);
		
		for(String prefix = "";tokens.hasMoreElements();prefix = CHILDREN_DELIM){
			String token = tokens.nextToken();
			if(token.equals(id)){
				if(tokens.hasMoreElements()){
					buf.append(prefix+tokens.nextToken());
					prefix = CHILDREN_DELIM;
				}
			}
			buf.append(prefix+token);
		}
		
		setColumn(_id.toString(),CHILDREN,buf.toString(),false);
	}
	
	public void up(NodeID _id,Node _child)
	{
		String children = getChildrenRaw(_id);
		String id = _child.getID().toString();
		CharBuffer buf = CharBuffer.allocate(children.length());
		StringTokenizer tokens = new StringTokenizer(children,CHILDREN_DELIM);
		
		String prev = null;
		String prefix = "";
		while(tokens.hasMoreElements()){
			if(prev == null){
				prev = tokens.nextToken();
				continue;
			}
		
			String token = tokens.nextToken();
			if(token.equals(id)){
				buf.append(prefix+prev);
				prev = token;
			}else{
				buf.append(prefix+prev);
			}
		
			prefix = CHILDREN_DELIM;
		}
	}
	
	public void setClassName(NodeID _id,String _className)
	{
		setColumn(_id.toString(),CLASSNAME,_className,false);
	}
	
	public Node createNode()
	{
		return new CassNode(this,new CassDecNodeID(this));
	}
	
	public List<Node> getChildren(NodeID _id)
	{
		LinkedList<Node> children = new LinkedList<Node>();
		
		String raw = getChildrenRaw(_id);
		StringTokenizer tokens = new StringTokenizer(raw,CHILDREN_DELIM);
		
		while(tokens.hasMoreElements()){
			String first = tokens.nextToken();
			StringTokenizer sparator = new StringTokenizer(first,"@");
			String uuid = sparator.nextToken();
			long ver = Long.parseLong(sparator.nextToken());
			NodeID id = new CassDecNodeID(this,uuid,ver);
			children.add(new CassNode(this,id));
		}
		
		return children;
	}
	
	public CassNode cloneNode(NodeID _id)
	{
		NodeID newID = _id.update();
		
		
		return null;
	}
	
	public CassLink createLink(NodeID _id)
	{
		return null;
	}
	
	public boolean isChild(NodeID _id,Node _child)
	{
		String raw = getChildrenRaw(_id);
		if(raw.indexOf(_child.getID().toString()) != -1){
			return true;
		}
		return false;
	}
	
	public String getClassName(NodeID _id)
	{
		return getColumn(_id.toString(),CLASSNAME,false);
	}
	
	public String getChildrenRaw(NodeID _id)
	{
		return getColumn(_id.toString(),CHILDREN,false);
	}
	
	public String getTitle(NodeID _id)
	{
		return getColumn(_id.toString(),TITLE,false);
	}
	
	public void setTitle(NodeID _id,String _title)
	{
		setColumn(_id.toString(),TITLE,_title,false);
	}
	
	public String getAttribute(NodeID _id,String _attr)
	{
		return getColumn(_id.toString(),ATTR_PREFIX+_attr,false);
	}
	
	public void setAttribute(NodeID _id,String _attr,String _value)
	{
		setColumn(_id.toString(),ATTR_PREFIX+_attr,_value,false);
	}
	
	public Set<String> getAttributeKeys(NodeID _id)
	{
		Set<String> attrs = new HashSet<String>();
		
		SliceRange sr = new SliceRange();
		sr.setStart(new byte[0]);
		sr.setFinish(new byte[0]);
		
		SlicePredicate sp = new SlicePredicate();
		sp.setSlice_range(sr);
		
		try{
			List<ColumnOrSuperColumn> columns = m_cassandra.get_slice(m_keySpace,_id.toString(),new ColumnParent(m_colFamily),sp,ConsistencyLevel.ONE);
			for(ColumnOrSuperColumn column : columns){
				String key = new String(column.column.name);
				if(key.startsWith(ATTR_PREFIX)){
					attrs.add(key);
				}
			}
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return attrs;
	}
	
	public NodeID getTipIDFromUUID(String _uuid)
	{
		SliceRange sliceRange = new SliceRange();
		sliceRange.setStart(_uuid.getBytes());
		sliceRange.setFinish(new byte[0]);
		
		SlicePredicate slicePredicate = new SlicePredicate();
		slicePredicate.setSlice_range(sliceRange);
		
		try {
			List<KeySlice> values = m_cassandra.get_range_slice(m_keySpace,new ColumnParent(m_colFamily),slicePredicate,"","",1,ConsistencyLevel.ONE);
			String key = values.get(0).getKey();
			
			String tip = key.replaceFirst((_uuid+"@").intern(),"");
			return new CassDecNodeID(this,_uuid,Long.parseLong(tip));
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return null;
	}
	
	private String getColumn(String _row,String _colName,boolean _retry)
	{
		ColumnPath path = new ColumnPath();
		path.column_family = m_colFamily;
		path.column = _colName.getBytes();
		
		String value = null;
		try{
			ColumnOrSuperColumn column;
			column = m_cassandra.get(m_keySpace,_row,path,ConsistencyLevel.ONE);
			value = column.column.toString();
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return value;
	}
	
	private boolean setColumn(String _row,String _colName,String _value,boolean _retry)
	{
		ColumnPath path = new ColumnPath();
		path.column_family = m_colFamily;
		path.column = _colName.getBytes();
	
		try{
			m_cassandra.insert(m_keySpace,_row,path,_value.getBytes(),System.currentTimeMillis()/1000,ConsistencyLevel.ONE);
			return true;
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return false;
	}
	
	public CassNode createNode(String _id)
	{
		return null;
	}

	@Override
	public Node useContents()
	{
		String uuid = getColumn(TREEINFO,ROOTNODE,false);
		NodeID rootID = getTipIDFromUUID(uuid);
		return new CassNode(this,rootID);
	}
}