ObsiGate/backend/services/vaults.py

"""Vault listing and directory browsing services.

Single source of truth consumed by both the REST routes (``/api/vaults``,
``/api/browse/{vault}``) and the AI tool layer (``list_vaults``,
``list_directory``).
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from backend.services.errors import ServiceError
from backend.services.paths import resolve_safe_path


def list_accessible_vaults(user: dict[str, Any]) -> list[dict[str, Any]]:
    """Return the vaults *user* may access, with summary metadata."""
    from backend.auth.middleware import check_vault_access
    from backend.indexer import index

    result: list[dict[str, Any]] = []
    for name, data in index.items():
        if not check_vault_access(name, user):
            continue
        result.append({
            "name": name,
            "file_count": len(data.get("files", [])),
            "tag_count": len(data.get("tags", {})),
            "type": data.get("config", {}).get("type", "VAULT"),
        })
    return result


def get_vault_root(vault_name: str) -> Path:
    """Return the filesystem root of *vault_name* or raise ``not_found``."""
    from backend.indexer import get_vault_data

    data = get_vault_data(vault_name)
    if not data:
        raise ServiceError(
            f"Vault '{vault_name}' not found",
            code="not_found",
            status=404,
            details={"vault": vault_name},
        )
    return Path(data["path"])


def browse_directory(vault_name: str, path: str = "") -> dict[str, Any]:
    """Return the direct children of a vault directory (directories first)."""
    from backend.indexer import SUPPORTED_EXTENSIONS
    from backend.vault_settings import get_vault_setting

    root = get_vault_root(vault_name)
    target = resolve_safe_path(root, path) if path else root.resolve()

    if not target.exists():
        raise ServiceError(
            f"Path not found: {path}",
            code="not_found",
            status=404,
            details={"vault": vault_name, "path": path},
        )

    hide_hidden = (get_vault_setting(vault_name) or {}).get("hideHiddenFiles", False)

    items: list[dict[str, Any]] = []
    try:
        for entry in sorted(target.iterdir(), key=lambda e: (not e.is_dir(), e.name.lower())):
            if hide_hidden and entry.name.startswith("."):
                continue
            rel = str(entry.relative_to(root)).replace("\\", "/")
            if entry.is_dir():
                # Count only direct children (files and subdirs) for performance.
                try:
                    file_count = sum(
                        1 for child in entry.iterdir()
                        if (not hide_hidden or not child.name.startswith("."))
                        and (child.is_file() and (child.suffix.lower() in SUPPORTED_EXTENSIONS or child.name.lower() in ("dockerfile", "makefile"))
                             or child.is_dir())
                    )
                except PermissionError:
                    file_count = 0
                items.append({
                    "name": entry.name,
                    "path": rel,
                    "type": "directory",
                    "children_count": file_count,
                })
            elif entry.suffix.lower() in SUPPORTED_EXTENSIONS or entry.name.lower() in ("dockerfile", "makefile"):
                items.append({
                    "name": entry.name,
                    "path": rel,
                    "type": "file",
                    "size": entry.stat().st_size,
                    "extension": entry.suffix.lower(),
                })
    except PermissionError:
        raise ServiceError("Permission denied", code="permission_denied", status=403) from None

    return {"vault": vault_name, "path": path, "items": items}


def _indexed_tags(vault_name: str, rel_path: str) -> list[str]:
    """Tags of a file as stored in the search index (#158).

    Empty list when the file is not indexed yet (binary formats, index still
    building) — the listing itself comes from the filesystem, tags are a
    decoration used by the navigation page facets/filters.

    Args:
        vault_name: Vault name.
        rel_path: Path of the file relative to the vault root (``/`` separated).

    Returns:
        The file's tags, or an empty list when unknown.
    """
    try:
        from backend.search import get_inverted_index

        info = get_inverted_index().doc_info.get(f"{vault_name}::{rel_path}")
    except Exception:
        return []
    return list((info or {}).get("tags") or [])


def list_all_files(
    vault_name: str,
    dir: str = "",
    limit: int = 200,
    recursive: bool = True,
) -> dict[str, Any]:
    """List files in a vault directory sorted by modification time (newest first).

    Unlike :func:`browse_directory`, this returns files only (with metadata:
    size, mtime, extension) and can recurse into subdirectories. Hidden files
    and ignored directories are skipped.

    Args:
        vault_name: Name of the vault.
        dir: Relative directory path within the vault (empty = root).
        limit: Maximum number of files to return.
        recursive: Recurse into subdirectories when True.

    Returns:
        Dict with ``vault``, ``directory``, ``recursive``, ``count`` and
        ``files``.

    Raises:
        ServiceError: ``not_found`` (404) if the directory does not exist,
            ``permission_denied`` (403) if it cannot be read.
    """
    from datetime import datetime, timezone

    from backend.indexer import IGNORED_DIRS

    root = get_vault_root(vault_name)
    dir_path = resolve_safe_path(root, dir) if dir else root

    if not dir_path.exists() or not dir_path.is_dir():
        raise ServiceError(
            f"Directory not found: {dir}",
            code="not_found",
            status=404,
            details={"vault": vault_name, "path": dir},
        )

    files: list[dict[str, Any]] = []
    dir_prefix = (dir or "").strip("/")

    try:
        iterator = dir_path.rglob("*") if recursive else dir_path.iterdir()
        for entry in iterator:
            if not entry.is_file():
                continue
            if entry.name.startswith("."):
                continue
            if entry.name in IGNORED_DIRS:
                continue

            if recursive and dir_prefix:
                skip = False
                try:
                    for part in entry.relative_to(dir_path).parts[:-1]:
                        if part.startswith(".") or part in IGNORED_DIRS:
                            skip = True
                            break
                except ValueError:
                    pass
                if skip:
                    continue

            stat = entry.stat()
            rel_path = str(entry.relative_to(root)).replace("\\", "/")

            if recursive and dir_prefix:
                try:
                    rel_to_dir = str(entry.parent.relative_to(dir_path)).replace("\\", "/")
                except ValueError:
                    rel_to_dir = ""
            else:
                rel_to_dir = ""

            ext = entry.suffix.lower() if entry.suffix else ""

            file_entry = {
                "name": entry.name,
                "path": rel_path,
                "vault": vault_name,
                "size": stat.st_size,
                "modified": stat.st_mtime,
                "modified_iso": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
                "extension": ext.lstrip(".") if ext else "",
                "tags": _indexed_tags(vault_name, rel_path),
            }
            if rel_to_dir and rel_to_dir != ".":
                file_entry["rel_dir"] = rel_to_dir

            files.append(file_entry)
    except PermissionError:
        raise ServiceError(
            "Permission denied reading directory",
            code="permission_denied",
            status=403,
            details={"vault": vault_name, "path": dir},
        ) from None

    files.sort(key=lambda f: f["modified"], reverse=True)
    files = files[:limit]

    return {
        "vault": vault_name,
        "directory": dir,
        "recursive": recursive,
        "count": len(files),
        "files": files,
    }