webmail-client/backend/app/search/meili.py

# INTERFACE VERSION: 1
"""Meilisearch SearchBackend (SPEC §5.2, §8; M2).

Typo-tolerant free-text search with highlighted snippets over a single ``messages``
index. Every query is filtered by ``user_id`` (invariant I-7). Structured §8.2
operators that map to filterable attributes — ``is`` / ``has`` / ``in`` / date /
size — become Meilisearch filters; the field operators (``from`` / ``to`` / ``cc``
/ ``subject`` / ``filename``) fold into the query text, since Meilisearch 1.8 has
no substring filter (the PG backend remains available for exact operator queries).
"""

from __future__ import annotations

import asyncio
import base64
import datetime as dt
import re
from typing import Any
from uuid import UUID

from meilisearch_python_sdk import AsyncClient

from app.config import get_settings
from app.search.base import SearchBackend, SearchDoc, SearchHit, SearchQuery

_INDEX = "messages"

# Priority order matters: subject/senders rank above the body.
_SEARCHABLE = [
    "subject",
    "from_names",
    "from_emails",
    "to_names",
    "to_emails",
    "cc_names",
    "cc_emails",
    "body",
    "attachment_names",
]
_FILTERABLE = [
    "user_id",
    "account_id",
    "special",
    "has_attachments",
    "is_unread",
    "is_starred",
    "date",
    "size_bytes",
]
_TEXT_FIELD_OPS = {"from", "to", "cc", "subject", "filename"}
_RELATIVE_DAYS = {"d": 1, "w": 7, "m": 30, "y": 365}


def _emails(field: list[dict[str, str]]) -> list[str]:
    return [(a.get("email") or "").lower() for a in field if a.get("email")]


def _names(field: list[dict[str, str]]) -> list[str]:
    return [a.get("name") or "" for a in field if a.get("name")]


def _to_doc(doc: SearchDoc) -> dict[str, Any]:
    return {
        "id": str(doc.message_id),
        "thread_id": str(doc.thread_id),
        "user_id": str(doc.user_id),
        "account_id": str(doc.account_id),
        "special": doc.mailbox_special_use or "",
        "subject": doc.subject,
        "body": doc.body_text,
        "from_names": _names(doc.from_),
        "from_emails": _emails(doc.from_),
        "to_names": _names(doc.to),
        "to_emails": _emails(doc.to),
        "cc_names": _names(doc.cc),
        "cc_emails": _emails(doc.cc),
        "attachment_names": doc.attachment_names,
        "has_attachments": doc.has_attachments,
        "is_unread": doc.is_unread,
        "is_starred": doc.is_starred,
        "date": doc.date,
        "size_bytes": doc.size_bytes,
    }


def _parse_date_unix(value: str) -> int | None:
    try:
        d = dt.datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=dt.UTC)
    except ValueError:
        return None
    return int(d.timestamp())


def _parse_relative_days(value: str) -> int | None:
    m = re.fullmatch(r"(\d+)([dwmy])", value.strip().lower())
    return int(m.group(1)) * _RELATIVE_DAYS[m.group(2)] if m else None


def _parse_size(value: str) -> int | None:
    m = re.fullmatch(r"(\d+)([km])", value.strip().lower())
    if not m:
        return None
    return int(m.group(1)) * (1024 if m.group(2) == "k" else 1024 * 1024)


def _op_filter(key: str, value: str) -> str | None:
    """A single §8.2 operator -> a Meilisearch filter fragment (or None)."""
    now = int(dt.datetime.now(dt.UTC).timestamp())
    if key == "in":
        return f'special = "{value.lower()}"'
    if key == "is":
        v = value.lower()
        if v == "unread":
            return "is_unread = true"
        if v == "read":
            return "is_unread = false"
        if v == "starred":
            return "is_starred = true"
        return None  # 'answered' is not indexed in SearchDoc
    if key == "has":
        return "has_attachments = true" if value.lower() == "attachment" else None
    if key == "before":
        u = _parse_date_unix(value)
        return f"date < {u}" if u is not None else None
    if key == "after":
        u = _parse_date_unix(value)
        return f"date >= {u}" if u is not None else None
    if key == "newer_than":
        days = _parse_relative_days(value)
        return f"date >= {now - days * 86400}" if days is not None else None
    if key == "older_than":
        days = _parse_relative_days(value)
        return f"date < {now - days * 86400}" if days is not None else None
    if key == "larger":
        n = _parse_size(value)
        return f"size_bytes > {n}" if n is not None else None
    if key == "smaller":
        n = _parse_size(value)
        return f"size_bytes < {n}" if n is not None else None
    return None


def _build_filter(groups: list[list[dict[str, Any]]], user_id: UUID) -> str:
    ands = [f'user_id = "{user_id}"']
    for group in groups:
        ors: list[str] = []
        for clause in group:
            if clause.get("type") == "text":
                continue
            frag = _op_filter(str(clause.get("key")), str(clause.get("value", "")))
            if frag is None:
                continue
            ors.append(f"NOT ({frag})" if clause.get("negate") else frag)
        if ors:
            ands.append("(" + " OR ".join(ors) + ")")
    return " AND ".join(ands)


def _query_text(groups: list[list[dict[str, Any]]]) -> str:
    """Free text for Meili's `q`: text terms (negated as -term) + folded field ops."""
    terms: list[str] = []
    for group in groups:
        for clause in group:
            value = str(clause.get("value", "")).strip()
            if not value:
                continue
            if clause.get("type") == "text":
                terms.append(f"-{value}" if clause.get("negate") else value)
            elif clause.get("key") in _TEXT_FIELD_OPS and not clause.get("negate"):
                terms.append(value)
    return " ".join(terms).strip()


class MeiliSearchBackend(SearchBackend):
    def __init__(self) -> None:
        cfg = get_settings()
        self._client = AsyncClient(cfg.meili_url, cfg.meili_key or None)
        self._index = self._client.index(_INDEX)
        self._ready = False
        self._lock = asyncio.Lock()

    async def _ensure(self) -> None:
        """Create the index and apply settings once (idempotent)."""
        if self._ready:
            return
        async with self._lock:
            if self._ready:
                return
            indexes = await self._client.get_indexes()
            names = {i.uid for i in indexes} if indexes else set()
            if _INDEX not in names:
                await self._client.create_index(_INDEX, primary_key="id")
            await self._index.update_searchable_attributes(_SEARCHABLE)
            await self._index.update_sortable_attributes(["date"])
            task = await self._index.update_filterable_attributes(_FILTERABLE)
            await self._client.wait_for_task(task.task_uid)
            self._ready = True

    async def index(self, doc: SearchDoc) -> None:
        await self._ensure()
        await self._index.add_documents([_to_doc(doc)])

    async def delete(self, message_id: UUID) -> None:
        await self._ensure()
        await self._index.delete_document(str(message_id))

    async def delete_user(self, user_id: UUID) -> None:
        await self._ensure()
        await self._index.delete_documents_by_filter(f'user_id = "{user_id}"')

    async def query(
        self, q: SearchQuery, *, limit: int, cursor: str | None
    ) -> tuple[list[SearchHit], str | None]:
        await self._ensure()
        groups = q.filters.get("groups", []) if isinstance(q.filters, dict) else []
        assert isinstance(groups, list)
        offset = _decode_cursor(cursor)
        res = await self._index.search(
            _query_text(groups),
            filter=_build_filter(groups, q.user_id),
            offset=offset,
            limit=limit + 1,
            sort=["date:desc"],  # stable order for offset paging; typo matching still applies
            attributes_to_retrieve=["id", "thread_id"],
            attributes_to_highlight=["subject", "body"],
            attributes_to_crop=["body"],
            crop_length=30,
            highlight_pre_tag="<mark>",
            highlight_post_tag="</mark>",
            show_ranking_score=True,
        )
        rows = res.hits
        has_more = len(rows) > limit
        rows = rows[:limit]
        hits = [_to_hit(row) for row in rows]
        next_cursor = _encode_cursor(offset + limit) if has_more else None
        return hits, next_cursor


def _to_hit(row: dict[str, Any]) -> SearchHit:
    fmt = row.get("_formatted") or {}
    highlights: dict[str, str] = {}
    subject = fmt.get("subject") or ""
    if "<mark>" in subject:
        highlights["subject"] = subject
    body = fmt.get("body") or ""
    if "<mark>" in body:
        highlights["body"] = body
    return SearchHit(
        message_id=UUID(row["id"]),
        thread_id=UUID(row["thread_id"]),
        score=float(row.get("_rankingScore") or 0.0),
        highlights=highlights,
    )


def _encode_cursor(offset: int) -> str:
    return base64.urlsafe_b64encode(str(offset).encode()).decode()


def _decode_cursor(cursor: str | None) -> int:
    if not cursor:
        return 0
    try:
        return max(0, int(base64.urlsafe_b64decode(cursor.encode()).decode()))
    except (ValueError, TypeError):
        return 0