From 1e3f80880809cfe476cf60737fe1a5c27cbd35f4 Mon Sep 17 00:00:00 2001
From: Cecilia Krum <cecilia@permanent.org>
Date: Thu, 20 Aug 2026 12:24:28 -0500
Subject: [PATCH] PER-10765: Use SQS fair queueing

SQS now offers "fair queueing" (see
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-fair-queues.html). This
means that we can group messages into different tenants, in this case
by uploader (account subject, so an individual human, not an archive).
Then: "When a tenant has a disproportionately large number of in-flight
messages compared to others, Amazon SQS identifies that tenant as a
noisy neighbor and prioritizes message delivery for other
tenants. This approach reduces the dwell time impact to the other
tenants."

So, if Person A is uploading a thousand files, Person B doesn't have
to wait for their one file.

This will do nothing by itself because right now our backup isn't in
SQS but instead inside Archivematica, but there's another change
coming to fix that!

Note also that I don't know if we'll ever use `backfill-ledger` again,
but we may want to have that `admin` tenant for other purposes in the
future and I don't think it will hurt anything.
---
 .../src/archive/service/backfill_ledger.ts    |  1 +
 packages/event_send/src/service.ts            |  1 +
 packages/publisher-utils/src/index.test.ts    | 32 +++++++++++++++++++
 packages/publisher-utils/src/index.ts         |  4 +++
 4 files changed, 38 insertions(+)

diff --git a/packages/api/src/archive/service/backfill_ledger.ts b/packages/api/src/archive/service/backfill_ledger.ts
index 929c1883..c787234a 100644
--- a/packages/api/src/archive/service/backfill_ledger.ts
+++ b/packages/api/src/archive/service/backfill_ledger.ts
@@ -35,6 +35,7 @@ export const backfillLedger = async (archiveId: string): Promise<void> => {
 					id: record.recordId,
 					body: JSON.stringify(message),
 					attributes: { Entity: "record", Action: "create" },
+					messageGroupId: "admin",
 				})
 				.catch((err: unknown) => {
 					logger.error(err);
diff --git a/packages/event_send/src/service.ts b/packages/event_send/src/service.ts
index c1fe79ac..16cc4c2f 100644
--- a/packages/event_send/src/service.ts
+++ b/packages/event_send/src/service.ts
@@ -78,6 +78,7 @@ export const sendEvents = async (): Promise<void> => {
 			body: event.body,
 		}),
 		attributes: { Entity: event.entity, Action: event.action },
+		messageGroupId: event.actorId,
 	}));
 
 	for (const event of events.rows) {
diff --git a/packages/publisher-utils/src/index.test.ts b/packages/publisher-utils/src/index.test.ts
index 5d94de0f..b3ccbd61 100644
--- a/packages/publisher-utils/src/index.test.ts
+++ b/packages/publisher-utils/src/index.test.ts
@@ -185,6 +185,38 @@ describe("batchPublishMessages", () => {
 			process.env["AWS_ENDPOINT_URL"] = originalEndpoint;
 		}
 	});
+
+	test("should set MessageGroupId only on messages that provide one", async () => {
+		mockSend.mockResolvedValue({ Failed: [] });
+
+		const messages = [
+			{ id: "1", body: "message 1", messageGroupId: "tenant-a" },
+			{ id: "2", body: "message 2" },
+		];
+
+		await publisherClient.batchPublishMessages("topic", messages);
+
+		const {
+			mock: {
+				calls: [firstCallArguments],
+			},
+		} = mockSend as {
+			mock: { calls: Array<Array<{ __input: PublishBatchInput }>> };
+		};
+		expect(firstCallArguments).toBeDefined();
+		if (firstCallArguments !== undefined) {
+			expect(firstCallArguments[0]).toBeDefined();
+			if (firstCallArguments[0] !== undefined) {
+				expect(firstCallArguments[0].__input).toEqual({
+					TopicArn: "topic",
+					PublishBatchRequestEntries: [
+						{ Id: "1", Message: "message 1", MessageGroupId: "tenant-a" },
+						{ Id: "2", Message: "message 2" },
+					],
+				});
+			}
+		}
+	});
 });
 
 describe("publishMessage", () => {
diff --git a/packages/publisher-utils/src/index.ts b/packages/publisher-utils/src/index.ts
index 2e3a427d..17bfc837 100644
--- a/packages/publisher-utils/src/index.ts
+++ b/packages/publisher-utils/src/index.ts
@@ -11,6 +11,7 @@ export interface Message {
 	id: string;
 	body: string;
 	attributes?: Record<string, string>;
+	messageGroupId?: string;
 }
 
 const CHUNK_SIZE = 10;
@@ -50,6 +51,9 @@ const batchPublishMessages = async (
 							]),
 						);
 					}
+					if (message.messageGroupId !== undefined) {
+						entry.MessageGroupId = message.messageGroupId;
+					}
 					return entry;
 				}),
 			});