dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-14 13:40:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

Respond with some data for sliding sync profiles extension

Browse Source
This commit is contained in:
Jason Robinson
2026-07-15 16:58:32 +03:00
parent 6b484bfa4f
commit 890b09b0cf
4 changed files with 343 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/sliding_sync/extensions.py
+166 -3
View File
@@ -25,7 +25,12 @@ from typing import (
from typing_extensions import TypeAlias, assert_never
from synapse.api.constants import AccountDataTypes, EduTypes, StickyEvent
from synapse.api.constants import (
AccountDataTypes,
EduTypes,
ProfileUpdateAction,
StickyEvent,
)
from synapse.events.utils import FilteredEvent
from synapse.handlers.receipts import ReceiptEventSource
from synapse.logging.opentracing import trace
@@ -33,12 +38,15 @@ from synapse.storage.databases.main.receipts import ReceiptInRoom
from synapse.types import (
Absent,
DeviceListUpdates,
JsonDict,
JsonMapping,
JsonValue,
MultiWriterStreamToken,
SlidingSyncStreamToken,
StrCollection,
StreamToken,
ThreadSubscriptionsToken,
UserID,
)
from synapse.types.handlers.sliding_sync import (
HaveSentRoomFlag,
@@ -1070,6 +1078,52 @@ class SlidingSyncExtensionHandler:
),
)
async def _get_profiles_extension_initial_sync_response(
self,
user_id: UserID,
fields: set[str],
) -> dict[str, JsonDict | None]:
"""
Build an initial sync response for the profiles extension.
Args:
user_id: The syncing user UserID
fields: A set of fields to include in the response.
Returns:
A dictionary containing the profile updates in an `updated` dictionary.
"""
response: dict[str, JsonDict | None] = {}
# TODO should be filtered for the rooms for this sync
profile_user_ids = await self.store.get_local_users_who_share_room_with_user(
user_id.to_string(),
)
# Ensure we're in the list even if we don't belong to any rooms
profile_user_ids.add(user_id.to_string())
profile_data_by_user = await self.store.get_profile_data_for_users(
profile_user_ids
)
# Serialise the profile updates into the sync response format.
for profile_user_id in profile_user_ids:
profile_data = profile_data_by_user.get(profile_user_id)
if profile_data is None:
# Don't generate anything for users with no profile data
# in initial sync.
continue
per_user_updates: dict[str, JsonValue | dict[str, JsonValue]] = {}
for field_name in fields:
if field_name in profile_data.keys():
per_user_updates[field_name] = profile_data[field_name]
if per_user_updates:
response[profile_user_id] = {
"updated": per_user_updates,
}
return response
async def get_profiles_extension_response(
self,
sync_config: SlidingSyncConfig,
@@ -1078,9 +1132,118 @@ class SlidingSyncExtensionHandler:
to_token: StreamToken,
from_token: SlidingSyncStreamToken | None,
) -> SlidingSyncResult.Extensions.ProfilesExtension | None:
"""
Generate a response for the profiles extension.
Args:
sync_config: The Sliding Sync config.
profiles_request: The profiles extension request.
all_interested_room_ids: Set of rooms the sync request is interested in.
to_token: The stream token to generate a response until.
from_token: The stream token to generate a response from.
Returns:
A SlidingSyncResult.Extensions.ProfilesExtension object containing
all the users who have profile updates.
"""
if not profiles_request.enabled:
return None
return SlidingSyncResult.Extensions.ProfilesExtension(
users={},
user_id = sync_config.user.to_string()
if not profiles_request.fields:
return SlidingSyncResult.Extensions.ProfilesExtension(
users={},
)
fields = set(profiles_request.fields)
response: dict[str, JsonDict | None] = {}
if from_token is None:
# Initial sync
return SlidingSyncResult.Extensions.ProfilesExtension(
users=await self._get_profiles_extension_initial_sync_response(
user_id=sync_config.user,
fields=fields,
),
)
# Incremental sync
updates = await self.store.get_profile_updates_for_user_and_fields(
from_id=from_token.stream_token.profile_updates_key,
to_id=to_token.profile_updates_key,
user_id=user_id,
field_names=set(fields),
)
profile_user_ids = set()
updated_users = {
update.user_id
for update in updates
if update.action == ProfileUpdateAction.UPDATE.value
}
# Add users with updates
profile_user_ids.update(updated_users)
updated_user_fields: dict[str, set[str]] = {}
# Set fields from updates
for update in updates:
# Skip the update if there is no field update (a joined or left room
# action), the client didn't ask for this field, or we're not
# interested in this user.
if (
not update.field_name
or update.field_name not in fields
or update.user_id not in profile_user_ids
):
continue
updated_user_fields.setdefault(update.user_id, set()).add(update.field_name)
profile_data_by_user = await self.store.get_profile_data_for_users(
profile_user_ids
)
# TODO lazy loading
is_lazy = False
# Serialise the profile updates into the sync response format.
for profile_user_id in profile_user_ids:
profile_data = profile_data_by_user.get(profile_user_id)
if profile_data is None:
# No profile data for this user, just return a blank dictionary
# in incremental sync, telling the clients to remove all profile
# information for this user.
response[profile_user_id] = None
continue
per_user_updates: dict[str, JsonValue | dict[str, JsonValue]] = {}
if is_lazy:
# TODO lazy cache
# Include all the fields the client asked for
fields = set(profile_data.keys()).intersection(fields)
for field_name in fields:
per_user_updates[field_name] = profile_data.get(field_name)
else:
# Include only the diff, unless the user recently joined,
# then send all the fields the client asked for.
# We don't use a cache here as for non-lazy sync we always
# send changes and/or fields the client asked for, if relevant
# as above joined condition.
fields = (
fields
# TODO joined_room_user_ids
if profile_user_id in []
else set(updated_user_fields.get(profile_user_id, []))
)
fields = set(profile_data.keys()).intersection(fields)
for field_name in fields:
per_user_updates[field_name] = profile_data[field_name]
if per_user_updates:
response[profile_user_id] = {
"updated": per_user_updates,
}
return SlidingSyncResult.Extensions.ProfilesExtension(
users=response,
)
synapse/rest/client/sync.py
+14 -4
View File
@@ -1149,18 +1149,28 @@ class SlidingSyncRestServlet(RestServlet):
serialized_extensions[
"org.matrix.msc4262.profiles"
] = await self._serialise_profiles(
requester, extensions.profiles, ref_rooms_results
extensions.profiles,
)
return serialized_extensions
async def _serialise_profiles(
self,
requester: Requester,
profiles: SlidingSyncResult.Extensions.ProfilesExtension,
ref_rooms_results: Mapping[str, SlidingSyncResult.RoomResult],
) -> JsonMapping:
return profiles.users
"""
Serialise the profiles extension response.
Args:
profiles: The generated profiles response object.
Returns:
A dictionary containing the response `users` with the
generated profile updates.
"""
return {
"users": profiles.users,
}
async def _serialise_sticky_events(
self,
synapse/types/handlers/sliding_sync.py
+1 -1
View File
@@ -450,7 +450,7 @@ class SlidingSyncResult:
users: map (user_id -> [profile_updates])
"""
users: Mapping[str, JsonMapping]
users: Mapping[str, JsonMapping | None]
def __bool__(self) -> bool:
return bool(self.users)
tests/rest/client/sliding_sync/test_extension_profiles.py
+162 -10
View File
@@ -20,6 +20,7 @@ from twisted.internet.testing import MemoryReactor
import synapse.rest.admin
from synapse.rest.client import login, profile, room, sync
from synapse.server import HomeServer
from synapse.types import UserID, create_requester
from synapse.util.clock import Clock
from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
@@ -55,7 +56,15 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase):
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
self.store = hs.get_datastores().main
self.profile_handler = self.hs.get_profile_handler()
self.user = self.register_user("user", "password")
self.tok = self.login("user", "password")
self.other_user = self.register_user("other_user", "password")
self.other_tok = self.login("other_user", "password")
self.joined_room = self.helper.create_room_as(self.user, tok=self.tok)
self.helper.join(
room=self.joined_room, user=self.other_user, tok=self.other_tok
)
super().prepare(reactor, clock, hs)
@parameterized.expand(
@@ -69,26 +78,45 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase):
Test that no profile extension response is returned
if the feature is not enabled.
"""
user1_id = self.register_user("user1", "pass")
user1_tok = self.login(user1_id, "pass")
if is_initial:
self.get_success(
self.profile_handler.set_field(
target_user=UserID.from_string(self.other_user),
requester=create_requester(self.other_user),
field_name="field",
new_value="value",
)
)
# Make an initial Sliding Sync request with the profiles extension enabled
sync_body = {
"lists": {},
"extensions": {
"org.matrix.msc4262.profiles": {
"enabled": True,
"fields": ["field"],
},
},
}
response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles"))
response_body, from_token = self.do_sync(sync_body, tok=self.tok)
self.assertIsNone(
response_body["extensions"].get("org.matrix.msc4262.profiles")
)
if not is_initial:
self.get_success(
self.profile_handler.set_field(
target_user=UserID.from_string(self.other_user),
requester=create_requester(self.other_user),
field_name="field",
new_value="value",
)
)
# Make an incremental Sliding Sync request
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok)
self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles"))
self.assertIsNone(
response_body["extensions"].get("org.matrix.msc4262.profiles")
)
@override_config({"include_profile_updates_in_sync": True})
def test_no_data_initial_sync(self) -> None:
@@ -105,11 +133,14 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase):
"extensions": {
"org.matrix.msc4262.profiles": {
"enabled": True,
"fields": ["field"],
},
},
}
response_body, _ = self.do_sync(sync_body, tok=user1_tok)
self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles"))
self.assertIsNone(
response_body["extensions"].get("org.matrix.msc4262.profiles")
)
@override_config({"include_profile_updates_in_sync": True})
def test_no_data_incremental_sync(self) -> None:
@@ -125,6 +156,7 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase):
"extensions": {
"org.matrix.msc4262.profiles": {
"enabled": True,
"fields": ["field"],
}
},
}
@@ -133,4 +165,124 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase):
# Make an incremental Sliding Sync request with the profiles extension enabled
response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok)
self.assertIsNone(response_body["extensions"].get("org.matrix.msc4262.profiles"))
self.assertIsNone(
response_body["extensions"].get("org.matrix.msc4262.profiles")
)
@parameterized.expand(
[
True,
False,
]
)
@override_config({"include_profile_updates_in_sync": True})
def test_updated_fields_are_sent(self, is_initial: bool) -> None:
"""
Test that profile extension response returns field updates
in incremental and initial sync.
"""
if is_initial:
self.get_success(
self.profile_handler.set_field(
target_user=UserID.from_string(self.other_user),
requester=create_requester(self.other_user),
field_name="field",
new_value="value",
)
)
# Make an initial Sliding Sync request with the profiles extension enabled
sync_body = {
"lists": {},
"extensions": {
"org.matrix.msc4262.profiles": {
"enabled": True,
"fields": ["field"],
},
},
}
response_body, from_token = self.do_sync(sync_body, tok=self.tok)
if is_initial:
self.assertEqual(
response_body["extensions"]["org.matrix.msc4262.profiles"]["users"][
"@other_user:test"
],
{
"updated": {
"field": "value",
}
},
)
if not is_initial:
self.get_success(
self.profile_handler.set_field(
target_user=UserID.from_string(self.other_user),
requester=create_requester(self.other_user),
field_name="field",
new_value="value",
)
)
# Make an incremental Sliding Sync request
response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok)
self.assertEqual(
response_body["extensions"]["org.matrix.msc4262.profiles"]["users"][
"@other_user:test"
],
{
"updated": {
"field": "value",
}
},
)
@parameterized.expand(
[
True,
False,
]
)
@override_config({"include_profile_updates_in_sync": True})
def test_updated_fields_are_not_sent_if_not_requested(
self, is_initial: bool
) -> None:
"""
Test that profile extension response doesn't return field updates we didn't
request in initial and incremental sync.
"""
if is_initial:
self.get_success(
self.profile_handler.set_field(
target_user=UserID.from_string(self.other_user),
requester=create_requester(self.other_user),
field_name="anotherfield",
new_value="value",
)
)
# Make an initial Sliding Sync request with the profiles extension enabled
sync_body = {
"lists": {},
"extensions": {
"org.matrix.msc4262.profiles": {
"enabled": True,
"fields": ["field"],
},
},
}
response_body, from_token = self.do_sync(sync_body, tok=self.tok)
if is_initial:
response_body["extensions"].get("org.matrix.msc4262.profiles")
if not is_initial:
self.get_success(
self.profile_handler.set_field(
target_user=UserID.from_string(self.other_user),
requester=create_requester(self.other_user),
field_name="anotherfield",
new_value="value",
)
)
# Make an incremental Sliding Sync request
response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok)
response_body["extensions"].get("org.matrix.msc4262.profiles")