force_pubsub/
handler.rs

1//! Pub/Sub API handler.
2
3use std::pin::Pin;
4use std::sync::Arc;
5use std::sync::Once;
6use tonic::transport::{Channel, ClientTlsConfig};
7
8/// Ensures rustls has a process-level [`CryptoProvider`](rustls::crypto::CryptoProvider).
9///
10/// The unified workspace dependency graph enables both the `ring` and `aws-lc-rs`
11/// rustls providers (the latter is pulled in transitively by the AWS SDK in
12/// `force-lake`), so rustls cannot automatically pick one and panics when tonic
13/// builds its TLS config. We install `ring` as the default once, deterministically.
14fn ensure_crypto_provider() {
15    static INSTALL: Once = Once::new();
16    INSTALL.call_once(|| {
17        // Ignore the error: another provider may already be installed, which is fine.
18        let _ = rustls::crypto::ring::default_provider().install_default();
19    });
20}
21
22use force::auth::Authenticator;
23use force::session::Session;
24use serde::Serialize;
25use serde::de::DeserializeOwned;
26use serde_json::Value;
27use tokio::sync::OnceCell;
28use tokio_stream::Stream;
29
30use crate::config::{PubSubConfig, ReplayPreset};
31use crate::error::{PubSubError, Result};
32use crate::interceptor;
33use crate::publish_sink::{PublishSink, open_publish_stream};
34use crate::publisher::publish_unary;
35use crate::schema_cache::SchemaCache;
36use crate::subscriber::{subscribe_dynamic, subscribe_typed_dynamic};
37use crate::types::{PubSubEvent, PublishResponse};
38
39use crate::proto::eventbus_v1::{SchemaRequest, TopicRequest, pub_sub_client::PubSubClient};
40
41/// JSON structure of Salesforce's `/services/oauth2/userinfo` response (relevant fields only).
42#[derive(serde::Deserialize)]
43struct UserInfo {
44    organization_id: String,
45}
46
47/// Fetch the 18-char org ID from the Salesforce userinfo endpoint.
48///
49/// Used by [`PubSubHandler::get_tenant_id`] as the initialiser for its [`OnceCell`].
50async fn fetch_tenant_id<A: Authenticator>(session: &Arc<Session<A>>) -> Result<String> {
51    let token = session.token_manager().token().await?;
52    let userinfo_url = format!("{}/services/oauth2/userinfo", token.instance_url());
53
54    let resp = reqwest::Client::new()
55        .get(&userinfo_url)
56        .bearer_auth(token.as_str())
57        .send()
58        .await
59        .map_err(|e| PubSubError::Config(format!("userinfo request failed: {e}")))?;
60
61    if !resp.status().is_success() {
62        return Err(PubSubError::Config(format!(
63            "userinfo returned status {}",
64            resp.status()
65        )));
66    }
67
68    let body = force::http::read_capped_body(resp, 1024 * 1024)
69        .await
70        .map_err(|e| PubSubError::Config(format!("userinfo parse failed: {e}")))?;
71
72    let info: UserInfo = serde_json::from_str(&body)
73        .map_err(|e| PubSubError::Config(format!("userinfo parse failed: {e}")))?;
74
75    Ok(info.organization_id)
76}
77
78/// Public-facing topic metadata (mirrors proto without leaking generated types).
79#[derive(Debug, Clone)]
80pub struct TopicInfo {
81    /// Topic name (e.g., `/event/MyEvent__e`).
82    pub topic_name: String,
83    /// Topic URI.
84    pub topic_uri: String,
85    /// Whether events can be published to this topic.
86    pub can_publish: bool,
87    /// Whether events can be subscribed to on this topic.
88    pub can_subscribe: bool,
89    /// Current Avro schema ID for this topic.
90    pub schema_id: String,
91}
92
93/// Public-facing schema metadata.
94#[derive(Debug, Clone)]
95pub struct SchemaInfo {
96    /// Schema ID.
97    pub schema_id: String,
98    /// Avro schema JSON string.
99    pub schema_json: String,
100}
101
102/// Entry point for all Salesforce Pub/Sub operations.
103///
104/// Obtained by calling [`PubSubHandler::connect`] with a [`Session`] and [`PubSubConfig`].
105/// The handler is cheaply cloneable — clones share the same gRPC channel and schema cache.
106#[derive(Clone)]
107pub struct PubSubHandler<A: Authenticator> {
108    pub(crate) session: Arc<Session<A>>,
109    /// Configuration for this handler, used by subscribe/publish operations in later tasks.
110    pub(crate) config: PubSubConfig,
111    /// Shared schema cache, populated during subscribe/publish operations in later tasks.
112    pub schema_cache: SchemaCache,
113    pub(crate) channel: Channel,
114    /// Lazily fetched org ID (18-char) from `/services/oauth2/userinfo`.
115    ///
116    /// Populated on the first call to [`Self::get_tenant_id`] and reused thereafter.
117    tenant_id: Arc<OnceCell<String>>,
118}
119
120impl<A: Authenticator> PubSubHandler<A> {
121    /// Connect to the Pub/Sub gRPC endpoint and return a handler.
122    ///
123    /// Validates configuration and establishes the gRPC channel. This is async
124    /// because channel creation involves a DNS lookup and TLS handshake.
125    ///
126    /// # Errors
127    ///
128    /// Returns `PubSubError::Config` if `batch_size` is out of range (1–100).
129    /// Returns `PubSubError::Connect` if the gRPC channel cannot be established.
130    pub async fn connect(session: Arc<Session<A>>, config: PubSubConfig) -> Result<Self> {
131        if config.batch_size < 1 || config.batch_size > 100 {
132            return Err(PubSubError::Config(
133                "batch_size must be between 1 and 100".to_string(),
134            ));
135        }
136
137        let endpoint = Channel::from_shared(config.endpoint.clone())
138            .map_err(|e| PubSubError::Config(format!("invalid endpoint: {e}")))?;
139        let endpoint = if endpoint.uri().scheme_str() == Some("https") {
140            ensure_crypto_provider();
141            endpoint.tls_config(ClientTlsConfig::new().with_webpki_roots())?
142        } else {
143            endpoint
144        };
145        let channel = endpoint.connect().await?;
146
147        Ok(Self {
148            session,
149            config,
150            schema_cache: SchemaCache::new(),
151            channel,
152            tenant_id: Arc::new(OnceCell::new()),
153        })
154    }
155
156    /// Build a gRPC client for each call (channels are cheap to clone).
157    fn grpc_client(&self) -> PubSubClient<Channel> {
158        PubSubClient::new(self.channel.clone())
159    }
160
161    /// Fetch the org's 18-char tenant ID from `/services/oauth2/userinfo`, caching the result.
162    ///
163    /// Salesforce's userinfo response includes `organization_id` which is the 18-char org ID
164    /// required as the `tenantid` gRPC metadata header.
165    ///
166    /// # Errors
167    ///
168    /// Returns [`PubSubError::Config`] if the userinfo endpoint cannot be reached or the
169    /// response does not contain a valid `organization_id` field.
170    pub(crate) async fn get_tenant_id(&self) -> Result<&str> {
171        self.tenant_id
172            .get_or_try_init(|| fetch_tenant_id(&self.session))
173            .await
174            .map(String::as_str)
175    }
176
177    /// Build a tonic request with all three required Pub/Sub auth headers.
178    ///
179    /// Fetches a fresh token and (lazily) the tenant ID, then delegates to
180    /// [`crate::interceptor::build_metadata`].
181    async fn auth_request<T>(&self, message: T) -> Result<tonic::Request<T>> {
182        let token = self.session.token_manager().token().await?;
183        let tenant_id = self.get_tenant_id().await?.to_string();
184        let meta = interceptor::build_metadata(&token, token.instance_url(), &tenant_id)?;
185        let mut req = tonic::Request::new(message);
186        *req.metadata_mut() = meta;
187        Ok(req)
188    }
189
190    /// Fetch metadata about a Pub/Sub topic.
191    ///
192    /// # Errors
193    ///
194    /// Returns `PubSubError::Transport` if the gRPC call fails.
195    pub async fn get_topic(&self, topic_name: &str) -> Result<TopicInfo> {
196        let req = self
197            .auth_request(TopicRequest {
198                topic_name: topic_name.to_string(),
199            })
200            .await?;
201
202        let resp = self.grpc_client().get_topic(req).await?;
203        let info = resp.into_inner();
204        Ok(TopicInfo {
205            topic_name: info.topic_name,
206            topic_uri: info.topic_uri,
207            can_publish: info.can_publish,
208            can_subscribe: info.can_subscribe,
209            schema_id: info.schema_id,
210        })
211    }
212
213    /// Fetch an Avro schema by its ID.
214    ///
215    /// Results are **not** automatically cached here — call [`SchemaCache::parse_and_insert`]
216    /// with the returned `schema_json` to cache it.
217    ///
218    /// # Errors
219    ///
220    /// Returns `PubSubError::Transport` if the gRPC call fails (including schema not found).
221    pub async fn get_schema(&self, schema_id: &str) -> Result<SchemaInfo> {
222        let req = self
223            .auth_request(SchemaRequest {
224                schema_id: schema_id.to_string(),
225            })
226            .await?;
227
228        let resp = self.grpc_client().get_schema(req).await?;
229        let info = resp.into_inner();
230        Ok(SchemaInfo {
231            schema_id: info.schema_id,
232            schema_json: info.schema_json,
233        })
234    }
235}
236
237impl<A: Authenticator + Send + Sync + 'static> PubSubHandler<A> {
238    /// Publish events to a topic via the unary Publish RPC.
239    ///
240    /// Automatically resolves the Avro schema for `topic` by calling `GetTopic`
241    /// to obtain the schema ID, then fetching the schema via `GetSchema` if it
242    /// is not already cached.
243    ///
244    /// # Errors
245    ///
246    /// Returns `PubSubError::Transport` if the `GetTopic` or `GetSchema` RPC fails.
247    /// Returns `PubSubError::Avro` if encoding fails.
248    pub async fn publish<T: Serialize + Send>(
249        &self,
250        topic: &str,
251        events: Vec<T>,
252    ) -> Result<PublishResponse> {
253        let topic_info = self.get_topic(topic).await?;
254        let schema_id = &topic_info.schema_id;
255
256        // Ensure the schema is in the cache (fetches from GetSchema if not).
257        let token = self.session.token_manager().token().await?;
258        let tenant_id = self.get_tenant_id().await?.to_string();
259        let meta = interceptor::build_metadata(&token, token.instance_url(), &tenant_id)?;
260        self.schema_cache
261            .get_or_fetch(schema_id, &self.channel, meta)
262            .await?;
263
264        publish_unary(
265            &self.session,
266            &self.channel,
267            &self.schema_cache,
268            schema_id,
269            topic,
270            events,
271            &tenant_id,
272        )
273        .await
274    }
275
276    /// Subscribe to a topic, yielding decoded events as [`serde_json::Value`].
277    ///
278    /// The returned stream emits [`PubSubEvent<Value>`] items. Use [`ReplayPreset`]
279    /// to control where playback starts.
280    ///
281    /// # Errors
282    ///
283    /// Returns [`PubSubError::Config`] if the tenant ID cannot be fetched from userinfo.
284    pub async fn subscribe(
285        &self,
286        topic: &str,
287        replay: ReplayPreset,
288    ) -> Result<Pin<Box<dyn Stream<Item = Result<PubSubEvent<Value>>> + Send>>> {
289        let tenant_id = self.get_tenant_id().await?.to_string();
290        Ok(subscribe_dynamic(
291            Arc::clone(&self.session),
292            self.config.clone(),
293            self.schema_cache.clone(),
294            self.channel.clone(),
295            topic.to_string(),
296            replay,
297            tenant_id,
298        ))
299    }
300
301    /// Subscribe to a topic, yielding typed events deserialized as `T`.
302    ///
303    /// # Errors
304    ///
305    /// Returns [`PubSubError::Config`] if the tenant ID cannot be fetched from userinfo.
306    pub async fn subscribe_typed<T>(
307        &self,
308        topic: &str,
309        replay: ReplayPreset,
310    ) -> Result<Pin<Box<dyn Stream<Item = Result<PubSubEvent<T>>> + Send>>>
311    where
312        T: DeserializeOwned + Send + 'static,
313    {
314        let tenant_id = self.get_tenant_id().await?.to_string();
315        Ok(subscribe_typed_dynamic(
316            Arc::clone(&self.session),
317            self.config.clone(),
318            self.schema_cache.clone(),
319            self.channel.clone(),
320            topic.to_string(),
321            replay,
322            tenant_id,
323        ))
324    }
325
326    /// Open a bidirectional streaming `PublishStream` RPC and return a [`PublishSink`].
327    ///
328    /// The returned sink allows callers to send multiple batches of events to
329    /// `topic` without the per-call overhead of the unary [`Self::publish`] RPC.
330    /// The server streams back [`PublishResponse`] acknowledgements, which are
331    /// accessible via [`PublishSink::responses`].
332    ///
333    /// # Errors
334    ///
335    /// - [`PubSubError::Config`] if the tenant ID cannot be fetched.
336    /// - [`PubSubError::Transport`] if the gRPC stream cannot be opened.
337    pub async fn publish_stream<T: Serialize + Send + 'static>(
338        &self,
339        topic: &str,
340    ) -> Result<PublishSink<T>> {
341        let token = self.session.token_manager().token().await?;
342        let tenant_id = self.get_tenant_id().await?.to_string();
343
344        open_publish_stream(
345            Arc::clone(&self.session),
346            self.channel.clone(),
347            self.schema_cache.clone(),
348            tenant_id,
349            topic.to_string(),
350            &token,
351        )
352        .await
353    }
354}
force_pubsub/handler.rs

force_pubsub/
handler.rs
