# 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="", highlight_post_tag="", 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 "" in subject: highlights["subject"] = subject body = fmt.get("body") or "" if "" 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