4532

Committed 23 Sep 2019 - 19:39 coverage decreased (-49.7%) to 17.596%

Build # 4532

Build Type

buildkite

Committed by Richard van der Hoff

Commit Message

update changelog

Pull Request Pull Request #6079: Add submit_url response parameter to msisdn /requestToken

Run Details

359 of 12986 branches covered (2.76%)

Branch coverage included in aggregate %.

0 of 7 new or added lines in 1 file covered. (0.0%)

18869 existing lines in 281 files now uncovered.

8809 of 39116 relevant lines covered (22.52%)

0.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

20.63

/synapse/storage/account_data.py

# -*- coding: utf-8 -*-
# Copyright 2014-2016 OpenMarket Ltd
# Copyright 2018 New Vector 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.

import abc
import logging

from canonicaljson import json

from twisted.internet import defer

from synapse.storage._base import SQLBaseStore
from synapse.storage.util.id_generators import StreamIdGenerator
from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
from synapse.util.caches.stream_change_cache import StreamChangeCache

logger = logging.getLogger(__name__)


class AccountDataWorkerStore(SQLBaseStore):
    """This is an abstract base class where subclasses must implement
    `get_max_account_data_stream_id` which can be called in the initializer.
    """

    # This ABCMeta metaclass ensures that we cannot be instantiated without
    # the abstract methods being implemented.
    __metaclass__ = abc.ABCMeta

    def __init__(self, db_conn, hs):
        account_max = self.get_max_account_data_stream_id()
        self._account_data_stream_cache = StreamChangeCache(
            "AccountDataAndTagsChangeCache", account_max
        )

        super(AccountDataWorkerStore, self).__init__(db_conn, hs)

    @abc.abstractmethod
    def get_max_account_data_stream_id(self):
        """Get the current max stream ID for account data stream

        Returns:
            int
        """
        raise NotImplementedError()

    @cached()
    def get_account_data_for_user(self, user_id):
        """Get all the client account_data for a user.

        Args:
            user_id(str): The user to get the account_data for.
        Returns:
            A deferred pair of a dict of global account_data and a dict
            mapping from room_id string to per room account_data dicts.
        """

        def get_account_data_for_user_txn(txn):
            rows = self._simple_select_list_txn(
                txn,
                "account_data",
                {"user_id": user_id},
                ["account_data_type", "content"],
            )

            global_account_data = {
                row["account_data_type"]: json.loads(row["content"]) for row in rows
            }

            rows = self._simple_select_list_txn(
                txn,
                "room_account_data",
                {"user_id": user_id},
                ["room_id", "account_data_type", "content"],
            )

            by_room = {}
            for row in rows:
                room_data = by_room.setdefault(row["room_id"], {})
                room_data[row["account_data_type"]] = json.loads(row["content"])

            return global_account_data, by_room

        return self.runInteraction(
            "get_account_data_for_user", get_account_data_for_user_txn
        )

    @cachedInlineCallbacks(num_args=2, max_entries=5000)
    def get_global_account_data_by_type_for_user(self, data_type, user_id):
        """
        Returns:
            Deferred: A dict
        """
        result = yield self._simple_select_one_onecol(
            table="account_data",
            keyvalues={"user_id": user_id, "account_data_type": data_type},
            retcol="content",
            desc="get_global_account_data_by_type_for_user",
            allow_none=True,
        )

        if result:
            return json.loads(result)
        else:
            return None

    @cached(num_args=2)
    def get_account_data_for_room(self, user_id, room_id):
        """Get all the client account_data for a user for a room.

        Args:
            user_id(str): The user to get the account_data for.
            room_id(str): The room to get the account_data for.
        Returns:
            A deferred dict of the room account_data
        """

        def get_account_data_for_room_txn(txn):
            rows = self._simple_select_list_txn(
                txn,
                "room_account_data",
                {"user_id": user_id, "room_id": room_id},
                ["account_data_type", "content"],
            )

            return {
                row["account_data_type"]: json.loads(row["content"]) for row in rows
            }

        return self.runInteraction(
            "get_account_data_for_room", get_account_data_for_room_txn
        )

    @cached(num_args=3, max_entries=5000)
    def get_account_data_for_room_and_type(self, user_id, room_id, account_data_type):
        """Get the client account_data of given type for a user for a room.

        Args:
            user_id(str): The user to get the account_data for.
            room_id(str): The room to get the account_data for.
            account_data_type (str): The account data type to get.
        Returns:
            A deferred of the room account_data for that type, or None if
            there isn't any set.
        """

        def get_account_data_for_room_and_type_txn(txn):
            content_json = self._simple_select_one_onecol_txn(
                txn,
                table="room_account_data",
                keyvalues={
                    "user_id": user_id,
                    "room_id": room_id,
                    "account_data_type": account_data_type,
                },
                retcol="content",
                allow_none=True,
            )

            return json.loads(content_json) if content_json else None

        return self.runInteraction(
            "get_account_data_for_room_and_type", get_account_data_for_room_and_type_txn
        )

    def get_all_updated_account_data(
        self, last_global_id, last_room_id, current_id, limit
    ):
        """Get all the client account_data that has changed on the server
        Args:
            last_global_id(int): The position to fetch from for top level data
            last_room_id(int): The position to fetch from for per room data
            current_id(int): The position to fetch up to.
        Returns:
            A deferred pair of lists of tuples of stream_id int, user_id string,
            room_id string, type string, and content string.
        """
        if last_room_id == current_id and last_global_id == current_id:
            return defer.succeed(([], []))

        def get_updated_account_data_txn(txn):
            sql = (
                "SELECT stream_id, user_id, account_data_type, content"
                " FROM account_data WHERE ? < stream_id AND stream_id <= ?"
                " ORDER BY stream_id ASC LIMIT ?"
            )
            txn.execute(sql, (last_global_id, current_id, limit))
            global_results = txn.fetchall()

            sql = (
                "SELECT stream_id, user_id, room_id, account_data_type, content"
                " FROM room_account_data WHERE ? < stream_id AND stream_id <= ?"
                " ORDER BY stream_id ASC LIMIT ?"
            )
            txn.execute(sql, (last_room_id, current_id, limit))
            room_results = txn.fetchall()
            return global_results, room_results

        return self.runInteraction(
            "get_all_updated_account_data_txn", get_updated_account_data_txn
        )

    def get_updated_account_data_for_user(self, user_id, stream_id):
        """Get all the client account_data for a that's changed for a user

        Args:
            user_id(str): The user to get the account_data for.
            stream_id(int): The point in the stream since which to get updates
        Returns:
            A deferred pair of a dict of global account_data and a dict
            mapping from room_id string to per room account_data dicts.
        """

        def get_updated_account_data_for_user_txn(txn):
            sql = (
                "SELECT account_data_type, content FROM account_data"
                " WHERE user_id = ? AND stream_id > ?"
            )

            txn.execute(sql, (user_id, stream_id))

            global_account_data = {row[0]: json.loads(row[1]) for row in txn}

            sql = (
                "SELECT room_id, account_data_type, content FROM room_account_data"
                " WHERE user_id = ? AND stream_id > ?"
            )

            txn.execute(sql, (user_id, stream_id))

            account_data_by_room = {}
            for row in txn:
                room_account_data = account_data_by_room.setdefault(row[0], {})
                room_account_data[row[1]] = json.loads(row[2])

            return global_account_data, account_data_by_room

        changed = self._account_data_stream_cache.has_entity_changed(
            user_id, int(stream_id)
        )
        if not changed:
            return {}, {}

        return self.runInteraction(
            "get_updated_account_data_for_user", get_updated_account_data_for_user_txn
        )

    @cachedInlineCallbacks(num_args=2, cache_context=True, max_entries=5000)
    def is_ignored_by(self, ignored_user_id, ignorer_user_id, cache_context):
        ignored_account_data = yield self.get_global_account_data_by_type_for_user(
            "m.ignored_user_list",
            ignorer_user_id,
            on_invalidate=cache_context.invalidate,
        )
        if not ignored_account_data:
            return False

        return ignored_user_id in ignored_account_data.get("ignored_users", {})


class AccountDataStore(AccountDataWorkerStore):
    def __init__(self, db_conn, hs):
        self._account_data_id_gen = StreamIdGenerator(
            db_conn, "account_data_max_stream_id", "stream_id"
        )

        super(AccountDataStore, self).__init__(db_conn, hs)

    def get_max_account_data_stream_id(self):
        """Get the current max stream id for the private user data stream

        Returns:
            A deferred int.
        """
        return self._account_data_id_gen.get_current_token()

    @defer.inlineCallbacks
    def add_account_data_to_room(self, user_id, room_id, account_data_type, content):
        """Add some account_data to a room for a user.
        Args:
            user_id(str): The user to add a tag for.
            room_id(str): The room to add a tag for.
            account_data_type(str): The type of account_data to add.
            content(dict): A json object to associate with the tag.
        Returns:
            A deferred that completes once the account_data has been added.
        """
        content_json = json.dumps(content)

        with self._account_data_id_gen.get_next() as next_id:
            # no need to lock here as room_account_data has a unique constraint
            # on (user_id, room_id, account_data_type) so _simple_upsert will
            # retry if there is a conflict.
            yield self._simple_upsert(
                desc="add_room_account_data",
                table="room_account_data",
                keyvalues={
                    "user_id": user_id,
                    "room_id": room_id,
                    "account_data_type": account_data_type,
                },
                values={"stream_id": next_id, "content": content_json},
                lock=False,
            )

            # it's theoretically possible for the above to succeed and the
            # below to fail - in which case we might reuse a stream id on
            # restart, and the above update might not get propagated. That
            # doesn't sound any worse than the whole update getting lost,
            # which is what would happen if we combined the two into one
            # transaction.
            yield self._update_max_stream_id(next_id)

            self._account_data_stream_cache.entity_has_changed(user_id, next_id)
            self.get_account_data_for_user.invalidate((user_id,))
            self.get_account_data_for_room.invalidate((user_id, room_id))
            self.get_account_data_for_room_and_type.prefill(
                (user_id, room_id, account_data_type), content
            )

        result = self._account_data_id_gen.get_current_token()
        return result

    @defer.inlineCallbacks
    def add_account_data_for_user(self, user_id, account_data_type, content):
        """Add some account_data to a room for a user.
        Args:
            user_id(str): The user to add a tag for.
            account_data_type(str): The type of account_data to add.
            content(dict): A json object to associate with the tag.
        Returns:
            A deferred that completes once the account_data has been added.
        """
        content_json = json.dumps(content)

        with self._account_data_id_gen.get_next() as next_id:
            # no need to lock here as account_data has a unique constraint on
            # (user_id, account_data_type) so _simple_upsert will retry if
            # there is a conflict.
            yield self._simple_upsert(
                desc="add_user_account_data",
                table="account_data",
                keyvalues={"user_id": user_id, "account_data_type": account_data_type},
                values={"stream_id": next_id, "content": content_json},
                lock=False,
            )

            # it's theoretically possible for the above to succeed and the
            # below to fail - in which case we might reuse a stream id on
            # restart, and the above update might not get propagated. That
            # doesn't sound any worse than the whole update getting lost,
            # which is what would happen if we combined the two into one
            # transaction.
            yield self._update_max_stream_id(next_id)

            self._account_data_stream_cache.entity_has_changed(user_id, next_id)
            self.get_account_data_for_user.invalidate((user_id,))
            self.get_global_account_data_by_type_for_user.invalidate(
                (account_data_type, user_id)
            )

        result = self._account_data_id_gen.get_current_token()
        return result

    def _update_max_stream_id(self, next_id):
        """Update the max stream_id

        Args:
            next_id(int): The the revision to advance to.
        """

        def _update(txn):
            update_max_id_sql = (
                "UPDATE account_data_max_stream_id"
                " SET stream_id = ?"
                " WHERE stream_id < ?"
            )
            txn.execute(update_max_id_sql, (next_id, next_id))

        return self.runInteraction("update_account_data_max_stream_id", _update)

1	# -- coding: utf-8 --
2	# Copyright 2014-2016 OpenMarket Ltd
3	# Copyright 2018 New Vector Ltd
4	#
5	# Licensed under the Apache License, Version 2.0 (the "License");
6	# you may not use this file except in compliance with the License.
7	# You may obtain a copy of the License at
8	#
9	# http://www.apache.org/licenses/LICENSE-2.0
10	#
11	# Unless required by applicable law or agreed to in writing, software
12	# distributed under the License is distributed on an "AS IS" BASIS,
13	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	# See the License for the specific language governing permissions and
15	# limitations under the License.
16
17	import abc	1×
18	import logging	1×
19
20	from canonicaljson import json	1×
21
22	from twisted.internet import defer	1×
23
24	from synapse.storage._base import SQLBaseStore	1×
25	from synapse.storage.util.id_generators import StreamIdGenerator	1×
26	from synapse.util.caches.descriptors import cached, cachedInlineCallbacks	1×
27	from synapse.util.caches.stream_change_cache import StreamChangeCache	1×
28
29	logger = logging.getLogger(__name__)	1×
30
31
32	class AccountDataWorkerStore(SQLBaseStore):	1×
33	"""This is an abstract base class where subclasses must implement
34	`get_max_account_data_stream_id` which can be called in the initializer.
35	"""
36
37	# This ABCMeta metaclass ensures that we cannot be instantiated without
38	# the abstract methods being implemented.
39	__metaclass__ = abc.ABCMeta	1×
40
41	def __init__(self, db_conn, hs):	1×
UNCOV 42	account_max = self.get_max_account_data_stream_id()	!
UNCOV 43	self._account_data_stream_cache = StreamChangeCache(	!
44	"AccountDataAndTagsChangeCache", account_max
45	)
46
UNCOV 47	super(AccountDataWorkerStore, self).__init__(db_conn, hs)	!
48
49	@abc.abstractmethod	1×
50	def get_max_account_data_stream_id(self):
51	"""Get the current max stream ID for account data stream
52
53	Returns:
54	int
55	"""
56	raise NotImplementedError()	!
57
58	@cached()	1×
59	def get_account_data_for_user(self, user_id):
60	"""Get all the client account_data for a user.
61
62	Args:
63	user_id(str): The user to get the account_data for.
64	Returns:
65	A deferred pair of a dict of global account_data and a dict
66	mapping from room_id string to per room account_data dicts.
67	"""
68
UNCOV 69	def get_account_data_for_user_txn(txn):	!
UNCOV 70	rows = self._simple_select_list_txn(	!
71	txn,
72	"account_data",
73	{"user_id": user_id},
74	["account_data_type", "content"],
75	)
76
UNCOV 77	global_account_data = {	Branches [[0, 77], [0, 81]] missed. !
78	row["account_data_type"]: json.loads(row["content"]) for row in rows
79	}
80
UNCOV 81	rows = self._simple_select_list_txn(	!
82	txn,
83	"room_account_data",
84	{"user_id": user_id},
85	["room_id", "account_data_type", "content"],
86	)
87
UNCOV 88	by_room = {}	!
UNCOV 89	for row in rows:	Branches [[0, 90], [0, 93]] missed. !
UNCOV 90	room_data = by_room.setdefault(row["room_id"], {})	!
UNCOV 91	room_data[row["account_data_type"]] = json.loads(row["content"])	!
92
UNCOV 93	return global_account_data, by_room	!
94
UNCOV 95	return self.runInteraction(	!
96	"get_account_data_for_user", get_account_data_for_user_txn
97	)
98
99	@cachedInlineCallbacks(num_args=2, max_entries=5000)	1×
100	def get_global_account_data_by_type_for_user(self, data_type, user_id):
101	"""
102	Returns:
103	Deferred: A dict
104	"""
UNCOV 105	result = yield self._simple_select_one_onecol(	!
106	table="account_data",
107	keyvalues={"user_id": user_id, "account_data_type": data_type},
108	retcol="content",
109	desc="get_global_account_data_by_type_for_user",
110	allow_none=True,
111	)
112
UNCOV 113	if result:	Branches [[0, 114], [0, 116]] missed. !
UNCOV 114	return json.loads(result)	!
115	else:
UNCOV 116	return None	!
117
118	@cached(num_args=2)	1×
119	def get_account_data_for_room(self, user_id, room_id):
120	"""Get all the client account_data for a user for a room.
121
122	Args:
123	user_id(str): The user to get the account_data for.
124	room_id(str): The room to get the account_data for.
125	Returns:
126	A deferred dict of the room account_data
127	"""
128
UNCOV 129	def get_account_data_for_room_txn(txn):	!
UNCOV 130	rows = self._simple_select_list_txn(	!
131	txn,
132	"room_account_data",
133	{"user_id": user_id, "room_id": room_id},
134	["account_data_type", "content"],
135	)
136
UNCOV 137	return {	Branches [[0, 137], [0, 129]] missed. !
138	row["account_data_type"]: json.loads(row["content"]) for row in rows
139	}
140
UNCOV 141	return self.runInteraction(	!
142	"get_account_data_for_room", get_account_data_for_room_txn
143	)
144
145	@cached(num_args=3, max_entries=5000)	1×
146	def get_account_data_for_room_and_type(self, user_id, room_id, account_data_type):
147	"""Get the client account_data of given type for a user for a room.
148
149	Args:
150	user_id(str): The user to get the account_data for.
151	room_id(str): The room to get the account_data for.
152	account_data_type (str): The account data type to get.
153	Returns:
154	A deferred of the room account_data for that type, or None if
155	there isn't any set.
156	"""
157
UNCOV 158	def get_account_data_for_room_and_type_txn(txn):	!
UNCOV 159	content_json = self._simple_select_one_onecol_txn(	!
160	txn,
161	table="room_account_data",
162	keyvalues={
163	"user_id": user_id,
164	"room_id": room_id,
165	"account_data_type": account_data_type,
166	},
167	retcol="content",
168	allow_none=True,
169	)
170
UNCOV 171	return json.loads(content_json) if content_json else None	!
172
UNCOV 173	return self.runInteraction(	!
174	"get_account_data_for_room_and_type", get_account_data_for_room_and_type_txn
175	)
176
177	def get_all_updated_account_data(	1×
178	self, last_global_id, last_room_id, current_id, limit
179	):
180	"""Get all the client account_data that has changed on the server
181	Args:
182	last_global_id(int): The position to fetch from for top level data
183	last_room_id(int): The position to fetch from for per room data
184	current_id(int): The position to fetch up to.
185	Returns:
186	A deferred pair of lists of tuples of stream_id int, user_id string,
187	room_id string, type string, and content string.
188	"""
189	if last_room_id == current_id and last_global_id == current_id:	Branches [[0, 190], [0, 192]] missed. !
190	return defer.succeed(([], []))	!
191
192	def get_updated_account_data_txn(txn):	!
193	sql = (	!
194	"SELECT stream_id, user_id, account_data_type, content"
195	" FROM account_data WHERE ? < stream_id AND stream_id <= ?"
196	" ORDER BY stream_id ASC LIMIT ?"
197	)
198	txn.execute(sql, (last_global_id, current_id, limit))	!
199	global_results = txn.fetchall()	!
200
201	sql = (	!
202	"SELECT stream_id, user_id, room_id, account_data_type, content"
203	" FROM room_account_data WHERE ? < stream_id AND stream_id <= ?"
204	" ORDER BY stream_id ASC LIMIT ?"
205	)
206	txn.execute(sql, (last_room_id, current_id, limit))	!
207	room_results = txn.fetchall()	!
208	return global_results, room_results	!
209
210	return self.runInteraction(	!
211	"get_all_updated_account_data_txn", get_updated_account_data_txn
212	)
213
214	def get_updated_account_data_for_user(self, user_id, stream_id):	1×
215	"""Get all the client account_data for a that's changed for a user
216
217	Args:
218	user_id(str): The user to get the account_data for.
219	stream_id(int): The point in the stream since which to get updates
220	Returns:
221	A deferred pair of a dict of global account_data and a dict
222	mapping from room_id string to per room account_data dicts.
223	"""
224
UNCOV 225	def get_updated_account_data_for_user_txn(txn):	!
UNCOV 226	sql = (	!
227	"SELECT account_data_type, content FROM account_data"
228	" WHERE user_id = ? AND stream_id > ?"
229	)
230
UNCOV 231	txn.execute(sql, (user_id, stream_id))	!
232
UNCOV 233	global_account_data = {row[0]: json.loads(row[1]) for row in txn}	Branches [[0, 233], [0, 235]] missed. !
234
UNCOV 235	sql = (	!
236	"SELECT room_id, account_data_type, content FROM room_account_data"
237	" WHERE user_id = ? AND stream_id > ?"
238	)
239
UNCOV 240	txn.execute(sql, (user_id, stream_id))	!
241
UNCOV 242	account_data_by_room = {}	!
UNCOV 243	for row in txn:	Branches [[0, 244], [0, 247]] missed. !
UNCOV 244	room_account_data = account_data_by_room.setdefault(row[0], {})	!
UNCOV 245	room_account_data[row[1]] = json.loads(row[2])	!
246
UNCOV 247	return global_account_data, account_data_by_room	!
248
UNCOV 249	changed = self._account_data_stream_cache.has_entity_changed(	!
250	user_id, int(stream_id)
251	)
UNCOV 252	if not changed:	Branches [[0, 253], [0, 255]] missed. !
UNCOV 253	return {}, {}	!
254
UNCOV 255	return self.runInteraction(	!
256	"get_updated_account_data_for_user", get_updated_account_data_for_user_txn
257	)
258
259	@cachedInlineCallbacks(num_args=2, cache_context=True, max_entries=5000)	1×
260	def is_ignored_by(self, ignored_user_id, ignorer_user_id, cache_context):
UNCOV 261	ignored_account_data = yield self.get_global_account_data_by_type_for_user(	!
262	"m.ignored_user_list",
263	ignorer_user_id,
264	on_invalidate=cache_context.invalidate,
265	)
UNCOV 266	if not ignored_account_data:	Branches [[0, 267], [0, 269]] missed. !
UNCOV 267	return False	!
268
269	return ignored_user_id in ignored_account_data.get("ignored_users", {})	!
270
271
272	class AccountDataStore(AccountDataWorkerStore):	1×
273	def __init__(self, db_conn, hs):	1×
UNCOV 274	self._account_data_id_gen = StreamIdGenerator(	!
275	db_conn, "account_data_max_stream_id", "stream_id"
276	)
277
UNCOV 278	super(AccountDataStore, self).__init__(db_conn, hs)	!
279
280	def get_max_account_data_stream_id(self):	1×
281	"""Get the current max stream id for the private user data stream
282
283	Returns:
284	A deferred int.
285	"""
UNCOV 286	return self._account_data_id_gen.get_current_token()	!
287
288	@defer.inlineCallbacks	1×
289	def add_account_data_to_room(self, user_id, room_id, account_data_type, content):
290	"""Add some account_data to a room for a user.
291	Args:
292	user_id(str): The user to add a tag for.
293	room_id(str): The room to add a tag for.
294	account_data_type(str): The type of account_data to add.
295	content(dict): A json object to associate with the tag.
296	Returns:
297	A deferred that completes once the account_data has been added.
298	"""
UNCOV 299	content_json = json.dumps(content)	!
300
UNCOV 301	with self._account_data_id_gen.get_next() as next_id:	!
302	# no need to lock here as room_account_data has a unique constraint
303	# on (user_id, room_id, account_data_type) so _simple_upsert will
304	# retry if there is a conflict.
UNCOV 305	yield self._simple_upsert(	!
306	desc="add_room_account_data",
307	table="room_account_data",
308	keyvalues={
309	"user_id": user_id,
310	"room_id": room_id,
311	"account_data_type": account_data_type,
312	},
313	values={"stream_id": next_id, "content": content_json},
314	lock=False,
315	)
316
317	# it's theoretically possible for the above to succeed and the
318	# below to fail - in which case we might reuse a stream id on
319	# restart, and the above update might not get propagated. That
320	# doesn't sound any worse than the whole update getting lost,
321	# which is what would happen if we combined the two into one
322	# transaction.
UNCOV 323	yield self._update_max_stream_id(next_id)	!
324
UNCOV 325	self._account_data_stream_cache.entity_has_changed(user_id, next_id)	!
UNCOV 326	self.get_account_data_for_user.invalidate((user_id,))	!
UNCOV 327	self.get_account_data_for_room.invalidate((user_id, room_id))	!
UNCOV 328	self.get_account_data_for_room_and_type.prefill(	!
329	(user_id, room_id, account_data_type), content
330	)
331
UNCOV 332	result = self._account_data_id_gen.get_current_token()	!
UNCOV 333	return result	!
334
335	@defer.inlineCallbacks	1×
336	def add_account_data_for_user(self, user_id, account_data_type, content):
337	"""Add some account_data to a room for a user.
338	Args:
339	user_id(str): The user to add a tag for.
340	account_data_type(str): The type of account_data to add.
341	content(dict): A json object to associate with the tag.
342	Returns:
343	A deferred that completes once the account_data has been added.
344	"""
UNCOV 345	content_json = json.dumps(content)	!
346
UNCOV 347	with self._account_data_id_gen.get_next() as next_id:	!
348	# no need to lock here as account_data has a unique constraint on
349	# (user_id, account_data_type) so _simple_upsert will retry if
350	# there is a conflict.
UNCOV 351	yield self._simple_upsert(	!
352	desc="add_user_account_data",
353	table="account_data",
354	keyvalues={"user_id": user_id, "account_data_type": account_data_type},
355	values={"stream_id": next_id, "content": content_json},
356	lock=False,
357	)
358
359	# it's theoretically possible for the above to succeed and the
360	# below to fail - in which case we might reuse a stream id on
361	# restart, and the above update might not get propagated. That
362	# doesn't sound any worse than the whole update getting lost,
363	# which is what would happen if we combined the two into one
364	# transaction.
UNCOV 365	yield self._update_max_stream_id(next_id)	!
366
UNCOV 367	self._account_data_stream_cache.entity_has_changed(user_id, next_id)	!
UNCOV 368	self.get_account_data_for_user.invalidate((user_id,))	!
UNCOV 369	self.get_global_account_data_by_type_for_user.invalidate(	!
370	(account_data_type, user_id)
371	)
372
UNCOV 373	result = self._account_data_id_gen.get_current_token()	!
UNCOV 374	return result	!
375
376	def _update_max_stream_id(self, next_id):	1×
377	"""Update the max stream_id
378
379	Args:
380	next_id(int): The the revision to advance to.
381	"""
382
UNCOV 383	def _update(txn):	!
UNCOV 384	update_max_id_sql = (	!
385	"UPDATE account_data_max_stream_id"
386	" SET stream_id = ?"
387	" WHERE stream_id < ?"
388	)
UNCOV 389	txn.execute(update_max_id_sql, (next_id, next_id))	!
390
UNCOV 391	return self.runInteraction("update_account_data_max_stream_id", _update)	!

matrix-org / synapse / 4532

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous