diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index c9203b915f..8bc724ad35 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -34,6 +34,7 @@ from synapse.api.errors import ( StoreError, SynapseError, ) +from synapse.replication.http.profile import ReplicationProfileRecordFieldUpdates from synapse.storage.databases.main.media_repository import LocalMedia, RemoteMedia from synapse.storage.roommember import ProfileInfo from synapse.types import ( @@ -101,8 +102,17 @@ class ProfileHandler: self._update_join_states_task, UPDATE_JOIN_STATES_ACTION_NAME ) self._worker_locks = hs.get_worker_locks_handler() + self._is_profile_worker = ( + hs.get_instance_name() in hs.config.worker.writers.profile_updates + ) + self._record_profile_updates_client = ( + ReplicationProfileRecordFieldUpdates.make_client(self.hs) + ) + self._profile_updates_writer_instance = ( + self.hs.config.worker.writers.profile_updates[0] + ) - async def _record_profile_updates( + async def record_profile_updates( self, user_id: UserID, updated_fields: set[str] ) -> None: """ @@ -298,7 +308,7 @@ class ProfileHandler: ) await self.store.set_profile_displayname(target_user, displayname_to_set) - await self._record_profile_updates( + await self._dispatch_record_profile_updates( target_user, {ProfileFields.DISPLAYNAME}, ) @@ -411,7 +421,7 @@ class ProfileHandler: ) await self.store.set_profile_avatar_url(target_user, avatar_url_to_set) - await self._record_profile_updates( + await self._dispatch_record_profile_updates( target_user, {ProfileFields.AVATAR_URL}, ) @@ -544,7 +554,9 @@ class ProfileHandler: profile_updates.append((field_name, None)) await self.store.delete_profile(target_user) - await self._record_profile_updates( + + # Record profile updates for the profile update stream + await self._dispatch_record_profile_updates( target_user, {field_name for field_name, _value in profile_updates} ) @@ -555,6 +567,33 @@ class ProfileHandler: deactivation=True, ) + async def _dispatch_record_profile_updates( + self, user_id: UserID, updated_fields: set[str] + ) -> None: + """ + Dispatch the recording of profile updates, either directly via the current + instance, if we're a profile worker, otherwise push via replication. + + Args: + user_id: The user whose profile has had updates. + updated_fields: A set of the names of the fields that were updated. + + Returns: + None + """ + if self._is_profile_worker: + await self.record_profile_updates( + user_id, + updated_fields, + ) + else: + # Offload to the right worker via http replication + await self._record_profile_updates_client( + instance_name=self._profile_updates_writer_instance, + user_id=user_id.to_string(), + updated_fields=updated_fields, + ) + @cached() async def check_avatar_size_and_mime_type(self, mxc: str) -> bool: """Check that the size and content type of the avatar at the given MXC URI are @@ -754,7 +793,7 @@ class ProfileHandler: raise AuthError(403, "Cannot set another user's profile") await self.store.set_profile_field(target_user, field_name, new_value) - await self._record_profile_updates(target_user, {field_name}) + await self._dispatch_record_profile_updates(target_user, {field_name}) # Custom fields do not propagate into the user directory *or* rooms. profile = await self.store.get_profileinfo(target_user) @@ -790,7 +829,7 @@ class ProfileHandler: raise AuthError(400, "Cannot set another user's profile") await self.store.delete_profile_field(target_user, field_name) - await self._record_profile_updates(target_user, {field_name}) + await self._dispatch_record_profile_updates(target_user, {field_name}) # Custom fields do not propagate into the user directory *or* rooms. profile = await self.store.get_profileinfo(target_user) diff --git a/synapse/replication/http/profile.py b/synapse/replication/http/profile.py index a692836489..8b27b08873 100644 --- a/synapse/replication/http/profile.py +++ b/synapse/replication/http/profile.py @@ -21,7 +21,7 @@ from twisted.web.server import Request from synapse.api.constants import Membership from synapse.http.server import HttpServer from synapse.replication.http._base import ReplicationEndpoint -from synapse.types import JsonDict, JsonValue, UserID, create_requester +from synapse.types import JsonDict, UserID if TYPE_CHECKING: from synapse.server import HomeServer @@ -29,79 +29,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -class ReplicationProfileSetFieldValue(ReplicationEndpoint): - """Set profile field for a user. - - The POST looks like: - - POST /_synapse/replication/profile_set_field_value/ - - { - "requester_id": "@user:domain.tld", - "field_name": "displayname", - "new_value": "User Display Name", - "by_admin": False, - "propagate": False, - "authenticated_entity": "@admin:domain.tld", - } - - 200 OK - - {} - """ - - NAME = "profile_set_field_value" - PATH_ARGS = ("user_id",) - METHOD = "POST" - CACHE = False - - def __init__(self, hs: "HomeServer"): - super().__init__(hs) - - self._profile_handler = hs.get_profile_handler() - - @staticmethod - async def _serialize_payload( # type: ignore[override] - user_id: str, - requester_id: str, - field_name: str, - new_value: JsonValue | dict[str, JsonValue], - by_admin: bool = False, - propagate: bool = False, - authenticated_entity: str | None = None, - ) -> JsonDict: - return { - "requester_id": requester_id, - "field_name": field_name, - "new_value": new_value, - "by_admin": by_admin, - "propagate": propagate, - "authenticated_entity": authenticated_entity, - } - - async def _handle_request( # type: ignore[override] - self, request: Request, content: JsonDict, user_id: str - ) -> tuple[int, JsonDict]: - # Create a requester object with potentially an authenticated_entity, - # ie an admin who has done the request on behalf of the user. - requester = create_requester( - user_id=user_id, - authenticated_entity=content["authenticated_entity"] - if content["by_admin"] - else None, - ) - await self._profile_handler.set_field( - target_user=UserID.from_string(user_id), - requester=requester, - field_name=content["field_name"], - new_value=content["new_value"], - by_admin=content["by_admin"], - propagate=content["propagate"], - ) - - return (200, {}) - - class ReplicationProfileUserRoomMembershipChange(ReplicationEndpoint): """Store user profile update action regarding membership changes. @@ -159,6 +86,54 @@ class ReplicationProfileUserRoomMembershipChange(ReplicationEndpoint): return (200, {}) +class ReplicationProfileRecordFieldUpdates(ReplicationEndpoint): + """Record user profile field updates for the profile updates stream. + + The POST looks like: + + POST /_synapse/replication/profile_record_field_updates/ + + { + "updated_fields": ["list", "of", "fields"] + } + + 200 OK + + {} + """ + + NAME = "profile_record_field_updates" + PATH_ARGS = ("user_id",) + METHOD = "POST" + CACHE = False + + def __init__(self, hs: "HomeServer"): + super().__init__(hs) + + self._profile_handler = hs.get_profile_handler() + + @staticmethod + async def _serialize_payload( # type: ignore[override] + user_id: str, + updated_fields: set[str], + ) -> JsonDict: + assert len(updated_fields) > 0 + return { + "updated_fields": list(updated_fields), + } + + async def _handle_request( # type: ignore[override] + self, request: Request, content: JsonDict, user_id: str + ) -> tuple[int, JsonDict]: + assert len(content["updated_fields"]) > 0 + await self._profile_handler.record_profile_updates( + user_id=UserID.from_string(user_id), + updated_fields=set(content["updated_fields"]), + ) + + return (200, {}) + + def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: - ReplicationProfileSetFieldValue(hs).register(http_server) ReplicationProfileUserRoomMembershipChange(hs).register(http_server) + ReplicationProfileRecordFieldUpdates(hs).register(http_server) diff --git a/synapse/rest/client/profile.py b/synapse/rest/client/profile.py index 1767b13eb6..daab2c6651 100644 --- a/synapse/rest/client/profile.py +++ b/synapse/rest/client/profile.py @@ -35,7 +35,6 @@ from synapse.http.servlet import ( parse_json_object_from_request, ) from synapse.http.site import SynapseRequest -from synapse.replication.http.profile import ReplicationProfileSetFieldValue from synapse.rest.client._base import client_patterns from synapse.types import JsonDict, JsonValue, UserID from synapse.util.stringutils import is_namedspaced_grammar @@ -210,33 +209,14 @@ class ProfileFieldRestServlet(RestServlet): Codes.USER_ACCOUNT_SUSPENDED, ) - if self._is_profile_worker: - await self.profile_handler.set_field( - target_user=user, - requester=requester, - field_name=field_name, - new_value=new_value, - by_admin=is_admin, - propagate=propagate, - ) - else: - # Offload to the right worker via http replication - set_profile_data_client = ReplicationProfileSetFieldValue.make_client( - self.hs - ) - profile_updates_writer_instance = ( - self.hs.config.worker.writers.profile_updates[0] - ) - await set_profile_data_client( - instance_name=profile_updates_writer_instance, - user_id=user.to_string(), - requester_id=requester.user.to_string(), - field_name=field_name, - new_value=new_value, - by_admin=is_admin, - propagate=propagate, - authenticated_entity=requester.authenticated_entity, - ) + await self.profile_handler.set_field( + target_user=user, + requester=requester, + field_name=field_name, + new_value=new_value, + by_admin=is_admin, + propagate=propagate, + ) return 200, {} @@ -282,33 +262,14 @@ class ProfileFieldRestServlet(RestServlet): Codes.USER_ACCOUNT_SUSPENDED, ) - if self._is_profile_worker: - await self.profile_handler.set_field( - target_user=user, - requester=requester, - field_name=field_name, - new_value="", - by_admin=is_admin, - propagate=propagate, - ) - else: - # Offload to the right worker via http replication - set_profile_data_client = ReplicationProfileSetFieldValue.make_client( - self.hs - ) - profile_updates_writer_instance = ( - self.hs.config.worker.writers.profile_updates[0] - ) - await set_profile_data_client( - instance_name=profile_updates_writer_instance, - user_id=user.to_string(), - requester_id=requester.user.to_string(), - field_name=field_name, - new_value="", - by_admin=is_admin, - propagate=propagate, - authenticated_entity=requester.authenticated_entity, - ) + await self.profile_handler.set_field( + target_user=user, + requester=requester, + field_name=field_name, + new_value="", + by_admin=is_admin, + propagate=propagate, + ) return 200, {}