Refactor Kafka event listeners and email templates

.github/workflows/deploy.yml

@@ -0,0 +1,19 @@
+name: Deploy
+
+on:
+  push:
+    branches: [ main ]
+
+permissions:
+  id-token: write
+  contents: read
+  packages: write
+
+jobs:
+  deploy:
+    uses: Management-System-for-Rental-SEP490/.github/.github/workflows/deploy-java-service.yml@main
+    with:
+      service_name: notification-service
+      xmx: 512m
+    secrets:
+      GH_PACKAGES_TOKEN: ${{ secrets.GITHUB_TOKEN }}

src/main/java/com/isums/notificationservice/configurations/KafkaConsumerConfig.java

@@ -64,7 +64,8 @@ public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> dltKa
         );
 
         ExponentialBackOff backOff = new ExponentialBackOff(1_000L, 2.0);
-        backOff.setMaxAttempts(3);
+        backOff.setMaxInterval(60_000L);
+        backOff.setMaxAttempts(Long.MAX_VALUE);
 
         DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
 
@@ -74,7 +75,9 @@ public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> dltKa
                 tools.jackson.databind.exc.UnrecognizedPropertyException.class,
                 IllegalArgumentException.class,
                 org.springframework.messaging.converter.MessageConversionException.class,
-                PermanentEventFailureException.class
+                PermanentEventFailureException.class,
+                org.springframework.dao.DataIntegrityViolationException.class,
+                org.hibernate.exception.ConstraintViolationException.class
         );
 
         return handler;

src/main/java/com/isums/notificationservice/domains/events/ConfirmAndSendToTenantEvent.java

@@ -13,9 +13,11 @@
 @AllArgsConstructor
 @Builder
 public class ConfirmAndSendToTenantEvent {
-    private String messageId;
-    private UUID recipientUserId;
-    private UUID contractId;
+    private String messageId;
+    private UUID recipientUserId;
+    private String recipientEmail;
+    private String recipientName;
+    private UUID contractId;
     private String contractName;
     private String url;
     private String confirmUrl;
@@ -26,4 +28,4 @@ public class ConfirmAndSendToTenantEvent {
     // picks the right email template. Expected values: "VI", "VI_EN", "VI_JA".
     // Null = legacy event (pre BE-3) → fall back to VI.
     private String contractLanguage;
-}
+}

src/main/java/com/isums/notificationservice/domains/events/ContractReadyForLandlordSignatureEvent.java

@@ -12,6 +12,7 @@
 public class ContractReadyForLandlordSignatureEvent {
     private String messageId;
     private UUID contractId;
+    private UUID houseId;
     private UUID recipientUserId;
     private UUID tenantId;
     private String tenantName;

src/main/java/com/isums/notificationservice/domains/events/DepositRefundConfirmedEvent.java

@@ -0,0 +1,22 @@
+package com.isums.notificationservice.domains.events;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.UUID;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DepositRefundConfirmedEvent {
+    private UUID contractId;
+    private UUID houseId;
+    private UUID tenantId;
+    private String tenantEmail;
+    private Long refundAmount;
+    private String note;
+    private String messageId;
+}

src/main/java/com/isums/notificationservice/domains/events/DepositRefundPaidEvent.java

@@ -0,0 +1,25 @@
+package com.isums.notificationservice.domains.events;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.Instant;
+import java.util.UUID;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DepositRefundPaidEvent {
+    private UUID contractId;
+    private UUID houseId;
+    private UUID tenantId;
+    private String tenantEmail;
+    private Long refundAmount;
+    private String paymentMethod;
+    private String note;
+    private Instant paidAt;
+    private String messageId;
+}

src/main/java/com/isums/notificationservice/domains/events/UserActivatedEvent.java

@@ -16,6 +16,7 @@ public record UserActivatedEvent(
         String email,
         String name,
         String password,
+        String locale,
         String firstRentPaymentUrl,
         Long firstRentAmount,
         Instant firstRentDueDate

src/main/java/com/isums/notificationservice/infrastructures/kafka/ContractEventListener.java

@@ -4,12 +4,19 @@
 import com.isums.notificationservice.domains.events.ContractCancelledByTenantEvent;
 import com.isums.notificationservice.domains.events.ContractCompletedEvent;
 import com.isums.notificationservice.domains.events.ContractReadyForLandlordSignatureEvent;
+import com.isums.notificationservice.domains.events.DepositRefundConfirmedEvent;
 import com.isums.notificationservice.domains.events.InspectionDoneNotifyEvent;
 import com.isums.notificationservice.domains.events.InspectionScheduledEvent;
+import com.isums.notificationservice.domains.enums.LocaleType;
+import com.isums.notificationservice.infrastructures.abstracts.EmailService;
 import com.isums.notificationservice.infrastructures.abstracts.ManagerNotificationService;
+import com.isums.notificationservice.infrastructures.grpcs.UserGrpcClient;
 import com.isums.notificationservice.services.NotificationRecipientResolver;
+import com.isums.userservice.grpc.UserResponse;
 import common.kafkas.IdempotencyService;
 import common.kafkas.KafkaListenerHelper;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -29,6 +36,8 @@
 public class ContractEventListener {
 
     private final ManagerNotificationService notificationService;
+    private final EmailService emailService;
+    private final UserGrpcClient userGrpcClient;
     private final NotificationRecipientResolver recipientResolver;
     private final ObjectMapper objectMapper;
     private final IdempotencyService idempotencyService;
@@ -150,16 +159,33 @@ public void handleReadyForLandlordSignature(
             if (event.getDocumentId() != null && !event.getDocumentId().isBlank()) {
                 metadata.put("documentId", event.getDocumentId());
             }
+            if (event.getHouseId() != null) {
+                metadata.put("houseId", event.getHouseId().toString());
+            }
 
-            notificationService.send(
-                    event.getRecipientUserId(),
-                    NotificationCategory.CONTRACT_READY_FOR_LANDLORD_SIGNATURE,
-                    "Khách thuê đã xác nhận CCCD",
-                    "Hợp đồng " + contractLabel + " của " + tenantLabel + " đã sẵn sàng để chủ nhà ký.",
-                    "vi",
-                    "/contracts/" + event.getContractId(),
-                    metadata
-            );
+            List<UUID> recipientIds = recipientResolver.resolveLandlordAndManager(
+                    event.getHouseId(), event.getRecipientUserId());
+            if (recipientIds.isEmpty() && event.getRecipientUserId() != null) {
+                recipientIds = List.of(event.getRecipientUserId());
+            }
+
+            String title = "Khách thuê đã xác nhận CCCD";
+            String body = "Hợp đồng " + contractLabel + " của " + tenantLabel
+                    + " đã sẵn sàng để chủ nhà ký.";
+            String actionUrl = "/contracts/" + event.getContractId();
+
+            for (UUID recipientId : recipientIds) {
+                notificationService.send(
+                        recipientId,
+                        NotificationCategory.CONTRACT_READY_FOR_LANDLORD_SIGNATURE,
+                        title,
+                        body,
+                        "vi",
+                        actionUrl,
+                        metadata
+                );
+                sendReadyForLandlordSignatureEmail(recipientId, event, contractLabel, tenantLabel, actionUrl);
+            }
 
             idempotencyService.markProcessed(messageId);
             ack.acknowledge();
@@ -170,6 +196,92 @@ public void handleReadyForLandlordSignature(
         }
     }
 
+    @KafkaListener(topics = "contract.deposit-refund.confirmed",
+            groupId = "notification-group")
+    public void handleDepositRefundConfirmed(
+            ConsumerRecord<String, String> record, Acknowledgment ack) {
+
+        String messageId = kafkaHelper.extractMessageId(record);
+        try {
+            if (idempotencyService.isDuplicate(messageId)) {
+                ack.acknowledge();
+                return;
+            }
+
+            DepositRefundConfirmedEvent event = objectMapper.readValue(
+                    record.value(), DepositRefundConfirmedEvent.class);
+
+            Map<String, String> metadata = new HashMap<>();
+            metadata.put("contractId", event.getContractId().toString());
+            metadata.put("status", "DEPOSIT_REFUND_PENDING");
+            metadata.put("refundAmount", String.valueOf(event.getRefundAmount()));
+            if (event.getHouseId() != null) {
+                metadata.put("houseId", event.getHouseId().toString());
+            }
+            if (event.getTenantId() != null) {
+                metadata.put("tenantId", event.getTenantId().toString());
+            }
+
+            List<UUID> recipientIds = recipientResolver.resolveLandlordAndManager(event.getHouseId());
+            for (UUID recipientId : recipientIds) {
+                notificationService.send(
+                        recipientId,
+                        NotificationCategory.DEPOSIT_REFUND_CONFIRM,
+                        "Đã ghi nhận yêu cầu hoàn cọc",
+                        "Khoản hoàn cọc hợp đồng #" + shortId(event.getContractId())
+                                + " đã được tạo. Vui lòng theo dõi và xác nhận khi đã chuyển tiền.",
+                        "vi",
+                        "/contracts/" + event.getContractId() + "/deposit-refund",
+                        metadata
+                );
+            }
+
+            idempotencyService.markProcessed(messageId);
+            ack.acknowledge();
+            log.info("[Notification] handleDepositRefundConfirmed done messageId={} recipients={}",
+                    messageId, recipientIds.size());
+        } catch (Exception e) {
+            log.error("[Notification] handleDepositRefundConfirmed failed: {}", e.getMessage(), e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void sendReadyForLandlordSignatureEmail(
+            UUID recipientId,
+            ContractReadyForLandlordSignatureEvent event,
+            String contractLabel,
+            String tenantLabel,
+            String actionUrl) {
+        try {
+            UserResponse recipient = userGrpcClient.getUserById(recipientId);
+            if (recipient == null || recipient.getEmail() == null || recipient.getEmail().isBlank()) {
+                log.warn("[Notification] Skip ready email, recipient email missing recipientId={} contractId={}",
+                        recipientId, event.getContractId());
+                return;
+            }
+
+            emailService.sendEmail(
+                    recipient.getEmail(),
+                    "econtract_ready_for_landlord_signature",
+                    LocaleType.vi_VN,
+                    Map.of(
+                            "recipientName", safe(recipient.getName(), "anh/chị"),
+                            "tenantName", tenantLabel,
+                            "contractName", contractLabel,
+                            "contractNo", shortId(event.getContractId()),
+                            "actionUrl", actionUrl
+                    )
+            );
+        } catch (StatusRuntimeException e) {
+            if (isPermanentGrpcFailure(e)) {
+                log.warn("[Notification] Skip ready email, user lookup failed code={} recipientId={} contractId={}: {}",
+                        e.getStatus().getCode(), recipientId, event.getContractId(), e.getMessage());
+                return;
+            }
+            throw e;
+        }
+    }
+
     @KafkaListener(topics = "contract-completed-topic",
             groupId = "notification-group")
     public void handleContractCompleted(
@@ -287,5 +399,21 @@ public void handleContractCancelledByTenant(
             throw new RuntimeException(e);
         }
     }
-}
 
+    private static boolean isPermanentGrpcFailure(StatusRuntimeException e) {
+        Status.Code code = e.getStatus().getCode();
+        return code == Status.Code.NOT_FOUND
+                || code == Status.Code.INVALID_ARGUMENT
+                || code == Status.Code.PERMISSION_DENIED
+                || code == Status.Code.UNAUTHENTICATED
+                || code == Status.Code.FAILED_PRECONDITION;
+    }
+
+    private static String safe(String value, String fallback) {
+        return value != null && !value.isBlank() ? value.trim() : fallback;
+    }
+
+    private static String shortId(UUID id) {
+        return id != null ? id.toString().substring(0, 8).toUpperCase() : "N/A";
+    }
+}

src/main/java/com/isums/notificationservice/infrastructures/listeners/EContractEventListener.java

@@ -56,8 +56,11 @@ public void handleConfirmAndSendToTenant(ConsumerRecord<String, String> record,
                 return;
             }
 
-            if (event.getRecipientUserId() == null) {
-                log.error("[EContract] recipientUserId null, skip. contractId={}", event.getContractId());
+            String recipientEmail = safe(event.getRecipientEmail(), null);
+            String recipientName = safe(event.getRecipientName(), null);
+            if ((recipientEmail == null || recipientEmail.isBlank()) && event.getRecipientUserId() == null) {
+                log.error("[EContract] recipientUserId and recipientEmail null, skip. contractId={}",
+                        event.getContractId());
                 ack.acknowledge();
                 return;
             }
@@ -67,18 +70,39 @@ public void handleConfirmAndSendToTenant(ConsumerRecord<String, String> record,
                 return;
             }
 
-            UserResponse user = userGrpcClient.getUserById(event.getRecipientUserId());
-            if (user == null) {
-                log.error("[EContract] User not found userId={} contractId={}",
+            if ((recipientEmail == null || recipientEmail.isBlank()) && event.getRecipientUserId() != null) {
+                try {
+                    UserResponse user = userGrpcClient.getUserById(event.getRecipientUserId());
+                    if (user != null) {
+                        recipientEmail = safe(user.getEmail(), null);
+                        recipientName = safe(user.getName(), recipientName);
+                    }
+                } catch (StatusRuntimeException e) {
+                    if (isPermanentGrpcFailure(e)) {
+                        log.warn("[EContract] User lookup failed code={} userId={} contractId={}, using event email fallback: {}",
+                                e.getStatus().getCode(), event.getRecipientUserId(), event.getContractId(), e.getMessage());
+                        if (recipientEmail == null || recipientEmail.isBlank()) {
+                            throw new IllegalStateException(
+                                    "Recipient user not available yet and event has no email; retry later. userId="
+                                            + event.getRecipientUserId());
+                        }
+                    } else {
+                        throw e;
+                    }
+                }
+            }
+            if (recipientEmail == null || recipientEmail.isBlank()) {
+                log.error("[EContract] recipientEmail unavailable, skip. userId={} contractId={}",
                         event.getRecipientUserId(), event.getContractId());
+                idempotencyService.markProcessed(messageId);
                 ack.acknowledge();
                 return;
             }
 
             LocaleType locale = mapLocale(event.getContractLanguage());
 
             Map<String, Object> vars = new HashMap<>();
-            vars.put("tenantName", safe(user.getName(), fallbackTenantName(locale)));
+            vars.put("tenantName", safe(recipientName, fallbackTenantName(locale)));
             vars.put("contractName", safe(event.getContractName(), fallbackContractName(locale)));
             vars.put("contractNo", shortId(event.getContractId()));
             vars.put("propertyAddress", "N/A");
@@ -89,13 +113,13 @@ public void handleConfirmAndSendToTenant(ConsumerRecord<String, String> record,
             vars.put("expiresIn", expiresIn(locale));
             vars.put("landlordName", fallbackLandlordName(locale));
 
-            emailService.sendEmail(user.getEmail(), "econtract_view_confirm", locale, vars);
+            emailService.sendEmail(recipientEmail, "econtract_view_confirm", locale, vars);
 
             idempotencyService.markProcessed(messageId);
             ack.acknowledge();
 
             log.info("[EContract] Email sent messageId={} to={} contractId={}",
-                    messageId, user.getEmail(), event.getContractId());
+                    messageId, recipientEmail, event.getContractId());
 
         } catch (JacksonException e) {
             log.error("[EContract] Deserialization failed messageId={} raw={}: {}",