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

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 (
-        previous?.state === 'running' &&
-        previous.processRunId === runId &&
-        active.has(previous.attemptId)
-      ) {
+      if (isJoinableRunningAttempt(previous, runId, active)) {
         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 (
-        previous?.state === 'running' &&
-        previous.processRunId === processRunId &&
-        active.has(previous.attemptId)
-      ) {
+      if (isJoinableRunningAttempt(previous, processRunId, active)) {
         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 (
-        previous?.state === 'running' &&
-        previous.processRunId === runId &&
-        active.has(previous.attemptId)
-      ) {
+      if (isJoinableRunningAttempt(previous, runId, active)) {
         return { work: previous, launch: false };
       }
       const data = {