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

dittolive_ditto/ditto/init/
mod.rs
