Remove groups replication code. (#12900)

The replication logic for groups is no longer used, so the message
passing infrastructure can be removed.

4 files changed, 0 insertions, 86 deletions

diff --git a/synapse/replication/slave/storage/groups.py b/synapse/replication/slave/storage/groups.py
deleted file mode 100644
index d6f37d7479..0000000000
--- a/synapse/replication/slave/storage/groups.py
+++ /dev/null
@@ -1,58 +0,0 @@
-# Copyright 2016 OpenMarket Ltd
-#
-# 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.
-
-from typing import TYPE_CHECKING, Any, Iterable
-
-from synapse.replication.slave.storage._base import BaseSlavedStore
-from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
-from synapse.replication.tcp.streams import GroupServerStream
-from synapse.storage.database import DatabasePool, LoggingDatabaseConnection
-from synapse.storage.databases.main.group_server import GroupServerWorkerStore
-from synapse.util.caches.stream_change_cache import StreamChangeCache
-
-if TYPE_CHECKING:
-    from synapse.server import HomeServer
-
-
-class SlavedGroupServerStore(GroupServerWorkerStore, BaseSlavedStore):
-    def __init__(
-        self,
-        database: DatabasePool,
-        db_conn: LoggingDatabaseConnection,
-        hs: "HomeServer",
-    ):
-        super().__init__(database, db_conn, hs)
-
-        self.hs = hs
-
-        self._group_updates_id_gen = SlavedIdTracker(
-            db_conn, "local_group_updates", "stream_id"
-        )
-        self._group_updates_stream_cache = StreamChangeCache(
-            "_group_updates_stream_cache",
-            self._group_updates_id_gen.get_current_token(),
-        )
-
-    def get_group_stream_token(self) -> int:
-        return self._group_updates_id_gen.get_current_token()
-
-    def process_replication_rows(
-        self, stream_name: str, instance_name: str, token: int, rows: Iterable[Any]
-    ) -> None:
-        if stream_name == GroupServerStream.NAME:
-            self._group_updates_id_gen.advance(instance_name, token)
-            for row in rows:
-                self._group_updates_stream_cache.entity_has_changed(row.user_id, token)
-
-        return super().process_replication_rows(stream_name, instance_name, token, rows)
diff --git a/synapse/replication/tcp/client.py b/synapse/replication/tcp/client.py
index a52e25c1af..2f59245058 100644
--- a/synapse/replication/tcp/client.py
+++ b/synapse/replication/tcp/client.py
@@ -30,7 +30,6 @@ from synapse.replication.tcp.protocol import ClientReplicationStreamProtocol
 from synapse.replication.tcp.streams import (
     AccountDataStream,
     DeviceListsStream,
-    GroupServerStream,
     PushersStream,
     PushRulesStream,
     ReceiptsStream,
@@ -185,10 +184,6 @@ class ReplicationDataHandler:
             self.notifier.on_new_event(
                 StreamKeyType.DEVICE_LIST, token, rooms=all_room_ids
             )
-        elif stream_name == GroupServerStream.NAME:
-            self.notifier.on_new_event(
-                "groups_key", token, users=[row.user_id for row in rows]
-            )
         elif stream_name == PushersStream.NAME:
             for row in rows:
                 if row.deleted:
diff --git a/synapse/replication/tcp/streams/__init__.py b/synapse/replication/tcp/streams/__init__.py
index f41eabd85e..b1cd55bf6f 100644
--- a/synapse/replication/tcp/streams/__init__.py
+++ b/synapse/replication/tcp/streams/__init__.py
@@ -29,7 +29,6 @@ from synapse.replication.tcp.streams._base import (
     BackfillStream,
     CachesStream,
     DeviceListsStream,
-    GroupServerStream,
     PresenceFederationStream,
     PresenceStream,
     PushersStream,
@@ -61,7 +60,6 @@ STREAMS_MAP = {
         FederationStream,
         TagAccountDataStream,
         AccountDataStream,
-        GroupServerStream,
         UserSignatureStream,
     )
 }
@@ -81,6 +79,5 @@ __all__ = [
     "ToDeviceStream",
     "TagAccountDataStream",
     "AccountDataStream",
-    "GroupServerStream",
     "UserSignatureStream",
 ]
diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py
index 495f2f0285..398bebeaa6 100644
--- a/synapse/replication/tcp/streams/_base.py
+++ b/synapse/replication/tcp/streams/_base.py
@@ -585,26 +585,6 @@ class AccountDataStream(Stream):
         return updates, to_token, limited
 
 
-class GroupServerStream(Stream):
-    @attr.s(slots=True, frozen=True, auto_attribs=True)
-    class GroupsStreamRow:
-        group_id: str
-        user_id: str
-        type: str
-        content: JsonDict
-
-    NAME = "groups"
-    ROW_TYPE = GroupsStreamRow
-
-    def __init__(self, hs: "HomeServer"):
-        store = hs.get_datastores().main
-        super().__init__(
-            hs.get_instance_name(),
-            current_token_without_instance(store.get_group_stream_token),
-            store.get_all_groups_changes,
-        )
-
-
 class UserSignatureStream(Stream):
     """A user has signed their own device with their user-signing key"""