Add ephemeral messages support (MSC2228)

changelog.d/6409.feature

@@ -0,0 +1 @@
+Add ephemeral messages support by partially implementing [MSC2228](https://github.com/matrix-org/matrix-doc/pull/2228).

synapse/api/constants.py

@@ -145,3 +145,6 @@ class EventContentFields(object):
 
     # Labels for the event, cf https://github.com/matrix-org/matrix-doc/pull/2326
     LABELS = "org.matrix.labels"
+    # Timestamp to delete the event after
+    # cf https://github.com/matrix-org/matrix-doc/pull/2228
+    SELF_DESTRUCT_AFTER = "m.self_destruct_after"

synapse/config/server.py

@@ -372,6 +372,8 @@ class LimitRemoteRoomsConfig(object):
             "cleanup_extremities_with_dummy_events", True
         )
 
+        self.enable_ephemeral_messages = config.get("enable_ephemeral_messages", False)
+
     def has_tls_listener(self) -> bool:
         return any(l["tls"] for l in self.listeners)
 

synapse/handlers/federation.py

@@ -31,7 +31,12 @@
 from twisted.internet import defer
 
 from synapse import event_auth
-from synapse.api.constants import EventTypes, Membership, RejectedReason
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    Membership,
+    RejectedReason,
+)
 from synapse.api.errors import (
     AuthError,
     CodeMessageException,
@@ -141,6 +146,8 @@ def __init__(self, hs):
 
         self.third_party_event_rules = hs.get_third_party_event_rules()
 
+        self._message_handler = hs.get_message_handler()
+
     @defer.inlineCallbacks
     def on_receive_pdu(self, origin, pdu, sent_to_us_directly=False):
         """ Process a PDU received via a federation /send/ transaction, or
@@ -1707,6 +1714,13 @@ def _handle_new_event(
                     self.store.remove_push_actions_from_staging, event.event_id
                 )
 
+        # If there's an expiry timestamp, schedule the redaction of the event.
+        expiry_ts = event.content.get(EventContentFields.SELF_DESTRUCT_AFTER)
+        if isinstance(expiry_ts, int) and not event.is_state():
+            yield self._message_handler.schedule_deletion_expired(
+                event.event_id, expiry_ts
+            )
+
         return context
 
     @defer.inlineCallbacks

synapse/handlers/message.py

@@ -24,7 +24,13 @@
 from twisted.internet.defer import succeed
 
 from synapse import event_auth
-from synapse.api.constants import EventTypes, Membership, RelationTypes, UserTypes
+from synapse.api.constants import (
+    EventContentFields,
+    EventTypes,
+    Membership,
+    RelationTypes,
+    UserTypes,
+)
 from synapse.api.errors import (
     AuthError,
     Codes,
@@ -62,6 +68,12 @@ def __init__(self, hs):
         self.storage = hs.get_storage()
         self.state_store = self.storage.state
         self._event_serializer = hs.get_event_client_serializer()
+        self._ephemeral_events_enabled = hs.config.enable_ephemeral_messages
+
+        run_as_background_process(
+            "_schedule_deletions_expired_from_db",
+            self._schedule_deletions_expired_from_db,
+        )
 
     @defer.inlineCallbacks
     def get_room_data(
@@ -225,6 +237,76 @@ def get_joined_members(self, requester, room_id):
             for user_id, profile in iteritems(users_with_profile)
         }
 
+    @defer.inlineCallbacks
+    def schedule_deletion_expired(self, event_id, redaction_ts):
+        """Schedule the deletion of an expired event, or if that event should have
+        expired then delete the event immediately.
+
+        Args:
+            event_id (str): The ID of the event to schedule the deletion of.
+            redaction_ts (int): The timestamp to delete the event at.
+        """
+        if not self._ephemeral_events_enabled:
+            return
+
+        # Save the timestamp at which the event expires so that we can reschedule its
+        # deletion on startup if the server is stopped before the event is deleted.
+        yield self.store.insert_event_expiry(event_id, redaction_ts)
+
+        now_ms = self.clock.time_msec()
+        delay = (redaction_ts - now_ms) / 1000
+
+        if delay > 0:
+            # Figure out how many seconds we need to wait before redacting the event.
+            logger.info("Scheduling deletion of event %s in %.3fs", event_id, delay)
+            self.clock.call_later(delay, self._delete_expired_event, event_id)
+        else:
+            # If the event should have already been redacted, redact it now.
+            yield self._delete_expired_event(event_id)
+
+    @defer.inlineCallbacks
+    def _schedule_deletions_expired_from_db(self):
+        """Load the IDs of the events that have an expiry date from the database (and
+        their expiry timestamp) and either delete them (if the expiry date is now or in
+        the past) or schedule their deletion on the timestamp.
+        """
+        if not self._ephemeral_events_enabled:
+            return
+
+        events_to_expire = yield self.store.get_events_to_expire()
+
+        for event in events_to_expire:
+            yield self.schedule_deletion_expired(event["event_id"], event["expiry_ts"])
+
+    @defer.inlineCallbacks
+    def _delete_expired_event(self, event_id):
+        """Retrieve and delete an expired event from the database.
+
+        If we don't have the event in the database, log it and delete the expiry date
+        from the database (so that we don't try to delete it again).
+
+        Args:
+            event_id (str): The ID of the event to retrieve and delete.
+        """
+        if not self._ephemeral_events_enabled:
+            return
+
+        logger.info("Deleting expired event %s", event_id)
+
+        # Try to retrieve the event from the database.
+        event = yield self.store.get_event(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 delete it again in the future.
+            logger.warning("Can't delete event %s because we don't have it." % event_id)
+            yield self.store.delete_event_expiry(event)
+            return
+
+        # Delete the event. This function also deletes the expiry date from the database
+        # in the same database transaction.
+        yield self.store.delete_expired_event(event)
+
 
 # The duration (in ms) after which rooms should be removed
 # `_rooms_to_exclude_from_dummy_event_insertion` (with the effect that we will try
@@ -295,6 +377,8 @@ def __init__(self, hs):
                 5 * 60 * 1000,
             )
 
+        self._message_handler = hs.get_message_handler()
+
     @defer.inlineCallbacks
     def create_event(
         self,
@@ -736,6 +820,13 @@ def handle_new_client_event(
                     self.store.remove_push_actions_from_staging, event.event_id
                 )
 
+        # If there's an expiry timestamp, schedule the redaction of the event.
+        expiry_ts = event.content.get(EventContentFields.SELF_DESTRUCT_AFTER)
+        if isinstance(expiry_ts, int) and not event.is_state():
+            yield self._message_handler.schedule_deletion_expired(
+                event.event_id, expiry_ts
+            )
+
     @defer.inlineCallbacks
     def persist_and_notify_client_event(
         self, requester, event, context, ratelimit=True, extra_users=[]

synapse/storage/data_stores/main/events.py

@@ -1954,6 +1954,78 @@ def insert_labels_for_event_txn(
             ],
         )
 
+    def insert_event_expiry(self, event_id, expiry_ts):
+        """Save the expiry timestamp associated with a given event ID.
+
+        Args:
+            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(
+            table="event_expiry",
+            values={"event_id": event_id, "expiry_ts": expiry_ts},
+            desc="insert_event_expiry",
+        )
+
+    def delete_expired_event(self, event):
+        """Delete an event that has expired by replacing its entry in event_json with a
+        pruned version of its JSON representation, and delete its associated expiry
+        timestamp.
+
+        Args:
+             event (events.EventBase): The event to delete.
+        """
+        # Prune the event's dict then convert it to JSON.
+        pruned_json = encode_json(prune_event_dict(event.get_dict()))
+
+        def delete_expired_event_txn(txn):
+            # Update the event_json table to replace the event's JSON with the pruned
+            # JSON.
+            self._simple_update_one_txn(
+                txn,
+                table="event_json",
+                keyvalues={"event_id": event.event_id},
+                updatevalues={"json": pruned_json},
+            )
+
+            # Delete the expiry timestamp associated with this event from the database.
+            self._simple_delete_txn(
+                txn, table="event_expiry", keyvalues={"event_id": event.event_id}
+            )
+
+            # We need to invalidate the event cache entry for this event because we
+            # changed its content in the database.
+            self._get_event_cache.invalidate((event.event_id,))
+
+        return self.runInteraction("delete_expired_event", delete_expired_event_txn)
+
+    def delete_event_expiry(self, event_id):
+        """Delete the expiry timestamp associated with an event ID without deleting the
+        actual event.
+
+        Args:
+            event_id (str): The event ID to delete the associated expiry timestamp of.
+        """
+        return self._simple_delete(
+            table="event_expiry",
+            keyvalues={"event_id": event_id},
+            desc="delete_event_expiry",
+        )
+
+    def get_events_to_expire(self):
+        """Retrieve the IDs of the events we have an expiry timestamp for, along with
+        said timestamp.
+
+        Returns:
+            A list of dicts, each containing an event_id and an expiry_ts.
+        """
+        return self._simple_select_list(
+            table="event_expiry",
+            keyvalues=None,
+            retcols=["event_id", "expiry_ts"],
+            desc="get_events_to_expire",
+        )
+
 
 AllNewEventsResult = namedtuple(
     "AllNewEventsResult",

synapse/storage/data_stores/main/events_bg_updates.py

@@ -90,6 +90,10 @@ def __init__(self, db_conn, hs):
             "event_store_labels", self._event_store_labels
         )
 
+        self.register_background_update_handler(
+            "event_store_expiry", self._event_store_expiry
+        )
+
     @defer.inlineCallbacks
     def _background_reindex_fields_sender(self, progress, batch_size):
         target_min_stream_id = progress["target_min_stream_id_inclusive"]
@@ -573,3 +577,63 @@ def _event_store_labels_txn(txn):
             yield self._end_background_update("event_store_labels")
 
         return num_rows
+
+    @defer.inlineCallbacks
+    def _event_store_expiry(self, progress, batch_size):
+        """Background update handler which will store expiry timestamps for existing
+        events.
+        """
+        last_event_id = progress.get("last_event_id", "")
+
+        def _event_store_labels_txn(txn):
+            txn.execute(
+                """
+                SELECT event_id, json FROM event_json
+                LEFT JOIN event_expiry USING (event_id)
+                WHERE event_id > ? AND expiry_ts IS NULL
+                ORDER BY event_id LIMIT ?
+                """,
+                (last_event_id, batch_size),
+            )
+
+            results = list(txn)
+
+            nbrows = 0
+            last_row_event_id = ""
+            for (event_id, event_json_raw) in results:
+                try:
+                    event_json = json.loads(event_json_raw)
+
+                    expiry_ts = event_json.get(EventContentFields.SELF_DESTRUCT_AFTER)
+
+                    if isinstance(expiry_ts, int):
+                        self._simple_insert_txn(
+                            txn=txn,
+                            table="event_expiry",
+                            values={"event_id": event_id, "expiry_ts": expiry_ts},
+                        )
+                except Exception as e:
+                    logger.warning(
+                        "Unable to load event %s (no expiry timestamp will be imported):"
+                        " %s",
+                        event_id,
+                        e,
+                    )
+
+                nbrows += 1
+                last_row_event_id = event_id
+
+            self._background_update_progress_txn(
+                txn, "event_store_expiry", {"last_event_id": last_row_event_id}
+            )
+
+            return nbrows
+
+        num_rows = yield self.runInteraction(
+            desc="event_store_expiry", func=_event_store_labels_txn
+        )
+
+        if not num_rows:
+            yield self._end_background_update("event_store_expiry")
+
+        return num_rows

synapse/storage/data_stores/main/schema/delta/56/event_expiry.sql

@@ -0,0 +1,22 @@
+/* Copyright 2019 The Matrix.org Foundation C.I.C.
+ *
+ * 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.
+ */
+
+CREATE TABLE IF NOT EXISTS event_expiry (
+    event_id TEXT PRIMARY KEY,
+    expiry_ts BIGINT NOT NULL
+);
+
+INSERT INTO background_updates (update_name, progress_json) VALUES
+  ('event_store_expiry', '{}');