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 => { 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 => { 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> }; + }; + 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; + messageGroupId?: string; } const CHUNK_SIZE = 10; @@ -50,6 +51,9 @@ const batchPublishMessages = async ( ]), ); } + if (message.messageGroupId !== undefined) { + entry.MessageGroupId = message.messageGroupId; + } return entry; }), });