Fix MultipleResultsFound crash in get_activity_for_source

- Use .scalars().first() + LIMIT 1 instead of scalar_one_or_none()
  which crashes when multiple activities exist for the same source
- Allow re-Create after Delete (re-publish after unpublish)
- Add missing on_post_unpublished handler to root shared copy
- Sync add_follower upsert fix to root shared

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

events/handlers/federation_handlers.py

@@ -30,28 +30,23 @@ async def _try_publish(
 ) -> None:
     """Publish an AP activity if federation is available and user has a profile."""
     if not services.has("federation"):
-        log.warning("_try_publish: no federation service")
         return
 
     if not user_id:
-        log.warning("_try_publish: no user_id for %s#%s", source_type, source_id)
         return
 
     # Check user has an ActorProfile (chose a username)
     actor = await services.federation.get_actor_by_user_id(session, user_id)
     if not actor:
-        log.warning("_try_publish: no ActorProfile for user_id=%s", user_id)
         return
 
-    # Don't re-publish if we already have an activity for this source
+    # Don't re-publish if we already have a live activity for this source
     existing = await services.federation.get_activity_for_source(
         session, source_type, source_id,
     )
-    if existing and activity_type == "Create":
-        log.warning("_try_publish: already published %s#%s", source_type, source_id)
-        return  # Already published
+    if existing and activity_type == "Create" and existing.activity_type != "Delete":
+        return  # Already published (allow re-Create after Delete/unpublish)
 
-    log.warning("_try_publish: publishing %s/%s for %s#%d user=%s", activity_type, object_type, source_type, source_id, user_id)
     try:
         await services.federation.publish_activity(
             session,
@@ -62,7 +57,7 @@ async def _try_publish(
             source_type=source_type,
             source_id=source_id,
         )
-        log.warning(
+        log.info(
             "Published %s/%s for %s#%d by user %d",
             activity_type, object_type, source_type, source_id, user_id,
         )
@@ -178,12 +173,12 @@ async def on_post_unpublished(event: DomainEvent, session: AsyncSession) -> None
     if not actor:
         return
 
-    # Find the original Create activity for this post
+    # Find the original activity for this post
     existing = await services.federation.get_activity_for_source(
         session, "Post", event.aggregate_id,
     )
-    if not existing:
-        return  # Never published to federation, nothing to delete
+    if not existing or existing.activity_type == "Delete":
+        return  # Never published or already deleted
 
     try:
         await services.federation.publish_activity(
@@ -198,7 +193,7 @@ async def on_post_unpublished(event: DomainEvent, session: AsyncSession) -> None
             source_type="Post",
             source_id=event.aggregate_id,
         )
-        log.warning("Published Delete for Post#%d", event.aggregate_id)
+        log.info("Published Delete for Post#%d", event.aggregate_id)
     except Exception:
         log.exception("Failed to publish Delete for Post#%d", event.aggregate_id)
 

services/federation_impl.py

@@ -251,8 +251,9 @@ class SqlFederationService:
                     APActivity.source_type == source_type,
                     APActivity.source_id == source_id,
                 ).order_by(APActivity.created_at.desc())
+                .limit(1)
             )
-        ).scalar_one_or_none()
+        ).scalars().first()
         return _activity_to_dto(a) if a else None
 
     # -- Followers ------------------------------------------------------------