fix: 분산 시스템 미적용 및 데이터 페이로드 문제 수정

ImGdevel · ImGdevel · commit c548522cec8b · 2025-09-25T13:00:56.000+09:00
diff --git a/ProjectVG.Api/Middleware/WebSocketMiddleware.cs b/ProjectVG.Api/Middleware/WebSocketMiddleware.cs
@@ -1,6 +1,7 @@
 using ProjectVG.Application.Services.Session;
 using ProjectVG.Infrastructure.Auth;
 using ProjectVG.Infrastructure.Realtime.WebSocketConnection;
+using ProjectVG.Domain.Services.Server;
 using System.Net.WebSockets;
 
 namespace ProjectVG.Api.Middleware
@@ -12,19 +13,22 @@ public class WebSocketMiddleware
         private readonly ISessionManager _sessionManager;
         private readonly IWebSocketConnectionManager _connectionManager;
         private readonly IJwtProvider _jwtProvider;
+        private readonly IServerRegistrationService? _serverRegistrationService;
 
         public WebSocketMiddleware(
             RequestDelegate next,
             ILogger<WebSocketMiddleware> logger,
             ISessionManager sessionManager,
             IWebSocketConnectionManager connectionManager,
-            IJwtProvider jwtProvider)
+            IJwtProvider jwtProvider,
+            IServerRegistrationService? serverRegistrationService = null)
         {
             _next = next;
             _logger = logger;
             _sessionManager = sessionManager;
             _connectionManager = connectionManager;
             _jwtProvider = jwtProvider;
+            _serverRegistrationService = serverRegistrationService;
         }
 
         public async Task InvokeAsync(HttpContext context)
@@ -113,6 +117,22 @@ private async Task RegisterConnection(Guid userId, WebSocket socket)
                 _connectionManager.RegisterConnection(userIdString, connection);
                 _logger.LogInformation("[WebSocketMiddleware] 로컬 WebSocket 연결 등록 완료: UserId={UserId}", userId);
 
+                // 3. 분산 시스템: 사용자-서버 매핑 저장 (Redis)
+                if (_serverRegistrationService != null)
+                {
+                    try
+                    {
+                        var serverId = _serverRegistrationService.GetServerId();
+                        await _serverRegistrationService.SetUserServerAsync(userIdString, serverId);
+                        _logger.LogInformation("[WebSocketMiddleware] 사용자-서버 매핑 저장 완료: UserId={UserId}, ServerId={ServerId}", userId, serverId);
+                    }
+                    catch (Exception mapEx)
+                    {
+                        _logger.LogWarning(mapEx, "[WebSocketMiddleware] 사용자-서버 매핑 저장 실패: UserId={UserId}", userId);
+                        // 매핑 저장 실패는 로그만 남기고 연결은 계속 진행
+                    }
+                }
+
                 // [디버그] 등록 후 상태 확인
                 var isSessionActive = await _sessionManager.IsSessionActiveAsync(userId);
                 var hasLocalConnection = _connectionManager.HasLocalConnection(userIdString);
@@ -219,11 +239,25 @@ await socket.SendAsync(
                         _logger.LogDebug("세션 관리자에서 세션 삭제 완료: {UserId}", userId);
                     }
 
-                    // 2. 로컬 WebSocket 연결 해제
+                    // 2. 분산 시스템: 사용자-서버 매핑 제거 (Redis)
+                    if (_serverRegistrationService != null)
+                    {
+                        try
+                        {
+                            await _serverRegistrationService.RemoveUserServerAsync(userId);
+                            _logger.LogDebug("사용자-서버 매핑 제거 완료: {UserId}", userId);
+                        }
+                        catch (Exception mapEx)
+                        {
+                            _logger.LogWarning(mapEx, "사용자-서버 매핑 제거 실패: {UserId}", userId);
+                        }
+                    }
+
+                    // 3. 로컬 WebSocket 연결 해제
                     _connectionManager.UnregisterConnection(userId);
                     _logger.LogDebug("로컬 WebSocket 연결 해제 완료: {UserId}", userId);
 
-                    // 3. WebSocket 소켓 정리
+                    // 4. WebSocket 소켓 정리
                     if (socket.State == WebSocketState.Open || socket.State == WebSocketState.CloseReceived) {
                         await socket.CloseAsync(
                             WebSocketCloseStatus.NormalClosure,
diff --git a/ProjectVG.Application/ApplicationServiceCollectionExtensions.cs b/ProjectVG.Application/ApplicationServiceCollectionExtensions.cs
@@ -1,6 +1,7 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using StackExchange.Redis;
 using ProjectVG.Application.Services.Auth;
 using ProjectVG.Application.Services.Character;
 using ProjectVG.Application.Services.Chat;
@@ -75,7 +76,19 @@ private static void AddChatServices(IServiceCollection services)
 
         private static void AddDistributedServices(IServiceCollection services, IConfiguration configuration)
         {
-            services.AddSingleton<IMessageBroker, DistributedMessageBroker>();
+            // DistributedMessageBroker를 즉시 생성하도록 팩토리 패턴 사용
+            services.AddSingleton<IMessageBroker>(serviceProvider =>
+            {
+                var redis = serviceProvider.GetRequiredService<IConnectionMultiplexer>();
+                var connectionManager = serviceProvider.GetRequiredService<IWebSocketConnectionManager>();
+                var serverRegistration = serviceProvider.GetRequiredService<ProjectVG.Domain.Services.Server.IServerRegistrationService>();
+                var logger = serviceProvider.GetRequiredService<ILogger<DistributedMessageBroker>>();
+
+                logger.LogInformation("[DI] DistributedMessageBroker 팩토리에서 생성 시작");
+                var broker = new DistributedMessageBroker(redis, connectionManager, serverRegistration, logger);
+                logger.LogInformation("[DI] DistributedMessageBroker 팩토리에서 생성 완료");
+                return broker;
+            });
 
             services.AddSingleton<ISessionManager>(serviceProvider =>
             {
@@ -85,6 +98,9 @@ private static void AddDistributedServices(IServiceCollection services, IConfigu
             });
 
             AddWebSocketConnectionServices(services);
+
+            // MessageBroker 초기화를 강제하는 HostedService 등록
+            services.AddHostedService<MessageBrokerInitializationService>();
         }
 
         private static void AddWebSocketConnectionServices(IServiceCollection services)
diff --git a/ProjectVG.Application/Models/MessageBroker/BrokerMessage.cs b/ProjectVG.Application/Models/MessageBroker/BrokerMessage.cs
@@ -15,12 +15,25 @@ public class BrokerMessage
 
         public static BrokerMessage CreateUserMessage(string userId, object payload, string? sourceServerId = null)
         {
+            string payloadJson;
+
+            // WebSocketMessage인 경우 이미 올바른 형태이므로 그대로 직렬화
+            if (payload is ProjectVG.Application.Models.WebSocket.WebSocketMessage wsMessage)
+            {
+                payloadJson = JsonSerializer.Serialize(wsMessage);
+            }
+            else
+            {
+                // 다른 객체의 경우 그대로 직렬화 (불필요한 래핑 방지)
+                payloadJson = JsonSerializer.Serialize(payload);
+            }
+
             return new BrokerMessage
             {
                 MessageType = "user_message",
                 TargetUserId = userId,
                 SourceServerId = sourceServerId,
-                Payload = JsonSerializer.Serialize(payload),
+                Payload = payloadJson,
                 Headers = new Dictionary<string, string>
                 {
                     ["content-type"] = "application/json"
diff --git a/ProjectVG.Application/Services/MessageBroker/DistributedMessageBroker.cs b/ProjectVG.Application/Services/MessageBroker/DistributedMessageBroker.cs
@@ -5,6 +5,7 @@
 using ProjectVG.Application.Services.Session;
 using StackExchange.Redis;
 using System.Collections.Concurrent;
+using System.Text.Json;
 
 namespace ProjectVG.Application.Services.MessageBroker
 {
@@ -32,35 +33,58 @@ public DistributedMessageBroker(
             IServerRegistrationService serverRegistration,
             ILogger<DistributedMessageBroker> logger)
         {
-            _redis = redis;
-            _subscriber = redis.GetSubscriber();
-            _connectionManager = connectionManager;
-            _serverRegistration = serverRegistration;
-            _logger = logger;
-            _serverId = serverRegistration.GetServerId();
-
-            InitializeSubscriptions();
+            _redis = redis ?? throw new ArgumentNullException(nameof(redis));
+            _connectionManager = connectionManager ?? throw new ArgumentNullException(nameof(connectionManager));
+            _serverRegistration = serverRegistration ?? throw new ArgumentNullException(nameof(serverRegistration));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
+            _logger.LogInformation("[분산브로커] DistributedMessageBroker 생성자 시작");
+
+            try
+            {
+                _subscriber = redis.GetSubscriber();
+                _serverId = serverRegistration.GetServerId();
+
+                _logger.LogInformation("[분산브로커] Redis 연결 상태: IsConnected={IsConnected}, ServerId={ServerId}",
+                    redis.IsConnected, _serverId);
+
+                InitializeSubscriptions();
+
+                _logger.LogInformation("[분산브로커] DistributedMessageBroker 생성 완료: ServerId={ServerId}", _serverId);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "[분산브로커] DistributedMessageBroker 생성자 실패");
+                throw;
+            }
         }
 
         private void InitializeSubscriptions()
         {
             try
             {
+                _logger.LogInformation("[분산브로커] Redis 구독 초기화 시작: ServerId={ServerId}", _serverId);
+
                 // 이 서버로 오는 메시지 구독
                 var serverChannel = $"{SERVER_CHANNEL_PREFIX}:{_serverId}";
+                _logger.LogInformation("[분산브로커] 서버 채널 구독 시작: Channel={Channel}", serverChannel);
                 _subscriber.Subscribe(serverChannel, OnServerMessageReceived);
+                _logger.LogInformation("[분산브로커] 서버 채널 구독 완료: Channel={Channel}", serverChannel);
 
                 // 브로드캐스트 메시지 구독
+                _logger.LogInformation("[분산브로커] 브로드캐스트 채널 구독 시작: Channel={Channel}", BROADCAST_CHANNEL);
                 _subscriber.Subscribe(BROADCAST_CHANNEL, OnBroadcastMessageReceived);
+                _logger.LogInformation("[분산브로커] 브로드캐스트 채널 구독 완료: Channel={Channel}", BROADCAST_CHANNEL);
 
                 // 사용자별 메시지 패턴 구독 (현재 서버에 연결된 사용자들만)
                 // 사용자가 연결될 때 동적으로 구독하도록 변경 예정
 
-                _logger.LogInformation("분산 메시지 브로커 구독 초기화 완료: 서버 {ServerId}", _serverId);
+                _logger.LogInformation("[분산브로커] 분산 메시지 브로커 구독 초기화 완료: ServerId={ServerId}", _serverId);
             }
             catch (Exception ex)
             {
-                _logger.LogError(ex, "분산 메시지 브로커 구독 초기화 실패");
+                _logger.LogError(ex, "[분산브로커] 분산 메시지 브로커 구독 초기화 실패: ServerId={ServerId}", _serverId);
+                throw;
             }
         }
 
@@ -92,14 +116,14 @@ public async Task SendToUserAsync(string userId, object message)
                     return;
                 }
 
-                // 3. 해당 서버로 메시지 전송
+                // 3. 해당 서버로 메시지 전송 (서버별 채널 사용)
                 var brokerMessage = BrokerMessage.CreateUserMessage(userId, message, _serverId);
-                var userChannel = $"{USER_CHANNEL_PREFIX}:{userId}";
+                var serverChannel = $"{SERVER_CHANNEL_PREFIX}:{targetServerId}";
 
                 _logger.LogInformation("[분산브로커] Redis Pub 시작: Channel={Channel}, TargetServerId={TargetServerId}, SourceServerId={SourceServerId}",
-                    userChannel, targetServerId, _serverId);
+                    serverChannel, targetServerId, _serverId);
 
-                await _subscriber.PublishAsync(userChannel, brokerMessage.ToJson());
+                await _subscriber.PublishAsync(serverChannel, brokerMessage.ToJson());
 
                 _logger.LogInformation("[분산브로커] Redis Pub 완료: UserId={UserId}, TargetServerId={TargetServerId}", userId, targetServerId);
             }
@@ -245,9 +269,41 @@ private async void OnServerMessageReceived(RedisChannel channel, RedisValue mess
                 }
 
                 // 서버별 메시지 처리 로직
-                _logger.LogDebug("서버 메시지 수신: {MessageType}", brokerMessage.MessageType);
+                _logger.LogInformation("[분산브로커] 서버 메시지 수신: MessageType={MessageType}, SourceServerId={SourceServerId}",
+                    brokerMessage.MessageType, brokerMessage.SourceServerId);
 
-                // TODO: 서버별 메시지 타입에 따른 처리 로직 구현
+                // 사용자 메시지 처리
+                if (brokerMessage.MessageType == "user_message" && !string.IsNullOrEmpty(brokerMessage.TargetUserId))
+                {
+                    _logger.LogInformation("[분산브로커] 사용자 메시지 처리 시작: TargetUserId={TargetUserId}",
+                        brokerMessage.TargetUserId);
+
+                    // 해당 사용자가 이 서버에 연결되어 있는지 확인
+                    if (_connectionManager.HasLocalConnection(brokerMessage.TargetUserId))
+                    {
+                        _logger.LogInformation("[분산브로커] 원본 Payload JSON: {PayloadJson}", brokerMessage.Payload);
+
+                        // 원본 JSON 문자열을 직접 사용하여 메시지 전달
+                        await SendLocalMessageAsJson(brokerMessage.TargetUserId, brokerMessage.Payload);
+
+                        _logger.LogInformation("[분산브로커] 사용자 메시지 전달 완료: TargetUserId={TargetUserId}",
+                            brokerMessage.TargetUserId);
+                    }
+                    else
+                    {
+                        _logger.LogWarning("[분산브로커] 대상 사용자가 이 서버에 연결되어 있지 않음: TargetUserId={TargetUserId}, ServerId={ServerId}",
+                            brokerMessage.TargetUserId, _serverId);
+                    }
+                }
+                else if (brokerMessage.MessageType == "server_message")
+                {
+                    // 다른 서버별 메시지 타입 처리 (향후 확장)
+                    _logger.LogDebug("[분산브로커] 서버 메시지 처리: {MessageType}", brokerMessage.MessageType);
+                }
+                else
+                {
+                    _logger.LogWarning("[분산브로커] 알 수 없는 메시지 타입: {MessageType}", brokerMessage.MessageType);
+                }
             }
             catch (Exception ex)
             {
@@ -320,6 +376,64 @@ private async void OnBroadcastMessageReceived(RedisChannel channel, RedisValue m
             }
         }
 
+        private async Task SendLocalMessageAsJson(string userId, string payloadJson)
+        {
+            if (string.IsNullOrEmpty(payloadJson))
+            {
+                _logger.LogWarning("[분산브로커] SendLocalMessageAsJson: Payload가 비어있습니다. UserId={UserId}", userId);
+                return;
+            }
+
+            _logger.LogInformation("[분산브로커] SendLocalMessageAsJson 시작: UserId={UserId}, PayloadLength={PayloadLength}",
+                userId, payloadJson.Length);
+
+            try
+            {
+                // 원본 JSON이 이미 WebSocketMessage 형태인지 확인
+                using var document = JsonDocument.Parse(payloadJson);
+                var root = document.RootElement;
+
+                string messageText;
+
+                // WebSocketMessage 구조인지 확인 (type과 data 필드가 있는지)
+                if (root.TryGetProperty("type", out var typeProperty) &&
+                    root.TryGetProperty("data", out var dataProperty))
+                {
+                    // 이미 WebSocketMessage 형태이므로 그대로 사용
+                    messageText = payloadJson;
+                    _logger.LogInformation("[분산브로커] WebSocketMessage 형태 감지: Type={Type}", typeProperty.GetString());
+                }
+                else
+                {
+                    // 일반 객체이므로 WebSocketMessage로 래핑 (예상되지 않는 케이스)
+                    _logger.LogWarning("[분산브로커] 예상하지 못한 JSON 구조, WebSocketMessage로 래핑: UserId={UserId}", userId);
+                    var wrappedMessage = new WebSocketMessage("message", root);
+                    messageText = System.Text.Json.JsonSerializer.Serialize(wrappedMessage);
+                }
+
+                _logger.LogInformation("[분산브로커] 최종 전송 메시지: {MessageText}", messageText);
+
+                var success = await _connectionManager.SendTextAsync(userId, messageText);
+                if (success)
+                {
+                    _logger.LogInformation("[분산브로커] JSON 메시지 전송 완료: UserId={UserId}", userId);
+                }
+                else
+                {
+                    _logger.LogWarning("[분산브로커] JSON 메시지 전송 실패: UserId={UserId}", userId);
+                }
+            }
+            catch (JsonException ex)
+            {
+                _logger.LogError(ex, "[분산브로커] JSON 파싱 실패: UserId={UserId}, Payload={Payload}", userId, payloadJson);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "[분산브로커] SendLocalMessageAsJson 실패: UserId={UserId}", userId);
+                throw;
+            }
+        }
+
         private async Task SendLocalMessage(string userId, object? message)
         {
             if (message == null)
@@ -335,12 +449,18 @@ private async Task SendLocalMessage(string userId, object? message)
             {
                 string messageText;
 
+                // WebSocketMessage는 이미 올바른 형태이므로 그대로 직렬화
                 if (message is WebSocketMessage wsMessage)
                 {
                     messageText = System.Text.Json.JsonSerializer.Serialize(wsMessage);
+                    _logger.LogInformation("[분산브로커] WebSocketMessage 직렬화: UserId={UserId}, Type={Type}",
+                        userId, wsMessage.Type);
                 }
                 else
                 {
+                    // 다른 객체는 WebSocketMessage로 래핑 (하지만 ChatSuccessHandler에서는 이미 래핑됨)
+                    _logger.LogWarning("[분산브로커] 예상하지 못한 객체 타입: {MessageType}, UserId={UserId}",
+                        message.GetType().Name, userId);
                     var wrappedMessage = new WebSocketMessage("message", message);
                     messageText = System.Text.Json.JsonSerializer.Serialize(wrappedMessage);
                 }
diff --git a/ProjectVG.Application/Services/MessageBroker/MessageBrokerInitializationService.cs b/ProjectVG.Application/Services/MessageBroker/MessageBrokerInitializationService.cs
