dittolive_ditto/ditto/init/
mod.rs

1use std::{
2    ffi::CString,
3    future::Future,
4    os::raw::{c_uint, c_void},
5    str::FromStr,
6    sync::{Arc, Weak},
7    time::Duration,
8};
9
10use async_fn_traits::AsyncFn2;
11use async_trait::async_trait;
12use extern_c::extern_c;
13use ffi_sdk::{FsComponent, TransportConfigMode};
14
15pub use self::config::{DittoConfig, DittoConfigConnect};
16use crate::{
17    auth::DittoAuthenticator,
18    ditto::{
19        init::config::{ActualConfig, InternalConfig},
20        DittoFields,
21    },
22    error,
23    small_peer_info::SmallPeerInfo,
24    utils::{make_continuation, prelude::*},
25    warn,
26};
27
28pub(crate) mod config;
29
30impl Ditto {
31    /// Open a new Ditto instance using a [`DittoConfig`]
32    ///
33    /// # Example
34    ///
35    /// ```
36    /// # use dittolive_ditto::prelude::*;
37    /// # async fn example() -> anyhow::Result<()> {
38    /// // Load your database ID somehow, ENV is a good option
39    /// let database_id = std::env::var("DITTO_DATABASE_ID")?;
40    ///
41    /// // Choose one of the following types of connection config
42    /// let connect = DittoConfigConnect::Server {
43    ///     url: "https://example.com/your-server-url".parse().unwrap(),
44    /// };
45    /// let connect = DittoConfigConnect::SmallPeersOnly {
46    ///     private_key: Some("https://example.com/your-server-url".bytes().collect()),
47    /// };
48    /// let connect = DittoConfigConnect::SmallPeersOnly { private_key: None };
49    ///
50    /// let config = DittoConfig::new(database_id, connect);
51    /// let ditto = Ditto::open(config).await?;
52    /// # Ok(())
53    /// # }
54    /// ```
55    pub async fn open(config: DittoConfig) -> Result<Ditto, DittoError> {
56        config
57            .validate_system_parameter_value_shapes()
58            .map_err(|message| DittoError::from_str(ErrorKind::InvalidInput, message))?;
59        Self::init_sdk_version();
60        let default_root_dir = default_root_directory();
61
62        let customer_config = config;
63        let mut internal_config = InternalConfig {
64            legacy_persistence_directory: None,
65        };
66
67        // SDKS-3187: When the user hasn't set an explicit persistence directory,
68        // provide the v4 default path as a legacy fallback so that v4→v5 upgrades
69        // don't silently create a new empty database. The v4 Rust SDK used the
70        // executable's parent directory directly as the default root.
71        if customer_config.persistence_directory.is_none() {
72            if let Some(root) = &default_root_dir {
73                internal_config.legacy_persistence_directory = Some(root.clone());
74            }
75        }
76
77        let actual_config = ActualConfig {
78            customer_facing: customer_config,
79            internal: internal_config,
80        };
81
82        let config_cbor: &[u8] =
83            &serde_cbor::to_vec(&actual_config).expect("should serialize well-formed DittoConfig");
84        let (continuation, recv) = make_continuation();
85        let default_root_dir_ref = default_root_dir
86            .as_deref()
87            .and_then(|root| root.to_str())
88            .ok_or_else(|| {
89                DittoError::new(
90                    ErrorKind::IO,
91                    "Unable to resolve a default data directory on this platform".to_string(),
92                )
93            })?;
94        let default_root_dir_cstring = CString::from_str(default_root_dir_ref)
95            .expect("should construct CString from no-nulls &str");
96        let default_root_dir_cstr = &*default_root_dir_cstring;
97        let default_root_dir_charp: char_p::Ref<'_> = default_root_dir_cstr.into();
98
99        // CLIPPY: continuation.into() needed to pass CI
100        #[allow(clippy::useless_conversion)]
101        ffi_sdk::dittoffi_ditto_open_async_throws(
102            config_cbor.into(),
103            TransportConfigMode::PlatformIndependent,
104            default_root_dir_charp,
105            continuation.into(),
106        );
107
108        let ffi_result = recv.await.unwrap();
109        let ffi_ditto: repr_c::Box<ffi_sdk::Ditto> = ffi_result.into_rust_result()?;
110
111        Self::finish_open(ffi_ditto, actual_config.customer_facing)
112    }
113
114    /// Open a new Ditto instance using a [`DittoConfig`]
115    ///
116    /// This is a synchronous blocking variant of [`Ditto::open()`] that will wait until
117    /// initialization is complete.
118    ///
119    /// # Example
120    ///
121    /// ```
122    /// # use dittolive_ditto::prelude::*;
123    /// # fn example() -> anyhow::Result<()> {
124    /// // Load your database ID somehow, ENV is a good option
125    /// let database_id = std::env::var("DITTO_DATABASE_ID")?;
126    ///
127    /// // Choose one of the following types of connection config
128    /// let connect = DittoConfigConnect::Server {
129    ///     url: "https://example.com/your-server-url".parse().unwrap(),
130    /// };
131    /// let connect = DittoConfigConnect::SmallPeersOnly {
132    ///     private_key: Some("https://example.com/your-server-url".bytes().collect()),
133    /// };
134    /// let connect = DittoConfigConnect::SmallPeersOnly { private_key: None };
135    ///
136    /// let config = DittoConfig::new(database_id, connect);
137    /// let ditto = Ditto::open_sync(config)?;
138    /// # Ok(())
139    /// # }
140    /// ```
141    pub fn open_sync(config: DittoConfig) -> Result<Ditto, DittoError> {
142        config
143            .validate_system_parameter_value_shapes()
144            .map_err(|message| DittoError::from_str(ErrorKind::InvalidInput, message))?;
145        Self::init_sdk_version();
146        let default_root_dir = default_root_directory();
147
148        let customer_config = config;
149        let mut internal_config = InternalConfig {
150            legacy_persistence_directory: None,
151        };
152
153        // SDKS-3187: Same legacy fallback as open() above.
154        if customer_config.persistence_directory.is_none() {
155            if let Some(root) = &default_root_dir {
156                internal_config.legacy_persistence_directory = Some(root.clone());
157            }
158        }
159
160        let actual_config = ActualConfig {
161            customer_facing: customer_config,
162            internal: internal_config,
163        };
164
165        let config_cbor: &[u8] =
166            &serde_cbor::to_vec(&actual_config).expect("should serialize well-formed DittoConfig");
167        let default_root_dir_ref = default_root_dir
168            .as_deref()
169            .and_then(|root| root.to_str())
170            .ok_or_else(|| {
171                DittoError::new(
172                    ErrorKind::IO,
173                    "Unable to resolve a default data directory on this platform".to_string(),
174                )
175            })?;
176        let default_root_dir_cstring = CString::from_str(default_root_dir_ref)
177            .expect("should construct CString from no-nulls &str");
178        let default_root_dir_cstr = &*default_root_dir_cstring;
179        let default_root_dir_charp: char_p::Ref<'_> = default_root_dir_cstr.into();
180
181        let ffi_ditto: repr_c::Box<ffi_sdk::Ditto> = ffi_sdk::dittoffi_ditto_open_throws(
182            config_cbor.into(),
183            TransportConfigMode::PlatformIndependent,
184            default_root_dir_charp,
185        )
186        .into_rust_result()?;
187
188        Self::finish_open(ffi_ditto, actual_config.customer_facing)
189    }
190
191    fn finish_open(
192        ffi_ditto: repr_c::Box<ffi_sdk::Ditto>,
193        config: DittoConfig,
194    ) -> Result<Ditto, DittoError> {
195        let ditto: Arc<repr_c::Box<ffi_sdk::Ditto>> = Arc::new(ffi_ditto);
196        let has_auth = matches!(&config.connect, DittoConfigConnect::Server { .. });
197
198        let disk_usage = DiskUsage::new(ditto.retain(), FsComponent::Root);
199        let small_peer_info = SmallPeerInfo::new(ditto.retain());
200        let fields = Arc::new_cyclic(|weak_fields: &arc::Weak<_>| {
201            let store = Store::new(ditto.retain(), weak_fields.clone());
202            let sync = crate::sync::Sync::new(weak_fields.clone());
203            let presence = Arc::new(Presence::new(weak_fields.clone()));
204
205            DittoFields {
206                ditto: ditto.retain(),
207                has_auth,
208                config,
209                store,
210                sync,
211                presence,
212                disk_usage,
213                small_peer_info,
214            }
215        });
216
217        let ditto = Ditto {
218            fields,
219            is_shut_down_able: true,
220        };
221
222        Ok(ditto)
223    }
224}
225
226fn default_root_directory() -> Option<PathBuf> {
227    std::env::current_exe()
228        .ok()
229        .and_then(|abspath| abspath.parent().map(|x| x.to_path_buf()))
230}
231
232impl DittoAuthenticator {
233    /// Set a callback to notify the client when the authentication is expiring.
234    ///
235    /// When using `DittoConfigConnect::Server { .. }` mode, Ditto _requires_ you to register an
236    /// "expiration handler" for authentication. This handler is called for an initial
237    /// authentication and periodically thereafter when the current authentication is near to
238    /// expiration.
239    ///
240    /// For more details about authentication, see the [Ditto Auth and Authorization docs][0].
241    ///
242    /// [0]: https://docs.ditto.live/sdk/latest/auth-and-authorization/cloud-authentication#login
243    ///
244    /// For more details about the expiration handler, see the [`DittoAuthExpirationHandler`] trait.
245    ///
246    /// # Example
247    ///
248    /// ```
249    /// # #[allow(deprecated)]
250    /// # use tracing::error;
251    /// # use dittolive_ditto::prelude::*;
252    /// # fn main() -> anyhow::Result<()> {
253    /// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_online_ditto();
254    /// async fn sample_get_token() -> anyhow::Result<String> {
255    ///     // e.g. reqwest::get("https://example.com/token").await?.text().await?;
256    ///     Ok("token".to_string())
257    /// }
258    ///
259    /// let auth = ditto
260    ///     .auth()
261    ///     .expect("Auth is available in Server connect mode");
262    /// auth.set_expiration_handler(async |ditto: &Ditto, duration_remaining| {
263    ///     let auth = ditto
264    ///         .auth()
265    ///         .expect("Auth is available in Server connect mode");
266    ///
267    ///     // Call your auth service to get a new token
268    ///     let token = sample_get_token().await.unwrap();
269    ///
270    ///     // Where "my-provider" is the name of the Authentication Webhook
271    ///     // you've configured on the Ditto Portal
272    ///     let result = auth.login(&token, "my-provider");
273    ///
274    ///     if let Err(login_error) = result {
275    ///         // Handle login error, e.g.:
276    ///         error!("Failed to login: {}", login_error);
277    ///     }
278    /// });
279    /// # Ok(())
280    /// # }
281    /// ```
282    pub fn set_expiration_handler<F>(&self, handler: F)
283    where
284        F: DittoAuthExpirationHandler,
285    {
286        let Some(ditto) = self.ditto_fields.upgrade() else {
287            #[allow(deprecated)] // Workaround for patched tracing
288            {
289                error!("Failed to set expiration handler, Ditto has shut down");
290            }
291            return;
292        };
293
294        let login_provider = make_login_provider(self.ditto_fields.clone(), Arc::new(handler));
295        ffi_sdk::ditto_auth_set_login_provider(&ditto.ditto, Some(login_provider));
296    }
297
298    /// Clear the expiration handler, if set.
299    pub fn clear_expiration_handler(&self) {
300        let Some(ditto) = self.ditto_fields.upgrade() else {
301            #[allow(deprecated)] // Workaround for patched tracing
302            {
303                error!("Failed to clear expiration handler, Ditto has shut down");
304            }
305            return;
306        };
307
308        ffi_sdk::ditto_auth_set_login_provider(&ditto.ditto, None);
309    }
310}
311
312/// Trait describing types which can be used as an authentication expiration handler for Ditto.
313///
314/// When using `DittoConfigConnect::Server { .. }` mode, Ditto _requires_ you to register an
315/// "expiration handler" for authentication. This handler is called for an initial authentication
316/// and periodically thereafter when the current authentication is near expiration.
317///
318/// This trait is implemented for async closures and for functions returning an
319/// `impl Future<Output = ()>`, which take the expected arguments of an auth expiration handler.
320///
321/// Expiration handlers are expected to call `auth.login(...)` with a valid authentication token.
322///
323/// For more details about authentication, see the [Ditto Auth and Authorization docs][0].
324///
325/// [0]: https://docs.ditto.live/sdk/latest/auth-and-authorization/cloud-authentication#login
326///
327/// **NOTE**: Because expiration handlers are asynchronous, they must not block the thread.
328///
329/// # Example
330///
331/// ```
332/// # use std::time::Duration;
333/// # use dittolive_ditto::prelude::*;
334/// # let (_root, ditto) = dittolive_ditto::doctest_helpers::doctest_online_ditto();
335/// let auth = ditto
336///     .auth()
337///     .expect("Auth is available for DittoConfigConnect::Server mode");
338///
339/// // Option 1: Use an async closure
340/// auth.set_expiration_handler(async |ditto: &Ditto, duration| {
341///     // Your authentication handler code here
342/// });
343///
344/// // Option 2: Use a closure returning an async block
345/// auth.set_expiration_handler(|ditto: &Ditto, duration| async {
346///     // Your authentication handler code here
347/// });
348///
349/// // Option 3: Use a custom type and trait impl
350/// struct MyAuthHandler;
351/// impl DittoAuthExpirationHandler for MyAuthHandler {
352///     async fn on_expiration(&self, ditto: &Ditto, duration_remaining: Duration) {
353///         // Your authentication handler code here
354///     }
355/// }
356/// auth.set_expiration_handler(MyAuthHandler);
357/// ```
358pub trait DittoAuthExpirationHandler: 'static + Send + Sync {
359    /// Provide an async handler that will be called when the authentication is expiring.
360    fn on_expiration(
361        &self,
362        ditto: &Ditto,
363        duration_remaining: Duration,
364    ) -> impl Send + Future<Output = ()>;
365}
366
367impl<F> DittoAuthExpirationHandler for F
368where
369    F: 'static + Send + Sync,
370    F: for<'r> AsyncFn2<&'r Ditto, Duration, Output = (), OutputFuture: Send>,
371{
372    async fn on_expiration(&self, ditto: &Ditto, duration_remaining: Duration) {
373        self(ditto, duration_remaining).await
374    }
375}
376
377#[async_trait]
378pub(crate) trait DynDittoAuthExpirationHandler: 'static + Send + Sync {
379    async fn dyn_on_expiration(&self, ditto: &Ditto, duration_remaining: Duration);
380}
381
382#[async_trait]
383impl<F: DittoAuthExpirationHandler> DynDittoAuthExpirationHandler for F {
384    async fn dyn_on_expiration(&self, ditto: &Ditto, duration_remaining: Duration) {
385        self.on_expiration(ditto, duration_remaining).await
386    }
387}
388
389impl DittoAuthExpirationHandler for dyn '_ + DynDittoAuthExpirationHandler {
390    async fn on_expiration(&self, ditto: &Ditto, duration_remaining: Duration) {
391        self.dyn_on_expiration(ditto, duration_remaining).await
392    }
393}
394
395pub(crate) fn make_login_provider(
396    ditto_fields: Weak<DittoFields>,
397    auth_expiration_handler: Arc<dyn DynDittoAuthExpirationHandler>,
398) -> repr_c::Box<ffi_sdk::LoginProvider> {
399    struct LoginProviderCtx {
400        ditto_fields: Weak<DittoFields>,
401        auth_expiration_handler: Arc<dyn DynDittoAuthExpirationHandler>,
402    }
403
404    let login_provider_ctx = Arc::new(LoginProviderCtx {
405        auth_expiration_handler,
406        ditto_fields,
407    });
408
409    let ffi_ctx = Arc::as_ptr(&login_provider_ctx) as *mut c_void;
410    let ffi_retain = Some(extern_c(|ctx: *mut c_void| unsafe {
411        Arc::<LoginProviderCtx>::increment_strong_count(ctx.cast())
412    }) as unsafe extern "C" fn(_));
413    let ffi_release = Some(extern_c(|ctx: *mut c_void| unsafe {
414        Arc::<LoginProviderCtx>::decrement_strong_count(ctx.cast())
415    }) as unsafe extern "C" fn(_));
416
417    // This callback is just a "trigger" to initiate the caller's authentication handler
418    // The handler is async and spawned in a separate task
419    let ffi_handler = extern_c(|ctx: *mut c_void, secs_remaining: c_uint| {
420        let login_provider_ctx: &LoginProviderCtx = unsafe { &*ctx.cast() };
421        let auth_expiration_handler = login_provider_ctx.auth_expiration_handler.retain();
422        let Ok(ditto) = Ditto::upgrade(&login_provider_ctx.ditto_fields) else {
423            #[allow(deprecated)] // Workaround for patched tracing
424            {
425                error!("Failed to dispatch auth handler, Ditto has been shut down");
426            }
427            return;
428        };
429
430        dispatch_auth_handler(
431            auth_expiration_handler,
432            ditto,
433            Duration::from_secs(secs_remaining.into()),
434        );
435    });
436
437    unsafe {
438        ffi_sdk::ditto_auth_client_make_login_provider(
439            ffi_ctx,
440            ffi_retain,
441            ffi_release,
442            ffi_handler,
443        )
444    }
445}
446
447/// Dispatches the users's authentication expiration handler.
448///
449/// This function checks whether the current execution context is within a tokio runtime.
450///
451/// - If a tokio runtime is available, the handler is spawned onto a task.
452/// - If no tokio runtime is available, a temporary current-thread runtime is created and the
453///   dispatcher will block until the handler completes.
454///
455/// # Panics
456///
457/// This function will panic if:
458///
459/// - There is no current tokio runtime available, and
460/// - We fail to create a temporary current-thread runtime
461fn dispatch_auth_handler(
462    auth_expiration_handler: Arc<dyn DynDittoAuthExpirationHandler>,
463    ditto: Ditto,
464    duration_remaining: Duration,
465) {
466    // Check if we're in a tokio runtime context
467    match tokio::runtime::Handle::try_current() {
468        Ok(handle) => {
469            // We have a tokio runtime, spawn the handler as usual
470            handle.spawn(async move {
471                auth_expiration_handler
472                    .on_expiration(&ditto, duration_remaining)
473                    .await;
474            });
475        }
476        Err(_) => {
477            // No tokio runtime available, create a temporary current-thread runtime
478            #[allow(deprecated)] // Workaround for patched tracing
479            {
480                warn!(
481                    "No tokio runtime available for expiration handler. Creating temporary \
482                     runtime."
483                );
484            }
485
486            match tokio::runtime::Builder::new_current_thread()
487                .enable_all()
488                .build()
489            {
490                Ok(rt) => {
491                    rt.block_on(async move {
492                        auth_expiration_handler
493                            .on_expiration(&ditto, duration_remaining)
494                            .await;
495                    });
496                }
497                Err(e) => {
498                    panic!(
499                        "Failed to create tokio runtime for expiration handler: {}. Consider \
500                         running within a tokio runtime context.",
501                        e
502                    );
503                }
504            }
505        }
506    }
507}
508
509#[cfg(test)]
510mod tests {
511
512    #[test]
513    fn test_runtime_detection_behavior() {
514        // This test demonstrates the runtime detection behavior
515        println!("Testing runtime detection...");
516
517        // Check if we're in a tokio context (should be false in regular test)
518        match tokio::runtime::Handle::try_current() {
519            Ok(_handle) => {
520                println!("✓ Tokio runtime detected - handlers will be spawned");
521            }
522            Err(_) => {
523                println!("✗ No tokio runtime - will create temporary runtime");
524
525                // Demonstrate temporary runtime creation (as done in make_login_provider)
526                match tokio::runtime::Builder::new_current_thread()
527                    .enable_all()
528                    .build()
529                {
530                    Ok(_rt) => {
531                        println!("✓ Successfully created temporary runtime");
532                    }
533                    Err(e) => {
534                        println!("✗ Failed to create temporary runtime: {}", e);
535                    }
536                }
537            }
538        }
539    }
540
541    #[tokio::test]
542    async fn test_with_tokio_runtime_available() {
543        // This test runs with a tokio runtime available
544        println!("Testing with tokio runtime available...");
545
546        // This should succeed
547        match tokio::runtime::Handle::try_current() {
548            Ok(_handle) => {
549                println!("✓ Tokio runtime is available for async handlers");
550            }
551            Err(_) => {
552                panic!("Expected tokio runtime to be available in tokio::test");
553            }
554        }
555
556        println!("✓ Runtime detection works correctly in async context");
557    }
558
559    #[test]
560    fn test_improved_error_handling() {
561        // Test that we can create the same type of runtime as the implementation
562        let rt_result = tokio::runtime::Builder::new_current_thread()
563            .enable_all()
564            .build();
565
566        match rt_result {
567            Ok(rt) => {
568                println!("✓ Temporary runtime creation works");
569
570                // Demonstrate that we can use it for async work
571                rt.block_on(async {
572                    println!("✓ Async work executes successfully in temporary runtime");
573                });
574            }
575            Err(e) => {
576                println!("✗ Failed to create runtime: {}", e);
577            }
578        }
579    }
580}
dittolive_ditto/ditto/init/mod.rs

dittolive_ditto/ditto/init/
mod.rs
