Update emoji for domain when shortcode doesn't match uri (#359)

activities/models/emoji.py

@@ -245,6 +245,26 @@ class Emoji(StatorModel):
         # create
         shortcode = name.lower().strip(":")
         category = (icon.get("category") or "")[:100]
+
+        if not domain.local:
+            try:
+                emoji = cls.objects.get(shortcode=shortcode, domain=domain)
+            except cls.DoesNotExist:
+                pass
+            else:
+                # Domain previously provided this shortcode. Trample in the new emoji
+                if emoji.remote_url != icon["url"] or emoji.mimetype != mimetype:
+                    emoji.object_uri = data["id"]
+                    emoji.remote_url = icon["url"]
+                    emoji.mimetype = mimetype
+                    emoji.category = category
+                    emoji.transition_set_state("outdated")
+                    if emoji.file:
+                        emoji.file.delete(save=True)
+                    else:
+                        emoji.save()
+                return emoji
+
         emoji = cls.objects.create(
             shortcode=shortcode,
             domain=None if domain.local else domain,

api/views/accounts.py

@@ -134,7 +134,7 @@ def account_statuses(
         identity.posts.not_hidden()
         .unlisted(include_replies=not exclude_replies)
         .select_related("author")
-        .prefetch_related("attachments")
+        .prefetch_related("attachments", "mentions__domain", "emojis")
         .order_by("-created")
     )
     if pinned:

stator/models.py

@@ -253,6 +253,25 @@ class StatorModel(models.Model):
 
     atransition_perform = sync_to_async(transition_perform)
 
+    def transition_set_state(self, state: State | str):
+        """
+        Sets the instance to the given state name for when it is saved.
+        """
+        if isinstance(state, State):
+            state = state.name
+        if state not in self.state_graph.states:
+            raise ValueError(f"Invalid state {state}")
+        self.state = state  # type: ignore
+        self.state_changed = timezone.now()
+        self.state_locked_until = None
+
+        if self.state_graph.states[state].attempt_immediately:
+            self.state_attempted = None
+            self.state_ready = True
+        else:
+            self.state_attempted = timezone.now()
+            self.state_ready = False
+
     @classmethod
     def transition_perform_queryset(
         cls,