Pub/Sub Messaging

pgvis includes a general-purpose publish/subscribe system backed by Postgres LISTEN/NOTIFY. It works like Redis pub/sub — any client can publish messages to named channels, and any number of subscribers receive them in real time.

Multiple pgvis instances connected to the same Postgres database automatically form a shared message bus without additional infrastructure.

Enabling Pub/Sub

Pub/sub is disabled by default. Enable it via CLI flag or environment variable:

# CLI
pgvis serve --pubsub-enabled

# Environment variable
PGVIS_PUBSUB_ENABLED=true pgvis serve

Or in your TOML configuration:

[pubsub]
enabled = true
channel_prefix = "pgvis:"      # optional, default "pgvis:"
max_payload_bytes = 7500        # optional, default 7500
channel_buffer_size = 64        # optional, default 64
allowed_channels = []           # empty = allow all
authorize_function = "auth.pubsub_authorize"  # optional, see below

Authentication

The pub/sub endpoints verify the caller exactly like the data API: the same jwt_secret, anon_role, role_claim_key and jwt_aud settings apply, and a request without a token runs as anon_role (or gets 401 when no anon_role is configured). The verified role and JWT claims are what authorize_function and pg_notify run as — see Per-Channel Authorization.

The curl examples below assume an anon_role is configured; otherwise add -H "Authorization: Bearer $TOKEN".

REST API (Server-Sent Events)

The REST surface exposes pub/sub at the /pubsub path prefix.

Subscribe to a Channel

GET /pubsub/{channel}
Accept: text/event-stream

Opens a Server-Sent Events (SSE) stream. Messages arrive as event: message frames:

curl -N -H "Accept: text/event-stream" \
  http://localhost:3000/pubsub/chat

Response stream:

: keepalive

event: message
data: {"channel":"chat","payload":"hello world","timestamp":"2026-06-01T10:00:00.000Z"}

event: message
data: {"channel":"chat","payload":"{\"user\":\"alice\",\"text\":\"hi\"}","timestamp":"2026-06-01T10:00:01.123Z"}

The connection stays open until the client disconnects. A keepalive comment (: keepalive) is sent every keepalive_interval_secs (default 15 seconds) to prevent proxy timeouts. A subscriber that falls more than channel_buffer_size messages behind receives a : lagged N messages comment in place of the messages it missed.

Before the stream opens, pgvis validates the channel name, runs authorize_function with op = 'subscribe' (if configured) and reserves a subscriber slot; any failure is returned as an error response instead of a stream.

Publish to a Channel

POST /pubsub/{channel}
Content-Type: text/plain

your message payload here

The body is sent as-is to all subscribers of that channel.

Example:

# Publish a plain text message
curl -X POST http://localhost:3000/pubsub/chat \
  -H "Content-Type: text/plain" \
  -d "hello world"

# Publish JSON (just send it as the body)
curl -X POST http://localhost:3000/pubsub/notifications \
  -H "Content-Type: text/plain" \
  -d '{"event":"order_placed","order_id":42}'

Response: 200 OK with {"ok": true} on success, or an error response. The pg_notify runs as the caller’s database role, in the same statement as the authorize_function check when one is configured.

Channel Status

GET /pubsub
Accept: application/json

Returns the channels with subscribers on this instance and their counts:

{
  "total_subscribers": 4,
  "channels": [
    { "name": "chat", "subscribers": 3 },
    { "name": "notifications", "subscribers": 1 }
  ]
}

The status endpoint requires the same authentication as subscribe/publish (it reveals channel names); authorize_function is not consulted for it.

Errors

Pub/sub failures return a JSON body with a PGVIS_PUBSUB_* code:

{ "error": "pubsub error: access to channel 'admin' denied", "code": "PGVIS_PUBSUB_CHANNEL_DENIED" }

Code	Status	When
PGVIS_PUBSUB_INVALID_CHANNEL	400	Empty name, null byte, or prefix + name longer than 63 bytes
PGVIS_PUBSUB_PAYLOAD_TOO_LARGE	400	Payload over max_payload_bytes
PGVIS_PUBSUB_CHANNEL_DENIED	403	Channel not in allowed_channels, or authorize_function did not return true
PGVIS_PUBSUB_MAX_SUBSCRIBERS	503	max_subscribers or max_subscribers_per_identity reached
PGVIS_PUBSUB_CONNECTION_LOST	503	The listener connection is down
PGVIS_PUBSUB_NOT_AVAILABLE	501	Pub/sub unavailable

Authentication failures (401) use the data API’s PGRST3xx error format.

MCP Tools (AI Agents)

When pub/sub is enabled, two additional MCP tools become available on the Streamable HTTP MCP endpoint (serve --mcp-http). The stdio server (pgvis mcp) does not offer them.

MCP calls carry no JWT, so these tools run as the anon_role with no claims, and are refused when jwt_secret is set without an anon_role — the same rule as every other MCP tool.

pubsub_publish

Publish a message to a channel. Goes through the same validation, authorize_function check (op = 'publish') and caller-role pg_notify as POST /pubsub/{channel}. Refused when the server is read-only.

Parameter	Type	Required	Description
channel	string	✓	Target channel name
payload	string	✓	Message content (max 7500 bytes)

Example tool call:

{
  "name": "pubsub_publish",
  "arguments": {
    "channel": "alerts",
    "payload": "{\"level\":\"warning\",\"message\":\"disk usage 90%\"}"
  }
}

Response:

{
  "content": [
    { "type": "text", "text": "{\"ok\":true,\"channel\":\"alerts\",\"payload_bytes\":46}" }
  ]
}

pubsub_channels

List the channels this instance is currently listening on (channels with at least one local subscriber).

Parameter	Type	Required	Description
(none)

Response:

{
  "content": [
    {
      "type": "text",
      "text": "{\"channels\":[\"alerts\"],\"count\":1}"
    }
  ]
}

Agent Use Cases

AI agents can use pub/sub to:

• Notify external systems when they complete a task
• Coordinate multi-agent workflows by publishing status updates
• Trigger webhooks by publishing to channels that other services monitor
• Stream progress to a dashboard while processing long-running queries

Embedding in Rust

When embedding pgvis in your Rust application, the PubSubHub is available on the Components struct.

The hub is a trusted, in-process API. Channel validation and allowed_channels still apply, but hub.publish() sends NOTIFY as the DSN’s role with no authorize_function check, and hub.subscribe() skips authorize_function and the per-identity cap (only max_subscribers applies; use subscribe_as(channel, Some(identity)) / unsubscribe_as to count against max_subscribers_per_identity). Authorize your own callers before using it on their behalf.

Basic Setup

use pgvis_lib::Builder;
use pgvis_lib::pgvis_core::pubsub::PubSubConfig;
use pgvis_lib::pgvis_core::config::Config;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let config = Config {
        anon_role: Some("web_anon".into()),
        pubsub: PubSubConfig {
            enabled: true,
            channel_prefix: "myapp:".to_string(),
            ..Default::default()
        },
        ..Default::default()
    };

    let components = Builder::new("postgres://localhost/mydb")
        .config(config)
        .build_components()
        .await?;

    // The pub/sub hub is available for direct use
    if let Some(hub) = &components.pubsub {
        // Publish a message
        hub.publish("events", r#"{"type":"startup"}"#).await?;

        // Subscribe to a channel (returns a broadcast::Receiver)
        let mut rx = hub.subscribe("events").await?;

        // Receive messages
        tokio::spawn(async move {
            while let Ok(msg) = rx.recv().await {
                println!("Got: {} on {}", msg.payload, msg.channel);
            }
        });

        // Check status
        let status = hub.status().await;
        println!("Active channels: {}", status.channels.len());
    }

    // Serve the router (includes /pubsub endpoints)
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    axum::serve(listener, components.router).await?;
    Ok(())
}

Using the Backend Directly

For lower-level control, use PgPubSub directly without the hub:

use futures::StreamExt;
use pgvis_lib::pgvis_postgres::PgPubSub;
use pgvis_lib::pgvis_core::pubsub::{PubSubBackend, PubSubConfig};
use pgvis_lib::pgvis_core::config::PoolConfig;

let config = PubSubConfig::default();
let pool_config = PoolConfig::default();

let backend = PgPubSub::new(
    "postgres://localhost/mydb",
    &config,
    &pool_config,
)?;

// Listen to a channel at the Postgres level
backend.listen("my_channel").await?;

// Get a stream of notifications (all listened channels)
let mut stream = backend.notification_stream().await?;

// Publish (sends NOTIFY via a pooled connection, as the DSN's role)
backend.publish("my_channel", "hello").await?;

if let Some(msg) = stream.next().await {
    println!("{}: {}", msg.channel, msg.payload);
}

// Clean up
backend.unlisten("my_channel").await?;

Combining with Custom axum Routes

use pgvis_lib::Builder;
use pgvis_lib::pgvis_core::config::Config;
use pgvis_lib::pgvis_core::pubsub::PubSubConfig;
use axum::{Router, routing::post, extract::State, Json};
use std::sync::Arc;

#[derive(Clone)]
struct AppState {
    hub: Arc<pgvis_lib::pgvis_router::PubSubHub>,
}

async fn notify_users(
    State(state): State<AppState>,
    Json(payload): Json<serde_json::Value>,
) -> &'static str {
    let msg = serde_json::to_string(&payload).unwrap();
    state.hub.publish("user_updates", &msg).await.ok();
    "notified"
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let config = Config {
        anon_role: Some("web_anon".into()),
        pubsub: PubSubConfig { enabled: true, ..Default::default() },
        ..Default::default()
    };

    let components = Builder::new("postgres://localhost/mydb")
        .config(config)
        .build_components()
        .await?;

    let hub = components.pubsub.clone().expect("pubsub enabled");

    // Add custom routes that publish to pub/sub
    let custom_routes = Router::new()
        .route("/notify", post(notify_users))
        .with_state(AppState { hub });

    let app = components.router.merge(custom_routes);

    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

Configuration Reference

Field	Type	Default	Description
enabled	bool	false	Enable the pub/sub subsystem
channel_prefix	string	"pgvis:"	Prefix added to Postgres channel names
max_payload_bytes	usize	7500	Maximum payload size (Postgres limit is ~8000)
channel_buffer_size	usize	64	In-process broadcast buffer per channel
allowed_channels	string[]	[]	Channel allowlist (empty = allow all). Supports glob patterns: chat.*, events.*
authorize_function	string	none	SQL function fn(channel text, op text) returns boolean that authorizes each subscribe/publish as the caller’s role
max_subscribers	usize	1000	Maximum concurrent SSE subscribers on this instance
max_subscribers_per_identity	usize	100	Maximum concurrent SSE subscribers per JWT role + sub (0 = no limit); callers without sub are bounded by max_subscribers only
keepalive_interval_secs	u64	15	Interval between SSE : keepalive comments
reconnect_base_ms	u64	500	Listener reconnect backoff base
reconnect_max_ms	u64	30000	Listener reconnect backoff cap

Environment Variables

Variable	Maps to
PGVIS_PUBSUB_ENABLED	pubsub.enabled
PGVIS_PUBSUB_CHANNEL_PREFIX	pubsub.channel_prefix
PGVIS_PUBSUB__<FIELD>	Any pubsub.<field>, e.g. PGVIS_PUBSUB__AUTHORIZE_FUNCTION, PGVIS_PUBSUB__MAX_SUBSCRIBERS

CLI Flags

--pubsub-enabled             Enable pub/sub (default: false)
--pubsub-channel-prefix      Postgres channel prefix (default: "pgvis:")

How It Works

Publishing and delivery take different paths. REST and MCP publishes run through Backend::execute like any data request, so pg_notify executes as the caller’s role. Delivery comes back through PgPubSub’s dedicated LISTEN connection: the hub issues LISTEN for a channel when its first local subscriber joins, and UNLISTEN when the last one leaves.

flowchart TB
    pub(["REST or MCP publisher"]) --> exec["<b>Backend::execute</b><br/>authorize + pg_notify as caller"]
    app(["Embedded Rust app"]) -.->|"hub.publish / subscribe"| hub
    exec --> pg[("Postgres<br/>LISTEN / NOTIFY")]
    pg -.-> others(["Other pgvis instances<br/>on the same database"])
    pg -->|notification| listener["<b>PgPubSub</b><br/>dedicated LISTEN connection"]
    listener --> hub["<b>PubSubHub</b><br/>per-channel broadcast"]
    hub -.->|"LISTEN on first subscriber"| listener
    hub --> sse(["SSE subscribers"])

The sequence below follows one subscriber and one publisher on the same channel, with authorize_function configured:

sequenceDiagram
    participant S as Subscriber
    participant P as Publisher
    participant G as pgvis
    participant D as Postgres
    S->>G: GET /pubsub/chat
    G->>G: verify JWT + channel
    G->>D: authorize(subscribe) as caller
    D-->>G: true
    G->>D: LISTEN, if first subscriber
    G-->>S: 200 text/event-stream
    P->>G: POST /pubsub/chat
    G->>D: authorize + pg_notify
    D-->>G: ok
    G-->>P: 200 {"ok": true}
    D-->>G: notification
    G-->>S: event: message

1. Publish: REST and MCP validate the channel and payload, then run authorize_function (if set) and pg_notify in one statement as the caller’s role; anything but true is a 403. Embedded code calling hub.publish(channel, payload) validates and sends NOTIFY as the DSN’s role
2. Delivery: Postgres broadcasts the notification to ALL connected listeners (including other pgvis instances)
3. Fan-out: The dedicated listener task receives the notification and broadcasts it to all local subscribers via tokio::sync::broadcast
4. Subscribe: REST streams SSE frames; embedded code receives from the broadcast channel. MCP has no subscribe tool

Limitations

Constraint	Value	Reason
Max payload	~7,500 bytes	Postgres NOTIFY limit is 8,000 bytes including channel name
Max channel name	63 bytes including channel_prefix	Postgres identifier length limit
Delivery guarantee	At-most-once	No persistence; missed during disconnection
Ordering	Per-connection FIFO	Postgres guarantees order within a single session
No history	—	New subscribers only see future messages
Postgres only	—	SQLite backends do not support pub/sub

Patterns and Recipes

Chat Room

# Terminal 1: Subscribe
curl -N http://localhost:3000/pubsub/room.general

# Terminal 2: Publish
curl -X POST http://localhost:3000/pubsub/room.general \
  -d '{"user":"bob","text":"hey everyone"}'

Event Bus Between Microservices

# Service A publishes order events
curl -X POST http://localhost:3000/pubsub/orders.created \
  -d '{"order_id":123,"total":49.99}'

# Service B subscribes to order events (using EventSource in JS)
const es = new EventSource("http://localhost:3000/pubsub/orders.created");
es.addEventListener("message", (e) => {
  const order = JSON.parse(JSON.parse(e.data).payload);
  console.log("New order:", order.order_id);
});

Channel Allowlist (Security)

Restrict which channels clients can use:

[pubsub]
enabled = true
allowed_channels = ["chat.*", "notifications", "events.*"]

Only channels matching these patterns will be accepted (* matches any sequence of characters). Attempts to subscribe or publish to other channels return 403 Forbidden with PGVIS_PUBSUB_CHANNEL_DENIED.

Per-Channel Authorization (Security)

The allowlist applies to every caller alike. To decide per caller, let the database decide, as it does for tables:

[pubsub]
authorize_function = "auth.pubsub_authorize"

CREATE FUNCTION auth.pubsub_authorize(channel text, op text) RETURNS boolean
LANGUAGE sql STABLE AS $$
  SELECT channel = 'user.' || (nullif(current_setting('request.jwt.claims', true), '')::json->>'sub')
$$;

pgvis calls it with op = 'subscribe' or 'publish' in a transaction running as the caller’s role with their JWT claims set, exactly like a data API request. Anything but true (false, NULL, or an error) returns 403 Forbidden. Publishing always runs pg_notify as the caller’s role, in the same statement as the check. Subscriptions are authorized once, when the stream opens.

The caller’s role needs EXECUTE on the function (granted to PUBLIC by default); a permission error counts as a denial. MCP’s pubsub_publish goes through the same check, as the anon_role with no claims. The embedded PubSubHub API does not.

Without authorize_function, any caller that passes JWT verification may subscribe to or publish on every allowed channel; publishes still run pg_notify as the caller’s role.

Subscriber Limits

max_subscribers caps concurrent SSE streams on the instance. max_subscribers_per_identity additionally caps each identity — the verified role plus the JWT sub claim — so one user can’t take every slot. Callers without a sub (including anonymous ones) can’t be told apart, so they are bounded only by max_subscribers rather than sharing one per-identity pool. Either limit returns 503 with PGVIS_PUBSUB_MAX_SUBSCRIBERS; slots are released when the stream closes.

Cross-Instance Coordination

Since all pgvis instances share the same Postgres database, a message published on one instance is received by subscribers on ALL instances:

Instance A (port 3001): POST /pubsub/sync → "reload config"
Instance B (port 3002): GET /pubsub/sync → receives "reload config"
Instance C (port 3003): GET /pubsub/sync → receives "reload config"

This happens automatically — no additional clustering configuration needed.