Ensure ordered ActivityPub message delivery

Add ordering keys to outgoing ActivityPub sends so related events
for the same object are delivered in order across queue workers.
This covers create, update, and delete flows for posts and articles,
plus share/unshare, follow transitions, block/unblock, and
reaction undo paths.

Closes #162

Author: dahlia

federation/inbox/following.ts

@@ -215,7 +215,10 @@ export async function onFollowed(
       actor: fedCtx.getActorUri(followee.id),
       object: follow,
     }),
-    { excludeBaseUris: [new URL(fedCtx.canonicalOrigin)] },
+    {
+      orderingKey: rows[0].iri,
+      excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
+    },
   );
 }
 

models/article.ts

@@ -240,6 +240,7 @@ export async function createArticle(
       object: articleObject,
     }),
     {
+      orderingKey: post.iri,
       preferSharedInbox: true,
       excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
     },
@@ -343,6 +344,7 @@ export async function updateArticle(
       object: articleObject,
     }),
     {
+      orderingKey: post.iri,
       preferSharedInbox: true,
       excludeBaseUris: [
         new URL(fedCtx.origin),
@@ -563,11 +565,14 @@ export async function startArticleContentTranslation(
       ccs: articleObject.ccIds,
       object: articleObject,
     });
+    const orderingKey = fedCtx.getObjectUri(vocab.Article, { id: article.id })
+      .href;
     await fedCtx.sendActivity(
       { identifier: article.accountId },
       "followers",
       update,
       {
+        orderingKey,
         preferSharedInbox: true,
         excludeBaseUris: [
           new URL(fedCtx.origin),

models/blocking.ts

@@ -102,6 +102,7 @@ export async function block(
       toRecipient(blockee),
       block,
       {
+        orderingKey: rows[0].iri,
         excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
         fanout: "skip",
         preferSharedInbox: false,
@@ -140,7 +141,10 @@ export async function unblock(
           object: new URL(blockee.iri),
         }),
       }),
-      { excludeBaseUris: [new URL(fedCtx.canonicalOrigin)] },
+      {
+        orderingKey: rows[0].iri,
+        excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
+      },
     );
   }
   return rows[0];

models/following.ts

@@ -48,7 +48,10 @@ export async function follow(
         actor: fedCtx.getActorUri(follower.id),
         object: new URL(followee.iri),
       }),
-      { excludeBaseUris: [new URL(fedCtx.canonicalOrigin)] },
+      {
+        orderingKey: rows[0].iri,
+        excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
+      },
     );
   } else if (rows.length > 0 && followee.accountId != null) {
     await updateFolloweesCount(db, rows[0].followerId, 1);
@@ -131,7 +134,10 @@ export async function unfollow(
           object: new URL(followee.iri),
         }),
       }),
-      { excludeBaseUris: [new URL(fedCtx.canonicalOrigin)] },
+      {
+        orderingKey: rows[0].iri,
+        excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
+      },
     );
   }
   if (rows.length > 0) {
@@ -180,6 +186,7 @@ export async function removeFollower(
           object: fedCtx.getActorUri(followee.id),
         }),
       }),
+      { orderingKey: rows[0].iri },
     );
   }
   return rows[0];

models/note.ts

@@ -359,6 +359,7 @@ export async function createNote(
     ccs: noteObject.ccIds,
     object: noteObject,
   });
+  const orderingKey = post.iri;
   if (post.mentions.length > 0) {
     const directRecipients: Recipient[] = post.mentions.map((m) => ({
       id: new URL(m.actor.iri),
@@ -372,6 +373,7 @@ export async function createNote(
       directRecipients,
       activity,
       {
+        orderingKey,
         preferSharedInbox: false,
         excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
       },
@@ -383,6 +385,7 @@ export async function createNote(
       "followers",
       activity,
       {
+        orderingKey,
         preferSharedInbox: true,
         excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
       },
@@ -501,6 +504,7 @@ export async function updateNote(
       object: noteObject,
     }),
     {
+      orderingKey: post.iri,
       preferSharedInbox: true,
       excludeBaseUris: [
         new URL(fedCtx.origin),

models/post.ts

@@ -890,6 +890,7 @@ export async function sharePost(
     "followers",
     announce,
     {
+      orderingKey: share.iri,
       preferSharedInbox: true,
       excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
     },
@@ -898,7 +899,10 @@ export async function sharePost(
     { identifier: account.id },
     toRecipient(post.actor),
     announce,
-    { excludeBaseUris: [new URL(fedCtx.canonicalOrigin)] },
+    {
+      orderingKey: share.iri,
+      excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
+    },
   );
   return share;
 }
@@ -948,6 +952,7 @@ export async function unsharePost(
     "followers",
     undo,
     {
+      orderingKey: unshared[0].iri,
       preferSharedInbox: true,
       excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
     },
@@ -956,7 +961,10 @@ export async function unsharePost(
     { identifier: account.id },
     toRecipient(sharedPost.actor),
     undo,
-    { excludeBaseUris: [new URL(fedCtx.canonicalOrigin)] },
+    {
+      orderingKey: unshared[0].iri,
+      excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
+    },
   );
   return unshared[0];
 }
@@ -1587,6 +1595,7 @@ export async function deletePost(
     "followers",
     activity,
     {
+      orderingKey: post.iri,
       preferSharedInbox: true,
       excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
     },
@@ -1596,6 +1605,7 @@ export async function deletePost(
     recipients,
     activity,
     {
+      orderingKey: post.iri,
       preferSharedInbox: true,
       excludeBaseUris: [new URL(fedCtx.canonicalOrigin)],
     },

models/reaction.ts

@@ -235,6 +235,7 @@ export async function react(
     post,
   });
   if (activity == null) return rows[0];
+  const orderingKey = id.href;
   await ctx.sendActivity(
     { identifier: account.id },
     {
@@ -245,13 +246,18 @@ export async function react(
         : { sharedInbox: new URL(post.actor.sharedInboxUrl) },
     },
     activity,
-    { excludeBaseUris: [new URL(ctx.canonicalOrigin)], fanout: "skip" },
+    {
+      orderingKey,
+      excludeBaseUris: [new URL(ctx.canonicalOrigin)],
+      fanout: "skip",
+    },
   );
   await ctx.sendActivity(
     { identifier: account.id },
     "followers",
     activity,
     {
+      orderingKey,
       excludeBaseUris: [new URL(ctx.canonicalOrigin)],
       preferSharedInbox: true,
     },
@@ -292,6 +298,7 @@ export async function undoReaction(
     post,
   });
   if (activity?.id == null) return rows[0];
+  const orderingKey = activity.id.href;
   const undo = new vocab.Undo({
     id: new URL("#undo", activity.id),
     actor: ctx.getActorUri(account.id),
@@ -310,13 +317,18 @@ export async function undoReaction(
         : { sharedInbox: new URL(post.actor.sharedInboxUrl) },
     },
     undo,
-    { excludeBaseUris: [new URL(ctx.canonicalOrigin)], fanout: "skip" },
+    {
+      orderingKey,
+      excludeBaseUris: [new URL(ctx.canonicalOrigin)],
+      fanout: "skip",
+    },
   );
   await ctx.sendActivity(
     { identifier: account.id },
     "followers",
     undo,
     {
+      orderingKey,
       excludeBaseUris: [new URL(ctx.canonicalOrigin)],
       preferSharedInbox: true,
     },