dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-07-28 09:59:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Ensure delete_profile_upon_deactivation does profile deletion and profile updates stream updates in a single transaction

Browse Source
This commit is contained in:
Jason Robinson
2026-07-07 22:06:10 +03:00
parent 739824217e
commit ae4e686f6f
2 changed files with 115 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/profile.py
+23 -12
View File
@@ -571,28 +571,39 @@ class ProfileHandler:
# have it.
raise AuthError(400, "Cannot remove another user's profile")
profile_updates: list[tuple[str, JsonValue | None]] = []
current_profile: ProfileInfo | None = None
profile_updates: list[str] = []
profile_update_targets: dict = {"rooms": set(), "users": set()}
if self._msc4429_enabled:
if current_profile is None:
current_profile = await self.store.get_profileinfo(target_user)
current_profile = await self.store.get_profileinfo(target_user)
if current_profile.display_name is not None:
profile_updates.append((ProfileFields.DISPLAYNAME, None))
profile_updates.append(ProfileFields.DISPLAYNAME)
if current_profile.avatar_url is not None:
profile_updates.append((ProfileFields.AVATAR_URL, None))
profile_updates.append(ProfileFields.AVATAR_URL)
custom_fields = await self.store.get_profile_fields(target_user)
for field_name in custom_fields.keys():
profile_updates.append((field_name, None))
profile_updates.append(field_name)
await self.store.delete_profile(target_user)
if profile_updates:
profile_update_targets = await self.get_targets_for_profile_updates(
target_user
)
# Discard ourselves as we're deactivating this account
profile_update_targets["users"].discard(target_user.to_string())
# Record profile updates for the profile update stream
if len(profile_updates):
await self._dispatch_record_profile_updates(
target_user, {field_name for field_name, _value in profile_updates}
# Record the profile delete and update the profile updates stream
stream_id = await self.store.delete_profile(
user_id=target_user,
field_names=profile_updates,
target_users=profile_update_targets["users"],
)
if stream_id and profile_update_targets["rooms"]:
self._notifier.on_new_event(
StreamKeyType.PROFILE_UPDATES,
stream_id,
rooms=profile_update_targets["rooms"],
)
await self._third_party_rules.on_profile_update(
synapse/storage/databases/main/profile.py
+92 -30
View File
@@ -754,7 +754,7 @@ class ProfileWorkerStore(SQLBaseStore):
txn=txn,
user_id=user_id,
action=ProfileUpdateAction.UPDATE,
field_name=field_name,
field_names=[field_name],
target_users=target_users,
)
@@ -766,7 +766,7 @@ class ProfileWorkerStore(SQLBaseStore):
txn: LoggingTransaction,
user_id: UserID,
action: ProfileUpdateAction,
field_name: str | None,
field_names: list[str] | None,
target_users: set[str],
) -> int | None:
"""
@@ -777,38 +777,68 @@ class ProfileWorkerStore(SQLBaseStore):
user_id: User ID that made the profile update
action: The profile update action, either `update`, `left_room` or
`joined_room`
field_name: The field to set the value for, if ProfileUpdateAction.UPDATE
field_names: A list of fields that were set, if ProfileUpdateAction.UPDATE
target_users: Set of users to create profile update stream rows for
Returns:
The stream ID created in this transaction
The latest stream ID created in this transaction
"""
if not self._msc4429_enabled:
return None
if action == ProfileUpdateAction.UPDATE:
assert field_name is not None
assert field_names
else:
assert field_name is None
assert not field_names
# Record the profile update
stream_id = self._profile_updates_id_gen.get_next_txn(txn)
self.db_pool.simple_insert_txn(
inserted_ts = self.clock.time_msec()
if field_names:
stream_ids = self._profile_updates_id_gen.get_next_mult_txn(
txn, len(field_names)
)
values: list[tuple[int, str, str, str, str | None, int]] = [
(
stream_id,
self._instance_name,
user_id.to_string(),
action.value,
field_name,
inserted_ts,
)
for stream_id, field_name in zip(stream_ids, field_names)
]
else:
stream_ids = [self._profile_updates_id_gen.get_next_txn(txn)]
values = [
(
stream_ids[0],
self._instance_name,
user_id.to_string(),
action.value,
None,
inserted_ts,
)
]
self.db_pool.simple_insert_many_txn(
txn,
table="profile_updates",
values={
"stream_id": stream_id,
"instance_name": self._instance_name,
"user_id": user_id.to_string(),
"action": action.value,
"field_name": field_name,
"inserted_ts": self.clock.time_msec(),
},
keys=[
"stream_id",
"instance_name",
"user_id",
"action",
"field_name",
"inserted_ts",
],
values=values,
)
# Add per user tracking rows
# Add per user tracking rows pointing to the latest stream ID
inserted_ts = self.clock.time_msec()
values = [(stream_id, user_id, inserted_ts) for user_id in target_users]
per_user_values = [
(stream_ids[-1], user_id, inserted_ts) for user_id in target_users
]
self.db_pool.simple_insert_many_txn(
txn,
table="profile_updates_per_user",
@@ -817,9 +847,9 @@ class ProfileWorkerStore(SQLBaseStore):
"user_id",
"inserted_ts",
],
values=values,
values=per_user_values,
)
return stream_id
return stream_ids[-1]
async def set_profile_field(
self,
@@ -888,7 +918,7 @@ class ProfileWorkerStore(SQLBaseStore):
txn=txn,
user_id=user_id,
action=ProfileUpdateAction.UPDATE,
field_name=field_name,
field_names=[field_name],
target_users=target_users,
)
return stream_id
@@ -897,16 +927,48 @@ class ProfileWorkerStore(SQLBaseStore):
"delete_profile_field", delete_profile_field
)
async def delete_profile(self, user_id: UserID) -> None:
async def delete_profile(
self,
user_id: UserID,
field_names: list[str],
target_users: set[str],
) -> int | None:
"""
Deletes an entire user profile, including displayname, avatar_url and all custom fields.
Used at user deactivation when erasure is requested.
Deletes an entire user profile, including displayname, avatar_url and all
custom fields. Used at user deactivation when erasure is requested.
Args:
user_id: User ID whose profile is going to be deleted.
field_names: List of profile fields the user had.
target_users: List of users who share rooms and are interested in
profile updates for this user.
Returns:
Latest stream ID for the profile updates stream.
"""
await self.db_pool.simple_delete(
desc="delete_profile",
table="profiles",
keyvalues={"full_user_id": user_id.to_string()},
def _delete_profile_txn(txn: LoggingTransaction) -> int | None:
# Delete the profile
txn.execute(
"""
DELETE FROM profiles
WHERE full_user_id = ?
""",
(user_id.to_string(),),
)
# Update the profile updates stream
stream_id = self._record_profile_updates_txn(
txn=txn,
user_id=user_id,
action=ProfileUpdateAction.UPDATE,
field_names=field_names,
target_users=target_users,
)
return stream_id
return await self.db_pool.runInteraction(
"delete_profile",
_delete_profile_txn,
)
async def record_profile_updates_for_user_left_room(
@@ -972,7 +1034,7 @@ class ProfileWorkerStore(SQLBaseStore):
txn=txn,
user_id=user_id,
action=ProfileUpdateAction.LEFT_ROOM,
field_name=None,
field_names=[],
target_users=users_to_update,
)
return stream_id
@@ -1008,7 +1070,7 @@ class ProfileWorkerStore(SQLBaseStore):
txn=txn,
user_id=user_id,
action=ProfileUpdateAction.JOINED_ROOM,
field_name=None,
field_names=[],
target_users=users_to_update,
)
return stream_id