feat(api,userbot,frontend): search peers without chats and start tracking them

backend/src/api/app.py

@@ -20,6 +20,7 @@ from api.routers import (
     backfill,
     chats,
     custom_emoji,
+    discover,
     events,
     folders,
     media,
@@ -83,6 +84,7 @@ app.include_router(stories.router)
 app.include_router(profile.router)
 app.include_router(events.router)
 app.include_router(peers.router)
+app.include_router(discover.router)
 app.include_router(annotations.router)
 app.include_router(watches.router)
 

backend/src/api/routers/discover.py

@@ -0,0 +1,116 @@
+import asyncio
+import json
+from typing import Annotated
+
+import asyncpg
+from dishka.integrations.fastapi import DishkaRoute, FromDishka
+from fastapi import APIRouter, Query
+from pydantic import BaseModel
+
+from api.routers.policy import POLICY_CHANGED_CHANNEL
+from utils.jobs import enqueue
+from utils.policy import repository as policy_repository
+from utils.policy.defaults import TRACKING
+from utils.policy.models import ScopeType
+from utils.read import discover
+from utils.read.models import DiscoverItem
+
+router = APIRouter(prefix="/api", tags=["discover"], route_class=DishkaRoute)
+
+DEFAULT_LIMIT = 30
+REMOTE_TIMEOUT_SECONDS = 20.0
+POLL_INTERVAL_SECONDS = 0.2
+FINISHED = ("done", "failed", "canceled")
+
+_CHAT_POLICY_ID = """
+SELECT id FROM capture_policy
+WHERE account_id = $1 AND scope_type = 'chat' AND scope_id = $2
+"""
+
+AccountId = Annotated[int, Query()]
+
+
+class TrackRequest(BaseModel):
+    account_id: int
+    backfill: bool = True
+
+
+class SyncContactsRequest(BaseModel):
+    account_id: int
+
+
+async def _remote_ids(
+    pool: asyncpg.Pool, account_id: int, query: str, limit: int
+) -> list[int]:
+    job_id = await enqueue(
+        pool, account_id, "search_peers", {"query": query, "limit": limit}
+    )
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + REMOTE_TIMEOUT_SECONDS
+    while loop.time() < deadline:
+        await asyncio.sleep(POLL_INTERVAL_SECONDS)
+        row = await pool.fetchrow(
+            "SELECT status, progress FROM jobs WHERE id = $1", job_id
+        )
+        if row is not None and row["status"] in FINISHED:
+            await pool.execute("DELETE FROM jobs WHERE id = $1", job_id)
+            return json.loads(row["progress"]).get("ids", [])
+    return []
+
+
+@router.get("/discover")
+async def discover_peers(
+    pool: FromDishka[asyncpg.Pool],
+    account_id: AccountId,
+    query: Annotated[str, Query()] = "",
+    remote: Annotated[bool, Query()] = False,
+    limit: Annotated[int, Query()] = DEFAULT_LIMIT,
+) -> list[DiscoverItem]:
+    if not query.strip():
+        return []
+    items = await discover.search(pool, account_id, query, limit)
+    if not remote:
+        return items
+    known = {item.chat_id for item in items}
+    ids = await _remote_ids(pool, account_id, query, limit)
+    extra = await discover.by_ids(
+        pool, account_id, [chat_id for chat_id in ids if chat_id not in known]
+    )
+    return [*items, *extra]
+
+
+@router.get("/discover/{chat_id}")
+async def discover_chat(
+    pool: FromDishka[asyncpg.Pool], chat_id: int, account_id: AccountId
+) -> DiscoverItem:
+    return await discover.get_item(pool, account_id, chat_id)
+
+
+@router.post("/chats/{chat_id}/track", status_code=201)
+async def track_chat(
+    pool: FromDishka[asyncpg.Pool], chat_id: int, body: TrackRequest
+) -> DiscoverItem:
+    kind = await discover.chat_kind(pool, body.account_id, chat_id)
+    toggles = TRACKING[kind]
+    policy_id = await pool.fetchval(_CHAT_POLICY_ID, body.account_id, chat_id)
+    if policy_id is None:
+        await policy_repository.create_policy(
+            pool, body.account_id, ScopeType.CHAT, chat_id, toggles
+        )
+    else:
+        await policy_repository.update_policy(pool, policy_id, toggles)
+    await pool.execute(f"NOTIFY {POLICY_CHANGED_CHANNEL}")
+    await enqueue(pool, body.account_id, "enrich_chat", {"chat_id": chat_id})
+    if body.backfill:
+        await enqueue(
+            pool, body.account_id, "backfill", {"chat_id": chat_id, "media": True}
+        )
+    return await discover.get_item(pool, body.account_id, chat_id)
+
+
+@router.post("/contacts/sync", status_code=201)
+async def sync_contacts(
+    pool: FromDishka[asyncpg.Pool], body: SyncContactsRequest
+) -> dict[str, int]:
+    job_id = await enqueue(pool, body.account_id, "sync_contacts", {})
+    return {"job_id": job_id}

backend/src/userbot/modules/jobs/handlers/__init__.py

@@ -4,6 +4,8 @@ from userbot.modules.jobs.handlers import (
     fetch_avatar,
     fetch_custom_emoji,
     fetch_media,
+    search_peers,
+    sync_contacts,
     sync_dialogs,
     transcribe,
 )
@@ -14,6 +16,8 @@ __all__ = [
     "fetch_avatar",
     "fetch_custom_emoji",
     "fetch_media",
+    "search_peers",
+    "sync_contacts",
     "sync_dialogs",
     "transcribe",
 ]

backend/src/userbot/modules/jobs/handlers/search_peers.py

@@ -0,0 +1,102 @@
+import re
+
+from pyrogram import Client, raw
+from pyrogram.errors import BadRequest, Forbidden
+from pyrogram.types import Chat
+
+from userbot.modules.capture.context import CaptureContext
+from userbot.modules.jobs.context import JobContext
+from userbot.modules.jobs.registry import register
+from userbot.modules.profiles.snapshots import save_chat
+
+DEFAULT_LIMIT = 30
+_USERNAME = re.compile(r"^[a-z][a-z0-9_]{3,31}$", re.IGNORECASE)
+_PREFIXES = ("https://t.me/", "http://t.me/", "t.me/", "@")
+
+
+def _normalize(query: str) -> str:
+    text = query.strip()
+    for prefix in _PREFIXES:
+        if text.lower().startswith(prefix):
+            text = text[len(prefix) :]
+            break
+    return text.strip("/")
+
+
+_SOURCE_TYPES = (raw.types.User, raw.types.Chat, raw.types.Channel)
+
+
+def _source(
+    peer: raw.base.Peer, users: dict, chats: dict
+) -> raw.types.User | raw.types.Chat | raw.types.Channel | None:
+    if isinstance(peer, raw.types.PeerUser):
+        source = users.get(peer.user_id)
+    elif isinstance(peer, raw.types.PeerChannel):
+        source = chats.get(peer.channel_id)
+    elif isinstance(peer, raw.types.PeerChat):
+        source = chats.get(peer.chat_id)
+    else:
+        return None
+    return source if isinstance(source, _SOURCE_TYPES) else None
+
+
+async def _save_found(
+    client: Client, ctx: CaptureContext, peer: raw.base.Peer, users: dict, chats: dict
+) -> int | None:
+    source = _source(peer, users, chats)
+    if source is None:
+        return None
+    chat = Chat._parse_chat(client, source)  # noqa: SLF001
+    if chat is None or chat.id is None:
+        return None
+    await save_chat(ctx, chat)
+    return chat.id
+
+
+async def _resolve(client: Client, ctx: CaptureContext, query: str) -> int | None:
+    try:
+        chat = await client.get_chat(query)
+    except (BadRequest, Forbidden):
+        return None
+    if not isinstance(chat, Chat) or chat.id is None:
+        return None
+    await save_chat(ctx, chat)
+    return chat.id
+
+
+async def _search(
+    client: Client, query: str, limit: int
+) -> raw.base.contacts.Found | None:
+    try:
+        return await client.invoke(raw.functions.contacts.Search(q=query, limit=limit))
+    except (BadRequest, Forbidden):
+        return None
+
+
+@register("search_peers")
+async def search_peers(ctx: JobContext) -> None:
+    client = ctx.client
+    if client is None:
+        return
+    capture = getattr(client, "capture", None)
+    if capture is None:
+        return
+    query = _normalize(ctx.job.params.get("query", ""))
+    if not query:
+        await ctx.report_progress({"ids": [], "done": True})
+        return
+    limit = int(ctx.job.params.get("limit", DEFAULT_LIMIT))
+    found = await _search(client, query, limit)
+    ids: list[int] = []
+    if found is not None:
+        users = {user.id: user for user in found.users}
+        chats = {chat.id: chat for chat in found.chats}
+        for peer in (*found.my_results, *found.results):
+            peer_id = await _save_found(client, capture, peer, users, chats)
+            if peer_id is not None and peer_id not in ids:
+                ids.append(peer_id)
+    if _USERNAME.match(query):
+        resolved = await _resolve(client, capture, query)
+        if resolved is not None and resolved not in ids:
+            ids.insert(0, resolved)
+    await ctx.report_progress({"ids": ids, "done": True})

backend/src/userbot/modules/jobs/handlers/sync_contacts.py

@@ -0,0 +1,36 @@
+from pyrogram.types import User
+
+from userbot.modules.avatars import note_avatar
+from userbot.modules.jobs.context import JobContext
+from userbot.modules.jobs.registry import register
+from userbot.modules.profiles.parse import snapshot_from_high_level
+from userbot.modules.profiles.repository import write_profile
+
+
+@register("sync_contacts")
+async def sync_contacts(ctx: JobContext) -> None:
+    client = ctx.client
+    if client is None:
+        return
+    capture = getattr(client, "capture", None)
+    if capture is None:
+        return
+    contacts = await client.get_contacts()
+    processed = 0
+    for user in contacts:
+        if not isinstance(user, User):
+            continue
+        fields, photo_file_id, photo_unique_id = snapshot_from_high_level(user)
+        await write_profile(ctx.pool, ctx.account_id, user.id, fields, str(user))
+        if photo_file_id and photo_unique_id:
+            await note_avatar(
+                ctx.pool,
+                ctx.account_id,
+                user.id,
+                "peer",
+                photo_unique_id,
+                photo_file_id,
+            )
+        processed += 1
+    await capture.contacts.refresh()
+    await ctx.report_progress({"processed": processed, "done": True})

backend/src/userbot/modules/jobs/handlers/sync_dialogs.py

@@ -1,16 +1,14 @@
-from datetime import UTC, datetime
-
 from pyrogram import Client
 from pyrogram.errors import BadRequest, Forbidden
-from pyrogram.types import Chat, User
+from pyrogram.types import User
 
 from userbot.modules.avatars import note_avatar
 from userbot.modules.capture.context import CaptureContext
-from userbot.modules.groups.repository import insert_chat_history
 from userbot.modules.jobs.context import JobContext
 from userbot.modules.jobs.registry import register
-from userbot.modules.profiles.parse import snapshot_from_chat, snapshot_from_high_level
+from userbot.modules.profiles.parse import snapshot_from_high_level
 from userbot.modules.profiles.repository import write_profile
+from userbot.modules.profiles.snapshots import save_group, save_private
 
 SAVE_EVERY = 100
 USERS_BATCH = 200
@@ -21,16 +19,6 @@ ON CONFLICT (account_id, chat_id) DO UPDATE SET updated_at = now()
 """
 
 
-async def _save_private(ctx: CaptureContext, chat: Chat, chat_id: int) -> bool:
-    fields, photo_file_id, photo_unique_id = snapshot_from_chat(chat)
-    await write_profile(ctx.pool, ctx.account_id, chat_id, fields, str(chat))
-    if photo_file_id and photo_unique_id:
-        await note_avatar(
-            ctx.pool, ctx.account_id, chat_id, "peer", photo_unique_id, photo_file_id
-        )
-    return bool(fields.first_name or fields.last_name or fields.username)
-
-
 async def _enrich_users(client: Client, ctx: CaptureContext, ids: list[int]) -> None:
     for start in range(0, len(ids), USERS_BATCH):
         batch = ids[start : start + USERS_BATCH]
@@ -55,28 +43,6 @@ async def _enrich_users(client: Client, ctx: CaptureContext, ids: list[int]) ->
                 )
 
 
-async def _save_group(ctx: CaptureContext, chat: Chat, chat_id: int) -> None:
-    photo = chat.photo
-    photo_unique_id = photo.big_photo_unique_id if photo else None
-    photo_file_id = photo.big_file_id if photo else None
-    await insert_chat_history(
-        ctx.pool,
-        ctx.account_id,
-        chat_id,
-        0,
-        "meta",
-        chat.title,
-        photo_unique_id,
-        None,
-        datetime.now(UTC),
-        str(chat),
-    )
-    if photo_file_id and photo_unique_id:
-        await note_avatar(
-            ctx.pool, ctx.account_id, chat_id, "chat", photo_unique_id, photo_file_id
-        )
-
-
 @register("sync_dialogs")
 async def sync_dialogs(ctx: JobContext) -> None:
     client = ctx.client
@@ -94,10 +60,10 @@ async def sync_dialogs(ctx: JobContext) -> None:
         chat_id = chat.id
         try:
             if chat_id > 0:
-                if not await _save_private(capture, chat, chat_id):
+                if not await save_private(capture, chat):
                     nameless.append(chat_id)
             else:
-                await _save_group(capture, chat, chat_id)
+                await save_group(capture, chat)
         except (BadRequest, Forbidden):
             pass
         await ctx.pool.execute(_UPSERT_DIALOG, ctx.account_id, chat_id)

backend/src/userbot/modules/profiles/snapshots.py

@@ -0,0 +1,50 @@
+from datetime import UTC, datetime
+
+from pyrogram.types import Chat
+
+from userbot.modules.avatars import note_avatar
+from userbot.modules.capture.context import CaptureContext
+from userbot.modules.groups.repository import insert_chat_history
+from userbot.modules.profiles.parse import snapshot_from_chat
+from userbot.modules.profiles.repository import write_profile
+
+
+async def save_private(ctx: CaptureContext, chat: Chat) -> bool:
+    chat_id = chat.id or 0
+    fields, photo_file_id, photo_unique_id = snapshot_from_chat(chat)
+    await write_profile(ctx.pool, ctx.account_id, chat_id, fields, str(chat))
+    if photo_file_id and photo_unique_id:
+        await note_avatar(
+            ctx.pool, ctx.account_id, chat_id, "peer", photo_unique_id, photo_file_id
+        )
+    return bool(fields.first_name or fields.last_name or fields.username)
+
+
+async def save_group(ctx: CaptureContext, chat: Chat) -> None:
+    chat_id = chat.id or 0
+    photo = chat.photo
+    photo_unique_id = photo.big_photo_unique_id if photo else None
+    photo_file_id = photo.big_file_id if photo else None
+    await insert_chat_history(
+        ctx.pool,
+        ctx.account_id,
+        chat_id,
+        0,
+        "meta",
+        chat.title,
+        photo_unique_id,
+        None,
+        datetime.now(UTC),
+        str(chat),
+    )
+    if photo_file_id and photo_unique_id:
+        await note_avatar(
+            ctx.pool, ctx.account_id, chat_id, "chat", photo_unique_id, photo_file_id
+        )
+
+
+async def save_chat(ctx: CaptureContext, chat: Chat) -> None:
+    if (chat.id or 0) > 0:
+        await save_private(ctx, chat)
+    else:
+        await save_group(ctx, chat)

backend/src/utils/policy/defaults.py

@@ -16,3 +16,22 @@ DEFAULTS: dict[ChatKind, CaptureToggles] = {
         backfill=True,
     ),
 }
+
+TRACKING: dict[ChatKind, CaptureToggles] = {
+    ChatKind.CHANNEL: CaptureToggles(
+        messages=True,
+        media=True,
+        reactions=True,
+        track_edits_deletes=True,
+        backfill=True,
+    ),
+    ChatKind.GROUP: CaptureToggles(
+        messages=True,
+        media=True,
+        reactions=True,
+        track_edits_deletes=True,
+        profile_history=True,
+        backfill=True,
+    ),
+    ChatKind.DM: DEFAULTS[ChatKind.DM],
+}

backend/src/utils/read/chats.py

@@ -56,7 +56,9 @@ async def list_chats(
     rows = await pool.fetch(
         "WITH ids AS ("
         "SELECT DISTINCT chat_id FROM messages WHERE account_id = $1 "
-        "UNION SELECT chat_id FROM dialogs WHERE account_id = $1), "
+        "UNION SELECT chat_id FROM dialogs WHERE account_id = $1 "
+        "UNION SELECT scope_id FROM capture_policy WHERE account_id = $1 "
+        "AND scope_type = 'chat' AND scope_id IS NOT NULL), "
         "agg AS (SELECT chat_id, count(*) AS message_count, max(date) AS last_date "
         "FROM messages WHERE account_id = $1 GROUP BY chat_id) "
         "SELECT ids.chat_id, COALESCE(agg.message_count, 0) AS message_count, "

backend/src/utils/read/discover.py

@@ -0,0 +1,141 @@
+import asyncpg
+
+from utils.policy.models import ChatKind
+from utils.read.models import DiscoverItem
+
+_ESCAPE = str.maketrans({"\\": "\\\\", "%": r"\%", "_": r"\_"})
+
+_IS_BROADCAST = """
+SELECT COALESCE(raw->'chat'->>'type', raw->>'type') = 'ChatType.CHANNEL'
+FROM chat_history
+WHERE account_id = $1 AND chat_id = $2
+  AND COALESCE(raw->'chat'->>'type', raw->>'type') IS NOT NULL
+ORDER BY ts DESC LIMIT 1
+"""
+
+_IS_TRACKED = """
+SELECT EXISTS (
+    SELECT 1 FROM capture_policy
+    WHERE account_id = $1 AND scope_type = 'chat' AND scope_id = $2
+)
+"""
+
+_ITEMS = """
+WITH chat_meta AS (
+    SELECT DISTINCT ON (chat_id) chat_id, title,
+        COALESCE(raw->'chat'->>'username', raw->>'username') AS username,
+        COALESCE(raw->'chat'->>'type', raw->>'type') = 'ChatType.CHANNEL'
+            AS is_broadcast
+    FROM chat_history
+    WHERE account_id = $1 AND title IS NOT NULL
+    ORDER BY chat_id, ts DESC
+), hits AS (
+    SELECT p.peer_id AS chat_id,
+        COALESCE(NULLIF(concat_ws(' ', p.first_name, p.last_name), ''), p.username)
+            AS title,
+        p.username,
+        COALESCE((p.raw->>'is_bot')::bool, (p.raw->>'bot')::bool, false) AS is_bot,
+        COALESCE((p.raw->>'is_contact')::bool, (p.raw->>'contact')::bool, false)
+            AS is_contact,
+        false AS is_broadcast
+    FROM peers p
+    WHERE p.account_id = $1 AND (p.peer_id = ANY($3::bigint[]) OR ($2 <> '' AND (
+        concat_ws(' ', p.first_name, p.last_name) ILIKE $2
+        OR p.username ILIKE $2 OR p.phone ILIKE $2)))
+    UNION ALL
+    SELECT c.chat_id, c.title, c.username, false, false,
+        COALESCE(c.is_broadcast, false)
+    FROM chat_meta c
+    WHERE c.chat_id = ANY($3::bigint[])
+        OR ($2 <> '' AND (c.title ILIKE $2 OR c.username ILIKE $2))
+), merged AS (
+    SELECT chat_id, max(title) AS title, max(username) AS username,
+        bool_or(is_bot) AS is_bot, bool_or(is_contact) AS is_contact,
+        bool_or(is_broadcast) AS is_broadcast
+    FROM hits GROUP BY chat_id
+), counts AS (
+    SELECT chat_id, count(*) AS message_count FROM messages
+    WHERE account_id = $1 AND chat_id IN (SELECT chat_id FROM merged)
+    GROUP BY chat_id
+)
+SELECT m.chat_id, m.title, m.username, m.is_bot, m.is_contact, m.is_broadcast,
+    COALESCE(c.message_count, 0) AS message_count,
+    EXISTS (SELECT 1 FROM avatars a
+        WHERE a.account_id = $1 AND a.owner_id = m.chat_id) AS has_avatar,
+    EXISTS (SELECT 1 FROM dialogs d
+        WHERE d.account_id = $1 AND d.chat_id = m.chat_id) AS in_dialogs,
+    EXISTS (SELECT 1 FROM capture_policy cp WHERE cp.account_id = $1
+        AND cp.scope_type = 'chat' AND cp.scope_id = m.chat_id) AS tracked
+FROM merged m LEFT JOIN counts c ON c.chat_id = m.chat_id
+ORDER BY in_dialogs DESC, message_count DESC, is_contact DESC, m.title
+LIMIT $4
+"""
+
+
+def _kind(chat_id: int, *, is_broadcast: bool) -> str:
+    if chat_id > 0:
+        return "private"
+    return "channel" if is_broadcast else "group"
+
+
+def _to_item(row: asyncpg.Record) -> DiscoverItem:
+    return DiscoverItem(
+        chat_id=row["chat_id"],
+        title=row["title"],
+        username=row["username"],
+        kind=_kind(row["chat_id"], is_broadcast=row["is_broadcast"]),
+        is_bot=row["is_bot"],
+        is_contact=row["is_contact"],
+        has_avatar=row["has_avatar"],
+        message_count=row["message_count"],
+        in_dialogs=row["in_dialogs"],
+        tracked=row["tracked"],
+    )
+
+
+async def search(
+    pool: asyncpg.Pool, account_id: int, query: str, limit: int
+) -> list[DiscoverItem]:
+    text = query.strip()
+    if not text:
+        return []
+    rows = await pool.fetch(
+        _ITEMS, account_id, f"%{text.translate(_ESCAPE)}%", [], limit
+    )
+    return [_to_item(row) for row in rows]
+
+
+async def by_ids(
+    pool: asyncpg.Pool, account_id: int, ids: list[int]
+) -> list[DiscoverItem]:
+    if not ids:
+        return []
+    rows = await pool.fetch(_ITEMS, account_id, "", ids, len(ids))
+    by_id = {row["chat_id"]: _to_item(row) for row in rows}
+    return [by_id[chat_id] for chat_id in ids if chat_id in by_id]
+
+
+async def get_item(pool: asyncpg.Pool, account_id: int, chat_id: int) -> DiscoverItem:
+    known = await by_ids(pool, account_id, [chat_id])
+    if known:
+        return known[0]
+    kind = await chat_kind(pool, account_id, chat_id)
+    return DiscoverItem(
+        chat_id=chat_id,
+        title=None,
+        username=None,
+        kind="private" if kind is ChatKind.DM else kind.value,
+        is_bot=False,
+        is_contact=False,
+        has_avatar=False,
+        message_count=0,
+        in_dialogs=False,
+        tracked=bool(await pool.fetchval(_IS_TRACKED, account_id, chat_id)),
+    )
+
+
+async def chat_kind(pool: asyncpg.Pool, account_id: int, chat_id: int) -> ChatKind:
+    if chat_id > 0:
+        return ChatKind.DM
+    is_broadcast = await pool.fetchval(_IS_BROADCAST, account_id, chat_id)
+    return ChatKind.CHANNEL if is_broadcast else ChatKind.GROUP

backend/src/utils/read/models.py

@@ -38,6 +38,19 @@ class ChatListItem(BaseModel):
     last_sender_id: int | None
 
 
+class DiscoverItem(BaseModel):
+    chat_id: int
+    title: str | None
+    username: str | None
+    kind: str
+    is_bot: bool
+    is_contact: bool
+    has_avatar: bool
+    message_count: int
+    in_dialogs: bool
+    tracked: bool
+
+
 class EntityView(BaseModel):
     type: str
     offset: int