refactor: 일일 금융 동기화 N+1 쿼리 제거 및 계좌 저장 벌크화 (#233)

services/account-service/src/main/java/com/ntropy/account/domain/Batching.java

@@ -0,0 +1,31 @@
+package com.ntropy.account.domain;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * MyBatis {@code IN}/다중 {@code VALUES} 절에 넘길 리스트를 안전한 크기로 나눈다 (이슈 #233).
+ * MySQL packet 크기와 파라미터 수 제한을 고려해 사용자·계좌 수가 많아도 단일 쿼리가 과도하게
+ * 커지지 않도록 chunk 단위로 분할한다.
+ */
+public final class Batching {
+
+    private Batching() {
+    }
+
+    public static <T> List<List<T>> chunk(List<T> items, int size) {
+        Objects.requireNonNull(items, "items");
+        if (size <= 0) {
+            throw new IllegalArgumentException("chunk size는 양수여야 합니다.");
+        }
+        if (items.isEmpty()) {
+            return List.of();
+        }
+        List<List<T>> chunks = new ArrayList<>();
+        for (int i = 0; i < items.size(); i += size) {
+            chunks.add(items.subList(i, Math.min(i + size, items.size())));
+        }
+        return chunks;
+    }
+}

services/account-service/src/main/java/com/ntropy/account/mapper/AccountMapper.java

@@ -11,11 +11,18 @@ public interface AccountMapper {
 
     void upsert(Account account);
 
+    /** {@link #upsert}의 다중 VALUES bulk 버전 (이슈 #233). 계좌 수와 무관하게 쿼리 1회로 저장한다. */
+    void upsertAll(@Param("list") List<Account> accounts);
+
     void updateAccountDetails(Account account);
 
     Account findByConnectionIdAndAccountNoHash(@Param("codefConnectionId") Long codefConnectionId,
                                                @Param("accountNoHash") String accountNoHash);
 
+    /** {@link #findByConnectionIdAndAccountNoHash}의 일괄 조회 버전 (이슈 #233). */
+    List<Account> findByConnectionIdAndAccountNoHashes(@Param("codefConnectionId") Long codefConnectionId,
+                                                        @Param("accountNoHashes") List<String> accountNoHashes);
+
     Account findByIdAndUserIdAndProvider(@Param("id") Long id,
                                          @Param("userId") Long userId,
                                          @Param("provider") String provider);

services/account-service/src/main/java/com/ntropy/account/mapper/CodefConnectionMapper.java

@@ -1,5 +1,7 @@
 package com.ntropy.account.mapper;
 
+import java.util.List;
+
 import com.ntropy.account.domain.entity.CodefConnection;
 import org.apache.ibatis.annotations.Mapper;
 import org.apache.ibatis.annotations.Param;
@@ -14,4 +16,8 @@ public interface CodefConnectionMapper {
     void upsert(CodefConnection codefConnection);
 
     CodefConnection findByUserIdAndProvider(@Param("userId") Long userId, @Param("provider") String provider);
+
+    /** 일일 동기화의 사용자별 단건 조회를 대체하는 chunk 단위 일괄 조회 (이슈 #233). */
+    List<CodefConnection> findByUserIdsAndProvider(@Param("userIds") List<Long> userIds,
+                                                    @Param("provider") String provider);
 }

services/account-service/src/main/java/com/ntropy/account/service/AccountCollectionService.java

@@ -3,8 +3,10 @@
 import java.time.LocalDate;
 import java.time.YearMonth;
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.BooleanSupplier;
 
@@ -22,6 +24,7 @@
 import com.ntropy.account.client.codef.parser.LoanTransactionResponseParser;
 import com.ntropy.account.client.codef.parser.LoanTransactionResponseParser.ParsedLoan;
 import com.ntropy.account.domain.AccountGroup;
+import com.ntropy.account.domain.Batching;
 import com.ntropy.account.domain.ConnectionProvider;
 import com.ntropy.account.domain.PersonalBank;
 import com.ntropy.account.domain.entity.Account;
@@ -113,7 +116,22 @@ public List<AccountCollectionOutcome> collectForDailySync(Long userId, PersonalB
                                                                LocalDate transactionStartDate,
                                                                LocalDate transactionEndDate,
                                                                BooleanSupplier heartbeat) {
-        CodefConnection connection = requireCodefConnection(userId);
+        return collectForDailySync(
+                userId, bank, requireCodefConnection(userId), birthDate, transactionStartDate, transactionEndDate,
+                heartbeat
+        );
+    }
+
+    /**
+     * {@link #collectForDailySync(Long, PersonalBank, String, LocalDate, LocalDate, BooleanSupplier)}와 같지만,
+     * 호출자가 이미 조회한 {@link CodefConnection}을 그대로 받아 동일 사용자의 기관별 반복 조회에서
+     * {@link #requireCodefConnection}을 다시 호출하지 않는다 (이슈 #233).
+     */
+    public List<AccountCollectionOutcome> collectForDailySync(Long userId, PersonalBank bank,
+                                                               CodefConnection connection, String birthDate,
+                                                               LocalDate transactionStartDate,
+                                                               LocalDate transactionEndDate,
+                                                               BooleanSupplier heartbeat) {
         String normalizedBirthDate = bank.normalizeBirthDate(birthDate);
         List<SavedAccountContext> savedContexts = fetchAndSaveAccounts(userId, bank, connection, heartbeat);
 
@@ -155,6 +173,9 @@ private CodefConnection requireCodefConnection(Long userId) {
         return connection;
     }
 
+    /** MySQL packet 크기·MyBatis 파라미터 수를 고려한 계좌 bulk upsert/조회 batch 크기 (이슈 #233). */
+    private static final int ACCOUNT_UPSERT_BATCH_SIZE = 200;
+
     /** 실행마다 보유계좌를 재조회해 원문 계좌번호를 이 요청 흐름 안에서만 확보한다(저장하지 않음). */
     private List<SavedAccountContext> fetchAndSaveAccounts(Long userId, PersonalBank bank, CodefConnection connection,
                                                            BooleanSupplier heartbeat) {
@@ -170,13 +191,36 @@ private List<SavedAccountContext> fetchAndSaveAccounts(Long userId, PersonalBank
         List<ParsedAccount> parsedAccounts = AccountResponseParser.parse(
                 accountListResponse.path("data"), connection.getId(), userId, bank.getOrganizationCode()
         );
+        if (parsedAccounts.isEmpty()) {
+            requireLease(heartbeat);
+            return List.of();
+        }
+
+        Map<String, Account> savedByHash = new LinkedHashMap<>();
+        for (List<ParsedAccount> chunk : Batching.chunk(parsedAccounts, ACCOUNT_UPSERT_BATCH_SIZE)) {
+            requireLease(heartbeat);
+            List<Account> accountsToUpsert = chunk.stream().map(ParsedAccount::account).toList();
+            accountMapper.upsertAll(accountsToUpsert);
+            List<String> accountNoHashes = accountsToUpsert.stream().map(Account::getAccountNoHash).toList();
+            List<Account> savedAccounts =
+                    accountMapper.findByConnectionIdAndAccountNoHashes(connection.getId(), accountNoHashes);
+            Map<String, Account> savedChunkByHash = new LinkedHashMap<>();
+            for (Account saved : savedAccounts) {
+                savedChunkByHash.put(saved.getAccountNoHash(), saved);
+                savedByHash.put(saved.getAccountNoHash(), saved);
+            }
+            if (!savedChunkByHash.keySet().containsAll(accountNoHashes)) {
+                throw new IllegalStateException("CODEF 계좌 bulk upsert 결과가 누락되었습니다.");
+            }
+            requireLease(heartbeat);
+        }
 
         List<SavedAccountContext> savedContexts = new ArrayList<>();
         for (ParsedAccount parsed : parsedAccounts) {
-            accountMapper.upsert(parsed.account());
-            Account saved = accountMapper.findByConnectionIdAndAccountNoHash(
-                    connection.getId(), parsed.account().getAccountNoHash()
-            );
+            Account saved = savedByHash.get(parsed.account().getAccountNoHash());
+            if (saved == null) {
+                throw new IllegalStateException("CODEF 계좌 bulk upsert 결과를 매핑할 수 없습니다.");
+            }
             savedContexts.add(new SavedAccountContext(saved, parsed.rawAccountNo()));
         }
         requireLease(heartbeat);

services/account-service/src/main/java/com/ntropy/account/service/DailyCodefSyncService.java

@@ -15,6 +15,7 @@
 
 import com.ntropy.account.config.IncrementalSyncPolicy;
 import com.ntropy.account.domain.AccountSyncStatus;
+import com.ntropy.account.domain.Batching;
 import com.ntropy.account.domain.ConnectionProvider;
 import com.ntropy.account.domain.IncrementalSyncRangeCalculator;
 import com.ntropy.account.domain.InstitutionKeys;
@@ -46,6 +47,9 @@ public class DailyCodefSyncService {
 
     public static final String JOB_NAME = "daily-sync-codef";
 
+    /** 사용자 ID IN 절이 과도하게 커지지 않도록 나누는 chunk 크기 (이슈 #233). */
+    private static final int USER_ID_CHUNK_SIZE = 500;
+
     private final CodefConnectionMapper codefConnectionMapper;
     private final AccountSyncStateMapper accountSyncStateMapper;
     private final AccountTransactionMapper accountTransactionMapper;
@@ -62,9 +66,11 @@ public DailyFinancialSyncResult synchronize(List<Long> activeUserIds, LocalDate
         long processedTransactionCount = 0;
         boolean leaseLost = false;
 
+        Map<Long, CodefConnection> connectionsByUserId = fetchConnectionsByUserId(activeUserIds);
+
         userLoop:
         for (Long userId : activeUserIds) {
-            CodefConnection connection = codefConnectionMapper.findByUserIdAndProvider(userId, ConnectionProvider.CODEF.name());
+            CodefConnection connection = connectionsByUserId.get(userId);
             if (connection == null || connection.getConnectedId() == null || connection.getConnectedId().isBlank()) {
                 continue; // 이 provider의 동기화 대상이 아닌 사용자
             }
@@ -157,6 +163,17 @@ public DailyFinancialSyncResult synchronize(List<Long> activeUserIds, LocalDate
         );
     }
 
+    /** 활성 사용자의 CODEF 연결을 chunk 단위로 일괄 조회한다 (이슈 #233). */
+    private Map<Long, CodefConnection> fetchConnectionsByUserId(List<Long> userIds) {
+        Map<Long, CodefConnection> connectionsByUserId = new LinkedHashMap<>();
+        for (List<Long> chunk : Batching.chunk(userIds, USER_ID_CHUNK_SIZE)) {
+            for (CodefConnection connection : codefConnectionMapper.findByUserIdsAndProvider(chunk, ConnectionProvider.CODEF.name())) {
+                connectionsByUserId.put(connection.getUserId(), connection);
+            }
+        }
+        return connectionsByUserId;
+    }
+
     private InstitutionSyncOutcome synchronizeInstitution(Long userId, PersonalBank bank, CodefConnection connection,
                                                            LocalDate businessDate, LeaseHandle lease) {
         String organizationCode = bank.getOrganizationCode();
@@ -185,7 +202,7 @@ private InstitutionSyncOutcome synchronizeInstitution(Long userId, PersonalBank
         List<AccountCollectionOutcome> outcomes;
         try {
             outcomes = accountCollectionService.collectForDailySync(
-                    userId, bank, birthDate, startDate, businessDate, () -> leaseService.heartbeat(lease)
+                    userId, bank, connection, birthDate, startDate, businessDate, () -> leaseService.heartbeat(lease)
             );
         } catch (LeaseLostException leaseLostSignal) {
             throw leaseLostSignal; // 상위 루프에서 전체 중단 처리

services/account-service/src/main/java/com/ntropy/account/service/DailyNtropySyncService.java

@@ -16,6 +16,7 @@
 import com.ntropy.account.config.IncrementalSyncPolicy;
 import com.ntropy.account.domain.AccountGroup;
 import com.ntropy.account.domain.AccountSyncStatus;
+import com.ntropy.account.domain.Batching;
 import com.ntropy.account.domain.ConnectionProvider;
 import com.ntropy.account.domain.IncrementalSyncRangeCalculator;
 import com.ntropy.account.domain.InstitutionKeys;
@@ -46,6 +47,9 @@ public class DailyNtropySyncService {
     public static final String JOB_NAME = "daily-sync-ntropy";
     private static final String ORDINARY_DEPOSIT_TYPE_CODE = "11";
 
+    /** 사용자 ID IN 절이 과도하게 커지지 않도록 나누는 chunk 크기 (이슈 #233). */
+    private static final int USER_ID_CHUNK_SIZE = 500;
+
     private final CodefConnectionMapper codefConnectionMapper;
     private final AccountMapper accountMapper;
     private final AccountTransactionMapper accountTransactionMapper;
@@ -62,15 +66,18 @@ public DailyFinancialSyncResult synchronize(List<Long> activeUserIds, LocalDate
         long processedTransactionCount = 0;
         boolean leaseLost = false;
 
+        Map<Long, CodefConnection> connectionsByUserId = fetchConnectionsByUserId(activeUserIds);
+
         userLoop:
         for (Long userId : activeUserIds) {
-            CodefConnection connection = codefConnectionMapper.findByUserIdAndProvider(userId, ConnectionProvider.NTROPY.name());
+            CodefConnection connection = connectionsByUserId.get(userId);
             if (connection == null || connection.getConnectedId() == null || connection.getConnectedId().isBlank()) {
                 continue; // 이 provider의 동기화 대상이 아닌 사용자
             }
 
             boolean userHasFailure = false;
             boolean userHasSuccess = false;
+            Map<String, Account> ordinaryAccountsByOrganization = findOrdinaryAccountsByOrganization(userId);
 
             for (String organizationCode : InstitutionKeys.parse(connection.getRegisteredInstitutionKeys())) {
                 if (!leaseService.heartbeat(lease)) {
@@ -79,7 +86,10 @@ public DailyFinancialSyncResult synchronize(List<Long> activeUserIds, LocalDate
                 }
 
                 accountSyncStateMapper.insertIfAbsent(pendingSyncState(connection.getId(), organizationCode));
-                InstitutionGenerationOutcome outcome = generateForInstitution(userId, connection, organizationCode, businessDate);
+                InstitutionGenerationOutcome outcome = generateForInstitution(
+                        connection, organizationCode, businessDate,
+                        ordinaryAccountsByOrganization.get(organizationCode)
+                );
                 processedTransactionCount += outcome.transactionCount();
                 institutionResults.add(new InstitutionSyncResult(
                         organizationCode, connection.getId(), outcome.aggregate().status(),
@@ -140,14 +150,31 @@ public DailyFinancialSyncResult synchronize(List<Long> activeUserIds, LocalDate
         );
     }
 
-    private InstitutionGenerationOutcome generateForInstitution(Long userId, CodefConnection connection,
-                                                                 String organizationCode, LocalDate businessDate) {
-        Account ordinaryAccount = accountMapper.findByUserIdAndProvider(userId, ConnectionProvider.NTROPY.name()).stream()
-                .filter(account -> organizationCode.equals(account.getOrganizationCode()))
-                .filter(account -> account.getAccountGroup() == AccountGroup.DEPOSIT_TRUST)
-                .filter(account -> ORDINARY_DEPOSIT_TYPE_CODE.equals(account.getDepositTypeCode()))
-                .findFirst()
-                .orElse(null);
+    /** 사용자의 NTROPY 수시입출 계좌를 한 번만 조회해 기관코드 기준으로 그룹핑한다 (이슈 #233). */
+    private Map<String, Account> findOrdinaryAccountsByOrganization(Long userId) {
+        Map<String, Account> ordinaryAccountsByOrganization = new LinkedHashMap<>();
+        for (Account account : accountMapper.findByUserIdAndProvider(userId, ConnectionProvider.NTROPY.name())) {
+            if (account.getAccountGroup() == AccountGroup.DEPOSIT_TRUST
+                    && ORDINARY_DEPOSIT_TYPE_CODE.equals(account.getDepositTypeCode())) {
+                ordinaryAccountsByOrganization.putIfAbsent(account.getOrganizationCode(), account);
+            }
+        }
+        return ordinaryAccountsByOrganization;
+    }
+
+    /** 활성 사용자의 NTROPY 연결을 chunk 단위로 일괄 조회한다 (이슈 #233). */
+    private Map<Long, CodefConnection> fetchConnectionsByUserId(List<Long> userIds) {
+        Map<Long, CodefConnection> connectionsByUserId = new LinkedHashMap<>();
+        for (List<Long> chunk : Batching.chunk(userIds, USER_ID_CHUNK_SIZE)) {
+            for (CodefConnection connection : codefConnectionMapper.findByUserIdsAndProvider(chunk, ConnectionProvider.NTROPY.name())) {
+                connectionsByUserId.put(connection.getUserId(), connection);
+            }
+        }
+        return connectionsByUserId;
+    }
+
+    private InstitutionGenerationOutcome generateForInstitution(CodefConnection connection, String organizationCode,
+                                                                 LocalDate businessDate, Account ordinaryAccount) {
         if (ordinaryAccount == null) {
             return InstitutionGenerationOutcome.failed(InstitutionAggregate.failed("ORDINARY_ACCOUNT_NOT_FOUND"));
         }
@@ -160,7 +187,7 @@ private InstitutionGenerationOutcome generateForInstitution(Long userId, CodefCo
 
         try {
             List<AccountTransaction> transactions = transactionGenerator.generate(
-                    userId, ordinaryAccount.getId(), startDate, businessDate
+                    ordinaryAccount.getUserId(), ordinaryAccount.getId(), startDate, businessDate
             );
             if (!transactions.isEmpty()) {
                 accountTransactionMapper.insertAll(transactions);