feat(nextcloud): add Nextcloud-compatible API layer

Implement a complete Nextcloud client compatibility layer so that
Nextcloud desktop/mobile sync clients can connect to OxiCloud.

Key additions:
- Login Flow v2 (device auth) with OIDC bridge support
- WebDAV handler compatible with Nextcloud clients (PROPFIND, GET,
  PUT, DELETE, MKCOL, MOVE, COPY, HEAD, PROPPATCH)
- OCS API endpoints (user info, capabilities, notifications stubs,
  sharees, unified search)
- Basic Auth middleware with app password verification, account
  lockout integration, and blake3-keyed auth cache
- App password management: create, list, revoke via both native
  API (JWT-authenticated profile page) and Nextcloud OCS endpoints
- Nextcloud file ID mapping (oc:fileid) with persistent DB storage
- Chunked upload support (Nextcloud v2 chunking protocol)
- Trashbin WebDAV interface
- Avatar (SVG placeholder) and preview (redirect) handlers
- User profile page with app password management UI
- URL user validation on all DAV routes (403 on mismatch)
- Database schema for app_passwords and nextcloud_object_ids tables

All services are behind a `nextcloud.enabled` config flag and
cleanly separated under src/interfaces/nextcloud/.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/infrastructure/repositories/pg/app_password_pg_repository.rs

@@ -103,6 +103,34 @@ impl AppPasswordStoragePort for AppPasswordPgRepository {
         Ok(rows.into_iter().map(|r| r.into()).collect())
     }
 
+    async fn get_active_by_user_prefix(
+        &self,
+        user_id: &str,
+        prefix: &str,
+    ) -> Result<Vec<AppPassword>, DomainError> {
+        let rows = sqlx::query_as::<_, AppPasswordRow>(
+            r#"
+            SELECT id, user_id, label, password_hash, prefix, scopes,
+                   created_at, last_used_at, expires_at, active
+              FROM auth.app_passwords
+             WHERE user_id = $1
+               AND prefix = $2
+               AND active = TRUE
+               AND (expires_at IS NULL OR expires_at > NOW())
+             ORDER BY created_at DESC
+            "#,
+        )
+        .bind(user_id)
+        .bind(prefix)
+        .fetch_all(self.pool())
+        .await
+        .map_err(|e| {
+            DomainError::internal_error("AppPasswordPg", format!("get_active_by_prefix: {e}"))
+        })?;
+
+        Ok(rows.into_iter().map(|r| r.into()).collect())
+    }
+
     async fn touch_last_used(&self, id: &str) -> Result<(), DomainError> {
         sqlx::query("UPDATE auth.app_passwords SET last_used_at = NOW() WHERE id = $1")
             .bind(id)
@@ -112,12 +140,15 @@ impl AppPasswordStoragePort for AppPasswordPgRepository {
         Ok(())
     }
 
-    async fn revoke(&self, id: &str) -> Result<(), DomainError> {
-        let result = sqlx::query("UPDATE auth.app_passwords SET active = FALSE WHERE id = $1")
-            .bind(id)
-            .execute(self.pool())
-            .await
-            .map_err(|e| DomainError::internal_error("AppPasswordPg", format!("revoke: {e}")))?;
+    async fn revoke(&self, id: &str, user_id: &str) -> Result<(), DomainError> {
+        let result = sqlx::query(
+            "UPDATE auth.app_passwords SET active = FALSE WHERE id = $1 AND user_id = $2",
+        )
+        .bind(id)
+        .bind(user_id)
+        .execute(self.pool())
+        .await
+        .map_err(|e| DomainError::internal_error("AppPasswordPg", format!("revoke: {e}")))?;
 
         if result.rows_affected() == 0 {
             return Err(DomainError::not_found("AppPassword", id));
@@ -125,6 +156,19 @@ impl AppPasswordStoragePort for AppPasswordPgRepository {
         Ok(())
     }
 
+    async fn delete_by_user_and_id(&self, id: &str, user_id: &str) -> Result<bool, DomainError> {
+        let result = sqlx::query("DELETE FROM auth.app_passwords WHERE id = $1 AND user_id = $2")
+            .bind(id)
+            .bind(user_id)
+            .execute(self.pool())
+            .await
+            .map_err(|e| {
+                DomainError::internal_error("AppPasswordPg", format!("delete_by_user_and_id: {e}"))
+            })?;
+
+        Ok(result.rows_affected() > 0)
+    }
+
     async fn delete_expired(&self) -> Result<u64, DomainError> {
         let result = sqlx::query(
             r#"

src/infrastructure/repositories/pg/favorites_pg_repository.rs

@@ -1,4 +1,5 @@
 use sqlx::{PgPool, Row};
+use std::collections::HashSet;
 use std::sync::Arc;
 use tracing::error;
 use uuid::Uuid;
@@ -247,4 +248,37 @@ impl FavoritesRepositoryPort for FavoritesPgRepository {
 
         Ok(total_inserted)
     }
+
+    async fn batch_check_favorites(
+        &self,
+        user_id: &str,
+        item_ids: &[(&str, &str)],
+    ) -> Result<HashSet<String>> {
+        if item_ids.is_empty() {
+            return Ok(HashSet::new());
+        }
+
+        let user_uuid = Uuid::parse_str(user_id)?;
+
+        // Collect just the IDs for the IN clause
+        let ids: Vec<String> = item_ids.iter().map(|(id, _)| id.to_string()).collect();
+
+        let rows = sqlx::query(
+            "SELECT item_id FROM auth.user_favorites WHERE user_id = $1::TEXT AND item_id = ANY($2)",
+        )
+        .bind(user_uuid)
+        .bind(&ids)
+        .fetch_all(&*self.db_pool)
+        .await
+        .map_err(|e| {
+            error!("Database error batch-checking favorites: {}", e);
+            DomainError::new(
+                ErrorKind::InternalError,
+                "Favorites",
+                format!("Failed to batch-check favorites: {}", e),
+            )
+        })?;
+
+        Ok(rows.iter().map(|r| r.get::<String, _>("item_id")).collect())
+    }
 }

src/infrastructure/repositories/pg/file_blob_read_repository.rs

@@ -62,6 +62,25 @@ impl FileBlobReadRepository {
         }
     }
 
+    /// Creates a stub instance for testing — never hits PG.
+    #[cfg(test)]
+    pub fn new_stub() -> Self {
+        use crate::infrastructure::services::dedup_service::DedupService;
+        Self {
+            pool: Arc::new(
+                sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
+                    .max_connections(1)
+                    .connect_lazy("postgres://invalid:5432/none")
+                    .unwrap(),
+            ),
+            dedup: Arc::new(DedupService::new_stub()),
+            hash_cache: Cache::builder()
+                .max_capacity(10_000)
+                .time_to_idle(Duration::from_secs(30))
+                .build(),
+        }
+    }
+
     /// Build a `StoragePath` from the materialized folder path + file name.
     fn make_file_path(folder_path: Option<&str>, file_name: &str) -> StoragePath {
         match folder_path {
@@ -508,14 +527,24 @@ impl FileReadPort for FileBlobReadRepository {
             ));
         }
 
+        self.get_folder_id_by_path(&folder_path).await
+    }
+
+    async fn get_folder_id_by_path(&self, folder_path: &str) -> Result<String, DomainError> {
+        let folder_path = folder_path.trim_start_matches('/').trim_end_matches('/');
+
+        if folder_path.is_empty() {
+            return Err(DomainError::not_found("Folder", "empty path"));
+        }
+
         sqlx::query_scalar::<_, String>(
             "SELECT id::text FROM storage.folders WHERE path = $1 AND NOT is_trashed",
         )
-        .bind(&folder_path)
+        .bind(folder_path)
         .fetch_optional(self.pool.as_ref())
         .await
-        .map_err(|e| DomainError::internal_error("FileBlobRead", format!("parent lookup: {e}")))?
-        .ok_or_else(|| DomainError::not_found("Folder", format!("parent for path: {path}")))
+        .map_err(|e| DomainError::internal_error("FileBlobRead", format!("folder lookup: {e}")))?
+        .ok_or_else(|| DomainError::not_found("Folder", format!("path: {folder_path}")))
     }
 
     /// Direct SQL lookup using materialized folder paths.
@@ -1042,13 +1071,9 @@ mod tests {
     /// Only the moka `hash_cache` is exercised — no SQL is executed.
     fn make_repo() -> FileBlobReadRepository {
         let _folder_repo = Arc::new(FolderDbRepository::new_stub());
-        // StubDedupPort satisfies the trait but is never called in cache-only tests
-        let dedup: Arc<DedupService> = Arc::new(StubDedupPort);
-        // PgPool is required by the struct but we won't hit any SQL in these tests.
-        // We create a repo with a stub pool placeholder — only hash_cache is tested.
+        let dedup: Arc<DedupService> = Arc::new(DedupService::new_stub());
         FileBlobReadRepository {
             pool: Arc::new(
-                // Use an intentionally invalid URL; tests never reach PG.
                 sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
                     .max_connections(1)
                     .connect_lazy("postgres://invalid:5432/none")
@@ -1135,7 +1160,7 @@ mod tests {
                     .connect_lazy("postgres://invalid:5432/none")
                     .unwrap(),
             ),
-            dedup: Arc::new(StubDedupPort),
+            dedup: Arc::new(DedupService::new_stub()),
             hash_cache: Cache::builder()
                 .max_capacity(2) // only 2 entries
                 .build(),

src/infrastructure/repositories/pg/file_blob_write_repository.rs

@@ -39,6 +39,22 @@ impl FileBlobWriteRepository {
         }
     }
 
+    /// Creates a stub instance for testing — never hits PG.
+    #[cfg(test)]
+    pub fn new_stub() -> Self {
+        use crate::infrastructure::services::dedup_service::DedupService;
+        Self {
+            pool: Arc::new(
+                sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
+                    .max_connections(1)
+                    .connect_lazy("postgres://invalid:5432/none")
+                    .unwrap(),
+            ),
+            dedup: Arc::new(DedupService::new_stub()),
+            folder_repo: Arc::new(super::folder_db_repository::FolderDbRepository::new_stub()),
+        }
+    }
+
     /// Build a `StoragePath` from the materialized folder path + file name.
     fn make_file_path(folder_path: Option<&str>, file_name: &str) -> StoragePath {
         match folder_path {

src/infrastructure/repositories/pg/mod.rs

@@ -7,6 +7,7 @@ mod contact_persistence_dto;
 mod contact_pg_repository;
 mod device_code_pg_repository;
 mod favorites_pg_repository;
+mod nextcloud_object_id_repository;
 mod recent_items_pg_repository;
 mod session_pg_repository;
 mod settings_pg_repository;
@@ -32,6 +33,7 @@ pub use favorites_pg_repository::FavoritesPgRepository;
 pub use file_blob_read_repository::FileBlobReadRepository;
 pub use file_blob_write_repository::FileBlobWriteRepository;
 pub use folder_db_repository::FolderDbRepository;
+pub use nextcloud_object_id_repository::NextcloudObjectIdRepository;
 pub use recent_items_pg_repository::RecentItemsPgRepository;
 pub use session_pg_repository::SessionPgRepository;
 pub use settings_pg_repository::SettingsPgRepository;

src/infrastructure/repositories/pg/nextcloud_object_id_repository.rs

@@ -0,0 +1,73 @@
+use sqlx::{PgPool, Row};
+use std::sync::Arc;
+
+use crate::common::errors::{DomainError, ErrorKind, Result};
+
+pub struct NextcloudObjectIdRepository {
+    pool: Arc<PgPool>,
+}
+
+impl NextcloudObjectIdRepository {
+    pub fn new(pool: Arc<PgPool>) -> Self {
+        Self { pool }
+    }
+
+    pub async fn get_or_create(&self, object_type: &str, object_id: &str) -> Result<i64> {
+        let row = sqlx::query(
+            r#"
+            INSERT INTO storage.nextcloud_object_ids (object_type, object_id)
+            VALUES ($1, $2::uuid)
+            ON CONFLICT (object_type, object_id)
+            DO UPDATE SET object_id = EXCLUDED.object_id
+            RETURNING id
+            "#,
+        )
+        .bind(object_type)
+        .bind(object_id)
+        .fetch_one(&*self.pool)
+        .await
+        .map_err(|e| {
+            DomainError::new(
+                ErrorKind::DatabaseError,
+                "NextcloudFileId",
+                format!("Failed to get/create Nextcloud ID: {}", e),
+            )
+        })?;
+
+        Ok(row.get::<i64, _>("id"))
+    }
+
+    /// Get the OxiCloud object ID from a Nextcloud numeric ID.
+    pub async fn get_object_id(&self, nc_id: i64, object_type: &str) -> Result<String> {
+        let row = sqlx::query(
+            r#"
+            SELECT object_id
+            FROM storage.nextcloud_object_ids
+            WHERE id = $1 AND object_type = $2
+            "#,
+        )
+        .bind(nc_id)
+        .bind(object_type)
+        .fetch_optional(&*self.pool)
+        .await
+        .map_err(|e| {
+            DomainError::new(
+                ErrorKind::DatabaseError,
+                "NextcloudFileId",
+                format!("Failed to lookup Nextcloud ID: {}", e),
+            )
+        })?;
+
+        match row {
+            Some(row) => {
+                let uuid: sqlx::types::Uuid = row.get("object_id");
+                Ok(uuid.to_string())
+            }
+            None => Err(DomainError::new(
+                ErrorKind::NotFound,
+                "NextcloudFileId",
+                format!("No mapping found for Nextcloud ID: {}", nc_id),
+            )),
+        }
+    }
+}

src/infrastructure/repositories/pg/share_pg_repository.rs

@@ -22,6 +22,19 @@ impl SharePgRepository {
         Self { db_pool }
     }
 
+    /// Creates a stub instance for testing — never hits PG.
+    #[cfg(test)]
+    pub fn new_stub() -> Self {
+        Self {
+            db_pool: Arc::new(
+                sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
+                    .max_connections(1)
+                    .connect_lazy("postgres://invalid:5432/none")
+                    .unwrap(),
+            ),
+        }
+    }
+
     /// Maps a [`sqlx::postgres::PgRow`] to the domain [`Share`] entity.
     fn row_to_entity(row: &sqlx::postgres::PgRow) -> Result<Share, DomainError> {
         let id: String = row

src/infrastructure/repositories/pg/trash_db_repository.rs

@@ -30,6 +30,20 @@ impl TrashDbRepository {
         }
     }
 
+    /// Creates a stub instance for testing — never hits PG.
+    #[cfg(test)]
+    pub fn new_stub() -> Self {
+        Self {
+            pool: Arc::new(
+                sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
+                    .max_connections(1)
+                    .connect_lazy("postgres://invalid:5432/none")
+                    .unwrap(),
+            ),
+            retention_days: 30,
+        }
+    }
+
     /// Convert a trash_items view row into a TrashedItem entity.
     fn row_to_trashed_item(
         &self,

src/infrastructure/repositories/pg/user_pg_repository.rs

@@ -369,6 +369,57 @@ impl UserRepository for UserPgRepository {
         Ok(users)
     }
 
+    async fn search_users(&self, query: &str, limit: i64) -> UserRepositoryResult<Vec<User>> {
+        let pattern = format!("%{}%", query);
+        let rows = sqlx::query(
+            r#"
+            SELECT
+                id, username, email, password_hash, role::text as role_text,
+                storage_quota_bytes, storage_used_bytes,
+                created_at, updated_at, last_login_at, active,
+                oidc_provider, oidc_subject
+            FROM auth.users
+            WHERE username ILIKE $1 OR email ILIKE $1
+            ORDER BY username
+            LIMIT $2
+            "#,
+        )
+        .bind(&pattern)
+        .bind(limit)
+        .fetch_all(&*self.pool)
+        .await
+        .map_err(Self::map_sqlx_error)?;
+
+        let users = rows
+            .into_iter()
+            .map(|row| {
+                let role_str: Option<String> = row.try_get("role_text").unwrap_or(None);
+                let role = match role_str.as_deref() {
+                    Some("admin") => UserRole::Admin,
+                    _ => UserRole::User,
+                };
+
+                User::from_data_full(
+                    row.get("id"),
+                    row.get("username"),
+                    row.get("email"),
+                    row.get("password_hash"),
+                    role,
+                    row.get("storage_quota_bytes"),
+                    row.get("storage_used_bytes"),
+                    row.get("created_at"),
+                    row.get("updated_at"),
+                    row.get("last_login_at"),
+                    row.get("active"),
+                    row.get("oidc_provider"),
+                    row.get("oidc_subject"),
+                )
+            })
+            .collect();
+
+        Ok(users)
+    }
+
     /// Activates or deactivates a user
     async fn set_user_active_status(
         &self,
@@ -664,6 +715,12 @@ impl UserStoragePort for UserPgRepository {
             .map_err(DomainError::from)
     }
 
+    async fn search_users(&self, query: &str, limit: i64) -> Result<Vec<User>, DomainError> {
+        UserRepository::search_users(self, query, limit)
+            .await
+            .map_err(DomainError::from)
+    }
+
     async fn list_users_by_role(&self, role: &str) -> Result<Vec<User>, DomainError> {
         UserRepository::list_users_by_role(self, role)
             .await

src/infrastructure/services/dedup_service.rs

@@ -110,6 +110,23 @@ impl DedupService {
         }
     }
 
+    /// Creates a stub instance for testing — never hits PG or the filesystem.
+    #[cfg(test)]
+    pub fn new_stub() -> Self {
+        let stub_pool = Arc::new(
+            sqlx::pool::PoolOptions::<sqlx::Postgres>::new()
+                .max_connections(1)
+                .connect_lazy("postgres://invalid:5432/none")
+                .unwrap(),
+        );
+        Self {
+            blob_root: std::path::PathBuf::from("/tmp/oxicloud_stub_blobs"),
+            temp_root: std::path::PathBuf::from("/tmp/oxicloud_stub_temp"),
+            pool: stub_pool.clone(),
+            maintenance_pool: stub_pool,
+        }
+    }
+
     /// Initialize the service (create blob directories on the filesystem).
     pub async fn initialize(&self) -> Result<(), DomainError> {
         // Create directories
@@ -903,6 +920,10 @@ impl DedupPort for DedupService {
             .map_err(DomainError::from)
     }
 
+    fn blob_path(&self, hash: &str) -> PathBuf {
+        self.blob_path(hash)
+    }
+
     async fn get_stats(&self) -> DedupStatsDto {
         self.get_stats().await
     }

src/infrastructure/services/mod.rs

@@ -6,6 +6,7 @@ pub mod file_system_i18n_service;
 pub mod image_transcode_service;
 pub mod jwt_service;
 pub mod login_lockout_service;
+pub mod nextcloud_chunked_upload_service;
 pub mod oidc_service;
 pub mod password_hasher;
 pub mod path_resolver_service;

src/infrastructure/services/nextcloud_chunked_upload_service.rs

@@ -0,0 +1,223 @@
+use std::path::PathBuf;
+use tokio::fs;
+use tokio::io::AsyncWriteExt;
+
+use crate::common::errors::{DomainError, Result};
+
+#[derive(Clone)]
+pub struct NextcloudChunkedUploadService {
+    pub base_dir: PathBuf,
+}
+
+impl NextcloudChunkedUploadService {
+    pub fn new(base_dir: PathBuf) -> Self {
+        Self { base_dir }
+    }
+
+    pub fn new_stub() -> Self {
+        Self {
+            base_dir: PathBuf::from("./storage/.uploads/nextcloud"),
+        }
+    }
+
+    /// Validate that a path component contains no traversal characters.
+    fn validate_path_component(name: &str, label: &str) -> Result<()> {
+        if name.is_empty()
+            || name.contains('/')
+            || name.contains('\\')
+            || name.contains("..")
+            || name == "."
+        {
+            return Err(DomainError::validation_error(format!(
+                "ChunkedUpload: invalid {}: contains path traversal characters",
+                label
+            )));
+        }
+        Ok(())
+    }
+
+    /// Build a session directory path and verify it's inside base_dir.
+    fn safe_session_dir(&self, user: &str, upload_id: &str) -> Result<PathBuf> {
+        Self::validate_path_component(user, "username")?;
+        Self::validate_path_component(upload_id, "upload_id")?;
+        Ok(self.base_dir.join(user).join(upload_id))
+    }
+
+    /// Create a new upload session directory.
+    pub async fn create_session(&self, user: &str, upload_id: &str) -> Result<()> {
+        let session_dir = self.safe_session_dir(user, upload_id)?;
+        fs::create_dir_all(&session_dir)
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+        Ok(())
+    }
+
+    /// Store a chunk in the session directory.
+    pub async fn store_chunk(
+        &self,
+        user: &str,
+        upload_id: &str,
+        chunk_name: &str,
+        data: &[u8],
+    ) -> Result<()> {
+        Self::validate_path_component(chunk_name, "chunk_name")?;
+        let chunk_path = self.safe_session_dir(user, upload_id)?.join(chunk_name);
+        let mut file = fs::File::create(&chunk_path)
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+        file.write_all(data)
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+        Ok(())
+    }
+
+    /// Assemble all chunks in numeric order into a temp file.
+    ///
+    /// Returns `(temp_path, total_size)`. The caller is responsible for
+    /// cleaning up the temp file after use.
+    pub async fn assemble(&self, user: &str, upload_id: &str) -> Result<(PathBuf, u64)> {
+        let session_dir = self.safe_session_dir(user, upload_id)?;
+        let mut entries: Vec<String> = Vec::new();
+
+        let mut dir = fs::read_dir(&session_dir)
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+
+        while let Some(entry) = dir
+            .next_entry()
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?
+        {
+            let name = entry.file_name().to_string_lossy().to_string();
+            if name == ".file" {
+                continue; // Skip the assembly marker.
+            }
+            entries.push(name);
+        }
+
+        // Sort chunks numerically (Nextcloud sends them as "00001", "00002", ...).
+        entries.sort();
+
+        // Stream chunks to a temp file instead of buffering in memory.
+        let temp_path = session_dir.join(".assembled");
+        let mut out = fs::File::create(&temp_path)
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+
+        let mut total_size: u64 = 0;
+        for chunk_name in &entries {
+            let mut chunk_file = fs::File::open(session_dir.join(chunk_name))
+                .await
+                .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+            let copied = tokio::io::copy(&mut chunk_file, &mut out)
+                .await
+                .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+            total_size += copied;
+        }
+
+        out.flush()
+            .await
+            .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+
+        Ok((temp_path, total_size))
+    }
+
+    /// Delete the upload session directory.
+    pub async fn cleanup(&self, user: &str, upload_id: &str) -> Result<()> {
+        let session_dir = self.safe_session_dir(user, upload_id)?;
+        if session_dir.exists() {
+            fs::remove_dir_all(&session_dir)
+                .await
+                .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
+        }
+        Ok(())
+    }
+
+    /// Check if a session directory exists.
+    pub async fn session_exists(&self, user: &str, upload_id: &str) -> bool {
+        self.safe_session_dir(user, upload_id)
+            .map(|p| p.exists())
+            .unwrap_or(false)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_service() -> (NextcloudChunkedUploadService, tempfile::TempDir) {
+        let dir = tempfile::tempdir().expect("create temp dir");
+        let svc = NextcloudChunkedUploadService::new(dir.path().to_path_buf());
+        (svc, dir)
+    }
+
+    #[tokio::test]
+    async fn test_create_session() {
+        let (svc, _dir) = test_service();
+        svc.create_session("alice", "upload-001").await.unwrap();
+        assert!(svc.session_exists("alice", "upload-001").await);
+    }
+
+    #[tokio::test]
+    async fn test_session_not_exists_before_create() {
+        let (svc, _dir) = test_service();
+        assert!(!svc.session_exists("alice", "upload-999").await);
+    }
+
+    #[tokio::test]
+    async fn test_store_and_assemble_chunks() {
+        let (svc, _dir) = test_service();
+        svc.create_session("alice", "upload-002").await.unwrap();
+
+        svc.store_chunk("alice", "upload-002", "00001", b"Hello, ")
+            .await
+            .unwrap();
+        svc.store_chunk("alice", "upload-002", "00002", b"World!")
+            .await
+            .unwrap();
+
+        let (temp_path, size) = svc.assemble("alice", "upload-002").await.unwrap();
+        let assembled = fs::read(&temp_path).await.unwrap();
+        assert_eq!(assembled, b"Hello, World!");
+        assert_eq!(size, 13);
+    }
+
+    #[tokio::test]
+    async fn test_assemble_chunks_in_sorted_order() {
+        let (svc, _dir) = test_service();
+        svc.create_session("alice", "upload-003").await.unwrap();
+
+        // Store out of order.
+        svc.store_chunk("alice", "upload-003", "00003", b"C")
+            .await
+            .unwrap();
+        svc.store_chunk("alice", "upload-003", "00001", b"A")
+            .await
+            .unwrap();
+        svc.store_chunk("alice", "upload-003", "00002", b"B")
+            .await
+            .unwrap();
+
+        let (temp_path, size) = svc.assemble("alice", "upload-003").await.unwrap();
+        let assembled = fs::read(&temp_path).await.unwrap();
+        assert_eq!(assembled, b"ABC");
+        assert_eq!(size, 3);
+    }
+
+    #[tokio::test]
+    async fn test_cleanup_removes_session() {
+        let (svc, _dir) = test_service();
+        svc.create_session("alice", "upload-004").await.unwrap();
+        assert!(svc.session_exists("alice", "upload-004").await);
+
+        svc.cleanup("alice", "upload-004").await.unwrap();
+        assert!(!svc.session_exists("alice", "upload-004").await);
+    }
+
+    #[tokio::test]
+    async fn test_cleanup_nonexistent_session_is_ok() {
+        let (svc, _dir) = test_service();
+        // Should not error.
+        svc.cleanup("alice", "nonexistent").await.unwrap();
+    }
+}