govoplan-files/src/govoplan_files/backend/storage/connector_spaces.py

from __future__ import annotations

from collections.abc import Mapping
from typing import Any

from sqlalchemy import false, or_
from sqlalchemy.orm import Session

from govoplan_files.backend.db.models import FileConnectorSpace
from govoplan_files.backend.storage.access import ensure_owner_access, user_group_ids
from govoplan_files.backend.storage.common import FileStorageError, utcnow
from govoplan_files.backend.storage.connector_browse import normalize_connector_browse_path
from govoplan_files.backend.storage.connector_profiles import ConnectorProfile


SYNC_MODES = {"manual"}


def connector_space_owner_id(space: FileConnectorSpace) -> str:
    return space.owner_user_id if space.owner_type == "user" else space.owner_group_id  # type: ignore[return-value]


def create_connector_space(
    session: Session,
    *,
    tenant_id: str,
    owner_type: str,
    owner_id: str,
    user_id: str,
    label: str,
    profile: ConnectorProfile,
    library_id: str | None = None,
    remote_path: str | None = None,
    sync_mode: str = "manual",
    read_only: bool = True,
    metadata: Mapping[str, Any] | None = None,
    is_admin: bool = False,
) -> FileConnectorSpace:
    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)
    label = _normalize_label(label)
    sync_mode = _normalize_sync_mode(sync_mode)
    remote_path = normalize_connector_browse_path(remote_path)
    library_id = _clean_optional(library_id)

    existing = _owned_query(session, tenant_id=tenant_id, owner_type=owner_type, owner_id=owner_id).filter(
        FileConnectorSpace.label == label,
        FileConnectorSpace.deleted_at.is_(None),
    ).first()
    if existing is not None:
        raise FileStorageError(f"Connector space already exists: {label}")

    space = FileConnectorSpace(
        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,
        label=label,
        connector_profile_id=profile.id,
        provider=profile.provider,
        library_id=library_id,
        remote_path=remote_path,
        sync_mode=sync_mode,
        read_only=read_only,
        is_active=True,
        created_by_user_id=user_id,
        metadata_=dict(metadata or {}),
    )
    session.add(space)
    session.flush()
    return space


def list_connector_spaces_for_user(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    owner_type: str | None = None,
    owner_id: str | None = None,
    include_inactive: bool = False,
    is_admin: bool = False,
) -> list[FileConnectorSpace]:
    query = session.query(FileConnectorSpace).filter(
        FileConnectorSpace.tenant_id == tenant_id,
        FileConnectorSpace.deleted_at.is_(None),
    )
    if not include_inactive:
        query = query.filter(FileConnectorSpace.is_active.is_(True))

    if owner_type:
        if not owner_id:
            raise FileStorageError("owner_id is required when owner_type is set")
        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 = _owner_filter(query, owner_type, owner_id)
    elif not is_admin:
        group_ids = user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
        query = query.filter(
            or_(
                FileConnectorSpace.owner_user_id == user_id,
                FileConnectorSpace.owner_group_id.in_(group_ids) if group_ids else false(),
            )
        )

    return query.order_by(FileConnectorSpace.owner_type.asc(), FileConnectorSpace.label.asc()).all()


def get_connector_space_for_user(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    space_id: str,
    include_inactive: bool = False,
    is_admin: bool = False,
) -> FileConnectorSpace:
    query = session.query(FileConnectorSpace).filter(
        FileConnectorSpace.id == space_id,
        FileConnectorSpace.tenant_id == tenant_id,
        FileConnectorSpace.deleted_at.is_(None),
    )
    if not include_inactive:
        query = query.filter(FileConnectorSpace.is_active.is_(True))
    space = query.one_or_none()
    if space is None:
        raise FileStorageError("Connector space not found")
    ensure_owner_access(
        session,
        tenant_id=tenant_id,
        owner_type=space.owner_type,
        owner_id=connector_space_owner_id(space),
        user_id=user_id,
        is_admin=is_admin,
    )
    return space


def update_connector_space(
    session: Session,
    space: FileConnectorSpace,
    *,
    user_id: str,
    label: str | None = None,
    library_id: str | None = None,
    remote_path: str | None = None,
    sync_mode: str | None = None,
    is_active: bool | None = None,
    metadata: Mapping[str, Any] | None = None,
    is_admin: bool = False,
) -> FileConnectorSpace:
    ensure_owner_access(
        session,
        tenant_id=space.tenant_id,
        owner_type=space.owner_type,
        owner_id=connector_space_owner_id(space),
        user_id=user_id,
        is_admin=is_admin,
    )
    if label is not None:
        new_label = _normalize_label(label)
        existing = _owned_query(
            session,
            tenant_id=space.tenant_id,
            owner_type=space.owner_type,
            owner_id=connector_space_owner_id(space),
        ).filter(
            FileConnectorSpace.id != space.id,
            FileConnectorSpace.label == new_label,
            FileConnectorSpace.deleted_at.is_(None),
        ).first()
        if existing is not None:
            raise FileStorageError(f"Connector space already exists: {new_label}")
        space.label = new_label
    if library_id is not None:
        space.library_id = _clean_optional(library_id)
    if remote_path is not None:
        space.remote_path = normalize_connector_browse_path(remote_path)
    if sync_mode is not None:
        space.sync_mode = _normalize_sync_mode(sync_mode)
    if is_active is not None:
        space.is_active = bool(is_active)
    if metadata is not None:
        space.metadata_ = dict(metadata)
    session.add(space)
    session.flush()
    return space


def soft_delete_connector_space(
    session: Session,
    space: FileConnectorSpace,
    *,
    user_id: str,
    is_admin: bool = False,
) -> FileConnectorSpace:
    ensure_owner_access(
        session,
        tenant_id=space.tenant_id,
        owner_type=space.owner_type,
        owner_id=connector_space_owner_id(space),
        user_id=user_id,
        is_admin=is_admin,
    )
    space.deleted_at = utcnow()
    space.is_active = False
    session.add(space)
    session.flush()
    return space


def _owned_query(session: Session, *, tenant_id: str, owner_type: str, owner_id: str):
    query = session.query(FileConnectorSpace).filter(
        FileConnectorSpace.tenant_id == tenant_id,
        FileConnectorSpace.owner_type == owner_type,
    )
    return _owner_filter(query, owner_type, owner_id)


def _owner_filter(query, owner_type: str, owner_id: str):
    if owner_type == "user":
        return query.filter(FileConnectorSpace.owner_user_id == owner_id)
    if owner_type == "group":
        return query.filter(FileConnectorSpace.owner_group_id == owner_id)
    raise FileStorageError("Files must be owned by a user or group")


def _normalize_label(value: str) -> str:
    label = value.strip()
    if not label:
        raise FileStorageError("Connector space label is required")
    if len(label) > 255:
        raise FileStorageError("Connector space label is too long")
    return label


def _normalize_sync_mode(value: str) -> str:
    mode = value.strip().casefold()
    if mode not in SYNC_MODES:
        raise FileStorageError(f"Unsupported connector space sync mode: {value}")
    return mode


def _clean_optional(value: object) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None