diff --git a/Realtime/Broadcast/BroadcastOptions.cs b/Realtime/Broadcast/BroadcastOptions.cs index 1114de4..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 { @@ -19,6 +20,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 +56,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..6fb2b31 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 { /// @@ -30,15 +34,61 @@ public class ChannelOptions public JsonSerializerSettings SerializerSettings { get; } /// - /// The Channel Options (typically only called from within the ) + /// Whether the channel is private, i.e. authorized against the server's Row Level Security + /// policies. Private channels are required for broadcast replay. /// - /// - /// - /// - public ChannelOptions(ClientOptions clientOptions, Func retrieveAccessToken, JsonSerializerSettings serializerSettings) + public bool IsPrivate { get; } + + /// + /// 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. + /// The JSON serializer settings to be used for message serialization. + public ChannelOptions( + ClientOptions clientOptions, + Func retrieveAccessToken, + JsonSerializerSettings serializerSettings + ) : this(clientOptions, retrieveAccessToken, serializerSettings, false) + { + } + + private ChannelOptions( + ClientOptions clientOptions, + Func retrieveAccessToken, + JsonSerializerSettings serializerSettings, + bool isPrivate + ) { ClientOptions = clientOptions; SerializerSettings = serializerSettings; RetrieveAccessToken = retrieveAccessToken; + IsPrivate = isPrivate; } -} \ No newline at end of file + + /// + /// 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 d24d8a3..cc27b56 100644 --- a/Realtime/Channel/JoinPush.cs +++ b/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 = null) + private JoinPush(BroadcastOptions? broadcastOptions, PresenceOptions? presenceOptions, List? postgresChangesOptions, bool isPrivate) { Config = new JoinPushConfig { Broadcast = broadcastOptions, Presence = presenceOptions, - PostgresChanges = postgresChangesOptions ?? new List() + PostgresChanges = postgresChangesOptions ?? new List(), + 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)] @@ -31,5 +38,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..0a88d23 100644 --- a/Realtime/Client.cs +++ b/Realtime/Client.cs @@ -343,7 +343,17 @@ public void SetAuth(string jwt) /// The name of the Channel to join (totally arbitrary) /// /// - public RealtimeChannel Channel(string channelName) + 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. + /// + /// 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}"; @@ -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); diff --git a/Realtime/Interfaces/IRealtimeChannel.cs b/Realtime/Interfaces/IRealtimeChannel.cs index f6d5513..d523c32 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..c96f5d2 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..95e8c1d 100644 --- a/Realtime/Models/BaseBroadcast.cs +++ b/Realtime/Models/BaseBroadcast.cs @@ -32,4 +32,30 @@ public class BaseBroadcast /// [JsonProperty("payload")] public Dictionary? Payload { get; set; } + + /// + /// 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 BroadcastMeta? Meta { get; set; } +} + +/// +/// 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; } + + /// + /// 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 f02782b..3495a04 100644 --- a/Realtime/RealtimeChannel.cs +++ b/Realtime/RealtimeChannel.cs @@ -201,13 +201,27 @@ private void HandleSocketStateChanged(IRealtimeSocket _, SocketState state) /// /// public RealtimeBroadcast Register(bool broadcastSelf = false, - bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast + bool broadcastAck = false) where TBroadcastResponse : BaseBroadcast => + Register(new BroadcastOptions(broadcastSelf, broadcastAck)); + + /// + /// 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."); - 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(this, BroadcastOptions, Options.SerializerSettings); @@ -624,7 +638,9 @@ internal void Enqueue(Push push) /// /// 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)); /// /// Generates an auth push. diff --git a/RealtimeTests/ChannelBroadcastReplayTests.cs b/RealtimeTests/ChannelBroadcastReplayTests.cs new file mode 100644 index 0000000..58fef3b --- /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(DisplayName = "Channel: Registering broadcast replay on a public channel throws")] + public void ClientCannotRegisterReplayOnPublicChannel() + { + var channel = PublicChannel(); + + Assert.Throws( + () => channel.Register(WithReplay())); + } + + [TestMethod(DisplayName = "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 35cc481..6c7e9c1 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] @@ -37,7 +41,7 @@ public void CleanupTest() _socketClient!.Disconnect(); } - [TestMethod(DisplayName = "Channel: Can listen for broadcast")] + [TestMethod("Channel: Can listen for broadcast")] public async Task ClientCanListenForBroadcast() { var tsc = new TaskCompletionSource(); @@ -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,7 +83,61 @@ public async Task ClientCanListenForBroadcast() await Task.WhenAll(new[] { tsc.Task, tsc2.Task }); } - [TestMethod(DisplayName = "Channel: Send resolves when broadcast ack is not explicitly enabled")] + [TestMethod("Channel: Can listen for broadcast on a private channel")] + 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 = ChannelOptions.Private( + client1.Options, + () => Helpers.ApiKey, + new JsonSerializerSettings() + ); + 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 = ChannelOptions.Private( + client2.Options, + () => Helpers.ApiKey, + new JsonSerializerSettings() + ); + 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: Send resolves when broadcast ack is not explicitly enabled")] public async Task ChannelSendResolvesWithoutExplicitAck() { // Mirrors the most natural usage: `client.Channel(name)` -> `Subscribe()` -> `Send()`, @@ -92,24 +154,76 @@ public async Task ChannelSendResolvesWithoutExplicitAck() Assert.IsTrue(await sendTask); } - [TestMethod(DisplayName = "Channel: Payload returns a modeled response (if possible)")] + [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 = ChannelOptions.Private( + client1.Options, + () => null, + new JsonSerializerSettings() + ); + 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: Payload returns a modeled response (if possible)")] public async Task ChannelPayloadReturnsModel() { var tsc = new TaskCompletionSource(); 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/ClientTests.cs b/RealtimeTests/ClientTests.cs index 8e801bc..8755ede 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 ffd05f8..d568300 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 ec42e4e..0f6bf0a 100644 --- a/supabase/migrations/20250224164421_init.sql +++ b/supabase/migrations/20250224164421_init.sql @@ -184,11 +184,43 @@ WHERE username = name_param; $$ LANGUAGE SQL IMMUTABLE; --- Tables created by the `postgres` role in the `public` schema no longer inherit --- SELECT/INSERT/UPDATE/DELETE for the API roles under current Supabase CLI default privileges --- (https://supabase.com/changelog/45329-breaking-change-tables-not-exposed-to-data-and-graphql-api-automatically), --- so the privileges must be granted explicitly. -grant select, insert, update, delete on all tables in schema public to anon, authenticated, service_role; + +-- 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 + BEGIN + -- Scope the insert to the requested topic. + EXECUTE format('SET LOCAL realtime.topic TO %L', topic); + + INSERT INTO realtime.messages (payload, event, topic, private, extension) + VALUES (null, event, topic, private, 'broadcast'); + EXCEPTION + WHEN OTHERS THEN + RAISE WARNING 'ErrorSendingBroadcastMessage: %', SQLERRM; + 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); + +-- 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; grant usage, select on all sequences in schema public to anon, authenticated, service_role; grant execute on all functions in schema public to anon, authenticated, service_role;