[HCMPRE-2739] central instance changes for expense service

backend/expense/pom.xml

@@ -108,6 +108,11 @@
             <artifactId>lombok</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.egov.services</groupId>
+            <artifactId>services-common</artifactId>
+            <version>2.9.0-SNAPSHOT</version>
+        </dependency>
 
         <dependency>
             <groupId>com.fasterxml.jackson.datatype</groupId>

backend/expense/src/main/java/org/egov/digit/expense/ExpenseApplication.java

@@ -1,15 +1,13 @@
 package org.egov.digit.expense;
 
+import org.egov.common.utils.MultiStateInstanceUtil;
 import org.egov.tracer.config.TracerConfiguration;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Import;
 
-@Import({ TracerConfiguration.class })
+@Import({ TracerConfiguration.class, MultiStateInstanceUtil.class })
 @SpringBootApplication
-@ComponentScan(basePackages = { "org.egov.digit.expense", "org.egov.digit.expense.web.controllers",
-		"org.egov.digit.expense.config" })
 public class ExpenseApplication {
 
 	public static void main(String[] args) {

backend/expense/src/main/java/org/egov/digit/expense/config/Constants.java

@@ -7,6 +7,8 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.egov.common.utils.MultiStateInstanceUtil.SCHEMA_REPLACE_STRING;
+
 public class Constants {
 	public static final String REQUEST_INFO = "RequestInfo";
 	public static final String TENANT_ID = "tenantId";
@@ -95,21 +97,21 @@ public class Constants {
 			+ " li.lastmodifiedby as li_lastmodifiedby, li.lastmodifiedtime as li_lastmodifiedtime "
 
 
-			+ " FROM eg_expense_payment payment "
+			+ " FROM " + SCHEMA_REPLACE_STRING + ".eg_expense_payment payment "
 
-			+ INNER_JOIN + " eg_expense_payment_bill paymentbill ON paymentbill.paymentid = payment.id"
+			+ INNER_JOIN + " " + SCHEMA_REPLACE_STRING + ".eg_expense_payment_bill paymentbill ON paymentbill.paymentid = payment.id"
 					+ " AND paymentbill.tenantid = payment.tenantid "
 
-			+ INNER_JOIN + " eg_expense_payment_billdetail paymentbd ON paymentbd.paymentbillid = paymentbill.id"
+			+ INNER_JOIN + " " + SCHEMA_REPLACE_STRING + ".eg_expense_payment_billdetail paymentbd ON paymentbd.paymentbillid = paymentbill.id"
 					+ " AND paymentbd.tenantid = paymentbill.tenantid "
 
-			+ INNER_JOIN + " eg_expense_payment_lineitem li ON li.paymentbilldetailid = paymentbd.id"
+			+ INNER_JOIN + " " + SCHEMA_REPLACE_STRING + ".eg_expense_payment_lineitem li ON li.paymentbilldetailid = paymentbd.id"
 					+ " AND li.tenantid=paymentbd.tenantid ";
 
 	public static final String PAYMENT_COUNT_QUERY = "SELECT distinct(payment.id) " +
-			"FROM eg_expense_payment payment "
+			"FROM " + SCHEMA_REPLACE_STRING + ".eg_expense_payment payment "
 
-			+ INNER_JOIN + " eg_expense_payment_bill paymentbill ON paymentbill.paymentid = payment.id"
+			+ INNER_JOIN + " " + SCHEMA_REPLACE_STRING + ".eg_expense_payment_bill paymentbill ON paymentbill.paymentid = payment.id"
 					+ " AND paymentbill.tenantid = payment.tenantid ";
 
 
@@ -140,18 +142,20 @@ public class Constants {
 			+ " payer.createdby as payer_createdby, payer.createdtime as payer_createdtime, payer.lastmodifiedby as payer_lastmodifiedby, "
 			+ "payer.lastmodifiedtime as payer_lastmodifiedtime, payer.additionaldetails as payer_additionaldetails, payer.status as payer_status "
 
-			+ "FROM eg_expense_bill bill "
+			+ "FROM "+ SCHEMA_REPLACE_STRING + ".eg_expense_bill bill "
 
-			+ LEFT_JOIN + " EG_EXPENSE_PARTY PAYER ON bill.id = payer.parentid AND bill.tenantid = payer.tenantid "
+			+ LEFT_JOIN + " " + SCHEMA_REPLACE_STRING + ".EG_EXPENSE_PARTY PAYER ON bill.id = payer.parentid AND bill.tenantid = payer.tenantid "
 
-			+ LEFT_JOIN + " EG_EXPENSE_BILLDETAIL BD ON bill.id = bd.billid AND bd.tenantid = bill.tenantid "
+			+ LEFT_JOIN + " " + SCHEMA_REPLACE_STRING + ".EG_EXPENSE_BILLDETAIL BD ON bill.id = bd.billid AND bd.tenantid = bill.tenantid "
 
-			+ LEFT_JOIN + " EG_EXPENSE_LINEITEM LI ON bd.id = li.billdetailid AND bd.tenantid = li.tenantid "
+			+ LEFT_JOIN + " " + SCHEMA_REPLACE_STRING + ".EG_EXPENSE_LINEITEM LI ON bd.id = li.billdetailid AND bd.tenantid = li.tenantid "
 
-			+ LEFT_JOIN + " EG_EXPENSE_PARTY PAYEE ON bd.id = payee.parentid AND bd.tenantid = payee.tenantid ";
+			+ LEFT_JOIN + " " + SCHEMA_REPLACE_STRING + ".EG_EXPENSE_PARTY PAYEE ON bd.id = payee.parentid AND bd.tenantid = payee.tenantid ";
 
 	public static final String COUNT_WRAPPER = " SELECT COUNT(*) FROM ({INTERNAL_QUERY}) AS count ";
 
 	public static final String BILL_COUNT_QUERY = "SELECT distinct(bill.id) " +
-             "FROM eg_expense_bill bill ";
+             "FROM " + SCHEMA_REPLACE_STRING + ".eg_expense_bill bill ";
+
+	public static final String INVALID_TENANT_ID_ERR_CODE = "INVALID_TENANT_ID";
 }

backend/expense/src/main/java/org/egov/digit/expense/kafka/ExpenseProducer.java

@@ -1,22 +1,36 @@
 package org.egov.digit.expense.kafka;
 
+import lombok.extern.slf4j.Slf4j;
+import org.egov.common.utils.MultiStateInstanceUtil;
 import org.egov.tracer.kafka.CustomKafkaTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 // NOTE: If tracer is disabled change CustomKafkaTemplate to KafkaTemplate in autowiring
 
+@Slf4j
 @Service
 public class ExpenseProducer {
 
     private final CustomKafkaTemplate<String, Object> kafkaTemplate;
+    private final MultiStateInstanceUtil multiStateInstanceUtil;
 
     @Autowired
-    public ExpenseProducer(CustomKafkaTemplate<String, Object> kafkaTemplate) {
+    public ExpenseProducer(CustomKafkaTemplate<String, Object> kafkaTemplate, MultiStateInstanceUtil multiStateInstanceUtil) {
         this.kafkaTemplate = kafkaTemplate;
+        this.multiStateInstanceUtil = multiStateInstanceUtil;
     }
 
-    public void push(String topic, Object value) {
-        kafkaTemplate.send(topic, value);
+    /**
+     * Publishes a message to a Kafka topic based on the tenant-specific topic name.
+     *
+     * @param tenantId the unique identifier of the tenant for which the message is being published
+     * @param topic the base Kafka topic name to which the message needs to be pushed
+     * @param value the message payload to be sent to the Kafka topic
+     */
+    public void push(String tenantId, String topic, Object value) {
+        String updatedTopic = multiStateInstanceUtil.getStateSpecificTopicName(tenantId, topic);
+        log.info("The Kafka topic for the tenantId : {} is : {}", tenantId, updatedTopic);
+        this.kafkaTemplate.send(updatedTopic, value);
     }
 }

backend/expense/src/main/java/org/egov/digit/expense/repository/BillRepository.java

@@ -3,14 +3,19 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import org.egov.common.exception.InvalidTenantIdException;
+import org.egov.common.utils.MultiStateInstanceUtil;
 import org.egov.digit.expense.repository.querybuilder.BillQueryBuilder;
 import org.egov.digit.expense.repository.rowmapper.BillRowMapper;
 import org.egov.digit.expense.web.models.Bill;
 import org.egov.digit.expense.web.models.BillSearchRequest;
+import org.egov.tracer.model.CustomException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Repository;
 
+import static org.egov.digit.expense.config.Constants.INVALID_TENANT_ID_ERR_CODE;
+
 @Repository
 public class BillRepository {
 
@@ -20,23 +25,38 @@ public class BillRepository {
 
 	private final BillRowMapper searchBillRowMapper;
 
+	private final MultiStateInstanceUtil multiStateInstanceUtil;
+
 	@Autowired
-	public BillRepository(JdbcTemplate jdbcTemplate, BillQueryBuilder queryBuilder, BillRowMapper searchBillRowMapper) {
+	public BillRepository(JdbcTemplate jdbcTemplate, BillQueryBuilder queryBuilder, BillRowMapper searchBillRowMapper, MultiStateInstanceUtil multiStateInstanceUtil) {
 		this.jdbcTemplate = jdbcTemplate;
 		this.queryBuilder = queryBuilder;
 		this.searchBillRowMapper = searchBillRowMapper;
-	}
+        this.multiStateInstanceUtil = multiStateInstanceUtil;
+    }
 
 	public List<Bill> search(BillSearchRequest billSearchRequest, boolean isValidationSearch){
 
 		List<Object> preparedStatementValues = new ArrayList<>();
 		String queryStr = queryBuilder.getBillQuery(billSearchRequest, preparedStatementValues, false, isValidationSearch);
+		try {
+			// Applies schema replacement to the query string based on tenant ID
+			queryStr = multiStateInstanceUtil.replaceSchemaPlaceholder(queryStr, billSearchRequest.getBillCriteria().getTenantId());
+		} catch (InvalidTenantIdException e) {
+			throw new CustomException(INVALID_TENANT_ID_ERR_CODE, e.getMessage());
+		}
         return jdbcTemplate.query(queryStr, preparedStatementValues.toArray(), searchBillRowMapper);
 	}
 
 	public Integer searchCount(BillSearchRequest billSearchRequest){
 		List<Object> preparedStatementValues = new ArrayList<>();
 		String queryStr = queryBuilder.getSearchCountQueryString(billSearchRequest, preparedStatementValues);
+		try {
+			// Applies schema replacement to the query string based on tenant ID
+			queryStr = multiStateInstanceUtil.replaceSchemaPlaceholder(queryStr, billSearchRequest.getBillCriteria().getTenantId());
+		} catch (InvalidTenantIdException e) {
+			throw new CustomException(INVALID_TENANT_ID_ERR_CODE, e.getMessage());
+		}
 		return jdbcTemplate.queryForObject(queryStr, preparedStatementValues.toArray(), Integer.class);
 	}
 }

backend/expense/src/main/java/org/egov/digit/expense/repository/PaymentRepository.java

@@ -5,14 +5,19 @@
 
 import jakarta.validation.Valid;
 
+import org.egov.common.exception.InvalidTenantIdException;
+import org.egov.common.utils.MultiStateInstanceUtil;
 import org.egov.digit.expense.repository.querybuilder.PaymentQueryBuilder;
 import org.egov.digit.expense.repository.rowmapper.PaymentRowMapper;
 import org.egov.digit.expense.web.models.Payment;
 import org.egov.digit.expense.web.models.PaymentSearchRequest;
+import org.egov.tracer.model.CustomException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Repository;
 
+import static org.egov.digit.expense.config.Constants.INVALID_TENANT_ID_ERR_CODE;
+
 @Repository
 public class PaymentRepository {
 
@@ -22,23 +27,38 @@ public class PaymentRepository {
 
 	private final PaymentRowMapper paymentBillRowMapper;
 
+	private final MultiStateInstanceUtil multiStateInstanceUtil;
+
 	@Autowired
-	public PaymentRepository(JdbcTemplate jdbcTemplate, PaymentQueryBuilder queryBuilder, PaymentRowMapper paymentBillRowMapper) {
+	public PaymentRepository(JdbcTemplate jdbcTemplate, PaymentQueryBuilder queryBuilder, PaymentRowMapper paymentBillRowMapper, MultiStateInstanceUtil multiStateInstanceUtil) {
 		this.jdbcTemplate = jdbcTemplate;
 		this.queryBuilder = queryBuilder;
 		this.paymentBillRowMapper = paymentBillRowMapper;
-	}
+		this.multiStateInstanceUtil = multiStateInstanceUtil;
+    }
 
 	public List<Payment> search(@Valid PaymentSearchRequest paymentSearchRequest) {
 
 		List<Object> preparedStatementValues = new ArrayList<>();
 		String queryStr = queryBuilder.getPaymentQuery(paymentSearchRequest, preparedStatementValues, false);
-		return jdbcTemplate.query(queryStr, preparedStatementValues.toArray(), paymentBillRowMapper);
+        try {
+			// Applies schema replacement to the query string based on tenant ID
+            queryStr = multiStateInstanceUtil.replaceSchemaPlaceholder(queryStr, paymentSearchRequest.getPaymentCriteria().getTenantId());
+        } catch (InvalidTenantIdException e) {
+			throw new CustomException(INVALID_TENANT_ID_ERR_CODE, e.getMessage());
+        }
+        return jdbcTemplate.query(queryStr, preparedStatementValues.toArray(), paymentBillRowMapper);
 	}
 
 	public Integer count(@Valid PaymentSearchRequest paymentSearchRequest) {
 		List<Object> preparedStatementValues = new ArrayList<>();
 		String queryStr = queryBuilder.getSearchCountQueryString(paymentSearchRequest, preparedStatementValues);
+		try {
+			// Applies schema replacement to the query string based on tenant ID
+			queryStr = multiStateInstanceUtil.replaceSchemaPlaceholder(queryStr, paymentSearchRequest.getPaymentCriteria().getTenantId());
+		} catch (InvalidTenantIdException e) {
+			throw new CustomException(INVALID_TENANT_ID_ERR_CODE, e.getMessage());
+		}
 		return jdbcTemplate.queryForObject(queryStr, preparedStatementValues.toArray(), Integer.class);
 	}
 }

backend/expense/src/main/java/org/egov/digit/expense/service/BillService.java

@@ -64,6 +64,7 @@ public BillService(ExpenseProducer expenseProducer, Configuration config, BillVa
 	public BillResponse create(BillRequest billRequest) {
 
 		Bill bill = billRequest.getBill();
+		String tenantId = bill.getTenantId();
 		RequestInfo requestInfo = billRequest.getRequestInfo();
 		BillResponse response = null;
 
@@ -90,7 +91,7 @@ public BillResponse create(BillRequest billRequest) {
 			// produce full bill to different topic if indexing is required
 			produceBillsBatchWise(billRequest, config.getBillCreateTopic());
 		} else {
-			expenseProducer.push(config.getBillCreateTopic(), billRequest);
+			expenseProducer.push(tenantId, config.getBillCreateTopic(), billRequest);
 		}
 
 		response = BillResponse.builder()
@@ -109,6 +110,7 @@ public BillResponse create(BillRequest billRequest) {
 	public BillResponse update(BillRequest billRequest) {
 
 		Bill bill = billRequest.getBill();
+		String tenantId = bill.getTenantId();
 		RequestInfo requestInfo = billRequest.getRequestInfo();
 		BillResponse response = null;
 
@@ -131,7 +133,7 @@ public BillResponse update(BillRequest billRequest) {
 			 Every bill will have a batch of billDetails */
 			produceBillsBatchWise(billRequest, config.getBillUpdateTopic());
 		} else {
-			expenseProducer.push(config.getBillUpdateTopic(), billRequest);
+			expenseProducer.push(tenantId, config.getBillUpdateTopic(), billRequest);
 		}
 
 		response = BillResponse.builder()
@@ -199,13 +201,14 @@ private void enrichWfstatusForBills(List<Bill> bills, String tenantId, RequestIn
 	 */
 	private void produceBillsBatchWise(BillRequest billRequest, String topic) {
 		Bill bill = billRequest.getBill();
+		String tenantId = bill.getTenantId();
 		List<BillDetail> allBillDetails = new ArrayList<>(bill.getBillDetails());
 		// Breakdown the billDetails into batches and push to kafka
 		for (int i = 0; i < allBillDetails.size(); i += config.getBillBreakdownSize()) {
 			// Breakdown bill details into batches and push to kafka topic
 			List<BillDetail> currBatchBillDetails = allBillDetails.subList(i, Math.min(i + config.getBillBreakdownSize(), allBillDetails.size()));
 			bill.setBillDetails(currBatchBillDetails);
-			expenseProducer.push(topic, billRequest);
+			expenseProducer.push(tenantId, topic, billRequest);
 		}
 		bill.setBillDetails(allBillDetails);
 		log.info("All bill details pushed to kafka");

backend/expense/src/main/java/org/egov/digit/expense/service/NotificationService.java

@@ -38,6 +38,7 @@ public NotificationService(ExpenseProducer expenseProducer, NotificationUtil not
     public void sendNotificationForPurchaseBill(BillRequest billRequest){
 
         String action = billRequest.getWorkflow().getAction();
+        String tenantId = billRequest.getBill().getTenantId();
         if(action.equalsIgnoreCase(APPROVE_CODE) || action.equalsIgnoreCase(REJECT_CODE)) {
             String amount = String.valueOf(billRequest.getBill().getTotalAmount());
             String billNumber = billRequest.getBill().getBillNumber();
@@ -60,19 +61,20 @@ public void sendNotificationForPurchaseBill(BillRequest billRequest){
             }
             String customizedMessage = buildMessageReplaceVariables(message, billNumber, amount);
             SMSRequest smsRequest = SMSRequest.builder().mobileNumber(contactMobileNumber).message(customizedMessage).build();
-            expenseProducer.push(config.getSmsNotificationTopic(), smsRequest);
+            expenseProducer.push(tenantId, config.getSmsNotificationTopic(), smsRequest);
         }
     }
 
     public void sendNotificationForSupervisionBill(BillRequest billRequest){
+        String tenantId = billRequest.getBill().getTenantId();
         Map<String, String> cboDetails = notificationUtil.getCBOContactPersonDetails(billRequest);
         String amount = String.valueOf(billRequest.getBill().getTotalAmount());
         String billNumber = billRequest.getBill().getBillNumber();
         String message = getMessage(billRequest.getRequestInfo(), billRequest.getBill().getTenantId(), SUPERVISION_BILL_APPROVE_ON_CREATE_TO_CBO_LOCALIZATION_CODE);
         String contactMobileNumber = cboDetails.get(CONTACT_MOBILE_NUMBER);
         String customizedMessage = buildMessageReplaceVariables(message, billNumber, amount);
         SMSRequest smsRequest = SMSRequest.builder().mobileNumber(contactMobileNumber).message(customizedMessage).build();
-        expenseProducer.push(config.getSmsNotificationTopic(), smsRequest);
+        expenseProducer.push(tenantId, config.getSmsNotificationTopic(), smsRequest);
     }
 
     public String getMessage(RequestInfo requestInfo, String tenantId, String msgCode){

backend/expense/src/main/java/org/egov/digit/expense/service/PaymentService.java

@@ -59,10 +59,11 @@ public PaymentResponse create(@Valid PaymentRequest paymentRequest) {
 
         log.info("PaymentService::create");
         Payment payment = paymentRequest.getPayment();
+        String tenantId = payment.getTenantId();
         validator.validateCreateRequest(paymentRequest);
         enrichmentUtil.encrichCreatePayment(paymentRequest);
 
-        expenseProducer.push(config.getPaymentCreateTopic(), paymentRequest);
+        expenseProducer.push(tenantId, config.getPaymentCreateTopic(), paymentRequest);
         backUpdateBillForPayment(paymentRequest);
 
         return PaymentResponse.builder()
@@ -76,13 +77,14 @@ public PaymentResponse update(@Valid PaymentRequest paymentRequest) {
 
         log.info("PaymentService::update");
         Payment payment = paymentRequest.getPayment();
+        String tenantId = payment.getTenantId();
         List<Payment> paymentsFromSearch = validator.validateUpdateRequest(paymentRequest);
         enrichmentUtil.encrichUpdatePayment(paymentRequest, paymentsFromSearch.get(0));
         paymentRequest.setPayment(paymentsFromSearch.get(0));
         backUpdateBillForPayment(paymentRequest);
 
         /* only status update should be allowed here */
-        expenseProducer.push(config.getPaymentUpdateTopic(), paymentRequest);
+        expenseProducer.push(tenantId, config.getPaymentUpdateTopic(), paymentRequest);
         return PaymentResponse.builder()
                 .payments(Arrays.asList(payment))
                 .responseInfo(
@@ -113,6 +115,7 @@ private void backUpdateBillForPayment(@Valid PaymentRequest paymentRequest) {
         log.info("PaymentService::backUpdateBillForPayment");
         RequestInfo requestInfo = paymentRequest.getRequestInfo();
         Payment payment = paymentRequest.getPayment();
+        String tenantId = payment.getTenantId();
         String createdBy = paymentRequest.getRequestInfo().getUserInfo().getUuid();
         AuditDetails auditDetails = enrichmentUtil.getAuditDetails(createdBy, true);
 
@@ -181,7 +184,7 @@ private void backUpdateBillForPayment(@Valid PaymentRequest paymentRequest) {
                     .bill(bill)
                     .requestInfo(requestInfo)
                     .build();
-            expenseProducer.push(config.getBillUpdateTopic(), billRequest);
+            expenseProducer.push(tenantId, config.getBillUpdateTopic(), billRequest);
         }
     }