xmtp_mls_server_config/
storage_location.rs

1//! Resolve a client's database and attachment paths from a deployment record.
2
3use std::{
4    collections::BTreeMap,
5    path::{Path, PathBuf},
6    sync::{Arc, OnceLock},
7};
8
9use serde::{Deserialize, Serialize};
10use sha2::{Digest, Sha256};
11use tokio::sync::Mutex;
12use xmtp_attachments::sanitize_path_component_with_limit;
13
14/// Rust requires each layout's paths. The builder rejects empty paths.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub enum StorageLocation {
17    DataDir(PathBuf),
18    Explicit {
19        db_path: PathBuf,
20        attachments_dir: PathBuf,
21    },
22}
23
24#[derive(Clone, Debug, PartialEq, Eq)]
25pub struct ResolvedPaths {
26    pub db_path: PathBuf,
27    pub attachments_dir: PathBuf,
28}
29
30#[derive(Debug, thiserror::Error)]
31pub enum StorageLocationError {
32    #[error("storage location needs an inbox id before opening the database")]
33    InboxId,
34    #[error("data directory needs a backend URL")]
35    BackendUrl,
36    #[error("offline storage location has no deployment record for this backend URL")]
37    OfflineMissingDeployment,
38    #[error("store or attachments_dir conflicts with data_location")]
39    ConflictingStore,
40    #[error("storage location is missing {field}")]
41    MissingPath { field: &'static str },
42    #[error("deployment identifier differs from the opened data directory")]
43    DeploymentMismatch,
44    #[error("storage location cannot read or write its deployment record: {0}")]
45    Io(#[from] std::io::Error),
46    #[error("storage location cannot encode its deployment record: {0}")]
47    Json(#[from] serde_json::Error),
48    #[error("storage location cannot access OPFS")]
49    Opfs,
50}
51
52#[derive(Default, Serialize, Deserialize)]
53struct DeploymentFile {
54    version: u8,
55    deployments: BTreeMap<String, String>,
56}
57
58static RECORD_LOCK: OnceLock<Arc<Mutex<()>>> = OnceLock::new();
59
60#[cfg(all(test, not(target_arch = "wasm32")))]
61static NATIVE_DEPLOYMENT_WRITE_PAUSE: parking_lot::Mutex<
62    Option<(
63        std::sync::Arc<tokio::sync::Notify>,
64        std::sync::Arc<tokio::sync::Notify>,
65    )>,
66> = parking_lot::Mutex::new(None);
67
68#[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
69#[doc(hidden)]
70#[derive(Clone)]
71pub enum DeploymentWriteFault {
72    Write,
73    Sync,
74    Replace,
75    Pause {
76        entered: std::sync::Arc<tokio::sync::Notify>,
77        resume: std::sync::Arc<tokio::sync::Notify>,
78    },
79    PauseReplace {
80        entered: std::sync::Arc<tokio::sync::Notify>,
81        resume: std::sync::Arc<tokio::sync::Notify>,
82        settled: std::sync::Arc<tokio::sync::Notify>,
83    },
84}
85
86#[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
87static DEPLOYMENT_WRITE_FAULT: parking_lot::Mutex<Option<DeploymentWriteFault>> =
88    parking_lot::Mutex::new(None);
89
90#[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
91#[doc(hidden)]
92pub fn set_deployment_write_fault(fault: DeploymentWriteFault) {
93    *DEPLOYMENT_WRITE_FAULT.lock() = Some(fault);
94}
95
96#[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
97#[doc(hidden)]
98pub async fn write_deployments_for_test(
99    data_dir: &Path,
100    bytes: &[u8],
101) -> Result<(), StorageLocationError> {
102    write_file(data_dir, bytes).await
103}
104
105#[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
106#[doc(hidden)]
107pub async fn record_deployment_for_test(
108    data_dir: &Path,
109    backend_url: &str,
110    identifier: &str,
111) -> Result<(), StorageLocationError> {
112    DeploymentRecorder::new(data_dir.to_path_buf(), backend_url)
113        .record(identifier)
114        .await
115}
116
117/// The file name table applies before the suffix. The suffix distinguishes
118/// identifiers whose file name table result is the same.
119pub fn deployment_component(identifier: &str) -> String {
120    let name = sanitize_path_component_with_limit(identifier, 190).to_ascii_lowercase();
121    let digest = hex::encode(Sha256::digest(identifier.as_bytes()));
122    format!("{name}-{digest}")
123}
124
125/// Reads and writes the deployment identifier recorded for one backend URL
126/// in a data directory.
127#[derive(Clone, Debug)]
128pub struct DeploymentRecorder {
129    data_dir: PathBuf,
130    backend_url: String,
131    opened_identifier: Option<String>,
132}
133
134impl DeploymentRecorder {
135    /// A recorder for `backend_url`'s record in `data_dir`.
136    pub fn new(data_dir: PathBuf, backend_url: &str) -> Self {
137        Self {
138            data_dir,
139            backend_url: crate::server_configuration::normalized_url(backend_url).to_owned(),
140            opened_identifier: None,
141        }
142    }
143
144    /// Refuse to record an identifier other than the one the open database uses.
145    pub fn for_opened_identifier(mut self, identifier: String) -> Self {
146        self.opened_identifier = Some(identifier);
147        self
148    }
149
150    /// The identifier recorded for this backend URL, if any.
151    pub async fn lookup(&self) -> Result<Option<String>, StorageLocationError> {
152        if self.backend_url.is_empty() {
153            return Err(StorageLocationError::BackendUrl);
154        }
155        let _guard = RECORD_LOCK
156            .get_or_init(|| Arc::new(Mutex::new(())))
157            .lock()
158            .await;
159        Ok(read_file(&self.data_dir)
160            .await?
161            .deployments
162            .remove(&self.backend_url))
163    }
164
165    /// Record `identifier` for this backend URL.
166    pub async fn record(&self, identifier: &str) -> Result<(), StorageLocationError> {
167        if self
168            .opened_identifier
169            .as_deref()
170            .is_some_and(|opened| opened != identifier)
171        {
172            return Err(StorageLocationError::DeploymentMismatch);
173        }
174        if self.backend_url.is_empty() {
175            return Err(StorageLocationError::BackendUrl);
176        }
177        let guard = RECORD_LOCK
178            .get_or_init(|| Arc::new(Mutex::new(())))
179            .clone()
180            .lock_owned()
181            .await;
182        let mut file = read_file(&self.data_dir).await?;
183        file.version = 1;
184        file.deployments
185            .insert(self.backend_url.clone(), identifier.to_owned());
186        let bytes = serde_json::to_vec(&file)?;
187        #[cfg(not(target_arch = "wasm32"))]
188        {
189            write_file_with_guard(&self.data_dir, &bytes, Some(guard)).await
190        }
191        #[cfg(target_arch = "wasm32")]
192        {
193            write_file_with_guard(&self.data_dir, &bytes, Some(guard)).await
194        }
195    }
196}
197
198impl StorageLocation {
199    /// Fail when a required path is empty.
200    pub fn validate(&self) -> Result<(), StorageLocationError> {
201        match self {
202            Self::DataDir(data_dir) if data_dir.as_os_str().is_empty() => {
203                Err(StorageLocationError::MissingPath { field: "data_dir" })
204            }
205            Self::Explicit { db_path, .. } if db_path.as_os_str().is_empty() => {
206                Err(StorageLocationError::MissingPath { field: "db_path" })
207            }
208            Self::Explicit {
209                attachments_dir, ..
210            } if attachments_dir.as_os_str().is_empty() => Err(StorageLocationError::MissingPath {
211                field: "attachments_dir",
212            }),
213            _ => Ok(()),
214        }
215    }
216
217    /// The deployment record of a data directory; explicit paths have none.
218    pub fn recorder(&self, backend_url: &str) -> Option<DeploymentRecorder> {
219        match self {
220            Self::DataDir(path) => Some(DeploymentRecorder::new(path.clone(), backend_url)),
221            Self::Explicit { .. } => None,
222        }
223    }
224
225    /// Resolve from a recorded identifier. A missing record is an error;
226    /// the builder fetches GetConfiguration before it calls this on a miss.
227    pub async fn resolve(
228        &self,
229        backend_url: &str,
230        inbox_id: &str,
231    ) -> Result<ResolvedPaths, StorageLocationError> {
232        match self {
233            Self::Explicit {
234                db_path,
235                attachments_dir,
236            } => Ok(ResolvedPaths {
237                db_path: db_path.clone(),
238                attachments_dir: attachments_dir.clone(),
239            }),
240            Self::DataDir(path) => {
241                let recorder = DeploymentRecorder::new(path.clone(), backend_url);
242                let identifier = recorder
243                    .lookup()
244                    .await?
245                    .ok_or(StorageLocationError::OfflineMissingDeployment)?;
246                self.resolve_identifier(inbox_id, &identifier)
247            }
248        }
249    }
250
251    /// Resolve the paths of `inbox_id` on the deployment `identifier`.
252    pub fn resolve_identifier(
253        &self,
254        inbox_id: &str,
255        identifier: &str,
256    ) -> Result<ResolvedPaths, StorageLocationError> {
257        match self {
258            Self::Explicit {
259                db_path,
260                attachments_dir,
261            } => Ok(ResolvedPaths {
262                db_path: db_path.clone(),
263                attachments_dir: attachments_dir.clone(),
264            }),
265            Self::DataDir(path) => {
266                if inbox_id.is_empty() || !inbox_id.bytes().all(|b| b.is_ascii_hexdigit()) {
267                    return Err(StorageLocationError::InboxId);
268                }
269                let root = path
270                    .join(deployment_component(identifier))
271                    .join(inbox_id.to_ascii_lowercase());
272                Ok(ResolvedPaths {
273                    db_path: root.join("xmtp.db3"),
274                    attachments_dir: root.join("attachments"),
275                })
276            }
277        }
278    }
279}
280
281#[cfg(not(target_arch = "wasm32"))]
282async fn read_file(data_dir: &Path) -> Result<DeploymentFile, StorageLocationError> {
283    match tokio::fs::read(data_dir.join("deployments.json")).await {
284        Ok(bytes) => Ok(serde_json::from_slice(&bytes).unwrap_or_default()),
285        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(DeploymentFile::default()),
286        Err(error) => Err(error.into()),
287    }
288}
289
290#[cfg(not(target_arch = "wasm32"))]
291struct NativeDeploymentTempGuard {
292    path: PathBuf,
293    committed: bool,
294}
295
296#[cfg(not(target_arch = "wasm32"))]
297impl Drop for NativeDeploymentTempGuard {
298    fn drop(&mut self) {
299        if self.committed {
300            return;
301        }
302        if let Err(error) = std::fs::remove_file(&self.path)
303            && error.kind() != std::io::ErrorKind::NotFound
304        {
305            tracing::warn!(?error, path = %self.path.display(), "cannot remove temporary deployment record");
306        }
307    }
308}
309
310#[cfg(all(test, not(target_arch = "wasm32")))]
311async fn write_file(data_dir: &Path, bytes: &[u8]) -> Result<(), StorageLocationError> {
312    write_file_with_guard(data_dir, bytes, None).await
313}
314
315#[cfg(not(target_arch = "wasm32"))]
316async fn write_file_with_guard(
317    data_dir: &Path,
318    bytes: &[u8],
319    record_guard: Option<tokio::sync::OwnedMutexGuard<()>>,
320) -> Result<(), StorageLocationError> {
321    xmtp_attachments::create_private_directory(data_dir).await?;
322    let path = data_dir.join("deployments.json");
323    let temp = data_dir.join(format!(
324        ".deployments-{}.tmp",
325        xmtp_common::rand_string::<16>()
326    ));
327    let bytes = bytes.to_vec();
328    xmtp_common::task::spawn_blocking(move || -> std::io::Result<()> {
329        let _record_guard = record_guard;
330        use std::io::Write as _;
331
332        let mut options = std::fs::OpenOptions::new();
333        options.write(true).create_new(true);
334        #[cfg(unix)]
335        {
336            use std::os::unix::fs::OpenOptionsExt as _;
337            options.mode(0o600);
338        }
339        let mut guard = NativeDeploymentTempGuard {
340            path: temp.clone(),
341            committed: true,
342        };
343        let mut file = options.open(&temp)?;
344        guard.committed = false;
345        #[cfg(test)]
346        let pause = { NATIVE_DEPLOYMENT_WRITE_PAUSE.lock().take() };
347        #[cfg(test)]
348        if let Some((entered, resume)) = pause {
349            entered.notify_one();
350            futures::executor::block_on(resume.notified());
351        }
352        let write = (|| -> std::io::Result<()> {
353            #[cfg(unix)]
354            {
355                use std::os::unix::fs::PermissionsExt as _;
356                file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
357            }
358            file.write_all(&bytes)?;
359            file.sync_all()
360        })();
361        drop(file);
362        write?;
363        std::fs::rename(&temp, &path)?;
364        guard.committed = true;
365        Ok(())
366    })
367    .await
368    .map_err(std::io::Error::other)??;
369    Ok(())
370}
371
372#[cfg(target_arch = "wasm32")]
373async fn read_file(data_dir: &Path) -> Result<DeploymentFile, StorageLocationError> {
374    use xmtp_attachments::{LocalStore, OpfsStore};
375    let store = OpfsStore::new(&data_dir.to_string_lossy())
376        .await
377        .map_err(|_| StorageLocationError::Opfs)?;
378    if !store
379        .exists("deployments.json")
380        .await
381        .map_err(|_| StorageLocationError::Opfs)?
382    {
383        return Ok(DeploymentFile::default());
384    }
385    let file = store
386        .open_read("deployments.json")
387        .await
388        .map_err(|_| StorageLocationError::Opfs)?;
389    let mut bytes = Vec::new();
390    let mut offset = 0;
391    while offset < file.len() {
392        let chunk = file
393            .read_chunk(offset, 64 * 1024)
394            .await
395            .map_err(|_| StorageLocationError::Opfs)?;
396        if chunk.is_empty() {
397            return Err(StorageLocationError::Opfs);
398        }
399        offset += chunk.len() as u64;
400        bytes.extend_from_slice(&chunk);
401    }
402    Ok(serde_json::from_slice(&bytes).unwrap_or_default())
403}
404
405#[cfg(target_arch = "wasm32")]
406struct DeploymentTempGuard {
407    store: xmtp_attachments::OpfsStore,
408    path: String,
409    committed: bool,
410}
411
412#[cfg(target_arch = "wasm32")]
413impl Drop for DeploymentTempGuard {
414    fn drop(&mut self) {
415        if self.committed {
416            return;
417        }
418        use xmtp_attachments::LocalStore;
419        let store = self.store.clone();
420        let path = self.path.clone();
421        drop(xmtp_common::task::spawn(async move {
422            if let Err(error) = store.remove_file(&path).await {
423                tracing::warn!(?error, %path, "cannot remove temporary deployment record");
424            }
425        }));
426    }
427}
428
429#[cfg(target_arch = "wasm32")]
430async fn write_file(data_dir: &Path, bytes: &[u8]) -> Result<(), StorageLocationError> {
431    write_file_with_guard(data_dir, bytes, None).await
432}
433
434#[cfg(target_arch = "wasm32")]
435async fn write_file_with_guard(
436    data_dir: &Path,
437    bytes: &[u8],
438    record_guard: Option<tokio::sync::OwnedMutexGuard<()>>,
439) -> Result<(), StorageLocationError> {
440    use xmtp_attachments::{DownloadSink, LocalStore, OpfsStore};
441    let store = OpfsStore::new(&data_dir.to_string_lossy())
442        .await
443        .map_err(|_| StorageLocationError::Opfs)?;
444    let temp = format!(".tmp/deployments-{}", xmtp_common::rand_string::<16>());
445    let mut writer = store
446        .create_temp(&temp)
447        .await
448        .map_err(|_| StorageLocationError::Opfs)?;
449    let guard = DeploymentTempGuard {
450        store: store.clone(),
451        path: temp.clone(),
452        committed: false,
453    };
454    #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
455    let fault = { DEPLOYMENT_WRITE_FAULT.lock().take() };
456    let write_result: Result<(), StorageLocationError> = async {
457        #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
458        if let Some(DeploymentWriteFault::Pause { entered, resume }) = &fault {
459            entered.notify_one();
460            resume.notified().await;
461        }
462        #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
463        if matches!(fault, Some(DeploymentWriteFault::Write)) {
464            return Err(StorageLocationError::Opfs);
465        }
466        writer
467            .write(bytes)
468            .await
469            .map_err(|_| StorageLocationError::Opfs)?;
470        #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
471        if matches!(fault, Some(DeploymentWriteFault::Sync)) {
472            return Err(StorageLocationError::Opfs);
473        }
474        store
475            .sync(&mut writer)
476            .await
477            .map_err(|_| StorageLocationError::Opfs)?;
478        Ok(())
479    }
480    .await;
481    drop(writer);
482    let task = xmtp_common::task::spawn(async move {
483        let _record_guard = record_guard;
484        let mut guard = guard;
485        #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
486        if let Some(DeploymentWriteFault::PauseReplace {
487            entered, resume, ..
488        }) = &fault
489        {
490            entered.notify_one();
491            resume.notified().await;
492        }
493        let result = match write_result {
494            Err(error) => Err(error),
495            Ok(()) => {
496                #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
497                if matches!(fault, Some(DeploymentWriteFault::Replace)) {
498                    Err(StorageLocationError::Opfs)
499                } else {
500                    store
501                        .replace(&temp, "deployments.json")
502                        .await
503                        .map_err(|_| StorageLocationError::Opfs)
504                }
505                #[cfg(not(all(target_arch = "wasm32", feature = "test-hooks")))]
506                // OPFS file move replaces the destination in current browser engines.
507                store
508                    .replace(&temp, "deployments.json")
509                    .await
510                    .map_err(|_| StorageLocationError::Opfs)
511            }
512        };
513        if result.is_ok() {
514            guard.committed = true;
515        } else if let Err(error) = store.remove_file(&temp).await {
516            tracing::warn!(?error, %temp, "cannot remove temporary deployment record");
517        } else {
518            guard.committed = true;
519        }
520        #[cfg(all(target_arch = "wasm32", feature = "test-hooks"))]
521        if let Some(DeploymentWriteFault::PauseReplace { settled, .. }) = &fault {
522            settled.notify_one();
523        }
524        result
525    });
526    task.await.map_err(|_| StorageLocationError::Opfs)?
527}
528
529#[cfg(any(test, feature = "test-utils"))]
530#[doc(hidden)]
531pub mod test_support;
532
533#[cfg(test)]
534mod tests;
xmtp_mls_server_config/storage_location.rs

xmtp_mls_server_config/
storage_location.rs
