feat: AiRun 진행 상태 SSE 구독 추가

.env.example

@@ -31,6 +31,13 @@ AI_RUNTIME_MAX_CONCURRENT_CALLS=8
 AI_RUNTIME_CIRCUIT_BREAKER_FAILURE_THRESHOLD=5
 AI_RUNTIME_CIRCUIT_BREAKER_OPEN_DURATION=30s
 
+# Client용 AiRun 상태 SSE입니다. 실행 기준은 SSE가 아니라 DB와 GET /api/v1/ai-runs/{id}입니다.
+AI_RUN_SSE_TIMEOUT=5m
+AI_RUN_SSE_HEARTBEAT_INTERVAL=15s
+AI_RUN_SSE_HISTORY_RETENTION=10m
+AI_RUN_SSE_HISTORY_SIZE=20
+AI_RUN_SSE_MAX_CONNECTIONS=2
+
 # Transactional Outbox worker 설정입니다.
 # 일반 실행에서는 켜 두며, 운영 장애 조사 중 자동 처리를 멈춰야 할 때만 false로 둡니다.
 OUTBOX_ENABLED=true

src/main/java/com/fowoco/server/airun/api/AiRunController.java

@@ -2,20 +2,29 @@
 
 import com.fowoco.server.airun.application.AiCandidateDecisionCommand;
 import com.fowoco.server.airun.application.AiCandidateDecisionService;
+import com.fowoco.server.airun.application.AiRunPublicEvent;
 import com.fowoco.server.airun.application.AiRunService;
 import com.fowoco.server.auth.application.ActorContext;
 import com.fowoco.server.auth.application.port.ActorContextProvider;
+import com.fowoco.server.common.error.ApiException;
 import com.fowoco.server.common.web.RequestMetadata;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.media.Content;
+import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import io.swagger.v3.oas.annotations.responses.ApiResponses;
 import io.swagger.v3.oas.annotations.security.SecurityRequirement;
 import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.validation.Valid;
 import jakarta.servlet.http.HttpServletRequest;
+import jakarta.validation.Valid;
+import java.io.IOException;
 import java.net.URI;
+import java.util.Map;
 import java.util.UUID;
+import org.springframework.http.CacheControl;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -26,6 +35,7 @@
 import org.springframework.web.bind.annotation.RequestHeader;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 import org.springframework.web.servlet.support.ServletUriComponentsBuilder;
 
 @Tag(name = "AI Run", description = "자연어 업무 분석 실행·질문·답변")
@@ -37,15 +47,18 @@ public class AiRunController {
     private final AiRunService aiRunService;
     private final AiCandidateDecisionService candidateDecisionService;
     private final ActorContextProvider actorContextProvider;
+    private final AiRunEventStreamBroker eventStreamBroker;
 
     public AiRunController(
             AiRunService aiRunService,
             AiCandidateDecisionService candidateDecisionService,
-            ActorContextProvider actorContextProvider
+            ActorContextProvider actorContextProvider,
+            AiRunEventStreamBroker eventStreamBroker
     ) {
         this.aiRunService = aiRunService;
         this.candidateDecisionService = candidateDecisionService;
         this.actorContextProvider = actorContextProvider;
+        this.eventStreamBroker = eventStreamBroker;
     }
 
     @Operation(
@@ -69,7 +82,7 @@ public ResponseEntity<AiRunResponse> create(
             @Valid @RequestBody CreateAiRunRequest request,
             HttpServletRequest servletRequest
     ) {
-        AiRunResponse response = AiRunResponse.from(aiRunService.createAndExecute(
+        AiRunResponse response = AiRunResponse.from(aiRunService.createAndSchedule(
                 request.instruction(),
                 idempotencyKey,
                 actor(),
@@ -93,6 +106,74 @@ public AiRunResponse findById(@PathVariable UUID aiRunId) {
         return AiRunResponse.from(aiRunService.requireRun(aiRunId, actor()));
     }
 
+    @Operation(
+            operationId = "subscribeAiRunEvents",
+            summary = "AI 분석 공개 상태 SSE 구독",
+            description = "Server DB의 AiRun 상태를 화면용 단방향 event로 제공합니다. "
+                    + "연결이 끊겨도 실행은 계속되며 Client는 기존 상세 조회 API로 최종 상태를 확인합니다. "
+                    + "브라우저 기본 EventSource는 Authorization header를 넣을 수 없으므로 fetch 기반 SSE Client를 사용합니다."
+    )
+    @ApiResponses({
+            @ApiResponse(
+                    responseCode = "200",
+                    description = "상태 event stream",
+                    content = @Content(
+                            mediaType = MediaType.TEXT_EVENT_STREAM_VALUE,
+                            schema = @Schema(implementation = AiRunPublicEvent.class)
+                    )
+            ),
+            @ApiResponse(responseCode = "400", ref = "#/components/responses/BadRequest"),
+            @ApiResponse(responseCode = "404", ref = "#/components/responses/NotFound"),
+            @ApiResponse(responseCode = "429", description = "사용자·AiRun별 연결 수 제한 초과")
+    })
+    @PreAuthorize("hasAnyRole('ADMIN', 'HR', 'VIEWER')")
+    @GetMapping(path = "/{aiRunId}/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public ResponseEntity<SseEmitter> subscribeEvents(
+            @PathVariable UUID aiRunId,
+            @Parameter(description = "마지막으로 처리한 SSE event id")
+            @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId
+    ) {
+        try {
+            ActorContext actor = actor();
+            SseEmitter emitter = eventStreamBroker.subscribe(
+                    actor,
+                    aiRunService.requireRun(aiRunId, actor),
+                    lastEventId
+            );
+            return streamResponse(HttpStatus.OK, emitter);
+        } catch (ApiException exception) {
+            return streamResponse(exception.errorCode().status(), errorEmitter(exception));
+        }
+    }
+
+    private ResponseEntity<SseEmitter> streamResponse(
+            HttpStatus status,
+            SseEmitter emitter
+    ) {
+        return ResponseEntity.status(status)
+                .cacheControl(CacheControl.noCache())
+                .header("X-Accel-Buffering", "no")
+                .header(HttpHeaders.CONNECTION, "keep-alive")
+                .contentType(MediaType.TEXT_EVENT_STREAM)
+                .body(emitter);
+    }
+
+    private SseEmitter errorEmitter(ApiException exception) {
+        SseEmitter emitter = new SseEmitter(1_000L);
+        try {
+            emitter.send(SseEmitter.event()
+                    .name("ERROR")
+                    .data(Map.of(
+                            "code", exception.errorCode().code(),
+                            "message", exception.getMessage()
+                    )));
+            emitter.complete();
+        } catch (IOException sendFailure) {
+            emitter.completeWithError(sendFailure);
+        }
+        return emitter;
+    }
+
     @Operation(operationId = "answerAiRunQuestions", summary = "누락 Slot 답변 제출")
     @ApiResponses({
             @ApiResponse(responseCode = "202", description = "답변 저장 및 새 분석 시도"),

src/main/java/com/fowoco/server/airun/api/AiRunEventStreamBroker.java

@@ -0,0 +1,219 @@
+package com.fowoco.server.airun.api;
+
+import com.fowoco.server.airun.application.AiRunPublicEvent;
+import com.fowoco.server.airun.application.AiRunResult;
+import com.fowoco.server.airun.application.error.AiRunErrorCode;
+import com.fowoco.server.airun.application.port.AiRunPublicEventPublisher;
+import com.fowoco.server.auth.application.ActorContext;
+import com.fowoco.server.common.error.ApiException;
+import java.io.IOException;
+import java.time.Clock;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+@Component
+@EnableConfigurationProperties(AiRunEventStreamProperties.class)
+public class AiRunEventStreamBroker implements AiRunPublicEventPublisher {
+
+    private final AiRunEventStreamProperties properties;
+    private final Clock clock;
+    private final Map<RunKey, EventHistory> histories = new ConcurrentHashMap<>();
+    private final Map<SubscriptionKey, CopyOnWriteArrayList<StreamSession>> subscriptions =
+            new ConcurrentHashMap<>();
+
+    public AiRunEventStreamBroker(AiRunEventStreamProperties properties, Clock clock) {
+        this.properties = properties;
+        this.clock = clock;
+    }
+
+    @Override
+    public void publish(UUID companyId, AiRunResult run) {
+        RunKey runKey = new RunKey(companyId, run.aiRunId());
+        AiRunPublicEvent event = AiRunPublicEvent.from(run);
+        histories.computeIfAbsent(runKey, ignored -> new EventHistory())
+                .record(event, clock.instant(), properties.historySize());
+
+        subscriptions.forEach((key, sessions) -> {
+            if (key.companyId().equals(companyId) && key.aiRunId().equals(run.aiRunId())) {
+                sessions.forEach(session -> send(session, key, event));
+            }
+        });
+    }
+
+    public SseEmitter subscribe(ActorContext actor, AiRunResult current, String lastEventIdHeader) {
+        Long lastEventId = parseLastEventId(lastEventIdHeader);
+        RunKey runKey = new RunKey(actor.companyId(), current.aiRunId());
+        SubscriptionKey subscriptionKey = new SubscriptionKey(
+                actor.companyId(),
+                actor.actorId(),
+                current.aiRunId()
+        );
+        CopyOnWriteArrayList<StreamSession> sessions = subscriptions.computeIfAbsent(
+                subscriptionKey,
+                ignored -> new CopyOnWriteArrayList<>()
+        );
+        StreamSession session;
+        synchronized (sessions) {
+            if (sessions.size() >= properties.maxConnectionsPerUserRun()) {
+                throw new ApiException(AiRunErrorCode.AI_RUN_SSE_CONNECTION_LIMIT);
+            }
+            SseEmitter emitter = new SseEmitter(properties.timeout().toMillis());
+            session = new StreamSession(emitter, lastEventId == null ? -1L : lastEventId);
+            sessions.add(session);
+        }
+        registerCallbacks(subscriptionKey, session);
+
+        List<AiRunPublicEvent> replay = histories.getOrDefault(runKey, EventHistory.EMPTY)
+                .after(lastEventId == null ? -1L : lastEventId);
+        if (replay.isEmpty() && (lastEventId == null || current.version() > lastEventId)) {
+            replay = List.of(AiRunPublicEvent.from(current));
+        }
+        replay.forEach(event -> send(session, subscriptionKey, event));
+
+        AiRunPublicEvent currentEvent = AiRunPublicEvent.from(current);
+        if (currentEvent.terminal() && currentEvent.eventId() <= session.lastSentEventId()) {
+            complete(subscriptionKey, session);
+        }
+        return session.emitter();
+    }
+
+    @Scheduled(fixedDelayString = "${app.ai-run.sse.heartbeat-interval:15s}")
+    void heartbeatAndCleanup() {
+        subscriptions.forEach((key, sessions) -> sessions.forEach(session -> {
+            try {
+                session.emitter().send(SseEmitter.event().comment("heartbeat"));
+            } catch (IOException | IllegalStateException exception) {
+                complete(key, session);
+            }
+        }));
+
+        Instant cutoff = clock.instant().minus(properties.historyRetention());
+        histories.entrySet().removeIf(entry -> entry.getValue().lastPublishedAt().isBefore(cutoff));
+    }
+
+    private Long parseLastEventId(String value) {
+        if (value == null || value.isBlank()) {
+            return null;
+        }
+        try {
+            long parsed = Long.parseLong(value.strip());
+            if (parsed < 0) {
+                throw new NumberFormatException("negative event id");
+            }
+            return parsed;
+        } catch (NumberFormatException exception) {
+            throw new ApiException(AiRunErrorCode.AI_RUN_INVALID_LAST_EVENT_ID);
+        }
+    }
+
+    private void send(StreamSession session, SubscriptionKey key, AiRunPublicEvent event) {
+        try {
+            boolean sent = session.send(event);
+            if (sent && event.terminal()) {
+                complete(key, session);
+            }
+        } catch (IOException | IllegalStateException exception) {
+            complete(key, session);
+        }
+    }
+
+    private void registerCallbacks(SubscriptionKey key, StreamSession session) {
+        session.emitter().onCompletion(() -> remove(key, session));
+        session.emitter().onTimeout(() -> complete(key, session));
+        session.emitter().onError(ignored -> remove(key, session));
+    }
+
+    private void complete(SubscriptionKey key, StreamSession session) {
+        remove(key, session);
+        try {
+            session.emitter().complete();
+        } catch (IllegalStateException ignored) {
+            // 이미 완료된 연결입니다.
+        }
+    }
+
+    private void remove(SubscriptionKey key, StreamSession session) {
+        CopyOnWriteArrayList<StreamSession> sessions = subscriptions.get(key);
+        if (sessions == null) {
+            return;
+        }
+        sessions.remove(session);
+        if (sessions.isEmpty()) {
+            subscriptions.remove(key, sessions);
+        }
+    }
+
+    private record RunKey(UUID companyId, UUID aiRunId) {
+    }
+
+    private record SubscriptionKey(UUID companyId, UUID actorId, UUID aiRunId) {
+    }
+
+    private static final class StreamSession {
+        private final SseEmitter emitter;
+        private long lastSentEventId;
+
+        private StreamSession(SseEmitter emitter, long lastSentEventId) {
+            this.emitter = emitter;
+            this.lastSentEventId = lastSentEventId;
+        }
+
+        private synchronized boolean send(AiRunPublicEvent event) throws IOException {
+            if (event.eventId() <= lastSentEventId) {
+                return false;
+            }
+            emitter.send(SseEmitter.event()
+                    .id(Long.toString(event.eventId()))
+                    .name(event.type().name())
+                    .data(event));
+            lastSentEventId = event.eventId();
+            return true;
+        }
+
+        private SseEmitter emitter() {
+            return emitter;
+        }
+
+        private synchronized long lastSentEventId() {
+            return lastSentEventId;
+        }
+    }
+
+    private static final class EventHistory {
+        private static final EventHistory EMPTY = new EventHistory();
+
+        private final List<AiRunPublicEvent> events = new ArrayList<>();
+        private Instant lastPublishedAt = Instant.EPOCH;
+
+        private synchronized void record(AiRunPublicEvent event, Instant publishedAt, int limit) {
+            if (events.stream().noneMatch(existing -> existing.eventId() == event.eventId())) {
+                events.add(event);
+                events.sort(Comparator.comparingLong(AiRunPublicEvent::eventId));
+                while (events.size() > limit) {
+                    events.remove(0);
+                }
+            }
+            lastPublishedAt = publishedAt;
+        }
+
+        private synchronized List<AiRunPublicEvent> after(long eventId) {
+            return events.stream()
+                    .filter(event -> event.eventId() > eventId)
+                    .toList();
+        }
+
+        private synchronized Instant lastPublishedAt() {
+            return lastPublishedAt;
+        }
+    }
+}