<commit_before>////////////////////////////////////////////////////////////////////////////
//
// 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.
//
////////////////////////////////////////////////////////////////////////////

#include "global_notifier.hpp"

#include "admin_realm.hpp"

#include "impl/realm_coordinator.hpp"
#include "impl/transact_log_handler.hpp"
#include "object_schema.hpp"
#include "object_store.hpp"
#include "event_loop_dispatcher.hpp"
#include "results.hpp"
#include "util/event_loop_signal.hpp"

#include "sync/sync_manager.hpp"
#include "sync/sync_user.hpp"
#include "sync/sync_session.hpp"

#include <realm/lang_bind_helper.hpp>
#include <realm/util/file.hpp>
#include <realm/util/scope_exit.hpp>

#include <json.hpp>

#include <condition_variable>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <unordered_map>
#include <utility>
#include <vector>

using namespace realm;
using namespace realm::_impl;

namespace realm {
void to_json(nlohmann::json& j, VersionID v)
{
    j = {{"version", v.version}, {"index", v.index}};
}
void from_json(nlohmann::json const& j, VersionID& v)
{
    v.version = j["version"];
    v.index = j["index"];
}
}

class GlobalNotifier::Impl : public AdminRealmListener {
public:
    Impl(std::unique_ptr<Callback>, std::string local_root_dir,
         std::string server_base_url, std::shared_ptr<SyncUser> user,
         std::function<SyncBindSessionHandler> bind_callback);

    Realm::Config get_config(StringData virtual_path, util::Optional<Schema> schema) const;
    util::Optional<ChangeNotification> next_changed_realm();
    void release_version(std::string const& virtual_path, VersionID old_version, VersionID new_version);

public:
    void register_realm(StringData virtual_path) override;
    void error(std::exception_ptr err) override { m_target->error(err); }
    void download_complete() override { m_target->download_complete(); }

    const std::unique_ptr<Callback> m_target;
    const std::string m_server_base_url;
    std::shared_ptr<SyncUser> m_user;
    std::function<SyncBindSessionHandler> m_bind_callback;
    std::string m_regular_realms_dir;

    std::mutex m_work_queue_mutex;
    struct RealmToCalculate {
        std::string virtual_path;
        std::shared_ptr<_impl::RealmCoordinator> coordinator;
        std::unique_ptr<Replication> history;
        std::unique_ptr<SharedGroup> shared_group;
        std::queue<VersionID> versions;
    };
    std::queue<RealmToCalculate*> m_work_queue;
    std::unordered_map<std::string, RealmToCalculate> m_realms;

    struct SignalCallback {
        std::weak_ptr<Impl> notifier;
        void operator()()
        {
            if (auto alive = notifier.lock()) {
                GlobalNotifier notifier(alive);
                alive->m_target->realm_changed(&notifier);
            }
        }
    };

    std::shared_ptr<util::EventLoopSignal<SignalCallback>> m_signal;
};

GlobalNotifier::Impl::Impl(std::unique_ptr<Callback> async_target,
                           std::string local_root_dir, std::string server_base_url,
                           std::shared_ptr<SyncUser> user, std::function<SyncBindSessionHandler> bind_callback)
: AdminRealmListener(local_root_dir, server_base_url, user, bind_callback)
, m_target(std::move(async_target))
, m_server_base_url(std::move(server_base_url))
, m_user(std::move(user))
, m_bind_callback(std::move(bind_callback))
, m_regular_realms_dir(util::File::resolve("realms", local_root_dir)) // Throws
{
    util::try_make_dir(m_regular_realms_dir); // Throws
}

Realm::Config GlobalNotifier::Impl::get_config(StringData virtual_path, util::Optional<Schema> schema) const {
    Realm::Config config;
    if (schema) {
        config.schema = std::move(schema);
        config.schema_version = 0;
    }

    std::string file_path = m_regular_realms_dir + virtual_path.data() + ".realm";
    for (size_t pos = m_regular_realms_dir.size(); pos != file_path.npos; pos = file_path.find('/', pos + 1)) {
        file_path[pos] = '\0';
        util::try_make_dir(file_path);
        file_path[pos] = '/';
    }

    config.path = std::move(file_path);
    config.sync_config = std::make_unique<SyncConfig>(m_user, m_server_base_url + virtual_path.data());
    config.sync_config->bind_session_handler = m_bind_callback;
    config.schema_mode = SchemaMode::Additive;
    config.cache = false;
    config.automatic_change_notifications = false;
    return config;
}

void GlobalNotifier::Impl::register_realm(StringData path) {
    if (!m_target->realm_available(path))
        return;

    auto config = get_config(path, util::none);
    auto coordinator = _impl::RealmCoordinator::get_coordinator(config);
    auto info = &m_realms.emplace(path, RealmToCalculate{path, coordinator}).first->second;

    std::weak_ptr<Impl> weak_self = std::static_pointer_cast<Impl>(shared_from_this());
    coordinator->set_transaction_callback([=, weak_self = std::move(weak_self)](VersionID old_version, VersionID new_version) {
        auto self = weak_self.lock();
        if (!self)
            return;

        std::lock_guard<std::mutex> l(m_work_queue_mutex);
        if (!info->shared_group) {
            std::unique_ptr<Group> read_only_group;
            auto config = info->coordinator->get_config();
            config.force_sync_history = true; // FIXME: needed?
            config.schema = util::none;
            Realm::open_with_config(config, info->history, info->shared_group, read_only_group, nullptr);
            info->shared_group->begin_read(old_version);
        }
        info->versions.push(new_version);
        if (info->versions.size() == 1) {
            m_work_queue.push(info);
            m_signal->notify();
        }
    });
}

void GlobalNotifier::Impl::release_version(std::string const& virtual_path, VersionID old_version, VersionID new_version)
{
    auto it = m_realms.find(virtual_path);
    REALM_ASSERT(it != m_realms.end());

    auto& sg = it->second.shared_group;
    REALM_ASSERT(sg->get_version_of_current_transaction() == old_version);

    REALM_ASSERT(!it->second.versions.empty() && it->second.versions.front() == new_version);
    it->second.versions.pop();

    if (it->second.versions.empty()) {
        it->second.shared_group = nullptr;
        it->second.history = nullptr;
    }
    else {
        LangBindHelper::advance_read(*sg, new_version);
        m_work_queue.push(&it->second);
        m_signal->notify();
    }
}

GlobalNotifier::GlobalNotifier(std::unique_ptr<Callback> async_target,
                               std::string local_root_dir, std::string server_base_url,
                               std::shared_ptr<SyncUser> user, std::function<SyncBindSessionHandler> bind_callback)
: m_impl(std::make_shared<GlobalNotifier::Impl>(std::move(async_target),
                                                std::move(local_root_dir),
                                                std::move(server_base_url),<commit_msg>Fix use-after-free crashes in Global Notifier (#405)<commit_after>////////////////////////////////////////////////////////////////////////////
//
// 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.
//
////////////////////////////////////////////////////////////////////////////

#include "global_notifier.hpp"

#include "admin_realm.hpp"

#include "impl/realm_coordinator.hpp"
#include "impl/transact_log_handler.hpp"
#include "object_schema.hpp"
#include "object_store.hpp"
#include "event_loop_dispatcher.hpp"
#include "results.hpp"
#include "util/event_loop_signal.hpp"

#include "sync/sync_manager.hpp"
#include "sync/sync_user.hpp"
#include "sync/sync_session.hpp"

#include <realm/lang_bind_helper.hpp>
#include <realm/util/file.hpp>
#include <realm/util/scope_exit.hpp>

#include <json.hpp>

#include <condition_variable>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <unordered_map>
#include <utility>
#include <vector>

using namespace realm;
using namespace realm::_impl;

namespace realm {
void to_json(nlohmann::json& j, VersionID v)
{
    j = {{"version", v.version}, {"index", v.index}};
}
void from_json(nlohmann::json const& j, VersionID& v)
{
    v.version = j["version"];
    v.index = j["index"];
}
}

class GlobalNotifier::Impl : public AdminRealmListener {
public:
    Impl(std::unique_ptr<Callback>, std::string local_root_dir,
         std::string server_base_url, std::shared_ptr<SyncUser> user,
         std::function<SyncBindSessionHandler> bind_callback);

    Realm::Config get_config(StringData virtual_path, util::Optional<Schema> schema) const;
    util::Optional<ChangeNotification> next_changed_realm();
    void release_version(std::string const& virtual_path, VersionID old_version, VersionID new_version);

public:
    void register_realm(StringData virtual_path) override;
    void error(std::exception_ptr err) override { m_target->error(err); }
    void download_complete() override { m_target->download_complete(); }

    const std::unique_ptr<Callback> m_target;
    const std::string m_server_base_url;
    std::shared_ptr<SyncUser> m_user;
    std::function<SyncBindSessionHandler> m_bind_callback;
    std::string m_regular_realms_dir;

    std::mutex m_work_queue_mutex;
    struct RealmToCalculate {
        std::string virtual_path;
        std::shared_ptr<_impl::RealmCoordinator> coordinator;
        std::unique_ptr<Replication> history;
        std::unique_ptr<SharedGroup> shared_group;
        std::queue<VersionID> versions;
    };
    std::queue<RealmToCalculate*> m_work_queue;
    std::unordered_map<std::string, RealmToCalculate> m_realms;

    struct SignalCallback {
        std::weak_ptr<Impl> notifier;
        void operator()()
        {
            if (auto alive = notifier.lock()) {
                GlobalNotifier notifier(alive);
                alive->m_target->realm_changed(&notifier);
            }
        }
    };

    std::shared_ptr<util::EventLoopSignal<SignalCallback>> m_signal;
};

GlobalNotifier::Impl::Impl(std::unique_ptr<Callback> async_target,
                           std::string local_root_dir, std::string server_base_url,
                           std::shared_ptr<SyncUser> user, std::function<SyncBindSessionHandler> bind_callback)
: AdminRealmListener(local_root_dir, server_base_url, user, bind_callback)
, m_target(std::move(async_target))
, m_server_base_url(std::move(server_base_url))
, m_user(std::move(user))
, m_bind_callback(std::move(bind_callback))
, m_regular_realms_dir(util::File::resolve("realms", local_root_dir)) // Throws
{
    util::try_make_dir(m_regular_realms_dir); // Throws
}

Realm::Config GlobalNotifier::Impl::get_config(StringData virtual_path, util::Optional<Schema> schema) const {
    Realm::Config config;
    if (schema) {
        config.schema = std::move(schema);
        config.schema_version = 0;
    }

    std::string file_path = m_regular_realms_dir + virtual_path.data() + ".realm";
    for (size_t pos = m_regular_realms_dir.size(); pos != file_path.npos; pos = file_path.find('/', pos + 1)) {
        file_path[pos] = '\0';
        util::try_make_dir(file_path);
        file_path[pos] = '/';
    }

    config.path = std::move(file_path);
    config.sync_config = std::make_unique<SyncConfig>(m_user, m_server_base_url + virtual_path.data());
    config.sync_config->bind_session_handler = m_bind_callback;
    config.schema_mode = SchemaMode::Additive;
    config.cache = false;
    config.automatic_change_notifications = false;
    return config;
}

void GlobalNotifier::Impl::register_realm(StringData path) {
    if (!m_target->realm_available(path))
        return;

    auto config = get_config(path, util::none);
    auto coordinator = _impl::RealmCoordinator::get_coordinator(config);
    auto info = &m_realms.emplace(path, RealmToCalculate{path, coordinator}).first->second;

    std::weak_ptr<Impl> weak_self = std::static_pointer_cast<Impl>(shared_from_this());
    coordinator->set_transaction_callback([=, weak_self = std::move(weak_self)](VersionID old_version, VersionID new_version) {
        auto self = weak_self.lock();
        if (!self)
            return;

        std::lock_guard<std::mutex> l(m_work_queue_mutex);
        if (!info->shared_group) {
            std::unique_ptr<Group> read_only_group;
            auto config = info->coordinator->get_config();
            config.force_sync_history = true; // FIXME: needed?
            config.schema = util::none;
            Realm::open_with_config(config, info->history, info->shared_group, read_only_group, nullptr);
            info->shared_group->begin_read(old_version);
        }
        info->versions.push(new_version);
        if (info->versions.size() == 1) {
            m_work_queue.push(info);
            m_signal->notify();
        }
    });
}

void GlobalNotifier::Impl::release_version(std::string const& virtual_path, VersionID old_version, VersionID new_version)
{
    auto it = m_realms.find(virtual_path);
    REALM_ASSERT(it != m_realms.end());

    std::lock_guard<std::mutex> l(m_work_queue_mutex);

    auto& sg = it->second.shared_group;
    REALM_ASSERT(sg->get_version_of_current_transaction() == old_version);

    REALM_ASSERT(!it->second.versions.empty() && it->second.versions.front() == new_version);
    it->second.versions.pop();

    if (it->second.versions.empty()) {
        it->second.shared_group = nullptr;
        it->second.history = nullptr;
    }
    else {
        LangBindHelper::advance_read(*sg, new_version);
        m_work_queue.push(&it->second);
        m_signal->notify();
    }
}

GlobalNotifier::GlobalNotifier(std::unique_ptr<Callback> async_target,
                               std::string local_root_dir, std::string server_base_url,
                               std::shared_ptr<SyncUser> user, std::function<SyncBindSessionHandler> bind_callback)
: m_impl(std::make_shared<GlobalNotifier::Impl>(std::move(async_target),
                                                std::move(local_root_dir),
                                                std::move(server_base_url),
{'fraction_non_alphanumeric': 0.14438042254453937, 'fraction_numerical': 0.0017881978939002583, 'mean_word_length': 3.4281524926686218, 'pattern_counts': {'":': 0, '<': 86, '<?xml version=': 0, '>': 142, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5262229', 'n_tokens_mistral': 4643, 'n_tokens_neox': 4566, 'n_words': 857}