Skip to content

feat(realtime): throw exception when try to update postgres change #385

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Tr00d merged 3 commits into from
Aug 25, 2026
Merged

feat(realtime): throw exception when try to update postgres change #385

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
53762b6
feat(realtime): throw exception when try to update
diegofesanto Aug 25, 2026
d91dfc7
fix(realtime): fix quality report
diegofesanto Aug 25, 2026
8017f55
fix(realtime): fix all warning touched file
diegofesanto Aug 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
108 changes: 70 additions & 38 deletions packages/Realtime/Realtime.Tests/PostgresChanges/PostgresChangesDeliveryTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Realtime.Tests.Models;
using Supabase.Postgrest.Interfaces;
using Supabase.Realtime;
using Supabase.Realtime.Exceptions;
using Supabase.Realtime.Interfaces;
using Supabase.Realtime.PostgresChanges;
using static Supabase.Realtime.Constants;
Expand All @@ -28,70 +28,102 @@ public class PostgresChangesDeliveryTests
[TestInitialize]
public async Task InitializeTest()
{
restClient = Helpers.RestClient();
socketClient = Helpers.SocketClient();
await socketClient.ConnectAsync();
this.restClient = Helpers.RestClient();
this.socketClient = Helpers.SocketClient();
await this.socketClient.ConnectAsync();
}

[TestCleanup]
public void CleanupTest() => socketClient.Disconnect();
public void CleanupTest() => this.socketClient.Disconnect();

[TestMethod]
public async Task OnPostgresChange_ShouldModelPayload()
{
var tsc = new TaskCompletionSource<bool>();
var channel = socketClient.Channel("example");
var channel = this.socketClient.Channel("example");
channel.OnPostgresChange((_, changes) =>
{
var model = changes.Model<Todo>();
tsc.SetResult(model != null);
}, ListenType.Inserts, new PostgresChangesFilter { Table = "*" });
await channel.Subscribe();
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client Models a response? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client Models a response? ✅" });
Assert.IsTrue(await tsc.Task);
}

[TestMethod]
public async Task OnPostgresChange_ShouldThrowError_GivenOnChangeAfterSubscribe()
{
var tsc = new TaskCompletionSource<bool>();
var channel = this.socketClient.Channel("example");
channel.OnPostgresChange((_, changes) =>
{
var model = changes.Model<Todo>();
tsc.SetResult(model != null);
}, ListenType.Inserts, new PostgresChangesFilter { Table = "*" });

await channel.Subscribe();

var act = () => channel.OnPostgresChange((_, changes) =>
{
var model = changes.Model<Todo>();
tsc.SetResult(model != null);
}, ListenType.Inserts, new PostgresChangesFilter { Table = "*" });

Assert.Throws<RealtimeException>(act);
}

[TestMethod]
public async Task OnPostgresChange_ShouldThrowError_GivenRegisterChangesAfterSubscribe()
{
var channel = this.socketClient.Channel("example");
await channel.Subscribe();

var act = () => channel.RegisterPostgresChangesOptions(new PostgresChangesOptions("example"));
Assert.Throws<RealtimeException>(act);
}

[TestMethod]
public async Task OnPostgresChange_ShouldReceiveInsert()
{
var tsc = new TaskCompletionSource<bool>();
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
channel.OnPostgresChange((_, _) => tsc.SetResult(true), ListenType.Inserts,
new PostgresChangesFilter { Table = "todos" });
await channel.Subscribe();
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives insert callback? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives insert callback? ✅" });
Assert.IsTrue(await tsc.Task);
}

[TestMethod]
public async Task OnPostgresChange_ShouldReceiveFilteredInsert()
{
var tsc = new TaskCompletionSource<bool>();
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
channel.OnPostgresChange((_, changes) =>
{
Assert.AreEqual("Client receives filtered insert callback? ✅", changes.Model<Todo>()?.Details);
tsc.SetResult(true);
}, ListenType.Inserts,
new PostgresChangesFilter { Table = "todos", Filter = "details=eq.Client receives filtered insert callback? ✅" });
await channel.Subscribe();
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives insert callback? ✅" });
await restClient.Table<Todo>().Insert(new Todo { UserId = 2, Details = "Client receives filtered insert callback? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives insert callback? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 2, Details = "Client receives filtered insert callback? ✅" });
Assert.IsTrue(await tsc.Task);
}

[TestMethod]
public async Task OnPostgresChange_ShouldReceiveUpdateAndFilteredInsert()
{
var tsc = new TaskCompletionSource<bool>();
var response = await restClient.Table<Todo>()
var response = await this.restClient.Table<Todo>()
.Insert(new Todo { UserId = 1, Details = "Client receives insert callback? ✅" });
await restClient.Table<Todo>()
await this.restClient.Table<Todo>()
.Insert(new Todo { UserId = 2, Details = "Client receives filtered insert callback? ✅" });
var model = response.Models.First();
var oldDetails = model.Details;
var newDetails = $"I'm an updated item ✏️ - {DateTime.Now}";
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
channel.OnPostgresChange((_, changes) =>
{
Assert.AreEqual(oldDetails, changes.OldModel<Todo>()?.Details);
Expand All @@ -111,20 +143,20 @@ await restClient.Table<Todo>()
tsc.SetResult(true);
}, ListenType.Inserts, new PostgresChangesFilter { Table = "todos", Filter = $"details=eq.{filter}" });
await channel.Subscribe();
await restClient.Table<Todo>().Set(x => x.Details!, newDetails).Match(model).Update();
await this.restClient.Table<Todo>().Set(x => x.Details!, newDetails).Match(model).Update();
Assert.IsTrue(await tsc.Task);
}

[TestMethod]
public async Task OnPostgresChange_ShouldReceiveUpdate()
{
var tsc = new TaskCompletionSource<bool>();
var response = await restClient.Table<Todo>()
var response = await this.restClient.Table<Todo>()
.Insert(new Todo { UserId = 1, Details = "Client receives insert callback? ✅" });
var model = response.Models.First();
var oldDetails = model.Details;
var newDetails = $"I'm an updated item ✏️ - {DateTime.Now}";
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
channel.OnPostgresChange((_, changes) =>
{
Assert.AreEqual(oldDetails, changes.OldModel<Todo>()?.Details);
Expand All @@ -138,32 +170,32 @@ public async Task OnPostgresChange_ShouldReceiveUpdate()
tsc.SetResult(true);
}, ListenType.Updates, new PostgresChangesFilter { Table = "todos" });
await channel.Subscribe();
await restClient.Table<Todo>().Set(x => x.Details!, newDetails).Match(model).Update();
await this.restClient.Table<Todo>().Set(x => x.Details!, newDetails).Match(model).Update();
Assert.IsTrue(await tsc.Task);
}

[TestMethod]
public async Task OnPostgresChange_ShouldReceiveDelete()
{
var tsc = new TaskCompletionSource<bool>();
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
channel.OnPostgresChange((_, _) => tsc.SetResult(true), ListenType.Deletes,
new PostgresChangesFilter { Table = "todos" });
await channel.Subscribe();
var result = await restClient.Table<Todo>().Get();
var result = await this.restClient.Table<Todo>().Get();
var model = result.Models.Last();
await restClient.Table<Todo>().Match(model).Delete();
await this.restClient.Table<Todo>().Match(model).Delete();
Assert.IsTrue(await tsc.Task);
}

[TestMethod]
public async Task OnPostgresChange_ShouldReceiveFilteredDelete()
{
var tsc = new TaskCompletionSource<bool>();
var channel = socketClient.Channel("realtime", "public", "todos");
var todo1 = await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives callbacks 1? ✅" });
var todo2 = await restClient.Table<Todo>().Insert(new Todo { UserId = 2, Details = "Client receives callbacks 2? ✅" });
await restClient.Table<Todo>().Insert(new Todo { UserId = 3, Details = "Client receives callbacks 3? ✅" });
var channel = this.socketClient.Channel("realtime", "public", "todos");
var todo1 = await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives callbacks 1? ✅" });
var todo2 = await this.restClient.Table<Todo>().Insert(new Todo { UserId = 2, Details = "Client receives callbacks 2? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 3, Details = "Client receives callbacks 3? ✅" });
channel.OnPostgresChange((_, removed) =>
{
var result = removed.OldModel<Todo>();
Expand All @@ -173,8 +205,8 @@ public async Task OnPostgresChange_ShouldReceiveFilteredDelete()
}, ListenType.Deletes,
new PostgresChangesFilter { Table = "todos", Filter = $"details=eq.{todo1.Model?.Details}" });
await channel.Subscribe();
await restClient.Table<Todo>().Match(todo1.Models.First()).Delete();
await restClient.Table<Todo>().Match(todo2.Models.First()).Delete();
await this.restClient.Table<Todo>().Match(todo1.Models.First()).Delete();
await this.restClient.Table<Todo>().Match(todo2.Models.First()).Delete();
Assert.IsTrue(await tsc.Task);
}

Expand All @@ -184,7 +216,7 @@ public async Task OnPostgresChange_ShouldReceiveAllEvents_GivenWildcard()
var insertTsc = new TaskCompletionSource<bool>();
var updateTsc = new TaskCompletionSource<bool>();
var deleteTsc = new TaskCompletionSource<bool>();
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
channel.OnPostgresChange((_, changes) =>
{
switch (changes.Payload?.Data?.Type)
Expand All @@ -195,10 +227,10 @@ public async Task OnPostgresChange_ShouldReceiveAllEvents_GivenWildcard()
}
}, ListenType.All, new PostgresChangesFilter { Table = "todos" });
await channel.Subscribe();
var inserted = await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives wildcard callbacks? ✅" });
var inserted = await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives wildcard callbacks? ✅" });
var newModel = inserted.Models.First();
await restClient.Table<Todo>().Set(x => x.Details!, "And edits.").Match(newModel).Update();
await restClient.Table<Todo>().Match(newModel).Delete();
await this.restClient.Table<Todo>().Set(x => x.Details!, "And edits.").Match(newModel).Update();
await this.restClient.Table<Todo>().Match(newModel).Delete();
await Task.WhenAll(insertTsc.Task, updateTsc.Task, deleteTsc.Task);
Assert.IsTrue(insertTsc.Task.Result);
Assert.IsTrue(updateTsc.Task.Result);
Expand All @@ -213,7 +245,7 @@ public async Task OnPostgresChange_ShouldFanOutToMultipleInsertListeners()
var insertTask3 = new TaskCompletionSource<bool>();
const string filter1 = "Client receives callbacks 1? ✅";
const string filter2 = "Client receives callbacks 2? ✅";
var channel = socketClient.Channel("realtime", "public", "todos");
var channel = this.socketClient.Channel("realtime", "public", "todos");
var count = 0;
channel.OnPostgresChange((_, _) =>
{
Expand All @@ -227,9 +259,9 @@ public async Task OnPostgresChange_ShouldFanOutToMultipleInsertListeners()
insertTask3.SetResult(added.Model<Todo>()?.Details == filter2), ListenType.Inserts,
new PostgresChangesFilter { Table = "todos", Filter = $"details=eq.{filter2}" });
await channel.Subscribe();
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives wildcard callbacks? ✅" });
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = filter1 });
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = filter2 });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "Client receives wildcard callbacks? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = filter1 });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = filter2 });
await Task.WhenAll(insertTask1.Task, insertTask2.Task, insertTask3.Task);
Assert.IsTrue(insertTask1.Task.Result);
Assert.IsTrue(insertTask2.Task.Result);
Expand All @@ -240,11 +272,11 @@ public async Task OnPostgresChange_ShouldFanOutToMultipleInsertListeners()
public async Task OnPostgresChange_ShouldRegisterAndDeliver_GivenChainedSubscribe()
{
var tsc = new TaskCompletionSource<bool>();
await socketClient.Channel("public:todos")
await this.socketClient.Channel("public:todos")
.OnPostgresChange((_, changes) => tsc.TrySetResult(changes.Model<Todo>() != null),
ListenType.Inserts, new PostgresChangesFilter { Table = "todos" })
.Subscribe();
await restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "OnPostgresChange receives insert? ✅" });
await this.restClient.Table<Todo>().Insert(new Todo { UserId = 1, Details = "OnPostgresChange receives insert? ✅" });
Assert.IsTrue(await WithinTimeout(tsc.Task));
}

Expand Down
10 changes: 7 additions & 3 deletions packages/Realtime/Realtime/Exceptions/FailureHint.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
using System;
using Websocket.Client;

namespace Supabase.Realtime.Exceptions;
Expand Down Expand Up @@ -48,6 +47,11 @@ public enum Reason
/// If seen, please open an issue.
/// </summary>
ConnectionStale,

/// <summary>
/// Cannot make changes after subscribe
/// </summary>
StateInvalid,
}

/// <summary>
Expand All @@ -63,7 +67,7 @@ public static Reason Parse(DisconnectionInfo info)
DisconnectionType.NoMessageReceived => Reason.ConnectionStale,
DisconnectionType.Lost => Reason.ConnectionLost,
DisconnectionType.ByServer => Reason.Unknown,
_ => Reason.Unknown
_ => Reason.Unknown,
};
}
}
}
1 change: 1 addition & 0 deletions packages/Realtime/Realtime/PublicAPI.Unshipped.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ Supabase.Realtime.Channel.ChannelOptions.SerializerSettings.get -> System.Text.J
Supabase.Realtime.Client.SerializerSettings.get -> System.Text.Json.JsonSerializerOptions!
Supabase.Realtime.ClientOptions.WebSocketFactory.get -> Supabase.Realtime.Sockets.IWebSocketFactory?
Supabase.Realtime.ClientOptions.WebSocketFactory.set -> void
Supabase.Realtime.Exceptions.FailureHint.Reason.StateInvalid = 7 -> Supabase.Realtime.Exceptions.FailureHint.Reason
Supabase.Realtime.Interfaces.IRealtimeClient<TSocket, TChannel>.SerializerSettings.get -> System.Text.Json.JsonSerializerOptions!
Supabase.Realtime.PostgresChanges.PostgresChangesResponse.PostgresChangesResponse() -> void
Supabase.Realtime.PostgresChanges.PostgresChangesResponse.PostgresChangesResponse(System.Text.Json.JsonSerializerOptions! serializerSettings) -> void
Expand Down
11 changes: 9 additions & 2 deletions packages/Realtime/Realtime/RealtimeChannel.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -443,6 +443,13 @@ public IRealtimeChannel Register(PostgresChangesOptions postgresChangesOptions)
/// <param name="postgresChangesOptions"></param>
internal void RegisterPostgresChangesOptions(PostgresChangesOptions postgresChangesOptions)
{
if (this.IsJoined || this.IsJoining)
throw new RealtimeException(
$"Cannot add `postgres_changes` callbacks for {this.Topic} after `Subscribe()`.")
{
Reason = FailureHint.Reason.StateInvalid,
};

this.PostgresChangesOptions.Add(postgresChangesOptions);
this.BindPostgresChangesOptions(postgresChangesOptions);
}
Expand Down Expand Up @@ -534,7 +541,7 @@ public IRealtimeChannel Unsubscribe()

/// <summary>
/// Sends a `Push` request under this channel.
///
///
/// Maintains a buffer in the event push is called prior to the channel being joined.
/// </summary>
/// <param name="eventName"></param>
Expand Down Expand Up @@ -885,7 +892,7 @@ private void BindIdPostgresChanges(PhoenixPostgresChangeResponse joinResponse)
}

/// <summary>
/// Try to invoke the handler properly based on event type and socket response
/// Try to invoke the handler properly based on event type and socket response
/// </summary>
/// <param name="eventType"></param>
/// <param name="response"></param>
Expand Down
Loading