diff --git a/pom.xml b/pom.xml index f577db19..7883df9a 100644 --- a/pom.xml +++ b/pom.xml @@ -292,6 +292,12 @@ h2 runtime + + + org.apache.commons + commons-csv + 1.11.0 + org.apache.commons diff --git a/src/main/environment/common_ci.properties b/src/main/environment/common_ci.properties index faf5180f..34875d64 100644 --- a/src/main/environment/common_ci.properties +++ b/src/main/environment/common_ci.properties @@ -5,6 +5,12 @@ spring.datasource.username=@env.DATABASE_USERNAME@ spring.datasource.password=@env.DATABASE_PASSWORD@ spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver +## S3 storage for diagnostic documents pushed from a van +aws.s3.access-key=@env.AWS_S3_ACCESS_KEY@ +aws.s3.secret-key=@env.AWS_S3_SECRET_KEY@ +aws.s3.region=@env.AWS_S3_REGION@ +diagnostic.documents.s3.bucket=@env.DIAGNOSTIC_DOCUMENTS_S3_BUCKET@ + ## Carestream URLs, local carestreamOrderCreateURL=@env.COMMON_API@carestream/createOrder @@ -28,6 +34,11 @@ dataSyncUploadUrl=@env.MMU_CENTRAL_SERVER@dataSync/van-to-server ## Data download API, central dataSyncDownloadUrl=@env.MMU_CENTRAL_SERVER@dataSync/server-to-van +## Diagnostic document push (this server -> further central server) +diagnosticDocumentUploadUrl=@env.MMU_CENTRAL_SERVER@dataSync/diagnostic-documents +diagnosticDocument.push.batchSize=@env.DIAGNOSTIC_DOCUMENT_PUSH_BATCH_SIZE@ +diagnostic.documents.storage-root=@env.DIAGNOSTIC_DOCUMENTS_STORAGE_ROOT@ + ## TC specialist slot booking, # not required tcSpecialistSlotBook=@env.TM_API@schedule/bookSlot diff --git a/src/main/environment/common_docker.properties b/src/main/environment/common_docker.properties index 28f6b89e..be5f3608 100644 --- a/src/main/environment/common_docker.properties +++ b/src/main/environment/common_docker.properties @@ -5,6 +5,12 @@ spring.datasource.username=${DATABASE_USERNAME} spring.datasource.password=${DATABASE_PASSWORD} spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver +## S3 storage for diagnostic documents pushed from a van +aws.s3.access-key=${AWS_S3_ACCESS_KEY} +aws.s3.secret-key=${AWS_S3_SECRET_KEY} +aws.s3.region=${AWS_S3_REGION} +diagnostic.documents.s3.bucket=${DIAGNOSTIC_DOCUMENTS_S3_BUCKET} + ## Carestream URLs, local carestreamOrderCreateURL=${COMMON_API}carestream/createOrder @@ -28,6 +34,11 @@ dataSyncUploadUrl=${MMU_CENTRAL_SERVER}dataSync/van-to-server ## Data download API, central dataSyncDownloadUrl=${MMU_CENTRAL_SERVER}dataSync/server-to-van +## Diagnostic document push (this server -> further central server) +diagnosticDocumentUploadUrl=${MMU_CENTRAL_SERVER}dataSync/diagnostic-documents +diagnosticDocument.push.batchSize=${DIAGNOSTIC_DOCUMENT_PUSH_BATCH_SIZE} +diagnostic.documents.storage-root=${DIAGNOSTIC_DOCUMENTS_STORAGE_ROOT} + ## TC specialist slot booking, # not required tcSpecialistSlotBook=${TM_API}schedule/bookSlot diff --git a/src/main/environment/common_example.properties b/src/main/environment/common_example.properties index 294c0e45..515bb23c 100644 --- a/src/main/environment/common_example.properties +++ b/src/main/environment/common_example.properties @@ -7,6 +7,12 @@ spring.datasource.username= spring.datasource.password= spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver +## S3 storage for diagnostic documents pushed from a van (POST /dataSync/diagnostic-documents) +aws.s3.access-key= +aws.s3.secret-key= +aws.s3.region=ap-south-1 +diagnostic.documents.s3.bucket= + ## Carestream URLs, local #carestreamOrderCreateURL =http://localhost:1040/carestream/createOrder carestreamOrderCreateURL =http://localhost:8083/carestream/createOrder @@ -44,6 +50,10 @@ dataSyncUploadUrl=http://10.208.122.38:8080/mmu-api/dataSync/van-to-server #dataSyncDownloadUrl=http://localhost:82/dataSync/server-to-van dataSyncDownloadUrl=http://10.208.122.38:8080/mmu-api/dataSync/server-to-van +diagnosticDocumentUploadUrl=http://localhost:8087/mmu-api/dataSync/diagnostic-documents +diagnosticDocument.push.batchSize=3 +diagnostic.documents.storage-root= + ## TC specialist slot booking, # not required #tcSpecialistSlotBook=http://localhost:8080/schedule/bookSlot tcSpecialistSlotBook=http://10.208.122.38:8080/tm-api/schedule/bookSlot diff --git a/src/main/java/com/iemr/mmu/config/S3ClientConfig.java b/src/main/java/com/iemr/mmu/config/S3ClientConfig.java new file mode 100644 index 00000000..af02dcbb --- /dev/null +++ b/src/main/java/com/iemr/mmu/config/S3ClientConfig.java @@ -0,0 +1,42 @@ +package com.iemr.mmu.config; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.presigner.S3Presigner; + +/*** + * @purpose S3 client beans used to store diagnostic documents pushed from a van. Built + * once and shared, rather than re-built per request. + */ +@Configuration +public class S3ClientConfig { + + @Value("${aws.s3.access-key}") + private String accessKey; + + @Value("${aws.s3.secret-key}") + private String secretKey; + + @Value("${aws.s3.region}") + private String region; + + private StaticCredentialsProvider credentialsProvider() { + return StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey, secretKey)); + } + + @Bean + public S3Client diagnosticDocumentS3Client() { + return S3Client.builder().region(Region.of(region)).credentialsProvider(credentialsProvider()).build(); + } + + @Bean + public S3Presigner diagnosticDocumentS3Presigner() { + return S3Presigner.builder().region(Region.of(region)).credentialsProvider(credentialsProvider()).build(); + } +} diff --git a/src/main/java/com/iemr/mmu/controller/dataSyncActivity/StartSyncActivity.java b/src/main/java/com/iemr/mmu/controller/dataSyncActivity/StartSyncActivity.java index a7f2a7c7..0e9367d6 100644 --- a/src/main/java/com/iemr/mmu/controller/dataSyncActivity/StartSyncActivity.java +++ b/src/main/java/com/iemr/mmu/controller/dataSyncActivity/StartSyncActivity.java @@ -21,6 +21,8 @@ */ package com.iemr.mmu.controller.dataSyncActivity; +import java.util.Map; + import org.json.JSONObject; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,6 +33,7 @@ import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import com.google.gson.Gson; @@ -38,6 +41,8 @@ import com.iemr.mmu.service.dataSyncActivity.DownloadDataFromServerImpl; import com.iemr.mmu.service.dataSyncActivity.DownloadDataFromServerTransactionalImpl; import com.iemr.mmu.service.dataSyncActivity.UploadDataToServerImpl; +import com.iemr.mmu.service.dataSyncLayerCentral.DiagnosticDocumentFetchService; +import com.iemr.mmu.service.dataSyncLayerCentral.DiagnosticDocumentPushServiceImpl; import com.iemr.mmu.utils.CookieUtil; import com.iemr.mmu.utils.response.OutputResponse; @@ -60,7 +65,12 @@ public class StartSyncActivity { @Autowired private DownloadDataFromServerTransactionalImpl downloadDataFromServerTransactionalImpl; @Autowired + private DiagnosticDocumentPushServiceImpl diagnosticDocumentPushServiceImpl; + @Autowired + private DiagnosticDocumentFetchService diagnosticDocumentFetchService; + @Autowired private DownSyncDataFromServerImpl downSyncDataFromServerImpl; + private static final String GROUP_ID = "groupID"; private static final String PROVIDER_SERVICE_MAP_ID = "providerServiceMapID"; @@ -91,6 +101,45 @@ public String dataSyncToServer(@RequestBody String requestOBJ, return response.toStringWithSerialization(); } + @Operation(summary = "Push pending diagnostic documents (docsProcessed='N') to the further central server, in batches") + @PostMapping(value = { "/diagnostic-documents-to-server" }) + public String diagnosticDocumentsToServer(@RequestHeader(value = "Authorization") String authorization, + @RequestHeader(value = "ServerAuthorization") String serverAuthorization, + @RequestParam(required = false) Long villageId) { + OutputResponse response = new OutputResponse(); + try { + String s = diagnosticDocumentPushServiceImpl.pushPendingDocuments(serverAuthorization, villageId); + if (s != null) + response.setResponse(s); + else + response.setError(5000, "Error in diagnostic document push"); + } catch (Exception e) { + logger.error("Error in diagnostic document push : " + e); + response.setError(e); + } + return response.toString(); + } + + @Operation(summary = "Fetch a short-lived download URL for the latest successfully-pushed diagnostic document matching a beneficiary+documentType, from this van's own local record") + @GetMapping(value = { "/diagnostic-documents/download" }) + public String diagnosticDocumentDownloadUrl(@RequestParam Long beneficiaryId, @RequestParam String documentType, + @RequestHeader(value = "Authorization") String authorization) { + OutputResponse response = new OutputResponse(); + try { + Map download = diagnosticDocumentFetchService.getLatestDocumentDownload(beneficiaryId, + documentType); + if (download != null) + response.setResponse(new Gson().toJson(download)); + else + response.setError(5000, "No pushed document found for beneficiaryId=" + beneficiaryId + + ", documentType=" + documentType); + } catch (Exception e) { + logger.error("Error fetching diagnostic document download URL : " + e); + response.setError(e); + } + return response.toString(); + } + @Operation(summary = "Get data sync group details") @GetMapping(value = { "/getSyncGroupDetails" }) public String getSyncGroupDetails() { diff --git a/src/main/java/com/iemr/mmu/controller/dataSyncLayerCentral/MMUDataSyncVanToServer.java b/src/main/java/com/iemr/mmu/controller/dataSyncLayerCentral/MMUDataSyncVanToServer.java index 589e1dcb..1d2c8790 100644 --- a/src/main/java/com/iemr/mmu/controller/dataSyncLayerCentral/MMUDataSyncVanToServer.java +++ b/src/main/java/com/iemr/mmu/controller/dataSyncLayerCentral/MMUDataSyncVanToServer.java @@ -24,7 +24,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -35,11 +34,11 @@ import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncDataDigester; import com.iemr.mmu.data.syncActivity_syncLayer.SyncDownloadMaster; import com.iemr.mmu.data.syncActivity_syncLayer.SyncUploadDataDigester; +import com.iemr.mmu.service.dataSyncLayerCentral.DiagnosticDocumentIngestService; import com.iemr.mmu.service.dataSyncLayerCentral.FetchDownloadDataImpl; import com.iemr.mmu.service.dataSyncLayerCentral.GetDataFromVanAndSyncToDBImpl; import com.iemr.mmu.service.dataSyncLayerCentral.GetDownSyncDataFromCentralImpl; import com.iemr.mmu.service.dataSyncLayerCentral.GetMasterDataFromCentralForVanImpl; -import com.iemr.mmu.utils.CookieUtil; import com.iemr.mmu.utils.response.OutputResponse; import io.swagger.v3.oas.annotations.Operation; @@ -61,6 +60,8 @@ public class MMUDataSyncVanToServer { private FetchDownloadDataImpl fetchDownloadDataImpl; @Autowired private GetDownSyncDataFromCentralImpl getDownSyncDataFromCentralImpl; + @Autowired + private DiagnosticDocumentIngestService diagnosticDocumentIngestService; @Operation(summary = "Sync data from van-to-server") @PostMapping(value = { "/van-to-server" }, consumes = "application/json", produces = "application/json") @@ -81,6 +82,24 @@ public String dataSyncToServer(@RequestBody String requestOBJ, return response.toString(); } + @Operation(summary = "Receive diagnostic documents pushed from a van and store each in S3 (no database write here)") + @PostMapping(value = { "/diagnostic-documents" }, consumes = "application/json", produces = "application/json") + public String diagnosticDocumentsFromVan(@RequestBody String requestOBJ, + @RequestHeader(value = "Authorization") String Authorization) { + OutputResponse response = new OutputResponse(); + try { + String s = diagnosticDocumentIngestService.ingestDocuments(requestOBJ); + if (s != null) + response.setResponse(s); + else + response.setError(5000, "diagnostic document ingest failed"); + } catch (Exception e) { + response.setError(e); + logger.error("Diagnostic document ingest Exception" + e); + } + return response.toString(); + } + @Operation(summary = "Download data from server-to-van") @PostMapping(value = { "/server-to-van" }, consumes = "application/json", produces = "application/json") public String dataDownloadFromServer(@RequestBody SyncDownloadMaster syncDownloadMaster, diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentFetchService.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentFetchService.java new file mode 100644 index 00000000..a03526f9 --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentFetchService.java @@ -0,0 +1,62 @@ +package com.iemr.mmu.service.dataSyncLayerCentral; + +import java.time.Duration; +import java.util.Map; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import software.amazon.awssdk.services.s3.model.GetObjectRequest; +import software.amazon.awssdk.services.s3.presigner.S3Presigner; +import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest; + +/*** + * @purpose Hands back a short-lived presigned URL for the latest successfully-pushed + * diagnostic document matching a beneficiary+documentType, generated on demand from + * the object's S3 key (s3_path) - the bucket is private, so no permanent URL is + * ever persisted or handed out. + */ +@Service +public class DiagnosticDocumentFetchService { + + private static final Duration URL_VALIDITY = Duration.ofMinutes(15); + + @Value("${diagnostic.documents.s3.bucket}") + private String bucket; + + @Autowired + private DiagnosticDocumentRepository diagnosticDocumentRepository; + + @Autowired + private S3Presigner s3Presigner; + + /*** + * @return null if no successfully-pushed document matches, otherwise the download details + * (documentType, orderType, externalOrderId, contentType, lastModDate, downloadUrl, + * urlExpiresInSeconds) + */ + public Map getLatestDocumentDownload(Long beneficiaryId, String documentType) { + Map row = diagnosticDocumentRepository.findLatestDocument(beneficiaryId, documentType); + if (row == null) { + return null; + } + + String s3Key = (String) row.get("s3_path"); + GetObjectRequest getObjectRequest = GetObjectRequest.builder().bucket(bucket).key(s3Key).build(); + GetObjectPresignRequest presignRequest = GetObjectPresignRequest.builder().signatureDuration(URL_VALIDITY) + .getObjectRequest(getObjectRequest).build(); + String downloadUrl = s3Presigner.presignGetObject(presignRequest).url().toString(); + + Map result = new java.util.HashMap<>(); + result.put("externalOrderId", row.get("external_order_id")); + result.put("orderType", row.get("order_type")); + result.put("documentType", row.get("document_type")); + result.put("contentType", row.get("content_type")); + result.put("originalFileName", row.get("original_file_name")); + result.put("lastModDate", String.valueOf(row.get("last_mod_date"))); + result.put("downloadUrl", downloadUrl); + result.put("urlExpiresInSeconds", URL_VALIDITY.getSeconds()); + return result; + } +} \ No newline at end of file diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentIngestService.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentIngestService.java new file mode 100644 index 00000000..b8e596b5 --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentIngestService.java @@ -0,0 +1,117 @@ +package com.iemr.mmu.service.dataSyncLayerCentral; + +import java.lang.reflect.Type; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Base64; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; + +import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.s3.model.ServerSideEncryption; + +/*** + * @purpose Receives a batch of decrypted diagnostic documents pushed from a van and stores + * each in S3 - purely a storage relay, no database writes here. The pushing van + * persists its own record locally (DiagnosticDocumentPushServiceImpl.markPushedToCentral, + * keyed off the s3Path this returns in each ack) - the central server's own database is + * left untouched. + */ +@Service +public class DiagnosticDocumentIngestService { + + private final Logger logger = LoggerFactory.getLogger(this.getClass().getSimpleName()); + private static final Gson GSON = new Gson(); + + @Value("${diagnostic.documents.s3.bucket}") + private String bucket; + + @Autowired + private S3Client s3Client; + + public String ingestDocuments(String requestOBJ) throws Exception { + Type listType = new TypeToken>>() { + }.getType(); + List> items = GSON.fromJson(requestOBJ, listType); + if (items == null || items.isEmpty()) { + return GSON.toJson(new ArrayList<>()); + } + + List> acks = new ArrayList<>(); + for (Map item : items) { + acks.add(ingestOne(item)); + } + return GSON.toJson(acks); + } + + private Map ingestOne(Map item) { + Long diagnosticOrderId = asLong(item.get("diagnosticOrderId")); + String externalOrderId = (String) item.get("externalOrderId"); + String documentType = (String) item.get("documentType"); + + Map ack = new HashMap<>(); + ack.put("diagnosticOrderId", diagnosticOrderId); + ack.put("externalOrderId", externalOrderId); + ack.put("documentType", documentType); + + try { + byte[] plaintext = Base64.getDecoder().decode((String) item.get("fileContentBase64")); + + String sha256Hash = (String) item.get("sha256Hash"); + if (sha256Hash != null && !sha256Hash.equalsIgnoreCase(sha256Hex(plaintext))) { + ack.put("status", "FAILED"); + ack.put("error", "sha256 mismatch on receipt"); + return ack; + } + + Long beneficiaryId = asLong(item.get("beneficiaryId")); + Long villageId = asLong(item.get("villageId")); + String orderType = (String) item.get("orderType"); + String storedFileName = (String) item.get("storedFileName"); + String s3Key = villageId + "/" + beneficiaryId + "/" + orderType + "/" + documentType + "/" + + storedFileName; + String contentType = (String) item.get("contentType"); + + s3Client.putObject( + PutObjectRequest.builder().bucket(bucket).key(s3Key) + .contentType(contentType != null ? contentType : "application/octet-stream") + .serverSideEncryption(ServerSideEncryption.AES256).build(), + RequestBody.fromBytes(plaintext)); + + ack.put("status", "SUCCESS"); + ack.put("s3Path", s3Key); + } catch (Exception e) { + logger.error("Error ingesting diagnostic document: diagnosticOrderId=" + diagnosticOrderId + + ", documentType=" + documentType, e); + ack.put("status", "FAILED"); + ack.put("error", e.getMessage()); + } + return ack; + } + + private static Long asLong(Object value) { + return value == null ? null : ((Number) value).longValue(); + } + + private static String sha256Hex(byte[] data) throws Exception { + byte[] digest = MessageDigest.getInstance("SHA-256").digest(data); + StringBuilder hex = new StringBuilder(digest.length * 2); + for (byte b : digest) { + hex.append(String.format(Locale.ROOT, "%02x", b)); + } + return hex.toString(); + } +} diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentPushServiceImpl.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentPushServiceImpl.java new file mode 100644 index 00000000..adb7714c --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentPushServiceImpl.java @@ -0,0 +1,273 @@ +package com.iemr.mmu.service.dataSyncLayerCentral; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Service; +import org.springframework.web.client.RestTemplate; + +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import com.google.gson.reflect.TypeToken; +import com.iemr.mmu.utils.CryptoUtil; +import com.iemr.mmu.utils.RestTemplateUtil; + +import java.lang.reflect.Type; + +/*** + * @purpose Reads MMU-API's own locally-pending diagnostic documents (docsProcessed='N', + * shared db_iemr.tb_diagnostic_document table), decrypts the file off the shared + * filesystem, and pushes each batch to the further central server's + * /dataSync/diagnostic-documents endpoint - the same relay shape as + * UploadDataToServerImpl's push to dataSyncUploadUrl, just for this table instead + * of the generic sync-group config. MMU-API's own /dataSync/diagnostic-documents + * ingest endpoint (DiagnosticDocumentCentralIngestService) is untouched by this - + * this service is a separate outbound relay, not a caller of it. + */ +@Service +public class DiagnosticDocumentPushServiceImpl { + + private final Logger logger = LoggerFactory.getLogger(this.getClass().getSimpleName()); + + private static final Map CONTENT_TYPE_EXTENSIONS = new HashMap<>(); + static { + CONTENT_TYPE_EXTENSIONS.put("application/pdf", "pdf"); + CONTENT_TYPE_EXTENSIONS.put("image/jpeg", "jpg"); + CONTENT_TYPE_EXTENSIONS.put("image/png", "png"); + } + + @Value("${diagnostic.documents.storage-root}") + private String storageRoot; + + @Value("${diagnosticDocumentUploadUrl}") + private String diagnosticDocumentUploadUrl; + + @Value("${diagnosticDocument.push.batchSize:3}") + private int batchSize; + + @Autowired + private DiagnosticDocumentRepository diagnosticDocumentRepository; + + @Autowired + private CryptoUtil cryptoUtil; + + public String pushPendingDocuments(String Authorization, Long villageId) throws Exception { + List> pendingRows = diagnosticDocumentRepository.findPendingDocuments(); + boolean anyRowsFound = !pendingRows.isEmpty(); + int totalAttempted = 0; + int totalSucceeded = 0; + + for (int offset = 0; offset < pendingRows.size(); offset += batchSize) { + List> rows = pendingRows.subList(offset, Math.min(offset + batchSize, pendingRows.size())); + + List> payloadItems = new ArrayList<>(); + Map> rowsByAckKey = new HashMap<>(); + for (Map row : rows) { + Long rowId = asLong(row.get("id")); + String base64Plaintext; + try { + String storedPath = (String) row.get("stored_path"); + Path filePath = Paths.get(storageRoot, storedPath); + String encryptedPayload = new String(Files.readAllBytes(filePath), StandardCharsets.UTF_8); + base64Plaintext = cryptoUtil.decrypt(encryptedPayload); + } catch (Exception e) { + // File missing/unreadable off the shared filesystem - mark just this row + // failed and move on, rather than letting the exception abort the whole run + // (which would leave every other pending row, in this batch and beyond, + // completely untouched). + logger.warn("Skipping diagnostic document push, could not read file off disk: id={}, error={}", + rowId, e.getMessage()); + diagnosticDocumentRepository.markPushFailed(rowId, + "Could not read file off shared filesystem: " + e.getMessage()); + continue; + } + if (base64Plaintext == null) { + logger.warn("Skipping diagnostic document push, decrypt failed: id={}", rowId); + diagnosticDocumentRepository.markPushFailed(rowId, "Decrypt failed"); + continue; + } + + Long diagnosticOrderId = asLong(row.get("diagnostic_order_id")); + String externalOrderId = (String) row.get("external_order_id"); + String documentType = (String) row.get("document_type"); + String contentType = (String) row.get("content_type"); + + Map item = new HashMap<>(); + item.put("diagnosticOrderId", diagnosticOrderId); + item.put("externalOrderId", externalOrderId); + item.put("beneficiaryId", row.get("beneficiary_id")); + item.put("orderType", row.get("order_type")); + item.put("documentType", documentType); + item.put("storedFileName", row.get("stored_file_name")); + item.put("sha256Hash", row.get("sha256_hash")); + item.put("contentType", contentType); + item.put("fileExtension", extensionFor(contentType)); + item.put("originalFileName", row.get("original_file_name")); + item.put("vanID", row.get("vanID")); + item.put("parkingPlaceID", row.get("parkingPlaceID")); + item.put("vanSerialNo", row.get("vanSerialNo")); + item.put("villageId", villageId); + item.put("fileContentBase64", base64Plaintext); + payloadItems.add(item); + rowsByAckKey.put(ackKey(externalOrderId, documentType), row); + } + + if (payloadItems.isEmpty()) { + // Every row in this batch already got marked failed above (missing file or + // failed decrypt) - nothing left to send. + continue; + } + + // Counted here, not after the central round-trip below - these documents WERE + // successfully decrypted and queued for sending regardless of whether the central + // server subsequently accepts, rejects, or fails to respond to them. Otherwise a + // batch that decrypted fine but got rejected by the central server (e.g. an expired + // session) would leave totalAttempted at 0, and the caller would see the misleading + // "No documents could be decrypted for push" instead of the real per-row reason + // (already recorded in docSyncFailureReason). + totalAttempted += payloadItems.size(); + + String requestOBJ = new Gson().toJson(payloadItems); + List> acks; + try { + HttpEntity request = RestTemplateUtil.createRequestEntity(requestOBJ, Authorization, "datasync"); + RestTemplate restTemplate = new RestTemplate(); + ResponseEntity response = restTemplate.exchange(diagnosticDocumentUploadUrl, HttpMethod.POST, + request, String.class); + + if (response == null || !response.hasBody()) { + logger.warn("No response body from central server for diagnostic document push, marking batch failed"); + markBatchFailed(rowsByAckKey, "No response body from central server"); + continue; + } + + // Central server wraps every response in the shared OutputResponse envelope + // ({"data": [...], "statusCode":200, ...}) - the ack array lives under "data". + JsonElement parsedBody = JsonParser.parseString(response.getBody()); + if (!parsedBody.isJsonObject()) { + logger.warn("Unexpected response shape from central server for diagnostic document push, marking batch failed"); + markBatchFailed(rowsByAckKey, "Unexpected response shape from central server"); + continue; + } + JsonObject envelope = parsedBody.getAsJsonObject(); + if (!envelope.has("statusCode") || envelope.get("statusCode").getAsInt() != 200 + || !envelope.has("data")) { + logger.warn("Central server reported failure for diagnostic document push batch, marking batch failed: {}", + response.getBody()); + markBatchFailed(rowsByAckKey, "Central server reported failure: " + response.getBody()); + continue; + } + + Type ackListType = new TypeToken>>() { + }.getType(); + acks = new Gson().fromJson(envelope.get("data"), ackListType); + if (acks == null) { + markBatchFailed(rowsByAckKey, "Central server returned no acknowledgements"); + continue; + } + } catch (Exception e) { + logger.error("Error calling central server for diagnostic document push, marking batch failed", e); + markBatchFailed(rowsByAckKey, "Error calling central server: " + e.getMessage()); + continue; + } + + int batchSuccessCount = 0; + Map> unmatchedRowsByAckKey = new HashMap<>(rowsByAckKey); + for (Map ack : acks) { + String key = ackKey((String) ack.get("externalOrderId"), (String) ack.get("documentType")); + Map row = rowsByAckKey.get(key); + if (row == null) { + continue; + } + unmatchedRowsByAckKey.remove(key); + if ("SUCCESS".equalsIgnoreCase((String) ack.get("status"))) { + diagnosticDocumentRepository.markPushedToCentral(asLong(row.get("id")), + (String) ack.get("s3Path")); + batchSuccessCount++; + } else { + String reason = (String) ack.get("error"); + logger.warn("Central server rejected diagnostic document push: externalOrderId={}, documentType={}, error={}", + ack.get("externalOrderId"), ack.get("documentType"), reason); + diagnosticDocumentRepository.markPushFailed(asLong(row.get("id")), + reason != null ? reason : "Central server rejected the document"); + } + } + if (!unmatchedRowsByAckKey.isEmpty()) { + // Central sent back fewer acks than documents we sent - whatever wasn't + // accounted for must not be silently left at its previous status forever. + logger.warn( + "Diagnostic document push: {} row(s) in this batch got no matching ack back, marking failed", + unmatchedRowsByAckKey.size()); + markBatchFailed(unmatchedRowsByAckKey, "No acknowledgement received from central server"); + } + + totalSucceeded += batchSuccessCount; + logger.info("Diagnostic document push batch complete: attempted={}, succeeded={}", payloadItems.size(), + batchSuccessCount); + } + + if (!anyRowsFound) { + return "No pending diagnostic documents to sync"; + } + if (totalAttempted == 0) { + return "No documents could be decrypted for push"; + } + if (totalSucceeded == 0) { + // Documents WERE decrypted and sent, but none were accepted (e.g. the central + // server rejected every batch) - the specific reason for each row is recorded in + // its own docSyncFailureReason, this is just the overall-outcome summary. + return "Documents were sent but none were accepted by the central server"; + } + + logger.info("Diagnostic document push complete overall: attempted={}, succeeded={}", totalAttempted, + totalSucceeded); + return "Data successfully synced"; + } + + private static String ackKey(String externalOrderId, String documentType) { + return externalOrderId + "|" + documentType; + } + + /*** + * @purpose Called when the whole batch call to the central server fails (no/garbled + * response, non-200, or a thrown exception) - marks every row that was in this + * batch as failed, matching how /van-to-server marks a whole batch failed on a + * connection error, rather than leaving them stuck at docsProcessed='N' forever. + */ + private void markBatchFailed(Map> rowsByAckKey, String reason) { + for (Map row : rowsByAckKey.values()) { + diagnosticDocumentRepository.markPushFailed(asLong(row.get("id")), reason); + } + } + + private static Long asLong(Object value) { + return value == null ? null : ((Number) value).longValue(); + } + + private static String extensionFor(String contentType) { + if (contentType == null) { + return "bin"; + } + String extension = CONTENT_TYPE_EXTENSIONS.get(contentType.toLowerCase()); + if (extension != null) { + return extension; + } + int slashIndex = contentType.indexOf('/'); + return slashIndex >= 0 && slashIndex < contentType.length() - 1 ? contentType.substring(slashIndex + 1) : "bin"; + } +} diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentRepository.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentRepository.java new file mode 100644 index 00000000..53c30c5f --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DiagnosticDocumentRepository.java @@ -0,0 +1,95 @@ +package com.iemr.mmu.service.dataSyncLayerCentral; + +import java.util.List; +import java.util.Map; + +import javax.sql.DataSource; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; + +/*** + * @purpose Local (van) side persistence for the diagnostic-document push pipeline - the central + * server is a pure S3 storage relay (DiagnosticDocumentCentralIngestService) and writes + * nothing to its own database, so this repository only ever operates on the pushing + * van's own local db_iemr.tb_diagnostic_document rows. + */ +@Service +public class DiagnosticDocumentRepository { + + @Autowired + private DataSource dataSource; + + private JdbcTemplate jdbcTemplate; + + private JdbcTemplate getJdbcTemplate() { + if (this.jdbcTemplate == null) { + this.jdbcTemplate = new JdbcTemplate(dataSource); + } + return this.jdbcTemplate; + } + + /*** + * @purpose Latest successfully-pushed document for a beneficiary+documentType, used to hand + * back a fresh presigned URL on demand instead of persisting a permanent (and, for a + * private bucket, non-functional) URL. s3_path holds the S3 key for a + * docsProcessed='P' row (markPushedToCentral() below writes it there). + */ + public Map findLatestDocument(Long beneficiaryId, String documentType) { + String query = "SELECT id, external_order_id, order_type, document_type, s3_path, content_type, " + + "original_file_name, last_mod_date FROM db_iemr.tb_diagnostic_document " + + "WHERE beneficiary_id = ? AND document_type = ? AND docsProcessed = 'P' " + + "ORDER BY last_mod_date DESC, id DESC LIMIT 1"; + List> rows = getJdbcTemplate().queryForList(query, beneficiaryId, documentType); + return rows.isEmpty() ? null : rows.get(0); + } + + /*** + * @purpose Snapshot of every locally-pending row's metadata (docsProcessed='N' or 'F') at + * the start of a push run, taken in one query so a row this same run marks 'F' + * partway through (e.g. an auth failure) is never re-picked-up later in the same + * run - it only becomes eligible again on the next trigger, alongside whatever's + * newly 'N' by then. Only metadata is loaded here (id, hashes, filenames, etc.) - + * each row's actual file content is still read off disk and decrypted one batch at + * a time by the caller, so the whole backlog's decrypted content is never held in + * memory at once. + */ + public List> findPendingDocuments() { + String query = "SELECT id, diagnostic_order_id, external_order_id, beneficiary_id, order_type, document_type, " + + "stored_file_name, stored_path, sha256_hash, content_type, original_file_name, " + + "vanID, parkingPlaceID, vanSerialNo FROM db_iemr.tb_diagnostic_document " + + "WHERE docsProcessed = 'N' OR docsProcessed = 'F' ORDER BY id ASC"; + return getJdbcTemplate().queryForList(query); + } + + /*** + * @purpose Marks a document as successfully pushed on the LOCAL (van) row - the central + * server itself writes nothing to its own database. s3Path is the S3 key the + * central server's ack reported back, persisted here so the van's own local DB + * knows where the document ended up, not just that it did. + */ + public void markPushedToCentral(Long id, String s3Path) { + String update = "UPDATE db_iemr.tb_diagnostic_document SET processed = 'N', docsProcessed = 'P', " + + "s3_path = ?, docSyncedDate = NOW(), docSyncFailureReason = NULL, last_mod_date = NOW() WHERE id = ?"; + getJdbcTemplate().update(update, s3Path, id); + } + + /*** + * @purpose Marks a document push to the further central server as failed - docsProcessed='F', + * matching the P/F status convention used by the generic sync pipeline's + * DataSyncRepository.updateProcessedFlagInVan for /van-to-server. Persists why in + * docSyncFailureReason (mirroring the generic pipeline's own SyncFailureReason, kept + * separate since this pipeline's docsProcessed/docSyncedDate are their own dedicated + * columns) so a failure isn't only visible in the application log. Also clears + * s3_path - otherwise a row that succeeded once, then got re-attempted and failed, + * would keep showing a stale S3 key while docsProcessed says 'F'. docSyncedDate is + * left untouched - it records the last time this row was actually confirmed synced, + * if ever. + */ + public void markPushFailed(Long id, String reason) { + String update = "UPDATE db_iemr.tb_diagnostic_document SET processed = 'N', docsProcessed = 'F', " + + "s3_path = NULL, docSyncFailureReason = ?, last_mod_date = NOW() WHERE id = ?"; + getJdbcTemplate().update(update, reason, id); + } +} diff --git a/src/main/java/com/iemr/mmu/utils/CryptoUtil.java b/src/main/java/com/iemr/mmu/utils/CryptoUtil.java new file mode 100644 index 00000000..c964fdf2 --- /dev/null +++ b/src/main/java/com/iemr/mmu/utils/CryptoUtil.java @@ -0,0 +1,42 @@ +package com.iemr.mmu.utils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import javax.crypto.Cipher; +import javax.crypto.SecretKey; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.util.Base64; + +/** + * Decrypts diagnostic-document files written by FLW-API's own CryptoUtil + * (same AES/ECB/PKCS5Padding scheme and key) so they can be read directly + * off the shared filesystem before pushing to the further central server. + */ +@Service +public class CryptoUtil { + + private static final Logger logger = LoggerFactory.getLogger(CryptoUtil.class); + private static final String ALGORITHM = "AES"; + private static final String SECRET_KEY = "dev-envro-secret"; + + public String decrypt(String encryptedValue) { + try { + SecretKey secretKey = new SecretKeySpec(SECRET_KEY.getBytes(StandardCharsets.UTF_8), ALGORITHM); + Cipher cipher = Cipher.getInstance("AES/ECB/PKCS5Padding"); + cipher.init(Cipher.DECRYPT_MODE, secretKey); + byte[] decryptedBytes = cipher.doFinal(Base64.getDecoder().decode(encryptedValue)); + return removePadding(new String(decryptedBytes, StandardCharsets.UTF_8)); + } catch (Exception e) { + logger.error("Exception while decrypting diagnostic document", e); + return null; + } + } + + private String removePadding(String value) { + int paddingLength = value.charAt(value.length() - 1); + return value.substring(0, value.length() - paddingLength); + } +}