123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960 |
- # -*- coding: utf-8 -*-
- # Copyright 2015 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 logging
- from six import iteritems
- from canonicaljson import json
- from twisted.internet import defer
- from synapse.metrics.background_process_metrics import run_as_background_process
- from synapse.storage._base import LoggingTransaction, SQLBaseStore
- from synapse.util.caches.descriptors import cachedInlineCallbacks
- logger = logging.getLogger(__name__)
- DEFAULT_NOTIF_ACTION = ["notify", {"set_tweak": "highlight", "value": False}]
- DEFAULT_HIGHLIGHT_ACTION = [
- "notify",
- {"set_tweak": "sound", "value": "default"},
- {"set_tweak": "highlight"},
- ]
- def _serialize_action(actions, is_highlight):
- """Custom serializer for actions. This allows us to "compress" common actions.
- We use the fact that most users have the same actions for notifs (and for
- highlights).
- We store these default actions as the empty string rather than the full JSON.
- Since the empty string isn't valid JSON there is no risk of this clashing with
- any real JSON actions
- """
- if is_highlight:
- if actions == DEFAULT_HIGHLIGHT_ACTION:
- return "" # We use empty string as the column is non-NULL
- else:
- if actions == DEFAULT_NOTIF_ACTION:
- return ""
- return json.dumps(actions)
- def _deserialize_action(actions, is_highlight):
- """Custom deserializer for actions. This allows us to "compress" common actions
- """
- if actions:
- return json.loads(actions)
- if is_highlight:
- return DEFAULT_HIGHLIGHT_ACTION
- else:
- return DEFAULT_NOTIF_ACTION
- class EventPushActionsWorkerStore(SQLBaseStore):
- def __init__(self, db_conn, hs):
- super(EventPushActionsWorkerStore, self).__init__(db_conn, hs)
- # These get correctly set by _find_stream_orderings_for_times_txn
- self.stream_ordering_month_ago = None
- self.stream_ordering_day_ago = None
- cur = LoggingTransaction(
- db_conn.cursor(),
- name="_find_stream_orderings_for_times_txn",
- database_engine=self.database_engine,
- )
- self._find_stream_orderings_for_times_txn(cur)
- cur.close()
- self.find_stream_orderings_looping_call = self._clock.looping_call(
- self._find_stream_orderings_for_times, 10 * 60 * 1000
- )
- self._rotate_delay = 3
- self._rotate_count = 10000
- @cachedInlineCallbacks(num_args=3, tree=True, max_entries=5000)
- def get_unread_event_push_actions_by_room_for_user(
- self, room_id, user_id, last_read_event_id
- ):
- ret = yield self.runInteraction(
- "get_unread_event_push_actions_by_room",
- self._get_unread_counts_by_receipt_txn,
- room_id,
- user_id,
- last_read_event_id,
- )
- return ret
- def _get_unread_counts_by_receipt_txn(
- self, txn, room_id, user_id, last_read_event_id
- ):
- sql = (
- "SELECT stream_ordering"
- " FROM events"
- " WHERE room_id = ? AND event_id = ?"
- )
- txn.execute(sql, (room_id, last_read_event_id))
- results = txn.fetchall()
- if len(results) == 0:
- return {"notify_count": 0, "highlight_count": 0}
- stream_ordering = results[0][0]
- return self._get_unread_counts_by_pos_txn(
- txn, room_id, user_id, stream_ordering
- )
- def _get_unread_counts_by_pos_txn(self, txn, room_id, user_id, stream_ordering):
- # First get number of notifications.
- # We don't need to put a notif=1 clause as all rows always have
- # notif=1
- sql = (
- "SELECT count(*)"
- " FROM event_push_actions ea"
- " WHERE"
- " user_id = ?"
- " AND room_id = ?"
- " AND stream_ordering > ?"
- )
- txn.execute(sql, (user_id, room_id, stream_ordering))
- row = txn.fetchone()
- notify_count = row[0] if row else 0
- txn.execute(
- """
- SELECT notif_count FROM event_push_summary
- WHERE room_id = ? AND user_id = ? AND stream_ordering > ?
- """,
- (room_id, user_id, stream_ordering),
- )
- rows = txn.fetchall()
- if rows:
- notify_count += rows[0][0]
- # Now get the number of highlights
- sql = (
- "SELECT count(*)"
- " FROM event_push_actions ea"
- " WHERE"
- " highlight = 1"
- " AND user_id = ?"
- " AND room_id = ?"
- " AND stream_ordering > ?"
- )
- txn.execute(sql, (user_id, room_id, stream_ordering))
- row = txn.fetchone()
- highlight_count = row[0] if row else 0
- return {"notify_count": notify_count, "highlight_count": highlight_count}
- @defer.inlineCallbacks
- def get_push_action_users_in_range(self, min_stream_ordering, max_stream_ordering):
- def f(txn):
- sql = (
- "SELECT DISTINCT(user_id) FROM event_push_actions WHERE"
- " stream_ordering >= ? AND stream_ordering <= ?"
- )
- txn.execute(sql, (min_stream_ordering, max_stream_ordering))
- return [r[0] for r in txn]
- ret = yield self.runInteraction("get_push_action_users_in_range", f)
- return ret
- @defer.inlineCallbacks
- def get_unread_push_actions_for_user_in_range_for_http(
- self, user_id, min_stream_ordering, max_stream_ordering, limit=20
- ):
- """Get a list of the most recent unread push actions for a given user,
- within the given stream ordering range. Called by the httppusher.
- Args:
- user_id (str): The user to fetch push actions for.
- min_stream_ordering(int): The exclusive lower bound on the
- stream ordering of event push actions to fetch.
- max_stream_ordering(int): The inclusive upper bound on the
- stream ordering of event push actions to fetch.
- limit (int): The maximum number of rows to return.
- Returns:
- A promise which resolves to a list of dicts with the keys "event_id",
- "room_id", "stream_ordering", "actions".
- The list will be ordered by ascending stream_ordering.
- The list will have between 0~limit entries.
- """
- # find rooms that have a read receipt in them and return the next
- # push actions
- def get_after_receipt(txn):
- # find rooms that have a read receipt in them and return the next
- # push actions
- sql = (
- "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
- " ep.highlight "
- " FROM ("
- " SELECT room_id,"
- " MAX(stream_ordering) as stream_ordering"
- " FROM events"
- " INNER JOIN receipts_linearized USING (room_id, event_id)"
- " WHERE receipt_type = 'm.read' AND user_id = ?"
- " GROUP BY room_id"
- ") AS rl,"
- " event_push_actions AS ep"
- " WHERE"
- " ep.room_id = rl.room_id"
- " AND ep.stream_ordering > rl.stream_ordering"
- " AND ep.user_id = ?"
- " AND ep.stream_ordering > ?"
- " AND ep.stream_ordering <= ?"
- " ORDER BY ep.stream_ordering ASC LIMIT ?"
- )
- args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
- txn.execute(sql, args)
- return txn.fetchall()
- after_read_receipt = yield self.runInteraction(
- "get_unread_push_actions_for_user_in_range_http_arr", get_after_receipt
- )
- # There are rooms with push actions in them but you don't have a read receipt in
- # them e.g. rooms you've been invited to, so get push actions for rooms which do
- # not have read receipts in them too.
- def get_no_receipt(txn):
- sql = (
- "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
- " ep.highlight "
- " FROM event_push_actions AS ep"
- " INNER JOIN events AS e USING (room_id, event_id)"
- " WHERE"
- " ep.room_id NOT IN ("
- " SELECT room_id FROM receipts_linearized"
- " WHERE receipt_type = 'm.read' AND user_id = ?"
- " GROUP BY room_id"
- " )"
- " AND ep.user_id = ?"
- " AND ep.stream_ordering > ?"
- " AND ep.stream_ordering <= ?"
- " ORDER BY ep.stream_ordering ASC LIMIT ?"
- )
- args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
- txn.execute(sql, args)
- return txn.fetchall()
- no_read_receipt = yield self.runInteraction(
- "get_unread_push_actions_for_user_in_range_http_nrr", get_no_receipt
- )
- notifs = [
- {
- "event_id": row[0],
- "room_id": row[1],
- "stream_ordering": row[2],
- "actions": _deserialize_action(row[3], row[4]),
- }
- for row in after_read_receipt + no_read_receipt
- ]
- # Now sort it so it's ordered correctly, since currently it will
- # contain results from the first query, correctly ordered, followed
- # by results from the second query, but we want them all ordered
- # by stream_ordering, oldest first.
- notifs.sort(key=lambda r: r["stream_ordering"])
- # Take only up to the limit. We have to stop at the limit because
- # one of the subqueries may have hit the limit.
- return notifs[:limit]
- @defer.inlineCallbacks
- def get_unread_push_actions_for_user_in_range_for_email(
- self, user_id, min_stream_ordering, max_stream_ordering, limit=20
- ):
- """Get a list of the most recent unread push actions for a given user,
- within the given stream ordering range. Called by the emailpusher
- Args:
- user_id (str): The user to fetch push actions for.
- min_stream_ordering(int): The exclusive lower bound on the
- stream ordering of event push actions to fetch.
- max_stream_ordering(int): The inclusive upper bound on the
- stream ordering of event push actions to fetch.
- limit (int): The maximum number of rows to return.
- Returns:
- A promise which resolves to a list of dicts with the keys "event_id",
- "room_id", "stream_ordering", "actions", "received_ts".
- The list will be ordered by descending received_ts.
- The list will have between 0~limit entries.
- """
- # find rooms that have a read receipt in them and return the most recent
- # push actions
- def get_after_receipt(txn):
- sql = (
- "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
- " ep.highlight, e.received_ts"
- " FROM ("
- " SELECT room_id,"
- " MAX(stream_ordering) as stream_ordering"
- " FROM events"
- " INNER JOIN receipts_linearized USING (room_id, event_id)"
- " WHERE receipt_type = 'm.read' AND user_id = ?"
- " GROUP BY room_id"
- ") AS rl,"
- " event_push_actions AS ep"
- " INNER JOIN events AS e USING (room_id, event_id)"
- " WHERE"
- " ep.room_id = rl.room_id"
- " AND ep.stream_ordering > rl.stream_ordering"
- " AND ep.user_id = ?"
- " AND ep.stream_ordering > ?"
- " AND ep.stream_ordering <= ?"
- " ORDER BY ep.stream_ordering DESC LIMIT ?"
- )
- args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
- txn.execute(sql, args)
- return txn.fetchall()
- after_read_receipt = yield self.runInteraction(
- "get_unread_push_actions_for_user_in_range_email_arr", get_after_receipt
- )
- # There are rooms with push actions in them but you don't have a read receipt in
- # them e.g. rooms you've been invited to, so get push actions for rooms which do
- # not have read receipts in them too.
- def get_no_receipt(txn):
- sql = (
- "SELECT ep.event_id, ep.room_id, ep.stream_ordering, ep.actions,"
- " ep.highlight, e.received_ts"
- " FROM event_push_actions AS ep"
- " INNER JOIN events AS e USING (room_id, event_id)"
- " WHERE"
- " ep.room_id NOT IN ("
- " SELECT room_id FROM receipts_linearized"
- " WHERE receipt_type = 'm.read' AND user_id = ?"
- " GROUP BY room_id"
- " )"
- " AND ep.user_id = ?"
- " AND ep.stream_ordering > ?"
- " AND ep.stream_ordering <= ?"
- " ORDER BY ep.stream_ordering DESC LIMIT ?"
- )
- args = [user_id, user_id, min_stream_ordering, max_stream_ordering, limit]
- txn.execute(sql, args)
- return txn.fetchall()
- no_read_receipt = yield self.runInteraction(
- "get_unread_push_actions_for_user_in_range_email_nrr", get_no_receipt
- )
- # Make a list of dicts from the two sets of results.
- notifs = [
- {
- "event_id": row[0],
- "room_id": row[1],
- "stream_ordering": row[2],
- "actions": _deserialize_action(row[3], row[4]),
- "received_ts": row[5],
- }
- for row in after_read_receipt + no_read_receipt
- ]
- # Now sort it so it's ordered correctly, since currently it will
- # contain results from the first query, correctly ordered, followed
- # by results from the second query, but we want them all ordered
- # by received_ts (most recent first)
- notifs.sort(key=lambda r: -(r["received_ts"] or 0))
- # Now return the first `limit`
- return notifs[:limit]
- def get_if_maybe_push_in_range_for_user(self, user_id, min_stream_ordering):
- """A fast check to see if there might be something to push for the
- user since the given stream ordering. May return false positives.
- Useful to know whether to bother starting a pusher on start up or not.
- Args:
- user_id (str)
- min_stream_ordering (int)
- Returns:
- Deferred[bool]: True if there may be push to process, False if
- there definitely isn't.
- """
- def _get_if_maybe_push_in_range_for_user_txn(txn):
- sql = """
- SELECT 1 FROM event_push_actions
- WHERE user_id = ? AND stream_ordering > ?
- LIMIT 1
- """
- txn.execute(sql, (user_id, min_stream_ordering))
- return bool(txn.fetchone())
- return self.runInteraction(
- "get_if_maybe_push_in_range_for_user",
- _get_if_maybe_push_in_range_for_user_txn,
- )
- def add_push_actions_to_staging(self, event_id, user_id_actions):
- """Add the push actions for the event to the push action staging area.
- Args:
- event_id (str)
- user_id_actions (dict[str, list[dict|str])]): A dictionary mapping
- user_id to list of push actions, where an action can either be
- a string or dict.
- Returns:
- Deferred
- """
- if not user_id_actions:
- return
- # This is a helper function for generating the necessary tuple that
- # can be used to inert into the `event_push_actions_staging` table.
- def _gen_entry(user_id, actions):
- is_highlight = 1 if _action_has_highlight(actions) else 0
- return (
- event_id, # event_id column
- user_id, # user_id column
- _serialize_action(actions, is_highlight), # actions column
- 1, # notif column
- is_highlight, # highlight column
- )
- def _add_push_actions_to_staging_txn(txn):
- # We don't use _simple_insert_many here to avoid the overhead
- # of generating lists of dicts.
- sql = """
- INSERT INTO event_push_actions_staging
- (event_id, user_id, actions, notif, highlight)
- VALUES (?, ?, ?, ?, ?)
- """
- txn.executemany(
- sql,
- (
- _gen_entry(user_id, actions)
- for user_id, actions in iteritems(user_id_actions)
- ),
- )
- return self.runInteraction(
- "add_push_actions_to_staging", _add_push_actions_to_staging_txn
- )
- @defer.inlineCallbacks
- def remove_push_actions_from_staging(self, event_id):
- """Called if we failed to persist the event to ensure that stale push
- actions don't build up in the DB
- Args:
- event_id (str)
- """
- try:
- res = yield self._simple_delete(
- table="event_push_actions_staging",
- keyvalues={"event_id": event_id},
- desc="remove_push_actions_from_staging",
- )
- return res
- except Exception:
- # this method is called from an exception handler, so propagating
- # another exception here really isn't helpful - there's nothing
- # the caller can do about it. Just log the exception and move on.
- logger.exception(
- "Error removing push actions after event persistence failure"
- )
- def _find_stream_orderings_for_times(self):
- return run_as_background_process(
- "event_push_action_stream_orderings",
- self.runInteraction,
- "_find_stream_orderings_for_times",
- self._find_stream_orderings_for_times_txn,
- )
- def _find_stream_orderings_for_times_txn(self, txn):
- logger.info("Searching for stream ordering 1 month ago")
- self.stream_ordering_month_ago = self._find_first_stream_ordering_after_ts_txn(
- txn, self._clock.time_msec() - 30 * 24 * 60 * 60 * 1000
- )
- logger.info(
- "Found stream ordering 1 month ago: it's %d", self.stream_ordering_month_ago
- )
- logger.info("Searching for stream ordering 1 day ago")
- self.stream_ordering_day_ago = self._find_first_stream_ordering_after_ts_txn(
- txn, self._clock.time_msec() - 24 * 60 * 60 * 1000
- )
- logger.info(
- "Found stream ordering 1 day ago: it's %d", self.stream_ordering_day_ago
- )
- def find_first_stream_ordering_after_ts(self, ts):
- """Gets the stream ordering corresponding to a given timestamp.
- Specifically, finds the stream_ordering of the first event that was
- received on or after the timestamp. This is done by a binary search on
- the events table, since there is no index on received_ts, so is
- relatively slow.
- Args:
- ts (int): timestamp in millis
- Returns:
- Deferred[int]: stream ordering of the first event received on/after
- the timestamp
- """
- return self.runInteraction(
- "_find_first_stream_ordering_after_ts_txn",
- self._find_first_stream_ordering_after_ts_txn,
- ts,
- )
- @staticmethod
- def _find_first_stream_ordering_after_ts_txn(txn, ts):
- """
- Find the stream_ordering of the first event that was received on or
- after a given timestamp. This is relatively slow as there is no index
- on received_ts but we can then use this to delete push actions before
- this.
- received_ts must necessarily be in the same order as stream_ordering
- and stream_ordering is indexed, so we manually binary search using
- stream_ordering
- Args:
- txn (twisted.enterprise.adbapi.Transaction):
- ts (int): timestamp to search for
- Returns:
- int: stream ordering
- """
- txn.execute("SELECT MAX(stream_ordering) FROM events")
- max_stream_ordering = txn.fetchone()[0]
- if max_stream_ordering is None:
- return 0
- # We want the first stream_ordering in which received_ts is greater
- # than or equal to ts. Call this point X.
- #
- # We maintain the invariants:
- #
- # range_start <= X <= range_end
- #
- range_start = 0
- range_end = max_stream_ordering + 1
- # Given a stream_ordering, look up the timestamp at that
- # stream_ordering.
- #
- # The array may be sparse (we may be missing some stream_orderings).
- # We treat the gaps as the same as having the same value as the
- # preceding entry, because we will pick the lowest stream_ordering
- # which satisfies our requirement of received_ts >= ts.
- #
- # For example, if our array of events indexed by stream_ordering is
- # [10, <none>, 20], we should treat this as being equivalent to
- # [10, 10, 20].
- #
- sql = (
- "SELECT received_ts FROM events"
- " WHERE stream_ordering <= ?"
- " ORDER BY stream_ordering DESC"
- " LIMIT 1"
- )
- while range_end - range_start > 0:
- middle = (range_end + range_start) // 2
- txn.execute(sql, (middle,))
- row = txn.fetchone()
- if row is None:
- # no rows with stream_ordering<=middle
- range_start = middle + 1
- continue
- middle_ts = row[0]
- if ts > middle_ts:
- # we got a timestamp lower than the one we were looking for.
- # definitely need to look higher: X > middle.
- range_start = middle + 1
- else:
- # we got a timestamp higher than (or the same as) the one we
- # were looking for. We aren't yet sure about the point we
- # looked up, but we can be sure that X <= middle.
- range_end = middle
- return range_end
- class EventPushActionsStore(EventPushActionsWorkerStore):
- EPA_HIGHLIGHT_INDEX = "epa_highlight_index"
- def __init__(self, db_conn, hs):
- super(EventPushActionsStore, self).__init__(db_conn, hs)
- self.register_background_index_update(
- self.EPA_HIGHLIGHT_INDEX,
- index_name="event_push_actions_u_highlight",
- table="event_push_actions",
- columns=["user_id", "stream_ordering"],
- )
- self.register_background_index_update(
- "event_push_actions_highlights_index",
- index_name="event_push_actions_highlights_index",
- table="event_push_actions",
- columns=["user_id", "room_id", "topological_ordering", "stream_ordering"],
- where_clause="highlight=1",
- )
- self._doing_notif_rotation = False
- self._rotate_notif_loop = self._clock.looping_call(
- self._start_rotate_notifs, 30 * 60 * 1000
- )
- def _set_push_actions_for_event_and_users_txn(
- self, txn, events_and_contexts, all_events_and_contexts
- ):
- """Handles moving push actions from staging table to main
- event_push_actions table for all events in `events_and_contexts`.
- Also ensures that all events in `all_events_and_contexts` are removed
- from the push action staging area.
- Args:
- events_and_contexts (list[(EventBase, EventContext)]): events
- we are persisting
- all_events_and_contexts (list[(EventBase, EventContext)]): all
- events that we were going to persist. This includes events
- we've already persisted, etc, that wouldn't appear in
- events_and_context.
- """
- sql = """
- INSERT INTO event_push_actions (
- room_id, event_id, user_id, actions, stream_ordering,
- topological_ordering, notif, highlight
- )
- SELECT ?, event_id, user_id, actions, ?, ?, notif, highlight
- FROM event_push_actions_staging
- WHERE event_id = ?
- """
- if events_and_contexts:
- txn.executemany(
- sql,
- (
- (
- event.room_id,
- event.internal_metadata.stream_ordering,
- event.depth,
- event.event_id,
- )
- for event, _ in events_and_contexts
- ),
- )
- for event, _ in events_and_contexts:
- user_ids = self._simple_select_onecol_txn(
- txn,
- table="event_push_actions_staging",
- keyvalues={"event_id": event.event_id},
- retcol="user_id",
- )
- for uid in user_ids:
- txn.call_after(
- self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
- (event.room_id, uid),
- )
- # Now we delete the staging area for *all* events that were being
- # persisted.
- txn.executemany(
- "DELETE FROM event_push_actions_staging WHERE event_id = ?",
- ((event.event_id,) for event, _ in all_events_and_contexts),
- )
- @defer.inlineCallbacks
- def get_push_actions_for_user(
- self, user_id, before=None, limit=50, only_highlight=False
- ):
- def f(txn):
- before_clause = ""
- if before:
- before_clause = "AND epa.stream_ordering < ?"
- args = [user_id, before, limit]
- else:
- args = [user_id, limit]
- if only_highlight:
- if len(before_clause) > 0:
- before_clause += " "
- before_clause += "AND epa.highlight = 1"
- # NB. This assumes event_ids are globally unique since
- # it makes the query easier to index
- sql = (
- "SELECT epa.event_id, epa.room_id,"
- " epa.stream_ordering, epa.topological_ordering,"
- " epa.actions, epa.highlight, epa.profile_tag, e.received_ts"
- " FROM event_push_actions epa, events e"
- " WHERE epa.event_id = e.event_id"
- " AND epa.user_id = ? %s"
- " ORDER BY epa.stream_ordering DESC"
- " LIMIT ?" % (before_clause,)
- )
- txn.execute(sql, args)
- return self.cursor_to_dict(txn)
- push_actions = yield self.runInteraction("get_push_actions_for_user", f)
- for pa in push_actions:
- pa["actions"] = _deserialize_action(pa["actions"], pa["highlight"])
- return push_actions
- @defer.inlineCallbacks
- def get_time_of_last_push_action_before(self, stream_ordering):
- def f(txn):
- sql = (
- "SELECT e.received_ts"
- " FROM event_push_actions AS ep"
- " JOIN events e ON ep.room_id = e.room_id AND ep.event_id = e.event_id"
- " WHERE ep.stream_ordering > ?"
- " ORDER BY ep.stream_ordering ASC"
- " LIMIT 1"
- )
- txn.execute(sql, (stream_ordering,))
- return txn.fetchone()
- result = yield self.runInteraction("get_time_of_last_push_action_before", f)
- return result[0] if result else None
- @defer.inlineCallbacks
- def get_latest_push_action_stream_ordering(self):
- def f(txn):
- txn.execute("SELECT MAX(stream_ordering) FROM event_push_actions")
- return txn.fetchone()
- result = yield self.runInteraction("get_latest_push_action_stream_ordering", f)
- return result[0] or 0
- def _remove_push_actions_for_event_id_txn(self, txn, room_id, event_id):
- # Sad that we have to blow away the cache for the whole room here
- txn.call_after(
- self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
- (room_id,),
- )
- txn.execute(
- "DELETE FROM event_push_actions WHERE room_id = ? AND event_id = ?",
- (room_id, event_id),
- )
- def _remove_old_push_actions_before_txn(
- self, txn, room_id, user_id, stream_ordering
- ):
- """
- Purges old push actions for a user and room before a given
- stream_ordering.
- We however keep a months worth of highlighted notifications, so that
- users can still get a list of recent highlights.
- Args:
- txn: The transcation
- room_id: Room ID to delete from
- user_id: user ID to delete for
- stream_ordering: The lowest stream ordering which will
- not be deleted.
- """
- txn.call_after(
- self.get_unread_event_push_actions_by_room_for_user.invalidate_many,
- (room_id, user_id),
- )
- # We need to join on the events table to get the received_ts for
- # event_push_actions and sqlite won't let us use a join in a delete so
- # we can't just delete where received_ts < x. Furthermore we can
- # only identify event_push_actions by a tuple of room_id, event_id
- # we we can't use a subquery.
- # Instead, we look up the stream ordering for the last event in that
- # room received before the threshold time and delete event_push_actions
- # in the room with a stream_odering before that.
- txn.execute(
- "DELETE FROM event_push_actions "
- " WHERE user_id = ? AND room_id = ? AND "
- " stream_ordering <= ?"
- " AND ((stream_ordering < ? AND highlight = 1) or highlight = 0)",
- (user_id, room_id, stream_ordering, self.stream_ordering_month_ago),
- )
- txn.execute(
- """
- DELETE FROM event_push_summary
- WHERE room_id = ? AND user_id = ? AND stream_ordering <= ?
- """,
- (room_id, user_id, stream_ordering),
- )
- def _start_rotate_notifs(self):
- return run_as_background_process("rotate_notifs", self._rotate_notifs)
- @defer.inlineCallbacks
- def _rotate_notifs(self):
- if self._doing_notif_rotation or self.stream_ordering_day_ago is None:
- return
- self._doing_notif_rotation = True
- try:
- while True:
- logger.info("Rotating notifications")
- caught_up = yield self.runInteraction(
- "_rotate_notifs", self._rotate_notifs_txn
- )
- if caught_up:
- break
- yield self.hs.get_clock().sleep(self._rotate_delay)
- finally:
- self._doing_notif_rotation = False
- def _rotate_notifs_txn(self, txn):
- """Archives older notifications into event_push_summary. Returns whether
- the archiving process has caught up or not.
- """
- old_rotate_stream_ordering = self._simple_select_one_onecol_txn(
- txn,
- table="event_push_summary_stream_ordering",
- keyvalues={},
- retcol="stream_ordering",
- )
- # We don't to try and rotate millions of rows at once, so we cap the
- # maximum stream ordering we'll rotate before.
- txn.execute(
- """
- SELECT stream_ordering FROM event_push_actions
- WHERE stream_ordering > ?
- ORDER BY stream_ordering ASC LIMIT 1 OFFSET ?
- """,
- (old_rotate_stream_ordering, self._rotate_count),
- )
- stream_row = txn.fetchone()
- if stream_row:
- offset_stream_ordering, = stream_row
- rotate_to_stream_ordering = min(
- self.stream_ordering_day_ago, offset_stream_ordering
- )
- caught_up = offset_stream_ordering >= self.stream_ordering_day_ago
- else:
- rotate_to_stream_ordering = self.stream_ordering_day_ago
- caught_up = True
- logger.info("Rotating notifications up to: %s", rotate_to_stream_ordering)
- self._rotate_notifs_before_txn(txn, rotate_to_stream_ordering)
- # We have caught up iff we were limited by `stream_ordering_day_ago`
- return caught_up
- def _rotate_notifs_before_txn(self, txn, rotate_to_stream_ordering):
- old_rotate_stream_ordering = self._simple_select_one_onecol_txn(
- txn,
- table="event_push_summary_stream_ordering",
- keyvalues={},
- retcol="stream_ordering",
- )
- # Calculate the new counts that should be upserted into event_push_summary
- sql = """
- SELECT user_id, room_id,
- coalesce(old.notif_count, 0) + upd.notif_count,
- upd.stream_ordering,
- old.user_id
- FROM (
- SELECT user_id, room_id, count(*) as notif_count,
- max(stream_ordering) as stream_ordering
- FROM event_push_actions
- WHERE ? <= stream_ordering AND stream_ordering < ?
- AND highlight = 0
- GROUP BY user_id, room_id
- ) AS upd
- LEFT JOIN event_push_summary AS old USING (user_id, room_id)
- """
- txn.execute(sql, (old_rotate_stream_ordering, rotate_to_stream_ordering))
- rows = txn.fetchall()
- logger.info("Rotating notifications, handling %d rows", len(rows))
- # If the `old.user_id` above is NULL then we know there isn't already an
- # entry in the table, so we simply insert it. Otherwise we update the
- # existing table.
- self._simple_insert_many_txn(
- txn,
- table="event_push_summary",
- values=[
- {
- "user_id": row[0],
- "room_id": row[1],
- "notif_count": row[2],
- "stream_ordering": row[3],
- }
- for row in rows
- if row[4] is None
- ],
- )
- txn.executemany(
- """
- UPDATE event_push_summary SET notif_count = ?, stream_ordering = ?
- WHERE user_id = ? AND room_id = ?
- """,
- ((row[2], row[3], row[0], row[1]) for row in rows if row[4] is not None),
- )
- txn.execute(
- "DELETE FROM event_push_actions"
- " WHERE ? <= stream_ordering AND stream_ordering < ? AND highlight = 0",
- (old_rotate_stream_ordering, rotate_to_stream_ordering),
- )
- logger.info("Rotating notifications, deleted %s push actions", txn.rowcount)
- txn.execute(
- "UPDATE event_push_summary_stream_ordering SET stream_ordering = ?",
- (rotate_to_stream_ordering,),
- )
- def _action_has_highlight(actions):
- for action in actions:
- try:
- if action.get("set_tweak", None) == "highlight":
- return action.get("value", True)
- except AttributeError:
- pass
- return False
|