synapse_port_db

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 2015, 2016 OpenMarket Ltd
# Copyright 2018 New Vector Ltd
# Copyright 2019 The Matrix.org Foundation C.I.C.
#
# 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.

import argparse
import curses
import logging
import sys
import time
import traceback
from typing import Dict, Optional, Set

import yaml

from twisted.internet import defer, reactor

import synapse
from synapse.config.database import DatabaseConnectionConfig
from synapse.config.homeserver import HomeServerConfig
from synapse.logging.context import (
    LoggingContext,
    make_deferred_yieldable,
    run_in_background,
)
from synapse.storage.database import DatabasePool, make_conn
from synapse.storage.databases.main.client_ips import ClientIpBackgroundUpdateStore
from synapse.storage.databases.main.deviceinbox import DeviceInboxBackgroundUpdateStore
from synapse.storage.databases.main.devices import DeviceBackgroundUpdateStore
from synapse.storage.databases.main.end_to_end_keys import EndToEndKeyBackgroundStore
from synapse.storage.databases.main.events_bg_updates import (
    EventsBackgroundUpdatesStore,
)
from synapse.storage.databases.main.media_repository import (
    MediaRepositoryBackgroundUpdateStore,
)
from synapse.storage.databases.main.registration import (
    RegistrationBackgroundUpdateStore,
    find_max_generated_user_id_localpart,
)
from synapse.storage.databases.main.room import RoomBackgroundUpdateStore
from synapse.storage.databases.main.roommember import RoomMemberBackgroundUpdateStore
from synapse.storage.databases.main.search import SearchBackgroundUpdateStore
from synapse.storage.databases.main.state import MainStateBackgroundUpdateStore
from synapse.storage.databases.main.stats import StatsStore
from synapse.storage.databases.main.user_directory import (
    UserDirectoryBackgroundUpdateStore,
)
from synapse.storage.databases.state.bg_updates import StateBackgroundUpdateStore
from synapse.storage.engines import create_engine
from synapse.storage.prepare_database import prepare_database
from synapse.util import Clock
from synapse.util.versionstring import get_version_string

logger = logging.getLogger("synapse_port_db")


BOOLEAN_COLUMNS = {
    "events": ["processed", "outlier", "contains_url"],
    "rooms": ["is_public"],
    "event_edges": ["is_state"],
    "presence_list": ["accepted"],
    "presence_stream": ["currently_active"],
    "public_room_list_stream": ["visibility"],
    "devices": ["hidden"],
    "device_lists_outbound_pokes": ["sent"],
    "users_who_share_rooms": ["share_private"],
    "groups": ["is_public"],
    "group_rooms": ["is_public"],
    "group_users": ["is_public", "is_admin"],
    "group_summary_rooms": ["is_public"],
    "group_room_categories": ["is_public"],
    "group_summary_users": ["is_public"],
    "group_roles": ["is_public"],
    "local_group_membership": ["is_publicised", "is_admin"],
    "e2e_room_keys": ["is_verified"],
    "account_validity": ["email_sent"],
    "redactions": ["have_censored"],
    "room_stats_state": ["is_federatable"],
    "local_media_repository": ["safe_from_quarantine"],
    "users": ["shadow_banned"],
    "e2e_fallback_keys_json": ["used"],
}


APPEND_ONLY_TABLES = [
    "event_reference_hashes",
    "events",
    "event_json",
    "state_events",
    "room_memberships",
    "topics",
    "room_names",
    "rooms",
    "local_media_repository",
    "local_media_repository_thumbnails",
    "remote_media_cache",
    "remote_media_cache_thumbnails",
    "redactions",
    "event_edges",
    "event_auth",
    "received_transactions",
    "sent_transactions",
    "transaction_id_to_pdu",
    "users",
    "state_groups",
    "state_groups_state",
    "event_to_state_groups",
    "rejections",
    "event_search",
    "presence_stream",
    "push_rules_stream",
    "ex_outlier_stream",
    "cache_invalidation_stream_by_instance",
    "public_room_list_stream",
    "state_group_edges",
    "stream_ordering_to_exterm",
]


IGNORED_TABLES = {
    # We don't port these tables, as they're a faff and we can regenerate
    # them anyway.
    "user_directory",
    "user_directory_search",
    "user_directory_search_content",
    "user_directory_search_docsize",
    "user_directory_search_segdir",
    "user_directory_search_segments",
    "user_directory_search_stat",
    "user_directory_search_pos",
    "users_who_share_private_rooms",
    "users_in_public_room",
    # UI auth sessions have foreign keys so additional care needs to be taken,
    # the sessions are transient anyway, so ignore them.
    "ui_auth_sessions",
    "ui_auth_sessions_credentials",
    "ui_auth_sessions_ips",
}


# Error returned by the run function. Used at the top-level part of the script to
# handle errors and return codes.
end_error = None  # type: Optional[str]
# The exec_info for the error, if any. If error is defined but not exec_info the script
# will show only the error message without the stacktrace, if exec_info is defined but
# not the error then the script will show nothing outside of what's printed in the run
# function. If both are defined, the script will print both the error and the stacktrace.
end_error_exec_info = None


class Store(
    ClientIpBackgroundUpdateStore,
    DeviceInboxBackgroundUpdateStore,
    DeviceBackgroundUpdateStore,
    EventsBackgroundUpdatesStore,
    MediaRepositoryBackgroundUpdateStore,
    RegistrationBackgroundUpdateStore,
    RoomBackgroundUpdateStore,
    RoomMemberBackgroundUpdateStore,
    SearchBackgroundUpdateStore,
    StateBackgroundUpdateStore,
    MainStateBackgroundUpdateStore,
    UserDirectoryBackgroundUpdateStore,
    EndToEndKeyBackgroundStore,
    StatsStore,
):
    def execute(self, f, *args, **kwargs):
        return self.db_pool.runInteraction(f.__name__, f, *args, **kwargs)

    def execute_sql(self, sql, *args):
        def r(txn):
            txn.execute(sql, args)
            return txn.fetchall()

        return self.db_pool.runInteraction("execute_sql", r)

    def insert_many_txn(self, txn, table, headers, rows):
        sql = "INSERT INTO %s (%s) VALUES (%s)" % (
            table,
            ", ".join(k for k in headers),
            ", ".join("%s" for _ in headers),
        )

        try:
            txn.executemany(sql, rows)
        except Exception: