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

  Program:   Visualization Toolkit
  Module:    vtkPTSPipelineBuilder.cxx

  Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
  All rights reserved.
  See Copyright.txt or http://www.kitware.com/Copyright.htm 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 "vtkPTSPipelineBuilder.h"

#include "vtkClientSession.h"
#include "vtkRemotingCoreUtilities.h"
#include "vtkSMCoreUtilities.h"
#include "vtkSMParaViewPipelineController.h"
#include "vtkSMProperty.h"
#include "vtkSMPropertyHelper.h"
#include "vtkSMProxy.h"
#include "vtkSMProxyProperty.h"
#include "vtkSMSessionProxyManager.h"
#include "vtkSMSourceProxy.h"
#include "vtkSmartPointer.h"

#include "vtk_rxcpp.h" // for rxcpp
// clang-format off
#include VTK_REMOTING_RXCPP(rx-util.hpp) // for rxcpp::util::apply_to
// clang-format on

#ifdef VTK_MODULE_ENABLE_ParaView_RemotingServerManagerViews
#include "vtkSMApplyController.h"
#include "vtkSMParaViewPipelineControllerWithRendering.h"
#endif

// clang-format off
#include <vtk_fmt.h> // needed for `fmt`
#include VTK_FMT(fmt/core.h)
// clang-format on

#include <exception> // for std::exception_ptr

class vtkPTSPipelineBuilder::vtkInternals
{
public:
  const std::thread::id OwnerTID{ std::this_thread::get_id() };
  vtkSmartPointer<vtkClientSession> Session = { nullptr };

  bool IsInitialized() const { return Session != nullptr; }

  /**
   * Create a proxy and initialize it with vtkSMParaViewPipelineController.
   *
   * @section Triggers
   *
   * * on_next: pointer to the created proxy.
   *
   * * on_error: if \p _group & \p _name do not correspond to any proxy, or if
   * error occured during the
   * initialization proccess. Check the `std::exception_ptr` argument for
   * details.
   *
   * * on_completed: right after on_next since this is a single value observable
   */
  rxcpp::observable<vtkSMProxy*> CreateProxy(vtkSMParaViewPipelineController* controller,
    const std::string& _group, const std::string& _name);

  /**
   * Perform any steps required to complete the initialization steps of a source proxy such as
   * registration etc.
   *
   * @section Triggers
   *
   * * on_next: pointer to the created proxy.
   *
   * * on_error: if error occured during the initialization proccess. Check the
   * `std::exception_ptr` argument for details.
   *
   * * on_completed: right after on_next since this is a single value observable
   */
  rxcpp::observable<vtkSMSourceProxy*> CompleteInitialization(
    vtkSmartPointer<vtkSMParaViewPipelineController> controller,
    rxcpp::observable<vtkSMProxy*> observable);
};

rxcpp::observable<vtkSMProxy*> vtkPTSPipelineBuilder::vtkInternals::CreateProxy(
  vtkSMParaViewPipelineController* controller, const std::string& _group, const std::string& _name)
{
  const char* group = _group.c_str();
  const char* name = _name.c_str();
  vtkSMProxy* newProxy = nullptr;

  vtkSMSessionProxyManager* pxm = this->Session->GetProxyManager();
  if (pxm && pxm->GetPrototypeProxy(group, name))
  {
    newProxy = pxm->NewProxy(group, name);
  }
  if (!newProxy)
  {
    const std::string errorMsg =
      fmt::format("Could not create proxy`{}` of group `{}`", name, group);
    return rxcpp::observable<>::error<vtkSMProxy*>(std::runtime_error(errorMsg.c_str()));
  }

  const bool status = controller->PreInitializeProxy(newProxy);

  if (!status)
  {
    const std::string errorMsg =
      fmt::format("PreInitialization of proxy proxy`{}` of group `{}` failed", name, group);
    return rxcpp::observable<>::error<vtkSMProxy*>(std::runtime_error(errorMsg.c_str()));
  }

  return rxcpp::observable<>::just(newProxy);
}

//-----------------------------------------------------------------------------
rxcpp::observable<vtkSMSourceProxy*> vtkPTSPipelineBuilder::vtkInternals::CompleteInitialization(
  vtkSmartPointer<vtkSMParaViewPipelineController> controller,
  rxcpp::observable<vtkSMProxy*> observable)
{

  return observable
    .map([controller](vtkSMProxy* proxy) {
      auto proxyObservable = rxcpp::observable<>::just(proxy);
      auto initializationObservable = controller->PostInitializeProxy(proxy);
      return initializationObservable.zip(proxyObservable);
    })
    .switch_on_next()
    .map(rxcpp::util::apply_to([controller](bool status, vtkSMProxy* proxy) {
      if (status)
      {
        controller->RegisterPipelineProxy(proxy);
        return proxy;
      }
      else
      {
        const std::string errorMsg = fmt::format("Could not initialize proxy`{}` of group `{}`",
          proxy->GetXMLName(), proxy->GetXMLGroup());
        throw std::runtime_error(errorMsg.c_str());
      }
    }))
    .map([](vtkSMProxy* proxy) {
      vtkSMSourceProxy* source = vtkSMSourceProxy::SafeDownCast(proxy);
      if (!source)
      {
        const std::string errorMsg = fmt::format("Proxy`{}` of group `{}` is not a SourceProxy !",
          proxy->GetXMLName(), proxy->GetXMLGroup());
        throw std::runtime_error(errorMsg.c_str());
      }
#ifdef VTK_MODULE_ENABLE_ParaView_RemotingServerManagerViews
      vtkSMApplyController::MarkShowOnApply(source);
#endif
      return source;
    })
    .take(1);
}

//=============================================================================

//-----------------------------------------------------------------------------
vtkStandardNewMacro(vtkPTSPipelineBuilder);

//-----------------------------------------------------------------------------
vtkPTSPipelineBuilder::vtkPTSPipelineBuilder()
  : Internals(new vtkPTSPipelineBuilder::vtkInternals())
{
}

//-----------------------------------------------------------------------------
vtkPTSPipelineBuilder::~vtkPTSPipelineBuilder() = default;

//-----------------------------------------------------------------------------
void vtkPTSPipelineBuilder::SetSession(vtkClientSession* session)
{
  auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  if (internals.Session != session)
  {
    // TODO error handling
    vtkNew<vtkSMParaViewPipelineController> controller;
    controller->InitializeSession(session);
    internals.Session = session;
  }
}

//-----------------------------------------------------------------------------
vtkClientSession* vtkPTSPipelineBuilder::GetSession() const
{
  const auto& internals = (*this->Internals);
  vtkRemotingCoreUtilities::EnsureThread(internals.OwnerTID);
  return internals.Session;
}

//-----------------------------------------------------------------------------
rxcpp::observable<vtkSMSourceProxy*> vtkPTSPipelineBuilder::CreateSource(
  const std::string& group, const std::string& name)
{
  auto& internals = (*this->Internals);
  if (!internals.IsInitialized())
  {
    return rxcpp::observable<>::error<vtkSMSourceProxy*>(
      std::runtime_error("Session was not set!"));
  }
  vtkNew<vtkSMParaViewPipelineController> controller;
  auto observable =
    internals.CreateProxy(controller, group, name).map([](vtkSMProxy* proxy) { return proxy; });
  return internals.CompleteInitialization(controller, observable);
}

//-----------------------------------------------------------------------------
rxcpp::observable<vtkSMSourceProxy*> vtkPTSPipelineBuilder::CreateFilter(
  const std::string& group, const std::string& name, vtkSMSourceProxy* input)
{
  auto& internals = (*this->Internals);
  if (!internals.IsInitialized())
  {
    return rxcpp::observable<>::error<vtkSMSourceProxy*>(
      std::runtime_error("Session was not set!"));
  }
  vtkNew<vtkSMParaViewPipelineController> controller;
  auto observable = internals.CreateProxy(controller, group, name).map([input](vtkSMProxy* proxy) {
    vtkSMProperty* prop = proxy->GetProperty("Input");
    if (!prop)
    {
      const std::string errorMsg =
        fmt::format("Cannot find `Input` property of Proxy`{}` of group `{}` !",
          proxy->GetXMLName(), proxy->GetXMLGroup());
      throw std::runtime_error(errorMsg.c_str());
    }

    vtkSMPropertyHelper(proxy, "Input").Set(input, 0);
    return proxy;
  });

  return internals.CompleteInitialization(controller, observable);
}

//-----------------------------------------------------------------------------
rxcpp::observable<vtkSMSourceProxy*> vtkPTSPipelineBuilder::CreateReader(
  const std::string& group, const std::string& name, const std::string& filename)
{
  auto& internals = (*this->Internals);
  if (!internals.IsInitialized())
  {
    return rxcpp::observable<>::error<vtkSMSourceProxy*>(
      std::runtime_error("Session was not set!"));
  }
  vtkNew<vtkSMParaViewPipelineController> controller;
  auto observable =
    internals.CreateProxy(controller, group, name).map([filename](vtkSMProxy* proxy) {
      // TODO
      vtkSMProperty* prop = proxy->GetProperty("FileName");
      if (!prop)
      {
        const std::string errorMsg =
          fmt::format("Cannot find 'FileName' property of Proxy`{}` of group `{}`!",
            proxy->GetXMLName(), proxy->GetXMLGroup());
        throw std::runtime_error(errorMsg.c_str());
      }

      vtkSMPropertyHelper(proxy, "FileName").Set(filename.c_str());
      return proxy;
    });

  return internals.CompleteInitialization(controller, observable);
}

//-----------------------------------------------------------------------------
#ifdef VTK_MODULE_ENABLE_ParaView_RemotingServerManagerViews
rxcpp::observable<vtkSMViewProxy*> vtkPTSPipelineBuilder::CreateView(
  const std::string& group, const std::string& name)
{
  auto& internals = (*this->Internals);
  if (!internals.IsInitialized())
  {
    return rxcpp::observable<>::error<vtkSMViewProxy*>(std::runtime_error("Session was not set!"));
  }
  auto controller = vtk::TakeSmartPointer(vtkSMParaViewPipelineControllerWithRendering::New());
  return internals.CreateProxy(controller, group, name)
    .map([controller](vtkSMProxy* proxy) {
      auto proxyObservable = rxcpp::observable<>::just(proxy);
      auto initializationObservable = controller->PostInitializeProxy(proxy);
      return initializationObservable.zip(proxyObservable);
    })
    .switch_on_next()
    .map(rxcpp::util::apply_to([controller](bool status, vtkSMProxy* proxy) {
      if (status)
      {
        controller->RegisterViewProxy(proxy);
        return proxy;
      }
      else
      {
        const std::string errorMsg = fmt::format("Could not initialize proxy`{}` of group `{}`",
          proxy->GetXMLName(), proxy->GetXMLGroup());
        throw std::runtime_error(errorMsg.c_str());
      }
    }))
    .map([controller](vtkSMProxy* proxy) {
      vtkSMViewProxy* view = vtkSMViewProxy::SafeDownCast(proxy);
      if (!view)
      {
        const std::string errorMsg = fmt::format("Proxy`{}` of group `{}` is not a ViewProxy !",
          proxy->GetXMLName(), proxy->GetXMLGroup());
        throw std::runtime_error(errorMsg.c_str());
      }
      controller->AssignViewToLayout(view);
      return view;
    })
    .take(1);
}
#else
rxcpp::observable<vtkSMProxy*> vtkPTSPipelineBuilder::CreateView(
  const std::string& group, const std::string& name)
{
  return rxcpp::observable<>::error<vtkSMProxy*>(
    std::runtime_error("ParaT was compiled without rendering!"));
}
#endif

//-----------------------------------------------------------------------------
rxcpp::observable<vtkSMProxy*> vtkPTSPipelineBuilder::CreateRepresentation(
  vtkSMSourceProxy* producer, int outputPort, vtkSMViewProxy* view, const char* representationType)
{
#if VTK_MODULE_ENABLE_ParaView_RemotingServerManagerViews
  if (producer == nullptr)
  {
    return rxcpp::observable<>::error<vtkSMProxy*>(
      std::runtime_error("producer proxy is invalid!"));
  }

  vtkNew<vtkSMParaViewPipelineControllerWithRendering> controller;

  vtkSMProxy* representation = controller->Show(producer, outputPort, view, representationType);
  // ASYNC FIXME
  // representation is already Registered with the view. Here we increase the
  // reference count in order to avoid `double free` if representation is deleted
  // after the view is gone.
  representation->Register(nullptr);

  return rxcpp::observable<>::just(representation);

#else
  return rxcpp::observable<>::error<vtkSMProxy*>(
    std::runtime_error("ParaT was compiled without rendering!"));
#endif
}

//-----------------------------------------------------------------------------
rxcpp::observable<bool> vtkPTSPipelineBuilder::DeleteProxy(vtkSMProxy* proxy)
{
  const auto& internals = (*this->Internals);
  if (!internals.IsInitialized())
  {
    return rxcpp::observable<>::error<bool>(std::runtime_error("Session was not set!"));
  }

  if (!proxy)
  {
    return rxcpp::observable<>::error<bool>(std::runtime_error("Proxy is nullptr"));
  }
  vtkNew<vtkSMParaViewPipelineController> controller;

  // if there are consumers that are not representations we cannot delete the proxy
  for (unsigned int cc = 0, max = proxy->GetNumberOfConsumers(); cc < max; ++cc)
  {
    vtkSMProxy* consumer = proxy->GetConsumerProxy(cc);
    consumer = consumer ? consumer->GetTrueParentProxy() : nullptr;
    if (consumer)
    {
      if (strcmp(consumer->GetXMLGroup(), "representations") == 0)
      {
        const bool status = controller->UnRegisterProxy(consumer);
        if (!status)
        {
          return rxcpp::observable<>::just(false);
        }

        continue;
      }

      return rxcpp::observable<>::just(false);
    }
  }

  // break links between the producers of this proxy and this proxy
  for (unsigned int cc = 0, max = proxy->GetNumberOfProducers(); cc < max; ++cc)
  {
    if (vtkSMProxyProperty* property =
          vtkSMProxyProperty::SafeDownCast(proxy->GetProducerProperty(cc)))
    {
      property->RemoveAllProxies();
    }
  }

  const bool status = controller->UnRegisterProxy(proxy);

  return rxcpp::observable<>::just(status);
}
//-----------------------------------------------------------------------------
void vtkPTSPipelineBuilder::PrintSelf(ostream& os, vtkIndent indent)
{
  this->Superclass::PrintSelf(os, indent);
}
