You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

808 lines
30 KiB

  1. # Copyright 2016 OpenMarket Ltd
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. import logging
  15. from typing import TYPE_CHECKING, Dict, List, Mapping, Optional, Tuple, Union, cast
  16. import attr
  17. from typing_extensions import TypedDict
  18. from synapse.metrics.background_process_metrics import wrap_as_background_process
  19. from synapse.storage._base import SQLBaseStore
  20. from synapse.storage.database import (
  21. DatabasePool,
  22. LoggingDatabaseConnection,
  23. LoggingTransaction,
  24. make_tuple_comparison_clause,
  25. )
  26. from synapse.storage.databases.main.monthly_active_users import (
  27. MonthlyActiveUsersWorkerStore,
  28. )
  29. from synapse.types import JsonDict, UserID
  30. from synapse.util.caches.lrucache import LruCache
  31. if TYPE_CHECKING:
  32. from synapse.server import HomeServer
  33. logger = logging.getLogger(__name__)
  34. # Number of msec of granularity to store the user IP 'last seen' time. Smaller
  35. # times give more inserts into the database even for readonly API hits
  36. # 120 seconds == 2 minutes
  37. LAST_SEEN_GRANULARITY = 120 * 1000
  38. @attr.s(slots=True, frozen=True, auto_attribs=True)
  39. class DeviceLastConnectionInfo:
  40. """Metadata for the last connection seen for a user and device combination"""
  41. # These types must match the columns in the `devices` table
  42. user_id: str
  43. device_id: str
  44. ip: Optional[str]
  45. user_agent: Optional[str]
  46. last_seen: Optional[int]
  47. class LastConnectionInfo(TypedDict):
  48. """Metadata for the last connection seen for an access token and IP combination"""
  49. # These types must match the columns in the `user_ips` table
  50. access_token: str
  51. ip: str
  52. user_agent: str
  53. last_seen: int
  54. class ClientIpBackgroundUpdateStore(SQLBaseStore):
  55. def __init__(
  56. self,
  57. database: DatabasePool,
  58. db_conn: LoggingDatabaseConnection,
  59. hs: "HomeServer",
  60. ):
  61. super().__init__(database, db_conn, hs)
  62. self.db_pool.updates.register_background_index_update(
  63. "user_ips_device_index",
  64. index_name="user_ips_device_id",
  65. table="user_ips",
  66. columns=["user_id", "device_id", "last_seen"],
  67. )
  68. self.db_pool.updates.register_background_index_update(
  69. "user_ips_last_seen_index",
  70. index_name="user_ips_last_seen",
  71. table="user_ips",
  72. columns=["user_id", "last_seen"],
  73. )
  74. self.db_pool.updates.register_background_index_update(
  75. "user_ips_last_seen_only_index",
  76. index_name="user_ips_last_seen_only",
  77. table="user_ips",
  78. columns=["last_seen"],
  79. )
  80. self.db_pool.updates.register_background_update_handler(
  81. "user_ips_analyze", self._analyze_user_ip
  82. )
  83. self.db_pool.updates.register_background_update_handler(
  84. "user_ips_remove_dupes", self._remove_user_ip_dupes
  85. )
  86. # Register a unique index
  87. self.db_pool.updates.register_background_index_update(
  88. "user_ips_device_unique_index",
  89. index_name="user_ips_user_token_ip_unique_index",
  90. table="user_ips",
  91. columns=["user_id", "access_token", "ip"],
  92. unique=True,
  93. )
  94. # Drop the old non-unique index
  95. self.db_pool.updates.register_background_update_handler(
  96. "user_ips_drop_nonunique_index", self._remove_user_ip_nonunique
  97. )
  98. # Update the last seen info in devices.
  99. self.db_pool.updates.register_background_update_handler(
  100. "devices_last_seen", self._devices_last_seen_update
  101. )
  102. async def _remove_user_ip_nonunique(
  103. self, progress: JsonDict, batch_size: int
  104. ) -> int:
  105. def f(conn: LoggingDatabaseConnection) -> None:
  106. txn = conn.cursor()
  107. txn.execute("DROP INDEX IF EXISTS user_ips_user_ip")
  108. txn.close()
  109. await self.db_pool.runWithConnection(f)
  110. await self.db_pool.updates._end_background_update(
  111. "user_ips_drop_nonunique_index"
  112. )
  113. return 1
  114. async def _analyze_user_ip(self, progress: JsonDict, batch_size: int) -> int:
  115. # Background update to analyze user_ips table before we run the
  116. # deduplication background update. The table may not have been analyzed
  117. # for ages due to the table locks.
  118. #
  119. # This will lock out the naive upserts to user_ips while it happens, but
  120. # the analyze should be quick (28GB table takes ~10s)
  121. def user_ips_analyze(txn: LoggingTransaction) -> None:
  122. txn.execute("ANALYZE user_ips")
  123. await self.db_pool.runInteraction("user_ips_analyze", user_ips_analyze)
  124. await self.db_pool.updates._end_background_update("user_ips_analyze")
  125. return 1
  126. async def _remove_user_ip_dupes(self, progress: JsonDict, batch_size: int) -> int:
  127. # This works function works by scanning the user_ips table in batches
  128. # based on `last_seen`. For each row in a batch it searches the rest of
  129. # the table to see if there are any duplicates, if there are then they
  130. # are removed and replaced with a suitable row.
  131. # Fetch the start of the batch
  132. begin_last_seen: int = progress.get("last_seen", 0)
  133. def get_last_seen(txn: LoggingTransaction) -> Optional[int]:
  134. txn.execute(
  135. """
  136. SELECT last_seen FROM user_ips
  137. WHERE last_seen > ?
  138. ORDER BY last_seen
  139. LIMIT 1
  140. OFFSET ?
  141. """,
  142. (begin_last_seen, batch_size),
  143. )
  144. row = cast(Optional[Tuple[int]], txn.fetchone())
  145. if row:
  146. return row[0]
  147. else:
  148. return None
  149. # Get a last seen that has roughly `batch_size` since `begin_last_seen`
  150. end_last_seen = await self.db_pool.runInteraction(
  151. "user_ips_dups_get_last_seen", get_last_seen
  152. )
  153. # If it returns None, then we're processing the last batch
  154. last = end_last_seen is None
  155. logger.info(
  156. "Scanning for duplicate 'user_ips' rows in range: %s <= last_seen < %s",
  157. begin_last_seen,
  158. end_last_seen,
  159. )
  160. def remove(txn: LoggingTransaction) -> None:
  161. # This works by looking at all entries in the given time span, and
  162. # then for each (user_id, access_token, ip) tuple in that range
  163. # checking for any duplicates in the rest of the table (via a join).
  164. # It then only returns entries which have duplicates, and the max
  165. # last_seen across all duplicates, which can the be used to delete
  166. # all other duplicates.
  167. # It is efficient due to the existence of (user_id, access_token,
  168. # ip) and (last_seen) indices.
  169. # Define the search space, which requires handling the last batch in
  170. # a different way
  171. args: Tuple[int, ...]
  172. if last:
  173. clause = "? <= last_seen"
  174. args = (begin_last_seen,)
  175. else:
  176. assert end_last_seen is not None
  177. clause = "? <= last_seen AND last_seen < ?"
  178. args = (begin_last_seen, end_last_seen)
  179. # (Note: The DISTINCT in the inner query is important to ensure that
  180. # the COUNT(*) is accurate, otherwise double counting may happen due
  181. # to the join effectively being a cross product)
  182. txn.execute(
  183. """
  184. SELECT user_id, access_token, ip,
  185. MAX(device_id), MAX(user_agent), MAX(last_seen),
  186. COUNT(*)
  187. FROM (
  188. SELECT DISTINCT user_id, access_token, ip
  189. FROM user_ips
  190. WHERE {}
  191. ) c
  192. INNER JOIN user_ips USING (user_id, access_token, ip)
  193. GROUP BY user_id, access_token, ip
  194. HAVING count(*) > 1
  195. """.format(
  196. clause
  197. ),
  198. args,
  199. )
  200. res = cast(
  201. List[Tuple[str, str, str, Optional[str], str, int, int]], txn.fetchall()
  202. )
  203. # We've got some duplicates
  204. for i in res:
  205. user_id, access_token, ip, device_id, user_agent, last_seen, count = i
  206. # We want to delete the duplicates so we end up with only a
  207. # single row.
  208. #
  209. # The naive way of doing this would be just to delete all rows
  210. # and reinsert a constructed row. However, if there are a lot of
  211. # duplicate rows this can cause the table to grow a lot, which
  212. # can be problematic in two ways:
  213. # 1. If user_ips is already large then this can cause the
  214. # table to rapidly grow, potentially filling the disk.
  215. # 2. Reinserting a lot of rows can confuse the table
  216. # statistics for postgres, causing it to not use the
  217. # correct indices for the query above, resulting in a full
  218. # table scan. This is incredibly slow for large tables and
  219. # can kill database performance. (This seems to mainly
  220. # happen for the last query where the clause is simply `? <
  221. # last_seen`)
  222. #
  223. # So instead we want to delete all but *one* of the duplicate
  224. # rows. That is hard to do reliably, so we cheat and do a two
  225. # step process:
  226. # 1. Delete all rows with a last_seen strictly less than the
  227. # max last_seen. This hopefully results in deleting all but
  228. # one row the majority of the time, but there may be
  229. # duplicate last_seen
  230. # 2. If multiple rows remain, we fall back to the naive method
  231. # and simply delete all rows and reinsert.
  232. #
  233. # Note that this relies on no new duplicate rows being inserted,
  234. # but if that is happening then this entire process is futile
  235. # anyway.
  236. # Do step 1:
  237. txn.execute(
  238. """
  239. DELETE FROM user_ips
  240. WHERE user_id = ? AND access_token = ? AND ip = ? AND last_seen < ?
  241. """,
  242. (user_id, access_token, ip, last_seen),
  243. )
  244. if txn.rowcount == count - 1:
  245. # We deleted all but one of the duplicate rows, i.e. there
  246. # is exactly one remaining and so there is nothing left to
  247. # do.
  248. continue
  249. elif txn.rowcount >= count:
  250. raise Exception(
  251. "We deleted more duplicate rows from 'user_ips' than expected"
  252. )
  253. # The previous step didn't delete enough rows, so we fallback to
  254. # step 2:
  255. # Drop all the duplicates
  256. txn.execute(
  257. """
  258. DELETE FROM user_ips
  259. WHERE user_id = ? AND access_token = ? AND ip = ?
  260. """,
  261. (user_id, access_token, ip),
  262. )
  263. # Add in one to be the last_seen
  264. txn.execute(
  265. """
  266. INSERT INTO user_ips
  267. (user_id, access_token, ip, device_id, user_agent, last_seen)
  268. VALUES (?, ?, ?, ?, ?, ?)
  269. """,
  270. (user_id, access_token, ip, device_id, user_agent, last_seen),
  271. )
  272. self.db_pool.updates._background_update_progress_txn(
  273. txn, "user_ips_remove_dupes", {"last_seen": end_last_seen}
  274. )
  275. await self.db_pool.runInteraction("user_ips_dups_remove", remove)
  276. if last:
  277. await self.db_pool.updates._end_background_update("user_ips_remove_dupes")
  278. return batch_size
  279. async def _devices_last_seen_update(
  280. self, progress: JsonDict, batch_size: int
  281. ) -> int:
  282. """Background update to insert last seen info into devices table"""
  283. last_user_id: str = progress.get("last_user_id", "")
  284. last_device_id: str = progress.get("last_device_id", "")
  285. def _devices_last_seen_update_txn(txn: LoggingTransaction) -> int:
  286. # This consists of two queries:
  287. #
  288. # 1. The sub-query searches for the next N devices and joins
  289. # against user_ips to find the max last_seen associated with
  290. # that device.
  291. # 2. The outer query then joins again against user_ips on
  292. # user/device/last_seen. This *should* hopefully only
  293. # return one row, but if it does return more than one then
  294. # we'll just end up updating the same device row multiple
  295. # times, which is fine.
  296. where_args: List[Union[str, int]]
  297. where_clause, where_args = make_tuple_comparison_clause(
  298. [("user_id", last_user_id), ("device_id", last_device_id)],
  299. )
  300. sql = """
  301. SELECT
  302. last_seen, ip, user_agent, user_id, device_id
  303. FROM (
  304. SELECT
  305. user_id, device_id, MAX(u.last_seen) AS last_seen
  306. FROM devices
  307. INNER JOIN user_ips AS u USING (user_id, device_id)
  308. WHERE %(where_clause)s
  309. GROUP BY user_id, device_id
  310. ORDER BY user_id ASC, device_id ASC
  311. LIMIT ?
  312. ) c
  313. INNER JOIN user_ips AS u USING (user_id, device_id, last_seen)
  314. """ % {
  315. "where_clause": where_clause
  316. }
  317. txn.execute(sql, where_args + [batch_size])
  318. rows = cast(List[Tuple[int, str, str, str, str]], txn.fetchall())
  319. if not rows:
  320. return 0
  321. sql = """
  322. UPDATE devices
  323. SET last_seen = ?, ip = ?, user_agent = ?
  324. WHERE user_id = ? AND device_id = ?
  325. """
  326. txn.execute_batch(sql, rows)
  327. _, _, _, user_id, device_id = rows[-1]
  328. self.db_pool.updates._background_update_progress_txn(
  329. txn,
  330. "devices_last_seen",
  331. {"last_user_id": user_id, "last_device_id": device_id},
  332. )
  333. return len(rows)
  334. updated = await self.db_pool.runInteraction(
  335. "_devices_last_seen_update", _devices_last_seen_update_txn
  336. )
  337. if not updated:
  338. await self.db_pool.updates._end_background_update("devices_last_seen")
  339. return updated
  340. class ClientIpWorkerStore(ClientIpBackgroundUpdateStore, MonthlyActiveUsersWorkerStore):
  341. def __init__(
  342. self,
  343. database: DatabasePool,
  344. db_conn: LoggingDatabaseConnection,
  345. hs: "HomeServer",
  346. ):
  347. super().__init__(database, db_conn, hs)
  348. if hs.config.redis.redis_enabled:
  349. # If we're using Redis, we can shift this update process off to
  350. # the background worker
  351. self._update_on_this_worker = hs.config.worker.run_background_tasks
  352. else:
  353. # If we're NOT using Redis, this must be handled by the master
  354. self._update_on_this_worker = hs.get_instance_name() == "master"
  355. self.user_ips_max_age = hs.config.server.user_ips_max_age
  356. # (user_id, access_token, ip,) -> last_seen
  357. self.client_ip_last_seen = LruCache[Tuple[str, str, str], int](
  358. cache_name="client_ip_last_seen", max_size=50000
  359. )
  360. if hs.config.worker.run_background_tasks and self.user_ips_max_age:
  361. self._clock.looping_call(self._prune_old_user_ips, 5 * 1000)
  362. if self._update_on_this_worker:
  363. # This is the designated worker that can write to the client IP
  364. # tables.
  365. # (user_id, access_token, ip,) -> (user_agent, device_id, last_seen)
  366. self._batch_row_update: Dict[
  367. Tuple[str, str, str], Tuple[str, Optional[str], int]
  368. ] = {}
  369. self._client_ip_looper = self._clock.looping_call(
  370. self._update_client_ips_batch, 5 * 1000
  371. )
  372. self.hs.get_reactor().addSystemEventTrigger(
  373. "before", "shutdown", self._update_client_ips_batch
  374. )
  375. @wrap_as_background_process("prune_old_user_ips")
  376. async def _prune_old_user_ips(self) -> None:
  377. """Removes entries in user IPs older than the configured period."""
  378. if self.user_ips_max_age is None:
  379. # Nothing to do
  380. return
  381. if not await self.db_pool.updates.has_completed_background_update(
  382. "devices_last_seen"
  383. ):
  384. # Only start pruning if we have finished populating the devices
  385. # last seen info.
  386. return
  387. # We do a slightly funky SQL delete to ensure we don't try and delete
  388. # too much at once (as the table may be very large from before we
  389. # started pruning).
  390. #
  391. # This works by finding the max last_seen that is less than the given
  392. # time, but has no more than N rows before it, deleting all rows with
  393. # a lesser last_seen time. (We COALESCE so that the sub-SELECT always
  394. # returns exactly one row).
  395. sql = """
  396. DELETE FROM user_ips
  397. WHERE last_seen <= (
  398. SELECT COALESCE(MAX(last_seen), -1)
  399. FROM (
  400. SELECT last_seen FROM user_ips
  401. WHERE last_seen <= ?
  402. ORDER BY last_seen ASC
  403. LIMIT 5000
  404. ) AS u
  405. )
  406. """
  407. timestamp = self._clock.time_msec() - self.user_ips_max_age
  408. def _prune_old_user_ips_txn(txn: LoggingTransaction) -> None:
  409. txn.execute(sql, (timestamp,))
  410. await self.db_pool.runInteraction(
  411. "_prune_old_user_ips", _prune_old_user_ips_txn
  412. )
  413. async def _get_last_client_ip_by_device_from_database(
  414. self, user_id: str, device_id: Optional[str]
  415. ) -> Dict[Tuple[str, str], DeviceLastConnectionInfo]:
  416. """For each device_id listed, give the user_ip it was last seen on.
  417. The result might be slightly out of date as client IPs are inserted in batches.
  418. Args:
  419. user_id: The user to fetch devices for.
  420. device_id: If None fetches all devices for the user
  421. Returns:
  422. A dictionary mapping a tuple of (user_id, device_id) to DeviceLastConnectionInfo.
  423. """
  424. keyvalues = {"user_id": user_id}
  425. if device_id is not None:
  426. keyvalues["device_id"] = device_id
  427. res = cast(
  428. List[Tuple[str, Optional[str], Optional[str], str, Optional[int]]],
  429. await self.db_pool.simple_select_list(
  430. table="devices",
  431. keyvalues=keyvalues,
  432. retcols=("user_id", "ip", "user_agent", "device_id", "last_seen"),
  433. ),
  434. )
  435. return {
  436. (user_id, device_id): DeviceLastConnectionInfo(
  437. user_id=user_id,
  438. device_id=device_id,
  439. ip=ip,
  440. user_agent=user_agent,
  441. last_seen=last_seen,
  442. )
  443. for user_id, ip, user_agent, device_id, last_seen in res
  444. }
  445. async def _get_user_ip_and_agents_from_database(
  446. self, user: UserID, since_ts: int = 0
  447. ) -> List[LastConnectionInfo]:
  448. """Fetch the IPs and user agents for a user since the given timestamp.
  449. The result might be slightly out of date as client IPs are inserted in batches.
  450. Args:
  451. user: The user for which to fetch IP addresses and user agents.
  452. since_ts: The timestamp after which to fetch IP addresses and user agents,
  453. in milliseconds.
  454. Returns:
  455. A list of dictionaries, each containing:
  456. * `access_token`: The access token used.
  457. * `ip`: The IP address used.
  458. * `user_agent`: The last user agent seen for this access token and IP
  459. address combination.
  460. * `last_seen`: The timestamp at which this access token and IP address
  461. combination was last seen, in milliseconds.
  462. Only the latest user agent for each access token and IP address combination
  463. is available.
  464. """
  465. user_id = user.to_string()
  466. def get_recent(txn: LoggingTransaction) -> List[Tuple[str, str, str, int]]:
  467. txn.execute(
  468. """
  469. SELECT access_token, ip, user_agent, last_seen FROM user_ips
  470. WHERE last_seen >= ? AND user_id = ?
  471. ORDER BY last_seen
  472. DESC
  473. """,
  474. (since_ts, user_id),
  475. )
  476. return cast(List[Tuple[str, str, str, int]], txn.fetchall())
  477. rows = await self.db_pool.runInteraction(
  478. desc="get_user_ip_and_agents", func=get_recent
  479. )
  480. return [
  481. {
  482. "access_token": access_token,
  483. "ip": ip,
  484. "user_agent": user_agent,
  485. "last_seen": last_seen,
  486. }
  487. for access_token, ip, user_agent, last_seen in rows
  488. ]
  489. async def insert_client_ip(
  490. self,
  491. user_id: str,
  492. access_token: str,
  493. ip: str,
  494. user_agent: str,
  495. device_id: Optional[str],
  496. now: Optional[int] = None,
  497. ) -> None:
  498. """Record that `user_id` used `access_token` from this `ip` address.
  499. This method does two things.
  500. 1. It queues up a row to be upserted into the `client_ips` table. These happen
  501. periodically; see _update_client_ips_batch.
  502. 2. It immediately records this user as having taken action for the purposes of
  503. MAU tracking.
  504. Any DB writes take place on the background tasks worker, falling back to the
  505. main process. If we're not that worker, this method emits a replication payload
  506. to run this logic on that worker.
  507. Two caveats to note:
  508. - We only take action once per LAST_SEEN_GRANULARITY, to avoid spamming the
  509. DB with writes.
  510. - Requests using the sliding-sync proxy's user agent are excluded, as its
  511. requests are not directly driven by end-users. This is a hack and we're not
  512. very proud of it.
  513. """
  514. # The sync proxy continuously triggers /sync even if the user is not
  515. # present so should be excluded from user_ips entries.
  516. if user_agent == "sync-v3-proxy-":
  517. return
  518. if not now:
  519. now = int(self._clock.time_msec())
  520. key = (user_id, access_token, ip)
  521. try:
  522. last_seen = self.client_ip_last_seen.get(key)
  523. except KeyError:
  524. last_seen = None
  525. # Rate-limited inserts
  526. if last_seen is not None and (now - last_seen) < LAST_SEEN_GRANULARITY:
  527. return
  528. self.client_ip_last_seen.set(key, now)
  529. if self._update_on_this_worker:
  530. await self.populate_monthly_active_users(user_id)
  531. self._batch_row_update[key] = (user_agent, device_id, now)
  532. else:
  533. # We are not the designated writer-worker, so stream over replication
  534. self.hs.get_replication_command_handler().send_user_ip(
  535. user_id, access_token, ip, user_agent, device_id, now
  536. )
  537. @wrap_as_background_process("update_client_ips")
  538. async def _update_client_ips_batch(self) -> None:
  539. assert (
  540. self._update_on_this_worker
  541. ), "This worker is not designated to update client IPs"
  542. # If the DB pool has already terminated, don't try updating
  543. if not self.db_pool.is_running():
  544. return
  545. to_update = self._batch_row_update
  546. self._batch_row_update = {}
  547. if to_update:
  548. await self.db_pool.runInteraction(
  549. "_update_client_ips_batch", self._update_client_ips_batch_txn, to_update
  550. )
  551. def _update_client_ips_batch_txn(
  552. self,
  553. txn: LoggingTransaction,
  554. to_update: Mapping[Tuple[str, str, str], Tuple[str, Optional[str], int]],
  555. ) -> None:
  556. assert (
  557. self._update_on_this_worker
  558. ), "This worker is not designated to update client IPs"
  559. # Keys and values for the `user_ips` upsert.
  560. user_ips_keys = []
  561. user_ips_values = []
  562. # Keys and values for the `devices` update.
  563. devices_keys = []
  564. devices_values = []
  565. for entry in to_update.items():
  566. (user_id, access_token, ip), (user_agent, device_id, last_seen) = entry
  567. user_ips_keys.append((user_id, access_token, ip))
  568. user_ips_values.append((user_agent, device_id, last_seen))
  569. # Technically an access token might not be associated with
  570. # a device so we need to check.
  571. if device_id:
  572. devices_keys.append((user_id, device_id))
  573. devices_values.append((user_agent, last_seen, ip))
  574. self.db_pool.simple_upsert_many_txn(
  575. txn,
  576. table="user_ips",
  577. key_names=("user_id", "access_token", "ip"),
  578. key_values=user_ips_keys,
  579. value_names=("user_agent", "device_id", "last_seen"),
  580. value_values=user_ips_values,
  581. )
  582. if devices_values:
  583. self.db_pool.simple_update_many_txn(
  584. txn,
  585. table="devices",
  586. key_names=("user_id", "device_id"),
  587. key_values=devices_keys,
  588. value_names=("user_agent", "last_seen", "ip"),
  589. value_values=devices_values,
  590. )
  591. async def get_last_client_ip_by_device(
  592. self, user_id: str, device_id: Optional[str]
  593. ) -> Dict[Tuple[str, str], DeviceLastConnectionInfo]:
  594. """For each device_id listed, give the user_ip it was last seen on
  595. Args:
  596. user_id: The user to fetch devices for.
  597. device_id: If None fetches all devices for the user
  598. Returns:
  599. A dictionary mapping a tuple of (user_id, device_id) to DeviceLastConnectionInfo.
  600. """
  601. ret = await self._get_last_client_ip_by_device_from_database(user_id, device_id)
  602. if not self._update_on_this_worker:
  603. # Only the writing-worker has additional in-memory data to enhance
  604. # the result
  605. return ret
  606. # Update what is retrieved from the database with data which is pending
  607. # insertion, as if it has already been stored in the database.
  608. for key in self._batch_row_update:
  609. uid, _access_token, ip = key
  610. if uid == user_id:
  611. user_agent, did, last_seen = self._batch_row_update[key]
  612. if did is None:
  613. # These updates don't make it to the `devices` table
  614. continue
  615. if not device_id or did == device_id:
  616. ret[(user_id, did)] = DeviceLastConnectionInfo(
  617. user_id=user_id,
  618. ip=ip,
  619. user_agent=user_agent,
  620. device_id=did,
  621. last_seen=last_seen,
  622. )
  623. return ret
  624. async def get_user_ip_and_agents(
  625. self, user: UserID, since_ts: int = 0
  626. ) -> List[LastConnectionInfo]:
  627. """Fetch the IPs and user agents for a user since the given timestamp.
  628. Args:
  629. user: The user for which to fetch IP addresses and user agents.
  630. since_ts: The timestamp after which to fetch IP addresses and user agents,
  631. in milliseconds.
  632. Returns:
  633. A list of dictionaries, each containing:
  634. * `access_token`: The access token used.
  635. * `ip`: The IP address used.
  636. * `user_agent`: The last user agent seen for this access token and IP
  637. address combination.
  638. * `last_seen`: The timestamp at which this access token and IP address
  639. combination was last seen, in milliseconds.
  640. Only the latest user agent for each access token and IP address combination
  641. is available.
  642. """
  643. rows_from_db = await self._get_user_ip_and_agents_from_database(user, since_ts)
  644. if not self._update_on_this_worker:
  645. # Only the writing-worker has additional in-memory data to enhance
  646. # the result
  647. return rows_from_db
  648. results: Dict[Tuple[str, str], LastConnectionInfo] = {
  649. (connection["access_token"], connection["ip"]): connection
  650. for connection in rows_from_db
  651. }
  652. # Overlay data that is pending insertion on top of the results from the
  653. # database.
  654. user_id = user.to_string()
  655. for key in self._batch_row_update:
  656. uid, access_token, ip = key
  657. if uid == user_id:
  658. user_agent, _, last_seen = self._batch_row_update[key]
  659. if last_seen >= since_ts:
  660. results[(access_token, ip)] = {
  661. "access_token": access_token,
  662. "ip": ip,
  663. "user_agent": user_agent,
  664. "last_seen": last_seen,
  665. }
  666. return list(results.values())
  667. async def get_last_seen_for_user_id(self, user_id: str) -> Optional[int]:
  668. """Get the last seen timestamp for a user, if we have it."""
  669. return await self.db_pool.simple_select_one_onecol(
  670. table="user_ips",
  671. keyvalues={"user_id": user_id},
  672. retcol="MAX(last_seen)",
  673. allow_none=True,
  674. desc="get_last_seen_for_user_id",
  675. )