using System.Collections.Concurrent;
using System.Security.Claims;
using MediatR;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
using Knot.Modules.Chats.Application.Messages.Send;
using Knot.Modules.Chats.Domain;
using Knot.Shared.Kernel;
using Microsoft.Extensions.Caching.Memory;
namespace Knot.Modules.Chats.Infrastructure.SignalR;
///
/// Хаб SignalR для обработки сообщений и WebRTC сигналинга в реальном времени.
///
[Authorize]
public sealed class ChatHub : Hub
{
// Маппинг userId → список connectionId
private static readonly ConcurrentDictionary> _userConnections = new();
// chatId → (userId → ParticipantInfo)
private static readonly ConcurrentDictionary> _groupCallParticipants = new();
public static int OnlineUsersCount => _userConnections.Count;
private readonly ISender _sender;
private readonly IUserContext _userContext;
private readonly IChatRepository _chatRepository;
private readonly ILogger _logger;
private readonly IMemoryCache _cache;
public ChatHub(ISender sender, IUserContext userContext, IChatRepository chatRepository, ILogger logger, IMemoryCache cache)
{
_sender = sender;
_userContext = userContext;
_chatRepository = chatRepository;
_logger = logger;
_cache = cache;
}
public override async Task OnConnectedAsync()
{
if (_userContext.IsAuthenticated)
{
var userId = _userContext.UserId.ToString();
_userConnections.AddOrUpdate(
userId,
_ => new HashSet { Context.ConnectionId },
(_, set) => { lock (set) { set.Add(Context.ConnectionId); } return set; }
);
_cache.Set("Global_OnlineUsersCount", _userConnections.Count);
var userChats = await _chatRepository.GetUserChatsAsync(_userContext.UserId, Context.ConnectionAborted);
foreach (var chat in userChats)
{
await Groups.AddToGroupAsync(Context.ConnectionId, chat.Id.ToString());
}
_logger.LogInformation("User {UserId} connected with {ConnectionId}, added to {ChatCount} chats",
userId, Context.ConnectionId, userChats.Count);
await Clients.Others.SendAsync("user_online", new { userId });
}
await base.OnConnectedAsync();
}
public override async Task OnDisconnectedAsync(Exception? exception)
{
if (_userContext.IsAuthenticated)
{
var userId = _userContext.UserId.ToString();
if (_userConnections.TryGetValue(userId, out var set))
{
lock (set) { set.Remove(Context.ConnectionId); }
if (set.Count == 0)
{
_userConnections.TryRemove(userId, out _);
await Clients.Others.SendAsync("user_offline", new { userId, lastSeen = DateTime.UtcNow });
}
}
_cache.Set("Global_OnlineUsersCount", _userConnections.Count);
_logger.LogInformation("User {UserId} disconnected", userId);
}
await base.OnDisconnectedAsync(exception);
}
// ────────────────────────────────────────────────────────────────
// Chat methods
// ────────────────────────────────────────────────────────────────
[HubMethodName("send_message")]
public async Task SendMessage(SendMessageHubRequest request)
{
var attachments = request.Attachments?.Select(a =>
new AttachmentRequest(a.Type, a.Url, a.FileName, a.FileSize)).ToList();
var command = new SendMessageCommand(
request.ChatId,
_userContext.UserId,
request.Content,
request.Type,
attachments,
request.ReplyToId,
request.Quote,
request.ForwardedFromId);
await _sender.Send(command);
}
[HubMethodName("read_messages")]
public async Task ReadMessages(ReadMessagesRequest request)
{
if (request.MessageIds != null && request.MessageIds.Any())
{
var parsedIds = request.MessageIds
.Select(id => Guid.TryParse(id, out var parsed) ? parsed : Guid.Empty)
.Where(id => id != Guid.Empty)
.ToList();
if (parsedIds.Any())
{
var command = new Knot.Modules.Chats.Application.Messages.Read.ReadMessagesCommand(
request.ChatId, _userContext.UserId, parsedIds);
await _sender.Send(command);
}
}
await Clients.Group(request.ChatId.ToString()).SendAsync("messages_read", new
{
ChatId = request.ChatId.ToString(),
UserId = _userContext.UserId,
MessageIds = request.MessageIds ?? new List()
});
}
[HubMethodName("delete_messages")]
public async Task DeleteMessages(DeleteMessagesHubRequest request)
{
var parsedIds = request.MessageIds
.Select(id => Guid.TryParse(id, out var parsed) ? parsed : Guid.Empty)
.Where(id => id != Guid.Empty)
.ToList();
if (parsedIds.Any())
{
var command = new Knot.Modules.Chats.Application.Messages.Delete.DeleteMessagesCommand(
request.ChatId, _userContext.UserId, parsedIds, request.DeleteForAll);
await _sender.Send(command);
}
}
[HubMethodName("typing_start")]
public async Task TypingStart(string chatId)
{
await Clients.Group(chatId).SendAsync("user_typing", new { ChatId = chatId, UserId = _userContext.UserId });
}
[HubMethodName("typing_stop")]
public async Task TypingStop(string chatId)
{
await Clients.Group(chatId).SendAsync("user_stopped_typing", new { ChatId = chatId, UserId = _userContext.UserId });
}
[HubMethodName("join_chat")]
public async Task JoinChat(string chatId)
{
// Simple security check: check if user is member of chat (optional but recommended)
if (Guid.TryParse(chatId, out var chatGuid))
{
var userChats = await _chatRepository.GetUserChatsAsync(_userContext.UserId, Context.ConnectionAborted);
if (userChats.Any(c => c.Id == chatGuid))
{
await Groups.AddToGroupAsync(Context.ConnectionId, chatId);
}
}
}
[HubMethodName("add_reaction")]
public async Task AddReaction(AddReactionRequest request)
{
_logger.LogInformation("AddReaction called: MessageId={MessageId}, ChatId={ChatId}, Emoji={Emoji}, UserId={UserId}",
request.MessageId, request.ChatId, request.Emoji, _userContext.UserId);
var command = new Knot.Modules.Chats.Application.Messages.React.AddReactionCommand(
request.MessageId, _userContext.UserId, request.Emoji, request.ChatId);
var result = await _sender.Send(command);
if (result.IsFailure)
{
_logger.LogWarning("AddReaction failed: {Error}", result.Error.Description);
throw new HubException(result.Error.Description);
}
_logger.LogInformation("AddReaction completed successfully");
}
[HubMethodName("remove_reaction")]
public async Task RemoveReaction(RemoveReactionRequest request)
{
_logger.LogInformation("RemoveReaction called: MessageId={MessageId}, ChatId={ChatId}, Emoji={Emoji}, UserId={UserId}",
request.MessageId, request.ChatId, request.Emoji, _userContext.UserId);
var command = new Knot.Modules.Chats.Application.Messages.React.RemoveReactionCommand(
request.MessageId, _userContext.UserId, request.Emoji, request.ChatId);
var result = await _sender.Send(command);
if (result.IsFailure)
{
_logger.LogWarning("RemoveReaction failed: {Error}", result.Error.Description);
throw new HubException(result.Error.Description);
}
_logger.LogInformation("RemoveReaction completed successfully");
}
// ────────────────────────────────────────────────────────────────
// Friend signals (Proxy methods for real-time notification)
// ────────────────────────────────────────────────────────────────
[HubMethodName("friend_request")]
public async Task FriendRequest(FriendSignalRequest request)
{
await SendToUserAsync(request.FriendId, "friend_request_received", new { userId = _userContext.UserId });
}
[HubMethodName("friend_accepted")]
public async Task FriendAccepted(FriendSignalRequest request)
{
await SendToUserAsync(request.FriendId, "friend_request_accepted", new { userId = _userContext.UserId });
}
[HubMethodName("friend_removed")]
public async Task FriendRemoved(FriendSignalRequest request)
{
await SendToUserAsync(request.FriendId, "friend_removed_notify", new { userId = _userContext.UserId });
}
// ────────────────────────────────────────────────────────────────
// WebRTC signaling
// ────────────────────────────────────────────────────────────────
private async Task SendToUserAsync(string targetUserId, string method, object payload)
{
if (_userConnections.TryGetValue(targetUserId, out var connectionIds))
{
string[] ids;
lock (connectionIds) { ids = connectionIds.ToArray(); }
foreach (var connId in ids)
{
await Clients.Client(connId).SendAsync(method, payload);
}
}
}
[HubMethodName("call_offer")]
public async Task CallOffer(CallOfferRequest request)
{
// Try to get caller info from current user's claims
var displayName = Context.User?.FindFirstValue("name") ?? Context.User?.FindFirstValue(ClaimTypes.Name) ?? "User";
var avatar = Context.User?.FindFirstValue("avatar");
var username = Context.User?.FindFirstValue("unique_name") ?? Context.User?.Identity?.Name;
await SendToUserAsync(request.TargetUserId, "call_incoming", new
{
from = _userContext.UserId.ToString(),
offer = request.Offer,
callType = request.CallType,
chatId = request.ChatId,
callerInfo = new
{
id = _userContext.UserId.ToString(),
displayName = displayName,
avatar = avatar,
username = username
}
});
}
[HubMethodName("call_answer")]
public async Task CallAnswer(CallAnswerRequest request)
{
await SendToUserAsync(request.TargetUserId, "call_answered", new
{
from = _userContext.UserId.ToString(),
answer = request.Answer,
});
}
[HubMethodName("call_decline")]
public async Task CallDecline(TargetUserRequest request)
{
await SendToUserAsync(request.TargetUserId, "call_declined", new
{
from = _userContext.UserId.ToString(),
});
}
[HubMethodName("call_end")]
public async Task CallEnd(TargetUserRequest request)
{
await SendToUserAsync(request.TargetUserId, "call_ended", new
{
from = _userContext.UserId.ToString(),
});
}
[HubMethodName("ice_candidate")]
public async Task IceCandidate(IceCandidateRequest request)
{
await SendToUserAsync(request.TargetUserId, "ice_candidate", new
{
from = _userContext.UserId.ToString(),
candidate = request.Candidate,
});
}
[HubMethodName("renegotiate")]
public async Task Renegotiate(RenegotiateRequest request)
{
await SendToUserAsync(request.TargetUserId, "renegotiate", new
{
from = _userContext.UserId.ToString(),
offer = request.Offer,
});
}
[HubMethodName("renegotiate_answer")]
public async Task RenegotiateAnswer(RenegotiateAnswerRequest request)
{
await SendToUserAsync(request.TargetUserId, "renegotiate_answer", new
{
from = _userContext.UserId.ToString(),
answer = request.Answer,
});
}
[HubMethodName("call_type_changed")]
public async Task CallTypeChanged(CallTypeChangedRequest request)
{
await SendToUserAsync(request.TargetUserId, "call_type_changed", new
{
from = _userContext.UserId.ToString(),
callType = request.CallType,
isScreenSharing = request.IsScreenSharing
});
}
[HubMethodName("call_status")]
public async Task CallStatus(CallStatusRequest request)
{
await SendToUserAsync(request.TargetUserId, "call_status_updated", new
{
from = _userContext.UserId.ToString(),
isMuted = request.IsMuted,
isVideoOff = request.IsVideoOff
});
}
// ────────────────────────────────────────────────────────────────
// Group Call signals
// ────────────────────────────────────────────────────────────────
[HubMethodName("group_call_join")]
public async Task GroupCallJoin(GroupCallJoinRequest request)
{
var chatId = request.ChatId;
var userId = _userContext.UserId.ToString();
var displayName = Context.User?.FindFirstValue("name") ?? Context.User?.FindFirstValue(ClaimTypes.Name) ?? "User";
var avatar = Context.User?.FindFirstValue("avatar");
var username = Context.User?.FindFirstValue("unique_name") ?? Context.User?.Identity?.Name ?? "user";
var userInfo = new ParticipantInfo(userId, username, displayName, avatar);
var participants = _groupCallParticipants.GetOrAdd(chatId, _ => new ConcurrentDictionary());
var isFirst = participants.IsEmpty;
participants.TryAdd(userId, userInfo);
// Notify others
await Clients.Group(chatId).SendAsync("group_call_user_joined", new
{
chatId = chatId,
userId = userId,
userInfo = userInfo
});
// Send current participants to joiner (excluding self)
var others = participants.Values.Where(p => p.Id != userId).ToList();
await Clients.Caller.SendAsync("group_call_participants", new
{
chatId = chatId,
participants = others
});
// Broadcast active call participants to everyone in the chat
await Clients.Group(chatId).SendAsync("group_call_active", new
{
chatId = chatId,
participants = participants.Keys.ToList()
});
if (isFirst)
{
await Clients.Group(chatId).SendAsync("group_call_incoming", new
{
chatId = chatId,
from = userId,
callerInfo = userInfo,
callType = request.CallType
});
}
}
[HubMethodName("group_call_leave")]
public async Task GroupCallLeave(GroupLeaveRequest request)
{
var chatId = request.ChatId;
var userId = _userContext.UserId.ToString();
if (_groupCallParticipants.TryGetValue(chatId, out var participants))
{
participants.TryRemove(userId, out _);
// Broadcast updated participants list
await Clients.Group(chatId).SendAsync("group_call_active", new
{
chatId = chatId,
participants = participants.Keys.ToList()
});
if (participants.IsEmpty)
{
_groupCallParticipants.TryRemove(chatId, out _);
await Clients.Group(chatId).SendAsync("group_call_ended", new { chatId = chatId });
}
}
await Clients.Group(chatId).SendAsync("group_call_user_left", new
{
chatId = chatId,
userId = userId
});
}
[HubMethodName("group_call_offer")]
public async Task GroupCallOffer(GroupCallOfferRequest request)
{
await SendToUserAsync(request.TargetUserId, "group_call_offer", new
{
chatId = request.ChatId,
from = _userContext.UserId.ToString(),
offer = request.Offer
});
}
[HubMethodName("group_call_answer")]
public async Task GroupCallAnswer(GroupCallAnswerRequest request)
{
await SendToUserAsync(request.TargetUserId, "group_call_answer", new
{
chatId = request.ChatId,
from = _userContext.UserId.ToString(),
answer = request.Answer
});
}
[HubMethodName("group_ice_candidate")]
public async Task GroupIceCandidate(GroupIceCandidateRequest request)
{
await SendToUserAsync(request.TargetUserId, "group_ice_candidate", new
{
chatId = request.ChatId,
from = _userContext.UserId.ToString(),
candidate = request.Candidate
});
}
[HubMethodName("group_call_renegotiate")]
public async Task GroupRenegotiate(GroupRenegotiateRequest request)
{
await SendToUserAsync(request.TargetUserId, "group_call_renegotiate", new
{
chatId = request.ChatId,
from = _userContext.UserId.ToString(),
offer = request.Offer
});
}
[HubMethodName("group_call_renegotiate_answer")]
public async Task GroupRenegotiateAnswer(GroupRenegotiateAnswerRequest request)
{
await SendToUserAsync(request.TargetUserId, "group_call_renegotiate_answer", new
{
chatId = request.ChatId,
from = _userContext.UserId.ToString(),
answer = request.Answer
});
}
[HubMethodName("group_call_status")]
public async Task GroupCallStatus(GroupCallStatusRequest request)
{
var chatId = request.ChatId;
var userId = _userContext.UserId.ToString();
if (_groupCallParticipants.TryGetValue(chatId, out var participants))
{
if (participants.TryGetValue(userId, out var info))
{
// Update local state if needed (e.g. muted status)
participants[userId] = info with
{
IsMuted = request.IsMuted,
IsVideoOff = request.IsVideoOff
};
}
}
await Clients.Group(request.ChatId).SendAsync("group_call_status_updated", new {
chatId = request.ChatId,
userId = Context.UserIdentifier,
isMuted = request.IsMuted,
isVideoOff = request.IsVideoOff
});
}
[HubMethodName("group_call_status_params")]
public async Task GroupCallStatusParams(string chatId, bool isMuted, bool isVideoOff)
{
var userId = _userContext.UserId.ToString();
if (_groupCallParticipants.TryGetValue(chatId, out var participants))
{
if (participants.TryGetValue(userId, out var info))
{
// Update local state if needed (e.g. muted status)
participants[userId] = info with
{
IsMuted = isMuted,
IsVideoOff = isVideoOff
};
}
}
await Clients.Group(chatId).SendAsync("group_call_status_updated", new {
chatId = chatId,
userId = Context.UserIdentifier,
isMuted = isMuted,
isVideoOff = isVideoOff
});
}
[HubMethodName("get_group_call_status")]
public async Task GetGroupCallStatus(GetGroupCallStatusRequest request)
{
var chatId = request.ChatId;
if (_groupCallParticipants.TryGetValue(chatId, out var participants))
{
var others = participants.Values.ToList();
_logger.LogInformation("Found {Count} participants for chat {ChatId}", others.Count, chatId);
await Clients.Caller.SendAsync("group_call_active", new
{
chatId = chatId,
participants = others.Select(p => p.Id).ToList()
});
}
else
{
_logger.LogInformation("No active call for chat {ChatId}", chatId);
await Clients.Caller.SendAsync("group_call_active", new
{
chatId = chatId,
participants = new List()
});
}
}
[HubMethodName("screen_share_started")]
public async Task ScreenShareStarted(string chatId)
{
var userId = _userContext.UserId.ToString();
if (_groupCallParticipants.TryGetValue(chatId, out var participants))
{
if (participants.TryGetValue(userId, out var info))
{
participants[userId] = info with { IsSharingScreen = true };
}
}
await Clients.Group(chatId).SendAsync("screen_share_started", new
{
chatId = chatId,
userId = userId
});
}
[HubMethodName("screen_share_stopped")]
public async Task ScreenShareStopped(string chatId)
{
var userId = _userContext.UserId.ToString();
if (_groupCallParticipants.TryGetValue(chatId, out var participants))
{
if (participants.TryGetValue(userId, out var info))
{
participants[userId] = info with { IsSharingScreen = false };
}
}
await Clients.Group(chatId).SendAsync("screen_share_stopped", new
{
chatId = chatId,
userId = userId
});
}
// ────────────────────────────────────────────────────────────────
// Records
// ────────────────────────────────────────────────────────────────
public record AttachmentHubRequest(string Type, string Url, string? FileName, long? FileSize);
public record SendMessageHubRequest(
Guid ChatId,
string? Content,
string Type,
List? Attachments = null,
Guid? ReplyToId = null,
string? Quote = null,
Guid? ForwardedFromId = null);
public record ReadMessagesRequest(Guid ChatId, List? MessageIds);
public record CallOfferRequest(string TargetUserId, object Offer, string CallType, string? ChatId);
public record CallAnswerRequest(string TargetUserId, object Answer);
public record TargetUserRequest(string TargetUserId);
public record IceCandidateRequest(string TargetUserId, object Candidate);
public record RenegotiateRequest(string TargetUserId, object Offer);
public record RenegotiateAnswerRequest(string TargetUserId, object Answer);
public record CallTypeChangedRequest(string TargetUserId, string CallType, bool IsScreenSharing = false);
public record CallStatusRequest(string TargetUserId, bool IsMuted, bool IsVideoOff);
public record AddReactionRequest(Guid MessageId, Guid ChatId, string Emoji);
public record RemoveReactionRequest(Guid MessageId, Guid ChatId, string Emoji);
public record DeleteMessagesHubRequest(Guid ChatId, List MessageIds, bool DeleteForAll);
public record GroupCallJoinRequest(string ChatId, string CallType);
public record ParticipantInfo(string Id, string Username, string DisplayName, string? Avatar, bool IsSharingScreen = false, bool IsMuted = false, bool IsVideoOff = false);
public record GroupLeaveRequest(string ChatId);
public record GetGroupCallStatusRequest(string ChatId);
public record GroupCallStatusRequest(string ChatId, bool IsMuted, bool IsVideoOff);
public record GroupCallOfferRequest(string ChatId, string TargetUserId, object Offer);
public record GroupCallAnswerRequest(string ChatId, string TargetUserId, object Answer);
public record GroupIceCandidateRequest(string ChatId, string TargetUserId, object Candidate);
public record GroupRenegotiateRequest(string ChatId, string TargetUserId, object Offer);
public record GroupRenegotiateAnswerRequest(string ChatId, string TargetUserId, object Answer);
public record FriendSignalRequest(string FriendId);
}