✨ Realtime call participants

🐛 Fix update, delete message wont send websocket packet

DysonNetwork.Sphere/Account/Relationship.cs

@@ -2,7 +2,7 @@ using NodaTime;
 
 namespace DysonNetwork.Sphere.Account;
 
-public enum RelationshipStatus
+public enum RelationshipStatus : short
 {
     Friends = 100,
     Pending = 0,

DysonNetwork.Sphere/Account/RelationshipService.cs

@@ -88,8 +88,9 @@ public class RelationshipService(AppDatabase db, ICacheService cache)
         var relationship = await GetRelationship(accountId, relatedId, RelationshipStatus.Pending);
         if (relationship is null) throw new ArgumentException("Friend request was not found.");
     
-        db.AccountRelationships.Remove(relationship);
+        await db.AccountRelationships
-        await db.SaveChangesAsync();
+            .Where(r => r.AccountId == accountId && r.RelatedId == relatedId && r.Status == RelationshipStatus.Pending)
+            .ExecuteDeleteAsync();
         
         await cache.RemoveAsync($"{UserFriendsCacheKeyPrefix}{accountId}");
         await cache.RemoveAsync($"{UserFriendsCacheKeyPrefix}{relatedId}");

DysonNetwork.Sphere/Chat/ChatController.cs

@@ -230,6 +230,7 @@ public partial class ChatController(AppDatabase db, ChatService cs, FileService
             .Include(m => m.Sender)
             .Include(m => m.Sender.Account)
             .Include(m => m.Sender.Account.Profile).Include(message => message.Attachments)
+            .Include(message => message.ChatRoom)
             .FirstOrDefaultAsync(m => m.Id == messageId && m.ChatRoomId == roomId);
         if (message == null) return NotFound();
 
@@ -270,6 +271,12 @@ public partial class ChatController(AppDatabase db, ChatService cs, FileService
         message.EditedAt = SystemClock.Instance.GetCurrentInstant();
         db.Update(message);
         await db.SaveChangesAsync();
+        _ = cs.DeliverMessageAsync(
+            message,
+            message.Sender,
+            message.ChatRoom,
+            WebSocketPacketType.MessageUpdate
+        );
 
         return Ok(message);
     }
@@ -282,6 +289,7 @@ public partial class ChatController(AppDatabase db, ChatService cs, FileService
 
         var message = await db.ChatMessages
             .Include(m => m.Sender)
+            .Include(m => m.ChatRoom)
             .FirstOrDefaultAsync(m => m.Id == messageId && m.ChatRoomId == roomId);
         if (message == null) return NotFound();
 
@@ -290,6 +298,12 @@ public partial class ChatController(AppDatabase db, ChatService cs, FileService
 
         db.ChatMessages.Remove(message);
         await db.SaveChangesAsync();
+        _ = cs.DeliverMessageAsync(
+            message,
+            message.Sender,
+            message.ChatRoom,
+            WebSocketPacketType.MessageDelete
+        );
 
         return Ok();
     }

DysonNetwork.Sphere/Chat/ChatService.cs

@@ -39,7 +39,12 @@ public class ChatService(
         return message;
     }
 
-    public async Task DeliverMessageAsync(Message message, ChatMember sender, ChatRoom room)
+    public async Task DeliverMessageAsync(
+        Message message,
+        ChatMember sender,
+        ChatRoom room,
+        string type = WebSocketPacketType.MessageNew
+    )
     {
         using var scope = scopeFactory.CreateScope();
         var scopedDb = scope.ServiceProvider.GetRequiredService<AppDatabase>();
@@ -60,7 +65,7 @@ public class ChatService(
         {
             scopedWs.SendPacketToAccount(member.AccountId, new WebSocketPacket
             {
-                Type = "messages.new",
+                Type = type,
                 Data = message
             });
             tasks.Add(scopedNty.DeliveryNotification(new Notification
@@ -185,7 +190,7 @@ public class ChatService(
             SenderId = sender.Id,
             Meta = new Dictionary<string, object>
             {
-                { "call", call.Id }
+                { "call_id", call.Id },
             }
         }, sender, room);
 
@@ -228,7 +233,8 @@ public class ChatService(
             SenderId = call.SenderId,
             Meta = new Dictionary<string, object>
             {
-                { "call", call.Id }
+                { "call_id", call.Id },
+                { "duration", (call.EndedAt!.Value - call.CreatedAt).TotalSeconds }
             }
         }, call.Sender, call.Room);
     }

DysonNetwork.Sphere/Chat/Realtime/IRealtimeService.cs

@@ -37,6 +37,15 @@ public interface IRealtimeService
     /// <param name="isAdmin">The user is the admin of session</param>
     /// <returns>User-specific token for the session</returns>
     string GetUserToken(Account.Account account, string sessionId, bool isAdmin = false);
+    
+    /// <summary>
+    /// Processes incoming webhook requests from the realtime service provider
+    /// </summary>
+    /// <param name="body">The webhook request body content</param>
+    /// <param name="authHeader">The authentication header value</param>
+    /// <returns>Task representing the asynchronous operation</returns>
+    Task ReceiveWebhook(string body, string authHeader);
+    
 }
 
 /// <summary>

DysonNetwork.Sphere/Chat/Realtime/LivekitService.cs

@@ -1,4 +1,9 @@
+using DysonNetwork.Sphere.Connection;
+using DysonNetwork.Sphere.Storage;
 using Livekit.Server.Sdk.Dotnet;
+using Microsoft.EntityFrameworkCore;
+using NodaTime;
+using System.Text.Json;
 
 namespace DysonNetwork.Sphere.Chat.Realtime;
 
@@ -7,11 +12,22 @@ namespace DysonNetwork.Sphere.Chat.Realtime;
 /// </summary>
 public class LivekitRealtimeService : IRealtimeService
 {
+    private readonly AppDatabase _db;
+    private readonly ICacheService _cache;
+    private readonly WebSocketService _ws;
+
     private readonly ILogger<LivekitRealtimeService> _logger;
     private readonly RoomServiceClient _roomService;
     private readonly AccessToken _accessToken;
+    private readonly WebhookReceiver _webhookReceiver;
 
-    public LivekitRealtimeService(IConfiguration configuration, ILogger<LivekitRealtimeService> logger)
+    public LivekitRealtimeService(
+        IConfiguration configuration,
+        ILogger<LivekitRealtimeService> logger,
+        AppDatabase db,
+        ICacheService cache,
+        WebSocketService ws
+    )
     {
         _logger = logger;
 
@@ -25,6 +41,11 @@ public class LivekitRealtimeService : IRealtimeService
 
         _roomService = new RoomServiceClient(host, apiKey, apiSecret);
         _accessToken = new AccessToken(apiKey, apiSecret);
+        _webhookReceiver = new WebhookReceiver(apiKey, apiSecret);
+
+        _db = db;
+        _cache = cache;
+        _ws = ws;
     }
 
     /// <inheritdoc />
@@ -49,7 +70,7 @@ public class LivekitRealtimeService : IRealtimeService
             {
                 Name = roomName,
                 EmptyTimeout = 300, // 5 minutes
-                Metadata = System.Text.Json.JsonSerializer.Serialize(roomMetadata)
+                Metadata = JsonSerializer.Serialize(roomMetadata)
             });
 
             // Return session config
@@ -108,4 +129,238 @@ public class LivekitRealtimeService : IRealtimeService
             .WithTtl(TimeSpan.FromHours(1));
         return token.ToJwt();
     }
+
+    public async Task ReceiveWebhook(string body, string authHeader)
+    {
+        var evt = _webhookReceiver.Receive(body, authHeader);
+        if (evt is null) return;
+
+        switch (evt.Event)
+        {
+            case "room_finished":
+                var now = SystemClock.Instance.GetCurrentInstant();
+                await _db.ChatRealtimeCall
+                    .Where(c => c.SessionId == evt.Room.Name)
+                    .ExecuteUpdateAsync(s => s.SetProperty(p => p.EndedAt, now)
+                    );
+
+                // Also clean up participants list when the room is finished
+                await _cache.RemoveAsync(_GetParticipantsKey(evt.Room.Name));
+                break;
+
+            case "participant_joined":
+                if (evt.Participant != null)
+                {
+                    // Add the participant to cache
+                    await _AddParticipantToCache(evt.Room.Name, evt.Participant);
+                    _logger.LogInformation(
+                        "Participant joined room: {RoomName}, Participant: {ParticipantIdentity}",
+                        evt.Room.Name, evt.Participant.Identity);
+
+                    // Broadcast participant list update to all participants
+                    await _BroadcastParticipantUpdate(evt.Room.Name);
+                }
+                break;
+
+            case "participant_left":
+                if (evt.Participant != null)
+                {
+                    // Remove the participant from cache
+                    await _RemoveParticipantFromCache(evt.Room.Name, evt.Participant);
+                    _logger.LogInformation(
+                        "Participant left room: {RoomName}, Participant: {ParticipantIdentity}",
+                        evt.Room.Name, evt.Participant.Identity);
+
+                    // Broadcast participant list update to all participants
+                    await _BroadcastParticipantUpdate(evt.Room.Name);
+                }
+                break;
+        }
+    }
+
+    private static string _GetParticipantsKey(string roomName)
+        => $"RoomParticipants_{roomName}";
+
+    private async Task _AddParticipantToCache(string roomName, ParticipantInfo participant)
+    {
+        var participantsKey = _GetParticipantsKey(roomName);
+
+        // Try to acquire a lock to prevent race conditions when updating the participants list
+        await using var lockObj = await _cache.AcquireLockAsync(
+            $"{participantsKey}_lock",
+            TimeSpan.FromSeconds(10),
+            TimeSpan.FromSeconds(5));
+
+        if (lockObj == null)
+        {
+            _logger.LogWarning("Failed to acquire lock for updating participants list in room: {RoomName}", roomName);
+            return;
+        }
+
+        // Get the current participants list
+        var participants = await _cache.GetAsync<List<ParticipantCacheItem>>(participantsKey) ??
+                           new List<ParticipantCacheItem>();
+
+        // Check if the participant already exists
+        var existingIndex = participants.FindIndex(p => p.Identity == participant.Identity);
+        if (existingIndex >= 0)
+        {
+            // Update existing participant
+            participants[existingIndex] = CreateParticipantCacheItem(participant);
+        }
+        else
+        {
+            // Add new participant
+            participants.Add(CreateParticipantCacheItem(participant));
+        }
+
+        // Update cache with new list
+        await _cache.SetAsync(participantsKey, participants, TimeSpan.FromHours(6));
+
+        // Also add to a room group in cache for easy cleanup
+        await _cache.AddToGroupAsync(participantsKey, $"Room_{roomName}");
+    }
+
+    private async Task _RemoveParticipantFromCache(string roomName, ParticipantInfo participant)
+    {
+        var participantsKey = _GetParticipantsKey(roomName);
+
+        // Try to acquire a lock to prevent race conditions when updating the participants list
+        await using var lockObj = await _cache.AcquireLockAsync(
+            $"{participantsKey}_lock",
+            TimeSpan.FromSeconds(10),
+            TimeSpan.FromSeconds(5));
+
+        if (lockObj == null)
+        {
+            _logger.LogWarning("Failed to acquire lock for updating participants list in room: {RoomName}", roomName);
+            return;
+        }
+
+        // Get current participants list
+        var participants = await _cache.GetAsync<List<ParticipantCacheItem>>(participantsKey);
+        if (participants == null || !participants.Any())
+            return;
+
+        // Remove participant
+        participants.RemoveAll(p => p.Identity == participant.Identity);
+
+        // Update cache with new list
+        await _cache.SetAsync(participantsKey, participants, TimeSpan.FromHours(6));
+    }
+
+    // Helper method to get participants in a room
+    public async Task<List<ParticipantCacheItem>> GetRoomParticipantsAsync(string roomName)
+    {
+        var participantsKey = _GetParticipantsKey(roomName);
+        return await _cache.GetAsync<List<ParticipantCacheItem>>(participantsKey) ?? new List<ParticipantCacheItem>();
+    }
+
+    // Class to represent a participant in the cache
+    public class ParticipantCacheItem
+    {
+        public string Identity { get; set; } = null!;
+        public string Name { get; set; } = null!;
+        public Guid? AccountId { get; set; }
+        public ParticipantInfo.Types.State State { get; set; }
+        public Dictionary<string, string> Metadata { get; set; } = new();
+        public DateTime JoinedAt { get; set; }
+    }
+
+    private ParticipantCacheItem CreateParticipantCacheItem(ParticipantInfo participant)
+    {
+        // Try to parse account ID from metadata
+        Guid? accountId = null;
+        var metadata = new Dictionary<string, string>();
+        
+        if (!string.IsNullOrEmpty(participant.Metadata))
+        {
+            try
+            {
+                metadata = JsonSerializer.Deserialize<Dictionary<string, string>>(participant.Metadata) ?? 
+                           new Dictionary<string, string>();
+                
+                if (metadata.TryGetValue("account_id", out var accountIdStr))
+                {
+                    if (Guid.TryParse(accountIdStr, out var parsedId))
+                    {
+                        accountId = parsedId;
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Failed to parse participant metadata");
+            }
+        }
+
+        return new ParticipantCacheItem
+        {
+            Identity = participant.Identity,
+            Name = participant.Name,
+            AccountId = accountId,
+            State = participant.State,
+            Metadata = metadata,
+            JoinedAt = DateTime.UtcNow
+        };
+    }
+    
+    // Broadcast participant update to all participants in a room
+    private async Task _BroadcastParticipantUpdate(string roomName)
+    {
+        try
+        {
+            // Get the room ID from the session name
+            var roomInfo = await _db.ChatRealtimeCall
+                .Where(c => c.SessionId == roomName && c.EndedAt == null)
+                .Select(c => new { c.RoomId, c.Id })
+                .FirstOrDefaultAsync();
+            
+            if (roomInfo == null)
+            {
+                _logger.LogWarning("Could not find room info for session: {SessionName}", roomName);
+                return;
+            }
+            
+            // Get current participants
+            var participants = await GetRoomParticipantsAsync(roomName);
+            
+            // Get all room members who should receive this update
+            var roomMembers = await _db.ChatMembers
+                .Where(m => m.ChatRoomId == roomInfo.RoomId && m.LeaveAt == null)
+                .Select(m => m.AccountId)
+                .ToListAsync();
+            
+            // Create the update packet
+            var participantsDto = participants.Select(p => new
+            {
+                p.Identity,
+                p.Name,
+                p.AccountId,
+                State = p.State.ToString(),
+                p.JoinedAt
+            }).ToList();
+            
+            var updatePacket = new WebSocketPacket
+            {
+                Type = WebSocketPacketType.CallParticipantsUpdate,
+                Data = new Dictionary<string, object>
+                {
+                    { "room_id", roomInfo.RoomId },
+                    { "call_id", roomInfo.Id },
+                    { "participants", participantsDto }
+                }
+            };
+            
+            // Send the update to all members
+            foreach (var accountId in roomMembers)
+            {
+                _ws.SendPacketToAccount(accountId, updatePacket);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Error broadcasting participant update for room {RoomName}", roomName);
+        }
+    }
 }

DysonNetwork.Sphere/Chat/RealtimeCallController.cs

@@ -1,7 +1,9 @@
 using DysonNetwork.Sphere.Chat.Realtime;
+using Livekit.Server.Sdk.Dotnet;
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.EntityFrameworkCore;
+using Swashbuckle.AspNetCore.Annotations;
 
 namespace DysonNetwork.Sphere.Chat;
 
@@ -12,42 +14,104 @@ public class RealtimeChatConfiguration
 
 [ApiController]
 [Route("/chat/realtime")]
-public class RealtimeCallController(IConfiguration configuration, AppDatabase db, ChatService cs, IRealtimeService realtime) : ControllerBase
+public class RealtimeCallController(
+    IConfiguration configuration,
+    AppDatabase db,
+    ChatService cs,
+    IRealtimeService realtime
+) : ControllerBase
 {
     private readonly RealtimeChatConfiguration _config =
         configuration.GetSection("RealtimeChat").Get<RealtimeChatConfiguration>()!;
 
+    /// <summary>
+    /// This endpoint is especially designed for livekit webhooks,
+    /// for update the call participates and more.
+    /// Learn more at: https://docs.livekit.io/home/server/webhooks/
+    /// </summary>
+    [HttpPost("webhook")]
+    [SwaggerIgnore]
+    public async Task<IActionResult> WebhookReceiver()
+    {
+        using var reader = new StreamReader(Request.Body);
+        var postData = await reader.ReadToEndAsync();
+        var authHeader = Request.Headers.Authorization.ToString();
+        
+        await realtime.ReceiveWebhook(postData, authHeader);
+    
+        return Ok();
+    }
+
+    [HttpGet("{roomId:guid}")]
+    [Authorize]
+    public async Task<ActionResult<RealtimeCall>> GetOngoingCall(Guid roomId)
+    {
+        if (HttpContext.Items["CurrentUser"] is not Account.Account currentUser) return Unauthorized();
+
+        var member = await db.ChatMembers
+            .Where(m => m.AccountId == currentUser.Id && m.ChatRoomId == roomId)
+            .FirstOrDefaultAsync();
+
+        if (member == null || member.Role < ChatMemberRole.Member)
+            return StatusCode(403, "You need to be a member to view call status.");
+
+        var ongoingCall = await db.ChatRealtimeCall
+            .Where(c => c.RoomId == roomId)
+            .Where(c => c.EndedAt == null)
+            .Include(c => c.Room)
+            .Include(c => c.Sender)
+            .ThenInclude(c => c.Account)
+            .ThenInclude(c => c.Profile)
+            .FirstOrDefaultAsync();
+        if (ongoingCall is null) return NotFound();
+        return Ok(ongoingCall);
+    }
+
     [HttpGet("{roomId:guid}/join")]
     [Authorize]
     public async Task<ActionResult<JoinCallResponse>> JoinCall(Guid roomId)
     {
         if (HttpContext.Items["CurrentUser"] is not Account.Account currentUser) return Unauthorized();
-        
+
         // Check if the user is a member of the chat room
         var member = await db.ChatMembers
             .Where(m => m.AccountId == currentUser.Id && m.ChatRoomId == roomId)
             .FirstOrDefaultAsync();
-        
+
         if (member == null || member.Role < ChatMemberRole.Member)
             return StatusCode(403, "You need to be a member to join a call.");
-        
+
         // Get ongoing call
         var ongoingCall = await cs.GetCallOngoingAsync(roomId);
-        if (ongoingCall is null) 
+        if (ongoingCall is null)
             return NotFound("There is no ongoing call in this room.");
-        
+
         // Check if session ID exists
         if (string.IsNullOrEmpty(ongoingCall.SessionId))
             return BadRequest("Call session is not properly configured.");
-        
+
         var isAdmin = member.Role >= ChatMemberRole.Moderator;
         var userToken = realtime.GetUserToken(currentUser, ongoingCall.SessionId, isAdmin);
-        
+
         // Get LiveKit endpoint from configuration
-        string endpoint = _config.Endpoint ?? 
+        var endpoint = _config.Endpoint ??
-                     throw new InvalidOperationException("LiveKit endpoint configuration is missing");
+                   throw new InvalidOperationException("LiveKit endpoint configuration is missing");
-        
+
-        // Create response model
+        // Get current participants from the LiveKit service
+        var participants = new List<CallParticipant>();
+        if (realtime is LivekitRealtimeService livekitService)
+        {
+            var roomParticipants = await livekitService.GetRoomParticipantsAsync(ongoingCall.SessionId);
+            participants = roomParticipants.Select(p => new CallParticipant
+            {
+                Identity = p.Identity,
+                Name = p.Name,
+                AccountId = p.AccountId,
+                JoinedAt = p.JoinedAt
+            }).ToList();
+        }
+
+        // Create the response model
         var response = new JoinCallResponse
         {
             Provider = realtime.ProviderName,
@@ -55,15 +119,16 @@ public class RealtimeCallController(IConfiguration configuration, AppDatabase db
             Token = userToken,
             CallId = ongoingCall.Id,
             RoomName = ongoingCall.SessionId,
-            IsAdmin = isAdmin
+            IsAdmin = isAdmin,
+            Participants = participants
         };
-        
+
         return Ok(response);
     }
 
     [HttpPost("{roomId:guid}")]
     [Authorize]
-    public async Task<IActionResult> StartCall(Guid roomId)
+    public async Task<ActionResult<RealtimeCall>> StartCall(Guid roomId)
     {
         if (HttpContext.Items["CurrentUser"] is not Account.Account currentUser) return Unauthorized();
 
@@ -82,7 +147,7 @@ public class RealtimeCallController(IConfiguration configuration, AppDatabase db
 
     [HttpDelete("{roomId:guid}")]
     [Authorize]
-    public async Task<IActionResult> EndCall(Guid roomId)
+    public async Task<ActionResult<RealtimeCall>> EndCall(Guid roomId)
     {
         if (HttpContext.Items["CurrentUser"] is not Account.Account currentUser) return Unauthorized();
 
@@ -111,29 +176,60 @@ public class JoinCallResponse
     /// The service provider name (e.g., "LiveKit")
     /// </summary>
     public string Provider { get; set; } = null!;
-    
+
     /// <summary>
     /// The LiveKit server endpoint
     /// </summary>
     public string Endpoint { get; set; } = null!;
-    
+
     /// <summary>
     /// Authentication token for the user
     /// </summary>
     public string Token { get; set; } = null!;
-    
+
     /// <summary>
     /// The call identifier
     /// </summary>
     public Guid CallId { get; set; }
-    
+
     /// <summary>
     /// The room name in LiveKit
     /// </summary>
     public string RoomName { get; set; } = null!;
-    
+
     /// <summary>
     /// Whether the user is the admin of the call
     /// </summary>
     public bool IsAdmin { get; set; }
+    
+    /// <summary>
+    /// Current participants in the call
+    /// </summary>
+    public List<CallParticipant> Participants { get; set; } = new();
+}
+
+/// <summary>
+/// Represents a participant in a real-time call
+/// </summary>
+public class CallParticipant
+{
+    /// <summary>
+    /// The participant's identity (username)
+    /// </summary>
+    public string Identity { get; set; } = null!;
+    
+    /// <summary>
+    /// The participant's display name
+    /// </summary>
+    public string Name { get; set; } = null!;
+    
+    /// <summary>
+    /// The participant's account ID if available
+    /// </summary>
+    public Guid? AccountId { get; set; }
+    
+    /// <summary>
+    /// When the participant joined the call
+    /// </summary>
+    public DateTime JoinedAt { get; set; }
 }

DysonNetwork.Sphere/Connection/WebSocketPacket.cs

@@ -5,6 +5,10 @@ using NodaTime.Serialization.SystemTextJson;
 public class WebSocketPacketType
 {
     public const string Error = "error";
+    public const string MessageNew = "messages.new";
+    public const string MessageUpdate = "messages.update";
+    public const string MessageDelete = "messages.delete";
+    public const string CallParticipantsUpdate = "call.participants.update";
 }
 
 public class WebSocketPacket

DysonNetwork.Sphere/Migrations/20250525083412_ModifyRelationshipStatusType.cs

@@ -0,0 +1,34 @@
+using Microsoft.EntityFrameworkCore.Migrations;
+
+#nullable disable
+
+namespace DysonNetwork.Sphere.Migrations
+{
+    /// <inheritdoc />
+    public partial class ModifyRelationshipStatusType : Migration
+    {
+        /// <inheritdoc />
+        protected override void Up(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.AlterColumn<short>(
+                name: "status",
+                table: "account_relationships",
+                type: "smallint",
+                nullable: false,
+                oldClrType: typeof(int),
+                oldType: "integer");
+        }
+
+        /// <inheritdoc />
+        protected override void Down(MigrationBuilder migrationBuilder)
+        {
+            migrationBuilder.AlterColumn<int>(
+                name: "status",
+                table: "account_relationships",
+                type: "integer",
+                nullable: false,
+                oldClrType: typeof(short),
+                oldType: "smallint");
+        }
+    }
+}

DysonNetwork.Sphere/Migrations/AppDatabaseModelSnapshot.cs

@@ -635,8 +635,8 @@ namespace DysonNetwork.Sphere.Migrations
                         .HasColumnType("timestamp with time zone")
                         .HasColumnName("expired_at");
 
-                    b.Property<int>("Status")
+                    b.Property<short>("Status")
-                        .HasColumnType("integer")
+                        .HasColumnType("smallint")
                         .HasColumnName("status");
 
                     b.Property<Instant>("UpdatedAt")