systemprompt_cloud/api_client/
streams.rs

1//! Server-Sent Events (SSE) subscriptions for tenant provisioning
2//! and Paddle checkout pipelines.
3//!
4//! Copyright (c) systemprompt.io — Business Source License 1.1.
5//! See <https://systemprompt.io> for licensing details.
6
7use std::pin::Pin;
8
9use futures::stream::{Stream, StreamExt};
10use reqwest_eventsource::{Event, EventSource};
11use systemprompt_identifiers::TenantId;
12use systemprompt_models::modules::ApiPaths;
13
14use super::CloudApiClient;
15use super::types::{CheckoutEvent, ProvisioningEvent};
16use crate::error::{CloudError, CloudResult};
17
18impl CloudApiClient {
19    pub fn subscribe_provisioning_events(
20        &self,
21        tenant_id: &TenantId,
22    ) -> Pin<Box<dyn Stream<Item = CloudResult<ProvisioningEvent>> + Send + '_>> {
23        let url = format!("{}{}", self.api_url(), ApiPaths::tenant_events(tenant_id));
24        let token = self.token().to_owned();
25        let client = self.client.clone();
26
27        let stream = async_stream::stream! {
28            let request = client
29                .get(&url)
30                .header("Authorization", format!("Bearer {}", token))
31                .header("Accept", "text/event-stream");
32
33            let mut es = match EventSource::new(request) {
34                Ok(es) => es,
35                Err(e) => {
36                    yield Err(CloudError::SseStream { message: format!("Failed to create SSE connection: {e}") });
37                    return;
38                }
39            };
40
41            while let Some(event) = es.next().await {
42                match event {
43                    Ok(Event::Open) => {
44                        tracing::debug!("SSE connection opened");
45                    }
46                    Ok(Event::Message(message)) => {
47                        if message.event == "provisioning" || message.event == "message" {
48                            match serde_json::from_str::<ProvisioningEvent>(&message.data) {
49                                Ok(event) => yield Ok(event),
50                                Err(e) => {
51                                    tracing::warn!(error = %e, data = %message.data, "Failed to parse SSE event");
52                                }
53                            }
54                        } else if message.event == "heartbeat" {
55                            tracing::trace!("SSE heartbeat received");
56                        }
57                    }
58                    Err(reqwest_eventsource::Error::StreamEnded) => {
59                        tracing::debug!("SSE stream ended normally");
60                        break;
61                    }
62                    Err(e) => {
63                        tracing::warn!(error = %e, "SSE stream error");
64                        yield Err(CloudError::SseStream { message: e.to_string() });
65                        break;
66                    }
67                }
68            }
69        };
70
71        Box::pin(stream)
72    }
73
74    pub fn subscribe_checkout_events(
75        &self,
76        checkout_session_id: &systemprompt_identifiers::CheckoutSessionId,
77    ) -> Pin<Box<dyn Stream<Item = CloudResult<CheckoutEvent>> + Send + '_>> {
78        let url = format!(
79            "{}/api/v1/checkout/{}/events",
80            self.api_url(),
81            checkout_session_id.as_str()
82        );
83        let token = self.token().to_owned();
84        let client = self.client.clone();
85
86        let stream = async_stream::stream! {
87            tracing::debug!(url = %url, "Building SSE request");
88            let request = client
89                .get(&url)
90                .header("Authorization", format!("Bearer {}", token))
91                .header("Accept", "text/event-stream");
92
93            let mut es = match EventSource::new(request) {
94                Ok(es) => es,
95                Err(e) => {
96                    tracing::error!(error = %e, "Failed to create EventSource");
97                    yield Err(CloudError::SseStream { message: format!("Failed to create SSE connection: {e}") });
98                    return;
99                }
100            };
101
102            while let Some(event) = es.next().await {
103                match event {
104                    Ok(Event::Open) => {
105                        tracing::debug!("SSE connection opened");
106                    }
107                    Ok(Event::Message(message)) => {
108                        tracing::debug!(event_type = %message.event, "SSE message received");
109                        if message.event == "provisioning" {
110                            match serde_json::from_str::<CheckoutEvent>(&message.data) {
111                                Ok(event) => yield Ok(event),
112                                Err(e) => {
113                                    tracing::warn!(error = %e, "Failed to parse checkout event");
114                                }
115                            }
116                        }
117                    }
118                    Err(reqwest_eventsource::Error::StreamEnded) => {
119                        tracing::debug!("SSE stream ended");
120                        break;
121                    }
122                    Err(e) => {
123                        tracing::warn!(error = %e, "SSE stream error");
124                        yield Err(CloudError::SseStream { message: e.to_string() });
125                        break;
126                    }
127                }
128            }
129        };
130
131        Box::pin(stream)
132    }
133}
systemprompt_cloud/api_client/streams.rs

systemprompt_cloud/api_client/
streams.rs
