[feat][queue-service] program Kafka 연동 + ProgramMeta Aggregate 추가

build.gradle

@@ -47,6 +47,20 @@ repositories {
 			password = githubToken
 		}
 	}
+    maven {
+        url = 'https://maven.pkg.github.com/first-ticket/common-jpa'
+        credentials {
+            username = githubUser
+            password = githubToken
+        }
+    }
+    maven {
+        url = 'https://maven.pkg.github.com/first-ticket/common-messaging'
+        credentials {
+            username = githubUser
+            password = githubToken
+        }
+    }
 }
 
 ext {
@@ -55,30 +69,58 @@ ext {
 }
 
 dependencies {
+    // 공통 모듈
+    implementation 'com.first-ticket:common:0.0.4-SNAPSHOT'
+    implementation 'com.first-ticket:common-jpa:0.0.2-SNAPSHOT'
+    implementation 'com.first-ticket:common-messaging:0.0.2-SNAPSHOT'
+
+    // Spring Boot
 	implementation 'org.springframework.boot:spring-boot-starter-actuator'
 	implementation 'org.springframework.boot:spring-boot-starter-data-redis'
+    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
 	implementation 'org.springframework.boot:spring-boot-starter-validation'
 	implementation 'org.springframework.boot:spring-boot-starter-web'
+    implementation 'org.springframework.boot:spring-boot-starter-aop'
+
+    // Spring Cloud
 	implementation 'org.springframework.cloud:spring-cloud-starter-config'
 	implementation 'org.springframework.cloud:spring-cloud-starter-netflix-eureka-client'
-    implementation 'org.springframework.retry:spring-retry'
-    implementation 'org.springframework.boot:spring-boot-starter-aop'
-    implementation 'com.first-ticket:common:0.0.4-SNAPSHOT'
+
+    // kafka
+    implementation 'org.springframework.kafka:spring-kafka'
+
+    // flyway
+    implementation 'org.flywaydb:flyway-core'
+    implementation 'org.flywaydb:flyway-database-postgresql'
+
+    // 모니터링
+    implementation 'io.micrometer:micrometer-registry-prometheus'
+    implementation 'io.micrometer:micrometer-tracing-bridge-brave'
+    runtimeOnly 'io.zipkin.reporter2:zipkin-reporter-brave'
+
+    // JWT
+    implementation 'io.jsonwebtoken:jjwt-api:0.12.6'
+    runtimeOnly 'io.jsonwebtoken:jjwt-impl:0.12.6'
+    runtimeOnly 'io.jsonwebtoken:jjwt-jackson:0.12.6'
+
+    // Lombok
 	compileOnly 'org.projectlombok:lombok'
 	annotationProcessor 'org.projectlombok:lombok'
+
+    // PostgreSQL
+    runtimeOnly 'org.postgresql:postgresql'
+
+    // Test
 	testImplementation 'org.springframework.boot:spring-boot-starter-test'
+    testImplementation 'org.testcontainers:postgresql'
 	testImplementation 'org.springframework.boot:spring-boot-testcontainers'
 	testImplementation 'org.testcontainers:junit-jupiter'
+    testImplementation 'org.springframework.kafka:spring-kafka-test'
 	testImplementation 'org.springframework.restdocs:spring-restdocs-mockmvc'
 	asciidoctorExt 'org.springframework.restdocs:spring-restdocs-asciidoctor'
 	testCompileOnly 'org.projectlombok:lombok'
 	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
 	testAnnotationProcessor 'org.projectlombok:lombok'
-
-	// JWT
-	implementation 'io.jsonwebtoken:jjwt-api:0.12.6'
-	runtimeOnly 'io.jsonwebtoken:jjwt-impl:0.12.6'
-	runtimeOnly 'io.jsonwebtoken:jjwt-jackson:0.12.6'
 }
 
 dependencyManagement {

src/main/java/com/firstticket/queueservice/config/JpaConfig.java

@@ -0,0 +1,17 @@
+package com.firstticket.queueservice.config;
+
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+
+@Configuration
+@EntityScan(basePackages = {
+    "com.firstticket.queueservice",
+    "com.firstticket.common.messaging.inbox"
+})
+@EnableJpaRepositories(basePackages = {
+    "com.firstticket.queueservice",
+    "com.firstticket.common.messaging.inbox"
+})
+public class JpaConfig {
+}

src/main/java/com/firstticket/queueservice/programmeta/application/ProgramMetaService.java

@@ -0,0 +1,87 @@
+package com.firstticket.queueservice.programmeta.application;
+
+import com.firstticket.queueservice.programmeta.application.dto.CancelProgramCommand;
+import com.firstticket.queueservice.programmeta.application.dto.CreateProgramMetaCommand;
+import com.firstticket.queueservice.programmeta.application.dto.UpdateProgramTimeCommand;
+import com.firstticket.queueservice.programmeta.domain.ProgramMeta;
+import com.firstticket.queueservice.programmeta.domain.ProgramMetaRepository;
+import com.firstticket.queueservice.programmeta.domain.event.ProgramEvents;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * ProgramMeta 도메인의 Command 처리 서비스.
+ * Kafka Consumer 가 전달한 Command 를 받아 Aggregate 의 상태를 변경하고,
+ * 필요 시 도메인 이벤트를 발행한다.
+ */
+@Slf4j
+@RequiredArgsConstructor
+@Service
+public class ProgramMetaService {
+    private final ProgramMetaRepository programMetaRepository;
+    private final ProgramEvents programEvents;
+
+    /**
+     * program.created 처리.
+     * ProgramMeta 새로 생성하여 저장.
+     */
+    public void handleCreated(CreateProgramMetaCommand command) {
+        log.info("Program created. programId={}, status={}", command.programId(), command.status());
+
+        programMetaRepository.findById(command.programId())
+            .ifPresentOrElse(
+                existing -> log.info("ProgramMeta already exists. skip created. programId={}", command.programId()),
+                () -> {
+                    ProgramMeta programMeta = ProgramMeta.of(
+                        command.programId(),
+                        command.openAt(),
+                        command.closeAt(),
+                        command.status()
+                    );
+                    programMetaRepository.save(programMeta);
+                }
+            );
+    }
+
+    /**
+     * program.time.updated 처리.
+     * 기존 ProgramMeta 의 openAt / closeAt 갱신.
+     * Meta 가 존재하지 않으면 경고 로그만 남긴다 (이벤트 순서가 어긋난 경우 대비).
+     */
+    public void handleTimeUpdated(UpdateProgramTimeCommand command) {
+        log.info("Program time updated. programId={}, openAt={}, closeAt={}",
+            command.programId(), command.openAt(), command.closeAt());
+
+        programMetaRepository.findById(command.programId())
+            .ifPresentOrElse(
+                programMeta -> {
+                    programMeta.updateTime(command.openAt(), command.closeAt());
+                    programMetaRepository.save(programMeta);
+                },
+                () -> log.warn("ProgramMeta 없음 (time updated). programId={}", command.programId())
+            );
+    }
+
+    /**
+     * program.cancelled 처리.
+     * 1. ProgramMeta 의 status 를 CANCELLED 로 갱신.
+     * 2. ProgramCancelledEvent 발행 (queuetoken Aggregate 가 토큰 정리).
+     */
+    public void handleCancelled(CancelProgramCommand command) {
+        log.info("Program cancelled. programId={}", command.programId());
+
+        programMetaRepository.findById(command.programId())
+            .ifPresentOrElse(
+                programMeta -> {
+                    programMeta.cancel();
+                    programMetaRepository.save(programMeta);
+                },
+                () -> log.warn("ProgramMeta 없음 (cancelled). programId={}", command.programId())
+            );
+
+        // queuetoken Aggregate 에 이벤트 발급
+        programEvents.publishProgramCancelled(command.programId());
+    }
+
+}

src/main/java/com/firstticket/queueservice/programmeta/application/dto/CancelProgramCommand.java

@@ -0,0 +1,21 @@
+package com.firstticket.queueservice.programmeta.application.dto;
+
+import com.firstticket.queueservice.programmeta.domain.ProgramStatus;
+import com.firstticket.queueservice.programmeta.domain.vo.ProgramId;
+
+import java.util.UUID;
+
+/**
+ * program.cancelled 이벤트 처리용 Command.
+ */
+public record CancelProgramCommand(
+    ProgramId programId,
+    ProgramStatus status
+) {
+    public static CancelProgramCommand of(UUID programId, String status) {
+        return new CancelProgramCommand(
+            ProgramId.of(programId),
+            ProgramStatus.parse(status)
+        );
+    }
+}

src/main/java/com/firstticket/queueservice/programmeta/application/dto/CreateProgramMetaCommand.java

@@ -0,0 +1,34 @@
+package com.firstticket.queueservice.programmeta.application.dto;
+
+import com.firstticket.queueservice.programmeta.domain.ProgramStatus;
+import com.firstticket.queueservice.programmeta.domain.vo.ProgramId;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+/**
+ * program.created 이벤트 처리용 Command.
+ * Kafka Consumer 가 외부 Payload 를 변환하여 application 에 전달한다.
+ *
+ * <p>openAt / closeAt 은 생성 시점엔 스케줄 미정이라 null 가능.</p>
+ */
+public record CreateProgramMetaCommand(
+    ProgramId programId,
+    LocalDateTime openAt,
+    LocalDateTime closeAt,
+    ProgramStatus status
+) {
+    public static CreateProgramMetaCommand of(
+        UUID programId,
+        LocalDateTime openAt,
+        LocalDateTime closeAt,
+        String status
+    ) {
+        return new CreateProgramMetaCommand(
+            ProgramId.of(programId),
+            openAt,
+            closeAt,
+            ProgramStatus.parse(status)
+        );
+    }
+}

src/main/java/com/firstticket/queueservice/programmeta/application/dto/UpdateProgramTimeCommand.java

@@ -0,0 +1,28 @@
+package com.firstticket.queueservice.programmeta.application.dto;
+
+import com.firstticket.queueservice.programmeta.domain.vo.ProgramId;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+/**
+ * program.time.updated 이벤트 처리용 Command.
+ * 스케줄 (openAt / closeAt) 등록 / 변경 시 사용.
+ */
+public record UpdateProgramTimeCommand(
+    ProgramId programId,
+    LocalDateTime openAt,
+    LocalDateTime closeAt
+) {
+    public static UpdateProgramTimeCommand of(
+        UUID programId,
+        LocalDateTime openAt,
+        LocalDateTime closeAt
+    ) {
+        return new UpdateProgramTimeCommand(
+            ProgramId.of(programId),
+            openAt,
+            closeAt
+        );
+    }
+}

src/main/java/com/firstticket/queueservice/programmeta/domain/ProgramMeta.java

@@ -0,0 +1,93 @@
+package com.firstticket.queueservice.programmeta.domain;
+
+import com.firstticket.queueservice.programmeta.domain.vo.ProgramId;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+import java.util.Objects;
+
+/**
+ * Program 의 메타 정보 (Aggregate Root).
+ * program-service 의 이벤트로 갱신되는 캐시 / 읽기 모델.
+ *
+ * <p>원본은 program-service 가 소유하므로 queue-service 는 이 객체를
+ * 영구 저장하지 않으며, 필요 시 program 토픽의 처음부터 재구독으로 복구한다.</p>
+ */
+@Getter
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class ProgramMeta {
+
+    private ProgramId programId;
+    private LocalDateTime openAt;
+    private LocalDateTime closeAt;
+    private ProgramStatus status;
+
+    /**
+     * 새 ProgramMeta 생성. 일반적으로 program.created 이벤트 처리 시 호출.
+     *
+     * @param programId UUID 형태의 program 식별자
+     * @param openAt    예매 오픈 시각 (생성 시점엔 null 가능)
+     * @param closeAt   예매 종료 시각 (생성 시점엔 null 가능)
+     * @param status    program-service 가 전달한 상태 문자열 ("DRAFT" / "CANCELLED")
+     */
+    public static ProgramMeta of(ProgramId programId, LocalDateTime openAt,
+                                 LocalDateTime closeAt, ProgramStatus status) {
+        Objects.requireNonNull(programId, "programId는 null일 수 없습니다.");
+        Objects.requireNonNull(status, "status는 null일 수 없습니다.");
+        validateSchedule(openAt, closeAt);
+        return new ProgramMeta(
+            programId,
+            openAt,
+            closeAt,
+            status
+        );
+    }
+
+    /**
+     * 스케줄 갱신. program.time.updated 이벤트 처리 시 호출.
+     */
+    public void updateTime(LocalDateTime newOpenAt, LocalDateTime newCloseAt) {
+        validateSchedule(newOpenAt, newCloseAt);
+        this.openAt = newOpenAt;
+        this.closeAt = newCloseAt;
+    }
+
+    private static void validateSchedule(LocalDateTime openAt, LocalDateTime closeAt) {
+        if (openAt != null && closeAt != null && openAt.isAfter(closeAt)) {
+            throw new IllegalArgumentException("openAt은 closeAt보다 늦을 수 없습니다");
+        }
+    }
+
+    /**
+     * 프로그램 취소 처리. program.cancelled 이벤트 처리 시 호출.
+     */
+    public void cancel() {
+        this.status = ProgramStatus.CANCELLED;
+    }
+
+    /**
+     * 현재 시각 기준 활성 여부.
+     * CANCELLED 가 아니고, 스케줄이 설정됐고, 현재 시각이 openAt ~ closeAt 사이일 때 true.
+     */
+    public boolean isActiveAt(LocalDateTime now) {
+        if (status == ProgramStatus.CANCELLED) return false;
+        if (openAt == null || closeAt == null) return false;
+        return !openAt.isAfter(now) && !closeAt.isBefore(now);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof ProgramMeta that)) return false;
+        return programId.equals(that.programId);
+    }
+
+    @Override
+    public int hashCode() {
+        return programId.hashCode();
+    }
+}