text: ////////////////////////////////////////////////////////////////////////////
  //
  // Copyright 2016 Realm Inc.
  //
  // Licensed under the Apache License, Version 2.0 (the "License");
  // you may not use this file except in compliance with the License.
  // You may obtain a copy of the License at
  //
  // http://www.apache.org/licenses/LICENSE-2.0
  //
  // Unless required by applicable law or agreed to in writing, software
  // distributed under the License is distributed on an "AS IS" BASIS,
  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  // See the License for the specific language governing permissions and
  // limitations under the License.
  //
  ////////////////////////////////////////////////////////////////////////////
  
  #pragma once
  
  #include "js_class.hpp"
  #include "js_collection.hpp"
  #include "platform.hpp"
  #include "sync/partial_sync.hpp"
  #include "sync/sync_config.hpp"
  #include "sync/sync_manager.hpp"
  #include "sync/sync_session.hpp"
  #include "sync/sync_user.hpp"
  #include "util/event_loop_dispatcher.hpp"
  
  #include <realm/util/logger.hpp>
  #include <realm/util/uri.hpp>
  
  #if REALM_PLATFORM_NODE
  #include "impl/realm_coordinator.hpp"
  #include "node/js_global_notifier.hpp"
  #include "node/sync_logger.hpp"
  #endif
  
  #if REALM_ANDROID
  #include <jni.h>
  #include "./android/io_realm_react_RealmReactModule.h"
  #include "./android/jni_utils.hpp"
  
  extern jclass ssl_helper_class;
  #endif
  
  #include <mutex>
  #include <condition_variable>
  
  namespace realm {
  class Adapter;
  
  namespace js {
  template<typename T>
  inline realm::SyncManager& syncManagerShared(typename T::Context &ctx) {
      static std::once_flag flag;
      std::call_once(flag, [ctx] {
          auto realm_constructor = js::Value<T>::validated_to_object(ctx, js::Object<T>::get_global(ctx, "Realm"));
          std::string user_agent_binding_info;
          auto user_agent_function = js::Object<T>::get_property(ctx, realm_constructor, "_createUserAgentDescription");
          if (js::Value<T>::is_function(ctx, user_agent_function)) {
              auto result = js::Function<T>::call(ctx, js::Value<T>::to_function(ctx, user_agent_function), realm_constructor, 0, nullptr);
              user_agent_binding_info = js::Value<T>::validated_to_string(ctx, result);
          }
          ensure_directory_exists_for_file(default_realm_file_directory());
          SyncManager::shared().configure(default_realm_file_directory(), SyncManager::MetadataMode::NoEncryption, user_agent_binding_info);
      });
      return SyncManager::shared();
  }
  
  template<typename T>
  class AdapterClass : public ClassDefinition<T, Adapter> {
      using GlobalContextType = typename T::GlobalContext;
      using ContextType = typename T::Context;
      using FunctionType = typename T::Function;
      using ObjectType = typename T::Object;
      using ValueType = typename T::Value;
      using String = js::String<T>;
      using Object = js::Object<T>;
      using Value = js::Value<T>;
      using Function = js::Function<T>;
      using ReturnValue = js::ReturnValue<T>;
      using Arguments = js::Arguments<T>;
  
  
  public:
      std::string const name = "Adapter";
  
      static void constructor(ContextType, ObjectType, Arguments &);
  
      static void current(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void advance(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void realm_at_path(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void close(ContextType, ObjectType, Arguments &, ReturnValue &);
  
      MethodMap<T> const methods = {
          {"current", wrap<current>},
          {"advance", wrap<advance>},
          {"realmAtPath", wrap<realm_at_path>},
          {"close", wrap<close>},
      };
  };
  
  using SharedUser = std::shared_ptr<realm::SyncUser>;
  using WeakSession = std::weak_ptr<realm::SyncSession>;
  
  template<typename T>
  class UserClass : public ClassDefinition<T, SharedUser> {
      using GlobalContextType = typename T::GlobalContext;
      using ContextType = typename T::Context;
      using FunctionType = typename T::Function;
      using ObjectType = typename T::Object;
      using ValueType = typename T::Value;
      using String = js::String<T>;
      using Object = js::Object<T>;
      using Value = js::Value<T>;
      using Function = js::Function<T>;
      using ReturnValue = js::ReturnValue<T>;
      using Arguments = js::Arguments<T>;
  
  public:
      std::string const name = "User";
  
      static FunctionType create_constructor(ContextType);
  
      static void get_server(ContextType, ObjectType, ReturnValue &);
      static void get_identity(ContextType, ObjectType, ReturnValue &);
      static void get_token(ContextType, ObjectType, ReturnValue &);
      static void is_admin(ContextType, ObjectType, ReturnValue &);
      static void is_admin_token(ContextType, ObjectType, ReturnValue &);
  
      PropertyMap<T> const properties = {
          {"server", {wrap<get_server>, nullptr}},
          {"identity", {wrap<get_identity>, nullptr}},
          {"token", {wrap<get_token>, nullptr}},
          {"isAdmin", {wrap<is_admin>, nullptr}},
          {"isAdminToken", {wrap<is_admin_token>, nullptr}},
      };
  
      static void create_user(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void admin_user(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void get_existing_user(ContextType, ObjectType, Arguments &, ReturnValue&);
  
      MethodMap<T> const static_methods = {
          {"createUser", wrap<create_user>},
          {"_adminUser", wrap<admin_user>},
          {"_getExistingUser", wrap<get_existing_user>},
      };
  
      static void all_users(ContextType ctx, ObjectType object, ReturnValue &);
  
      PropertyMap<T> const static_properties = {
          {"all", {wrap<all_users>, nullptr}},
      };
  
      static void logout(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void session_for_on_disk_path(ContextType, ObjectType, Arguments &, ReturnValue &);
  
      MethodMap<T> const methods = {
          {"_logout", wrap<logout>},
          {"_sessionForOnDiskPath", wrap<session_for_on_disk_path>}
      };
  };
  
  template<typename T>
  void UserClass<T>::get_server(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      std::string server = get_internal<T, UserClass<T>>(object)->get()->server_url();
      return_value.set(server);
  }
  
  template<typename T>
  void UserClass<T>::get_identity(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      std::string identity = get_internal<T, UserClass<T>>(object)->get()->identity();
      return_value.set(identity);
  }
  
  template<typename T>
  void UserClass<T>::get_token(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      std::string token = get_internal<T, UserClass<T>>(object)->get()->refresh_token();
      return_value.set(token);
  }
  
  template<typename T>
  void UserClass<T>::is_admin(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      return_value.set(get_internal<T, UserClass<T>>(object)->get()->is_admin());
  }
  
  template<typename T>
  void UserClass<T>::is_admin_token(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      return_value.set(get_internal<T, UserClass<T>>(object)->get()->token_type() == SyncUser::TokenType::Admin);
  }
  
  template<typename T>
  void UserClass<T>::create_user(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_between(3, 5);
      SyncUserIdentifier userIdentifier {
          Value::validated_to_string(ctx, args[1], "identity"),
          Value::validated_to_string(ctx, args[0], "authServerUrl")
       };
      SharedUser *user = new SharedUser(syncManagerShared<T>(ctx).get_user(
          userIdentifier,
          Value::validated_to_string(ctx, args[2], "refreshToken")
      ));
  
      if (args.count == 5) {
          (*user)->set_is_admin(Value::validated_to_boolean(ctx, args[4], "isAdmin"));
      }
      return_value.set(create_object<T, UserClass<T>>(ctx, user));
  }
  
  template<typename T>
  void UserClass<T>::admin_user(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(2);
      SharedUser *user = new SharedUser(syncManagerShared<T>(ctx).get_admin_token_user(
          Value::validated_to_string(ctx, args[0], "authServerUrl"),
          Value::validated_to_string(ctx, args[1], "refreshToken")
      ));
      return_value.set(create_object<T, UserClass<T>>(ctx, user));
  }
  
  template<typename T>
  void UserClass<T>::get_existing_user(ContextType ctx, ObjectType, Arguments &args, ReturnValue &return_value) {
      args.validate_count(2);
      if (auto user = syncManagerShared<T>(ctx).get_existing_logged_in_user(SyncUserIdentifier{
              Value::validated_to_string(ctx, args[1], "identity"),
              Value::validated_to_string(ctx, args[0], "authServerUrl")})) {
          return_value.set(create_object<T, UserClass<T>>(ctx, new SharedUser(std::move(user))));
      }
  }
  
  template<typename T>
  void UserClass<T>::all_users(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      auto users = Object::create_empty(ctx);
      for (auto user : syncManagerShared<T>(ctx).all_logged_in_users()) {
          if (user->token_type() == SyncUser::TokenType::Normal) {
              Object::set_property(ctx, users, user->identity(), create_object<T, UserClass<T>>(ctx, new SharedUser(user)), ReadOnly | DontDelete);
          }
      }
      return_value.set(users);
  }
  
  template<typename T>
  void UserClass<T>::logout(ContextType, ObjectType this_object, Arguments &, ReturnValue &) {
      get_internal<T, UserClass<T>>(this_object)->get()->log_out();
  }
  
  template<typename T>
  class SessionClass : public ClassDefinition<T, WeakSession> {
      using ContextType = typename T::Context;
      using FunctionType = typename T::Function;
      using ObjectType = typename T::Object;
      using ValueType = typename T::Value;
      using String = js::String<T>;
      using Object = js::Object<T>;
      using Value = js::Value<T>;
      using ReturnValue = js::ReturnValue<T>;
      using Arguments = js::Arguments<T>;
  
  public:
      std::string const name = "Session";
      using ProgressHandler = void(uint64_t transferred_bytes, uint64_t transferrable_bytes);
      using StateHandler = void(SyncSession::PublicState old_state, SyncSession::PublicState new_state);
      using ConnectionHandler = void(SyncSession::ConnectionState new_state, SyncSession::ConnectionState old_state);
      using DownloadUploadCompletionHandler = void(std::error_code error);
  
      static FunctionType create_constructor(ContextType);
  
      static void get_config(ContextType, ObjectType, ReturnValue &);
      static void get_user(ContextType, ObjectType, ReturnValue &);
      static void get_url(ContextType, ObjectType, ReturnValue &);
      static void get_state(ContextType, ObjectType, ReturnValue &);
      static void get_connection_state(ContextType, ObjectType, ReturnValue &);
  
      static void simulate_error(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void refresh_access_token(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void add_progress_notification(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void remove_progress_notification(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void add_state_notification(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void remove_state_notification(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void add_connection_notification(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void remove_connection_notification(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void is_connected(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void resume(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void pause(ContextType ctx, ObjectType this_object, Arguments &, ReturnValue &);
      static void override_server(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void wait_for_download_completion(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void wait_for_upload_completion(ContextType, ObjectType, Arguments &, ReturnValue &);
  
  
      PropertyMap<T> const properties = {
          {"config", {wrap<get_config>, nullptr}},
          {"user", {wrap<get_user>, nullptr}},
          {"url", {wrap<get_url>, nullptr}},
          {"state", {wrap<get_state>, nullptr}},
          {"connectionState", {wrap<get_connection_state>, nullptr}},
      };
  
      MethodMap<T> const methods = {
          {"_simulateError", wrap<simulate_error>},
          {"_refreshAccessToken", wrap<refresh_access_token>},
          {"_overrideServer", wrap<override_server>},
          {"_waitForDownloadCompletion", wrap<wait_for_download_completion>},
          {"_waitForUploadCompletion", wrap<wait_for_upload_completion>},
          {"addProgressNotification", wrap<add_progress_notification>},
          {"removeProgressNotification", wrap<remove_progress_notification>},
          {"addConnectionNotification", wrap<add_connection_notification>},
          {"removeConnectionNotification", wrap<remove_connection_notification>},
          {"isConnected", wrap<is_connected>},
          {"resume", wrap<resume>},
          {"pause", wrap<pause>},
      };
  
  private:
      enum Direction { Upload, Download };
      static std::string get_connection_state_value(SyncSession::ConnectionState state);
      static void wait_for_completion(Direction direction, ContextType ctx, ObjectType this_object, Arguments& args);
  };
  
  template<typename T>
  class SyncSessionErrorHandlerFunctor {
  public:
      SyncSessionErrorHandlerFunctor(typename T::Context ctx, typename T::Function error_func)
      : m_ctx(Context<T>::get_global_context(ctx))
      , m_func(ctx, error_func)
      { }
  
      typename T::Function func() const { return m_func; }
  
      void operator()(std::shared_ptr<SyncSession> session, SyncError error) {
          HANDLESCOPE
  
          std::string name = "Error";
          auto error_object = Object<T>::create_empty(m_ctx);
  
          if (error.is_client_reset_requested()) {
              auto config_object = Object<T>::create_empty(m_ctx);
              Object<T>::set_property(m_ctx, config_object, "path", Value<T>::from_string(m_ctx, error.user_info[SyncError::c_recovery_file_path_key]));
              Object<T>::set_property(m_ctx, config_object, "readOnly", Value<T>::from_boolean(m_ctx, true));
              Object<T>::set_property(m_ctx, error_object, "config", config_object);
              name = "ClientReset";
          }
  
          Object<T>::set_property(m_ctx, error_object, "name", Value<T>::from_string(m_ctx, name));
          Object<T>::set_property(m_ctx, error_object, "message", Value<T>::from_string(m_ctx, error.message));
          Object<T>::set_property(m_ctx, error_object, "isFatal", Value<T>::from_boolean(m_ctx, error.is_fatal));
          Object<T>::set_property(m_ctx, error_object, "category", Value<T>::from_string(m_ctx, error.error_code.category().name()));
          Object<T>::set_property(m_ctx, error_object, "code", Value<T>::from_number(m_ctx, error.error_code.value()));
  
          auto user_info = Object<T>::create_empty(m_ctx);
          for (auto& kvp : error.user_info) {
              Object<T>::set_property(m_ctx, user_info, kvp.first, Value<T>::from_string(m_ctx, kvp.second));
          }
          Object<T>::set_property(m_ctx, error_object, "userInfo", user_info);
  
          typename T::Value arguments[2];
          arguments[0] = create_object<T, SessionClass<T>>(m_ctx, new WeakSession(session));
          arguments[1] = error_object;
  
          Function<T>::callback(m_ctx, m_func, typename T::Object(), 2, arguments);
      }
  private:
      const Protected<typename T::GlobalContext> m_ctx;
      const Protected<typename T::Function> m_func;
  };
  
  
  // An object of type SSLVerifyCallbackSyncThreadFunctor is registered with the sync client in order
  // to verify SSL certificates. The SSLVerifyCallbackSyncThreadFunctor object's operator() is called
  // on the sync client's event loop thread.
  template <typename T>
  class SSLVerifyCallbackSyncThreadFunctor {
  public:
      SSLVerifyCallbackSyncThreadFunctor(typename T::Context ctx, typename T::Function ssl_verify_func)
      : m_ctx(Context<T>::get_global_context(ctx))
      , m_func(ctx, ssl_verify_func)
      , m_event_loop_dispatcher {SSLVerifyCallbackSyncThreadFunctor<T>::main_loop_handler}
      , m_mutex{new std::mutex}
      , m_cond_var{new std::condition_variable}
      {
      }
  
      // This function is called on the sync client's event loop thread.
      bool operator ()(const std::string& server_address, sync::Session::port_type server_port, const char* pem_data, size_t pem_size, int preverify_ok, int depth)
      {
          const std::string pem_certificate {pem_data, pem_size};
          {
              std::lock_guard<std::mutex> lock {*m_mutex};
              m_ssl_certificate_callback_done = false;
          }
  
          // Dispatch the call to the main_loop_handler on the node.js thread.
          m_event_loop_dispatcher(this, server_address, server_port, pem_certificate, preverify_ok, depth);
  
          bool ssl_certificate_accepted = false;
          {
              // Wait for the return value of the callback function on the node.js main thread.
              // The sync client blocks during this wait.
              std::unique_lock<std::mutex> lock(*m_mutex);
              m_cond_var->wait(lock, [this] { return this->m_ssl_certificate_callback_done; });
              ssl_certificate_accepted = m_ssl_certificate_accepted;
          }
  
          return ssl_certificate_accepted;
      }
  
      // main_loop_handler is called on the node.js main thread.
      // main_loop_handler calls the user callback (m_func) and sends the return value
      // back to the sync client's event loop thread through a condition variable.
      static void main_loop_handler(SSLVerifyCallbackSyncThreadFunctor<T>* this_object,
                                    const std::string& server_address,
                                    sync::Session::port_type server_port,
                                    const std::string& pem_certificate,
                                    int preverify_ok,
                                    int depth)
      {
          HANDLESCOPE
  
          const Protected<typename T::GlobalContext>& ctx = this_object->m_ctx;
  
          typename T::Object ssl_certificate_object = Object<T>::create_empty(ctx);
          Object<T>::set_property(ctx, ssl_certificate_object, "serverAddress", Value<T>::from_string(ctx, server_address));
          Object<T>::set_property(ctx, ssl_certificate_object, "serverPort", Value<T>::from_number(ctx, double(server_port)));
          Object<T>::set_property(ctx, ssl_certificate_object, "pemCertificate", Value<T>::from_string(ctx, pem_certificate));
          Object<T>::set_property(ctx, ssl_certificate_object, "acceptedByOpenSSL", Value<T>::from_boolean(ctx, preverify_ok != 0));
          Object<T>::set_property(ctx, ssl_certificate_object, "depth", Value<T>::from_number(ctx, double(depth)));
  
          const int argc = 1;
          typename T::Value arguments[argc] = { ssl_certificate_object };
          typename T::Value ret_val = Function<T>::callback(ctx, this_object->m_func, typename T::Object(), 1, arguments);
          bool ret_val_bool = Value<T>::to_boolean(ctx, ret_val);
  
          {
              std::lock_guard<std::mutex> lock {*this_object->m_mutex};
              this_object->m_ssl_certificate_callback_done = true;
              this_object->m_ssl_certificate_accepted = ret_val_bool;
          }
  
          this_object->m_cond_var->notify_one();
      };
  
  
  private:
      const Protected<typename T::GlobalContext> m_ctx;
      const Protected<typename T::Function> m_func;
      util::EventLoopDispatcher<void(SSLVerifyCallbackSyncThreadFunctor<T>* this_object,
                                     const std::string& server_address,
                                     sync::Session::port_type server_port,
                                     const std::string& pem_certificate,
                                     int preverify_ok,
                                     int depth)> m_event_loop_dispatcher;
      bool m_ssl_certificate_callback_done = false;
      bool m_ssl_certificate_accepted = false;
      std::shared_ptr<std::mutex> m_mutex;
      std::shared_ptr<std::condition_variable> m_cond_var;
  };
  
  template<typename T>
  void UserClass<T>::session_for_on_disk_path(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(1);
      auto user = *get_internal<T, UserClass<T>>(this_object);
      if (auto session = user->session_for_on_disk_path(Value::validated_to_string(ctx, args[0]))) {
          return_value.set(create_object<T, SessionClass<T>>(ctx, new WeakSession(session)));
      } else {
          return_value.set_undefined();
      }
  }
  
  template<typename T>
  void SessionClass<T>::get_config(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
          ObjectType config = Object::create_empty(ctx);
          Object::set_property(ctx, config, "user", create_object<T, UserClass<T>>(ctx, new SharedUser(session->config().user)));
          Object::set_property(ctx, config, "url", Value::from_string(ctx, session->config().realm_url()));
          if (auto dispatcher = session->config().error_handler.template target<util::EventLoopDispatcher<SyncSessionErrorHandler>>()) {
              if (auto handler = dispatcher->func().template target<SyncSessionErrorHandlerFunctor<T>>()) {
                  Object::set_property(ctx, config, "error", handler->func());
              }
          }
          if (!session->config().custom_http_headers.empty()) {
              ObjectType custom_http_headers_object = Object::create_empty(ctx);
              for (auto it = session->config().custom_http_headers.begin(); it != session->config().custom_http_headers.end(); ++it) {
                  Object::set_property(ctx, custom_http_headers_object, it->first, Value::from_string(ctx, it->second));
              }
              Object::set_property(ctx, config, "custom_http_headers", custom_http_headers_object);
          }
          return_value.set(config);
      } else {
          return_value.set_undefined();
      }
  }
  
  template<typename T>
  void SessionClass<T>::get_user(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
          return_value.set(create_object<T, UserClass<T>>(ctx, new SharedUser(session->config().user)));
      } else {
          return_value.set_undefined();
      }
  }
  
  template<typename T>
  void SessionClass<T>::get_url(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
          if (util::Optional<std::string> url = session->full_realm_url()) {
              return_value.set(*url);
              return;
          }
      }
  
      return_value.set_undefined();
  }
  
  template<typename T>
  void SessionClass<T>::get_state(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      static const std::string invalid("invalid");
      static const std::string inactive("inactive");
      static const std::string active("active");
  
      return_value.set(invalid);
  
      if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
          if (session->state() == SyncSession::PublicState::Inactive) {
              return_value.set(inactive);
          } else {
              return_value.set(active);
          }
      }
  }
  
  template<typename T>
  std::string SessionClass<T>::get_connection_state_value(SyncSession::ConnectionState state) {
      switch(state) {
          case SyncSession::ConnectionState::Disconnected: return "disconnected";
          case SyncSession::ConnectionState::Connecting: return "connecting";
          case SyncSession::ConnectionState::Connected: return "connected";
      }
  }
  
  template<typename T>
  void SessionClass<T>::get_connection_state(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      return_value.set(get_connection_state_value(SyncSession::ConnectionState::Disconnected));
      if (auto session = get_internal<T, SessionClass<T>>(object)->lock()) {
          return_value.set(get_connection_state_value(session->connection_state()));
      }
  }
  
  template<typename T>
  void SessionClass<T>::simulate_error(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &) {
      args.validate_count(2);
  
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          std::error_code error_code(Value::validated_to_number(ctx, args[0]), realm::sync::protocol_error_category());
          std::string message = Value::validated_to_string(ctx, args[1]);
          SyncSession::OnlyForTesting::handle_error(*session, SyncError(error_code, message, false));
      }
  }
  
  template<typename T>
  void SessionClass<T>::refresh_access_token(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &) {
      args.validate_count(3);
  
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          std::string sync_label = Value::validated_to_string(ctx, args[2], "syncLabel");
          session->set_multiplex_identifier(std::move(sync_label));
  
          std::string access_token = Value::validated_to_string(ctx, args[0], "accessToken");
          std::string realm_url = Value::validated_to_string(ctx, args[1], "realmUrl");
          session->refresh_access_token(std::move(access_token), std::move(realm_url));
      }
  }
  
  template<typename T>
  void SessionClass<T>::add_progress_notification(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(3);
  
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
  
          std::string direction = Value::validated_to_string(ctx, args[0], "direction");
          std::string mode = Value::validated_to_string(ctx, args[1], "mode");
          SyncSession::NotifierType notifierType;
          if (direction == "download") {
              notifierType = SyncSession::NotifierType::download;
          }
          else if (direction == "upload") {
              notifierType = SyncSession::NotifierType::upload;
          }
          else {
              throw std::invalid_argument("Invalid argument 'direction'. Only 'download' and 'upload' progress notification directions are supported");
          }
  
          bool is_streaming = false;
          if (mode == "reportIndefinitely") {
              is_streaming = true;
          }
          else if (mode == "forCurrentlyOutstandingWork") {
              is_streaming = false;
          }
          else {
              throw std::invalid_argument("Invalid argument 'mode'. Only 'reportIndefinitely' and 'forCurrentlyOutstandingWork' progress notification modes are supported");
          }
  
          auto callback_function = Value::validated_to_function(ctx, args[2], "callback");
  
          Protected<FunctionType> protected_callback(ctx, callback_function);
          Protected<ObjectType> protected_this(ctx, this_object);
          Protected<typename T::GlobalContext> protected_ctx(Context<T>::get_global_context(ctx));
          std::function<ProgressHandler> progressFunc;
  
          util::EventLoopDispatcher<ProgressHandler> progress_handler([=](uint64_t transferred_bytes, uint64_t transferrable_bytes) {
              HANDLESCOPE
              ValueType callback_arguments[2];
              callback_arguments[0] = Value::from_number(protected_ctx, transferred_bytes);
              callback_arguments[1] = Value::from_number(protected_ctx, transferrable_bytes);
  
              Function<T>::callback(protected_ctx, protected_callback, typename T::Object(), 2, callback_arguments);
          });
  
          progressFunc = std::move(progress_handler);
  
          auto registrationToken = session->register_progress_notifier(std::move(progressFunc), notifierType, is_streaming);
          auto syncSession = create_object<T, SessionClass<T>>(ctx, new WeakSession(session));
          PropertyAttributes attributes = ReadOnly | DontEnum | DontDelete;
          Object::set_property(ctx, callback_function, "_syncSession", syncSession, attributes);
          Object::set_property(ctx, callback_function, "_registrationToken", Value::from_number(protected_ctx, registrationToken), attributes);
      }
  }
  
  template<typename T>
  void SessionClass<T>::remove_progress_notification(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(1);
      auto callback_function = Value::validated_to_function(ctx, args[0], "callback");
      auto syncSessionProp = Object::get_property(ctx, callback_function, "_syncSession");
      if (Value::is_undefined(ctx, syncSessionProp) || Value::is_null(ctx, syncSessionProp)) {
          return;
      }
  
      auto syncSession = Value::validated_to_object(ctx, syncSessionProp);
      auto registrationToken = Object::get_property(ctx, callback_function, "_registrationToken");
  
      if (auto session = get_internal<T, SessionClass<T>>(syncSession)->lock()) {
          auto reg = Value::validated_to_number(ctx, registrationToken);
          session->unregister_progress_notifier(reg);
      }
  }
  
  template<typename T>
  void SessionClass<T>::add_connection_notification(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      args.validate_count(1);
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          auto callback_function = Value::validated_to_function(ctx, args[0], "callback");
          Protected<FunctionType> protected_callback(ctx, callback_function);
          Protected<ObjectType> protected_this(ctx, this_object);
          Protected<typename T::GlobalContext> protected_ctx(Context<T>::get_global_context(ctx));
  
          std::function<ConnectionHandler> connectionFunc;
  
          util::EventLoopDispatcher<ConnectionHandler> connection_handler([=](SyncSession::ConnectionState old_state, SyncSession::ConnectionState new_state) {
              HANDLESCOPE
              ValueType callback_arguments[2];
              callback_arguments[0] = Value::from_string(protected_ctx, get_connection_state_value(new_state));
              callback_arguments[1] = Value::from_string(protected_ctx, get_connection_state_value(old_state));
              Function<T>::callback(protected_ctx, protected_callback, typename T::Object(), 2, callback_arguments);
          });
  
          connectionFunc = std::move(connection_handler);
  
          auto notificationToken = session->register_connection_change_callback(std::move(connectionFunc));
          auto syncSession = create_object<T, SessionClass<T>>(ctx, new WeakSession(session));
          PropertyAttributes attributes = ReadOnly | DontEnum | DontDelete;
          Object::set_property(ctx, callback_function, "_syncSession", syncSession, attributes);
          Object::set_property(ctx, callback_function, "_connectionNotificationToken", Value::from_number(protected_ctx, notificationToken), attributes);
      }
  }
  
  template<typename T>
  void SessionClass<T>::remove_connection_notification(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      args.validate_count(1);
      auto callback_function = Value::validated_to_function(ctx, args[0], "callback");
      auto syncSessionProp = Object::get_property(ctx, callback_function, "_syncSession");
      if (Value::is_undefined(ctx, syncSessionProp) || Value::is_null(ctx, syncSessionProp)) {
          return;
      }
      auto syncSession = Value::validated_to_object(ctx, syncSessionProp);
      auto registrationToken = Object::get_property(ctx, callback_function, "_connectionNotificationToken");
  
      if (auto session = get_internal<T, SessionClass<T>>(syncSession)->lock()) {
          auto reg = Value::validated_to_number(ctx, registrationToken);
          session->unregister_connection_change_callback(reg);
      }
  }
  
  template<typename T>
  void SessionClass<T>::is_connected(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      args.validate_count(0);
      return_value.set(false);
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          auto state = session->state();
          auto connection_state = session->connection_state();
          if (connection_state == SyncSession::ConnectionState::Connected
              && (state == SyncSession::PublicState::Active || state == SyncSession::PublicState::Dying)) {
              return_value.set(true);
          }
      }
  }
  
  template<typename T>
  void SessionClass<T>::resume(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      args.validate_count(0);
      return_value.set(false);
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          session->revive_if_needed();
      }
  }
  
  template<typename T>
  void SessionClass<T>::pause(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      args.validate_count(0);
      return_value.set(false);
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          session->log_out();
      }
  }
  
  template<typename T>
  void SessionClass<T>::override_server(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue&) {
      args.validate_count(2);
  
      std::string address = Value::validated_to_string(ctx, args[0], "address");
      double port = Value::validated_to_number(ctx, args[1], "port");
      if (port < 1 || port > 65535 || uint16_t(port) != port) {
          std::ostringstream message;
          message << "Invalid port number. Expected an integer in the range 1-65,535, got '" << port << "'";
          throw std::invalid_argument(message.str());
      }
  
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          session->override_server(std::move(address), uint16_t(port));
      }
  }
  
  template<typename T>
  void SessionClass<T>::wait_for_completion(Direction direction, ContextType ctx, ObjectType this_object, Arguments &args) {
      args.validate_count(1);
      if (auto session = get_internal<T, SessionClass<T>>(this_object)->lock()) {
          auto callback_function = Value::validated_to_function(ctx, args[0]);
          Protected<FunctionType> protected_callback(ctx, callback_function);
          Protected<ObjectType> protected_this(ctx, this_object);
          Protected<typename T::GlobalContext> protected_ctx(Context<T>::get_global_context(ctx));
  
          util::EventLoopDispatcher<DownloadUploadCompletionHandler> completion_handler([=](std::error_code error) {
              HANDLESCOPE
              ValueType callback_arguments[1];
              if (error) {
                  ObjectType error_object = Object::create_empty(protected_ctx);
                  Object::set_property(protected_ctx, error_object, "message", Value::from_string(protected_ctx, error.message()));
                  Object::set_property(protected_ctx, error_object, "errorCode", Value::from_number(protected_ctx, error.value()));
                  callback_arguments[0] = error_object;
              } else {
                  callback_arguments[0] = Value::from_undefined(protected_ctx);
              }
              Function<T>::callback(protected_ctx, protected_callback, typename T::Object(), 1, callback_arguments);
          });
  
          bool callback_registered;
          switch(direction) {
              case Upload:
                  callback_registered = session->wait_for_upload_completion(std::move(completion_handler));
                  break;
              case Download:
                  callback_registered = session->wait_for_download_completion(std::move(completion_handler));
                  break;
          }
          if (!callback_registered) {
              throw new logic_error("Could not register upload/download completion handler");
          }
          auto syncSession = create_object<T, SessionClass<T>>(ctx, new WeakSession(session));
          PropertyAttributes attributes = ReadOnly | DontEnum | DontDelete;
          Object::set_property(ctx, callback_function, "_syncSession", syncSession, attributes);
      }
  }
  
  template<typename T>
  void SessionClass<T>::wait_for_upload_completion(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue&) {
      wait_for_completion(Direction::Upload, ctx, this_object, args);
  }
  
  template<typename T>
  void SessionClass<T>::wait_for_download_completion(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue&) {
      wait_for_completion(Direction::Download, ctx, this_object, args);
  }
  
  template<typename T>
  class Subscription : public partial_sync::Subscription {
  public:
      Subscription(partial_sync::Subscription s, util::Optional<std::string> name) : partial_sync::Subscription(std::move(s)), m_name(name) {}
      Subscription(Subscription &&) = default;
  
      util::Optional<std::string> m_name;
      std::vector<std::pair<Protected<typename T::Function>, partial_sync::SubscriptionNotificationToken>> m_notification_tokens;
  };
  
  template<typename T>
  class SubscriptionClass : public ClassDefinition<T, Subscription<T>> {
      using GlobalContextType = typename T::GlobalContext;
      using ContextType = typename T::Context;
      using FunctionType = typename T::Function;
      using ObjectType = typename T::Object;
      using ValueType = typename T::Value;
      using String = js::String<T>;
      using Object = js::Object<T>;
      using Value = js::Value<T>;
      using Function = js::Function<T>;
      using ReturnValue = js::ReturnValue<T>;
      using Arguments = js::Arguments<T>;
  
  public:
      std::string const name = "Subscription";
  
      static FunctionType create_constructor(ContextType);
      static ObjectType create_instance(ContextType, partial_sync::Subscription, util::Optional<std::string>);
  
      static void get_state(ContextType, ObjectType, ReturnValue &);
      static void get_error(ContextType, ObjectType, ReturnValue &);
      static void get_name(ContextType, ObjectType, ReturnValue &);
  
      static void unsubscribe(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void add_listener(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void remove_listener(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void remove_all_listeners(ContextType, ObjectType, Arguments &, ReturnValue &);
  
      PropertyMap<T> const properties = {
          {"state", {wrap<get_state>, nullptr}},
          {"error", {wrap<get_error>, nullptr}},
          {"name",  {wrap<get_name>, nullptr}},
      };
  
      MethodMap<T> const methods = {
          {"unsubscribe", wrap<unsubscribe>},
          {"addListener", wrap<add_listener>},
          {"removeListener", wrap<remove_listener>},
          {"removeAllListeners", wrap<remove_all_listeners>},
      };
  };
  
  template<typename T>
  typename T::Object SubscriptionClass<T>::create_instance(ContextType ctx, partial_sync::Subscription subscription, util::Optional<std::string> name) {
      return create_object<T, SubscriptionClass<T>>(ctx, new Subscription<T>(std::move(subscription), name));
  }
  
  template<typename T>
  void SubscriptionClass<T>::get_state(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      auto subscription = get_internal<T, SubscriptionClass<T>>(object);
      return_value.set(static_cast<int8_t>(subscription->state()));
  }
  
  template<typename T>
  void SubscriptionClass<T>::get_error(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      auto subscription = get_internal<T, SubscriptionClass<T>>(object);
      if (auto error = subscription->error()) {
          try {
              std::rethrow_exception(error);
          }
          catch (const std::exception& e) {
              return_value.set(e.what());
          }
      }
      else {
          return_value.set_undefined();
      }
  }
  
  template<typename T>
  void SubscriptionClass<T>::get_name(ContextType ctx, ObjectType object, ReturnValue &return_value) {
      auto subscription = get_internal<T, SubscriptionClass<T>>(object);
      if (subscription->m_name == util::none) {
          // FIXME: should we reconstruct the name to match the one stored in __ResultSets?
          return_value.set_undefined();
      }
      else {
          return_value.set(subscription->m_name);
      }
  }
  
  template<typename T>
  void SubscriptionClass<T>::unsubscribe(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_maximum(0);
      auto subscription = get_internal<T, SubscriptionClass<T>>(this_object);
      partial_sync::unsubscribe(*subscription);
      return_value.set_undefined();
  }
  
  template<typename T>
  void SubscriptionClass<T>::add_listener(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_maximum(1);
      auto subscription = get_internal<T, SubscriptionClass<T>>(this_object);
  
      auto callback = Value::validated_to_function(ctx, args[0]);
      Protected<FunctionType> protected_callback(ctx, callback);
      Protected<ObjectType> protected_this(ctx, this_object);
      Protected<typename T::GlobalContext> protected_ctx(Context<T>::get_global_context(ctx));
  
      auto token = subscription->add_notification_callback([=]() {
          HANDLESCOPE
  
          ValueType arguments[2];
          arguments[0] = static_cast<ObjectType>(protected_this),
          arguments[1] = Value::from_number(protected_ctx, static_cast<double>(subscription->state()));
          Function::callback(protected_ctx, protected_callback, protected_this, 2, arguments);
      });
  
      subscription->m_notification_tokens.emplace_back(protected_callback, std::move(token));
  }
  
  template<typename T>
  void SubscriptionClass<T>::remove_listener(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_maximum(1);
      auto subscription = get_internal<T, SubscriptionClass<T>>(this_object);
  
      auto callback = Value::validated_to_function(ctx, args[0]);
      auto protected_function = Protected<FunctionType>(ctx, callback);
  
      auto& tokens = subscription->m_notification_tokens;
      auto compare = [&](auto&& token) {
          return typename Protected<FunctionType>::Comparator()(token.first, protected_function);
      };
      tokens.erase(std::remove_if(tokens.begin(), tokens.end(), compare), tokens.end());
  }
  
  template<typename T>
  void SubscriptionClass<T>::remove_all_listeners(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_maximum(0);
      auto subscription = get_internal<T, SubscriptionClass<T>>(this_object);
      subscription->m_notification_tokens.clear();
  }
  
  template<typename T>
  class SyncClass : public ClassDefinition<T, void*> {
      using GlobalContextType = typename T::GlobalContext;
      using ContextType = typename T::Context;
      using FunctionType = typename T::Function;
      using ObjectType = typename T::Object;
      using ValueType = typename T::Value;
      using String = js::String<T>;
      using Object = js::Object<T>;
      using Value = js::Value<T>;
      using Function = js::Function<T>;
      using ReturnValue = js::ReturnValue<T>;
      using Arguments = js::Arguments<T>;
  
  public:
      std::string const name = "Sync";
  
      static FunctionType create_constructor(ContextType);
  
      static void initialize_sync_manager(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void set_sync_log_level(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void set_sync_logger(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void set_sync_user_agent(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void initiate_client_reset(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void reconnect(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void has_existing_sessions(ContextType, ObjectType, Arguments &, ReturnValue &);
  
      static void create_global_notifier(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void local_listener_realms(ContextType, ObjectType, Arguments&, ReturnValue &);
      static void enable_multiplexing(ContextType, ObjectType, Arguments &, ReturnValue &);
      static void deserialize_change_set(ContextType, ObjectType, Arguments &, ReturnValue &);
  
      // private
      static std::function<SyncBindSessionHandler> session_bind_callback(ContextType ctx, ObjectType sync_constructor);
      static void populate_sync_config(ContextType, ObjectType realm_constructor, ObjectType config_object, Realm::Config&);
      static void populate_sync_config_for_ssl(ContextType, ObjectType config_object, SyncConfig&);
  
      MethodMap<T> const static_methods = {
          {"_hasExistingSessions", {wrap<has_existing_sessions>}},
          {"initiateClientReset", wrap<initiate_client_reset>},
          {"reconnect", wrap<reconnect>},
          {"setLogLevel", wrap<set_sync_log_level>},
          {"enableSessionMultiplexing", wrap<enable_multiplexing>},
          {"setUserAgent", wrap<set_sync_user_agent>},
          {"_initializeSyncManager", wrap<initialize_sync_manager>},
  
  #if REALM_PLATFORM_NODE
          {"setLogger", wrap<set_sync_logger>},
          {"setSyncLogger", wrap<set_sync_logger>},
          {"_createNotifier", wrap<create_global_notifier>},
          {"_localListenerRealms", wrap<local_listener_realms>},
          {"_deserializeChangeSet", wrap<deserialize_change_set>},
  #endif
      };
  };
  
  template<typename T>
  inline typename T::Function SyncClass<T>::create_constructor(ContextType ctx) {
      FunctionType sync_constructor = ObjectWrap<T, SyncClass<T>>::create_constructor(ctx);
  
      PropertyAttributes attributes = ReadOnly | DontEnum | DontDelete;
      Object::set_property(ctx, sync_constructor, "User", ObjectWrap<T, UserClass<T>>::create_constructor(ctx), attributes);
      Object::set_property(ctx, sync_constructor, "Session", ObjectWrap<T, SessionClass<T>>::create_constructor(ctx), attributes);
  #if REALM_PLATFORM_NODE
      Object::set_property(ctx, sync_constructor, "Adapter", ObjectWrap<T, AdapterClass<T>>::create_constructor(ctx), attributes);
  #endif
  
      return sync_constructor;
  }
  
  template<typename T>
  void SyncClass<T>::initialize_sync_manager(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue & return_value) {
      args.validate_count(1);
      std::string user_agent_binding_info = Value::validated_to_string(ctx, args[0]);
      ensure_directory_exists_for_file(default_realm_file_directory());
      SyncManager::shared().configure(default_realm_file_directory(), SyncManager::MetadataMode::NoEncryption, user_agent_binding_info);
  }
  
  template<typename T>
  void SyncClass<T>::initiate_client_reset(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue & return_value) {
      args.validate_count(1);
      std::string path = Value::validated_to_string(ctx, args[0]);
      if (!SyncManager::shared().immediately_run_file_actions(std::string(path))) {
          throw std::runtime_error(util::format("Realm was not configured correctly. Client Reset could not be run for Realm at: %1", path));
      }
  }
  
  template<typename T>
  void SyncClass<T>::set_sync_log_level(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(1);
      std::string log_level = Value::validated_to_string(ctx, args[0]);
      std::istringstream in(log_level); // Throws
      in.imbue(std::locale::classic()); // Throws
      in.unsetf(std::ios_base::skipws);
      util::Logger::Level log_level_2 = util::Logger::Level();
      in >> log_level_2; // Throws
      if (!in || !in.eof())
          throw std::runtime_error("Bad log level");
      syncManagerShared<T>(ctx).set_log_level(log_level_2);
  }
  
  #if REALM_PLATFORM_NODE
  template<typename T>
  void SyncClass<T>::set_sync_logger(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(1);
      auto callback_fn = Value::validated_to_function(ctx, args[0], "logger_callback");
      syncManagerShared<T>(ctx).set_logger_factory(*new realm::node::SyncLoggerFactory(ctx, callback_fn));
  }
  #endif
  
  template<typename T>
  void SyncClass<T>::set_sync_user_agent(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(1);
      std::string application_user_agent = Value::validated_to_string(ctx, args[0]);
      syncManagerShared<T>(ctx).set_user_agent(application_user_agent);
  }
  
  template<typename T>
  void SyncClass<T>::reconnect(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue &return_value) {
      args.validate_count(0);
      syncManagerShared<T>(ctx).reconnect();
  }
  
  template<typename T>
  void SyncClass<T>::has_existing_sessions(ContextType ctx, ObjectType this_object, Arguments &args, ReturnValue & return_value) {
      args.validate_count(0);
      return_value.set(syncManagerShared<T>(ctx).has_existing_sessions());
  }
  
  template<typename T>
  std::function<SyncBindSessionHandler> SyncClass<T>::session_bind_callback(ContextType ctx, ObjectType sync_constructor)
  {
      Protected<typename T::GlobalContext> protected_ctx(Context<T>::get_global_context(ctx));
      Protected<ObjectType> protected_sync_constructor(ctx, sync_constructor);
      return util::EventLoopDispatcher<SyncBindSessionHandler>([protected_ctx, protected_sync_constructor](const std::string& path, const realm::SyncConfig& config, std::shared_ptr<SyncSession>) {
          HANDLESCOPE
          ObjectType user_constructor = Object::validated_get_object(protected_ctx, protected_sync_constructor, "User");
          FunctionType refreshAccessToken = Object::validated_get_function(protected_ctx, user_constructor, "_refreshAccessToken");
  
          ValueType arguments[3];
          arguments[0] = create_object<T, UserClass<T>>(protected_ctx, new SharedUser(config.user));
          arguments[1] = Value::from_string(protected_ctx, path);
          arguments[2] = Value::from_string(protected_ctx, config.realm_url());
          Function::call(protected_ctx, refreshAccessToken, 3, arguments);
      });
  }
  
  template<typename T>
  void SyncClass<T>::populate_sync_config(ContextType ctx, ObjectType realm_constructor, ObjectType config_object, Realm::Config& config)
  {
      ValueType sync_config_value = Object::get_property(ctx, config_object, "sync");
      if (Value::is_boolean(ctx, sync_config_value)) {
          config.force_sync_history = Value::to_boolean(ctx, sync_config_value);
          if (config.force_sync_history) {
              config.schema_mode = SchemaMode::Additive;
          }
      } else if (!Value::is_undefined(ctx, sync_config_value)) {
          auto sync_config_object = Value::validated_to_object(ctx, sync_config_value);
  
          ObjectType sync_constructor = Object::validated_get_object(ctx, realm_constructor, "Sync");
          auto bind = session_bind_callback(ctx, sync_constructor);
  
          std::function<SyncSessionErrorHandler> error_handler;
          ValueType error_func = Object::get_property(ctx, sync_config_object, "error");
          if (!Value::is_undefined(ctx, error_func)) {
              error_handler = util::EventLoopDispatcher<SyncSessionErrorHandler>(SyncSessionErrorHandlerFunctor<T>(ctx, Value::validated_to_function(ctx, error_func)));
          }
  
          ObjectType user = Object::validated_get_object(ctx, sync_config_object, "user");
          SharedUser shared_user = *get_internal<T, UserClass<T>>(user);
          if (shared_user->state() != SyncUser::State::Active) {
              throw std::runtime_error("User is no longer valid.");
          }
  
          std::string raw_realm_url = Object::validated_get_string(ctx, sync_config_object, "url");
          if (shared_user->token_type() == SyncUser::TokenType::Admin) {
              size_t pos = raw_realm_url.find("/~/");
              if (pos != std::string::npos) {
                  raw_realm_url.replace(pos + 1, 1, "__auth");
              }
          }
  
          bool client_validate_ssl = true;
          ValueType validate_ssl_temp = Object::get_property(ctx, sync_config_object, "validate_ssl");
          if (!Value::is_undefined(ctx, validate_ssl_temp)) {
              client_validate_ssl = Value::validated_to_boolean(ctx, validate_ssl_temp, "validate_ssl");
          }
  
          util::Optional<std::string> ssl_trust_certificate_path;
          ValueType trust_certificate_path_temp = Object::get_property(ctx, sync_config_object, "ssl_trust_certificate_path");
           if (!Value::is_undefined(ctx, trust_certificate_path_temp)) {
              ssl_trust_certificate_path = std::string(Value::validated_to_string(ctx, trust_certificate_path_temp, "ssl_trust_certificate_path"));
          }
          else {
              ssl_trust_certificate_path = util::none;
          }
  
          std::function<sync::Session::SSLVerifyCallback> ssl_verify_callback;
          ValueType ssl_verify_func = Object::get_property(ctx, sync_config_object, "open_ssl_verify_callback");
          if (!Value::is_undefined(ctx, ssl_verify_func)) {
              SSLVerifyCallbackSyncThreadFunctor<T> ssl_verify_functor {ctx, Value::validated_to_function(ctx, ssl_verify_func)};
              ssl_verify_callback = std::move(ssl_verify_functor);
          }
  
          bool is_partial = false; // Change to `true` when `partial` is removed
          ValueType full_synchronization_value = Object::get_property(ctx, sync_config_object, "fullSynchronization");
          ValueType partial_value = Object::get_property(ctx, sync_config_object, "partial");
  
          // Disallow setting `partial` and `fullSynchronization` at the same time
          if (!Value::is_undefined(ctx, full_synchronization_value) && !Value::is_undefined(ctx, partial_value)) {
              throw std::invalid_argument("'partial' and 'fullSynchronization' were both set. 'partial' has been deprecated, use only 'fullSynchronization'");
          }
  
          if (!Value::is_undefined(ctx, partial_value)) {
              is_partial = Value::validated_to_boolean(ctx, partial_value);
          } else if (!Value::is_undefined(ctx, full_synchronization_value)) {
              is_partial = !Value::validated_to_boolean(ctx, full_synchronization_value);
          }
  
          bool disable_query_based_sync_url_checks = false;
          ValueType disable_query_based_sync_url_checks_value = Object::get_property(ctx, sync_config_object, "_disableQueryBasedSyncUrlChecks");
          if (!Value::is_undefined(ctx, disable_query_based_sync_url_checks_value)) {
              disable_query_based_sync_url_checks = Value::validated_to_boolean(ctx, disable_query_based_sync_url_checks_value);
          }
  
          if (disable_query_based_sync_url_checks) {
              config.sync_config = std::make_shared<SyncConfig>(shared_user, std::move(""));
              config.sync_config->reference_realm_url = std::move(raw_realm_url);
          }
          else {
              config.sync_config = std::make_shared<SyncConfig>(shared_user, std::move(raw_realm_url));
          }
          config.sync_config->bind_session_handler = std::move(bind);
          config.sync_config->error_handler = std::move(error_handler);
          config.sync_config->is_partial = is_partial;
  
          SyncSessionStopPolicy session_stop_policy = SyncSessionStopPolicy::AfterChangesUploaded;
          ValueType session_stop_policy_value = Object::get_property(ctx, sync_config_object, "_sessionStopPolicy");
          if (!Value::is_undefined(ctx, session_stop_policy_value)) {
              std::string stop_session = Value::validated_to_string(ctx, session_stop_policy_value, "_sessionStopPolicy");
              if (stop_session == std::string("immediately")) {
                  session_stop_policy = SyncSessionStopPolicy::Immediately;
              } else if (stop_session == std::string("never")) {
                  session_stop_policy = SyncSessionStopPolicy::LiveIndefinitely;
              } else if (stop_session == std::string("after-upload")) {
                  session_stop_policy = SyncSessionStopPolicy::AfterChangesUploaded;
              } else {
                  throw std::invalid_argument("Unknown argument for _sessionStopPolicy: " + stop_session);
              }
          }
          config.sync_config->stop_policy = session_stop_policy;
  
          ValueType custom_partial_sync_identifier_value = Object::get_property(ctx, sync_config_object, "customQueryBasedSyncIdentifier");
          if (!Value::is_undefined(ctx, custom_partial_sync_identifier_value)) {
              config.sync_config->custom_partial_sync_identifier = std::string(Value::validated_to_string(ctx, custom_partial_sync_identifier_value, "customQueryBasedSyncIdentifier"));
          }
  
          // Custom HTTP headers
          ValueType sync_custom_http_headers_value = Object::get_property(ctx, sync_config_object, "custom_http_headers");
          if (!Value::is_undefined(ctx, sync_custom_http_headers_value)) {
              auto sync_custom_http_headers = Value::validated_to_object(ctx, sync_custom_http_headers_value);
              auto property_names = Object::get_property_names(ctx, sync_custom_http_headers);
              std::map<std::string, std::string> http_headers;
              for (auto it = property_names.begin(); it != property_names.end(); ++it) {
                  auto name = *it;
                  ValueType prop_value = Object::get_property(ctx, sync_custom_http_headers, name);
                  auto value = Value::validated_to_string(ctx, prop_value);
                  http_headers[name] = value;
              }
              config.sync_config->custom_http_headers = std::move(http_headers);
          }
  
          // TODO: remove
          config.sync_config->client_validate_ssl = client_validate_ssl;
          config.sync_config->ssl_trust_certificate_path = ssl_trust_certificate_path;
          config.sync_config->ssl_verify_callback = std::move(ssl_verify_callback);
  
          ValueType ssl_config_value = Object::get_property(ctx, sync_config_object, "ssl");
          if (Value::is_object(ctx, ssl_config_value)) {
              auto ssl_config_object = Value::to_object(ctx, ssl_config_value);
              populate_sync_config_for_ssl(ctx, ssl_config_object, *config.sync_config);
          }
  
          config.schema_mode = SchemaMode::Additive;
          config.path = syncManagerShared<T>(ctx).path_for_realm(*shared_user, config.sync_config->realm_url());
  
          if (!config.encryption_key.empty()) {
              config.sync_config->realm_encryption_key = std::array<char, 64>();
              std::copy_n(config.encryption_key.begin(), config.sync_config->realm_encryption_key->size(), config.sync_config->realm_encryption_key->begin());
          }
  
  #if REALM_ANDROID
          // For React Native Android, if the user didn't define the ssl_verify_callback, we provide a default
          // implementation for him, otherwise all SSL validation will fail, since the Sync client doesn't have
          // access to the Android Keystore.
          // This default implementation will perform a JNI call to invoke a Java method defined at the `SSLHelper`
          // to perform the certificate verification.
          if (!config.sync_config->ssl_verify_callback) {
              auto ssl_verify_functor =
              [](const std::string server_address, realm::sync::Session::port_type server_port,
                 const char* pem_data, size_t pem_size, int preverify_ok, int depth) {
                  JNIEnv* env = realm::jni_util::JniUtils::get_env(true);
                  static jmethodID java_certificate_verifier = env->GetStaticMethodID(ssl_helper_class, "certificateVerifier", "(Ljava/lang/String;Ljava/lang/String;I)Z");
                  jstring jserver_address = env->NewStringUTF(server_address.c_str());
                  // deep copy the pem_data into a string so DeleteLocalRef delete the local reference not the original const char
                  std::string pem(pem_data, pem_size);
                  jstring jpem = env->NewStringUTF(pem.c_str());
  
                  bool isValid = env->CallStaticBooleanMethod(ssl_helper_class, java_certificate_verifier,
                                                              jserver_address,
                                                              jpem, depth) == JNI_TRUE;
                  env->DeleteLocalRef(jserver_address);
                  env->DeleteLocalRef(jpem);
                  return isValid;
              };
              config.sync_config->ssl_verify_callback = std::move(ssl_verify_functor);
          }
  #endif
      }
  }
  
  template<typename T>
  void SyncClass<T>::populate_sync_config_for_ssl(ContextType ctx, ObjectType config_object, SyncConfig& config)
  {
      ValueType validate_ssl = Object::get_property(ctx, config_object, "validate");
      if (Value::is_boolean(ctx, validate_ssl)) {
          config.client_validate_ssl = Value::to_boolean(ctx, validate_ssl);
      }
  
      ValueType certificate_path = Object::get_property(ctx, config_object, "certificatePath");
      if (Value::is_string(ctx, certificate_path)) {
          config.ssl_trust_certificate_path = std::string(Value::to_string(ctx, certificate_path));
      }
  
      ValueType validate_callback = Object::get_property(ctx, config_object, "validateCallback");
      if (Value::is_function(ctx, validate_callback)) {
          config.ssl_verify_callback = SSLVerifyCallbackSyncThreadFunctor<T> { ctx, Value::to_function(ctx, validate_callback) };
      }
  }
  
  template<typename T>
  void SyncClass<T>::enable_multiplexing(ContextType ctx, ObjectType this_object, Arguments& arguments, ReturnValue &return_value) {
      arguments.validate_count(0);
      SyncManager::shared().enable_session_multiplexing();
  }
  
  #if REALM_PLATFORM_NODE
  template<typename T>
  void SyncClass<T>::create_global_notifier(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      args.validate_maximum(5);
      std::string local_root_dir = normalize_realm_path(Value::validated_to_string(ctx, args[4], "listenerDirectory"));
      util::try_make_dir(local_root_dir);
  
      std::string server_base_url = Value::validated_to_string(ctx, args[0], "serverUrl");
  
      Uri uri(server_base_url);
      if (uri.get_scheme() != "realm:" && uri.get_scheme() != "realms:") {
          throw std::runtime_error("Server URL must be of the realm-scheme");
      }
  
      if (!uri.get_path().empty() || !uri.get_query().empty() || !uri.get_frag().empty()) {
          throw std::runtime_error("Server URL must only contain a host and port");
      }
  
      ObjectType user = Value::validated_to_object(ctx, args[1], "adminUser");
      if (!Object::template is_instance<UserClass<T>>(ctx, user)) {
          throw std::runtime_error("object must be of type Sync.User");
      }
      SharedUser shared_user = *get_internal<T, UserClass<T>>(user);
      if (shared_user->state() != SyncUser::State::Active) {
          throw std::runtime_error("User is no longer valid.");
      }
      if (!shared_user->is_admin()) {
          throw std::runtime_error("User needs to be an admin.");
      }
  
      auto user_callback = Value::validated_to_function(ctx, args[2], "callback");
  
      SyncConfig sync_config_template(shared_user, server_base_url);
      if (!Value::is_undefined(ctx, args[3])) {
          ObjectType ssl_config_object = Value::validated_to_object(ctx, args[3], "sslConfiguration");
          SyncClass<T>::populate_sync_config_for_ssl(ctx, ssl_config_object, sync_config_template);
      }
  
      sync_config_template.bind_session_handler = SyncClass<T>::session_bind_callback(ctx, this_object);
  
      auto notifier = std::make_unique<GlobalNotifier>(std::make_unique<GlobalNotifierCallback<T>>(ctx, Protected<FunctionType>{ctx, std::move(user_callback)}),
                                                       std::move(local_root_dir),
                                                       std::move(sync_config_template)); // Throws
      return_value.set(create_object<T, GlobalNotifierClass<T>>(ctx, notifier.get()));
      notifier.release();
  }
  
  template <typename T>
  void SyncClass<T>::local_listener_realms(ContextType ctx, ObjectType this_object, Arguments& args,
                                           ReturnValue& return_value)
  {
      args.validate_count(1);
  
      std::string local_root_dir = normalize_realm_path(Value::validated_to_string(ctx, args[0], "listenerDirectory"));
      std::string admin_realm_path = util::File::resolve("realms.realm", local_root_dir);
      // if the admin Realm doesn't exists, then there is no local Realm files to
      // return (notifier didn't run yet here).
      if (!util::File::exists(admin_realm_path)) {
          return_value.set_undefined();
          return;
      }
  
      // If the admin Realm is already open we need to get it from the
      // coordinator to get the matching sync configuration, but if it's not
      // already open we want to open it without creating a sync session.
      std::shared_ptr<Realm> realm;
      if (auto coordinator = realm::_impl::RealmCoordinator::get_existing_coordinator(admin_realm_path)) {
          realm = coordinator->get_realm();
      }
      else {
          Realm::Config config;
          config.cache = false;
          config.path = admin_realm_path;
          config.force_sync_history = true;
          config.schema_mode = SchemaMode::Additive;
          realm = Realm::get_shared_realm(config);
      }
  
      auto& group = realm->read_group();
      auto& table = *ObjectStore::table_for_object_type(group, "RealmFile");
      size_t path_col_ndx = table.get_column_index("path");
  
      std::vector<std::string> local_realms;
      for (size_t i = 0, size = table.size(); i < size; ++i) {
          auto virtual_path = table.get_string(path_col_ndx, i);
          auto id = sync::object_id_for_row(group, table, i);
          std::string file_path = util::format("%1/realms%2/%3.realm", local_root_dir, virtual_path, id.to_string());
  
          // filter out Realms not present locally
          if (util::File::exists(file_path)) {
              local_realms.push_back(virtual_path);
              local_realms.push_back(file_path);
          }
      }
  
      if (local_realms.empty()) {
          return_value.set_undefined();
          return;
      }
  
      auto arr = Nan::New<v8::Array>(local_realms.size());
      for (size_t i = 0; i < local_realms.size(); i++) {
          Nan::Set(arr, i, Nan::New<v8::String>(local_realms[i]).ToLocalChecked());
      }
      return_value.set(arr);
  }
  
  template<typename T>
  void SyncClass<T>::deserialize_change_set(ContextType ctx, ObjectType this_object, Arguments& args, ReturnValue &return_value) {
      std::string serialized = Value::validated_to_string(ctx, args[0], "serialized");
      return_value.set(create_object<T, ChangeObject<T>>(ctx, new GlobalNotifier::ChangeNotification(serialized)));
  }
  #endif
  
  } // js
  } // realm
  