#include <fstream>

#include "../../shared/LogFile.h"
#include "../../shared/GlobalConfigFile.h"
#include "../../shared/ReplyToClient.h"
#include "../data_framework/SimpleMarketData.h"
#include "../data_framework/GenericL1Data.h"
#include "../data_framework/GenericTosData.h"
#include "../data_framework/GenericHaltData.h"
#include "../data_framework/StandardCandles.h"
#include "../../shared/MarketHours.h"
#include "../misc_framework/CsvFileDataNodes.h"
#include "../misc_framework/TimerDataNode.h"
#include "../ProxyData.h"
#include "../NxCoreKafkaConsumer.h"
#include "../nasdaq_vf/StockTwitsDataNode.h"

#include "AlertMainControl.h"

////////////////////////////////////////////////////////////////////
// TestDataNode
//
// This is a rudimentary test to see if the market data is working.
// We try to connect and request data for one symbol.  It doesn't
// sound very interesting, but this fails more often than it should.
// I'd rather have this happen at a time of my chosing, rather than
// when we first start the alerts at or around 4:20 in the morning.
////////////////////////////////////////////////////////////////////

class TestDataNode : public DataNode
{
  TestDataNode(DataNodeArgument const &args);
  friend class DataNode;

public:
  static DataNodeLink *find();
};

DataNodeLink *TestDataNode::find()
{
  TestDataNode *unused;
  return findHelper(NULL, 0, unused, DataNodeArgument());
}

TestDataNode::TestDataNode(DataNodeArgument const &args)
{
  static const std::string SYMBOL = "$TIME";
  GenericTosDataNode *tosData;
  addAutoLink(GenericTosDataNode::find(tosData, SYMBOL));
}


////////////////////////////////////////////////////////////////////
// NormalStartMessage
////////////////////////////////////////////////////////////////////

class NormalStartMessage : public Request
{
private:
  const std::string _configFileName;
  const std::string _symbolListFileName;
public:
  NormalStartMessage(std::string const &configFileName,
		     std::string const &symbolListFileName) : 
    Request(NULL), _configFileName(configFileName), 
    _symbolListFileName(symbolListFileName) { }
  std::string const &getConfigFileName() const { return _configFileName; }
  std::string const &getSymbolListFileName() const 
  { return _symbolListFileName; }
};

////////////////////////////////////////////////////////////////////
// AlertMainControl
////////////////////////////////////////////////////////////////////

// TODO this should probably go in MiscSupport.[Ch]
static std::vector< std::string > loadSymbolList(std::string const &fileName)
{
  std::vector< std::string > result;
  std::ifstream stream(fileName.c_str(), std::ios_base::in);
  // TODO check for errors and report them to the log.
  while(stream)
    {
      std::string symbol;
      std::getline(stream, symbol);
      symbol = strtoupper(trim(symbol, " \r"));
      if (!symbol.empty())
 	result.push_back(symbol);
    }
  // TODO if there was an error part way through (not counting end of file as
  // an error) we should probably clear the result in addition to reporting to
  // the log.  That's consistent with the rest of our code.
  return result;
}


AlertMainControl::AlertMainControl() :
  ThreadClass("AlertMainControl"),
  _incoming(getName()),
  _pushAlertsThread("_pushAlertsThread"),
  _dataNodeManager(&_incoming, mtDataNodeEvent)
{
  initProxyData(&_dataNodeManager);
  initKafkaConsumer(&_dataNodeManager);
  startThread();
}

AlertMatrix *AlertMainControl::findMatrix(std::string const &name)
{
  AlertMatrix *&matrix = _matrix[name];
  if (!matrix)
    matrix = new AlertMatrix(&_pushAlertsThread);
  return matrix;
}

AlertMainControl::~AlertMainControl()
{
  Request *r = new Request(NULL);
  r->callbackId = mtQuit;
  _incoming.newRequest(r);
  waitForThread();
}

void AlertMainControl::normalStart(std::string const &configFileName,
				   std::string const &symbolListFileName)
{
  Request *r = new NormalStartMessage(configFileName, symbolListFileName);
  r->callbackId = mtNormalStart;
  _incoming.newRequest(r);
}

void AlertMainControl::startTest()
{
  Request *r = new Request(NULL);
  r->callbackId = mtStartTest;
  _incoming.newRequest(r);
}

void AlertMainControl::loadAlertsTypes(AlertMatrix *matrix,
				       std::string const &requirements)
{
  for (TwoDArray::StringList::const_iterator it = 
	 _config.getRowHeaders().begin();
       it != _config.getRowHeaders().end();
       it++)
    {
      if (_config.get("Requirements", *it) == requirements)
	{
	  const std::string factoryName = _config.get("Factory Name", *it);
	  AlertCategory category;
	  category.alertType = *it;
	  category.factory = 
	    GenericDataNodeFactory::findFactory(factoryName, true);
	  if (category.factory)
	    matrix->add(category);
	  else
	    { // findFactory() is not capable of printing a good error message.
	      TclList msg;
	      msg<<"AlertMainControl.C"
		 <<"AlertMainControl::loadAlertsTypes()"
		 <<"Can't find factory by name."
		 <<*it
		 <<factoryName;
	      LogFile::primary().sendString(msg);
	    }
	}
    }
  // Config file structure:  This is a csv file.  The row headers are the
  // internal names for the alert types.  Things like "NHP".  These appear in 
  // the database, and in the alert window config string.  The standard column
  // headers are as follows:
  // Factory Name -- This is how we find the implementation of the alerts, as
  //                 shown above.
  // Requirements -- This is how we segregate the alerts so some will be 
  //                 implemented by one matrix, and some by another.  The idea
  //                 is that another part of the code will test each symbol
  //                 before adding it to the matrix.  The symbol must satisfy
  //                 the "requirements", such as a certain minimum volume, or
  //                 it will not be added to that matrix.
  // Short Name -- Available for debugging.  In the Delphi version, this was
  //               the name on the check box when you manually turned these
  //               on or off.  Note that we have access to the official name
  //               in C++, if we want it.  We can steal that code from the
  //               ax_alert_server project.
  // Foreground Color -- Available for debugging.  In the Delphi version, when
  //                     we displayed an alert on the debugging console, we
  //                     displayed the text using this foreground color.  This
  //                     takes the format #xxxxxx, so it can be used as an
  //                     HTML or X-Windows color, or it can be parsed.  Like
  //                     all fields which are obsolete and/or for debugging,
  //                     this might be blank.  Note, the formatting used by
  //                     TI Pro is available as an XML file in the
  //                     ax_alert_server project.
  // Background Color -- Similar to Foreground Color.
  // Bold -- Available for debugging.  "1" for true, blank for false.  This is
  //         how we displayed the text describing an alert in the Delphi
  //         debugging console.
  // Italic -- Similar to Bold.
  // Underline -- Similar to Bold.
  // NoLateStart -- This is obsolete.  Even the first C++/CSV version of this
  //                file is proabably out of date.  At one time we would not
  //                try to start certain alerts if we had to restart the alert
  //                server in the middle of the day.  That functionalilty has
  //                been moved.  Now the alerts are always running, but the
  //                data is smarter.  Data nodes which splice historical data
  //                with live data are responsible for handling the late start.
  //                Delphi made this available to the debug console long after
  //                it was obsolete.  "1" for true, blank for false.
}

void testOneDurationString(int seconds, std::string description)
{
  TclList msg;
  msg<<__FILE__<<__LINE__<<__FUNCTION__
     <<seconds<<description<<Alert::durationString(0, seconds);
  LogFile::primary().sendString(msg);
}

void testDurationStrings()
{
  testOneDurationString(-1, "minus");
  testOneDurationString(0, "zero");
  testOneDurationString(1, "1 sec");
  testOneDurationString(2, "");
  testOneDurationString(59, "");
  testOneDurationString(60, "");
  testOneDurationString(61, "");
  testOneDurationString(89, "");
  testOneDurationString(90, "");
  testOneDurationString(91, "");
  testOneDurationString(119, "");
  testOneDurationString(120, "");
  testOneDurationString(121, "");
  testOneDurationString(122, "");
  testOneDurationString(90*60-1, "90*60-1");
  testOneDurationString(90*60, "90*60");
  testOneDurationString(90*60+1, "90*60+1");
  testOneDurationString(91*60, "91*60");
  testOneDurationString(92*60, "92*60");
  testOneDurationString(2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE, 
			"2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE");
  testOneDurationString(2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-1, 
			"2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-1");
  testOneDurationString(2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE+1, 
			"2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE+1");
  testOneDurationString(2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-29, 
			"2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-29");
  testOneDurationString(2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-30, 
			"2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-30");
  testOneDurationString(2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-31, 
			"2*MARKET_HOURS_HOUR-MARKET_HOURS_MINUTE-31");
  testOneDurationString(2*MARKET_HOURS_HOUR, "2*MARKET_HOURS_HOUR");
  testOneDurationString(2*MARKET_HOURS_HOUR+29, "2*MARKET_HOURS_HOUR+29");
  testOneDurationString(2*MARKET_HOURS_HOUR+30, "2*MARKET_HOURS_HOUR+30");
  testOneDurationString(2*MARKET_HOURS_HOUR+31, "2*MARKET_HOURS_HOUR+31");
  testOneDurationString(2*MARKET_HOURS_HOUR+2*MARKET_HOURS_MINUTE, 
			"2*MARKET_HOURS_HOUR+2*MARKET_HOURS_MINUTE");
  testOneDurationString(24*MARKET_HOURS_HOUR+1*MARKET_HOURS_MINUTE, 
			"24*MARKET_HOURS_HOUR+1*MARKET_HOURS_MINUTE");
  testOneDurationString(24*MARKET_HOURS_HOUR+1, "24*MARKET_HOURS_HOUR+1");
  testOneDurationString(25*MARKET_HOURS_HOUR-1, "25*MARKET_HOURS_HOUR-1");
  testOneDurationString(25*MARKET_HOURS_HOUR, "25*MARKET_HOURS_HOUR");
  testOneDurationString(25*MARKET_HOURS_HOUR+1, "25*MARKET_HOURS_HOUR+1");
  testOneDurationString(2*24*MARKET_HOURS_HOUR-1, "2*24*MARKET_HOURS_HOUR-1");
  testOneDurationString((2*24+1)*MARKET_HOURS_HOUR, 
			"(2*24+1)*MARKET_HOURS_HOUR");
}

// The idea is to make our initialization process go faster.  Requesting data
// from the market data proxy is a slow process.  One option is to start this
// program sooner, but the overnight data might not be ready yet.  However,
// today's symbol list should be very similar to yesterday's list.  So, we can
// start requesting data from the market data proxy immediately.  Later we
// can ask for just what we missed in the preload.
//
// Note that data nodes are already cached and reference counted and reused.
// We don't have to make any changes to the rest of the program.  We request
// the data now.  The program does the heavy lifting now.  Later, when the
// main part of the alerts starts requesting data, most of the time we'll do
// a single lookup in a std::map, and we'll bump a reference count.  That's
// very inexpensive, and doesn't affect the market data proxy server at all.
void AlertMainControl::initializePreLoad()
{
  _toPreload = loadSymbolList(defaultSymbolListFile());
  TclList msg;
  msg<<FLF<<_toPreload.size()<<"symbols";
  LogFile::primary().sendString(msg);
  scheduleNextPreLoadStep();
}

void AlertMainControl::scheduleNextPreLoadStep()
{
  if (_toPreload.empty())
    {
      TclList msg;
      msg<<FLF<<"finished";
      LogFile::primary().sendString(msg);
    }
  else
    {
      Request *r = new Request(NULL);
      r->callbackId = mtPreload;
      _incoming.newRequest(r); 
    }
}

void AlertMainControl::doNextPreLoadStep()
{
  const std::string symbol = _toPreload.back();
  _toPreload.pop_back();
  GenericTosDataNode *tosData;
  GenericTosDataNode::find(tosData, symbol);
  GenericL1DataNode *l1Data;
  GenericL1DataNode::find(l1Data, symbol);
  GenericHaltDataNode *haltData;
  GenericHaltDataNode::find(haltData, symbol);
  char const *byName[] = { "PutVolume", "CallVolume", "LimitUp", "LimitDown" };
  for (char const *name : byName)
    {
      DataNodeArgument factory = GenericDataNodeFactory::findFactory(name);
      factory = factory.replace(symbolPlaceholder, symbol);
      GenericDataNode *genericData;
      factory.getValue< GenericDataNodeFactory >().find(genericData);
    }
  // TODO are we concerned with the regional data?  NyseBid and NyseAsk
  scheduleNextPreLoadStep();
}

std::string AlertMainControl::defaultSymbolListFile()
{
  return FileOwnerDataNode::findDataFile(getConfigItem("symbol_list_file_name",
						       "Symbols.txt"));
}

void AlertMainControl::normalStartImpl(std::string configFileName,
				       std::string symbolListFileName)
{
  if (configFileName.empty())
    {
      configFileName = 
	getConfigItem("alert_config_file", "AlertList.csv");
      configFileName = FileOwnerDataNode::findDataFile(configFileName);
    }
  if (symbolListFileName.empty())
    symbolListFileName = defaultSymbolListFile();
  TclList msg;
  msg<<"AlertMainControl.C"
     <<"normalStartImpl"
     <<"configFileName"
     <<configFileName
     <<"symbolListFileName"
     <<symbolListFileName;
  LogFile::primary().sendString(msg);
  AlertMatrix *misc = findMatrix("misc");
  AlertMatrix *lowVolume = findMatrix("lowVolume");
  AlertMatrix *volumeConfirmed = findMatrix("volumeConfirmed");
  AlertMatrix *topList = findMatrix("topList");
  if (!(misc->empty() && lowVolume->empty() 
	&& volumeConfirmed->empty() && topList->empty()))
    { // This case is not precisely defined.
      msg.clear();
      msg<<"AlertMainControl.C"
	 <<"normalStartImpl"
	 <<"Aborted!"
	 <<"Already running.";
      LogFile::primary().sendString(msg);
      return;
    }
  _config.loadFromCSV(configFileName);
  loadAlertsTypes(misc, "Misc");
  loadAlertsTypes(lowVolume, "LowVolumeOnly");
  loadAlertsTypes(volumeConfirmed, "VolumeConfirmed");
  {
    AlertCategory topListCategory;
    topListCategory.factory = 
      GenericDataNodeFactory::findFactory("FastHeartbeat");
    topList->add(topListCategory);
  }
  int symbolCount = 0;
  for (std::string const &symbol : loadSymbolList(symbolListFileName))
    {
      TclList msg;
      symbolCount++;
      msg<<__FILE__<<__LINE__<<__FUNCTION__
	 <<"adding"<<symbolCount<<symbol;
      LogFile::primary().sendString(msg);
      misc->add(symbol);
      topList->add(symbol);
      if (!symbolIsIndex(symbol))
	{
	  const DataNode::Integer volume = getAverageDailyVolume(symbol);
	  if (volume >= 150000)
	    volumeConfirmed->add(symbol);
	  if (volume <= 3000000)
	    lowVolume->add(symbol);
	}
    }
  msg.clear();
  msg<<"AlertMainControl.C"
     <<"normalStartImpl"
     <<"success"
     <<DataNodeManager::dataNodeCount();;
  LogFile::primary().sendString(msg);
}

void AlertMainControl::threadFunction()
{
  _dataNodeManager.setAsDefault();
  GenericL1DataNode::registerDebugger(&_dataNodeManager);
  GenericTosDataNode::registerDebugger(&_dataNodeManager);
  StandardCandles::registerDebugger(&_dataNodeManager);
  StockTwitsDataNode::init();

  DataNodeLink *timerLink = TimerDataNode::find(this, 0);
  DataNodeLink *testDataNodeLink = NULL;

  //testDurationStrings();

  while (true)
    {
      while (Request *current = _incoming.getRequest())
        {
          switch (current->callbackId)
            {
	    case mtDataNodeEvent:
              {
		DataNodeManager::EventQueueListener *request =
                  dynamic_cast< DataNodeManager::EventQueueListener * >
		  (current);
                request->onEventQueue();
                break;
              }
	    case mtStartTest:
	      {
		if (testDataNodeLink)
		  { // Duplicate request.  Silently ignore it.
		  }
		else
		  {
		    initializePreLoad();
		    testDataNodeLink = TestDataNode::find();
		    // If there's a problem, we typically would not get to
		    // here.  The program would hang or, more often, fail an
		    // asertion.
		    TclList msg;
		    msg<<__FILE__<<__LINE__<<__FUNCTION__
		       <<"test data started";
		    LogFile::primary().sendString(msg);
		  }
		break;
	      }
	    case mtPreload:
	      {
		doNextPreLoadStep();
		break;
	      }
	    case mtNormalStart:
	      {
		NormalStartMessage *request =
                  dynamic_cast< NormalStartMessage * >(current);
		normalStartImpl(request->getConfigFileName(),
				request->getSymbolListFileName());
		break;
	      }
	    case mtQuit:
              {
		timerLink->release();
		if (testDataNodeLink)
		  testDataNodeLink->release();
		for (std::map< std::string, AlertMatrix * >::iterator it = 
		       _matrix.begin();
		     it != _matrix.end();
		     it++)
		  {
		    delete it->second;
		  }
		_matrix.clear();
                delete current;
                return;
              }
	    case DeleteSocketThread::callbackId:
	      {
		SocketClosedDataNode::doNotify(current->getSocketInfo());
		break;
	      }
	    }
	  delete current;
	}
      _incoming.waitForRequest();
    }
}

void AlertMainControl::onWakeup(int msgId)
{ // This is from the dead man timer.
  _pushAlertsThread.updateDeadManTimer();
}