Implement listen / notify for Postgres backend

[mcraveiro]

Hi sqlgen developers,

As per discussion in [1], I thought it may be easier to discuss this topic with some kind of experimental evidence. I had a quick go at this using qwen and came with with something which, at first glance, is not entirely unusable :-) As with OFFSET, I will iterate on it until it all goes green etc - but if you have any comments as I go along, or if you disagree with the approach entirely do let me know.

Cheers

Marco

[1] Using sqlgen with LISTEN/NOTIFY #104

[mcraveiro]

Assorted snippets with the conversation with qwen

It may be of use to others to see how I arrived at the present state. Here are some useful snippets of the chat I had with qwen; I've edited them lightly for readability.

Define overall approach

In this conversation qwen sketched out the overall approach.

🎯 Goal: Add LISTEN / NOTIFY Support (PostgreSQL-specific)

Since LISTEN / NOTIFY are PostgreSQL-specific asynchronous notification mechanisms, they don’t map to typical CRUD or query patterns. Therefore, support should:

• Be optional (only enabled with -DSQLGEN_POSTGRESQL=ON or similar),
• Integrate cleanly with the existing sqlgen::postgres::Connection,
• Expose a safe, non-blocking, and composable API,
• Respect the library’s preference for Result types (rfl::Result) and functional chaining.

---

🧠 Design Brainstorm

1. Core Operations

We need to support:

• LISTEN channel_name
• UNLISTEN channel_name (or UNLISTEN *)
• NOTIFY channel_name [, payload]
• Asynchronously receive notifications while the connection is idle or between queries

Note: PostgreSQL delivers NOTIFY messages asynchronously via the same connection, but they are only processed when you call PQconsumeInput() or similar.

---

2. Connection Extension

Extend sqlgen::postgres::Connection with:

Result<Nothing> listen(const std::string& channel) noexcept;
Result<Nothing> unlisten(const std:: string& channel) noexcept;
Result<Nothing> notify(const std::string& channel, const std::string& payload = "") noexcept;

These would internally issue LISTEN, UNLISTEN, and NOTIFY SQL commands.

✅ Consistent with existing execute() pattern.

---

3. Notification Polling / Callbacks

This is trickier. Options:

A. Polling API

Add a method:

std::vector<Notification> get_notifications();

Where Notification is:

struct Notification {
    std::string channel;
    std::string payload;
    std::string pid; // sender's backend PID (optional)
};

Internally, this would:

• Call PQconsumeInput()
• Drain PQnotifies() until null

✅ Simple, synchronous, no threading.
⚠️ User must poll manually (not event-driven).

B. Callback Registration

Allow the user to register a handler:

void on_notification(std::function<void(const Notification&)> handler);

But this implies:

• Running a background thread or integrating with an event loop (e.g., Asio)
• Violates sqlgen’s current sync-only, no-hidden-thread philosophy

❌ Likely too opinionated for a low-level library like sqlgen.

C. Async-Friendly (C++20 Coroutines or Sender-Receiver)

Possible in the future, but sqlgen doesn’t currently use coroutines.

❌ Out of scope for now.

---

4. Integration with Fluent Interface?

Unlike SELECT or INSERT, LISTEN isn’t composable in the same way. So it probably should not use the | operator chain.

Instead, treat it as a connection-side effect, like begin_transaction().

So:

auto conn = sqlgen::postgres::connect(...);
conn.listen("my_channel").value();

// Later...
auto notices = conn.get_notifications();
for (const auto& n : notices) {
    std::cout << "Got: " << n.payload << " on " << n.channel << "\n";
}

This aligns with existing patterns (execute, begin_transaction, etc.).

---

5. Safety & Validation

• Channel names in PostgreSQL must be valid identifiers.
• We should not allow arbitrary strings without sanitization.

But note: LISTEN my_channel does not support parameters — it’s not a prepared statement.
So the channel name is interpolated into the SQL string.

➡️ Therefore, we must validate that channel matches [a-zA-Z_][a-zA-Z0-9_]* (or allow quoted identifiers carefully).

Proposal: Only allow sqlgen::Identifier or sqlgen::AlphaNumeric-like wrapper for channel names, similar to how column names are handled.

Alternatively, quote the identifier:

auto safe_name = "\"" + escape_double_quotes(channel) + "\"";

But PostgreSQL LISTEN does not accept quoted identifiers! It treats them as literals.

From PG docs:

The channel name is case-sensitive and must match the name used in the NOTIFY command.

And:

The name must be a valid identifier ([a-z0-9_]+), unquoted.

So: restrict to safe identifier format at compile-time or runtime.

Suggestion:

struct ChannelName {
    std::string value;
    constexpr ChannelName(const char* s) : value(s) {
        // optional: static validation if used as literal
    }
};

Or reuse sqlgen::literals mechanism: "my_channel"_chan

But maybe overkill. Simpler: runtime validation + clear error.

---

6. Error Handling

All methods return rfl::Result<Nothing> or rfl::Result<std::vector<Notification>>, consistent with rest of API.

---

✅ Implementation

Implementation:

• listen → execute("LISTEN " + validate_and_format(channel))
• notify → execute("NOTIFY " + channel + ", " + quote_literal(payload))
• get_notifications() → loop over PQnotifies() after PQconsumeInput()

Note: PQnotifies() returns a queue — each call pops one notification.

---

🔒 Security Notes

• notify() payload must be quoted using PQescapeLiteral or equivalent.
• listen() channel must be validated as a PostgreSQL identifier.

---

🧩 Future Considerations

• Could expose a non-blocking wait via PQisBusy() + get_notifications() in a loop (for CLI tools).
• Integration with Asio or Qt event loop would be in user code, not sqlgen core.

---

Summary

Feature	Design
LISTEN / UNLISTEN	Connection methods returning Result<Nothing>
NOTIFY	Connection method with optional payload
Notifications	Retrieved via get_notifications()
Safety	Validate channel names, escape payload
Style	Matches existing sqlgen PostgreSQL API (sync, functional, no hidden threads)

Discussions on retrieving results

In this conversation we discussed the pro's and con's of owning the loop. We decided against adding any threading etc within sqlgen as it seems to go against the library design.

✅ Summary of Your Approach

• User owns the loop (e.g., in a CLI tool, daemon, or Asio-compatible I/O loop).
• select() (or poll()/epoll() on Linux) blocks until either:
  • Data arrives on the PostgreSQL socket (potentially a NOTIFY),
  • A timeout occurs,
  • The loop is explicitly broken (e.g., by signal or shutdown flag).
• After select() indicates readability, you:
  • Call PQconsumeInput() to buffer incoming data,
  • Drain all available PGnotify* objects via PQnotifies().

This is exactly how libpq documentation recommends handling asynchronous notifications in a single-threaded app.

[mcraveiro]

Explanation of LISTEN / NOTIFY

Actually it just occurred to me that you may not be entirely familiar with this pattern as it appears to be a postgres specific feature. Let me sketch it out quickly. From the docs [1]:

The NOTIFY command sends a notification event together with an optional “payload” string to each client application that has previously executed LISTEN channel for the specified channel name in the current database. Notifications are visible to all users.

NOTIFY provides a simple interprocess communication mechanism for a collection of processes accessing the same PostgreSQL database. A payload string can be sent along with the notification, and higher-level mechanisms for passing structured data can be built by using tables in the database to pass additional data from notifier to listener(s).

The information passed to the client for a notification event includes the notification channel name, the notifying session's server process PID, and the payload string, which is an empty string if it has not been specified.

It is up to the database designer to define the channel names that will be used in a given database and what each one means. Commonly, the channel name is the same as the name of some table in the database, and the notify event essentially means, “I changed this table, take a look at it to see what's new”. But no such association is enforced by the NOTIFY and LISTEN commands. For example, a database designer could use several different channel names to signal different sorts of changes to a single table. Alternatively, the payload string could be used to differentiate various cases.

When NOTIFY is used to signal the occurrence of changes to a particular table, a useful programming technique is to put the NOTIFY in a statement trigger that is triggered by table updates. In this way, notification happens automatically when the table is changed, and the application programmer cannot accidentally forget to do it.

Given this, the approach we intend to use in our code is as follows, we define triggers on a per table basis like so:

-- Trigger function to send notifications on currency changes
CREATE OR REPLACE FUNCTION oresdb.notify_currency_changes()
RETURNS TRIGGER AS $$
DECLARE
    notification_payload jsonb;
    entity_name text := 'ores.risk.currency';
    change_timestamp timestamptz := NOW();
BEGIN
    -- Construct the JSON payload
    notification_payload := jsonb_build_object(
        'entity', entity_name,
        'timestamp', to_char(change_timestamp, 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')
    );

    -- Notify on the 'ores_currencies' channel
    PERFORM pg_notify('ores_currencies', notification_payload::text);

    RETURN NULL; -- AFTER triggers can return NULL
END;
$$ LANGUAGE plpgsql;

-- Trigger to fire after INSERT, UPDATE, or DELETE on the currencies table
CREATE OR REPLACE TRIGGER currency_change_notify_trigger
AFTER INSERT OR UPDATE OR DELETE ON oresdb.currencies
FOR EACH ROW EXECUTE FUNCTION oresdb.notify_currency_changes();

When an insert is performed, we see it as follows:

[ores@localhost:5432 (13:26:41) oresdb ]$ listen ores_currencies;
LISTEN
[ores@localhost:5432 (13:26:52) oresdb ]$ SELECT pg_listening_channels();
 pg_listening_channels 
-----------------------
 ores_currencies
(1 row)

[ores@localhost:5432 (13:28:46) oresdb ]$ ;select * from currencies order by valid_from desc limit 3;
 iso_code |        name         | numeric_code | symbol | fraction_symbol | fractions_per_unit | rounding_type | rounding_precision | format | currency_type | modified_by |          valid_from           |        valid_to        
----------+---------------------+--------------+--------+-----------------+--------------------+---------------+--------------------+--------+---------------+-------------+-------------------------------+------------------------
 Test     | Test Currency       | 12345        |        |                 |                  0 |               |                  0 |        | Major         | newuser3    | 2025-12-05 12:47:39.886489+00 | 9999-12-31 23:59:59+00
 CUP      | Cuban peso          | 192          |        |                 |                100 | Closest       |                  2 |        |               | ores        | 2025-11-28 12:04:18.005205+00 | 9999-12-31 23:59:59+00
 CVE      | Cape Verdean escudo | 132          |        |                 |                100 | Closest       |                  2 |        |               | ores        | 2025-11-28 12:04:18.005205+00 | 9999-12-31 23:59:59+00
(3 rows)

<INSERT PERFORMED>

[ores@localhost:5432 (13:29:59) oresdb ]$ select * from currencies order by valid_from desc limit 3;
 iso_code |     name      | numeric_code | symbol | fraction_symbol | fractions_per_unit | rounding_type | rounding_precision | format | currency_type | modified_by |          valid_from           |        valid_to        
----------+---------------+--------------+--------+-----------------+--------------------+---------------+--------------------+--------+---------------+-------------+-------------------------------+------------------------
 Test2    | Test Listen   | 12345        |        |                 |                  0 |               |                  0 |        |               | newuser3    | 2025-12-05 13:29:54.898161+00 | 9999-12-31 23:59:59+00
 Test     | Test Currency | 12345        |        |                 |                  0 |               |                  0 |        | Major         | newuser3    | 2025-12-05 12:47:39.886489+00 | 9999-12-31 23:59:59+00
 CUP      | Cuban peso    | 192          |        |                 |                100 | Closest       |                  2 |        |               | ores        | 2025-11-28 12:04:18.005205+00 | 9999-12-31 23:59:59+00
(3 rows)

Asynchronous notification "ores_currencies" with payload "{"entity": "currency", "timestamp": "2025-12-05T13:29:54.898Z"}" received from server process with PID 39239.

The JSON object will then be retrieved via the sqlgen machinery, deserialised with rfl and processed by the application.

[1] NOTIFY — generate a notification

[mcraveiro]

Multiple connections required

Important point:

However, there's a more fundamental issue:
🟢 A PostgreSQL session cannot receive its own NOTIFY messages.

From the PostgreSQL docs:

“Notifications are delivered to all listening sessions except the one that executed the NOTIFY.”

[mcraveiro]

Exposing the native socket

After playing with this for a bit, it seems to me that we need to expose the native socket so that client code can wait on it.

Qwen analysis on exposing socket

Rather than embedding platform-specific I/O logic (select()/WSAEventSelect) directly into sqlgen::postgres::Connection, expose the underlying socket and let the user integrate it into their own event loop or waiting logic.

This approach:

---

✅ Matches sqlgen’s design philosophy

• Minimalist: sqlgen doesn’t manage threads, event loops, or I/O polling.
• Composable: Users bring their own concurrency (Asio, Qt, libuv, CLI loop, etc.).
• Portable: Avoids #ifdef _WIN32 in sqlgen core; pushes platform handling to the user (who already knows their environment).
• Consistent: Other database libraries (e.g., libpqxx) follow this pattern — provide the socket, let the app wait.

---

✅ Avoids cross-platform I/O complexity in sqlgen

• No need for winsock2.h, sys/select.h, poll(), or epoll() in sqlgen.
• No linking to ws2_32.lib unless the user chooses to use socket I/O.
• sqlgen remains a pure query/data layer, not a networking layer.

---

✅ What to expose

Add one method to sqlgen::postgres::Connection:

// Returns the underlying socket descriptor used by libpq.
// Valid only when connection is active.
// On POSIX: int; on Windows: SOCKET (but libpq returns int for portability).
// Returns -1 if no active connection.
int native_socket() const noexcept {
    return PQsocket(conn_);
}

🔸 Note: PQsocket() returns int on all platforms (including Windows), as per libpq docs.

So this is already cross-platform.

---

✅ User code handles waiting

The user then integrates like this (example with Asio):

auto conn = sqlgen::postgres::connect(creds).value();
conn->listen("mychan").value();

asio::io_context io;
asio::posix::stream_descriptor sock(io, conn->native_socket());

sock.async_wait(asio::posix::stream_descriptor::wait_read, [&](auto ec) {
    if (!ec) {
        // Now safe to call conn->get_notifications()
        auto notices = conn->get_notifications();
        // process...
    }
});
io.run();

Or in a CLI loop with select() (user writes the portable version they need).

---

✅ What to remove

• Remove wait_for_notification() from sqlgen core.
• Keep get_notifications() (it’s pure libpq, no I/O waiting).
• Keep listen()/notify() (they’re just SQL commands).

---

✅ Precedent in sqlgen

sqlgen already follows this pattern:

• It does not provide begin_background_listener().
• It does not abstract event loops.
• It exposes only data and query operations, not transport semantics.

---

✅ Recommendation

Do this:

// In Connection.hpp
int native_socket() const noexcept;

Don’t do this:

• Embed select()/poll()/Winsock in sqlgen.
• Provide a blocking wait_for_notification().

Instead, document the pattern:

To receive NOTIFY messages, call listen(), then monitor the socket returned by native_socket() using your event loop. When readable, call get_notifications() to drain pending messages.

This keeps sqlgen lean, portable, and composable — exactly as designed.

Would you like a PR-ready patch for native_socket() and updated documentation/example?

[mcraveiro]

Accessing the connection from the session

OK when I finally tried using all of this in anger, I immediately bumped into a problem :-) so I was thinking I could just start a session off of the pool as I do for normal code and just keep that session alive within my Postgres listener. However, from the session I don't seem to be able to access the connection with the present session interface.

Not quite sure what the right thing is here:

• should I add a connection() method which returns the ConnPtr conn_;? or
• is it best to expose only the required methods within the connection as we do for everything else?

For the latter, we would need:

  std::list<Notification> get_notifications() noexcept;

  rfl::Result<Nothing> listen(const std::string& channel) noexcept;

  rfl::Result<Nothing> unlisten(const std:: string& channel) noexcept;

  rfl::Result<Nothing> notify(const std::string& channel, const std::string& payload = "") noexcept;

  int native_socket() const noexcept;

I think I will start by just forwarding these calls from the session to the connection since that seems to be in keeping with the current approach whilst I wait for any commentary from you guys.

[mcraveiro]

Accessing the connection from the session

I think I will start by just forwarding these calls from the session to the connection since that seems to be in keeping with the current approach whilst I wait for any commentary from you guys.

Actually this is not even that sensible because the notification methods are Postgres specific. I am just going to do the quick thing and expose a getter for the connection for now and perhaps you guys can help me with the right solution for this at some point later on.

[mcraveiro]

Storing session as a long lived connection

Second problem: storing the session as a member variable in a listener class does not seem like the greatest of ideas. I mean, to be fair, you already say this in your docs [1]:

3. Session Lifetime: Keep sessions as short as possible:

// Good: Session is released immediately after use
session(pool).and_then(execute_query).value();

// Bad: Session is held longer than necessary
const auto sess = session(pool).value();
// ... other operations ...
sess.execute_query();

So this is making me believe that we should not really use the session interface for this use case:

• bit of a hack really, as we are just taking one connection off of the pool and keeping it around to listen to events.
• we can't return the connection to the pool and get another one - the whole point of notifications is that you keep listening in.
• we need some retry logic to obtain a new connection if the current one dies. This may involve some polling; but as of yet, I'm not quite sure how you'd find out the connection is alive, short of doing something with it. I need to find some kind of no-op operation on the connection (possibly related to LISTEN/NOTIFY) which can be used to validate its liveliness.

I'll try to put as much of this stuff in client code to avoid polluting sqlgen with it, get it to work end to end, and then present my findings so we can brainstorm as to what belongs where.

[1] Connection Pool

[mcraveiro]

Ready for review - Final thoughts

I have now removed all the unnecessary bits and added documentation. Let me know what you think.

Build is still red due to some intermittent issues. I will try to force a rebuild tomorrow, hopefully it will go green.