Postgres subscriptions (working stub example)

[Alorien1984]

Product
Hot Chocolate

A warm welcome to the esteemed developers and members of the HotChocolate community!
I'm trying to figure out the relationship between HotCgocolate subscriptions and the PostgresQL notification mechanism.

I have notifications being created in PostgresQL under the channel name 'hotchocolate_subscriptions'.
Example: NOTIFY "hotchocolate_subscriptions", 'This is payload!'

I have connected AddPostgresSubscriptions in this way:

    .AddPostgresSubscriptions((x, op) => op.ConnectionFactory = ct =>
    {
        var connection = new NpgsqlConnection(builder.Configuration.GetConnectionString("DefaultConnection"));
        return new ValueTask<NpgsqlConnection>(connection);
    })

I defined a subscription in the Subscription class this way:

    [Subscribe(With = nameof(SubscribePostgresTest))]
    public string OnMessageReceived(
        [EventMessage] string message)
    {
        return message;
    }
    
    public async ValueTask<ISourceStream<string>> SubscribePostgresTest(
        [Service] ITopicEventReceiver receiver)
    { 
        var result = await receiver.SubscribeAsync<string>("hotchocolate_subscriptions");
        return result;
    }

The project starts correctly, but I don't see any messages by the name of this channel in HotChocolate GraphQL UI.

Can I find a correct example somewhere? The documentation didn't give any answers.

Would you please be so kind as to you tell me what I am doing wrong?

Regards.

[juantxorena]

More than one year later, I have the same question/problem. The documentation is extremely lacking. Please provide some kind of working example, instead of having to read through the code to know how to use a theoretically supported feature.

[PascalSenn]

@juantxorena See #6404 (comment)

[PascalSenn]

Hi @Alorien1984

The Postgres integration for subscription should be used together with ITopicEventSender (and ITopicEventReceiver). To Send subscriptions you can use the SendAsync method, to listen to subscription events, you should use the ITopicEventReceiver and subscribe to the same topic that you push messages to.

In the documentation you find this here:
https://chillicream.com/docs/hotchocolate/v13/defining-a-schema/subscriptions/#publishing-events

You cannot directly notify the topic in postgres, as the structure of the encoded message might change.

[juantxorena]

Hi @PascalSenn

My intention was to launch the notify directly from postgresql, with a trigger or something when some tables are updated, because the app offering the graphql api is not the app writing in the database.

So it is my understanding that this is not supported? If that is the case, that's very disappointing. I think the documentation should say explicitly what is supported and what isn't (on this topic and on any other), and provide some more real life examples rather than the simplest "hello world"-like examples.

[PascalSenn]

It's probably possible if you send it in the exact format that the subscription provider requires it to be in. But it's like calling a private method with reflection. Once we change something in the structure with an update, your trigger will suddenly stop working correctly.

Regarding the documentation, we are always looking for people that help out with the documentation. So feel free to open a pull request if you think it can be improved. You seem to be already deep in the topic, you just have to type your knowledge down now.

[juantxorena]

Well, I've been with graphql in general and hotchocolate in particular for 4 days, so I doubt I can help with anything. Plus, I'm only interested in subscriptions, in particular because I thought that the project fully supported notify/listen from postgresql. If this is not the case, either I will implement the service for listening to notifies myself, or most probably I will ditch graphql altogether.

BTW, why should the notify format change? Postgresql has native json support, which npgsql mappes cleanly to classes. If the subscription expects a class that represents an event, why not use a json object as the notify payload?

[derekh2]

@Alorien1984 I am unsure what the AddPostgresSubscriptions is for but I you may be able to do what you're want by making a HostedService that uses Postgres listen and then forwards the updates using the ITopicEventSender class to send the message.

using System.Text.Json;
using HotChocolate.Subscriptions;
using Npgsql;

namespace collection_service.GraphQL;

public class PostgresNotificationService : BackgroundService
{
    private static readonly string POSTGRES_TOPIC = "export_tracking_update";
    private readonly ILogger<PostgresNotificationService> logger;
    private readonly ITopicEventSender sender;
    private readonly NpgsqlConnection connection;

    public PostgresNotificationService(
        ILogger<PostgresNotificationService> logger, 
        ITopicEventSender sender,
        NpgsqlDataSource dataSource
    )
    {
        this.logger = logger;
        this.sender = sender;
        connection = dataSource.CreateConnection();
    }

    private async void HandleNotificationReceived(object data, NpgsqlNotificationEventArgs events)
    {
        try 
        {
            // internal topic notification to GraphQL.
            await sender.SendAsync("export_tracking_update", JsonSerializer.Deserialize<Data>(events.Payload));
        } 
        catch (Exception e)
        {
            logger.LogError(e, "An exception occurred while deserializing the Postgres payload.");
        }
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await connection.OpenAsync(stoppingToken);
        connection.Notification += HandleNotificationReceived;

        try
        {
            logger.LogInformation("Started listening for PostgreSQL notifications on channel '{Topic}'.", POSTGRES_TOPIC);

            // continuously wait for notifications until the application is stopped
            while (!stoppingToken.IsCancellationRequested)
            {
                // listening for notifications on Postgres topic channel
                using var command = new NpgsqlCommand($"LISTEN {POSTGRES_TOPIC}", connection);
                await command.ExecuteNonQueryAsync(stoppingToken);
            }
        }
        catch (Exception e)
        {
            logger.LogError(e, "An exception occurred while listening for Postgres notifications.");
        }
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        // close connection when the service is stopped
        if (connection != null && connection.State == System.Data.ConnectionState.Open)
        {
            await connection.CloseAsync();
        }

        logger.LogInformation("PostgreSQL notification listener stopped.");

        await base.StopAsync(cancellationToken);
    }
}

[derekh2]

Here's the related database function/trigger I created using DBeaver. This turns the updated/created row into JSON before calling pg_notify with the data on the same topic we specified in that class above:

-- DROP FUNCTION data_movement_data.notify_graphql_clients();

CREATE OR REPLACE FUNCTION export_tracking.notify_graphql_clients()
 RETURNS trigger
 LANGUAGE plpgsql
AS $function$
	BEGIN
	    -- Send a notification to any listeners about the updated row in JSON format
	    PERFORM pg_notify('export_tracking_update', row_to_json(NEW)::text);
		RETURN NULL;
	END;
$function$
;

And then the trigger that calls that on UPDATE / INSERT / DELETE:

create trigger data_export_tracking_update after
insert
    or
delete
    or
update
    on
    export_tracking.export_tracking_data for each row execute function export_tracking.notify_graphql_clients()