SolarNetwork/Swarm
2
3
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

✨ Implement realtime chat

Browse Source
This commit is contained in:
LittleSheep
2025-05-25 05:51:13 +08:00
parent 59bc9edd4b
commit 9e7ba820c4
13 changed files with 3849 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
DysonNetwork.Sphere/Chat/ChatService.cs
View File

@@ -1,4 +1,5 @@
using DysonNetwork.Sphere.Account;
using DysonNetwork.Sphere.Chat.Realtime;
using DysonNetwork.Sphere.Connection;
using DysonNetwork.Sphere.Storage;
using Microsoft.EntityFrameworkCore;
@@ -6,7 +7,12 @@ using NodaTime;
namespace DysonNetwork.Sphere.Chat;
public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory scopeFactory)
public class ChatService(
AppDatabase db,
FileService fs,
IServiceScopeFactory scopeFactory,
IRealtimeService realtime
)
{
public async Task<Message> SendMessageAsync(Message message, ChatMember sender, ChatRoom room)
{
@@ -105,10 +111,10 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
.Where(m => m.AccountId == userId)
.Select(m => new { m.ChatRoomId, m.LastReadAt })
.ToListAsync();
var lastReadAt = members.ToDictionary(m => m.ChatRoomId, m => m.LastReadAt);
var roomsId = lastReadAt.Keys.ToList();
return await db.ChatMessages
.Where(m => roomsId.Contains(m.ChatRoomId))
.GroupBy(m => m.ChatRoomId)
@@ -124,7 +130,7 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
.Where(m => m.AccountId == userId)
.Select(m => m.ChatRoomId)
.ToListAsync();
var messages = await db.ChatMessages
.IgnoreQueryFilters()
.Include(m => m.Sender)
@@ -137,7 +143,7 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
m => m!.ChatRoomId,
m => m
);
return messages;
}
@@ -147,13 +153,33 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
{
RoomId = room.Id,
SenderId = sender.Id,
ProviderName = realtime.ProviderName
};
try
{
var sessionConfig = await realtime.CreateSessionAsync(room.Id, new Dictionary<string, object>
{
{ "room_id", room.Id },
{ "user_id", sender.AccountId },
});
// Store session details
call.SessionId = sessionConfig.SessionId;
call.UpstreamConfig = sessionConfig.Parameters;
}
catch (Exception ex)
{
// Log the exception but continue with call creation
throw new InvalidOperationException($"Failed to create {realtime.ProviderName} session: {ex.Message}");
}
db.ChatRealtimeCall.Add(call);
await db.SaveChangesAsync();
await SendMessageAsync(new Message
{
Type = "realtime.start",
Type = "call.start",
ChatRoomId = room.Id,
SenderId = sender.Id,
Meta = new Dictionary<string, object>
@@ -169,14 +195,34 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
{
var call = await GetCallOngoingAsync(roomId);
if (call is null) throw new InvalidOperationException("No ongoing call was not found.");
// End the realtime session if it exists
if (!string.IsNullOrEmpty(call.SessionId) && !string.IsNullOrEmpty(call.ProviderName))
{
try
{
var config = new RealtimeSessionConfig
{
SessionId = call.SessionId,
Parameters = call.UpstreamConfig
};
await realtime.EndSessionAsync(call.SessionId, config);
}
catch (Exception ex)
{
// Log the exception but continue with call ending
throw new InvalidOperationException($"Failed to end {call.ProviderName} session: {ex.Message}");
}
}
call.EndedAt = SystemClock.Instance.GetCurrentInstant();
db.ChatRealtimeCall.Update(call);
await db.SaveChangesAsync();
await SendMessageAsync(new Message
{
Type = "realtime.ended",
Type = "call.ended",
ChatRoomId = call.RoomId,
SenderId = call.SenderId,
Meta = new Dictionary<string, object>
@@ -185,7 +231,7 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
}
}, call.Sender, call.Room);
}
public async Task<RealtimeCall?> GetCallOngoingAsync(Guid roomId)
{
return await db.ChatRealtimeCall
@@ -195,7 +241,7 @@ public class ChatService(AppDatabase db, FileService fs, IServiceScopeFactory sc
.Include(c => c.Sender)
.FirstOrDefaultAsync();
}
public async Task<SyncResponse> GetSyncDataAsync(Guid roomId, long lastSyncTimestamp)
{
var timestamp = Instant.FromUnixTimeMilliseconds(lastSyncTimestamp);

56
DysonNetwork.Sphere/Chat/Realtime/IRealtimeService.cs Normal file
View File

@@ -0,0 +1,56 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
namespace DysonNetwork.Sphere.Chat.Realtime;
/// <summary>
/// Interface for real-time communication services (like Cloudflare, Agora, Twilio, etc.)
/// </summary>
public interface IRealtimeService
{
/// <summary>
/// Service provider name
/// </summary>
string ProviderName { get; }
/// <summary>
/// Creates a new real-time session
/// </summary>
/// <param name="roomId">The room identifier</param>
/// <param name="metadata">Additional metadata to associate with the session</param>
/// <returns>Session configuration data</returns>
Task<RealtimeSessionConfig> CreateSessionAsync(Guid roomId, Dictionary<string, object> metadata);
/// <summary>
/// Ends an existing real-time session
/// </summary>
/// <param name="sessionId">The session identifier</param>
/// <param name="config">The session configuration</param>
Task EndSessionAsync(string sessionId, RealtimeSessionConfig config);
/// <summary>
/// Gets a token for user to join the session
/// </summary>
/// <param name="account">The user identifier</param>
/// <param name="sessionId">The session identifier</param>
/// <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>
/// Common configuration object for real-time sessions
/// </summary>
public class RealtimeSessionConfig
{
/// <summary>
/// Service-specific session identifier
/// </summary>
public string SessionId { get; set; } = null!;
/// <summary>
/// Additional provider-specific configuration parameters
/// </summary>
public Dictionary<string, object> Parameters { get; set; } = new();
}

111
DysonNetwork.Sphere/Chat/Realtime/LivekitService.cs Normal file
View File

@@ -0,0 +1,111 @@
using Livekit.Server.Sdk.Dotnet;
namespace DysonNetwork.Sphere.Chat.Realtime;
/// <summary>
/// LiveKit implementation of the real-time communication service
/// </summary>
public class LivekitRealtimeService : IRealtimeService
{
private readonly ILogger<LivekitRealtimeService> _logger;
private readonly RoomServiceClient _roomService;
private readonly AccessToken _accessToken;
public LivekitRealtimeService(IConfiguration configuration, ILogger<LivekitRealtimeService> logger)
{
_logger = logger;
// Get LiveKit configuration from appsettings
var host = configuration["RealtimeChat:Endpoint"] ??
throw new ArgumentNullException("Endpoint configuration is required");
var apiKey = configuration["RealtimeChat:ApiKey"] ??
throw new ArgumentNullException("ApiKey configuration is required");
var apiSecret = configuration["RealtimeChat:ApiSecret"] ??
throw new ArgumentNullException("ApiSecret configuration is required");
_roomService = new RoomServiceClient(host, apiKey, apiSecret);
_accessToken = new AccessToken(apiKey, apiSecret);
}
/// <inheritdoc />
public string ProviderName => "LiveKit";
/// <inheritdoc />
public async Task<RealtimeSessionConfig> CreateSessionAsync(Guid roomId, Dictionary<string, object> metadata)
{
try
{
var roomName = $"Call_{roomId.ToString().Replace("-", "")}";
// Convert metadata to a string dictionary for LiveKit
var roomMetadata = new Dictionary<string, string>();
foreach (var item in metadata)
{
roomMetadata[item.Key] = item.Value?.ToString() ?? string.Empty;
}
// Create room in LiveKit
var room = await _roomService.CreateRoom(new CreateRoomRequest
{
Name = roomName,
EmptyTimeout = 300, // 5 minutes
Metadata = System.Text.Json.JsonSerializer.Serialize(roomMetadata)
});
// Return session config
return new RealtimeSessionConfig
{
SessionId = room.Name,
Parameters = new Dictionary<string, object>
{
{ "sid", room.Sid },
{ "emptyTimeout", room.EmptyTimeout },
{ "creationTime", room.CreationTime }
}
};
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to create LiveKit room for roomId: {RoomId}", roomId);
throw;
}
}
/// <inheritdoc />
public async Task EndSessionAsync(string sessionId, RealtimeSessionConfig config)
{
try
{
// Delete the room in LiveKit
await _roomService.DeleteRoom(new DeleteRoomRequest
{
Room = sessionId
});
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to end LiveKit session: {SessionId}", sessionId);
throw;
}
}
/// <inheritdoc />
public string GetUserToken(Account.Account account, string sessionId, bool isAdmin = false)
{
var token = _accessToken.WithIdentity(account.Name)
.WithName(account.Nick)
.WithGrants(new VideoGrants
{
RoomJoin = true,
CanPublish = true,
CanPublishData = true,
CanSubscribe = true,
CanSubscribeMetrics = true,
RoomAdmin = isAdmin,
Room = sessionId
})
.WithAttributes(new Dictionary<string, string> { { "account_id", account.Id.ToString() } })
.WithTtl(TimeSpan.FromHours(1));
return token.ToJwt();
}
}

39
DysonNetwork.Sphere/Chat/RealtimeCall.cs
View File

@@ -1,3 +1,9 @@
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations.Schema;
using System.Text.Json;
using System.Text.Json.Serialization;
using DysonNetwork.Sphere.Chat.Realtime;
using NodaTime;
namespace DysonNetwork.Sphere.Chat;
@@ -5,11 +11,40 @@ namespace DysonNetwork.Sphere.Chat;
public class RealtimeCall : ModelBase
{
public Guid Id { get; set; } = Guid.NewGuid();
public string? Title { get; set; }
public Instant? EndedAt { get; set; }
public Guid SenderId { get; set; }
public ChatMember Sender { get; set; } = null!;
public Guid RoomId { get; set; }
public ChatRoom Room { get; set; } = null!;
/// <summary>
/// Provider name (e.g., "cloudflare", "agora", "twilio")
/// </summary>
public string? ProviderName { get; set; }
/// <summary>
/// Service provider's session identifier
/// </summary>
public string? SessionId { get; set; }
/// <summary>
/// JSONB column containing provider-specific configuration
/// </summary>
[Column(name: "upstream", TypeName = "jsonb")]
public string? UpstreamConfigJson { get; set; }
/// <summary>
/// Deserialized upstream configuration
/// </summary>
[NotMapped]
public Dictionary<string, object> UpstreamConfig
{
get => string.IsNullOrEmpty(UpstreamConfigJson)
? new Dictionary<string, object>()
: JsonSerializer.Deserialize<Dictionary<string, object>>(UpstreamConfigJson) ?? new Dictionary<string, object>();
set => UpstreamConfigJson = value.Count > 0
? JsonSerializer.Serialize(value)
: null;
}
}

83
DysonNetwork.Sphere/Chat/RealtimeCallController.cs
View File

@@ -1,3 +1,4 @@
using DysonNetwork.Sphere.Chat.Realtime;
using Microsoft.AspNetCore.Authorization;
using Microsoft.AspNetCore.Mvc;
using Microsoft.EntityFrameworkCore;
@@ -6,16 +7,60 @@ namespace DysonNetwork.Sphere.Chat;
public class RealtimeChatConfiguration
{
public string Provider { get; set; } = null!;
public string Endpoint { get; set; } = null!;
}
[ApiController]
[Route("/chat/realtime")]
public class RealtimeCallController(IConfiguration configuration, AppDatabase db, ChatService cs) : ControllerBase
public class RealtimeCallController(IConfiguration configuration, AppDatabase db, ChatService cs, IRealtimeService realtime) : ControllerBase
{
private readonly RealtimeChatConfiguration _config =
configuration.GetSection("RealtimeChat").Get<RealtimeChatConfiguration>()!;
[HttpGet("{roomId:guid}/join")]
[Authorize]
public async Task<IActionResult> 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)
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 ??
throw new InvalidOperationException("LiveKit endpoint configuration is missing");
// Create response model
var response = new JoinCallResponse
{
Provider = realtime.ProviderName,
Endpoint = endpoint,
Token = userToken,
CallId = ongoingCall.Id,
RoomName = ongoingCall.SessionId,
IsAdmin = isAdmin
};
return Ok(response);
}
[HttpPost("{roomId:guid}")]
[Authorize]
public async Task<IActionResult> StartCall(Guid roomId)
@@ -57,4 +102,38 @@ public class RealtimeCallController(IConfiguration configuration, AppDatabase db
return BadRequest(exception.Message);
}
}
}
// Response model for joining a call
public class JoinCallResponse
{
/// <summary>
/// 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; }
}