refactor(be): 라이더 배달 로직 리팩토링 (#141)

* refactor: rider 반경 검색 해당 아이디 중 상태가 OFF인 아이디 검색 제외

* refactor: kafka 설계 redis pub/sub으로 전환

* refactor: 배달료 계산 공식 전환 -> 기본 배달료 + a

Author: Chaeruin

backend/src/main/java/com/deliveranything/domain/delivery/event/event/RedisPublisher.java

@@ -0,0 +1,30 @@
+package com.deliveranything.domain.delivery.event.event;
+
+import com.deliveranything.domain.delivery.event.dto.RiderNotificationDto;
+import com.deliveranything.global.exception.CustomException;
+import com.deliveranything.global.exception.ErrorCode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.List;
+import lombok.RequiredArgsConstructor;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+@RequiredArgsConstructor
+public class RedisPublisher {
+
+  private final StringRedisTemplate redisTemplate;
+  private final ObjectMapper objectMapper;
+
+  public void publish(List<RiderNotificationDto> dtoList) {
+    dtoList.forEach(dto -> {
+      try {
+        String channel = "order-events";
+        String message = objectMapper.writeValueAsString(dto);
+        redisTemplate.convertAndSend(channel, message);
+      } catch (Exception e) {
+        throw new CustomException(ErrorCode.REDIS_MESSAGE_PROCESSING_ERROR);
+      }
+    });
+  }
+}

backend/src/main/java/com/deliveranything/domain/delivery/event/event/RiderEtaPublisher.java

@@ -1,23 +1,22 @@
 package com.deliveranything.domain.delivery.handler;
 
 import com.deliveranything.domain.delivery.event.dto.OrderDeliveryCreatedEvent;
-import com.deliveranything.domain.delivery.event.event.RiderEtaPublisher;
+import com.deliveranything.domain.delivery.event.event.RedisPublisher;
 import com.deliveranything.domain.delivery.service.OrderNotificationService;
 import lombok.RequiredArgsConstructor;
-import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Component;
 
 @Component
 @RequiredArgsConstructor
 public class OrderDeliveryEventHandler {
 
   private final OrderNotificationService orderNotificationService;
-  private final RiderEtaPublisher riderEtaPublisher;
+  private final RedisPublisher redisPublisher; // Kafka → Redis 변경
 
-  @KafkaListener(topics = "order-events", groupId = "delivery-service")
+  @EventListener // @KafkaListener → @EventListener로 변경
   public void handleOrderCreated(OrderDeliveryCreatedEvent orderEvent) {
-    // Kafka 발행 분리
     orderNotificationService.processOrderEvent(orderEvent)
-        .subscribe(riderEtaPublisher::publish);
+        .subscribe(redisPublisher::publish); // Kafka → Redis 변경
   }
 }

backend/src/main/java/com/deliveranything/domain/delivery/handler/OrderDeliveryEventHandler.java

@@ -0,0 +1,36 @@
+package com.deliveranything.domain.delivery.handler;
+
+import com.deliveranything.domain.delivery.event.dto.RiderNotificationDto;
+import com.deliveranything.domain.delivery.websocket.RiderWebSocketPublisher;
+import com.deliveranything.global.exception.CustomException;
+import com.deliveranything.global.exception.ErrorCode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class RedisSubscriber implements MessageListener {
+
+  private final RiderWebSocketPublisher webSocketPublisher;
+  private final ObjectMapper objectMapper;
+
+  @Override
+  public void onMessage(Message message, byte[] pattern) {
+    try {
+      String channel = new String(message.getChannel());
+      String body = new String(message.getBody());
+
+      if ("order-events".equals(channel)) {
+        RiderNotificationDto dto = objectMapper.readValue(body, RiderNotificationDto.class);
+        webSocketPublisher.publishToRider(dto.riderId(), dto);
+      }
+    } catch (Exception e) {
+      throw new CustomException(ErrorCode.REDIS_MESSAGE_PROCESSING_ERROR);
+    }
+  }
+}

backend/src/main/java/com/deliveranything/domain/delivery/handler/RedisSubscriber.java

@@ -31,7 +31,7 @@ public Mono<List<RiderNotificationDto>> processOrderEvent(OrderDeliveryCreatedEv
 
     return distanceKmMono.flatMap(distanceMap -> {
       double distanceKm = distanceMap.getOrDefault("distance", 0.0);
-      int expectedCharge = (int) (distanceKm * 1000); // 임시 배달료 계산
+      int expectedCharge = 1000 + (int) (distanceKm * 500); // 임시 배달료 계산
 
       // 2️⃣ 반경 내 라이더 ETA 조회
       return reactiveRiderEtaService.findNearbyRidersEta(customerLat, customerLon, 3.0)

backend/src/main/java/com/deliveranything/domain/delivery/handler/RiderEtaConsumer.java

@@ -2,6 +2,11 @@
 
 import static com.deliveranything.domain.delivery.service.RiderLocationService.RIDER_GEO_KEY;
 
+import com.deliveranything.domain.user.entity.profile.RiderProfile;
+import com.deliveranything.domain.user.enums.RiderToggleStatus;
+import com.deliveranything.domain.user.repository.RiderProfileRepository;
+import com.deliveranything.global.exception.CustomException;
+import com.deliveranything.global.exception.ErrorCode;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -23,6 +28,7 @@ public class ReactiveRiderEtaService {
 
   private final StringRedisTemplate redisTemplate;
   private final EtaService etaService;
+  private final RiderProfileRepository riderProfileRepository;
 
   //반경 내 라이더 검색 후 ETA 계산
   // Kafka 발행(2차 구현)
@@ -46,6 +52,11 @@ public Mono<Map<String, Double>> findNearbyRidersEta(double customerLat, double
 
     for (GeoResult<RedisGeoCommands.GeoLocation<String>> result : nearbyRiders) {
       RedisGeoCommands.GeoLocation<String> loc = result.getContent();
+      RiderProfile riderProfile = riderProfileRepository.findById(Long.parseLong(loc.getName()))
+          .orElseThrow(() -> new CustomException(ErrorCode.RIDER_NOT_FOUND));
+      if (riderProfile.getToggleStatus() == RiderToggleStatus.OFF) {
+        continue; // OFF 상태 라이더는 제외
+      }
       riderIds.add(loc.getName());
       riderPoints.add(loc.getPoint());
     }

backend/src/main/java/com/deliveranything/domain/delivery/service/OrderNotificationService.java

@@ -12,6 +12,7 @@ public class RiderWebSocketPublisher {
 
   private final SimpMessagingTemplate messagingTemplate;
 
+  // 여기 SSE 변경 보류
   public void publishToRider(String riderId, RiderNotificationDto dto) {
     String destination = "/topic/rider/" + riderId;
     messagingTemplate.convertAndSend(destination, dto);

backend/src/main/java/com/deliveranything/domain/delivery/service/ReactiveRiderEtaService.java

@@ -1,17 +1,24 @@
 package com.deliveranything.global.config;
 
+import com.deliveranything.domain.delivery.handler.RedisSubscriber;
+import lombok.RequiredArgsConstructor;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 
 @Configuration
+@RequiredArgsConstructor
 public class RedisConfig {
 
+  private final RedisSubscriber redisSubscriber;
+
   @Bean
   public LettuceConnectionFactory redisConnectionFactory() {
     return new LettuceConnectionFactory(); // application.yml 기반으로 host/port 세팅
@@ -34,4 +41,17 @@ public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connec
   public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory connectionFactory) {
     return new StringRedisTemplate(connectionFactory);
   }
+
+  // Redis Pub/Sub용 추가
+  @Bean
+  public RedisMessageListenerContainer redisMessageListenerContainer(
+      RedisConnectionFactory connectionFactory) {
+    RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+    container.setConnectionFactory(connectionFactory);
+
+    // order-events 채널 구독
+    container.addMessageListener(redisSubscriber, new PatternTopic("order-events"));
+
+    return container;
+  }
 }

backend/src/main/java/com/deliveranything/domain/delivery/websocket/RiderWebSocketPublisher.java

@@ -43,7 +43,11 @@ public enum ErrorCode {
   STORE_CATEGORY_NOT_FOUND(HttpStatus.NOT_FOUND, "STORE-CATEGORY-404", "상점 카테고리를 찾을 수 없습니다."),
 
   // SSE 관련 오류
-  SSE_SUBSCRIBE_UNAVAILABLE(HttpStatus.SERVICE_UNAVAILABLE, "SSE-503", "SSE 연결에 실패했습니다.");
+  SSE_SUBSCRIBE_UNAVAILABLE(HttpStatus.SERVICE_UNAVAILABLE, "SSE-503", "SSE 연결에 실패했습니다."),
+
+  // Redis 관련 오류
+  REDIS_MESSAGE_PROCESSING_ERROR(HttpStatus.INTERNAL_SERVER_ERROR, "REDIS-500",
+      "Redis 메시지 처리 중 오류가 발생했습니다.");
 
   private final HttpStatus httpStatus;
   private final String code;