Merge pull request #73 from argon-chat/feature/lockdown_and_fixes

Lockdown feature + EF Optimization+ Fixes

Argon.Server.sln.DotSettings

@@ -92,4 +92,5 @@
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Rebalance/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Splitted/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=twirp/@EntryIndexedValue">True</s:Boolean>
-	<s:Boolean x:Key="/Default/UserDictionary/Words/=unmute/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=unmute/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=YUUKI/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

src/Argon.Api/Entities/ApplicationDbContext.cs

@@ -15,15 +15,19 @@ public class ApplicationDbContext(DbContextOptions<ApplicationDbContext> options
     public DbSet<Channel>        Channels               { get; set; }
     public DbSet<ServerMember>   UsersToServerRelations { get; set; }
 
-    public DbSet<ServerMemberArchetype>      ServerMemberArchetypes      { get; set; }
-    public DbSet<Archetype>                  Archetypes                  { get; set; }
+    public DbSet<ServerMemberArchetype>       ServerMemberArchetypes       { get; set; }
+    public DbSet<Archetype>                   Archetypes                   { get; set; }
     public DbSet<ChannelEntitlementOverwrite> ChannelEntitlementOverwrites { get; set; }
 
 
     protected override void OnModelCreating(ModelBuilder modelBuilder)
     {
         modelBuilder.Entity<ServerMemberArchetype>()
-           .HasKey(x => new { x.ServerMemberId, x.ArchetypeId });
+           .HasKey(x => new
+            {
+                x.ServerMemberId,
+                x.ArchetypeId
+            });
 
         modelBuilder.Entity<ServerMemberArchetype>()
            .HasOne(x => x.ServerMember)
@@ -60,7 +64,11 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
            .HasForeignKey(c => c.ServerId);
 
         modelBuilder.Entity<Channel>()
-           .HasIndex(x => new { x.Id, x.ServerId });
+           .HasIndex(x => new
+            {
+                x.Id,
+                x.ServerId
+            });
 
         modelBuilder.Entity<ChannelEntitlementOverwrite>()
            .HasOne(cpo => cpo.Channel)
@@ -116,7 +124,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 IsMentionable = true,
                 IsHidden      = false,
                 Description   = "Default role for everyone in this server",
-                CreatedAt     = DateTime.UtcNow
+                CreatedAt     = new DateTime(2024, 11, 23, 16, 1, 14, 205, DateTimeKind.Utc).AddTicks(8411)
             }
         ]);
 
@@ -133,7 +141,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 IsMentionable = false,
                 IsHidden      = true,
                 Description   = "Default role for owner in this server",
-                CreatedAt     = DateTime.UtcNow
+                CreatedAt     = new DateTime(2024, 11, 23, 16, 1, 14, 205, DateTimeKind.Utc).AddTicks(8382)
             }
         ]);
     }

src/Argon.Api/Features/Orleanse/Streams/ClientArgonStream.cs

@@ -0,0 +1,46 @@
+namespace Argon.Api.Features.Rpc;
+
+using Orleans.Streams;
+using System.Threading.Channels;
+using Contracts;
+
+public sealed class ClientArgonStream<T> : IArgonStream<T> where T : IArgonEvent
+{
+    private StreamSubscriptionHandle<T> clientHandler { get; set; }
+    private Channel<T>                  channel       { get; } = Channel.CreateUnbounded<T>();
+    public async Task OnNextAsync(T item, StreamSequenceToken? token = null)
+        => await channel.Writer.WriteAsync(item);
+
+    public Task OnCompletedAsync()
+    {
+        channel.Writer.Complete();
+        return Task.CompletedTask;
+    }
+    public Task OnErrorAsync(Exception ex)
+    {
+        channel.Writer.Complete(ex);
+        return Task.CompletedTask;
+    }
+
+    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken ct = default)
+    {
+        while (await channel.Reader.WaitToReadAsync(ct))
+        {
+            while (channel.Reader.TryRead(out var serverEvent))
+                yield return serverEvent;
+        }
+    }
+
+    public async ValueTask DisposeAsync()
+        => await clientHandler.UnsubscribeAsync();
+
+
+    internal async ValueTask<IArgonStream<T>> BindClient(IAsyncStream<T> stream)
+    {
+        clientHandler = await stream.SubscribeAsync(this);
+        return this;
+    }
+
+    public ValueTask Fire(T ev)
+        => throw new NotImplementedException($"Client stream cannot be fire event");
+}

src/Argon.Api/Features/Orleanse/Streams/IArgonStream.cs

@@ -0,0 +1,12 @@
+namespace Argon.Api.Features.Rpc;
+
+using ActualLab.Rpc;
+using Contracts;
+using Orleans.Streams;
+
+public interface IArgonStream<T> : 
+    IAsyncObserver<T>, IAsyncEnumerable<T>, IAsyncDisposable where T : IArgonEvent
+{
+    public RpcStream<T> AsRpcStream() => new(this);
+    ValueTask           Fire(T ev);
+}

src/Argon.Api/Features/Orleanse/Streams/IStreamExtension.cs

@@ -0,0 +1,39 @@
+namespace Argon.Api.Features.Rpc;
+
+using Contracts;
+
+public interface IStreamExtension
+{
+    ValueTask<IArgonStream<IArgonEvent>> CreateClientStream(Guid primary);
+}
+
+public interface IStreamExtension<T> where T : Grain, IGrainWithGuidKey
+{
+    ValueTask<IArgonStream<IArgonEvent>> CreateServerStream();
+    ValueTask<IArgonStream<IArgonEvent>> CreateServerStreamFor(Guid targetId);
+}
+public static class ArgonStreamExtensions
+{
+    public static IStreamExtension<T> Streams<T>(this T grain) where T : Grain, IGrainWithGuidKey
+        => new StreamForGrainExtension<T>(grain);
+
+    public static IStreamExtension Streams(this IClusterClient clusterClient)
+        => new StreamForClusterClientExtension(clusterClient);
+}
+public readonly struct StreamForGrainExtension<T>(T grain) : IStreamExtension<T> where T : Grain, IGrainWithGuidKey
+{
+    public ValueTask<IArgonStream<IArgonEvent>> CreateServerStream()
+        => CreateServerStreamFor(grain.GetPrimaryKey());
+
+    public async ValueTask<IArgonStream<IArgonEvent>> CreateServerStreamFor(Guid targetId)
+        => new ServerArgonStream<IArgonEvent>(grain.GetStreamProvider(IArgonEvent.ProviderId)
+           .GetStream<IArgonEvent>(StreamId.Create(IArgonEvent.Namespace, targetId)));
+}
+
+public readonly struct StreamForClusterClientExtension(IClusterClient? client) : IStreamExtension
+{
+    public ValueTask<IArgonStream<IArgonEvent>> CreateClientStream(Guid primary)
+        => new ClientArgonStream<IArgonEvent>().BindClient(client
+           .GetStreamProvider(IArgonEvent.ProviderId)
+           .GetStream<IArgonEvent>(StreamId.Create(IArgonEvent.Namespace, primary)));
+}

src/Argon.Api/Features/Orleanse/Streams/ServerArgonStream.cs

@@ -0,0 +1,23 @@
+namespace Argon.Api.Features.Rpc;
+
+using Contracts;
+using Orleans.Streams;
+
+public sealed class ServerArgonStream<T>(IAsyncStream<IArgonEvent> stream) : IArgonStream<T> where T : IArgonEvent
+{
+    public Task OnNextAsync(T item, StreamSequenceToken? token = null)
+        => stream.OnNextAsync(item, token);
+
+    public Task OnCompletedAsync()
+        => stream.OnCompletedAsync();
+    public Task OnErrorAsync(Exception ex)
+        => stream.OnErrorAsync(ex);
+
+    public IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken ct = default)
+        => throw new NotImplementedException($"Server stream cannot create async enumerator");
+
+    public async ValueTask DisposeAsync() {} // nothing any to dispose
+
+    public async ValueTask Fire(T ev)
+        => await OnNextAsync(ev);
+}

src/Argon.Api/Grains/FusionGrain.cs

@@ -3,17 +3,21 @@ namespace Argon.Api.Grains;
 using Contracts;
 using Extensions;
 using Features.Jwt;
+using Features.Rpc;
 using Interfaces;
 using R3;
 using static DeactivationReasonCode;
 
 public class FusionGrain(IGrainFactory grainFactory) : Grain, IFusionSessionGrain
 {
     private DateTimeOffset _latestSignalTime = DateTimeOffset.UtcNow;
-    private DisposableBag disposableBag;
+    private DisposableBag  disposableBag;
+
     private Guid _userId;
     private Guid _machineId;
 
+    private IArgonStream<IArgonEvent> userStream;
+
     public async ValueTask SelfDestroy()
     {
         var servers = await grainFactory
@@ -23,7 +27,7 @@ public async ValueTask SelfDestroy()
             await grainFactory
                .GetGrain<IServerGrain>(server)
                .SetUserStatus(_userId, UserStatus.Offline);
-        _userId = default;
+        _userId    = default;
         _machineId = default;
         GrainContext.Deactivate(new(ApplicationRequested, "omae wa mou shindeiru"));
     }
@@ -38,8 +42,11 @@ public async ValueTask BeginRealtimeSession(Guid userId, Guid machineKey, UserSt
     {
         this.RegisterGrainTimer(OnValidateActiveAsync, TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(2))
            .AddTo(ref disposableBag);
-        this._userId   = userId;
+        this._userId    = userId;
         this._machineId = machineKey;
+
+        userStream      = await this.Streams().CreateServerStreamFor(_userId);
+
         await grainFactory
            .GetGrain<IUserMachineSessions>(userId)
            .IndicateLastActive(machineKey);

src/Argon.Api/Grains/Interfaces/IUserGrain.cs

@@ -6,19 +6,15 @@ namespace Argon.Api.Grains.Interfaces;
 [Alias("Argon.Api.Grains.Interfaces.IUserGrain")]
 public interface IUserGrain : IGrainWithGuidKey
 {
-    [Alias("UpdateUser")]
+    [Alias(nameof(UpdateUser))]
     Task<User> UpdateUser(UserEditInput input);
 
-    [Alias("DeleteUser")]
-    Task DeleteUser();
+    [Alias(nameof(GetMe))]
+    Task<User> GetMe();
 
-    [Alias("GetUser")]
-    Task<User> GetUser();
-
-    [Alias("GetMyServers")]
+    [Alias(nameof(GetMyServers))]
     Task<List<Server>> GetMyServers();
 
-    [Alias("GetMyServersIds")]
+    [Alias(nameof(GetMyServersIds))]
     Task<List<Guid>> GetMyServersIds();
-}
-
+}