|
@@ -130,6 +130,8 @@ class EventsStore(
|
|
|
if self.hs.config.redaction_retention_period is not None:
|
|
|
hs.get_clock().looping_call(_censor_redactions, 5 * 60 * 1000)
|
|
|
|
|
|
+ self._ephemeral_messages_enabled = hs.config.enable_ephemeral_messages
|
|
|
+
|
|
|
@defer.inlineCallbacks
|
|
|
def _read_forward_extremities(self):
|
|
|
def fetch(txn):
|
|
@@ -940,6 +942,12 @@ class EventsStore(
|
|
|
txn, event.event_id, labels, event.room_id, event.depth
|
|
|
)
|
|
|
|
|
|
+ if self._ephemeral_messages_enabled:
|
|
|
+ # If there's an expiry timestamp on the event, store it.
|
|
|
+ expiry_ts = event.content.get(EventContentFields.SELF_DESTRUCT_AFTER)
|
|
|
+ if isinstance(expiry_ts, int) and not event.is_state():
|
|
|
+ self._insert_event_expiry_txn(txn, event.event_id, expiry_ts)
|
|
|
+
|
|
|
# Insert into the room_memberships table.
|
|
|
self._store_room_members_txn(
|
|
|
txn,
|
|
@@ -1101,12 +1109,7 @@ class EventsStore(
|
|
|
def _update_censor_txn(txn):
|
|
|
for redaction_id, event_id, pruned_json in updates:
|
|
|
if pruned_json:
|
|
|
- self._simple_update_one_txn(
|
|
|
- txn,
|
|
|
- table="event_json",
|
|
|
- keyvalues={"event_id": event_id},
|
|
|
- updatevalues={"json": pruned_json},
|
|
|
- )
|
|
|
+ self._censor_event_txn(txn, event_id, pruned_json)
|
|
|
|
|
|
self._simple_update_one_txn(
|
|
|
txn,
|
|
@@ -1117,6 +1120,22 @@ class EventsStore(
|
|
|
|
|
|
yield self.runInteraction("_update_censor_txn", _update_censor_txn)
|
|
|
|
|
|
+ def _censor_event_txn(self, txn, event_id, pruned_json):
|
|
|
+ """Censor an event by replacing its JSON in the event_json table with the
|
|
|
+ provided pruned JSON.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ txn (LoggingTransaction): The database transaction.
|
|
|
+ event_id (str): The ID of the event to censor.
|
|
|
+ pruned_json (str): The pruned JSON
|
|
|
+ """
|
|
|
+ self._simple_update_one_txn(
|
|
|
+ txn,
|
|
|
+ table="event_json",
|
|
|
+ keyvalues={"event_id": event_id},
|
|
|
+ updatevalues={"json": pruned_json},
|
|
|
+ )
|
|
|
+
|
|
|
@defer.inlineCallbacks
|
|
|
def count_daily_messages(self):
|
|
|
"""
|
|
@@ -1957,6 +1976,101 @@ class EventsStore(
|
|
|
],
|
|
|
)
|
|
|
|
|
|
+ def _insert_event_expiry_txn(self, txn, event_id, expiry_ts):
|
|
|
+ """Save the expiry timestamp associated with a given event ID.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ txn (LoggingTransaction): The database transaction to use.
|
|
|
+ event_id (str): The event ID the expiry timestamp is associated with.
|
|
|
+ expiry_ts (int): The timestamp at which to expire (delete) the event.
|
|
|
+ """
|
|
|
+ return self._simple_insert_txn(
|
|
|
+ txn=txn,
|
|
|
+ table="event_expiry",
|
|
|
+ values={"event_id": event_id, "expiry_ts": expiry_ts},
|
|
|
+ )
|
|
|
+
|
|
|
+ @defer.inlineCallbacks
|
|
|
+ def expire_event(self, event_id):
|
|
|
+ """Retrieve and expire an event that has expired, and delete its associated
|
|
|
+ expiry timestamp. If the event can't be retrieved, delete its associated
|
|
|
+ timestamp so we don't try to expire it again in the future.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ event_id (str): The ID of the event to delete.
|
|
|
+ """
|
|
|
+ # Try to retrieve the event's content from the database or the event cache.
|
|
|
+ event = yield self.get_event(event_id)
|
|
|
+
|
|
|
+ def delete_expired_event_txn(txn):
|
|
|
+ # Delete the expiry timestamp associated with this event from the database.
|
|
|
+ self._delete_event_expiry_txn(txn, event_id)
|
|
|
+
|
|
|
+ if not event:
|
|
|
+ # If we can't find the event, log a warning and delete the expiry date
|
|
|
+ # from the database so that we don't try to expire it again in the
|
|
|
+ # future.
|
|
|
+ logger.warning(
|
|
|
+ "Can't expire event %s because we don't have it.", event_id
|
|
|
+ )
|
|
|
+ return
|
|
|
+
|
|
|
+ # Prune the event's dict then convert it to JSON.
|
|
|
+ pruned_json = encode_json(prune_event_dict(event.get_dict()))
|
|
|
+
|
|
|
+ # Update the event_json table to replace the event's JSON with the pruned
|
|
|
+ # JSON.
|
|
|
+ self._censor_event_txn(txn, event.event_id, pruned_json)
|
|
|
+
|
|
|
+ # We need to invalidate the event cache entry for this event because we
|
|
|
+ # changed its content in the database. We can't call
|
|
|
+ # self._invalidate_cache_and_stream because self.get_event_cache isn't of the
|
|
|
+ # right type.
|
|
|
+ txn.call_after(self._get_event_cache.invalidate, (event.event_id,))
|
|
|
+ # Send that invalidation to replication so that other workers also invalidate
|
|
|
+ # the event cache.
|
|
|
+ self._send_invalidation_to_replication(
|
|
|
+ txn, "_get_event_cache", (event.event_id,)
|
|
|
+ )
|
|
|
+
|
|
|
+ yield self.runInteraction("delete_expired_event", delete_expired_event_txn)
|
|
|
+
|
|
|
+ def _delete_event_expiry_txn(self, txn, event_id):
|
|
|
+ """Delete the expiry timestamp associated with an event ID without deleting the
|
|
|
+ actual event.
|
|
|
+
|
|
|
+ Args:
|
|
|
+ txn (LoggingTransaction): The transaction to use to perform the deletion.
|
|
|
+ event_id (str): The event ID to delete the associated expiry timestamp of.
|
|
|
+ """
|
|
|
+ return self._simple_delete_txn(
|
|
|
+ txn=txn, table="event_expiry", keyvalues={"event_id": event_id}
|
|
|
+ )
|
|
|
+
|
|
|
+ def get_next_event_to_expire(self):
|
|
|
+ """Retrieve the entry with the lowest expiry timestamp in the event_expiry
|
|
|
+ table, or None if there's no more event to expire.
|
|
|
+
|
|
|
+ Returns: Deferred[Optional[Tuple[str, int]]]
|
|
|
+ A tuple containing the event ID as its first element and an expiry timestamp
|
|
|
+ as its second one, if there's at least one row in the event_expiry table.
|
|
|
+ None otherwise.
|
|
|
+ """
|
|
|
+
|
|
|
+ def get_next_event_to_expire_txn(txn):
|
|
|
+ txn.execute(
|
|
|
+ """
|
|
|
+ SELECT event_id, expiry_ts FROM event_expiry
|
|
|
+ ORDER BY expiry_ts ASC LIMIT 1
|
|
|
+ """
|
|
|
+ )
|
|
|
+
|
|
|
+ return txn.fetchone()
|
|
|
+
|
|
|
+ return self.runInteraction(
|
|
|
+ desc="get_next_event_to_expire", func=get_next_event_to_expire_txn
|
|
|
+ )
|
|
|
+
|
|
|
|
|
|
AllNewEventsResult = namedtuple(
|
|
|
"AllNewEventsResult",
|