fix(api): retry commands that clash with a concurrent write

Services run commands through a CommandRunner that runs one again (up to
5 times, waiting a little longer each time) when its aggregate changed
underneath, e.g. a saga writing to the same ticket. Each command loads its
aggregate fresh and stores its events in one transaction, so a
ConcurrencyError means nothing was stored and running it again is safe.
People no longer get a 409 for it, and the e2e suite stops flaking.

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

libs/api/shared/application/src/index.ts

@@ -2,3 +2,4 @@ export * from './lib/ports/event-store';
 export * from './lib/aggregate-rehydrator';
 export * from './lib/shared.module';
 export * from './lib/event-recorder';
+export * from './lib/command-runner';

libs/api/shared/application/src/lib/command-runner.spec.ts

@@ -0,0 +1,48 @@
+import { Command, CommandBus } from '@nestjs/cqrs';
+import {
+  BusinessRuleError,
+  ConcurrencyError,
+} from '@task-stream/api/shared/domain';
+import { COMMAND_ATTEMPTS, CommandRunner } from './command-runner';
+
+class Rename extends Command<string> {}
+
+const busFailing = (errors: Error[]) => {
+  const execute = jest.fn(async () => {
+    const error = errors.shift();
+    if (error) throw error;
+    return 'done';
+  });
+  return { bus: { execute } as unknown as CommandBus, execute };
+};
+
+describe('CommandRunner', () => {
+  it('runs a command again while its aggregate changed underneath', async () => {
+    const { bus, execute } = busFailing([
+      new ConcurrencyError('Aggregate', 'a'),
+      new ConcurrencyError('Aggregate', 'a'),
+    ]);
+    await expect(new CommandRunner(bus).execute(new Rename())).resolves.toBe(
+      'done',
+    );
+    expect(execute).toHaveBeenCalledTimes(3);
+  });
+
+  it('gives up after a few conflicts, and never retries other errors', async () => {
+    const conflicts = Array.from(
+      { length: COMMAND_ATTEMPTS },
+      () => new ConcurrencyError('Aggregate', 'a'),
+    );
+    const stubborn = busFailing(conflicts);
+    await expect(
+      new CommandRunner(stubborn.bus).execute(new Rename()),
+    ).rejects.toBeInstanceOf(ConcurrencyError);
+    expect(stubborn.execute).toHaveBeenCalledTimes(COMMAND_ATTEMPTS);
+
+    const refused = busFailing([new BusinessRuleError('No')]);
+    await expect(
+      new CommandRunner(refused.bus).execute(new Rename()),
+    ).rejects.toBeInstanceOf(BusinessRuleError);
+    expect(refused.execute).toHaveBeenCalledTimes(1);
+  });
+});

libs/api/shared/application/src/lib/command-runner.ts

@@ -0,0 +1,31 @@
+import { Injectable } from '@nestjs/common';
+import { Command, CommandBus } from '@nestjs/cqrs';
+import { ConcurrencyError } from '@task-stream/api/shared/domain';
+
+/** Attempts before a conflict reaches the caller (as a 409). */
+export const COMMAND_ATTEMPTS = 5;
+
+/**
+ * Runs commands for the application services, and runs one again when its
+ * aggregate changed underneath (someone, or a saga, wrote to it at the same
+ * moment). Every command loads its aggregate fresh and stores its events in
+ * one transaction, so a ConcurrencyError means nothing was stored: running
+ * it again on the new state is safe.
+ */
+@Injectable()
+export class CommandRunner {
+  constructor(private readonly commandBus: CommandBus) {}
+
+  async execute<R>(command: Command<R>): Promise<R> {
+    for (let attempt = 1; ; attempt++) {
+      try {
+        return await this.commandBus.execute(command);
+      } catch (error) {
+        if (!(error instanceof ConcurrencyError) || attempt >= COMMAND_ATTEMPTS)
+          throw error;
+        // A little later each time, so the other writer can finish.
+        await new Promise((resolve) => setTimeout(resolve, 25 * attempt));
+      }
+    }
+  }
+}

libs/api/shared/application/src/lib/shared.module.ts

@@ -1,12 +1,13 @@
 import { DynamicModule, Global, Module, Type } from '@nestjs/common';
 import { CqrsModule } from '@nestjs/cqrs';
 import { AggregateRehydrator } from './aggregate-rehydrator';
+import { CommandRunner } from './command-runner';
 import { EventRecorder } from './event-recorder';
 
 /**
  * Shared building blocks for every bounded context: the event store (from
  * the infrastructure module plugged in by the composition root), the
- * AggregateRehydrator and the EventRecorder.
+ * AggregateRehydrator, the EventRecorder and the CommandRunner.
  */
 @Global()
 @Module({})
@@ -17,8 +18,13 @@ export class SharedModule {
     return {
       module: SharedModule,
       imports: [CqrsModule, infrastructureModule],
-      providers: [AggregateRehydrator, EventRecorder],
-      exports: [infrastructureModule, AggregateRehydrator, EventRecorder],
+      providers: [AggregateRehydrator, EventRecorder, CommandRunner],
+      exports: [
+        infrastructureModule,
+        AggregateRehydrator,
+        EventRecorder,
+        CommandRunner,
+      ],
     };
   }
 }