forked from MirrorHub/synapse
Add groups to sync stream
This commit is contained in:
parent
b3bf6a1218
commit
c544188ee3
8 changed files with 161 additions and 8 deletions
|
@ -108,6 +108,17 @@ class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
|
|||
return True
|
||||
|
||||
|
||||
class GroupsSyncResult(collections.namedtuple("GroupsSyncResult", [
|
||||
"join",
|
||||
"invite",
|
||||
"leave",
|
||||
])):
|
||||
__slots__ = []
|
||||
|
||||
def __nonzero__(self):
|
||||
return self.join or self.invite or self.leave
|
||||
|
||||
|
||||
class SyncResult(collections.namedtuple("SyncResult", [
|
||||
"next_batch", # Token for the next sync
|
||||
"presence", # List of presence events for the user.
|
||||
|
@ -119,6 +130,7 @@ class SyncResult(collections.namedtuple("SyncResult", [
|
|||
"device_lists", # List of user_ids whose devices have chanegd
|
||||
"device_one_time_keys_count", # Dict of algorithm to count for one time keys
|
||||
# for this device
|
||||
"groups",
|
||||
])):
|
||||
__slots__ = []
|
||||
|
||||
|
@ -134,7 +146,8 @@ class SyncResult(collections.namedtuple("SyncResult", [
|
|||
self.archived or
|
||||
self.account_data or
|
||||
self.to_device or
|
||||
self.device_lists
|
||||
self.device_lists or
|
||||
self.groups
|
||||
)
|
||||
|
||||
|
||||
|
@ -560,6 +573,8 @@ class SyncHandler(object):
|
|||
user_id, device_id
|
||||
)
|
||||
|
||||
yield self._generate_sync_entry_for_groups(sync_result_builder)
|
||||
|
||||
defer.returnValue(SyncResult(
|
||||
presence=sync_result_builder.presence,
|
||||
account_data=sync_result_builder.account_data,
|
||||
|
@ -568,10 +583,56 @@ class SyncHandler(object):
|
|||
archived=sync_result_builder.archived,
|
||||
to_device=sync_result_builder.to_device,
|
||||
device_lists=device_lists,
|
||||
groups=sync_result_builder.groups,
|
||||
device_one_time_keys_count=one_time_key_counts,
|
||||
next_batch=sync_result_builder.now_token,
|
||||
))
|
||||
|
||||
@measure_func("_generate_sync_entry_for_groups")
|
||||
@defer.inlineCallbacks
|
||||
def _generate_sync_entry_for_groups(self, sync_result_builder):
|
||||
user_id = sync_result_builder.sync_config.user.to_string()
|
||||
since_token = sync_result_builder.since_token
|
||||
now_token = sync_result_builder.now_token
|
||||
|
||||
if since_token and since_token.groups_key:
|
||||
results = yield self.store.get_groups_changes_for_user(
|
||||
user_id, since_token.groups_key, now_token.groups_key,
|
||||
)
|
||||
else:
|
||||
results = yield self.store.get_all_groups_for_user(
|
||||
user_id, now_token.groups_key,
|
||||
)
|
||||
|
||||
invited = {}
|
||||
joined = {}
|
||||
left = {}
|
||||
for result in results:
|
||||
membership = result["membership"]
|
||||
group_id = result["group_id"]
|
||||
gtype = result["type"]
|
||||
content = result["content"]
|
||||
|
||||
if membership == "join":
|
||||
if gtype == "membership":
|
||||
content.pop("membership", None)
|
||||
invited[group_id] = content["content"]
|
||||
else:
|
||||
joined.setdefault(group_id, {})[gtype] = content
|
||||
elif membership == "invite":
|
||||
if gtype == "membership":
|
||||
content.pop("membership", None)
|
||||
invited[group_id] = content["content"]
|
||||
else:
|
||||
if gtype == "membership":
|
||||
left[group_id] = content["content"]
|
||||
|
||||
sync_result_builder.groups = GroupsSyncResult(
|
||||
join=joined,
|
||||
invite=invited,
|
||||
leave=left,
|
||||
)
|
||||
|
||||
@measure_func("_generate_sync_entry_for_device_list")
|
||||
@defer.inlineCallbacks
|
||||
def _generate_sync_entry_for_device_list(self, sync_result_builder):
|
||||
|
@ -1260,6 +1321,7 @@ class SyncResultBuilder(object):
|
|||
self.invited = []
|
||||
self.archived = []
|
||||
self.device = []
|
||||
self.groups = None
|
||||
|
||||
|
||||
class RoomSyncResultBuilder(object):
|
||||
|
|
|
@ -199,6 +199,11 @@ class SyncRestServlet(RestServlet):
|
|||
"invite": invited,
|
||||
"leave": archived,
|
||||
},
|
||||
"groups": {
|
||||
"join": sync_result.groups.join,
|
||||
"invite": sync_result.groups.invite,
|
||||
"leave": sync_result.groups.leave,
|
||||
},
|
||||
"device_one_time_keys_count": sync_result.device_one_time_keys_count,
|
||||
"next_batch": sync_result.next_batch.to_string(),
|
||||
}
|
||||
|
|
|
@ -136,6 +136,9 @@ class DataStore(RoomMemberStore, RoomStore,
|
|||
db_conn, "pushers", "id",
|
||||
extra_tables=[("deleted_pushers", "stream_id")],
|
||||
)
|
||||
self._group_updates_id_gen = StreamIdGenerator(
|
||||
db_conn, "local_group_updates", "stream_id",
|
||||
)
|
||||
|
||||
if isinstance(self.database_engine, PostgresEngine):
|
||||
self._cache_id_gen = StreamIdGenerator(
|
||||
|
@ -236,6 +239,18 @@ class DataStore(RoomMemberStore, RoomStore,
|
|||
prefilled_cache=curr_state_delta_prefill,
|
||||
)
|
||||
|
||||
_group_updates_prefill, min_group_updates_id = self._get_cache_dict(
|
||||
db_conn, "local_group_updates",
|
||||
entity_column="user_id",
|
||||
stream_column="stream_id",
|
||||
max_value=self._group_updates_id_gen.get_current_token(),
|
||||
limit=1000,
|
||||
)
|
||||
self._group_updates_stream_cache = StreamChangeCache(
|
||||
"_group_updates_stream_cache", min_group_updates_id,
|
||||
prefilled_cache=_group_updates_prefill,
|
||||
)
|
||||
|
||||
cur = LoggingTransaction(
|
||||
db_conn.cursor(),
|
||||
name="_find_stream_orderings_for_times_txn",
|
||||
|
|
|
@ -776,7 +776,7 @@ class GroupServerStore(SQLBaseStore):
|
|||
remote_attestation (dict): If remote group then store the remote
|
||||
attestation from the group, else None.
|
||||
"""
|
||||
def _register_user_group_membership_txn(txn):
|
||||
def _register_user_group_membership_txn(txn, next_id):
|
||||
# TODO: Upsert?
|
||||
self._simple_delete_txn(
|
||||
txn,
|
||||
|
@ -798,6 +798,19 @@ class GroupServerStore(SQLBaseStore):
|
|||
},
|
||||
)
|
||||
|
||||
self._simple_insert_txn(
|
||||
txn,
|
||||
table="local_group_updates",
|
||||
values={
|
||||
"stream_id": next_id,
|
||||
"group_id": group_id,
|
||||
"user_id": user_id,
|
||||
"type": "membership",
|
||||
"content": json.dumps({"membership": membership, "content": content}),
|
||||
}
|
||||
)
|
||||
self._group_updates_stream_cache.entity_has_changed(user_id, next_id)
|
||||
|
||||
# TODO: Insert profile to ensure it comes down stream if its a join.
|
||||
|
||||
if membership == "join":
|
||||
|
@ -840,10 +853,11 @@ class GroupServerStore(SQLBaseStore):
|
|||
},
|
||||
)
|
||||
|
||||
yield self.runInteraction(
|
||||
"register_user_group_membership",
|
||||
_register_user_group_membership_txn,
|
||||
)
|
||||
with self._group_updates_id_gen.get_next() as next_id:
|
||||
yield self.runInteraction(
|
||||
"register_user_group_membership",
|
||||
_register_user_group_membership_txn, next_id,
|
||||
)
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def create_group(self, group_id, user_id, name, avatar_url, short_description,
|
||||
|
@ -937,3 +951,47 @@ class GroupServerStore(SQLBaseStore):
|
|||
retcol="group_id",
|
||||
desc="get_joined_groups",
|
||||
)
|
||||
|
||||
def get_all_groups_for_user(self, user_id, now_token):
|
||||
def _get_all_groups_for_user_txn(txn):
|
||||
sql = """
|
||||
SELECT group_id, type, membership, u.content
|
||||
FROM local_group_updates AS u
|
||||
INNER JOIN local_group_membership USING (group_id, user_id)
|
||||
WHERE user_id = ? AND membership != 'leave'
|
||||
AND stream_id <= ?
|
||||
"""
|
||||
txn.execute(sql, (user_id, now_token,))
|
||||
return self.cursor_to_dict(txn)
|
||||
return self.runInteraction(
|
||||
"get_all_groups_for_user", _get_all_groups_for_user_txn,
|
||||
)
|
||||
|
||||
def get_groups_changes_for_user(self, user_id, from_token, to_token):
|
||||
from_token = int(from_token)
|
||||
has_changed = self._group_updates_stream_cache.has_entity_changed(
|
||||
user_id, from_token,
|
||||
)
|
||||
if not has_changed:
|
||||
return []
|
||||
|
||||
def _get_groups_changes_for_user_txn(txn):
|
||||
sql = """
|
||||
SELECT group_id, membership, type, u.content
|
||||
FROM local_group_updates AS u
|
||||
INNER JOIN local_group_membership USING (group_id, user_id)
|
||||
WHERE user_id = ? AND ? < stream_id AND stream_id <= ?
|
||||
"""
|
||||
txn.execute(sql, (user_id, from_token, to_token,))
|
||||
return [{
|
||||
"group_id": group_id,
|
||||
"membership": membership,
|
||||
"type": gtype,
|
||||
"content": json.loads(content_json),
|
||||
} for group_id, membership, gtype, content_json in txn]
|
||||
return self.runInteraction(
|
||||
"get_groups_changes_for_user", _get_groups_changes_for_user_txn,
|
||||
)
|
||||
|
||||
def get_group_stream_token(self):
|
||||
return self._group_updates_id_gen.get_current_token()
|
||||
|
|
|
@ -155,3 +155,12 @@ CREATE TABLE local_group_membership (
|
|||
|
||||
CREATE INDEX local_group_membership_u_idx ON local_group_membership(user_id, group_id);
|
||||
CREATE INDEX local_group_membership_g_idx ON local_group_membership(group_id);
|
||||
|
||||
|
||||
CREATE TABLE local_group_updates (
|
||||
stream_id BIGINT NOT NULL,
|
||||
group_id TEXT NOT NULL,
|
||||
user_id TEXT NOT NULL,
|
||||
type TEXT NOT NULL,
|
||||
content TEXT NOT NULL
|
||||
);
|
||||
|
|
|
@ -45,6 +45,7 @@ class EventSources(object):
|
|||
push_rules_key, _ = self.store.get_push_rules_stream_token()
|
||||
to_device_key = self.store.get_to_device_stream_token()
|
||||
device_list_key = self.store.get_device_stream_token()
|
||||
groups_key = self.store.get_group_stream_token()
|
||||
|
||||
token = StreamToken(
|
||||
room_key=(
|
||||
|
@ -65,6 +66,7 @@ class EventSources(object):
|
|||
push_rules_key=push_rules_key,
|
||||
to_device_key=to_device_key,
|
||||
device_list_key=device_list_key,
|
||||
groups_key=groups_key,
|
||||
)
|
||||
defer.returnValue(token)
|
||||
|
||||
|
|
|
@ -171,6 +171,7 @@ class StreamToken(
|
|||
"push_rules_key",
|
||||
"to_device_key",
|
||||
"device_list_key",
|
||||
"groups_key",
|
||||
))
|
||||
):
|
||||
_SEPARATOR = "_"
|
||||
|
@ -209,6 +210,7 @@ class StreamToken(
|
|||
or (int(other.push_rules_key) < int(self.push_rules_key))
|
||||
or (int(other.to_device_key) < int(self.to_device_key))
|
||||
or (int(other.device_list_key) < int(self.device_list_key))
|
||||
or (int(other.groups_key) < int(self.groups_key))
|
||||
)
|
||||
|
||||
def copy_and_advance(self, key, new_value):
|
||||
|
|
|
@ -1032,7 +1032,7 @@ class RoomMessageListTestCase(RestTestCase):
|
|||
|
||||
@defer.inlineCallbacks
|
||||
def test_topo_token_is_accepted(self):
|
||||
token = "t1-0_0_0_0_0_0_0_0"
|
||||
token = "t1-0_0_0_0_0_0_0_0_0"
|
||||
(code, response) = yield self.mock_resource.trigger_get(
|
||||
"/rooms/%s/messages?access_token=x&from=%s" %
|
||||
(self.room_id, token))
|
||||
|
@ -1044,7 +1044,7 @@ class RoomMessageListTestCase(RestTestCase):
|
|||
|
||||
@defer.inlineCallbacks
|
||||
def test_stream_token_is_accepted_for_fwd_pagianation(self):
|
||||
token = "s0_0_0_0_0_0_0_0"
|
||||
token = "s0_0_0_0_0_0_0_0_0"
|
||||
(code, response) = yield self.mock_resource.trigger_get(
|
||||
"/rooms/%s/messages?access_token=x&from=%s" %
|
||||
(self.room_id, token))
|
||||
|
|
Loading…
Reference in a new issue