diff --git a/pom.xml b/pom.xml index 59ea5dce..7883df9a 100644 --- a/pom.xml +++ b/pom.xml @@ -5,7 +5,7 @@ 4.0.0 com.iemr.mmu mmu-api - 3.6.2 + 3.9.0 war MMU-API @@ -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 918156fa..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 @@ -51,6 +62,15 @@ dataSyncTransactionDownloadUrl=@env.MMU_CENTRAL_SERVER@dataSync/server-to-van-tr ## inventory transaction processed flag update URL, central, mmu dataSyncProcessedFlagUpdate=@env.MMU_CENTRAL_SERVER@dataSync/updateProcessedFlagPostDownload +## down-sync (central -> local) data API, central, mmu +downSyncDataUrl=@env.MMU_CENTRAL_SERVER@dataSync/server-to-van-downsync + +## down-sync (central -> local) flag update API, central, mmu +downSyncFlagUpdateUrl=@env.MMU_CENTRAL_SERVER@dataSync/updateDownSyncFlagPostDownload + +## mark an up-synced record as already delivered to the van (central only) +downSync.markDeliveredOnUpSync=true + fileBasePath =@env.MMU_FILE_BASE_PATH@ diff --git a/src/main/environment/common_docker.properties b/src/main/environment/common_docker.properties index 52298c5f..be5f3608 100644 --- a/src/main/environment/common_docker.properties +++ b/src/main/environment/common_docker.properties @@ -1,87 +1,107 @@ -# local env -# DB Connections -spring.datasource.url=${DATABASE_URL} -spring.datasource.username=${DATABASE_USERNAME} -spring.datasource.password=${DATABASE_PASSWORD} -spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver - -## Carestream URLs, local -carestreamOrderCreateURL=${COMMON_API}carestream/createOrder - -## Identity - Common URLs -# Registration -registrationUrl =${COMMON_API}beneficiary/createBeneficiary - -registrarQuickSearchByIdUrl =${COMMON_API}beneficiary/searchUserByID - -registrarQuickSearchByPhoneNoUrl =${COMMON_API}beneficiary/searchUserByPhone - -getBenImageFromIdentity =${IDENTITY_API_BASE_URL}id/benImageByBenRegID - -beneficiaryEditUrl =${COMMON_API}beneficiary/updateBenefciaryDetails - -registrarAdvanceSearchUrl =${COMMON_API}beneficiary/searchBeneficiary - -## Data Sync API, central -dataSyncUploadUrl=${MMU_CENTRAL_SERVER}dataSync/van-to-server - -## Data download API, central -dataSyncDownloadUrl=${MMU_CENTRAL_SERVER}dataSync/server-to-van - -## TC specialist slot booking, # not required -tcSpecialistSlotBook=${TM_API}schedule/bookSlot - -## TC specialist slot cancelling, # not required -tcSpecialistSlotCancel=${TM_API}schedule/cancelBookedSlot - -## benGenUrl, common, central -benGenUrlCentral=${COMMON_API_OPEN_SYNC}beneficiary/generateBeneficiaryIDs - -## benImportUrl, identity, local -benImportUrlLocal=${IDENTITY_API_BASE_URL}id/saveGeneratedBenIDToLocalServer -mmucentralserver = ${MMU_CENTRAL_SERVER}common/get/Case-sheet/centralServerTMCaseSheet - -tmCentralServer = ${TM_CENTRAL_SERVER}common/get/Case-sheet/printData - -specialistSign=${COMMON_API}signature1/getSignClass - -## inventory transaction download URL, central, mmu -dataSyncTransactionDownloadUrl=${MMU_CENTRAL_SERVER}dataSync/server-to-van-transactional - -## inventory transaction processed flag update URL, central, mmu -dataSyncProcessedFlagUpdate=${MMU_CENTRAL_SERVER}dataSync/updateProcessedFlagPostDownload - - -fileBasePath =${MMU_FILE_BASE_PATH} - -## data sync batch size -BATCH_SIZE=30 -snomedCTPageSize=10 -nurseWL=7 -docWL=7 -pharmaWL=7 -labWL=7 -radioWL=7 -oncoWL=7 - -## For File Sync -serverIP=${FILE_SYNC_SERVER_IP} -serverDomain=${FILE_SYNC_SERVER_DOMAIN} -serverUserName=${FILE_SYNC_SERVER_USERNAME} -serverPassword=${FILE_SYNC_SERVER_PASSWORD} -localFolderToSync=${LOCAL_FOLDER_TO_SYNC} -serverFolder=MMUSync -getServerCredentialURL=${MMU_API}fileSyncController/getServerCredential -tmReferCheckValue=TM -### Redis IP -spring.redis.host=${REDIS_HOST} - -jwt.secret=${JWT_SECRET_KEY} - - -#ELK logging file name -logging.path=logs -logging.file.name=${MMU_API_LOGGING_FILE_NAME} - -springdoc.api-docs.enabled=${SWAGGER_DOC_ENABLED} +# local env +# DB Connections +spring.datasource.url=${DATABASE_URL} +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 + +## Identity - Common URLs +# Registration +registrationUrl =${COMMON_API}beneficiary/createBeneficiary + +registrarQuickSearchByIdUrl =${COMMON_API}beneficiary/searchUserByID + +registrarQuickSearchByPhoneNoUrl =${COMMON_API}beneficiary/searchUserByPhone + +getBenImageFromIdentity =${IDENTITY_API_BASE_URL}id/benImageByBenRegID + +beneficiaryEditUrl =${COMMON_API}beneficiary/updateBenefciaryDetails + +registrarAdvanceSearchUrl =${COMMON_API}beneficiary/searchBeneficiary + +## Data Sync API, central +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 + +## TC specialist slot cancelling, # not required +tcSpecialistSlotCancel=${TM_API}schedule/cancelBookedSlot + +## benGenUrl, common, central +benGenUrlCentral=${COMMON_API_OPEN_SYNC}beneficiary/generateBeneficiaryIDs + +## benImportUrl, identity, local +benImportUrlLocal=${IDENTITY_API_BASE_URL}id/saveGeneratedBenIDToLocalServer +mmucentralserver = ${MMU_CENTRAL_SERVER}common/get/Case-sheet/centralServerTMCaseSheet + +tmCentralServer = ${TM_CENTRAL_SERVER}common/get/Case-sheet/printData + +specialistSign=${COMMON_API}signature1/getSignClass + +## inventory transaction download URL, central, mmu +dataSyncTransactionDownloadUrl=${MMU_CENTRAL_SERVER}dataSync/server-to-van-transactional + +## inventory transaction processed flag update URL, central, mmu +dataSyncProcessedFlagUpdate=${MMU_CENTRAL_SERVER}dataSync/updateProcessedFlagPostDownload + +## down-sync (central -> local) data API, central, mmu +downSyncDataUrl=${MMU_CENTRAL_SERVER}dataSync/server-to-van-downsync + +## down-sync (central -> local) flag update API, central, mmu +downSyncFlagUpdateUrl=${MMU_CENTRAL_SERVER}dataSync/updateDownSyncFlagPostDownload + +## mark an up-synced record as already delivered to the van (central only) +downSync.markDeliveredOnUpSync=true + + +fileBasePath =${MMU_FILE_BASE_PATH} + +## data sync batch size +BATCH_SIZE=30 +snomedCTPageSize=10 +nurseWL=7 +docWL=7 +pharmaWL=7 +labWL=7 +radioWL=7 +oncoWL=7 + +## For File Sync +serverIP=${FILE_SYNC_SERVER_IP} +serverDomain=${FILE_SYNC_SERVER_DOMAIN} +serverUserName=${FILE_SYNC_SERVER_USERNAME} +serverPassword=${FILE_SYNC_SERVER_PASSWORD} +localFolderToSync=${LOCAL_FOLDER_TO_SYNC} +serverFolder=MMUSync +getServerCredentialURL=${MMU_API}fileSyncController/getServerCredential +tmReferCheckValue=TM +### Redis IP +spring.redis.host=${REDIS_HOST} + +jwt.secret=${JWT_SECRET_KEY} + + +#ELK logging file name +logging.path=logs +logging.file.name=${MMU_API_LOGGING_FILE_NAME} + +springdoc.api-docs.enabled=${SWAGGER_DOC_ENABLED} springdoc.swagger-ui.enabled=${SWAGGER_DOC_ENABLED} \ No newline at end of file diff --git a/src/main/environment/common_example.properties b/src/main/environment/common_example.properties index 7157c564..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 @@ -64,6 +74,15 @@ dataSyncTransactionDownloadUrl=http://10.208.122.38:8080/mmu-api/dataSync/server ## inventory transaction processed flag update URL, central, mmu dataSyncProcessedFlagUpdate=http://10.208.122.38:8080/mmu-api/dataSync/updateProcessedFlagPostDownload +## down-sync (central -> local) data API, central, mmu +downSyncDataUrl=http://10.208.122.38:8080/mmu-api/dataSync/server-to-van-downsync + +## down-sync (central -> local) flag update API, central, mmu +downSyncFlagUpdateUrl=http://10.208.122.38:8080/mmu-api/dataSync/updateDownSyncFlagPostDownload + +## mark an up-synced record as already delivered to the van (central only) +downSync.markDeliveredOnUpSync=true + ## fetching the TM case sheet for MMU. mmucentralserver = /mmu-api/common/get/Case-sheet/centralServerTMCaseSheet 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 83c5dcfe..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,12 +33,16 @@ 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; +import com.iemr.mmu.service.dataSyncActivity.DownSyncDataFromServerImpl; 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; @@ -58,6 +64,13 @@ public class StartSyncActivity { private DownloadDataFromServerImpl downloadDataFromServerImpl; @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"; @@ -88,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() { @@ -225,4 +277,59 @@ public String downloadTransactionToLocal(@RequestBody String requestOBJ, return response.toString(); } + + @Operation(summary = "Start the down-sync of masters & transactional data from central to local") + @PostMapping(value = { "/startDownSync" }) + public String startDownSync(@RequestBody String requestOBJ, + @RequestHeader(value = "Authorization") String authorization, + @RequestHeader(value = "ServerAuthorization") String serverAuthorization, HttpServletRequest request) { + OutputResponse response = new OutputResponse(); + try { + String jwtToken = CookieUtil.getJwtTokenFromCookie(request); + + if (serverAuthorization == null || serverAuthorization.trim().isEmpty()) { + logger.error("Down-sync : ServerAuthorization is empty - do the data sync server login first"); + response.setError(5000, + "Server authorization is missing. Kindly do the data sync server login and try again."); + return response.toString(); + } + logger.info("Down-sync : ServerAuthorization received, length {}", serverAuthorization.trim().length()); + + JSONObject obj = new JSONObject(requestOBJ); + if (obj.has("vanID") && obj.get("vanID") != null) { + Integer providerServiceMapID = obj.has(PROVIDER_SERVICE_MAP_ID) && !obj.isNull(PROVIDER_SERVICE_MAP_ID) + ? obj.getInt(PROVIDER_SERVICE_MAP_ID) + : null; + + String s = downSyncDataFromServerImpl.startDownSync(serverAuthorization, jwtToken, obj.getInt("vanID"), + providerServiceMapID); + + if (s != null && s.equalsIgnoreCase("inProgress")) { + response.setError(5000, "Down-sync is already in progress, kindly wait for it to finish"); + } else { + response.setResponse(new Gson().toJson(downSyncDataFromServerImpl.getDownSyncStatus())); + } + } else { + response.setError(5000, "vanID is missing, Kindly contact the administrator."); + } + } catch (Exception e) { + logger.error("Error in down-sync : {}", e.getMessage(), e); + response.setError(e); + } + return response.toString(); + } + + @Operation(summary = "Down-sync progress check") + @GetMapping(value = { "/checkDownSyncProgress" }) + public String checkDownSyncProgress() { + OutputResponse response = new OutputResponse(); + try { + response.setResponse(new Gson().toJson(downSyncDataFromServerImpl.getDownSyncStatus())); + } catch (Exception e) { + logger.error("Error in down-sync progress check : {}", e.getMessage(), e); + response.setError(e); + } + return response.toString(); + } + } 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 ea943d0e..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; @@ -32,12 +31,14 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +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; @@ -57,6 +58,10 @@ public class MMUDataSyncVanToServer { private GetMasterDataFromCentralForVanImpl getMasterDataFromCentralForVanImpl; @Autowired 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") @@ -77,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, @@ -156,4 +179,52 @@ public String updateProcessedFlagPostDownload(@RequestBody SyncUploadDataDigeste return response.toStringWithSerialization(); } + + @Operation(summary = "Down-sync data of one configured table from central to van") + @PostMapping(value = { "/server-to-van-downsync" }, consumes = "application/json", produces = "application/json") + public String downSyncDataFromServer(@RequestBody DownSyncDataDigester downSyncDataDigester, + @RequestHeader(value = "Authorization") String Authorization) { + OutputResponse response = new OutputResponse(); + try { + if (downSyncDataDigester != null) { + String s = getDownSyncDataFromCentralImpl.getDownSyncDataForVan(downSyncDataDigester); + if (s != null) + response.setResponse(s); + else + response.setError(5000, "Error in down-sync for table " + downSyncDataDigester.getSchemaName() + "." + + downSyncDataDigester.getTableName()); + } else { + response.setError(5000, "Invalid request"); + } + } catch (Exception e) { + if (downSyncDataDigester != null) + logger.error("Error in down-sync for table {}.{}", downSyncDataDigester.getSchemaName(), + downSyncDataDigester.getTableName(), e); + response.setError(e); + } + return response.toStringWithSerialization(); + } + + @Operation(summary = "Update the down-sync flag at central post delivery to the van") + @PostMapping(value = { + "/updateDownSyncFlagPostDownload" }, consumes = "application/json", produces = "application/json") + public String updateDownSyncFlagPostDownload(@RequestBody DownSyncDataDigester downSyncDataDigester, + @RequestHeader(value = "Authorization") String Authorization) { + OutputResponse response = new OutputResponse(); + try { + if (downSyncDataDigester != null) { + int i = getDownSyncDataFromCentralImpl.updateDownSyncFlagPostDownload(downSyncDataDigester); + response.setResponse(String.valueOf(i)); + } else { + response.setError(5000, "Invalid request"); + } + } catch (Exception e) { + if (downSyncDataDigester != null) + logger.error("Error while updating the down-sync flag for {}.{}", downSyncDataDigester.getSchemaName(), + downSyncDataDigester.getTableName(), e); + response.setError(e); + } + return response.toStringWithSerialization(); + } + } diff --git a/src/main/java/com/iemr/mmu/data/nurse/BeneficiaryVisitDetail.java b/src/main/java/com/iemr/mmu/data/nurse/BeneficiaryVisitDetail.java index dd17ef7f..ff7778ec 100644 --- a/src/main/java/com/iemr/mmu/data/nurse/BeneficiaryVisitDetail.java +++ b/src/main/java/com/iemr/mmu/data/nurse/BeneficiaryVisitDetail.java @@ -141,6 +141,22 @@ public class BeneficiaryVisitDetail { @Column(name = "VisitFlowStatusFlag", insertable = false) private String visitFlowStatusFlag; + @Expose + @Column(name = "NurseID") + private Long nurseID; + + @Expose + @Column(name = "DoctorID") + private Long doctorID; + + @Expose + @Column(name = "PharmacistID") + private Long pharmacistID; + + @Expose + @Column(name = "LabTechnicianID") + private Long labTechnicianID; + @Expose @Column(name = "VanSerialNo") private Long vanSerialNo; @@ -535,4 +551,36 @@ public void setFiles(ArrayList> files) { this.files = files; } + public Long getNurseID() { + return nurseID; + } + + public void setNurseID(Long nurseID) { + this.nurseID = nurseID; + } + + public Long getDoctorID() { + return doctorID; + } + + public void setDoctorID(Long doctorID) { + this.doctorID = doctorID; + } + + public Long getPharmacistID() { + return pharmacistID; + } + + public void setPharmacistID(Long pharmacistID) { + this.pharmacistID = pharmacistID; + } + + public Long getLabTechnicianID() { + return labTechnicianID; + } + + public void setLabTechnicianID(Long labTechnicianID) { + this.labTechnicianID = labTechnicianID; + } + } diff --git a/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncDataDigester.java b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncDataDigester.java new file mode 100644 index 00000000..c250cf73 --- /dev/null +++ b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncDataDigester.java @@ -0,0 +1,166 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.data.syncActivity_syncLayer; + +import java.util.List; + +import com.google.gson.annotations.Expose; + +public class DownSyncDataDigester { + + @Expose + private String schemaName; + @Expose + private String tableName; + /** columns to SELECT from central */ + @Expose + private String serverColumnName; + /** primary key column of the table (same name in central & local) */ + @Expose + private String vanAutoIncColumnName; + /** MASTER / TRANSACTIONAL */ + @Expose + private String tableType; + @Expose + private Integer vanID; + @Expose + private Integer providerServiceMapID; + /** populated only on the acknowledgement call */ + @Expose + private List records; + private Long lastFetchedID; + private Integer batchSize; + + public DownSyncDataDigester() { + } + + public static DownSyncDataDigester forDownload(DownSyncTableDetail tableDetail, String serverColumnName, + Integer vanID, Integer providerServiceMapID, Long lastFetchedID, Integer batchSize) { + DownSyncDataDigester digester = new DownSyncDataDigester(); + digester.schemaName = tableDetail.getSchemaName(); + digester.tableName = tableDetail.getTableName(); + digester.serverColumnName = serverColumnName; + digester.vanAutoIncColumnName = tableDetail.getVanAutoIncColumnName(); + digester.tableType = tableDetail.getTableType(); + digester.vanID = vanID; + digester.providerServiceMapID = providerServiceMapID; + digester.lastFetchedID = lastFetchedID; + digester.batchSize = batchSize; + return digester; + } + + public static DownSyncDataDigester forAck(DownSyncTableDetail tableDetail, Integer vanID, + List records) { + DownSyncDataDigester digester = new DownSyncDataDigester(); + digester.schemaName = tableDetail.getSchemaName(); + digester.tableName = tableDetail.getTableName(); + digester.vanAutoIncColumnName = tableDetail.getVanAutoIncColumnName(); + digester.tableType = tableDetail.getTableType(); + digester.vanID = vanID; + digester.records = records; + return digester; + } + + public String getSchemaName() { + return schemaName; + } + + public void setSchemaName(String schemaName) { + this.schemaName = schemaName; + } + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public String getServerColumnName() { + return serverColumnName; + } + + public void setServerColumnName(String serverColumnName) { + this.serverColumnName = serverColumnName; + } + + public String getVanAutoIncColumnName() { + return vanAutoIncColumnName; + } + + public void setVanAutoIncColumnName(String vanAutoIncColumnName) { + this.vanAutoIncColumnName = vanAutoIncColumnName; + } + + public String getTableType() { + return tableType; + } + + public void setTableType(String tableType) { + this.tableType = tableType; + } + + public Integer getVanID() { + return vanID; + } + + public void setVanID(Integer vanID) { + this.vanID = vanID; + } + + public Integer getProviderServiceMapID() { + return providerServiceMapID; + } + + public void setProviderServiceMapID(Integer providerServiceMapID) { + this.providerServiceMapID = providerServiceMapID; + } + + public Long getLastFetchedID() { + return lastFetchedID; + } + + public void setLastFetchedID(Long lastFetchedID) { + this.lastFetchedID = lastFetchedID; + } + + public Integer getBatchSize() { + return batchSize; + } + + public void setBatchSize(Integer batchSize) { + this.batchSize = batchSize; + } + + public List getRecords() { + return records; + } + + public void setRecords(List records) { + this.records = records; + } + + public boolean isMasterTable() { + return tableType == null || DownSyncTableDetail.TABLE_TYPE_MASTER.equalsIgnoreCase(tableType); + } +} diff --git a/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncRecordAck.java b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncRecordAck.java new file mode 100644 index 00000000..e0de23c2 --- /dev/null +++ b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncRecordAck.java @@ -0,0 +1,104 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.data.syncActivity_syncLayer; + +import com.google.gson.annotations.Expose; + +public class DownSyncRecordAck { + + public static final String STATUS_PROCESSED = "P"; + public static final String STATUS_FAILED = "F"; + public static final String STATUS_RETRY = "U"; + public static final String CONFLICT = "CONFLICT"; + + @Expose + private Long centralID; + @Expose + private Long vanSerialNo; + @Expose + private String status; + @Expose + private String failureReason; + + public DownSyncRecordAck() { + } + + public DownSyncRecordAck(Long centralID, Long vanSerialNo, String status, String failureReason) { + this.centralID = centralID; + this.vanSerialNo = vanSerialNo; + this.status = status; + this.failureReason = failureReason; + } + + public static DownSyncRecordAck success(Long centralID, Long vanSerialNo) { + return new DownSyncRecordAck(centralID, vanSerialNo, STATUS_PROCESSED, null); + } + + public static DownSyncRecordAck failure(Long centralID, Long vanSerialNo, String failureReason) { + return new DownSyncRecordAck(centralID, vanSerialNo, STATUS_FAILED, failureReason); + } + + + public static DownSyncRecordAck retryable(Long centralID, Long vanSerialNo, String failureReason) { + return new DownSyncRecordAck(centralID, vanSerialNo, STATUS_RETRY, failureReason); + } + + public boolean isRetryable() { + return STATUS_RETRY.equalsIgnoreCase(status); + } + + public boolean isSuccess() { + return STATUS_PROCESSED.equalsIgnoreCase(status); + } + + public Long getCentralID() { + return centralID; + } + + public void setCentralID(Long centralID) { + this.centralID = centralID; + } + + public Long getVanSerialNo() { + return vanSerialNo; + } + + public void setVanSerialNo(Long vanSerialNo) { + this.vanSerialNo = vanSerialNo; + } + + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + + public String getFailureReason() { + return failureReason; + } + + public void setFailureReason(String failureReason) { + this.failureReason = failureReason; + } +} diff --git a/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncTableDetail.java b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncTableDetail.java new file mode 100644 index 00000000..1e22fdb6 --- /dev/null +++ b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncTableDetail.java @@ -0,0 +1,215 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.data.syncActivity_syncLayer; + +import com.google.gson.annotations.Expose; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.Table; +import jakarta.persistence.Transient; + +/*** + * Down-sync table configuration (central -> local). Analogous to + * {@link SyncUtilityClass} (m_synctabledetail) which drives the up-sync. + */ +@Entity +@Table(name = "m_downsynctabledetail") +public class DownSyncTableDetail { + + public static final String TABLE_TYPE_MASTER = "MASTER"; + public static final String TABLE_TYPE_TRANSACTIONAL = "TRANSACTIONAL"; + public static final String LAST_MOD_COLUMN = "LastModDate"; + + @Id + @GeneratedValue(strategy = GenerationType.AUTO) + @Expose + @Column(name = "DownSyncTableDetailID", updatable = false) + private Integer downSyncTableDetailID; + @Expose + @Column(name = "SchemaName") + private String schemaName; + @Expose + @Column(name = "TableName") + private String tableName; + /** columns to SELECT from central */ + @Expose + @Column(name = "ServerColumnName") + private String serverColumnName; + /** columns in local, positionally mapped with {@link #serverColumnName} */ + @Expose + @Column(name = "VanColumnName") + private String vanColumnName; + /** local auto-increment PK column - skipped on INSERT */ + @Expose + @Column(name = "VanAutoIncColumnName") + private String vanAutoIncColumnName; + /** MASTER = full pull, no VanID filter / TRANSACTIONAL = VanID + DownSynced filter */ + @Expose + @Column(name = "TableType") + private String tableType; + + @Expose + @Column(name = "FkColumnMapping") + private String fkColumnMapping; + + @Expose + @Column(name = "PreserveCentralPK") + private Boolean preserveCentralPK; + + + @Expose + @Column(name = "SyncOrder") + private Integer syncOrder; + @Expose + @Column(name = "IsActive") + private Boolean isActive; + + @Transient + @Expose + private Integer vanID; + @Transient + @Expose + private Integer providerServiceMapID; + + public DownSyncTableDetail() { + } + + public boolean isMasterTable() { + return tableType == null || TABLE_TYPE_MASTER.equalsIgnoreCase(tableType); + } + + public boolean isTransactionalTable() { + return TABLE_TYPE_TRANSACTIONAL.equalsIgnoreCase(tableType); + } + + public Integer getDownSyncTableDetailID() { + return downSyncTableDetailID; + } + + public void setDownSyncTableDetailID(Integer downSyncTableDetailID) { + this.downSyncTableDetailID = downSyncTableDetailID; + } + + public String getSchemaName() { + return schemaName; + } + + public void setSchemaName(String schemaName) { + this.schemaName = schemaName; + } + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public String getServerColumnName() { + return serverColumnName; + } + + public void setServerColumnName(String serverColumnName) { + this.serverColumnName = serverColumnName; + } + + public String getVanColumnName() { + return vanColumnName; + } + + public void setVanColumnName(String vanColumnName) { + this.vanColumnName = vanColumnName; + } + + public String getLastModColumnName() { + return LAST_MOD_COLUMN; + } + + public String getVanAutoIncColumnName() { + return vanAutoIncColumnName; + } + + public void setVanAutoIncColumnName(String vanAutoIncColumnName) { + this.vanAutoIncColumnName = vanAutoIncColumnName; + } + + public boolean isPreserveCentralPK() { + return Boolean.TRUE.equals(preserveCentralPK); + } + + public void setPreserveCentralPK(Boolean preserveCentralPK) { + this.preserveCentralPK = preserveCentralPK; + } + + public String getFkColumnMapping() { + return fkColumnMapping; + } + + public void setFkColumnMapping(String fkColumnMapping) { + this.fkColumnMapping = fkColumnMapping; + } + + public String getTableType() { + return tableType; + } + + public void setTableType(String tableType) { + this.tableType = tableType; + } + + public Integer getSyncOrder() { + return syncOrder; + } + + public void setSyncOrder(Integer syncOrder) { + this.syncOrder = syncOrder; + } + + public Boolean getIsActive() { + return isActive; + } + + public void setIsActive(Boolean isActive) { + this.isActive = isActive; + } + + public Integer getVanID() { + return vanID; + } + + public void setVanID(Integer vanID) { + this.vanID = vanID; + } + + public Integer getProviderServiceMapID() { + return providerServiceMapID; + } + + public void setProviderServiceMapID(Integer providerServiceMapID) { + this.providerServiceMapID = providerServiceMapID; + } +} diff --git a/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncTableResult.java b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncTableResult.java new file mode 100644 index 00000000..a9691fbb --- /dev/null +++ b/src/main/java/com/iemr/mmu/data/syncActivity_syncLayer/DownSyncTableResult.java @@ -0,0 +1,171 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.data.syncActivity_syncLayer; + +import java.util.LinkedHashMap; +import java.util.Map; + +import com.google.gson.annotations.Expose; + +/*** + * The outcome of down-syncing one table, reported to the screen. + * + *

+ * A table can fail in two quite different ways and the difference matters when + * reading the result: + * + *

    + *
  • the table itself failed - central could not even return the rows (a schema + * mismatch, say), so nothing was delivered and no record was flagged;
    • + *
  • the table was delivered but individual records failed, in which case the + * table shows a record failure count and each distinct reason with a tally.
    • + *
+ */ +public class DownSyncTableResult { + + public static final String STATUS_SUCCESS = "SUCCESS"; + public static final String STATUS_FAILED = "FAILED"; + public static final String STATUS_PARTIAL = "PARTIAL"; + public static final String STATUS_CONFLICT = "CONFLICT"; + + @Expose + private String groupName; + @Expose + private String schemaName; + @Expose + private String tableName; + @Expose + private String status; + @Expose + private int inserted; + @Expose + private int updated; + @Expose + private int skipped; + @Expose + private int conflicts; + @Expose + private int failedRecords; + /** why the table as a whole failed; null when the table itself was fine */ + @Expose + private String failureReason; + /** distinct record-level failure reasons, each with the number of records */ + @Expose + private Map recordFailureReasons = new LinkedHashMap<>(); + private Map recordWarnings = new LinkedHashMap<>(); + + public DownSyncTableResult(String groupName, String schemaName, String tableName) { + this.groupName = groupName; + this.schemaName = schemaName; + this.tableName = tableName; + this.status = STATUS_SUCCESS; + } + + /** the table could not be processed at all - nothing was delivered */ + public void tableFailed(String reason) { + this.status = STATUS_FAILED; + this.failureReason = reason; + } + + /** one record failed; the table itself is fine unless every record failed */ + public void recordFailed(String reason) { + this.failedRecords++; + String key = (reason == null || reason.trim().isEmpty()) ? "Unknown error" : reason.trim(); + recordFailureReasons.merge(key, 1, Integer::sum); + if (STATUS_SUCCESS.equals(this.status)) + this.status = STATUS_PARTIAL; + } + + public void recordWarning(String reason) { + recordWarnings.merge(reason, 1, Integer::sum); + } + + public Map getRecordWarnings() { + return recordWarnings; + } + + public int getRecordsProcessed() { + return inserted + updated + skipped + conflicts + failedRecords; + } + + public void addInserted() { + inserted++; + } + + public void addUpdated() { + updated++; + } + + public void addSkipped() { + skipped++; + } + + public void addConflict() { + conflicts++; + if (STATUS_SUCCESS.equals(this.status)) + this.status = STATUS_CONFLICT; + } + + public String getGroupName() { + return groupName; + } + + public String getSchemaName() { + return schemaName; + } + + public String getTableName() { + return tableName; + } + + public String getStatus() { + return status; + } + + public int getInserted() { + return inserted; + } + + public int getUpdated() { + return updated; + } + + public int getSkipped() { + return skipped; + } + + public int getConflicts() { + return conflicts; + } + + public int getFailedRecords() { + return failedRecords; + } + + public String getFailureReason() { + return failureReason; + } + + public Map getRecordFailureReasons() { + return recordFailureReasons; + } +} diff --git a/src/main/java/com/iemr/mmu/repo/login/UserLoginRepo.java b/src/main/java/com/iemr/mmu/repo/login/UserLoginRepo.java index 6f6390ee..d14e3c3f 100644 --- a/src/main/java/com/iemr/mmu/repo/login/UserLoginRepo.java +++ b/src/main/java/com/iemr/mmu/repo/login/UserLoginRepo.java @@ -15,6 +15,9 @@ public interface UserLoginRepo extends CrudRepository { @Query(" SELECT u FROM Users u WHERE u.userID = :userID AND u.deleted = false ") public Users getUserByUserID(@Param("userID") Long userID); + @Query(" SELECT u FROM Users u WHERE u.userName = :UserName AND u.deleted = false ") + public Users getUserByUsername(@Param("UserName") String username); + @Query(nativeQuery = true,value = "select rolename from m_role where roleid in (select roleid from m_userservicerolemapping where userid=:userID)") List getRoleNamebyUserId(@Param("userID") Long userID); diff --git a/src/main/java/com/iemr/mmu/repo/nurse/BenVisitDetailRepo.java b/src/main/java/com/iemr/mmu/repo/nurse/BenVisitDetailRepo.java index c4a03286..75d075f9 100644 --- a/src/main/java/com/iemr/mmu/repo/nurse/BenVisitDetailRepo.java +++ b/src/main/java/com/iemr/mmu/repo/nurse/BenVisitDetailRepo.java @@ -101,4 +101,16 @@ public int updateFileID(@Param("fileIDs") String fileIDs, @Param("regID") Long r @Query(" UPDATE BeneficiaryVisitDetail set vanSerialNo = :benVisitID WHERE benVisitID = :benVisitID") int updateVanSerialNo(@Param("benVisitID") Long benVisitID); + // store responsible doctor's user ID against the visit + @Transactional + @Modifying + @Query("UPDATE BeneficiaryVisitDetail set doctorID = :doctorID where visitCode = :visitCode ") + public Integer updateDoctorID(@Param("doctorID") Long doctorID, @Param("visitCode") Long visitCode); + + // store responsible lab technician's user ID against the visit + @Transactional + @Modifying + @Query("UPDATE BeneficiaryVisitDetail set labTechnicianID = :labTechnicianID where visitCode = :visitCode ") + public Integer updateLabTechnicianID(@Param("labTechnicianID") Long labTechnicianID, @Param("visitCode") Long visitCode); + } diff --git a/src/main/java/com/iemr/mmu/repo/syncActivity_syncLayer/DownSyncTableDetailRepo.java b/src/main/java/com/iemr/mmu/repo/syncActivity_syncLayer/DownSyncTableDetailRepo.java new file mode 100644 index 00000000..683bb10a --- /dev/null +++ b/src/main/java/com/iemr/mmu/repo/syncActivity_syncLayer/DownSyncTableDetailRepo.java @@ -0,0 +1,41 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.repo.syncActivity_syncLayer; + +import java.util.ArrayList; + +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.CrudRepository; +import org.springframework.data.repository.query.Param; +import org.springframework.stereotype.Repository; + +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncTableDetail; + +@Repository +public interface DownSyncTableDetailRepo extends CrudRepository { + + @Query(" SELECT d FROM DownSyncTableDetail d WHERE d.isActive = true ORDER BY d.syncOrder, d.downSyncTableDetailID ") + ArrayList getActiveDownSyncTables(); + + @Query(" SELECT d FROM DownSyncTableDetail d WHERE d.isActive = true AND lower(d.tableName) = lower(:tableName) ") + ArrayList getActiveDownSyncTableByName(@Param("tableName") String tableName); +} diff --git a/src/main/java/com/iemr/mmu/service/common/transaction/CommonDoctorServiceImpl.java b/src/main/java/com/iemr/mmu/service/common/transaction/CommonDoctorServiceImpl.java index cab7a2d3..7302abc4 100644 --- a/src/main/java/com/iemr/mmu/service/common/transaction/CommonDoctorServiceImpl.java +++ b/src/main/java/com/iemr/mmu/service/common/transaction/CommonDoctorServiceImpl.java @@ -115,6 +115,34 @@ public class CommonDoctorServiceImpl { @Autowired private CookieUtil cookieUtil; + @Autowired + private com.iemr.mmu.repo.nurse.BenVisitDetailRepo benVisitDetailRepo; + @Autowired + private com.iemr.mmu.repo.login.UserLoginRepo userLoginRepo; + + /** + * Resolve the numeric user ID of the responsible staff member from the username + * captured in createdBy. Returns null if it cannot be resolved so an unknown + * staff member never blocks the flow. + */ + private Long resolveUserId(String username) { + if (username == null || username.trim().isEmpty()) + return null; + com.iemr.mmu.data.login.Users user = userLoginRepo.getUserByUsername(username.trim()); + return user != null ? user.getUserID() : null; + } + + /** + * Store the responsible doctor's user ID on the visit record, taken from createdBy. + */ + private void storeDoctorIDOnVisit(CommonUtilityClass commonUtilityClass) { + if (commonUtilityClass == null || commonUtilityClass.getVisitCode() == null) + return; + Long doctorID = resolveUserId(commonUtilityClass.getCreatedBy()); + if (doctorID != null) + benVisitDetailRepo.updateDoctorID(doctorID, commonUtilityClass.getVisitCode()); + } + @Autowired public void setSnomedServiceImpl(SnomedServiceImpl snomedServiceImpl) { this.snomedServiceImpl = snomedServiceImpl; @@ -778,6 +806,9 @@ public int updateBenFlowtableAfterDocDataSave(CommonUtilityClass commonUtilityCl Long tmpBenVisitID = commonUtilityClass.getBenVisitID(); Long tmpbeneficiaryRegID = commonUtilityClass.getBeneficiaryRegID(); + // Store the responsible doctor's user ID against the visit + storeDoctorIDOnVisit(commonUtilityClass); + // checking if test is prescribed if (isTestPrescribed) { docFlag = (short) 2; @@ -840,6 +871,9 @@ public int updateBenFlowtableAfterDocDataUpdate(CommonUtilityClass commonUtility Long tmpBenVisitID = commonUtilityClass.getBenVisitID(); Long tmpbeneficiaryRegID = commonUtilityClass.getBeneficiaryRegID(); + // Store the responsible doctor's user ID against the visit + storeDoctorIDOnVisit(commonUtilityClass); + if (commonUtilityClass.getIsSpecialist() != null && commonUtilityClass.getIsSpecialist() == true) { if (isTestPrescribed) tcSpecialistFlag = (short) 2; diff --git a/src/main/java/com/iemr/mmu/service/common/transaction/CommonNurseServiceImpl.java b/src/main/java/com/iemr/mmu/service/common/transaction/CommonNurseServiceImpl.java index 72e3c8f2..a7efa9d4 100644 --- a/src/main/java/com/iemr/mmu/service/common/transaction/CommonNurseServiceImpl.java +++ b/src/main/java/com/iemr/mmu/service/common/transaction/CommonNurseServiceImpl.java @@ -145,6 +145,22 @@ public class CommonNurseServiceImpl implements CommonNurseService { private Integer TMReferredWL; @Autowired private BenVisitDetailRepo benVisitDetailRepo; + + @Autowired + private com.iemr.mmu.repo.login.UserLoginRepo userLoginRepo; + + /** + * Resolve the numeric user ID of the responsible staff member from the username + * captured in createdBy. Returns null if it cannot be resolved so an unknown + * staff member never blocks the save. + */ + private Long resolveUserId(String username) { + if (username == null || username.trim().isEmpty()) + return null; + com.iemr.mmu.data.login.Users user = userLoginRepo.getUserByUsername(username.trim()); + return user != null ? user.getUserID() : null; + } + @Autowired private BenChiefComplaintRepo benChiefComplaintRepo; @Autowired @@ -258,6 +274,10 @@ public Long saveBeneficiaryVisitDetails(BeneficiaryVisitDetail beneficiaryVisitD } beneficiaryVisitDetail.setReportFilePath(sb.toString()); + // Store the responsible nurse's user ID (resolved from the createdBy username) + if (beneficiaryVisitDetail.getNurseID() == null) + beneficiaryVisitDetail.setNurseID(resolveUserId(beneficiaryVisitDetail.getCreatedBy())); + response = benVisitDetailRepo.save(beneficiaryVisitDetail); benVisitDetailRepo.updateVanSerialNo(response.getBenVisitID()); diff --git a/src/main/java/com/iemr/mmu/service/dataSyncActivity/DataSyncRepository.java b/src/main/java/com/iemr/mmu/service/dataSyncActivity/DataSyncRepository.java index dc36c7c1..75a74842 100644 --- a/src/main/java/com/iemr/mmu/service/dataSyncActivity/DataSyncRepository.java +++ b/src/main/java/com/iemr/mmu/service/dataSyncActivity/DataSyncRepository.java @@ -23,9 +23,12 @@ import java.sql.Timestamp; import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.sql.PreparedStatement; + import javax.sql.DataSource; import org.slf4j.Logger; @@ -33,9 +36,12 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.annotation.CreatedDate; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.support.GeneratedKeyHolder; +import org.springframework.jdbc.support.KeyHolder; import org.springframework.stereotype.Service; import com.iemr.mmu.repo.syncActivity_syncLayer.SyncUtilityClassRepo; +import com.iemr.mmu.utils.validator.SqlIdentifierValidator; /*** * @@ -137,4 +143,239 @@ public int[] updateLatestMasterInLocal(String query, List syncDataList // ---------------------------------- End of Download Repository + // ---------------------------------- Down-Sync Repository (central -> local) + + public Map getSyncGroupNamesByTable() { + jdbcTemplate = getJdbcTemplate(); + + String query = " SELECT LOWER(TRIM(s.TableName)) AS TableName, g.SyncTableGroupName " + + " FROM db_iemr.m_synctabledetail s " + + " JOIN db_iemr.m_synctablegroup g ON g.SyncTableGroupID = s.SyncTableGroupID " + + " WHERE IFNULL(s.Deleted, b'0') = b'0' AND IFNULL(g.Deleted, b'0') = b'0' " + + " AND s.TableName IS NOT NULL "; + + Map groups = new LinkedHashMap<>(); + for (Map row : jdbcTemplate.queryForList(query)) { + Object table = row.get("TableName"); + Object group = row.get("SyncTableGroupName"); + if (table != null && group != null) + groups.putIfAbsent(String.valueOf(table), String.valueOf(group)); + } + return groups; + } + + public String resolveLastModColumn(String schema, String table) { + jdbcTemplate = getJdbcTemplate(); + + String query = " SELECT COLUMN_NAME FROM information_schema.COLUMNS " + + " WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " + + " AND COLUMN_NAME IN ('LastModDate', 'last_mod_date') " + + " ORDER BY FIELD(COLUMN_NAME, 'LastModDate', 'last_mod_date') "; + + List found = jdbcTemplate.queryForList(query, String.class, schema, table); + return (found == null || found.isEmpty()) ? null : found.get(0); + } + + public List getDownSyncColumns(String schema, String table) { + jdbcTemplate = getJdbcTemplate(); + + String query = " SELECT COLUMN_NAME FROM information_schema.COLUMNS " + + " WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " + + " AND COLUMN_NAME NOT IN ('CentralID', 'DownSynced', 'DownSyncDate', 'DownSyncFailureReason', " + + " 'LastDownSyncDate', 'Processed', 'SyncFailureReason', 'SyncedBy', 'SyncedDate') " + + " ORDER BY COLUMN_NAME "; + + return jdbcTemplate.queryForList(query, String.class, schema, table); + } + + public Map getLocalRecordForDownSync(String schema, String table, String autoIncColumnName, + Object centralID, Object vanID, String lastModColumn, boolean preserveCentralPK) { + if (centralID == null) + return null; + + jdbcTemplate = getJdbcTemplate(); + // schema, table & column names cannot be bound as query parameters, so each of + // them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String validPkColumn = SqlIdentifierValidator.validatedColumnName(autoIncColumnName); + String validLastModColumn = SqlIdentifierValidator.validatedColumnName(lastModColumn); + + StringBuilder query = new StringBuilder(" SELECT ").append(validPkColumn).append(", CentralID, Processed, ") + .append(validLastModColumn).append(" AS LastModDate, LastDownSyncDate FROM ").append(validSchema) + .append(".").append(validTable); + + List params = new ArrayList<>(); + + if (preserveCentralPK) { + query.append(" WHERE ( CentralID = ? OR ").append(validPkColumn).append(" = ? ) "); + params.add(centralID); + params.add(centralID); + } else { + query.append(" WHERE CentralID = ? AND VanID = ? "); + params.add(centralID); + params.add(vanID); + } + + List> resultSet = jdbcTemplate.queryForList(query.toString(), params.toArray()); + if (resultSet == null || resultSet.isEmpty()) + return null; + + return resultSet.get(0); + } + + /*** + * Stamps the central key on a local row that was matched by its primary key, so + * the later runs of the down-sync find it on the CentralID match itself. + */ + public int stampCentralIDInLocal(String schema, String table, String autoIncColumnName, Object localID, + Object centralID) { + if (localID == null || centralID == null) + return 0; + + jdbcTemplate = getJdbcTemplate(); + // schema, table & column names cannot be bound as query parameters, so each of + // them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String validPkColumn = SqlIdentifierValidator.validatedColumnName(autoIncColumnName); + + String query = " UPDATE " + validSchema + "." + validTable + " SET CentralID = ? WHERE " + validPkColumn + + " = ? AND CentralID IS NULL "; + + return jdbcTemplate.update(query, centralID, localID); + } + + public Long insertDownSyncRecordInLocal(final String query, final Object[] params) { + jdbcTemplate = getJdbcTemplate(); + KeyHolder keyHolder = new GeneratedKeyHolder(); + + jdbcTemplate.update(connection -> { + PreparedStatement ps = connection.prepareStatement(query, PreparedStatement.RETURN_GENERATED_KEYS); + for (int i = 0; i < params.length; i++) { + ps.setObject(i + 1, params[i]); + } + return ps; + }, keyHolder); + + Number key = keyHolder.getKey(); + return key != null ? key.longValue() : null; + } + + public int updateDownSyncRecordInLocal(String query, Object[] params) { + jdbcTemplate = getJdbcTemplate(); + return jdbcTemplate.update(query, params); + } + + public int updateVanSerialNoInLocal(String schema, String table, String autoIncColumnName, Object localID) { + jdbcTemplate = getJdbcTemplate(); + // schema, table & column names cannot be bound as query parameters, so each of + // them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String validPkColumn = SqlIdentifierValidator.validatedColumnName(autoIncColumnName); + + String query = " UPDATE " + validSchema + "." + validTable + " SET VanSerialNo = ? WHERE " + + validPkColumn + " = ? "; + return jdbcTemplate.update(query, localID, localID); + } + + + public Long resolveLocalIdForCentralValue(String schema, String table, String pkColumn, Object centralValue, + Object vanID) { + if (centralValue == null) + return null; + + jdbcTemplate = getJdbcTemplate(); + // schema, table & column names cannot be bound as query parameters, so each of + // them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String validPkColumn = SqlIdentifierValidator.validatedColumnName(pkColumn); + String from = " FROM " + validSchema + "." + validTable + " WHERE "; + + Long resolved = queryForFirstLong(" SELECT " + validPkColumn + from + " CentralID = ? AND VanID = ? ", + centralValue, vanID); + if (resolved != null) + return resolved; + + resolved = queryForFirstLong(" SELECT " + validPkColumn + from + " VanSerialNo = ? AND VanID = ? ", + centralValue, vanID); + if (resolved != null) + return resolved; + + return queryForFirstLong(" SELECT " + validPkColumn + from + validPkColumn + " = ? AND VanID = ? ", + centralValue, vanID); + } + + public int countConflictsInLocal(String schema, String table, Object vanID) { + jdbcTemplate = getJdbcTemplate(); + // schema & table cannot be bound as query parameters, so both are validated + // before they are concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + + String query = " SELECT COUNT(*) FROM " + validSchema + "." + validTable + + " WHERE VanID = ? AND Processed = 'F' AND SyncFailureReason = 'CONFLICT' "; + Integer count = jdbcTemplate.queryForObject(query, Integer.class, vanID); + return count == null ? 0 : count; + } + + public List> getForeignKeysOfTable(String schema, String table) { + jdbcTemplate = getJdbcTemplate(); + String query = " SELECT k.CONSTRAINT_NAME, k.COLUMN_NAME AS CHILD_COLUMN," + + " k.REFERENCED_TABLE_SCHEMA AS PARENT_SCHEMA, k.REFERENCED_TABLE_NAME AS PARENT_TABLE," + + " k.REFERENCED_COLUMN_NAME AS PARENT_COLUMN " + + " FROM information_schema.KEY_COLUMN_USAGE k " + + " WHERE k.TABLE_SCHEMA = ? AND k.TABLE_NAME = ? AND k.REFERENCED_TABLE_NAME IS NOT NULL "; + + List> foreignKeys = jdbcTemplate.queryForList(query, schema, table); + if (foreignKeys == null) + return new ArrayList<>(); + + Map columnsPerConstraint = new LinkedHashMap<>(); + for (Map foreignKey : foreignKeys) { + columnsPerConstraint.merge(String.valueOf(foreignKey.get("CONSTRAINT_NAME")), 1, Integer::sum); + } + + List> singleColumn = new ArrayList<>(); + for (Map foreignKey : foreignKeys) { + if (columnsPerConstraint.get(String.valueOf(foreignKey.get("CONSTRAINT_NAME"))) == 1) + singleColumn.add(foreignKey); + } + return singleColumn; + } + + public boolean isColumnNullable(String schema, String table, String column) { + jdbcTemplate = getJdbcTemplate(); + List nullable = jdbcTemplate.queryForList( + " SELECT IS_NULLABLE FROM information_schema.COLUMNS " + + " WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_NAME = ? ", + String.class, schema, table, column); + return !nullable.isEmpty() && "YES".equalsIgnoreCase(nullable.get(0)); + } + + private Long queryForFirstLong(String query, Object... params) { + List found = jdbcTemplate.queryForList(query, Long.class, params); + return (found == null || found.isEmpty()) ? null : found.get(0); + } + + public int markDownSyncConflictInLocal(String schema, String table, String autoIncColumnName, Object localID) { + jdbcTemplate = getJdbcTemplate(); + // schema, table & column names cannot be bound as query parameters, so each of + // them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String validPkColumn = SqlIdentifierValidator.validatedColumnName(autoIncColumnName); + + String query = " UPDATE " + validSchema + "." + validTable + + " SET Processed = 'F', SyncFailureReason = 'CONFLICT'," + + " DownSynced = 'F', DownSyncFailureReason = 'CONFLICT' WHERE " + validPkColumn + + " = ? "; + return jdbcTemplate.update(query, localID); + } + + // End of Down-Sync Repository + + } \ No newline at end of file diff --git a/src/main/java/com/iemr/mmu/service/dataSyncActivity/DownSyncDataFromServer.java b/src/main/java/com/iemr/mmu/service/dataSyncActivity/DownSyncDataFromServer.java new file mode 100644 index 00000000..3fcabcf8 --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncActivity/DownSyncDataFromServer.java @@ -0,0 +1,32 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.service.dataSyncActivity; + +import java.util.Map; + +public interface DownSyncDataFromServer { + + String startDownSync(String serverAuthorization, String jwtToken, Integer vanID, Integer providerServiceMapID) + throws Exception; + + Map getDownSyncStatus(); +} diff --git a/src/main/java/com/iemr/mmu/service/dataSyncActivity/DownSyncDataFromServerImpl.java b/src/main/java/com/iemr/mmu/service/dataSyncActivity/DownSyncDataFromServerImpl.java new file mode 100644 index 00000000..c3c1e1d0 --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncActivity/DownSyncDataFromServerImpl.java @@ -0,0 +1,946 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.service.dataSyncActivity; + +import java.sql.Timestamp; +import java.text.DateFormat; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.json.JSONObject; +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.context.annotation.PropertySource; +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.GsonBuilder; +import com.google.gson.reflect.TypeToken; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncDataDigester; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncRecordAck; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncTableDetail; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncTableResult; +import com.iemr.mmu.repo.syncActivity_syncLayer.DownSyncTableDetailRepo; +import com.iemr.mmu.utils.RestTemplateUtil; + + +@Service +@PropertySource("classpath:application.properties") +public class DownSyncDataFromServerImpl implements DownSyncDataFromServer { + + private Logger logger = LoggerFactory.getLogger(this.getClass().getSimpleName()); + + private static final String VAN_SERIAL_NO = "VanSerialNo"; + private static final String VAN_ID = "VanID"; + private static final String LAST_MOD_DATE = "LastModDate"; + private static final String DATA_SYNC_CALL = "datasync"; + private static final String LAST_DOWN_SYNC_DATE = "LastDownSyncDate"; + private static final String DOWN_SYNCED = "DownSynced"; + private static final String DOWN_SYNC_DATE = "DownSyncDate"; + private static final String DOWN_SYNC_FAILURE_REASON = "DownSyncFailureReason"; + private static final String PROCESSED = "Processed"; + private static final String SYNC_FAILURE_REASON = "SyncFailureReason"; + private static final String CENTRAL_ID = "CentralID"; + private static final String PROVISIONED = "Provisioned"; + private static final String RESERVED = "Reserved"; + private static final String RESERVED_FOR = "ReservedFor"; + private static final String RESERVED_BY_ID = "ReservedById"; + private static final String RESERVED_ON = "ReservedOn"; + + private static final int ACK_BATCH_SIZE = 500; + + private static final int FETCH_BATCH_SIZE = 1000; + + @Value("${downSyncDataUrl}") + private String downSyncDataUrl; + + @Value("${downSyncFlagUpdateUrl}") + private String downSyncFlagUpdateUrl; + + @Autowired + private DownSyncTableDetailRepo downSyncTableDetailRepo; + + @Autowired + private DataSyncRepository dataSyncRepository; + + private static final AtomicBoolean IN_PROGRESS = new AtomicBoolean(false); + private static int totalCounter = 0; + private static int progressCounter = 0; + private static int insertedCounter = 0; + private static int updatedCounter = 0; + private static int transactionalFetchedCounter = 0; + private static int masterUpdatedCounter = 0; + private static int conflictCounter = 0; + private static int skippedCounter = 0; + private static int failedTableCounter = 0; + private static int failedRecordCounter = 0; + private static StringBuilder failedTables = new StringBuilder(); + private static final List tableResults = + Collections.synchronizedList(new ArrayList<>()); + private static DownSyncTableResult currentResult; + private static String currentTable = ""; + private static int outstandingConflictCounter = 0; + private static final Map outstandingConflictTables = new LinkedHashMap<>(); + + public String startDownSync(String serverAuthorization, String jwtToken, Integer vanID, + Integer providerServiceMapID) throws Exception { + + if (vanID == null) + throw new Exception("vanID is mandatory for down-sync. Kindly contact the administrator."); + + final ArrayList downSyncTables = downSyncTableDetailRepo.getActiveDownSyncTables(); + if (downSyncTables == null || downSyncTables.isEmpty()) + throw new Exception("No table is configured for down-sync in m_downsynctabledetail."); + + if (!IN_PROGRESS.compareAndSet(false, true)) + return "inProgress"; + + totalCounter = downSyncTables.size(); + progressCounter = 0; + insertedCounter = 0; + updatedCounter = 0; + transactionalFetchedCounter = 0; + masterUpdatedCounter = 0; + conflictCounter = 0; + skippedCounter = 0; + failedTableCounter = 0; + failedRecordCounter = 0; + failedTables = new StringBuilder(); + tableResults.clear(); + currentResult = null; + currentTable = ""; + + final Map syncGroups = dataSyncRepository.getSyncGroupNamesByTable(); + + try { + for (DownSyncTableDetail tableDetail : downSyncTables) { + currentTable = tableDetail.getSchemaName() + "." + tableDetail.getTableName(); + + String groupName = syncGroups.getOrDefault(tableDetail.getTableName().toLowerCase(), + tableDetail.isMasterTable() ? "Masters" : "Other"); + currentResult = new DownSyncTableResult(groupName, tableDetail.getSchemaName(), + tableDetail.getTableName()); + tableResults.add(currentResult); + + try { + downSyncTable(tableDetail, vanID, providerServiceMapID, serverAuthorization, jwtToken); + } catch (Exception e) { + failedTableCounter++; + failedTables.append(tableDetail.getTableName()).append(" | "); + currentResult.tableFailed(shorten(e.getMessage())); + logger.error("Down-sync failed for " + currentTable + ". Exception : " + e.getMessage(), e); + } + progressCounter++; + } + countOutstandingConflicts(downSyncTables, vanID); + } finally { + IN_PROGRESS.set(false); + currentTable = ""; + } + + logger.info("Down-sync finished : {} of {} tables succeeded, {} records delivered for van {}, {} inserted," + + " {} updated, {} skipped, {} conflicts, {} failed records, {} master rows refreshed", + progressCounter - failedTableCounter, totalCounter, transactionalFetchedCounter, vanID, + insertedCounter, updatedCounter, skippedCounter, conflictCounter, failedRecordCounter, + masterUpdatedCounter); + + return buildSummary(); + } + + public Map getDownSyncStatus() { + Map resultMap = new LinkedHashMap<>(); + resultMap.put("percentage", totalCounter == 0 ? 0 : Math.floor((progressCounter * 100.0) / totalCounter)); + resultMap.put("inProgress", IN_PROGRESS.get()); + resultMap.put("currentTable", currentTable); + resultMap.put("totalTables", totalCounter); + resultMap.put("completedTables", progressCounter); + resultMap.put("successTableCount", progressCounter - failedTableCounter); + resultMap.put("failedTableCount", failedTableCounter); + resultMap.put("failedRecordCount", failedRecordCounter); + resultMap.put("failedTables", failedTables.toString()); + synchronized (tableResults) { + resultMap.put("tableResults", new ArrayList<>(tableResults)); + } + resultMap.put("recordsInserted", insertedCounter); + resultMap.put("recordsUpdated", updatedCounter); + resultMap.put("recordsSkipped", skippedCounter); + resultMap.put("conflicts", conflictCounter); + resultMap.put("outstandingConflicts", outstandingConflictCounter); + synchronized (outstandingConflictTables) { + resultMap.put("outstandingConflictTables", new LinkedHashMap<>(outstandingConflictTables)); + } + resultMap.put("conflictsPending", conflictCounter > 0 || outstandingConflictCounter > 0); + resultMap.put("recordsFoundForVan", transactionalFetchedCounter); + resultMap.put("summary", buildSummary()); + resultMap.put("message", buildSummary()); + return resultMap; + } + + private void countOutstandingConflicts(List downSyncTables, Integer vanID) { + outstandingConflictCounter = 0; + outstandingConflictTables.clear(); + + for (DownSyncTableDetail tableDetail : downSyncTables) { + if (!tableDetail.isTransactionalTable()) + continue; + try { + int conflicts = dataSyncRepository.countConflictsInLocal(tableDetail.getSchemaName(), + tableDetail.getTableName(), vanID); + if (conflicts > 0) { + outstandingConflictCounter += conflicts; + outstandingConflictTables.put(tableDetail.getTableName(), conflicts); + } + } catch (Exception e) { + logger.warn("Could not count outstanding conflicts for {}.{} : {}", tableDetail.getSchemaName(), + tableDetail.getTableName(), e.getMessage()); + } + } + } + + private String buildSummary() { + StringBuilder summary = new StringBuilder(); + + if (transactionalFetchedCounter == 0 && failedTableCounter == 0) { + summary.append("There is no record to down-sync for your van"); + if (outstandingConflictCounter > 0) + summary.append(". ").append(outstandingConflictCounter) + .append(outstandingConflictCounter == 1 ? " record is in conflict" + : " records are in conflict") + .append(" and needs review before it can sync"); + return summary.toString(); + } + + summary.append(progressCounter - failedTableCounter).append(" of ").append(totalCounter) + .append(" tables synced, ").append(insertedCounter).append(" inserted, ").append(updatedCounter) + .append(" updated, ").append(skippedCounter).append(" unchanged"); + + if (failedTableCounter > 0) + summary.append(". ").append(failedTableCounter).append(" table(s) failed"); + if (failedRecordCounter > 0) + summary.append(". ").append(failedRecordCounter).append(" record(s) failed"); + if (conflictCounter > 0) + summary.append(". ").append(conflictCounter) + .append(conflictCounter == 1 ? " new conflict" : " new conflicts"); + if (outstandingConflictCounter > 0) + summary.append(". ").append(outstandingConflictCounter) + .append(outstandingConflictCounter == 1 ? " record is in conflict" : " records are in conflict") + .append(" and needs review before it can sync"); + + return summary.toString(); + } + + /*** + * Down-syncs one configured table. + */ + private void downSyncTable(DownSyncTableDetail tableDetail, Integer vanID, Integer providerServiceMapID, + String serverAuthorization, String jwtToken) throws Exception { + + List serverColumns = splitColumns(tableDetail.getServerColumnName()); + List vanColumns = splitColumns(tableDetail.getVanColumnName()); + + if (serverColumns.isEmpty() && vanColumns.isEmpty()) { + List resolved = dataSyncRepository.getDownSyncColumns(tableDetail.getSchemaName(), + tableDetail.getTableName()); + if (resolved.isEmpty()) + throw new Exception("No column found locally for " + tableDetail.getSchemaName() + "." + + tableDetail.getTableName() + + ". The table is configured for down-sync but does not exist in the local DB."); + + serverColumns = resolved; + vanColumns = resolved; + } + + if (serverColumns.isEmpty() || vanColumns.isEmpty()) + throw new Exception("Only one of ServerColumnName / VanColumnName is configured for " + + tableDetail.getTableName() + ". Configure both, with the same columns in the same order, " + + "or leave both empty to have them resolved from the schema."); + if (serverColumns.size() != vanColumns.size()) + throw new Exception("ServerColumnName & VanColumnName column count does not match for " + + tableDetail.getTableName() + " (" + serverColumns.size() + " vs " + vanColumns.size() + + "). Central and local must expose the same number of columns, in the same order."); + + String columnList = String.join(",", serverColumns); + + if (tableDetail.isMasterTable()) { + // masters are a full pull with no primary key to page on, and are small + List> dataFromCentral = fetchDataFromCentral(tableDetail, columnList, vanID, + providerServiceMapID, serverAuthorization, jwtToken, null, null); + + if (dataFromCentral.isEmpty()) { + logger.info("Nothing to down-sync for {}.{}", tableDetail.getSchemaName(), + tableDetail.getTableName()); + return; + } + upsertMasterData(tableDetail, serverColumns, vanColumns, dataFromCentral); + return; + } + + String pkColumn = tableDetail.getVanAutoIncColumnName(); + requireIdentifier(pkColumn == null ? null : pkColumn.trim(), "VanAutoIncColumnName", tableDetail); + + Long lastFetchedID = null; + int page = 0; + int totalFetched = 0; + + while (true) { + List> dataFromCentral = fetchDataFromCentral(tableDetail, columnList, vanID, + providerServiceMapID, serverAuthorization, jwtToken, lastFetchedID, FETCH_BATCH_SIZE); + + if (dataFromCentral.isEmpty()) + break; + + page++; + totalFetched += dataFromCentral.size(); + transactionalFetchedCounter += dataFromCentral.size(); + + List acks = saveTransactionalData(tableDetail, serverColumns, vanColumns, + dataFromCentral, vanID); + acknowledgeToCentral(tableDetail, vanID, acks, serverAuthorization, jwtToken); + + Long highestID = highestCentralID(dataFromCentral, pkColumn.trim()); + if (highestID == null || (lastFetchedID != null && highestID <= lastFetchedID)) + break; + lastFetchedID = highestID; + + if (dataFromCentral.size() < FETCH_BATCH_SIZE) + break; + } + + if (totalFetched == 0) + logger.info("Nothing to down-sync for {}.{}", tableDetail.getSchemaName(), tableDetail.getTableName()); + else + logger.info("Down-synced {} records of {}.{} in {} page(s)", totalFetched, tableDetail.getSchemaName(), + tableDetail.getTableName(), page); + } + + private Long highestCentralID(List> records, String pkColumn) { + Long highest = null; + for (Map record : records) { + Long id = toLong(record.get(resolveKeyIgnoringCase(record, pkColumn))); + if (id != null && (highest == null || id > highest)) + highest = id; + } + return highest; + } + + private List> fetchDataFromCentral(DownSyncTableDetail tableDetail, String serverColumnName, + Integer vanID, Integer providerServiceMapID, String serverAuthorization, String jwtToken, + Long lastFetchedID, Integer batchSize) throws Exception { + + DownSyncDataDigester digester = DownSyncDataDigester.forDownload(tableDetail, serverColumnName, vanID, + providerServiceMapID, lastFetchedID, batchSize); + + RestTemplate restTemplate = new RestTemplate(); + HttpEntity request = RestTemplateUtil.createRequestEntity(digester, serverAuthorization, + DATA_SYNC_CALL); + ResponseEntity response = restTemplate.exchange(downSyncDataUrl, HttpMethod.POST, request, String.class); + + if (response == null || !response.hasBody()) + throw new Exception("Empty response from central for " + tableDetail.getTableName()); + + JSONObject responseObj = new JSONObject(response.getBody()); + if (!responseObj.has("data") || !responseObj.has("statusCode") || responseObj.getInt("statusCode") != 200) + throw new Exception("Down-sync API failed for " + tableDetail.getTableName() + " : " + + responseObj.optString("errorMessage")); + + Gson gson = new GsonBuilder().serializeNulls().create(); + List> data = gson.fromJson(responseObj.get("data").toString(), + new TypeToken>>() { + }.getType()); + + return data != null ? data : new ArrayList<>(); + } + + /*** + * MASTER tables : central authoritative, upserted in one batch. + */ + private void upsertMasterData(DownSyncTableDetail tableDetail, List serverColumns, List vanColumns, + List> dataFromCentral) { + + StringBuilder placeHolders = new StringBuilder(); + StringBuilder onDuplicate = new StringBuilder(); + for (int i = 0; i < vanColumns.size(); i++) { + if (i > 0) { + placeHolders.append(", "); + onDuplicate.append(", "); + } + placeHolders.append("?"); + onDuplicate.append(vanColumns.get(i)).append(" = VALUES(").append(vanColumns.get(i)).append(")"); + } + + String query = " INSERT INTO " + tableDetail.getSchemaName() + "." + tableDetail.getTableName() + " ( " + + String.join(", ", vanColumns) + " ) VALUES ( " + placeHolders + " ) ON DUPLICATE KEY UPDATE " + + onDuplicate; + + List batch = new ArrayList<>(); + for (Map record : dataFromCentral) { + Object[] values = new Object[serverColumns.size()]; + for (int i = 0; i < serverColumns.size(); i++) { + values[i] = normalize(record.get(serverColumns.get(i))); + } + batch.add(values); + } + + int[] result = dataSyncRepository.updateLatestMasterInLocal(query, batch); + masterUpdatedCounter += result != null ? result.length : 0; + logger.info("Down-synced {} master records into {}.{}", batch.size(), tableDetail.getSchemaName(), + tableDetail.getTableName()); + } + + + private List saveTransactionalData(DownSyncTableDetail tableDetail, List serverColumns, + List vanColumns, List> dataFromCentral, Integer vanID) throws Exception { + + String pkColumn = tableDetail.getVanAutoIncColumnName(); + if (pkColumn == null || pkColumn.trim().isEmpty()) + throw new Exception( + "VanAutoIncColumnName is not configured for transactional table " + tableDetail.getTableName()); + pkColumn = pkColumn.trim(); + + requireIdentifier(pkColumn, "VanAutoIncColumnName", tableDetail); + + String lastModColumn = dataSyncRepository.resolveLastModColumn(tableDetail.getSchemaName(), + tableDetail.getTableName()); + if (lastModColumn == null) + throw new Exception(tableDetail.getSchemaName() + "." + tableDetail.getTableName() + + " has neither LastModDate nor last_mod_date, so a change cannot be dated"); + requireIdentifier(lastModColumn, "modification-time column", tableDetail); + + Map fkMapping = resolveFkColumnMapping(tableDetail); + Map fkCache = new LinkedHashMap<>(); + + List acks = new ArrayList<>(); + + for (Map incoming : dataFromCentral) { + Map record = incoming; + Long centralID = toLong(record.get(resolveKeyIgnoringCase(record, pkColumn))); + + try { + if (centralID == null) + throw new Exception("Central sent no value for the primary key '" + pkColumn + + "'. Check VanAutoIncColumnName in m_downsynctabledetail against the real column of " + + tableDetail.getTableName() + " - writing the record without it inserts a duplicate on" + + " every later run."); + + Map localRecord = dataSyncRepository.getLocalRecordForDownSync( + tableDetail.getSchemaName(), tableDetail.getTableName(), pkColumn, centralID, vanID, + lastModColumn, tableDetail.isPreserveCentralPK()); + + // a pointer at another table's primary key means a different number here + // than at central, so it is rewritten before the row is written + StringBuilder unresolved = new StringBuilder(); + Map mapped = translateForeignKeys(tableDetail, fkMapping, record, vanID, fkCache, + unresolved); + + if (localRecord == null) { + if (mapped == null) + throw new Exception("Cannot place this record : " + shorten(unresolved.toString()) + + " Its parent row has not reached this van yet."); + + Long localID = insertRecord(tableDetail, serverColumns, vanColumns, pkColumn, mapped, centralID); + insertedCounter++; + if (currentResult != null) + currentResult.addInserted(); + acks.add(DownSyncRecordAck.success(centralID, localID)); + continue; + } + + Long localID = toLong(localRecord.get(pkColumn)); + + // a row matched by its primary key alone carries no central key yet + if (localRecord.get(CENTRAL_ID) == null) + dataSyncRepository.stampCentralIDInLocal(tableDetail.getSchemaName(), tableDetail.getTableName(), + pkColumn, localID, centralID); + + if (!isCentralCopyNewer(record, localRecord, lastModColumn)) { + skippedCounter++; + if (currentResult != null) + currentResult.addSkipped(); + acks.add(DownSyncRecordAck.success(centralID, localID)); + continue; + } + + if (hasUnsentLocalChanges(localRecord)) { + dataSyncRepository.markDownSyncConflictInLocal(tableDetail.getSchemaName(), + tableDetail.getTableName(), pkColumn, localID); + conflictCounter++; + if (currentResult != null) + currentResult.addConflict(); + acks.add(DownSyncRecordAck.failure(centralID, localID, DownSyncRecordAck.CONFLICT)); + logger.warn("Down-sync conflict for {}.{}, local id {}", tableDetail.getSchemaName(), + tableDetail.getTableName(), localID); + continue; + } + + // on an update an unresolvable pointer is left at its current local value + // rather than being overwritten with an id that means nothing here + updateRecord(tableDetail, serverColumns, vanColumns, pkColumn, mapped != null ? mapped : record, + localID, mapped == null ? fkMapping.keySet() : null); + updatedCounter++; + if (currentResult != null) + currentResult.addUpdated(); + acks.add(DownSyncRecordAck.success(centralID, localID)); + + } catch (Exception e) { + failedRecordCounter++; + String reason = shorten(e.getMessage()); + if (currentResult != null) + currentResult.recordFailed(reason); + acks.add(DownSyncRecordAck.failure(centralID, null, reason)); + logger.error("Down-sync failed for record " + centralID + " of " + tableDetail.getTableName() + " : " + + e.getMessage(), e); + } + } + + return acks; + } + + private boolean isCentralCopyNewer(Map centralRecord, Map localRecord, + String lastModColumn) { + Timestamp centralLastModDate = toTimestamp(centralRecord.get(lastModColumn)); + if (centralLastModDate == null) + return false; + + Timestamp localLastModDate = toTimestamp(localRecord.get(LAST_MOD_DATE)); + if (localLastModDate != null) + return centralLastModDate.after(localLastModDate); + + Timestamp localLastDownSyncDate = toTimestamp(localRecord.get(LAST_DOWN_SYNC_DATE)); + return localLastDownSyncDate == null || centralLastModDate.after(localLastDownSyncDate); + } + + private boolean hasUnsentLocalChanges(Map localRecord) { + String localProcessed = localRecord.get(PROCESSED) != null ? String.valueOf(localRecord.get(PROCESSED)) : null; + return "N".equalsIgnoreCase(localProcessed); + } + + private Long insertRecord(DownSyncTableDetail tableDetail, List serverColumns, List vanColumns, + String pkColumn, Map record, Long centralID) { + + List columns = new ArrayList<>(); + List values = new ArrayList<>(); + + boolean preservePK = tableDetail.isPreserveCentralPK(); + + for (int i = 0; i < vanColumns.size(); i++) { + String vanColumn = vanColumns.get(i); + if ((vanColumn.equalsIgnoreCase(pkColumn) && !preservePK) || isDownSyncManagedColumn(vanColumn)) + continue; + + columns.add(vanColumn); + values.add(normalize(record.get(serverColumns.get(i)))); + } + + columns.add(CENTRAL_ID); + values.add(centralID); + columns.add(PROCESSED); + values.add("P"); + columns.add(SYNC_FAILURE_REASON); + values.add(null); + columns.add(LAST_DOWN_SYNC_DATE); + values.add(Timestamp.valueOf(LocalDateTime.now())); + columns.add(DOWN_SYNCED); + values.add(DownSyncRecordAck.STATUS_PROCESSED); + columns.add(DOWN_SYNC_DATE); + values.add(Timestamp.valueOf(LocalDateTime.now())); + + StringBuilder placeHolders = new StringBuilder(); + for (int i = 0; i < columns.size(); i++) { + if (i > 0) + placeHolders.append(", "); + placeHolders.append("?"); + } + + String query = " INSERT INTO " + tableDetail.getSchemaName() + "." + tableDetail.getTableName() + " ( " + + String.join(", ", columns) + " ) VALUES ( " + placeHolders + " ) "; + + Long localID = dataSyncRepository.insertDownSyncRecordInLocal(query, values.toArray()); + + if (preservePK && localID == null) + localID = centralID; + + if (localID == null && preservePK) + localID = centralID; + + if (localID != null) + dataSyncRepository.updateVanSerialNoInLocal(tableDetail.getSchemaName(), tableDetail.getTableName(), + pkColumn, localID); + + return localID; + } + + private void updateRecord(DownSyncTableDetail tableDetail, List serverColumns, List vanColumns, + String pkColumn, Map record, Long localID, java.util.Set columnsToLeaveAlone) { + + StringBuilder setClause = new StringBuilder(); + List values = new ArrayList<>(); + + for (int i = 0; i < vanColumns.size(); i++) { + String vanColumn = vanColumns.get(i); + if (vanColumn.equalsIgnoreCase(pkColumn) || isDownSyncManagedColumn(vanColumn) + || isVanOwnedColumn(vanColumn)) + continue; + if (columnsToLeaveAlone != null && containsIgnoringCase(columnsToLeaveAlone, vanColumn)) + continue; + + if (setClause.length() > 0) + setClause.append(", "); + setClause.append(vanColumn).append(" = ?"); + values.add(normalize(record.get(serverColumns.get(i)))); + } + + setClause.append(setClause.length() > 0 ? ", " : "").append(PROCESSED).append(" = 'P', ") + .append(SYNC_FAILURE_REASON).append(" = NULL, ").append(LAST_DOWN_SYNC_DATE).append(" = ?, ") + .append(DOWN_SYNCED).append(" = 'P', ").append(DOWN_SYNC_DATE).append(" = ? "); + values.add(Timestamp.valueOf(LocalDateTime.now())); + values.add(Timestamp.valueOf(LocalDateTime.now())); + values.add(localID); + + String query = " UPDATE " + tableDetail.getSchemaName() + "." + tableDetail.getTableName() + " SET " + setClause + + " WHERE " + pkColumn + " = ? "; + + dataSyncRepository.updateDownSyncRecordInLocal(query, values.toArray()); + } + + private Map resolveFkColumnMapping(DownSyncTableDetail tableDetail) throws Exception { + Map configured = parseFkColumnMapping(tableDetail); + if (!configured.isEmpty()) + return configured; + + Map mapping = new LinkedHashMap<>(); + String ownPK = tableDetail.getVanAutoIncColumnName() == null ? null + : tableDetail.getVanAutoIncColumnName().trim(); + + for (Map foreignKey : dataSyncRepository.getForeignKeysOfTable(tableDetail.getSchemaName(), + tableDetail.getTableName())) { + + String childColumn = asTrimmedString(foreignKey.get("CHILD_COLUMN")); + String parentSchema = asTrimmedString(foreignKey.get("PARENT_SCHEMA")); + String parentTable = asTrimmedString(foreignKey.get("PARENT_TABLE")); + String parentColumn = asTrimmedString(foreignKey.get("PARENT_COLUMN")); + + if (childColumn == null || parentSchema == null || parentTable == null || parentColumn == null) + continue; + if (childColumn.equalsIgnoreCase(ownPK)) + continue; + + String parentPK = downSyncedTransactionalPrimaryKey(parentSchema, parentTable); + if (parentPK == null || !parentPK.equalsIgnoreCase(parentColumn)) + continue; + + requireIdentifier(childColumn, "foreign key column", tableDetail); + mapping.put(childColumn, parentSchema + "." + parentTable); + } + + if (!mapping.isEmpty()) + logger.info("Down-sync will rewrite {} of {}.{} from the table\'s own foreign keys", mapping, + tableDetail.getSchemaName(), tableDetail.getTableName()); + + return mapping; + } + + private String downSyncedTransactionalPrimaryKey(String parentSchema, String parentTable) { + ArrayList parents = downSyncTableDetailRepo.getActiveDownSyncTableByName(parentTable); + for (DownSyncTableDetail parent : parents) { + if (!parentSchema.equalsIgnoreCase(parent.getSchemaName()) || !parent.isTransactionalTable()) + continue; + if (parent.getVanAutoIncColumnName() != null + && parent.getVanAutoIncColumnName().trim().matches("^[a-zA-Z_][a-zA-Z0-9_]*$")) + return parent.getVanAutoIncColumnName().trim(); + } + return null; + } + + private String asTrimmedString(Object value) { + if (value == null) + return null; + String trimmed = String.valueOf(value).trim(); + return trimmed.isEmpty() ? null : trimmed; + } + + private Map parseFkColumnMapping(DownSyncTableDetail tableDetail) throws Exception { + Map mapping = new LinkedHashMap<>(); + String config = tableDetail.getFkColumnMapping(); + if (config == null || config.trim().isEmpty()) + return mapping; + + for (String entry : config.split(";")) { + if (entry == null || entry.trim().isEmpty()) + continue; + + String[] parts = entry.trim().split(":"); + if (parts.length != 2 || parts[1].trim().split("\\.").length != 2) + throw new Exception("FkColumnMapping of " + tableDetail.getTableName() + " holds '" + entry.trim() + + "', which is not of the form childColumn:schema.parentTable"); + + requireIdentifier(parts[0].trim(), "FkColumnMapping child column", tableDetail); + mapping.put(parts[0].trim(), parts[1].trim()); + } + return mapping; + } + + + private Map translateForeignKeys(DownSyncTableDetail tableDetail, Map fkMapping, + Map record, Integer vanID, Map cache, StringBuilder unresolved) { + + if (fkMapping.isEmpty()) + return record; + + Map translated = new LinkedHashMap<>(record); + + for (Map.Entry fk : fkMapping.entrySet()) { + String childColumn = resolveKeyIgnoringCase(record, fk.getKey()); + if (childColumn == null) + continue; + + Long centralValue = toLong(record.get(childColumn)); + if (centralValue == null) + continue; + + String parent = fk.getValue(); + String cacheKey = parent + "#" + centralValue; + Long localValue = cache.get(cacheKey); + + if (localValue == null) { + String parentSchema = parent.substring(0, parent.indexOf('.')); + String parentTable = parent.substring(parent.indexOf('.') + 1); + String parentPK = parentPrimaryKey(parentSchema, parentTable); + + if (parentPK == null) { + unresolved.append(fk.getKey()).append(" -> ").append(parent) + .append(" (no VanAutoIncColumnName configured for the parent); "); + continue; + } + + localValue = dataSyncRepository.resolveLocalIdForCentralValue(parentSchema, parentTable, parentPK, + centralValue, vanID); + if (localValue != null) + cache.put(cacheKey, localValue); + } + + if (localValue == null) { + // A pointer the van cannot place is only fatal if the column demands a + // value. Where it is optional - a consent or an image the van has no copy + // of - the record is worth more than the pointer, so it lands with the + // column empty and the loss is reported rather than hidden. + if (dataSyncRepository.isColumnNullable(tableDetail.getSchemaName(), tableDetail.getTableName(), + childColumn)) { + translated.put(childColumn, null); + String warning = fk.getKey() + " could not be placed (" + parent + + " has no such row on this van) and was left empty"; + if (currentResult != null) + currentResult.recordWarning(warning); + logger.warn("{}.{} : {}", tableDetail.getSchemaName(), tableDetail.getTableName(), warning); + continue; + } + + unresolved.append(fk.getKey()).append('=').append(centralValue).append(" not found in ").append(parent) + .append("; "); + continue; + } + + translated.put(childColumn, localValue); + } + + return unresolved.length() > 0 ? null : translated; + } + + /*** + * The parent's local primary key, taken from its own down-sync configuration. + */ + private String parentPrimaryKey(String parentSchema, String parentTable) { + ArrayList parents = downSyncTableDetailRepo.getActiveDownSyncTableByName(parentTable); + for (DownSyncTableDetail parent : parents) { + if (parentSchema.equalsIgnoreCase(parent.getSchemaName()) && parent.getVanAutoIncColumnName() != null + && parent.getVanAutoIncColumnName().trim().matches("^[a-zA-Z_][a-zA-Z0-9_]*$")) + return parent.getVanAutoIncColumnName().trim(); + } + return null; + } + + /*** + * Column names differ in case between the tables (vanSerialNo vs VanSerialNo), + * and a map lookup is case-sensitive - so never index a result row by a + * hard-coded name without this. + */ + private String resolveKeyIgnoringCase(Map record, String column) { + if (record == null || column == null) + return null; + if (record.containsKey(column)) + return column; + for (String key : record.keySet()) { + if (key.equalsIgnoreCase(column)) + return key; + } + return null; + } + + private boolean containsIgnoringCase(java.util.Set columns, String column) { + for (String candidate : columns) { + if (candidate.equalsIgnoreCase(column)) + return true; + } + return false; + } + + private static String shorten(String message) { + if (message == null || message.trim().isEmpty()) + return "Unknown error"; + String trimmed = message.trim(); + return trimmed.length() > 250 ? trimmed.substring(0, 250) : trimmed; + } + + private void requireIdentifier(String value, String what, DownSyncTableDetail tableDetail) throws Exception { + if (value == null || !value.matches("^[a-zA-Z_][a-zA-Z0-9_]*$")) + throw new Exception("The down-sync configuration of " + tableDetail.getTableName() + " holds an invalid " + + what); + } + + private boolean isVanOwnedColumn(String column) { + return VAN_ID.equalsIgnoreCase(column) || PROVISIONED.equalsIgnoreCase(column) + || RESERVED.equalsIgnoreCase(column) || RESERVED_FOR.equalsIgnoreCase(column) + || RESERVED_BY_ID.equalsIgnoreCase(column) || RESERVED_ON.equalsIgnoreCase(column); + } + + private boolean isDownSyncManagedColumn(String column) { + return VAN_SERIAL_NO.equalsIgnoreCase(column) || CENTRAL_ID.equalsIgnoreCase(column) + || PROCESSED.equalsIgnoreCase(column) || SYNC_FAILURE_REASON.equalsIgnoreCase(column) + || LAST_DOWN_SYNC_DATE.equalsIgnoreCase(column) || DOWN_SYNCED.equalsIgnoreCase(column) + || DOWN_SYNC_DATE.equalsIgnoreCase(column) || DOWN_SYNC_FAILURE_REASON.equalsIgnoreCase(column); + } + + private void acknowledgeToCentral(DownSyncTableDetail tableDetail, Integer vanID, List acks, + String serverAuthorization, String jwtToken) throws Exception { + + if (acks == null || acks.isEmpty()) + return; + + RestTemplate restTemplate = new RestTemplate(); + int acknowledged = 0; + + for (int from = 0; from < acks.size(); from += ACK_BATCH_SIZE) { + List batch = acks.subList(from, Math.min(from + ACK_BATCH_SIZE, acks.size())); + + DownSyncDataDigester digester = DownSyncDataDigester.forAck(tableDetail, vanID, batch); + HttpEntity request = RestTemplateUtil.createRequestEntity(digester, serverAuthorization, + DATA_SYNC_CALL); + + ResponseEntity response; + try { + response = restTemplate.exchange(downSyncFlagUpdateUrl, HttpMethod.POST, request, String.class); + } catch (Exception e) { + // the batches already accepted stay acknowledged - say how far we got, so + // the gap between the two sides is visible rather than guessed at + throw new Exception("Down-sync flag update failed for " + tableDetail.getTableName() + " after " + + acknowledged + " of " + acks.size() + " records : " + e.getMessage(), e); + } + + if (response == null || !response.hasBody()) + throw new Exception("Empty response while updating the down-sync flag for " + + tableDetail.getTableName() + " after " + acknowledged + " of " + acks.size() + " records"); + + JSONObject responseObj = new JSONObject(response.getBody()); + if (!responseObj.has("statusCode") || responseObj.getInt("statusCode") != 200) + throw new Exception("Could not update the down-sync flag at central for " + + tableDetail.getTableName() + " after " + acknowledged + " of " + acks.size() + " records : " + + responseObj.optString("errorMessage")); + + acknowledged += batch.size(); + } + + logger.info("Down-sync flag updated at central for {} records of {} in {} batch(es)", acknowledged, + tableDetail.getTableName(), (acks.size() + ACK_BATCH_SIZE - 1) / ACK_BATCH_SIZE); + } + + private List splitColumns(String columns) { + List columnList = new ArrayList<>(); + if (columns == null) + return columnList; + + for (String column : Arrays.asList(columns.split(","))) { + if (column != null && !column.trim().isEmpty()) + columnList.add(column.trim()); + } + return columnList; + } + + private Object normalize(Object value) { + if (value instanceof Double) { + double doubleValue = (Double) value; + if (!Double.isInfinite(doubleValue) && !Double.isNaN(doubleValue) && doubleValue == Math.floor(doubleValue) + && Math.abs(doubleValue) < 1e15) + return (long) doubleValue; + } + return value; + } + + private Long toLong(Object value) { + Object normalized = normalize(value); + if (normalized == null) + return null; + if (normalized instanceof Number) + return ((Number) normalized).longValue(); + try { + return Long.valueOf(String.valueOf(normalized).trim()); + } catch (NumberFormatException e) { + return null; + } + } + + private Timestamp toTimestamp(Object value) { + if (value == null) + return null; + if (value instanceof Timestamp) + return (Timestamp) value; + if (value instanceof java.util.Date) + return new Timestamp(((java.util.Date) value).getTime()); + + String stringValue = String.valueOf(value).trim(); + if (stringValue.isEmpty() || "null".equalsIgnoreCase(stringValue)) + return null; + + String candidate = stringValue.replace("T", " ").replace("Z", ""); + try { + return Timestamp.valueOf(candidate); + } catch (IllegalArgumentException e) { + try { + // gson's default rendering of a java.sql.Timestamp + return new Timestamp(DateFormat.getDateTimeInstance().parse(stringValue).getTime()); + } catch (java.text.ParseException pe) { + logger.warn("Could not parse the timestamp '{}' received from central", stringValue); + return null; + } + } + } +} diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DataSyncRepositoryCentralDownload.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DataSyncRepositoryCentralDownload.java index 565e2466..c76962b4 100644 --- a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DataSyncRepositoryCentralDownload.java +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/DataSyncRepositoryCentralDownload.java @@ -28,6 +28,9 @@ import javax.sql.DataSource; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncRecordAck; +import com.iemr.mmu.utils.validator.SqlIdentifierValidator; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -54,6 +57,8 @@ private JdbcTemplate getJdbcTemplate() { private Logger logger = LoggerFactory.getLogger(this.getClass().getSimpleName()); + private static final String DOWN_SYNC_TRANSACTIONAL = "TRANSACTIONAL"; + // Data Upload Repository public int checkRecordIsAlreadyPresentOrNot(String schemaName, String tableName, String vanSerialNo, String vanID, String vanAutoIncColumnName, int syncFacilityID) { @@ -205,4 +210,182 @@ else if (masterType.equalsIgnoreCase("P")) { } // End of Data Download Repository -} \ No newline at end of file + + // ---------------------------------- Down-Sync Repository (central -> local) + + public String resolveLastModColumn(String schema, String table) { + jdbcTemplate = getJdbcTemplate(); + + String query = " SELECT COLUMN_NAME FROM information_schema.COLUMNS " + + " WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " + + " AND COLUMN_NAME IN ('LastModDate', 'last_mod_date') " + + " ORDER BY FIELD(COLUMN_NAME, 'LastModDate', 'last_mod_date') "; + + List found = jdbcTemplate.queryForList(query, String.class, schema, table); + return (found == null || found.isEmpty()) ? null : found.get(0); + } + + public List> getDownSyncDataFromTable(String schema, String table, String columnNames, + String tableType, Integer vanID, String lastModColumn, String pkColumn, Long lastFetchedID, + Integer batchSize) throws Exception { + jdbcTemplate = getJdbcTemplate(); + + if (schema == null || table == null) + throw new Exception("Invalid down-sync request. Schema/table info is missing"); + + // schema, table & column names cannot be bound as query parameters, so each of + // them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String columns = (columnNames == null || columnNames.trim().isEmpty()) ? " * " + : SqlIdentifierValidator.validatedColumnList(columnNames); + List> resultSetList; + + if (DOWN_SYNC_TRANSACTIONAL.equalsIgnoreCase(tableType)) { + if (vanID == null) + throw new Exception( + "Invalid down-sync request. VanID is mandatory for transactional table " + validTable); + + String validLastModColumn = SqlIdentifierValidator.validatedColumnName(lastModColumn); + boolean paged = SqlIdentifierValidator.isValidIdentifier(pkColumn) && batchSize != null && batchSize > 0; + + StringBuilder query = new StringBuilder(" SELECT ").append(columns).append(" FROM ").append(validSchema) + .append(".").append(validTable) + .append(" WHERE VanID = ? AND ( DownSynced IS NULL OR DownSynced IN ('N', 'U', 'F') ") + .append(" OR ( DownSynced = 'P' AND DownSyncDate IS NOT NULL AND ").append(validLastModColumn) + .append(" > DownSyncDate ) ) ") + .append(" AND ( DownSyncFailureReason IS NULL OR DownSyncFailureReason <> '") + .append(DownSyncRecordAck.CONFLICT).append("' ) "); + + List params = new ArrayList<>(); + params.add(vanID); + + if (paged) { + String validPkColumn = SqlIdentifierValidator.validatedColumnName(pkColumn); + if (lastFetchedID != null) { + query.append(" AND ").append(validPkColumn).append(" > ? "); + params.add(lastFetchedID); + } + query.append(" ORDER BY ").append(validPkColumn).append(" LIMIT ").append(batchSize.intValue()); + } + + logger.info("Down-sync select query for {}.{} : {}", validSchema, validTable, query); + resultSetList = jdbcTemplate.queryForList(query.toString(), params.toArray()); + } else { + String query = " SELECT " + columns + " FROM " + validSchema + "." + validTable; + logger.info("Down-sync select query for {}.{} : {}", validSchema, validTable, query); + resultSetList = jdbcTemplate.queryForList(query); + } + + logger.info("Down-sync record count for {}.{} : {}", validSchema, validTable, resultSetList.size()); + return resultSetList; + } + + public int updateDownSyncFlagPostDownload(String schema, String table, String pkColumnName, + List records) throws Exception { + jdbcTemplate = getJdbcTemplate(); + + if (schema == null || table == null || pkColumnName == null) + throw new Exception("Invalid request. Schema/table/primary key info is missing"); + if (records == null || records.isEmpty()) + return 0; + + // schema, table & primary key column cannot be bound as query parameters, so + // each of them is validated before it is concatenated into the query + String validSchema = SqlIdentifierValidator.validatedSchemaName(schema); + String validTable = SqlIdentifierValidator.validatedTableName(table); + String validPkColumn = SqlIdentifierValidator.validatedColumnName(pkColumnName); + + String lastModColumn = resolveLastModColumn(schema, table); + String holdLastModDate = lastModColumn == null ? "" + : ", " + SqlIdentifierValidator.validatedColumnName(lastModColumn) + " = " + + SqlIdentifierValidator.validatedColumnName(lastModColumn) + " "; + + List successWithSerialNo = new ArrayList<>(); + List successWithoutSerialNo = new ArrayList<>(); + List failed = new ArrayList<>(); + List retryable = new ArrayList<>(); + + for (DownSyncRecordAck record : records) { + if (record == null || record.getCentralID() == null) + continue; + + if (record.isSuccess()) { + if (record.getVanSerialNo() != null) + successWithSerialNo.add(new Object[] { record.getVanSerialNo(), record.getCentralID() }); + else + successWithoutSerialNo.add(new Object[] { record.getCentralID() }); + } else if (record.isRetryable()) { + retryable.add(new Object[] { record.getFailureReason(), record.getCentralID() }); + } else { + failed.add(new Object[] { record.getFailureReason(), record.getCentralID() }); + } + } + + int updatedRows = 0; + + if (!successWithSerialNo.isEmpty()) { + String query = " UPDATE " + validSchema + "." + validTable + + " SET DownSynced = 'P', DownSyncDate = now(), DownSyncFailureReason = NULL, VanSerialNo = ? " + + holdLastModDate + " WHERE " + validPkColumn + " = ? "; + updatedRows += countUpdatedRows(jdbcTemplate.batchUpdate(query, successWithSerialNo)); + } + + if (!successWithoutSerialNo.isEmpty()) { + String query = " UPDATE " + validSchema + "." + validTable + + " SET DownSynced = 'P', DownSyncDate = now(), DownSyncFailureReason = NULL " + holdLastModDate + + " WHERE " + validPkColumn + " = ? "; + updatedRows += countUpdatedRows(jdbcTemplate.batchUpdate(query, successWithoutSerialNo)); + } + + if (!retryable.isEmpty()) { + String query = " UPDATE " + validSchema + "." + validTable + + " SET DownSynced = 'U', DownSyncFailureReason = ? " + holdLastModDate + " WHERE " + validPkColumn + + " = ? "; + updatedRows += countUpdatedRows(jdbcTemplate.batchUpdate(query, retryable)); + } + + if (!failed.isEmpty()) { + String query = " UPDATE " + validSchema + "." + validTable + + " SET DownSynced = 'F', DownSyncFailureReason = ? " + holdLastModDate + " WHERE " + validPkColumn + + " = ? "; + updatedRows += countUpdatedRows(jdbcTemplate.batchUpdate(query, failed)); + } + + logger.info("Down-sync flag updated for {}.{}. Records : {}, rows : {}", validSchema, validTable, + records.size(), updatedRows); + return updatedRows; + } + + public int markDownSyncedPostUpSync(String schema, String table, List vanSerialNoAndVanID) { + if (schema == null || table == null || vanSerialNoAndVanID == null || vanSerialNoAndVanID.isEmpty()) + return 0; + + jdbcTemplate = getJdbcTemplate(); + // schema & table cannot be bound as query parameters, so both are validated + // before they are concatenated into the query + String lastModColumn = resolveLastModColumn(schema, table); + String holdLastModDate = lastModColumn == null ? "" + : ", " + SqlIdentifierValidator.validatedColumnName(lastModColumn) + " = " + + SqlIdentifierValidator.validatedColumnName(lastModColumn) + " "; + + String query = " UPDATE " + SqlIdentifierValidator.validatedSchemaName(schema) + "." + + SqlIdentifierValidator.validatedTableName(table) + + " SET DownSynced = 'P', DownSyncDate = now(), DownSyncFailureReason = NULL " + holdLastModDate + + " WHERE VanSerialNo = ? AND VanID = ? "; + return countUpdatedRows(jdbcTemplate.batchUpdate(query, vanSerialNoAndVanID)); + } + + private int countUpdatedRows(int[] batchResult) { + int count = 0; + if (batchResult != null) { + for (int rows : batchResult) { + if (rows > 0) + count += rows; + } + } + return count; + } + + // End of Down-Sync Repository +} 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/service/dataSyncLayerCentral/GetDataFromVanAndSyncToDBImpl.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDataFromVanAndSyncToDBImpl.java index cca5634d..3205007c 100644 --- a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDataFromVanAndSyncToDBImpl.java +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDataFromVanAndSyncToDBImpl.java @@ -31,6 +31,7 @@ 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.fasterxml.jackson.databind.ObjectMapper; @@ -46,6 +47,12 @@ public class GetDataFromVanAndSyncToDBImpl implements GetDataFromVanAndSyncToDB @Autowired private DataSyncRepositoryCentral dataSyncRepositoryCentral; + @Autowired + private GetDownSyncDataFromCentral getDownSyncDataFromCentral; + + @Value("${downSync.markDeliveredOnUpSync:true}") + private boolean markDownSyncedOnUpSync; + private static final Map> TABLE_GROUPS = new HashMap<>(); static { TABLE_GROUPS.put(1, @@ -326,6 +333,7 @@ private boolean performGenericTableSync(SyncUploadDataDigester syncUploadDataDig Map insertIndexMap = new HashMap<>(); Map updateIndexMap = new HashMap<>(); + Map vanIDByResultIndex = new HashMap<>(); boolean overallSuccess = true; @@ -463,6 +471,7 @@ private boolean performGenericTableSync(SyncUploadDataDigester syncUploadDataDig int currentSyncResultIndex = syncResults.size(); syncResults.add(new SyncResult(schemaName, syncTableName, vanSerialNo, syncUploadDataDigester.getSyncedBy(), false, "Pending")); + vanIDByResultIndex.put(currentSyncResultIndex, vanID); if (recordCheck == 0) { insertIndexMap.put(currentSyncResultIndex, syncDataListInsert.size()); @@ -591,9 +600,35 @@ private boolean performGenericTableSync(SyncUploadDataDigester syncUploadDataDig } logger.info("Sync results for table {}: {}", syncTableName, syncResults); + + flagUpSyncedRecordsForDownSync(schemaName, syncTableName, syncResults, vanIDByResultIndex); + return overallSuccess; } + private void flagUpSyncedRecordsForDownSync(String schemaName, String tableName, List syncResults, + Map vanIDByResultIndex) { + if (!markDownSyncedOnUpSync || vanIDByResultIndex.isEmpty()) + return; + + List vanSerialNoAndVanID = new ArrayList<>(); + for (Map.Entry entry : vanIDByResultIndex.entrySet()) { + int index = entry.getKey(); + if (index >= syncResults.size()) + continue; + + SyncResult result = syncResults.get(index); + if (result == null || !result.isSuccess() || result.getVanSerialNo() == null + || entry.getValue() == null || "null".equalsIgnoreCase(entry.getValue())) + continue; + + vanSerialNoAndVanID.add(new Object[] { result.getVanSerialNo(), entry.getValue() }); + } + + if (!vanSerialNoAndVanID.isEmpty()) + getDownSyncDataFromCentral.markDownSyncedPostUpSync(schemaName, tableName, vanSerialNoAndVanID); + } + private String getVanSerialNo(Object[] record, int vanSerialIndex, SyncResult originalResult) { if (vanSerialIndex >= 0 && vanSerialIndex < record.length) { return String.valueOf(record[vanSerialIndex]); diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDownSyncDataFromCentral.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDownSyncDataFromCentral.java new file mode 100644 index 00000000..553ee73e --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDownSyncDataFromCentral.java @@ -0,0 +1,35 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.service.dataSyncLayerCentral; + +import java.util.List; + +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncDataDigester; + +public interface GetDownSyncDataFromCentral { + + String getDownSyncDataForVan(DownSyncDataDigester downSyncDataDigester) throws Exception; + + int updateDownSyncFlagPostDownload(DownSyncDataDigester downSyncDataDigester) throws Exception; + + int markDownSyncedPostUpSync(String schemaName, String tableName, List vanSerialNoAndVanID); +} diff --git a/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDownSyncDataFromCentralImpl.java b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDownSyncDataFromCentralImpl.java new file mode 100644 index 00000000..8dc1015f --- /dev/null +++ b/src/main/java/com/iemr/mmu/service/dataSyncLayerCentral/GetDownSyncDataFromCentralImpl.java @@ -0,0 +1,141 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ +package com.iemr.mmu.service.dataSyncLayerCentral; + +import java.sql.Timestamp; +import java.time.LocalDateTime; +import java.util.ArrayList; +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.stereotype.Service; + +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.JsonPrimitive; +import com.google.gson.JsonSerializer; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncDataDigester; +import com.iemr.mmu.data.syncActivity_syncLayer.DownSyncTableDetail; +import com.iemr.mmu.repo.syncActivity_syncLayer.DownSyncTableDetailRepo; + + +@Service +public class GetDownSyncDataFromCentralImpl implements GetDownSyncDataFromCentral { + + private Logger logger = LoggerFactory.getLogger(this.getClass().getSimpleName()); + + @Autowired + private DataSyncRepositoryCentralDownload dataSyncRepositoryCentralDownload; + + @Autowired + private DownSyncTableDetailRepo downSyncTableDetailRepo; + + public String getDownSyncDataForVan(DownSyncDataDigester downSyncDataDigester) throws Exception { + if (downSyncDataDigester == null || downSyncDataDigester.getSchemaName() == null + || downSyncDataDigester.getTableName() == null) + throw new Exception("Invalid down-sync request. Either schema or table info is missing/wrong"); + + if (!downSyncDataDigester.isMasterTable() && downSyncDataDigester.getVanID() == null) + throw new Exception("Invalid down-sync request. VanID is mandatory for a transactional table"); + + List> resultSetList = dataSyncRepositoryCentralDownload.getDownSyncDataFromTable( + downSyncDataDigester.getSchemaName(), downSyncDataDigester.getTableName(), + downSyncDataDigester.getServerColumnName(), downSyncDataDigester.getTableType(), + downSyncDataDigester.getVanID(), + resolveLastModColumn(downSyncDataDigester.getSchemaName(), downSyncDataDigester.getTableName()), + downSyncDataDigester.getVanAutoIncColumnName(), downSyncDataDigester.getLastFetchedID(), + downSyncDataDigester.getBatchSize()); + + return downSyncGson().toJson(resultSetList); + } + + private String resolveLastModColumn(String schemaName, String tableName) throws Exception { + String lastModColumn = dataSyncRepositoryCentralDownload.resolveLastModColumn(schemaName, tableName); + + if (lastModColumn == null) + throw new Exception(schemaName + "." + tableName + + " has neither LastModDate nor last_mod_date, so the down-sync cannot tell when a record was" + + " last changed"); + + if (!lastModColumn.matches("^[a-zA-Z_][a-zA-Z0-9_]*$")) + throw new Exception(schemaName + "." + tableName + " holds an invalid modification-time column name"); + + return lastModColumn; + } + + private Gson downSyncGson() { + GsonBuilder gsonBuilder = new GsonBuilder(); + gsonBuilder.serializeNulls(); + gsonBuilder.registerTypeAdapter(Timestamp.class, + (JsonSerializer) (src, type, context) -> new JsonPrimitive(src.toString())); + gsonBuilder.registerTypeAdapter(java.sql.Date.class, + (JsonSerializer) (src, type, context) -> new JsonPrimitive(src.toString())); + gsonBuilder.registerTypeAdapter(java.sql.Time.class, + (JsonSerializer) (src, type, context) -> new JsonPrimitive(src.toString())); + gsonBuilder.registerTypeAdapter(LocalDateTime.class, + (JsonSerializer) (src, type, context) -> new JsonPrimitive( + Timestamp.valueOf(src).toString())); + return gsonBuilder.create(); + } + + public int updateDownSyncFlagPostDownload(DownSyncDataDigester downSyncDataDigester) throws Exception { + if (downSyncDataDigester == null || downSyncDataDigester.getSchemaName() == null + || downSyncDataDigester.getTableName() == null || downSyncDataDigester.getRecords() == null) + throw new Exception("Invalid request. Schema, table or record info is missing/wrong"); + + String pkColumnName = downSyncDataDigester.getVanAutoIncColumnName(); + if (pkColumnName == null || pkColumnName.trim().isEmpty()) + throw new Exception("Invalid request. Primary key column of " + downSyncDataDigester.getTableName() + + " is missing in the request"); + + return dataSyncRepositoryCentralDownload.updateDownSyncFlagPostDownload(downSyncDataDigester.getSchemaName(), + downSyncDataDigester.getTableName(), pkColumnName.trim(), downSyncDataDigester.getRecords()); + } + + @Override + public int markDownSyncedPostUpSync(String schemaName, String tableName, List vanSerialNoAndVanID) { + if (schemaName == null || tableName == null || vanSerialNoAndVanID == null || vanSerialNoAndVanID.isEmpty()) + return 0; + + try { + ArrayList tableDetails = downSyncTableDetailRepo + .getActiveDownSyncTableByName(tableName); + boolean downSyncEnabled = false; + for (DownSyncTableDetail tableDetail : tableDetails) { + if (tableDetail.isTransactionalTable()) + downSyncEnabled = true; + } + if (!downSyncEnabled) + return 0; + + return dataSyncRepositoryCentralDownload.markDownSyncedPostUpSync(schemaName, tableName, + vanSerialNoAndVanID); + } catch (Exception e) { + logger.warn("Could not mark the DownSynced flag for {}.{} post up-sync : {}", schemaName, tableName, + e.getMessage()); + return 0; + } + } +} diff --git a/src/main/java/com/iemr/mmu/service/labtechnician/LabTechnicianServiceImpl.java b/src/main/java/com/iemr/mmu/service/labtechnician/LabTechnicianServiceImpl.java index 1026f63c..96f3b5c1 100644 --- a/src/main/java/com/iemr/mmu/service/labtechnician/LabTechnicianServiceImpl.java +++ b/src/main/java/com/iemr/mmu/service/labtechnician/LabTechnicianServiceImpl.java @@ -47,6 +47,22 @@ public class LabTechnicianServiceImpl implements LabTechnicianService { private LabResultEntryRepo labResultEntryRepo; private CommonBenStatusFlowServiceImpl commonBenStatusFlowServiceImpl; + @Autowired + private com.iemr.mmu.repo.nurse.BenVisitDetailRepo benVisitDetailRepo; + @Autowired + private com.iemr.mmu.repo.login.UserLoginRepo userLoginRepo; + + /** + * Resolve the numeric user ID of the responsible staff member from the username + * captured in createdBy. Returns null if it cannot be resolved. + */ + private Long resolveUserId(String username) { + if (username == null || username.trim().isEmpty()) + return null; + com.iemr.mmu.data.login.Users user = userLoginRepo.getUserByUsername(username.trim()); + return user != null ? user.getUserID() : null; + } + @Autowired public void setCommonBenStatusFlowServiceImpl(CommonBenStatusFlowServiceImpl commonBenStatusFlowServiceImpl) { this.commonBenStatusFlowServiceImpl = commonBenStatusFlowServiceImpl; @@ -345,6 +361,12 @@ public Integer saveLabTestResult(JsonObject requestOBJ) throws Exception { labResultSaveFlag = saveLabTestResult(wrapperLabResults); if (labResultSaveFlag == 1) { + // Store the responsible lab technician's user ID against the visit + if (wrapperLabResults.getVisitCode() != null) { + Long labTechnicianID = resolveUserId(wrapperLabResults.getCreatedBy()); + if (labTechnicianID != null) + benVisitDetailRepo.updateLabTechnicianID(labTechnicianID, wrapperLabResults.getVisitCode()); + } int i = updateBenFlowStatusFlagAfterLabResultEntry(wrapperLabResults.getLabCompleted(), wrapperLabResults.getBenFlowID(), wrapperLabResults.getBeneficiaryRegID(), wrapperLabResults.getVisitID(), wrapperLabResults.getNurseFlag(), 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); + } +} diff --git a/src/main/java/com/iemr/mmu/utils/validator/SqlIdentifierValidator.java b/src/main/java/com/iemr/mmu/utils/validator/SqlIdentifierValidator.java new file mode 100644 index 00000000..ed058a30 --- /dev/null +++ b/src/main/java/com/iemr/mmu/utils/validator/SqlIdentifierValidator.java @@ -0,0 +1,103 @@ +/* +* AMRIT – Accessible Medical Records via Integrated Technology +* Integrated EHR (Electronic Health Records) Solution +* +* Copyright (C) "Piramal Swasthya Management and Research Institute" +* +* This file is part of AMRIT. +* +* This program is free software: you can redistribute it and/or modify +* it under the terms of the GNU General Public License as published by +* the Free Software Foundation, either version 3 of the License, or +* (at your option) any later version. +* +* This program is distributed in the hope that it will be useful, +* but WITHOUT ANY WARRANTY; without even the implied warranty of +* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +* GNU General Public License for more details. +* +* You should have received a copy of the GNU General Public License +* along with this program. If not, see https://www.gnu.org/licenses/. +*/ + +package com.iemr.mmu.utils.validator; + +import java.util.Set; +import java.util.regex.Pattern; + +/*** + * Validates the schema, table and column names that the data-sync layer has to + * concatenate into its dynamic SQL, because a schema/table/column name cannot be + * supplied as a prepared-statement parameter. Every such identifier reaching a + * query must first be passed through this class, so that nothing but a plain SQL + * identifier can ever become part of a statement. + */ +public final class SqlIdentifierValidator { + + private static final Pattern IDENTIFIER_PATTERN = Pattern.compile("^[a-zA-Z_][a-zA-Z0-9_]*$"); + + private static final int MAX_IDENTIFIER_LENGTH = 64; + + private static final Set VALID_SCHEMAS = Set.of("public", "db_iemr", "db_identity", "apl_db_iemr", + "apl_db_identity", "db_iemr_sync", "db_identity_sync"); + + private SqlIdentifierValidator() { + // utility class + } + + public static boolean isValidIdentifier(String identifier) { + return identifier != null && identifier.length() <= MAX_IDENTIFIER_LENGTH + && IDENTIFIER_PATTERN.matcher(identifier).matches(); + } + + /*** + * Returns the allow-listed constant matching the given schema name, so that the + * value concatenated into a query never originates from the caller. + */ + public static String validatedSchemaName(String schemaName) { + if (schemaName != null) { + for (String allowedSchema : VALID_SCHEMAS) { + if (allowedSchema.equalsIgnoreCase(schemaName)) { + return allowedSchema; + } + } + } + throw new IllegalArgumentException("Invalid schema name provided for the data-sync query"); + } + + public static String validatedIdentifier(String identifier, String identifierType) { + if (!isValidIdentifier(identifier)) { + throw new IllegalArgumentException( + "Invalid " + identifierType + " provided for the data-sync query"); + } + return identifier; + } + + public static String validatedTableName(String tableName) { + return validatedIdentifier(tableName, "table name"); + } + + public static String validatedColumnName(String columnName) { + return validatedIdentifier(columnName, "column name"); + } + + /*** + * Validates a comma separated column list, as used in the SELECT clause of the + * sync queries. Returns the trimmed, comma separated list built back from the + * validated names. + */ + public static String validatedColumnList(String columnNames) { + if (columnNames == null || columnNames.trim().isEmpty()) { + throw new IllegalArgumentException("Invalid column list provided for the data-sync query"); + } + + StringBuilder validatedColumns = new StringBuilder(); + for (String columnName : columnNames.split(",")) { + if (validatedColumns.length() > 0) { + validatedColumns.append(", "); + } + validatedColumns.append(validatedColumnName(columnName.trim())); + } + return validatedColumns.toString(); + } +}