feat(backend,frontend): search everywhere in buttons, links and message data, a jump from search or an alert lights up the found words and buttons

backend/src/utils/search/found.py

@@ -0,0 +1,65 @@
+import re
+from collections.abc import Iterator
+from typing import Any
+
+from utils.search.models import Found, FoundWhere
+
+HIDDEN_KEYS = ("chat", "from_user", "sender_chat")
+SKIPPED_KEYS = ("_", "file_id", "file_unique_id")
+ENUM = re.compile(r"^[A-Z][A-Za-z]+\.[A-Z_]+$")
+
+_HEADS: dict[str, FoundWhere] = {
+    "text": "text",
+    "caption": "text",
+    "web_page": "preview",
+}
+_LEAVES: dict[str, FoundWhere] = {"url": "link", "file_name": "file"}
+
+type Path = tuple[str | int, ...]
+
+
+def _scalars(node: object, path: Path = ()) -> Iterator[tuple[Path, str, object]]:
+    if isinstance(node, dict):
+        items = ((k, v) for k, v in node.items() if k not in SKIPPED_KEYS)
+    elif isinstance(node, list):
+        items = enumerate(node)
+    else:
+        return
+    for key, value in items:
+        if isinstance(value, dict | list):
+            yield from _scalars(value, (*path, key))
+        elif isinstance(value, str | int | float) and not isinstance(value, bool):
+            yield (*path, key), str(value), node
+
+
+def _where(path: Path) -> FoundWhere:
+    head, leaf = str(path[0]), str(path[-1])
+    if head == "reply_markup":
+        return "button"
+    if leaf in _LEAVES:
+        return _LEAVES[leaf]
+    if head.startswith("forward"):
+        return "forward"
+    if head.startswith("reply_to"):
+        return "reply"
+    return _HEADS.get(head, "data")
+
+
+def find(raw: dict[str, Any], needle: str) -> Found | None:
+    folded = needle.casefold()
+    fallback = None
+    for path, value, owner in _scalars(raw):
+        if folded not in value.casefold() or ENUM.match(value):
+            continue
+        where = _where(path)
+        button = owner.get("text") if isinstance(owner, dict) else None
+        found = Found(
+            where=where,
+            path=".".join(map(str, path)),
+            value=value,
+            button=button if where == "button" else None,
+        )
+        if where != "text":
+            return found
+        fallback = fallback or found
+    return fallback

backend/src/utils/search/models.py

@@ -10,7 +10,10 @@ _EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
 _MICROSECOND = timedelta(microseconds=1)
 
 ChatKind = Literal["private", "group", "channel"]
-HitSource = Literal["text", "stt", "version"]
+HitSource = Literal["text", "stt", "version", "data"]
+FoundWhere = Literal[
+    "text", "button", "link", "forward", "reply", "file", "preview", "data"
+]
 
 
 def encode_cursor(date: datetime, chat_id: int, message_id: int) -> str:
@@ -34,6 +37,7 @@ class SearchFilters(BaseModel):
     date_from: datetime | None = None
     date_to: datetime | None = None
     regex: str | None = None
+    everywhere: bool = False
     cursor: str | None = None
     limit: int = DEFAULT_LIMIT
 
@@ -45,6 +49,13 @@ class SearchFilters(BaseModel):
         return value
 
 
+class Found(BaseModel):
+    where: FoundWhere
+    path: str
+    value: str
+    button: str | None = None
+
+
 class SearchHit(BaseModel):
     chat_id: int
     message_id: int
@@ -56,12 +67,14 @@ class SearchHit(BaseModel):
     edited_at: datetime | None
     deleted_at: datetime | None
     source: HitSource
-    extracted_text: str | None
-    version_text: str | None
-    version_at: datetime | None
+    extracted_text: str | None = None
+    version_text: str | None = None
+    version_at: datetime | None = None
+    found: Found | None = None
 
 
 class SearchPage(BaseModel):
     hits: list[SearchHit]
     next_cursor: str | None
     terms: list[str]
+    scanned_to: datetime | None = None

backend/src/utils/search/repository.py

@@ -1,5 +1,12 @@
+import json
+import re
+from datetime import UTC, datetime, timedelta
+from time import monotonic
+
 import asyncpg
 
+from utils.read.message_view import load_raw
+from utils.search.found import ENUM, HIDDEN_KEYS, SKIPPED_KEYS, find
 from utils.search.models import (
     MAX_LIMIT,
     ChatKind,
@@ -10,6 +17,12 @@ from utils.search.models import (
     encode_cursor,
 )
 
+MIN_NEEDLE = 3
+_WINDOW = timedelta(days=7)
+_BUDGET_SECONDS = 2.5
+_BOTTOM = -(2**63)
+_LIKE_SPECIAL = re.compile(r"([\\%_])")
+
 _PARSE = """
 SELECT (regexp_replace(websearch_to_tsquery('russian', q)::text,
                        '''((?:[^'']|''''){3,})''', '''\\1'':*', 'g')::tsquery
@@ -76,6 +89,24 @@ _WITH_MEDIA = (
 
 _SOURCES = ("text", "stt", "version")
 
+_OLDEST = "SELECT min(date) FROM message_search WHERE account_id = $1"
+
+_VALUE_MATCH = (
+    'strict $.** ? (@.type() == "object").keyvalue() ? ('
+    + "".join(f'@.key != "{key}" && ' for key in SKIPPED_KEYS)
+    + '(@.value.type() == "string" || @.value.type() == "number") && '
+    f"!(@.value.string() like_regex {json.dumps(ENUM.pattern)}) && "
+    '@.value.string() like_regex {needle} flag "iq")'
+)
+
+_EVERYWHERE = (
+    f"SELECT {_COLS}, s.raw - {{hidden}}::text[] AS raw "  # noqa: S608
+    "FROM messages s WHERE {where} "
+    "AND CASE WHEN s.raw::text ILIKE {like} "
+    "THEN jsonb_path_exists(s.raw - {hidden}::text[], {path}::jsonpath) "
+    f"ELSE false END ORDER BY {_KEY_ORDER} LIMIT {{limit}}"
+)
+
 
 class _Binder:
     def __init__(self) -> None:
@@ -130,9 +161,72 @@ def _branch(sql: str, limit: str, order: str = _KEY_ORDER, **parts: str) -> str:
 async def search_messages(pool: asyncpg.Pool, filters: SearchFilters) -> SearchPage:
     async with pool.acquire() as conn, conn.transaction():
         await conn.execute("SET LOCAL jit = off")
+        if filters.everywhere:
+            return await _everywhere(conn, filters)
         return await _search(conn, filters)
 
 
+async def _channels(conn: asyncpg.Connection, filters: SearchFilters) -> list[int]:
+    if filters.kind in {"group", "channel"}:
+        return await conn.fetchval(_CHANNELS, filters.account_id)
+    return []
+
+
+def _data_hit(row: asyncpg.Record, needle: str) -> SearchHit:
+    data = dict(row)
+    raw = load_raw(data.pop("raw"))
+    return SearchHit(**data, source="data", found=find(raw, needle))
+
+
+async def _everywhere(conn: asyncpg.Connection, filters: SearchFilters) -> SearchPage:
+    needle = (filters.query or "").strip()
+    floor = filters.date_from or await conn.fetchval(_OLDEST, filters.account_id)
+    if len(needle) < MIN_NEEDLE or floor is None:
+        return SearchPage(hits=[], next_cursor=None, terms=[])
+    channels = await _channels(conn, filters)
+    quoted = json.dumps(needle, ensure_ascii=False)
+    like = _LIKE_SPECIAL.sub(r"\\\1", quoted[1:-1])
+    path = _VALUE_MATCH.format(needle=quoted)
+    size = min(filters.limit, MAX_LIMIT)
+    cursor = filters.cursor
+    top = decode_cursor(cursor)[0] if cursor else None
+    hits: list[SearchHit] = []
+    started = monotonic()
+    while True:
+        lower = max((top or filters.date_to or datetime.now(UTC)) - _WINDOW, floor)
+        bind = _Binder()
+        conds = _conditions(
+            filters.model_copy(update={"cursor": cursor}), channels, bind
+        )
+        conds.append(f"s.date >= {bind(lower)}")
+        if top is not None:
+            conds.append(f"s.date <= {bind(top)}")
+        limit = size - len(hits)
+        sql = _EVERYWHERE.format(
+            where=" AND ".join(conds),
+            hidden=bind(list(HIDDEN_KEYS)),
+            like=bind(f"%{like}%"),
+            path=bind(path),
+            limit=bind(limit),
+        )
+        account = bind(filters.account_id)
+        rows = await conn.fetch(
+            _WITH_MEDIA.format(hits=sql, account=account), *bind.params
+        )
+        hits += [_data_hit(row, needle) for row in rows]
+        if len(rows) >= limit:
+            last = rows[-1]
+            top = last["date"]
+            cursor = encode_cursor(top, last["chat_id"], last["message_id"])
+        elif lower > floor:
+            top = lower
+            cursor = encode_cursor(lower, _BOTTOM, _BOTTOM)
+        else:
+            return SearchPage(hits=hits, next_cursor=None, terms=[], scanned_to=floor)
+        if len(hits) >= size or monotonic() - started > _BUDGET_SECONDS:
+            return SearchPage(hits=hits, next_cursor=cursor, terms=[], scanned_to=top)
+
+
 async def _search(conn: asyncpg.Connection, filters: SearchFilters) -> SearchPage:
     query = (filters.query or "").strip()
     terms: list[str] = []
@@ -144,9 +238,7 @@ async def _search(conn: asyncpg.Connection, filters: SearchFilters) -> SearchPag
             return SearchPage(hits=[], next_cursor=None, terms=terms)
         tsquery = parsed["tsquery"]
 
-    channels: list[int] = []
-    if filters.kind in {"group", "channel"}:
-        channels = await conn.fetchval(_CHANNELS, filters.account_id)
+    channels = await _channels(conn, filters)
     bind = _Binder()
     where = " AND ".join(_conditions(filters, channels, bind))
     size = min(filters.limit, MAX_LIMIT)