feat(mounts): P3 — WebDAV/NextCloud path resolution + browse/download

Mounts are now browsable and downloadable over both WebDAV surfaces
(/webdav/ and NextCloud /remote.php/dav) and NextCloud clients. The work
funnels through the path-based service methods all three surfaces already use,
so both WebDAV handlers gain mount support with no handler changes.

- get_folder_by_path / get_file_by_path resolve a path that descends past a
  mount root to a synthetic ext: DTO (the mount root itself stays a real row)
- list_folders_paginated_with_perms / list_files_batch_with_perms branch to the
  provider, so PROPFIND Depth:1 enumerates mount directory contents in both
  WebDAV handlers
- get_file_stream / get_file_range_stream branch ext: file ids to the provider,
  so WebDAV GET streams mount content (Pin<Box<dyn Stream>> re-boxed)
- router injected into FileRetrievalService; new MountRouter::find_path delegates
  to the registry's (drive_id, mount_path) index
- WebDAV mkdir/delete/move/rename by path work via path→ext:id→the P2 service
  methods (no extra wiring)

Fixes a leading-slash normalization bug in the registry path index: materialized
folder paths arrive both as `Personal/Media` and `/Personal/Media`; keys and
lookups now normalize the leading slash (would have broken real WebDAV paths).

Known follow-up: WebDAV PUT (upload/update by path) still ingests to the CAS;
streaming a WebDAV PUT straight to the provider needs a pre-ingest branch in the
two WebDAV PUT handlers (mirrors the REST upload branch).

Integration test: get_folder_by_path/get_file_by_path resolution, PROPFIND
Depth:1 folder+file listing, and content streaming on a real mount + Postgres.

src/application/services/external_mount_router.rs

@@ -78,6 +78,17 @@ impl MountRouter {
             ResolvedId::MountRoot { .. } | ResolvedId::MountChild { .. }
         )
     }
+
+    /// Path-based lookup for the protocol surfaces (WebDAV / NextCloud): does
+    /// `internal_path` descend into a mount within `drive_id`? Returns the mount
+    /// config plus the remainder relpath (empty when the path IS the mount root).
+    pub fn find_path(
+        &self,
+        drive_id: uuid::Uuid,
+        internal_path: &str,
+    ) -> Option<(Arc<MountConfig>, String)> {
+        self.registry.find_mount_for_path(drive_id, internal_path)
+    }
 }
 
 #[cfg(test)]

src/application/services/file_retrieval_service.rs

@@ -37,6 +37,9 @@ pub struct FileRetrievalService {
     content_cache: Option<Arc<FileContentCache>>,
     transcode: Option<Arc<ImageTranscodeService>>,
     authz: Option<Arc<PgAclEngine>>,
+    /// External-mount classifier for path-based resolution (WebDAV/NextCloud).
+    /// `None` in the simple/test constructor → no mount support.
+    mount_router: Option<Arc<crate::application::services::external_mount_router::MountRouter>>,
 }
 
 impl FileRetrievalService {
@@ -49,6 +52,7 @@ impl FileRetrievalService {
             content_cache: None,
             transcode: None,
             authz: None,
+            mount_router: None,
         }
     }
 
@@ -65,9 +69,20 @@ impl FileRetrievalService {
             content_cache: Some(content_cache),
             transcode: Some(transcode),
             authz: Some(authz),
+            mount_router: None,
         }
     }
 
+    /// Injects the external-mount classifier so path-based lookups
+    /// (`get_file_by_path`) can resolve mount paths to the provider.
+    pub fn with_mount_router(
+        mut self,
+        router: Arc<crate::application::services::external_mount_router::MountRouter>,
+    ) -> Self {
+        self.mount_router = Some(router);
+        self
+    }
+
     /// Test-only constructor: authorization engine without the cache/transcode
     /// tiers. The external-mount read methods only consult `authz` + the
     /// provider, so this is sufficient to exercise their authorization.
@@ -81,6 +96,7 @@ impl FileRetrievalService {
             content_cache: None,
             transcode: None,
             authz: Some(authz),
+            mount_router: None,
         }
     }
 
@@ -186,6 +202,22 @@ impl FileRetrievalService {
         cfg.provider.open_read_stream(node_id, range).await
     }
 
+    /// If `id` is an `ext:` mount FILE id, return the mount config + node id.
+    /// `None` for native ids, mount roots, or when no router is wired.
+    fn mount_file_node(
+        &self,
+        id: &str,
+    ) -> Option<(
+        Arc<crate::application::services::mount_registry::MountConfig>,
+        NodeId,
+    )> {
+        use crate::application::services::external_mount_router::ResolvedId;
+        match self.mount_router.as_ref()?.classify(id) {
+            ResolvedId::MountChild { cfg, node_id } => Some((cfg, node_id)),
+            _ => None,
+        }
+    }
+
     /// Try to transcode image content to WebP and return transcoded variant.
     async fn try_transcode(
         &self,
@@ -349,6 +381,26 @@ impl FileRetrievalUseCase for FileRetrievalService {
         // `drive_id` scope axis prevents cross-drive resolution — without
         // it, `find_file_by_path` would return a non-deterministic row
         // when the same path exists in multiple drives.
+        // External mount: a path descending past a mount root resolves on the
+        // provider (stat). The mount root itself has no file at its path.
+        if let Some(router) = &self.mount_router
+            && let Some((cfg, remainder)) = router.find_path(drive_id, path)
+            && !remainder.is_empty()
+        {
+            let node = cfg.provider.resolve_path(&remainder);
+            let stat = cfg.provider.stat(&node).await?;
+            if stat.is_dir {
+                return Err(DomainError::not_found("File", path));
+            }
+            let parent = crate::application::services::mount_dto::mount_parent_id(
+                &cfg,
+                stat.node_id.as_str(),
+            );
+            return Ok(crate::application::services::mount_dto::mount_file_dto(
+                &cfg, &parent, &stat,
+            ));
+        }
+
         if let Some(file) = self.file_read.find_file_by_path(path, drive_id).await? {
             return Ok(FileDto::from(file));
         }
@@ -388,6 +440,11 @@ impl FileRetrievalUseCase for FileRetrievalService {
         &self,
         id: &str,
     ) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
+        if let Some((cfg, node)) = self.mount_file_node(id) {
+            let s = cfg.provider.open_read_stream(&node, None).await?;
+            // `Pin<Box<dyn Stream>>` is itself a `Stream`, so re-box it.
+            return Ok(Box::new(s));
+        }
         self.file_read.get_file_stream(id).await
     }
 
@@ -446,6 +503,13 @@ impl FileRetrievalUseCase for FileRetrievalService {
         start: u64,
         end: Option<u64>,
     ) -> Result<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>, DomainError> {
+        if let Some((cfg, node)) = self.mount_file_node(id) {
+            // The native range convention is exclusive-end; the provider wants
+            // an inclusive end.
+            let range = Some((start, end.map(|e| e.saturating_sub(1))));
+            let s = cfg.provider.open_read_stream(&node, range).await?;
+            return Ok(Box::new(s));
+        }
         self.file_read.get_file_range_stream(id, start, end).await
     }
 
@@ -490,6 +554,44 @@ impl FileRetrievalUseCase for FileRetrievalService {
         offset: i64,
         limit: i64,
     ) -> Result<Vec<FileDto>, DomainError> {
+        // External mount: list files from the provider (WebDAV/NextCloud
+        // PROPFIND Depth:1 file loop). Authz collapses on the mount root.
+        if let Some(fid) = folder_id
+            && let Some(router) = &self.mount_router
+        {
+            use crate::application::services::external_mount_router::ResolvedId;
+            let resolved = match router.classify(fid) {
+                ResolvedId::Regular => None,
+                ResolvedId::MountRoot { cfg } => Some((
+                    cfg,
+                    crate::domain::services::external_mount_id::NodeId::default(),
+                )),
+                ResolvedId::MountChild { cfg, node_id } => Some((cfg, node_id)),
+            };
+            if let Some((cfg, node)) = resolved {
+                if let Some(authz) = &self.authz {
+                    authz
+                        .require(
+                            Subject::User(owner_id),
+                            Permission::Read,
+                            Resource::Folder(cfg.mount_id),
+                        )
+                        .await?;
+                }
+                let entries = cfg.provider.list_dir(&node).await?;
+                let files: Vec<FileDto> = entries
+                    .iter()
+                    .filter(|e| !e.is_dir)
+                    .skip(offset.max(0) as usize)
+                    .take(limit.max(0) as usize)
+                    .map(|e| {
+                        crate::application::services::mount_dto::mount_entry_file_dto(&cfg, fid, e)
+                    })
+                    .collect();
+                return Ok(files);
+            }
+        }
+
         if folder_id.is_some() {
             // folder id is defined, check permissions
             self.require_target_folder_perm(folder_id, Permission::Read, owner_id)

src/application/services/folder_service.rs

@@ -8,7 +8,7 @@ use crate::application::ports::external_mount_ports::MountEntry;
 use crate::application::ports::folder_ports::FolderUseCase;
 use crate::application::services::external_mount_router::{MountRouter, ResolvedId};
 use crate::application::services::mount_dto::{
-    audit_mount_write, mount_folder_dto, mount_parent_id,
+    audit_mount_write, mount_entry_folder_dto, mount_folder_dto, mount_parent_id,
 };
 use crate::application::services::mount_registry::MountConfig;
 use crate::common::errors::{DomainError, ErrorKind};
@@ -67,6 +67,16 @@ impl FolderService {
             .await
     }
 
+    /// If `id` addresses a mount directory (root or `ext:` child), return the
+    /// mount config and the node id of that directory. `None` for native ids.
+    fn mount_node_for(&self, id: &str) -> Option<(Arc<MountConfig>, NodeId)> {
+        match self.mount_router.classify(id) {
+            ResolvedId::Regular => None,
+            ResolvedId::MountRoot { cfg } => Some((cfg, NodeId::default())),
+            ResolvedId::MountChild { cfg, node_id } => Some((cfg, node_id)),
+        }
+    }
+
     /// Resolve a move destination within the SAME mount as `cfg`, returning the
     /// destination parent's node id. Errors (`UnsupportedOperation`) if the
     /// destination is absent, native, or in a different mount.
@@ -353,6 +363,21 @@ impl FolderUseCase for FolderService {
         path: &str,
         drive_id: Uuid,
     ) -> Result<FolderDto, DomainError> {
+        // External mount: a path that descends past a mount root (non-empty
+        // remainder) resolves on the provider. The mount root itself is a real
+        // folder row, so the empty-remainder case falls through to the DB.
+        if let Some((cfg, remainder)) = self.mount_router.find_path(drive_id, path)
+            && !remainder.is_empty()
+        {
+            let node = cfg.provider.resolve_path(&remainder);
+            let stat = cfg.provider.stat(&node).await?;
+            if !stat.is_dir {
+                return Err(DomainError::not_found("Folder", path));
+            }
+            let parent = mount_parent_id(&cfg, stat.node_id.as_str());
+            return Ok(mount_folder_dto(&cfg, &parent, &stat));
+        }
+
         let storage_path = StoragePath::from_string(path);
 
         let folder = self
@@ -472,6 +497,32 @@ impl FolderUseCase for FolderService {
     {
         let pagination = pagination.validate_and_adjust();
 
+        // External mount: list subdirectories from the provider (used by the
+        // WebDAV/NextCloud PROPFIND Depth:1 folder loop).
+        if let Some(pid) = parent_id
+            && let Some((cfg, node)) = self.mount_node_for(pid)
+        {
+            self.require_mount_perm(&cfg, Permission::Read, owner_id)
+                .await?;
+            let entries = cfg.provider.list_dir(&node).await?;
+            let mut dirs: Vec<FolderDto> = entries
+                .iter()
+                .filter(|e| e.is_dir)
+                .map(|e| mount_entry_folder_dto(&cfg, pid, e))
+                .collect();
+            let total = dirs.len();
+            let (offset, limit) = (pagination.offset(), pagination.limit());
+            let page: Vec<FolderDto> = dirs.drain(..).skip(offset).take(limit).collect();
+            return Ok(
+                crate::application::dtos::pagination::PaginatedResponseDto::new(
+                    page,
+                    pagination.page,
+                    pagination.page_size,
+                    total,
+                ),
+            );
+        }
+
         if let Some(parent_id_unwrapped) = parent_id {
             self.authz
                 .require(
@@ -1530,6 +1581,107 @@ mod mount_authz_integration {
         assert!(!host.path().join("evil.txt").exists());
     }
 
+    /// P3: the WebDAV/NextCloud-facing path + listing methods resolve mount
+    /// paths and enumerate provider children (PROPFIND Depth:1), and content
+    /// streams from the provider.
+    #[tokio::test]
+    async fn webdav_path_resolution_and_listing() {
+        use crate::application::dtos::pagination::PaginationRequestDto;
+        use crate::application::ports::file_ports::FileRetrievalUseCase;
+        use crate::application::services::file_retrieval_service::FileRetrievalService;
+        use futures::TryStreamExt;
+
+        let (_c, pool) = fresh_db().await;
+        let host = tempfile::tempdir().unwrap();
+        std::fs::create_dir(host.path().join("sub")).unwrap();
+        std::fs::write(host.path().join("a.txt"), b"top").unwrap();
+        std::fs::write(host.path().join("sub/b.txt"), b"nested!").unwrap();
+
+        let p = provision_folder(&pool, "owner", "Media").await;
+        insert_mount(&pool, &p, host.path().to_str().unwrap()).await;
+        let registry = Arc::new(MountRegistry::empty());
+        registry
+            .reload(
+                &ExternalMountPgRepository::new(pool.clone()),
+                &DefaultMountProviderFactory::new(),
+            )
+            .await;
+        let router = Arc::new(MountRouter::new(registry));
+        let folder_service = FolderService::new(
+            Arc::new(FolderDbRepository::new(pool.clone())),
+            acl(&pool),
+            router.clone(),
+        );
+        let retrieval = FileRetrievalService::new_with_authz_for_test(
+            Arc::new(FileBlobReadRepository::new_stub()),
+            acl(&pool),
+        )
+        .with_mount_router(router.clone());
+
+        // The mount root's materialized path; descend into it.
+        let root = folder_service
+            .get_folder(&p.mount_folder_id.to_string())
+            .await
+            .unwrap();
+
+        // get_folder_by_path resolves a mount subdirectory → synthetic ext: id.
+        let sub = folder_service
+            .get_folder_by_path(&format!("{}/sub", root.path), p.drive_id)
+            .await
+            .expect("resolve sub dir by path");
+        assert!(sub.id.starts_with("ext:"));
+        assert_eq!(sub.name, "sub");
+
+        // get_file_by_path resolves a mount file.
+        let file = retrieval
+            .get_file_by_path(&format!("{}/a.txt", root.path), p.drive_id)
+            .await
+            .expect("resolve file by path");
+        assert!(file.id.starts_with("ext:"));
+        assert_eq!(file.size, 3);
+
+        // PROPFIND Depth:1 folder loop: list subdirectories of the mount root.
+        let dirs = folder_service
+            .list_folders_paginated_with_perms(
+                Some(&p.mount_folder_id.to_string()),
+                p.owner_id,
+                &PaginationRequestDto::default(),
+            )
+            .await
+            .expect("list mount subdirs");
+        assert_eq!(
+            dirs.items
+                .iter()
+                .map(|d| d.name.as_str())
+                .collect::<Vec<_>>(),
+            ["sub"]
+        );
+
+        // PROPFIND Depth:1 file loop: list files of the mount root.
+        let files = retrieval
+            .list_files_batch_with_perms(Some(&p.mount_folder_id.to_string()), p.owner_id, 0, 100)
+            .await
+            .expect("list mount files");
+        assert_eq!(
+            files.iter().map(|f| f.name.as_str()).collect::<Vec<_>>(),
+            ["a.txt"]
+        );
+
+        // Content streams from the provider (WebDAV GET) — resolve the nested
+        // file by path, then stream it by its ext: id.
+        let nested = retrieval
+            .get_file_by_path(&format!("{}/sub/b.txt", root.path), p.drive_id)
+            .await
+            .expect("nested file");
+        use futures::TryStreamExt as _;
+        let content: Vec<u8> = Box::into_pin(retrieval.get_file_stream(&nested.id).await.unwrap())
+            .map_ok(|b| b.to_vec())
+            .try_concat()
+            .await
+            .unwrap();
+        assert_eq!(content, b"nested!");
+    }
+
     /// P2: a move that would cross the mount boundary is forbidden.
     #[tokio::test]
     async fn cross_boundary_move_forbidden() {

src/application/services/mount_dto.rs

@@ -14,7 +14,7 @@ use crate::application::dtos::display_helpers::{
 };
 use crate::application::dtos::file_dto::FileDto;
 use crate::application::dtos::folder_dto::FolderDto;
-use crate::application::ports::external_mount_ports::MountStat;
+use crate::application::ports::external_mount_ports::{MountEntry, MountStat};
 use crate::application::services::mount_registry::MountConfig;
 use crate::domain::services::external_mount_id::{
     encode_child_id, virtual_file_etag, virtual_folder_etag,
@@ -71,6 +71,56 @@ pub fn mount_folder_dto(cfg: &MountConfig, parent_id: &str, stat: &MountStat) ->
     }
 }
 
+/// Build a `FolderDto` from a directory listing entry. `parent_id` is the
+/// id-string of the directory being listed.
+pub fn mount_entry_folder_dto(cfg: &MountConfig, parent_id: &str, entry: &MountEntry) -> FolderDto {
+    FolderDto {
+        etag: virtual_folder_etag(entry.modified_at),
+        id: encode_child_id(cfg.mount_id, entry.node_id.clone()),
+        name: entry.name.clone(),
+        path: String::new(),
+        parent_id: Some(parent_id.to_owned()),
+        owner_id: Some(cfg.owner_id.to_string()),
+        drive_id: cfg.drive_id,
+        created_at: entry.created_at,
+        modified_at: entry.modified_at,
+        is_root: false,
+        icon_class: Arc::from("fas fa-folder"),
+        icon_special_class: Arc::from("folder-icon"),
+        category: Arc::from("Folder"),
+        created_by: None,
+        updated_by: None,
+    }
+}
+
+/// Build a `FileDto` from a directory listing entry (mime sniffed from name).
+pub fn mount_entry_file_dto(cfg: &MountConfig, parent_id: &str, entry: &MountEntry) -> FileDto {
+    let name = entry.name.as_str();
+    let mime = mime_guess::from_path(name)
+        .first_or_octet_stream()
+        .to_string();
+    FileDto {
+        id: encode_child_id(cfg.mount_id, entry.node_id.clone()),
+        name: name.to_owned(),
+        path: String::new(),
+        size: entry.size,
+        mime_type: Arc::from(mime.as_str()),
+        folder_id: Some(parent_id.to_owned()),
+        created_at: entry.created_at,
+        modified_at: entry.modified_at,
+        icon_class: Arc::from(icon_class_for(name, &mime)),
+        icon_special_class: Arc::from(icon_special_class_for(name, &mime)),
+        category: Arc::from(category_for(name, &mime)),
+        size_formatted: format_file_size(entry.size),
+        owner_id: Some(cfg.owner_id.to_string()),
+        sort_date: None,
+        content_hash: String::new(),
+        etag: virtual_file_etag(entry.size, entry.modified_at),
+        created_by: None,
+        updated_by: None,
+    }
+}
+
 /// Build a `FileDto` for a mount file from its stat. Virtual files have no blob
 /// hash (`content_hash` empty) and a size+mtime etag.
 pub fn mount_file_dto(cfg: &MountConfig, parent_id: &str, stat: &MountStat) -> FileDto {

src/application/services/mount_registry.rs

@@ -90,6 +90,11 @@ impl MountRegistry {
         internal_path: &str,
     ) -> Option<(Arc<MountConfig>, String)> {
         let index = self.inner.load();
+        // Normalize away a leading slash: materialized folder paths arrive both
+        // as `Personal/Media` (raw `folders.path`) and `/Personal/Media`
+        // (FolderDto / WebDAV internal paths). The index keys are stored without
+        // a leading slash (see `reload`).
+        let internal_path = internal_path.trim_start_matches('/');
         // Walk ancestor paths from the full path up to the root, longest first,
         // so the deepest matching mount wins.
         let mut candidate = internal_path;
@@ -144,7 +149,15 @@ impl MountRegistry {
                     continue;
                 }
             };
-            by_path.insert((rec.drive_id, rec.mount_path.clone()), rec.mount_folder_id);
+            // Store the path key without a leading slash so lookups normalize
+            // consistently (see `find_mount_for_path`).
+            by_path.insert(
+                (
+                    rec.drive_id,
+                    rec.mount_path.trim_start_matches('/').to_string(),
+                ),
+                rec.mount_folder_id,
+            );
             by_folder.insert(
                 rec.mount_folder_id,
                 Arc::new(MountConfig {

src/common/di.rs

@@ -531,12 +531,15 @@ impl AppServiceFactory {
         // bridge (which looks file metadata up by id) can be wired into the
         // dispatcher they receive. It depends only on repos + core, never on
         // the upload service, so the reorder is safe.
-        let file_retrieval_service = Arc::new(FileRetrievalService::new_with_cache(
+        let file_retrieval_service = Arc::new(
-            repos.file_read_repository.clone(),
+            FileRetrievalService::new_with_cache(
-            core.file_content_cache.clone(),
+                repos.file_read_repository.clone(),
-            core.image_transcode_service.clone(),
+                core.file_content_cache.clone(),
-            authz.clone(),
+                core.image_transcode_service.clone(),
-        ));
+                authz.clone(),
+            )
+            .with_mount_router(mount_router.clone()),
+        );
 
         // Effective lifecycle dispatcher: the core hooks (thumbnails, metadata)
         // plus, when the plugins feature is enabled, the WASM plugin bridge.