Skip to content

Commit 19be711

Browse files
oskardudyczoskardudycz
oskardudycz
authored and
oskardudycz
committed
Imlemented double hop correctly and tested it with integration tests
1 parent f018f00 commit 19be711

10 files changed

Lines changed: 951 additions & 37 deletions

File tree

‎src/packages/emmett-postgresql/src/eventStore/consumers/postgreSQLEventStoreConsumer.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ export type PostgreSQLEventStoreConsumer<
8383
HandlerContext,
8484
StoredMessage
8585
>,
86-
) => PostgreSQLProcessor<Input>;
86+
) => PostgreSQLProcessor<Input | Output>;
8787
}> &
8888
(AnyEvent extends ConsumerMessageType
8989
? Readonly<{
@@ -247,7 +247,7 @@ export const postgreSQLEventStoreConsumer = <
247247
HandlerContext,
248248
StoredMessage
249249
>,
250-
): PostgreSQLProcessor<Input> => {
250+
): PostgreSQLProcessor<Input | Output> => {
251251
const processor = postgreSQLWorkflowProcessor(options);
252252

253253
processors.push(

‎src/packages/emmett-postgresql/src/eventStore/consumers/postgreSQLEventStoreConsumer.workflow.int.spec.ts‎

Lines changed: 222 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import {
22
assertEqual,
33
assertThatArray,
4+
WorkflowHandler,
45
type WorkflowOptions,
56
} from '@event-driven-io/emmett';
67
import { getPostgreSQLStartedContainer } from '@event-driven-io/emmett-testcontainers';
@@ -89,8 +90,9 @@ void describe('PostgreSQL event store workflow processor', () => {
8990
GroupCheckoutOutput
9091
>({
9192
...workflowProcessorOptions,
93+
separateInputInboxFromProcessing: true,
9294
stopAfter: (message) =>
93-
message.type === 'InitiateGroupCheckout' &&
95+
message.type === 'GroupCheckoutInitiated' &&
9496
message.data.groupCheckoutId === groupCheckoutId,
9597
});
9698

@@ -149,9 +151,10 @@ void describe('PostgreSQL event store workflow processor', () => {
149151
GroupCheckoutOutput
150152
>({
151153
...workflowProcessorOptions,
154+
separateInputInboxFromProcessing: true,
152155
processorId: `workflow-${groupCheckoutId}-initiate`,
153156
stopAfter: (message) =>
154-
message.type === 'InitiateGroupCheckout' &&
157+
message.type === 'GroupCheckoutInitiated' &&
155158
message.data.groupCheckoutId === groupCheckoutId,
156159
});
157160

@@ -186,10 +189,11 @@ void describe('PostgreSQL event store workflow processor', () => {
186189
GroupCheckoutOutput
187190
>({
188191
...workflowProcessorOptions,
192+
separateInputInboxFromProcessing: true,
189193
processorId: `workflow-${groupCheckoutId}-complete`,
190194
stopAfter: (message) =>
191-
message.type === 'GuestCheckedOut' &&
192-
message.data.guestStayAccountId === guestId,
195+
message.type === 'GroupCheckoutCompleted' &&
196+
message.data.groupCheckoutId === groupCheckoutId,
193197
});
194198

195199
try {
@@ -271,4 +275,218 @@ void describe('PostgreSQL event store workflow processor', () => {
271275
}
272276
},
273277
);
278+
279+
void it(
280+
'processes messages directly in regular mode (separateInputInboxFromProcessing: false)',
281+
withDeadline,
282+
async () => {
283+
const groupCheckoutId = uuid();
284+
const guestStayAccountIds = [uuid()];
285+
const now = new Date();
286+
287+
const consumer = postgreSQLEventStoreConsumer({
288+
connectionString,
289+
});
290+
291+
consumer.workflowProcessor<
292+
GroupCheckoutInput,
293+
GroupCheckout,
294+
GroupCheckoutOutput
295+
>({
296+
...workflowProcessorOptions,
297+
separateInputInboxFromProcessing: false,
298+
stopAfter: (message) =>
299+
message.type === 'InitiateGroupCheckout' &&
300+
message.data.groupCheckoutId === groupCheckoutId,
301+
});
302+
303+
try {
304+
const consumerPromise = consumer.start();
305+
306+
await eventStore.appendToStream(`groupCheckout-${groupCheckoutId}`, [
307+
{
308+
type: 'InitiateGroupCheckout',
309+
data: {
310+
groupCheckoutId,
311+
clerkId: 'clerk-1',
312+
guestStayAccountIds,
313+
now,
314+
},
315+
},
316+
]);
317+
318+
await consumerPromise;
319+
320+
const { events } = await eventStore.readStream(
321+
`emt:workflow:${groupCheckoutId}`,
322+
);
323+
324+
assertThatArray(events).isNotEmpty();
325+
// In regular mode, input IS stored with workflow prefix, but together with outputs
326+
// This is the single-operation: input + outputs appended together
327+
assertEqual(
328+
events[0]!.type,
329+
'GroupCheckoutWorkflow:InitiateGroupCheckout',
330+
);
331+
assertEqual(events[1]!.type, 'GroupCheckoutInitiated');
332+
assertEqual(events[2]!.type, 'CheckOut');
333+
334+
// Verify we got all messages in one operation (NOT double-hop)
335+
assertThatArray(events).hasSize(3);
336+
} finally {
337+
await consumer.close();
338+
}
339+
},
340+
);
341+
342+
void it(
343+
'stores input first then processes in double-hop mode (separateInputInboxFromProcessing: true)',
344+
withDeadline,
345+
async () => {
346+
const groupCheckoutId = uuid();
347+
const guestStayAccountIds = [uuid(), uuid()];
348+
const now = new Date();
349+
350+
const consumer = postgreSQLEventStoreConsumer({
351+
connectionString,
352+
});
353+
354+
consumer.workflowProcessor<
355+
GroupCheckoutInput,
356+
GroupCheckout,
357+
GroupCheckoutOutput
358+
>({
359+
...workflowProcessorOptions,
360+
separateInputInboxFromProcessing: true,
361+
stopAfter: (message) =>
362+
message.type === 'GroupCheckoutInitiated' &&
363+
message.data.groupCheckoutId === groupCheckoutId,
364+
});
365+
366+
try {
367+
const consumerPromise = consumer.start();
368+
369+
await eventStore.appendToStream(`groupCheckout-${groupCheckoutId}`, [
370+
{
371+
type: 'InitiateGroupCheckout',
372+
data: {
373+
groupCheckoutId,
374+
clerkId: 'clerk-1',
375+
guestStayAccountIds,
376+
now,
377+
},
378+
},
379+
]);
380+
381+
await consumerPromise;
382+
383+
const { events } = await eventStore.readStream(
384+
`emt:workflow:${groupCheckoutId}`,
385+
);
386+
387+
assertThatArray(events).isNotEmpty();
388+
// In double-hop mode, first message should be the prefixed input
389+
assertEqual(
390+
events[0]!.type,
391+
'GroupCheckoutWorkflow:InitiateGroupCheckout',
392+
);
393+
// Then the workflow outputs
394+
assertEqual(events[1]!.type, 'GroupCheckoutInitiated');
395+
assertEqual(events[2]!.type, 'CheckOut');
396+
assertEqual(events[3]!.type, 'CheckOut');
397+
} finally {
398+
await consumer.close();
399+
}
400+
},
401+
);
402+
403+
void it(
404+
'processes external events in double-hop mode after storing with prefix',
405+
withDeadline,
406+
async () => {
407+
const groupCheckoutId = uuid();
408+
const guestId = uuid();
409+
const now = new Date();
410+
411+
const handleWorkflow = WorkflowHandler(workflowProcessorOptions);
412+
await handleWorkflow(eventStore, {
413+
type: 'InitiateGroupCheckout',
414+
data: {
415+
groupCheckoutId,
416+
clerkId: 'clerk-1',
417+
guestStayAccountIds: [guestId],
418+
now,
419+
},
420+
});
421+
422+
const consumer = postgreSQLEventStoreConsumer({
423+
connectionString,
424+
});
425+
426+
consumer.workflowProcessor<
427+
GroupCheckoutInput,
428+
GroupCheckout,
429+
GroupCheckoutOutput
430+
>({
431+
...workflowProcessorOptions,
432+
separateInputInboxFromProcessing: true,
433+
stopAfter: (message) =>
434+
message.type === 'GroupCheckoutCompleted' &&
435+
message.data.groupCheckoutId === groupCheckoutId,
436+
});
437+
438+
try {
439+
const consumerPromise = consumer.start();
440+
441+
await eventStore.appendToStream(`guestStay-${guestId}`, [
442+
{
443+
type: 'GuestCheckedOut',
444+
data: {
445+
guestStayAccountId: guestId,
446+
checkedOutAt: now,
447+
groupCheckoutId,
448+
},
449+
},
450+
]);
451+
452+
await consumerPromise;
453+
454+
const { events } = await eventStore.readStream(
455+
`emt:workflow:${groupCheckoutId}`,
456+
);
457+
458+
const eventTypes = events.map((e) => e.type);
459+
// Verify both prefixed inputs are stored
460+
assertThatArray(eventTypes).containsElements([
461+
'GroupCheckoutWorkflow:InitiateGroupCheckout',
462+
'GroupCheckoutInitiated',
463+
'GroupCheckoutWorkflow:GuestCheckedOut',
464+
'GroupCheckoutCompleted',
465+
]);
466+
467+
// Verify the prefixed messages appear before their corresponding outputs
468+
const initiateIndex = eventTypes.indexOf(
469+
'GroupCheckoutWorkflow:InitiateGroupCheckout',
470+
);
471+
const initiatedIndex = eventTypes.indexOf('GroupCheckoutInitiated');
472+
assertEqual(
473+
initiateIndex < initiatedIndex,
474+
true,
475+
'Prefixed input should appear before its output',
476+
);
477+
478+
const guestCheckedOutIndex = eventTypes.indexOf(
479+
'GroupCheckoutWorkflow:GuestCheckedOut',
480+
);
481+
const completedIndex = eventTypes.indexOf('GroupCheckoutCompleted');
482+
assertEqual(
483+
guestCheckedOutIndex < completedIndex,
484+
true,
485+
'Prefixed external input should appear before completion output',
486+
);
487+
} finally {
488+
await consumer.close();
489+
}
490+
},
491+
);
274492
});

‎src/packages/emmett-postgresql/src/eventStore/consumers/postgreSQLProcessor.ts‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -298,7 +298,7 @@ const postgreSQLProcessingScope = (options: {
298298
client,
299299
transaction: transaction as PgTransaction,
300300
messageStore: getPostgreSQLEventStore(connectionString, {
301-
connectionOptions: { dumbo: pool as PgPool },
301+
connectionOptions: { client },
302302
}),
303303
},
304304
});
@@ -471,7 +471,7 @@ export const postgreSQLWorkflowProcessor = <
471471
HandlerContext,
472472
StoredMessage
473473
>,
474-
): PostgreSQLProcessor<Input> => {
474+
): PostgreSQLProcessor<Input | Output> => {
475475
const {
476476
processorId = options.processorId ??
477477
getWorkflowId({
@@ -523,12 +523,12 @@ export const postgreSQLWorkflowProcessor = <
523523
processorId,
524524
partition,
525525
}) as unknown as MessageProcessingScope<HandlerContext>,
526-
checkpoints: postgreSQLCheckpointer<Input>() as Checkpointer<
527-
Input,
526+
checkpoints: postgreSQLCheckpointer<Input | Output>() as Checkpointer<
527+
Input | Output,
528528
MetaDataType,
529529
HandlerContext
530530
>,
531-
}) as PostgreSQLProcessor<Input>;
531+
}) as PostgreSQLProcessor<Input | Output>;
532532
};
533533

534534
export const postgreSQLReactor = <

‎src/packages/emmett/src/processors/processors.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ export type MessageProcessor<
8181
id: string;
8282
instanceId: string;
8383
type: string;
84+
canHandle?: string[];
8485
init: (options: Partial<HandlerContext>) => Promise<void>;
8586
start: (
8687
options: Partial<HandlerContext>,
@@ -366,6 +367,7 @@ export const reactor = <
366367
id: processorId,
367368
instanceId,
368369
type,
370+
canHandle,
369371
init,
370372
start: async (
371373
startOptions: Partial<HandlerContext>,

0 commit comments

Comments
 (0)