"""Conversation sharing API. A user creates a share code for *one* of their conversations; redeeming the code copies that conversation into the redeemer's account as an independent copy they fully own and can keep chatting in. See ``app/gateway/thread_copy.py`` for the checkpoint duplication. """ from __future__ import annotations import logging import mimetypes import shutil import uuid from pathlib import Path from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import FileResponse from pydantic import BaseModel, Field from app.gateway.deps import get_checkpointer, get_thread_share_store, get_thread_store from app.gateway.thread_copy import copy_thread_state from deerflow.config.paths import get_paths from deerflow.persistence.thread_meta.base import ThreadMetaStore from deerflow.persistence.thread_shares.base import ThreadShareStore from deerflow.runtime import serialize_channel_values from deerflow.runtime.user_context import get_effective_user_id logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/thread-shares", tags=["thread-shares"]) # Public, unauthenticated router. Its prefix is whitelisted in # ``auth_middleware._PUBLIC_PATH_PREFIXES`` — only mount read-only endpoints # that are intentionally world-readable here. public_router = APIRouter(prefix="/api/public/shares", tags=["thread-shares-public"]) _UNTITLED = "未命名对话" # Valid share modes. "import" copies the conversation into the recipient's # account; "view" is a public read-only static page anyone can open. _SHARE_MODES = ("import", "view") def _current_user_id(request: Request) -> str: user = getattr(request.state, "user", None) if user is not None: return str(user.id) return get_effective_user_id() # --------------------------------------------------------------------------- # Request / response models # --------------------------------------------------------------------------- class CreateShareRequest(BaseModel): thread_id: str = Field(..., description="The conversation to share") mode: str = Field( default="import", description='"import" — recipient copies the conversation; "view" — public read-only page', ) class ShareResponse(BaseModel): share_code: str = Field(..., description="Short code that recipients enter to import") owner_user_id: str = Field(..., description="User who created the share") thread_id: str = Field(..., description="The shared conversation") revoked: bool = Field(default=False, description="Whether the code has been revoked") mode: str = Field(default="import", description='"import" or "view"') created_at: str | None = Field(default=None, description="Creation timestamp (ISO-8601)") class PublicShareSnapshotResponse(BaseModel): """Read-only conversation snapshot served to anonymous viewers.""" share_code: str title: str = Field(..., description="Conversation title") mode: str = Field(default="view") messages: list[dict] = Field(default_factory=list, description="Serialized conversation messages") artifacts: list[str] = Field(default_factory=list, description="Artifact virtual paths produced in the conversation") created_at: str | None = Field(default=None, description="When the share was created (ISO-8601)") class SharesListResponse(BaseModel): shares: list[ShareResponse] class SharePreviewResponse(BaseModel): share_code: str shared_by: str = Field(..., description="Owner user id of the shared conversation") title: str = Field(..., description="Title of the shared conversation") class ImportResultResponse(BaseModel): imported: int = Field(..., description="1 when the conversation was copied, else 0") skipped: int = Field(..., description="1 when it was already imported, else 0") # --------------------------------------------------------------------------- # Owner side — create / list / revoke # --------------------------------------------------------------------------- @router.post("", response_model=ShareResponse, summary="Create a share code for one conversation") async def create_share( body: CreateShareRequest, request: Request, share_store: ThreadShareStore = Depends(get_thread_share_store), thread_store: ThreadMetaStore = Depends(get_thread_store), ) -> ShareResponse: me = _current_user_id(request) mode = body.mode if body.mode in _SHARE_MODES else "import" # Only the conversation's owner may share it. meta = await thread_store.get(body.thread_id, user_id=me) if meta is None: raise HTTPException(status_code=404, detail="Conversation not found") # Reuse an existing active code (of the same mode) so re-sharing is idempotent. existing = await share_store.get_active_for_thread(me, body.thread_id, mode=mode) if existing is not None: return ShareResponse(**existing) record = await share_store.create(me, body.thread_id, mode=mode) return ShareResponse(**record) @router.get("", response_model=SharesListResponse, summary="List my conversation share codes") async def list_shares( request: Request, share_store: ThreadShareStore = Depends(get_thread_share_store), ) -> SharesListResponse: records = await share_store.list_for_owner(_current_user_id(request)) return SharesListResponse(shares=[ShareResponse(**r) for r in records]) @router.delete("/{share_code}", summary="Revoke a conversation share code") async def revoke_share( share_code: str, request: Request, share_store: ThreadShareStore = Depends(get_thread_share_store), ) -> dict[str, bool]: ok = await share_store.revoke(share_code, _current_user_id(request)) if not ok: raise HTTPException(status_code=404, detail="Share code not found") return {"success": True} # --------------------------------------------------------------------------- # Recipient side — preview / import # --------------------------------------------------------------------------- async def _resolve_active_share(share_code: str, share_store: ThreadShareStore) -> dict: share = await share_store.get(share_code) if share is None: raise HTTPException(status_code=404, detail="Share code not found") if share.get("revoked"): raise HTTPException(status_code=410, detail="This share code has been revoked") return share @router.get("/{share_code}/preview", response_model=SharePreviewResponse, summary="Preview a shared conversation") async def preview_share( share_code: str, share_store: ThreadShareStore = Depends(get_thread_share_store), thread_store: ThreadMetaStore = Depends(get_thread_store), ) -> SharePreviewResponse: share = await _resolve_active_share(share_code, share_store) owner = share["owner_user_id"] meta = await thread_store.get(share["thread_id"], user_id=owner) if meta is None: raise HTTPException(status_code=404, detail="The shared conversation no longer exists") return SharePreviewResponse( share_code=share_code, shared_by=owner, title=meta.get("display_name") or _UNTITLED, ) def _copy_thread_files(origin_thread_id: str, owner_user_id: str, new_thread_id: str, recipient_user_id: str) -> None: """Copy a thread's user-data directory (uploads + outputs/artifacts).""" paths = get_paths() src_data = paths.thread_dir(origin_thread_id, user_id=owner_user_id) / "user-data" if not src_data.is_dir(): return dst_data = paths.thread_dir(new_thread_id, user_id=recipient_user_id) / "user-data" shutil.copytree(src_data, dst_data, dirs_exist_ok=True) @router.post("/{share_code}/import", response_model=ImportResultResponse, summary="Import a shared conversation") async def import_share( share_code: str, request: Request, share_store: ThreadShareStore = Depends(get_thread_share_store), thread_store: ThreadMetaStore = Depends(get_thread_store), ) -> ImportResultResponse: share = await _resolve_active_share(share_code, share_store) if share.get("mode") == "view": raise HTTPException(status_code=400, detail="This is a view-only share — open its link to read it instead of importing") owner = share["owner_user_id"] origin_id = share["thread_id"] recipient = _current_user_id(request) if owner == recipient: raise HTTPException(status_code=400, detail="You cannot import your own share code") src = await thread_store.get(origin_id, user_id=owner) if src is None: raise HTTPException(status_code=404, detail="The shared conversation no longer exists") # Dedup: skip if this user already imported this conversation. mine = await thread_store.search(user_id=recipient, limit=2000) already = {(t.get("metadata") or {}).get("shared_origin") for t in mine} if origin_id in already: return ImportResultResponse(imported=0, skipped=1) new_id = str(uuid.uuid4()) checkpointer = get_checkpointer(request) await copy_thread_state(checkpointer, origin_id, new_id) metadata = dict(src.get("metadata") or {}) metadata.update({"shared_from_code": share_code, "shared_origin": origin_id, "shared_by": owner}) await thread_store.create( new_id, assistant_id=src.get("assistant_id"), user_id=recipient, display_name=src.get("display_name"), metadata=metadata, ) try: _copy_thread_files(origin_id, owner, new_id, recipient) except Exception as exc: # noqa: BLE001 — file copy failure must not lose the conversation logger.warning("Failed to copy files for thread %s -> %s: %s", origin_id, new_id, exc) logger.info("Share %s imported by %s as thread %s", share_code, recipient, new_id) return ImportResultResponse(imported=1, skipped=0) # --------------------------------------------------------------------------- # Public read-only view — no authentication required # --------------------------------------------------------------------------- async def _load_thread_snapshot(checkpointer, thread_id: str) -> tuple[list[dict], list[str]]: """Return (messages, artifacts) from a thread's latest checkpoint. Reads straight from the checkpointer — the same source the live chat streams from — so the static snapshot matches the conversation exactly. """ config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} checkpoint_tuple = await checkpointer.aget_tuple(config) if checkpoint_tuple is None: return [], [] checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} values = serialize_channel_values(checkpoint.get("channel_values", {}) or {}) messages = values.get("messages") artifacts = values.get("artifacts") return ( [m for m in messages if isinstance(m, dict)] if isinstance(messages, list) else [], [a for a in artifacts if isinstance(a, str)] if isinstance(artifacts, list) else [], ) @public_router.get( "/{share_code}", response_model=PublicShareSnapshotResponse, summary="Public read-only snapshot of a view-shared conversation", ) async def get_public_share_snapshot( share_code: str, share_store: ThreadShareStore = Depends(get_thread_share_store), thread_store: ThreadMetaStore = Depends(get_thread_store), checkpointer=Depends(get_checkpointer), ) -> PublicShareSnapshotResponse: """Serve a frozen, read-only conversation. Anyone with the link can read it. Only ``mode="view"`` shares are exposed; missing, revoked, or import-mode codes all return 404 so the endpoint never reveals which case applies. """ share = await share_store.get(share_code) if share is None or share.get("revoked") or share.get("mode") != "view": raise HTTPException(status_code=404, detail="Share not found") meta = await thread_store.get(share["thread_id"], user_id=share["owner_user_id"]) if meta is None: raise HTTPException(status_code=404, detail="The shared conversation no longer exists") messages, artifacts = await _load_thread_snapshot(checkpointer, share["thread_id"]) return PublicShareSnapshotResponse( share_code=share_code, title=meta.get("display_name") or _UNTITLED, mode="view", messages=messages, artifacts=artifacts, created_at=share.get("created_at"), ) # Content types that can execute scripts in the app origin — always served as # a download attachment so a shared file can never run as a top-level document. _PUBLIC_SHARE_ACTIVE_TYPES = {"text/html", "application/xhtml+xml", "image/svg+xml"} def _public_share_disposition(mime_type: str) -> str: """Pick a Content-Disposition for a public-share file. Active content (HTML/XHTML/SVG) is forced to ``attachment`` so it cannot execute as a top-level document; everything else (images, Markdown, text, PDF, …) is served ``inline`` so viewers can open it directly. """ return "attachment" if mime_type in _PUBLIC_SHARE_ACTIVE_TYPES else "inline" @public_router.get( "/{share_code}/files/{path:path}", summary="Public file attachment from a view-shared conversation", ) async def get_public_share_file( share_code: str, path: str, share_store: ThreadShareStore = Depends(get_thread_share_store), ) -> FileResponse: """Serve a file referenced by a view-shared conversation. A ``view`` share is a public read-only page, so the files its conversation references (images, Markdown, text, PDF, …) must be openable. Path resolution is confined to the owner's thread directory, so this cannot read arbitrary files. Active content (HTML/XHTML/SVG) is forced to download so it can never execute as a top-level document in the app origin; everything else is served inline. """ share = await share_store.get(share_code) if share is None or share.get("revoked") or share.get("mode") != "view": raise HTTPException(status_code=404, detail="Share not found") # Resolve the virtual path inside the *owner's* thread directory. try: actual = get_paths().resolve_virtual_path(share["thread_id"], path, user_id=share["owner_user_id"]) except ValueError as exc: status = 403 if "traversal" in str(exc) else 400 raise HTTPException(status_code=status, detail=str(exc)) actual_path = Path(actual) if not actual_path.exists() or not actual_path.is_file(): raise HTTPException(status_code=404, detail="File not found") mime_type = mimetypes.guess_type(actual_path.name)[0] or "application/octet-stream" return FileResponse( actual_path, media_type=mime_type, content_disposition_type=_public_share_disposition(mime_type), headers={"Cache-Control": "public, max-age=300"}, )