feat: enable config to set channel to private

Realtime/Broadcast/BroadcastOptions.cs

@@ -3,7 +3,8 @@
 namespace Supabase.Realtime.Broadcast;
 
 /// <summary>
-/// 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.
 /// </summary>
 public class BroadcastOptions
 {
@@ -19,6 +20,32 @@ public class BroadcastOptions
     [JsonProperty("ack")]
     public bool BroadcastAck { get; set; } = false;
 
+    /// <summary>
+    /// replay option instructs server to replay broadcast messages
+    /// </summary>
+    [JsonProperty("replay", NullValueHandling = NullValueHandling.Ignore)]
+    public ReplayOptions? Replay { get; set; }
+
+    /// <summary>
+    /// Options for replaying events in broadcast configurations.
+    /// </summary>
+    public class ReplayOptions
+    {
+        /// <summary>
+        /// Specifies the starting point in time, in milliseconds since the Unix epoch,
+        /// from which events should be replayed in the broadcast configuration.
+        /// </summary>
+        [JsonProperty("since")]
+        public long Since { get; set; }
+
+        /// <summary>
+        /// 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.
+        /// </summary>
+        [JsonProperty("limit", NullValueHandling = NullValueHandling.Ignore)]
+        public int? Limit { get; set; }
+    }
+
     /// <summary>
     /// Initializes broadcast options
     /// </summary>
@@ -29,4 +56,4 @@ public BroadcastOptions(bool broadcastSelf = false, bool broadcastAck = false)
         BroadcastSelf = broadcastSelf;
         BroadcastAck = broadcastAck;
     }
-}
+}

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;
 
 /// <summary>
-/// Channel Options
+/// Represents configuration options for a Realtime channel.
 /// </summary>
+/// <remarks>
+/// This class contains all the necessary configuration options for establishing and maintaining
+/// a Realtime channel connection, including authentication, parameters, and serialization settings.
+/// </remarks>
 public class ChannelOptions
 {
     /// <summary>
@@ -30,15 +34,61 @@ public class ChannelOptions
     public JsonSerializerSettings SerializerSettings { get; }
 
     /// <summary>
-    /// The Channel Options (typically only called from within the <see cref="Client"/>)
+    /// Whether the channel is private, i.e. authorized against the server's Row Level Security
+    /// policies. Private channels are required for broadcast replay.
     /// </summary>
-    /// <param name="clientOptions"></param>
-    /// <param name="retrieveAccessToken"></param>
-    /// <param name="serializerSettings"></param>
-    public ChannelOptions(ClientOptions clientOptions, Func<string?> retrieveAccessToken, JsonSerializerSettings serializerSettings)
+    public bool IsPrivate { get; }
+
+    /// <summary>
+    /// The Channel Options (typically only called from within the <see cref="Client"/>). Creates
+    /// options for a public channel; use <see cref="Private"/> for a private one.
+    /// </summary>
+    /// <param name="clientOptions">The client configuration options.</param>
+    /// <param name="retrieveAccessToken">A function that returns the current access token.</param>
+    /// <param name="serializerSettings">The JSON serializer settings to be used for message serialization.</param>
+    public ChannelOptions(
+        ClientOptions clientOptions,
+        Func<string?> retrieveAccessToken,
+        JsonSerializerSettings serializerSettings
+    ) : this(clientOptions, retrieveAccessToken, serializerSettings, false)
+    {
+    }
+
+    private ChannelOptions(
+        ClientOptions clientOptions,
+        Func<string?> retrieveAccessToken,
+        JsonSerializerSettings serializerSettings,
+        bool isPrivate
+    )
     {
         ClientOptions = clientOptions;
         SerializerSettings = serializerSettings;
         RetrieveAccessToken = retrieveAccessToken;
+        IsPrivate = isPrivate;
     }
-}
+
+    /// <summary>
+    /// Creates options for a public channel.
+    /// </summary>
+    /// <param name="clientOptions">The client configuration options.</param>
+    /// <param name="retrieveAccessToken">A function that returns the current access token.</param>
+    /// <param name="serializerSettings">The JSON serializer settings to be used for message serialization.</param>
+    public static ChannelOptions Public(
+        ClientOptions clientOptions,
+        Func<string?> retrieveAccessToken,
+        JsonSerializerSettings serializerSettings
+    ) => new(clientOptions, retrieveAccessToken, serializerSettings, false);
+
+    /// <summary>
+    /// Creates options for a private channel, i.e. one authorized against the server's Row Level
+    /// Security policies. Required for broadcast replay.
+    /// </summary>
+    /// <param name="clientOptions">The client configuration options.</param>
+    /// <param name="retrieveAccessToken">A function that returns the current access token.</param>
+    /// <param name="serializerSettings">The JSON serializer settings to be used for message serialization.</param>
+    public static ChannelOptions Private(
+        ClientOptions clientOptions,
+        Func<string?> retrieveAccessToken,
+        JsonSerializerSettings serializerSettings
+    ) => new(clientOptions, retrieveAccessToken, serializerSettings, true);
+}

Realtime/Channel/JoinPush.cs

@@ -11,16 +11,23 @@ internal class JoinPush
 	[JsonProperty("config")]
 	public JoinPushConfig Config { get; private set; }
 
-	public JoinPush(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List<PostgresChangesOptions>? postgresChangesOptions = null)
+	private JoinPush(BroadcastOptions? broadcastOptions, PresenceOptions? presenceOptions, List<PostgresChangesOptions>? postgresChangesOptions, bool isPrivate)
 	{
 		Config = new JoinPushConfig
 		{
 			Broadcast = broadcastOptions,
 			Presence = presenceOptions,
-			PostgresChanges = postgresChangesOptions ?? new List<PostgresChangesOptions>()
+			PostgresChanges = postgresChangesOptions ?? new List<PostgresChangesOptions>(),
+			IsPrivate = isPrivate
 		};
 	}
 
+	public static JoinPush ForPublicChannel(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List<PostgresChangesOptions>? postgresChangesOptions = null)
+		=> new(broadcastOptions, presenceOptions, postgresChangesOptions, isPrivate: false);
+
+	public static JoinPush ForPrivateChannel(BroadcastOptions? broadcastOptions = null, PresenceOptions? presenceOptions = null, List<PostgresChangesOptions>? postgresChangesOptions = null)
+		=> new(broadcastOptions, presenceOptions, postgresChangesOptions, isPrivate: true);
+
 	internal class JoinPushConfig
 	{
 		[JsonProperty("broadcast", NullValueHandling = NullValueHandling.Ignore)]
@@ -31,5 +38,8 @@ internal class JoinPushConfig
 
 		[JsonProperty("postgres_changes", NullValueHandling = NullValueHandling.Ignore)]
 		public List<PostgresChangesOptions> PostgresChanges { get; set; } = new List<PostgresChangesOptions> { };
+
+		[JsonProperty("private", NullValueHandling = NullValueHandling.Ignore)]
+		public bool? IsPrivate { get; set; }
 	}
 }

Realtime/Client.cs

@@ -343,7 +343,17 @@ public void SetAuth(string jwt)
     /// <param name="channelName">The name of the Channel to join (totally arbitrary)</param>
     /// <returns></returns>
     /// <exception cref="Exception"></exception>
-    public RealtimeChannel Channel(string channelName)
+    public RealtimeChannel Channel(string channelName) =>
+        Channel(channelName, ChannelOptions.Public(Options, () => AccessToken, SerializerSettings));
+
+    /// <summary>
+    /// Adds a RealtimeChannel subscription with custom options - if a subscription exists with the same signature, the existing subscription will be returned.
+    /// </summary>
+    /// <param name="channelName">The name of the Channel to join</param>
+    /// <param name="options">Custom channel options for configuring the subscription</param>
+    /// <returns>A RealtimeChannel instance representing the subscription</returns>
+    /// <exception cref="Exception">Thrown when Socket is null, indicating Connect() was not called</exception>
+    public RealtimeChannel Channel(string channelName, ChannelOptions options)
     {
         var topic = $"realtime:{channelName}";
 
@@ -353,8 +363,7 @@ public RealtimeChannel Channel(string channelName)
         if (Socket == null)
             throw new Exception("Socket must exist, was `Connect` called?");
 
-        var subscription = new RealtimeChannel(Socket!, topic,
-            new ChannelOptions(Options, () => AccessToken, SerializerSettings));
+        var subscription = new RealtimeChannel(Socket!, topic, options);
         _subscriptions.Add(topic, subscription);
 
         return subscription;
@@ -383,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);

Realtime/Interfaces/IRealtimeChannel.cs

@@ -215,6 +215,16 @@ public interface IRealtimeChannel
     RealtimeBroadcast<TBroadcastResponse> Register<TBroadcastResponse>(bool broadcastSelf = false,
         bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast;
 
+
+    /// <summary>
+    /// Registers the channel with the specified configuration options.
+    /// </summary>
+    /// <param name="options">The configuration options for the broadcast registration.</param>
+    /// <typeparam name="TBroadcastResponse">The type of the broadcast response, which must inherit from <see cref="BaseBroadcast"/>.</typeparam>
+    /// <returns>A <see cref="RealtimeBroadcast{TBroadcastResponse}"/> instance for managing the broadcast.</returns>
+    public RealtimeBroadcast<TBroadcastResponse> Register<TBroadcastResponse>(BroadcastOptions options)
+        where TBroadcastResponse : BaseBroadcast;
+
     /// <summary>
     /// Register presence options, must be called to use <see cref="IRealtimePresence"/>, and prior to <see cref="Subscribe"/>
     /// </summary>

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<TSocket, TChannel>: IGettableHeaders
     /// <returns></returns>
     TChannel Channel(string channelName);
 
+    /// <summary>
+    /// Adds a RealtimeChannel subscription with custom options - if a subscription exists with the same signature, the existing subscription will be returned.
+    /// </summary>
+    /// <param name="channelName">The name of the Channel to join</param>
+    /// <param name="options">Custom channel options for configuring the subscription</param>
+    /// <returns>A RealtimeChannel instance representing the subscription</returns>
+    /// <exception cref="Exception">Thrown when Socket is null, indicating Connect() was not called</exception>
+    TChannel Channel(string channelName, ChannelOptions options);
+
     /// <summary>
     /// Shorthand initialization of a channel with postgres_changes options already set. 
     /// </summary>

Realtime/Models/BaseBroadcast.cs

@@ -32,4 +32,30 @@ public class BaseBroadcast
 	/// </summary>
 	[JsonProperty("payload")]
 	public Dictionary<string, object>? Payload { get; set; }
+
+	/// <summary>
+	/// Additional metadata associated with a broadcast event. Populated by the server when a
+	/// message is replayed from history on a private channel; otherwise absent.
+	/// </summary>
+	[JsonProperty("meta", NullValueHandling = NullValueHandling.Ignore)]
+	public BroadcastMeta? Meta { get; set; }
+}
+
+/// <summary>
+/// Server-supplied metadata attached to a broadcast event, present when the message was replayed
+/// from history on a private channel.
+/// </summary>
+public class BroadcastMeta
+{
+	/// <summary>
+	/// The unique identifier the server assigned to the broadcast message.
+	/// </summary>
+	[JsonProperty("id")]
+	public string? Id { get; set; }
+
+	/// <summary>
+	/// Whether this event was replayed from history rather than received live.
+	/// </summary>
+	[JsonProperty("replayed")]
+	public bool Replayed { get; set; }
 }

Realtime/RealtimeChannel.cs

@@ -201,13 +201,27 @@ private void HandleSocketStateChanged(IRealtimeSocket _, SocketState state)
     /// <returns></returns>
     /// <exception cref="InvalidOperationException"></exception>
     public RealtimeBroadcast<TBroadcastResponse> Register<TBroadcastResponse>(bool broadcastSelf = false,
-        bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast
+        bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast =>
+        Register<TBroadcastResponse>(new BroadcastOptions(broadcastSelf, broadcastAck));
+
+    /// <summary>
+    /// Registers the channel for broadcast with the specified options.
+    /// </summary>
+    /// <typeparam name="TBroadcastResponse">The type of the broadcast response, which must inherit from <see cref="BaseBroadcast"/>.</typeparam>
+    /// <param name="options">The broadcast options to configure the channel's broadcast behavior.</param>
+    /// <returns>Returns an instance of <see cref="RealtimeBroadcast{TBroadcastResponse}"/> initialized with the specified broadcast options.</returns>
+    /// <exception cref="InvalidOperationException">Thrown if the method is called multiple times for the same channel.</exception>
+    public RealtimeBroadcast<TBroadcastResponse> Register<TBroadcastResponse>(BroadcastOptions options) 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);
+        if (!Options.IsPrivate && options.Replay != null)
+            throw new InvalidOperationException(
+                $"Broadcast replay requires a private channel, but '{Topic}' is public.");
+
+        BroadcastOptions = options;
 
         var instance =
             new RealtimeBroadcast<TBroadcastResponse>(this, BroadcastOptions, Options.SerializerSettings);
@@ -624,7 +638,9 @@ internal void Enqueue(Push push)
     /// </summary>
     /// <returns></returns>
     private Push GenerateJoinPush() => new(Socket, this, ChannelEventJoin,
-        payload: new JoinPush(BroadcastOptions, PresenceOptions, PostgresChangesOptions));
+        payload: Options.IsPrivate
+            ? Channel.JoinPush.ForPrivateChannel(BroadcastOptions, PresenceOptions, PostgresChangesOptions)
+            : Channel.JoinPush.ForPublicChannel(BroadcastOptions, PresenceOptions, PostgresChangesOptions));
 
     /// <summary>
     /// Generates an auth push.

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;
+
+/// <summary>
+/// Client-side validation of broadcast replay registration. These tests exercise the guard on
+/// <see cref="RealtimeChannel.Register{TBroadcastResponse}(BroadcastOptions)"/> without a live
+/// server: the channel is built directly against an unconnected socket, so no stack is required.
+/// </summary>
+[TestClass]
+public class ChannelBroadcastReplayTests
+{
+    [TestMethod(DisplayName = "Channel: Registering broadcast replay on a public channel throws")]
+    public void ClientCannotRegisterReplayOnPublicChannel()
+    {
+        var channel = PublicChannel();
+
+        Assert.Throws<InvalidOperationException>(
+            () => channel.Register<BroadcastExample>(WithReplay()));
+    }
+
+    [TestMethod(DisplayName = "Channel: Registering broadcast replay on a private channel is allowed")]
+    public void ClientCanRegisterReplayOnPrivateChannel()
+    {
+        var channel = PrivateChannel();
+
+        var broadcast = channel.Register<BroadcastExample>(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);
+    }
+}