#include <fstream>
#include <sstream>
#include "../../shared/SimpleLogFile.h"
#include "../../shared/ThreadMonitor.h"

#include "BlackList.h"


/////////////////////////////////////////////////////////////////////
// BlackList::FileWatcher
//
// It's tempting to move this logic to a background thread.  The
// batch thread seems appropriate.
//
// Performance notes:
// * Once per minute we load black-list.txt from disk.  This should
//   be short and in the cache, so it shouldn't matter much.  We
//   compare it to the last version to know if there was a change or
//   not.  Most of the time we are expecting no changes, so we go
//   back to sleep.  This could easily get moved from the DataNode
//   thead to a shared background thread.
// * Parsing the file and updating the set could also be done in the
//   background thread, although this does not happen vey much.
//   If we moved the first item into a seperate thread, we should
//   do the parsing in that thread too.
// * Each time there is a change in the file we will notify one
//   BlackList object per stock symbol, and it will cache the result,
//   so that part is about is as good as it gets.
//
// Note:  We don't have to poll the file.  We could use an API
// and ask linux to watch the file for us.  See
// https://man7.org/linux/man-pages/man7/fanotify.7.html
// and https://man7.org/linux/man-pages/man7/inotify.7.html
// I chose polling because that's simple and reliable.
/////////////////////////////////////////////////////////////////////

static const std::string FILE_NAME = "data/black-list.txt";

std::string BlackList::FileWatcher::getFileContents()
{ // Read whole ASCII file into C++ std::string:
  // https://stackoverflow.com/a/2602258/971955
  std::ifstream t(FILE_NAME);
  std::stringstream buffer;
  buffer << t.rdbuf();
  return buffer.str();
}

void BlackList::FileWatcher::addStrings()
{
  bool somethingWasAdded = false;
  const std::vector< std::string > lines = explode("\n", _body);
  for (std::string const &line : lines)
  {
    const std::string symbol = trim(line, " \r\t");
    if (!symbol.empty())
    {
      const bool newlyAdded = _blackListed.insert(symbol).second;
      if (newlyAdded)
      {
	somethingWasAdded = true;
	sendToLogFile(TclList()<<FLF<<symbol);
      }
    }
  }
  if (somethingWasAdded)
  {
    notifyListeners();
  }
}

void BlackList::FileWatcher::onBroadcast(BroadcastMessage &message, int msgId)
{
  //sendToLogFile(TclList()<<"starting BlackList::FileWatcher::onBroadcast()"
  //                       <<msgId);
  const std::string newBody = getFileContents();
  if (newBody != _body)
  {
    TclList msg;
    msg<<FLF<<"file changed"
       <<"_body"<<_body.length()<<"newBody"<<newBody.length()
      //<<"_body"<<_body<<"newBody"<<newBody
       <<(_waitingForConfirmation?
	  "consecutive changes":
	  "first change in a while");
    sendToLogFile(msg);
    _body = newBody;
    _waitingForConfirmation = true;
    // Wait just in case someone was in the process of copying or editing the
    // file while we were reading it.  We need two identical reads in a row
    // before we do anything with the data.
    getManager()->getTimerThread().requestBroadcastAfter(_singleTimer, 5000);
  }
  else if (_waitingForConfirmation)
  {
    TclList msg;
    msg<<FLF<<"file confirmed and debounced";//<<_body;
    sendToLogFile(msg);
    _waitingForConfirmation = false;
    addStrings();
  }
  else
  {
    //TclList msg;
    //msg<<FLF<<"No work to do!"<<"_body"<<_body.length();
    //sendToLogFile(msg);
    ThreadMonitor::find().increment("BlackList idle");
  }
  //sendToLogFile(TclList()<<"BlackList::FileWatcher::onBroadcast() done");
}

BlackList::FileWatcher::FileWatcher(DataNodeArgument const &args) :
  _periodicTimer(getOwnerChannel() + ".p", &getManager()->getTimerThread()),
  _singleTimer(getOwnerChannel() + ".s", &getManager()->getTimerThread()),
  _body(getFileContents()),
  _waitingForConfirmation(false)
{
  assert(!args);

  // The timer thread only allows one outstanding request per channel.
  // Sending a second request will cancel the previous request.
  // Requesting a single extra wakeup would cancel the periodic wakup
  // request.  That's why I created two different timers.
  // 
  // They both respond the same way, with id = 0.  It was just simpler
  // that way.
  registerForBroadcast(_periodicTimer, 0);
  registerForBroadcast(_singleTimer, 0);

  getManager()->getTimerThread()
    .requestPeriodicBroadcast(_periodicTimer, 60000);
  addStrings();
}

DataNodeLink *BlackList::FileWatcher::find(DataNodeListener *listener,
					   int msgId,
					   FileWatcher *&node)
{
  return findHelper(listener, msgId, node, DataNodeArgument());
}


/////////////////////////////////////////////////////////////////////
// BlackList
/////////////////////////////////////////////////////////////////////

void BlackList::onWakeup(int msgId)
{
  if (_fileWatcher->isBlackListed(_symbol) && !_blackListed)
  { // It is now black listed, but it was not before.
    _blackListed = true;
    notifyListeners();
  }
}

BlackList::BlackList(DataNodeArgument const &args) : 
  _symbol(args.getStringValue())
{
  addAutoLink(FileWatcher::find(this, 0, _fileWatcher));
  _blackListed = _fileWatcher->isBlackListed(_symbol);
}

DataNodeLink *BlackList::find(DataNodeListener *listener,
			      int msgId,
			      BlackList *&node,
			      std::string const &symbol)
{
  return findHelper(listener, msgId, node, symbol);
}

DataNodeLink *BlackList::find(BlackList *&node,
			      std::string const &symbol)
{
  return findHelper(NULL, 0, node, symbol);
}