path: root/whiteboard.cpp

#include "whiteboard.h"

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <dirent.h>
#include <sys/types.h>

#include <chrono>
#include <initializer_list>
#include <iostream>
#include <functional>
#include <filesystem>
#include <mutex>
#include <regex>
#include <sstream>
#include <string>
#include <thread>
#include <unordered_map>

#include <boost/algorithm/string/predicate.hpp>
#include <boost/algorithm/string/trim.hpp>
#include <boost/property_tree/xml_parser.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/websocket.hpp>
#include <boost/beast/version.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/buffers_iterator.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/ip/tcp.hpp>

#include <fmt/core.h>

#include "libreichwein/base64.h"
#include "libreichwein/file.h"
#include "libreichwein/xml.h"

#include "config.h"
#include "diff.h"
#include "qrcode.h"
#include "storage.h"

namespace pt = boost::property_tree;
using namespace std::string_literals;
namespace fs = std::filesystem;

namespace {

 void usage() {
  std::cout <<
   "Usage: \n"
   " whiteboard [options]\n"
   "\n"
   "Options:\n"
   " -c <path> : specify configuration file including path\n"
   " -C : clean up database according to timeout rules (config: maxage)\n"
   " -h : this help\n"
   "\n"
   "Without options, whiteboard will be started as websocket application"
   << std::endl;
 }

} // namespace

Whiteboard::Whiteboard()
{
}

// contents of cleanup thread; looping
void Whiteboard::storage_cleanup()
{
 while(true) {
  {
   std::lock_guard<std::mutex> lock(m_storage_mutex);
   if (!m_storage)
    throw std::runtime_error("Storage not initialized");
   m_storage->cleanup();
  }
  std::this_thread::sleep_for(std::chrono::minutes(10));
 }
}

pt::ptree make_ptree(const std::initializer_list<std::pair<std::string, std::string>>& key_values)
{
 pt::ptree ptree;
 for (const auto& i: key_values) {
  ptree.put(fmt::format("serverinfo.{}", i.first), i.second);
 }

 return ptree;
}

std::string make_xml(const std::initializer_list<std::pair<std::string, std::string>>& key_values)
{
 pt::ptree ptree{make_ptree(key_values)};
 return Reichwein::XML::plain_xml(ptree);
}

void Whiteboard::notify_other_connections_diff(Whiteboard::connection& c, const std::string& id, const Diff& diff)
{
 std::for_each(m_registry.begin(id), m_registry.end(id), [&](const Whiteboard::connection& ci)
               {
                if (c != ci) {
                 boost::beast::flat_buffer buffer;
                 pt::ptree ptree {make_ptree({
                   {"type", "getdiff"},
                   {"revision", std::to_string(m_storage->getRevision(id)) }
                   })};
                 ptree.put_child("serverinfo.diff", diff.get_structure().get_child("diff"));
                 boost::beast::ostream(buffer) << Reichwein::XML::plain_xml(ptree);
                 std::lock_guard<std::mutex> lock(m_websocket_mutex);
                 try {
                  ci->write(buffer.data());
                 } catch (const std::exception& ex) {
                  std::cerr << "Warning: Notify getdiff write for " << ci << " not possible, id " << id << std::endl;
                  m_registry.dump();
                 }
                }
               });
}

void Whiteboard::notify_other_connections_pos(Whiteboard::connection& c, const std::string& id)
{
 std::for_each(m_registry.begin(id), m_registry.end(id), [&](const Whiteboard::connection& ci)
               {
                if (c != ci) {
                 boost::beast::flat_buffer buffer;
                 boost::beast::ostream(buffer) << make_xml({
                   {"type", "getpos"},
                   {"pos", std::to_string(m_storage->getCursorPos(id)) }
                   });
                 std::lock_guard<std::mutex> lock(m_websocket_mutex);
                 try {
                  ci->write(buffer.data());
                 } catch (const std::exception& ex) {
                  std::cerr << "Warning: Notify getpos write for " << ci << " not possible, id " << id << std::endl;
                  m_registry.dump();
                 }
                }
               });
}

std::string Whiteboard::handle_request(Whiteboard::connection& c, const std::string& request)
{
 try {
  std::lock_guard<std::mutex> lock(m_storage_mutex);
  if (!m_storage)
   throw std::runtime_error("Storage not initialized");
  
  pt::ptree xml;
  std::istringstream ss{request};
  pt::xml_parser::read_xml(ss, xml);

  std::string command {xml.get<std::string>("request.command")};

  if (command == "modify") {
   std::string id {xml.get<std::string>("request.id")};
   
   int baserev {xml.get<int>("request.baserev")};
   if (baserev != m_storage->getRevision(id))
    return make_xml({{"type", "error"}, {"message", "Bad base revision ("s + std::to_string(baserev) + "). Current: "s + std::to_string(m_storage->getRevision(id)) }});

   pt::ptree ptree;
   ptree.put_child("diff", xml.get_child("request.diff"));
   Diff d{ptree};
   std::string data {m_storage->getDocument(id)};
   data = d.apply(data);

   m_storage->setDocument(id, data);
   m_registry.setId(c, id);
   notify_other_connections_diff(c, id, d);
   
   int pos {xml.get<int>("request.pos")};
   if (m_storage->getCursorPos(id) != pos) {
    m_storage->setCursorPos(id, pos);
    notify_other_connections_pos(c, id);
   }
   return make_xml({{"type", "modify"}, {"revision", std::to_string(m_storage->getRevision(id)) }});
  } else if (command == "cursorpos") {
   std::string id {xml.get<std::string>("request.id")};
   int pos {xml.get<int>("request.pos")};
   if (m_storage->getCursorPos(id) != pos) {
    m_storage->setCursorPos(id, pos);
    notify_other_connections_pos(c, id);
   }
   return {};
  } else if (command == "getfile") {
   std::string id {xml.get<std::string>("request.id")};

   std::string filedata;
   try {
    filedata = m_storage->getDocument(id);
   } catch (const std::runtime_error&) {
    m_storage->setDocument(id, filedata);
   }

   if (filedata.size() > 30000000)
    throw std::runtime_error("File too big");
   m_registry.setId(c, id);

   return make_xml({
                   {"type", "getfile"},
                   {"data", filedata},
                   {"revision", std::to_string(m_storage->getRevision(id)) },
                   {"pos", std::to_string(m_storage->getCursorPos(id)) }
                   }); 
  } else if (command == "getpos") {
   std::string id {xml.get<std::string>("request.id")};

   return make_xml({
                   {"type", "getpos"},
                   {"pos", std::to_string(m_storage->getCursorPos(id)) }
                   }); 
  } else if (command == "newid") {
   return make_xml({{"type", "newid"}, {"id", m_storage->generate_id()}}); 
  } else if (command == "qrcode") {
   std::string url{xml.get<std::string>("request.url")};
   
   if (url.size() > 1000)
    throw std::runtime_error("URL too big");

   std::string pngdata {QRCode::getQRCode(url)};

   return make_xml({{"type", "qrcode"}, {"png", Reichwein::Base64::encode64(pngdata)}});
  } else if (command == "getversion") {
   return make_xml({
                   {"type", "version"},
                   {"version", WHITEBOARD_VERSION }
                   }); 
  } else {
   throw std::runtime_error("Bad command: "s + command);
  }

 } catch (const std::exception& ex) {
  return make_xml({{"type", "error"}, {"message", "Message handling error: "s + ex.what()}}); 
 }
}

void Whiteboard::do_session(boost::asio::ip::tcp::socket socket)
{
 try {
  // Construct the stream by moving in the socket
  std::shared_ptr ws{std::make_shared<boost::beast::websocket::stream<boost::asio::ip::tcp::socket>>(std::move(socket))};
  ConnectionRegistry::RegistryGuard guard(m_registry, ws);

  // Set a decorator to change the Server of the handshake
  ws->set_option(boost::beast::websocket::stream_base::decorator(
      [](boost::beast::websocket::response_type& res)
      {
       res.set(boost::beast::http::field::server,
           std::string("Reichwein.IT Whiteboard"));
      }));

  boost::beast::http::request_parser<boost::beast::http::string_body> parser;
  boost::beast::http::request<boost::beast::http::string_body> req;
  boost::beast::flat_buffer buffer;

  boost::beast::http::read(ws->next_layer(), buffer, parser);
  req = parser.get();

  ws->accept(req);

  while (true) {
   boost::beast::flat_buffer buffer;

   ws->read(buffer);

   ws->text(ws->got_text());
   std::string data(boost::asio::buffers_begin(buffer.data()), boost::asio::buffers_end(buffer.data()));
   data = handle_request(ws, data);
   if (buffer.data().size() > 0) {
    buffer.consume(buffer.size());
   }
   if (data.size() > 0) {
    boost::beast::ostream(buffer) << data;
    std::lock_guard<std::mutex> lock(m_websocket_mutex);
    ws->write(buffer.data());
   }
  }
 } catch (boost::beast::system_error const& se) {
  // This indicates that the session was closed
  if (se.code() != boost::beast::websocket::error::closed && se.code() != boost::asio::error::eof)
   std::cerr << "Boost system_error in session: " << se.code().message() << std::endl;
 } catch (std::exception const& ex) {
  std::cerr << "Error in session: " << ex.what() << std::endl;
 }
}

// the actual main() for testability
int Whiteboard::run(int argc, char* argv[])
{
 try {
  bool flag_cleanup{};
  fs::path configFile;

  if (argc == 2) {
   if (argv[1] == "-h"s || argv[1] == "-?"s) {
    usage();
    exit(0);
   } else if (argv[1] == "-C"s) {
    flag_cleanup = true;
   }
  } else if (argc == 3) {
   if (argv[1] == "-c"s) {
    configFile = argv[2];
   }
  }

  if (configFile.empty())
   m_config = std::make_unique<Config>();
  else
   m_config = std::make_unique<Config>(configFile);

  m_storage = std::make_unique<Storage>(*m_config);

  if (flag_cleanup) {
   m_storage->cleanup();
   exit(0);
  }

  std::thread storage_cleanup_thread(std::bind(&Whiteboard::storage_cleanup, this));

  QRCode::init();

  auto const address = boost::asio::ip::make_address(m_config->getListenAddress());
  auto const port = static_cast<unsigned short>(m_config->getListenPort());

  // The io_context is required for all I/O
  boost::asio::io_context ioc{m_config->getThreads()};

  // The acceptor receives incoming connections
  boost::asio::ip::tcp::acceptor acceptor{ioc, {address, port}};
  while (true) {
   // This will receive the new connection
   boost::asio::ip::tcp::socket socket{ioc};

   // Block until we get a connection
   acceptor.accept(socket);

   // Launch the session, transferring ownership of the socket
   std::thread(
       &Whiteboard::do_session, this,
       std::move(socket)).detach();
  }

  storage_cleanup_thread.join();
 } catch (const std::exception& ex) {
  std::cerr << "Error: " << ex.what() << std::endl;
 }

 return 0;
}