ctms/backend/app/services/onlyoffice_collaboration_service.py

from __future__ import annotations

import hashlib
import hmac
import json
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit, urlunsplit

import httpx
from fastapi import HTTPException, Request, status
from jose import JWTError, jwt
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.models.collaboration import (
    CollaborationCallbackReceipt,
    CollaborationFile,
    CollaborationRevision,
    CollaborationSession,
    CollaborationShareLink,
)
from app.models.user import User
from app.schemas.collaboration import CollaborationCallbackPayload, CollaborationEditorConfigRead
from app.services import collaboration_service, ledger_access, onlyoffice_service


def collaboration_document_key(file_id: uuid.UUID, generation: int) -> str:
    fingerprint = f"{settings.ONLYOFFICE_INSTANCE_ID or ''}:collaboration:{file_id}:{generation}"
    return f"ctms-collab-{hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()}"


def _content_url(session_id: uuid.UUID) -> str:
    return (
        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
        f"/internal/onlyoffice/collaboration/sessions/{session_id}/content"
    )


def _callback_url(session_id: uuid.UUID) -> str:
    return (
        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
        f"/internal/onlyoffice/collaboration/sessions/{session_id}/callback"
    )


async def _active_session(
    db: AsyncSession, item: CollaborationFile, user_id: uuid.UUID
) -> CollaborationSession:
    session = await db.scalar(
        select(CollaborationSession).where(
            CollaborationSession.file_id == item.id,
            CollaborationSession.generation == item.generation,
        ).order_by(CollaborationSession.created_at.desc())
    )
    if session:
        if session.status != "ACTIVE":
            session.status = "ACTIVE"
            session.closed_at = None
            await db.commit()
            await db.refresh(session)
        return session
    if not item.current_revision_id:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作文件尚无可编辑内容")
    session = CollaborationSession(
        file_id=item.id,
        base_revision_id=item.current_revision_id,
        document_key=collaboration_document_key(item.id, item.generation),
        generation=item.generation,
        started_by=user_id,
    )
    db.add(session)
    await db.commit()
    await db.refresh(session)
    return session


async def build_editor_config(
    db: AsyncSession, item: CollaborationFile, user
) -> CollaborationEditorConfigRead:
    if ledger_access.is_ledger(item):
        await ledger_access.require_access(db, item, user)
    await onlyoffice_service.ensure_onlyoffice_available()
    revision = await db.get(CollaborationRevision, item.current_revision_id)
    if not revision or not Path(revision.file_uri).exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在")
    can_edit = await collaboration_service.can_edit_file(db, item, user)
    can_request_edit = await collaboration_service.can_request_edit_file(db, item, user)
    can_download = await collaboration_service.can_export_file(db, item, user)
    can_save_as = can_download and await collaboration_service.can_create_file(db, item, user)
    session = await _active_session(db, item, user.id)
    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
    config: dict[str, Any] = {
        "type": "desktop",
        "documentType": item.file_type,
        "document": {
            "fileType": item.extension,
            "key": session.document_key,
            "title": item.title,
            "url": _content_url(session.id),
            "permissions": {
                "chat": False,
                "copy": can_download,
                "comment": can_edit,
                "download": can_download,
                # In view mode ONLYOFFICE displays "Edit current file" only
                # when edit=true and onRequestEditRights is registered. CTMS
                # handles that event as an approval request, not an escalation.
                "edit": can_edit or can_request_edit,
                "fillForms": False,
                "modifyContentControl": can_edit,
                "modifyFilter": can_edit,
                "print": can_download,
                "protect": False,
                "review": False,
            },
        },
        "editorConfig": {
            "callbackUrl": _callback_url(session.id),
            "coEditing": {"mode": "fast", "change": False},
            "customization": {
                "autosave": True,
                "chat": False,
                "comments": can_edit,
                "forcesave": can_edit,
                "help": False,
                "plugins": False,
            },
            "lang": "zh-CN",
            "mode": "edit" if can_edit else "view",
            "user": {"id": str(user.id), "name": user.full_name},
        },
    }
    config["token"] = jwt.encode(
        {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())},
        settings.ONLYOFFICE_JWT_SECRET or "",
        algorithm="HS256",
    )
    return CollaborationEditorConfigRead(
        file_id=item.id,
        file_name=item.title,
        access_mode="edit" if can_edit else "view",
        can_save_as=can_save_as,
        can_download=can_download,
        can_request_edit=can_request_edit,
        expires_at=expires_at,
        config=config,
    )


async def build_shared_editor_config(
    db: AsyncSession,
    item: CollaborationFile,
    link: CollaborationShareLink,
    *,
    client_id: str,
    display_name: str,
) -> CollaborationEditorConfigRead:
    if ledger_access.is_ledger(item):
        version = link.token_version
        item = await db.scalar(select(CollaborationFile).where(
            CollaborationFile.id == item.id,
        ).with_for_update().execution_options(populate_existing=True))
        await db.refresh(link)
        if (not item or item.status != "ACTIVE" or not link.enabled or link.token_version != version
                or (link.expires_at and link.expires_at <= datetime.now(timezone.utc))):
            raise HTTPException(404, "共享链接不存在或已失效")
    await onlyoffice_service.ensure_onlyoffice_available()
    revision = await db.get(CollaborationRevision, item.current_revision_id)
    if not revision or not Path(revision.file_uri).exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享文件内容不存在")
    can_edit = link.access_mode == "EDIT"
    session = await _active_session(db, item, item.owner_id)
    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
    if link.expires_at and link.expires_at < expires_at:
        expires_at = link.expires_at
    external_user_id = f"share-{link.id.hex[:12]}-{client_id[:32]}"
    if ledger_access.is_ledger(item):
        known_users = json.loads(session.active_users or "[]")
        session.active_users = json.dumps(list(dict.fromkeys([*known_users, external_user_id])))
        await db.commit()
    config: dict[str, Any] = {
        "type": "desktop",
        "documentType": item.file_type,
        "document": {
            "fileType": item.extension,
            "key": session.document_key,
            "title": item.title,
            "url": _content_url(session.id),
            "permissions": {
                "chat": False,
                "copy": item.allow_export,
                "comment": can_edit,
                "download": item.allow_export,
                "edit": can_edit,
                "fillForms": False,
                "modifyContentControl": can_edit,
                "modifyFilter": can_edit,
                "print": item.allow_export,
                "protect": False,
                "review": False,
            },
        },
        "editorConfig": {
            "callbackUrl": _callback_url(session.id),
            "coEditing": {"mode": "fast", "change": False},
            "customization": {
                "autosave": can_edit,
                "chat": False,
                "comments": can_edit,
                "forcesave": can_edit,
                "help": False,
                "plugins": False,
            },
            "lang": "zh-CN",
            "mode": "edit" if can_edit else "view",
            "user": {"id": external_user_id, "name": display_name},
        },
    }
    config["token"] = jwt.encode(
        {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())},
        settings.ONLYOFFICE_JWT_SECRET or "",
        algorithm="HS256",
    )
    return CollaborationEditorConfigRead(
        file_id=item.id,
        file_name=item.title,
        access_mode="edit" if can_edit else "view",
        can_save_as=False,
        can_download=item.allow_export,
        expires_at=expires_at,
        config=config,
    )


async def get_session_content(
    db: AsyncSession, session_id: uuid.UUID, authorization: str | None
) -> tuple[CollaborationRevision, CollaborationFile]:
    session = await db.get(CollaborationSession, session_id)
    if not session:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在")
    onlyoffice_service.validate_outbox_token(authorization, _content_url(session_id))
    revision = await db.get(CollaborationRevision, session.base_revision_id)
    item = await db.get(CollaborationFile, session.file_id)
    if not revision or not item or not Path(revision.file_uri).exists():
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在")
    if ledger_access.is_ledger(item) and session.generation != item.generation:
        raise HTTPException(403, "台账会话已失效，请重新打开")
    return revision, item


def validate_callback_token(token: str | None, payload: CollaborationCallbackPayload) -> dict[str, Any]:
    if not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="缺少 ONLYOFFICE 回调签名")
    value = token.strip()
    if " " in value:
        scheme, credential = value.split(" ", 1)
        if scheme.lower() != "bearer":
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名格式无效")
        value = credential.strip()
    try:
        decoded = onlyoffice_service.decode_onlyoffice_token(value)
    except JWTError as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名无效") from exc
    signed = decoded.get("payload") if isinstance(decoded.get("payload"), dict) else decoded
    if not isinstance(signed, dict):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调载荷无效")
    signed_key = signed.get("key")
    signed_status = signed.get("status")
    if not isinstance(signed_key, str) or not hmac.compare_digest(signed_key, payload.key):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调 key 不匹配")
    if not isinstance(signed_status, int) or signed_status != payload.status:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调状态不匹配")
    if payload.url:
        signed_url = signed.get("url")
        if not isinstance(signed_url, str) or not hmac.compare_digest(signed_url, payload.url):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调文件地址不匹配")
    return decoded


def _callback_fingerprint(payload: CollaborationCallbackPayload) -> str:
    normalized = payload.model_dump(mode="json", exclude_none=True)
    return hashlib.sha256(json.dumps(normalized, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def _url_origin_matches(actual, expected) -> bool:
    actual_port = actual.port or (443 if actual.scheme == "https" else 80)
    expected_port = expected.port or (443 if expected.scheme == "https" else 80)
    return (
        actual.scheme == expected.scheme
        and actual.hostname
        and actual.hostname.lower() == (expected.hostname or "").lower()
        and actual_port == expected_port
    )


def _validate_result_url(url: str) -> str:
    actual = urlsplit(url)
    if (
        actual.scheme not in {"http", "https"}
        or actual.username
        or actual.password
        or actual.fragment
        or not actual.hostname
    ):
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任")

    internal = urlsplit(settings.ONLYOFFICE_INTERNAL_URL.rstrip("/"))
    if _url_origin_matches(actual, internal):
        return urlunsplit((internal.scheme, internal.netloc, actual.path, actual.query, ""))

    public = urlsplit(settings.FRONTEND_PUBLIC_URL.rstrip("/"))
    proxy_prefix = "/onlyoffice/"
    if not _url_origin_matches(actual, public) or not actual.path.startswith(proxy_prefix):
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任")
    internal_path = f"{internal.path.rstrip('/')}/{actual.path[len(proxy_prefix):]}"
    return urlunsplit((internal.scheme, internal.netloc, internal_path, actual.query, ""))


async def _download_result(url: str) -> bytes:
    download_url = _validate_result_url(url)
    try:
        async with httpx.AsyncClient(timeout=30.0, follow_redirects=False) as client:
            async with client.stream("GET", download_url) as response:
                if response.status_code != status.HTTP_200_OK:
                    raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败")
                content = bytearray()
                async for chunk in response.aiter_bytes():
                    content.extend(chunk)
                    if len(content) > settings.COLLABORATION_MAX_FILE_BYTES:
                        raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="ONLYOFFICE 保存文件超出限制")
    except httpx.HTTPError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败") from exc
    return bytes(content)


async def _callback_user(
    db: AsyncSession, payload: CollaborationCallbackPayload, session: CollaborationSession
) -> User | None:
    has_public_share_user = False
    for value in payload.users or json.loads(session.active_users or "[]"):
        if value.startswith("share-"):
            has_public_share_user = True
            continue
        try:
            user = await db.get(User, uuid.UUID(value))
        except (ValueError, TypeError):
            user = None
        if user:
            return user
    if has_public_share_user:
        return None
    user = await db.get(User, session.started_by)
    if not user:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话用户不存在")
    return user


async def _ledger_callback_can_edit(db, item, payload, session):
    users = payload.users or json.loads(session.active_users or "[]")
    if not users:
        return await ledger_access.can_edit(db, item, await _callback_user(db, payload, session))
    for value in users:
        try:
            actor = await db.get(User, uuid.UUID(value))
        except (ValueError, TypeError):
            continue
        if await ledger_access.can_edit(db, item, actor):
            return True
    link = await db.scalar(select(CollaborationShareLink).where(
        CollaborationShareLink.file_id == item.id,
        CollaborationShareLink.enabled.is_(True),
        CollaborationShareLink.access_mode == "EDIT",
    ))
    if not link or (link.expires_at and link.expires_at <= datetime.now(timezone.utc)):
        return False
    prefix = f"share-{link.id.hex[:12]}-"
    return any(value.startswith(prefix) for value in users)


async def process_callback(
    db: AsyncSession,
    session_id: uuid.UUID,
    payload: CollaborationCallbackPayload,
) -> dict[str, int]:
    # Lock the file before writing sessions, matching editor initialization and
    # ledger permission changes. This also serializes callbacks across generations.
    session = await db.scalar(
        select(CollaborationSession)
        .join(CollaborationFile, CollaborationFile.id == CollaborationSession.file_id)
        .where(CollaborationSession.id == session_id)
        .with_for_update(of=CollaborationFile)
    )
    if not session:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在")
    if not hmac.compare_digest(session.document_key, payload.key):
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话 key 不匹配")
    fingerprint = _callback_fingerprint(payload)
    duplicate = await db.scalar(select(CollaborationCallbackReceipt.id).where(
        CollaborationCallbackReceipt.session_id == session.id,
        CollaborationCallbackReceipt.fingerprint == fingerprint,
    ))
    if duplicate:
        return {"error": 0}

    item = await db.get(CollaborationFile, session.file_id)
    if not item:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件不存在")
    if ledger_access.is_ledger(item):
        # Same lock as grant/settings changes: a stale callback cannot race a revocation.
        item = await db.scalar(select(CollaborationFile).where(
            CollaborationFile.id == item.id,
        ).with_for_update().execution_options(populate_existing=True))
    session.last_callback_at = datetime.now(timezone.utc)
    result = "ACKNOWLEDGED"
    saved_revision_id = None

    if payload.status in {2, 6}:
        if not payload.url:
            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存回调缺少文件地址")
        if session.generation != item.generation:
            result = "STALE"
        elif ledger_access.is_ledger(item) and (
            item.status != "ACTIVE" or not await _ledger_callback_can_edit(db, item, payload, session)
        ):
            result = "ACCESS_REVOKED"
        else:
            content = await _download_result(payload.url)
            actor = await _callback_user(db, payload, session)
            source = "SESSION_CLOSE" if payload.status == 2 else "FORCE_SAVE"
            if actor is None:
                source = "SHARE_SESSION_CLOSE" if payload.status == 2 else "SHARE_FORCE_SAVE"
            revision, created = await collaboration_service.append_revision(
                db, item, content, source=source, created_by=actor.id if actor else None
            )
            saved_revision_id = revision.id
            result = "SAVED" if created else "UNCHANGED"
            if payload.status == 2:
                item.generation += 1
                session.status = "CLOSED"
                session.closed_at = datetime.now(timezone.utc)
            else:
                # 强制保存不结束当前共同编辑会话；同步基线可保证 Document
                # Server 缓存重建时仍从最近一次持久化内容恢复。
                session.base_revision_id = revision.id
    elif payload.status == 4:
        session.status = "CLOSED"
        session.closed_at = datetime.now(timezone.utc)
        result = "UNCHANGED"
    elif payload.status in {3, 7}:
        session.status = "ERROR"
        result = "ERROR"

    if payload.users or not ledger_access.is_ledger(item):
        session.active_users = json.dumps(payload.users, ensure_ascii=True)
    db.add(CollaborationCallbackReceipt(
        session_id=session.id,
        fingerprint=fingerprint,
        callback_status=payload.status,
        result=result,
        revision_id=saved_revision_id,
    ))
    await db.commit()
    # ONLYOFFICE 要求回调处理器在接收并记录状态后固定确认成功。
    # status 3/7 表示文档服务自身保存失败，不应通过 error=1 制造重试环。
    return {"error": 0}