// ...

        SecurityContext context = SecurityContextHolder.createEmptyContext();
        context.setAuthentication(createAuthentication(username));
        SecurityContextHolder.setContext(context);
    }

    // Authentication 객체 생성 (UPAT 생성)
    private Authentication createAuthentication(String username) {
        UserDetails userDetails = userDetailsService.loadUserByUsername(username);
        return new UsernamePasswordAuthenticationToken(userDetails, null, userDetails.getAuthorities());
    }

// Auth Filter in API Gateway

// 인증 요청 Kafka 전송
return kafkaProducerTemplate.send(topic, id.toString(), tokenDTO)
        .then(Mono.defer(() -> {
            // 인증 응답 대기
            return kafkaReceiver
                    .receive()
                    .filter(record -> record.key().equals(id.toString()))
                    .next()
                    .map(ConsumerRecord::value)
                    .flatMap(userInfoDTO -> {
                        // ...

// Auth Filter in API Gateway

// 인증 요청 Kafka 전송
return kafkaProducerTemplate.send(topic, tokenDTO)
        .flatMap(result -> {
            // Kafka에 메시지 전송 후 Redis에서 결과를 비동기적으로 조회
            return checkRedisForUserInfo(id)
                    .flatMap(userInfoDTO -> {
                    // ...

// redis 캐시 조회 메소드
private Mono<UserInfoDTO> checkRedisForUserInfo(String id) {
    return Mono.defer(() -> {
            UserInfoDTO userInfo = userInfoTemplate.opsForValue().get(REDIS_ACCESS_KEY + id);

                if (userInfo != null) {
                    log.info("Redis에서 사용자 정보 조회 성공 - UserInfo: {}", userInfo);  // 성공 시 로그 추가

                    if (!id.equals(userInfo.getId())) {
                        userInfoTemplate.delete(REDIS_ACCESS_KEY + id);
                        userInfoTemplate.opsForValue().set(REDIS_ACCESS_KEY + userInfo.getId(), userInfo, 120 * 30, TimeUnit.SECONDS);
                    }

                    return Mono.just(userInfo);
                } else {
                    log.info("Redis에서 사용자 정보가 없음 - ID: {}\n 조회하는 시간: {}", id, System.nanoTime());  // 데이터가 없는 경우 로그 추가
                    return Mono.empty();
                }
            })
            .repeatWhenEmpty(flux -> flux
                    .delayElements(Duration.ofMillis(50)) // 0.05초 간격으로 재시도
                    .take(200)) // 재시도
            .timeout(Duration.ofSeconds(10)) // 타임아웃 설정
            .onErrorResume(e ->
                    Mono.error(new ResponseStatusException(
                            HttpStatus.INTERNAL_SERVER_ERROR, "레디스 유저 임시정보 조회 에러", e)));
    }

// chat instance

@Getter
@Setter
@Entity
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@Table(name = "openchats")
@EntityListeners(AuditingEntityListener.class)
public class ChatRoom {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(name = "id")
    private Long id;

    // ...

    @Column(name = "personnel")
    private Integer personnel;

// message instance

@Slf4j(topic = "ChatMessageService")
@Service
@RequiredArgsConstructor
public class ChatMessageService {

    // ...

    public void enter(ChatMessageDTO.Enter enter, @AuthenticationPrincipal UserDetailsImpl userDetails) {
        String username = userDetails.getUsername();
        String role = userDetails.getRole();

        // ...

        // GraphQL을 통해 chat instance에 있는 채팅방 엔티티 인스턴스 필드 업데이트
        graphqlService.incrementPersonnel(enter.getChatId(), username, role);

// 인메모리 데이터베이스 h2 기반 동시성 제어 테스트
// 메모리 기반이기 떄문에 테스트 종료시, 자동 휘발됨(데이터베이스까지!)
@Slf4j
@SpringBootTest
@ActiveProfiles("test")
public class ConcurrencyTest {

    @Autowired
    private ChatRoomRepository chatRoomRepository;

    @Autowired
    private ChatGraphqlServiceImpl chatGraphqlService;

    // ...

    @BeforeEach
    @DisplayName("임의의 채팅방 객체 생성")
    void setUp() {
        ChatRoomDTO dto = new ChatRoomDTO(TITLE, MAX_PERSONNEL);
        ChatRoom chatRoom = new ChatRoom(dto, "openUser");

        chatRoomRepository.save(chatRoom);
    }

    // ...

    @Test
    @DisplayName("트래픽이 몰리는 상황에서 정합성이 지켜지는지 테스트")
    void testWithNoLock() throws InterruptedException {
        // given & when...

        // 스레드 풀 및 동시 시작 장치
        ExecutorService executorService = Executors.newFixedThreadPool(CLIENT);
        CountDownLatch countDownLatch = new CountDownLatch(CLIENT);

        // 성공 결과를 담을 자료구조
        List<GraphqlDTO> result = new ArrayList<>();

        for (int i = 0; i < CLIENT; i++) {
            executorService.submit(() -> {
               try {
                   GraphqlDTO success =
                           chatGraphqlService
                                   .incrementPersonnel(chatRoom.getId().toString());
                   result.add(success);
                   log.info("입장 성공!: {}", success);
               } catch (Exception e) {
                   log.error("입장 실패! : {}", e.getMessage());
               } finally {
                   countDownLatch.countDown(); // 카운트 감소
               }
            });
        }

        countDownLatch.await(); // 스레드 종료 대기
        executorService.shutdown(); // 서비스 종료

        // then
        assertThat(result.size())
                .describedAs("예상 인원 수: %d, 실제 인원 수: %d", MAX_PERSONNEL, result.size())
                .isGreaterThan(MAX_PERSONNEL);
    }

@Slf4j
@Component
@RequiredArgsConstructor
public class DistributedLockFacade {

    private final RedissonClient redissonClient;
    private final ChatGraphqlService chatGraphqlService;

    public GraphqlDTO incrementPersonnel(String id) {

        RLock lock = redissonClient.getLock(id);

        try {
            boolean available = lock.tryLock(10, 1, TimeUnit.SECONDS);

            if (!available) {
                log.error("락 획득 실패");
                return null;
            }

            log.info("락 획득 성공");
            return chatGraphqlService.incrementPersonnel(id);
        } catch (InterruptedException e) {
            log.error("락 획득 예외 발생: {}", e.getMessage());
            throw new RuntimeException(e);
        } finally {
            log.info("락 잠금 해제");
            lock.unlock();
        }
    }
}

// message instance

@Getter
@Entity
@NoArgsConstructor(access = AccessLevel.PROTECTED)
@Table(name = "chat_message")
public class ChatMessage {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    @Column(name = "id")
    private Long id;

    @Column(name = "chat_id")
    private String chatId;

    @Column(name = "type")
    private ChatMessageType type;

    @Column(name = "usernmae")
    private String username;

    @Column(name = "message")
    private String message;

    @Column(name = "created")
    private LocalDateTime createdAt;

    public ChatMessage(ChatMessageDTO dto) {
        this.chatId = dto.getChatId();
        this.type = dto.getType();
        this.username = dto.getUsername();
        this.message = dto.getMessage();
        this.createdAt = dto.getCreatedAt();
    }
}

@Repository
public interface ChatMessageRepository extends JpaRepository<ChatMessage, Long> {

    // 네이티브 쿼리 적용
    @Query(value = "SELECT * FROM chat_message WHERE chat_id = :chatId AND created >= :startTime AND created <= :endTime", nativeQuery = true)
    List<ChatMessage> findMessagesByChatIdAndTimeRange(String chatId, LocalDateTime startTime, LocalDateTime endTime);
}

public record ChatUserReadDTO(String chatId, String email, Boolean read, Boolean leave) {
}

@Slf4j(topic = "WEBSOCKET_EVENT_LISTENING")
@Controller
@RequiredArgsConstructor
public class WebSocketEventListener {

    private final EventQueueService eventQueueService;

    @EventListener
    public void handleWebSocketSubscribeEvent(SessionSubscribeEvent event) throws HttpResponseException {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
        String destination = accessor.getDestination();
        String sessionId = accessor.getSessionId();

        if (destination == null || !destination.startsWith("/sub/participant/")) {
            throw new HttpResponseException(
                    HttpStatus.SC_SERVICE_UNAVAILABLE,
                    "Server Error, 구독 경로 확인 필요"
            );
        }

        log.info("구독 시점의 세션 아이디: {}", sessionId);

        String id = destination.replaceFirst("/sub/participant/", "");
        log.info("구독할 때의 경로 아이디 갖고오기: {}", id);

        // ...

@Slf4j(topic = "QUEUE_SERVICE")
@Component
@RequiredArgsConstructor
public class EventQueueService {

    private final Map<String, String> sessionIdChannelMap = new ConcurrentHashMap<>();
    private final Map<String, Boolean> subscriptionFlags = new ConcurrentHashMap<>();
    private final Map<String, Queue<List<UserReadDTO>>> messageQueue = new ConcurrentHashMap<>();
    private final SimpMessageSendingOperations messagingTemplate;

    // 세션 아이디 - 채널 저장
    public void setSessionIdChannelMap(String sessionId, String chatId) {
        sessionIdChannelMap.put(sessionId, chatId);
    }

    // ...

@Slf4j
@RequiredArgsConstructor
@Service
public class RedisMessageListenerService implements MessageListener {

    private final EventQueueService eventQueueService;
    private final SimpMessageSendingOperations messagingTemplate;

    // ...

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String channel = new String(message.getChannel());
        String id = channel.replaceFirst("chat_", "");

        String body = new String(message.getBody());

        log.info("채널 확인: {} // 파싱: {}", channel, id);

        try {
            // ...

            if (eventQueueService.isSubscriptionComplete(id)) {
                // 구독 완료 -> 바로 전송
                log.info("{}번 구독 확인, 송신", id);
                messagingTemplate.convertAndSend("/sub/participant/" + id, data);
            } else {
                // 구독 미완료 -> 대기열에 저장
                log.info("{}번 구독 미완료, 데이터 대기열에 저장: {}", id, data);
                eventQueueService.enqueueMessage(id, data);
            }

@Slf4j(topic = "QUEUE_SERVICE")
@Component
@RequiredArgsConstructor
public class EventQueueService {

    private final Map<String, String> sessionIdChannelMap = new ConcurrentHashMap<>();
    private final Map<String, Boolean> subscriptionFlags = new ConcurrentHashMap<>();
    private final Map<String, Queue<List<UserReadDTO>>> messageQueue = new ConcurrentHashMap<>();
    private final SimpMessageSendingOperations messagingTemplate;

    // 세션 아이디 - 채널 저장
    public void setSessionIdChannelMap(String sessionId, String chatId) {
        sessionIdChannelMap.put(sessionId, chatId);
    }

    // 세션 아이디 - 채널 및 채널의 플래그, 큐 삭제
    public void removeAllFields(String sessionId) {
        String chatId = sessionIdChannelMap.get(sessionId);

        if (chatId == null) {
            throw new IllegalArgumentException("채팅 아이디가 세션 맵에 저장 안 됨");
        }

        subscriptionFlags.remove(chatId);
        messageQueue.remove(chatId);
        sessionIdChannelMap.remove(sessionId);
    }

    // 구독 완료 처리
    public void onSubscriptionComplete(String clientId) {
        subscriptionFlags.put(clientId, true);
        flushQueueForClient(clientId);
    }

    // 구독 완료 여부 확인
    public boolean isSubscriptionComplete(String clientId) {
        return subscriptionFlags.getOrDefault(clientId, false);
    }

    // 대기열에 메시지 저장
    public void enqueueMessage(String clientId, List<UserReadDTO> message) {
        messageQueue.computeIfAbsent(clientId, k -> new ConcurrentLinkedQueue<>()).add(message);
    }

    // 대기열 처리
    private void flushQueueForClient(String clientId) {
        log.info("{}번 큐 플러시 과정 돌입", clientId);
        Queue<List<UserReadDTO>> queue = messageQueue.get(clientId);
        if (queue != null) {
            while (!queue.isEmpty()) {
                log.info("{}번 큐 데이터 산입 확인: {}", clientId, queue.size());
                List<UserReadDTO> message = queue.poll();
                // 대기열에 쌓인 메시지를 클라이언트로 전송하는 로직
                sendMessageToClient(clientId, message);
                log.info("*** 데이터 송신 완료 ***");
            }
        }
    }

    private void sendMessageToClient(String clientId, List<UserReadDTO> message) {
        // WebSocket 클라이언트로 메시지를 전송하는 로직
        log.info("대기열 이후 최종 메세지 송신: {}번", clientId);
        messagingTemplate.convertAndSend("/sub/participant/" + clientId, message);
    }

}