feat(files): orchestrate connector folder sync

src/govoplan_files/backend/routes/connector_io.py

@@ -9,6 +9,10 @@ from govoplan_core.audit.logging import audit_from_principal
 from govoplan_files.backend.schemas import (
     FileConnectorBrowseItem,
     FileConnectorBrowseResponse,
+    FileConnectorFolderSyncItemResponse,
+    FileConnectorFolderSyncRequest,
+    FileConnectorFolderSyncResponse,
+    FileConnectorFolderSyncSummary,
     FileConnectorImportRequest,
     FileConnectorSyncResponse,
     FileConnectorWriteRequest,
@@ -28,6 +32,11 @@ from govoplan_files.backend.storage.connector_imports import (
     ConnectorImportError,
     ConnectorImportUnsupported,
 )
+from govoplan_files.backend.storage.connector_folder_sync import (
+    connector_relative_path,
+    discover_connector_folder,
+    join_connector_path,
+)
 from govoplan_files.backend.storage.connector_deployment import (
     connector_effective_endpoint_url,
 )
@@ -43,6 +52,7 @@ from govoplan_files.backend.storage.files import (
     sync_file_asset_from_source,
 )
 from govoplan_files.backend.storage.connector_spaces import (
+    connector_space_owner_id,
     get_connector_space_for_user,
 )
 from govoplan_files.backend.storage.connector_writes import write_connector_file
@@ -54,6 +64,7 @@ from govoplan_files.backend.route_support import (
     _audit_connector_sync,
     _connector_browse_next_token,
     _connector_policy_error,
+    _connector_space_policy_decision,
     _download_connector_payload,
     _ensure_campaign_file_access,
     _http_error,
@@ -64,6 +75,259 @@ from govoplan_files.backend.route_support import (
 router = APIRouter(prefix="/files", tags=["files"])
 
 
+@router.post(
+    "/connector-spaces/{space_id}/sync",
+    response_model=FileConnectorFolderSyncResponse,
+)
+def sync_connector_space_folder(
+    space_id: str,
+    payload: FileConnectorFolderSyncRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:upload")),
+):
+    try:
+        space = get_connector_space_for_user(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            space_id=space_id,
+            is_admin=_is_admin(principal),
+        )
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc, not_found=True) from exc
+
+    try:
+        if space.sync_mode != "manual":
+            raise FileStorageError("This connector space is not configured for manual sync")
+        profile = _visible_connector_profile(
+            session, principal, space.connector_profile_id
+        )
+        remote_path = join_connector_path(space.remote_path, payload.path)
+        target_folder = join_connector_path(payload.target_folder)
+        decision = _connector_space_policy_decision(
+            profile,
+            library_id=space.library_id,
+            remote_path=remote_path,
+            operation="sync",
+        )
+        if not decision.allowed:
+            raise ConnectorPolicyDenied(decision)
+        discovery = discover_connector_folder(
+            profile,
+            path=remote_path,
+            library_id=space.library_id,
+            recursive=payload.recursive,
+            max_files=payload.max_files,
+            max_depth=payload.max_depth,
+        )
+    except ConnectorPolicyDenied as exc:
+        session.rollback()
+        raise _connector_policy_error(exc) from exc
+    except ConnectorBrowseUnsupported as exc:
+        session.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=str(exc)
+        ) from exc
+    except (ConnectorBrowseError, FileStorageError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+    owner_id = connector_space_owner_id(space)
+    items: list[FileConnectorFolderSyncItemResponse] = [
+        FileConnectorFolderSyncItemResponse(
+            source_path=skipped.path,
+            action="skipped",
+            detail=skipped.reason,
+        )
+        for skipped in discovery.skipped
+    ]
+    counts = {
+        "created": 0,
+        "updated": 0,
+        "unchanged": 0,
+        "skipped": len(discovery.skipped),
+        "conflicts": 0,
+        "policy_denied": 0,
+        "failed": 0,
+    }
+
+    for source in discovery.files:
+        source_path = normalize_connector_browse_path(source.path)
+        try:
+            relative_path = connector_relative_path(
+                space_root=space.remote_path,
+                item_path=source_path,
+            )
+            target_path = join_connector_path(target_folder, relative_path)
+        except (ConnectorBrowseError, ValueError) as exc:
+            counts["failed"] += 1
+            items.append(
+                FileConnectorFolderSyncItemResponse(
+                    source_path=source_path,
+                    action="failed",
+                    detail=str(exc),
+                )
+            )
+            continue
+
+        try:
+            with session.begin_nested():
+                file_payload = FileConnectorImportRequest(
+                    library_id=space.library_id or "",
+                    path=source_path,
+                    owner_type=space.owner_type,
+                    owner_id=owner_id,
+                    target_path=target_path,
+                    conflict_strategy=payload.conflict_strategy,
+                    source_revision=source.etag,
+                    metadata={
+                        **dict(source.metadata),
+                        **payload.metadata,
+                        "connector_space_id": space.id,
+                        "browse_name": source.name,
+                        "browse_path": source_path,
+                        "browse_modified_at": source.modified_at,
+                        "browse_etag": source.etag,
+                        "folder_sync": True,
+                    },
+                )
+                _source_path, downloaded, metadata = _download_connector_payload(
+                    profile, file_payload, operation="sync"
+                )
+                stored, sync_action, previous_version_id = (
+                    sync_file_asset_from_source(
+                        session,
+                        tenant_id=principal.tenant_id,
+                        owner_type=space.owner_type,
+                        owner_id=owner_id,
+                        user_id=principal.user.id,
+                        filename=downloaded.filename,
+                        data=downloaded.data,
+                        display_path=target_path,
+                        content_type=downloaded.content_type,
+                        metadata=metadata,
+                        conflict_strategy=payload.conflict_strategy,
+                        is_admin=_is_admin(principal),
+                    )
+                )
+                _audit_connector_sync(
+                    session,
+                    principal,
+                    stored.asset,
+                    sync_action=sync_action,
+                    previous_version_id=previous_version_id,
+                )
+                file_response = _asset_response(
+                    session, stored.asset, include_shares=True
+                )
+            counts[sync_action] += 1
+            items.append(
+                FileConnectorFolderSyncItemResponse(
+                    source_path=source_path,
+                    target_path=target_path,
+                    action=sync_action,
+                    file=file_response,
+                    previous_version_id=previous_version_id,
+                    current_version_id=stored.version.id,
+                    source_revision=file_response.source_revision,
+                )
+            )
+        except ConnectorPolicyDenied as exc:
+            counts["policy_denied"] += 1
+            items.append(
+                FileConnectorFolderSyncItemResponse(
+                    source_path=source_path,
+                    target_path=target_path,
+                    action="policy_denied",
+                    source_revision=source.etag,
+                    detail=str(exc),
+                    policy_decision=exc.decision.to_dict(),
+                )
+            )
+        except FileStorageError as exc:
+            detail = str(exc)
+            if detail.startswith("Skipped upload target:"):
+                action = "skipped"
+                counts["skipped"] += 1
+            elif detail.startswith("Target file already exists:"):
+                action = "conflict"
+                counts["conflicts"] += 1
+            else:
+                action = "failed"
+                counts["failed"] += 1
+            items.append(
+                FileConnectorFolderSyncItemResponse(
+                    source_path=source_path,
+                    target_path=target_path,
+                    action=action,
+                    source_revision=source.etag,
+                    detail=detail,
+                )
+            )
+        except (
+            ConnectorImportError,
+            UnsafeFilePathError,
+            OSError,
+            ValueError,
+            json.JSONDecodeError,
+        ) as exc:
+            counts["failed"] += 1
+            items.append(
+                FileConnectorFolderSyncItemResponse(
+                    source_path=source_path,
+                    target_path=target_path,
+                    action="failed",
+                    source_revision=source.etag,
+                    detail=str(exc),
+                )
+            )
+
+    summary = FileConnectorFolderSyncSummary(
+        discovered=len(discovery.files),
+        **counts,
+    )
+    audit_from_principal(
+        session,
+        principal,
+        action="files.connector.folder_synced",
+        object_type="file_connector_space",
+        object_id=space.id,
+        details={
+            "connector_profile_id": profile.id,
+            "provider": profile.provider,
+            "library_id": space.library_id,
+            "remote_path": remote_path,
+            "target_folder": target_folder,
+            "recursive": payload.recursive,
+            "truncated": discovery.truncated,
+            "summary": summary.model_dump(),
+            "results": [
+                {
+                    "source_path": item.source_path,
+                    "target_path": item.target_path,
+                    "action": item.action,
+                    "file_id": item.file.id if item.file else None,
+                    "current_version_id": item.current_version_id,
+                }
+                for item in items
+            ],
+        },
+    )
+    session.commit()
+    return FileConnectorFolderSyncResponse(
+        connector_space_id=space.id,
+        connector_profile_id=profile.id,
+        provider=profile.provider,
+        remote_path=remote_path,
+        target_folder=target_folder,
+        recursive=payload.recursive,
+        truncated=discovery.truncated,
+        summary=summary,
+        items=items,
+    )
+
+
 @router.post(
     "/connector-spaces/{space_id}/write-back",
     response_model=FileConnectorWriteResponse,