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 "fetch-task-db.hpp"
#include "core/logging.hpp"

#include <ndn-cxx/util/digest.hpp>
#include <ndn-cxx/util/string-helper.hpp>

#include <boost/lexical_cast.hpp>

namespace ndn {
namespace chronoshare {

_LOG_INIT(Dispatcher);

namespace fs = boost::filesystem;

static const name::Component CHRONOSHARE_APP = name::Component("chronoshare");
static const Name BROADCAST_DOMAIN = "/ndn/multicast";

static const time::seconds DEFAULT_SYNC_INTEREST_INTERVAL = time::seconds(10);
static const time::seconds DEFAULT_AUTO_DISCOVERY_INTERVAL = time::seconds(60);

Dispatcher::Dispatcher(const std::string& localUserName, const std::string& sharedFolder,
                       const fs::path& rootDir, Face& face)
  : m_face(face)
  , m_rootDir(rootDir)
  , m_ioService(face.getIoService())
  , m_scheduler(m_ioService)
  , m_autoDiscovery(m_scheduler)
  , m_objectManager(face, m_keyChain, rootDir, CHRONOSHARE_APP.toUri())
  , m_localUserName(localUserName)
  , m_sharedFolder(sharedFolder)
{
  m_syncLog = make_shared<SyncLog>(m_rootDir, localUserName);
  m_actionLog =
    make_shared<ActionLog>(m_face, 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);
  syncPrefix.append(CHRONOSHARE_APP);
  syncPrefix.append(sharedFolder);

  m_server = make_unique<ContentServer>(m_face, m_actionLog, rootDir, m_localUserName,
                                        m_sharedFolder, CHRONOSHARE_APP, m_keyChain);
  m_server->registerPrefix(Name("/"));
  m_server->registerPrefix(Name(BROADCAST_DOMAIN));

  m_stateServer = make_unique<StateServer>(m_face, m_actionLog, rootDir, m_localUserName, m_sharedFolder,
                                           CHRONOSHARE_APP, m_objectManager, m_keyChain);

  m_core = make_unique<SyncCore>(m_face, m_syncLog, localUserName, Name("/"), syncPrefix,
                                 bind(&Dispatcher::Did_SyncLog_StateChange, this, _1),
                                 DEFAULT_SYNC_INTEREST_INTERVAL);

  FetchTaskDbPtr actionTaskDb = make_shared<FetchTaskDb>(m_rootDir, "action");
  m_actionFetcher =
    make_shared<FetchManager>(m_face, 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_face, 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);

  _LOG_DEBUG("registering prefix discovery in Dispatcher");
  m_autoDiscovery = m_scheduler.scheduleEvent(DEFAULT_AUTO_DISCOVERY_INTERVAL,
                                              bind(&Dispatcher::DiscoverPrefix, this));
}

Dispatcher::~Dispatcher()
{
  _LOG_DEBUG("Enter destructor of dispatcher");
}

void
Dispatcher::DiscoverPrefix()
{
  _LOG_DEBUG("Query for local prefix");
  Interest interest(Name("/localhop/nfd/rib/routable-prefixes"));
  m_face.expressInterest(interest,
      bind(&Dispatcher::Handle_Prefix_Discovery, this, _1, _2),
      bind([this] {m_autoDiscovery = m_scheduler.scheduleEvent(DEFAULT_AUTO_DISCOVERY_INTERVAL,
                                                               bind(&Dispatcher::DiscoverPrefix, this));
                   _LOG_DEBUG("query routable-prefixes, nack received"); }),
      bind([this] {m_autoDiscovery = m_scheduler.scheduleEvent(DEFAULT_AUTO_DISCOVERY_INTERVAL,
                                                               bind(&Dispatcher::DiscoverPrefix, this));
                   _LOG_DEBUG("query routable-prefixes interest timeout"); }));
}

void
Dispatcher::Handle_Prefix_Discovery(const Interest& interest, const Data& data)
{
  const Block& content = data.getContent();
  content.parse();

  Name prefix;

  for (auto& element : content.elements()) {
    prefix = Name(element);
    break;
  }

  _LOG_DEBUG("local prefix:" << prefix << prefix.empty());
  if(prefix.empty()) {
    _LOG_DEBUG("Local prefix is empty");
    return;
  }

  _LOG_DEBUG("local prefix:" << prefix << "discovered");
  Did_LocalPrefix_Updated(prefix);

  m_autoDiscovery = m_scheduler.scheduleEvent(DEFAULT_AUTO_DISCOVERY_INTERVAL,
                                              bind(&Dispatcher::DiscoverPrefix, this));
}

void
Dispatcher::Did_LocalPrefix_Updated(const Name& forwardingHint)
{
  Name effectiveForwardingHint;
  if (m_localUserName.size() >= forwardingHint.size() &&
      m_localUserName.getSubName(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);
}

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

void
Dispatcher::Did_LocalFile_AddOrModify(const fs::path& relativeFilePath)
{
  m_ioService.post(bind(&Dispatcher::Did_LocalFile_AddOrModify_Execute, this, relativeFilePath));
}

void
Dispatcher::Did_LocalFile_AddOrModify_Execute(fs::path relativeFilePath)
{
  _LOG_DEBUG(m_localUserName << " calls LocalFile_AddOrModify_Execute");
  fs::path absolutePath = m_rootDir / relativeFilePath;
  _LOG_DEBUG("relativeFilePath : " << relativeFilePath);
  _LOG_DEBUG("absolutePath : " << absolutePath);
  if (!fs::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) {
    fs::ifstream input(absolutePath);
    if (*util::Sha256(input).computeDigest() ==
        Buffer(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<fs::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;
  ConstBufferPtr hash;
  _LOG_DEBUG("absolutePath: " << absolutePath << " m_localUserName: " << m_localUserName);
  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 (const fs::filesystem_error& error) {
    _LOG_ERROR("File operations failed on [" << relativeFilePath << "](ignoring)");
  }

  _LOG_DEBUG("LocalFile_AddOrModify_Execute Finished!");
}

void
Dispatcher::Did_LocalFile_Delete(const fs::path& relativeFilePath)
{
  m_ioService.post(bind(&Dispatcher::Did_LocalFile_Delete_Execute, this, relativeFilePath));
}

void
Dispatcher::Did_LocalFile_Delete_Execute(fs::path relativeFilePath)
{
  fs::path absolutePath = m_rootDir / relativeFilePath;
  if (fs::exists(absolutePath)) {
    _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();
}

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

void
Dispatcher::Did_SyncLog_StateChange(SyncStateMsgPtr stateMsg)
{
  m_ioService.post(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(Block((const unsigned char*)state.name().c_str(), state.name().size()));

      // fetch actions with oldSeq + 1 to newSeq(inclusive)
      Name actionNameBase = Name("/");
      actionNameBase.append(userName).append(CHRONOSHARE_APP).append("action").append(m_sharedFolder);
      m_actionFetcher->Enqueue(userName, actionNameBase, std::max<uint64_t>(oldSeq + 1, 1), newSeq,
                               FetchManager::PRIORITY_HIGH);
    }
  }
}

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

  ActionItemPtr action = m_actionLog->AddRemoteAction(deviceName, seqno, actionData);
  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) {
    ConstBufferPtr hash =
      make_shared<Buffer>(action->file_hash().c_str(), action->file_hash().size());

    Name fileNameBase = Name("/");
    fileNameBase.append(deviceName).append(CHRONOSHARE_APP).append("file");
    //      fileNameBase.append(name::Component(hash));
    fileNameBase.appendImplicitSha256Digest(hash);

    std::string hashStr = toHex(*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 " << toHex(*hash));
        m_objectDbMap[*hash] = make_shared<ObjectDb>(m_rootDir / ".chronoshare", hashStr);
      }

      m_fileFetcher->Enqueue(deviceName, fileNameBase, 0, action->seg_num() - 1,
                             FetchManager::PRIORITY_NORMAL);
    }
  }
}

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

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

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

void
Dispatcher::Did_FetchManager_FileSegmentFetch(const Name& deviceName,
                                              const Name& fileSegmentBaseName,
                                              uint32_t segment,
                                              shared_ptr<Data>
                                                fileSegmentData)
{
  m_ioService.post(bind(&Dispatcher::Did_FetchManager_FileSegmentFetch_Execute, this, deviceName,
                        fileSegmentBaseName, segment, fileSegmentData));
}

void
Dispatcher::Did_FetchManager_FileSegmentFetch_Execute(Name deviceName,
                                                      Name fileSegmentBaseName,
                                                      uint32_t segment,
                                                      shared_ptr<Data> fileSegmentData)
{
  // fileSegmentBaseName:  /<device_name>/<appname>/file/<hash>

  Buffer hash(fileSegmentBaseName.get(-1).value(), fileSegmentBaseName.get(-1).value_size());

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

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

  std::map<Buffer, shared_ptr<ObjectDb>>::iterator db = m_objectDbMap.find(hash);
  if (db != m_objectDbMap.end()) {
    db->second->saveContentObject(deviceName, segment, *fileSegmentData);
  }
  else {
    _LOG_ERROR("no db available for this content object: " << fileSegmentBaseName << ", size: "
                                                           << fileSegmentData->getContent().size());
  }

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

void
Dispatcher::Did_FetchManager_FileFetchComplete(const Name& deviceName, const Name& fileBaseName)
{
  m_ioService.post(
    bind(&Dispatcher::Did_FetchManager_FileFetchComplete_Execute, this, deviceName, fileBaseName));
}

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

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

  Buffer hash(fileBaseName.get(-1).value(), fileBaseName.get(-1).value_size());

  _LOG_DEBUG("Extracted hash: " << toHex(hash));

  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: " << toHex(hash));
  }

  FileItemsPtr filesToAssemble = m_fileState->LookupFilesForHash(hash);

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

    try {
      if (fs::exists(filePath) && fs::last_write_time(filePath) == file->mtime() &&
          fs::status(filePath).permissions() == static_cast<fs::perms>(file->mode())) {
          fs::ifstream input(filePath, std::ios::in | std::ios::binary);
          if (*util::Sha256(input).computeDigest() == hash) {
            _LOG_DEBUG("Asking to assemble a file, but file already exists on a filesystem");
            continue;
          }
      }
    }
    catch (const fs::filesystem_error& error) {
      _LOG_ERROR("File operations failed on [" << filePath << "](ignoring)");
    }

    if (ObjectDb::doesExist(m_rootDir / ".chronoshare", deviceName, toHex(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<fs::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
    }
  }
}

} // namespace chronoshare
} // namespace ndn