[Feat/#113] 아웃박스 릴레이 추가

bootstrap/src/main/java/kr/ac/kookmin/stream/bootstrap/config/SchedulingConfig.java

@@ -0,0 +1,12 @@
+package kr.ac.kookmin.stream.bootstrap.config;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * {@code @Scheduled} 작업(아웃박스 릴레이 등)을 켠다.
+ */
+@Configuration
+@EnableScheduling
+public class SchedulingConfig {
+}

core/common/src/main/java/kr/ac/kookmin/stream/common/OutboxWriter.java

@@ -0,0 +1,13 @@
+package kr.ac.kookmin.stream.common;
+
+import kr.ac.kookmin.stream.common.event.DomainEvent;
+
+/**
+ * 도메인 이벤트를 아웃박스에 기록한다. 구현은 {@code infrastructure:outbox}가 맡는다.
+ * <p>
+ * 원 비즈니스 쓰기와 같은 트랜잭션 안에서 불러야 한다. 트랜잭션이 롤백되면 기록도 함께 사라지고,
+ * 커밋되면 릴레이가 이벤트를 다시 발행한다. 트랜잭션 밖에서 부르면 실패한다.
+ */
+public interface OutboxWriter {
+    void record(DomainEvent event);
+}

core/common/src/main/java/kr/ac/kookmin/stream/common/event/DomainEvent.java

@@ -0,0 +1,10 @@
+package kr.ac.kookmin.stream.common.event;
+
+/**
+ * 다른 도메인이 반응할 수 있는 도메인의 사실. 아웃박스에 기록되어 릴레이가 다시 발행한다.
+ * <p>
+ * 아웃박스에 JSON으로 저장되므로 ID와 원시값만 담는 record로 만든다. 클래스 이름이 역직렬화 키라서,
+ * 이름이나 패키지를 바꿀 때는 아직 발행되지 않은 아웃박스 행이 없는지 확인한다.
+ */
+public interface DomainEvent {
+}

core/common/src/main/java/kr/ac/kookmin/stream/common/event/package-info.java

@@ -0,0 +1,8 @@
+/**
+ * 크로스 도메인 이벤트 타입. 발행 도메인과 구독 도메인이 서로를 모른 채 이 타입만 공유한다.
+ * Modulith는 모듈의 하위 패키지를 비공개로 보므로 named interface로 열어 둔다.
+ */
+@NamedInterface("event")
+package kr.ac.kookmin.stream.common.event;
+
+import org.springframework.modulith.NamedInterface;

docs/conventions/architecture.md

@@ -355,18 +355,18 @@ Data Access    {Domain}Repository / {Domain}Client 인터페이스  → core:dom
   EventServiceImpl → outboxWriter.record(event)              // core:common 포트
       ↓ 같은 트랜잭션에서 outbox 테이블에 저장 (원 비즈니스 쓰기와 원자적)
 릴레이 (infrastructure:outbox)
-  OutboxRelay(@Scheduled 폴러) → 미발행 레코드 조회
-      → applicationEventPublisher.publishEvent(event)        // 인프로세스 재발행
-      → 전달 성공 시 발행 완료 마킹, 실패 시 다음 폴에서 재시도
-구독 (예: welfare 도메인 — 공지/알림 반영)
-  @TransactionalEventListener void on(EventApplicationConfirmed e)  // spring-tx, Modulith 의존 없음
+  OutboxRelay(@Scheduled 폴러) → PENDING 레코드 조회
+      → applicationEventPublisher.publishEvent(event)        // 트랜잭션 밖에서 인프로세스 재발행, 리스너 동기 실행
+      → 리스너가 예외 없이 끝나면 PUBLISHED, 실패하면 attempts+1 후 다음 폴에서 재시도 (한도 초과 시 FAILED)
+구독 (예: member 도메인 — 푸시 알림)
+  @EventListener void on(EventApplied e)                     // spring-context, Modulith 의존 없음
 ```
 
-- **이벤트 타입은 `core:common.event`에 둔다.** 발행/구독 도메인이 서로를 의존하지 않고 이벤트 타입만 공유한다.
+- **이벤트 타입은 `core:common.event`에 두고 `DomainEvent`를 구현한다.** 발행/구독 도메인이 서로를 의존하지 않고 이벤트 타입만 공유한다. `common.event`는 `@NamedInterface("event")`로 열어 두었다(Modulith는 모듈의 하위 패키지를 비공개로 본다).
 - 이벤트는 **사실**이다. ID·원시값만 담고 애그리거트/값 타입을 싣지 않는다. 커맨드(요청)를 이벤트로 쓰지 않는다.
 - **발행은 `OutboxWriter` 포트로 기록**한다(원 비즈니스 트랜잭션과 원자적). 인프로세스 재발행은 릴레이만 수행한다.
-- **구독은 `@TransactionalEventListener`**(spring-tx)로 한다. `core:domain`에 Modulith 이벤트 의존을 넣지 않는다.
-- 소비 리스너는 **멱등**하게 작성한다(at-least-once).
+- **구독은 `@EventListener`**로 한다. 릴레이가 트랜잭션 밖에서 발행하므로 `@TransactionalEventListener`는 호출되지 않는다. 릴레이는 리스너가 끝난 뒤에 발행 완료로 표시하므로, 리스너가 예외를 던지면 재시도된다. `core:domain`에 Modulith 이벤트 의존을 넣지 않는다.
+- 소비 리스너는 **멱등**하게 작성한다(at-least-once). 리스너가 끝난 뒤 완료 표시 전에 서버가 내려가면 한 번 더 전달된다. 푸시 알림처럼 드문 중복을 감수할 수 있으면 그대로 둔다.
 - 아웃박스 테이블 스키마는 `flyway-migration.md`로 관리한다.
 - 최종적 일관성 경로다. "한 요청 안에서 쓰기 직후 결과를 즉시 조회해 응답에 실어야" 하면 이벤트가 아니라 6-1의 동기 UseCase로 처리한다.
 

docs/conventions/flyway-migration.md

@@ -145,20 +145,23 @@ CREATE INDEX idx_user_departments_user_id ON user_departments (user_id);
 `infrastructure:outbox` 릴레이가 사용하는 공용 테이블(`architecture.md` 6-2절). 특정 도메인에 종속되지 않는다.
 
 ```sql
--- V2__create_outbox_table.sql
+-- V18__create_outbox_events_table.sql
 CREATE TABLE outbox_events (
-    id BIGINT AUTO_INCREMENT PRIMARY KEY,
-    event_type VARCHAR(200) NOT NULL,
+    outbox_event_id BIGINT AUTO_INCREMENT PRIMARY KEY,
+    event_type VARCHAR(200) NOT NULL,                -- 이벤트 클래스 FQCN. 역직렬화에 쓴다
     payload JSON NOT NULL,
-    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',   -- PENDING / PUBLISHED
+    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',   -- PENDING / PUBLISHED / FAILED
+    attempts INT NOT NULL DEFAULT 0,                 -- 발행에 실패한 횟수
     created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
     published_at DATETIME NULL
 );
-CREATE INDEX idx_outbox_events_status_id ON outbox_events (status, id);
+CREATE INDEX idx_outbox_events_status_outbox_event_id ON outbox_events (status, outbox_event_id);
 ```
 
-- 발행 도메인은 자기 트랜잭션 안에서 `OutboxWriter`로 기록(원 비즈니스 쓰기와 원자적).
+- 발행 도메인은 자기 트랜잭션 안에서 `OutboxWriter`로 기록(원 비즈니스 쓰기와 원자적). 트랜잭션 밖에서 부르면 실패한다.
 - 릴레이(`@Scheduled` 폴러)는 `status = 'PENDING'`을 조회해 인프로세스 재발행 후 `PUBLISHED`로 마킹한다. at-least-once이므로 소비 리스너는 멱등해야 한다.
+- 리스너가 실패하면 `attempts`를 올리고 다음 폴에서 다시 발행한다. 한도(5회)에 닿으면 `FAILED`로 바꿔 더는 재시도하지 않는다. 원인을 고친 뒤 `PENDING`으로 되돌리면 다시 발행된다.
+- `PUBLISHED`가 된 지 7일 지난 행은 매일 새벽 정리한다. `FAILED`는 원인을 확인할 수 있게 남긴다.
 
 ---
 

gradle/libs.versions.toml

@@ -25,6 +25,7 @@ springBootStarterAspectj = { module = "org.springframework.boot:spring-boot-star
 springBootStarterSecurity = { module = "org.springframework.boot:spring-boot-starter-security" }
 springBootStarterDataJpa = { module = "org.springframework.boot:spring-boot-starter-data-jpa" }
 springBootStarterFlyway = { module = "org.springframework.boot:spring-boot-starter-flyway" }
+springBootStarterJackson = { module = "org.springframework.boot:spring-boot-starter-jackson" }
 springBootStarterTest = { module = "org.springframework.boot:spring-boot-starter-test" }
 springBootStarterSecurityTest = { module = "org.springframework.boot:spring-boot-starter-security-test" }
 springBootStarterDataJpaTest = { module = "org.springframework.boot:spring-boot-starter-data-jpa-test" }

infrastructure/db/src/main/resources/db/migration/V18__create_outbox_events_table.sql

@@ -0,0 +1,12 @@
+-- 도메인 이벤트를 원 비즈니스 트랜잭션과 함께 기록하는 아웃박스. 릴레이(infrastructure:outbox)가 PENDING을 읽어 다시 발행한다.
+-- 리스너가 계속 실패하는 이벤트는 attempts가 한도에 닿으면 FAILED로 바꿔 더는 재시도하지 않는다.
+CREATE TABLE outbox_events (
+    outbox_event_id BIGINT AUTO_INCREMENT PRIMARY KEY,
+    event_type VARCHAR(200) NOT NULL,                -- 이벤트 클래스 FQCN. 역직렬화에 쓴다
+    payload JSON NOT NULL,
+    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',   -- PENDING / PUBLISHED / FAILED
+    attempts INT NOT NULL DEFAULT 0,                 -- 발행에 실패한 횟수
+    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    published_at DATETIME NULL
+);
+CREATE INDEX idx_outbox_events_status_outbox_event_id ON outbox_events (status, outbox_event_id);

infrastructure/outbox/build.gradle.kts

@@ -10,6 +10,7 @@ dependencies {
     implementation(platform(libs.springBootDependenciesBom))
     implementation(libs.springContext)
     implementation(libs.springBootStarterDataJpa)
+    implementation(libs.springBootStarterJackson)   // 이벤트를 JSON으로 저장·복원 (Jackson 3, JsonMapper)
 
     testImplementation(platform(libs.springBootDependenciesBom))
     testImplementation(libs.springBootStarterDataJpaTest)

infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxCleaner.java

@@ -0,0 +1,31 @@
+package kr.ac.kookmin.stream.outbox;
+
+import java.time.LocalDateTime;
+import kr.ac.kookmin.stream.common.DateUtil;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 발행을 마친 아웃박스 행을 정리한다. 실패(FAILED) 행은 원인을 확인할 수 있게 남겨 둔다.
+ */
+@Component
+@RequiredArgsConstructor
+public class OutboxCleaner {
+
+    private static final Logger log = LoggerFactory.getLogger(OutboxCleaner.class);
+
+    private static final long RETENTION_DAYS = 7;
+
+    private final OutboxEventJpaRepository outboxEventJpaRepository;
+
+    // 발행이 몰리지 않는 새벽에 돈다
+    @Scheduled(cron = "0 0 4 * * *", zone = "Asia/Seoul")
+    public void deletePublished() {
+        LocalDateTime before = LocalDateTime.now(DateUtil.KST).minusDays(RETENTION_DAYS);
+        int deleted = outboxEventJpaRepository.deleteByStatusAndPublishedAtBefore(OutboxStatus.PUBLISHED, before);
+        log.info("발행 완료 아웃박스 정리: deleted={}, before={}", deleted, before);
+    }
+}

infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxEventJpaEntity.java

@@ -0,0 +1,78 @@
+package kr.ac.kookmin.stream.outbox;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Index;
+import jakarta.persistence.Table;
+import java.time.LocalDateTime;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import org.hibernate.annotations.JdbcTypeCode;
+import org.hibernate.type.SqlTypes;
+
+// created_at은 DB 기본값으로만 채우고 읽지 않아서 매핑하지 않는다
+@Entity
+@Table(
+    name = "outbox_events",
+    indexes = {
+        @Index(name = "idx_outbox_events_status_outbox_event_id", columnList = "status, outbox_event_id")
+    }
+)
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+public class OutboxEventJpaEntity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "outbox_event_id")
+    private Long id;
+
+    @Column(name = "event_type", nullable = false, length = 200)
+    private String eventType;
+
+    /** 이벤트를 직렬화한 JSON 원문을 문자열 그대로 읽고 쓴다. */
+    @JdbcTypeCode(SqlTypes.JSON)
+    @Column(nullable = false, columnDefinition = "json")
+    private String payload;
+
+    @Enumerated(EnumType.STRING)
+    @Column(nullable = false, length = 20)
+    private OutboxStatus status;
+
+    @Column(nullable = false)
+    private int attempts;
+
+    @Column(name = "published_at")
+    private LocalDateTime publishedAt;
+
+    private OutboxEventJpaEntity(String eventType, String payload) {
+        this.eventType = eventType;
+        this.payload = payload;
+        this.status = OutboxStatus.PENDING;
+    }
+
+    public static OutboxEventJpaEntity pending(String eventType, String payload) {
+        return new OutboxEventJpaEntity(eventType, payload);
+    }
+
+    public void markPublished(LocalDateTime publishedAt) {
+        this.status = OutboxStatus.PUBLISHED;
+        this.publishedAt = publishedAt;
+    }
+
+    // 실패 횟수가 한도에 닿으면 FAILED로 바꾸고 true를 돌려준다
+    public boolean recordFailure(int maxAttempts) {
+        this.attempts++;
+        if (attempts < maxAttempts) {
+            return false;
+        }
+        this.status = OutboxStatus.FAILED;
+        return true;
+    }
+}

infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxEventJpaRepository.java

@@ -0,0 +1,24 @@
+package kr.ac.kookmin.stream.outbox;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.transaction.annotation.Transactional;
+
+public interface OutboxEventJpaRepository extends JpaRepository<OutboxEventJpaEntity, Long> {
+
+    List<OutboxEventJpaEntity> findTop100ByStatusOrderByIdAsc(OutboxStatus status);
+
+    // 파생 delete는 행을 읽어 한 건씩 지우므로 벌크 DELETE로 지운다. 호출하는 정리 작업에 트랜잭션이 없어 여기서 연다
+    @Transactional
+    @Modifying
+    @Query("""
+        DELETE FROM OutboxEventJpaEntity o
+        WHERE o.status = :status
+        AND o.publishedAt < :before
+        """)
+    int deleteByStatusAndPublishedAtBefore(@Param("status") OutboxStatus status, @Param("before") LocalDateTime before);
+}

infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxRelay.java

@@ -0,0 +1,71 @@
+package kr.ac.kookmin.stream.outbox;
+
+import java.time.LocalDateTime;
+import kr.ac.kookmin.stream.common.DateUtil;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.util.ClassUtils;
+import tools.jackson.databind.json.JsonMapper;
+
+/**
+ * 아웃박스에 쌓인 이벤트를 인프로세스로 다시 발행한다.
+ * <p>
+ * 트랜잭션 밖에서 발행하고 리스너({@code @EventListener})는 이 스레드에서 동기로 실행된다. 리스너가 예외 없이 끝나야
+ * 발행 완료로 표시하므로, 리스너가 실패하면 다음 폴링에서 다시 발행된다(at-least-once). 리스너가 끝난 뒤 완료 표시 전에
+ * 서버가 내려가면 같은 이벤트가 한 번 더 발행될 수 있다.
+ * <p>
+ * 환경마다 인스턴스가 하나라고 보고 행을 선점하지 않는다. 인스턴스를 늘리면 두 릴레이가 같은 행을 발행하지 않도록 선점 방식을 넣어야 한다.
+ */
+@Component
+@RequiredArgsConstructor
+public class OutboxRelay {
+
+    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);
+
+    private static final long POLL_INTERVAL_MILLIS = 1000;
+    private static final int MAX_ATTEMPTS = 5;
+
+    private final OutboxEventJpaRepository outboxEventJpaRepository;
+    private final ApplicationEventPublisher applicationEventPublisher;
+    private final JsonMapper jsonMapper;
+
+    @Scheduled(fixedDelay = POLL_INTERVAL_MILLIS)
+    public void relay() {
+        outboxEventJpaRepository.findTop100ByStatusOrderByIdAsc(OutboxStatus.PENDING)
+            .forEach(this::relay);
+    }
+
+    // 한 이벤트가 실패해도 다음 이벤트는 계속 발행한다. 완료·실패 표시는 save가 건별 트랜잭션으로 반영한다
+    private void relay(OutboxEventJpaEntity outboxEvent) {
+        try {
+            applicationEventPublisher.publishEvent(deserialize(outboxEvent));
+            outboxEvent.markPublished(LocalDateTime.now(DateUtil.KST));
+        } catch (RuntimeException e) {
+            logFailure(outboxEvent, outboxEvent.recordFailure(MAX_ATTEMPTS), e);
+        }
+        outboxEventJpaRepository.save(outboxEvent);
+    }
+
+    private Object deserialize(OutboxEventJpaEntity outboxEvent) {
+        try {
+            Class<?> eventType = ClassUtils.forName(outboxEvent.getEventType(), ClassUtils.getDefaultClassLoader());
+            return jsonMapper.readValue(outboxEvent.getPayload(), eventType);
+        } catch (ClassNotFoundException e) {
+            throw new IllegalStateException("아웃박스 이벤트 타입을 찾을 수 없습니다: " + outboxEvent.getEventType(), e);
+        }
+    }
+
+    private void logFailure(OutboxEventJpaEntity outboxEvent, boolean gaveUp, RuntimeException e) {
+        if (gaveUp) {
+            log.error("아웃박스 이벤트 발행을 포기함(FAILED): outboxEventId={}, eventType={}, attempts={}",
+                outboxEvent.getId(), outboxEvent.getEventType(), outboxEvent.getAttempts(), e);
+            return;
+        }
+        log.warn("아웃박스 이벤트 발행 실패(재시도 예정): outboxEventId={}, eventType={}, attempts={}",
+            outboxEvent.getId(), outboxEvent.getEventType(), outboxEvent.getAttempts(), e);
+    }
+}

infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxStatus.java

@@ -0,0 +1,7 @@
+package kr.ac.kookmin.stream.outbox;
+
+public enum OutboxStatus {
+    PENDING,    // 아직 발행하지 못함. 릴레이가 다음 폴링에서 다시 발행한다
+    PUBLISHED,  // 리스너까지 예외 없이 끝남
+    FAILED      // 발행 실패가 한도에 닿아 더는 재시도하지 않음. 원인을 고친 뒤 PENDING으로 되돌리면 다시 발행된다
+}

infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxWriterImpl.java

@@ -0,0 +1,25 @@
+package kr.ac.kookmin.stream.outbox;
+
+import kr.ac.kookmin.stream.common.OutboxWriter;
+import kr.ac.kookmin.stream.common.event.DomainEvent;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+import tools.jackson.databind.json.JsonMapper;
+
+@Component
+@RequiredArgsConstructor
+public class OutboxWriterImpl implements OutboxWriter {
+
+    private final OutboxEventJpaRepository outboxEventJpaRepository;
+    private final JsonMapper jsonMapper;
+
+    // 원 비즈니스 쓰기와 함께 커밋·롤백되어야 하므로 진행 중인 트랜잭션이 없으면 실패시킨다
+    @Override
+    @Transactional(propagation = Propagation.MANDATORY)
+    public void record(DomainEvent event) {
+        String payload = jsonMapper.writeValueAsString(event);
+        outboxEventJpaRepository.save(OutboxEventJpaEntity.pending(event.getClass().getName(), payload));
+    }
+}