1+ package com .wipro .fhir .service .elasticsearch ;
2+
3+ import co .elastic .clients .elasticsearch .ElasticsearchClient ;
4+ import co .elastic .clients .elasticsearch ._types .Refresh ;
5+ import co .elastic .clients .elasticsearch .core .GetRequest ;
6+ import co .elastic .clients .elasticsearch .core .GetResponse ;
7+ import co .elastic .clients .elasticsearch .core .UpdateRequest ;
8+ import com .fasterxml .jackson .databind .ObjectMapper ;
9+ import org .slf4j .Logger ;
10+ import org .slf4j .LoggerFactory ;
11+ import org .springframework .beans .factory .annotation .Autowired ;
12+ import org .springframework .beans .factory .annotation .Value ;
13+ import org .springframework .scheduling .annotation .Async ;
14+ import org .springframework .stereotype .Service ;
15+
16+ import java .util .HashMap ;
17+ import java .util .Map ;
18+
19+ /**
20+ * Lightweight ES sync service for FHIR-API
21+ * Only updates ABHA-related fields without fetching full beneficiary data
22+ */
23+ @ Service
24+ public class AbhaElasticsearchSyncService {
25+
26+ private static final Logger logger = LoggerFactory .getLogger (AbhaElasticsearchSyncService .class );
27+
28+ @ Autowired
29+ private ElasticsearchClient esClient ;
30+
31+ @ Value ("${elasticsearch.index.beneficiary}" )
32+ private String beneficiaryIndex ;
33+
34+ @ Value ("${elasticsearch.enabled}" )
35+ private boolean esEnabled ;
36+
37+ private final ObjectMapper objectMapper = new ObjectMapper ();
38+
39+ /**
40+ * Update ABHA details in Elasticsearch after ABHA is created/updated
41+ * This method updates only ABHA fields, doesn't require full beneficiary data
42+ */
43+ @ Async ("esAsyncExecutor" )
44+ public void updateAbhaInElasticsearch (Long benRegId , String healthId , String healthIdNumber , String createdDate ) {
45+ if (!esEnabled ) {
46+ logger .debug ("Elasticsearch is disabled, skipping ABHA sync" );
47+ return ;
48+ }
49+
50+ if (benRegId == null ) {
51+ logger .warn ("benRegId is null, cannot sync ABHA to ES" );
52+ return ;
53+ }
54+
55+ int maxRetries = 3 ;
56+ int retryDelay = 2000 ; // 2 seconds
57+
58+ for (int attempt = 1 ; attempt <= maxRetries ; attempt ++) {
59+ try {
60+ logger .info ("Syncing ABHA details to ES for benRegId: {} (attempt {}/{})" , benRegId , attempt , maxRetries );
61+
62+ Map <String , Object > abhaData = new HashMap <>();
63+ abhaData .put ("healthID" , healthId );
64+ abhaData .put ("abhaID" , healthIdNumber );
65+ abhaData .put ("abhaCreatedDate" , createdDate );
66+
67+ String documentId = String .valueOf (benRegId );
68+ boolean exists = checkDocumentExists (documentId );
69+
70+ if (exists ) {
71+ UpdateRequest <Object , Object > updateRequest = UpdateRequest .of (u -> u
72+ .index (beneficiaryIndex )
73+ .id (documentId )
74+ .doc (abhaData )
75+ .refresh (Refresh .True )
76+ .docAsUpsert (false )
77+ .retryOnConflict (3 )
78+ );
79+
80+ esClient .update (updateRequest , Object .class );
81+ logger .info ("Successfully updated ABHA in ES: benRegId={}" , benRegId );
82+ return ;
83+
84+ } else {
85+ logger .warn ("Document not found in ES for benRegId={} (attempt {}/{})" , benRegId , attempt , maxRetries );
86+ if (attempt < maxRetries ) {
87+ Thread .sleep (retryDelay * attempt );
88+ }
89+ }
90+
91+ } catch (java .net .SocketTimeoutException e ) {
92+ logger .error ("Timeout updating ABHA in ES for benRegId {} (attempt {}/{}): {}" ,
93+ benRegId , attempt , maxRetries , e .getMessage ());
94+ if (attempt < maxRetries ) {
95+ try {
96+ Thread .sleep (retryDelay * attempt );
97+ } catch (InterruptedException ie ) {
98+ Thread .currentThread ().interrupt ();
99+ return ;
100+ }
101+ }
102+ } catch (Exception e ) {
103+ logger .error ("Error updating ABHA in ES for benRegId {} (attempt {}/{}): {}" ,
104+ benRegId , attempt , maxRetries , e .getMessage ());
105+ if (attempt == maxRetries ) {
106+ logger .error ("Failed to sync ABHA after {} attempts for benRegId {}" , maxRetries , benRegId );
107+ }
108+ }
109+ }
110+ }
111+
112+ /**
113+ * Check if document exists in ES
114+ */
115+ private boolean checkDocumentExists (String documentId ) {
116+ try {
117+ GetRequest getRequest = GetRequest .of (g -> g
118+ .index (beneficiaryIndex )
119+ .id (documentId )
120+ );
121+
122+ GetResponse <Object > response = esClient .get (getRequest , Object .class );
123+ return response .found ();
124+
125+ } catch (Exception e ) {
126+ logger .debug ("Document not found or error checking: {}" , e .getMessage ());
127+ return false ;
128+ }
129+ }
130+
131+ /**
132+ * Retry sync after 5 seconds if document wasn't found
133+ * (Handles race condition where ABHA is saved before beneficiary is synced to ES)
134+ */
135+ @ Async ("esAsyncExecutor" )
136+ private void retryAfterDelay (Long benRegId , String healthId , String healthIdNumber , String createdDate ) {
137+ try {
138+ Thread .sleep (5000 ); // Wait 5 seconds
139+ logger .info ("Retrying ABHA sync for benRegId: {}" , benRegId );
140+ updateAbhaInElasticsearch (benRegId , healthId , healthIdNumber , createdDate );
141+ } catch (InterruptedException e ) {
142+ Thread .currentThread ().interrupt ();
143+ logger .error ("Retry interrupted for benRegId: {}" , benRegId );
144+ }
145+ }
146+
147+ /**
148+ * Update multiple ABHA addresses (comma-separated)
149+ */
150+ @ Async ("esAsyncExecutor" )
151+ public void updateMultipleAbhaAddresses (Long benRegId , String commaSeparatedHealthIds ,
152+ String healthIdNumber , String createdDate ) {
153+ if (!esEnabled || benRegId == null ) {
154+ return ;
155+ }
156+
157+ // For multiple ABHA addresses, store as comma-separated string
158+ updateAbhaInElasticsearch (benRegId , commaSeparatedHealthIds , healthIdNumber , createdDate );
159+ }
160+ }
0 commit comments