fraiseql_core/runtime/subscription/
mod.rs

1//! Subscription runtime for event-driven GraphQL subscriptions.
2//!
3//! FraiseQL subscriptions are **compiled projections of database events**, not
4//! traditional resolver-based subscriptions. Events originate from database
5//! transactions (via LISTEN/NOTIFY or CDC) and are delivered through transport
6//! adapters.
7//!
8//! # Architecture
9//!
10//! ```text
11//! Database Transaction (INSERT/UPDATE/DELETE)
12//!     ↓ (commits)
13//! LISTEN/NOTIFY (PostgreSQL)
14//!     ↓
15//! SubscriptionManager (event routing)
16//!     ↓
17//! SubscriptionMatcher (filter evaluation)
18//!     ↓ (parallel delivery)
19//! ├─ graphql-ws Adapter (WebSocket)
20//! ├─ Webhook Adapter (HTTP POST)
21//! └─ Kafka Adapter (event streaming)
22//! ```
23//!
24//! # Example
25//!
26//! ```text
27//! // Illustrative — subscription infrastructure requires a live schema + transport.
28//! // Use SubscriptionManager::new(Arc::new(schema)) for the full API.
29//!
30//! // Create subscription manager
31//! let manager = SubscriptionManager::new(Arc::new(schema));
32//!
33//! // Subscribe to events (synchronous, not async)
34//! let subscription_id = manager.subscribe(
35//!     "OrderCreated",
36//!     user_context_json,
37//!     variables_json,
38//!     "connection-id",
39//! )?;
40//!
41//! // Receive events via broadcast channel
42//! let mut receiver = manager.receiver();
43//! while let Ok(payload) = receiver.recv().await {
44//!     if payload.subscription_id == subscription_id {
45//!         // Deliver to client
46//!     }
47//! }
48//!
49//! // Unsubscribe (synchronous)
50//! manager.unsubscribe(subscription_id)?;
51//! ```
52
53use thiserror::Error;
54
55mod kafka;
56mod manager;
57pub mod protocol;
58#[cfg(test)]
59mod tests;
60mod transport;
61mod types;
62mod webhook;
63
64pub use kafka::{KafkaAdapter, KafkaConfig, KafkaMessage};
65pub use manager::SubscriptionManager;
66pub use transport::{BoxDynTransportAdapter, DeliveryResult, TransportAdapter, TransportManager};
67pub use types::{
68    ActiveSubscription, ChangeSpineEnvelope, SubscriptionEvent, SubscriptionId,
69    SubscriptionOperation, SubscriptionPayload,
70};
71pub use webhook::{WebhookAdapter, WebhookPayload, WebhookTransportConfig};
72/// Backward-compatible type alias — use [`WebhookTransportConfig`] in new code.
73pub type WebhookConfig = WebhookTransportConfig;
74
75/// Extract `(field, value)` equality conditions from an RLS `WhereClause`.
76///
77/// Walks the clause tree and collects all `Field { op: Eq }` nodes.
78/// `And` nodes are recursively flattened. Other clause types (nested `Or`,
79/// non-Eq operators) are ignored — they cannot be represented as simple
80/// field-value pairs and will not filter subscription events.
81///
82/// The caller should evaluate the RLS policy at subscribe time and pass
83/// the result through this function to produce conditions for
84/// [`ActiveSubscription::with_rls_conditions`].
85///
86/// # Errors
87///
88/// This function is infallible — unsupported clause shapes are silently
89/// skipped, which is a safe default (fewer conditions = more events
90/// delivered, never fewer).
91#[must_use]
92pub fn extract_rls_conditions(clause: &crate::db::WhereClause) -> Vec<(String, serde_json::Value)> {
93    let mut conditions = Vec::new();
94    collect_eq_conditions(clause, &mut conditions);
95    conditions
96}
97
98fn collect_eq_conditions(
99    clause: &crate::db::WhereClause,
100    out: &mut Vec<(String, serde_json::Value)>,
101) {
102    use crate::db::{WhereClause, WhereOperator};
103    match clause {
104        WhereClause::Field {
105            path,
106            operator: WhereOperator::Eq,
107            value,
108        } => {
109            // Use the last path component as the field name (e.g., ["tenant_id"] → "tenant_id")
110            if let Some(field) = path.last() {
111                out.push((field.clone(), value.clone()));
112            }
113        },
114        WhereClause::And(clauses) => {
115            for c in clauses {
116                collect_eq_conditions(c, out);
117            }
118        },
119        _ => {
120            // Or, Not, non-Eq operators — cannot be represented as simple field-value pairs.
121            // Safe default: skip (delivers more events, never fewer).
122        },
123    }
124}
125
126// =============================================================================
127// Error Types
128// =============================================================================
129
130/// Errors that can occur during subscription operations.
131#[non_exhaustive]
132#[derive(Debug, Error)]
133pub enum SubscriptionError {
134    /// Subscription type not found in schema.
135    #[error("Subscription not found: {0}")]
136    SubscriptionNotFound(String),
137
138    /// Authentication required for subscription.
139    #[error("Authentication required for subscription: {0}")]
140    AuthenticationRequired(String),
141
142    /// User not authorized for subscription.
143    #[error("Not authorized for subscription: {0}")]
144    Forbidden(String),
145
146    /// Invalid subscription variables.
147    #[error("Invalid subscription variables: {0}")]
148    InvalidVariables(String),
149
150    /// Subscription already exists.
151    #[error("Subscription already exists: {0}")]
152    AlreadyExists(String),
153
154    /// Subscription not active.
155    #[error("Subscription not active: {0}")]
156    NotActive(String),
157
158    /// Internal subscription error.
159    #[error("Subscription error: {0}")]
160    Internal(String),
161
162    /// Channel send error.
163    #[error("Failed to send event: {0}")]
164    SendError(String),
165
166    /// Database connection error.
167    #[error("Database connection error: {0}")]
168    DatabaseConnection(String),
169
170    /// Listener already running.
171    #[error("Listener already running")]
172    ListenerAlreadyRunning,
173
174    /// Listener not running.
175    #[error("Listener not running")]
176    ListenerNotRunning,
177
178    /// Failed to parse notification payload.
179    #[error("Failed to parse notification: {0}")]
180    InvalidNotification(String),
181
182    /// Failed to deliver event to transport.
183    #[error("Failed to deliver to {transport}: {reason}")]
184    DeliveryFailed {
185        /// Transport that failed.
186        transport: String,
187        /// Reason for failure.
188        reason:    String,
189    },
190}
fraiseql_core/runtime/subscription/mod.rs

fraiseql_core/runtime/subscription/
mod.rs
