PrimitiveListNotifier::PrimitiveListNotifier(TableRef table, std::shared_ptr<Realm> realm)
: CollectionNotifier(std::move(realm))
, m_prev_size(table->size())
{
    set_table(*table->get_parent_table());
    m_table_handover = source_shared_group().export_table_for_handover(table);
}

void ObjectStore::delete_data_for_object(Group *group, StringData object_type) {
    TableRef table = table_for_object_type(group, object_type);
    if (table) {
        group->remove_table(table->get_index_in_group());
        set_primary_key_for_object(group, object_type, "");
    }
}

void Permissions::get_permissions(std::shared_ptr<SyncUser> user,
                                  PermissionResultsCallback callback,
                                  const ConfigMaker& make_config)
{
    auto realm = Permissions::permission_realm(user, make_config);
    auto table = ObjectStore::table_for_object_type(realm->read_group(), "Permission");
    auto results = std::make_shared<NotificationWrapper<Results>>(std::move(realm), *table);

    // `get_permissions` works by temporarily adding an async notifier to the permission Realm.
    // This notifier will run the `async` callback until the Realm contains permissions or
    // an error happens. When either of these two things happen, the notifier will be
    // unregistered by nulling out the `results_wrapper` container.
    auto async = [results, callback=std::move(callback)](CollectionChangeSet, std::exception_ptr ex) mutable {
        if (ex) {
            callback(Results(), ex);
            results.reset();
            return;
        }
        if (results->size() > 0) {
            // We monitor the raw results. The presence of a `__management` Realm indicates
            // that the permissions have been downloaded (hence, we wait until size > 0).
            TableRef table = ObjectStore::table_for_object_type(results->get_realm()->read_group(), "Permission");
            size_t col_idx = table->get_descriptor()->get_column_index("path");
            auto query = !(table->column<StringData>(col_idx).ends_with("/__permission")
                           || table->column<StringData>(col_idx).ends_with("/__management"));
            // Call the callback with our new permissions object. This object will exclude the
            // private Realms.
            callback(results->filter(std::move(query)), nullptr);
            results.reset();
        }
    };
    results->add_notification_callback(std::move(async));
}

bool ObjectStore::update_indexes(Group *group, Schema &schema) {
    bool changed = false;
    for (auto& object_schema : schema) {
        TableRef table = table_for_object_type(group, object_schema.name);
        if (!table) {
            continue;
        }

        for (auto& property : object_schema.properties) {
            if (property.requires_index() == table->has_search_index(property.table_column)) {
                continue;
            }

            changed = true;
            if (property.requires_index()) {
                try {
                    table->add_search_index(property.table_column);
                }
                catch (LogicError const&) {
                    throw ObjectStoreException(ObjectStoreException::Kind::RealmPropertyTypeNotIndexable, {
                        {"object_type", object_schema.name},
                        {"property_name", property.name},
                        {"property_type", string_for_property_type(property.type)}
                    });
                }
            }
            else {
                table->remove_search_index(property.table_column);
            }
        }
    }
    return changed;
}

SyncFileActionMetadataResults SyncMetadataManager::all_pending_actions() const
{
    SharedRealm realm = Realm::get_shared_realm(get_configuration());
    TableRef table = ObjectStore::table_for_object_type(realm->read_group(), c_sync_fileActionMetadata);
    Results results(realm, table->where());
    return SyncFileActionMetadataResults(std::move(results), std::move(realm), m_file_action_schema);
}

ValueType Object::get_property_value_impl(ContextType& ctx, const Property &property)
{
    verify_attached();

    size_t column = property.table_column;
    if (is_nullable(property.type) && m_row.is_null(column))
        return ctx.null_value();
    if (is_array(property.type) && property.type != PropertyType::LinkingObjects)
        return ctx.box(List(m_realm, *m_row.get_table(), column, m_row.get_index()));

    switch (property.type & ~PropertyType::Flags) {
        case PropertyType::Bool:   return ctx.box(m_row.get_bool(column));
        case PropertyType::Int:    return ctx.box(m_row.get_int(column));
        case PropertyType::Float:  return ctx.box(m_row.get_float(column));
        case PropertyType::Double: return ctx.box(m_row.get_double(column));
        case PropertyType::String: return ctx.box(m_row.get_string(column));
        case PropertyType::Data:   return ctx.box(m_row.get_binary(column));
        case PropertyType::Date:   return ctx.box(m_row.get_timestamp(column));
        case PropertyType::Any:    return ctx.box(m_row.get_mixed(column));
        case PropertyType::Object: {
            auto linkObjectSchema = m_realm->schema().find(property.object_type);
            TableRef table = ObjectStore::table_for_object_type(m_realm->read_group(), property.object_type);
            return ctx.box(Object(m_realm, *linkObjectSchema, table->get(m_row.get_link(column))));
        }
        case PropertyType::LinkingObjects: {
            auto target_object_schema = m_realm->schema().find(property.object_type);
            auto link_property = target_object_schema->property_for_name(property.link_origin_property_name);
            TableRef table = ObjectStore::table_for_object_type(m_realm->read_group(), target_object_schema->name);
            auto tv = m_row.get_table()->get_backlink_view(m_row.get_index(), table.get(), link_property->table_column);
            return ctx.box(Results(m_realm, std::move(tv)));
        }
        default: REALM_UNREACHABLE();
    }
}

ObjectSchema::ObjectSchema(Group *group, const std::string &name) : name(name) {
    TableRef tableRef = ObjectStore::table_for_object_type(group, name);
    Table *table = tableRef.get();

    size_t count = table->get_column_count();
    properties.reserve(count);
    for (size_t col = 0; col < count; col++) {
        Property property;
        property.name = table->get_column_name(col).data();
        property.type = (PropertyType)table->get_column_type(col);
        property.is_indexed = table->has_search_index(col);
        property.is_primary = false;
        property.table_column = col;
        if (property.type == PropertyTypeObject || property.type == PropertyTypeArray) {
            // set link type for objects and arrays
            realm::TableRef linkTable = table->get_link_target(col);
            property.object_type = ObjectStore::object_type_for_table_name(linkTable->get_name().data());
        }
        properties.push_back(move(property));
    }

    primary_key = realm::ObjectStore::get_primary_key_for_object(group, name);
    if (primary_key.length()) {
        auto primary_key_prop = primary_key_property();
        if (!primary_key_prop) {
            throw ObjectStoreValidationException({"No property matching primary key '" + primary_key + "'"}, name);
        }
        primary_key_prop->is_primary = true;
    }
}

uint64_t ObjectStore::get_schema_version(Group *group) {
    TableRef table = group->get_table(c_metadataTableName);
    if (!table || table->get_column_count() == 0) {
        return ObjectStore::NotVersioned;
    }
    return table->get_int(c_versionColumnIndex, c_zeroRowIndex);
}

void SyncUserMetadata::remove()
{
    m_invalid = true;
    m_realm->begin_transaction();
    TableRef table = ObjectStore::table_for_object_type(m_realm->read_group(), c_sync_userMetadata);
    table->move_last_over(m_row.get_index());
    m_realm->commit_transaction();
    m_realm = nullptr;
}

void SyncFileActionMetadata::remove()
{
    REALM_ASSERT(m_realm);
    m_realm->verify_thread();
    m_realm->begin_transaction();
    TableRef table = ObjectStore::table_for_object_type(m_realm->read_group(), c_sync_fileActionMetadata);
    table->move_last_over(m_row.get_index());
    m_realm->commit_transaction();
    m_realm = nullptr;
}

SyncFileActionMetadata::SyncFileActionMetadata(const SyncMetadataManager& manager,
                                               Action action,
                                               const std::string& original_name,
                                               const std::string& url,
                                               const std::string& user_identity,
                                               util::Optional<std::string> new_name)
: m_schema(manager.m_file_action_schema)
{
    size_t raw_action = static_cast<size_t>(action);

    // Open the Realm.
    m_realm = Realm::get_shared_realm(manager.get_configuration());

    // Retrieve or create the row for this object.
    TableRef table = ObjectStore::table_for_object_type(m_realm->read_group(), c_sync_fileActionMetadata);
    m_realm->begin_transaction();
    size_t row_idx = table->find_first_string(m_schema.idx_original_name, original_name);
    if (row_idx == not_found) {
        row_idx = table->add_empty_row();
        table->set_string(m_schema.idx_original_name, row_idx, original_name);
    }
    table->set_string(m_schema.idx_new_name, row_idx, new_name);
    table->set_int(m_schema.idx_action, row_idx, raw_action);
    table->set_string(m_schema.idx_url, row_idx, url);
    table->set_string(m_schema.idx_user_identity, row_idx, user_identity);
    m_realm->commit_transaction();
    m_row = table->get(row_idx);
}

StringData ObjectStore::get_primary_key_for_object(Group *group, StringData object_type) {
    TableRef table = group->get_table(c_primaryKeyTableName);
    if (!table) {
        return "";
    }
    size_t row = table->find_first_string(c_primaryKeyObjectClassColumnIndex, object_type);
    if (row == not_found) {
        return "";
    }
    return table->get_string(c_primaryKeyPropertyNameColumnIndex, row);
}

util::Optional<SyncFileActionMetadata> SyncFileActionMetadata::metadata_for_path(const std::string& original_name, const SyncMetadataManager& manager)
{
    auto realm = Realm::get_shared_realm(manager.get_configuration());
    auto schema = manager.m_file_action_schema;
    TableRef table = ObjectStore::table_for_object_type(realm->read_group(), c_sync_fileActionMetadata);
    size_t row_idx = table->find_first_string(schema.idx_original_name, original_name);
    if (row_idx == not_found) {
        return none;
    }
    return SyncFileActionMetadata(std::move(schema), std::move(realm), table->get(row_idx));
}                   

SyncUserMetadataResults SyncMetadataManager::get_users(bool marked) const
{
    // Open the Realm.
    SharedRealm realm = Realm::get_shared_realm(get_configuration());

    TableRef table = ObjectStore::table_for_object_type(realm->read_group(), c_sync_userMetadata);
    Query query = table->where().equal(m_user_schema.idx_marked_for_removal, marked);

    Results results(realm, std::move(query));
    return SyncUserMetadataResults(std::move(results), std::move(realm), m_user_schema);
}

void ObjectStore::set_primary_key_for_object(Group& group, StringData object_type, StringData primary_key) {
    TableRef table = group.get_table(c_primaryKeyTableName);

    size_t row = table->find_first_string(c_primaryKeyObjectClassColumnIndex, object_type);

#if REALM_ENABLE_SYNC
    // sync::create_table* functions should have already updated the pk table.
    if (sync::has_object_ids(group)) {
        if (primary_key.size() == 0)
            REALM_ASSERT(row == not_found);
        else {
             REALM_ASSERT(row != not_found);
             REALM_ASSERT(table->get_string(c_primaryKeyPropertyNameColumnIndex, row) == primary_key);
        }
        return;
    }
#endif // REALM_ENABLE_SYNC

    if (row == not_found && primary_key.size()) {
        row = table->add_empty_row();
        table->set_string_unique(c_primaryKeyObjectClassColumnIndex, row, object_type);
        table->set_string(c_primaryKeyPropertyNameColumnIndex, row, primary_key);
        return;
    }
    // set if changing, or remove if setting to nil
    if (primary_key.size() == 0) {
        if (row != not_found) {
            table->move_last_over(row);
        }
    }
    else {
        table->set_string(c_primaryKeyPropertyNameColumnIndex, row, primary_key);
    }
}

static void copy_property_to_property(Property &old_property, Property &new_property, TableRef table) {
    size_t old_column = old_property.table_column, new_column = new_property.table_column;
    size_t count = table->size();
    for (size_t i = 0; i < count; i++) {
        T old_value = get_value<T>(table, i, old_column);
        set_value(table, i, new_column, old_value);
    }
}

void Stack::pushTable(const TableRef &value) {
	const int ref = value.getRef();
	if (ref != LUA_REFNIL) {
		lua_getref(&_luaState, ref);
	} else {
		lua_pushnil(&_luaState);
	}
}

void ObjectStore::create_metadata_tables(Group *group) {
    TableRef table = group->get_or_add_table(c_primaryKeyTableName);
    if (table->get_column_count() == 0) {
        table->add_column(type_String, c_primaryKeyObjectClassColumnName);
        table->add_column(type_String, c_primaryKeyPropertyNameColumnName);
    }

    table = group->get_or_add_table(c_metadataTableName);
    if (table->get_column_count() == 0) {
        table->add_column(type_Int, c_versionColumnName);

        // set initial version
        table->add_empty_row();
        table->set_int(c_versionColumnIndex, c_zeroRowIndex, ObjectStore::NotVersioned);
    }
}

Variables ScriptManager::callFunction(const Common::UString &name, const Variables &params) {
	assert(!name.empty());
	assert(_luaState && _regNestingLevel == 0);

	std::vector<Common::UString> parts;
	Common::UString::split(name, '.', parts);
	if (parts.empty()) {
		error("Lua call \"%s\" failed: bad name", name.c_str());
		return Variables();
	}

	const Common::UString funcName = parts.back();
	parts.pop_back();

	if (parts.empty()) {
		return getGlobalFunction(funcName).call(params);
	}

	TableRef table = getGlobalTable(parts[0]);
	for (uint32 i = 1; i < parts.size(); ++i) {
		table = table.getTableAt(parts[i]);
	}
	return table.getFunctionAt(funcName).call(params);
}

void ObjectStore::set_primary_key_for_object(Group *group, StringData object_type, StringData primary_key) {
    TableRef table = group->get_table(c_primaryKeyTableName);

    // get row or create if new object and populate
    size_t row = table->find_first_string(c_primaryKeyObjectClassColumnIndex, object_type);
    if (row == not_found && primary_key.size()) {
        row = table->add_empty_row();
        table->set_string(c_primaryKeyObjectClassColumnIndex, row, object_type);
    }

    // set if changing, or remove if setting to nil
    if (primary_key.size() == 0) {
        if (row != not_found) {
            table->remove(row);
        }
    }
    else {
        table->set_string(c_primaryKeyPropertyNameColumnIndex, row, primary_key);
    }
}