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;
42
43/// A subscription field in the GraphQL schema.
44#[derive(Debug, Clone)]
45pub struct SubscriptionField {
46    /// The GraphQL field name (e.g., "users", "orders")
47    pub name: String,
48    /// The source table name
49    pub table_name: String,
50    /// The source schema name
51    pub schema_name: String,
52    /// The return type (e.g., "Users", "Orders")
53    pub return_type: String,
54    /// Description for documentation
55    pub description: Option<String>,
56}
57
58impl SubscriptionField {
59    /// Create a new subscription field for a table.
60    pub fn for_table(schema: &str, table: &str, type_name: &str) -> Self {
61        Self {
62            name: table.to_string(),
63            table_name: table.to_string(),
64            schema_name: schema.to_string(),
65            return_type: type_name.to_string(),
66            description: Some(format!("Subscribe to changes on the {} table", table)),
67        }
68    }
69
70    /// Get the PostgreSQL channel name for this subscription.
71    pub fn channel_name(&self) -> String {
72        table_channel_name(&self.schema_name, &self.table_name)
73    }
74}
75
76/// Generate subscription fields for all tables in the schema.
77pub fn generate_subscription_fields(
78    _schema_cache: &SchemaCache,
79    generated: &GeneratedSchema,
80) -> Vec<SubscriptionField> {
81    let mut fields = Vec::new();
82
83    for (type_name, obj_type) in &generated.object_types {
84        let table = &obj_type.table;
85
86        // Only create subscriptions for tables, not views (views can be added later)
87        if !table.is_view {
88            fields.push(SubscriptionField::for_table(
89                &table.schema,
90                &table.name,
91                type_name,
92            ));
93        }
94    }
95
96    fields
97}
98
99/// Payload structure for table change notifications.
100#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
101pub struct TableChangePayload {
102    /// The operation type: INSERT, UPDATE, or DELETE
103    pub operation: String,
104    /// The table name
105    pub table: String,
106    /// The schema name
107    pub schema: String,
108    /// The old row data (for UPDATE and DELETE)
109    #[serde(skip_serializing_if = "Option::is_none")]
110    pub old: Option<serde_json::Value>,
111    /// The new row data (for INSERT and UPDATE)
112    #[serde(skip_serializing_if = "Option::is_none")]
113    pub new: Option<serde_json::Value>,
114}
115
116impl TableChangePayload {
117    /// Parse a notification payload.
118    pub fn from_payload(payload: &str) -> Result<Self, serde_json::Error> {
119        serde_json::from_str(payload)
120    }
121
122    /// Get the data to return to the client.
123    ///
124    /// For INSERT and UPDATE, returns the new row.
125    /// For DELETE, returns the old row.
126    pub fn data(&self) -> Option<&serde_json::Value> {
127        match self.operation.as_str() {
128            "DELETE" => self.old.as_ref(),
129            _ => self.new.as_ref(),
130        }
131    }
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137
138    #[test]
139    fn test_subscription_field_channel_name() {
140        let field = SubscriptionField::for_table("public", "users", "Users");
141        assert_eq!(field.channel_name(), "postrust_public_users");
142    }
143
144    #[test]
145    fn test_table_change_payload_parsing() {
146        let json = r#"{
147            "operation": "INSERT",
148            "table": "users",
149            "schema": "public",
150            "new": {"id": 1, "name": "Alice"}
151        }"#;
152
153        let payload = TableChangePayload::from_payload(json).unwrap();
154        assert_eq!(payload.operation, "INSERT");
155        assert_eq!(payload.table, "users");
156        assert!(payload.new.is_some());
157        assert!(payload.old.is_none());
158    }
159
160    #[test]
161    fn test_table_change_payload_data() {
162        let insert_payload = TableChangePayload {
163            operation: "INSERT".to_string(),
164            table: "users".to_string(),
165            schema: "public".to_string(),
166            old: None,
167            new: Some(serde_json::json!({"id": 1})),
168        };
169        assert!(insert_payload.data().is_some());
170
171        let delete_payload = TableChangePayload {
172            operation: "DELETE".to_string(),
173            table: "users".to_string(),
174            schema: "public".to_string(),
175            old: Some(serde_json::json!({"id": 1})),
176            new: None,
177        };
178        assert!(delete_payload.data().is_some());
179    }
180}
postrust_graphql/subscription/mod.rs

postrust_graphql/subscription/
mod.rs
