diff --git a/sync-server/.env b/sync-server/.env
deleted file mode 100644
index b47c1b5c..00000000
--- a/sync-server/.env
+++ /dev/null
@@ -1 +0,0 @@
-DATABASE_URL=sqlite://db.sqlite3
diff --git a/sync-server/Cargo.toml b/sync-server/Cargo.toml
index 7c2c1ab5..a9aac137 100644
--- a/sync-server/Cargo.toml
+++ b/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"
diff --git a/sync-server/src/app_state.rs b/sync-server/src/app_state.rs
index 5293ee1c..45a02fb3 100644
--- a/sync-server/src/app_state.rs
+++ b/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,
diff --git a/sync-server/src/app_state/cursors.rs b/sync-server/src/app_state/cursors.rs
index 231933f7..0a86db35 100644
--- a/sync-server/src/app_state/cursors.rs
+++ b/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;
diff --git a/sync-server/src/app_state/database.rs b/sync-server/src/app_state/database.rs
index e5c492cb..3154570e 100644
--- a/sync-server/src/app_state/database.rs
+++ b/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
diff --git a/sync-server/src/app_state/database/history.rs b/sync-server/src/app_state/database/history.rs
new file mode 100644
index 00000000..861c68f5
--- /dev/null
+++ b/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)
+    }
+}
diff --git a/sync-server/src/app_state/database/models.rs b/sync-server/src/app_state/database/models.rs
index 5330a8dc..084b1cb6 100644
--- a/sync-server/src/app_state/database/models.rs
+++ b/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(optional, as = "Option<f64>")]
-    pub end_event_id: Option<VaultUpdateId>,
+    #[ts(as = "f64")]
+    pub end_event_id: VaultUpdateId,
 
     pub events: Vec<EventRecord>,
 }
diff --git a/sync-server/src/app_state/database/queries.rs b/sync-server/src/app_state/database/queries.rs
index 162b0fe6..5e5a8a2a 100644
--- a/sync-server/src/app_state/database/queries.rs
+++ b/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)) => {
-                Ok(Some((fingerprint, Self::event_by_id(tx, event_id).await?)))
-            }
+            Some((fingerprint, event_id)) => Ok(Some((
+                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 file_manifest = Self::current_file_manifest(&mut tx).await?;
+        let head_event_id = Self::get_latest_event_id(&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)
-    }
-}
diff --git a/sync-server/src/app_state/database/tests.rs b/sync-server/src/app_state/database/tests.rs
index 25de0e5f..810d484a 100644
--- a/sync-server/src/app_state/database/tests.rs
+++ b/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::run_migrations(&pool).await.unwrap();
+    Database::initialize_schema(&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!(
-        batch.end_event_id,
-        batch.events.last().map(|event| event.event_id)
-    );
+    assert_eq!(batch.end_event_id, batch.events.last().unwrap().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
-    // migrations. Even already-open, unrelated vaults must remain available.
-    let external = Database::open_database(&config, &Database::database_path(&config, &vault))
+    // A second process holds the writer lock while this process initializes
+    // the schema. Even already-open, unrelated vaults must remain available.
+    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!(
diff --git a/sync-server/src/app_state/websocket/broadcasts.rs b/sync-server/src/app_state/websocket/broadcasts.rs
index f97370f3..6617ff2e 100644
--- a/sync-server/src/app_state/websocket/broadcasts.rs
+++ b/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) {
diff --git a/sync-server/src/app_state/websocket/utils.rs b/sync-server/src/app_state/websocket/utils.rs
index adfab14a..2ea57000 100644
--- a/sync-server/src/app_state/websocket/utils.rs
+++ b/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(
diff --git a/sync-server/src/errors.rs b/sync-server/src/errors.rs
index c505b8ae..e49defab 100644
--- a/sync-server/src/errors.rs
+++ b/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 {
diff --git a/sync-server/src/server.rs b/sync-server/src/server.rs
index 0a6145c6..0cfa87ef 100644
--- a/sync-server/src/server.rs
+++ b/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",
diff --git a/sync-server/src/server/auth.rs b/sync-server/src/server/auth.rs
index c601ef54..74c424e9 100644
--- a/sync-server/src/server/auth.rs
+++ b/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
diff --git a/sync-server/src/server/endpoints/events.rs b/sync-server/src/server/endpoints/events.rs
index 3c2ffd30..8ce27115 100644
--- a/sync-server/src/server/endpoints/events.rs
+++ b/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),
diff --git a/sync-server/src/server/endpoints/fetch_document_version_content.rs b/sync-server/src/server/endpoints/fetch_document_version_content.rs
index 9e981c5f..08599bfb 100644
--- a/sync-server/src/server/endpoints/fetch_document_version_content.rs
+++ b/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
-        .map_err(server_error)?
-        .map_or_else(
-            || {
-                Err(not_found_error(anyhow!(
-                    "Document with vault update id `{vault_update_id}` not found",
-                )))
-            },
-            Ok,
-        )?;
+        .await?
+        .ok_or_else(|| {
+            not_found_error(anyhow!(
+                "Document with vault update id `{vault_update_id}` not found",
+            ))
+        })?;
 
     if result.document_id != document_id {
         return Err(not_found_error(anyhow!(
diff --git a/sync-server/src/server/endpoints/fetch_latest_document_metadata.rs b/sync-server/src/server/endpoints/fetch_latest_document_metadata.rs
index 75b1a1e5..1c029847 100644
--- a/sync-server/src/server/endpoints/fetch_latest_document_metadata.rs
+++ b/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
-        .map_err(server_error)?;
+        .await?;
 
     let metadata = Database::get_latest_document_metadata(&mut tx, &document_id)
-        .await
-        .map_err(server_error)?
+        .await?
         .ok_or_else(|| not_found_error(anyhow!("Document does not exist")))?;
 
     Ok(Json(metadata))
diff --git a/sync-server/src/server/endpoints/fetch_latest_document_version.rs b/sync-server/src/server/endpoints/fetch_latest_document_version.rs
index 870ac669..25f80e81 100644
--- a/sync-server/src/server/endpoints/fetch_latest_document_version.rs
+++ b/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
-        .map_err(server_error)?
-        .map_or_else(
-            || {
-                Err(not_found_error(anyhow!(
-                    "Document with id `{document_id}` not found",
-                )))
-            },
-            Ok,
-        )?;
+        .await?
+        .ok_or_else(|| not_found_error(anyhow!("Document with id `{document_id}` not found")))?;
 
     Ok(Json(latest_version.into()))
 }
diff --git a/sync-server/src/server/endpoints/get_file_manifest.rs b/sync-server/src/server/endpoints/get_file_manifest.rs
index 3f706696..d54ed06a 100644
--- a/sync-server/src/server/endpoints/get_file_manifest.rs
+++ b/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
-        .map_err(server_error)?;
+        .await?;
 
-    Database::current_file_manifest(&mut transaction)
+    Database::get_current_file_manifest(&mut transaction)
         .await
         .map(Json)
         .map_err(server_error)
diff --git a/sync-server/src/server/endpoints/ping.rs b/sync-server/src/server/endpoints/ping.rs
index 412ed0d2..a49c25a0 100644
--- a/sync-server/src/server/endpoints/ping.rs
+++ b/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
-        .is_some_and(|auth_header| auth(&state, auth_header.token(), &vault_id).is_ok());
+    let is_authenticated = maybe_auth_header.is_some_and(|auth_header| {
+        authenticate_user(&state, auth_header.token(), &vault_id).is_ok()
+    });
 
     Ok(Json(PingResponse {
         server_version: env!("CARGO_PKG_VERSION").to_owned(),
diff --git a/sync-server/src/server/endpoints/put_file_content.rs b/sync-server/src/server/endpoints/put_file_content.rs
index 45656a92..feed8277 100644
--- a/sync-server/src/server/endpoints/put_file_content.rs
+++ b/sync-server/src/server/endpoints/put_file_content.rs
@@ -1,4 +1,7 @@
-use super::{VaultPath, utils::find_already_processed_event};
+use super::{
+    VaultPath,
+    utils::{commit_and_notify, find_already_processed_event, fingerprint_request},
+};
 use crate::{
     app_state::{
         AppState,
@@ -24,7 +27,6 @@ use axum_extra::TypedHeader;
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use log::debug;
 use reconcile_text::{BuiltinTokenizer, EditedText, NumberOrText};
-use sha2::{Digest, Sha256};
 
 #[axum::debug_handler]
 pub async fn put_file_content(
@@ -36,16 +38,9 @@ pub async fn put_file_content(
 ) -> Result<Json<DocumentUpdateResponse>, SyncServerError> {
     debug!("Pushing document `{document_id}` in vault `{vault_id}`");
 
-    let fingerprint = Sha256::digest(
-        serde_json::to_vec(&("content", document_id, &push))
-            .map_err(|error| server_error(error.into()))?,
-    );
+    let fingerprint = fingerprint_request(&("content", document_id, &push))?;
 
-    let mut tx = state
-        .database
-        .create_write_transaction(&vault_id)
-        .await
-        .map_err(server_error)?;
+    let mut tx = state.database.create_write_transaction(&vault_id).await?;
 
     if let Some(event) =
         find_already_processed_event(&mut tx, push.request_id, &fingerprint).await?
@@ -63,8 +58,7 @@ pub async fn put_file_content(
     let parent = state
         .database
         .get_latest_document_version(&vault_id, &document_id, Some(&mut tx))
-        .await
-        .map_err(server_error)?;
+        .await?;
 
     if let Some(parent) = &parent {
         if Some(parent.vault_update_id) != push.parent_version_id {
@@ -82,16 +76,13 @@ pub async fn put_file_content(
         let content = decode_content(parent, push.content);
         (tx, content)
     })
-    .await
-    .map_err(|error| server_error(error.into()))?;
+    .await?;
 
     let mut tx = tx;
     let content = content.map_err(client_error)?;
 
     let version = StoredDocumentVersion {
-        vault_update_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint)
-            .await
-            .map_err(server_error)?,
+        vault_update_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint).await?,
         document_id,
         content,
         updated_date: chrono::Utc::now(),
@@ -99,17 +90,11 @@ pub async fn put_file_content(
         device_id: device.0,
     };
 
-    Database::insert_document_version(&mut tx, &version)
-        .await
-        .map_err(server_error)?;
+    Database::insert_document_version(&mut tx, &version).await?;
 
     let response = DocumentUpdateResponse::Accepted((&version).into());
 
-    tx.commit()
-        .await
-        .map_err(|error| server_error(error.into()))?;
-
-    state.broadcasts.notify_about_vault_update(vault_id).await;
+    commit_and_notify(tx, &state, vault_id).await?;
 
     Ok(Json(response))
 }
@@ -388,7 +373,9 @@ mod tests {
             .await
             .unwrap();
         assert_eq!(
-            Database::latest_event_id(&mut transaction).await.unwrap(),
+            Database::get_latest_event_id(&mut transaction)
+                .await
+                .unwrap(),
             2
         );
     }
@@ -450,7 +437,7 @@ mod tests {
         }
         let events = reopened
             .database
-            .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap();
         assert_eq!(events.head_event_id, latest.vault_update_id);
@@ -477,7 +464,7 @@ mod tests {
         let events = vault
             .state
             .database
-            .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap();
         assert_eq!(events.head_event_id, 1);
@@ -514,7 +501,7 @@ mod tests {
         let events = vault
             .state
             .database
-            .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap();
         assert_eq!(events.head_event_id, 0);
@@ -538,7 +525,7 @@ mod tests {
         let events = vault
             .state
             .database
-            .events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
+            .get_events_after(&vault.vault, 0, DEFAULT_EVENTS_PAGE_SIZE)
             .await
             .unwrap();
         assert_eq!(events.events.len(), 1);
diff --git a/sync-server/src/server/endpoints/put_file_manifest.rs b/sync-server/src/server/endpoints/put_file_manifest.rs
index 9ba9156a..b08cf4eb 100644
--- a/sync-server/src/server/endpoints/put_file_manifest.rs
+++ b/sync-server/src/server/endpoints/put_file_manifest.rs
@@ -4,9 +4,11 @@ use axum::{
     extract::{Path, State},
 };
 use log::debug;
-use sha2::{Digest, Sha256};
 
-use super::{VaultPath, utils::find_already_processed_event};
+use super::{
+    VaultPath,
+    utils::{commit_and_notify, find_already_processed_event, fingerprint_request},
+};
 use crate::{
     app_state::{
         AppState,
@@ -15,7 +17,7 @@ use crate::{
             models::{FileManifest, VaultEvent},
         },
     },
-    errors::{SyncServerError, client_error, server_error},
+    errors::{SyncServerError, client_error},
     server::{requests::PushFileManifest, responses::FileManifestUpdateResponse},
 };
 
@@ -28,16 +30,9 @@ pub async fn put_file_manifest(
     debug!("Pushing file manifest for vault `{vault_id}`");
 
     // BTreeMap canonicalizes file manifest entries for request fingerprinting.
-    let fingerprint = Sha256::digest(
-        serde_json::to_vec(&("file_manifest", &push))
-            .map_err(|error| server_error(error.into()))?,
-    );
+    let fingerprint = fingerprint_request(&("file_manifest", &push))?;
 
-    let mut tx = state
-        .database
-        .create_write_transaction(&vault_id)
-        .await
-        .map_err(server_error)?;
+    let mut tx = state.database.create_write_transaction(&vault_id).await?;
 
     if let Some(event) =
         find_already_processed_event(&mut tx, push.request_id, &fingerprint).await?
@@ -54,9 +49,7 @@ pub async fn put_file_manifest(
         };
     }
 
-    let latest = Database::current_file_manifest(&mut tx)
-        .await
-        .map_err(server_error)?;
+    let latest = Database::get_current_file_manifest(&mut tx).await?;
 
     if latest.file_manifest_id != push.parent_file_manifest_id {
         return Ok(Json(FileManifestUpdateResponse::StaleBase(latest)));
@@ -64,8 +57,7 @@ pub async fn put_file_manifest(
 
     if let Some(id) =
         Database::get_missing_document(&mut tx, &push.entries.keys().copied().collect::<Vec<_>>())
-            .await
-            .map_err(server_error)?
+            .await?
     {
         return Err(client_error(anyhow!(
             "File manifest references missing content: {id}"
@@ -73,25 +65,17 @@ pub async fn put_file_manifest(
     }
 
     let file_manifest = FileManifest {
-        file_manifest_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint)
-            .await
-            .map_err(server_error)?,
+        file_manifest_id: Database::allocate_event(&mut tx, push.request_id, &fingerprint).await?,
         entries: push.entries,
     };
 
-    Database::insert_file_manifest(&mut tx, &file_manifest)
-        .await
-        .map_err(server_error)?;
+    Database::insert_file_manifest(&mut tx, &file_manifest).await?;
 
     let response = FileManifestUpdateResponse::Accepted {
         file_manifest_id: file_manifest.file_manifest_id,
     };
 
-    tx.commit()
-        .await
-        .map_err(|error| server_error(error.into()))?;
-
-    state.broadcasts.notify_about_vault_update(vault_id).await;
+    commit_and_notify(tx, &state, vault_id).await?;
 
     Ok(Json(response))
 }
diff --git a/sync-server/src/server/endpoints/utils.rs b/sync-server/src/server/endpoints/utils.rs
index c0df4216..7f895133 100644
--- a/sync-server/src/server/endpoints/utils.rs
+++ b/sync-server/src/server/endpoints/utils.rs
@@ -1,18 +1,23 @@
 use anyhow::anyhow;
 
 use crate::{
-    app_state::database::{Database, Transaction, models::VaultEvent},
+    app_state::{
+        AppState,
+        database::{
+            Database, Transaction,
+            models::{VaultEvent, VaultId},
+        },
+    },
     errors::{SyncServerError, client_error, server_error},
 };
+use sha2::{Digest, Sha256};
 
 pub(super) async fn find_already_processed_event(
     tx: &mut Transaction<'_>,
     request_id: uuid::Uuid,
     fingerprint: &[u8],
 ) -> Result<Option<VaultEvent>, SyncServerError> {
-    let Some((stored_fingerprint, event)) = Database::get_request_event(tx, request_id)
-        .await
-        .map_err(server_error)?
+    let Some((stored_fingerprint, event)) = Database::get_request_event(tx, request_id).await?
     else {
         return Ok(None);
     };
@@ -25,3 +30,24 @@ pub(super) async fn find_already_processed_event(
 
     Ok(Some(event))
 }
+
+/// Hash the canonical wire payload, including the operation and document ID.
+pub(super) fn fingerprint_request(
+    payload: &impl serde::Serialize,
+) -> Result<[u8; 32], SyncServerError> {
+    let encoded = serde_json::to_vec(payload).map_err(|error| server_error(error.into()))?;
+    Ok(Sha256::digest(encoded).into())
+}
+
+/// Never advertise a write before its event and data are durably committed.
+pub(super) async fn commit_and_notify(
+    tx: Transaction<'_>,
+    state: &AppState,
+    vault: VaultId,
+) -> Result<(), SyncServerError> {
+    tx.commit().await?;
+
+    state.broadcasts.notify_about_vault_update(vault).await;
+
+    Ok(())
+}
diff --git a/sync-server/src/server/endpoints/vault_snapshot.rs b/sync-server/src/server/endpoints/vault_snapshot.rs
index 93e5e4d1..b4071d6b 100644
--- a/sync-server/src/server/endpoints/vault_snapshot.rs
+++ b/sync-server/src/server/endpoints/vault_snapshot.rs
@@ -19,7 +19,7 @@ pub async fn vault_snapshot(
 
     state
         .database
-        .vault_snapshot(&vault_id)
+        .get_vault_snapshot(&vault_id)
         .await
         .map(Json)
         .map_err(server_error)
diff --git a/sync-server/src/server/endpoints/websocket.rs b/sync-server/src/server/endpoints/websocket.rs
index 0ef24919..326f2e1b 100644
--- a/sync-server/src/server/endpoints/websocket.rs
+++ b/sync-server/src/server/endpoints/websocket.rs
@@ -1,7 +1,7 @@
 use crate::{
     app_state::{
         AppState,
-        database::models::VaultId,
+        database::models::{DeviceId, VaultId},
         websocket::{
             broadcasts::Notification,
             models::{CursorPositionFromServer, WebSocketClientMessage, WebSocketServerMessage},
@@ -9,7 +9,7 @@ use crate::{
         },
     },
     consts::WEBSOCKET_HANDSHAKE_TIMEOUT,
-    errors::{SyncServerError, client_error, server_error},
+    errors::{SyncServerError, client_error},
 };
 use axum::{
     extract::{
@@ -19,7 +19,10 @@ use axum::{
     http::StatusCode,
     response::{IntoResponse, Response},
 };
-use futures::StreamExt;
+use futures::{
+    StreamExt,
+    stream::{SplitSink, SplitStream},
+};
 use log::{debug, info};
 use std::time::Duration;
 use tokio::sync::broadcast::error::RecvError;
@@ -44,13 +47,13 @@ pub async fn websocket_handler(
 
     Ok(ws.on_upgrade(move |socket| async move {
         let _permit = permit;
-        if let Err(error) = websocket(state, socket, vault_id).await {
+        if let Err(error) = serve_connection(state, socket, vault_id).await {
             debug!("WebSocket disconnected: {error}");
         }
     }))
 }
 
-async fn websocket(
+async fn serve_connection(
     state: AppState,
     socket: WebSocket,
     vault: VaultId,
@@ -68,53 +71,16 @@ async fn websocket(
 
     let device = authenticated.handshake.device_id;
     let session = state.cursors.register_connection(&vault, &device).await;
-    let mut last_checkpoint = None;
-    let mut notifications = state.broadcasts.get_receiver(vault.clone()).await;
-    let mut timer = tokio::time::interval(Duration::from_secs(2));
-    timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-
     info!("WebSocket connected to vault {vault}");
 
-    let result = async {
-        loop {
-            let checkpoint = state.database.history_checkpoint(&vault).await.map_err(server_error)?;
-            if last_checkpoint.as_ref() != Some(&checkpoint) {
-                send_update_over_websocket(&WebSocketServerMessage::VaultChanged, &mut sender).await?;
-                last_checkpoint = Some(checkpoint);
-            }
-
-            tokio::select! {
-                _ = timer.tick() => {},
-                notification = notifications.recv() => match notification {
-                    Ok(update) => if let Notification::Cursors(mut positions) = update {
-                        positions.clients.retain(|c| c.device_id != device);
-                        send_update_over_websocket(&WebSocketServerMessage::CursorPositions(positions), &mut sender).await?;
-                    },
-
-                    Err(RecvError::Lagged(_)) => {}, // Check the durable head on the next iteration.
-                    Err(RecvError::Closed) => break,
-                },
-
-                incoming = receiver.next() => match incoming {
-                    Some(Ok(Message::Text(text))) => {
-                        let message: WebSocketClientMessage = serde_json::from_str(&text).map_err(|error| client_error(error.into()))?;
-                        match message {
-                            WebSocketClientMessage::Handshake(_) => return Err(client_error(anyhow::anyhow!("Unexpected handshake"))),
-                            WebSocketClientMessage::CursorPositions(positions) => {
-                                state.cursors.update_cursors(vault.clone(), authenticated.user.name.clone(), &device, session, positions.documents_with_cursors).await;
-                                let clients = state.cursors.get_cursors(&vault).await.into_iter().filter(|c| c.device_id != device).collect();
-                                send_update_over_websocket(&WebSocketServerMessage::CursorPositions(CursorPositionFromServer { clients }), &mut sender).await?;
-                            }
-                        }
-                    },
-                    Some(Ok(Message::Ping(_) | Message::Pong(_))) => {},
-                    _ => break,
-                }
-            }
-        }
-
-        Ok(())
-    }.await;
+    let connection = AuthenticatedConnection {
+        state: &state,
+        vault: &vault,
+        device: &device,
+        user_name: &authenticated.user.name,
+        session,
+    };
+    let result = connection.forward_updates(&mut sender, &mut receiver).await;
 
     state
         .cursors
@@ -123,3 +89,101 @@ async fn websocket(
 
     result
 }
+
+/// A session owns cursor presence until it disconnects or a newer session replaces it.
+struct AuthenticatedConnection<'a> {
+    state: &'a AppState,
+    vault: &'a VaultId,
+    device: &'a DeviceId,
+    user_name: &'a str,
+    session: uuid::Uuid,
+}
+
+impl AuthenticatedConnection<'_> {
+    async fn forward_updates(
+        &self,
+        sender: &mut SplitSink<WebSocket, Message>,
+        receiver: &mut SplitStream<WebSocket>,
+    ) -> Result<(), SyncServerError> {
+        let mut last_checkpoint = None;
+        let mut notifications = self.state.broadcasts.get_receiver(self.vault.clone()).await;
+        let mut timer = tokio::time::interval(Duration::from_secs(2));
+        timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+        loop {
+            let checkpoint = self
+                .state
+                .database
+                .get_history_checkpoint(self.vault)
+                .await?;
+            if last_checkpoint.as_ref() != Some(&checkpoint) {
+                send_update_over_websocket(&WebSocketServerMessage::VaultChanged, sender).await?;
+                last_checkpoint = Some(checkpoint);
+            }
+
+            tokio::select! {
+                _ = timer.tick() => {},
+                notification = notifications.recv() => {
+                    match notification {
+                        Ok(Notification::Cursors(positions)) => {
+                            self.send_other_clients_cursors(positions, sender).await?;
+                        }
+                        // Recheck the durable head after either a notification or lag.
+                        Ok(Notification::VaultUpdate) | Err(RecvError::Lagged(_)) => {},
+                        Err(RecvError::Closed) => break,
+                    }
+                }
+                incoming = receiver.next() => {
+                    match incoming {
+                        Some(Ok(Message::Text(text))) => {
+                            self.handle_client_message(&text, sender).await?;
+                        }
+                        Some(Ok(Message::Ping(_) | Message::Pong(_))) => {},
+                        _ => break,
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn send_other_clients_cursors(
+        &self,
+        mut positions: CursorPositionFromServer,
+        sender: &mut SplitSink<WebSocket, Message>,
+    ) -> Result<(), SyncServerError> {
+        positions
+            .clients
+            .retain(|client| client.device_id != *self.device);
+        send_update_over_websocket(&WebSocketServerMessage::CursorPositions(positions), sender)
+            .await
+    }
+
+    async fn handle_client_message(
+        &self,
+        text: &str,
+        sender: &mut SplitSink<WebSocket, Message>,
+    ) -> Result<(), SyncServerError> {
+        let message: WebSocketClientMessage =
+            serde_json::from_str(text).map_err(|error| client_error(error.into()))?;
+        let WebSocketClientMessage::CursorPositions(positions) = message else {
+            return Err(client_error(anyhow::anyhow!("Unexpected handshake")));
+        };
+
+        self.state
+            .cursors
+            .update_cursors(
+                self.vault.clone(),
+                self.user_name.to_owned(),
+                self.device,
+                self.session,
+                positions.documents_with_cursors,
+            )
+            .await;
+
+        let clients = self.state.cursors.get_cursors(self.vault).await;
+        self.send_other_clients_cursors(CursorPositionFromServer { clients }, sender)
+            .await
+    }
+}
diff --git a/sync-server/src/server/history.rs b/sync-server/src/server/history.rs
index 943a2d83..2f851c8f 100644
--- a/sync-server/src/server/history.rs
+++ b/sync-server/src/server/history.rs
@@ -50,7 +50,7 @@ pub async fn history_middleware(
     if response.status().is_success() {
         let checkpoint = state
             .database
-            .history_checkpoint(&vault_id)
+            .get_history_checkpoint(&vault_id)
             .await
             .map_err(server_error)?;
         response.headers_mut().insert(
diff --git a/sync-server/src/server/responses.rs b/sync-server/src/server/responses.rs
index e8d85687..ad0742b3 100644
--- a/sync-server/src/server/responses.rs
+++ b/sync-server/src/server/responses.rs
@@ -20,7 +20,7 @@ pub struct PingResponse {
     /// List of file extensions that are allowed to be merged.
     pub mergeable_file_extensions: Vec<String>,
 
-    /// API version ensuring backwards & forwards compatibility between the client
+    /// API version that must match exactly between the client
     /// and server.
     pub supported_api_version: u32,
 }
diff --git a/sync-server/src/server/tests.rs b/sync-server/src/server/tests.rs
index 01d5f842..16d44623 100644
--- a/sync-server/src/server/tests.rs
+++ b/sync-server/src/server/tests.rs
@@ -17,7 +17,7 @@ async fn app(directory: &tempfile::TempDir) -> Router {
     })).unwrap();
     config.database.databases_directory_path = directory.path().join("databases");
     let state = AppState::try_new(config).await.unwrap();
-    get_authed_routes(state.clone()).with_state(state)
+    create_authenticated_routes(state.clone()).with_state(state)
 }
 
 async fn request(app: Router, path: &str, token: &str, body: Option<Value>) -> (StatusCode, Value) {
diff --git a/sync-server/src/utils/portable_path.rs b/sync-server/src/utils/portable_path.rs
index 7c170848..63d15c09 100644
--- a/sync-server/src/utils/portable_path.rs
+++ b/sync-server/src/utils/portable_path.rs
@@ -7,7 +7,7 @@ pub const INTERNAL_DIRECTORY: &str = ".vault-link-sync";
 
 // Use the same locale-independent Unicode upper-case mapping in the client.
 // NFC -> uppercase -> NFC also catches ß/SS and final/ordinary sigma aliases.
-pub fn path_key(path: &str) -> String {
+pub fn get_path_key(path: &str) -> String {
     path.nfc()
         .collect::<String>()
         .to_uppercase()
@@ -15,8 +15,19 @@ pub fn path_key(path: &str) -> String {
         .collect()
 }
 
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum PathKind {
+    File,
+    Directory,
+}
+
+struct PathNode {
+    spelling: String,
+    kind: PathKind,
+}
+
 pub fn validate_file_manifest(entries: &FileManifestEntries) -> Result<()> {
-    let mut nodes: BTreeMap<String, (String, bool)> = BTreeMap::new();
+    let mut nodes: BTreeMap<String, PathNode> = BTreeMap::new();
 
     for path in entries.values() {
         ensure!(!path.is_empty(), "Path must not be empty");
@@ -29,59 +40,35 @@ pub fn validate_file_manifest(entries: &FileManifestEntries) -> Result<()> {
         let parts: Vec<_> = path.split('/').collect();
 
         ensure!(
-            path_key(parts[0]) != path_key(INTERNAL_DIRECTORY),
+            get_path_key(parts[0]) != get_path_key(INTERNAL_DIRECTORY),
             "Reserved sync directory"
         );
 
         for (index, part) in parts.iter().enumerate() {
-            ensure!(
-                !part.is_empty() && *part != "." && *part != "..",
-                "Invalid path component"
-            );
-
-            ensure!(!part.ends_with(['.', ' ']), "Trailing dot or space in path");
-
-            ensure!(
-                !part
-                    .chars()
-                    .any(|c| c.is_control() || c == '\\' || "<>:\"|?*".contains(c)),
-                "Non-portable filename"
-            );
-
-            let stem = part
-                .split('.')
-                .next()
-                .unwrap_or_default()
-                .trim_end_matches(' ')
-                .to_uppercase();
-
-            ensure!(
-                !matches!(
-                    stem.as_str(),
-                    "CON" | "PRN" | "AUX" | "NUL" | "CONIN$" | "CONOUT$"
-                ),
-                "Reserved filename"
-            );
-
-            if stem.starts_with("COM") || stem.starts_with("LPT") {
-                ensure!(
-                    !matches!(
-                        &stem[3..],
-                        "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9" | "¹" | "²" | "³"
-                    ),
-                    "Reserved filename"
-                );
-            }
+            validate_component(part)?;
 
             let prefix = parts[..=index].join("/");
-            let is_file = index == parts.len() - 1;
+            let kind = if index == parts.len() - 1 {
+                PathKind::File
+            } else {
+                PathKind::Directory
+            };
 
-            if let Some((spelling, previous_is_file)) = nodes.get(&path_key(&prefix)) {
-                if *spelling != prefix || *previous_is_file || is_file {
+            if let Some(previous) = nodes.get(&get_path_key(&prefix)) {
+                if previous.spelling != prefix
+                    || previous.kind == PathKind::File
+                    || kind == PathKind::File
+                {
                     bail!("Conflicting path: {path}");
                 }
             } else {
-                nodes.insert(path_key(&prefix), (prefix, is_file));
+                nodes.insert(
+                    get_path_key(&prefix),
+                    PathNode {
+                        spelling: prefix,
+                        kind,
+                    },
+                );
             }
         }
     }
@@ -89,9 +76,53 @@ pub fn validate_file_manifest(entries: &FileManifestEntries) -> Result<()> {
     Ok(())
 }
 
+/// Validate one filename independently of the manifest's path relationships.
+fn validate_component(part: &str) -> Result<()> {
+    ensure!(
+        !part.is_empty() && part != "." && part != "..",
+        "Invalid path component"
+    );
+
+    ensure!(!part.ends_with(['.', ' ']), "Trailing dot or space in path");
+
+    ensure!(
+        !part
+            .chars()
+            .any(|c| c.is_control() || c == '\\' || "<>:\"|?*".contains(c)),
+        "Non-portable filename"
+    );
+
+    let stem = part
+        .split('.')
+        .next()
+        .unwrap_or_default()
+        .trim_end_matches(' ')
+        .to_uppercase();
+
+    ensure!(
+        !matches!(
+            stem.as_str(),
+            "CON" | "PRN" | "AUX" | "NUL" | "CONIN$" | "CONOUT$"
+        ),
+        "Reserved filename"
+    );
+
+    if stem.starts_with("COM") || stem.starts_with("LPT") {
+        ensure!(
+            !matches!(
+                &stem[3..],
+                "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9" | "¹" | "²" | "³"
+            ),
+            "Reserved filename"
+        );
+    }
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{FileManifestEntries, path_key, validate_file_manifest};
+    use super::{FileManifestEntries, get_path_key, validate_file_manifest};
     use uuid::Uuid;
 
     fn manifest(paths: &[&str]) -> FileManifestEntries {
@@ -110,7 +141,7 @@ mod tests {
             ("Straße.md", "STRASSE.MD"),
             ("σς.md", "ΣΣ.MD"),
         ] {
-            assert_eq!(path_key(path), expected, "path: {path:?}");
+            assert_eq!(get_path_key(path), expected, "path: {path:?}");
         }
     }