initial commit after split

src/govoplan_files/__init__.py

@@ -0,0 +1 @@
+"""govoplan-files module package."""

src/govoplan_files/backend/__init__.py

@@ -0,0 +1 @@
+"""Backend integration for this GovOPlaN module."""

src/govoplan_files/backend/db/__init__.py

@@ -0,0 +1 @@
+from govoplan_files.backend.db.models import *

src/govoplan_files/backend/db/models.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+def new_uuid() -> str:
+    return str(uuid.uuid4())
+
+
+class FileBlob(Base, TimestampMixin):
+    __tablename__ = "file_blobs"
+    __table_args__ = (UniqueConstraint("tenant_id", "checksum_sha256", "size_bytes", name="uq_file_blobs_tenant_checksum_size"),)
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    storage_backend: Mapped[str] = mapped_column(String(50), nullable=False)
+    storage_bucket: Mapped[str | None] = mapped_column(String(255))
+    storage_key: Mapped[str] = mapped_column(String(1000), nullable=False)
+    checksum_sha256: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
+    content_type: Mapped[str | None] = mapped_column(String(255))
+    ref_count: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
+    retained_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
+class FileFolder(Base, TimestampMixin):
+    __tablename__ = "file_folders"
+    __table_args__ = (
+        Index(
+            "uq_file_folders_active_user_path",
+            "tenant_id", "owner_user_id", "path",
+            unique=True,
+            sqlite_where=text("owner_type = 'user' AND deleted_at IS NULL"),
+            postgresql_where=text("owner_type = 'user' AND deleted_at IS NULL"),
+        ),
+        Index(
+            "uq_file_folders_active_group_path",
+            "tenant_id", "owner_group_id", "path",
+            unique=True,
+            sqlite_where=text("owner_type = 'group' AND deleted_at IS NULL"),
+            postgresql_where=text("owner_type = 'group' AND deleted_at IS NULL"),
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    owner_type: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
+    owner_user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    owner_group_id: Mapped[str | None] = mapped_column(ForeignKey("groups.id", ondelete="SET NULL"), nullable=True, index=True)
+    path: Mapped[str] = mapped_column(String(1000), nullable=False, index=True)
+    created_by_user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
+
+
+class FileAsset(Base, TimestampMixin):
+    __tablename__ = "file_assets"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    owner_type: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
+    owner_user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    owner_group_id: Mapped[str | None] = mapped_column(ForeignKey("groups.id", ondelete="SET NULL"), nullable=True, index=True)
+    current_version_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
+    display_path: Mapped[str] = mapped_column(String(1000), nullable=False, index=True)
+    filename: Mapped[str] = mapped_column(String(500), nullable=False, index=True)
+    description: Mapped[str | None] = mapped_column(Text)
+    created_by_user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
+
+
+class FileVersion(Base, TimestampMixin):
+    __tablename__ = "file_versions"
+    __table_args__ = (UniqueConstraint("file_asset_id", "version_number", name="uq_file_versions_asset_number"),)
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    file_asset_id: Mapped[str] = mapped_column(ForeignKey("file_assets.id", ondelete="CASCADE"), nullable=False, index=True)
+    blob_id: Mapped[str] = mapped_column(ForeignKey("file_blobs.id", ondelete="RESTRICT"), nullable=False, index=True)
+    version_number: Mapped[int] = mapped_column(Integer, nullable=False)
+    filename_at_upload: Mapped[str] = mapped_column(String(500), nullable=False)
+    display_path_at_upload: Mapped[str] = mapped_column(String(1000), nullable=False)
+    content_type: Mapped[str | None] = mapped_column(String(255))
+    size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
+    checksum_sha256: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    created_by_user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+
+
+class FileShare(Base, TimestampMixin):
+    __tablename__ = "file_shares"
+    __table_args__ = (UniqueConstraint("file_asset_id", "target_type", "target_id", "revoked_at", name="uq_file_shares_active_target"),)
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    file_asset_id: Mapped[str] = mapped_column(ForeignKey("file_assets.id", ondelete="CASCADE"), nullable=False, index=True)
+    target_type: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
+    target_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    permission: Mapped[str] = mapped_column(String(20), default="read", nullable=False)
+    created_by_user_id: Mapped[str | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+
+
+class CampaignAttachmentUse(Base, TimestampMixin):
+    __tablename__ = "campaign_attachment_uses"
+    __table_args__ = (UniqueConstraint("campaign_job_id", "file_version_id", "filename_used", "use_stage", name="uq_campaign_attachment_uses_job_file_stage"),)
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False, index=True)
+    campaign_id: Mapped[str] = mapped_column(ForeignKey("campaigns.id", ondelete="CASCADE"), nullable=False, index=True)
+    campaign_version_id: Mapped[str] = mapped_column(ForeignKey("campaign_versions.id", ondelete="CASCADE"), nullable=False, index=True)
+    campaign_job_id: Mapped[str | None] = mapped_column(ForeignKey("campaign_jobs.id", ondelete="SET NULL"), nullable=True, index=True)
+    entry_index: Mapped[int | None] = mapped_column(Integer)
+    entry_id: Mapped[str | None] = mapped_column(String(255), index=True)
+    file_asset_id: Mapped[str] = mapped_column(ForeignKey("file_assets.id", ondelete="RESTRICT"), nullable=False, index=True)
+    file_version_id: Mapped[str] = mapped_column(ForeignKey("file_versions.id", ondelete="RESTRICT"), nullable=False, index=True)
+    file_blob_id: Mapped[str] = mapped_column(ForeignKey("file_blobs.id", ondelete="RESTRICT"), nullable=False, index=True)
+    filename_used: Mapped[str] = mapped_column(String(500), nullable=False)
+    checksum_sha256: Mapped[str] = mapped_column(String(64), nullable=False)
+    size_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
+    content_type: Mapped[str | None] = mapped_column(String(255))
+    use_stage: Mapped[str] = mapped_column(String(20), default="built", nullable=False, index=True)
+    used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+
+
+__all__ = [
+    "CampaignAttachmentUse",
+    "FileAsset",
+    "FileBlob",
+    "FileFolder",
+    "FileShare",
+    "FileVersion",
+]

src/govoplan_files/backend/manifest.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from govoplan_core.core.modules import FrontendModule, MigrationSpec, ModuleContext, ModuleManifest, NavItem, PermissionDefinition, RoleTemplate
+from govoplan_core.db.base import Base
+from govoplan_files.backend.db import models as file_models  # noqa: F401 - populate Files ORM metadata
+
+
+def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
+    module_id, resource, action = scope.split(":", 2)
+    return PermissionDefinition(
+        scope=scope,
+        label=label,
+        description=description,
+        category="Files",
+        level="tenant",
+        module_id=module_id,
+        resource=resource,
+        action=action,
+    )
+
+
+PERMISSIONS = (
+    _permission("files:file:read", "View files", "List, search and preview managed files."),
+    _permission("files:file:download", "Download files", "Download managed files and generated archives."),
+    _permission("files:file:upload", "Upload files", "Upload new managed file versions."),
+    _permission("files:file:organize", "Organize files", "Create folders, rename, move or copy managed files."),
+    _permission("files:file:share", "Share files", "Grant or revoke managed file shares."),
+    _permission("files:file:delete", "Delete files", "Delete or hide managed files and folders where policy allows it."),
+    _permission("files:file:admin", "Administer file spaces", "Administer all file spaces in the tenant."),
+)
+
+ROLE_TEMPLATES = (
+    RoleTemplate(
+        slug="file_manager",
+        name="File manager",
+        description="Manage tenant file spaces without campaign delivery rights.",
+        permissions=(
+            "files:file:read",
+            "files:file:download",
+            "files:file:upload",
+            "files:file:organize",
+            "files:file:share",
+            "files:file:delete",
+        ),
+    ),
+    RoleTemplate(
+        slug="file_viewer",
+        name="File viewer",
+        description="Read and download permitted files.",
+        permissions=("files:file:read", "files:file:download"),
+    ),
+)
+
+
+def _files_router(context: ModuleContext):
+    from govoplan_files.backend.runtime import configure_runtime
+
+    configure_runtime(settings=context.settings)
+    from govoplan_files.backend.router import router
+
+    return router
+
+
+manifest = ModuleManifest(
+    id="files",
+    name="Files",
+    version="1.0.0",
+    dependencies=("access",),
+    optional_dependencies=(),
+    permissions=PERMISSIONS,
+    route_factory=_files_router,
+    role_templates=ROLE_TEMPLATES,
+    nav_items=(NavItem(path="/files", label="Files", icon="folder", required_any=("files:file:read",), order=40),),
+    frontend=FrontendModule(
+        module_id="files",
+        package_name="@govoplan/files-webui",
+        nav_items=(NavItem(path="/files", label="Files", icon="folder", required_any=("files:file:read",), order=40),),
+    ),
+    migration_spec=MigrationSpec(
+        module_id="files",
+        metadata=Base.metadata,
+        script_location=str(Path(__file__).with_name("migrations") / "versions"),
+    ),
+)
+
+
+def get_manifest() -> ModuleManifest:
+    return manifest
+

src/govoplan_files/backend/migrations/README.md

@@ -0,0 +1,3 @@
+# Files migrations
+
+New Files-owned database migrations belong in `versions/` and are registered through the Files module manifest. Historical migrations still live in the legacy application until those tables are fully split from the old `multi-seal-mail` repository.

src/govoplan_files/backend/router.py

@@ -0,0 +1,668 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from io import BytesIO
+from typing import Literal
+from urllib.parse import quote
+from fastapi import APIRouter, Depends, File as FastAPIFile, Form, HTTPException, UploadFile, status
+from fastapi.responses import FileResponse, StreamingResponse
+from starlette.background import BackgroundTask
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth.dependencies import ApiPrincipal, has_scope, require_scope
+from govoplan_files.backend.schemas import (
+    ArchiveRequest,
+    BulkDeleteRequest,
+    BulkDeleteResponse,
+    ConflictResolutionRequest,
+    FileAssetResponse,
+    FileFolderCreateRequest,
+    FileFolderDeleteRequest,
+    FileFolderDeleteResponse,
+    FileFolderResponse,
+    FileFoldersResponse,
+    FileListResponse,
+    FileShareRequest,
+    FileShareResponse,
+    FileSpaceResponse,
+    FileSpacesResponse,
+    FileUploadResponse,
+    PatternMatchResponse,
+    PatternResolveRequest,
+    PatternResolveResponse,
+    RenamePreviewItem,
+    RenameRequest,
+    RenameResponse,
+    TransferRequest,
+    TransferResponse,
+    _conflict_resolutions,
+)
+from govoplan_core.db.models import Group, UserGroupMembership
+from govoplan_campaign.backend.db.models import Campaign, CampaignShare
+from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare
+from govoplan_core.db.session import get_session
+from govoplan_files.backend.runtime import settings
+from govoplan_files.backend.storage.paths import UnsafeFilePathError, filename_from_path, normalize_logical_path
+from govoplan_files.backend.storage.access import ensure_group_access, user_group_ids
+from govoplan_files.backend.storage.archives import create_zip_file, extract_zip_upload
+from govoplan_files.backend.storage.common import FileStorageError
+from govoplan_files.backend.storage.files import (
+    asset_is_audit_relevant,
+    create_file_asset,
+    current_version_and_blob,
+    get_asset_for_user,
+    list_assets_for_user,
+    read_asset_bytes,
+    share_file,
+    soft_delete_assets,
+)
+from govoplan_files.backend.storage.folders import create_folder, list_folders_for_user, soft_delete_folder
+from govoplan_files.backend.storage.search import resolve_patterns
+from govoplan_files.backend.storage.transfers import rename_selection, transfer_selection
+
+router = APIRouter(prefix="/files", tags=["files"])
+
+
+
+def _is_admin(principal: ApiPrincipal) -> bool:
+    return has_scope(principal, "files:file:admin")
+
+
+async def _read_limited_upload(upload: UploadFile, *, max_bytes: int) -> bytes:
+    data = await upload.read(max_bytes + 1)
+    if len(data) > max_bytes:
+        raise HTTPException(
+            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            detail=f"Upload exceeds limit of {max_bytes} bytes",
+        )
+    return data
+
+
+
+
+def _cleanup_temp_file(path: str) -> None:
+    try:
+        os.unlink(path)
+    except FileNotFoundError:
+        pass
+
+
+def _attachment_disposition(filename: str) -> str:
+    safe = filename.replace("\\", "_").replace("/", "_").replace("\r", "_").replace("\n", "_").strip() or "download"
+    ascii_name = "".join(
+        char if 32 <= ord(char) < 127 and char not in {'"', "\\", ";"} else "_"
+        for char in safe
+    ).strip() or "download"
+    encoded = quote(safe, safe="")
+    return f'attachment; filename="{ascii_name}"; filename*=UTF-8' + "''" + encoded
+
+
+def _ensure_campaign_file_access(session: Session, principal: ApiPrincipal, campaign_id: str | None) -> None:
+    if not campaign_id:
+        return
+    if not has_scope(principal, "campaigns:campaign:read"):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: campaign:read")
+    campaign = session.get(Campaign, campaign_id)
+    if not campaign or campaign.tenant_id != principal.tenant_id:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign not found")
+    if has_scope(principal, "tenant:*"):
+        return
+    if campaign.owner_user_id == principal.user.id:
+        return
+    group_ids = {
+        row[0]
+        for row in session.query(UserGroupMembership.group_id)
+        .filter(UserGroupMembership.tenant_id == principal.tenant_id, UserGroupMembership.user_id == principal.user.id)
+        .all()
+    }
+    if campaign.owner_group_id and campaign.owner_group_id in group_ids:
+        return
+    share = (
+        session.query(CampaignShare)
+        .filter(
+            CampaignShare.tenant_id == principal.tenant_id,
+            CampaignShare.campaign_id == campaign.id,
+            CampaignShare.revoked_at.is_(None),
+            or_(
+                (CampaignShare.target_type == "user") & (CampaignShare.target_id == principal.user.id),
+                (CampaignShare.target_type == "group") & (CampaignShare.target_id.in_(group_ids)),
+            ),
+        )
+        .first()
+    )
+    if share is None:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No access to this campaign")
+
+
+def _http_error(exc: Exception, *, not_found: bool = False) -> HTTPException:
+    code = status.HTTP_404_NOT_FOUND if not_found else status.HTTP_400_BAD_REQUEST
+    return HTTPException(status_code=code, detail=str(exc))
+
+
+def _owner_id(asset: FileAsset) -> str:
+    return asset.owner_user_id if asset.owner_type == "user" else asset.owner_group_id  # type: ignore[return-value]
+
+
+def _asset_response(session: Session, asset: FileAsset, *, include_shares: bool = False) -> FileAssetResponse:
+    version, blob = current_version_and_blob(session, asset)
+    shares: list[FileShareResponse] = []
+    if include_shares:
+        rows = session.query(FileShare).filter(FileShare.file_asset_id == asset.id).order_by(FileShare.created_at.desc()).all()
+        shares = [
+            FileShareResponse(
+                id=row.id,
+                target_type=row.target_type,
+                target_id=row.target_id,
+                permission=row.permission,
+                created_at=row.created_at.isoformat(),
+                revoked_at=row.revoked_at.isoformat() if row.revoked_at else None,
+            )
+            for row in rows
+        ]
+    return FileAssetResponse(
+        id=asset.id,
+        tenant_id=asset.tenant_id,
+        owner_type=asset.owner_type,
+        owner_id=_owner_id(asset),
+        display_path=asset.display_path,
+        filename=asset.filename,
+        description=asset.description,
+        size_bytes=blob.size_bytes,
+        content_type=blob.content_type,
+        checksum_sha256=blob.checksum_sha256,
+        version_id=version.id,
+        created_at=asset.created_at.isoformat(),
+        updated_at=asset.updated_at.isoformat(),
+        deleted_at=asset.deleted_at.isoformat() if asset.deleted_at else None,
+        audit_relevant=asset_is_audit_relevant(session, asset),
+        metadata=asset.metadata_ or {},
+        shares=shares,
+    )
+
+
+def _folder_owner_id(folder: FileFolder) -> str:
+    return folder.owner_user_id if folder.owner_type == "user" else folder.owner_group_id  # type: ignore[return-value]
+
+
+def _folder_response(folder: FileFolder) -> FileFolderResponse:
+    return FileFolderResponse(
+        id=folder.id,
+        tenant_id=folder.tenant_id,
+        owner_type=folder.owner_type,
+        owner_id=_folder_owner_id(folder),
+        path=folder.path,
+        created_at=folder.created_at.isoformat(),
+        updated_at=folder.updated_at.isoformat(),
+        deleted_at=folder.deleted_at.isoformat() if folder.deleted_at else None,
+    )
+
+
+def _ensure_list_owner_access(session: Session, principal: ApiPrincipal, owner_type: str | None, owner_id: str | None) -> None:
+    if not owner_type:
+        return
+    if owner_type == "user" and owner_id and owner_id != principal.user.id and not _is_admin(principal):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="No access to this user file space")
+    if owner_type == "group" and owner_id:
+        try:
+            ensure_group_access(
+                session,
+                tenant_id=principal.tenant_id,
+                group_id=owner_id,
+                user_id=principal.user.id,
+                is_admin=_is_admin(principal),
+            )
+        except FileStorageError as exc:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
+
+
+@router.get("/spaces", response_model=FileSpacesResponse)
+def list_file_spaces(
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
+):
+    spaces = [
+        FileSpaceResponse(
+            id=f"user:{principal.user.id}",
+            label="My files",
+            owner_type="user",
+            owner_id=principal.user.id,
+            description="Files owned by your user account.",
+        )
+    ]
+    group_ids = user_group_ids(session, tenant_id=principal.tenant_id, user_id=principal.user.id, include_admin_groups=_is_admin(principal))
+    if group_ids:
+        groups = session.query(Group).filter(Group.tenant_id == principal.tenant_id, Group.id.in_(group_ids)).order_by(Group.name.asc()).all()
+        spaces.extend(
+            FileSpaceResponse(
+                id=f"group:{group.id}",
+                label=f"{group.name} files",
+                owner_type="group",
+                owner_id=group.id,
+                description="Files owned by this group.",
+            )
+            for group in groups
+        )
+    return FileSpacesResponse(spaces=spaces)
+
+
+@router.get("/folders", response_model=FileFoldersResponse)
+def list_file_folders(
+    owner_type: Literal["user", "group"],
+    owner_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
+):
+    try:
+        folders = list_folders_for_user(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            owner_type=owner_type,
+            owner_id=owner_id,
+            is_admin=_is_admin(principal),
+        )
+        return FileFoldersResponse(folders=[_folder_response(folder) for folder in folders])
+    except FileStorageError as exc:
+        raise _http_error(exc) from exc
+
+
+@router.post("/folders", response_model=FileFolderResponse)
+def create_file_folder(
+    payload: FileFolderCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:organize")),
+):
+    try:
+        folder = create_folder(
+            session,
+            tenant_id=principal.tenant_id,
+            owner_type=payload.owner_type,
+            owner_id=payload.owner_id,
+            user_id=principal.user.id,
+            path=payload.path,
+            is_admin=_is_admin(principal),
+        )
+        session.commit()
+        return _folder_response(folder)
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.post("/folders/delete", response_model=FileFolderDeleteResponse)
+def delete_file_folder(
+    payload: FileFolderDeleteRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:delete")),
+):
+    try:
+        deleted_folders, deleted_files = soft_delete_folder(
+            session,
+            tenant_id=principal.tenant_id,
+            owner_type=payload.owner_type,
+            owner_id=payload.owner_id,
+            user_id=principal.user.id,
+            path=payload.path,
+            recursive=payload.recursive,
+            is_admin=_is_admin(principal),
+        )
+        session.commit()
+        return FileFolderDeleteResponse(deleted_folders=deleted_folders, deleted_files=deleted_files)
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.get("", response_model=FileListResponse)
+def list_files(
+    owner_type: Literal["user", "group"] | None = None,
+    owner_id: str | None = None,
+    campaign_id: str | None = None,
+    path_prefix: str | None = None,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
+):
+    _ensure_list_owner_access(session, principal, owner_type, owner_id)
+    _ensure_campaign_file_access(session, principal, campaign_id)
+    assets = list_assets_for_user(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=principal.user.id,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        campaign_id=campaign_id,
+        path_prefix=path_prefix,
+        is_admin=_is_admin(principal),
+    )
+    return FileListResponse(files=[_asset_response(session, asset, include_shares=True) for asset in assets])
+
+
+@router.post("/upload", response_model=FileUploadResponse)
+async def upload_files(
+    files: list[UploadFile] = FastAPIFile(...),
+    owner_type: Literal["user", "group"] = Form(default="user"),
+    owner_id: str | None = Form(default=None),
+    path: str = Form(default=""),
+    campaign_id: str | None = Form(default=None),
+    unpack_zip: bool = Form(default=False),
+    conflict_strategy: Literal["reject", "overwrite", "rename"] = Form(default="reject"),
+    conflict_resolutions_json: str | None = Form(default=None),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:upload")),
+):
+    target_owner = owner_id or principal.user.id
+    uploaded_assets: list[FileAsset] = []
+    try:
+        raw_resolutions = json.loads(conflict_resolutions_json) if conflict_resolutions_json else []
+        upload_resolutions = _conflict_resolutions([ConflictResolutionRequest(**item) for item in raw_resolutions])
+        for upload in files:
+            filename = upload.filename or "file"
+            content_type = upload.content_type or None
+            upload_limit = settings.file_upload_zip_max_bytes if unpack_zip and filename.lower().endswith(".zip") else settings.file_upload_max_bytes
+            data = await _read_limited_upload(upload, max_bytes=upload_limit)
+            if unpack_zip and filename.lower().endswith(".zip"):
+                extracted = extract_zip_upload(
+                    session,
+                    tenant_id=principal.tenant_id,
+                    owner_type=owner_type,
+                    owner_id=target_owner,
+                    user_id=principal.user.id,
+                    zip_data=data,
+                    folder=path,
+                    campaign_id=campaign_id,
+                    conflict_strategy=conflict_strategy,
+                    conflict_resolutions=upload_resolutions,
+                    is_admin=_is_admin(principal),
+                    max_file_bytes=settings.file_upload_max_bytes,
+                    max_total_bytes=settings.file_upload_zip_max_bytes,
+                )
+                uploaded_assets.extend(item.asset for item in extracted)
+                continue
+            stored = create_file_asset(
+                session,
+                tenant_id=principal.tenant_id,
+                owner_type=owner_type,
+                owner_id=target_owner,
+                user_id=principal.user.id,
+                filename=filename,
+                data=data,
+                folder=path,
+                content_type=content_type,
+                campaign_id=campaign_id,
+                conflict_strategy=conflict_strategy,
+                conflict_resolutions=upload_resolutions,
+                is_admin=_is_admin(principal),
+            )
+            uploaded_assets.append(stored.asset)
+        session.commit()
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+    return FileUploadResponse(files=[_asset_response(session, asset, include_shares=True) for asset in uploaded_assets])
+
+
+@router.post("/upload-zip", response_model=FileUploadResponse)
+async def upload_zip(
+    file: UploadFile = FastAPIFile(...),
+    owner_type: Literal["user", "group"] = Form(default="user"),
+    owner_id: str | None = Form(default=None),
+    path: str = Form(default=""),
+    campaign_id: str | None = Form(default=None),
+    conflict_strategy: Literal["reject", "overwrite", "rename"] = Form(default="reject"),
+    conflict_resolutions_json: str | None = Form(default=None),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:upload")),
+):
+    data = await _read_limited_upload(file, max_bytes=settings.file_upload_zip_max_bytes)
+    target_owner = owner_id or principal.user.id
+    try:
+        raw_resolutions = json.loads(conflict_resolutions_json) if conflict_resolutions_json else []
+        upload_resolutions = _conflict_resolutions([ConflictResolutionRequest(**item) for item in raw_resolutions])
+        extracted = extract_zip_upload(
+            session,
+            tenant_id=principal.tenant_id,
+            owner_type=owner_type,
+            owner_id=target_owner,
+            user_id=principal.user.id,
+            zip_data=data,
+            folder=path,
+            campaign_id=campaign_id,
+            conflict_strategy=conflict_strategy,
+            conflict_resolutions=upload_resolutions,
+            is_admin=_is_admin(principal),
+            max_file_bytes=settings.file_upload_max_bytes,
+            max_total_bytes=settings.file_upload_zip_max_bytes,
+        )
+        session.commit()
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+    return FileUploadResponse(files=[_asset_response(session, item.asset, include_shares=True) for item in extracted])
+
+
+@router.get("/{file_id}", response_model=FileAssetResponse)
+def get_file(
+    file_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
+):
+    try:
+        asset = get_asset_for_user(session, tenant_id=principal.tenant_id, user_id=principal.user.id, asset_id=file_id, is_admin=_is_admin(principal))
+        return _asset_response(session, asset, include_shares=True)
+    except FileStorageError as exc:
+        raise _http_error(exc, not_found=True) from exc
+
+
+@router.get("/{file_id}/download")
+def download_file(
+    file_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:download")),
+):
+    try:
+        asset = get_asset_for_user(session, tenant_id=principal.tenant_id, user_id=principal.user.id, asset_id=file_id, is_admin=_is_admin(principal))
+        data, _, blob = read_asset_bytes(session, asset)
+    except FileStorageError as exc:
+        raise _http_error(exc, not_found=True) from exc
+    headers = {"Content-Disposition": _attachment_disposition(asset.filename)}
+    return StreamingResponse(BytesIO(data), media_type=blob.content_type or "application/octet-stream", headers=headers)
+
+
+@router.delete("/{file_id}", response_model=BulkDeleteResponse)
+def delete_file(
+    file_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:delete")),
+):
+    try:
+        asset = get_asset_for_user(session, tenant_id=principal.tenant_id, user_id=principal.user.id, asset_id=file_id, require_write=True, is_admin=_is_admin(principal))
+        count = soft_delete_assets(session, [asset])
+        session.commit()
+        return BulkDeleteResponse(deleted_count=count)
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc, not_found=True) from exc
+
+
+@router.post("/bulk-delete", response_model=BulkDeleteResponse)
+def bulk_delete_files(
+    payload: BulkDeleteRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:delete")),
+):
+    try:
+        assets = [
+            get_asset_for_user(session, tenant_id=principal.tenant_id, user_id=principal.user.id, asset_id=file_id, require_write=True, is_admin=_is_admin(principal))
+            for file_id in payload.file_ids
+        ]
+        count = soft_delete_assets(session, assets)
+        session.commit()
+        return BulkDeleteResponse(deleted_count=count)
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.post("/{file_id}/shares", response_model=FileShareResponse)
+def create_share(
+    file_id: str,
+    payload: FileShareRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:share")),
+):
+    try:
+        asset = get_asset_for_user(session, tenant_id=principal.tenant_id, user_id=principal.user.id, asset_id=file_id, require_write=True, is_admin=_is_admin(principal))
+        share = share_file(
+            session,
+            tenant_id=principal.tenant_id,
+            asset=asset,
+            target_type=payload.target_type,
+            target_id=payload.target_id,
+            permission=payload.permission,
+            user_id=principal.user.id,
+        )
+        session.commit()
+        return FileShareResponse(
+            id=share.id,
+            target_type=share.target_type,
+            target_id=share.target_id,
+            permission=share.permission,
+            created_at=share.created_at.isoformat(),
+            revoked_at=share.revoked_at.isoformat() if share.revoked_at else None,
+        )
+    except FileStorageError as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.post("/bulk-rename", response_model=RenameResponse)
+def bulk_rename(
+    payload: RenameRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:organize")),
+):
+    try:
+        plan = rename_selection(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            file_ids=payload.file_ids,
+            folder_paths=payload.folder_paths,
+            owner_type=payload.owner_type,
+            owner_id=payload.owner_id,
+            mode=payload.mode,
+            new_name=payload.new_name,
+            find=payload.find,
+            replacement=payload.replacement,
+            prefix=payload.prefix,
+            suffix=payload.suffix,
+            recursive=payload.recursive,
+            dry_run=payload.dry_run,
+            is_admin=_is_admin(principal),
+        )
+        if not payload.dry_run:
+            session.commit()
+        return RenameResponse(
+            dry_run=payload.dry_run,
+            items=[
+                RenamePreviewItem(
+                    kind=item.kind,
+                    id=item.id,
+                    file_id=item.id if item.kind == "file" else None,
+                    folder_path=item.old_path if item.kind == "folder" else None,
+                    old_path=item.old_path,
+                    new_path=item.new_path,
+                )
+                for item in plan
+            ],
+        )
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.post("/transfer", response_model=TransferResponse)
+def transfer_files(
+    payload: TransferRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:organize")),
+):
+    try:
+        files, folders = transfer_selection(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            operation=payload.operation,
+            file_ids=payload.file_ids,
+            folder_paths=payload.folder_paths,
+            source_owner_type=payload.source_owner_type,
+            source_owner_id=payload.source_owner_id,
+            target_owner_type=payload.target_owner_type,
+            target_owner_id=payload.target_owner_id,
+            target_folder=payload.target_folder,
+            conflict_strategy=payload.conflict_strategy,
+            conflict_resolutions=_conflict_resolutions(payload.conflict_resolutions),
+            is_admin=_is_admin(principal),
+        )
+        session.commit()
+        return TransferResponse(operation=payload.operation, files=files, folders=folders)
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        session.rollback()
+        raise _http_error(exc) from exc
+
+
+@router.post("/archive.zip")
+def download_archive(
+    payload: ArchiveRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:download")),
+):
+    try:
+        assets = [
+            get_asset_for_user(session, tenant_id=principal.tenant_id, user_id=principal.user.id, asset_id=file_id, is_admin=_is_admin(principal))
+            for file_id in payload.file_ids
+        ]
+        tmp = tempfile.NamedTemporaryFile(prefix="multimailer-files-", suffix=".zip", delete=False)
+        tmp_path = tmp.name
+        tmp.close()
+        try:
+            create_zip_file(session, assets, tmp_path)
+        except Exception:
+            _cleanup_temp_file(tmp_path)
+            raise
+    except FileStorageError as exc:
+        raise _http_error(exc) from exc
+    filename = filename_from_path(normalize_logical_path(payload.filename, fallback_filename="files.zip"))
+    headers = {"Content-Disposition": _attachment_disposition(filename)}
+    return FileResponse(tmp_path, media_type="application/zip", headers=headers, background=BackgroundTask(_cleanup_temp_file, tmp_path))
+
+
+@router.post("/resolve-patterns", response_model=PatternResolveResponse)
+def resolve_file_patterns(
+    payload: PatternResolveRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(require_scope("files:file:read")),
+):
+    _ensure_list_owner_access(session, principal, payload.owner_type, payload.owner_id)
+    _ensure_campaign_file_access(session, principal, payload.campaign_id)
+    try:
+        assets = list_assets_for_user(
+            session,
+            tenant_id=principal.tenant_id,
+            user_id=principal.user.id,
+            owner_type=payload.owner_type,
+            owner_id=payload.owner_id,
+            campaign_id=payload.campaign_id,
+            path_prefix=payload.path_prefix,
+            is_admin=_is_admin(principal),
+        )
+        resolved, unmatched = resolve_patterns(assets, payload.patterns, base_path=payload.path_prefix, case_sensitive=payload.case_sensitive)
+        return PatternResolveResponse(
+            patterns=[PatternMatchResponse(pattern=item.pattern, matches=[_asset_response(session, asset) for asset in item.matches]) for item in resolved],
+            unmatched=[_asset_response(session, asset) for asset in unmatched] if payload.include_unmatched else [],
+        )
+    except (FileStorageError, UnsafeFilePathError, ValueError) as exc:
+        raise _http_error(exc) from exc

src/govoplan_files/backend/runtime.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Any
+
+_runtime_settings: object | None = None
+
+
+def configure_runtime(*, settings: object | None = None) -> None:
+    global _runtime_settings
+    if settings is not None:
+        _runtime_settings = settings
+
+
+def get_settings() -> object:
+    if _runtime_settings is not None:
+        return _runtime_settings
+    try:
+        from govoplan_core.settings import settings as legacy_settings
+    except ModuleNotFoundError as exc:
+        raise RuntimeError("GovOPlaN Files runtime settings are not configured") from exc
+    return legacy_settings
+
+
+class SettingsProxy:
+    def __getattr__(self, name: str) -> Any:
+        return getattr(get_settings(), name)
+
+
+settings = SettingsProxy()

src/govoplan_files/backend/schemas.py

@@ -0,0 +1,186 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+from govoplan_files.backend.storage.common import FileConflictResolution
+
+
+class FileSpaceResponse(BaseModel):
+    id: str
+    label: str
+    owner_type: Literal["user", "group"]
+    owner_id: str
+    description: str | None = None
+
+
+class FileSpacesResponse(BaseModel):
+    spaces: list[FileSpaceResponse]
+
+
+class FileShareResponse(BaseModel):
+    id: str
+    target_type: str
+    target_id: str
+    permission: str
+    created_at: str
+    revoked_at: str | None = None
+
+
+class FileAssetResponse(BaseModel):
+    id: str
+    tenant_id: str
+    owner_type: str
+    owner_id: str
+    display_path: str
+    filename: str
+    description: str | None = None
+    size_bytes: int
+    content_type: str | None = None
+    checksum_sha256: str
+    version_id: str
+    created_at: str
+    updated_at: str
+    deleted_at: str | None = None
+    audit_relevant: bool = False
+    metadata: dict[str, Any] | None = None
+    shares: list[FileShareResponse] = Field(default_factory=list)
+
+
+class FileFolderResponse(BaseModel):
+    id: str
+    tenant_id: str
+    owner_type: str
+    owner_id: str
+    path: str
+    created_at: str
+    updated_at: str
+    deleted_at: str | None = None
+
+
+class FileFoldersResponse(BaseModel):
+    folders: list[FileFolderResponse]
+
+
+class FileFolderCreateRequest(BaseModel):
+    owner_type: Literal["user", "group"]
+    owner_id: str
+    path: str
+
+
+class FileFolderDeleteRequest(BaseModel):
+    owner_type: Literal["user", "group"]
+    owner_id: str
+    path: str
+    recursive: bool = True
+
+
+class FileFolderDeleteResponse(BaseModel):
+    deleted_folders: int
+    deleted_files: int
+
+
+class FileListResponse(BaseModel):
+    files: list[FileAssetResponse]
+
+
+class FileUploadResponse(BaseModel):
+    files: list[FileAssetResponse]
+
+
+class BulkDeleteRequest(BaseModel):
+    file_ids: list[str]
+
+
+class BulkDeleteResponse(BaseModel):
+    deleted_count: int
+
+
+class ConflictResolutionRequest(BaseModel):
+    target_path: str
+    action: Literal["overwrite", "rename", "skip"]
+    new_path: str | None = None
+
+
+def _conflict_resolutions(items: list[ConflictResolutionRequest] | None) -> list[FileConflictResolution]:
+    return [FileConflictResolution(target_path=item.target_path, action=item.action, new_path=item.new_path) for item in items or []]
+
+
+class FileShareRequest(BaseModel):
+    target_type: Literal["user", "group", "campaign", "tenant"]
+    target_id: str
+    permission: Literal["read", "write", "manage"] = "read"
+
+
+class RenameRequest(BaseModel):
+    file_ids: list[str] = Field(default_factory=list)
+    folder_paths: list[str] = Field(default_factory=list)
+    owner_type: Literal["user", "group"] | None = None
+    owner_id: str | None = None
+    mode: Literal["direct", "prefix", "suffix", "replace"]
+    new_name: str | None = None
+    find: str | None = None
+    replacement: str = ""
+    prefix: str = ""
+    suffix: str = ""
+    recursive: bool = False
+    dry_run: bool = True
+
+
+class RenamePreviewItem(BaseModel):
+    kind: Literal["file", "folder"]
+    id: str
+    file_id: str | None = None
+    folder_path: str | None = None
+    old_path: str
+    new_path: str
+
+
+class RenameResponse(BaseModel):
+    dry_run: bool
+    items: list[RenamePreviewItem]
+
+
+class TransferRequest(BaseModel):
+    operation: Literal["move", "copy"]
+    file_ids: list[str] = Field(default_factory=list)
+    folder_paths: list[str] = Field(default_factory=list)
+    source_owner_type: Literal["user", "group"]
+    source_owner_id: str
+    target_owner_type: Literal["user", "group"]
+    target_owner_id: str
+    target_folder: str = ""
+    conflict_strategy: Literal["reject", "overwrite", "rename"] = "reject"
+    conflict_resolutions: list[ConflictResolutionRequest] = Field(default_factory=list)
+
+
+class TransferResponse(BaseModel):
+    operation: str
+    files: int
+    folders: int
+
+
+class ArchiveRequest(BaseModel):
+    file_ids: list[str]
+    filename: str = "files.zip"
+
+
+class PatternResolveRequest(BaseModel):
+    patterns: list[str]
+    owner_type: Literal["user", "group"] | None = None
+    owner_id: str | None = None
+    campaign_id: str | None = None
+    path_prefix: str | None = None
+    include_unmatched: bool = True
+    case_sensitive: bool = False
+
+
+class PatternMatchResponse(BaseModel):
+    pattern: str
+    matches: list[FileAssetResponse]
+
+
+class PatternResolveResponse(BaseModel):
+    patterns: list[PatternMatchResponse]
+    unmatched: list[FileAssetResponse] = Field(default_factory=list)

src/govoplan_files/backend/storage/__init__.py

@@ -0,0 +1 @@
+"""Managed file storage services for GovOPlaN Files."""

src/govoplan_files/backend/storage/access.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.db.models import Group, UserGroupMembership
+from govoplan_files.backend.storage.common import FileStorageError
+
+
+def user_group_ids(session: Session, *, tenant_id: str, user_id: str, include_admin_groups: bool = False) -> list[str]:
+    if include_admin_groups:
+        return [row.id for row in session.query(Group).filter(Group.tenant_id == tenant_id).order_by(Group.name.asc()).all()]
+    return [
+        row.group_id
+        for row in session.query(UserGroupMembership)
+        .filter(UserGroupMembership.tenant_id == tenant_id, UserGroupMembership.user_id == user_id)
+        .all()
+    ]
+
+
+def ensure_group_access(session: Session, *, tenant_id: str, group_id: str, user_id: str, is_admin: bool = False) -> None:
+    group = session.get(Group, group_id)
+    if not group or group.tenant_id != tenant_id:
+        raise FileStorageError("Group not found")
+    if is_admin:
+        return
+    membership = (
+        session.query(UserGroupMembership)
+        .filter(UserGroupMembership.tenant_id == tenant_id, UserGroupMembership.user_id == user_id, UserGroupMembership.group_id == group_id)
+        .one_or_none()
+    )
+    if membership is None:
+        raise FileStorageError("No access to this group file space")
+
+
+def ensure_owner_access(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, user_id: str, is_admin: bool = False) -> None:
+    owner_type = owner_type.lower().strip()
+    if owner_type == "user":
+        if owner_id != user_id and not is_admin:
+            raise FileStorageError("No access to this user file space")
+        return
+    if owner_type == "group":
+        ensure_group_access(session, tenant_id=tenant_id, group_id=owner_id, user_id=user_id, is_admin=is_admin)
+        return
+    raise FileStorageError("Files must be owned by a user or group")

src/govoplan_files/backend/storage/archives.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import mimetypes
+import zipfile
+from io import BytesIO
+from pathlib import Path
+from typing import Iterable
+
+from sqlalchemy.orm import Session
+
+from govoplan_files.backend.db.models import FileAsset
+from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError, UploadedStoredFile
+from govoplan_files.backend.storage.backends import get_storage_backend
+from govoplan_files.backend.storage.files import create_file_asset, current_version_and_blob
+from govoplan_files.backend.storage.paths import filename_from_path, normalize_folder, normalize_logical_path
+
+
+_ZIP_READ_CHUNK_SIZE = 1024 * 1024
+
+
+def _read_zip_member(
+    archive: zipfile.ZipFile,
+    info: zipfile.ZipInfo,
+    *,
+    max_file_bytes: int,
+    max_total_bytes: int,
+    current_total: int,
+) -> tuple[bytes, int]:
+    parts: list[bytes] = []
+    actual_size = 0
+    with archive.open(info) as source:
+        while True:
+            read_size = min(_ZIP_READ_CHUNK_SIZE, max_file_bytes + 1 - actual_size)
+            chunk = source.read(read_size)
+            if not chunk:
+                break
+            actual_size += len(chunk)
+            if actual_size > max_file_bytes:
+                raise FileStorageError(f"ZIP member {info.filename!r} exceeds per-file limit")
+            if current_total + actual_size > max_total_bytes:
+                raise FileStorageError("ZIP is too large after extraction")
+            parts.append(chunk)
+    return b"".join(parts), current_total + actual_size
+
+
+def create_zip_file(session: Session, assets: Iterable[FileAsset], output_path: str | Path) -> None:
+    backend = get_storage_backend()
+    with zipfile.ZipFile(output_path, mode="w", compression=zipfile.ZIP_DEFLATED) as archive:
+        for asset in assets:
+            _version, blob = current_version_and_blob(session, asset)
+            info = zipfile.ZipInfo(asset.display_path)
+            info.compress_type = zipfile.ZIP_DEFLATED
+            with archive.open(info, "w") as member:
+                for chunk in backend.iter_bytes(blob.storage_key):
+                    if chunk:
+                        member.write(chunk)
+
+
+def extract_zip_upload(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    user_id: str,
+    zip_data: bytes,
+    folder: str | None,
+    campaign_id: str | None,
+    conflict_strategy: str = "reject",
+    conflict_resolutions: Iterable[FileConflictResolution] | None = None,
+    is_admin: bool = False,
+    max_files: int = 1000,
+    max_file_bytes: int = 50 * 1024 * 1024,
+    max_total_bytes: int = 250 * 1024 * 1024,
+) -> list[UploadedStoredFile]:
+    uploaded: list[UploadedStoredFile] = []
+    total = 0
+    base_folder = normalize_folder(folder)
+    try:
+        with zipfile.ZipFile(BytesIO(zip_data)) as archive:
+            infos = [info for info in archive.infolist() if not info.is_dir()]
+            if len(infos) > max_files:
+                raise FileStorageError(f"ZIP contains too many files (limit {max_files})")
+            for info in infos:
+                if info.flag_bits & 0x1:
+                    raise FileStorageError("Encrypted ZIP uploads are not supported")
+                if info.file_size < 0:
+                    raise FileStorageError("Invalid ZIP member")
+                if info.file_size > max_file_bytes:
+                    raise FileStorageError(f"ZIP member {info.filename!r} exceeds per-file limit")
+                if total + info.file_size > max_total_bytes:
+                    raise FileStorageError("ZIP is too large after extraction")
+                inner_path = normalize_logical_path(info.filename)
+                target_path = f"{base_folder}/{inner_path}" if base_folder else inner_path
+                data, total = _read_zip_member(
+                    archive,
+                    info,
+                    max_file_bytes=max_file_bytes,
+                    max_total_bytes=max_total_bytes,
+                    current_total=total,
+                )
+                uploaded.append(
+                    create_file_asset(
+                        session,
+                        tenant_id=tenant_id,
+                        owner_type=owner_type,
+                        owner_id=owner_id,
+                        user_id=user_id,
+                        filename=filename_from_path(inner_path),
+                        data=data,
+                        display_path=target_path,
+                        content_type=mimetypes.guess_type(inner_path)[0] or "application/octet-stream",
+                        campaign_id=campaign_id,
+                        conflict_strategy=conflict_strategy,
+                        conflict_resolutions=conflict_resolutions,
+                        is_admin=is_admin,
+                    )
+                )
+    except zipfile.BadZipFile as exc:
+        raise FileStorageError("Invalid ZIP upload") from exc
+    return uploaded

src/govoplan_files/backend/storage/backends.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable, Protocol
+
+import boto3
+
+from govoplan_files.backend.runtime import settings
+
+
+class StorageBackendError(RuntimeError):
+    pass
+
+
+class StorageBackend(Protocol):
+    name: str
+
+    def put_bytes(self, key: str, data: bytes, *, content_type: str | None = None) -> None: ...
+    def get_bytes(self, key: str) -> bytes: ...
+    def iter_bytes(self, key: str, *, chunk_size: int = 1024 * 1024) -> Iterable[bytes]: ...
+    def delete(self, key: str) -> None: ...
+    def exists(self, key: str) -> bool: ...
+
+
+@dataclass(slots=True)
+class LocalFilesystemStorageBackend:
+    root: Path
+    name: str = "local"
+
+    def __post_init__(self) -> None:
+        self.root = self.root.expanduser().resolve()
+        self.root.mkdir(parents=True, exist_ok=True)
+
+    def _path(self, key: str) -> Path:
+        path = (self.root / key).resolve()
+        if not path.is_relative_to(self.root):
+            raise StorageBackendError("Storage key escapes local storage root")
+        return path
+
+    def put_bytes(self, key: str, data: bytes, *, content_type: str | None = None) -> None:
+        path = self._path(key)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+
+    def get_bytes(self, key: str) -> bytes:
+        path = self._path(key)
+        if not path.exists() or not path.is_file():
+            raise StorageBackendError("Stored object does not exist")
+        return path.read_bytes()
+
+    def iter_bytes(self, key: str, *, chunk_size: int = 1024 * 1024) -> Iterable[bytes]:
+        path = self._path(key)
+        if not path.exists() or not path.is_file():
+            raise StorageBackendError("Stored object does not exist")
+        with path.open("rb") as handle:
+            while True:
+                chunk = handle.read(chunk_size)
+                if not chunk:
+                    break
+                yield chunk
+
+    def delete(self, key: str) -> None:
+        path = self._path(key)
+        if path.exists() and path.is_file():
+            path.unlink()
+
+    def exists(self, key: str) -> bool:
+        path = self._path(key)
+        return path.exists() and path.is_file()
+
+
+@dataclass(slots=True)
+class S3StorageBackend:
+    bucket: str
+    endpoint_url: str
+    region_name: str
+    access_key_id: str
+    secret_access_key: str
+    name: str = "s3"
+
+    @property
+    def client(self):
+        return boto3.client(
+            "s3",
+            endpoint_url=self.endpoint_url,
+            region_name=self.region_name,
+            aws_access_key_id=self.access_key_id,
+            aws_secret_access_key=self.secret_access_key,
+        )
+
+    def put_bytes(self, key: str, data: bytes, *, content_type: str | None = None) -> None:
+        kwargs = {"Bucket": self.bucket, "Key": key, "Body": data}
+        if content_type:
+            kwargs["ContentType"] = content_type
+        self.client.put_object(**kwargs)
+
+    def get_bytes(self, key: str) -> bytes:
+        try:
+            obj = self.client.get_object(Bucket=self.bucket, Key=key)
+            return obj["Body"].read()
+        except Exception as exc:  # pragma: no cover - depends on S3 backend
+            raise StorageBackendError(str(exc)) from exc
+
+    def iter_bytes(self, key: str, *, chunk_size: int = 1024 * 1024) -> Iterable[bytes]:
+        try:
+            obj = self.client.get_object(Bucket=self.bucket, Key=key)
+            body = obj["Body"]
+            while True:
+                chunk = body.read(chunk_size)
+                if not chunk:
+                    break
+                yield chunk
+        except Exception as exc:  # pragma: no cover - depends on S3 backend
+            raise StorageBackendError(str(exc)) from exc
+
+    def delete(self, key: str) -> None:
+        self.client.delete_object(Bucket=self.bucket, Key=key)
+
+    def exists(self, key: str) -> bool:
+        try:
+            self.client.head_object(Bucket=self.bucket, Key=key)
+            return True
+        except Exception:
+            return False
+
+
+def get_storage_backend() -> StorageBackend:
+    backend = settings.file_storage_backend.lower().strip()
+    if backend in {"local", "filesystem", "fs"}:
+        return LocalFilesystemStorageBackend(Path(settings.file_storage_local_root))
+    if backend in {"s3", "garage"}:
+        return S3StorageBackend(
+            bucket=settings.file_storage_s3_bucket or settings.s3_bucket,
+            endpoint_url=settings.file_storage_s3_endpoint_url or settings.s3_endpoint_url,
+            region_name=settings.file_storage_s3_region or settings.s3_region,
+            access_key_id=settings.file_storage_s3_access_key_id or settings.s3_access_key_id,
+            secret_access_key=settings.file_storage_s3_secret_access_key or settings.s3_secret_access_key,
+        )
+    raise StorageBackendError(f"Unsupported file storage backend: {settings.file_storage_backend}")

src/govoplan_files/backend/storage/campaign_attachments.py

@@ -0,0 +1,334 @@
+from __future__ import annotations
+
+import copy
+import json
+import shutil
+import tempfile
+from collections import defaultdict
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+from pathlib import Path, PurePosixPath
+from typing import Any, Iterator
+
+from sqlalchemy.orm import Session
+
+from govoplan_files.backend.db.models import FileAsset
+from govoplan_files.backend.storage.files import current_version_and_blob, list_assets_for_user, read_asset_bytes
+from govoplan_files.backend.storage.paths import normalize_folder, normalize_logical_path, safe_storage_component
+
+
+MANAGED_SOURCE_PREFIX = "managed:"
+
+
+@dataclass(frozen=True, slots=True)
+class ManagedAttachmentFile:
+    local_path: str
+    asset_id: str
+    version_id: str
+    blob_id: str
+    display_path: str
+    relative_path: str
+    filename: str
+    owner_type: str
+    owner_id: str
+    checksum_sha256: str
+    size_bytes: int
+    content_type: str | None
+
+    def as_dict(self) -> dict[str, Any]:
+        payload = asdict(self)
+        payload.pop("local_path", None)
+        return payload
+
+
+@dataclass(slots=True)
+class PreparedCampaignSnapshot:
+    path: Path
+    raw_json: dict[str, Any]
+    managed_files_by_local_path: dict[str, ManagedAttachmentFile]
+    shared_assets: list[FileAsset]
+
+
+def parse_managed_source(value: object) -> tuple[str, str] | None:
+    if not isinstance(value, str) or not value.startswith(MANAGED_SOURCE_PREFIX):
+        return None
+    parts = value.split(":", 2)
+    if len(parts) != 3 or parts[1] not in {"user", "group"} or not parts[2].strip():
+        return None
+    return parts[1], parts[2].strip()
+
+
+def _asset_owner_id(asset: FileAsset) -> str | None:
+    if asset.owner_type == "user":
+        return asset.owner_user_id
+    if asset.owner_type == "group":
+        return asset.owner_group_id
+    return None
+
+
+def _relative_asset_path(asset: FileAsset, logical_root: str) -> str | None:
+    display_path = normalize_logical_path(asset.display_path)
+    root = normalize_folder(logical_root)
+    if not root:
+        return display_path
+    prefix = f"{root}/"
+    if not display_path.startswith(prefix):
+        return None
+    return display_path[len(prefix) :]
+
+
+def _safe_local_target(root: Path, relative_path: str) -> Path:
+    parts = PurePosixPath(normalize_logical_path(relative_path)).parts
+    if not parts or any(part in {"", ".", ".."} for part in parts):
+        raise ValueError(f"Unsafe managed attachment path: {relative_path!r}")
+    target = root.joinpath(*parts).resolve()
+    resolved_root = root.resolve()
+    if not target.is_relative_to(resolved_root):
+        raise ValueError(f"Managed attachment path escapes materialization root: {relative_path!r}")
+    return target
+
+
+def _iter_rule_dicts(attachments: dict[str, Any], raw_json: dict[str, Any]):
+    global_rules = attachments.get("global")
+    if isinstance(global_rules, list):
+        for rule in global_rules:
+            if isinstance(rule, dict):
+                yield rule
+
+    entries = raw_json.get("entries")
+    inline = entries.get("inline") if isinstance(entries, dict) else None
+    if isinstance(inline, list):
+        for entry in inline:
+            if not isinstance(entry, dict):
+                continue
+            rules = entry.get("attachments")
+            if isinstance(rules, list):
+                for rule in rules:
+                    if isinstance(rule, dict):
+                        yield rule
+
+
+def _selected_base_path(
+    rule: dict[str, Any],
+    prepared_by_id: dict[str, tuple[str, str]],
+    prepared_by_old_path: dict[str, list[tuple[str, str]]],
+    first_prepared: tuple[str, str] | None,
+) -> tuple[str, str] | None:
+    base_path_id = str(rule.get("base_path_id") or "").strip()
+    if base_path_id and base_path_id in prepared_by_id:
+        return prepared_by_id[base_path_id]
+
+    base_dir = str(rule.get("base_dir") or ".").strip() or "."
+    candidates = prepared_by_old_path.get(base_dir)
+    if candidates:
+        return candidates[0]
+    if base_dir in {"", "."}:
+        return first_prepared
+    return None
+
+
+def prepare_campaign_snapshot(
+    session: Session,
+    *,
+    tenant_id: str,
+    campaign_id: str,
+    raw_json: dict[str, Any],
+    destination: Path,
+    include_bytes: bool,
+) -> PreparedCampaignSnapshot:
+    """Create a temporary file-oriented campaign snapshot for managed attachments.
+
+    The existing mailer resolver deliberately remains file-oriented. Managed
+    campaign-shared file versions are materialized into an isolated tree and a
+    copied campaign JSON is rewritten to point to those directories. The
+    returned manifest preserves exact asset/version/blob identity so build and
+    audit code never has to guess by filename.
+    """
+
+    destination = destination.expanduser().resolve()
+    destination.mkdir(parents=True, exist_ok=True)
+    materialized_root = destination / "managed-attachments"
+    materialized_root.mkdir(parents=True, exist_ok=True)
+
+    prepared_json = copy.deepcopy(raw_json if isinstance(raw_json, dict) else {})
+    attachments = prepared_json.get("attachments")
+    if not isinstance(attachments, dict):
+        attachments = {}
+        prepared_json["attachments"] = attachments
+    base_paths = attachments.get("base_paths")
+    if not isinstance(base_paths, list):
+        base_paths = []
+
+    shared_assets = list_assets_for_user(
+        session,
+        tenant_id=tenant_id,
+        user_id="",
+        campaign_id=campaign_id,
+        is_admin=True,
+    )
+
+    assets_by_owner: dict[tuple[str, str], list[FileAsset]] = defaultdict(list)
+    for asset in shared_assets:
+        owner_id = _asset_owner_id(asset)
+        if owner_id:
+            assets_by_owner[(asset.owner_type, owner_id)].append(asset)
+
+    manifest: dict[str, ManagedAttachmentFile] = {}
+    prepared_by_id: dict[str, tuple[str, str]] = {}
+    prepared_by_old_path: dict[str, list[tuple[str, str]]] = {}
+    first_prepared: tuple[str, str] | None = None
+
+    for index, item in enumerate(base_paths):
+        if not isinstance(item, dict):
+            continue
+        parsed_source = parse_managed_source(item.get("source"))
+        if parsed_source is None:
+            continue
+        owner_type, owner_id = parsed_source
+        old_path = str(item.get("path") or ".").strip() or "."
+        logical_root = "" if old_path in {"", ".", "/"} else normalize_folder(old_path)
+        base_path_id = str(item.get("id") or f"base-path-{index + 1}")
+        local_root = materialized_root / f"{index + 1:03d}-{safe_storage_component(base_path_id)}"
+        local_root.mkdir(parents=True, exist_ok=True)
+        local_root_string = str(local_root.resolve())
+
+        prepared = (base_path_id, local_root_string)
+        prepared_by_id[base_path_id] = prepared
+        prepared_by_old_path.setdefault(old_path, []).append(prepared)
+        if first_prepared is None:
+            first_prepared = prepared
+
+        item["path"] = local_root_string
+
+        for asset in assets_by_owner.get((owner_type, owner_id), []):
+            relative_path = _relative_asset_path(asset, logical_root)
+            if not relative_path:
+                continue
+            target = _safe_local_target(local_root, relative_path)
+            target.parent.mkdir(parents=True, exist_ok=True)
+            if include_bytes:
+                data, version, blob = read_asset_bytes(session, asset)
+                target.write_bytes(data)
+            else:
+                version, blob = current_version_and_blob(session, asset)
+                target.touch()
+            local_key = str(target.resolve())
+            manifest[local_key] = ManagedAttachmentFile(
+                local_path=local_key,
+                asset_id=asset.id,
+                version_id=version.id,
+                blob_id=blob.id,
+                display_path=asset.display_path,
+                relative_path=normalize_logical_path(relative_path),
+                filename=asset.filename,
+                owner_type=asset.owner_type,
+                owner_id=owner_id,
+                checksum_sha256=blob.checksum_sha256,
+                size_bytes=blob.size_bytes,
+                content_type=blob.content_type,
+            )
+
+    for rule in _iter_rule_dicts(attachments, prepared_json):
+        selected = _selected_base_path(rule, prepared_by_id, prepared_by_old_path, first_prepared)
+        if selected is None:
+            continue
+        base_path_id, local_root_string = selected
+        rule["base_path_id"] = base_path_id
+        rule["base_dir"] = local_root_string
+
+    if first_prepared is not None:
+        attachments["base_path"] = first_prepared[1]
+
+    snapshot_path = destination / "campaign.json"
+    snapshot_path.write_text(json.dumps(prepared_json, ensure_ascii=False, indent=2), encoding="utf-8")
+    return PreparedCampaignSnapshot(
+        path=snapshot_path,
+        raw_json=prepared_json,
+        managed_files_by_local_path=manifest,
+        shared_assets=shared_assets,
+    )
+
+
+@contextmanager
+def prepared_campaign_snapshot(
+    session: Session,
+    *,
+    tenant_id: str,
+    campaign_id: str,
+    raw_json: dict[str, Any],
+    include_bytes: bool,
+    prefix: str = "multimailer-managed-campaign-",
+) -> Iterator[PreparedCampaignSnapshot]:
+    temp_dir = Path(tempfile.mkdtemp(prefix=prefix))
+    try:
+        yield prepare_campaign_snapshot(
+            session,
+            tenant_id=tenant_id,
+            campaign_id=campaign_id,
+            raw_json=raw_json,
+            destination=temp_dir,
+            include_bytes=include_bytes,
+        )
+    finally:
+        shutil.rmtree(temp_dir, ignore_errors=True)
+
+
+def managed_match_payloads(
+    matches: list[str],
+    manifest: dict[str, ManagedAttachmentFile],
+) -> list[dict[str, Any]]:
+    payloads: list[dict[str, Any]] = []
+    for match in matches:
+        item = manifest.get(str(Path(match).resolve()))
+        if item is not None:
+            payloads.append(item.as_dict())
+    return payloads
+
+
+def public_attachment_summary_payload(value: Any) -> dict[str, Any]:
+    """Return an attachment summary without temporary materialization paths.
+
+    Managed builds use isolated local directories internally. Queue, review and
+    audit payloads must expose the stable managed paths and immutable IDs
+    instead of those temporary paths. Legacy filesystem attachments remain
+    unchanged for backwards compatibility.
+    """
+
+    if hasattr(value, "model_dump"):
+        payload = value.model_dump(mode="json")
+    elif isinstance(value, dict):
+        payload = copy.deepcopy(value)
+    else:
+        return {}
+
+    managed_matches = payload.get("managed_matches")
+    if not isinstance(managed_matches, list) or not managed_matches:
+        return payload
+
+    logical_matches: list[str] = []
+    for item in managed_matches:
+        if not isinstance(item, dict):
+            continue
+        display_path = str(item.get("display_path") or item.get("relative_path") or item.get("filename") or "").strip()
+        if display_path:
+            logical_matches.append(display_path)
+
+    payload["matches"] = logical_matches
+    # These values point into a deleted temporary materialization directory.
+    # The named source plus managed match metadata are the stable references.
+    payload["base_path"] = None
+    payload["directory"] = payload.get("base_path_name") or "managed"
+    return payload
+
+
+def annotate_built_messages_with_managed_files(
+    built_messages: list[Any],
+    manifest: dict[str, ManagedAttachmentFile],
+) -> None:
+    """Attach exact managed-file identities to built attachment summaries."""
+
+    for built in built_messages:
+        draft = getattr(built, "draft", None)
+        for attachment in getattr(draft, "attachments", []) if draft is not None else []:
+            matches = list(getattr(attachment, "matches", []) or [])
+            attachment.managed_matches = managed_match_payloads(matches, manifest)

src/govoplan_files/backend/storage/campaign_usage.py

@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from pathlib import PurePosixPath
+from typing import Iterable
+
+from sqlalchemy.orm import Session
+
+from govoplan_campaign.backend.db.models import CampaignJob
+from govoplan_files.backend.db.models import CampaignAttachmentUse, FileAsset, FileBlob, FileVersion
+from govoplan_files.backend.storage.common import utcnow
+from govoplan_files.backend.storage.files import current_version_and_blob, list_assets_for_user
+
+
+def _candidate_match_keys(raw_match: str) -> set[str]:
+    cleaned = raw_match.replace("\\", "/").strip().strip("/")
+    result = {cleaned}
+    if cleaned:
+        result.add(PurePosixPath(cleaned).name)
+    return {item for item in result if item}
+
+
+AttachmentUseKey = tuple[str, str, str, str]
+
+
+def _attachment_use_key(*, job_id: str, file_version_id: str, filename_used: str, stage: str) -> AttachmentUseKey:
+    return job_id, file_version_id, filename_used, stage
+
+
+def _known_use_keys_for_jobs(session: Session, jobs: Iterable[CampaignJob], *, stage: str) -> set[AttachmentUseKey]:
+    job_ids = {job.id for job in jobs if job.id}
+    if not job_ids:
+        return set()
+
+    keys = {
+        _attachment_use_key(
+            job_id=job_id,
+            file_version_id=file_version_id,
+            filename_used=filename_used,
+            stage=use_stage,
+        )
+        for job_id, file_version_id, filename_used, use_stage in (
+            session.query(
+                CampaignAttachmentUse.campaign_job_id,
+                CampaignAttachmentUse.file_version_id,
+                CampaignAttachmentUse.filename_used,
+                CampaignAttachmentUse.use_stage,
+            )
+            .filter(
+                CampaignAttachmentUse.campaign_job_id.in_(job_ids),
+                CampaignAttachmentUse.use_stage == stage,
+            )
+            .all()
+        )
+        if job_id is not None
+    }
+    for pending in session.new:
+        if not isinstance(pending, CampaignAttachmentUse):
+            continue
+        if pending.campaign_job_id not in job_ids or pending.use_stage != stage:
+            continue
+        keys.add(
+            _attachment_use_key(
+                job_id=pending.campaign_job_id,
+                file_version_id=pending.file_version_id,
+                filename_used=pending.filename_used,
+                stage=pending.use_stage,
+            )
+        )
+    return keys
+
+
+def _known_use_keys(session: Session, job: CampaignJob, *, stage: str) -> set[AttachmentUseKey]:
+    return _known_use_keys_for_jobs(session, [job], stage=stage)
+
+
+def _add_use(
+    session: Session,
+    job: CampaignJob,
+    *,
+    asset: FileAsset,
+    version: FileVersion,
+    blob: FileBlob,
+    filename_used: str,
+    stage: str,
+    known_keys: set[AttachmentUseKey],
+) -> None:
+    key = _attachment_use_key(
+        job_id=job.id,
+        file_version_id=version.id,
+        filename_used=filename_used,
+        stage=stage,
+    )
+    if key in known_keys:
+        return
+    known_keys.add(key)
+    session.add(
+        CampaignAttachmentUse(
+            tenant_id=job.tenant_id,
+            campaign_id=job.campaign_id,
+            campaign_version_id=job.campaign_version_id,
+            campaign_job_id=job.id,
+            entry_index=job.entry_index,
+            entry_id=job.entry_id,
+            file_asset_id=asset.id,
+            file_version_id=version.id,
+            file_blob_id=blob.id,
+            filename_used=filename_used,
+            checksum_sha256=blob.checksum_sha256,
+            size_bytes=blob.size_bytes,
+            content_type=blob.content_type,
+            use_stage=stage,
+        )
+    )
+
+
+def record_campaign_attachment_uses_for_jobs(
+    session: Session,
+    jobs: Iterable[CampaignJob],
+    *,
+    stage: str = "built",
+) -> None:
+    """Record immutable attachment evidence for multiple jobs efficiently.
+
+    Build operations may create thousands of jobs in one transaction. Loading
+    every campaign-shared asset and flushing once per recipient turns that into
+    an avoidable query/flush multiplier. This batch path resolves the exact
+    managed IDs once, fetches each referenced table in bulk, and reuses legacy
+    filename maps per campaign only when an older snapshot needs them.
+    """
+
+    job_list = [job for job in jobs if job.id]
+    if not job_list:
+        return
+
+    managed_by_job: dict[str, list[dict[str, object]]] = defaultdict(list)
+    fallback_attachments_by_job: dict[str, list[dict[str, object]]] = defaultdict(list)
+    job_by_id = {job.id: job for job in job_list}
+    asset_ids: set[str] = set()
+    version_ids: set[str] = set()
+    blob_ids: set[str] = set()
+
+    for job in job_list:
+        attachments = job.resolved_attachments or []
+        if not isinstance(attachments, list):
+            continue
+        for attachment in attachments:
+            if not isinstance(attachment, dict):
+                continue
+            managed_matches = attachment.get("managed_matches")
+            if isinstance(managed_matches, list) and managed_matches:
+                for item in managed_matches:
+                    if not isinstance(item, dict):
+                        continue
+                    asset_id = str(item.get("asset_id") or "")
+                    version_id = str(item.get("version_id") or "")
+                    blob_id = str(item.get("blob_id") or "")
+                    if not asset_id or not version_id or not blob_id:
+                        continue
+                    managed_by_job[job.id].append(item)
+                    asset_ids.add(asset_id)
+                    version_ids.add(version_id)
+                    blob_ids.add(blob_id)
+            else:
+                fallback_attachments_by_job[job.id].append(attachment)
+
+    assets_by_id = {
+        item.id: item
+        for item in session.query(FileAsset).filter(FileAsset.id.in_(asset_ids)).all()
+    } if asset_ids else {}
+    versions_by_id = {
+        item.id: item
+        for item in session.query(FileVersion).filter(FileVersion.id.in_(version_ids)).all()
+    } if version_ids else {}
+    blobs_by_id = {
+        item.id: item
+        for item in session.query(FileBlob).filter(FileBlob.id.in_(blob_ids)).all()
+    } if blob_ids else {}
+    known_keys = _known_use_keys_for_jobs(session, job_list, stage=stage)
+
+    for job_id, items in managed_by_job.items():
+        job = job_by_id[job_id]
+        for item in items:
+            asset = assets_by_id.get(str(item.get("asset_id") or ""))
+            version = versions_by_id.get(str(item.get("version_id") or ""))
+            blob = blobs_by_id.get(str(item.get("blob_id") or ""))
+            if not asset or not version or not blob:
+                continue
+            if asset.tenant_id != job.tenant_id or version.tenant_id != job.tenant_id or blob.tenant_id != job.tenant_id:
+                continue
+            if version.file_asset_id != asset.id or version.blob_id != blob.id:
+                continue
+            _add_use(
+                session,
+                job,
+                asset=asset,
+                version=version,
+                blob=blob,
+                filename_used=str(item.get("filename") or asset.filename),
+                stage=stage,
+                known_keys=known_keys,
+            )
+
+    assets_by_campaign: dict[tuple[str, str], dict[str, FileAsset]] = {}
+    for job_id, attachments in fallback_attachments_by_job.items():
+        job = job_by_id[job_id]
+        campaign_key = (job.tenant_id, job.campaign_id)
+        by_key = assets_by_campaign.get(campaign_key)
+        if by_key is None:
+            assets = list_assets_for_user(
+                session,
+                tenant_id=job.tenant_id,
+                user_id="",
+                campaign_id=job.campaign_id,
+                is_admin=True,
+            )
+            by_key = {}
+            for asset in assets:
+                by_key[asset.display_path.strip("/")] = asset
+                by_key.setdefault(asset.filename, asset)
+            assets_by_campaign[campaign_key] = by_key
+
+        for attachment in attachments:
+            matches = attachment.get("matches") if isinstance(attachment.get("matches"), list) else []
+            for raw in matches:
+                if not isinstance(raw, str):
+                    continue
+                asset = next((by_key[key] for key in _candidate_match_keys(raw) if key in by_key), None)
+                if not asset:
+                    continue
+                version, blob = current_version_and_blob(session, asset)
+                _add_use(
+                    session,
+                    job,
+                    asset=asset,
+                    version=version,
+                    blob=blob,
+                    filename_used=asset.filename,
+                    stage=stage,
+                    known_keys=known_keys,
+                )
+
+
+def record_campaign_attachment_uses_for_job(session: Session, job: CampaignJob, *, stage: str = "built") -> None:
+    """Record immutable managed file versions used by one built/sent job."""
+
+    record_campaign_attachment_uses_for_jobs(session, [job], stage=stage)
+
+
+def mark_job_attachment_uses_sent(session: Session, job: CampaignJob) -> None:
+    record_campaign_attachment_uses_for_job(session, job, stage="built")
+    # Sessions use autoflush=False. Flush any compatibility-built evidence so
+    # the following query can copy it to the sent stage in the same call.
+    session.flush()
+    now = utcnow()
+    uses = (
+        session.query(CampaignAttachmentUse)
+        .filter(
+            CampaignAttachmentUse.tenant_id == job.tenant_id,
+            CampaignAttachmentUse.campaign_job_id == job.id,
+            CampaignAttachmentUse.use_stage == "built",
+        )
+        .all()
+    )
+    sent_keys = _known_use_keys(session, job, stage="sent")
+    for use in uses:
+        key = _attachment_use_key(
+            job_id=job.id,
+            file_version_id=use.file_version_id,
+            filename_used=use.filename_used,
+            stage="sent",
+        )
+        if key in sent_keys:
+            continue
+        sent_keys.add(key)
+        session.add(
+            CampaignAttachmentUse(
+                tenant_id=use.tenant_id,
+                campaign_id=use.campaign_id,
+                campaign_version_id=use.campaign_version_id,
+                campaign_job_id=use.campaign_job_id,
+                entry_index=use.entry_index,
+                entry_id=use.entry_id,
+                file_asset_id=use.file_asset_id,
+                file_version_id=use.file_version_id,
+                file_blob_id=use.file_blob_id,
+                filename_used=use.filename_used,
+                checksum_sha256=use.checksum_sha256,
+                size_bytes=use.size_bytes,
+                content_type=use.content_type,
+                use_stage="sent",
+                used_at=now,
+            )
+        )

src/govoplan_files/backend/storage/common.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+from govoplan_files.backend.db.models import FileAsset, FileBlob, FileVersion
+
+
+class FileStorageError(RuntimeError):
+    pass
+
+
+@dataclass(slots=True)
+class UploadedStoredFile:
+    asset: FileAsset
+    version: FileVersion
+    blob: FileBlob
+
+
+@dataclass(slots=True)
+class ResolvedPattern:
+    pattern: str
+    matches: list[FileAsset]
+
+
+@dataclass(slots=True)
+class FileConflictResolution:
+    target_path: str
+    action: str
+    new_path: str | None = None
+
+
+@dataclass(slots=True)
+class RenamePlanItem:
+    kind: str
+    id: str
+    old_path: str
+    new_path: str
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)

src/govoplan_files/backend/storage/files.py

@@ -0,0 +1,500 @@
+from __future__ import annotations
+
+import hashlib
+import mimetypes
+from pathlib import PurePosixPath
+from typing import Any, Iterable
+from uuid import uuid4
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.db.models import Group, Tenant, User
+from govoplan_campaign.backend.db.models import Campaign
+from govoplan_files.backend.db.models import CampaignAttachmentUse, FileAsset, FileBlob, FileShare, FileVersion
+from govoplan_files.backend.runtime import settings
+from govoplan_files.backend.storage.access import ensure_owner_access, user_group_ids
+from govoplan_files.backend.storage.backends import get_storage_backend
+from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError, UploadedStoredFile, utcnow
+from govoplan_files.backend.storage.paths import filename_from_path, join_folder_filename, normalize_folder, normalize_logical_path, safe_storage_component
+
+
+def _asset_query_for_owner(session: Session, *, tenant_id: str, owner_type: str, owner_id: str):
+    query = session.query(FileAsset).filter(FileAsset.tenant_id == tenant_id, FileAsset.owner_type == owner_type)
+    if owner_type == "user":
+        return query.filter(FileAsset.owner_user_id == owner_id)
+    if owner_type == "group":
+        return query.filter(FileAsset.owner_group_id == owner_id)
+    raise FileStorageError("Unsupported owner type")
+
+
+def _storage_bucket_name() -> str:
+    return settings.file_storage_s3_bucket or settings.s3_bucket
+
+
+def _storage_backend_name() -> str:
+    return settings.file_storage_backend.lower().strip()
+
+
+def _storage_key(*, tenant_id: str, checksum: str, filename: str) -> str:
+    return f"tenants/{tenant_id}/files/{checksum[:2]}/{uuid4().hex}-{safe_storage_component(filename)}"
+
+
+def _get_or_create_blob(
+    session: Session,
+    *,
+    tenant_id: str,
+    data: bytes,
+    filename: str,
+    content_type: str | None,
+) -> FileBlob:
+    checksum = hashlib.sha256(data).hexdigest()
+    size = len(data)
+    blob = (
+        session.query(FileBlob)
+        .filter(FileBlob.tenant_id == tenant_id, FileBlob.checksum_sha256 == checksum, FileBlob.size_bytes == size)
+        .one_or_none()
+    )
+    if blob:
+        blob.ref_count += 1
+        session.add(blob)
+        return blob
+
+    storage_key = _storage_key(tenant_id=tenant_id, checksum=checksum, filename=filename)
+    backend = get_storage_backend()
+    backend.put_bytes(storage_key, data, content_type=content_type)
+    blob = FileBlob(
+        tenant_id=tenant_id,
+        storage_backend=_storage_backend_name(),
+        storage_bucket=_storage_bucket_name(),
+        storage_key=storage_key,
+        checksum_sha256=checksum,
+        size_bytes=size,
+        content_type=content_type,
+        ref_count=1,
+    )
+    session.add(blob)
+    session.flush()
+    return blob
+
+
+def create_file_asset(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    user_id: str,
+    filename: str,
+    data: bytes,
+    folder: str | None = None,
+    display_path: str | None = None,
+    content_type: str | None = None,
+    description: str | None = None,
+    metadata: dict[str, Any] | None = None,
+    campaign_id: str | None = None,
+    conflict_strategy: str = "reject",
+    conflict_resolutions: Iterable[FileConflictResolution] | None = None,
+    is_admin: bool = False,
+) -> UploadedStoredFile:
+    owner_type = owner_type.lower().strip()
+    ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)
+
+    safe_filename = filename_from_path(normalize_logical_path(filename, fallback_filename="file"))
+    logical_path = normalize_logical_path(display_path) if display_path else join_folder_filename(folder, safe_filename)
+    if not content_type:
+        content_type = mimetypes.guess_type(safe_filename)[0] or "application/octet-stream"
+
+    conflict_strategy = _normalize_conflict_strategy(conflict_strategy)
+    resolutions = _resolution_by_path(conflict_resolutions)
+    resolution = resolutions.get(logical_path)
+    action = resolution.action if resolution else conflict_strategy
+    if resolution and resolution.new_path and action == "rename":
+        logical_path = normalize_logical_path(resolution.new_path)
+    elif _active_asset_exists(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=logical_path):
+        if action == "reject":
+            raise FileStorageError(f"Target file already exists: {logical_path}")
+        if action == "overwrite":
+            _soft_delete_conflicting_asset(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, path=logical_path)
+        elif action == "rename":
+            logical_path = _next_available_logical_path(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, desired_path=logical_path)
+        elif action == "skip":
+            raise FileStorageError(f"Skipped upload target: {logical_path}")
+    elif action == "rename":
+        logical_path = _next_available_logical_path(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, desired_path=logical_path)
+
+    blob = _get_or_create_blob(session, tenant_id=tenant_id, data=data, filename=safe_filename, content_type=content_type)
+    asset = FileAsset(
+        tenant_id=tenant_id,
+        owner_type=owner_type,
+        owner_user_id=owner_id if owner_type == "user" else None,
+        owner_group_id=owner_id if owner_type == "group" else None,
+        display_path=logical_path,
+        filename=filename_from_path(logical_path),
+        description=description,
+        created_by_user_id=user_id,
+        metadata_=metadata or {},
+    )
+    session.add(asset)
+    session.flush()
+    version = FileVersion(
+        tenant_id=tenant_id,
+        file_asset_id=asset.id,
+        blob_id=blob.id,
+        version_number=1,
+        filename_at_upload=safe_filename,
+        display_path_at_upload=logical_path,
+        content_type=content_type,
+        size_bytes=blob.size_bytes,
+        checksum_sha256=blob.checksum_sha256,
+        created_by_user_id=user_id,
+    )
+    session.add(version)
+    session.flush()
+    asset.current_version_id = version.id
+    session.add(asset)
+    if campaign_id:
+        share_file(session, tenant_id=tenant_id, asset=asset, target_type="campaign", target_id=campaign_id, permission="read", user_id=user_id)
+    return UploadedStoredFile(asset=asset, version=version, blob=blob)
+
+
+def get_asset_for_user(session: Session, *, tenant_id: str, user_id: str, asset_id: str, require_write: bool = False, is_admin: bool = False) -> FileAsset:
+    asset = session.get(FileAsset, asset_id)
+    if not asset or asset.tenant_id != tenant_id or asset.deleted_at is not None:
+        raise FileStorageError("File not found")
+    if is_admin:
+        return asset
+    group_ids = user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
+    owns = (asset.owner_type == "user" and asset.owner_user_id == user_id) or (asset.owner_type == "group" and asset.owner_group_id in group_ids)
+    if owns:
+        return asset
+    permission_values = ["read", "write", "manage"] if not require_write else ["write", "manage"]
+    share = (
+        session.query(FileShare)
+        .filter(
+            FileShare.tenant_id == tenant_id,
+            FileShare.file_asset_id == asset.id,
+            FileShare.revoked_at.is_(None),
+            FileShare.permission.in_(permission_values),
+            or_(
+                (FileShare.target_type == "user") & (FileShare.target_id == user_id),
+                (FileShare.target_type == "group") & (FileShare.target_id.in_(group_ids)),
+                (FileShare.target_type == "tenant") & (FileShare.target_id == tenant_id),
+            ),
+        )
+        .first()
+    )
+    if not share:
+        raise FileStorageError("No access to this file")
+    return asset
+
+
+def list_assets_for_user(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    owner_type: str | None = None,
+    owner_id: str | None = None,
+    campaign_id: str | None = None,
+    path_prefix: str | None = None,
+    include_deleted: bool = False,
+    is_admin: bool = False,
+) -> list[FileAsset]:
+    query = session.query(FileAsset).filter(FileAsset.tenant_id == tenant_id)
+    if not include_deleted:
+        query = query.filter(FileAsset.deleted_at.is_(None))
+    if owner_type:
+        query = query.filter(FileAsset.owner_type == owner_type)
+        if owner_type == "user" and owner_id:
+            query = query.filter(FileAsset.owner_user_id == owner_id)
+        if owner_type == "group" and owner_id:
+            query = query.filter(FileAsset.owner_group_id == owner_id)
+    if campaign_id:
+        query = query.join(FileShare, FileShare.file_asset_id == FileAsset.id).filter(
+            FileShare.tenant_id == tenant_id,
+            FileShare.target_type == "campaign",
+            FileShare.target_id == campaign_id,
+            FileShare.revoked_at.is_(None),
+        )
+    elif not is_admin and not owner_type:
+        group_ids = user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
+        query = query.outerjoin(FileShare, FileShare.file_asset_id == FileAsset.id).filter(
+            or_(
+                (FileAsset.owner_type == "user") & (FileAsset.owner_user_id == user_id),
+                (FileAsset.owner_type == "group") & (FileAsset.owner_group_id.in_(group_ids)),
+                (FileShare.revoked_at.is_(None)) & (FileShare.target_type == "user") & (FileShare.target_id == user_id),
+                (FileShare.revoked_at.is_(None)) & (FileShare.target_type == "group") & (FileShare.target_id.in_(group_ids)),
+                (FileShare.revoked_at.is_(None)) & (FileShare.target_type == "tenant") & (FileShare.target_id == tenant_id),
+            )
+        )
+    if path_prefix:
+        prefix = normalize_folder(path_prefix)
+        if prefix:
+            query = query.filter(FileAsset.display_path.like(f"{prefix}/%"))
+    return query.order_by(FileAsset.display_path.asc(), FileAsset.updated_at.desc()).all()
+
+
+def current_version_and_blob(session: Session, asset: FileAsset) -> tuple[FileVersion, FileBlob]:
+    if not asset.current_version_id:
+        raise FileStorageError("File has no current version")
+    version = session.get(FileVersion, asset.current_version_id)
+    if not version:
+        raise FileStorageError("File version not found")
+    blob = session.get(FileBlob, version.blob_id)
+    if not blob:
+        raise FileStorageError("File blob not found")
+    return version, blob
+
+
+def read_asset_bytes(session: Session, asset: FileAsset) -> tuple[bytes, FileVersion, FileBlob]:
+    version, blob = current_version_and_blob(session, asset)
+    backend = get_storage_backend()
+    return backend.get_bytes(blob.storage_key), version, blob
+
+
+def share_file(
+    session: Session,
+    *,
+    tenant_id: str,
+    asset: FileAsset,
+    target_type: str,
+    target_id: str,
+    permission: str,
+    user_id: str,
+) -> FileShare:
+    target_type = target_type.lower().strip()
+    permission = permission.lower().strip()
+    if asset.tenant_id != tenant_id:
+        raise FileStorageError("File not found")
+    if target_type not in {"user", "group", "campaign", "tenant"}:
+        raise FileStorageError("Unsupported share target")
+    if permission not in {"read", "write", "manage"}:
+        raise FileStorageError("Unsupported file permission")
+    if target_type == "user":
+        target_user = session.get(User, target_id)
+        if not target_user or target_user.tenant_id != tenant_id or not target_user.is_active:
+            raise FileStorageError("User not found")
+    if target_type == "group":
+        group = session.get(Group, target_id)
+        if not group or group.tenant_id != tenant_id or not group.is_active:
+            raise FileStorageError("Group not found")
+    if target_type == "tenant":
+        tenant = session.get(Tenant, target_id)
+        if target_id != tenant_id or not tenant or not tenant.is_active:
+            raise FileStorageError("Tenant not found")
+    if target_type == "campaign":
+        campaign = session.get(Campaign, target_id)
+        if not campaign or campaign.tenant_id != tenant_id:
+            raise FileStorageError("Campaign not found")
+    existing = (
+        session.query(FileShare)
+        .filter(
+            FileShare.tenant_id == tenant_id,
+            FileShare.file_asset_id == asset.id,
+            FileShare.target_type == target_type,
+            FileShare.target_id == target_id,
+            FileShare.revoked_at.is_(None),
+        )
+        .one_or_none()
+    )
+    if existing:
+        existing.permission = permission
+        session.add(existing)
+        return existing
+    share = FileShare(
+        tenant_id=tenant_id,
+        file_asset_id=asset.id,
+        target_type=target_type,
+        target_id=target_id,
+        permission=permission,
+        created_by_user_id=user_id,
+    )
+    session.add(share)
+    return share
+
+
+def soft_delete_assets(session: Session, assets: Iterable[FileAsset]) -> int:
+    count = 0
+    now = utcnow()
+    for asset in assets:
+        if asset.deleted_at is None:
+            asset.deleted_at = now
+            session.add(asset)
+            count += 1
+    return count
+
+
+def asset_is_audit_relevant(session: Session, asset: FileAsset) -> bool:
+    return (
+        session.query(CampaignAttachmentUse)
+        .filter(CampaignAttachmentUse.file_asset_id == asset.id, CampaignAttachmentUse.use_stage == "sent")
+        .first()
+        is not None
+    )
+
+
+def _asset_owner_id(asset: FileAsset) -> str:
+    if asset.owner_type == "user" and asset.owner_user_id:
+        return asset.owner_user_id
+    if asset.owner_type == "group" and asset.owner_group_id:
+        return asset.owner_group_id
+    raise FileStorageError("File has no valid owner")
+
+
+def _active_asset_exists(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_asset_id: str | None = None) -> bool:
+    return _active_asset_at_path(
+        session,
+        tenant_id=tenant_id,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        path=path,
+        exclude_asset_id=exclude_asset_id,
+    ) is not None
+
+
+def _active_asset_at_path(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_asset_id: str | None = None) -> FileAsset | None:
+    query = _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter(
+        FileAsset.deleted_at.is_(None),
+        FileAsset.display_path == normalize_logical_path(path),
+    )
+    if exclude_asset_id:
+        query = query.filter(FileAsset.id != exclude_asset_id)
+    return query.first()
+
+
+def _soft_delete_conflicting_asset(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_asset_id: str | None = None) -> None:
+    asset = _active_asset_at_path(
+        session,
+        tenant_id=tenant_id,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        path=path,
+        exclude_asset_id=exclude_asset_id,
+    )
+    if asset is not None:
+        asset.deleted_at = utcnow()
+        session.add(asset)
+
+
+def _split_logical_path(path: str) -> tuple[str, str, str, str]:
+    normalized = normalize_logical_path(path)
+    logical = PurePosixPath(normalized)
+    folder = "" if str(logical.parent) == "." else str(logical.parent)
+    filename = logical.name
+    suffixes = "".join(PurePosixPath(filename).suffixes)
+    stem = filename[: -len(suffixes)] if suffixes else filename
+    return folder, filename, stem, suffixes
+
+
+def _candidate_renamed_path(path: str, counter: int) -> str:
+    folder, _filename, stem, suffixes = _split_logical_path(path)
+    suffix = " copy" if counter == 1 else f" copy {counter}"
+    next_name = f"{stem}{suffix}{suffixes}"
+    return normalize_logical_path(f"{folder}/{next_name}" if folder else next_name)
+
+
+def _next_available_logical_path(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    desired_path: str,
+    reserved_paths: set[str] | None = None,
+    exclude_asset_id: str | None = None,
+) -> str:
+    reserved = reserved_paths or set()
+    desired = normalize_logical_path(desired_path)
+    if desired not in reserved and not _active_asset_exists(
+        session,
+        tenant_id=tenant_id,
+        owner_type=owner_type,
+        owner_id=owner_id,
+        path=desired,
+        exclude_asset_id=exclude_asset_id,
+    ):
+        return desired
+    counter = 1
+    while True:
+        candidate = _candidate_renamed_path(desired, counter)
+        if candidate not in reserved and not _active_asset_exists(
+            session,
+            tenant_id=tenant_id,
+            owner_type=owner_type,
+            owner_id=owner_id,
+            path=candidate,
+            exclude_asset_id=exclude_asset_id,
+        ):
+            return candidate
+        counter += 1
+
+
+def _resolution_by_path(conflict_resolutions: Iterable[FileConflictResolution] | None) -> dict[str, FileConflictResolution]:
+    result: dict[str, FileConflictResolution] = {}
+    for item in conflict_resolutions or []:
+        target_path = normalize_logical_path(item.target_path)
+        action = item.action.lower().strip()
+        if action not in {"overwrite", "rename", "skip"}:
+            raise FileStorageError("Unsupported conflict resolution")
+        result[target_path] = FileConflictResolution(target_path=target_path, action=action, new_path=item.new_path)
+    return result
+
+
+def _normalize_conflict_strategy(strategy: str | None) -> str:
+    normalized = (strategy or "reject").lower().strip()
+    if normalized not in {"reject", "overwrite", "rename"}:
+        raise FileStorageError("Unsupported conflict strategy")
+    return normalized
+
+
+def _copy_asset_to_path(
+    session: Session,
+    asset: FileAsset,
+    *,
+    tenant_id: str,
+    target_owner_type: str,
+    target_owner_id: str,
+    target_path: str,
+    user_id: str,
+) -> FileAsset:
+    version, blob = current_version_and_blob(session, asset)
+    blob.ref_count += 1
+    session.add(blob)
+    normalized_path = normalize_logical_path(target_path)
+    copied = FileAsset(
+        tenant_id=tenant_id,
+        owner_type=target_owner_type,
+        owner_user_id=target_owner_id if target_owner_type == "user" else None,
+        owner_group_id=target_owner_id if target_owner_type == "group" else None,
+        display_path=normalized_path,
+        filename=filename_from_path(normalized_path),
+        description=asset.description,
+        created_by_user_id=user_id,
+        metadata_=dict(asset.metadata_ or {}),
+    )
+    session.add(copied)
+    session.flush()
+    copied_version = FileVersion(
+        tenant_id=tenant_id,
+        file_asset_id=copied.id,
+        blob_id=blob.id,
+        version_number=1,
+        filename_at_upload=version.filename_at_upload,
+        display_path_at_upload=normalized_path,
+        content_type=version.content_type,
+        size_bytes=blob.size_bytes,
+        checksum_sha256=blob.checksum_sha256,
+        created_by_user_id=user_id,
+    )
+    session.add(copied_version)
+    session.flush()
+    copied.current_version_id = copied_version.id
+    session.add(copied)
+    return copied
+
+
+def rename_asset(asset: FileAsset, *, new_path: str) -> None:
+    normalized = normalize_logical_path(new_path)
+    asset.display_path = normalized
+    asset.filename = filename_from_path(normalized)

src/govoplan_files/backend/storage/folders.py

@@ -0,0 +1,183 @@
+from __future__ import annotations
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_files.backend.db.models import FileAsset, FileFolder
+from govoplan_files.backend.storage.access import ensure_owner_access
+from govoplan_files.backend.storage.common import FileStorageError, utcnow
+from govoplan_files.backend.storage.files import _asset_query_for_owner
+from govoplan_files.backend.storage.paths import normalize_folder
+
+
+def _owner_filter(query, owner_type: str, owner_id: str):
+    if owner_type == "user":
+        return query.filter(FileFolder.owner_user_id == owner_id)
+    if owner_type == "group":
+        return query.filter(FileFolder.owner_group_id == owner_id)
+    raise FileStorageError("Unsupported owner type")
+
+
+def create_folder(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    user_id: str,
+    path: str,
+    is_admin: bool = False,
+) -> FileFolder:
+    owner_type = owner_type.lower().strip()
+    ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)
+    normalized = normalize_folder(path)
+    if not normalized:
+        raise FileStorageError("Folder path is required")
+    query = session.query(FileFolder).filter(FileFolder.tenant_id == tenant_id, FileFolder.owner_type == owner_type, FileFolder.path == normalized)
+    query = _owner_filter(query, owner_type, owner_id)
+    active_existing = query.filter(FileFolder.deleted_at.is_(None)).first()
+    if active_existing is not None:
+        raise FileStorageError(f"Folder already exists: {normalized}")
+    deleted_existing = query.filter(FileFolder.deleted_at.is_not(None)).first()
+    if deleted_existing is not None:
+        deleted_existing.deleted_at = None
+        session.add(deleted_existing)
+        session.flush()
+        return deleted_existing
+    folder = FileFolder(
+        tenant_id=tenant_id,
+        owner_type=owner_type,
+        owner_user_id=owner_id if owner_type == "user" else None,
+        owner_group_id=owner_id if owner_type == "group" else None,
+        path=normalized,
+        created_by_user_id=user_id,
+        metadata_={},
+    )
+    session.add(folder)
+    session.flush()
+    return folder
+
+
+def list_folders_for_user(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    owner_type: str,
+    owner_id: str,
+    include_deleted: bool = False,
+    is_admin: bool = False,
+) -> list[FileFolder]:
+    owner_type = owner_type.lower().strip()
+    ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)
+    query = session.query(FileFolder).filter(FileFolder.tenant_id == tenant_id, FileFolder.owner_type == owner_type)
+    query = _owner_filter(query, owner_type, owner_id)
+    if not include_deleted:
+        query = query.filter(FileFolder.deleted_at.is_(None))
+    return query.order_by(FileFolder.path.asc()).all()
+
+
+def soft_delete_folder(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    user_id: str,
+    path: str,
+    recursive: bool = True,
+    is_admin: bool = False,
+) -> tuple[int, int]:
+    owner_type = owner_type.lower().strip()
+    ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id, user_id=user_id, is_admin=is_admin)
+    normalized = normalize_folder(path)
+    if not normalized:
+        raise FileStorageError("Folder path is required")
+    prefix = f"{normalized}/"
+    now = utcnow()
+
+    folder_query = session.query(FileFolder).filter(FileFolder.tenant_id == tenant_id, FileFolder.owner_type == owner_type, FileFolder.deleted_at.is_(None))
+    folder_query = _owner_filter(folder_query, owner_type, owner_id)
+    if recursive:
+        folder_query = folder_query.filter(or_(FileFolder.path == normalized, FileFolder.path.like(f"{prefix}%")))
+    else:
+        child_exists = folder_query.filter(FileFolder.path.like(f"{prefix}%")).first() is not None
+        file_exists = _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter(FileAsset.display_path.like(f"{prefix}%")).first() is not None
+        if child_exists or file_exists:
+            raise FileStorageError("Folder is not empty")
+        folder_query = folder_query.filter(FileFolder.path == normalized)
+
+    folders = folder_query.all()
+    for folder in folders:
+        folder.deleted_at = now
+        session.add(folder)
+
+    file_query = _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter(FileAsset.deleted_at.is_(None), FileAsset.display_path.like(f"{prefix}%"))
+    assets = file_query.all() if recursive else []
+    for asset in assets:
+        asset.deleted_at = now
+        session.add(asset)
+
+    return len(folders), len(assets)
+
+
+def _active_folder_exists(session: Session, *, tenant_id: str, owner_type: str, owner_id: str, path: str, exclude_paths: set[str] | None = None) -> bool:
+    normalized = normalize_folder(path)
+    if not normalized:
+        return False
+    query = session.query(FileFolder).filter(
+        FileFolder.tenant_id == tenant_id,
+        FileFolder.owner_type == owner_type,
+        FileFolder.deleted_at.is_(None),
+        FileFolder.path == normalized,
+    )
+    query = _owner_filter(query, owner_type, owner_id)
+    if exclude_paths:
+        query = query.filter(FileFolder.path.notin_(exclude_paths))
+    return query.first() is not None
+
+
+def _folder_query_for_owner(session: Session, *, tenant_id: str, owner_type: str, owner_id: str):
+    query = session.query(FileFolder).filter(FileFolder.tenant_id == tenant_id, FileFolder.owner_type == owner_type)
+    return _owner_filter(query, owner_type, owner_id)
+
+
+def _ensure_target_folder_hierarchy(
+    session: Session,
+    *,
+    tenant_id: str,
+    owner_type: str,
+    owner_id: str,
+    user_id: str,
+    path: str,
+) -> None:
+    parts = normalize_folder(path).split("/") if normalize_folder(path) else []
+    for index in range(1, len(parts) + 1):
+        partial = "/".join(parts[:index])
+        query = _folder_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter(FileFolder.path == partial)
+        matches = query.order_by(FileFolder.deleted_at.asc(), FileFolder.created_at.asc(), FileFolder.id.asc()).all()
+        existing = next((folder for folder in matches if folder.deleted_at is None), matches[0] if matches else None)
+        if existing:
+            # Historical databases may contain a soft-deleted row alongside an
+            # active row, or even multiple active rows from concurrent creates.
+            # Keep one canonical row and retire the rest instead of letting
+            # one_or_none() abort a move/copy operation.
+            if existing.deleted_at is not None:
+                existing.deleted_at = None
+            now = utcnow()
+            for duplicate in matches:
+                if duplicate.id != existing.id and duplicate.deleted_at is None:
+                    duplicate.deleted_at = now
+                    session.add(duplicate)
+            session.add(existing)
+            continue
+        folder = FileFolder(
+            tenant_id=tenant_id,
+            owner_type=owner_type,
+            owner_user_id=owner_id if owner_type == "user" else None,
+            owner_group_id=owner_id if owner_type == "group" else None,
+            path=partial,
+            created_by_user_id=user_id,
+            metadata_={},
+        )
+        session.add(folder)

src/govoplan_files/backend/storage/paths.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import re
+from pathlib import PurePosixPath
+from uuid import uuid4
+
+_SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9_.@ -]+")
+
+
+class UnsafeFilePathError(ValueError):
+    pass
+
+
+def normalize_logical_path(path: str | None, *, fallback_filename: str | None = None) -> str:
+    """Return a safe tenant-relative logical path using POSIX separators.
+
+    The logical path is metadata, not a filesystem path. It never starts with a
+    slash and cannot contain path traversal components. It is used for browsing,
+    wildcard matching and attachment rules.
+    """
+
+    raw = (path or "").replace("\\", "/").strip()
+    if not raw and fallback_filename:
+        raw = fallback_filename
+    if not raw:
+        raise UnsafeFilePathError("File path is empty")
+    if raw.startswith("/"):
+        raw = raw.lstrip("/")
+    parts: list[str] = []
+    for part in raw.split("/"):
+        clean = part.strip()
+        if not clean or clean == ".":
+            continue
+        if clean == "..":
+            raise UnsafeFilePathError("Path traversal is not allowed")
+        parts.append(clean)
+    if not parts:
+        raise UnsafeFilePathError("File path is empty")
+    return "/".join(parts)
+
+
+def normalize_folder(path: str | None) -> str:
+    raw = (path or "").replace("\\", "/").strip().strip("/")
+    if not raw:
+        return ""
+    normalized = normalize_logical_path(raw)
+    return "" if normalized == "." else normalized
+
+
+def filename_from_path(path: str) -> str:
+    name = PurePosixPath(path).name
+    if not name or name in {".", ".."}:
+        raise UnsafeFilePathError("Invalid filename")
+    return name
+
+
+def join_folder_filename(folder: str | None, filename: str) -> str:
+    safe_name = sanitize_filename(filename)
+    safe_folder = normalize_folder(folder)
+    return f"{safe_folder}/{safe_name}" if safe_folder else safe_name
+
+
+def sanitize_filename(filename: str | None) -> str:
+    raw = (filename or "file").replace("\\", "/").split("/")[-1].strip()
+    raw = raw.strip(".") or "file"
+    safe = _SAFE_NAME_RE.sub("_", raw)
+    safe = re.sub(r"\s+", " ", safe).strip()
+    return safe or f"file-{uuid4().hex}"
+
+
+def safe_storage_component(value: str | None, fallback: str = "file") -> str:
+    safe = sanitize_filename(value or fallback)
+    return safe.replace(" ", "_")[:180]

src/govoplan_files/backend/storage/search.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import re
+from typing import Iterable
+
+from govoplan_files.backend.db.models import FileAsset
+from govoplan_files.backend.storage.common import ResolvedPattern
+from govoplan_files.backend.storage.paths import normalize_folder, normalize_logical_path
+
+
+def _normalize_pattern(pattern: str) -> str:
+    if pattern.strip() in {"", "*"}:
+        return "*"
+    return normalize_logical_path(pattern, fallback_filename="*")
+
+
+def _logical_glob_regex(pattern: str, *, case_sensitive: bool = False) -> re.Pattern[str]:
+    """Compile Multi Seal Mail logical globs.
+
+    `*` and `?` stay within one folder segment. `**` crosses folder
+    boundaries, and `**/` also matches the current folder so `**/*.pdf`
+    returns direct and nested PDF files.
+    """
+
+    pattern = _normalize_pattern(pattern)
+    pieces = ["^"]
+    index = 0
+    while index < len(pattern):
+        char = pattern[index]
+        if char == "*":
+            if index + 1 < len(pattern) and pattern[index + 1] == "*":
+                index += 2
+                if index < len(pattern) and pattern[index] == "/":
+                    pieces.append("(?:.*/)?")
+                    index += 1
+                else:
+                    pieces.append(".*")
+                continue
+            pieces.append("[^/]*")
+        elif char == "?":
+            pieces.append("[^/]")
+        else:
+            pieces.append(re.escape(char))
+        index += 1
+    pieces.append("$")
+    flags = 0 if case_sensitive else re.IGNORECASE
+    return re.compile("".join(pieces), flags)
+
+
+def _relative_display_path(asset: FileAsset, base_path: str | None) -> str:
+    path = normalize_logical_path(asset.display_path)
+    base = normalize_folder(base_path)
+    if not base:
+        return path
+    prefix = f"{base}/"
+    if path.startswith(prefix):
+        return path[len(prefix) :]
+    return path
+
+
+def match_assets(assets: Iterable[FileAsset], pattern: str, *, base_path: str | None = None, case_sensitive: bool = False) -> list[FileAsset]:
+    regex = _logical_glob_regex(pattern, case_sensitive=case_sensitive)
+    normalized_pattern = _normalize_pattern(pattern)
+    has_path_context = base_path is not None or "/" in normalized_pattern or "**" in normalized_pattern
+    matches: list[FileAsset] = []
+    for asset in assets:
+        candidates = [_relative_display_path(asset, base_path)] if has_path_context else [asset.display_path, asset.filename]
+        if any(regex.match(candidate) for candidate in candidates):
+            matches.append(asset)
+    return matches
+
+
+def resolve_patterns(assets: list[FileAsset], patterns: list[str], *, base_path: str | None = None, case_sensitive: bool = False) -> tuple[list[ResolvedPattern], list[FileAsset]]:
+    resolved = [ResolvedPattern(pattern=pattern, matches=match_assets(assets, pattern, base_path=base_path, case_sensitive=case_sensitive)) for pattern in patterns]
+    matched_ids = {asset.id for item in resolved for asset in item.matches}
+    unmatched = [asset for asset in assets if asset.id not in matched_ids]
+    return resolved, unmatched

src/govoplan_files/backend/storage/services.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+# Compatibility facade for existing storage imports. New storage code should
+# prefer importing from the focused modules directly.
+
+from govoplan_files.backend.storage.access import ensure_group_access, ensure_owner_access, user_group_ids
+from govoplan_files.backend.storage.archives import create_zip_file, extract_zip_upload
+from govoplan_files.backend.storage.campaign_usage import mark_job_attachment_uses_sent, record_campaign_attachment_uses_for_job
+from govoplan_files.backend.storage.common import (
+    FileConflictResolution,
+    FileStorageError,
+    RenamePlanItem,
+    ResolvedPattern,
+    UploadedStoredFile,
+    utcnow,
+)
+from govoplan_files.backend.storage.files import (
+    _active_asset_at_path,
+    _active_asset_exists,
+    _asset_owner_id,
+    _asset_query_for_owner,
+    _candidate_renamed_path,
+    _copy_asset_to_path,
+    _get_or_create_blob,
+    _next_available_logical_path,
+    _normalize_conflict_strategy,
+    _resolution_by_path,
+    _soft_delete_conflicting_asset,
+    _split_logical_path,
+    _storage_backend_name,
+    _storage_bucket_name,
+    _storage_key,
+    asset_is_audit_relevant,
+    create_file_asset,
+    current_version_and_blob,
+    get_asset_for_user,
+    list_assets_for_user,
+    read_asset_bytes,
+    rename_asset,
+    share_file,
+    soft_delete_assets,
+)
+from govoplan_files.backend.storage.folders import (
+    _active_folder_exists,
+    _ensure_target_folder_hierarchy,
+    _folder_query_for_owner,
+    _owner_filter,
+    create_folder,
+    list_folders_for_user,
+    soft_delete_folder,
+)
+from govoplan_files.backend.storage.search import match_assets, resolve_patterns
+from govoplan_files.backend.storage.transfers import build_rename_preview, rename_selection, transfer_selection
+
+__all__ = [
+    "FileConflictResolution",
+    "FileStorageError",
+    "RenamePlanItem",
+    "ResolvedPattern",
+    "UploadedStoredFile",
+    "asset_is_audit_relevant",
+    "build_rename_preview",
+    "create_file_asset",
+    "create_folder",
+    "create_zip_file",
+    "current_version_and_blob",
+    "ensure_group_access",
+    "ensure_owner_access",
+    "extract_zip_upload",
+    "get_asset_for_user",
+    "list_assets_for_user",
+    "list_folders_for_user",
+    "mark_job_attachment_uses_sent",
+    "match_assets",
+    "read_asset_bytes",
+    "record_campaign_attachment_uses_for_job",
+    "rename_asset",
+    "rename_selection",
+    "resolve_patterns",
+    "share_file",
+    "soft_delete_assets",
+    "soft_delete_folder",
+    "transfer_selection",
+    "user_group_ids",
+    "utcnow",
+]

src/govoplan_files/backend/storage/transfers.py

@@ -0,0 +1,446 @@
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+from typing import Iterable
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_files.backend.db.models import FileAsset, FileFolder
+from govoplan_files.backend.storage.access import ensure_owner_access
+from govoplan_files.backend.storage.common import FileConflictResolution, FileStorageError, RenamePlanItem, utcnow
+from govoplan_files.backend.storage.files import (
+    _active_asset_exists,
+    _asset_owner_id,
+    _asset_query_for_owner,
+    _candidate_renamed_path,
+    _copy_asset_to_path,
+    _next_available_logical_path,
+    _normalize_conflict_strategy,
+    _resolution_by_path,
+    _soft_delete_conflicting_asset,
+    get_asset_for_user,
+    rename_asset,
+)
+from govoplan_files.backend.storage.folders import _active_folder_exists, _ensure_target_folder_hierarchy, _folder_query_for_owner
+from govoplan_files.backend.storage.paths import filename_from_path, join_folder_filename, normalize_folder, normalize_logical_path
+
+
+def transfer_selection(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    operation: str,
+    file_ids: list[str],
+    folder_paths: list[str],
+    source_owner_type: str,
+    source_owner_id: str,
+    target_owner_type: str,
+    target_owner_id: str,
+    target_folder: str,
+    conflict_strategy: str = "reject",
+    conflict_resolutions: Iterable[FileConflictResolution] | None = None,
+    is_admin: bool = False,
+) -> tuple[int, int]:
+    """Move or copy files/folders between user/group file spaces.
+
+    Folder transfers preserve the selected folder's basename below the target
+    folder. File transfers place files directly in the target folder. Existing
+    active target paths are handled by the requested conflict strategy. Copies
+    create new file assets/versions that reference the existing immutable blob.
+    """
+
+    operation = operation.lower().strip()
+    if operation not in {"move", "copy"}:
+        raise FileStorageError("Unsupported transfer operation")
+    source_owner_type = source_owner_type.lower().strip()
+    target_owner_type = target_owner_type.lower().strip()
+    source_folder_paths = [normalize_folder(path) for path in folder_paths if normalize_folder(path)]
+    target_folder = normalize_folder(target_folder)
+    conflict_strategy = _normalize_conflict_strategy(conflict_strategy)
+    conflict_resolution_map = _resolution_by_path(conflict_resolutions)
+
+    ensure_owner_access(session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id, user_id=user_id, is_admin=is_admin)
+    ensure_owner_access(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, is_admin=is_admin)
+
+    if operation == "move" and source_owner_type == target_owner_type and source_owner_id == target_owner_id:
+        for folder_path in source_folder_paths:
+            if target_folder == folder_path or target_folder.startswith(f"{folder_path}/"):
+                raise FileStorageError("Cannot move a folder into itself or one of its child folders")
+
+    assets_by_id: dict[str, FileAsset] = {}
+    for file_id in file_ids:
+        asset = get_asset_for_user(
+            session,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            asset_id=file_id,
+            require_write=operation == "move",
+            is_admin=is_admin,
+        )
+        assets_by_id[asset.id] = asset
+
+    folder_asset_targets: dict[str, str] = {}
+    folder_target_paths: dict[str, str] = {}
+    for folder_path in source_folder_paths:
+        folder_basename = PurePosixPath(folder_path).name
+        target_prefix = normalize_folder(f"{target_folder}/{folder_basename}" if target_folder else folder_basename)
+        folder_target_paths[folder_path] = target_prefix
+        if operation == "copy" or not (source_owner_type == target_owner_type and source_owner_id == target_owner_id and target_prefix == folder_path):
+            if _active_folder_exists(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, path=target_prefix):
+                resolution = conflict_resolution_map.get(target_prefix)
+                action = resolution.action if resolution else conflict_strategy
+                if action == "skip":
+                    folder_target_paths.pop(folder_path, None)
+                    continue
+                if action == "rename":
+                    # Rename the selected folder root while preserving its contents below it.
+                    counter = 1
+                    candidate = target_prefix
+                    while _active_folder_exists(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, path=candidate):
+                        candidate = _candidate_renamed_path(target_prefix, counter)
+                        counter += 1
+                    target_prefix = normalize_folder(candidate)
+                    folder_target_paths[folder_path] = target_prefix
+                elif action == "reject":
+                    raise FileStorageError(f"Target folder already exists: {target_prefix}")
+                # overwrite on folders means merge into the existing folder; conflicting files are still handled below.
+        source_assets = _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id).filter(
+            FileAsset.deleted_at.is_(None),
+            FileAsset.display_path.like(f"{folder_path}/%"),
+        ).all()
+        for asset in source_assets:
+            relative = asset.display_path[len(folder_path) + 1 :]
+            folder_asset_targets[asset.id] = normalize_logical_path(f"{target_prefix}/{relative}")
+            assets_by_id[asset.id] = asset
+
+    direct_file_targets: dict[str, str] = {}
+    for file_id, asset in assets_by_id.items():
+        if file_id in folder_asset_targets:
+            continue
+        direct_file_targets[file_id] = normalize_logical_path(join_folder_filename(target_folder, filename_from_path(asset.display_path)))
+
+    all_targets = {**folder_asset_targets, **direct_file_targets}
+    if not all_targets and not source_folder_paths:
+        return 0, 0
+
+    resolved_targets: dict[str, str] = {}
+    skipped_asset_ids: set[str] = set()
+    reserved_targets: set[str] = set()
+    for asset_id, target_path in all_targets.items():
+        source_asset = assets_by_id[asset_id]
+        same_owner = (source_asset.owner_type == target_owner_type and _asset_owner_id(source_asset) == target_owner_id)
+        exclude = source_asset.id if operation == "move" and same_owner else None
+        normalized_target = normalize_logical_path(target_path)
+        resolution = conflict_resolution_map.get(normalized_target)
+        action = resolution.action if resolution else conflict_strategy
+        if resolution and resolution.new_path and action == "rename":
+            normalized_target = normalize_logical_path(resolution.new_path)
+        conflict = _active_asset_exists(
+            session,
+            tenant_id=tenant_id,
+            owner_type=target_owner_type,
+            owner_id=target_owner_id,
+            path=normalized_target,
+            exclude_asset_id=exclude,
+        ) or normalized_target in reserved_targets
+        if conflict:
+            if action == "reject":
+                raise FileStorageError(f"Target file already exists: {normalized_target}")
+            if action == "skip":
+                skipped_asset_ids.add(asset_id)
+                continue
+            if action == "overwrite":
+                _soft_delete_conflicting_asset(
+                    session,
+                    tenant_id=tenant_id,
+                    owner_type=target_owner_type,
+                    owner_id=target_owner_id,
+                    path=normalized_target,
+                    exclude_asset_id=exclude,
+                )
+            elif action == "rename":
+                normalized_target = _next_available_logical_path(
+                    session,
+                    tenant_id=tenant_id,
+                    owner_type=target_owner_type,
+                    owner_id=target_owner_id,
+                    desired_path=normalized_target,
+                    reserved_paths=reserved_targets,
+                    exclude_asset_id=exclude,
+                )
+        elif action == "rename":
+            normalized_target = _next_available_logical_path(
+                session,
+                tenant_id=tenant_id,
+                owner_type=target_owner_type,
+                owner_id=target_owner_id,
+                desired_path=normalized_target,
+                reserved_paths=reserved_targets,
+                exclude_asset_id=exclude,
+            )
+        reserved_targets.add(normalized_target)
+        resolved_targets[asset_id] = normalized_target
+    all_targets = resolved_targets
+
+    copied_or_moved_files = 0
+    copied_or_moved_folders = 0
+
+    # Create target folder hierarchy first, including selected folder roots and
+    # any nested folders/files parents.
+    same_owner_transfer = source_owner_type == target_owner_type and source_owner_id == target_owner_id
+    for target_path in folder_target_paths.values():
+        path_to_create = target_path
+        if operation == "move" and same_owner_transfer:
+            parent = str(PurePosixPath(target_path).parent)
+            path_to_create = "" if parent == "." else parent
+        if path_to_create:
+            _ensure_target_folder_hierarchy(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, path=path_to_create)
+        copied_or_moved_folders += 1
+    for target_path in all_targets.values():
+        parent = str(PurePosixPath(target_path).parent)
+        if parent and parent != ".":
+            _ensure_target_folder_hierarchy(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, path=parent)
+
+    if operation == "copy":
+        for asset_id, target_path in all_targets.items():
+            _copy_asset_to_path(session, assets_by_id[asset_id], tenant_id=tenant_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, target_path=target_path, user_id=user_id)
+            copied_or_moved_files += 1
+        return copied_or_moved_files, copied_or_moved_folders
+
+    now = utcnow()
+    for asset_id, target_path in all_targets.items():
+        asset = assets_by_id[asset_id]
+        same_owner = asset.owner_type == target_owner_type and _asset_owner_id(asset) == target_owner_id
+        if same_owner:
+            if normalize_logical_path(asset.display_path) == target_path:
+                continue
+            rename_asset(asset, new_path=target_path)
+            session.add(asset)
+        else:
+            _copy_asset_to_path(session, asset, tenant_id=tenant_id, target_owner_type=target_owner_type, target_owner_id=target_owner_id, target_path=target_path, user_id=user_id)
+            asset.deleted_at = now
+            session.add(asset)
+        copied_or_moved_files += 1
+
+    # Move/copy persisted folder records after files. For cross-owner moves, create
+    # target records and soft-delete source records. For same-owner moves, rename
+    # them in place.
+    for source_path, target_prefix in folder_target_paths.items():
+        folder_rows = _folder_query_for_owner(session, tenant_id=tenant_id, owner_type=source_owner_type, owner_id=source_owner_id).filter(
+            FileFolder.deleted_at.is_(None),
+            or_(FileFolder.path == source_path, FileFolder.path.like(f"{source_path}/%")),
+        ).all()
+        for folder in folder_rows:
+            relative = "" if folder.path == source_path else folder.path[len(source_path) + 1 :]
+            new_path = normalize_folder(f"{target_prefix}/{relative}" if relative else target_prefix)
+            if operation == "move" and source_owner_type == target_owner_type and source_owner_id == target_owner_id:
+                folder.path = new_path
+                session.add(folder)
+            else:
+                _ensure_target_folder_hierarchy(session, tenant_id=tenant_id, owner_type=target_owner_type, owner_id=target_owner_id, user_id=user_id, path=new_path)
+                if operation == "move":
+                    folder.deleted_at = now
+                    session.add(folder)
+    return copied_or_moved_files, copied_or_moved_folders
+
+
+def _rename_basename(name: str, *, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str:
+    if mode == "direct":
+        cleaned = normalize_logical_path(new_name or "", fallback_filename="item")
+        if "/" in cleaned:
+            raise FileStorageError("Rename expects a name, not a path")
+        return cleaned
+    stem_path = PurePosixPath(name)
+    suffixes = "".join(stem_path.suffixes)
+    stem = name[: -len(suffixes)] if suffixes else name
+    if mode == "prefix":
+        return prefix + name
+    if mode == "suffix":
+        return f"{stem}{suffix}{suffixes}"
+    if mode == "replace":
+        return name.replace(find or "", replacement) if find else name
+    raise FileStorageError("Unsupported rename mode")
+
+
+def _rename_path(path: str, *, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str:
+    normalized = normalize_logical_path(path)
+    logical = PurePosixPath(normalized)
+    folder = "" if str(logical.parent) == "." else str(logical.parent)
+    next_name = _rename_basename(logical.name, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+    return normalize_logical_path(f"{folder}/{next_name}" if folder else next_name)
+
+
+def _rename_relative_recursive(relative_path: str, *, mode: str, new_name: str | None = None, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str:
+    parts = normalize_logical_path(relative_path).split("/")
+    return normalize_logical_path(
+        "/".join(
+            _rename_basename(part, mode=mode, new_name=new_name if len(parts) == 1 else None, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+            for part in parts
+            if part
+        )
+    )
+
+
+def build_rename_preview(asset: FileAsset, *, mode: str, find: str | None = None, replacement: str = "", prefix: str = "", suffix: str = "") -> str:
+    return _rename_path(asset.display_path, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+
+
+def _collapse_folder_roots(folder_paths: list[str]) -> list[str]:
+    roots = sorted({normalize_folder(path) for path in folder_paths if normalize_folder(path)}, key=lambda item: (item.count("/"), item))
+    collapsed: list[str] = []
+    for path in roots:
+        if any(path == root or path.startswith(f"{root}/") for root in collapsed):
+            continue
+        collapsed.append(path)
+    return collapsed
+
+
+def _path_under_root(path: str, root: str) -> bool:
+    normalized = normalize_logical_path(path)
+    return normalized == root or normalized.startswith(f"{root}/")
+
+
+def _folder_new_path_for_root(path: str, root: str, new_root: str, *, recursive: bool, mode: str, find: str | None, replacement: str, prefix: str, suffix: str) -> str:
+    if path == root:
+        return normalize_folder(new_root)
+    relative = path[len(root) + 1 :]
+    if recursive:
+        relative = _rename_relative_recursive(relative, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+    return normalize_folder(f"{new_root}/{relative}")
+
+
+def _file_new_path_for_root(path: str, root: str, new_root: str, *, recursive: bool, mode: str, find: str | None, replacement: str, prefix: str, suffix: str) -> str:
+    relative = path[len(root) + 1 :]
+    if recursive:
+        relative = _rename_relative_recursive(relative, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+    return normalize_logical_path(f"{new_root}/{relative}")
+
+
+def rename_selection(
+    session: Session,
+    *,
+    tenant_id: str,
+    user_id: str,
+    file_ids: list[str],
+    folder_paths: list[str],
+    owner_type: str | None,
+    owner_id: str | None,
+    mode: str,
+    new_name: str | None = None,
+    find: str | None = None,
+    replacement: str = "",
+    prefix: str = "",
+    suffix: str = "",
+    recursive: bool = False,
+    dry_run: bool = True,
+    is_admin: bool = False,
+) -> list[RenamePlanItem]:
+    mode = mode.lower().strip()
+    if mode not in {"direct", "prefix", "suffix", "replace"}:
+        raise FileStorageError("Unsupported rename mode")
+    selected_file_ids = list(dict.fromkeys(file_ids))
+    selected_folder_roots = _collapse_folder_roots(folder_paths)
+    if not selected_file_ids and not selected_folder_roots:
+        raise FileStorageError("No files or folders selected")
+    if selected_folder_roots and (not owner_type or not owner_id):
+        raise FileStorageError("Folder rename requires an owner file space")
+    if mode == "direct" and (len(selected_file_ids) + len(selected_folder_roots)) != 1:
+        raise FileStorageError("Direct rename requires exactly one selected item")
+
+    assets: dict[str, FileAsset] = {}
+    for file_id in selected_file_ids:
+        asset = get_asset_for_user(session, tenant_id=tenant_id, user_id=user_id, asset_id=file_id, require_write=True, is_admin=is_admin)
+        assets[asset.id] = asset
+
+    owner_type_norm = owner_type.lower().strip() if owner_type else None
+    owner_id_norm = owner_id
+    folder_rows_by_path: dict[str, FileFolder] = {}
+    affected_folder_paths: set[str] = set()
+    if selected_folder_roots and owner_type_norm and owner_id_norm:
+        ensure_owner_access(session, tenant_id=tenant_id, owner_type=owner_type_norm, owner_id=owner_id_norm, user_id=user_id, is_admin=is_admin)
+        for root in selected_folder_roots:
+            rows = _folder_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type_norm, owner_id=owner_id_norm).filter(
+                FileFolder.deleted_at.is_(None),
+                or_(FileFolder.path == root, FileFolder.path.like(f"{root}/%")),
+            ).all()
+            if not rows and not _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type_norm, owner_id=owner_id_norm).filter(FileAsset.deleted_at.is_(None), FileAsset.display_path.like(f"{root}/%")).first():
+                raise FileStorageError(f"Folder not found: {root}")
+            for row in rows:
+                folder_rows_by_path[row.path] = row
+                affected_folder_paths.add(row.path)
+            for asset in _asset_query_for_owner(session, tenant_id=tenant_id, owner_type=owner_type_norm, owner_id=owner_id_norm).filter(FileAsset.deleted_at.is_(None), FileAsset.display_path.like(f"{root}/%")).all():
+                assets[asset.id] = asset
+
+    folder_plan: dict[str, str] = {}
+    root_target_paths: dict[str, str] = {}
+    for root in selected_folder_roots:
+        if mode == "direct":
+            root_parent = "" if str(PurePosixPath(root).parent) == "." else str(PurePosixPath(root).parent)
+            root_new_name = _rename_basename(PurePosixPath(root).name, mode=mode, new_name=new_name)
+            new_root = normalize_folder(f"{root_parent}/{root_new_name}" if root_parent else root_new_name)
+        else:
+            new_root = normalize_folder(_rename_path(root, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix))
+        root_target_paths[root] = new_root
+        if root in affected_folder_paths:
+            folder_plan[root] = new_root
+        for path in sorted(affected_folder_paths, key=lambda item: item.count("/")):
+            if path != root and _path_under_root(path, root):
+                folder_plan[path] = _folder_new_path_for_root(path, root, new_root, recursive=recursive, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+
+    file_plan: dict[str, str] = {}
+    for asset in assets.values():
+        matched_root = next((root for root in sorted(selected_folder_roots, key=len, reverse=True) if _path_under_root(asset.display_path, root)), None)
+        if matched_root:
+            root_new = root_target_paths.get(matched_root)
+            if not root_new:
+                continue
+            file_plan[asset.id] = _file_new_path_for_root(asset.display_path, matched_root, root_new, recursive=recursive, mode=mode, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+        else:
+            file_plan[asset.id] = _rename_path(asset.display_path, mode=mode, new_name=new_name, find=find, replacement=replacement, prefix=prefix, suffix=suffix)
+
+    # Detect duplicate targets and existing active target conflicts before applying.
+    target_files: set[str] = set()
+    target_folders: set[str] = set()
+    affected_file_ids = set(file_plan)
+    for asset_id, target in file_plan.items():
+        normalized = normalize_logical_path(target)
+        if normalized in target_files:
+            raise FileStorageError(f"Rename would create duplicate file path: {normalized}")
+        target_files.add(normalized)
+        asset = assets[asset_id]
+        if _active_asset_exists(session, tenant_id=tenant_id, owner_type=asset.owner_type, owner_id=_asset_owner_id(asset), path=normalized, exclude_asset_id=asset.id):
+            raise FileStorageError(f"Target file already exists: {normalized}")
+        if _active_folder_exists(session, tenant_id=tenant_id, owner_type=asset.owner_type, owner_id=_asset_owner_id(asset), path=normalized):
+            raise FileStorageError(f"Target path is already a folder: {normalized}")
+    if selected_folder_roots and owner_type_norm and owner_id_norm:
+        for old_path, target in folder_plan.items():
+            normalized = normalize_folder(target)
+            if normalized in target_folders:
+                raise FileStorageError(f"Rename would create duplicate folder path: {normalized}")
+            target_folders.add(normalized)
+            if _active_folder_exists(session, tenant_id=tenant_id, owner_type=owner_type_norm, owner_id=owner_id_norm, path=normalized, exclude_paths=set(folder_plan.keys())):
+                raise FileStorageError(f"Target folder already exists: {normalized}")
+            if _active_asset_exists(session, tenant_id=tenant_id, owner_type=owner_type_norm, owner_id=owner_id_norm, path=normalized):
+                raise FileStorageError(f"Target path is already a file: {normalized}")
+
+    plan: list[RenamePlanItem] = []
+    for old_path, new_path in sorted(folder_plan.items()):
+        plan.append(RenamePlanItem(kind="folder", id=old_path, old_path=old_path, new_path=new_path))
+    for asset_id, new_path in file_plan.items():
+        asset = assets[asset_id]
+        plan.append(RenamePlanItem(kind="file", id=asset.id, old_path=asset.display_path, new_path=new_path))
+
+    if dry_run:
+        return plan
+
+    for old_path, new_path in folder_plan.items():
+        folder = folder_rows_by_path.get(old_path)
+        if folder:
+            folder.path = normalize_folder(new_path)
+            session.add(folder)
+    for asset_id, new_path in file_plan.items():
+        rename_asset(assets[asset_id], new_path=new_path)
+        session.add(assets[asset_id])
+    return plan