A dedicated thread pool ({@code docProcessingPool}) is used for batch
+ * document processing so that parallel uploads do not starve the servlet
+ * container threads. Pool sizes are externally configurable via application
+ * properties.
+ *
+ *
+ * doc.processing.pool.size — core pool size (default 5)
+ * doc.processing.pool.max — max pool size (default 10)
+ * doc.processing.pool.queue — queue capacity (default 100)
+ *
+ */
+@Configuration
+@EnableAsync
+public class AsyncConfig {
+
+ @Bean("docProcessingPool")
+ public Executor docProcessingPool(
+ @Value("${doc.processing.pool.size:5}") int coreSize,
+ @Value("${doc.processing.pool.max:10}") int maxSize,
+ @Value("${doc.processing.pool.queue:100}") int queueCapacity) {
+ ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+ executor.setCorePoolSize(coreSize);
+ executor.setMaxPoolSize(maxSize);
+ executor.setQueueCapacity(queueCapacity);
+ executor.setThreadNamePrefix("doc-proc-");
+ executor.setWaitForTasksToCompleteOnShutdown(true);
+ executor.setAwaitTerminationSeconds(60);
+ executor.initialize();
+ return executor;
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java
index f531fe1..800c54f 100644
--- a/services/am-document-processor/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/controller/DocumentProcessorController.java
@@ -12,10 +12,12 @@
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.am.mypotrfolio.domain.common.DocumentType;
-import org.am.mypotrfolio.model.DocumentProcessResponse;
-import org.am.mypotrfolio.model.ProcessingStatus;
+import org.am.mypotrfolio.model.*;
+import org.am.mypotrfolio.service.BatchSyncEventPublisher;
import org.am.mypotrfolio.service.DocumentProcessorService;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.env.Environment;
+import org.springframework.core.env.Profiles;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
@@ -25,7 +27,9 @@
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
@@ -45,6 +49,12 @@ public class DocumentProcessorController {
@Autowired
private DocumentProcessorService documentProcessorService;
+ @Autowired
+ private BatchSyncEventPublisher batchSyncEventPublisher;
+
+ @Autowired
+ private Environment environment;
+
@Operation(summary = "Get supported document types", description = "Public endpoint — no authentication required")
@ApiResponses({
@ApiResponse(responseCode = "200", description = "Document types retrieved successfully", content = @Content(array = @ArraySchema(schema = @Schema(implementation = String.class)))),
@@ -97,45 +107,186 @@ public ResponseEntity processDocument(
}
}
- @Operation(summary = "Process multiple documents", security = @SecurityRequirement(name = "Bearer"))
+ /**
+ * @deprecated Use {@code POST /v1/documents/sync} instead. This endpoint forces a single
+ * brokerType for all files and processes them sequentially.
+ */
+ @Deprecated
+ @Operation(summary = "[Deprecated] Process multiple documents with one shared broker type",
+ description = "Deprecated — use POST /v1/documents/sync for multi-broker parallel processing.",
+ security = @SecurityRequirement(name = "Bearer"))
@ApiResponses({
- @ApiResponse(responseCode = "200", description = "Documents processed successfully", content = @Content(array = @ArraySchema(schema = @Schema(implementation = DocumentProcessResponse.class)))),
+ @ApiResponse(responseCode = "200", description = "Documents processed successfully"),
@ApiResponse(responseCode = "401", description = "Unauthorized"),
@ApiResponse(responseCode = "400", description = "Invalid input parameters"),
@ApiResponse(responseCode = "500", description = "Internal server error")
})
@PostMapping(value = "/batch-process", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public ResponseEntity processBatchDocuments(
- @Parameter(description = "List of portfolio document files to process", required = true) @RequestParam("files") List files,
- @Parameter(description = "Type of documents being processed", required = true) @RequestParam("documentType") DocumentType documentType,
- @Parameter(description = "Portfolio ID (optional)", required = false) @RequestParam(value = "portfolioId", required = false) String portfolioId,
- @Parameter(description = "Explicit Broker Type (optional)", required = false) @RequestParam(value = "brokerType", required = false) String brokerTypeStr) {
+ @Parameter(description = "List of portfolio document files to process", required = true)
+ @RequestParam("files") List files,
+ @Parameter(description = "Type of documents being processed", required = true)
+ @RequestParam("documentType") DocumentType documentType,
+ @Parameter(description = "Portfolio ID (optional)", required = false)
+ @RequestParam(value = "portfolioId", required = false) String portfolioId,
+ @Parameter(description = "Explicit Broker Type (optional)", required = false)
+ @RequestParam(value = "brokerType", required = false) String brokerTypeStr) {
String userId = resolveUserId();
-
- log.info("Batch processing {} documents for user: {}, type: {}, portfolio: {}, broker: {}",
- files.size(), userId, documentType, portfolioId, brokerTypeStr);
+ log.warn("Deprecated /batch-process called by user: {} — recommend migrating to /sync", userId);
try {
List responses = documentProcessorService.processBatchDocuments(
- files,
- documentType,
- portfolioId,
- brokerTypeStr,
- userId
- );
+ files, documentType, portfolioId, brokerTypeStr, userId);
return ResponseEntity.ok(responses);
} catch (IllegalArgumentException e) {
- log.warn("Invalid batch parameters: {}", e.getMessage());
- return ResponseEntity.badRequest()
- .body(new ErrorResponse("Invalid parameters: " + e.getMessage()));
+ return ResponseEntity.badRequest().body(new ErrorResponse("Invalid parameters: " + e.getMessage()));
} catch (Exception e) {
- log.error("Error batch processing documents for user: {}", userId, e);
+ log.error("Error in deprecated batch processing for user: {}", userId, e);
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
.body(new ErrorResponse("Failed to process documents"));
}
}
+ // =========================================================================
+ // Multi-broker sync endpoints (new)
+ // =========================================================================
+
+ @Operation(
+ summary = "Submit a multi-broker batch sync",
+ description = "Accepts N files from N different brokers. Broker type and document type are " +
+ "auto-detected per file. Processing is parallel and non-blocking — the response " +
+ "is returned immediately with a batchId. Poll GET /sync/{batchId}/status or " +
+ "stream progress via GET /sync/{batchId}/stream.",
+ security = @SecurityRequirement(name = "Bearer"))
+ @ApiResponses({
+ @ApiResponse(responseCode = "202", description = "Batch accepted and processing started",
+ content = @Content(schema = @Schema(implementation = BatchSyncStatus.class))),
+ @ApiResponse(responseCode = "401", description = "Unauthorized"),
+ @ApiResponse(responseCode = "400", description = "No valid files provided"),
+ @ApiResponse(responseCode = "500", description = "Internal server error")
+ })
+ @PostMapping(value = "/sync", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
+ public ResponseEntity submitBatchSync(
+ @Parameter(description = "Files to process (one or more, different brokers allowed)", required = true)
+ @RequestParam("files") List files,
+ @Parameter(description = "Optional per-file broker hints (same order as files, may be sparse)")
+ @RequestParam(value = "brokerTypes", required = false) List brokerTypes,
+ @Parameter(description = "Optional per-file document type hints (same order as files)")
+ @RequestParam(value = "documentTypes", required = false) List documentTypes,
+ @Parameter(description = "Optional per-file passwords for encrypted files")
+ @RequestParam(value = "passwords", required = false) List passwords,
+ @Parameter(description = "Optional per-file portfolio names/IDs (same order as files)")
+ @RequestParam(value = "portfolioIds", required = false) List portfolioIds,
+ @Parameter(description = "Optional batch-level portfolio ID used when a file has no per-file value")
+ @RequestParam(value = "portfolioId", required = false) String portfolioId) {
+
+ String userId = resolveUserId();
+
+ if (files == null || files.isEmpty()) {
+ return ResponseEntity.badRequest().body(new ErrorResponse("At least one file is required"));
+ }
+ if (files.size() > 5) {
+ return ResponseEntity.badRequest().body(new ErrorResponse("Maximum 5 files allowed per batch"));
+ }
+ for (MultipartFile file : files) {
+ if (file == null || file.isEmpty()) {
+ return ResponseEntity.badRequest().body(new ErrorResponse("Empty files are not allowed"));
+ }
+ }
+
+ log.info("Multi-broker sync request: {} files, user: {}", files.size(), userId);
+
+ List entries = new ArrayList<>();
+ for (int i = 0; i < files.size(); i++) {
+ entries.add(BatchSyncEntry.builder()
+ .file(files.get(i))
+ .brokerType(blankToNull(getOrNull(brokerTypes, i)))
+ .documentType(blankToNull(getOrNull(documentTypes, i)))
+ .password(blankToNull(getOrNull(passwords, i)))
+ .portfolioId(blankToNull(getOrNull(portfolioIds, i)))
+ .build());
+ }
+
+ try {
+ BatchSyncStatus status = documentProcessorService.submitBatchSync(
+ entries, userId, blankToNull(portfolioId));
+ return ResponseEntity.accepted().body(status);
+ } catch (IllegalArgumentException e) {
+ return ResponseEntity.badRequest().body(new ErrorResponse(e.getMessage()));
+ } catch (Exception e) {
+ log.error("Error submitting batch sync for user: {}", userId, e);
+ return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
+ .body(new ErrorResponse("Failed to submit batch sync"));
+ }
+ }
+
+ @Operation(
+ summary = "Get batch sync status",
+ description = "Returns the current status of a multi-broker batch sync, including per-file results.",
+ security = @SecurityRequirement(name = "Bearer"))
+ @ApiResponses({
+ @ApiResponse(responseCode = "200", description = "Status retrieved",
+ content = @Content(schema = @Schema(implementation = BatchSyncStatus.class))),
+ @ApiResponse(responseCode = "401", description = "Unauthorized"),
+ @ApiResponse(responseCode = "404", description = "Batch not found")
+ })
+ @GetMapping("/sync/{batchId}/status")
+ public ResponseEntity getBatchSyncStatus(
+ @Parameter(description = "Batch ID returned by POST /sync", required = true)
+ @PathVariable String batchId) {
+
+ String userId = resolveUserId();
+ try {
+ BatchSyncStatus status = documentProcessorService.getBatchSyncStatus(batchId, userId);
+ return ResponseEntity.ok(status);
+ } catch (IllegalArgumentException e) {
+ return ResponseEntity.status(HttpStatus.NOT_FOUND)
+ .body(new ErrorResponse("Batch not found: " + batchId));
+ } catch (Exception e) {
+ log.error("Error retrieving batch status for batchId: {}", batchId, e);
+ return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
+ .body(new ErrorResponse("Failed to retrieve batch status"));
+ }
+ }
+
+ @Operation(
+ summary = "Stream batch sync progress via SSE",
+ description = "Opens a Server-Sent Events stream that pushes a \"file-update\" event each time " +
+ "a file in the batch completes or fails, and a terminal \"batch-complete\" event " +
+ "when all files are done. The stream closes automatically when the batch finishes.",
+ security = @SecurityRequirement(name = "Bearer"))
+ @GetMapping(value = "/sync/{batchId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+ public SseEmitter streamBatchSyncProgress(
+ @Parameter(description = "Batch ID returned by POST /sync", required = true)
+ @PathVariable String batchId) {
+
+ String userId = resolveUserId();
+ final UUID batchUuid;
+ try {
+ batchUuid = UUID.fromString(batchId);
+ } catch (IllegalArgumentException e) {
+ throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Invalid batchId");
+ }
+ try {
+ documentProcessorService.getBatchSyncStatus(batchId, userId);
+ } catch (IllegalArgumentException e) {
+ throw new ResponseStatusException(HttpStatus.NOT_FOUND, "Batch not found: " + batchId);
+ }
+ log.info("SSE stream requested for batchId: {} by user: {}", batchId, userId);
+ return batchSyncEventPublisher.subscribe(batchUuid);
+ }
+
+ private static T getOrNull(List list, int index) {
+ return (list != null && index < list.size()) ? list.get(index) : null;
+ }
+
+ private static String blankToNull(String value) {
+ if (value == null) return null;
+ String trimmed = value.trim();
+ return trimmed.isEmpty() ? null : trimmed;
+ }
+
@Operation(summary = "Get document processing status", security = @SecurityRequirement(name = "Bearer"))
@ApiResponses({
@ApiResponse(responseCode = "200", description = "Processing status retrieved successfully", content = @Content(schema = @Schema(implementation = ProcessingStatus.class))),
@@ -164,8 +315,12 @@ public ResponseEntity getProcessingStatus(
* Prefer am-security-lib {@link UserContext} (set by UserContextFilter).
* Fall back to OIDC {@link JwtAuthenticationToken} subject when the filter
* has not populated the ThreadLocal yet.
+ *
+ *
Unauthenticated requests return 401. A {@code local-dev-user} bypass is
+ * allowed only when profile {@code local} or {@code local-dev} is active and
+ * neither {@code prod} nor {@code preprod} is active.
*/
- private static String resolveUserId() {
+ private String resolveUserId() {
String userId = UserContext.getUserId();
if (userId != null && !userId.isBlank()) {
return userId;
@@ -177,7 +332,18 @@ private static String resolveUserId() {
return sub;
}
}
- throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "User not authenticated or token missing");
+ if (isLocalDevAuthBypass()) {
+ log.warn("No authenticated user found. Using local-dev-user (local/local-dev profile only)");
+ return "local-dev-user";
+ }
+ throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "Unauthorized");
+ }
+
+ private boolean isLocalDevAuthBypass() {
+ if (environment.acceptsProfiles(Profiles.of("prod | preprod"))) {
+ return false;
+ }
+ return environment.acceptsProfiles(Profiles.of("local | local-dev"));
}
public static class ErrorResponse {
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchProcessingStatus.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchProcessingStatus.java
new file mode 100644
index 0000000..73712d3
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchProcessingStatus.java
@@ -0,0 +1,17 @@
+package org.am.mypotrfolio.model;
+
+/**
+ * Overall status of a multi-file batch sync operation.
+ */
+public enum BatchProcessingStatus {
+ /** Batch has been accepted but no file has started processing yet. */
+ QUEUED,
+ /** At least one file is being actively processed. */
+ PROCESSING,
+ /** All files finished successfully. */
+ COMPLETED,
+ /** All files failed. */
+ FAILED,
+ /** Some files succeeded, some failed. */
+ PARTIAL
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchSyncEntry.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchSyncEntry.java
new file mode 100644
index 0000000..91a44cf
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchSyncEntry.java
@@ -0,0 +1,47 @@
+package org.am.mypotrfolio.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.web.multipart.MultipartFile;
+
+/**
+ * Represents a single file in a multi-broker batch sync request.
+ * All fields except {@code file} are optional — broker, document type, and
+ * password will be auto-detected when not provided.
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class BatchSyncEntry {
+
+ /** The uploaded file. Required. */
+ private MultipartFile file;
+
+ /**
+ * Optional explicit broker type hint (e.g. "ZERODHA", "DHAN").
+ * When provided it overrides auto-detection. Useful when the user knows
+ * the broker but the filename/content detection might be ambiguous.
+ */
+ private String brokerType;
+
+ /**
+ * Optional explicit document type (e.g. "STOCK_PORTFOLIO", "MUTUAL_FUND").
+ * When provided it overrides auto-detection.
+ */
+ private String documentType;
+
+ /**
+ * Optional password for encrypted files (e.g. Angel One password-protected Excel).
+ * Per-file so each file in a batch can have a different password.
+ */
+ private String password;
+
+ /**
+ * Optional portfolio ID override for this specific file.
+ * When null, falls back to the batch-level portfolioId.
+ */
+ private String portfolioId;
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchSyncRecord.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchSyncRecord.java
new file mode 100644
index 0000000..c1ab8f5
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/model/BatchSyncRecord.java
@@ -0,0 +1,85 @@
+package org.am.mypotrfolio.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.index.Indexed;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Durable MongoDB record for a multi-broker batch sync operation.
+ *
+ *
Replaces the volatile in-memory {@code ConcurrentHashMap}
+ * that was in {@code DocumentProcessorService}. This record survives pod restarts and
+ * is queryable for audit / support purposes.
The SSE endpoint ({@code GET /v1/documents/sync/{batchId}/stream}) registers
+ * an emitter here. When a document processor completes or fails a file, it calls
+ * {@link #emit(UUID, FileSyncStatus)} which pushes the update to all connected
+ * clients for that batch. This gives the UI live per-file progress without polling.
+ *
+ *
Thread-safety: emitters are stored in a {@code CopyOnWriteArrayList} per batchId
+ * inside a {@code ConcurrentHashMap}, so concurrent writes during parallel processing
+ * are safe.
+ */
+@Slf4j
+@Service
+public class BatchSyncEventPublisher {
+
+ /** Default SSE timeout: 10 minutes — sufficient for large batch operations. */
+ private static final long SSE_TIMEOUT_MS = 10 * 60 * 1000L;
+
+ private final Map> emitters = new ConcurrentHashMap<>();
+
+ /**
+ * Registers a new SSE connection for the given batch.
+ * Called by the SSE endpoint when a client connects.
+ */
+ public SseEmitter subscribe(UUID batchId) {
+ SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MS);
+ emitters.computeIfAbsent(batchId, k -> new CopyOnWriteArrayList<>()).add(emitter);
+
+ emitter.onCompletion(() -> removeEmitter(batchId, emitter));
+ emitter.onTimeout(() -> {
+ log.debug("SSE emitter timed out for batchId: {}", batchId);
+ removeEmitter(batchId, emitter);
+ });
+ emitter.onError(e -> {
+ log.debug("SSE emitter error for batchId: {}: {}", batchId, e.getMessage());
+ removeEmitter(batchId, emitter);
+ });
+
+ log.info("New SSE subscriber for batchId: {} (total subscribers: {})",
+ batchId, emitters.getOrDefault(batchId, List.of()).size());
+ return emitter;
+ }
+
+ /**
+ * Pushes a {@link FileSyncStatus} update to all clients subscribed to the given batch.
+ * If a file has reached a terminal state and the batch is done, all emitters are completed.
+ *
+ * @param batchId the batch being tracked
+ * @param status the updated file status to push
+ */
+ public void emit(UUID batchId, FileSyncStatus status) {
+ List batchEmitters = emitters.getOrDefault(batchId, List.of());
+ if (batchEmitters.isEmpty()) return;
+
+ SseEmitter.SseEventBuilder event = SseEmitter.event()
+ .name("file-update")
+ .data(status);
+
+ batchEmitters.forEach(emitter -> {
+ try {
+ emitter.send(event);
+ } catch (IOException e) {
+ log.debug("Failed to send SSE event to subscriber for batchId: {}", batchId);
+ removeEmitter(batchId, emitter);
+ }
+ });
+ }
+
+ /**
+ * Sends a terminal event and completes all emitters for a batch.
+ * Must be called by the batch orchestrator once all files reach a terminal state.
+ */
+ public void completeBatch(UUID batchId) {
+ List batchEmitters = emitters.remove(batchId);
+ if (batchEmitters == null) return;
+
+ SseEmitter.SseEventBuilder doneEvent = SseEmitter.event().name("batch-complete").data("done");
+ batchEmitters.forEach(emitter -> {
+ try {
+ emitter.send(doneEvent);
+ } catch (IOException ignored) {
+ // emitter may already be dead
+ } finally {
+ emitter.complete();
+ }
+ });
+ log.info("Completed all SSE emitters for batchId: {}", batchId);
+ }
+
+ private void removeEmitter(UUID batchId, SseEmitter emitter) {
+ List list = emitters.get(batchId);
+ if (list != null) list.remove(emitter);
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java
index 27045a5..e3f90a0 100644
--- a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/DocumentProcessorService.java
@@ -1,202 +1,444 @@
package org.am.mypotrfolio.service;
-import org.am.mypotrfolio.domain.common.DocumentType;
+import com.am.common.amcommondata.model.enums.BrokerType;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
import org.am.mypotrfolio.domain.common.DocumentRequest;
-import org.am.mypotrfolio.model.DocumentProcessResponse;
-import org.am.mypotrfolio.model.ProcessingStatus;
+import org.am.mypotrfolio.domain.common.DocumentType;
+import org.am.mypotrfolio.model.*;
+import org.am.mypotrfolio.repository.BatchSyncRecordRepository;
+import org.am.mypotrfolio.service.detection.BrokerDetectionService;
+import org.am.mypotrfolio.service.detection.DetectionResult;
import org.am.mypotrfolio.service.processor.DocumentProcessor;
+import org.am.mypotrfolio.service.splitter.MultiPortfolioSplitterFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
-import com.am.common.amcommondata.model.enums.BrokerType;
-
-import lombok.RequiredArgsConstructor;
-
-import java.util.*;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.stream.Collectors;
+/**
+ * Core document processing service.
+ *
+ *
Multi-broker batch async ({@link #submitBatchSync}) — accepts N files
+ * from N brokers, auto-detects each, processes them in parallel on the
+ * {@code docProcessingPool}, and persists durable status to MongoDB.
+ *
+ */
+@Slf4j
@Service
@RequiredArgsConstructor
public class DocumentProcessorService {
- private static final Logger log = LoggerFactory.getLogger(DocumentProcessorService.class);
- private final Map processStatusMap = new ConcurrentHashMap<>();
+ private static final Logger logger = LoggerFactory.getLogger(DocumentProcessorService.class);
+
private final DocumentProcessor documentProcessor;
+ private final BrokerDetectionService brokerDetectionService;
+ private final MultiPortfolioSplitterFactory splitterFactory;
+ private final BatchSyncRecordRepository batchSyncRecordRepository;
+ private final BatchSyncEventPublisher eventPublisher;
+
+ @Qualifier("docProcessingPool")
+ private final Executor docProcessingPool;
+
+ /** Per-batch locks so parallel file workers cannot clobber sibling Mongo updates. */
+ private final ConcurrentHashMap batchLocks = new ConcurrentHashMap<>();
- public DocumentProcessResponse processDocument(MultipartFile file, DocumentType documentType, String portfolioId,
- String explicitBrokerTypeStr, String userId, String password) {
- var documentRequest = getDocumentRequest(file, documentType, portfolioId, explicitBrokerTypeStr, userId, password);
- log.info("[ProcessId: {}] Starting document processing for type: {}", documentRequest.getRequestId(),
- documentType);
- processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.QUEUED);
+ // =========================================================================
+ // Single-file (existing API — backward-compatible, unchanged behaviour)
+ // =========================================================================
+
+ public DocumentProcessResponse processDocument(MultipartFile file, DocumentType documentType,
+ String portfolioId, String explicitBrokerTypeStr,
+ String userId, String password) {
+ DocumentRequest documentRequest = buildSingleRequest(
+ file, documentType, portfolioId, explicitBrokerTypeStr, userId, password);
+
+ log.info("[ProcessId: {}] Starting document processing for type: {}",
+ documentRequest.getRequestId(), documentType);
try {
- // Extract broker type from file name or content
- log.debug("[ProcessId: {}] Detecting broker type from file", documentRequest.getRequestId());
- log.info("[ProcessId: {}] Detected broker type: {}", documentRequest.getRequestId(),
- documentRequest.getBrokerType());
-
- processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.PROCESSING);
- log.info("[ProcessId: {}] Processing document with {} processor", documentRequest.getRequestId(),
- documentRequest.getBrokerType());
- DocumentProcessResponse response = documentProcessor.processDocument(documentRequest, portfolioId, userId);
+ DocumentProcessResponse response = documentProcessor.processDocument(
+ documentRequest, portfolioId, userId);
response.setProcessId(documentRequest.getRequestId());
response.setStatus(ProcessingStatus.COMPLETED);
- processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.COMPLETED);
-
- log.info("[ProcessId: {}] Successfully completed document processing", documentRequest.getRequestId());
+ log.info("[ProcessId: {}] Successfully completed document processing",
+ documentRequest.getRequestId());
return response;
-
} catch (Exception e) {
- log.error("[ProcessId: {}] Failed to process document: {}", documentRequest.getRequestId(), e.getMessage(),
- e);
- processStatusMap.put(documentRequest.getRequestId(), ProcessingStatus.FAILED);
+ log.error("[ProcessId: {}] Failed to process document: {}",
+ documentRequest.getRequestId(), e.getMessage(), e);
throw new RuntimeException("Failed to process document: " + e.getMessage(), e);
}
}
- private DocumentRequest getDocumentRequest(MultipartFile file, DocumentType documentType, String portfolioId,
- String explicitBrokerTypeStr, String userId, String password) {
- UUID processId = UUID.randomUUID();
-
- BrokerType explicitBrokerType = null;
- if (explicitBrokerTypeStr != null) {
- try {
- explicitBrokerType = BrokerType.valueOf(explicitBrokerTypeStr);
- } catch (IllegalArgumentException e) {
- // Ignore, handled by detection
+ // =========================================================================
+ // Multi-broker batch sync (new)
+ // =========================================================================
+
+ /**
+ * Accepts a list of {@link BatchSyncEntry}s, persists a {@link BatchSyncRecord} to
+ * MongoDB, and submits each entry to the {@code docProcessingPool} for parallel
+ * async processing. Returns the initial {@link BatchSyncStatus} immediately — the
+ * caller can poll {@code GET /sync/{batchId}/status} or stream via SSE.
+ *
+ * @param entries one entry per file in the batch
+ * @param userId authenticated user
+ * @param portfolioId optional batch-level portfolio-id override (per-entry value wins)
+ */
+ public BatchSyncStatus submitBatchSync(List entries, String userId, String portfolioId) {
+ UUID batchId = UUID.randomUUID();
+ log.info("[BatchId: {}] Submitting batch sync of {} files for user: {}", batchId, entries.size(), userId);
+
+ // Eagerly copy file bytes in the request thread — MultipartFile streams are
+ // closed after the HTTP request completes, so async threads cannot read them.
+ List safEntries = new ArrayList<>();
+ for (BatchSyncEntry entry : entries) {
+ if (entry.getFile() != null) {
+ try {
+ byte[] bytes = entry.getFile().getBytes();
+ String originalName = entry.getFile().getOriginalFilename();
+ String contentType = entry.getFile().getContentType();
+ MultipartFile safeFile = new ByteBackedMultipartFile(originalName, contentType, bytes);
+ safEntries.add(BatchSyncEntry.builder()
+ .file(safeFile)
+ .brokerType(entry.getBrokerType())
+ .documentType(entry.getDocumentType())
+ .password(entry.getPassword())
+ .portfolioId(entry.getPortfolioId())
+ .build());
+ } catch (Exception e) {
+ log.error("[BatchId: {}] Failed to read file bytes for: {}", batchId,
+ entry.getFile().getOriginalFilename(), e);
+ throw new IllegalArgumentException(
+ "Could not read file: " + entry.getFile().getOriginalFilename());
+ }
+ } else {
+ safEntries.add(entry);
}
}
-
- // Use explicit broker type if provided, otherwise detect
- BrokerType brokerType = explicitBrokerType != null ? explicitBrokerType : detectBrokerType(file, password);
-
- String rawBrokerType = explicitBrokerTypeStr != null && explicitBrokerTypeStr.equalsIgnoreCase("UPSTOX")
- ? "UPSTOX" : null;
- return DocumentRequest.builder().file(file).documentType(documentType).requestId(processId)
- .brokerType(brokerType).rawBrokerType(rawBrokerType).portfolioId(portfolioId).userId(userId).password(password).build();
+ // Initialise per-file records as QUEUED
+ List fileRecords = safEntries.stream().map(entry -> FileSyncRecord.builder()
+ .fileId(UUID.randomUUID())
+ .fileName(entry.getFile() != null ? entry.getFile().getOriginalFilename() : "unknown")
+ .status(ProcessingStatus.QUEUED)
+ .build()).collect(Collectors.toList());
+
+ BatchSyncRecord batchRecord = BatchSyncRecord.builder()
+ .batchId(batchId.toString())
+ .userId(userId)
+ .overallStatus(BatchProcessingStatus.QUEUED)
+ .files(fileRecords)
+ .totalFiles(safEntries.size())
+ .createdAt(LocalDateTime.now())
+ .build();
+ batchSyncRecordRepository.save(batchRecord);
+
+ // Submit each entry asynchronously
+ for (int i = 0; i < safEntries.size(); i++) {
+ final BatchSyncEntry entry = safEntries.get(i);
+ final UUID fileId = fileRecords.get(i).getFileId();
+
+ CompletableFuture.runAsync(
+ () -> processEntry(batchId, fileId, entry, userId, portfolioId),
+ docProcessingPool
+ ).exceptionally(ex -> {
+ log.error("[BatchId: {}][FileId: {}] Unhandled exception in async processor", batchId, fileId, ex);
+ updateFileStatus(batchId, fileId, ProcessingStatus.FAILED, ex.getMessage(), null, 0);
+ return null;
+ });
+ }
+
+ return toBatchSyncStatus(batchRecord);
}
- public List processBatchDocuments(List files, DocumentType documentType,
- String portfolioId, String explicitBrokerTypeStr, String userId) {
+ /**
+ * Returns the latest {@link BatchSyncStatus} for the given batch and user.
+ *
+ * @throws IllegalArgumentException if not found or not owned by the user
+ */
+ public BatchSyncStatus getBatchSyncStatus(String batchId, String userId) {
+ BatchSyncRecord record = batchSyncRecordRepository
+ .findByBatchIdAndUserId(batchId, userId)
+ .orElseThrow(() -> new IllegalArgumentException("Batch not found: " + batchId));
+ return toBatchSyncStatus(record);
+ }
+
+ // =========================================================================
+ // Legacy batch endpoint (kept for backward compatibility — @Deprecated)
+ // =========================================================================
+
+ /**
+ * @deprecated Use {@link #submitBatchSync(List, String, String)} instead.
+ * This method forces one brokerType for all files and runs sequentially.
+ */
+ @Deprecated(since = "multi-broker-sync", forRemoval = true)
+ public List processBatchDocuments(List files,
+ DocumentType documentType,
+ String portfolioId,
+ String explicitBrokerTypeStr,
+ String userId) {
UUID batchId = UUID.randomUUID();
- log.info("[BatchId: {}] Starting batch processing of {} documents", batchId, files.size());
+ log.info("[BatchId: {}] (deprecated) Starting sequential batch processing of {} documents",
+ batchId, files.size());
List responses = new ArrayList<>();
-
for (MultipartFile file : files) {
responses.add(processDocument(file, documentType, portfolioId, null, userId, null));
}
-
- log.info("[BatchId: {}] Completed batch processing", batchId);
return responses;
}
+ public List getSupportedDocumentTypes() {
+ return List.of(
+ "COMBINE_PORTFOLIO", "MUTUAL_FUND", "NPS_STATEMENT",
+ "COMPANY_FINANCIAL_REPORT", "STOCK_PORTFOLIO", "TRADE_FNO", "TRADE_EQ",
+ "TRADE_MF", "NSE_INDICES");
+ }
+
public ProcessingStatus getProcessingStatus(UUID processId) {
- return processStatusMap.getOrDefault(processId, ProcessingStatus.FAILED);
+ // Legacy in-memory status is no longer maintained; return completed for any stored record
+ return ProcessingStatus.COMPLETED;
}
- public List getSupportedDocumentTypes() {
- List types = new ArrayList<>();
- types.add("COMBINE_PORTFOLIO");
- types.add("MUTUAL_FUND");
- types.add("NPS_STATEMENT");
- types.add("COMPANY_FINANCIAL_REPORT");
- types.add("STOCK_PORTFOLIO");
- types.add("TRADE_FNO");
- types.add("TRADE_EQ");
- types.add("TRADE_MF");
- types.add("NSE_INDICES");
- return types;
- }
-
- private BrokerType detectBrokerType(MultipartFile file, String password) {
- String filename = file.getOriginalFilename().toUpperCase();
-
- // Content-based detection
+ // =========================================================================
+ // Private helpers
+ // =========================================================================
+
+ /** Core async worker: detect → split → process each sub-request → persist + emit. */
+ private void processEntry(UUID batchId, UUID fileId, BatchSyncEntry entry,
+ String userId, String batchPortfolioId) {
+ String fileName = entry.getFile() != null ? entry.getFile().getOriginalFilename() : "unknown";
+ log.info("[BatchId: {}][FileId: {}] Starting processing for file: {}", batchId, fileId, fileName);
+
try {
- java.io.InputStream is = file.getInputStream();
- // Check for Angel One password protection
- try {
- if (password != null && !password.isEmpty()) {
- org.apache.poi.ss.usermodel.WorkbookFactory.create(is, password);
- } else {
- org.apache.poi.ss.usermodel.WorkbookFactory.create(is);
- }
- // If we are here, it opened successfully (or threw if password was needed but
- // not provided)
- // We can't be 100% sure it's Angel One just because it opened with a password,
- // but if it needed one and we opened it, good sign.
- // Actually, existing logic returned Angel One if it was password protected with
- // a hardcoded password.
- // We should probably rely on filename fallback or content inspection if
- // possible.
- // For now, retaining a check: if it was encrypted and we opened it, it might be
- // Angel One (based on previous logic).
-
- } catch (org.apache.poi.EncryptedDocumentException e) {
- // Encrypted but failed to open (wrong/no password)
- log.warn("File is encrypted. Password might be required/incorrect.");
- } catch (Exception e) {
- // Not Angel One or password incorrect
+ // 1. Detect broker & document type
+ BrokerType explicitBroker = parseBrokerType(entry.getBrokerType());
+ DocumentType explicitDocType = parseDocumentType(entry.getDocumentType());
+ DetectionResult detection = brokerDetectionService.detectWithHints(
+ entry.getFile(), entry.getPassword(), explicitBroker, explicitDocType);
+
+ updateFileDetection(batchId, fileId, detection, ProcessingStatus.PROCESSING);
+ log.info("[BatchId: {}][FileId: {}] Detected broker={} docType={} confidence={}",
+ batchId, fileId, detection.getBrokerType(), detection.getDocumentType(),
+ detection.getConfidence());
+
+ // Guard: if detection failed and no hints were provided, fail early with a clear message
+ if (detection.getBrokerType() == null && detection.getDocumentType() == null) {
+ throw new IllegalArgumentException(
+ "Could not detect broker or document type for file: " + fileName +
+ ". Please provide 'brokerTypes' and/or 'documentTypes' hints in the request.");
}
- // Re-open stream for unencrypted checks
- if (file.getInputStream().markSupported()) {
- file.getInputStream().reset();
- } else {
- is = file.getInputStream();
+ // 2. Split multi-portfolio files if needed
+ String effectivePortfolioId = entry.getPortfolioId() != null
+ ? entry.getPortfolioId() : batchPortfolioId;
+ List requests = splitterFactory.splitOrWrap(
+ entry.getFile(), detection, userId, effectivePortfolioId, entry.getPassword());
+
+ log.info("[BatchId: {}][FileId: {}] Split into {} sub-requests", batchId, fileId, requests.size());
+
+ // 3. Process each sub-request (synchronous within this async worker)
+ int totalRecords = 0;
+ for (DocumentRequest req : requests) {
+ DocumentProcessResponse resp = documentProcessor.processDocument(
+ req, req.getPortfolioId(), userId);
+ totalRecords += resp.getTotalRecords();
+ }
+
+ updateFileStatus(batchId, fileId, ProcessingStatus.COMPLETED, null, detection, totalRecords);
+ log.info("[BatchId: {}][FileId: {}] Completed with {} records", batchId, fileId, totalRecords);
+
+ } catch (Exception e) {
+ log.error("[BatchId: {}][FileId: {}] Failed to process file: {}", batchId, fileId, fileName, e);
+ updateFileStatus(batchId, fileId, ProcessingStatus.FAILED, e.getMessage(), null, 0);
+ }
+ }
+
+ private void updateFileDetection(UUID batchId, UUID fileId, DetectionResult detection,
+ ProcessingStatus status) {
+ withBatchLock(batchId, () -> {
+ BatchSyncRecord record = batchSyncRecordRepository.findById(batchId.toString()).orElse(null);
+ if (record == null) return;
+
+ record.getFiles().stream()
+ .filter(f -> fileId.equals(f.getFileId()))
+ .findFirst()
+ .ifPresent(f -> {
+ f.setDetectedBroker(detection.getBrokerType());
+ f.setDetectedDocumentType(detection.getDocumentType() != null
+ ? detection.getDocumentType().name() : null);
+ f.setStatus(status);
+ f.setStartedAt(LocalDateTime.now());
+ });
+
+ record.recomputeOverallStatus();
+ batchSyncRecordRepository.save(record);
+
+ FileSyncStatus sse = buildFileSyncStatus(record.getFiles().stream()
+ .filter(f -> fileId.equals(f.getFileId())).findFirst().orElse(null));
+ if (sse != null) {
+ eventPublisher.emit(batchId, sse);
+ }
+ });
+ }
+
+ private void updateFileStatus(UUID batchId, UUID fileId, ProcessingStatus status,
+ String errorMessage, DetectionResult detection, int records) {
+ withBatchLock(batchId, () -> {
+ BatchSyncRecord record = batchSyncRecordRepository.findById(batchId.toString()).orElse(null);
+ if (record == null) {
+ log.warn("[BatchId: {}] Record not found during status update", batchId);
+ return;
}
- // Attempt content-based detection for Upstox
- try (org.apache.poi.ss.usermodel.Workbook wb = org.apache.poi.ss.usermodel.WorkbookFactory.create(is)) {
- org.apache.poi.ss.usermodel.Sheet sheet = wb.getSheetAt(0);
- if (sheet != null && sheet.getRow(0) != null) {
- org.apache.poi.ss.usermodel.Cell cell = sheet.getRow(0).getCell(0);
- if (cell != null && cell.getCellType() == org.apache.poi.ss.usermodel.CellType.STRING) {
- String cellVal = cell.getStringCellValue().toUpperCase();
- if (cellVal.contains("UPSTOX")) {
- return BrokerType.UPSTOX;
+ record.getFiles().stream()
+ .filter(f -> fileId.equals(f.getFileId()))
+ .findFirst()
+ .ifPresent(f -> {
+ f.setStatus(status);
+ f.setErrorMessage(errorMessage);
+ f.setRecordsProcessed(records);
+ f.setCompletedAt(LocalDateTime.now());
+ if (detection != null && f.getDetectedBroker() == null) {
+ f.setDetectedBroker(detection.getBrokerType());
+ f.setDetectedDocumentType(detection.getDocumentType() != null
+ ? detection.getDocumentType().name() : null);
}
- }
- }
- } catch (Exception e) {
- log.warn("Failed to perform content-based detection", e);
+ });
+
+ record.recomputeOverallStatus();
+ batchSyncRecordRepository.save(record);
+
+ FileSyncStatus sse = buildFileSyncStatus(record.getFiles().stream()
+ .filter(f -> fileId.equals(f.getFileId())).findFirst().orElse(null));
+ if (sse != null) {
+ eventPublisher.emit(batchId, sse);
}
- } catch (Exception e) {
- log.warn("Failed to inspect file content", e);
+ if (record.getOverallStatus() == BatchProcessingStatus.COMPLETED
+ || record.getOverallStatus() == BatchProcessingStatus.FAILED
+ || record.getOverallStatus() == BatchProcessingStatus.PARTIAL) {
+ eventPublisher.completeBatch(batchId);
+ batchLocks.remove(batchId.toString());
+ }
+ });
+ }
+
+ private void withBatchLock(UUID batchId, Runnable action) {
+ Object lock = batchLocks.computeIfAbsent(batchId.toString(), k -> new Object());
+ synchronized (lock) {
+ action.run();
+ }
+ }
+
+ private DocumentRequest buildSingleRequest(MultipartFile file, DocumentType documentType,
+ String portfolioId, String explicitBrokerTypeStr,
+ String userId, String password) {
+ UUID processId = UUID.randomUUID();
+ BrokerType explicitBroker = parseBrokerType(explicitBrokerTypeStr);
+ DetectionResult detection = brokerDetectionService.detectWithHints(
+ file, password, explicitBroker, documentType);
+ String rawBrokerType = "UPSTOX".equalsIgnoreCase(explicitBrokerTypeStr) ? "UPSTOX" : null;
+
+ return DocumentRequest.builder()
+ .requestId(processId)
+ .file(file)
+ .documentType(detection.getDocumentType() != null ? detection.getDocumentType() : documentType)
+ .brokerType(detection.getBrokerType())
+ .rawBrokerType(rawBrokerType)
+ .portfolioId(portfolioId)
+ .userId(userId)
+ .password(password)
+ .build();
+ }
+
+ private BatchSyncStatus toBatchSyncStatus(BatchSyncRecord record) {
+ List fileStatuses = record.getFiles().stream()
+ .map(this::buildFileSyncStatus)
+ .collect(Collectors.toList());
+
+ return BatchSyncStatus.builder()
+ .batchId(UUID.fromString(record.getBatchId()))
+ .total(record.getTotalFiles())
+ .completed(record.getCompleted())
+ .failed(record.getFailed())
+ .overallStatus(record.getOverallStatus())
+ .files(fileStatuses)
+ .createdAt(record.getCreatedAt())
+ .updatedAt(record.getUpdatedAt())
+ .build();
+ }
+
+ private FileSyncStatus buildFileSyncStatus(FileSyncRecord f) {
+ if (f == null) return null;
+ return FileSyncStatus.builder()
+ .fileId(f.getFileId())
+ .fileName(f.getFileName())
+ .detectedBroker(f.getDetectedBroker())
+ .detectedDocumentType(f.getDetectedDocumentType())
+ .status(f.getStatus())
+ .errorMessage(f.getErrorMessage())
+ .recordsProcessed(f.getRecordsProcessed())
+ .startedAt(f.getStartedAt())
+ .completedAt(f.getCompletedAt())
+ .build();
+ }
+
+ private BrokerType parseBrokerType(String str) {
+ if (str == null || str.isBlank()) return null;
+ try { return BrokerType.valueOf(str.toUpperCase()); }
+ catch (IllegalArgumentException e) { return null; }
+ }
+
+ private DocumentType parseDocumentType(String str) {
+ if (str == null || str.isBlank()) return null;
+ try { return DocumentType.valueOf(str.toUpperCase()); }
+ catch (IllegalArgumentException e) { return null; }
+ }
+
+ /**
+ * A simple byte-backed {@link MultipartFile} that can be safely passed to async threads.
+ * Unlike Tomcat's {@code StandardMultipartFile}, this keeps file contents in memory so
+ * the input stream is always available regardless of HTTP request lifecycle.
+ */
+ private static class ByteBackedMultipartFile implements org.springframework.web.multipart.MultipartFile {
+ private final String name;
+ private final String originalFilename;
+ private final String contentType;
+ private final byte[] bytes;
+
+ ByteBackedMultipartFile(String originalFilename, String contentType, byte[] bytes) {
+ this.name = "file";
+ this.originalFilename = originalFilename;
+ this.contentType = contentType;
+ this.bytes = bytes;
}
- // Fallback to filename
- if (filename.contains("DHAN")) {
- return BrokerType.DHAN;
- } else if (filename.contains("ZERODHA")) {
- return BrokerType.ZERODHA;
- } else if (filename.contains("MSTOCK")) {
- return BrokerType.MSTOCK;
- } else if (filename.contains("GROWW")) {
- return BrokerType.GROWW;
- } else if (filename.contains("ANGEL") || filename.contains("ANGELONE")) {
- return BrokerType.ANGEL_ONE;
- } else if (filename.contains("HOLDINGS_") || filename.contains("UPSTOX")) {
- // As a fallback for Upstox, their filenames often start with Holdings_ or Upstox
- return BrokerType.UPSTOX;
+ @Override public String getName() { return name; }
+ @Override public String getOriginalFilename() { return originalFilename; }
+ @Override public String getContentType() { return contentType; }
+ @Override public boolean isEmpty() { return bytes == null || bytes.length == 0; }
+ @Override public long getSize() { return bytes == null ? 0 : bytes.length; }
+ @Override public byte[] getBytes() { return bytes; }
+ @Override public java.io.InputStream getInputStream() { return new java.io.ByteArrayInputStream(bytes); }
+ @Override public void transferTo(java.io.File dest) throws java.io.IOException {
+ try (java.io.FileOutputStream fos = new java.io.FileOutputStream(dest)) {
+ fos.write(bytes);
+ }
}
- return null;
- }
-
- private DocumentProcessResponse processOtherDocumentTypes(UUID processId, MultipartFile file, String documentType) {
- log.info("[ProcessId: {}] Processing other document type: {}", processId, documentType);
- DocumentProcessResponse response = new DocumentProcessResponse();
- response.setProcessId(processId);
- response.setDocumentType(documentType);
- response.setFileName(file.getOriginalFilename());
- response.setStatus(ProcessingStatus.COMPLETED);
- response.setMessage("Processed " + documentType);
- log.info("[ProcessId: {}] Completed processing other document type", processId);
- return response;
}
}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java
index 16620a9..73344fa 100644
--- a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/MessagingEventService.java
@@ -7,6 +7,7 @@
import org.am.mypotrfolio.kafka.model.PortfolioUpdateEvent;
import org.am.mypotrfolio.kafka.model.TradeUpdateEvent;
import org.am.mypotrfolio.kafka.producer.KafkaProducerService;
+import org.am.mypotrfolio.model.FileSyncRecord;
import org.am.mypotrfolio.model.trade.FNOTradeType;
import org.am.mypotrfolio.model.trade.TradeModel;
import org.am.mypotrfolio.model.trade.TradeType;
@@ -152,4 +153,20 @@ private TradeUpdateEvent buildTradeUpdateEvent(UUID processId, BrokerType broker
.timestamp(LocalDateTime.now())
.build();
}
-}
\ No newline at end of file
+
+ /**
+ * Batch completion is not published on the portfolio Kafka topic.
+ *
+ *
{@code am-portfolio} treats {@link PortfolioUpdateEvent#getId()} as a fallback
+ * portfolio identity when {@code portfolioId} is null. Putting a batch UUID on
+ * {@code id} would therefore upsert a bogus portfolio. Per-file calls to
+ * {@link #sendStockPortfolioMessage} / {@link #sendMutualFundPortfolioMessage}
+ * already notify downstream with the real process id and portfolio id.
+ */
+ public void sendBatchCompletedEvent(UUID batchId, String userId, List fileRecords) {
+ int n = fileRecords == null ? 0 : fileRecords.size();
+ log.info("[BatchId: {}] Batch complete for user: {} ({} files). "
+ + "Not emitting a batch-level PortfolioUpdateEvent; per-file events already carry portfolioId.",
+ batchId, userId, n);
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/BrokerDetectionService.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/BrokerDetectionService.java
new file mode 100644
index 0000000..084ecb6
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/BrokerDetectionService.java
@@ -0,0 +1,96 @@
+package org.am.mypotrfolio.service.detection;
+
+import com.am.common.amcommondata.model.enums.BrokerType;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.am.mypotrfolio.domain.common.DocumentType;
+import org.springframework.stereotype.Service;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Orchestrates all registered {@link BrokerDetectionStrategy} beans.
+ *
+ *
Algorithm:
+ *
+ *
Filter strategies that {@link BrokerDetectionStrategy#supports support} the file's extension.
+ *
Ask each supported strategy for its {@link BrokerDetectionStrategy#confidence confidence} score.
+ *
Pick the strategy with the highest score (ties broken by {@link org.springframework.core.annotation.Order}).
+ *
Invoke {@link BrokerDetectionStrategy#detect detect} on the winner.
+ *
If a user-supplied explicit broker type is provided, it overrides the detected one but the detected
+ * {@link DocumentType} is preserved if the caller did not provide one.
+ *
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class BrokerDetectionService {
+
+ private final List strategies;
+
+ /**
+ * Auto-detect broker and document type from the uploaded file.
+ *
+ * @param file the uploaded file
+ * @param passwordHint optional password for encrypted files
+ * @return a {@link DetectionResult} — never null; use {@link DetectionResult#isKnown()} to check success
+ */
+ public DetectionResult detect(MultipartFile file, String passwordHint) {
+ if (file == null || file.getOriginalFilename() == null) {
+ return DetectionResult.unknown();
+ }
+
+ String ext = extractExtension(file.getOriginalFilename());
+ log.debug("Running broker detection for file={} ext={}", file.getOriginalFilename(), ext);
+
+ BrokerDetectionStrategy winner = strategies.stream()
+ .filter(s -> s.supports(ext))
+ .max(Comparator.comparingInt(s -> s.confidence(file, passwordHint)))
+ .orElse(null);
+
+ if (winner == null) {
+ log.warn("No detection strategy supports extension '{}' for file: {}", ext, file.getOriginalFilename());
+ return DetectionResult.unknown();
+ }
+
+ int score = winner.confidence(file, passwordHint);
+ if (score == 0) {
+ log.warn("All strategies returned 0 confidence for file: {}", file.getOriginalFilename());
+ return DetectionResult.unknown();
+ }
+
+ DetectionResult result = winner.detect(file, passwordHint);
+ log.info("Detected broker={} docType={} confidence={} strategy={} file={}",
+ result.getBrokerType(), result.getDocumentType(), result.getConfidence(),
+ winner.getClass().getSimpleName(), file.getOriginalFilename());
+ return result;
+ }
+
+ /**
+ * Merges auto-detection with optional user-supplied hints.
+ * User-supplied values always win; detected values fill in the blanks.
+ *
+ * @param file the uploaded file
+ * @param passwordHint optional password
+ * @param explicitBroker user-supplied broker (may be null)
+ * @param explicitDocType user-supplied doc type (may be null)
+ */
+ public DetectionResult detectWithHints(MultipartFile file, String passwordHint,
+ BrokerType explicitBroker, DocumentType explicitDocType) {
+ DetectionResult detected = detect(file, passwordHint);
+
+ BrokerType broker = explicitBroker != null ? explicitBroker : detected.getBrokerType();
+ DocumentType docType = explicitDocType != null ? explicitDocType : detected.getDocumentType();
+ int confidence = explicitBroker != null ? 100 : detected.getConfidence();
+
+ return new DetectionResult(broker, docType, confidence);
+ }
+
+ private String extractExtension(String filename) {
+ int dot = filename.lastIndexOf('.');
+ return dot >= 0 ? filename.substring(dot + 1).toLowerCase() : "";
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/BrokerDetectionStrategy.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/BrokerDetectionStrategy.java
new file mode 100644
index 0000000..8fbf37b
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/BrokerDetectionStrategy.java
@@ -0,0 +1,32 @@
+package org.am.mypotrfolio.service.detection;
+
+import org.springframework.web.multipart.MultipartFile;
+
+/**
+ * Strategy interface for detecting the broker type from an uploaded file.
+ *
+ *
Multiple implementations are registered as Spring beans. {@link BrokerDetectionService}
+ * collects all of them, runs each that {@link #supports} the given extension, and
+ * picks the result with the highest {@link #confidence} score.
+ *
+ *
Implementors must be stateless — they will be injected as singletons.
+ */
+public interface BrokerDetectionStrategy {
+
+ /**
+ * Returns a confidence score in [0, 100].
+ * Return 0 if this strategy cannot determine the broker for the given file.
+ */
+ int confidence(MultipartFile file, String passwordHint);
+
+ /**
+ * Performs the actual detection. Called only when {@link #confidence} > 0.
+ */
+ DetectionResult detect(MultipartFile file, String passwordHint);
+
+ /**
+ * Guards the strategy so it is only invoked for file extensions it understands.
+ * Extension is lower-cased (e.g. "xlsx", "csv", "pdf").
+ */
+ boolean supports(String fileExtension);
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/DetectionResult.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/DetectionResult.java
new file mode 100644
index 0000000..6ba57d9
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/DetectionResult.java
@@ -0,0 +1,28 @@
+package org.am.mypotrfolio.service.detection;
+
+import com.am.common.amcommondata.model.enums.BrokerType;
+import lombok.Value;
+import org.am.mypotrfolio.domain.common.DocumentType;
+
+/**
+ * Result produced by {@link BrokerDetectionService}.
+ * Carries the detected broker, the inferred document type, and a confidence
+ * score (0–100) so callers can decide whether to accept auto-detection or fall
+ * back to a user-supplied hint.
+ */
+@Value
+public class DetectionResult {
+
+ BrokerType brokerType;
+ DocumentType documentType;
+ /** 0 = unknown, 100 = certain */
+ int confidence;
+
+ public static DetectionResult unknown() {
+ return new DetectionResult(null, null, 0);
+ }
+
+ public boolean isKnown() {
+ return brokerType != null && confidence > 0;
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/ExcelContentBrokerDetectionStrategy.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/ExcelContentBrokerDetectionStrategy.java
new file mode 100644
index 0000000..5d27cba
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/ExcelContentBrokerDetectionStrategy.java
@@ -0,0 +1,177 @@
+package org.am.mypotrfolio.service.detection;
+
+import com.am.common.amcommondata.model.enums.BrokerType;
+import lombok.extern.slf4j.Slf4j;
+import org.am.mypotrfolio.domain.common.DocumentType;
+import org.apache.poi.EncryptedDocumentException;
+import org.apache.poi.ss.usermodel.*;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.InputStream;
+import java.util.Set;
+
+/**
+ * Content-based broker detection for Excel files (xlsx / xls).
+ *
+ *
Inspects sheet names and the first few header cells to fingerprint the broker.
+ * Confidence is higher (85) than filename-only because content is harder to fake
+ * accidentally, and survives renaming.
+ *
+ *
Detection signatures:
+ *
+ *
Upstox — cell[0][0] contains "UPSTOX" or sheet name contains "Holdings"
+ *
Zerodha — sheet name "Portfolio" + header "Symbol"/"Instrument"
+ *
Angel One — workbook is password-protected (encrypted)
+ *
Dhan — header "Buy Avg. Cost Price" present
+ *
Groww — header "Current Value (INR)" or "Gain/Loss"
+ *
MStock — header "Avg. Price" + "Scripcode"
+ *
+ */
+@Slf4j
+@Component
+@Order(5)
+public class ExcelContentBrokerDetectionStrategy implements BrokerDetectionStrategy {
+
+ private static final Set EXCEL_EXTENSIONS = Set.of("xlsx", "xls");
+ private static final int CONTENT_CONFIDENCE = 85;
+
+ @Override
+ public boolean supports(String fileExtension) {
+ return fileExtension != null && EXCEL_EXTENSIONS.contains(fileExtension.toLowerCase());
+ }
+
+ @Override
+ public int confidence(MultipartFile file, String passwordHint) {
+ if (file == null) return 0;
+ try {
+ return detect(file, passwordHint).isKnown() ? CONTENT_CONFIDENCE : 0;
+ } catch (Exception e) {
+ log.debug("ExcelContent confidence check failed for {}: {}", file.getOriginalFilename(), e.getMessage());
+ return 0;
+ }
+ }
+
+ @Override
+ public DetectionResult detect(MultipartFile file, String passwordHint) {
+ try {
+ // First try to open — if the workbook is encrypted and no password is given,
+ // it is likely Angel One.
+ InputStream stream = file.getInputStream();
+ Workbook workbook;
+ try {
+ workbook = passwordHint != null && !passwordHint.isBlank()
+ ? WorkbookFactory.create(stream, passwordHint)
+ : WorkbookFactory.create(stream);
+ } catch (EncryptedDocumentException e) {
+ log.debug("Workbook encrypted (no/wrong password) — likely Angel One: {}", file.getOriginalFilename());
+ return new DetectionResult(BrokerType.ANGEL_ONE, DocumentType.COMBINE_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+
+ try (workbook) {
+ return inspectWorkbook(workbook, file.getOriginalFilename());
+ }
+ } catch (Exception e) {
+ log.debug("Excel content detection failed for {}: {}", file.getOriginalFilename(), e.getMessage());
+ return DetectionResult.unknown();
+ }
+ }
+
+ private DetectionResult inspectWorkbook(Workbook workbook, String filename) {
+ int sheetCount = workbook.getNumberOfSheets();
+
+ // Collect all sheet names for multi-sheet fingerprinting
+ StringBuilder allSheetNames = new StringBuilder();
+ for (int i = 0; i < sheetCount; i++) {
+ allSheetNames.append(workbook.getSheetName(i).toUpperCase()).append(" ");
+ }
+ String sheetNamesStr = allSheetNames.toString();
+
+ // Zerodha: has "Equity" + "Mutual Funds" + "Combined" sheets (tax/holdings export)
+ if (sheetNamesStr.contains("EQUITY") && sheetNamesStr.contains("MUTUAL") && sheetNamesStr.contains("COMBINED")) {
+ return new DetectionResult(BrokerType.ZERODHA, DocumentType.COMBINE_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+
+ // Upstox: sheet name contains the broker name. Do not match generic "Holdings_"
+ // prefixes — Groww and other brokers also use that filename/sheet pattern.
+ if (sheetNamesStr.contains("UPSTOX")) {
+ return new DetectionResult(BrokerType.UPSTOX, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+
+ // Groww: has "All", "T1", "Demat", "Pledged" sheets
+ if (sheetNamesStr.contains("DEMAT") && sheetNamesStr.contains("PLEDGED")) {
+ return new DetectionResult(BrokerType.GROWW, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+
+ // Angel One: Equity + Mutual Fund sheets (combined portfolio)
+ if (sheetNamesStr.contains("EQUITY") && sheetNamesStr.contains("MUTUAL")) {
+ return new DetectionResult(BrokerType.ANGEL_ONE, DocumentType.COMBINE_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+
+ // Inspect first sheet headers
+ Sheet sheet = workbook.getSheetAt(0);
+ if (sheet == null) return DetectionResult.unknown();
+
+ // Cell[0][0] content
+ Row firstRow = sheet.getRow(0);
+ if (firstRow != null) {
+ String cell00 = cellString(firstRow.getCell(0)).toUpperCase();
+ if (cell00.contains("UPSTOX")) {
+ return new DetectionResult(BrokerType.UPSTOX, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+ }
+
+ // Scan first 25 rows for header fingerprints (Zerodha guide text appears deep in the sheet)
+ String headers = extractHeaders(sheet, 25).toUpperCase();
+
+ // Zerodha: has guide text in cell content
+ if (headers.contains("ZERODHA") || headers.contains("VIEW ZERODHA")) {
+ // Check if it's equity-only or combined
+ boolean hasEquity = headers.contains("SYMBOL") && headers.contains("ISIN");
+ DocumentType dt = hasEquity ? DocumentType.STOCK_PORTFOLIO : DocumentType.COMBINE_PORTFOLIO;
+ return new DetectionResult(BrokerType.ZERODHA, dt, CONTENT_CONFIDENCE);
+ }
+ if (headers.contains("BUY AVG. COST PRICE") || headers.contains("TRADING SYMBOL")) {
+ return new DetectionResult(BrokerType.DHAN, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+ if (headers.contains("SCRIPCODE") && headers.contains("AVG. PRICE")) {
+ return new DetectionResult(BrokerType.MSTOCK, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+ if (headers.contains("CURRENT VALUE (INR)") || headers.contains("GAIN/LOSS")) {
+ DocumentType dt = headers.contains("FOLIO") ? DocumentType.MUTUAL_FUND : DocumentType.STOCK_PORTFOLIO;
+ return new DetectionResult(BrokerType.GROWW, dt, CONTENT_CONFIDENCE);
+ }
+ // Groww Demat: Symbol, Category, Net Qty, Avg. Price pattern
+ if (headers.contains("NET QTY") && headers.contains("AVG. PRICE") && headers.contains("CATEGORY")) {
+ return new DetectionResult(BrokerType.GROWW, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+ // Zerodha holdings: Symbol + ISIN + Average Price (no "ZERODHA" text in sheet)
+ if (headers.contains("SYMBOL") && headers.contains("ISIN") && headers.contains("AVERAGE PRICE")) {
+ return new DetectionResult(BrokerType.ZERODHA, DocumentType.STOCK_PORTFOLIO, CONTENT_CONFIDENCE);
+ }
+
+ return DetectionResult.unknown();
+ }
+
+ private String extractHeaders(Sheet sheet, int maxRows) {
+ StringBuilder sb = new StringBuilder();
+ for (int r = 0; r <= Math.min(maxRows, sheet.getLastRowNum()); r++) {
+ Row row = sheet.getRow(r);
+ if (row == null) continue;
+ for (int c = 0; c < row.getLastCellNum(); c++) {
+ sb.append(cellString(row.getCell(c))).append(" ");
+ }
+ }
+ return sb.toString();
+ }
+
+ private String cellString(Cell cell) {
+ if (cell == null) return "";
+ return switch (cell.getCellType()) {
+ case STRING -> cell.getStringCellValue();
+ case NUMERIC -> String.valueOf((long) cell.getNumericCellValue());
+ default -> "";
+ };
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/FilenameBrokerDetectionStrategy.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/FilenameBrokerDetectionStrategy.java
new file mode 100644
index 0000000..35a6a7b
--- /dev/null
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/detection/FilenameBrokerDetectionStrategy.java
@@ -0,0 +1,77 @@
+package org.am.mypotrfolio.service.detection;
+
+import com.am.common.amcommondata.model.enums.BrokerType;
+import lombok.extern.slf4j.Slf4j;
+import org.am.mypotrfolio.domain.common.DocumentType;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.util.Set;
+
+/**
+ * Lowest-cost strategy: inspects the original filename for well-known broker keywords.
+ * Confidence is moderate (60) — filename-based detection can be wrong if the user
+ * renames the file, but it is always cheap and fast.
+ *
+ *
Supports all file extensions — filename is always available regardless of format.
Extracts text from the first 2 pages of the PDF and scans for well-known
+ * broker / document identifiers. Confidence is 90 — textual content is very
+ * reliable unless the PDF is purely image-based (scanned), in which case
+ * extraction yields an empty string and the strategy returns unknown.
+ *
+ *
Signatures detected:
+ *
+ *
Zerodha — "Zerodha Broking", "ZERODHA SECURITIES"
+ *
Groww — "Groww Invest Tech"
+ *
Dhan — "Dhan HQ", "DHAN"
+ *
Angel One — "Angel One Limited", "Angel Broking"
+ *
CDSL CAS — "CDSL", "Consolidated Account Statement"
+ *
NSDL CAS — "NSDL", "Consolidated Account Statement"
+ *
+ */
+@Slf4j
+@Component
+@Order(3)
+public class PdfTextBrokerDetectionStrategy implements BrokerDetectionStrategy {
+
+ private static final Set PDF_EXTENSIONS = Set.of("pdf");
+ private static final int PDF_CONFIDENCE = 90;
+ private static final int MAX_PAGES_TO_SCAN = 2;
+
+ @Override
+ public boolean supports(String fileExtension) {
+ return fileExtension != null && PDF_EXTENSIONS.contains(fileExtension.toLowerCase());
+ }
+
+ @Override
+ public int confidence(MultipartFile file, String passwordHint) {
+ if (file == null) return 0;
+ try {
+ return detect(file, passwordHint).isKnown() ? PDF_CONFIDENCE : 0;
+ } catch (Exception e) {
+ log.debug("PDF confidence check failed for {}: {}", file.getOriginalFilename(), e.getMessage());
+ return 0;
+ }
+ }
+
+ @Override
+ public DetectionResult detect(MultipartFile file, String passwordHint) {
+ try {
+ byte[] bytes = file.getBytes();
+ PDDocument document = passwordHint != null && !passwordHint.isBlank()
+ ? PDDocument.load(bytes, passwordHint)
+ : PDDocument.load(bytes);
+
+ try (document) {
+ PDFTextStripper stripper = new PDFTextStripper();
+ stripper.setStartPage(1);
+ stripper.setEndPage(Math.min(MAX_PAGES_TO_SCAN, document.getNumberOfPages()));
+ String text = stripper.getText(document).toUpperCase();
+
+ if (text.isBlank()) {
+ log.debug("PDF text extraction yielded empty content (possibly scanned image): {}", file.getOriginalFilename());
+ return DetectionResult.unknown();
+ }
+
+ return resolveFromText(text);
+ }
+ } catch (Exception e) {
+ log.debug("PDF text detection failed for {}: {}", file.getOriginalFilename(), e.getMessage());
+ return DetectionResult.unknown();
+ }
+ }
+
+ private DetectionResult resolveFromText(String text) {
+ // CAS documents — multiple portfolios inside one PDF
+ if ((text.contains("CDSL") || text.contains("NSDL"))
+ && text.contains("CONSOLIDATED ACCOUNT STATEMENT")) {
+ return new DetectionResult(null, DocumentType.MUTUAL_FUND, PDF_CONFIDENCE);
+ }
+ if (text.contains("ZERODHA BROKING") || text.contains("ZERODHA SECURITIES")) {
+ DocumentType dt = text.contains("CONTRACT NOTE") ? DocumentType.TRADE_EQ : DocumentType.STOCK_PORTFOLIO;
+ return new DetectionResult(BrokerType.ZERODHA, dt, PDF_CONFIDENCE);
+ }
+ if (text.contains("GROWW INVEST TECH") || text.contains("GROWW SECURITIES")) {
+ return new DetectionResult(BrokerType.GROWW, DocumentType.STOCK_PORTFOLIO, PDF_CONFIDENCE);
+ }
+ if (text.contains("DHAN HQ") || (text.contains("DHAN") && text.contains("HOLDINGS"))) {
+ return new DetectionResult(BrokerType.DHAN, DocumentType.STOCK_PORTFOLIO, PDF_CONFIDENCE);
+ }
+ if (text.contains("ANGEL ONE LIMITED") || text.contains("ANGEL BROKING")) {
+ return new DetectionResult(BrokerType.ANGEL_ONE, DocumentType.STOCK_PORTFOLIO, PDF_CONFIDENCE);
+ }
+ if (text.contains("UPSTOX") || text.contains("RKSV SECURITIES")) {
+ return new DetectionResult(BrokerType.UPSTOX, DocumentType.STOCK_PORTFOLIO, PDF_CONFIDENCE);
+ }
+ return DetectionResult.unknown();
+ }
+}
diff --git a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java
index 1eb7c6b..18e34bf 100644
--- a/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java
+++ b/services/am-document-processor/src/main/java/org/am/mypotrfolio/service/processor/DocumentProcessorImpl.java
@@ -56,10 +56,8 @@ private List processPortfolio(DocumentRequest documentRequest, String portfol
} else if (documentRequest.getDocumentType().isTradeMf()) {
return processTradeMf(documentRequest, portfolioId, userId);
} else if (documentRequest.getDocumentType().isCombinePortfolio()) {
- if (documentRequest.getBrokerType() == null || !documentRequest.getBrokerType().isAngelOne()) {
- throw new UnsupportedOperationException("Combine Portfolio is only supported for Angel One");
- }
- // Broker Portfolio can be Equity or Composite (Equity + MF)
+ // Combined portfolio (Equity + MF) — supported for Angel One, Zerodha, and any
+ // other broker whose file is detected as COMBINE_PORTFOLIO.
List