package compbio.cassandra;

import java.util.Calendar;

import org.apache.log4j.Logger;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Host;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;

import com.datastax.driver.core.Session;
import compbio.engine.ProteoCachePropertyHelperManager;
import compbio.util.PropertyHelper;

public class CassandraNativeConnector {
	private static Cluster cluster;
	private static Session session;
	private static final PropertyHelper ph = ProteoCachePropertyHelperManager.getPropertyHelper();
	private static Logger log = Logger.getLogger(CassandraNativeConnector.class);

	public static String CASSANDRA_HOSTNAME = "localhost";

	public static Session getSession () {
		return session;
	}

	/*
	 * connect to the cluster and look whether all tables exist
	 */
	public void Connect() {

		String cassandrahostname = ph.getProperty("cassandra.host");
		if (null != cassandrahostname) {
			CASSANDRA_HOSTNAME = cassandrahostname;
		}

		cluster = Cluster.builder().addContactPoint(CASSANDRA_HOSTNAME).build();

		Metadata metadata = cluster.getMetadata();
		System.out.printf("Connected to cluster: %s\n", metadata.getClusterName());
		for (Host host : metadata.getAllHosts()) {
			System.out.printf("Datatacenter: %s; Host: %s; Rack: %s\n", host.getDatacenter(), host.getAddress(), host.getRack());
		}
		session = cluster.connect();
		CreateMainTables();
		System.out.println("Cassandra connected");
	}

	private void CreateMainTables() {
		session.execute("CREATE KEYSPACE IF NOT EXISTS ProteinKeyspace WITH replication = {'class':'SimpleStrategy', 'replication_factor':3};");
		session.execute("USE ProteinKeyspace");

		session.execute("CREATE TABLE IF NOT EXISTS MainParameters "
				+ "(Name ascii, Value ascii, PRIMARY KEY(Name));");
		
		session.execute("CREATE TABLE IF NOT EXISTS ProteinRow "
				+ "(Protein ascii, JobID ascii, Predictions map<ascii,ascii>, PRIMARY KEY(JobID));");

		session.execute("CREATE TABLE IF NOT EXISTS ProteinLog "
				+ "(JobID ascii, DataBegin ascii, DataEnd ascii, ip ascii, FinalStatus ascii, "
				+ "ExecutionStatus ascii, Protein ascii, PRIMARY KEY(JobID));");

		session.execute("CREATE TABLE IF NOT EXISTS ProteinData "
				+ "(jobtime bigint, JobID ascii, Protein ascii, PRIMARY KEY(JobID));");

		session.execute("CREATE TABLE IF NOT EXISTS JpredArchive "
				+ "(JobID ascii, Protein varchar, IP ascii, StartTime bigint, ExecTime int, alignment map<ascii,ascii>, "
				+ "predictions map<ascii,ascii>, archive blob, LOG varchar, PRIMARY KEY(JobID));");
		
		session.execute("CREATE TABLE IF NOT EXISTS JobDateInfo "
				+ "(jobday bigint, Total bigint, Program varchar, Version varchar, PRIMARY KEY(jobday));");

		session.execute("CREATE INDEX IF NOT EXISTS ProteinSeq ON ProteinRow (protein);");
		session.execute("CREATE INDEX IF NOT EXISTS JobDateStamp ON ProteinData (jobtime);");
	}

	public void Closing() {
		session.shutdown();
		cluster.shutdown();
		System.out.println("Cassandra has been shut down");
	}

	/*
	 * getting earlest date of jobs from the db
	 */
	public static long getEarliestDateInDB() {
		String com = "SELECT * FROM MainParameters WHERE Name = 'EarliestJobDate';";
		System.out.println("Command: " + com);
		ResultSet results = session.execute(com);

		if (!results.isExhausted()) {
			Row r = results.one();
			return Long.parseLong(r.getString("Value"));
		}
		Calendar cal = Calendar.getInstance();
		return cal.getTimeInMillis();
	}
	
}