xmtp_mls/
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-utils"))]
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-utils"))]
87static DEPLOYMENT_WRITE_FAULT: parking_lot::Mutex<Option<DeploymentWriteFault>> =
88    parking_lot::Mutex::new(None);
89
90#[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
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#[derive(Clone, Debug)]
126pub(crate) struct DeploymentRecorder {
127    data_dir: PathBuf,
128    backend_url: String,
129    opened_identifier: Option<String>,
130}
131
132impl DeploymentRecorder {
133    pub(crate) fn new(data_dir: PathBuf, backend_url: &str) -> Self {
134        Self {
135            data_dir,
136            backend_url: crate::server_configuration::normalized_url(backend_url).to_owned(),
137            opened_identifier: None,
138        }
139    }
140
141    pub(crate) fn for_opened_identifier(mut self, identifier: String) -> Self {
142        self.opened_identifier = Some(identifier);
143        self
144    }
145
146    pub(crate) async fn lookup(&self) -> Result<Option<String>, StorageLocationError> {
147        if self.backend_url.is_empty() {
148            return Err(StorageLocationError::BackendUrl);
149        }
150        let _guard = RECORD_LOCK
151            .get_or_init(|| Arc::new(Mutex::new(())))
152            .lock()
153            .await;
154        Ok(read_file(&self.data_dir)
155            .await?
156            .deployments
157            .remove(&self.backend_url))
158    }
159
160    pub(crate) async fn record(&self, identifier: &str) -> Result<(), StorageLocationError> {
161        if self
162            .opened_identifier
163            .as_deref()
164            .is_some_and(|opened| opened != identifier)
165        {
166            return Err(StorageLocationError::DeploymentMismatch);
167        }
168        if self.backend_url.is_empty() {
169            return Err(StorageLocationError::BackendUrl);
170        }
171        let guard = RECORD_LOCK
172            .get_or_init(|| Arc::new(Mutex::new(())))
173            .clone()
174            .lock_owned()
175            .await;
176        let mut file = read_file(&self.data_dir).await?;
177        file.version = 1;
178        file.deployments
179            .insert(self.backend_url.clone(), identifier.to_owned());
180        let bytes = serde_json::to_vec(&file)?;
181        #[cfg(not(target_arch = "wasm32"))]
182        {
183            write_file_with_guard(&self.data_dir, &bytes, Some(guard)).await
184        }
185        #[cfg(target_arch = "wasm32")]
186        {
187            write_file_with_guard(&self.data_dir, &bytes, Some(guard)).await
188        }
189    }
190}
191
192impl StorageLocation {
193    pub(crate) fn validate(&self) -> Result<(), StorageLocationError> {
194        match self {
195            Self::DataDir(data_dir) if data_dir.as_os_str().is_empty() => {
196                Err(StorageLocationError::MissingPath { field: "data_dir" })
197            }
198            Self::Explicit { db_path, .. } if db_path.as_os_str().is_empty() => {
199                Err(StorageLocationError::MissingPath { field: "db_path" })
200            }
201            Self::Explicit {
202                attachments_dir, ..
203            } if attachments_dir.as_os_str().is_empty() => Err(StorageLocationError::MissingPath {
204                field: "attachments_dir",
205            }),
206            _ => Ok(()),
207        }
208    }
209
210    pub(crate) fn recorder(&self, backend_url: &str) -> Option<DeploymentRecorder> {
211        match self {
212            Self::DataDir(path) => Some(DeploymentRecorder::new(path.clone(), backend_url)),
213            Self::Explicit { .. } => None,
214        }
215    }
216
217    /// Resolve from a recorded identifier. A missing record is an error;
218    /// the builder fetches GetConfiguration before it calls this on a miss.
219    pub async fn resolve(
220        &self,
221        backend_url: &str,
222        inbox_id: &str,
223    ) -> Result<ResolvedPaths, StorageLocationError> {
224        match self {
225            Self::Explicit {
226                db_path,
227                attachments_dir,
228            } => Ok(ResolvedPaths {
229                db_path: db_path.clone(),
230                attachments_dir: attachments_dir.clone(),
231            }),
232            Self::DataDir(path) => {
233                let recorder = DeploymentRecorder::new(path.clone(), backend_url);
234                let identifier = recorder
235                    .lookup()
236                    .await?
237                    .ok_or(StorageLocationError::OfflineMissingDeployment)?;
238                self.resolve_identifier(inbox_id, &identifier)
239            }
240        }
241    }
242
243    pub(crate) fn resolve_identifier(
244        &self,
245        inbox_id: &str,
246        identifier: &str,
247    ) -> Result<ResolvedPaths, StorageLocationError> {
248        match self {
249            Self::Explicit {
250                db_path,
251                attachments_dir,
252            } => Ok(ResolvedPaths {
253                db_path: db_path.clone(),
254                attachments_dir: attachments_dir.clone(),
255            }),
256            Self::DataDir(path) => {
257                if inbox_id.is_empty() || !inbox_id.bytes().all(|b| b.is_ascii_hexdigit()) {
258                    return Err(StorageLocationError::InboxId);
259                }
260                let root = path
261                    .join(deployment_component(identifier))
262                    .join(inbox_id.to_ascii_lowercase());
263                Ok(ResolvedPaths {
264                    db_path: root.join("xmtp.db3"),
265                    attachments_dir: root.join("attachments"),
266                })
267            }
268        }
269    }
270}
271
272#[cfg(not(target_arch = "wasm32"))]
273async fn read_file(data_dir: &Path) -> Result<DeploymentFile, StorageLocationError> {
274    match tokio::fs::read(data_dir.join("deployments.json")).await {
275        Ok(bytes) => Ok(serde_json::from_slice(&bytes).unwrap_or_default()),
276        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(DeploymentFile::default()),
277        Err(error) => Err(error.into()),
278    }
279}
280
281#[cfg(not(target_arch = "wasm32"))]
282struct NativeDeploymentTempGuard {
283    path: PathBuf,
284    committed: bool,
285}
286
287#[cfg(not(target_arch = "wasm32"))]
288impl Drop for NativeDeploymentTempGuard {
289    fn drop(&mut self) {
290        if self.committed {
291            return;
292        }
293        if let Err(error) = std::fs::remove_file(&self.path)
294            && error.kind() != std::io::ErrorKind::NotFound
295        {
296            tracing::warn!(?error, path = %self.path.display(), "cannot remove temporary deployment record");
297        }
298    }
299}
300
301#[cfg(all(test, not(target_arch = "wasm32")))]
302async fn write_file(data_dir: &Path, bytes: &[u8]) -> Result<(), StorageLocationError> {
303    write_file_with_guard(data_dir, bytes, None).await
304}
305
306#[cfg(not(target_arch = "wasm32"))]
307async fn write_file_with_guard(
308    data_dir: &Path,
309    bytes: &[u8],
310    record_guard: Option<tokio::sync::OwnedMutexGuard<()>>,
311) -> Result<(), StorageLocationError> {
312    xmtp_attachments::create_private_directory(data_dir).await?;
313    let path = data_dir.join("deployments.json");
314    let temp = data_dir.join(format!(
315        ".deployments-{}.tmp",
316        xmtp_common::rand_string::<16>()
317    ));
318    let bytes = bytes.to_vec();
319    xmtp_common::task::spawn_blocking(move || -> std::io::Result<()> {
320        let _record_guard = record_guard;
321        use std::io::Write as _;
322
323        let mut options = std::fs::OpenOptions::new();
324        options.write(true).create_new(true);
325        #[cfg(unix)]
326        {
327            use std::os::unix::fs::OpenOptionsExt as _;
328            options.mode(0o600);
329        }
330        let mut guard = NativeDeploymentTempGuard {
331            path: temp.clone(),
332            committed: true,
333        };
334        let mut file = options.open(&temp)?;
335        guard.committed = false;
336        #[cfg(test)]
337        let pause = { NATIVE_DEPLOYMENT_WRITE_PAUSE.lock().take() };
338        #[cfg(test)]
339        if let Some((entered, resume)) = pause {
340            entered.notify_one();
341            futures::executor::block_on(resume.notified());
342        }
343        let write = (|| -> std::io::Result<()> {
344            #[cfg(unix)]
345            {
346                use std::os::unix::fs::PermissionsExt as _;
347                file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
348            }
349            file.write_all(&bytes)?;
350            file.sync_all()
351        })();
352        drop(file);
353        write?;
354        std::fs::rename(&temp, &path)?;
355        guard.committed = true;
356        Ok(())
357    })
358    .await
359    .map_err(std::io::Error::other)??;
360    Ok(())
361}
362
363#[cfg(target_arch = "wasm32")]
364async fn read_file(data_dir: &Path) -> Result<DeploymentFile, StorageLocationError> {
365    use xmtp_attachments::{LocalStore, OpfsStore};
366    let store = OpfsStore::new(&data_dir.to_string_lossy())
367        .await
368        .map_err(|_| StorageLocationError::Opfs)?;
369    if !store
370        .exists("deployments.json")
371        .await
372        .map_err(|_| StorageLocationError::Opfs)?
373    {
374        return Ok(DeploymentFile::default());
375    }
376    let file = store
377        .open_read("deployments.json")
378        .await
379        .map_err(|_| StorageLocationError::Opfs)?;
380    let mut bytes = Vec::new();
381    let mut offset = 0;
382    while offset < file.len() {
383        let chunk = file
384            .read_chunk(offset, 64 * 1024)
385            .await
386            .map_err(|_| StorageLocationError::Opfs)?;
387        if chunk.is_empty() {
388            return Err(StorageLocationError::Opfs);
389        }
390        offset += chunk.len() as u64;
391        bytes.extend_from_slice(&chunk);
392    }
393    Ok(serde_json::from_slice(&bytes).unwrap_or_default())
394}
395
396#[cfg(target_arch = "wasm32")]
397struct DeploymentTempGuard {
398    store: xmtp_attachments::OpfsStore,
399    path: String,
400    committed: bool,
401}
402
403#[cfg(target_arch = "wasm32")]
404impl Drop for DeploymentTempGuard {
405    fn drop(&mut self) {
406        if self.committed {
407            return;
408        }
409        use xmtp_attachments::LocalStore;
410        let store = self.store.clone();
411        let path = self.path.clone();
412        drop(xmtp_common::task::spawn(async move {
413            if let Err(error) = store.remove_file(&path).await {
414                tracing::warn!(?error, %path, "cannot remove temporary deployment record");
415            }
416        }));
417    }
418}
419
420#[cfg(target_arch = "wasm32")]
421async fn write_file(data_dir: &Path, bytes: &[u8]) -> Result<(), StorageLocationError> {
422    write_file_with_guard(data_dir, bytes, None).await
423}
424
425#[cfg(target_arch = "wasm32")]
426async fn write_file_with_guard(
427    data_dir: &Path,
428    bytes: &[u8],
429    record_guard: Option<tokio::sync::OwnedMutexGuard<()>>,
430) -> Result<(), StorageLocationError> {
431    use xmtp_attachments::{DownloadSink, LocalStore, OpfsStore};
432    let store = OpfsStore::new(&data_dir.to_string_lossy())
433        .await
434        .map_err(|_| StorageLocationError::Opfs)?;
435    let temp = format!(".tmp/deployments-{}", xmtp_common::rand_string::<16>());
436    let mut writer = store
437        .create_temp(&temp)
438        .await
439        .map_err(|_| StorageLocationError::Opfs)?;
440    let guard = DeploymentTempGuard {
441        store: store.clone(),
442        path: temp.clone(),
443        committed: false,
444    };
445    #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
446    let fault = { DEPLOYMENT_WRITE_FAULT.lock().take() };
447    let write_result: Result<(), StorageLocationError> = async {
448        #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
449        if let Some(DeploymentWriteFault::Pause { entered, resume }) = &fault {
450            entered.notify_one();
451            resume.notified().await;
452        }
453        #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
454        if matches!(fault, Some(DeploymentWriteFault::Write)) {
455            return Err(StorageLocationError::Opfs);
456        }
457        writer
458            .write(bytes)
459            .await
460            .map_err(|_| StorageLocationError::Opfs)?;
461        #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
462        if matches!(fault, Some(DeploymentWriteFault::Sync)) {
463            return Err(StorageLocationError::Opfs);
464        }
465        store
466            .sync(&mut writer)
467            .await
468            .map_err(|_| StorageLocationError::Opfs)?;
469        Ok(())
470    }
471    .await;
472    drop(writer);
473    let task = xmtp_common::task::spawn(async move {
474        let _record_guard = record_guard;
475        let mut guard = guard;
476        #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
477        if let Some(DeploymentWriteFault::PauseReplace {
478            entered, resume, ..
479        }) = &fault
480        {
481            entered.notify_one();
482            resume.notified().await;
483        }
484        let result = match write_result {
485            Err(error) => Err(error),
486            Ok(()) => {
487                #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
488                if matches!(fault, Some(DeploymentWriteFault::Replace)) {
489                    Err(StorageLocationError::Opfs)
490                } else {
491                    store
492                        .replace(&temp, "deployments.json")
493                        .await
494                        .map_err(|_| StorageLocationError::Opfs)
495                }
496                #[cfg(not(all(target_arch = "wasm32", feature = "test-utils")))]
497                // OPFS file move replaces the destination in current browser engines.
498                store
499                    .replace(&temp, "deployments.json")
500                    .await
501                    .map_err(|_| StorageLocationError::Opfs)
502            }
503        };
504        if result.is_ok() {
505            guard.committed = true;
506        } else if let Err(error) = store.remove_file(&temp).await {
507            tracing::warn!(?error, %temp, "cannot remove temporary deployment record");
508        } else {
509            guard.committed = true;
510        }
511        #[cfg(all(target_arch = "wasm32", feature = "test-utils"))]
512        if let Some(DeploymentWriteFault::PauseReplace { settled, .. }) = &fault {
513            settled.notify_one();
514        }
515        result
516    });
517    task.await.map_err(|_| StorageLocationError::Opfs)?
518}
519
520#[cfg(test)]
521mod tests;
522
523#[cfg(all(test, not(target_arch = "wasm32")))]
524mod admission_tests;
xmtp_mls/storage_location.rs

xmtp_mls/
storage_location.rs
