src/dispatcher.cpp

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
 * Copyright (c) 2013-2017, Regents of the University of California.
 *
 * This file is part of ChronoShare, a decentralized file sharing application over NDN.
 *
 * ChronoShare is free software: you can redistribute it and/or modify it under the terms
 * of the GNU General Public License as published by the Free Software Foundation, either
 * version 3 of the License, or (at your option) any later version.
 *
 * ChronoShare is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE.  See the GNU General Public License for more details.
 *
 * You should have received copies of the GNU General Public License along with
 * ChronoShare, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
 *
 * See AUTHORS.md for complete list of ChronoShare authors and contributors.
 */

#include "dispatcher.hpp"
#include "ccnx-discovery.hpp"
#include "fetch-task-db.hpp"
#include "logging.hpp"

#include <boost/lexical_cast.hpp>
#include <boost/make_shared.hpp>

using namespace Ndnx;
using namespace std;
using namespace boost;

_LOG_INIT(Dispatcher);

static const string CHRONOSHARE_APP = "chronoshare";
static const string BROADCAST_DOMAIN = "/ndn/broadcast";

static const int CONTENT_FRESHNESS = 1800;                 // seconds
const static double DEFAULT_SYNC_INTEREST_INTERVAL = 10.0; // seconds;

Dispatcher::Dispatcher(const std::string& localUserName, const std::string& sharedFolder,
                       const filesystem::path& rootDir, Ccnx::CcnxWrapperPtr ccnx,
                       bool enablePrefixDiscovery)
  : m_ccnx(ccnx)
  , m_core(NULL)
  , m_rootDir(rootDir)
  , m_executor(
      1) // creates problems with file assembly. need to ensure somehow that FinishExectute is called after all Segment_Execute finished
  , m_objectManager(ccnx, rootDir, CHRONOSHARE_APP)
  , m_localUserName(localUserName)
  , m_sharedFolder(sharedFolder)
  , m_server(NULL)
  , m_enablePrefixDiscovery(enablePrefixDiscovery)
{
  m_syncLog = make_shared<SyncLog>(m_rootDir, localUserName);
  m_actionLog =
    make_shared<ActionLog>(m_ccnx, m_rootDir, m_syncLog, sharedFolder, CHRONOSHARE_APP,
                           // bind (&Dispatcher::Did_ActionLog_ActionApply_AddOrModify, this, _1, _2, _3, _4, _5, _6, _7),
                           ActionLog::OnFileAddedOrChangedCallback(), // don't really need this callback
                           bind(&Dispatcher::Did_ActionLog_ActionApply_Delete, this, _1));
  m_fileState = m_actionLog->GetFileState();

  Name syncPrefix = Name(BROADCAST_DOMAIN)(CHRONOSHARE_APP)(sharedFolder);

  // m_server needs a different ccnx face
  m_server = new ContentServer(make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName,
                               m_sharedFolder, CHRONOSHARE_APP, CONTENT_FRESHNESS);
  m_server->registerPrefix(Name("/"));
  m_server->registerPrefix(Name(BROADCAST_DOMAIN));

  m_stateServer =
    new StateServer(make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName,
                    m_sharedFolder, CHRONOSHARE_APP, m_objectManager, CONTENT_FRESHNESS);
  // no need to register, right now only listening on localhost prefix

  m_core = new SyncCore(m_syncLog, localUserName, Name("/"), syncPrefix,
                        bind(&Dispatcher::Did_SyncLog_StateChange, this, _1), ccnx,
                        DEFAULT_SYNC_INTEREST_INTERVAL);

  FetchTaskDbPtr actionTaskDb = make_shared<FetchTaskDb>(m_rootDir, "action");
  m_actionFetcher =
    make_shared<FetchManager>(m_ccnx, bind(&SyncLog::LookupLocator, &*m_syncLog, _1),
                              Name(BROADCAST_DOMAIN), // no appname suffix now
                              3,
                              bind(&Dispatcher::Did_FetchManager_ActionFetch, this, _1, _2, _3, _4),
                              FetchManager::FinishCallback(), actionTaskDb);

  FetchTaskDbPtr fileTaskDb = make_shared<FetchTaskDb>(m_rootDir, "file");
  m_fileFetcher =
    make_shared<FetchManager>(m_ccnx, bind(&SyncLog::LookupLocator, &*m_syncLog, _1),
                              Name(BROADCAST_DOMAIN), // no appname suffix now
                              3, bind(&Dispatcher::Did_FetchManager_FileSegmentFetch, this, _1, _2,
                                      _3, _4),
                              bind(&Dispatcher::Did_FetchManager_FileFetchComplete, this, _1, _2),
                              fileTaskDb);


  if (m_enablePrefixDiscovery) {
    _LOG_DEBUG("registering prefix discovery in Dispatcher");
    string tag = "dispatcher" + m_localUserName.toString();
    Ccnx::CcnxDiscovery::registerCallback(
      TaggedFunction(bind(&Dispatcher::Did_LocalPrefix_Updated, this, _1), tag));
  }

  m_executor.start();
}

Dispatcher::~Dispatcher()
{
  _LOG_DEBUG("Enter destructor of dispatcher");
  m_executor.shutdown();

  // _LOG_DEBUG (">>");

  if (m_enablePrefixDiscovery) {
    _LOG_DEBUG("deregistering prefix discovery in Dispatcher");
    string tag = "dispatcher" + m_localUserName.toString();
    Ccnx::CcnxDiscovery::deregisterCallback(
      TaggedFunction(bind(&Dispatcher::Did_LocalPrefix_Updated, this, _1), tag));
  }

  if (m_core != NULL) {
    delete m_core;
    m_core = NULL;
  }

  if (m_server != NULL) {
    delete m_server;
    m_server = NULL;
  }

  if (m_stateServer != NULL) {
    delete m_stateServer;
    m_stateServer = NULL;
  }
}

void
Dispatcher::Did_LocalPrefix_Updated(const Ccnx::Name& forwardingHint)
{
  Name effectiveForwardingHint;
  if (m_localUserName.size() >= forwardingHint.size() &&
      m_localUserName.getPartialName(0, forwardingHint.size()) == forwardingHint) {
    effectiveForwardingHint = Name("/"); // "directly" accesible
  }
  else {
    effectiveForwardingHint = forwardingHint;
  }

  Name oldLocalPrefix = m_syncLog->LookupLocalLocator();

  if (oldLocalPrefix == effectiveForwardingHint) {
    _LOG_DEBUG(
      "Got notification about prefix change from " << oldLocalPrefix << " to: " << forwardingHint
                                                   << ", but effective prefix didn't change");
    return;
  }

  if (effectiveForwardingHint == Name("/") || effectiveForwardingHint == Name(BROADCAST_DOMAIN)) {
    _LOG_DEBUG("Basic effective prefix [" << effectiveForwardingHint
                                          << "]. Updating local prefix, but don't reregister");
    m_syncLog->UpdateLocalLocator(effectiveForwardingHint);
    return;
  }

  _LOG_DEBUG("LocalPrefix changed from: " << oldLocalPrefix << " to: " << effectiveForwardingHint);

  m_server->registerPrefix(effectiveForwardingHint);
  m_syncLog->UpdateLocalLocator(effectiveForwardingHint);

  if (oldLocalPrefix == Name("/") || oldLocalPrefix == Name(BROADCAST_DOMAIN)) {
    _LOG_DEBUG("Don't deregister basic prefix: " << oldLocalPrefix);
    return;
  }
  m_server->deregisterPrefix(oldLocalPrefix);
}

// moved to state-server
// void
// Dispatcher::Restore_LocalFile (FileItemPtr file)
// {
//   m_executor.execute (bind (&Dispatcher::Restore_LocalFile_Execute, this, file));
// }

/////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////

void
Dispatcher::Did_LocalFile_AddOrModify(const filesystem::path& relativeFilePath)
{
  m_executor.execute(bind(&Dispatcher::Did_LocalFile_AddOrModify_Execute, this, relativeFilePath));
}

void
Dispatcher::Did_LocalFile_AddOrModify_Execute(filesystem::path relativeFilePath)
{
  _LOG_DEBUG(m_localUserName << " calls LocalFile_AddOrModify_Execute");
  filesystem::path absolutePath = m_rootDir / relativeFilePath;
  if (!filesystem::exists(absolutePath)) {
    //BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Update non exist file: " + absolutePath.string() ));
    _LOG_DEBUG("Update non exist file: " << absolutePath.string());
    return;
  }

  FileItemPtr currentFile = m_fileState->LookupFile(relativeFilePath.generic_string());
  if (currentFile &&
      *Hash::FromFileContent(absolutePath) ==
        Hash(currentFile->file_hash().c_str(), currentFile->file_hash().size())
      // The following two are commented out to prevent front end from reporting intermediate files
      // should enable it if there is other way to prevent this
      // && last_write_time (absolutePath) == currentFile->mtime ()
      // && status (absolutePath).permissions () == static_cast<filesystem::perms> (currentFile->mode ())
      ) {
    _LOG_ERROR("Got notification about the same file [" << relativeFilePath << "]");
    return;
  }

  if (currentFile && !currentFile->is_complete()) {
    _LOG_ERROR("Got notification about incomplete file [" << relativeFilePath << "]");
    return;
  }


  int seg_num;
  HashPtr hash;
  tie(hash, seg_num) = m_objectManager.localFileToObjects(absolutePath, m_localUserName);

  try {
    m_actionLog->AddLocalActionUpdate(relativeFilePath.generic_string(),
                                      *hash,
                                      last_write_time(absolutePath),
#if BOOST_VERSION >= 104900
                                      status(absolutePath).permissions(),
#else
                                      0,
#endif
                                      seg_num);

    // notify SyncCore to propagate the change
    m_core->localStateChangedDelayed();
  }
  catch (filesystem::filesystem_error& error) {
    _LOG_ERROR("File operations failed on [" << relativeFilePath << "] (ignoring)");
  }
}

void
Dispatcher::Did_LocalFile_Delete(const filesystem::path& relativeFilePath)
{
  m_executor.execute(bind(&Dispatcher::Did_LocalFile_Delete_Execute, this, relativeFilePath));
}

void
Dispatcher::Did_LocalFile_Delete_Execute(filesystem::path relativeFilePath)
{
  filesystem::path absolutePath = m_rootDir / relativeFilePath;
  if (filesystem::exists(absolutePath)) {
    //BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Delete notification but file exists: " + absolutePath.string() ));
    _LOG_ERROR("DELETE command, but file still exists: " << absolutePath.string());
    return;
  }

  m_actionLog->AddLocalActionDelete(relativeFilePath.generic_string());
  // notify SyncCore to propagate the change
  m_core->localStateChangedDelayed();
}

/////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////

/**
 * Callbacks:
 *
 * - from SyncLog: when state changes -> to fetch missing actions
 *
 * - from FetchManager/Actions: when action is fetched -> to request a file, specified by the action
 *                                                     -> to add action to the action log
 *
 * - from ActionLog/Delete:      when action applied (file state changed, file deleted)           -> to delete local file
 *
 * - from ActionLog/AddOrUpdate: when action applied (file state changes, file added or modified) -> do nothing?
 *
 * - from FetchManager/Files: when file segment is retrieved -> save it in ObjectDb
 *                            when file fetch is completed   -> if file belongs to FileState, then assemble it to filesystem. Don't do anything otherwise
 */

void
Dispatcher::Did_SyncLog_StateChange(SyncStateMsgPtr stateMsg)
{
  m_executor.execute(bind(&Dispatcher::Did_SyncLog_StateChange_Execute, this, stateMsg));
}

void
Dispatcher::Did_SyncLog_StateChange_Execute(SyncStateMsgPtr stateMsg)
{
  int size = stateMsg->state_size();
  int index = 0;
  // iterate and fetch the actions
  for (; index < size; index++) {
    SyncState state = stateMsg->state(index);
    if (state.has_old_seq() && state.has_seq()) {
      uint64_t oldSeq = state.old_seq();
      uint64_t newSeq = state.seq();
      Name userName(reinterpret_cast<const unsigned char*>(state.name().c_str()),
                    state.name().size());

      // fetch actions with oldSeq + 1 to newSeq (inclusive)
      Name actionNameBase = Name("/")(userName)(CHRONOSHARE_APP)("action")(m_sharedFolder);

      m_actionFetcher->Enqueue(userName, actionNameBase, std::max<uint64_t>(oldSeq + 1, 1), newSeq,
                               FetchManager::PRIORITY_HIGH);
    }
  }
}


void
Dispatcher::Did_FetchManager_ActionFetch(const Ccnx::Name& deviceName,
                                         const Ccnx::Name& actionBaseName, uint32_t seqno,
                                         Ccnx::PcoPtr actionPco)
{
  /// @todo Errors and exception checking
  _LOG_DEBUG("Received action deviceName: " << deviceName << ", actionBaseName: " << actionBaseName
                                            << ", seqno: "
                                            << seqno);

  ActionItemPtr action = m_actionLog->AddRemoteAction(deviceName, seqno, actionPco);
  if (!action) {
    _LOG_ERROR("AddRemoteAction did not insert action, ignoring");
    return;
  }
  // trigger may invoke Did_ActionLog_ActionApply_Delete or Did_ActionLog_ActionApply_AddOrModify callbacks

  if (action->action() == ActionItem::UPDATE) {
    Hash hash(action->file_hash().c_str(), action->file_hash().size());

    Name fileNameBase =
      Name("/")(deviceName)(CHRONOSHARE_APP)("file")(hash.GetHash(), hash.GetHashBytes());

    string hashStr = lexical_cast<string>(hash);
    if (ObjectDb::DoesExist(m_rootDir / ".chronoshare", deviceName, hashStr)) {
      _LOG_DEBUG(
        "File already exists in the database. No need to refetch, just directly applying the action");
      Did_FetchManager_FileFetchComplete(deviceName, fileNameBase);
    }
    else {
      if (m_objectDbMap.find(hash) == m_objectDbMap.end()) {
        _LOG_DEBUG("create ObjectDb for " << hash);
        m_objectDbMap[hash] = make_shared<ObjectDb>(m_rootDir / ".chronoshare", hashStr);
      }

      m_fileFetcher->Enqueue(deviceName, fileNameBase, 0, action->seg_num() - 1,
                             FetchManager::PRIORITY_NORMAL);
    }
  }
  // if necessary (when version number is the highest) delete will be applied through the trigger in m_actionLog->AddRemoteAction call
}

void
Dispatcher::Did_ActionLog_ActionApply_Delete(const std::string& filename)
{
  m_executor.execute(bind(&Dispatcher::Did_ActionLog_ActionApply_Delete_Execute, this, filename));
}

void
Dispatcher::Did_ActionLog_ActionApply_Delete_Execute(std::string filename)
{
  _LOG_DEBUG("Action to delete " << filename);

  filesystem::path absolutePath = m_rootDir / filename;
  try {
    if (filesystem::exists(absolutePath)) {
      // need some protection from local detection of removal
      remove(absolutePath);

      // hack to remove empty parent dirs
      filesystem::path parentPath = absolutePath.parent_path();
      while (parentPath > m_rootDir) {
        if (filesystem::is_empty(parentPath)) {
          filesystem::remove(parentPath);
          parentPath = parentPath.parent_path();
        }
        else {
          break;
        }
      }
    }
    // don't exist
  }
  catch (filesystem::filesystem_error& error) {
    _LOG_ERROR("File operations failed when removing [" << absolutePath << "] (ignoring)");
  }
}

void
Dispatcher::Did_FetchManager_FileSegmentFetch(const Ccnx::Name& deviceName,
                                              const Ccnx::Name& fileSegmentBaseName,
                                              uint32_t segment, Ccnx::PcoPtr fileSegmentPco)
{
  m_executor.execute(bind(&Dispatcher::Did_FetchManager_FileSegmentFetch_Execute, this, deviceName,
                          fileSegmentBaseName, segment, fileSegmentPco));
}

void
Dispatcher::Did_FetchManager_FileSegmentFetch_Execute(Ccnx::Name deviceName,
                                                      Ccnx::Name fileSegmentBaseName,
                                                      uint32_t segment, Ccnx::PcoPtr fileSegmentPco)
{
  // fileSegmentBaseName:  /<device_name>/<appname>/file/<hash>

  const Bytes& hashBytes = fileSegmentBaseName.getCompFromBack(0);
  Hash hash(head(hashBytes), hashBytes.size());

  _LOG_DEBUG("Received segment deviceName: " << deviceName << ", segmentBaseName: " << fileSegmentBaseName
                                             << ", segment: "
                                             << segment);

  // _LOG_DEBUG ("Looking up objectdb for " << hash);

  map<Hash, ObjectDbPtr>::iterator db = m_objectDbMap.find(hash);
  if (db != m_objectDbMap.end()) {
    db->second->saveContentObject(deviceName, segment, fileSegmentPco->buf());
  }
  else {
    _LOG_ERROR("no db available for this content object: " << fileSegmentBaseName << ", size: "
                                                           << fileSegmentPco->buf().size());
  }

  // ObjectDb objectDb (m_rootDir / ".chronoshare", lexical_cast<string> (hash));
  // objectDb.saveContentObject(deviceName, segment, fileSegmentPco->buf ());
}

void
Dispatcher::Did_FetchManager_FileFetchComplete(const Ccnx::Name& deviceName,
                                               const Ccnx::Name& fileBaseName)
{
  m_executor.execute(
    bind(&Dispatcher::Did_FetchManager_FileFetchComplete_Execute, this, deviceName, fileBaseName));
}

void
Dispatcher::Did_FetchManager_FileFetchComplete_Execute(Ccnx::Name deviceName, Ccnx::Name fileBaseName)
{
  // fileBaseName:  /<device_name>/<appname>/file/<hash>

  _LOG_DEBUG("Finished fetching " << deviceName << ", fileBaseName: " << fileBaseName);

  const Bytes& hashBytes = fileBaseName.getCompFromBack(0);
  Hash hash(head(hashBytes), hashBytes.size());
  _LOG_DEBUG("Extracted hash: " << hash.shortHash());

  if (m_objectDbMap.find(hash) != m_objectDbMap.end()) {
    // remove the db handle
    m_objectDbMap.erase(hash); // to commit write
  }
  else {
    _LOG_ERROR("no db available for this file: " << hash);
  }

  FileItemsPtr filesToAssemble = m_fileState->LookupFilesForHash(hash);

  for (FileItems::iterator file = filesToAssemble->begin(); file != filesToAssemble->end(); file++) {
    boost::filesystem::path filePath = m_rootDir / file->filename();

    try {
      if (filesystem::exists(filePath) && filesystem::last_write_time(filePath) == file->mtime() &&
#if BOOST_VERSION >= 104900
          filesystem::status(filePath).permissions() == static_cast<filesystem::perms>(file->mode()) &&
#endif
          *Hash::FromFileContent(filePath) == hash) {
        _LOG_DEBUG("Asking to assemble a file, but file already exists on a filesystem");
        continue;
      }
    }
    catch (filesystem::filesystem_error& error) {
      _LOG_ERROR("File operations failed on [" << filePath << "] (ignoring)");
    }

    if (ObjectDb::DoesExist(m_rootDir / ".chronoshare", deviceName,
                            boost::lexical_cast<string>(hash))) {
      bool ok = m_objectManager.objectsToLocalFile(deviceName, hash, filePath);
      if (ok) {
        last_write_time(filePath, file->mtime());
#if BOOST_VERSION >= 104900
        permissions(filePath, static_cast<filesystem::perms>(file->mode()));
#endif

        m_fileState->SetFileComplete(file->filename());
      }
      else {
        _LOG_ERROR("Notified about complete fetch, but file cannot be restored from the database: ["
                   << filePath
                   << "]");
      }
    }
    else {
      _LOG_ERROR(filePath << " supposed to have all segments, but not");
      // should abort for debugging
    }
  }
}

// moved to state-server
// void
// Dispatcher::Restore_LocalFile_Execute (FileItemPtr file)
// {
//   _LOG_DEBUG ("Got request to restore local file [" << file->filename () << "]");
//   // the rest will gracefully fail if object-db is missing or incomplete

//   boost::filesystem::path filePath = m_rootDir / file->filename ();
//   Name deviceName (file->device_name ().c_str (), file->device_name ().size ());
//   Hash hash (file->file_hash ().c_str (), file->file_hash ().size ());

//   try
//     {
//       if (filesystem::exists (filePath) &&
//           filesystem::last_write_time (filePath) == file->mtime () &&
//           filesystem::status (filePath).permissions () == static_cast<filesystem::perms> (file->mode ()) &&
//           *Hash::FromFileContent (filePath) == hash)
//         {
//           _LOG_DEBUG ("Asking to assemble a file, but file already exists on a filesystem");
//           return;
//         }
//     }
//   catch (filesystem::filesystem_error &error)
//     {
//       _LOG_ERROR ("File operations failed on [" << filePath << "] (ignoring)");
//     }

//   m_objectManager.objectsToLocalFile (deviceName, hash, filePath);

//   last_write_time (filePath, file->mtime ());
//   permissions (filePath, static_cast<filesystem::perms> (file->mode ()));

// }