feat: improve mobile native impl (#14481)

fix #13529 

#### PR Dependency Tree


* **PR #14481** 👈

This tree was auto-generated by
[Charcoal](https://github.com/danerwilliams/charcoal)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Mobile blob caching with file-backed storage for faster loads and
reduced network usage
* Blob decoding with lazy refresh on token-read failures for improved
reliability
  * Full-text search/indexing exposed to mobile apps
* Document sync APIs and peer clock management for robust cross-device
sync

* **Tests**
* Added unit tests covering payload decoding, cache safety, and
concurrency

* **Dependencies**
* Added an LRU cache dependency and a new mobile-shared package for
shared mobile logic
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

packages/frontend/mobile-native/src/storage/blobs.rs

@@ -0,0 +1,182 @@
+use super::*;
+
+#[uniffi::export(async_runtime = "tokio")]
+impl DocStoragePool {
+  pub async fn get_blob(&self, universal_id: String, key: String) -> Result<Option<Blob>> {
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    {
+      use affine_nbstore::Blob as NbBlob;
+      enum BlobEncodeOutcome {
+        Cached(Blob),
+        Inline(NbBlob),
+      }
+
+      let universal_id_for_cache = universal_id.clone();
+      let key_for_cache = key.clone();
+      if let Ok(Some(blob)) = self
+        .run_mobile_cache_io(
+          move |cache| Ok(cache.get_blob(&universal_id_for_cache, &key_for_cache)),
+          "Failed to read mobile blob cache",
+        )
+        .await
+      {
+        return Ok(Some(blob));
+      }
+
+      let Some(blob) = self
+        .inner
+        .get(universal_id.clone())
+        .await?
+        .get_blob(key.clone())
+        .await?
+      else {
+        return Ok(None);
+      };
+
+      if !should_cache_payload_as_file(blob.data.len()) {
+        return Ok(Some(blob.into()));
+      }
+
+      let universal_id_for_cache = universal_id.clone();
+      let key_for_fallback = key.clone();
+      return match self
+        .run_mobile_cache_io(
+          move |cache| {
+            Ok(match cache.cache_blob(&universal_id_for_cache, &blob) {
+              Ok(cached) => BlobEncodeOutcome::Cached(cached),
+              Err(_) => BlobEncodeOutcome::Inline(blob),
+            })
+          },
+          "Failed to cache blob file",
+        )
+        .await
+      {
+        Ok(BlobEncodeOutcome::Cached(cached)) => Ok(Some(cached)),
+        Ok(BlobEncodeOutcome::Inline(blob)) => Ok(Some(blob.into())),
+        Err(_) => Ok(
+          self
+            .inner
+            .get(universal_id)
+            .await?
+            .get_blob(key_for_fallback)
+            .await?
+            .map(Into::into),
+        ),
+      };
+    }
+
+    #[cfg(not(any(target_os = "android", target_os = "ios", test)))]
+    {
+      Ok(self.inner.get(universal_id).await?.get_blob(key).await?.map(Into::into))
+    }
+  }
+
+  pub async fn set_blob(&self, universal_id: String, blob: SetBlob) -> Result<()> {
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    let key = blob.key.clone();
+    let blob = NbSetBlob {
+      key: blob.key,
+      data: Into::<Data>::into(self.decode_blob_data(&universal_id, &blob.data).await?),
+      mime: blob.mime,
+    };
+    self.inner.get(universal_id.clone()).await?.set_blob(blob).await?;
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    {
+      let universal_id_for_cache = universal_id;
+      let _ = self
+        .run_mobile_cache_io(
+          move |cache| {
+            cache.invalidate_blob(&universal_id_for_cache, &key);
+            Ok(())
+          },
+          "Failed to invalidate mobile blob cache entry",
+        )
+        .await;
+    }
+    Ok(())
+  }
+
+  pub async fn delete_blob(&self, universal_id: String, key: String, permanently: bool) -> Result<()> {
+    self
+      .inner
+      .get(universal_id.clone())
+      .await?
+      .delete_blob(key.clone(), permanently)
+      .await?;
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    {
+      let universal_id_for_cache = universal_id;
+      let _ = self
+        .run_mobile_cache_io(
+          move |cache| {
+            cache.invalidate_blob(&universal_id_for_cache, &key);
+            Ok(())
+          },
+          "Failed to invalidate mobile blob cache entry",
+        )
+        .await;
+    }
+    Ok(())
+  }
+
+  pub async fn release_blobs(&self, universal_id: String) -> Result<()> {
+    self.inner.get(universal_id.clone()).await?.release_blobs().await?;
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    {
+      let universal_id_for_cache = universal_id;
+      let _ = self
+        .run_mobile_cache_io(
+          move |cache| {
+            cache.clear_workspace_cache(&universal_id_for_cache);
+            Ok(())
+          },
+          "Failed to clear mobile blob cache workspace",
+        )
+        .await;
+    }
+    Ok(())
+  }
+
+  pub async fn list_blobs(&self, universal_id: String) -> Result<Vec<ListedBlob>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .list_blobs()
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn set_blob_uploaded_at(
+    &self,
+    universal_id: String,
+    peer: String,
+    blob_id: String,
+    uploaded_at: Option<i64>,
+  ) -> Result<()> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .set_blob_uploaded_at(peer, blob_id, uploaded_at.map(millis_to_naive_utc).transpose()?)
+        .await?,
+    )
+  }
+
+  pub async fn get_blob_uploaded_at(&self, universal_id: String, peer: String, blob_id: String) -> Result<Option<i64>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_blob_uploaded_at(peer, blob_id)
+        .await?
+        .map(|t| t.and_utc().timestamp_millis()),
+    )
+  }
+}

packages/frontend/mobile-native/src/storage/docs.rs

@@ -0,0 +1,105 @@
+use super::*;
+
+#[uniffi::export(async_runtime = "tokio")]
+impl DocStoragePool {
+  pub async fn get_doc_snapshot(&self, universal_id: String, doc_id: String) -> Result<Option<DocRecord>> {
+    let Some(record) = self
+      .inner
+      .get(universal_id.clone())
+      .await?
+      .get_doc_snapshot(doc_id)
+      .await?
+    else {
+      return Ok(None);
+    };
+
+    let timestamp = record.timestamp.and_utc().timestamp_millis();
+    let bin = self
+      .encode_doc_data(&universal_id, &record.doc_id, timestamp, &record.bin)
+      .await?;
+    Ok(Some(DocRecord {
+      doc_id: record.doc_id,
+      bin,
+      timestamp,
+    }))
+  }
+
+  pub async fn set_doc_snapshot(&self, universal_id: String, snapshot: DocRecord) -> Result<bool> {
+    let doc_record = NbDocRecord {
+      doc_id: snapshot.doc_id,
+      bin: Into::<Data>::into(self.decode_base64_payload(&snapshot.bin)?),
+      timestamp: millis_to_naive_utc(snapshot.timestamp)?,
+    };
+    Ok(self.inner.get(universal_id).await?.set_doc_snapshot(doc_record).await?)
+  }
+
+  pub async fn get_doc_updates(&self, universal_id: String, doc_id: String) -> Result<Vec<DocUpdate>> {
+    let updates = self
+      .inner
+      .get(universal_id.clone())
+      .await?
+      .get_doc_updates(doc_id)
+      .await?;
+
+    let mut converted = Vec::with_capacity(updates.len());
+    for update in updates {
+      let timestamp = update.timestamp.and_utc().timestamp_millis();
+      let bin = self
+        .encode_doc_data(&universal_id, &update.doc_id, timestamp, &update.bin)
+        .await?;
+      converted.push(DocUpdate {
+        doc_id: update.doc_id,
+        timestamp,
+        bin,
+      });
+    }
+    Ok(converted)
+  }
+
+  pub async fn mark_updates_merged(&self, universal_id: String, doc_id: String, updates: Vec<i64>) -> Result<u32> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .mark_updates_merged(
+          doc_id,
+          updates
+            .into_iter()
+            .map(millis_to_naive_utc)
+            .collect::<Result<Vec<_>>>()?,
+        )
+        .await?,
+    )
+  }
+
+  pub async fn delete_doc(&self, universal_id: String, doc_id: String) -> Result<()> {
+    Ok(self.inner.get(universal_id).await?.delete_doc(doc_id).await?)
+  }
+
+  pub async fn get_doc_clocks(&self, universal_id: String, after: Option<i64>) -> Result<Vec<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_doc_clocks(after.map(millis_to_naive_utc).transpose()?)
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn get_doc_clock(&self, universal_id: String, doc_id: String) -> Result<Option<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_doc_clock(doc_id)
+        .await?
+        .map(Into::into),
+    )
+  }
+}

packages/frontend/mobile-native/src/storage/indexer.rs

@@ -0,0 +1,100 @@
+use super::*;
+
+#[uniffi::export(async_runtime = "tokio")]
+impl DocStoragePool {
+  pub async fn crawl_doc_data(&self, universal_id: String, doc_id: String) -> Result<CrawlResult> {
+    let result = self
+      .inner
+      .get(universal_id.clone())
+      .await?
+      .crawl_doc_data(&doc_id)
+      .await?;
+    Ok(result.into())
+  }
+
+  pub async fn fts_add_document(
+    &self,
+    universal_id: String,
+    index_name: String,
+    doc_id: String,
+    text: String,
+    index: bool,
+  ) -> Result<()> {
+    self
+      .inner
+      .get(universal_id)
+      .await?
+      .fts_add(&index_name, &doc_id, &text, index)
+      .await?;
+    Ok(())
+  }
+
+  pub async fn fts_delete_document(&self, universal_id: String, index_name: String, doc_id: String) -> Result<()> {
+    self
+      .inner
+      .get(universal_id)
+      .await?
+      .fts_delete(&index_name, &doc_id)
+      .await?;
+    Ok(())
+  }
+
+  pub async fn fts_get_document(
+    &self,
+    universal_id: String,
+    index_name: String,
+    doc_id: String,
+  ) -> Result<Option<String>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .fts_get(&index_name, &doc_id)
+        .await?,
+    )
+  }
+
+  pub async fn fts_search(&self, universal_id: String, index_name: String, query: String) -> Result<Vec<SearchHit>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .fts_search(&index_name, &query)
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn fts_get_matches(
+    &self,
+    universal_id: String,
+    index_name: String,
+    doc_id: String,
+    query: String,
+  ) -> Result<Vec<MatchRange>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .fts_get_matches(&index_name, &doc_id, &query)
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn fts_flush_index(&self, universal_id: String) -> Result<()> {
+    self.inner.get(universal_id).await?.flush_index().await?;
+    Ok(())
+  }
+
+  pub async fn fts_index_version(&self) -> Result<u32> {
+    Ok(SqliteDocStorage::index_version())
+  }
+}

packages/frontend/mobile-native/src/storage/lifecycle.rs

@@ -0,0 +1,74 @@
+use super::*;
+
+#[uniffi::export(async_runtime = "tokio")]
+impl DocStoragePool {
+  /// Initialize the database and run migrations.
+  pub async fn connect(&self, universal_id: String, path: String) -> Result<()> {
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    {
+      let universal_id_for_cache = universal_id.clone();
+      let path_for_cache = path.clone();
+      self
+        .run_mobile_cache_io(
+          move |cache| cache.register_workspace(&universal_id_for_cache, &path_for_cache),
+          "Failed to initialize mobile blob cache",
+        )
+        .await?;
+    }
+
+    if let Err(err) = self.inner.connect(universal_id.clone(), path).await {
+      #[cfg(any(target_os = "android", target_os = "ios", test))]
+      {
+        let universal_id_for_cache = universal_id.clone();
+        let _ = self
+          .run_mobile_cache_io(
+            move |cache| {
+              cache.invalidate_workspace(&universal_id_for_cache);
+              Ok(())
+            },
+            "Failed to rollback mobile blob cache workspace",
+          )
+          .await;
+      }
+      return Err(err.into());
+    }
+
+    Ok(())
+  }
+
+  pub async fn disconnect(&self, universal_id: String) -> Result<()> {
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    {
+      let universal_id_for_cache = universal_id.clone();
+      let _ = self
+        .run_mobile_cache_io(
+          move |cache| {
+            cache.invalidate_workspace(&universal_id_for_cache);
+            Ok(())
+          },
+          "Failed to clear mobile blob cache workspace",
+        )
+        .await;
+    }
+    self.inner.disconnect(universal_id).await?;
+    Ok(())
+  }
+
+  pub async fn set_space_id(&self, universal_id: String, space_id: String) -> Result<()> {
+    Ok(self.inner.get(universal_id).await?.set_space_id(space_id).await?)
+  }
+
+  pub async fn push_update(&self, universal_id: String, doc_id: String, update: String) -> Result<i64> {
+    let decoded_update = self.decode_base64_payload(&update)?;
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .push_update(doc_id, decoded_update)
+        .await?
+        .and_utc()
+        .timestamp_millis(),
+    )
+  }
+}

packages/frontend/mobile-native/src/storage/mod.rs

@@ -0,0 +1,90 @@
+mod blobs;
+mod docs;
+mod indexer;
+mod lifecycle;
+mod peers;
+
+#[cfg(any(target_os = "android", target_os = "ios", test))]
+use std::sync::Arc;
+
+use affine_nbstore::{
+  Data, DocRecord as NbDocRecord, SetBlob as NbSetBlob, pool::SqliteDocStoragePool, storage::SqliteDocStorage,
+};
+use chrono::{DateTime, NaiveDateTime, Utc};
+
+#[cfg(any(target_os = "android", target_os = "ios", test))]
+use crate::cache::{MobileBlobCache, is_mobile_binary_file_token, should_cache_payload_as_file};
+use crate::{
+  Blob, CrawlResult, DocClock, DocRecord, DocUpdate, ListedBlob, MatchRange, Result, SearchHit, SetBlob, UniffiError,
+  payload_codec::{decode_base64_data, encode_base64_data},
+};
+
+fn millis_to_naive_utc(millis: i64) -> Result<NaiveDateTime> {
+  DateTime::<Utc>::from_timestamp_millis(millis)
+    .ok_or(UniffiError::TimestampDecodingError)
+    .map(|value| value.naive_utc())
+}
+
+#[derive(uniffi::Object)]
+pub struct DocStoragePool {
+  inner: SqliteDocStoragePool,
+  #[cfg(any(target_os = "android", target_os = "ios", test))]
+  mobile_blob_cache: Arc<MobileBlobCache>,
+}
+
+#[uniffi::export]
+pub fn new_doc_storage_pool() -> DocStoragePool {
+  DocStoragePool {
+    inner: Default::default(),
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    mobile_blob_cache: Arc::new(MobileBlobCache::new()),
+  }
+}
+
+impl DocStoragePool {
+  #[cfg(any(target_os = "android", target_os = "ios", test))]
+  async fn run_mobile_cache_io<T, F>(&self, task: F, context: &'static str) -> Result<T>
+  where
+    T: Send + 'static,
+    F: FnOnce(Arc<MobileBlobCache>) -> std::io::Result<T> + Send + 'static,
+  {
+    let cache = Arc::clone(&self.mobile_blob_cache);
+    tokio::task::spawn_blocking(move || task(cache))
+      .await
+      .map_err(|err| UniffiError::Err(format!("{context}: {err}")))?
+      .map_err(|err| UniffiError::Err(format!("{context}: {err}")))
+  }
+
+  pub(crate) fn decode_base64_payload(&self, data: &str) -> Result<Vec<u8>> {
+    decode_base64_data(data)
+  }
+
+  pub(crate) async fn decode_blob_data(&self, universal_id: &str, data: &str) -> Result<Vec<u8>> {
+    #[cfg(any(target_os = "android", target_os = "ios", test))]
+    if is_mobile_binary_file_token(data) {
+      let universal_id = universal_id.to_string();
+      let data = data.to_string();
+      return self
+        .run_mobile_cache_io(
+          move |cache| cache.read_binary_file(&universal_id, &data),
+          "Failed to read mobile file token",
+        )
+        .await;
+    }
+    #[cfg(not(any(target_os = "android", target_os = "ios", test)))]
+    let _ = universal_id;
+
+    self.decode_base64_payload(data)
+  }
+
+  pub(crate) async fn encode_doc_data(
+    &self,
+    universal_id: &str,
+    doc_id: &str,
+    timestamp: i64,
+    data: &[u8],
+  ) -> Result<String> {
+    let _ = (universal_id, doc_id, timestamp);
+    Ok(encode_base64_data(data))
+  }
+}

packages/frontend/mobile-native/src/storage/peers.rs

@@ -0,0 +1,152 @@
+use super::*;
+
+#[uniffi::export(async_runtime = "tokio")]
+impl DocStoragePool {
+  pub async fn get_peer_remote_clocks(&self, universal_id: String, peer: String) -> Result<Vec<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_peer_remote_clocks(peer)
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn get_peer_remote_clock(
+    &self,
+    universal_id: String,
+    peer: String,
+    doc_id: String,
+  ) -> Result<Option<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_peer_remote_clock(peer, doc_id)
+        .await?
+        .map(Into::into),
+    )
+  }
+
+  pub async fn set_peer_remote_clock(
+    &self,
+    universal_id: String,
+    peer: String,
+    doc_id: String,
+    clock: i64,
+  ) -> Result<()> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .set_peer_remote_clock(peer, doc_id, millis_to_naive_utc(clock)?)
+        .await?,
+    )
+  }
+
+  pub async fn get_peer_pulled_remote_clocks(&self, universal_id: String, peer: String) -> Result<Vec<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_peer_pulled_remote_clocks(peer)
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn get_peer_pulled_remote_clock(
+    &self,
+    universal_id: String,
+    peer: String,
+    doc_id: String,
+  ) -> Result<Option<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_peer_pulled_remote_clock(peer, doc_id)
+        .await?
+        .map(Into::into),
+    )
+  }
+
+  pub async fn set_peer_pulled_remote_clock(
+    &self,
+    universal_id: String,
+    peer: String,
+    doc_id: String,
+    clock: i64,
+  ) -> Result<()> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .set_peer_pulled_remote_clock(peer, doc_id, millis_to_naive_utc(clock)?)
+        .await?,
+    )
+  }
+
+  pub async fn get_peer_pushed_clock(
+    &self,
+    universal_id: String,
+    peer: String,
+    doc_id: String,
+  ) -> Result<Option<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_peer_pushed_clock(peer, doc_id)
+        .await?
+        .map(Into::into),
+    )
+  }
+
+  pub async fn get_peer_pushed_clocks(&self, universal_id: String, peer: String) -> Result<Vec<DocClock>> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .get_peer_pushed_clocks(peer)
+        .await?
+        .into_iter()
+        .map(Into::into)
+        .collect(),
+    )
+  }
+
+  pub async fn set_peer_pushed_clock(
+    &self,
+    universal_id: String,
+    peer: String,
+    doc_id: String,
+    clock: i64,
+  ) -> Result<()> {
+    Ok(
+      self
+        .inner
+        .get(universal_id)
+        .await?
+        .set_peer_pushed_clock(peer, doc_id, millis_to_naive_utc(clock)?)
+        .await?,
+    )
+  }
+
+  pub async fn clear_clocks(&self, universal_id: String) -> Result<()> {
+    Ok(self.inner.get(universal_id).await?.clear_clocks().await?)
+  }
+}