/*=========================================================================

  Program:   ParaView
  Module:    vtkRemoteFileSystemProvider.cxx

  Copyright (c) Kitware, Inc.
  All rights reserved.
  See Copyright.txt or http://www.paraview.org/HTML/Copyright.html for details.

     This software is distributed WITHOUT ANY WARRANTY; without even
     the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
     PURPOSE.  See the above copyright notice for more information.

=========================================================================*/
#include "vtkRemoteFileSystemProvider.h"

#include "vtkDirectory.h"
#include "vtkNJson.h"
#include "vtkPVDataInformation.h"
#include "vtkPVFileInformation.h"
#include "vtkPVLogger.h"
#include "vtkPacket.h"
#include "vtkService.h"
#include "vtkSmartPointer.h"
#include "vtksys/Status.hxx"
#include "vtksys/SystemTools.hxx"

#include <exception>

class vtkRemoteFileSystemProvider::vtkInternals
{
public:
  rxcpp::composite_subscription Subscription;

  vtkInternals() = default;

  ~vtkInternals() { this->Subscription.unsubscribe(); }

  void Preview(const vtkPacket& packet){};

  vtkPacket Process(const vtkPacket& packet);

private:
  vtkNJson MakeDirectory(const std::string& fullpath) const;
  vtkNJson DeleteDirectory(const std::string& fullpath) const;
  vtkNJson RenameDirectory(const std::string& oldFullPath, const std::string& newFullPath) const;
  vtkNJson ListDirectory(const vtkNJson& request) const;
};

//----------------------------------------------------------------------------
vtkNJson vtkRemoteFileSystemProvider::vtkInternals::MakeDirectory(const std::string& fullpath) const
{
  vtkVLogF(PARAVIEW_LOG_PROVIDER_VERBOSITY(), "MakeDirectory(%s)", fullpath.c_str());
  vtkNJson reply;
  vtksys::Status status = vtksys::SystemTools::MakeDirectory(fullpath);
  reply["status"] = status.IsSuccess();
  if (!status.IsSuccess())
  {
    reply["__vtk_error_message__"] = status.GetString();
  }
  return reply;
}

//----------------------------------------------------------------------------
vtkNJson vtkRemoteFileSystemProvider::vtkInternals::DeleteDirectory(
  const std::string& fullpath) const
{
  vtkVLogF(PARAVIEW_LOG_PROVIDER_VERBOSITY(), "DeleteDirectory(%s)", fullpath.c_str());
  vtkNJson reply;
  vtksys::Status status = vtksys::SystemTools::RemoveADirectory(fullpath);
  reply["status"] = status.IsSuccess();
  if (!status.IsSuccess())
  {
    reply["__vtk_error_message__"] = status.GetString();
  }
  return reply;
}

//----------------------------------------------------------------------------
vtkNJson vtkRemoteFileSystemProvider::vtkInternals::RenameDirectory(
  const std::string& oldFullPath, const std::string& newFullPath) const
{
  vtkVLogF(PARAVIEW_LOG_PROVIDER_VERBOSITY(), "RenameDirectory(%s, %s)", oldFullPath.c_str(),
    newFullPath.c_str());
  vtkNJson reply;
  const bool status = (vtkDirectory::Rename(oldFullPath.c_str(), newFullPath.c_str()) == 1);
  reply["status"] = status;
  if (!status)
  {
    reply["__vtk_error_message__"] = std::strerror(errno);
  }
  return reply;
}

//-----------------------------------------------------------------------------
vtkNJson vtkRemoteFileSystemProvider::vtkInternals::ListDirectory(const vtkNJson& request) const
{
  vtkNew<vtkPVFileInformation> info;
  // initialize the information object.
  info->LoadState(request.at("state"));

  info->GatherInformation(nullptr);
  auto state = info->SaveInformation();
  // TODO how to capture error ?
  return state;
}
//----------------------------------------------------------------------------
bool vtkRemoteFileSystemProvider::ParseResponse(
  const vtkPacket& packet, vtkPVFileInformation* information)
{
  return information->LoadInformation(packet.GetJSON());
}
//----------------------------------------------------------------------------
bool vtkRemoteFileSystemProvider::ParseResponse(
  const vtkNJson& state, vtkPVFileInformation* information)
{
  return information->LoadInformation(state);
}
//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::vtkInternals::Process(const vtkPacket& packet)
{
  const auto& json = packet.GetJSON();
  vtkVLogScopeF(PARAVIEW_LOG_PROVIDER_VERBOSITY(), "Process: %s", json.dump(-1).c_str());
  try
  {
    const auto& type = json.at("type").get<std::string>();

    if (type == "vtk-remote-filesystem-make-directory")
    {
      return this->MakeDirectory(json.at("path").get<std::string>());
    }

    if (type == "vtk-remote-filesystem-delete-directory")
    {
      return this->DeleteDirectory(json.at("path").get<std::string>());
    }

    if (type == "vtk-remote-filesystem-rename-directory")
    {
      return this->RenameDirectory(
        json.at("old_path").get<std::string>(), json.at("new_path").get<std::string>());
    }

    if (type == "vtk-remote-filesystem-list-directory")
    {
      return this->ListDirectory(json);
    }

    const std::string errorMsg = "unknown rpc type" + type;
    return { { "__vtk_error_message__", errorMsg } };
  }
  catch (std::exception& exception)
  {
    const std::string errorMsg = exception.what();
    return { { "__vtk_error_message__", errorMsg } };
  }
  return { { "__vtk_error_message__", "Uncaught Error" } };
}

//============================================================================
vtkObjectFactoryNewMacro(vtkRemoteFileSystemProvider);
//----------------------------------------------------------------------------
vtkRemoteFileSystemProvider::vtkRemoteFileSystemProvider()
  : Internals(new vtkRemoteFileSystemProvider::vtkInternals())
{
}

//----------------------------------------------------------------------------
vtkRemoteFileSystemProvider::~vtkRemoteFileSystemProvider()
{
  vtkVLogF(PARAVIEW_LOG_PROVIDER_VERBOSITY(),
    "vtkRemoteFileSystemProvider::~vtkRemoteFileSystemProvider()");
}

//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::MakeDirectory(const std::string& fullpath)
{
  vtkNJson message;
  message["type"] = "vtk-remote-filesystem-make-directory";
  message["path"] = fullpath;
  return { message };
}

//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::DeleteDirectory(const std::string& fullpath)
{
  vtkNJson message;
  message["type"] = "vtk-remote-filesystem-delete-directory";
  message["path"] = fullpath;
  return { message };
}

//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::RenameDirectory(
  const std::string& oldFullPath, const std::string& newFullPath)
{
  vtkNJson message;
  message["type"] = "vtk-remote-filesystem-rename-directory";
  message["old_path"] = oldFullPath;
  message["new_path"] = newFullPath;
  return { message };
}

//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::ListDirectory(vtkPVFileInformation* information)
{
  vtkNJson message;
  message["type"] = "vtk-remote-filesystem-list-directory";
  message["state"] = information->SaveState();
  message["run_on_rpc"] = information->CanRunOnRPC();
  return { message };
}

//----------------------------------------------------------------------------
void vtkRemoteFileSystemProvider::InitializeInternal(vtkService* service)
{
  assert(service != nullptr);
  auto& internals = (*this->Internals);

  auto observable = service->GetRequestObservable(/*skipEventLoop*/ true);

  // since we want to handle each message in order it is received we create a
  // single processing stream.
  internals.Subscription =
    observable
      .filter([](const vtkServiceReceiver& receiver) {
        auto& json = receiver.GetPacket().GetJSON();
        auto iter = json.find("type");
        return iter != json.end() &&
          iter.value().get<std::string>().find("vtk-remote-filesystem-") != std::string::npos;
      })
      .map([this](const vtkServiceReceiver& receiver) {
        this->Preview(receiver.GetPacket());
        return receiver;
      })
      .filter([this](const vtkServiceReceiver& receiver) {
        // handle "run_on_rpc" requests.
        const auto& json = receiver.GetPacket().GetJSON();
        if (json.value("run_on_rpc", false))
        {
          receiver.Respond(this->ProcessOnRPC(receiver.GetPacket()));
          return false;
        }
        return true;
      })
      .observe_on(service->GetRunLoopScheduler())
      .subscribe([this](const vtkServiceReceiver& receiver) {
        receiver.Respond(this->Process(receiver.GetPacket()));
      });
}

//----------------------------------------------------------------------------
void vtkRemoteFileSystemProvider::Preview(const vtkPacket& packet)
{
  auto& internals = (*this->Internals);
  internals.Preview(packet);
}

//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::ProcessOnRPC(const vtkPacket& packet)
{
  auto& internals = (*this->Internals);
  return internals.Process(packet);
}

//----------------------------------------------------------------------------
vtkPacket vtkRemoteFileSystemProvider::Process(const vtkPacket& packet)
{
  auto& internals = (*this->Internals);
  return internals.Process(packet);
}

//----------------------------------------------------------------------------
void vtkRemoteFileSystemProvider::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
}
