"""Canonical owner resolution for a thread's filesystem paths.

Task-deeplink conversations (those whose ``threads_meta.metadata.taskId`` is
set — rwfx / 3qfx / xdfx) are **shared across all users**: any logged-in
account may open the same conversation (enforced by the ``taskId`` escape
branches in :mod:`deerflow.persistence.thread_meta.sql`). But the agent's
generated files (md / artifacts) and uploads live under a *per-user* directory::

    {base_dir}/users/{user_id}/threads/{thread_id}/user-data/...

If that ``{user_id}`` were the **current login**, account B opening a shared
task conversation would resolve to an empty ``users/B/...`` directory and 404
on every file account A generated — which is exactly the "切换账号后生成的 md
文件找不到" bug.

To keep the *file* layer consistent with the *conversation-sharing* layer, all
path resolution for a task-shared thread must use one stable user bucket: the
thread's **creator** (``threads_meta.user_id``). That value is fixed at thread
creation and never changes, so every account resolves to the same physical
directory.

Personal (non-task) threads are unaffected: their creator is always the only /
current user, so :func:`resolve_path_user_id` returns exactly what
``get_effective_user_id`` would have. On any failure (no DB, missing row,
memory-mode store) it falls back to the current login, so behaviour degrades
safely to the pre-fix semantics.
"""

from __future__ import annotations

import logging
import threading

from deerflow.runtime.user_context import get_effective_user_id

logger = logging.getLogger(__name__)

# thread_id -> creator user_id, populated only for task-shared threads. The
# (taskId, creator) pair is immutable for a thread, so positive results are
# cached forever; the fallback (current login) is never cached because a row may
# appear after the first lookup.
_owner_cache: dict[str, str] = {}
_owner_cache_lock = threading.Lock()
_OWNER_CACHE_MAX = 4096
_LOCAL_ONLY_THREAD_PREFIXES = ("open-", "__skill-review__")


def _is_local_only_thread(thread_id: str | None) -> bool:
    """Return True for synthetic threads that are never persisted in threads_meta."""
    return bool(thread_id) and str(thread_id).startswith(_LOCAL_ONLY_THREAD_PREFIXES)


def _cache_get(thread_id: str) -> str | None:
    with _owner_cache_lock:
        return _owner_cache.get(thread_id)


def _cache_put(thread_id: str, owner: str) -> None:
    with _owner_cache_lock:
        # Crude bound: clear wholesale rather than track LRU order. The mapping
        # is tiny and immutable, so a rare full clear just re-warms lazily.
        if len(_owner_cache) >= _OWNER_CACHE_MAX:
            _owner_cache.clear()
        _owner_cache[thread_id] = owner


async def _fetch_task_owner(thread_id: str) -> str | None:
    """Return the creator user_id iff *thread_id* is a task-shared thread.

    Returns ``None`` for personal threads, untracked threads, the memory-mode
    store, or any DB error — callers then fall back to the current login.
    """
    from deerflow.persistence.engine import get_session_factory
    from deerflow.persistence.thread_meta import ThreadMetaRepository

    session_factory = get_session_factory()
    if session_factory is None:
        return None
    store = ThreadMetaRepository(session_factory)
    # user_id=None bypasses the owner filter so we can read any thread's row.
    row = await store.get(thread_id, user_id=None)
    if not row:
        return None
    metadata = row.get("metadata") or {}
    owner = row.get("user_id")
    if owner and _is_creator_shared_thread(metadata):
        return str(owner)
    return None


def _is_creator_shared_thread(metadata: dict) -> bool:
    """True iff a thread's files should resolve to its **creator** (shared bucket).

    Two kinds of threads are shared to one stable directory rather than the
    current login:

    - **Task-deeplink conversations** (``metadata.taskId``) — rwfx / 3qfx / xdfx,
      explicitly shared across all users;
    - **Roundtable 会商 system threads** (``metadata.thread_type == "roundtable"``)
      — the Step 2 seats / coordinator and Step 3 report / summary / dashboard /
      structure threads. Their sandbox artifacts (md / 流程图 / 报告) live under the
      creator's directory, but a历史草稿 reopened later — or a taskId会商 draft opened
      by **another account** — must still be able to view them. Resolving to the
      creator keeps the *file* layer consistent with the *draft-sharing* layer.

    Roundtable thread ids are random UUIDs surfaced only inside the owner's draft
    (personal drafts are owner-scoped; taskId drafts are already shared), so this
    does not let an unrelated account *discover* another user's 会商 files — it only
    lets a caller that already holds the thread id read the creator's bucket.
    """
    return bool(metadata.get("taskId")) or metadata.get("thread_type") == "roundtable"


def resolve_path_user_id(thread_id: str | None) -> str:
    """Canonical user bucket for *thread_id*'s files (sync).

    For a task-shared thread this is the thread creator, so all accounts that
    share the conversation read/write the same physical directory. For every
    other case it is the current login (``get_effective_user_id``).
    """
    if not thread_id or _is_local_only_thread(thread_id):
        return get_effective_user_id()
    cached = _cache_get(thread_id)
    if cached is not None:
        return cached
    try:
        from deerflow.persistence.engine import run_db_blocking

        owner = run_db_blocking(_fetch_task_owner(thread_id))
    except Exception:
        logger.debug("Failed to resolve task-shared owner for thread %s", thread_id, exc_info=True)
        owner = None
    if owner:
        _cache_put(thread_id, owner)
        return owner
    return get_effective_user_id()


async def aresolve_path_user_id(thread_id: str | None) -> str:
    """Async variant of :func:`resolve_path_user_id` for FastAPI handlers."""
    if not thread_id or _is_local_only_thread(thread_id):
        return get_effective_user_id()
    cached = _cache_get(thread_id)
    if cached is not None:
        return cached
    try:
        owner = await _fetch_task_owner(thread_id)
    except Exception:
        logger.debug("Failed to resolve task-shared owner for thread %s", thread_id, exc_info=True)
        owner = None
    if owner:
        _cache_put(thread_id, owner)
        return owner
    return get_effective_user_id()