Clean up Rust by renaming functions, extracting logic, and auto into-ing errors

sync-server/.env

@@ -1 +0,0 @@
-DATABASE_URL=sqlite://db.sqlite3

sync-server/Cargo.toml

@@ -21,7 +21,7 @@ tower-http = { version = "0.6.1", features = ["cors", "trace", "limit", "timeout
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.20", features = ["fmt", "env-filter"]}
 humantime-serde = "1.1.1"
-sqlx = { version = "0.8.6", features = ["sqlite", "runtime-tokio", "uuid", "chrono"] }
+sqlx = { version = "0.8.6", default-features = false, features = ["sqlite", "runtime-tokio", "uuid", "chrono", "macros"] }
 chrono = { version = "0.4.41", features = ["serde"] }
 rand = "0.9.0"
 regex = "1.12.2"

sync-server/src/app_state.rs

@@ -24,7 +24,7 @@ impl AppState {
         let database = Database::try_new(&config.database).await?;
         let cursors: Cursors = Cursors::new(&config.database, &broadcasts);
 
-        Cursors::start_background_task(cursors.clone());
+        Cursors::start_cursor_expiration_task(cursors.clone());
 
         Ok(Self {
             config,

sync-server/src/app_state/cursors.rs

@@ -95,7 +95,7 @@ impl Cursors {
             .unwrap_or_default()
     }
 
-    pub fn start_background_task(self) {
+    pub fn start_cursor_expiration_task(self) {
         tokio::spawn(async move {
             loop {
                 self.remove_expired_cursors().await;

sync-server/src/app_state/database.rs

@@ -9,13 +9,14 @@ use anyhow::{Context as _, Result};
 use log::info;
 use models::VaultId;
 use sha2::{Digest, Sha256};
-use sqlx::{ConnectOptions, sqlite::SqliteConnectOptions};
+use sqlx::{ConnectOptions, Executor, sqlite::SqliteConnectOptions};
 use sqlx::{Pool, Sqlite, sqlite::SqlitePoolOptions};
 use tokio::sync::Mutex;
 use tokio::time::Instant;
 
 pub mod models;
 
+mod history;
 mod mutations;
 mod queries;
 
@@ -83,7 +84,7 @@ impl Database {
         Ok(database)
     }
 
-    fn database_path(config: &DatabaseConfig, vault: &str) -> PathBuf {
+    fn get_database_path(config: &DatabaseConfig, vault: &str) -> PathBuf {
         // Hash the normalized vault name into a fixed-length ASCII filename so
         // filesystem case/Unicode aliasing (e.g. composed vs. decomposed é on
         // macOS) cannot make separately authorized vaults share a database.
@@ -149,7 +150,7 @@ impl Database {
         let vault = normalize_string(vault);
         validate_vault_id(&vault)?;
 
-        if let Some(pool) = self.existing_pool(&vault).await {
+        if let Some(pool) = self.find_cached_pool(&vault).await {
             return Ok(pool);
         }
 
@@ -160,11 +161,11 @@ impl Database {
 
         let _opening = opening.lock().await;
 
-        if let Some(pool) = self.existing_pool(&vault).await {
+        if let Some(pool) = self.find_cached_pool(&vault).await {
             return Ok(pool);
         }
 
-        let file_name = Self::database_path(&self.config, &vault);
+        let file_name = Self::get_database_path(&self.config, &vault);
         let pool = Self::open_database(&self.config, &file_name).await?;
 
         self.connection_pools.lock().await.insert(
@@ -177,7 +178,7 @@ impl Database {
         Ok(pool)
     }
 
-    async fn existing_pool(&self, vault: &VaultId) -> Option<Pool<Sqlite>> {
+    async fn find_cached_pool(&self, vault: &VaultId) -> Option<Pool<Sqlite>> {
         self.connection_pools
             .lock()
             .await

sync-server/src/app_state/database/history.rs

@@ -0,0 +1,48 @@
+//! Checkpoints distinguish restored histories even when numeric event IDs are reused.
+use super::Database;
+use super::models::VaultId;
+use anyhow::Result;
+
+const EMPTY_HISTORY_CHECKPOINT: &str = "0:empty";
+
+impl Database {
+    pub async fn get_history_checkpoint(&self, vault: &VaultId) -> Result<String> {
+        let mut tx = self.create_readonly_transaction(vault).await?;
+
+        let row: Option<(i64, String)> = sqlx::query_as(
+            "SELECT event_id, event_token FROM events ORDER BY event_id DESC LIMIT 1",
+        )
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        Ok(row.map_or_else(
+            || EMPTY_HISTORY_CHECKPOINT.to_owned(),
+            |(id, token)| format!("{id}:{token}"),
+        ))
+    }
+
+    pub async fn contains_checkpoint(&self, vault: &VaultId, checkpoint: &str) -> Result<bool> {
+        if checkpoint == EMPTY_HISTORY_CHECKPOINT {
+            return Ok(true);
+        }
+
+        let Some((id, token)) = checkpoint.split_once(':') else {
+            return Ok(false);
+        };
+
+        let Ok(id) = id.parse::<i64>() else {
+            return Ok(false);
+        };
+
+        let mut tx = self.create_readonly_transaction(vault).await?;
+        let exists: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM events WHERE event_id = ? AND event_token = ?)",
+        )
+        .bind(id)
+        .bind(token)
+        .fetch_one(&mut *tx)
+        .await?;
+
+        Ok(exists)
+    }
+}

sync-server/src/app_state/database/models.rs

@@ -125,9 +125,8 @@ pub struct EventBatch {
     pub head_event_id: VaultUpdateId,
 
     /// Last event in this page; `head_event_id` is the vault's current head.
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[ts(as = "f64")]
-    #[ts(optional, as = "Option<f64>")]
+    pub end_event_id: VaultUpdateId,
-    pub end_event_id: Option<VaultUpdateId>,
 
     pub events: Vec<EventRecord>,
 }

sync-server/src/app_state/database/queries.rs

@@ -9,8 +9,6 @@ use anyhow::{Context as _, Result, ensure};
 
 use crate::consts::MAX_EVENTS_PAGE_BYTES;
 
-const EMPTY_HISTORY_CHECKPOINT: &str = "0:empty";
-
 impl Database {
     pub async fn get_missing_document(
         tx: &mut Transaction<'_>,
@@ -36,9 +34,10 @@ impl Database {
                 .fetch_optional(&mut **tx)
                 .await?;
         match row {
-            Some((fingerprint, event_id)) => {
+            Some((fingerprint, event_id)) => Ok(Some((
-                Ok(Some((fingerprint, Self::event_by_id(tx, event_id).await?)))
+                fingerprint,
-            }
+                Self::get_event_by_id(tx, event_id).await?,
+            ))),
             None => Ok(None),
         }
     }
@@ -96,7 +95,7 @@ impl Database {
         })
     }
 
-    pub async fn current_file_manifest(tx: &mut Transaction<'_>) -> Result<FileManifest> {
+    pub async fn get_current_file_manifest(tx: &mut Transaction<'_>) -> Result<FileManifest> {
         let Some(file_manifest_id) = sqlx::query_scalar::<_, VaultUpdateId>(
             "SELECT file_manifest_id FROM file_manifests ORDER BY file_manifest_id DESC LIMIT 1",
         )
@@ -106,12 +105,12 @@ impl Database {
             return Ok(FileManifest::default());
         };
 
-        Self::file_manifest_by_id(tx, file_manifest_id)
+        Self::get_file_manifest_by_id(tx, file_manifest_id)
             .await?
             .context("Latest file manifest is missing")
     }
 
-    async fn file_manifest_by_id(
+    async fn get_file_manifest_by_id(
         tx: &mut Transaction<'_>,
         file_manifest_id: VaultUpdateId,
     ) -> Result<Option<FileManifest>> {
@@ -140,7 +139,10 @@ impl Database {
         }))
     }
 
-    async fn event_by_id(tx: &mut Transaction<'_>, event_id: VaultUpdateId) -> Result<VaultEvent> {
+    async fn get_event_by_id(
+        tx: &mut Transaction<'_>,
+        event_id: VaultUpdateId,
+    ) -> Result<VaultEvent> {
         let document = sqlx::query_as::<_, DocumentVersionWithoutContent>(
             "SELECT vault_update_id, document_id, updated_date, user_id, device_id,
                     length(content) AS content_size
@@ -152,13 +154,13 @@ impl Database {
         if let Some(document) = document {
             return Ok(VaultEvent::Content { document });
         }
-        let file_manifest = Self::file_manifest_by_id(tx, event_id)
+        let file_manifest = Self::get_file_manifest_by_id(tx, event_id)
             .await?
             .context("Event has no normalized content or file manifest")?;
         Ok(VaultEvent::FileManifest { file_manifest })
     }
 
-    pub async fn latest_event_id(tx: &mut Transaction<'_>) -> Result<VaultUpdateId> {
+    pub async fn get_latest_event_id(tx: &mut Transaction<'_>) -> Result<VaultUpdateId> {
         Ok(
             sqlx::query_scalar("SELECT COALESCE(MAX(event_id), 0) FROM events")
                 .fetch_one(&mut **tx)
@@ -166,10 +168,10 @@ impl Database {
         )
     }
 
-    pub async fn vault_snapshot(&self, vault: &VaultId) -> Result<VaultSnapshot> {
+    pub async fn get_vault_snapshot(&self, vault: &VaultId) -> Result<VaultSnapshot> {
         let mut tx = self.create_readonly_transaction(vault).await?;
-        let head_event_id = Self::latest_event_id(&mut tx).await?;
+        let head_event_id = Self::get_latest_event_id(&mut tx).await?;
-        let file_manifest = Self::current_file_manifest(&mut tx).await?;
+        let file_manifest = Self::get_current_file_manifest(&mut tx).await?;
 
         let documents = sqlx::query_as::<_, DocumentVersionWithoutContent>(
             "SELECT d.vault_update_id, d.document_id, d.updated_date, d.user_id, d.device_id,
@@ -189,14 +191,14 @@ impl Database {
         })
     }
 
-    pub async fn events_after(
+    pub async fn get_events_after(
         &self,
         vault: &VaultId,
         after: VaultUpdateId,
         page_size: usize,
     ) -> Result<EventBatch> {
         let mut tx = self.create_readonly_transaction(vault).await?;
-        let head_event_id = Self::latest_event_id(&mut tx).await?;
+        let head_event_id = Self::get_latest_event_id(&mut tx).await?;
 
         ensure!(
             (0..=head_event_id).contains(&after),
@@ -218,7 +220,7 @@ impl Database {
             let event = EventRecord {
                 event_id,
                 request_id: request_id.parse()?,
-                event: Self::event_by_id(&mut tx, event_id).await?,
+                event: Self::get_event_by_id(&mut tx, event_id).await?,
             };
             let size = serde_json::to_vec(&event)?.len();
             // One manifest is atomic and may exceed this budget. Always allow
@@ -233,50 +235,8 @@ impl Database {
 
         Ok(EventBatch {
             head_event_id,
-            end_event_id: Some(end_event_id),
+            end_event_id,
             events,
         })
     }
 }
-
-impl Database {
-    pub async fn history_checkpoint(&self, vault: &VaultId) -> Result<String> {
-        let mut tx = self.create_readonly_transaction(vault).await?;
-
-        let row: Option<(i64, String)> = sqlx::query_as(
-            "SELECT event_id, event_token FROM events ORDER BY event_id DESC LIMIT 1",
-        )
-        .fetch_optional(&mut *tx)
-        .await?;
-
-        Ok(row.map_or_else(
-            || EMPTY_HISTORY_CHECKPOINT.to_owned(),
-            |(id, token)| format!("{id}:{token}"),
-        ))
-    }
-
-    pub async fn contains_checkpoint(&self, vault: &VaultId, checkpoint: &str) -> Result<bool> {
-        if checkpoint == EMPTY_HISTORY_CHECKPOINT {
-            return Ok(true);
-        }
-
-        let Some((id, token)) = checkpoint.split_once(':') else {
-            return Ok(false);
-        };
-
-        let Ok(id) = id.parse::<i64>() else {
-            return Ok(false);
-        };
-
-        let mut tx = self.create_readonly_transaction(vault).await?;
-        let exists: bool = sqlx::query_scalar(
-            "SELECT EXISTS(SELECT 1 FROM events WHERE event_id = ? AND event_token = ?)",
-        )
-        .bind(id)
-        .bind(token)
-        .fetch_one(&mut *tx)
-        .await?;
-
-        Ok(exists)
-    }
-}

sync-server/src/app_state/database/tests.rs

@@ -87,7 +87,7 @@ async fn rejects_invalid_vault_names_at_the_database_boundary() {
         "a\0b",
     ] {
         assert!(
-            database.vault_snapshot(&name.to_owned()).await.is_err(),
+            database.get_vault_snapshot(&name.to_owned()).await.is_err(),
             "accepted {name:?}"
         );
     }
@@ -96,15 +96,15 @@ async fn rejects_invalid_vault_names_at_the_database_boundary() {
 }
 
 #[tokio::test]
-async fn migrations_can_be_reapplied_without_losing_requests_or_content() {
+async fn schema_initialization_preserves_existing_requests_and_content() {
     let directory = tempfile::tempdir().unwrap();
     let config = config(&directory);
     let database = Database::try_new(&config).await.unwrap();
-    let vault = "migrations".to_owned();
+    let vault = "initialization".to_owned();
     let version = insert_content(&database, &vault, uuid::Uuid::new_v4(), b"only copy").await;
     let pool = database.get_connection_pool(&vault).await.unwrap();
-    Database::run_migrations(&pool).await.unwrap();
+    Database::initialize_schema(&pool).await.unwrap();
-    Database::run_migrations(&pool).await.unwrap();
+    Database::initialize_schema(&pool).await.unwrap();
     let reopened = Database::try_new(&config).await.unwrap();
     assert_eq!(
         reopened
@@ -117,7 +117,7 @@ async fn migrations_can_be_reapplied_without_losing_requests_or_content() {
     );
     assert_eq!(
         reopened
-            .events_after(&vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap()
             .events
@@ -155,7 +155,7 @@ async fn snapshot_returns_latest_referenced_metadata_and_byte_lengths() {
         .unwrap();
     tx.commit().await.unwrap();
 
-    let snapshot = database.vault_snapshot(&vault).await.unwrap();
+    let snapshot = database.get_vault_snapshot(&vault).await.unwrap();
     assert_eq!(snapshot.head_event_id, manifest.file_manifest_id);
     assert_eq!(snapshot.file_manifest.entries, manifest.entries);
     assert_eq!(snapshot.documents.len(), 2);
@@ -179,7 +179,7 @@ async fn replay_pages_are_bounded_and_cover_every_event_in_order() {
     let mut after = 0;
     while after < 150 {
         let batch = database
-            .events_after(&vault, after, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault, after, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap();
         assert!(batch.events.len() <= 64, "catchup must bound each page");
@@ -192,7 +192,7 @@ async fn replay_pages_are_bounded_and_cover_every_event_in_order() {
     }
     assert!(
         database
-            .events_after(&vault, after, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault, after, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap()
             .events
@@ -204,7 +204,10 @@ async fn replay_pages_are_bounded_and_cover_every_event_in_order() {
 async fn idle_cleanup_does_not_wait_for_an_active_vault() {
     let directory = tempfile::tempdir().unwrap();
     let database = Database::try_new(&config(&directory)).await.unwrap();
-    database.vault_snapshot(&"other".to_owned()).await.unwrap();
+    database
+        .get_vault_snapshot(&"other".to_owned())
+        .await
+        .unwrap();
     let vault = "slow".to_owned();
     let transaction = database.create_write_transaction(&vault).await.unwrap();
     database
@@ -219,7 +222,7 @@ async fn idle_cleanup_does_not_wait_for_an_active_vault() {
         .expect("cleanup must leave an active pool alone");
     let other = tokio::time::timeout(
         Duration::from_millis(200),
-        database.vault_snapshot(&"other".to_owned()),
+        database.get_vault_snapshot(&"other".to_owned()),
     )
     .await
     .expect("other vaults must remain accessible")
@@ -299,16 +302,13 @@ async fn replay_byte_budget_prevents_accumulating_large_historical_manifests() {
     }
     tx.commit().await.unwrap();
     let batch = database
-        .events_after(&vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
+        .get_events_after(&vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
         .await
         .unwrap();
     assert!(!batch.events.is_empty());
     assert!(batch.events.len() < 8);
     assert!(serde_json::to_vec(&batch).unwrap().len() < MAX_EVENTS_PAGE_BYTES + 1024);
-    assert_eq!(
+    assert_eq!(batch.end_event_id, batch.events.last().unwrap().event_id);
-        batch.end_event_id,
-        batch.events.last().map(|event| event.event_id)
-    );
 }
 
 #[tokio::test]
@@ -316,11 +316,14 @@ async fn opening_a_locked_vault_does_not_block_unrelated_vaults() {
     let directory = tempfile::tempdir().unwrap();
     let config = config(&directory);
     let database = Database::try_new(&config).await.unwrap();
-    database.vault_snapshot(&"other".to_owned()).await.unwrap();
+    database
+        .get_vault_snapshot(&"other".to_owned())
+        .await
+        .unwrap();
     let vault = "locked-opening".to_owned();
-    // A second process holds the writer lock while this process lazily checks
+    // A second process holds the writer lock while this process initializes
-    // migrations. Even already-open, unrelated vaults must remain available.
+    // the schema. Even already-open, unrelated vaults must remain available.
-    let external = Database::open_database(&config, &Database::database_path(&config, &vault))
+    let external = Database::open_database(&config, &Database::get_database_path(&config, &vault))
         .await
         .unwrap();
     let transaction = external.begin_with("BEGIN IMMEDIATE").await.unwrap();
@@ -330,7 +333,7 @@ async fn opening_a_locked_vault_does_not_block_unrelated_vaults() {
     let opening_is_blocked = !opening.is_finished();
     let other = tokio::time::timeout(
         Duration::from_millis(300),
-        database.vault_snapshot(&"other".to_owned()),
+        database.get_vault_snapshot(&"other".to_owned()),
     )
     .await;
     transaction.rollback().await.unwrap();
@@ -346,7 +349,7 @@ async fn opening_a_locked_vault_does_not_block_unrelated_vaults() {
     );
     assert_eq!(
         other
-            .expect("a blocked migration must not lock every vault")
+            .expect("a blocked database open must not lock every vault")
             .unwrap()
             .head_event_id,
         0
@@ -360,7 +363,7 @@ async fn concurrent_vault_opens_share_a_pool_and_cancelled_opens_can_retry() {
     config.max_connections_per_vault = 1;
     let database = Database::try_new(&config).await.unwrap();
     let vault = "cancelled-opening".to_owned();
-    let external = Database::open_database(&config, &Database::database_path(&config, &vault))
+    let external = Database::open_database(&config, &Database::get_database_path(&config, &vault))
         .await
         .unwrap();
     let transaction = external.begin_with("BEGIN IMMEDIATE").await.unwrap();
@@ -401,7 +404,7 @@ async fn concurrent_vault_opens_share_a_pool_and_cancelled_opens_can_retry() {
 async fn one_corrupt_vault_does_not_prevent_startup_or_healthy_vault_access() {
     let directory = tempfile::tempdir().unwrap();
     let config = config(&directory);
-    let bad = Database::database_path(&config, "broken");
+    let bad = Database::get_database_path(&config, "broken");
     tokio::fs::create_dir_all(bad.parent().unwrap())
         .await
         .unwrap();
@@ -409,7 +412,12 @@ async fn one_corrupt_vault_does_not_prevent_startup_or_healthy_vault_access() {
         .await
         .unwrap();
     let database = Database::try_new(&config).await.unwrap();
-    assert!(database.vault_snapshot(&"broken".to_owned()).await.is_err());
+    assert!(
+        database
+            .get_vault_snapshot(&"broken".to_owned())
+            .await
+            .is_err()
+    );
     let saved = insert_content(&database, "healthy", uuid::Uuid::new_v4(), b"safe").await;
     assert_eq!(saved.content, b"safe");
     assert_eq!(

sync-server/src/app_state/websocket/broadcasts.rs

@@ -39,7 +39,7 @@ impl Broadcasts {
         limit.try_acquire_owned().ok()
     }
 
-    async fn sender(&self, vault: VaultId) -> broadcast::Sender<Notification> {
+    async fn get_or_create_sender(&self, vault: VaultId) -> broadcast::Sender<Notification> {
         self.senders
             .lock()
             .await
@@ -49,12 +49,12 @@ impl Broadcasts {
     }
 
     pub async fn get_receiver(&self, vault: VaultId) -> broadcast::Receiver<Notification> {
-        self.sender(vault).await.subscribe()
+        self.get_or_create_sender(vault).await.subscribe()
     }
 
     pub async fn send(&self, vault: VaultId, notification: Notification) {
         // No listeners (or lagging listeners) is normal; they replay the log.
-        let _ = self.sender(vault).await.send(notification);
+        let _ = self.get_or_create_sender(vault).await.send(notification);
     }
 
     pub async fn notify_about_vault_update(&self, vault: VaultId) {

sync-server/src/app_state/websocket/utils.rs

@@ -8,7 +8,7 @@ use crate::{
     app_state::{AppState, database::models::VaultId},
     config::user_config::User,
     errors::{SyncServerError, server_error, unauthenticated_error},
-    server::auth::auth,
+    server::auth::authenticate_user,
 };
 
 pub struct AuthenticatedWebSocketHandshake {
@@ -28,7 +28,7 @@ pub fn get_authenticated_handshake(
 
         match message {
             WebSocketClientMessage::Handshake(handshake) => {
-                let user = auth(state, handshake.token.trim(), vault_id)?;
+                let user = authenticate_user(state, handshake.token.trim(), vault_id)?;
                 Ok(AuthenticatedWebSocketHandshake { handshake, user })
             }
             WebSocketClientMessage::CursorPositions(_) => Err(unauthenticated_error(

sync-server/src/errors.rs

@@ -31,6 +31,26 @@ pub enum SyncServerError {
     PermissionDeniedError(#[source] anyhow::Error),
 }
 
+// Infrastructure failures are server errors. Request validation and missing
+// resources opt into their HTTP status explicitly at the point of validation.
+impl From<anyhow::Error> for SyncServerError {
+    fn from(error: anyhow::Error) -> Self {
+        server_error(error)
+    }
+}
+
+impl From<sqlx::Error> for SyncServerError {
+    fn from(error: sqlx::Error) -> Self {
+        server_error(error.into())
+    }
+}
+
+impl From<tokio::task::JoinError> for SyncServerError {
+    fn from(error: tokio::task::JoinError) -> Self {
+        server_error(error.into())
+    }
+}
+
 impl SyncServerError {
     pub fn serialize(&self) -> SerializedError {
         match self {

sync-server/src/server.rs

@@ -46,7 +46,7 @@ pub async fn create_server(config: Config) -> Result<()> {
     let server_config = app_state.config.server.clone();
 
     let app = Router::new()
-        .nest("/", get_authed_routes(app_state.clone()))
+        .nest("/", create_authenticated_routes(app_state.clone()))
         .route("/", get(endpoints::index::index))
         .route("/vaults/:vault_id/ping", get(endpoints::ping::ping))
         .route(
@@ -100,7 +100,7 @@ pub async fn create_server(config: Config) -> Result<()> {
     start_server(app, &server_config).await
 }
 
-fn get_authed_routes(app_state: AppState) -> Router<AppState> {
+fn create_authenticated_routes(app_state: AppState) -> Router<AppState> {
     Router::new()
         .route(
             "/vaults/:vault_id/file-manifest",

sync-server/src/server/auth.rs

@@ -32,14 +32,18 @@ pub async fn auth_middleware(
             .ok_or_else(|| unauthenticated_error(anyhow::anyhow!("Missing vault_id")))?,
     );
 
-    let user = auth(&state, token, &vault_id)?;
+    let user = authenticate_user(&state, token, &vault_id)?;
 
     req.extensions_mut().insert(user);
 
     Ok(next.run(req).await)
 }
 
-pub fn auth(state: &AppState, token: &str, vault_id: &VaultId) -> Result<User, SyncServerError> {
+pub fn authenticate_user(
+    state: &AppState,
+    token: &str,
+    vault_id: &VaultId,
+) -> Result<User, SyncServerError> {
     validate_vault_id(vault_id).map_err(client_error)?;
     let user = state
         .config

sync-server/src/server/endpoints/events.rs

@@ -35,7 +35,7 @@ pub async fn events(
 
     state
         .database
-        .events_after(
+        .get_events_after(
             &vault_id,
             query.after,
             query.page_size.unwrap_or(DEFAULT_EVENTS_PAGE_SIZE),

sync-server/src/server/endpoints/fetch_document_version_content.rs

@@ -11,7 +11,7 @@ use crate::{
         AppState,
         database::models::{DocumentId, VaultUpdateId},
     },
-    errors::{SyncServerError, not_found_error, server_error},
+    errors::{SyncServerError, not_found_error},
 };
 
 #[axum::debug_handler]
@@ -30,16 +30,12 @@ pub async fn fetch_document_version_content(
     let result = state
         .database
         .get_document_version(&vault_id, vault_update_id, None)
-        .await
+        .await?
-        .map_err(server_error)?
+        .ok_or_else(|| {
-        .map_or_else(
+            not_found_error(anyhow!(
-            || {
-                Err(not_found_error(anyhow!(
                 "Document with vault update id `{vault_update_id}` not found",
-                )))
+            ))
-            },
+        })?;
-            Ok,
-        )?;
 
     if result.document_id != document_id {
         return Err(not_found_error(anyhow!(

sync-server/src/server/endpoints/fetch_latest_document_metadata.rs

@@ -14,7 +14,7 @@ use crate::{
             models::{DocumentId, DocumentVersionWithoutContent},
         },
     },
-    errors::{SyncServerError, not_found_error, server_error},
+    errors::{SyncServerError, not_found_error},
 };
 
 #[axum::debug_handler]
@@ -27,12 +27,10 @@ pub async fn fetch_latest_document_metadata(
     let mut tx = state
         .database
         .create_readonly_transaction(&vault_id)
-        .await
+        .await?;
-        .map_err(server_error)?;
 
     let metadata = Database::get_latest_document_metadata(&mut tx, &document_id)
-        .await
+        .await?
-        .map_err(server_error)?
         .ok_or_else(|| not_found_error(anyhow!("Document does not exist")))?;
 
     Ok(Json(metadata))

sync-server/src/server/endpoints/fetch_latest_document_version.rs

@@ -11,7 +11,7 @@ use crate::{
         AppState,
         database::models::{DocumentId, DocumentVersion},
     },
-    errors::{SyncServerError, not_found_error, server_error},
+    errors::{SyncServerError, not_found_error},
 };
 
 #[axum::debug_handler]
@@ -24,16 +24,8 @@ pub async fn fetch_latest_document_version(
     let latest_version = state
         .database
         .get_latest_document_version(&vault_id, &document_id, None)
-        .await
+        .await?
-        .map_err(server_error)?
+        .ok_or_else(|| not_found_error(anyhow!("Document with id `{document_id}` not found")))?;
-        .map_or_else(
-            || {
-                Err(not_found_error(anyhow!(
-                    "Document with id `{document_id}` not found",
-                )))
-            },
-            Ok,
-        )?;
 
     Ok(Json(latest_version.into()))
 }

sync-server/src/server/endpoints/get_file_manifest.rs

@@ -23,10 +23,9 @@ pub async fn get_file_manifest(
     let mut transaction = state
         .database
         .create_readonly_transaction(&vault_id)
-        .await
+        .await?;
-        .map_err(server_error)?;
 
-    Database::current_file_manifest(&mut transaction)
+    Database::get_current_file_manifest(&mut transaction)
         .await
         .map(Json)
         .map_err(server_error)

sync-server/src/server/endpoints/ping.rs

@@ -9,7 +9,7 @@ use axum_extra::{
 use log::debug;
 
 use super::VaultPath;
-use crate::server::{auth::auth, responses::PingResponse};
+use crate::server::{auth::authenticate_user, responses::PingResponse};
 use crate::{app_state::AppState, consts::SUPPORTED_API_VERSION, errors::SyncServerError};
 
 #[axum::debug_handler]
@@ -20,8 +20,9 @@ pub async fn ping(
 ) -> Result<Json<PingResponse>, SyncServerError> {
     debug!("Pinging vault `{vault_id}`");
 
-    let is_authenticated = maybe_auth_header
+    let is_authenticated = maybe_auth_header.is_some_and(|auth_header| {
-        .is_some_and(|auth_header| auth(&state, auth_header.token(), &vault_id).is_ok());
+        authenticate_user(&state, auth_header.token(), &vault_id).is_ok()
+    });
 
     Ok(Json(PingResponse {
         server_version: env!("CARGO_PKG_VERSION").to_owned(),