ObsiGate/backend/file_chat.py

# backend/file_chat.py — historique de discussion (#169, #190)
"""Chat history persisted under ``data/chats/``.

One JSON document per ``(vault, path)`` pair, keyed by a SHA-256 of both so
the filename never carries user-controlled path separators. Writes are
atomic (tmp + move) and the message list is capped at
:data:`MAX_MESSAGES` to bound growth.

#190 adds the **general chat**: the same store addressed with the reserved
sentinels (:data:`GLOBAL_VAULT` / :data:`GLOBAL_PATH`), so no second
implementation. Messages may carry an ``attachment`` (image/video/url)
uploaded under ``data/chat_uploads/``.
"""

from __future__ import annotations

import hashlib
import html
import json
import logging
import re
import shutil
import threading
import time
import uuid
from pathlib import Path
from typing import Any
from urllib.parse import urljoin, urlparse

import httpx

from backend.render import _render_markdown

logger = logging.getLogger("obsigate.file_chat")

CHAT_DIR = Path("data/chats")
MAX_MESSAGES = 500  # retention ceiling per file (oldest dropped first)
MAX_TEXT = 4000  # characters per message

# ponytail: global lock over read-modify-write — chat writes are HTTP-only and
# serialized anyway, this just makes losing a message to a future thread (or a
# watcher hook) impossible; per-file locks if it ever becomes contended.
_LOCK = threading.RLock()

# #190 — general (non file-bound) conversation, stored like any other one.
GLOBAL_VAULT = "__global__"
GLOBAL_PATH = "general"

# #191 — private (2 users) conversations reuse the same store: the vault is
# the reserved sentinel and the path is the sorted username pair, so the
# storage key never depends on who asks.
DM_VAULT = "__dm__"

# #190 — attachments (image/video) live outside the vaults.
UPLOAD_DIR = Path("data/chat_uploads")
MAX_UPLOAD_BYTES = 25 * 1024 * 1024  # 25 MB per attachment
ALLOWED_ATTACH_EXT = {
    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".svg",
    ".mp4", ".webm", ".ogg", ".mov", ".m4v",
}

# #191 — link preview fetch budget
PREVIEW_TIMEOUT = 5.0  # seconds
PREVIEW_MAX_BYTES = 512 * 1024  # only the head of the page is parsed
PREVIEW_IMAGE_MAX = 2 * 1024 * 1024  # ponytail: 2 MB ceiling on a thumbnail


def _chat_file(vault: str, path: str) -> Path:
    """Return the chat file for *(vault, path)* (hashed, traversal-proof)."""
    CHAT_DIR.mkdir(parents=True, exist_ok=True)
    key = hashlib.sha256(f"{vault}\0{path}".encode()).hexdigest()[:32]
    return CHAT_DIR / f"{key}.json"


def _read(vault: str, path: str) -> dict[str, Any]:
    """Load the raw chat document (empty structure when missing/corrupt)."""
    file = _chat_file(vault, path)
    if not file.exists():
        return {"vault": vault, "path": path, "messages": []}
    try:
        doc = json.loads(file.read_text(encoding="utf-8"))
        if not isinstance(doc.get("messages"), list):
            raise TypeError("messages is not a list")  # caught by the handler below
        return doc
    except Exception as e:
        logger.error("Failed to read chat for %s/%s: %s", vault, path, e)
        return {"vault": vault, "path": path, "messages": []}


def _write(file: Path, doc: dict[str, Any]) -> None:
    """Atomically persist *doc* (tmp file + rename)."""
    try:
        tmp = file.with_suffix(".tmp")
        tmp.write_text(json.dumps(doc, ensure_ascii=False, indent=1), encoding="utf-8")
        shutil.move(str(tmp), str(file))
    except Exception as e:
        logger.error("Failed to write chat %s: %s", file.name, e)


def get_messages(vault: str, path: str) -> list[dict[str, Any]]:
    """Return the chat history for *(vault, path)* (chronological).

    Every message carries its rendered ``html`` (#193): same markdown pipeline
    as a document (mistune + sanitizer), computed on read so a template change
    applies to the whole history without rewriting the JSON store.
    """
    return [_decorate(m, vault) for m in _read(vault, path).get("messages", [])]


def _decorate(msg: dict[str, Any], vault: str) -> dict[str, Any]:
    """Return a copy of *msg* with its sanitized markdown ``html`` (#193).

    The stored message is left untouched (``html`` is never persisted). A
    rendering failure must never break the chat: the message goes out with an
    empty ``html`` and the client falls back to plain text.
    """
    try:
        html = _render_markdown(msg.get("text", ""), vault)
    except Exception as e:  # pragma: no cover - defensive
        logger.warning("chat markdown rendering failed: %s", e)
        html = ""
    return {**msg, "html": html}


def add_message(
    vault: str,
    path: str,
    user: str,
    text: str,
    attachment: dict[str, Any] | None = None,
    preview: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Append a message and persist it. Returns the stored message.

    The list is capped at :data:`MAX_MESSAGES` (oldest dropped first).
    *attachment* (#190) is ``{name, url, mime, kind}`` for image/video/url;
    *preview* (#191) is the OpenGraph card of the first URL in *text*.
    """
    text = (text or "").strip()[:MAX_TEXT]
    msg: dict[str, Any] = {
        "id": uuid.uuid4().hex[:12],
        "user": user or "anonyme",
        "text": text,
        "ts": time.time(),
    }
    if attachment:
        msg["attachment"] = attachment
    if preview:
        msg["preview"] = preview
    return _append(vault, path, msg)


def _append(
    vault: str,
    path: str,
    msg: dict[str, Any],
) -> dict[str, Any]:
    """Cap, persist and return *msg* (shared by file, global and DM chats).

    The read-modify-write of the whole document happens under ``_LOCK`` so a
    concurrent writer can never drop a message (same class of bug as
    BUG-029 on ``users.json``).
    """
    with _LOCK:
        doc = _read(vault, path)
        messages = list(doc.get("messages", []))
        messages.append(msg)
        if len(messages) > MAX_MESSAGES:
            messages = messages[-MAX_MESSAGES:]
        doc["messages"] = messages
        _write(_chat_file(vault, path), doc)
    return _decorate(msg, vault)  # #193 — le html part avec l'écho SSE


# --- #192 : accusé de réception ---------------------------------------------

def get_read(vault: str, path: str) -> dict[str, float]:
    """Return ``{username: last_read_ts}`` for a conversation (#192)."""
    return {str(u): float(ts) for u, ts in (_read(vault, path).get("read") or {}).items()}


def mark_read(vault: str, path: str, user: str) -> dict[str, float]:
    """Record that *user* has seen the conversation (#192).

    Returns the whole read map so the caller can broadcast it on SSE: a
    sender learns their messages were received as soon as the recipient
    displays the conversation.
    """
    if not user:
        return get_read(vault, path)
    with _LOCK:
        doc = _read(vault, path)
        read = {str(u): float(ts) for u, ts in (doc.get("read") or {}).items()}
        read[user] = time.time()
        doc["read"] = read
        _write(_chat_file(vault, path), doc)
        return read


# --- #191 : messages privés (2 utilisateurs) -------------------------------

def dm_path(user_a: str, user_b: str) -> str:
    """Storage path for the private conversation between two users.

    The pair is sorted so both participants address the same document.
    """
    return "|".join(sorted([user_a, user_b]))


def get_dm_messages(user_a: str, user_b: str) -> list[dict[str, Any]]:
    """Return the private history between two users (chronological)."""
    return get_messages(DM_VAULT, dm_path(user_a, user_b))


def add_dm_message(
    user_a: str,
    user_b: str,
    author: str,
    text: str,
    attachment: dict[str, Any] | None = None,
    preview: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Append a private message. Returns the stored message."""
    return add_message(DM_VAULT, dm_path(user_a, user_b), author, text, attachment, preview)


# --- #191 : suppression -----------------------------------------------------

def delete_message(vault: str, path: str, message_id: str) -> bool:
    """Remove one message from a conversation. True when it existed."""
    with _LOCK:
        doc = _read(vault, path)
        messages = list(doc.get("messages", []))
        kept = [m for m in messages if m.get("id") != message_id]
        if len(kept) == len(messages):
            return False
        doc["messages"] = kept
        _write(_chat_file(vault, path), doc)
    return True


# --- #190 : chat général (conversation centrale, hors fichier) -------------

def get_global_messages() -> list[dict[str, Any]]:
    """Return the general-chat history (chronological)."""
    return get_messages(GLOBAL_VAULT, GLOBAL_PATH)


def add_global_message(
    user: str,
    text: str,
    attachment: dict[str, Any] | None = None,
    preview: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Append a message to the general chat. Returns the stored message."""
    return add_message(GLOBAL_VAULT, GLOBAL_PATH, user, text, attachment, preview)


def save_attachment(filename: str, data: bytes) -> dict[str, Any]:
    """Persist an uploaded attachment under :data:`UPLOAD_DIR`.

    Returns ``{name, url, mime, kind}``. The stored name is a fresh UUID
    (never the client name), the extension must be in
    :data:`ALLOWED_ATTACH_EXT` and the size is capped at
    :data:`MAX_UPLOAD_BYTES`.

    Raises:
        ValueError: extension refused, empty file or size exceeded.
    """
    ext = Path(filename or "").suffix.lower()
    if ext not in ALLOWED_ATTACH_EXT:
        raise ValueError(f"extension refusée : {ext or '(aucune)'}")
    if not data:
        raise ValueError("fichier vide")
    if len(data) > MAX_UPLOAD_BYTES:
        raise ValueError(f"fichier trop lourd (max {MAX_UPLOAD_BYTES // (1024 * 1024)} MB)")
    UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
    name = f"{uuid.uuid4().hex}{ext}"
    (UPLOAD_DIR / name).write_bytes(data)
    kind = "video" if ext in {".mp4", ".webm", ".ogg", ".mov", ".m4v"} else "image"
    return {
        "name": name,
        "url": f"/api/chat/attachment/{name}",
        "mime": _MIME_BY_EXT.get(ext, "application/octet-stream"),
        "kind": kind,
    }


# --- #191 : link preview ----------------------------------------------------

_URL_RE = re.compile(r"https?://[^\s<>\"']+")
_PREVIEW_CACHE: dict[str, dict[str, Any] | None] = {}
PREVIEW_CACHE_MAX = 200


def _og(content: str, prop: str) -> str:
    """Extract one OpenGraph/``<title>`` value from an HTML head (regex)."""
    for pattern in (
        rf'<meta[^>]+(?:property|name)="{prop}"[^>]+content="([^"]*)"',
        rf'<meta[^>]+content="([^"]*)"[^>]+(?:property|name)="{prop}"',
    ):
        m = re.search(pattern, content, re.IGNORECASE)
        if m:
            return html.unescape(m.group(1)).strip()[:300]
    if prop == "og:title":
        m = re.search(r"<title[^>]*>([^<]*)</title>", content, re.IGNORECASE)
        if m:
            return html.unescape(m.group(1)).strip()[:300]
    return ""


# BUG-109 — content-type → extension (the attachment allow-list decides).
_IMG_EXT_BY_MIME = {
    "image/png": ".png",
    "image/jpeg": ".jpg",
    "image/gif": ".gif",
    "image/webp": ".webp",
    "image/svg+xml": ".svg",
}


def _proxy_image(img_url: str, page_url: str) -> str:
    """Download *img_url* into ``chat_uploads`` and return a same-origin URL.

    The response CSP is ``img-src 'self' data: blob:``: a remote ``og:image``
    would be blocked by the browser (BUG-109). Relative and protocol-relative
    values are resolved against *page_url* first. Raises ``ValueError`` /
    ``SSRFError`` on any failure — the caller keeps the card and drops only
    the thumbnail.
    """
    from backend.tools.web import USER_AGENT, _assert_public_http_url

    full = urljoin(page_url, img_url)
    _assert_public_http_url(full)
    resp = httpx.get(
        full,
        headers={"User-Agent": USER_AGENT},
        timeout=PREVIEW_TIMEOUT,
        follow_redirects=True,
    )
    if resp.status_code >= 400:
        raise ValueError(f"HTTP {resp.status_code}")
    data = resp.content
    if not data or len(data) > PREVIEW_IMAGE_MAX:
        raise ValueError("image vide ou trop lourde")
    mime = (resp.headers.get("content-type") or "").split(";")[0].strip().lower()
    ext = _IMG_EXT_BY_MIME.get(mime) or Path(urlparse(full).path).suffix.lower()
    # save_attachment(): allow-list d'extensions + nom UUID (jamais le nom distant)
    return str(save_attachment(f"preview{ext}", data)["url"])


def build_preview(text: str) -> dict[str, Any] | None:
    """Fetch OpenGraph metadata for the first URL in *text* (#191).

    SSRF-guarded (reuses the web-tool guard), size/time capped, cached in a
    bounded dict. Returns ``{url, title, description, image, site}`` or
    ``None`` when there is no URL / the fetch fails (never raises: a dead
    link must not block the message). ``image`` is a **same-origin**
    ``/api/chat/attachment/...`` URL (BUG-109), empty when the thumbnail
    could not be fetched.
    """
    m = _URL_RE.search(text or "")
    if not m:
        return None
    url = m.group(0).rstrip(".,;:!?)")
    if url in _PREVIEW_CACHE:
        cached = _PREVIEW_CACHE[url]
        return dict(cached) if cached else None
    try:
        from backend.tools.web import USER_AGENT, _assert_public_http_url

        _assert_public_http_url(url)
        resp = httpx.get(
            url,
            headers={"User-Agent": USER_AGENT, "Accept": "text/html,*/*"},
            timeout=PREVIEW_TIMEOUT,
            follow_redirects=True,
        )
        if resp.status_code >= 400:
            raise ValueError(f"HTTP {resp.status_code}")
        body = resp.text[:PREVIEW_MAX_BYTES]
        # BUG-109 : vignette téléchargée côté serveur — une image distante
        # échouerait à la CSP. Échec isolé = carte sans vignette.
        image = _og(body, "og:image")
        try:
            image = _proxy_image(image, url) if image else ""
        except Exception as ie:
            logger.debug("preview image failed for %s: %s", url, ie)
            image = ""
        preview = {
            "url": url,
            "title": _og(body, "og:title") or _og(body, "og:site_name"),
            "description": _og(body, "og:description"),
            "image": image,
            "site": _og(body, "og:site_name") or (url.split("/")[2] if "/" in url[8:] else url),
        }
        if not preview["title"]:
            raise ValueError("pas de titre")
    except Exception as e:
        logger.debug("link preview failed for %s: %s", url, e)
        preview = None
    if len(_PREVIEW_CACHE) >= PREVIEW_CACHE_MAX:
        _PREVIEW_CACHE.pop(next(iter(_PREVIEW_CACHE)))  # oldest first (dict order)
    _PREVIEW_CACHE[url] = preview
    return dict(preview) if preview else None


def attachment_path(name: str) -> Path | None:
    """Resolve an attachment by its stored name (UUID+ext only, no traversal)."""
    p = Path(name)
    if p.name != name or p.suffix.lower() not in ALLOWED_ATTACH_EXT:
        return None
    file = UPLOAD_DIR / p.name
    return file if file.exists() else None


# Extension → MIME (literals only; ``mimetypes`` guesses poorly for a few).
_MIME_BY_EXT = {
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
    ".gif": "image/gif",
    ".webp": "image/webp",
    ".svg": "image/svg+xml",
    ".mp4": "video/mp4",
    ".webm": "video/webm",
    ".ogg": "video/ogg",
    ".mov": "video/quicktime",
    ".m4v": "video/x-m4v",
}