[add]: loan_repayment event

app/api/integrations/kafka.py

@@ -43,9 +43,9 @@ class KafkaIntegration:
 
 
     @staticmethod
-    def send_loan_request(loan_data, request_id):
+    def send_loan_request(loan_data, request_id, topic):
         """
-        Send loan request to PROCESS_PAYMENT topic
+        Send loan request to topic
 
         Args:
             loan_data: Loan request payload as dict
@@ -58,7 +58,7 @@ class KafkaIntegration:
 
             # Sending loan request message to Kafka
             producer.produce(
-                topic="PROCESS_PAYMENT",
+                topic=topic,
                 key=str(request_id),
                 value=json.dumps(loan_data).encode("utf-8"),
                 callback=KafkaIntegration.delivery_report,

app/api/services/base_service.py

@@ -3,6 +3,7 @@ from app.api.enums import TransactionType
 from flask import jsonify
 from marshmallow import ValidationError
 import logging
+from app.api.integrations import KafkaIntegration
 
 logger = logging.getLogger(__name__)
 
@@ -53,3 +54,8 @@ class BaseService:
             type=cls.TRANSACTION_TYPE,
             channel=validated_data.get("channel"),
         )
+    
+    @classmethod
+    def async_send_to_kafka(cls, loan_data, request_id, topic):
+        KafkaIntegration.send_loan_request(loan_data = loan_data, request_id = request_id, topic = topic)
+        KafkaIntegration.flush()

app/api/services/provide_loan.py

@@ -5,7 +5,6 @@ from app.api.services.base_service import BaseService
 from app.api.enums import TransactionType   
 from app.utils.logger import logger
 from app.api.schemas.provide_loan import ProvideLoanSchema
-from app.api.integrations import KafkaIntegration
 from threading import Thread
 
 
@@ -58,7 +57,7 @@ class ProvideLoanService(BaseService):
 
             # KafkaIntegration.send_loan_request(loan_data = response_data, request_id = request_id)
             # Call Kafka in a background thread
-            thread = Thread(target=ProvideLoanService.async_send_to_kafka, args=(response_data, request_id))
+            thread = Thread(target=ProvideLoanService.async_send_to_kafka, args=(response_data, request_id, "PROCESS_PAYMENT"))
             thread.start()
 
             return response_data
@@ -85,8 +84,6 @@ class ProvideLoanService(BaseService):
                 }) , 500
         
     
-    def async_send_to_kafka(loan_data, request_id):
-        KafkaIntegration.send_loan_request(loan_data = loan_data, request_id = request_id)
-        KafkaIntegration.flush()
+ 
 
 

app/api/services/repayment.py

@@ -3,7 +3,8 @@ from marshmallow import ValidationError
 from app.utils.logger import logger
 from app.api.schemas.repayment import RepaymentSchema
 from app.api.services.base_service import BaseService
-from app.api.enums import TransactionType      
+from app.api.enums import TransactionType 
+from threading import Thread     
 
 class RepaymentService(BaseService):
     TRANSACTION_TYPE = TransactionType.REPAYMENT 
@@ -23,6 +24,11 @@ class RepaymentService(BaseService):
             validated_data = RepaymentService.validate_data(data, RepaymentSchema())
             account_id = validated_data.get('accountId')
             customer_id =  validated_data.get('customerId')
+            customer = RepaymentService.get_or_create_customer(validated_data)
+            account = customer.accounts[0]
+            validated_data['accountId'] = account.id
+            request_id =  validated_data.get('requestId')
+
 
             if (RepaymentService.validate_account_ownership(account_id = account_id, customer_id = customer_id)):
                 transaction = RepaymentService.log_transaction(validated_data = validated_data)
@@ -51,6 +57,10 @@ class RepaymentService(BaseService):
             #     message="Repayment processed successfully"
             # )
 
+            # Call Kafka in a background thread
+            thread = Thread(target=RepaymentService.async_send_to_kafka, args=(response_data, request_id, "LOAN_REPAYMENT"))
+            thread.start()
+
             return response_data
 
         except ValidationError as err: