diff --git a/build.gradle b/build.gradle index 5cd07d9..113a24f 100644 --- a/build.gradle +++ b/build.gradle @@ -32,6 +32,7 @@ repositories { } dependencies { + implementation 'org.springframework.kafka:spring-kafka' implementation 'org.springframework.ai:spring-ai-starter-model-google-genai' implementation 'org.jetbrains.kotlin:kotlin-reflect' implementation 'org.springframework.boot:spring-boot-starter-actuator' @@ -39,6 +40,7 @@ dependencies { implementation 'io.github.cdimascio:dotenv-java:3.0.0' implementation 'org.springframework.boot:spring-boot-starter-data-jpa' implementation 'org.springframework.boot:spring-boot-starter-security' + implementation 'org.redisson:redisson:3.39.0' implementation 'io.jsonwebtoken:jjwt-api:0.12.6' runtimeOnly 'io.jsonwebtoken:jjwt-impl:0.12.6' runtimeOnly 'io.jsonwebtoken:jjwt-jackson:0.12.6' diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index a7922aa..191fb28 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -14,6 +14,44 @@ services: - catchtable-data:/var/lib/postgresql/data - ./data/init-data.sql:/docker-entrypoint-initdb.d/init-data.sql - ./data:/tmp/data + networks: + - catchtable-dev-net + + kafka: + image: apache/kafka:3.7.0 + container_name: catchtable-kafka-dev + user: "0" + ports: + - "9092:9092" + environment: + KAFKA_CLUSTER_ID: "MkU3OEV5NURaZW52OTYyM2RjY1JqcA" + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: 'broker,controller' + KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' + + KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093' + + KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://localhost:9092' + + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT' + KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' + + KAFKA_CONTROLLER_QUORUM_VOTERS: '1@127.0.0.1:9093' + volumes: + - kafka-data-dev:/tmp/kafka-logs + networks: + - catchtable-dev-net + +networks: + catchtable-dev-net: + driver: bridge + + redis: + image: redis:7-alpine + container_name: catchtable-redis + ports: + - "${REDIS_PORT:-6379}:6379" volumes: catchtable-data: + kafka-data-dev: \ No newline at end of file diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 8e747c9..322743e 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -7,21 +7,22 @@ services: env_file: .env expose: - "8080" - # t3.small(2GB) 환경. 모니터링 스택과 공존하기 위해 app은 600M로 제한. - # OTel Agent가 약 80MB 추가 사용하므로 500M → 600M로 상향. + # t3.small 메모리 한도 고려한 app 한도 (OTel Agent 80MB 포함 600M) environment: - # JVM 옵션은 Dockerfile의 ENV에 기본값으로 정의됨 (javaagent + 메모리 튜닝). - # 여기선 OpenTelemetry Agent의 송신 대상만 지정. OTEL_SERVICE_NAME: catchtable-backend - OTEL_EXPORTER_OTLP_ENDPOINT: http://jaeger:4317 + # OTel Collector로 전송 (Collector에서 path 기반 노이즈 필터링 후 Jaeger로 전달) + OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4317 OTEL_EXPORTER_OTLP_PROTOCOL: grpc OTEL_TRACES_EXPORTER: otlp OTEL_METRICS_EXPORTER: none # 메트릭은 Micrometer/Prometheus로 처리 OTEL_LOGS_EXPORTER: none # 로그는 Promtail/Loki로 처리 OTEL_RESOURCE_ATTRIBUTES: deployment.environment=production - # 샘플링: 전체 요청 대비 트레이스 수집 비율 (1.0=100%, 부하 시 0.1로 낮추기 권장) + # 샘플링 100% — 노이즈 제거는 Collector에서 path 기반으로 처리 OTEL_TRACES_SAMPLER: parentbased_traceidratio OTEL_TRACES_SAMPLER_ARG: "1.0" + OTEL_INSTRUMENTATION_SPRING_SCHEDULING_ENABLED: "false" + OTEL_INSTRUMENTATION_LOGBACK_APPENDER_ENABLED: "true" + OTEL_INSTRUMENTATION_LOGBACK_MDC_ENABLED: "true" deploy: resources: limits: @@ -56,6 +57,10 @@ services: depends_on: app: condition: service_healthy + deploy: + resources: + limits: + memory: 50M restart: unless-stopped networks: - catchtable-net @@ -67,12 +72,12 @@ services: - ./certbot/conf:/etc/letsencrypt - ./certbot/www:/var/www/certbot entrypoint: "/bin/sh -c 'trap exit TERM; while :; do certbot renew --quiet; sleep 12h & wait $${!}; done;'" + deploy: + resources: + limits: + memory: 30M restart: unless-stopped - # ============================================================ - # 모니터링 스택 (Prometheus + Loki + Promtail + Jaeger + Grafana) - # ============================================================ - prometheus: image: prom/prometheus:v2.55.0 container_name: catchtable-prometheus @@ -106,7 +111,7 @@ services: deploy: resources: limits: - memory: 250M + memory: 200M restart: unless-stopped networks: - catchtable-net @@ -130,6 +135,24 @@ services: networks: - catchtable-net + otel-collector: + image: otel/opentelemetry-collector-contrib:0.115.0 + container_name: catchtable-otel-collector + volumes: + - ./monitoring/otel-collector-config.yml:/etc/otelcol-contrib/config.yaml:ro + command: ["--config=/etc/otelcol-contrib/config.yaml"] + expose: + - "4317" # OTLP gRPC (app에서 전송) + depends_on: + - jaeger + deploy: + resources: + limits: + memory: 150M + restart: unless-stopped + networks: + - catchtable-net + jaeger: image: jaegertracing/all-in-one:1.62.0 container_name: catchtable-jaeger @@ -140,7 +163,7 @@ services: BADGER_EPHEMERAL: "false" BADGER_DIRECTORY_VALUE: /badger/data BADGER_DIRECTORY_KEY: /badger/key - BADGER_SPAN_STORE_TTL: 168h0m0s + BADGER_SPAN_STORE_TTL: 72h0m0s volumes: - jaeger-data:/badger ports: @@ -151,7 +174,7 @@ services: deploy: resources: limits: - memory: 350M + memory: 250M restart: unless-stopped networks: - catchtable-net @@ -181,57 +204,71 @@ services: - catchtable-net # ============================================================ - # 고도화 작업 시작 시 주석 해제 (5대 고도화 항목 1, 3번 진행 시) + # 고도화 인프라 (Kafka는 해당 작업 시작 시 주석 해제) # ============================================================ - # redis: - # image: redis:7-alpine - # container_name: catchtable-redis - # command: - # - redis-server - # - --maxmemory - # - 80mb - # - --maxmemory-policy - # - allkeys-lru - # - --appendonly - # - "yes" - # expose: - # - "6379" - # volumes: - # - redis-data:/data - # deploy: - # resources: - # limits: - # memory: 100M - # restart: unless-stopped - # networks: - # - catchtable-net - - # kafka: - # image: bitnami/kafka:3.8 - # container_name: catchtable-kafka - # environment: - # # KRaft 모드 (Zookeeper 불필요, 메모리 절약) - # KAFKA_CFG_NODE_ID: 0 - # KAFKA_CFG_PROCESS_ROLES: controller,broker - # KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093 - # KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 - # KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 - # KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER - # KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT - # # 메모리 다이어트 - # KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m" - # expose: - # - "9092" - # volumes: - # - kafka-data:/bitnami/kafka - # deploy: - # resources: - # limits: - # memory: 400M - # restart: unless-stopped - # networks: - # - catchtable-net + redis: + image: redis:7-alpine + container_name: catchtable-redis + command: + - redis-server + - --maxmemory + - 80mb + - --maxmemory-policy + - allkeys-lru + - --appendonly + - "yes" + expose: + - "6379" + volumes: + - redis-data:/data + deploy: + resources: + limits: + memory: 100M + restart: unless-stopped + networks: + - catchtable-net + + kafka: + image: apache/kafka:3.7.0 + container_name: catchtable-kafka-prod + user: "0" # 볼륨 마운트 폴더에 로그를 쓰고 지울 수 있도록 최고 관리자 권한 부여 + + # 외부(내 PC 등)로 포트를 무방비하게 노출하지 않고, 오직 'catchtable-net' 내부에서만 통신하도록 격리 + expose: + - "9092" + - "9093" + + environment: + # KRaft 모드 핵심 식별자 설정 + KAFKA_CLUSTER_ID: "MkU3OEV5NURaZW52OTYyM2RjY1JqcA" # 운영 환경 고유 클러스터 ID + KAFKA_NODE_ID: 1 + KAFKA_PROCESS_ROLES: 'broker,controller' + KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' + + # 모든 인터페이스(0.0.0.0)로 귀를 열어두고, 운영 서버 안의 다른 컨테이너(Spring Boot)에는 서비스 명인 'kafka:9092'로 광고합니다. + KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093' + KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:9092' + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT' + KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' + KAFKA_CONTROLLER_QUORUM_VOTERS: '1@127.0.0.1:9093' # 루프백 주소로 바인딩 충돌 방지 + + # 메모리 제한 설정 (서버 스펙에 맞게 조정 가능, 최소 256m~512m 추천) + KAFKA_HEAP_OPTS: "-Xms256m -Xmx256m" + # 임시 폴더인 /tmp 대신 컨테이너가 내려가도 데이터와 오프셋 장부가 보존되도록 볼륨 볼륨 마운트 경로 고정 + volumes: + - kafka-prod-data:/var/lib/kafka/data + + # 리소스 상한선 설정 (메모리 릭으로 인한 전체 서버 다운 방지) + deploy: + resources: + limits: + memory: 400M + + restart: unless-stopped + networks: + - catchtable-net networks: catchtable-net: @@ -243,5 +280,5 @@ volumes: jaeger-data: grafana-data: promtail-positions: - # redis-data: - # kafka-data: + kafka-prod-data: + redis-data: \ No newline at end of file diff --git a/grafana/provisioning/datasources/datasources.yml b/grafana/provisioning/datasources/datasources.yml index f073717..7c5bc1c 100644 --- a/grafana/provisioning/datasources/datasources.yml +++ b/grafana/provisioning/datasources/datasources.yml @@ -17,11 +17,10 @@ datasources: editable: true jsonData: derivedFields: - # 로그 안의 traceId(OTel) 를 클릭하면 Jaeger 트레이스로 점프 - datasourceUid: jaeger - matcherRegex: "traceId=(\\w+)" + matcherRegex: 'trace_id=(\w+)' name: TraceID - url: "$${__value.raw}" + url: '$${__value.raw}' - name: Jaeger type: jaeger diff --git a/monitoring/loki-config.yml b/monitoring/loki-config.yml index 91393a4..5fe44af 100644 --- a/monitoring/loki-config.yml +++ b/monitoring/loki-config.yml @@ -27,9 +27,9 @@ schema_config: period: 24h limits_config: - retention_period: 168h # 7일 + retention_period: 72h reject_old_samples: true - reject_old_samples_max_age: 168h + reject_old_samples_max_age: 72h allow_structured_metadata: true volume_enabled: true @@ -41,8 +41,5 @@ compactor: retention_delete_worker_count: 150 delete_request_store: filesystem -ruler: - alertmanager_url: http://localhost:9093 - analytics: reporting_enabled: false diff --git a/monitoring/otel-collector-config.yml b/monitoring/otel-collector-config.yml new file mode 100644 index 0000000..48f8f89 --- /dev/null +++ b/monitoring/otel-collector-config.yml @@ -0,0 +1,36 @@ +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + +processors: + # OTel Semantic Conventions v1.23.0 이후 속성명 변경 (http.target → url.path 등) 호환 위해 + # coalesce(??)로 구버전·신버전 속성 둘 다 시도. + filter/drop_noise: + error_mode: ignore + traces: + span: + - 'IsMatch(attributes["http.target"] ?? attributes["url.path"] ?? "", "^/actuator.*")' + - 'IsMatch(attributes["http.target"] ?? attributes["url.path"] ?? "", "^/favicon.*")' + - '(attributes["http.status_code"] ?? attributes["http.response.status_code"]) == 404 and not IsMatch(attributes["http.target"] ?? attributes["url.path"] ?? "", "^/(api|swagger-ui|api-docs).*")' + + batch: + timeout: 5s + send_batch_size: 1024 + +exporters: + otlp/jaeger: + endpoint: jaeger:4317 + tls: + insecure: true + +service: + pipelines: + traces: + receivers: [otlp] + processors: [filter/drop_noise, batch] + exporters: [otlp/jaeger] + telemetry: + logs: + level: info diff --git a/monitoring/prometheus.yml b/monitoring/prometheus.yml index ff73fee..df22205 100644 --- a/monitoring/prometheus.yml +++ b/monitoring/prometheus.yml @@ -1,20 +1,15 @@ global: scrape_interval: 15s - evaluation_interval: 15s external_labels: cluster: catchtable env: prod scrape_configs: - # Spring Boot Actuator (Micrometer) - job_name: 'catchtable-app' metrics_path: '/actuator/prometheus' static_configs: - targets: ['app:8080'] - labels: - service: catchtable-backend - # Prometheus 자체 메트릭 - job_name: 'prometheus' static_configs: - targets: ['localhost:9090'] diff --git a/monitoring/promtail-config.yml b/monitoring/promtail-config.yml index 5942c9a..24bfa45 100644 --- a/monitoring/promtail-config.yml +++ b/monitoring/promtail-config.yml @@ -9,7 +9,6 @@ clients: - url: http://loki:3100/loki/api/v1/push scrape_configs: - # Docker 컨테이너 stdout 자동 수집 - job_name: docker docker_sd_configs: - host: unix:///var/run/docker.sock diff --git a/new.sh b/new.sh new file mode 100644 index 0000000..61d5e67 --- /dev/null +++ b/new.sh @@ -0,0 +1,2 @@ + docker-compose -f docker-compose.dev.yml up -d + \ No newline at end of file diff --git a/src/main/java/com/catchtable/global/config/KafkaConfig.java b/src/main/java/com/catchtable/global/config/KafkaConfig.java new file mode 100644 index 0000000..5b724e7 --- /dev/null +++ b/src/main/java/com/catchtable/global/config/KafkaConfig.java @@ -0,0 +1,58 @@ +package com.catchtable.global.config; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.*; + +import java.util.HashMap; +import java.util.Map; + +@EnableKafka +@Configuration +public class KafkaConfig { + + @Value("${spring.kafka.bootstrap-servers:localhost:9092}") + private String bootstrapServers; + + @Bean + public KafkaTemplate kafkaTemplate() { + return new KafkaTemplate<>(producerFactory()); + } + + @Bean + public ProducerFactory producerFactory() { + Map config = new HashMap<>(); + config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonSerializer"); + return new DefaultKafkaProducerFactory<>(config); + } + + @Bean + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory()); + return factory; + } + + @Bean + public ConsumerFactory consumerFactory() { + Map config = new HashMap<>(); + config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + config.put(ConsumerConfig.GROUP_ID_CONFIG, "catchtable-notification-group"); + config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.springframework.kafka.support.serializer.JsonDeserializer"); + config.put("spring.json.trusted.packages", "com.catchtable.notification.event,java.lang.String,java.lang.Object"); + config.put("spring.json.use.type.headers", true); + + return new DefaultKafkaConsumerFactory<>(config); + } +} diff --git a/src/main/java/com/catchtable/global/config/MetricsConfig.java b/src/main/java/com/catchtable/global/config/MetricsConfig.java new file mode 100644 index 0000000..8054730 --- /dev/null +++ b/src/main/java/com/catchtable/global/config/MetricsConfig.java @@ -0,0 +1,16 @@ +package com.catchtable.global.config; + +import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry; +import io.github.resilience4j.micrometer.tagged.TaggedCircuitBreakerMetrics; +import io.micrometer.core.instrument.binder.MeterBinder; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class MetricsConfig { + + @Bean + public MeterBinder circuitBreakerMetrics(CircuitBreakerRegistry circuitBreakerRegistry) { + return TaggedCircuitBreakerMetrics.ofCircuitBreakerRegistry(circuitBreakerRegistry)::bindTo; + } +} diff --git a/src/main/java/com/catchtable/global/config/RedissonConfig.java b/src/main/java/com/catchtable/global/config/RedissonConfig.java new file mode 100644 index 0000000..99acd9f --- /dev/null +++ b/src/main/java/com/catchtable/global/config/RedissonConfig.java @@ -0,0 +1,26 @@ +package com.catchtable.global.config; + +import org.redisson.Redisson; +import org.redisson.api.RedissonClient; +import org.redisson.config.Config; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +public class RedissonConfig { + + @Value("${spring.data.redis.host}") + private String host; + + @Value("${spring.data.redis.port}") + private int port; + + @Bean(destroyMethod = "shutdown") + public RedissonClient redissonClient() { + Config config = new Config(); + config.useSingleServer() + .setAddress("redis://" + host + ":" + port); + return Redisson.create(config); + } +} diff --git a/src/main/java/com/catchtable/notification/service/NotificationKafkaConsumer.java b/src/main/java/com/catchtable/notification/service/NotificationKafkaConsumer.java new file mode 100644 index 0000000..deb0e81 --- /dev/null +++ b/src/main/java/com/catchtable/notification/service/NotificationKafkaConsumer.java @@ -0,0 +1,137 @@ +package com.catchtable.notification.service; + +import com.catchtable.notification.entity.NotificationType; +import com.catchtable.notification.event.ReservationCanceledEvent; +import com.catchtable.notification.event.ReservationChangedEvent; +import com.catchtable.notification.event.ReservationConfirmedEvent; +import com.catchtable.notification.event.ReservationVisitedEvent; +import com.catchtable.notification.event.VacancyEvent; +import com.catchtable.user.entity.User; +import com.catchtable.user.repository.UserRepository; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.Optional; + +@Slf4j +@Service +@RequiredArgsConstructor +public class NotificationKafkaConsumer { + + private final NotificationService notificationService; + private final UserRepository userRepository; + + @KafkaListener(topics = "notification.reservation.confirmed", groupId = "catchtable-notification-group") + @Transactional + public void handleReservationConfirmed(@Payload ReservationConfirmedEvent event) { + log.info("[Kafka Consumer] 예약 확정 이벤트 수신: reservationId={}", event.getReservationId()); + Optional userOpt = findUser(event.getUserId()); + if (userOpt.isEmpty()) return; + + String title = "예약이 확정되었습니다."; + String content = String.format("'%s' 매장 %s %s 예약이 확정되었습니다.", + event.getStoreName(), + event.getRemainDate(), + event.getRemainTime()); + + notificationService.createNotification( + userOpt.get(), + NotificationType.RESERVATION_CONFIRMED, + title, + content, + event.getReservationId() + ); + } + + @KafkaListener(topics = "notification.reservation.canceled", groupId = "catchtable-notification-group") + @Transactional + public void handleReservationCanceled(@Payload ReservationCanceledEvent event) { + log.info("[Kafka Consumer] 예약 취소 이벤트 수신: reservationId={}", event.getReservationId()); + Optional userOpt = findUser(event.getUserId()); + if (userOpt.isEmpty()) return; + + String title = "예약이 취소되었습니다."; + String content = String.format("'%s' 매장 %s %s 예약이 취소되었습니다.", + event.getStoreName(), + event.getRemainDate(), + event.getRemainTime()); + + notificationService.createNotification( + userOpt.get(), + NotificationType.RESERVATION_CANCELED, + title, + content, + event.getReservationId() + ); + } + + @KafkaListener(topics = "notification.reservation.changed", groupId = "catchtable-notification-group") + @Transactional + public void handleReservationChanged(@Payload ReservationChangedEvent event) { + log.info("[Kafka Consumer] 예약 변경 이벤트 수신: newReservationId={}", event.getNewReservationId()); + Optional userOpt = findUser(event.getUserId()); + if (userOpt.isEmpty()) return; + + String title = "예약이 변경되었습니다."; + String content = String.format("'%s' 매장 예약이 %s %s에서 %s %s으로 변경되었습니다.", + event.getStoreName(), + event.getOldRemainDate(), + event.getOldRemainTime(), + event.getNewRemainDate(), + event.getNewRemainTime()); + + notificationService.createNotification( + userOpt.get(), + NotificationType.RESERVATION_CHANGED, + title, + content, + event.getNewReservationId() + ); + } + + @KafkaListener(topics = "notification.reservation.visited", groupId = "catchtable-notification-group") + @Transactional + public void handleReservationVisited(@Payload ReservationVisitedEvent event) { + log.info("[Kafka Consumer] 방문 완료 이벤트 수신: reservationId={}", event.getReservationId()); + Optional userOpt = findUser(event.getUserId()); + if (userOpt.isEmpty()) return; + + String title = "방문은 즐거우셨나요?"; + String content = String.format("'%s' 매장 방문이 완료되었습니다. 소중한 경험을 리뷰로 남겨주세요!", + event.getStoreName()); + + notificationService.createNotification( + userOpt.get(), + NotificationType.RESERVATION_VISITED, + title, + content, + event.getReservationId() + ); + } + + @KafkaListener(topics = "notification.vacancy.opened", groupId = "catchtable-notification-group") + @Transactional + public void handleVacancyOpened(@Payload VacancyEvent event) { + // 이 부분은 Redis 도입 후, Redis에서 구독자 목록을 가져와서 처리해야 합니다. + // 현재는 구현하지 않고 로그만 남깁니다. + log.info("[Kafka Consumer] 빈자리 발생 이벤트 수신: remainId={}", event.getRemainId()); + // TODO: Redis에서 remainId에 해당하는 구독자(userId) 목록 조회 + // TODO: 각 구독자에게 알림 생성 (notificationService.createNotification) + } + + /** + * 사용자 조회. 없으면 빈 Optional 반환 + 로그만 남기고 메시지 정상 소모. + * 예외를 던지면 Kafka가 무한 재시도하므로 (존재하지 않는 사용자는 재시도해도 성공 못 함) drop. + */ + private Optional findUser(Long userId) { + Optional user = userRepository.findById(userId); + if (user.isEmpty()) { + log.warn("[Kafka Consumer] 사용자 정보 없음, 메시지 스킵: userId={}", userId); + } + return user; + } +} diff --git a/src/main/java/com/catchtable/notification/service/NotificationKafkaPublisher.java b/src/main/java/com/catchtable/notification/service/NotificationKafkaPublisher.java new file mode 100644 index 0000000..c89c1bb --- /dev/null +++ b/src/main/java/com/catchtable/notification/service/NotificationKafkaPublisher.java @@ -0,0 +1,58 @@ +package com.catchtable.notification.service; + +import com.catchtable.notification.event.ReservationCanceledEvent; +import com.catchtable.notification.event.ReservationChangedEvent; +import com.catchtable.notification.event.ReservationConfirmedEvent; +import com.catchtable.notification.event.ReservationVisitedEvent; +import com.catchtable.notification.event.VacancyEvent; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; + +/** + * ApplicationEvent를 받아 Kafka로 변환 발행. + * @TransactionalEventListener(AFTER_COMMIT) → DB 트랜잭션이 커밋된 후에만 Kafka 메시지 전송. + * 트랜잭션 롤백 시 Kafka 메시지도 발행 안 되어 ghost 알림(데이터 정합성 깨짐) 방지. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class NotificationKafkaPublisher { + + private final KafkaTemplate kafkaTemplate; + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void publishReservationConfirmed(ReservationConfirmedEvent event) { + kafkaTemplate.send("notification.reservation.confirmed", event); + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void publishReservationCanceled(ReservationCanceledEvent event) { + kafkaTemplate.send("notification.reservation.canceled", event); + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void publishReservationChanged(ReservationChangedEvent event) { + kafkaTemplate.send("notification.reservation.changed", event); + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void publishReservationVisited(ReservationVisitedEvent event) { + kafkaTemplate.send("notification.reservation.visited", event); + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + @Transactional(propagation = Propagation.REQUIRES_NEW) + public void publishVacancyOpened(VacancyEvent event) { + kafkaTemplate.send("notification.vacancy.opened", event); + } +} diff --git a/src/main/java/com/catchtable/remain/scheduler/StoreRemainGenerationScheduler.java b/src/main/java/com/catchtable/remain/scheduler/StoreRemainGenerationScheduler.java index 7847282..6e9c912 100644 --- a/src/main/java/com/catchtable/remain/scheduler/StoreRemainGenerationScheduler.java +++ b/src/main/java/com/catchtable/remain/scheduler/StoreRemainGenerationScheduler.java @@ -14,17 +14,17 @@ import java.util.List; /** - * 매일 새벽 4시에 "오늘 포함 30일" 범위의 예약 슬롯을 자동 생성한다. - * 항상 오늘부터 30일치 슬롯이 DB에 존재하도록 보충 (차집합 기반이라 이미 있는 날짜는 스킵). + * 매일 새벽 4시에 "오늘 포함 90일" 범위의 예약 슬롯을 자동 생성한다. + * 항상 오늘부터 90일치 슬롯이 DB에 존재하도록 보충 (차집합 기반이라 이미 있는 날짜는 스킵). * 매장 목록 조회와 영업시간 파싱은 범위 루프 시작 전 단 한 번만 수행해 - * 30회 반복 조회/파싱 비용을 제거한다. + * 90회 반복 조회/파싱 비용을 제거한다. */ @Slf4j @Component @RequiredArgsConstructor public class StoreRemainGenerationScheduler { - private static final int RANGE_DAYS = 30; + private static final int RANGE_DAYS = 90; private final StoreRemainService storeRemainService; private final StoreRepository storeRepository;