refactor: share sync reservation join rule (WealthLedger/backend#119)

src/integrations/tbank/instrument-metadata.repository.ts

@@ -7,10 +7,10 @@ import type {
   TbankInstrument as MetadataRow,
   Prisma,
 } from '../../../generated/prisma/client';
-import type { TbankSyncAttempt } from './tbank-access.service';
 import type { TbankInstrument } from './tbank.client';
 
 import { PrismaService } from '../../prisma/prisma.service';
+import { isJoinableRunningAttempt, type TbankSyncAttempt } from './tbank-access.service';
 import { TbankErrorCode } from './tbank-error-code';
 
 const FIRST_BATCH_OFFSET = 0;
@@ -131,11 +131,7 @@ export class InstrumentMetadataRepository {
       const previous = await tx.tbankSyncWork.findUnique({
         where: { userId_revision_scopeKey: scope },
       });
-      if (
+      if (isJoinableRunningAttempt(previous, runId, active)) {
-        previous?.state === 'running' &&
-        previous.processRunId === runId &&
-        active.has(previous.attemptId)
-      ) {
         return { work: previous, launch: false };
       }
       const data = {

src/integrations/tbank/tbank-access.service.ts

@@ -25,6 +25,22 @@ export type TbankSyncAttempt = {
   userId: string;
 };
 
+// Join rule shared by sync reservations: the same process joins its own running attempt
+// only while that attempt is still active in memory. Set and Map both satisfy `has`.
+export function isJoinableRunningAttempt<
+  T extends { attemptId: string; processRunId: string; state: string },
+>(
+  previous: null | T,
+  processRunId: string,
+  active: { has(attemptId: string): boolean },
+): previous is T {
+  return (
+    previous?.state === 'running' &&
+    previous.processRunId === processRunId &&
+    active.has(previous.attemptId)
+  );
+}
+
 @Injectable()
 export class TbankAccessService {
   constructor(
@@ -62,11 +78,7 @@ export class TbankAccessService {
       const previous = await tx.tbankSyncWork.findUnique({
         where: { userId_revision_scopeKey: scope },
       });
-      if (
+      if (isJoinableRunningAttempt(previous, processRunId, active)) {
-        previous?.state === 'running' &&
-        previous.processRunId === processRunId &&
-        active.has(previous.attemptId)
-      ) {
         return { work: previous, launch: false };
       }
       const data = {

src/integrations/tbank/tbank.api.ts

@@ -3,6 +3,7 @@ export { InstrumentMetadataService } from './instrument-metadata.service';
 export { metadataKey } from './instrument-metadata.service';
 export type { MetadataIdentity } from './instrument-metadata.service';
 export {
+  isJoinableRunningAttempt,
   TbankAccessService,
   type TbankSyncAttempt,
 } from './tbank-access.service';

src/modules/investments/instruments/infrastructure/catalog-discovery.repository.ts

@@ -3,9 +3,9 @@ import { randomUUID } from 'node:crypto';
 import { isDeepStrictEqual } from 'node:util';
 
 import type { Prisma } from '../../../../../generated/prisma/client';
-import type { TbankBond, TbankShare, TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
 
 import { TbankErrorCode } from '../../../../integrations/tbank/tbank-error-code';
+import { isJoinableRunningAttempt, type TbankBond, type TbankShare, type TbankSyncAttempt } from '../../../../integrations/tbank/tbank.api';
 import { PrismaService } from '../../../../prisma/prisma.service';
 
 const UNCHANGED_REVISION_INCREMENT = 0;
@@ -293,11 +293,7 @@ export class CatalogDiscoveryRepository {
       const previous = await tx.tbankSyncWork.findUnique({
         where: { userId_revision_scopeKey: scope },
       });
-      if (
+      if (isJoinableRunningAttempt(previous, runId, active)) {
-        previous?.state === 'running' &&
-        previous.processRunId === runId &&
-        active.has(previous.attemptId)
-      ) {
         return { work: previous, launch: false };
       }
       const data = {

test/tbank-membership-reservation.e2e-spec.ts

@@ -0,0 +1,107 @@
+import { ConfigService } from '@nestjs/config';
+
+import { TbankAccessService } from '../src/integrations/tbank/tbank-access.service';
+import { TbankTokenService } from '../src/integrations/tbank/tbank-token.service';
+import { PrismaService } from '../src/prisma/prisma.service';
+
+// Characterizes the membership reserve protocol on real SQLite transactions.
+// Join by the same process while the attempt is active; launch after it is not.
+describe('membership reservation protocol (SQLite)', () => {
+  const prisma = new PrismaService(
+    new ConfigService({ DATABASE_URL: process.env.DATABASE_URL }),
+  );
+  const tokens = new TbankTokenService(
+    new ConfigService({
+      TOKEN_ENCRYPTION_KEY: process.env.TOKEN_ENCRYPTION_KEY,
+    }) as never,
+  );
+  const access = new TbankAccessService(prisma, tokens);
+  const userId = 'synthetic-membership-reservation';
+  const revision = `${userId}-revision`;
+
+  beforeAll(async () => {
+    await prisma.$connect();
+    await prisma.tbankSyncWork.deleteMany({ where: { userId } });
+    await prisma.tbankConnection.deleteMany({ where: { userId } });
+    await prisma.user.deleteMany({ where: { id: userId } });
+    await prisma.user.create({
+      data: {
+        id: userId,
+        email: `${userId}@example.invalid`,
+        passwordHash: 'synthetic',
+      },
+    });
+    await prisma.tbankConnection.create({
+      data: {
+        userId,
+        revision,
+        encryptedToken: tokens.encrypt('synthetic-membership'),
+        connectedAt: new Date(),
+      },
+    });
+  });
+
+  afterAll(async () => {
+    await prisma.tbankSyncWork.deleteMany({ where: { userId } });
+    await prisma.tbankConnection.deleteMany({ where: { userId } });
+    await prisma.user.deleteMany({ where: { id: userId } });
+    await prisma.$disconnect();
+  });
+
+  it('launches once, joins the active attempt by the same process, and relaunches after it finishes', async () => {
+    const processRunId = 'run-membership-a';
+    const first = await access.reserveMembership(
+      userId,
+      revision,
+      processRunId,
+      new Set(),
+      true,
+    );
+    expect(first?.launch).toBe(true);
+    const attemptId = first?.work.attemptId ?? '';
+
+    const joined = await access.reserveMembership(
+      userId,
+      revision,
+      processRunId,
+      new Set([attemptId]),
+      true,
+    );
+    expect(joined?.launch).toBe(false);
+    expect(joined?.work.id).toBe(first?.work.id);
+    expect(joined?.work.attemptId).toBe(attemptId);
+
+    // The attempt is no longer in the active set: this must not join it.
+    const relaunched = await access.reserveMembership(
+      userId,
+      revision,
+      processRunId,
+      new Set(),
+      true,
+    );
+    expect(relaunched?.launch).toBe(true);
+    expect(relaunched?.work.attemptId).not.toBe(attemptId);
+  });
+
+  it('does not join an attempt started by another process', async () => {
+    const other = await access.reserveMembership(
+      userId,
+      revision,
+      'run-membership-b',
+      new Set(),
+      true,
+    );
+    expect(other?.launch).toBe(true);
+    const attemptId = other?.work.attemptId ?? '';
+
+    const foreign = await access.reserveMembership(
+      userId,
+      revision,
+      'run-membership-c',
+      new Set([attemptId]),
+      true,
+    );
+    expect(foreign?.launch).toBe(true);
+    expect(foreign?.work.attemptId).not.toBe(attemptId);
+  });
+});