package treecms.proto.cassandra.v1;

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.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.NotFoundException;
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;
	
	public static final String TREEINFO = "TreeInfo";
	public static final String ROOTNODE = "RootNode";
	public static final String MARKER = "Marker"; // because of can't create null row entry.
	public static final String CLASSNAME = "ClassName";
	public static final String TITLE = "Title";
	public static final String CHILDREN = "Children";
	public static final String LINKUUID = "LinkUUID";
	public static final String CHILDREN_DELIM = ",";
	
	private static final String ATTR_PREFIX = "_";
	
	public static void main(String _args[])
	{
		CassBrowser cass = CassBrowser.createInstance("localhost","9160","TreeCMSKS","TreeCMSCF03");
		Node node = cass.useContents();
		node.setTitle("fugafuga");
		node.setClassName("hogehoge");
		
		node.setAttribute("fuga","higa".getBytes());
		node.setAttribute("Title","higa".getBytes());
		
		System.out.println(node.getTitle());
		
		for(String key : node.getAttributeKeys()){
			System.out.println(key);
			System.out.println(node.getAttribute(key));
		}
		
		Node clone = node.cloneNode();
		System.out.println(clone.getID().toString());
		clone.setTitle("cloned1");
		
		LinkedList<Node> children = new LinkedList<Node>();
		Node child1 = clone.createNode();
		child1.setTitle("fugafuga");
		children.add(child1);
		Node child2 = clone.createNode();
		child1.setTitle("hugahuga");
		children.add(child2);
		
		clone.addChildren(children);
		
		Node child3 = clone.createNode();
		child3.setTitle("xfgafuga");
		
		clone.addChild(child3);
	}
	
	private CassBrowser(String _host,String _port) throws TTransportException
	{
		m_host = _host;
		m_port = _port;
		m_cassandra = null;
		
		connect();
	}
	
	private boolean initialize()
	{
		try{
			ColumnPath column = new ColumnPath();
			column.column_family = m_colFamily;
			column.column = ROOTNODE.getBytes();
			
			m_cassandra.get(m_keySpace,TREEINFO,column,ConsistencyLevel.ALL);
			
		}catch(NotFoundException _e){
			_e.printStackTrace();
			System.out.println("Initializing ColumFamily["+m_colFamily+"]");
			
			Node root = createNode();
			ColumnPath path = new ColumnPath();
			path.column = ROOTNODE.getBytes();
			path.column_family = m_colFamily;
			
			try {
				m_cassandra.insert(m_keySpace,TREEINFO,path,root.getID().getUUID().getBytes(),System.currentTimeMillis()/1000,ConsistencyLevel.ALL);
			}catch(Exception _e2){
				_e2.printStackTrace();
				return false;
			}
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return true;
	}
	
	public Cassandra.Client getClient()
	{
		return m_cassandra;
	}
	
	public String getKeySpace()
	{
		return m_keySpace;
	}
	
	public String getColumnFamily()
	{
		return m_colFamily;
	}
	
	public synchronized void connect() throws TTransportException 
	{
		if(m_cassandra != null && m_cassandra.getOutputProtocol().getTransport().isOpen()){
			return;
		}
		
		TTransport tr = 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;
			if(!browser.initialize()){
				return null;
			}
		}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(children);
		
		if(buf.length() != 0){
			buf.append(CHILDREN_DELIM);
		}
		
		String prefix = "";
		for(int i = 0;i < _children.size();i ++){
			String key = _children.get(i).getID().toString();
			if(set.get(key) != null){
				return;
			}
			
			buf.append(prefix).append(key);
			prefix = CHILDREN_DELIM;
		}
		
		setColumn(_id.toString(),CHILDREN,buf.toString().getBytes());
	}
	
	public void clearChildren(NodeID _id)
	{
		setColumn(_id.toString(),CHILDREN,"".getBytes());
	}
	
	public boolean removeChild(NodeID _id,Node _child)
	{
		String raw = getChildrenRaw(_id);
		StringTokenizer token = new StringTokenizer(raw,CHILDREN_DELIM);
		
		//CharBuffer buf = CharBuffer.allocate(raw.length());
		StringBuffer buf = new StringBuffer();
		
		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().getBytes());
		}
		
		return changed;
	}
	
	public void replace(NodeID _id,Node _child1,Node _child2)
	{
		String children = getChildrenRaw(_id);
		//CharBuffer buf = CharBuffer.allocate(children.length());
		StringBuffer buf = new StringBuffer();
		StringTokenizer tokens = new StringTokenizer(children,CHILDREN_DELIM);

		boolean flag = false;
		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);
				flag = true;
			}else{
				buf.append(prefix+token);
			}
		}
		
		if(flag){
//			setColumn(_id.toString(),CHILDREN,new String(buf.array()));
			setColumn(_id.toString(),CHILDREN,buf.toString().getBytes());
		}
	}
	
	public void down(NodeID _id,Node _child)
	{
		String children = getChildrenRaw(_id);
		String id = _child.getID().toString();
		//CharBuffer buf = CharBuffer.allocate(children.length());
		StringBuffer buf = new StringBuffer();
		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()){
					String next = tokens.nextToken();
					buf.append(prefix+next);
					buf.append(CHILDREN_DELIM+_child.getID().toString());
				}else{
					buf.append(prefix+_child.getID().toString());
				}
				continue;
			}
			
			buf.append(prefix+token);
		}
		
		//setColumn(_id.toString(),CHILDREN,new String(buf.array()));
		setColumn(_id.toString(),CHILDREN,buf.toString().getBytes());
	}
	
	public void up(NodeID _id,Node _child)
	{
		String children = getChildrenRaw(_id);
		String id = _child.getID().toString();
		//CharBuffer buf = CharBuffer.allocate(children.length());
		StringBuffer buf = new StringBuffer();
		StringTokenizer tokens = new StringTokenizer(children,CHILDREN_DELIM);
		
		int diff = 0;
		while(tokens.hasMoreElements()){
			String token = tokens.nextToken();
			if(token.equals(id)){
				int size = buf.length();
				buf.insert(size - diff,CHILDREN_DELIM+token);
			}else{
				buf.append(CHILDREN_DELIM+token);
			}
		
			diff = buf.length() - diff;
		}
		
		String result = buf.toString().substring(1);
		setColumn(_id.toString(),CHILDREN,result.getBytes());
	}
	
	public void setClassName(NodeID _id,String _className)
	{
		setColumn(_id.toString(),CLASSNAME,_className.getBytes());
	}
	
	public Node createNode()
	{
		NodeID newID = new CassDecNodeID(this);
	
		try{
			ColumnPath path = new ColumnPath();
			path.column_family = m_colFamily;
			path.column = MARKER.getBytes();
			
			m_cassandra.insert(m_keySpace,newID.toString(),path,"null".getBytes(),System.currentTimeMillis()/1000,ConsistencyLevel.ALL);
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return new CassNode(this,newID);
	}
	
	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();
		
		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.ALL);
			Map<String,List<ColumnOrSuperColumn>> cfmap = new HashMap<String,List<ColumnOrSuperColumn>>();
			cfmap.put(m_colFamily,columns);
			
			m_cassandra.batch_insert(m_keySpace,newID.toString(),cfmap,ConsistencyLevel.ALL);
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return new CassNode(this,newID);
	}
	
	public CassLink createLink(NodeID _id)
	{
		//must create link entries on cassandra
		Node link = createNode();
		setNode(link.getID(),_id);
		return new CassLink(link.getID(),this);
	}
	
	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)
	{
		byte[] res = getColumn(_id.toString(),CLASSNAME);
		if(res == null){
			return "";
		}
		return new String(res);
	}
	
	public String getChildrenRaw(NodeID _id)
	{
		byte[] res = getColumn(_id.toString(),CHILDREN);
		if(res == null){
			return "";
		}
		return new String(res);
	}
	
	public String getTitle(NodeID _id)
	{
		byte[] res = getColumn(_id.toString(),TITLE);
		if(res == null){
			return "";
		}
		return new String(res);
	}
	
	public void setTitle(NodeID _id,String _title)
	{
		setColumn(_id.toString(),TITLE,_title.getBytes());
	}
	
	public byte[] getAttribute(NodeID _id,String _attr)
	{
		return getColumn(_id.toString(),ATTR_PREFIX+_attr);
	}
	
	public void setAttribute(NodeID _id,String _attr,byte[] _value)
	{
		setColumn(_id.toString(),ATTR_PREFIX+_attr,_value);
	}
	
	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.ALL);
			for(ColumnOrSuperColumn column : columns){
				String key = new String(column.column.name);
				if(key.startsWith(ATTR_PREFIX)){
					attrs.add(key.replace(ATTR_PREFIX,""));
				}
			}
		}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,_uuid,"",1,ConsistencyLevel.ALL);
			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 byte[] getColumn(String _row,String _colName)
	{
		ColumnPath path = new ColumnPath();
		path.column_family = m_colFamily;
		path.column = _colName.getBytes();
		
		byte[] value = null;
		try{
			ColumnOrSuperColumn column;
			column = m_cassandra.get(m_keySpace,_row,path,ConsistencyLevel.ALL);
			value = column.column.value; 
		}catch(NotFoundException _e){
			System.out.println("column not found ["+_row+"]["+_colName+"]");
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return value;
	}
	
	private boolean setColumn(String _row,String _colName,byte[] _value)
	{
		ColumnPath path = new ColumnPath();
		path.column_family = m_colFamily;
		path.column = _colName.getBytes();
	
		try{
			m_cassandra.insert(m_keySpace,_row,path,_value,System.currentTimeMillis()/1000,ConsistencyLevel.ALL);
			return true;
		}catch(Exception _e){
			_e.printStackTrace();
		}
		
		return false;
	}
	
	public Node getNodeByID(NodeID _id)
	{
		return new CassNode(this,_id);
	}

	@Override
	public Node useContents()
	{
		String uuid = new String(getColumn(TREEINFO,ROOTNODE));
		NodeID rootID = getTipIDFromUUID(uuid);
		return new CassNode(this,rootID);
	}
	
	public Node getNode(NodeID m_id)
	{
		String uuid = new String(getColumn(m_id.toString(),LINKUUID));
		NodeID id = getTipIDFromUUID(uuid);
		return new CassNode(this,id);
	}
	
	public void setNode(NodeID m_id,NodeID _link)
	{
		setColumn(m_id.toString(),LINKUUID,_link.getUUID().getBytes());
	}
}