emoji: Add migration to reupload all RealmEmoji and ensure .author.

Fixes #19732.

tools/semgrep.yml

@@ -39,6 +39,7 @@ rules:
         - zerver/migrations/0206_stream_rendered_description.py
         - zerver/migrations/0209_user_profile_no_empty_password.py
         - zerver/migrations/0260_missed_message_addresses_from_redis_to_db.py
+        - zerver/migrations/0376_set_realmemoji_author_and_reupload_realmemoji.py
         - pgroonga/migrations/0002_html_escape_subject.py
 
   - id: logging-format

zerver/lib/upload.py

@@ -12,6 +12,7 @@ import urllib
 from datetime import timedelta
 from mimetypes import guess_extension, guess_type
 from typing import IO, Any, Callable, Optional, Tuple
+from urllib.parse import urljoin
 
 import boto3
 import botocore
@@ -32,6 +33,7 @@ from PIL.Image import DecompressionBombError
 
 from zerver.lib.avatar_hash import user_avatar_path
 from zerver.lib.exceptions import ErrorCode, JsonableError
+from zerver.lib.outgoing_http import OutgoingSession
 from zerver.lib.utils import assert_is_not_none
 from zerver.models import (
     Attachment,
@@ -1143,3 +1145,52 @@ def upload_export_tarball(
 
 def delete_export_tarball(export_path: str) -> Optional[str]:
     return upload_backend.delete_export_tarball(export_path)
+
+
+def get_emoji_file_content(
+    session: OutgoingSession, emoji_url: str, emoji_id: int, logger: logging.Logger
+) -> bytes:
+    original_emoji_url = emoji_url + ".original"
+
+    logger.info("Downloading %s", original_emoji_url)
+    response = session.get(original_emoji_url)
+    if response.status_code == 200:
+        assert type(response.content) == bytes
+        return response.content
+
+    logger.info("Error fetching emoji from URL %s", original_emoji_url)
+    logger.info("Trying %s instead", emoji_url)
+    response = session.get(emoji_url)
+    if response.status_code == 200:
+        assert type(response.content) == bytes
+        return response.content
+    logger.info("Error fetching emoji from URL %s", emoji_url)
+    logger.error("Could not fetch emoji %s", emoji_id)
+    raise AssertionError(f"Could not fetch emoji {emoji_id}")
+
+
+def handle_reupload_emojis_event(realm: Realm, logger: logging.Logger) -> None:
+    from zerver.lib.emoji import get_emoji_url
+
+    session = OutgoingSession(role="reupload_emoji", timeout=3, max_retries=3)
+
+    query = RealmEmoji.objects.filter(realm=realm).order_by("id")
+
+    for realm_emoji in query:
+        logger.info("Processing emoji %s", realm_emoji.id)
+        emoji_filename = realm_emoji.file_name
+        emoji_url = get_emoji_url(emoji_filename, realm_emoji.realm_id)
+        if emoji_url.startswith("/"):
+            emoji_url = urljoin(realm_emoji.realm.uri, emoji_url)
+
+        emoji_file_content = get_emoji_file_content(session, emoji_url, realm_emoji.id, logger)
+
+        emoji_bytes_io = io.BytesIO(emoji_file_content)
+
+        user_profile = realm_emoji.author
+        # When this runs, emojis have already been migrated to always have .author set.
+        assert user_profile is not None
+
+        logger.info("Reuploading emoji %s", realm_emoji.id)
+        realm_emoji.is_animated = upload_emoji_image(emoji_bytes_io, emoji_filename, user_profile)
+        realm_emoji.save(update_fields=["is_animated"])

zerver/migrations/0376_set_realmemoji_author_and_reupload_realmemoji.py

@@ -0,0 +1,49 @@
+from django.db import migrations
+from django.db.backends.postgresql.schema import DatabaseSchemaEditor
+from django.db.migrations.state import StateApps
+
+from zerver.lib.queue import queue_json_publish
+
+
+def set_emoji_author(apps: StateApps, schema_editor: DatabaseSchemaEditor) -> None:
+    """
+    This migration establishes the invariant that all RealmEmoji objects have .author set
+    and queues events for reuploading all RealmEmoji.
+    """
+    RealmEmoji = apps.get_model("zerver", "RealmEmoji")
+    Realm = apps.get_model("zerver", "Realm")
+    UserProfile = apps.get_model("zerver", "UserProfile")
+    ROLE_REALM_OWNER = 100
+
+    realm_emoji_to_update = []
+    for realm_emoji in RealmEmoji.objects.all():
+        if realm_emoji.author_id is None:
+            user_profile = (
+                UserProfile.objects.filter(
+                    realm_id=realm_emoji.realm_id, is_active=True, role=ROLE_REALM_OWNER
+                )
+                .order_by("id")
+                .first()
+            )
+            realm_emoji.author_id = user_profile.id
+            realm_emoji_to_update.append(realm_emoji)
+
+    RealmEmoji.objects.bulk_update(realm_emoji_to_update, ["author_id"])
+
+    for realm_id in Realm.objects.order_by("id").values_list("id", flat=True):
+        event = {
+            "type": "reupload_realm_emoji",
+            "realm_id": realm_id,
+        }
+        queue_json_publish("deferred_work", event)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("zerver", "0375_invalid_characters_in_stream_names"),
+    ]
+
+    operations = [
+        migrations.RunPython(set_emoji_author, reverse_code=migrations.RunPython.noop),
+    ]

zerver/worker/queue_processors.py

@@ -88,6 +88,7 @@ from zerver.lib.send_email import (
     send_future_email,
 )
 from zerver.lib.timestamp import timestamp_to_datetime
+from zerver.lib.upload import handle_reupload_emojis_event
 from zerver.lib.url_preview import preview as url_preview
 from zerver.models import (
     Message,
@@ -1064,6 +1065,13 @@ class DeferredWorker(QueueProcessingWorker):
                 user_profile.realm.string_id,
                 time.time() - start,
             )
+        elif event["type"] == "reupload_realm_emoji":
+            # This is a special event queued by the migration for reuploading emojis.
+            # We don't want to run the necessary code in the actual migration, so it simply
+            # queues the necessary event, and the actual work is done here in the queue worker.
+            realm = Realm.objects.get(id=event["realm_id"])
+            logger.info("Processing reupload_realm_emoji event for realm %s", realm.id)
+            handle_reupload_emojis_event(realm, logger)
 
         end = time.time()
         logger.info("deferred_work processed %s event (%dms)", event["type"], (end - start) * 1000)