Remove pushers when deleting 3pid from account

When a user deletes an email from their account it will
now also remove all pushers for that email and that user
(even if these pushers were created by a different client)

synapse/handlers/auth.py

@@ -1459,6 +1459,9 @@ async def delete_threepid(
         )
 
         await self.store.user_delete_threepid(user_id, medium, address)
+        await self.store.delete_all_pushers_with_pushkey_and_user_id(
+            pushkey=address, user_id=user_id
+        )
         return result
 
     async def hash(self, password: str) -> str:

synapse/storage/databases/main/pusher.py

@@ -478,6 +478,52 @@ def delete_pusher_txn(txn, stream_id):
                 "delete_pusher", delete_pusher_txn, stream_id
             )
 
+    async def delete_all_pushers_with_pushkey_and_user_id(
+        self, pushkey: str, user_id: str
+    ) -> None:
+        """Delete all pushers associated with an account with a certain pushkey."""
+
+        # We want to generate a row in `deleted_pushers` for each pusher we're
+        # deleting, so we fetch the list now so we can generate the appropriate
+        # number of stream IDs.
+        #
+        # Note: technically there could be a race here between adding/deleting
+        # pushers, but a) the worst case if we don't stop a pusher until the
+        # next restart and b) this is only called when we're deactivating an
+        # account.
+
+        pushers = list(await self.get_pushers_by({"user_name": user_id, "pushkey": pushkey}))
+
+        def delete_pushers_txn(txn, stream_ids):
+            self._invalidate_cache_and_stream(  # type: ignore
+                txn, self.get_if_user_has_pusher, (user_id,)
+            )
+
+            self.db_pool.simple_delete_txn(
+                txn,
+                table="pushers",
+                keyvalues={"user_name": user_id},
+            )
+
+            self.db_pool.simple_insert_many_txn(
+                txn,
+                table="deleted_pushers",
+                values=[
+                    {
+                        "stream_id": stream_id,
+                        "app_id": pusher.app_id,
+                        "pushkey": pushkey,
+                        "user_id": user_id,
+                    }
+                    for stream_id, pusher in zip(stream_ids, pushers)
+                ],
+            )
+
+        async with self._pushers_id_gen.get_next_mult(len(pushers)) as stream_ids:
+            await self.db_pool.runInteraction(
+                "delete_all_pushers_for_user", delete_pushers_txn, stream_ids
+            )
+
     async def delete_all_pushers_for_user(self, user_id: str) -> None:
         """Delete all pushers associated with an account."""