#include "program_base.hpp"
#include <iomanip>

namespace dbtoaster {

/******************************************************************************
	map_t
******************************************************************************/

ProgramBase::map_t::map_t(ProgramBase::serialize_fn_t _serialize_fn) :
        serialize_fn(_serialize_fn)
		, isOutput(false)
		, isTraced(false)
{}

/******************************************************************************
	logger_t
******************************************************************************/

ProgramBase::logger_t::logger_t(const path& fp, bool ln, bool le) :
		log_stream(new ProgramBase::logger_t::file_stream_t(fp.filename().c_str()))
		, log_relation_name(ln)
		, log_event_type(le) 
{
	if (!log_stream) {
		cerr << "failed to open file path " << fp << endl;
	} else {
		cout << "logging to " << fp << endl;
	}
}

void ProgramBase::logger_t::log(string& relation_name, const event_t& evt) {
	if (!log_stream)
		return;

    if (log_relation_name)
        (*log_stream) << relation_name << "|";
    if (log_event_type)
        (*log_stream) << evt.type << "|";
//  (*log_stream) << setprecision(15) << evt.data << endl;  
    for(size_t i = 0 ; i < evt.data.size(); ++i) {
        (*log_stream) << std::setprecision(15) << evt.data[i];
        if (i < evt.data.size() - 1) (*log_stream) << "|";
    }
    (*log_stream) << endl;
}

/******************************************************************************
	trigger_t
******************************************************************************/

ProgramBase::trigger_t::trigger_t(
			string r_name, 
			event_type ev_type, 
			ProgramBase::trigger_fn_t t_fn,
			std::shared_ptr<ProgramBase::logger_t> t_logger) :
	name(string(event_name[ev_type]) + "_" + r_name)
	,fn(t_fn)
	,logger(t_logger)
{}

void ProgramBase::trigger_t::trigger_t::log(
			string& relation_name, 
			const event_t& evt) {
	if (!logger)
		return;
	logger->log(relation_name, evt);
}

/******************************************************************************
	relation_t
******************************************************************************/

ProgramBase::relation_t::relation_t(
			string r_name, 
			bool r_is_table, 
			relation_id_t r_id,
			ProgramBase::trigger_fn_t ins_trigger_fn, 
			ProgramBase::trigger_fn_t del_trigger_fn,
			std::shared_ptr<ProgramBase::logger_t> ins_logger,
			std::shared_ptr<ProgramBase::logger_t> del_logger) :
	name(r_name)
	, is_table(r_is_table)
	, id(r_id) 
{
	trigger[insert_tuple] =
			ins_trigger_fn ?
				std::shared_ptr<ProgramBase::trigger_t>(
					new ProgramBase::trigger_t(r_name, insert_tuple,
									ins_trigger_fn, ins_logger)) :
				std::shared_ptr<ProgramBase::trigger_t>();
	trigger[delete_tuple] =
			del_trigger_fn ?
				std::shared_ptr<ProgramBase::trigger_t>(
					new ProgramBase::trigger_t(r_name, delete_tuple,
									del_trigger_fn, del_logger)) :
					std::shared_ptr<ProgramBase::trigger_t>();
}

/******************************************************************************
	ProgramBase
******************************************************************************/

relation_id_t ProgramBase::get_relation_id(string r_name) {
	map<string, std::shared_ptr<relation_t> >::iterator it =
			relations_by_name.find(r_name);
	return (it != relations_by_name.end()) ? it->second->id : -1;
}

string ProgramBase::get_relation_name(relation_id_t s_id) {
	map<relation_id_t, std::shared_ptr<relation_t> >::iterator it =
			relations_by_id.find(s_id);
	return (it != relations_by_id.end()) ? it->second->name : "";
}

void ProgramBase::add_relation(
			string r_name, 
			bool is_table, 
			relation_id_t s_id) {
	if (relations_by_name.find(r_name) != relations_by_name.end()) {
		cerr << "Found existing relation " << r_name << endl;
		return;
	}

	relation_id_t id = (s_id != -1) ? s_id : next_relation_id++;
	if (relations_by_id.find(id) != relations_by_id.end()) {
		cerr << "Found existing relation " << r_name << " with id " << id
				<< endl;
		return;
	}

	ProgramBase::relation_ptr_t r = 
			std::shared_ptr<ProgramBase::relation_t>(
				new ProgramBase::relation_t(r_name, is_table, id));
	relations_by_name[r_name] = r;
	relations_by_id[id] = r;
}

void ProgramBase::add_trigger( 
			string r_name, 
			event_type ev_type, 
			ProgramBase::trigger_fn_t fn) {
	map<string, ProgramBase::relation_ptr_t>::iterator it =
			relations_by_name.find(r_name);
	if (it == relations_by_name.end()) {
		cerr << "Relation not found: " << r_name << endl;
		return;
	}
	ProgramBase::relation_ptr_t r = it->second;

	static std::shared_ptr<ProgramBase::logger_t> g_log = 
			std::shared_ptr<ProgramBase::logger_t>();
	std::shared_ptr<ProgramBase::logger_t> log = 
			std::shared_ptr<ProgramBase::logger_t>();
	if (run_opts->global()) {
		if (!g_log) {
			path global_file = run_opts->get_log_file("", "Events", true);
			g_log = std::shared_ptr<ProgramBase::logger_t>(
					new ProgramBase::logger_t(global_file, true, true));
		}
		log = g_log;
	} else if (run_opts->logged_streams.find(r_name)
			!= run_opts->logged_streams.end()) {
		if (run_opts->unified()) {
			event_type other_type =
					ev_type == insert_tuple ? delete_tuple : insert_tuple;
			std::shared_ptr<ProgramBase::logger_t> other_log = 
					r->trigger[other_type]->logger;

			if (other_log)
				log = other_log;
			else
				log = std::shared_ptr<ProgramBase::logger_t>(
						new ProgramBase::logger_t(
								run_opts->get_log_file(r->name),	
								false, true));
		} else {
			log = std::shared_ptr<ProgramBase::logger_t>(
						new ProgramBase::logger_t(
								run_opts->get_log_file(r->name, ev_type),
								false, false));
		}
	}

	r->trigger[ev_type] = 
		std::shared_ptr<ProgramBase::trigger_t>(
			new ProgramBase::trigger_t(r->name, ev_type, fn, log));
}

void ProgramBase::add_source(
				std::shared_ptr<streams::source> source, 
				bool is_table_source) {
	if( is_table_source )   table_multiplexer.add_source(source);
	else                    stream_multiplexer.add_source(source);
}

ProgramBase::ProgramBase(int argc, char* argv[]) :
	run_opts(new runtime::runtime_options(argc, argv))
	, stream_multiplexer(12345, 10)
	, table_multiplexer(12345, 10)
	, next_relation_id(0)
	, tuple_count(0)
	, log_count_every(run_opts->log_tuple_count_every)
#ifdef DBT_PROFILE
	, window_size( run_opts->get_stats_window_size() )
	, stats_period( run_opts->get_stats_period() )
	, stats_file( run_opts->get_stats_file() )
	, exec_stats(new trigger_exec_stats("exec", window_size, 
										stats_period, stats_file))
	, ivc_stats(new trigger_exec_stats("ivc", window_size, 
									   stats_period, stats_file))
	, delta_stats(new delta_size_stats("delta_sz", window_size,
									   stats_period, stats_file))
#endif // DBT_PROFILE
{
}

void ProgramBase::process_streams() {
	if(!stream_multiplexer.eventList->empty()) {
		std::list<event_t>::iterator it = stream_multiplexer.eventList->begin();
		std::list<event_t>::iterator it_end = stream_multiplexer.eventList->end();
		for(;it != it_end; ++it) {
			process_stream_event(*it);
		}
	}
	if(!stream_multiplexer.eventQue->empty()) {
		std::list<event_t>::iterator it = stream_multiplexer.eventQue->begin();
		std::list<event_t>::iterator it_end = stream_multiplexer.eventQue->end();
		for(;it != it_end; ++it) {
			process_stream_event(*it);
		}
	}
	// XXX memory leak
	// but if we assume that program finishes at this point
	// we can ignore it
	// stream_multiplexer.eventList->clear();
	// stream_multiplexer.eventQue->clear();
#ifdef DBT_PROFILE
	exec_stats->save_now();
#endif // DBT_PROFILE
}

void ProgramBase::process_tables() {
	std::list<event_t>::iterator it = table_multiplexer.eventList->begin();
	std::list<event_t>::iterator it_end = table_multiplexer.eventList->end();
	for(;it != it_end; ++it) {
		process_event(*it,true);
	}
	if(!table_multiplexer.eventQue->empty()) {
		table_multiplexer.eventQue->sort(compare_event_timestamp_order);
		it = table_multiplexer.eventQue->begin();
		it_end = table_multiplexer.eventQue->end();
		for(;it != it_end; ++it) {
			process_event(*it,true);
		}
	}
	// XXX memory leak
	// but if we assume that program finishes at this point
	// we can ignore it
	// table_multiplexer.eventList->clear();
	// table_multiplexer.eventQue->clear();
}

void ProgramBase::set_log_count_every(
			unsigned int _log_count_every) {
	log_count_every = _log_count_every;
}

void ProgramBase::process_event(const event_t& evt, const bool process_table) {
	map<relation_id_t, 
				 std::shared_ptr<ProgramBase::relation_t> >::iterator r_it =
			relations_by_id.find(evt.id);
	if( r_it != relations_by_id.end() &&
		r_it->second->is_table == process_table )
	{
		if(r_it->second->trigger[evt.type]) {
			std::shared_ptr<ProgramBase::trigger_t> trig = 
				r_it->second->trigger[evt.type];

			#ifdef DBT_TRACE
			cout << trig->name << ": " << evt.data << endl;
			#endif // DBT_TRACE
			trig->log(r_it->second->name, evt);

			(trig->fn)(evt.data);
		} else {
			cerr << "Could not find " << event_name[evt.type]
					<< " handler for relation " << evt.id << endl;
		}
	} else {
		cerr << "Could not find " << event_name[evt.type]
				<< " handler for relation " << evt.id << endl;
	}
}

void ProgramBase::process_stream_event(const event_t& _evt) {
	#ifdef DBT_TRACE
	trace(cout, false);
	#else
	if (run_opts->is_traced())
		trace(run_opts->get_trace_file(), true);
	#endif // DBT_TRACE

	process_event(_evt, false);

	if (log_count_every && (tuple_count % log_count_every == 0)) {
		struct timeval tp;
		gettimeofday(&tp, NULL);
		cout << tuple_count << " tuples processed at " << tp.tv_sec << "s+"
				<< tp.tv_usec << "us" << endl;
	}
	tuple_count += 1;

	IProgram::process_stream_event(_evt);
}

void ProgramBase::trace(const path& trace_file, bool debug) {
	if (strcmp(trace_file.filename().c_str(), "-")) {
		std::ofstream ofs(trace_file.filename().c_str());
		trace(ofs, debug);
	} else {
		trace(cout, debug);
	}
}

void ProgramBase::trace(std::ostream &ofs, bool debug) {
	std::unique_ptr<dbtoaster::xml_oarchive> oa;

	map<string, ProgramBase::map_ptr_t>::iterator it = 
			maps_by_name.begin();
	for (; it != maps_by_name.end(); ++it)
	#ifndef DBT_TRACE_ALL
	if ((!debug && it->second->isOutput) || (debug && it->second->isTraced)) 
	#endif
	{
		if (!oa.get())
			oa = std::unique_ptr<dbtoaster::xml_oarchive>(&ofs);
		it->second->serialize_fn(*oa);
	}
}

bool ProgramBase::is_async() {
	return run_opts->async;
}

bool ProgramBase::is_no_output() {
	return run_opts->no_output;
}

}
