From c5354c0912368d5af4e3f90e8b650b086e5383ca Mon Sep 17 00:00:00 2001 From: Diego Santo Date: Wed, 5 Nov 2025 21:23:20 -0300 Subject: [PATCH 1/3] feat: enable config to set channel to private - insert config to get replay from broadcast --- Realtime/Broadcast/BroadcastOptions.cs | 30 ++- Realtime/Channel/ChannelOptions.cs | 52 ++++- Realtime/Channel/JoinPush.cs | 8 +- Realtime/Client.cs | 23 ++ Realtime/Interfaces/IRealtimeChannel.cs | 10 + Realtime/Interfaces/IRealtimeClient.cs | 10 + Realtime/Models/BaseBroadcast.cs | 16 ++ Realtime/RealtimeChannel.cs | 29 ++- RealtimeTests/ChannelBroadcastTests.cs | 244 ++++++++++++++++++-- RealtimeTests/Helpers.cs | 21 +- supabase/migrations/20250224164421_init.sql | 33 ++- 11 files changed, 439 insertions(+), 37 deletions(-) diff --git a/Realtime/Broadcast/BroadcastOptions.cs b/Realtime/Broadcast/BroadcastOptions.cs index 1114de4..53fc358 100644 --- a/Realtime/Broadcast/BroadcastOptions.cs +++ b/Realtime/Broadcast/BroadcastOptions.cs @@ -3,7 +3,7 @@ namespace Supabase.Realtime.Broadcast; /// -/// Options +/// Options /// public class BroadcastOptions { @@ -19,6 +19,32 @@ public class BroadcastOptions [JsonProperty("ack")] public bool BroadcastAck { get; set; } = false; + /// + /// replay option instructs server to replay broadcast messages + /// + [JsonProperty("replay", NullValueHandling = NullValueHandling.Ignore)] + public ReplayOptions? Replay { get; set; } + + /// + /// Options for replaying events in broadcast configurations. + /// + public class ReplayOptions + { + /// + /// Specifies the starting point in time, in milliseconds since the Unix epoch, + /// from which events should be replayed in the broadcast configuration. + /// + [JsonProperty("since")] + public long Since { get; set; } + + /// + /// Specifies the maximum number of events to be replayed during broadcast. + /// When set to null, there is no limit to the number of events replayed. + /// + [JsonProperty("limit", NullValueHandling = NullValueHandling.Ignore)] + public int? Limit { get; set; } + } + /// /// Initializes broadcast options /// @@ -29,4 +55,4 @@ public BroadcastOptions(bool broadcastSelf = false, bool broadcastAck = false) BroadcastSelf = broadcastSelf; BroadcastAck = broadcastAck; } -} \ No newline at end of file +} diff --git a/Realtime/Channel/ChannelOptions.cs b/Realtime/Channel/ChannelOptions.cs index 48868c1..f621bdb 100644 --- a/Realtime/Channel/ChannelOptions.cs +++ b/Realtime/Channel/ChannelOptions.cs @@ -1,12 +1,16 @@ -using Newtonsoft.Json; -using System; +using System; using System.Collections.Generic; +using Newtonsoft.Json; namespace Supabase.Realtime.Channel; /// -/// Channel Options +/// Represents configuration options for a Realtime channel. /// +/// +/// This class contains all the necessary configuration options for establishing and maintaining +/// a Realtime channel connection, including authentication, parameters, and serialization settings. +/// public class ChannelOptions { /// @@ -29,16 +33,48 @@ public class ChannelOptions /// public JsonSerializerSettings SerializerSettings { get; } + /// + /// Gets a value indicating whether the channel is private. + /// + /// + /// true if the channel is private; otherwise, false. + /// + public bool IsPrivate { get; } = false; + + /// + /// The Channel Options (typically only called from within the ) + /// + /// The client configuration options. + /// A function that returns the current access token. + /// The JSON serializer settings to be used for message serialization. + public ChannelOptions( + ClientOptions clientOptions, + Func retrieveAccessToken, + JsonSerializerSettings serializerSettings + ) + { + ClientOptions = clientOptions; + SerializerSettings = serializerSettings; + RetrieveAccessToken = retrieveAccessToken; + } + /// /// The Channel Options (typically only called from within the ) /// - /// - /// - /// - public ChannelOptions(ClientOptions clientOptions, Func retrieveAccessToken, JsonSerializerSettings serializerSettings) + /// The client configuration options. + /// A function that returns the current access token. + /// The JSON serializer settings to be used for message serialization. + /// A value indicating whether the channel is private. + public ChannelOptions( + ClientOptions clientOptions, + Func retrieveAccessToken, + JsonSerializerSettings serializerSettings, + bool isPrivate + ) { ClientOptions = clientOptions; SerializerSettings = serializerSettings; RetrieveAccessToken = retrieveAccessToken; + IsPrivate = isPrivate; } -} \ No newline at end of file +} diff --git a/Realtime/Channel/JoinPush.cs b/Realtime/Channel/JoinPush.cs index d24d8a3..7401790 100644 --- a/Realtime/Channel/JoinPush.cs +++ b/Realtime/Channel/JoinPush.cs @@ -11,13 +11,14 @@ internal class JoinPush [JsonProperty("config")] public JoinPushConfig Config { get; private set; } - public JoinPush(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List? postgresChangesOptions = null) + public JoinPush(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List? postgresChangesOptions = null, bool? isPrivate = null) { Config = new JoinPushConfig { Broadcast = broadcastOptions, Presence = presenceOptions, - PostgresChanges = postgresChangesOptions ?? new List() + PostgresChanges = postgresChangesOptions ?? new List(), + IsPrivate = isPrivate }; } @@ -31,5 +32,8 @@ internal class JoinPushConfig [JsonProperty("postgres_changes", NullValueHandling = NullValueHandling.Ignore)] public List PostgresChanges { get; set; } = new List { }; + + [JsonProperty("private", NullValueHandling = NullValueHandling.Ignore)] + public bool? IsPrivate { get; set; } } } \ No newline at end of file diff --git a/Realtime/Client.cs b/Realtime/Client.cs index 1e54549..ca9c6ed 100644 --- a/Realtime/Client.cs +++ b/Realtime/Client.cs @@ -360,6 +360,29 @@ public RealtimeChannel Channel(string channelName) return subscription; } + /// + /// Adds a RealtimeChannel subscription with custom options - if a subscription exists with the same signature, the existing subscription will be returned. + /// + /// The name of the Channel to join + /// Custom channel options for configuring the subscription + /// A RealtimeChannel instance representing the subscription + /// Thrown when Socket is null, indicating Connect() was not called + public RealtimeChannel Channel(string channelName, ChannelOptions options) + { + var topic = $"realtime:{channelName}"; + + if (_subscriptions.TryGetValue(topic, out var channel)) + return channel; + + if (Socket == null) + throw new Exception("Socket must exist, was `Connect` called?"); + + var subs = new RealtimeChannel(Socket!, topic, options); + _subscriptions.Add(topic, subs); + + return subs; + } + /// /// Adds a RealtimeChannel subscription - if a subscription exists with the same signature, the existing subscription will be returned. /// diff --git a/Realtime/Interfaces/IRealtimeChannel.cs b/Realtime/Interfaces/IRealtimeChannel.cs index f6d5513..896e70a 100644 --- a/Realtime/Interfaces/IRealtimeChannel.cs +++ b/Realtime/Interfaces/IRealtimeChannel.cs @@ -215,6 +215,16 @@ public interface IRealtimeChannel RealtimeBroadcast Register(bool broadcastSelf = false, bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast; + + /// + /// Registers the channel with the specified configuration options. + /// + /// The configuration options for the broadcast registration. + /// The type of the broadcast response, which must inherit from . + /// A instance for managing the broadcast. + public RealtimeBroadcast Register(BroadcastOptions options) + where TBroadcastResponse : BaseBroadcast; + /// /// Register presence options, must be called to use , and prior to /// diff --git a/Realtime/Interfaces/IRealtimeClient.cs b/Realtime/Interfaces/IRealtimeClient.cs index a8f9cf7..c78cde5 100644 --- a/Realtime/Interfaces/IRealtimeClient.cs +++ b/Realtime/Interfaces/IRealtimeClient.cs @@ -5,6 +5,7 @@ using System.Net.WebSockets; using System.Threading.Tasks; using Supabase.Core.Interfaces; +using Supabase.Realtime.Channel; using Supabase.Realtime.Exceptions; using static Supabase.Realtime.Constants; @@ -85,6 +86,15 @@ public interface IRealtimeClient: IGettableHeaders /// TChannel Channel(string channelName); + /// + /// Adds a RealtimeChannel subscription with custom options - if a subscription exists with the same signature, the existing subscription will be returned. + /// + /// The name of the Channel to join + /// Custom channel options for configuring the subscription + /// A RealtimeChannel instance representing the subscription + /// Thrown when Socket is null, indicating Connect() was not called + TChannel Channel(string channelName, ChannelOptions options); + /// /// Shorthand initialization of a channel with postgres_changes options already set. /// diff --git a/Realtime/Models/BaseBroadcast.cs b/Realtime/Models/BaseBroadcast.cs index 62b0013..ed24cad 100644 --- a/Realtime/Models/BaseBroadcast.cs +++ b/Realtime/Models/BaseBroadcast.cs @@ -32,4 +32,20 @@ public class BaseBroadcast /// [JsonProperty("payload")] public Dictionary? Payload { get; set; } + + /// + /// Additional metadata associated with a broadcast event. + /// + [JsonProperty("meta", NullValueHandling = NullValueHandling.Ignore)] + public Meta? Meta { get; set; } +} + + +public class Meta +{ + [JsonProperty("id")] + public string Id { get; set; } + + [JsonProperty("replayed")] + public bool Replayed { get; set; } } \ No newline at end of file diff --git a/Realtime/RealtimeChannel.cs b/Realtime/RealtimeChannel.cs index c38ea8f..b981667 100644 --- a/Realtime/RealtimeChannel.cs +++ b/Realtime/RealtimeChannel.cs @@ -218,6 +218,33 @@ public RealtimeBroadcast Register(bool b return instance; } + /// + /// Registers the channel for broadcast with the specified options. + /// + /// The type of the broadcast response, which must inherit from . + /// The broadcast options to configure the channel's broadcast behavior. + /// Returns an instance of initialized with the specified broadcast options. + /// Thrown if the method is called multiple times for the same channel. + public RealtimeBroadcast Register(BroadcastOptions options) where TBroadcastResponse : BaseBroadcast + { + if (_broadcast != null) + throw new InvalidOperationException( + "Register can only be called with broadcast options for a channel once."); + + if (!Options.IsPrivate && options.Replay != null) + throw new InvalidOperationException($"tried to use replay on public channel '{Topic}'. It must be a private channel."); + + BroadcastOptions = options; + + var instance = + new RealtimeBroadcast(this, BroadcastOptions, Options.SerializerSettings); + _broadcast = instance; + + BroadcastHandler = (_, response) => _broadcast.TriggerReceived(response); + + return instance; + } + /// /// Registers a instance - allowing presence responses to be parsed and state to be tracked. /// @@ -592,7 +619,7 @@ internal void Enqueue(Push push) /// /// private Push GenerateJoinPush() => new(Socket, this, ChannelEventJoin, - payload: new JoinPush(BroadcastOptions, PresenceOptions, PostgresChangesOptions)); + payload: new JoinPush(BroadcastOptions, PresenceOptions, PostgresChangesOptions, Options.IsPrivate)); /// /// Generates an auth push. diff --git a/RealtimeTests/ChannelBroadcastTests.cs b/RealtimeTests/ChannelBroadcastTests.cs index aa8b3b6..8535cc2 100644 --- a/RealtimeTests/ChannelBroadcastTests.cs +++ b/RealtimeTests/ChannelBroadcastTests.cs @@ -1,10 +1,13 @@ using System; +using System.Collections.Generic; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using Newtonsoft.Json; -using Supabase.Postgrest.Interfaces; using RealtimeTests.Models; +using Supabase.Postgrest.Interfaces; using Supabase.Realtime; +using Supabase.Realtime.Broadcast; +using Supabase.Realtime.Channel; using Supabase.Realtime.Interfaces; using Supabase.Realtime.Models; using Supabase.Realtime.PostgresChanges; @@ -14,7 +17,8 @@ namespace RealtimeTests; public class BroadcastExample : BaseBroadcast { - [JsonProperty("userId")] public string? UserId { get; set; } + [JsonProperty("userId")] + public string? UserId { get; set; } } [TestClass] @@ -48,23 +52,27 @@ public async Task ClientCanListenForBroadcast() var channel1 = _socketClient!.Channel("online-users"); var broadcast1 = channel1.Register(true, true); - broadcast1.AddBroadcastEventHandler((_, _) => - { - var broadcast = broadcast1.Current(); - if (broadcast?.UserId != guid1 && broadcast?.Event == "user") - tsc.TrySetResult(true); - }); + broadcast1.AddBroadcastEventHandler( + (_, _) => + { + var broadcast = broadcast1.Current(); + if (broadcast?.UserId != guid1 && broadcast?.Event == "user") + tsc.TrySetResult(true); + } + ); var client2 = Helpers.SocketClient(); await client2.ConnectAsync(); var channel2 = client2.Channel("online-users"); var broadcast2 = channel2.Register(true, true); - broadcast2.AddBroadcastEventHandler((_, _) => - { - var broadcast = broadcast2.Current(); - if (broadcast?.UserId != guid2 && broadcast?.Event == "user") - tsc2.TrySetResult(true); - }); + broadcast2.AddBroadcastEventHandler( + (_, _) => + { + var broadcast = broadcast2.Current(); + if (broadcast?.UserId != guid2 && broadcast?.Event == "user") + tsc2.TrySetResult(true); + } + ); await channel1.Subscribe(); await channel2.Subscribe(); @@ -75,6 +83,195 @@ public async Task ClientCanListenForBroadcast() await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); } + [TestMethod("Channel: Can listen for broadcast")] + public async Task ClientCanListenForBroadcastPrivate() + { + var tsc = new TaskCompletionSource(); + var tsc2 = new TaskCompletionSource(); + + var guid1 = Guid.NewGuid().ToString(); + var guid2 = Guid.NewGuid().ToString(); + + var client1 = Helpers.PrivateSocketClient(); + await client1.ConnectAsync(); + var options1 = new ChannelOptions( + client1.Options, + () => Helpers.ApiKey, + new JsonSerializerSettings(), + true + ); + var channel1 = client1.Channel("online-users", options1); + var broadcast1 = channel1.Register(true, true); + broadcast1.AddBroadcastEventHandler( + (_, _) => + { + var broadcast = broadcast1.Current(); + if (broadcast?.UserId != guid1 && broadcast?.Event == "user") + tsc.TrySetResult(true); + } + ); + + var client2 = Helpers.PrivateSocketClient(); + await client2.ConnectAsync(); + var options2 = new ChannelOptions( + _socketClient!.Options, + () => Helpers.ApiKey, + new JsonSerializerSettings(), + true + ); + var channel2 = client2.Channel("online-users", options2); + var broadcast2 = channel2.Register(true, true); + broadcast2.AddBroadcastEventHandler( + (_, _) => + { + var broadcast = broadcast2.Current(); + if (broadcast?.UserId != guid2 && broadcast?.Event == "user") + tsc2.TrySetResult(true); + } + ); + + await channel1.Subscribe(); + await channel2.Subscribe(); + + await broadcast1.Send("user", new BroadcastExample { UserId = guid1 }); + await broadcast2.Send("user", new BroadcastExample { UserId = guid2 }); + + await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); + } + + [TestMethod("Channel: Cannot listen for private broadcast")] + public async Task ClientCannotListenForBroadcastPrivate() + { + var tsc = new TaskCompletionSource(); + var tsc2 = new TaskCompletionSource(); + + var guid1 = Guid.NewGuid().ToString(); + var guid2 = Guid.NewGuid().ToString(); + + var channel1 = _socketClient!.Channel("online-users"); + var broadcast1 = channel1.Register(true, true); + broadcast1.AddBroadcastEventHandler( + (_, broadcast) => + { + if (broadcast is not BroadcastExample broad1) + { + tsc.TrySetResult(true); + return; + } + + if (broad1.UserId == guid1 && broad1.Event == "user1") + tsc.TrySetResult(true); + } + ); + + var client2 = Helpers.PrivateSocketClient(); + await client2.ConnectAsync(); + var options = new ChannelOptions( + client2.Options, + () => Helpers.ApiKey, + new JsonSerializerSettings(), + true + ); + var channel2 = client2.Channel("online-users", options); + var broadcast2 = channel2.Register(true, true); + broadcast2.AddBroadcastEventHandler( + (sender, broad) => + { + if (broad is not BroadcastExample broadcast) + { + tsc2.TrySetResult(false); + return; + } + + if (broadcast.UserId == guid2 && broadcast.Event == "user2") + tsc2.TrySetResult(true); + } + ); + + await channel1.Subscribe(); + await channel2.Subscribe(); + + await broadcast1.Send("user1", new BroadcastExample { UserId = guid1 }); + await broadcast2.Send("user2", new BroadcastExample { UserId = guid2 }); + + await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); + } + + [TestMethod("Channel: Can listen history for private broadcast")] + public async Task ClientCanListenHistoryForBroadcastPrivate() + { + var send = new Dictionary + { + { "event", "user" }, + { "topic", "online-users" }, + { "private", true } + }; + await _restClient!.Rpc("send", send); + + var tsc = new TaskCompletionSource(); + + var client1 = Helpers.PrivateSocketClient(); + await client1.ConnectAsync(); + var options1 = new ChannelOptions( + client1.Options, + () => null, + new JsonSerializerSettings(), + true + ); + var broadcastOptions = new BroadcastOptions + { + BroadcastAck = true, + BroadcastSelf = true, + Replay = new BroadcastOptions.ReplayOptions + { + Limit = 10, + Since = DateTimeOffset.UtcNow.AddDays(-3).ToUnixTimeMilliseconds() + } + }; + var channel1 = client1.Channel("online-users", options1); + var broadcast1 = channel1.Register(broadcastOptions); + broadcast1.AddBroadcastEventHandler( + (_, _) => + { + + var broadcast = broadcast1.Current(); + if (broadcast is { Event: "user", Meta.Replayed: true }) + tsc.TrySetResult(true); + } + ); + + await channel1.Subscribe(); + + await Task.WhenAll(tsc.Task); + } + + [TestMethod("Channel: Cannot listen broadcast replay on public channel")] + public async Task ClientCannotListenForBroadcastReplay() + { + var client1 = Helpers.PrivateSocketClient(); + await client1.ConnectAsync(); + var options1 = new ChannelOptions( + client1.Options, + () => null, + new JsonSerializerSettings(), + false + ); + var broadcastOptions = new BroadcastOptions + { + BroadcastAck = true, + BroadcastSelf = true, + Replay = new BroadcastOptions.ReplayOptions + { + Limit = 10, + Since = DateTimeOffset.UtcNow.AddDays(-3).ToUnixTimeMilliseconds() + } + }; + var channel1 = client1.Channel("online-users", options1); + var action = () => channel1.Register(broadcastOptions); + + Assert.ThrowsException(action); + } + [TestMethod("Channel: Payload returns a modeled response (if possible)")] public async Task ChannelPayloadReturnsModel() { @@ -82,17 +279,22 @@ public async Task ChannelPayloadReturnsModel() var channel = _socketClient!.Channel("example"); channel.Register(new PostgresChangesOptions("public", "*")); - channel.AddPostgresChangeHandler(ListenType.Inserts, (_, changes) => - { - var model = changes.Model(); - tsc.SetResult(model != null); - }); + channel.AddPostgresChangeHandler( + ListenType.Inserts, + (_, changes) => + { + var model = changes.Model(); + tsc.SetResult(model != null); + } + ); await channel.Subscribe(); - await _restClient!.Table().Insert(new Todo { UserId = 1, Details = "Client Models a response? ✅" }); + await _restClient! + .Table() + .Insert(new Todo { UserId = 1, Details = "Client Models a response? ✅" }); var check = await tsc.Task; Assert.IsTrue(check); } -} \ No newline at end of file +} diff --git a/RealtimeTests/Helpers.cs b/RealtimeTests/Helpers.cs index ffd05f8..3b16077 100644 --- a/RealtimeTests/Helpers.cs +++ b/RealtimeTests/Helpers.cs @@ -1,4 +1,5 @@ -using System.Diagnostics; +using System.Collections.Generic; +using System.Diagnostics; using Supabase.Realtime; using Supabase.Realtime.Socket; using Client = Supabase.Realtime.Client; @@ -7,7 +8,8 @@ namespace RealtimeTests; internal static class Helpers { - private const string ApiKey = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0"; + public const string ApiKeyAnon = "sb_publishable_ACJWlzQHlZjBrEguHvfOxg_3BJgxAaH"; + public const string ApiKey = "sb_secret_N7UND0UgjKTVK-Uodkm0Hg_xSvEMPvz"; private const string SocketEndpoint = "ws://127.0.0.1:54321/realtime/v1"; private const string RestEndpoint = "http://localhost:54321/rest/v1"; @@ -15,6 +17,21 @@ internal static class Helpers public static Supabase.Postgrest.Client RestClient() => new(RestEndpoint, new Supabase.Postgrest.ClientOptions()); public static Client SocketClient() + { + var client = new Client(SocketEndpoint, new ClientOptions + { + Parameters = new SocketOptionsParameters + { + ApiKey = ApiKeyAnon + } + }); + + client.AddDebugHandler((_, message, _) => Debug.WriteLine(message)); + + return client; + } + + public static Client PrivateSocketClient() { var client = new Client(SocketEndpoint, new ClientOptions { diff --git a/supabase/migrations/20250224164421_init.sql b/supabase/migrations/20250224164421_init.sql index 718e09b..08fc4a1 100644 --- a/supabase/migrations/20250224164421_init.sql +++ b/supabase/migrations/20250224164421_init.sql @@ -182,4 +182,35 @@ SELECT status from users WHERE username = name_param; $$ - LANGUAGE SQL IMMUTABLE; \ No newline at end of file + LANGUAGE SQL IMMUTABLE; + + +CREATE OR REPLACE FUNCTION public.send( + event text, + topic text, + private boolean +) +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN +BEGIN + -- Set the topic configuration +EXECUTE format('SET LOCAL realtime.topic TO %L', topic); + +-- Attempt to insert the message +INSERT INTO realtime.messages (payload, event, topic, private, extension) +VALUES (null, event, topic, private, 'broadcast'); +EXCEPTION + WHEN OTHERS THEN + -- Capture and notify the error + RAISE WARNING 'ErrorSendingBroadcastMessage: %', SQLERRM; +END; +END; +$$; + +CREATE POLICY messages_insert_all +ON realtime.messages +FOR INSERT +TO PUBLIC +WITH CHECK (true); From eb98c4b54a01874226e2a0e5c664861340c9f48d Mon Sep 17 00:00:00 2001 From: tr00d Date: Thu, 6 Aug 2026 08:23:55 +0200 Subject: [PATCH 2/3] refactor: improve private channel design, test rework --- Realtime/Broadcast/BroadcastOptions.cs | 3 +- Realtime/Channel/ChannelOptions.cs | 50 ++++++--- Realtime/Channel/JoinPush.cs | 12 +- Realtime/Client.cs | 32 ++---- Realtime/Interfaces/IRealtimeChannel.cs | 2 +- Realtime/Interfaces/IRealtimeClient.cs | 2 +- Realtime/Models/BaseBroadcast.cs | 22 +++- Realtime/RealtimeChannel.cs | 31 ++---- RealtimeTests/ChannelBroadcastReplayTests.cs | 56 ++++++++++ RealtimeTests/ChannelBroadcastTests.cs | 110 ++----------------- RealtimeTests/ClientTests.cs | 5 - RealtimeTests/Helpers.cs | 2 +- supabase/migrations/20250224164421_init.sql | 43 +++++--- 13 files changed, 173 insertions(+), 197 deletions(-) create mode 100644 RealtimeTests/ChannelBroadcastReplayTests.cs diff --git a/Realtime/Broadcast/BroadcastOptions.cs b/Realtime/Broadcast/BroadcastOptions.cs index 53fc358..448519f 100644 --- a/Realtime/Broadcast/BroadcastOptions.cs +++ b/Realtime/Broadcast/BroadcastOptions.cs @@ -3,7 +3,8 @@ namespace Supabase.Realtime.Broadcast; /// -/// Options +/// Configures broadcast behavior for a channel: whether the client receives its own messages, +/// whether the server acknowledges sends, and whether past messages are replayed from history. /// public class BroadcastOptions { diff --git a/Realtime/Channel/ChannelOptions.cs b/Realtime/Channel/ChannelOptions.cs index f621bdb..6fb2b31 100644 --- a/Realtime/Channel/ChannelOptions.cs +++ b/Realtime/Channel/ChannelOptions.cs @@ -34,15 +34,14 @@ public class ChannelOptions public JsonSerializerSettings SerializerSettings { get; } /// - /// Gets a value indicating whether the channel is private. + /// Whether the channel is private, i.e. authorized against the server's Row Level Security + /// policies. Private channels are required for broadcast replay. /// - /// - /// true if the channel is private; otherwise, false. - /// - public bool IsPrivate { get; } = false; + public bool IsPrivate { get; } /// - /// The Channel Options (typically only called from within the ) + /// The Channel Options (typically only called from within the ). Creates + /// options for a public channel; use for a private one. /// /// The client configuration options. /// A function that returns the current access token. @@ -51,21 +50,11 @@ public ChannelOptions( ClientOptions clientOptions, Func retrieveAccessToken, JsonSerializerSettings serializerSettings - ) + ) : this(clientOptions, retrieveAccessToken, serializerSettings, false) { - ClientOptions = clientOptions; - SerializerSettings = serializerSettings; - RetrieveAccessToken = retrieveAccessToken; } - /// - /// The Channel Options (typically only called from within the ) - /// - /// The client configuration options. - /// A function that returns the current access token. - /// The JSON serializer settings to be used for message serialization. - /// A value indicating whether the channel is private. - public ChannelOptions( + private ChannelOptions( ClientOptions clientOptions, Func retrieveAccessToken, JsonSerializerSettings serializerSettings, @@ -77,4 +66,29 @@ bool isPrivate RetrieveAccessToken = retrieveAccessToken; IsPrivate = isPrivate; } + + /// + /// Creates options for a public channel. + /// + /// The client configuration options. + /// A function that returns the current access token. + /// The JSON serializer settings to be used for message serialization. + public static ChannelOptions Public( + ClientOptions clientOptions, + Func retrieveAccessToken, + JsonSerializerSettings serializerSettings + ) => new(clientOptions, retrieveAccessToken, serializerSettings, false); + + /// + /// Creates options for a private channel, i.e. one authorized against the server's Row Level + /// Security policies. Required for broadcast replay. + /// + /// The client configuration options. + /// A function that returns the current access token. + /// The JSON serializer settings to be used for message serialization. + public static ChannelOptions Private( + ClientOptions clientOptions, + Func retrieveAccessToken, + JsonSerializerSettings serializerSettings + ) => new(clientOptions, retrieveAccessToken, serializerSettings, true); } diff --git a/Realtime/Channel/JoinPush.cs b/Realtime/Channel/JoinPush.cs index 7401790..cc27b56 100644 --- a/Realtime/Channel/JoinPush.cs +++ b/Realtime/Channel/JoinPush.cs @@ -11,17 +11,23 @@ internal class JoinPush [JsonProperty("config")] public JoinPushConfig Config { get; private set; } - public JoinPush(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List? postgresChangesOptions = null, bool? isPrivate = null) + private JoinPush(BroadcastOptions? broadcastOptions, PresenceOptions? presenceOptions, List? postgresChangesOptions, bool isPrivate) { Config = new JoinPushConfig { Broadcast = broadcastOptions, Presence = presenceOptions, PostgresChanges = postgresChangesOptions ?? new List(), - IsPrivate = isPrivate + IsPrivate = isPrivate }; } + public static JoinPush ForPublicChannel(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List? postgresChangesOptions = null) + => new(broadcastOptions, presenceOptions, postgresChangesOptions, isPrivate: false); + + public static JoinPush ForPrivateChannel(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List? postgresChangesOptions = null) + => new(broadcastOptions, presenceOptions, postgresChangesOptions, isPrivate: true); + internal class JoinPushConfig { [JsonProperty("broadcast", NullValueHandling = NullValueHandling.Ignore)] @@ -32,7 +38,7 @@ internal class JoinPushConfig [JsonProperty("postgres_changes", NullValueHandling = NullValueHandling.Ignore)] public List PostgresChanges { get; set; } = new List { }; - + [JsonProperty("private", NullValueHandling = NullValueHandling.Ignore)] public bool? IsPrivate { get; set; } } diff --git a/Realtime/Client.cs b/Realtime/Client.cs index ca9c6ed..0a88d23 100644 --- a/Realtime/Client.cs +++ b/Realtime/Client.cs @@ -343,22 +343,8 @@ public void SetAuth(string jwt) /// The name of the Channel to join (totally arbitrary) /// /// - public RealtimeChannel Channel(string channelName) - { - var topic = $"realtime:{channelName}"; - - if (_subscriptions.TryGetValue(topic, out var channel)) - return channel; - - if (Socket == null) - throw new Exception("Socket must exist, was `Connect` called?"); - - var subscription = new RealtimeChannel(Socket!, topic, - new ChannelOptions(Options, () => AccessToken, SerializerSettings)); - _subscriptions.Add(topic, subscription); - - return subscription; - } + public RealtimeChannel Channel(string channelName) => + Channel(channelName, ChannelOptions.Public(Options, () => AccessToken, SerializerSettings)); /// /// Adds a RealtimeChannel subscription with custom options - if a subscription exists with the same signature, the existing subscription will be returned. @@ -370,17 +356,17 @@ public RealtimeChannel Channel(string channelName) public RealtimeChannel Channel(string channelName, ChannelOptions options) { var topic = $"realtime:{channelName}"; - + if (_subscriptions.TryGetValue(topic, out var channel)) return channel; - + if (Socket == null) throw new Exception("Socket must exist, was `Connect` called?"); - var subs = new RealtimeChannel(Socket!, topic, options); - _subscriptions.Add(topic, subs); - - return subs; + var subscription = new RealtimeChannel(Socket!, topic, options); + _subscriptions.Add(topic, subscription); + + return subscription; } /// @@ -406,7 +392,7 @@ public RealtimeChannel Channel(string database = "realtime", string schema = "pu var changesOptions = new PostgresChangesOptions(schema, table, filter: column != null && value != null ? $"{column}=eq.{value}" : null, parameters: parameters); - var options = new ChannelOptions(Options, () => AccessToken, SerializerSettings); + var options = ChannelOptions.Public(Options, () => AccessToken, SerializerSettings); var subscription = new RealtimeChannel(Socket!, key, options); subscription.Register(changesOptions); diff --git a/Realtime/Interfaces/IRealtimeChannel.cs b/Realtime/Interfaces/IRealtimeChannel.cs index 896e70a..d523c32 100644 --- a/Realtime/Interfaces/IRealtimeChannel.cs +++ b/Realtime/Interfaces/IRealtimeChannel.cs @@ -224,7 +224,7 @@ RealtimeBroadcast Register(bool broadcas /// A instance for managing the broadcast. public RealtimeBroadcast Register(BroadcastOptions options) where TBroadcastResponse : BaseBroadcast; - + /// /// Register presence options, must be called to use , and prior to /// diff --git a/Realtime/Interfaces/IRealtimeClient.cs b/Realtime/Interfaces/IRealtimeClient.cs index c78cde5..c96f5d2 100644 --- a/Realtime/Interfaces/IRealtimeClient.cs +++ b/Realtime/Interfaces/IRealtimeClient.cs @@ -94,7 +94,7 @@ public interface IRealtimeClient: IGettableHeaders /// A RealtimeChannel instance representing the subscription /// Thrown when Socket is null, indicating Connect() was not called TChannel Channel(string channelName, ChannelOptions options); - + /// /// Shorthand initialization of a channel with postgres_changes options already set. /// diff --git a/Realtime/Models/BaseBroadcast.cs b/Realtime/Models/BaseBroadcast.cs index ed24cad..95e8c1d 100644 --- a/Realtime/Models/BaseBroadcast.cs +++ b/Realtime/Models/BaseBroadcast.cs @@ -34,18 +34,28 @@ public class BaseBroadcast public Dictionary? Payload { get; set; } /// - /// Additional metadata associated with a broadcast event. + /// Additional metadata associated with a broadcast event. Populated by the server when a + /// message is replayed from history on a private channel; otherwise absent. /// [JsonProperty("meta", NullValueHandling = NullValueHandling.Ignore)] - public Meta? Meta { get; set; } + public BroadcastMeta? Meta { get; set; } } - -public class Meta +/// +/// Server-supplied metadata attached to a broadcast event, present when the message was replayed +/// from history on a private channel. +/// +public class BroadcastMeta { + /// + /// The unique identifier the server assigned to the broadcast message. + /// [JsonProperty("id")] - public string Id { get; set; } - + public string? Id { get; set; } + + /// + /// Whether this event was replayed from history rather than received live. + /// [JsonProperty("replayed")] public bool Replayed { get; set; } } \ No newline at end of file diff --git a/Realtime/RealtimeChannel.cs b/Realtime/RealtimeChannel.cs index b981667..b4bf8cd 100644 --- a/Realtime/RealtimeChannel.cs +++ b/Realtime/RealtimeChannel.cs @@ -201,22 +201,8 @@ private void HandleSocketStateChanged(IRealtimeSocket _, SocketState state) /// /// public RealtimeBroadcast Register(bool broadcastSelf = false, - bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast - { - if (_broadcast != null) - throw new InvalidOperationException( - "Register can only be called with broadcast options for a channel once."); - - BroadcastOptions = new BroadcastOptions(broadcastSelf, broadcastAck); - - var instance = - new RealtimeBroadcast(this, BroadcastOptions, Options.SerializerSettings); - _broadcast = instance; - - BroadcastHandler = (_, response) => _broadcast.TriggerReceived(response); - - return instance; - } + bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast => + Register(new BroadcastOptions(broadcastSelf, broadcastAck)); /// /// Registers the channel for broadcast with the specified options. @@ -232,9 +218,10 @@ public RealtimeBroadcast Register(Broadc "Register can only be called with broadcast options for a channel once."); if (!Options.IsPrivate && options.Replay != null) - throw new InvalidOperationException($"tried to use replay on public channel '{Topic}'. It must be a private channel."); - - BroadcastOptions = options; + throw new InvalidOperationException( + $"Broadcast replay requires a private channel, but '{Topic}' is public."); + + BroadcastOptions = options; var instance = new RealtimeBroadcast(this, BroadcastOptions, Options.SerializerSettings); @@ -244,7 +231,7 @@ public RealtimeBroadcast Register(Broadc return instance; } - + /// /// Registers a instance - allowing presence responses to be parsed and state to be tracked. /// @@ -619,7 +606,9 @@ internal void Enqueue(Push push) /// /// private Push GenerateJoinPush() => new(Socket, this, ChannelEventJoin, - payload: new JoinPush(BroadcastOptions, PresenceOptions, PostgresChangesOptions, Options.IsPrivate)); + payload: Options.IsPrivate + ? Channel.JoinPush.ForPrivateChannel(BroadcastOptions, PresenceOptions, PostgresChangesOptions) + : Channel.JoinPush.ForPublicChannel(BroadcastOptions, PresenceOptions, PostgresChangesOptions)); /// /// Generates an auth push. diff --git a/RealtimeTests/ChannelBroadcastReplayTests.cs b/RealtimeTests/ChannelBroadcastReplayTests.cs new file mode 100644 index 0000000..4929fd7 --- /dev/null +++ b/RealtimeTests/ChannelBroadcastReplayTests.cs @@ -0,0 +1,56 @@ +using System; +using Microsoft.VisualStudio.TestTools.UnitTesting; +using Newtonsoft.Json; +using Supabase.Realtime; +using Supabase.Realtime.Broadcast; +using Supabase.Realtime.Channel; + +namespace RealtimeTests; + +/// +/// Client-side validation of broadcast replay registration. These tests exercise the guard on +/// without a live +/// server: the channel is built directly against an unconnected socket, so no stack is required. +/// +[TestClass] +public class ChannelBroadcastReplayTests +{ + [TestMethod("Channel: Registering broadcast replay on a public channel throws")] + public void ClientCannotRegisterReplayOnPublicChannel() + { + var channel = PublicChannel(); + + Assert.ThrowsException( + () => channel.Register(WithReplay())); + } + + [TestMethod("Channel: Registering broadcast replay on a private channel is allowed")] + public void ClientCanRegisterReplayOnPrivateChannel() + { + var channel = PrivateChannel(); + + var broadcast = channel.Register(WithReplay()); + + Assert.IsNotNull(broadcast); + } + + private static BroadcastOptions WithReplay() => new() + { + Replay = new BroadcastOptions.ReplayOptions { Limit = 10, Since = 0 } + }; + + private static RealtimeChannel PublicChannel() => + Channel(ChannelOptions.Public(ClientOptions(), () => null, new JsonSerializerSettings())); + + private static RealtimeChannel PrivateChannel() => + Channel(ChannelOptions.Private(ClientOptions(), () => null, new JsonSerializerSettings())); + + private static ClientOptions ClientOptions() => new(); + + private static RealtimeChannel Channel(ChannelOptions options) + { + var socket = new RealtimeSocket("ws://localhost:54321/realtime/v1", options.ClientOptions); + + return new RealtimeChannel(socket, "realtime:online-users", options); + } +} diff --git a/RealtimeTests/ChannelBroadcastTests.cs b/RealtimeTests/ChannelBroadcastTests.cs index 8535cc2..0494bae 100644 --- a/RealtimeTests/ChannelBroadcastTests.cs +++ b/RealtimeTests/ChannelBroadcastTests.cs @@ -83,7 +83,7 @@ public async Task ClientCanListenForBroadcast() await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); } - [TestMethod("Channel: Can listen for broadcast")] + [TestMethod("Channel: Can listen for broadcast on a private channel")] public async Task ClientCanListenForBroadcastPrivate() { var tsc = new TaskCompletionSource(); @@ -94,11 +94,10 @@ public async Task ClientCanListenForBroadcastPrivate() var client1 = Helpers.PrivateSocketClient(); await client1.ConnectAsync(); - var options1 = new ChannelOptions( + var options1 = ChannelOptions.Private( client1.Options, () => Helpers.ApiKey, - new JsonSerializerSettings(), - true + new JsonSerializerSettings() ); var channel1 = client1.Channel("online-users", options1); var broadcast1 = channel1.Register(true, true); @@ -113,11 +112,10 @@ public async Task ClientCanListenForBroadcastPrivate() var client2 = Helpers.PrivateSocketClient(); await client2.ConnectAsync(); - var options2 = new ChannelOptions( - _socketClient!.Options, + var options2 = ChannelOptions.Private( + client2.Options, () => Helpers.ApiKey, - new JsonSerializerSettings(), - true + new JsonSerializerSettings() ); var channel2 = client2.Channel("online-users", options2); var broadcast2 = channel2.Register(true, true); @@ -139,64 +137,6 @@ public async Task ClientCanListenForBroadcastPrivate() await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); } - [TestMethod("Channel: Cannot listen for private broadcast")] - public async Task ClientCannotListenForBroadcastPrivate() - { - var tsc = new TaskCompletionSource(); - var tsc2 = new TaskCompletionSource(); - - var guid1 = Guid.NewGuid().ToString(); - var guid2 = Guid.NewGuid().ToString(); - - var channel1 = _socketClient!.Channel("online-users"); - var broadcast1 = channel1.Register(true, true); - broadcast1.AddBroadcastEventHandler( - (_, broadcast) => - { - if (broadcast is not BroadcastExample broad1) - { - tsc.TrySetResult(true); - return; - } - - if (broad1.UserId == guid1 && broad1.Event == "user1") - tsc.TrySetResult(true); - } - ); - - var client2 = Helpers.PrivateSocketClient(); - await client2.ConnectAsync(); - var options = new ChannelOptions( - client2.Options, - () => Helpers.ApiKey, - new JsonSerializerSettings(), - true - ); - var channel2 = client2.Channel("online-users", options); - var broadcast2 = channel2.Register(true, true); - broadcast2.AddBroadcastEventHandler( - (sender, broad) => - { - if (broad is not BroadcastExample broadcast) - { - tsc2.TrySetResult(false); - return; - } - - if (broadcast.UserId == guid2 && broadcast.Event == "user2") - tsc2.TrySetResult(true); - } - ); - - await channel1.Subscribe(); - await channel2.Subscribe(); - - await broadcast1.Send("user1", new BroadcastExample { UserId = guid1 }); - await broadcast2.Send("user2", new BroadcastExample { UserId = guid2 }); - - await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); - } - [TestMethod("Channel: Can listen history for private broadcast")] public async Task ClientCanListenHistoryForBroadcastPrivate() { @@ -207,16 +147,15 @@ public async Task ClientCanListenHistoryForBroadcastPrivate() { "private", true } }; await _restClient!.Rpc("send", send); - + var tsc = new TaskCompletionSource(); var client1 = Helpers.PrivateSocketClient(); await client1.ConnectAsync(); - var options1 = new ChannelOptions( + var options1 = ChannelOptions.Private( client1.Options, () => null, - new JsonSerializerSettings(), - true + new JsonSerializerSettings() ); var broadcastOptions = new BroadcastOptions { @@ -233,7 +172,7 @@ public async Task ClientCanListenHistoryForBroadcastPrivate() broadcast1.AddBroadcastEventHandler( (_, _) => { - + var broadcast = broadcast1.Current(); if (broadcast is { Event: "user", Meta.Replayed: true }) tsc.TrySetResult(true); @@ -241,37 +180,10 @@ public async Task ClientCanListenHistoryForBroadcastPrivate() ); await channel1.Subscribe(); - + await Task.WhenAll(tsc.Task); } - [TestMethod("Channel: Cannot listen broadcast replay on public channel")] - public async Task ClientCannotListenForBroadcastReplay() - { - var client1 = Helpers.PrivateSocketClient(); - await client1.ConnectAsync(); - var options1 = new ChannelOptions( - client1.Options, - () => null, - new JsonSerializerSettings(), - false - ); - var broadcastOptions = new BroadcastOptions - { - BroadcastAck = true, - BroadcastSelf = true, - Replay = new BroadcastOptions.ReplayOptions - { - Limit = 10, - Since = DateTimeOffset.UtcNow.AddDays(-3).ToUnixTimeMilliseconds() - } - }; - var channel1 = client1.Channel("online-users", options1); - var action = () => channel1.Register(broadcastOptions); - - Assert.ThrowsException(action); - } - [TestMethod("Channel: Payload returns a modeled response (if possible)")] public async Task ChannelPayloadReturnsModel() { diff --git a/RealtimeTests/ClientTests.cs b/RealtimeTests/ClientTests.cs index 390e173..364fdc3 100644 --- a/RealtimeTests/ClientTests.cs +++ b/RealtimeTests/ClientTests.cs @@ -1,8 +1,6 @@ using System; using System.Collections.Generic; using System.Diagnostics; -using System.Net; -using System.Net.Sockets; using System.Threading.Tasks; using Microsoft.VisualStudio.TestTools.UnitTesting; using Supabase.Realtime.Exceptions; @@ -18,9 +16,6 @@ public class ClientTests [TestInitialize] public async Task InitializeTest() { - Console.WriteLine(); - Console.WriteLine(Dns.GetHostEntryAsync(Dns.GetHostName()).GetAwaiter().GetResult().AddressList[0]); - client = Helpers.SocketClient(); client.AddDebugHandler((sender, message, exception) => Debug.WriteLine(message)); diff --git a/RealtimeTests/Helpers.cs b/RealtimeTests/Helpers.cs index 3b16077..d568300 100644 --- a/RealtimeTests/Helpers.cs +++ b/RealtimeTests/Helpers.cs @@ -30,7 +30,7 @@ public static Client SocketClient() return client; } - + public static Client PrivateSocketClient() { var client = new Client(SocketEndpoint, new ClientOptions diff --git a/supabase/migrations/20250224164421_init.sql b/supabase/migrations/20250224164421_init.sql index 08fc4a1..7c96485 100644 --- a/supabase/migrations/20250224164421_init.sql +++ b/supabase/migrations/20250224164421_init.sql @@ -183,34 +183,41 @@ from users WHERE username = name_param; $$ LANGUAGE SQL IMMUTABLE; - - + + +-- Test fixture: emulates Supabase's realtime.send() so broadcast-replay tests can seed +-- messages into realtime.messages over REST. Local stack only. CREATE OR REPLACE FUNCTION public.send( event text, topic text, private boolean ) -RETURNS void -LANGUAGE plpgsql -AS $$ -BEGIN + RETURNS void + LANGUAGE plpgsql +AS +$$ BEGIN - -- Set the topic configuration -EXECUTE format('SET LOCAL realtime.topic TO %L', topic); + BEGIN + -- Scope the insert to the requested topic. + EXECUTE format('SET LOCAL realtime.topic TO %L', topic); --- Attempt to insert the message -INSERT INTO realtime.messages (payload, event, topic, private, extension) -VALUES (null, event, topic, private, 'broadcast'); -EXCEPTION + INSERT INTO realtime.messages (payload, event, topic, private, extension) + VALUES (null, event, topic, private, 'broadcast'); + EXCEPTION WHEN OTHERS THEN - -- Capture and notify the error RAISE WARNING 'ErrorSendingBroadcastMessage: %', SQLERRM; -END; + END; END; $$; +-- Test fixture: allow the seed inserts above. Local stack only. CREATE POLICY messages_insert_all -ON realtime.messages -FOR INSERT -TO PUBLIC -WITH CHECK (true); + ON realtime.messages + FOR INSERT + TO PUBLIC + WITH CHECK (true); + +-- Test fixture: the postgres-changes tests insert into the sample tables over REST as an +-- unauthenticated (anon) client. Newer Supabase CLIs no longer grant anon/authenticated DML on +-- public tables by default, so grant it here. Local stack only. +GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO anon, authenticated, service_role; From ceb6389f64e61e5614d545cde23cc267de6a6812 Mon Sep 17 00:00:00 2001 From: tr00d Date: Thu, 6 Aug 2026 08:26:59 +0200 Subject: [PATCH 3/3] chore: deal with merge conflicts --- RealtimeTests/ChannelBroadcastReplayTests.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/RealtimeTests/ChannelBroadcastReplayTests.cs b/RealtimeTests/ChannelBroadcastReplayTests.cs index 4929fd7..58fef3b 100644 --- a/RealtimeTests/ChannelBroadcastReplayTests.cs +++ b/RealtimeTests/ChannelBroadcastReplayTests.cs @@ -15,16 +15,16 @@ namespace RealtimeTests; [TestClass] public class ChannelBroadcastReplayTests { - [TestMethod("Channel: Registering broadcast replay on a public channel throws")] + [TestMethod(DisplayName = "Channel: Registering broadcast replay on a public channel throws")] public void ClientCannotRegisterReplayOnPublicChannel() { var channel = PublicChannel(); - Assert.ThrowsException( + Assert.Throws( () => channel.Register(WithReplay())); } - [TestMethod("Channel: Registering broadcast replay on a private channel is allowed")] + [TestMethod(DisplayName = "Channel: Registering broadcast replay on a private channel is allowed")] public void ClientCanRegisterReplayOnPrivateChannel() { var channel = PrivateChannel();