postrust_graphql/subscription/
mod.rs

1//! GraphQL subscriptions using PostgreSQL LISTEN/NOTIFY.
2//!
3//! This module provides realtime data synchronization through GraphQL subscriptions,
4//! powered by PostgreSQL's native LISTEN/NOTIFY mechanism.
5//!
6//! ## Architecture
7//!
8//! ```text
9//! ┌──────────────┐   WebSocket   ┌──────────────┐   LISTEN/NOTIFY   ┌────────────┐
10//! │   Client     │◀────────────▶│   Postrust   │◀────────────────▶│  PostgreSQL│
11//! │  (Browser)   │              │   Server     │                   │  Database  │
12//! └──────────────┘              └──────────────┘                   └────────────┘
13//! ```
14//!
15//! ## Usage
16//!
17//! 1. Create notification triggers on your tables using [`broker::create_notify_trigger_sql`]
18//! 2. Start the [`NotifyBroker`] to listen for database notifications
19//! 3. Use GraphQL subscriptions to receive realtime updates
20//!
21//! ## Example
22//!
23//! ```graphql
24//! subscription {
25//!   users {
26//!     id
27//!     name
28//!     email
29//!   }
30//! }
31//! ```
32
33pub mod broker;
34
35pub use broker::{
36    create_notify_trigger_sql, drop_notify_trigger_sql, table_channel_name, BrokerError,
37    NotifyBroker, PgNotification,
38};
39
40use crate::schema::GeneratedSchema;
41use postrust_core::schema_cache::SchemaCache;
42use std::sync::atomic::{AtomicI64, Ordering};
43
44/// Live queries open right now, across every schema this process has built.
45///
46/// Process-wide because a subscription outlives the schema it was opened on:
47/// a reload swaps the schema and the subscription carries on, so a count kept
48/// on the state would lose it. One increment when a subscription opens and
49/// one decrement when its stream is dropped, however it ends -- the client
50/// closing, an error, the connection going -- so it is never on a request's
51/// path.
52static ACTIVE_SUBSCRIPTIONS: AtomicI64 = AtomicI64::new(0);
53
54/// How many GraphQL subscriptions are open, for a host's metrics.
55pub fn active_subscriptions() -> i64 {
56    ACTIVE_SUBSCRIPTIONS.load(Ordering::Relaxed)
57}
58
59/// Held by an open subscription; dropping it closes the count.
60pub(crate) struct ActiveSubscription(());
61
62impl ActiveSubscription {
63    pub(crate) fn open() -> Self {
64        ACTIVE_SUBSCRIPTIONS.fetch_add(1, Ordering::Relaxed);
65        Self(())
66    }
67}
68
69impl Drop for ActiveSubscription {
70    fn drop(&mut self) {
71        ACTIVE_SUBSCRIPTIONS.fetch_sub(1, Ordering::Relaxed);
72    }
73}
74
75/// A subscription field in the GraphQL schema.
76#[derive(Debug, Clone)]
77pub struct SubscriptionField {
78    /// The GraphQL field name (e.g., "users", "orders")
79    pub name: String,
80    /// The source table name
81    pub table_name: String,
82    /// The source schema name
83    pub schema_name: String,
84    /// The return type (e.g., "Users", "Orders")
85    pub return_type: String,
86    /// Description for documentation
87    pub description: Option<String>,
88}
89
90impl SubscriptionField {
91    /// Create a new subscription field for a table.
92    pub fn for_table(schema: &str, table: &str, type_name: &str) -> Self {
93        Self {
94            name: table.to_string(),
95            table_name: table.to_string(),
96            schema_name: schema.to_string(),
97            return_type: type_name.to_string(),
98            description: Some(format!("Subscribe to changes on the {} table", table)),
99        }
100    }
101
102    /// Get the PostgreSQL channel name for this subscription.
103    pub fn channel_name(&self) -> String {
104        table_channel_name(&self.schema_name, &self.table_name)
105    }
106}
107
108/// Generate subscription fields for all tables in the schema.
109pub fn generate_subscription_fields(
110    _schema_cache: &SchemaCache,
111    generated: &GeneratedSchema,
112) -> Vec<SubscriptionField> {
113    let mut fields = Vec::new();
114
115    for (type_name, obj_type) in &generated.object_types {
116        let table = &obj_type.table;
117
118        // Only create subscriptions for tables, not views (views can be added later)
119        if !table.is_view {
120            fields.push(SubscriptionField::for_table(
121                &table.schema,
122                &table.name,
123                type_name,
124            ));
125        }
126    }
127
128    fields
129}
130
131/// Payload structure for table change notifications.
132#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
133pub struct TableChangePayload {
134    /// The operation type: INSERT, UPDATE, or DELETE
135    pub operation: String,
136    /// The table name
137    pub table: String,
138    /// The schema name
139    pub schema: String,
140    /// The old row data (for UPDATE and DELETE)
141    #[serde(skip_serializing_if = "Option::is_none")]
142    pub old: Option<serde_json::Value>,
143    /// The new row data (for INSERT and UPDATE)
144    #[serde(skip_serializing_if = "Option::is_none")]
145    pub new: Option<serde_json::Value>,
146}
147
148impl TableChangePayload {
149    /// Parse a notification payload.
150    pub fn from_payload(payload: &str) -> Result<Self, serde_json::Error> {
151        serde_json::from_str(payload)
152    }
153
154    /// Get the data to return to the client.
155    ///
156    /// For INSERT and UPDATE, returns the new row.
157    /// For DELETE, returns the old row.
158    pub fn data(&self) -> Option<&serde_json::Value> {
159        match self.operation.as_str() {
160            "DELETE" => self.old.as_ref(),
161            _ => self.new.as_ref(),
162        }
163    }
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    #[test]
171    fn test_subscription_field_channel_name() {
172        let field = SubscriptionField::for_table("public", "users", "Users");
173        assert_eq!(field.channel_name(), "postrust_public_users");
174    }
175
176    #[test]
177    fn test_table_change_payload_parsing() {
178        let json = r#"{
179            "operation": "INSERT",
180            "table": "users",
181            "schema": "public",
182            "new": {"id": 1, "name": "Alice"}
183        }"#;
184
185        let payload = TableChangePayload::from_payload(json).unwrap();
186        assert_eq!(payload.operation, "INSERT");
187        assert_eq!(payload.table, "users");
188        assert!(payload.new.is_some());
189        assert!(payload.old.is_none());
190    }
191
192    #[test]
193    fn test_table_change_payload_data() {
194        let insert_payload = TableChangePayload {
195            operation: "INSERT".to_string(),
196            table: "users".to_string(),
197            schema: "public".to_string(),
198            old: None,
199            new: Some(serde_json::json!({"id": 1})),
200        };
201        assert!(insert_payload.data().is_some());
202
203        let delete_payload = TableChangePayload {
204            operation: "DELETE".to_string(),
205            table: "users".to_string(),
206            schema: "public".to_string(),
207            old: Some(serde_json::json!({"id": 1})),
208            new: None,
209        };
210        assert!(delete_payload.data().is_some());
211    }
212}
postrust_graphql/subscription/mod.rs

postrust_graphql/subscription/
mod.rs
