feat: Transactional Outbox 기반 메시지 발행 신뢰성 보강

docker-compose.yml

@@ -23,6 +23,7 @@ services:
       RABBITMQ_PREFETCH: ${RABBITMQ_PREFETCH:-250}
       MOCK_PROVIDER_FORCE_FAILURE: ${MOCK_PROVIDER_FORCE_FAILURE:-false}
       MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS: ${MOCK_PROVIDER_FAIL_FIRST_ATTEMPTS:-0}
+      OUTBOX_PUBLISH_INTERVAL_MS: ${OUTBOX_PUBLISH_INTERVAL_MS:-1000}
 
     healthcheck:
       test:

src/main/java/com/backendsystemdesignlab/notification/NotificationSystemApplication.java

@@ -2,8 +2,10 @@
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 @SpringBootApplication
+@EnableScheduling
 public class NotificationSystemApplication {
 
 	public static void main(String[] args) {

src/main/java/com/backendsystemdesignlab/notification/messaging/DeliveryMessageHandler.java

@@ -23,6 +23,8 @@ public class DeliveryMessageHandler {
 
     public void handle(DeliveryMessage message) {
 
+        if (transactionService.isAlreadyProcessed(message.deliveryId())) return;
+
         boolean success = send(message);
 
         if (success) {
@@ -43,12 +45,14 @@ public void handle(DeliveryMessage message) {
 
     private boolean send(DeliveryMessage message) {
 
+        String idempotencyKey = "notification-delivery-" + message.deliveryId();
+
         try {
             ProviderResult result =
                     switch (message.channel()) {
-                        case PUSH -> pushProvider.send(message.destination());
-                        case SMS -> smsProvider.send(message.destination());
-                        case EMAIL -> emailProvider.send(message.destination());
+                        case PUSH -> pushProvider.send(message.destination(), idempotencyKey);
+                        case SMS -> smsProvider.send(message.destination(), idempotencyKey);
+                        case EMAIL -> emailProvider.send(message.destination(), idempotencyKey);
                     };
             return result.success();
         } catch (RuntimeException e) {

src/main/java/com/backendsystemdesignlab/notification/notification/provider/EmailProvider.java

@@ -1,5 +1,5 @@
 package com.backendsystemdesignlab.notification.notification.provider;
 
 public interface EmailProvider {
-    ProviderResult send(String email);
+    ProviderResult send(String email, String idempotencyKey);
 }

src/main/java/com/backendsystemdesignlab/notification/notification/provider/PushProvider.java

@@ -1,5 +1,5 @@
 package com.backendsystemdesignlab.notification.notification.provider;
 
 public interface PushProvider {
-    ProviderResult send(String deviceToken);
+    ProviderResult send(String deviceToken, String idempotencyKey);
 }

src/main/java/com/backendsystemdesignlab/notification/notification/provider/SmsProvider.java

@@ -1,5 +1,5 @@
 package com.backendsystemdesignlab.notification.notification.provider;
 
 public interface SmsProvider {
-    ProviderResult send(String phoneNumber);
+    ProviderResult send(String phoneNumber, String idempotencyKey);
 }

src/main/java/com/backendsystemdesignlab/notification/notification/provider/mock/MockEmailProvider.java

@@ -6,6 +6,7 @@
 import org.springframework.stereotype.Component;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -14,6 +15,7 @@ public class MockEmailProvider implements EmailProvider {
 
     private final boolean forceFailure;
     private final int failFirstAttempts;
+    private final Set<String> processedKeys = ConcurrentHashMap.newKeySet();
 
     private final Map<String, AtomicInteger> attempts = new ConcurrentHashMap<>();
 
@@ -25,8 +27,13 @@ public MockEmailProvider(
     }
 
     @Override
-    public ProviderResult send(String email) {
+    public ProviderResult send(String email,  String idempotencyKey) {
         simulateDelay();
+
+        if (processedKeys.contains(idempotencyKey)) {
+            return new ProviderResult(true);
+        }
+
         if (forceFailure) return new ProviderResult(false);
 
         int currentAttempt = attempts.computeIfAbsent(email, key -> new AtomicInteger()).incrementAndGet(); // 증가 연산을 원자적 처리
@@ -35,6 +42,7 @@ public ProviderResult send(String email) {
             return new ProviderResult(false);
         }
 
+        processedKeys.add(idempotencyKey);
         return new ProviderResult(true);
     }
 

src/main/java/com/backendsystemdesignlab/notification/notification/provider/mock/MockPushProvider.java

@@ -5,19 +5,29 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
 @Component
 public class MockPushProvider implements PushProvider {
 
     private final boolean forceFailure;
+    private final Set<String> processedKeys = ConcurrentHashMap.newKeySet();
 
     public MockPushProvider(@Value("${mock.provider.force-failure:false}") boolean forceFailure) {
         this.forceFailure = forceFailure;
     }
 
     @Override
-    public ProviderResult send(String deviceToken) {
+    public ProviderResult send(String deviceToken, String idempotencyKey) {
         simulateDelay();
+
+        if (processedKeys.contains(idempotencyKey)) {
+            return new ProviderResult(true);
+        }
+
         if (forceFailure) return new ProviderResult(false);
+        processedKeys.add(idempotencyKey);
         return new ProviderResult(true);
     }
 

src/main/java/com/backendsystemdesignlab/notification/notification/provider/mock/MockSmsProvider.java

@@ -5,19 +5,29 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
 @Component
 public class MockSmsProvider implements SmsProvider {
 
     private final boolean forceFailure;
+    private final Set<String> processedKeys =
+            ConcurrentHashMap.newKeySet();
 
     public MockSmsProvider(@Value("${mock.provider.force-failure:false}") boolean forceFailure) {
         this.forceFailure = forceFailure;
     }
 
     @Override
-    public ProviderResult send(String deviceToken) {
+    public ProviderResult send(String phoneNumber, String idempotencyKey) {
         simulateDelay();
+        if (processedKeys.contains(idempotencyKey)) {
+            return new ProviderResult(true);
+        }
         if (forceFailure) return new ProviderResult(false);
+
+        processedKeys.add(idempotencyKey);
         return new ProviderResult(true);
     }
 

src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationService.java

@@ -28,7 +28,7 @@ public SendNotificationResponse send(SendNotificationRequest request) {
             );
         }
 
-        deliveryPublisher.publishAll(prepared.notificationId(), prepared.deliveries());
+//        deliveryPublisher.publishAll(prepared.notificationId(), prepared.deliveries());
 
         return new SendNotificationResponse(
                 prepared.notificationId(),

src/main/java/com/backendsystemdesignlab/notification/notification/service/NotificationTransactionService.java

@@ -8,6 +8,8 @@
 import com.backendsystemdesignlab.notification.notification.dto.SendNotificationRequest;
 import com.backendsystemdesignlab.notification.notification.repository.NotificationDeliveryRepository;
 import com.backendsystemdesignlab.notification.notification.repository.NotificationRepository;
+import com.backendsystemdesignlab.notification.outbox.OutboxEvent;
+import com.backendsystemdesignlab.notification.outbox.OutboxEventRepository;
 import com.backendsystemdesignlab.notification.user.domain.NotificationChannel;
 import com.backendsystemdesignlab.notification.user.domain.NotificationPreference;
 import com.backendsystemdesignlab.notification.user.domain.User;
@@ -22,6 +24,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 @Service
@@ -33,6 +36,7 @@ public class NotificationTransactionService {
     private final NotificationPreferenceRepository preferenceRepository;
     private final NotificationRepository notificationRepository;
     private final NotificationDeliveryRepository deliveryRepository;
+    private final OutboxEventRepository outboxEventRepository;
 
     @Transactional
     public PreparedNotification prepare(SendNotificationRequest request) {
@@ -93,6 +97,19 @@ public PreparedNotification prepare(SendNotificationRequest request) {
 
         deliveryRepository.flush(); // DB의 ID를 얻기 위함 (delivery.getId())
 
+        for (NotificationDelivery delivery : deliveries) {
+
+            OutboxEvent outboxEvent = new OutboxEvent(
+                    UUID.randomUUID().toString(),
+                    notification.getId(),
+                    delivery.getId(),
+                    delivery.getChannel(),
+                    delivery.getDestination()
+            );
+
+            outboxEventRepository.save(outboxEvent);
+        }
+
         notification.startProcessing();
 
         List<DeliveryCommand> commands = deliveries.stream()
@@ -216,4 +233,29 @@ private void createEmailDelivery(User user, Notification notification, List<Noti
                 )
         );
     }
+
+    @Transactional(readOnly = true)
+    public boolean isAlreadyProcessed(Long deliveryId) {
+        NotificationDelivery delivery = deliveryRepository.findById(deliveryId)
+                .orElseThrow(() -> new IllegalArgumentException("Delivery not found: " + deliveryId));
+
+        return delivery.getStatus() == DeliveryStatus.SENT || delivery.getStatus() == DeliveryStatus.FAILED;
+    }
+
+    @Transactional
+    public void recordPublishFinalFailure(Long notificationId, Long deliveryId) {
+        Notification notification = notificationRepository.findByIdForUpdate(notificationId)
+                .orElseThrow(() -> new IllegalArgumentException("알림을 찾을 수 없습니다."));
+
+        NotificationDelivery delivery = deliveryRepository.findById(deliveryId)
+                .orElseThrow(() -> new IllegalArgumentException("전송 정보를 찾을 수 없습니다."));
+
+        if (delivery.getStatus() != DeliveryStatus.PENDING) {
+            return;
+        }
+
+        delivery.markFailed();
+
+        updateNotificationStatus(notificationId, notification);
+    }
 }

src/main/java/com/backendsystemdesignlab/notification/outbox/OutboxEvent.java

@@ -0,0 +1,89 @@
+package com.backendsystemdesignlab.notification.outbox;
+
+import com.backendsystemdesignlab.notification.user.domain.NotificationChannel;
+import jakarta.persistence.*;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+@Entity
+@Table(
+        name = "outbox_events",
+        indexes = {
+                @Index(
+                        name = "idx_outbox_status_created_at",
+                        columnList = "status, created_at"
+                )
+        }
+)
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+public class OutboxEvent {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(nullable = false, unique = true)
+    private String messageId;
+
+    @Column(nullable = false)
+    private Long notificationId;
+
+    @Column(nullable = false)
+    private Long deliveryId;
+
+    @Enumerated(EnumType.STRING)
+    @Column(nullable = false)
+    private NotificationChannel channel;
+
+    @Column(nullable = false)
+    private String destination;
+
+    @Enumerated(EnumType.STRING)
+    @Column(nullable = false)
+    private OutboxStatus status;
+
+    @Column(nullable = false)
+    private LocalDateTime createdAt;
+
+    private LocalDateTime publishedAt;
+
+    @Column(nullable = false)
+    private int attemptCount;
+
+    private String lastError;
+
+    public OutboxEvent(
+            String messageId,
+            Long notificationId,
+            Long deliveryId,
+            NotificationChannel channel,
+            String destination
+    ) {
+        this.messageId = messageId;
+        this.notificationId = notificationId;
+        this.deliveryId = deliveryId;
+        this.channel = channel;
+        this.destination = destination;
+        this.status = OutboxStatus.PENDING;
+        this.attemptCount = 0;
+        this.createdAt = LocalDateTime.now();
+    }
+
+    public void markPublished() {
+        this.status = OutboxStatus.PUBLISHED;
+        this.publishedAt = LocalDateTime.now();
+    }
+
+    public void recordFailure(String error) {
+        this.attemptCount++;
+        this.lastError = error;
+    }
+
+    public void markFailed() {
+        this.status = OutboxStatus.FAILED;
+    }
+}

src/main/java/com/backendsystemdesignlab/notification/outbox/OutboxEventRepository.java

@@ -0,0 +1,10 @@
+package com.backendsystemdesignlab.notification.outbox;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.List;
+
+public interface OutboxEventRepository extends JpaRepository<OutboxEvent, Long> {
+
+    List<OutboxEvent> findTop100ByStatusOrderByCreatedAtAsc(OutboxStatus status);
+}