+ * 원 비즈니스 쓰기와 같은 트랜잭션 안에서 불러야 한다. 트랜잭션이 롤백되면 기록도 함께 사라지고, + * 커밋되면 릴레이가 이벤트를 다시 발행한다. 트랜잭션 밖에서 부르면 실패한다. + */ +public interface OutboxWriter { + void record(DomainEvent event); +} diff --git a/core/common/src/main/java/kr/ac/kookmin/stream/common/event/DomainEvent.java b/core/common/src/main/java/kr/ac/kookmin/stream/common/event/DomainEvent.java new file mode 100644 index 00000000..ba600d0c --- /dev/null +++ b/core/common/src/main/java/kr/ac/kookmin/stream/common/event/DomainEvent.java @@ -0,0 +1,10 @@ +package kr.ac.kookmin.stream.common.event; + +/** + * 다른 도메인이 반응할 수 있는 도메인의 사실. 아웃박스에 기록되어 릴레이가 다시 발행한다. + *
+ * 아웃박스에 JSON으로 저장되므로 ID와 원시값만 담는 record로 만든다. 클래스 이름이 역직렬화 키라서,
+ * 이름이나 패키지를 바꿀 때는 아직 발행되지 않은 아웃박스 행이 없는지 확인한다.
+ */
+public interface DomainEvent {
+}
diff --git a/core/common/src/main/java/kr/ac/kookmin/stream/common/event/package-info.java b/core/common/src/main/java/kr/ac/kookmin/stream/common/event/package-info.java
new file mode 100644
index 00000000..f7879bd6
--- /dev/null
+++ b/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;
diff --git a/docs/conventions/architecture.md b/docs/conventions/architecture.md
index 0bc04259..fa3b17ca 100644
--- a/docs/conventions/architecture.md
+++ b/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로 처리한다.
diff --git a/docs/conventions/flyway-migration.md b/docs/conventions/flyway-migration.md
index f1140f05..148dd988 100644
--- a/docs/conventions/flyway-migration.md
+++ b/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`는 원인을 확인할 수 있게 남긴다.
---
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 6f2b6e1f..685b5e6f 100644
--- a/gradle/libs.versions.toml
+++ b/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" }
diff --git a/infrastructure/db/src/main/resources/db/migration/V18__create_outbox_events_table.sql b/infrastructure/db/src/main/resources/db/migration/V18__create_outbox_events_table.sql
new file mode 100644
index 00000000..6212f14e
--- /dev/null
+++ b/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);
diff --git a/infrastructure/outbox/build.gradle.kts b/infrastructure/outbox/build.gradle.kts
index 16b7387c..f4c9a959 100644
--- a/infrastructure/outbox/build.gradle.kts
+++ b/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)
diff --git a/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxCleaner.java b/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxCleaner.java
new file mode 100644
index 00000000..c659b0e1
--- /dev/null
+++ b/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);
+ }
+}
diff --git a/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxEventJpaEntity.java b/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxEventJpaEntity.java
new file mode 100644
index 00000000..58fb7a9e
--- /dev/null
+++ b/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;
+ }
+}
diff --git a/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxEventJpaRepository.java b/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxEventJpaRepository.java
new file mode 100644
index 00000000..3ffa2262
--- /dev/null
+++ b/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
+ * 트랜잭션 밖에서 발행하고 리스너({@code @EventListener})는 이 스레드에서 동기로 실행된다. 리스너가 예외 없이 끝나야
+ * 발행 완료로 표시하므로, 리스너가 실패하면 다음 폴링에서 다시 발행된다(at-least-once). 리스너가 끝난 뒤 완료 표시 전에
+ * 서버가 내려가면 같은 이벤트가 한 번 더 발행될 수 있다.
+ *
+ * 환경마다 인스턴스가 하나라고 보고 행을 선점하지 않는다. 인스턴스를 늘리면 두 릴레이가 같은 행을 발행하지 않도록 선점 방식을 넣어야 한다.
+ */
+@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);
+ }
+}
diff --git a/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxStatus.java b/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxStatus.java
new file mode 100644
index 00000000..dc727b51
--- /dev/null
+++ b/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으로 되돌리면 다시 발행된다
+}
diff --git a/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxWriterImpl.java b/infrastructure/outbox/src/main/java/kr/ac/kookmin/stream/outbox/OutboxWriterImpl.java
new file mode 100644
index 00000000..30c5a9a4
--- /dev/null
+++ b/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));
+ }
+}