////////////////////////////////////////////////////////////////////////////
//
// Copyright 2015 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 <catch2/catch_all.hpp>

#include "util/test_file.hpp"
#include "util/test_utils.hpp"
#include "util/event_loop.hpp"
#include "util/index_helpers.hpp"

#include <realm/object-store/binding_context.hpp>
#include <realm/object-store/list.hpp>
#include <realm/object-store/object.hpp>
#include <realm/object-store/object_schema.hpp>
#include <realm/object-store/property.hpp>
#include <realm/object-store/results.hpp>
#include <realm/object-store/schema.hpp>

#include <realm/object-store/impl/realm_coordinator.hpp>
#include <realm/object-store/impl/object_accessor_impl.hpp>

#include <realm/version.hpp>
#include <realm/db.hpp>

#include <cstdint>

using namespace realm;
using util::any_cast;

TEST_CASE("list", "[list]") {
    InMemoryTestFile config;
    config.automatic_change_notifications = false;
    auto r = Realm::get_shared_realm(config);
    r->update_schema({
        {"origin", {{"array", PropertyType::Array | PropertyType::Object, "target"}}},
        {"target", {{"value", PropertyType::Int}, {"value2", PropertyType::Int}}},
        {"other_origin", {{"array", PropertyType::Array | PropertyType::Object, "other_target"}}},
        {"other_target", {{"value", PropertyType::Int}}},
    });

    auto& coordinator = *_impl::RealmCoordinator::get_coordinator(config.path);

    auto origin = r->read_group().get_table("class_origin");
    auto target = r->read_group().get_table("class_target");
    auto other_origin = r->read_group().get_table("class_other_origin");
    auto other_target = r->read_group().get_table("class_other_target");
    ColKey col_link = origin->get_column_key("array");
    ColKey col_target_value = target->get_column_key("value");
    ColKey other_col_link = other_origin->get_column_key("array");
    ColKey other_col_value = other_target->get_column_key("value");

    r->begin_transaction();

    std::vector<ObjKey> target_keys;
    target->create_objects(10, target_keys);
    for (int i = 0; i < 10; ++i)
        target->get_object(target_keys[i]).set_all(i);

    Obj obj = origin->create_object();
    auto lv = obj.get_linklist_ptr(col_link);
    for (int i = 0; i < 10; ++i)
        lv->add(target_keys[i]);
    auto lv2 = origin->create_object().get_linklist_ptr(col_link);
    for (int i = 0; i < 10; ++i)
        lv2->add(target_keys[i]);

    ObjKeys other_target_keys({3, 5, 7, 9, 11, 13, 15, 17, 19, 21});
    other_target->create_objects(other_target_keys);
    for (int i = 0; i < 10; ++i)
        other_target->get_object(other_target_keys[i]).set_all(i);

    Obj other_obj = other_origin->create_object();
    auto other_lv = other_obj.get_linklist_ptr(other_col_link);
    for (int i = 0; i < 10; ++i)
        other_lv->add(other_target_keys[i]);

    r->commit_transaction();

    auto r2 = coordinator.get_realm();
    auto r2_lv = r2->read_group().get_table("class_origin")->get_object(0).get_linklist_ptr(col_link);

    auto write = [&](auto&& f) {
        r->begin_transaction();
        f();
        r->commit_transaction();
        advance_and_notify(*r);
    };

    SECTION("add_notification_block()") {
        CollectionChangeSet change;
        List lst(r, obj, col_link);

        auto require_change = [&] {
            auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            advance_and_notify(*r);
            return token;
        };

        auto require_no_change = [&] {
            bool first = true;
            auto token = lst.add_notification_callback([&, first](CollectionChangeSet) mutable {
                REQUIRE(first);
                first = false;
            });
            advance_and_notify(*r);
            return token;
        };

        SECTION("modifying the list sends a change notifications") {
            auto token = require_change();
            write([&] {
                if (lv2->size() > 5)
                    lst.remove(5);
            });
            REQUIRE_INDICES(change.deletions, 5);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("modifying a different list doesn't send a change notification") {
            auto token = require_no_change();
            write([&] {
                if (lv2->size() > 5)
                    lv2->remove(5);
            });
        }

        SECTION("clearing list sets cleared flag") {
            auto token = require_change();
            write([&] {
                lst.remove_all();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
            REQUIRE(change.collection_was_cleared);
        }

        SECTION("clearing list followed by insertion does not set cleared flag") {
            auto token = require_change();
            write([&] {
                lst.remove_all();
                Obj obj = target->get_object(target_keys[5]);
                lst.add(obj);
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
            REQUIRE_INDICES(change.insertions, 0);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("removing all elements from list does not set cleared flag") {
            auto token = require_change();
            write([&] {
                auto size = lst.size();
                for (size_t i = 0; i < size; i++)
                    lst.remove(0);
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("deleting the list sends a change notification") {
            auto token = require_change();
            write([&] {
                obj.remove();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);

            // Should not resend delete all notification after another commit
            change = {};
            write([&] {
                target->create_object();
            });
            REQUIRE(change.empty());
        }

        SECTION("deleting list before first run of notifier reports deletions") {
            auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            advance_and_notify(*r);
            write([&] {
                origin->begin()->remove();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
            REQUIRE(change.collection_root_was_deleted);
        }

        SECTION("deleting an empty list triggers the notifier") {
            size_t notifier_count = 0;
            auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
                ++notifier_count;
            });
            advance_and_notify(*r);
            write([&] {
                lst.delete_all();
            });
            REQUIRE(!change.collection_root_was_deleted);
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
            REQUIRE(notifier_count == 2);
            REQUIRE(lst.size() == 0);

            write([&] {
                origin->begin()->remove();
            });
            REQUIRE(change.deletions.count() == 0);
            REQUIRE(change.collection_root_was_deleted);
            REQUIRE(notifier_count == 3);

            // Should not resend delete notification after another commit
            change = {};
            write([&] {
                target->create_object();
            });
            REQUIRE(change.empty());
        }

        SECTION("modifying one of the target rows sends a change notification") {
            auto token = require_change();
            write([&] {
                lst.get(5).set(col_target_value, 6);
            });
            REQUIRE_INDICES(change.modifications, 5);
        }

        SECTION("deleting a target row sends a change notification") {
            auto token = require_change();
            write([&] {
                target->remove_object(target_keys[5]);
            });
            REQUIRE_INDICES(change.deletions, 5);
        }

        SECTION("adding a row and then modifying the target row does not mark the row as modified") {
            auto token = require_change();
            write([&] {
                Obj obj = target->get_object(target_keys[5]);
                lst.add(obj);
                obj.set(col_target_value, 10);
            });
            REQUIRE_INDICES(change.insertions, 10);
            REQUIRE_INDICES(change.modifications, 5);
        }

        SECTION("modifying and then moving a row reports move/insert but not modification") {
            auto token = require_change();
            write([&] {
                target->get_object(target_keys[5]).set(col_target_value, 10);
                lst.move(5, 8);
            });
            REQUIRE_INDICES(change.insertions, 8);
            REQUIRE_INDICES(change.deletions, 5);
            REQUIRE_MOVES(change, {5, 8});
            REQUIRE(change.modifications.empty());
        }

        SECTION("modifying a row which appears multiple times in a list marks them all as modified") {
            r->begin_transaction();
            lst.add(target_keys[5]);
            r->commit_transaction();

            auto token = require_change();
            write([&] {
                target->get_object(target_keys[5]).set(col_target_value, 10);
            });
            REQUIRE_INDICES(change.modifications, 5, 10);
        }

        SECTION("deleting a row which appears multiple times in a list marks them all as modified") {
            r->begin_transaction();
            lst.add(target_keys[5]);
            r->commit_transaction();

            auto token = require_change();
            write([&] {
                target->remove_object(target_keys[5]);
            });
            REQUIRE_INDICES(change.deletions, 5, 10);
        }

        SECTION("clearing the target table sends a change notification") {
            auto token = require_change();
            write([&] {
                target->clear();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
        }

        SECTION("moving a target row does not send a change notification") {
            // Remove a row from the LV so that we have one to delete that's not in the list
            r->begin_transaction();
            if (lv->size() > 2)
                lv->remove(2);
            r->commit_transaction();

            auto token = require_no_change();
            write([&] {
                target->remove_object(target_keys[2]);
            });
        }

        SECTION("multiple LinkViews for the same LinkList can get notifications") {
            r->begin_transaction();
            target->clear();
            std::vector<ObjKey> keys;
            target->create_objects(5, keys);
            r->commit_transaction();

            auto get_list = [&] {
                auto r = Realm::get_shared_realm(config);
                auto obj = r->read_group().get_table("class_origin")->get_object(0);
                return List(r, obj, col_link);
            };
            auto change_list = [&] {
                r->begin_transaction();
                if (lv->size()) {
                    target->get_object(lv->size() - 1).set(col_target_value, int64_t(lv->size()));
                }
                lv->add(keys[lv->size()]);
                r->commit_transaction();
            };

            List lists[3];
            NotificationToken tokens[3];
            CollectionChangeSet changes[3];

            for (int i = 0; i < 3; ++i) {
                lists[i] = get_list();
                tokens[i] = lists[i].add_notification_callback([i, &changes](CollectionChangeSet c) {
                    changes[i] = std::move(c);
                });
                change_list();
            }

            // Each of the Lists now has a different source version and state at
            // that version, so they should all see different changes despite
            // being for the same LinkList
            for (auto& list : lists)
                advance_and_notify(*list.get_realm());

            REQUIRE_INDICES(changes[0].insertions, 0, 1, 2);
            REQUIRE(changes[0].modifications.empty());

            REQUIRE_INDICES(changes[1].insertions, 1, 2);
            REQUIRE_INDICES(changes[1].modifications, 0);

            REQUIRE_INDICES(changes[2].insertions, 2);
            REQUIRE_INDICES(changes[2].modifications, 1);

            // After making another change, they should all get the same notification
            change_list();
            for (auto& list : lists)
                advance_and_notify(*list.get_realm());

            for (int i = 0; i < 3; ++i) {
                REQUIRE_INDICES(changes[i].insertions, 3);
                REQUIRE_INDICES(changes[i].modifications, 2);
            }
        }

        SECTION("multiple callbacks for the same Lists can be skipped individually") {
            auto token = require_no_change();
            auto token2 = require_change();

            r->begin_transaction();
            lv->add(target_keys[0]);
            token.suppress_next();
            r->commit_transaction();

            advance_and_notify(*r);
            REQUIRE_INDICES(change.insertions, 10);
        }

        SECTION("multiple Lists for the same LinkView can be skipped individually") {
            auto token = require_no_change();

            List list2(r, obj, col_link);
            auto token2 = list2.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            advance_and_notify(*r);

            r->begin_transaction();
            lv->add(target_keys[0]);
            token.suppress_next();
            r->commit_transaction();

            advance_and_notify(*r);
            REQUIRE_INDICES(change.insertions, 10);
        }

        SECTION("skipping only effects the current transaction even if no notification would occur anyway") {
            auto token = require_change();

            // would not produce a notification even if it wasn't skipped because no changes were made
            r->begin_transaction();
            token.suppress_next();
            r->commit_transaction();
            advance_and_notify(*r);
            REQUIRE(change.empty());

            // should now produce a notification
            r->begin_transaction();
            lv->add(target_keys[0]);
            r->commit_transaction();
            advance_and_notify(*r);
            REQUIRE_INDICES(change.insertions, 10);
        }

        SECTION("modifying a different table does not send a change notification") {
            auto token = require_no_change();
            write([&] {
                other_lv->add(other_target_keys[0]);
            });
        }

        SECTION("changes are reported correctly for multiple tables") {
            List list2(r, *other_lv);
            CollectionChangeSet other_changes;
            auto token1 = list2.add_notification_callback([&](CollectionChangeSet c) {
                other_changes = std::move(c);
            });
            auto token2 = require_change();

            write([&] {
                lv->add(target_keys[1]);

                other_origin->create_object();
                if (other_lv->size() > 0)
                    other_lv->insert(1, other_target_keys[0]);

                lv->add(target_keys[2]);
            });
            REQUIRE_INDICES(change.insertions, 10, 11);
            REQUIRE_INDICES(other_changes.insertions, 1);

            write([&] {
                lv->add(target_keys[3]);
                other_obj.remove();
                lv->add(target_keys[4]);
            });
            REQUIRE_INDICES(change.insertions, 12, 13);
            REQUIRE_INDICES(other_changes.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

            write([&] {
                lv->add(target_keys[5]);
                other_origin->clear();
                lv->add(target_keys[6]);
            });
            REQUIRE_INDICES(change.insertions, 14, 15);
        }

        SECTION("tables-of-interest are tracked properly for multiple source versions") {
            // Add notifiers for different tables at different versions to verify
            // that the tables of interest are updated correctly as we process
            // new notifiers
            CollectionChangeSet changes1, changes2;
            auto token1 = lst.add_notification_callback([&](CollectionChangeSet c) {
                changes1 = std::move(c);
            });

            r2->begin_transaction();
            r2->read_group().get_table("class_target")->get_object(target_keys[0]).set(col_target_value, 10);
            r2->read_group()
                .get_table("class_other_target")
                ->get_object(other_target_keys[1])
                .set(other_col_value, 10);
            r2->commit_transaction();

            List list2(r2, r2->read_group().get_table("class_other_origin")->get_object(0), other_col_link);
            auto token2 = list2.add_notification_callback([&](CollectionChangeSet c) {
                changes2 = std::move(c);
            });

            auto r3 = coordinator.get_realm();
            r3->begin_transaction();
            r3->read_group().get_table("class_target")->get_object(target_keys[2]).set(col_target_value, 10);
            r3->read_group()
                .get_table("class_other_target")
                ->get_object(other_target_keys[3])
                .set(other_col_value, 10);
            r3->commit_transaction();

            advance_and_notify(*r);
            advance_and_notify(*r2);

            REQUIRE_INDICES(changes1.modifications, 0, 2);
            REQUIRE_INDICES(changes2.modifications, 3);
        }

        SECTION("modifications are reported for rows that are moved and then moved back in a second transaction") {
            auto token = require_change();

            r2->begin_transaction();
            r2_lv->get_object(5).set(col_target_value, 10);
            r2_lv->get_object(1).set(col_target_value, 10);
            r2_lv->move(5, 8);
            r2_lv->move(1, 2);
            r2->commit_transaction();

            coordinator.on_change();

            r2->begin_transaction();
            if (r2_lv->size() > 8)
                r2_lv->move(8, 5);
            r2->commit_transaction();
            advance_and_notify(*r);

            REQUIRE_INDICES(change.deletions, 1);
            REQUIRE_INDICES(change.insertions, 2);
            REQUIRE_INDICES(change.modifications, 5);
            REQUIRE_MOVES(change, {1, 2});
        }

        SECTION("changes are sent in initial notification") {
            auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            r2->begin_transaction();
            r2_lv->remove(5);
            r2->commit_transaction();
            advance_and_notify(*r);
            REQUIRE_INDICES(change.deletions, 5);
        }

        SECTION("changes are sent in initial notification after removing and then re-adding callback") {
            auto token = lst.add_notification_callback([&](CollectionChangeSet) {
                REQUIRE(false);
            });
            token = {};

            auto write = [&] {
                r2->begin_transaction();
                r2_lv->remove(5);
                r2->commit_transaction();
            };

            SECTION("add new callback before transaction") {
                token = lst.add_notification_callback([&](CollectionChangeSet c) {
                    change = c;
                });

                write();

                advance_and_notify(*r);
                REQUIRE_INDICES(change.deletions, 5);
            }

            SECTION("add new callback after transaction") {
                write();

                token = lst.add_notification_callback([&](CollectionChangeSet c) {
                    change = c;
                });

                advance_and_notify(*r);
                REQUIRE_INDICES(change.deletions, 5);
            }

            SECTION("add new callback after transaction and after changeset was calculated") {
                write();
                coordinator.on_change();

                token = lst.add_notification_callback([&](CollectionChangeSet c) {
                    change = c;
                });

                advance_and_notify(*r);
                REQUIRE_INDICES(change.deletions, 5);
            }
        }
    }

    SECTION("sorted add_notification_block()") {
        List lst(r, *lv);
        Results results = lst.sort({{{col_target_value}}, {false}});

        int notification_calls = 0;
        CollectionChangeSet change;
        auto token = results.add_notification_callback([&](CollectionChangeSet c) {
            change = c;
            ++notification_calls;
        });

        advance_and_notify(*r);

        SECTION("add duplicates") {
            write([&] {
                lst.add(target_keys[5]);
                lst.add(target_keys[5]);
                lst.add(target_keys[5]);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.insertions, 5, 6, 7);
        }

        SECTION("change order by modifying target") {
            write([&] {
                lst.get(5).set(col_target_value, 15);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.deletions, 4);
            REQUIRE_INDICES(change.insertions, 0);
        }

        SECTION("swap") {
            write([&] {
                lst.swap(1, 2);
            });
            REQUIRE(notification_calls == 1);
        }

        SECTION("move") {
            write([&] {
                lst.move(5, 3);
            });
            REQUIRE(notification_calls == 1);
        }
    }

    SECTION("filtered add_notification_block()") {
        List lst(r, *lv);
        Results results = lst.filter(target->where().less(col_target_value, 9));

        int notification_calls = 0;
        CollectionChangeSet change;
        auto token = results.add_notification_callback([&](CollectionChangeSet c) {
            change = c;
            ++notification_calls;
        });

        advance_and_notify(*r);

        SECTION("add duplicates") {
            write([&] {
                lst.add(target_keys[5]);
                lst.add(target_keys[5]);
                lst.add(target_keys[5]);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.insertions, 9, 10, 11);
        }

        SECTION("swap") {
            write([&] {
                lst.swap(1, 2);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.deletions, 2);
            REQUIRE_INDICES(change.insertions, 1);

            write([&] {
                lst.swap(5, 8);
            });
            REQUIRE(notification_calls == 3);
            REQUIRE_INDICES(change.deletions, 5, 8);
            REQUIRE_INDICES(change.insertions, 5, 8);
        }

        SECTION("move") {
            write([&] {
                lst.move(5, 3);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.deletions, 5);
            REQUIRE_INDICES(change.insertions, 3);
        }

        SECTION("move non-matching entry") {
            write([&] {
                lst.move(9, 3);
            });
            REQUIRE(notification_calls == 1);
        }
    }

    SECTION("Keypath filtered change notifications") {
        ColKey col_target_value2 = target->get_column_key("value2");
        List list(r, obj, col_link);

        // Creating KeyPathArrays:
        // 1. Property pairs
        std::pair<TableKey, ColKey> pair_origin_link(origin->get_key(), col_link);
        std::pair<TableKey, ColKey> pair_target_value(target->get_key(), col_target_value);
        std::pair<TableKey, ColKey> pair_target_value2(target->get_key(), col_target_value2);
        // 2. KeyPaths
        auto key_path_origin_link = {pair_origin_link};
        auto key_path_target_value = {pair_target_value};
        auto key_path_target_value2 = {pair_target_value2};
        // 3. Aggregated `KeyPathArray`
        KeyPathArray key_path_array_origin_to_target_value = {key_path_origin_link, key_path_target_value};
        KeyPathArray key_path_array_target_value = {key_path_target_value};
        KeyPathArray key_path_array_target_value2 = {key_path_target_value2};

        // For the keypath filtered notifications we need to check three scenarios:
        // - no callbacks have filters (this part is covered by other sections)
        // - some callbacks have filters
        // - all callbacks have filters
        CollectionChangeSet collection_change_set_without_filter;
        CollectionChangeSet collection_change_set_with_empty_filter;
        CollectionChangeSet collection_change_set_with_filter_on_target_value;

        // Note that in case not all callbacks have filters we do accept false positive notifications by design.
        // Distinguishing between these two cases would be a big change for little value.
        SECTION("some callbacks have filters") {
            auto require_change_no_filter = [&] {
                auto token = list.add_notification_callback([&](CollectionChangeSet c) {
                    collection_change_set_without_filter = c;
                });
                advance_and_notify(*r);
                return token;
            };

            auto require_change_target_value_filter = [&] {
                auto token = list.add_notification_callback(
                    [&](CollectionChangeSet c) {
                        collection_change_set_with_filter_on_target_value = c;
                    },
                    key_path_array_target_value);
                advance_and_notify(*r);
                return token;
            };

            SECTION("modifying table 'target', property 'value' "
                    "-> DOES send a notification") {
                auto token1 = require_change_no_filter();
                auto token2 = require_change_target_value_filter();
                write([&] {
                    list.get(0).set(col_target_value, 42);
                });
                REQUIRE_INDICES(collection_change_set_without_filter.modifications, 0);
                REQUIRE_INDICES(collection_change_set_without_filter.modifications_new, 0);
                REQUIRE_INDICES(collection_change_set_with_filter_on_target_value.modifications, 0);
                REQUIRE_INDICES(collection_change_set_with_filter_on_target_value.modifications_new, 0);
            }

            SECTION("modifying table 'target', property 'value2' "
                    "-> DOES send a notification") {
                auto token1 = require_change_no_filter();
                auto token2 = require_change_target_value_filter();
                write([&] {
                    list.get(0).set(col_target_value2, 42);
                });
                REQUIRE_INDICES(collection_change_set_without_filter.modifications, 0);
                REQUIRE_INDICES(collection_change_set_without_filter.modifications_new, 0);
                REQUIRE_INDICES(collection_change_set_with_filter_on_target_value.modifications, 0);
                REQUIRE_INDICES(collection_change_set_with_filter_on_target_value.modifications_new, 0);
            }
        }

        // In case all callbacks do have filters we expect every callback to only get called when the corresponding
        // filter is hit. Compared to the above 'some callbacks have filters' case we do not expect false positives
        // here.
        SECTION("all callbacks have filters") {
            auto require_change = [&] {
                auto token = list.add_notification_callback(
                    [&](CollectionChangeSet c) {
                        collection_change_set_with_filter_on_target_value = c;
                    },
                    key_path_array_target_value);
                advance_and_notify(*r);
                return token;
            };

            auto require_no_change = [&] {
                bool first = true;
                auto token = list.add_notification_callback(
                    [&, first](CollectionChangeSet) mutable {
                        REQUIRE(first);
                        first = false;
                    },
                    key_path_array_target_value2);
                advance_and_notify(*r);
                return token;
            };

            SECTION("modifying table 'target', property 'value' "
                    "-> DOES send a notification for 'value'") {
                auto token = require_change();
                write([&] {
                    list.get(0).set(col_target_value, 42);
                });
                REQUIRE_INDICES(collection_change_set_with_filter_on_target_value.modifications, 0);
                REQUIRE_INDICES(collection_change_set_with_filter_on_target_value.modifications_new, 0);
            }

            SECTION("modifying table 'target', property 'value' "
                    "-> does NOT send a notification for 'value'") {
                auto token = require_no_change();
                write([&] {
                    list.get(0).set(col_target_value, 42);
                });
            }
        }

        SECTION("callback with empty keypatharray") {
            auto shallow_require_change = [&] {
                auto token = list.add_notification_callback(
                    [&](CollectionChangeSet c) {
                        collection_change_set_with_empty_filter = c;
                    },
                    KeyPathArray());
                advance_and_notify(*r);
                return token;
            };

            auto shallow_require_no_change = [&] {
                bool first = true;
                auto token = list.add_notification_callback(
                    [&first](CollectionChangeSet) mutable {
                        REQUIRE(first);
                        first = false;
                    },
                    KeyPathArray());
                advance_and_notify(*r);
                return token;
            };

            SECTION("modifying table 'target', property 'value' "
                    "-> does NOT send a notification for 'value'") {
                auto token = shallow_require_no_change();
                write([&] {
                    list.get(0).set(col_target_value, 42);
                });
            }

            SECTION("modifying table 'target', property 'value' "
                    "-> does NOT send a notification for 'value'") {
                auto token = shallow_require_no_change();
                write([&] {
                    list.get(0).set(col_target_value, 42);
                });
            }

            SECTION("deleting a target row with shallow listener sends a change notification") {
                auto token = shallow_require_change();
                write([&] {
                    list.remove(5);
                });
                REQUIRE_INDICES(collection_change_set_with_empty_filter.deletions, 5);
            }

            SECTION("adding a row with shallow listener sends a change notifcation") {
                auto token = shallow_require_change();
                write([&] {
                    Obj obj = target->get_object(target_keys[5]);
                    list.add(obj);
                });
                REQUIRE_INDICES(collection_change_set_with_empty_filter.insertions, 10);
            }

            SECTION("moving a row with shallow listener sends a change notifcation") {
                auto token = shallow_require_change();
                write([&] {
                    list.move(5, 8);
                });
                REQUIRE_INDICES(collection_change_set_with_empty_filter.insertions, 8);
                REQUIRE_INDICES(collection_change_set_with_empty_filter.deletions, 5);
                REQUIRE_MOVES(collection_change_set_with_empty_filter, {5, 8});
            }
        }

        SECTION("linked filter") {
            CollectionChangeSet collection_change_set_linked_filter;
            Object object(r, obj);

            auto require_change_origin_to_target = [&] {
                auto token = object.add_notification_callback(
                    [&](CollectionChangeSet c) {
                        collection_change_set_linked_filter = c;
                    },
                    key_path_array_origin_to_target_value);
                advance_and_notify(*r);
                return token;
            };

            auto token = require_change_origin_to_target();

            write([&] {
                auto foo = obj.get_linklist(col_link);
                ObjKey obj_key = foo.get(0);
                TableRef target_table = foo.get_target_table();
                Obj target_object = target_table->get_object(obj_key);
                target_object.set(col_target_value, 42);
            });
            REQUIRE_INDICES(collection_change_set_linked_filter.modifications, 0);
            REQUIRE_INDICES(collection_change_set_linked_filter.modifications_new, 0);
        }
    }

    SECTION("sort()") {
        auto objectschema = &*r->schema().find("target");
        List list(r, *lv);
        auto results = list.sort({{{col_target_value}}, {false}});

        REQUIRE(&results.get_object_schema() == objectschema);
        REQUIRE(results.get_mode() == Results::Mode::Collection);
        REQUIRE(results.size() == 10);

        // Aggregates don't inherently have to convert to TableView, but do
        // because aggregates aren't implemented for Collection
        REQUIRE(results.sum(col_target_value) == 45);
        REQUIRE(results.get_mode() == Results::Mode::TableView);

        // Reset to Collection mode to test implicit conversion to TableView on get()
        results = list.sort({{{col_target_value}}, {false}});
        for (size_t i = 0; i < 10; ++i)
            REQUIRE(results.get(i).get_key() == target_keys[9 - i]);
        REQUIRE_EXCEPTION(results.get(10), OutOfBounds, "Requested index 10 calling get() on Results when max is 9");
        REQUIRE(results.get_mode() == Results::Mode::TableView);

        // Zero sort columns should leave it in Collection mode
        results = list.sort(SortDescriptor());
        for (size_t i = 0; i < 10; ++i)
            REQUIRE(results.get(i).get_key() == target_keys[i]);
        REQUIRE_EXCEPTION(results.get(10), OutOfBounds, "Requested index 10 calling get() on Results when max is 9");
        REQUIRE(results.get_mode() == Results::Mode::Collection);
    }

    SECTION("distinct()") {
        // Make it so that there's actually duplicate values in the target
        write([&] {
            for (int i = 0; i < 10; ++i)
                target->get_object(i).set_all(i / 2);
        });

        auto objectschema = &*r->schema().find("target");
        List list(r, *lv);
        auto results = list.as_results().distinct(DistinctDescriptor({{col_target_value}}));
        REQUIRE(&results.get_object_schema() == objectschema);
        REQUIRE(results.get_mode() == Results::Mode::Collection);

        SECTION("size()") {
            REQUIRE(results.size() == 5);
        }

        SECTION("aggregates") {
            REQUIRE(results.sum(col_target_value) == 10);
        }

        SECTION("get()") {
            for (size_t i = 0; i < 5; ++i)
                REQUIRE(results.get(i).get_key() == target_keys[i * 2]);
            REQUIRE_EXCEPTION(results.get(5), OutOfBounds,
                              "Requested index 5 calling get() on Results when max is 4");
            REQUIRE(results.get_mode() == Results::Mode::TableView);
        }

        SECTION("clear()") {
            REQUIRE(target->size() == 10);
            write([&] {
                results.clear();
            });
            REQUIRE(target->size() == 5);

            // After deleting the first object with each distinct value, the
            // results should now contain the second object with each distinct
            // value (which in this case means that the size hasn't changed)
            REQUIRE(results.size() == 5);
            for (size_t i = 0; i < 5; ++i)
                REQUIRE(results.get(i).get_key() == target_keys[(i + 1) * 2 - 1]);
        }

        SECTION("empty distinct descriptor does nothing") {
            results = list.as_results().distinct(DistinctDescriptor());
            for (size_t i = 0; i < 10; ++i)
                REQUIRE(results.get(i).get_key() == target_keys[i]);
            REQUIRE_EXCEPTION(results.get(10), OutOfBounds,
                              "Requested index 10 calling get() on Results when max is 9");
            REQUIRE(results.get_mode() == Results::Mode::Collection);
        }
    }

    SECTION("filter()") {
        auto objectschema = &*r->schema().find("target");
        List list(r, *lv);
        auto results = list.filter(target->where().greater(col_target_value, 5));

        REQUIRE(&results.get_object_schema() == objectschema);
        REQUIRE(results.get_mode() == Results::Mode::Query);
        REQUIRE(results.size() == 4);

        for (size_t i = 0; i < 4; ++i) {
            REQUIRE(results.get(i).get_key() == target_keys[i + 6]);
        }
    }

    SECTION("snapshot()") {
        auto objectschema = &*r->schema().find("target");
        List list(r, *lv);

        auto snapshot = list.snapshot();
        REQUIRE(&snapshot.get_object_schema() == objectschema);
        REQUIRE(snapshot.get_mode() == Results::Mode::TableView);
        REQUIRE(snapshot.size() == 10);

        r->begin_transaction();
        for (size_t i = 0; i < 5; ++i) {
            list.remove(0);
        }
        REQUIRE(snapshot.size() == 10);
        for (size_t i = 0; i < snapshot.size(); ++i) {
            REQUIRE(snapshot.get(i).is_valid());
        }
        for (size_t i = 0; i < 5; ++i) {
            target->remove_object(target_keys[i]);
        }
        REQUIRE(snapshot.size() == 10);
        for (size_t i = 0; i < 5; ++i) {
            REQUIRE(!snapshot.get(i).is_valid());
        }
        for (size_t i = 5; i < 10; ++i) {
            REQUIRE(snapshot.get(i).is_valid());
        }
        list.add(target_keys[5]);
        REQUIRE(snapshot.size() == 10);
    }

    SECTION("snapshot() after deletion") {
        List list(r, *lv);

        auto snapshot = list.snapshot();

        for (size_t i = 0; i < snapshot.size(); ++i) {
            r->begin_transaction();
            Obj obj = snapshot.get<Obj>(i);
            obj.remove();
            r->commit_transaction();
        }

        auto snapshot2 = list.snapshot();
        CHECK(snapshot2.size() == 0);
        CHECK(list.size() == 0);
    }

    SECTION("get_object_schema()") {
        List list(r, *lv);
        auto objectschema = &*r->schema().find("target");
        REQUIRE(&list.get_object_schema() == objectschema);
    }

    SECTION("delete_at()") {
        List list(r, *lv);
        r->begin_transaction();
        auto initial_view_size = lv->size();
        auto initial_target_size = target->size();
        list.delete_at(1);
        REQUIRE(lv->size() == initial_view_size - 1);
        REQUIRE(target->size() == initial_target_size - 1);
        r->cancel_transaction();
    }

    SECTION("delete_all()") {
        List list(r, *lv);
        r->begin_transaction();
        list.delete_all();
        REQUIRE(lv->size() == 0);
        REQUIRE(target->size() == 0);
        r->cancel_transaction();
    }

    SECTION("as_results().clear()") {
        List list(r, *lv);
        r->begin_transaction();
        list.as_results().clear();
        REQUIRE(lv->size() == 0);
        REQUIRE(target->size() == 0);
        r->cancel_transaction();
    }

    SECTION("snapshot().clear()") {
        List list(r, *lv);
        r->begin_transaction();
        auto snapshot = list.snapshot();
        snapshot.clear();
        REQUIRE(snapshot.size() == 10);
        REQUIRE(list.size() == 0);
        REQUIRE(lv->size() == 0);
        REQUIRE(target->size() == 0);
        r->cancel_transaction();
    }

    SECTION("add(RowExpr)") {
        List list(r, *lv);
        r->begin_transaction();
        SECTION("adds rows from the correct table") {
            list.add(target_keys[5]);
            REQUIRE(list.size() == 11);
            REQUIRE(list.get(10).get_key() == target_keys[5]);
        }

        SECTION("throws for rows from the wrong table") {
            REQUIRE_EXCEPTION(list.add(obj), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
        }
        r->cancel_transaction();
    }

    SECTION("insert(RowExpr)") {
        List list(r, *lv);
        r->begin_transaction();

        SECTION("insert rows from the correct table") {
            list.insert(0, target_keys[5]);
            REQUIRE(list.size() == 11);
            REQUIRE(list.get(0).get_key() == target_keys[5]);
        }

        SECTION("throws for rows from the wrong table") {
            REQUIRE_EXCEPTION(list.insert(0, obj), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
        }

        SECTION("throws for out of bounds insertions") {
            REQUIRE_EXCEPTION(list.insert(11, target_keys[5]), OutOfBounds,
                              "Requested index 11 calling insert() on list 'origin.array' when max is 10");
            REQUIRE_NOTHROW(list.insert(10, target_keys[5]));
        }
        r->cancel_transaction();
    }

    SECTION("set(RowExpr)") {
        List list(r, *lv);
        r->begin_transaction();

        SECTION("assigns for rows from the correct table") {
            list.set(0, target_keys[5]);
            REQUIRE(list.size() == 10);
            REQUIRE(list.get(0).get_key() == target_keys[5]);
        }

        SECTION("throws for rows from the wrong table") {
            REQUIRE_EXCEPTION(list.set(0, obj), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
        }

        SECTION("throws for out of bounds sets") {
            REQUIRE_EXCEPTION(list.set(10, target_keys[5]), OutOfBounds,
                              "Requested index 10 calling set() on list 'origin.array' when max is 9");
        }
        r->cancel_transaction();
    }

    SECTION("find(RowExpr)") {
        List list(r, *lv);
        Obj obj1 = target->get_object(target_keys[1]);
        Obj obj5 = target->get_object(target_keys[5]);

        SECTION("returns index in list for values in the list") {
            REQUIRE(list.find(obj5) == 5);
        }

        SECTION("returns index in list and not index in table") {
            r->begin_transaction();
            list.remove(1);
            REQUIRE(list.find(obj5) == 4);
            REQUIRE(list.as_results().index_of(obj5) == 4);
            r->cancel_transaction();
        }

        SECTION("returns npos for values not in the list") {
            r->begin_transaction();
            list.remove(1);
            REQUIRE(list.find(obj1) == npos);
            REQUIRE(list.as_results().index_of(obj1) == npos);
            r->cancel_transaction();
        }

        SECTION("throws for row in wrong table") {
            REQUIRE_EXCEPTION(list.find(obj), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
            REQUIRE_EXCEPTION(list.as_results().index_of(obj), ObjectTypeMismatch,
                              "Object of type 'origin' does not match Results type 'target'");
        }
    }

    SECTION("find(Query)") {
        List list(r, *lv);

        SECTION("returns index in list for values in the list") {
            REQUIRE(list.find(std::move(target->where().equal(col_target_value, 5))) == 5);
        }

        SECTION("returns index in list and not index in table") {
            r->begin_transaction();
            list.remove(1);
            REQUIRE(list.find(std::move(target->where().equal(col_target_value, 5))) == 4);
            r->cancel_transaction();
        }

        SECTION("returns npos for values not in the list") {
            REQUIRE(list.find(std::move(target->where().equal(col_target_value, 11))) == npos);
        }
    }

    SECTION("add(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());
        r->begin_transaction();

        SECTION("adds boxed RowExpr") {
            list.add(ctx, std::any(target->get_object(target_keys[5])));
            REQUIRE(list.size() == 11);
            REQUIRE(list.get(10).get_key().value == 5);
        }

        SECTION("adds boxed realm::Object") {
            realm::Object obj(r, list.get_object_schema(), target->get_object(target_keys[5]));
            list.add(ctx, std::any(obj));
            REQUIRE(list.size() == 11);
            REQUIRE(list.get(10).get_key() == target_keys[5]);
        }

        SECTION("creates new object for dictionary") {
            list.add(ctx, std::any(AnyDict{{"value", INT64_C(20)}, {"value2", INT64_C(20)}}));
            REQUIRE(list.size() == 11);
            REQUIRE(target->size() == 11);
            REQUIRE(list.get(10).get<Int>(col_target_value) == 20);
        }

        SECTION("throws for object in wrong table") {
            REQUIRE_EXCEPTION(list.add(ctx, std::any(origin->get_object(0))), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
            realm::Object object(r, *r->schema().find("origin"), origin->get_object(0));
            REQUIRE_EXCEPTION(list.add(ctx, std::any(object)), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
        }

        r->cancel_transaction();
    }

    SECTION("find(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());

        SECTION("returns index in list for boxed RowExpr") {
            REQUIRE(list.find(ctx, std::any(target->get_object(target_keys[5]))) == 5);
        }

        SECTION("returns index in list for boxed Object") {
            realm::Object obj(r, *r->schema().find("origin"), target->get_object(target_keys[5]));
            REQUIRE(list.find(ctx, std::any(obj)) == 5);
        }

        SECTION("does not insert new objects for dictionaries") {
            REQUIRE(list.find(ctx, std::any(AnyDict{{"value", INT64_C(20)}})) == npos);
            REQUIRE(target->size() == 10);
        }

        SECTION("throws for object in wrong table") {
            REQUIRE_EXCEPTION(list.find(ctx, std::any(obj)), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
        }
    }

    SECTION("get(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());

        Object obj;
        REQUIRE_NOTHROW(obj = util::any_cast<Object&&>(list.get(ctx, 1)));
        REQUIRE(obj.is_valid());
        REQUIRE(obj.get_obj().get_key() == target_keys[1]);
    }
}

TEST_CASE("nested List") {
    InMemoryTestFile config;
    config.automatic_change_notifications = false;
    auto r = Realm::get_shared_realm(config);
    r->update_schema({
        {"table",
         {{"pk", PropertyType::Int, Property::IsPrimary{true}},
          {"any", PropertyType::Mixed | PropertyType::Nullable}}},
    });

    auto& coordinator = *_impl::RealmCoordinator::get_coordinator(config.path);

    auto table = r->read_group().get_table("class_table");
    ColKey col_any = table->get_column_key("any");

    r->begin_transaction();

    Obj obj = table->create_object_with_primary_key(47);
    obj.set_collection(col_any, CollectionType::List);
    auto top_list = obj.get_list<Mixed>(col_any);
    top_list.insert(0, "Hello");
    top_list.insert_collection(1, CollectionType::List);
    top_list.insert(2, "Godbye");
    top_list.insert_collection(3, CollectionType::List);
    top_list.insert_collection(4, CollectionType::List);
    auto l0 = obj.get_list_ptr<Mixed>(Path{"any", 1});
    l0->insert_collection(0, CollectionType::Dictionary);
    auto d = l0->get_dictionary(0);
    d->insert_collection("list", CollectionType::List);
    d->get_list("list")->add(Mixed(5));
    auto l1 = obj.get_list_ptr<Mixed>(Path{"any", 3});

    r->commit_transaction();

    auto r2 = coordinator.get_realm();

    auto write = [&](auto&& f) {
        r->begin_transaction();
        f();
        r->commit_transaction();
        advance_and_notify(*r);
    };

    SECTION("add_notification_block()") {
        CollectionChangeSet change;
        List lst0(r, l0);
        List lst1(r, l1);

        auto require_change = [&] {
            auto token = lst0.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            advance_and_notify(*r);
            return token;
        };

        auto require_no_change = [&] {
            bool first = true;
            auto token = lst0.add_notification_callback([&, first](CollectionChangeSet) mutable {
                REQUIRE(first);
                first = false;
            });
            advance_and_notify(*r);
            return token;
        };

        SECTION("modifying the list sends a change notifications") {
            auto token = require_change();
            write([&] {
                lst0.add(Mixed(8));
            });
            REQUIRE_INDICES(change.insertions, 1);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("inserting in sub structure sends a change notifications") {
            // Check that notifications on Results are correct
            Results res = lst0.as_results();
            CollectionChangeSet change1;
            auto token_res = res.add_notification_callback([&](CollectionChangeSet c) {
                change1 = c;
            });

            auto token = require_change();
            write([&] {
                lst0.get_dictionary(0).get_list("list").add(Mixed(42));
            });
            REQUIRE_INDICES(change.modifications, 0);
            REQUIRE_INDICES(change1.modifications, 0);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("modifying in sub structure sends a change notifications") {
            auto token = require_change();
            write([&] {
                lst0.get_dictionary(0).get_list("list").set(0, Mixed(42));
            });
            REQUIRE_INDICES(change.modifications, 0);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("clearing in sub structure sends a change notifications") {
            auto token = require_change();
            write([&] {
                lst0.get_dictionary(0).get_list("list").remove_all();
            });
            REQUIRE_INDICES(change.modifications, 0);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("deleting sub structure sends a change notifications") {
            auto token = require_change();
            write([&] {
                lst0.get_dictionary(0).erase("list");
            });
            REQUIRE_INDICES(change.modifications, 0);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("creating and modifying sub structure results in insert change only") {
            auto token = require_change();
            write([&] {
                lst0.insert_collection(1, CollectionType::Dictionary);
                lst0.get_dictionary(1).insert("Value", Mixed(42));
            });
            REQUIRE_INDICES(change.insertions, 1);
            REQUIRE(change.modifications.empty());
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("modifying another list does not send notifications") {
            auto token = require_no_change();
            write([&] {
                lst1.add(Mixed(47));
            });
        }

        SECTION("modifying the list sends a change notifications - even when index changes") {
            auto token = require_change();
            write([&] {
                obj.get_collection_ptr(col_any)->insert_collection(0, CollectionType::List);
                lst0.add(Mixed(8));
            });
            REQUIRE_INDICES(change.insertions, 1);
            REQUIRE(!change.collection_was_cleared);
        }

        SECTION("a notifier can be attached in a different transaction") {
            {
                r2->begin_transaction();
                auto t = r2->read_group().get_table("class_table");
                auto l = t->get_object_with_primary_key(47).get_list<Mixed>("any");
                l.remove(0);
                r2->commit_transaction();
            }

            auto token = require_change();
            write([&] {
                lst0.add(Mixed(8));
            });
            REQUIRE_INDICES(change.insertions, 1);
            REQUIRE(!change.collection_was_cleared);
        }
        SECTION("remove item from collection") {
            auto token = require_change();
            write([&] {
                lst0.add(Mixed(8));
            });
            REQUIRE_INDICES(change.insertions, 1);
            write([&] {
                lst0.remove(1);
            });
            REQUIRE_INDICES(change.deletions, 1);
        }
        SECTION("erase from containing list") {
            auto token = require_change();
            write([&] {
                lst0.add(Mixed(8));
            });
            REQUIRE_INDICES(change.insertions, 1);
            write([&] {
                top_list.set(1, 42);
            });
            REQUIRE_INDICES(change.deletions, 0, 1);
            REQUIRE(change.collection_root_was_deleted);
        }
        SECTION("remove containing object") {
            auto token = require_change();
            write([&] {
                lst0.add(Mixed(8));
            });
            REQUIRE_INDICES(change.insertions, 1);
            write([&] {
                obj.remove();
            });
            REQUIRE_INDICES(change.deletions, 0, 1);
            REQUIRE(change.collection_root_was_deleted);
        }
    }
}

TEST_CASE("embedded List", "[list]") {
    InMemoryTestFile config;
    config.automatic_change_notifications = false;
    auto r = Realm::get_shared_realm(config);
    r->update_schema({
        {"origin",
         {{"pk", PropertyType::Int, Property::IsPrimary{true}},
          {"array", PropertyType::Array | PropertyType::Object, "target"}}},
        {"target", ObjectSchema::ObjectType::Embedded, {{"value", PropertyType::Int}}},
        {"other_origin",
         {{"id", PropertyType::Int, Property::IsPrimary{true}},
          {"array", PropertyType::Array | PropertyType::Object, "other_target"}}},
        {"other_target", ObjectSchema::ObjectType::Embedded, {{"value", PropertyType::Int}}},
    });

    auto& coordinator = *_impl::RealmCoordinator::get_coordinator(config.path);

    auto origin = r->read_group().get_table("class_origin");
    auto target = r->read_group().get_table("class_target");
    auto other_origin = r->read_group().get_table("class_other_origin");
    ColKey col_link = origin->get_column_key("array");
    ColKey col_value = target->get_column_key("value");
    ColKey other_col_link = other_origin->get_column_key("array");

    r->begin_transaction();

    Obj obj = origin->create_object_with_primary_key(0);
    auto lv = obj.get_linklist_ptr(col_link);
    for (int i = 0; i < 10; ++i)
        lv->create_and_insert_linked_object(i).set_all(i);
    auto lv2 = origin->create_object_with_primary_key(1).get_linklist_ptr(col_link);
    for (int i = 0; i < 10; ++i)
        lv2->create_and_insert_linked_object(i).set_all(i);


    Obj other_obj = other_origin->create_object_with_primary_key(1);
    auto other_lv = other_obj.get_linklist_ptr(other_col_link);
    for (int i = 0; i < 10; ++i)
        other_lv->create_and_insert_linked_object(i).set_all(i);

    r->commit_transaction();
    lv->size();
    lv2->size();
    other_lv->size();

    auto r2 = coordinator.get_realm();
    auto r2_lv = r2->read_group().get_table("class_origin")->get_object(0).get_linklist_ptr(col_link);

    auto write = [&](auto&& f) {
        r->begin_transaction();
        f();
        r->commit_transaction();
        advance_and_notify(*r);
    };

    SECTION("add_notification_block()") {
        CollectionChangeSet change;
        List lst(r, obj, col_link);

        auto require_change = [&] {
            auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            advance_and_notify(*r);
            return token;
        };

        auto require_no_change = [&] {
            bool first = true;
            auto token = lst.add_notification_callback([&, first](CollectionChangeSet) mutable {
                REQUIRE(first);
                first = false;
            });
            advance_and_notify(*r);
            return token;
        };

        SECTION("modifying the list sends a change notifications") {
            auto token = require_change();
            write([&] {
                lst.remove(5);
            });
            REQUIRE_INDICES(change.deletions, 5);
        }

        SECTION("modifying a different list doesn't send a change notification") {
            auto token = require_no_change();
            write([&] {
                lv2->remove(5);
            });
        }

        SECTION("deleting the list sends a change notification") {
            auto token = require_change();
            write([&] {
                obj.remove();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);

            // Should not resend delete all notification after another commit
            change = {};
            write([&] {
                lv2->size();
                lv2->create_and_insert_linked_object(0);
            });
            REQUIRE(change.empty());
        }

        SECTION("deleting list before first run of notifier reports deletions") {
            auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
                change = c;
            });
            advance_and_notify(*r);
            write([&] {
                origin->begin()->remove();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
        }

        SECTION("modifying one of the target rows sends a change notification") {
            auto token = require_change();
            write([&] {
                lst.get(5).set(col_value, 6);
            });
            REQUIRE_INDICES(change.modifications, 5);
        }

        SECTION("deleting a target row sends a change notification") {
            auto token = require_change();
            write([&] {
                target->remove_object(lv->get(5));
            });
            REQUIRE_INDICES(change.deletions, 5);
        }

        SECTION("modifying and then moving a row reports move/insert but not modification") {
            auto token = require_change();
            write([&] {
                target->get_object(lv->get(5)).set(col_value, 10);
                lst.move(5, 8);
            });
            REQUIRE_INDICES(change.insertions, 8);
            REQUIRE_INDICES(change.deletions, 5);
            REQUIRE_MOVES(change, {5, 8});
            REQUIRE(change.modifications.empty());
        }

        SECTION("clearing the target table sends a change notification") {
            auto token = require_change();
            write([&] {
                target->clear();
            });
            REQUIRE_INDICES(change.deletions, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
        }
    }

    SECTION("sorted add_notification_block()") {
        List lst(r, *lv);
        Results results = lst.sort({{{col_value}}, {false}});

        int notification_calls = 0;
        CollectionChangeSet change;
        auto token = results.add_notification_callback([&](CollectionChangeSet c) {
            change = c;
            ++notification_calls;
        });

        advance_and_notify(*r);

        SECTION("change order by modifying target") {
            write([&] {
                lst.get(5).set(col_value, 15);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.deletions, 4);
            REQUIRE_INDICES(change.insertions, 0);
        }

        SECTION("swap") {
            write([&] {
                lst.swap(1, 2);
            });
            REQUIRE(notification_calls == 1);
        }

        SECTION("move") {
            write([&] {
                lst.move(5, 3);
            });
            REQUIRE(notification_calls == 1);
        }
    }

    SECTION("filtered add_notification_block()") {
        List lst(r, *lv);
        Results results = lst.filter(target->where().less(col_value, 9));

        int notification_calls = 0;
        CollectionChangeSet change;
        auto token = results.add_notification_callback([&](CollectionChangeSet c) {
            change = c;
            ++notification_calls;
        });

        advance_and_notify(*r);

        SECTION("swap") {
            write([&] {
                lst.swap(1, 2);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.deletions, 2);
            REQUIRE_INDICES(change.insertions, 1);

            write([&] {
                lst.swap(5, 8);
            });
            REQUIRE(notification_calls == 3);
            REQUIRE_INDICES(change.deletions, 5, 8);
            REQUIRE_INDICES(change.insertions, 5, 8);
        }

        SECTION("move") {
            write([&] {
                lst.move(5, 3);
            });
            REQUIRE(notification_calls == 2);
            REQUIRE_INDICES(change.deletions, 5);
            REQUIRE_INDICES(change.insertions, 3);
        }

        SECTION("move non-matching entry") {
            write([&] {
                lst.move(9, 3);
            });
            REQUIRE(notification_calls == 1);
        }
    }

    auto initial_view_size = lv->size();
    auto initial_target_size = target->size();
    SECTION("delete_at()") {
        List list(r, *lv);
        r->begin_transaction();
        list.delete_at(1);
        REQUIRE(lv->size() == initial_view_size - 1);
        REQUIRE(target->size() == initial_target_size - 1);
        r->cancel_transaction();
    }

    SECTION("delete_all()") {
        List list(r, *lv);
        r->begin_transaction();
        list.delete_all();
        REQUIRE(lv->size() == 0);
        REQUIRE(target->size() == initial_target_size - 10);
        r->cancel_transaction();
    }

    SECTION("as_results().clear()") {
        List list(r, *lv);
        r->begin_transaction();
        list.as_results().clear();
        REQUIRE(lv->size() == 0);
        REQUIRE(target->size() == initial_target_size - 10);
        r->cancel_transaction();
    }

    SECTION("snapshot().clear()") {
        List list(r, *lv);
        r->begin_transaction();
        auto snapshot = list.snapshot();
        snapshot.clear();
        REQUIRE(snapshot.size() == 10);
        REQUIRE(list.size() == 0);
        REQUIRE(lv->size() == 0);
        REQUIRE(target->size() == initial_target_size - 10);
        r->cancel_transaction();
    }

    SECTION("add(), insert(), and set() to existing object is not allowed") {
        List list(r, *lv);
        r->begin_transaction();
        REQUIRE_EXCEPTION(list.add(target->get_object(0)), IllegalOperation,
                          "Cannot insert an already managed object into list of embedded objects 'origin.array'");
        REQUIRE_EXCEPTION(list.insert(0, target->get_object(0)), IllegalOperation,
                          "Cannot insert an already managed object into list of embedded objects 'origin.array'");
        REQUIRE_EXCEPTION(list.set(0, target->get_object(0)), IllegalOperation,
                          "Cannot insert an already managed object into list of embedded objects 'origin.array'");
        r->cancel_transaction();
    }

    SECTION("find(RowExpr)") {
        List list(r, *lv);
        Obj obj1 = target->get_object(1);
        Obj obj5 = target->get_object(5);

        SECTION("returns index in list for values in the list") {
            REQUIRE(list.find(obj5) == 5);
        }

        SECTION("returns index in list and not index in table") {
            r->begin_transaction();
            list.remove(1);
            REQUIRE(list.find(obj5) == 4);
            REQUIRE(list.as_results().index_of(obj5) == 4);
            r->cancel_transaction();
        }

        SECTION("returns npos for values not in the list") {
            r->begin_transaction();
            list.remove(1);
            REQUIRE(list.find(obj1) == npos);
            REQUIRE_EXCEPTION(list.as_results().index_of(obj1), StaleAccessor,
                              "Attempting to access an invalid object");
            r->cancel_transaction();
        }

        SECTION("throws for row in wrong table") {
            REQUIRE_EXCEPTION(list.find(obj), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
            REQUIRE_EXCEPTION(list.as_results().index_of(obj), ObjectTypeMismatch,
                              "Object of type 'origin' does not match Results type 'target'");
        }
    }

    SECTION("find(Query)") {
        List list(r, *lv);

        SECTION("returns index in list for values in the list") {
            REQUIRE(list.find(std::move(target->where().equal(col_value, 5))) == 5);
        }

        SECTION("returns index in list and not index in table") {
            r->begin_transaction();
            list.remove(1);
            REQUIRE(list.find(std::move(target->where().equal(col_value, 5))) == 4);
            r->cancel_transaction();
        }

        SECTION("returns npos for values not in the list") {
            REQUIRE(list.find(std::move(target->where().equal(col_value, 11))) == npos);
        }
    }

    SECTION("add(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());
        r->begin_transaction();

        auto initial_target_size = target->size();
        SECTION("rejects boxed Obj and Object") {
            REQUIRE_THROW_LOGIC_ERROR_WITH_CODE(list.add(ctx, std::any(target->get_object(5))),
                                                ErrorCodes::IllegalOperation);
            REQUIRE_THROW_LOGIC_ERROR_WITH_CODE(list.add(ctx, std::any(Object(r, target->get_object(5)))),
                                                ErrorCodes::IllegalOperation);
        }

        SECTION("creates new object for dictionary") {
            list.add(ctx, std::any(AnyDict{{"value", INT64_C(20)}}));
            REQUIRE(list.size() == 11);
            REQUIRE(target->size() == initial_target_size + 1);
            REQUIRE(list.get(10).get<Int>(col_value) == 20);
        }

        r->cancel_transaction();
    }

    SECTION("set(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());
        r->begin_transaction();

        auto initial_target_size = target->size();
        SECTION("rejects boxed Obj and Object") {
            REQUIRE_THROW_LOGIC_ERROR_WITH_CODE(list.set(ctx, 0, std::any(target->get_object(5))),
                                                ErrorCodes::IllegalOperation);
            REQUIRE_THROW_LOGIC_ERROR_WITH_CODE(list.set(ctx, 0, std::any(Object(r, target->get_object(5)))),
                                                ErrorCodes::IllegalOperation);
        }

        SECTION("creates new object for update mode All") {
            auto old_object = list.get<Obj>(0);
            list.set(ctx, 0, std::any(AnyDict{{"value", INT64_C(20)}}));
            REQUIRE(list.size() == 10);
            REQUIRE(target->size() == initial_target_size);
            REQUIRE(list.get(0).get<Int>(col_value) == 20);
            REQUIRE_FALSE(old_object.is_valid());
        }

        SECTION("mutates the existing object for update mode Modified") {
            auto old_object = list.get<Obj>(0);
            list.set(ctx, 0, std::any(AnyDict{{"value", INT64_C(20)}}), CreatePolicy::UpdateModified);
            REQUIRE(list.size() == 10);
            REQUIRE(target->size() == initial_target_size);
            REQUIRE(list.get(0).get<Int>(col_value) == 20);
            REQUIRE(old_object.is_valid());
            REQUIRE(list.get(0) == old_object);
        }

        r->cancel_transaction();
    }

    SECTION("find(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());

        SECTION("returns index in list for boxed Obj") {
            REQUIRE(list.find(ctx, std::any(list.get(5))) == 5);
        }

        SECTION("returns index in list for boxed Object") {
            realm::Object obj(r, *r->schema().find("origin"), list.get(5));
            REQUIRE(list.find(ctx, std::any(obj)) == 5);
        }

        SECTION("does not insert new objects for dictionaries") {
            auto initial_target_size = target->size();
            REQUIRE(list.find(ctx, std::any(AnyDict{{"value", INT64_C(20)}})) == npos);
            REQUIRE(target->size() == initial_target_size);
        }

        SECTION("throws for object in wrong table") {
            REQUIRE_EXCEPTION(list.find(ctx, std::any(obj)), ObjectTypeMismatch,
                              "Object of type (origin) does not match List type (target)");
        }
    }

    SECTION("get(Context)") {
        List list(r, *lv);
        CppContext ctx(r, &list.get_object_schema());

        Object obj;
        REQUIRE_NOTHROW(obj = util::any_cast<Object&&>(list.get(ctx, 1)));
        REQUIRE(obj.is_valid());
        REQUIRE(obj.get_obj().get<int64_t>(col_value) == 1);
    }
}


TEST_CASE("list of embedded objects", "[list]") {
    Schema schema{
        {"parent",
         {
             {"array", PropertyType::Object | PropertyType::Array, "embedded"},
         }},
        {"embedded",
         ObjectSchema::ObjectType::Embedded,
         {
             {"value", PropertyType::Int},
         }},
    };

    InMemoryTestFile config;
    config.automatic_change_notifications = false;
    config.schema_mode = SchemaMode::Automatic;
    config.schema = schema;
    auto realm = Realm::get_shared_realm(config);
    auto parent_table = realm->read_group().get_table("class_parent");
    ColKey col_array = parent_table->get_column_key("array");
    auto embedded_table = realm->read_group().get_table("class_embedded");
    ColKey col_value = embedded_table->get_column_key("value");
    realm->begin_transaction();
    auto parent = parent_table->create_object();
    realm->commit_transaction();

    auto list = List(realm, parent, col_array);

    auto add_two_elements = [&] {
        auto first = list.add_embedded();
        first.set(col_value, 1);

        auto second = list.add_embedded();
        second.set(col_value, 2);
    };

    auto insert_three_elements = [&] {
        // Insert at position 0, shifting all elements back
        auto beginning = list.insert_embedded(0);
        beginning.set(col_value, 0);

        // Insert at position 2, so it's between the originally inserted items
        auto middle = list.insert_embedded(2);
        middle.set(col_value, 10);

        // Insert at the end of the list (i.e. list.size())
        auto end = list.insert_embedded(4);
        end.set(col_value, 20);
    };

    SECTION("add to list") {
        realm->begin_transaction();
        add_two_elements();
        realm->commit_transaction();

        REQUIRE(list.size() == 2);
        REQUIRE(list.get(0).get<int64_t>(col_value) == 1);
        REQUIRE(list.get(1).get<int64_t>(col_value) == 2);
    }

    SECTION("insert in list") {
        realm->begin_transaction();
        add_two_elements();
        insert_three_elements();
        realm->commit_transaction();

        REQUIRE(list.size() == 5);
        REQUIRE(list.get(0).get<int64_t>(col_value) == 0);  // inserted beginning
        REQUIRE(list.get(1).get<int64_t>(col_value) == 1);  // added first
        REQUIRE(list.get(2).get<int64_t>(col_value) == 10); // inserted middle
        REQUIRE(list.get(3).get<int64_t>(col_value) == 2);  // added second
        REQUIRE(list.get(4).get<int64_t>(col_value) == 20); // inserted end
    }

    SECTION("set in list") {
        realm->begin_transaction();

        add_two_elements();
        insert_three_elements();

        auto originalAt2 = list.get(2);
        auto newAt2 = list.set_embedded(2);
        newAt2.set(col_value, 100);

        realm->commit_transaction();

        REQUIRE(originalAt2.is_valid() == false);
        REQUIRE(newAt2.is_valid() == true);

        REQUIRE(list.size() == 5);
        REQUIRE(list.get(0).get<int64_t>(col_value) == 0);   // inserted at beginning
        REQUIRE(list.get(1).get<int64_t>(col_value) == 1);   // added first
        REQUIRE(list.get(2).get<int64_t>(col_value) == 100); // set at 2
        REQUIRE(list.get(3).get<int64_t>(col_value) == 2);   // added second
        REQUIRE(list.get(4).get<int64_t>(col_value) == 20);  // inserted at end
    }

    SECTION("invalid indices") {
        realm->begin_transaction();
        // Insertions
        REQUIRE_EXCEPTION(
            list.insert_embedded(-1), OutOfBounds, // Negative
            util::format("Requested index %1 calling insert() on list 'parent.array' when max is 0", size_t(-1)));
        REQUIRE_EXCEPTION(list.insert_embedded(1), OutOfBounds, // At index > size()
                          "Requested index 1 calling insert() on list 'parent.array' when max is 0");

        // Sets
        REQUIRE_EXCEPTION(
            list.set_embedded(-1), OutOfBounds, // Negative
            util::format("Requested index %1 calling set() on list 'parent.array' when empty", size_t(-1)));
        REQUIRE_EXCEPTION(list.set_embedded(0), OutOfBounds, // At index == size()
                          "Requested index 0 calling set() on list 'parent.array' when empty");
        REQUIRE_EXCEPTION(list.set_embedded(1), OutOfBounds, // At index > size()
                          "Requested index 1 calling set() on list 'parent.array' when empty");
        realm->cancel_transaction();
    }
}

#if REALM_ENABLE_SYNC

TEST_CASE("list with unresolved links", "[list]") {
    TestSyncManager init_sync_manager({}, {false});
    auto& server = init_sync_manager.sync_server();

    SyncTestFile config1(init_sync_manager, "shared");
    config1.schema = Schema{
        {"origin",
         {{"_id", PropertyType::Int, Property::IsPrimary(true)},
          {"array", PropertyType::Array | PropertyType::Object, "target"}}},
        {"target", {{"_id", PropertyType::Int, Property::IsPrimary(true)}, {"value", PropertyType::Int}}},
    };

    SyncTestFile config2(init_sync_manager, "shared");

    auto r1 = Realm::get_shared_realm(config1);
    auto r2 = Realm::get_shared_realm(config2);

    auto origin = r1->read_group().get_table("class_origin");
    auto target = r1->read_group().get_table("class_target");
    ColKey col_link = origin->get_column_key("array");
    ColKey col_target_value = target->get_column_key("value");

    r1->begin_transaction();

    std::vector<ObjKey> target_keys;
    for (int64_t i = 0; i < 11; ++i) {
        target_keys.push_back(target->create_object_with_primary_key(i).set(col_target_value, i).get_key());
    }
    auto origin_obj = origin->create_object_with_primary_key(100);
    auto ll = origin_obj.get_linklist(col_link);
    for (int i = 0; i < 10; ++i) {
        ll.add(target_keys[i]);
    }
    target->invalidate_object(target_keys[2]); // Entry 2 in list will be unresolved
    r1->commit_transaction();

    server.start();

    util::EventLoop::main().run_until([&] {
        if (auto table = r2->read_group().get_table("class_target")) {
            return table->size() > 0;
        }
        return false;
    });

    auto table = r2->read_group().get_table("class_origin");
    Obj obj = *table->begin();
    auto col = table->get_column_key("array");
    CollectionChangeSet change;
    List lst(r2, obj, col);
    bool called = false;

    auto require_change = [&] {
        auto token = lst.add_notification_callback([&](CollectionChangeSet c) {
            if (!c.empty()) {
                change = c;
                called = true;
            }
        });
        return token;
    };

    auto write = [&](auto&& f) {
        r1->begin_transaction();
        f();
        r1->commit_transaction();
        called = false;
        util::EventLoop::main().run_until([&] {
            return called;
        });
    };

    SECTION("adjust index of deleted entry") {
        auto token = require_change();
        write([&] {
            ll.remove(5);
        });
        REQUIRE_INDICES(change.deletions, 5);
    }

    SECTION("adjust index of inserted entry") {
        auto token = require_change();
        write([&] {
            ll.insert(5, target_keys[10]);
        });
        REQUIRE_INDICES(change.insertions, 5);
    }

    SECTION("adjust index of modified entry") {
        auto token = require_change();
        write([&] {
            ll.set(5, target_keys[10]);
        });
        REQUIRE_INDICES(change.modifications, 5);
    }

    SECTION("invalidating an object is seen as a deletion") {
        auto token = require_change();
        write([&] {
            target->invalidate_object(target_keys[6]);
        });
        REQUIRE_INDICES(change.deletions, 5);
    }

    SECTION("resurrecting an object is seen as an insertion") {
        auto token = require_change();
        write([&] {
            target->create_object_with_primary_key(2);
        });
        REQUIRE_INDICES(change.insertions, 2);
    }

    SECTION("inserting an unresolved link is not seen") {
        auto token = require_change();
        write([&] {
            origin_obj.get_list<ObjKey>(col_link).insert(7, target->get_objkey_from_primary_key(100));
            // We will have to make other modifications for the notifier to be called
            ll.set(6, target_keys[10]);
        });
        REQUIRE(change.insertions.empty());
        REQUIRE_INDICES(change.modifications, 6);
    }

    SECTION("erasing an unresolved link is not seen") {
        auto token = require_change();
        write([&] {
            origin_obj.get_list<ObjKey>(col_link).remove(2);
            // We will have to make other modifications for the notifier to be called
            ll.set(6, target_keys[10]);
        });
        REQUIRE(change.deletions.empty());
        REQUIRE_INDICES(change.modifications, 6);
    }
}
#endif
