diff --git a/backend/src/api/mcp/server.py b/backend/src/api/mcp/server.py index 3bf060d..7915d54 100644 --- a/backend/src/api/mcp/server.py +++ b/backend/src/api/mcp/server.py @@ -25,6 +25,7 @@ from utils.read import ( chats, media, peers, + people, presence, presence_stats, social, @@ -397,8 +398,8 @@ async def get_person(account_id: int, peer_id: int) -> dict[str, Any]: of their deleted messages, and the private chat if there is one. """ pool = await _pool() - overview = await peers.get_overview(pool, account_id, peer_id) - groups = await peers.get_groups(pool, account_id, peer_id) + overview = await people.get_overview(pool, account_id, peer_id) + groups = await people.get_groups(pool, account_id, peer_id) return {**overview.model_dump(mode="json"), **groups.model_dump(mode="json")} diff --git a/backend/src/api/routers/peers.py b/backend/src/api/routers/peers.py index ca2dd8f..e098340 100644 --- a/backend/src/api/routers/peers.py +++ b/backend/src/api/routers/peers.py @@ -4,7 +4,7 @@ import asyncpg from dishka.integrations.fastapi import DishkaRoute, FromDishka from fastapi import APIRouter, HTTPException, Query -from utils.read import peers +from utils.read import peers, people from utils.read.models import ( PeerGroups, PeerHistoryView, @@ -47,18 +47,18 @@ async def peer_history( async def peer_overview( pool: FromDishka[asyncpg.Pool], peer_id: int, account_id: AccountId ) -> PeerOverview: - return await peers.get_overview(pool, account_id, peer_id) + return await people.get_overview(pool, account_id, peer_id) @router.get("/peers/{peer_id}/groups") async def peer_groups( pool: FromDishka[asyncpg.Pool], peer_id: int, account_id: AccountId ) -> PeerGroups: - return await peers.get_groups(pool, account_id, peer_id) + return await people.get_groups(pool, account_id, peer_id) @router.get("/chats/{chat_id}/changes") async def chat_changes( pool: FromDishka[asyncpg.Pool], chat_id: int, account_id: AccountId ) -> list[ProfileChange]: - return await peers.get_chat_changes(pool, account_id, chat_id) + return await people.get_chat_changes(pool, account_id, chat_id) diff --git a/backend/src/utils/read/peers.py b/backend/src/utils/read/peers.py index 611f6b1..65b36a8 100644 --- a/backend/src/utils/read/peers.py +++ b/backend/src/utils/read/peers.py @@ -1,23 +1,6 @@ -from collections.abc import Callable -from datetime import datetime -from itertools import pairwise - import asyncpg -from utils.read import chats -from utils.read.avatars import avatar_history -from utils.read.models import ( - DirectChatView, - MembershipView, - Page, - PeerGroups, - PeerHistoryView, - PeerOverview, - PeerView, - ProfileChange, - SharedGroupView, - StoryView, -) +from utils.read.models import Page, PeerHistoryView, PeerView, StoryView HISTORY_LIMIT = 500 @@ -45,53 +28,6 @@ ORDER BY observed_at DESC LIMIT $3 """ -_CHAT_TITLES = """ -SELECT ts, title, previous, actor_id, message_id -FROM ( - SELECT ts, title, - CASE WHEN event = 'title' THEN actor_id END AS actor_id, - CASE WHEN event = 'title' THEN message_id END AS message_id, - lag(title) OVER (ORDER BY ts, message_id) AS previous - FROM chat_history - WHERE account_id = $1 AND chat_id = $2 AND event IN ('meta', 'title') - AND title IS NOT NULL -) titles -WHERE previous IS DISTINCT FROM title AND previous IS NOT NULL -ORDER BY ts DESC -LIMIT $3 -""" - -_MEMBERSHIPS = """ -SELECT m.chat_id, m.message_id, m.event, m.actor_id, m.ts, - COALESCE(NULLIF(trim(concat_ws(' ', a.first_name, a.last_name)), ''), - a.username) AS actor_name -FROM memberships m -LEFT JOIN peers a ON a.account_id = m.account_id AND a.peer_id = m.actor_id -WHERE m.account_id = $1 AND m.user_id = $2 -ORDER BY m.ts DESC -LIMIT 200 -""" - -_DIRECT_CHAT = """ -SELECT $2::bigint AS chat_id, COALESCE(cs.message_count, 0) AS message_count, - cs.last_date -FROM (SELECT 1) one -LEFT JOIN chat_stats cs ON cs.account_id = $1 AND cs.chat_id = $2 -WHERE cs.chat_id IS NOT NULL - OR EXISTS (SELECT 1 FROM dialogs WHERE account_id = $1 AND chat_id = $2) -""" - -_SENT = """ -SELECT chat_id, count(*) AS message_count, max(date) AS last_date, - max(message_id) AS last_message_id, - count(*) FILTER (WHERE deleted_at IS NOT NULL) AS deleted_count -FROM messages -WHERE account_id = $1 AND sender_id = $2 -GROUP BY chat_id -""" - -type Point = tuple[datetime, str | None] - async def get_peer( pool: asyncpg.Pool, account_id: int, peer_id: int @@ -126,135 +62,6 @@ async def get_peer_history( return [PeerHistoryView(**dict(row)) for row in rows] -def _full_name(state: PeerHistoryView) -> str | None: - parts = [part for part in (state.first_name, state.last_name) if part] - return " ".join(parts) or None - - -def _timeline( - states: list[PeerHistoryView], value: Callable[[PeerHistoryView], str | None] -) -> list[Point]: - points: list[Point] = [] - for state in states: - current = value(state) - if not points or points[-1][1] != current: - points.append((state.observed_at, current)) - return points - - -def _without_gaps(points: list[Point]) -> list[Point]: - kept: list[Point] = [] - for at, current in points: - gap = len(kept) > 1 and kept[-1][1] is None and kept[-2][1] == current - if gap: - kept.pop() - elif not kept or kept[-1][1] != current: - kept.append((at, current)) - return kept - - -def _known(points: list[Point]) -> list[Point]: - return _without_gaps([point for point in points if point[1] is not None]) - - -def _changes(field: str, points: list[Point]) -> list[ProfileChange]: - return [ - ProfileChange(field=field, old=old, new=new, changed_at=at) - for (_, old), (at, new) in pairwise(points) - ] - - -def profile_changes(states: list[PeerHistoryView]) -> list[ProfileChange]: - ordered = sorted(states, key=lambda state: state.observed_at) - alive = [state for state in ordered if not state.is_deleted_account] - deleted = _timeline( - ordered, lambda state: "deleted" if state.is_deleted_account else None - ) - changes = [ - *_changes("name", _without_gaps(_timeline(alive, _full_name))), - *_changes( - "username", _without_gaps(_timeline(alive, lambda state: state.username)) - ), - *_changes("phone", _known(_timeline(alive, lambda state: state.phone))), - *[change for change in _changes("deleted", deleted) if change.new is not None], - ] - return sorted(changes, key=lambda change: change.changed_at, reverse=True) - - -async def get_chat_changes( - pool: asyncpg.Pool, account_id: int, chat_id: int -) -> list[ProfileChange]: - rows = await pool.fetch(_CHAT_TITLES, account_id, chat_id, HISTORY_LIMIT) - return [ - ProfileChange( - field="title", - old=row["previous"], - new=row["title"], - changed_at=row["ts"], - actor_id=row["actor_id"], - message_id=row["message_id"], - ) - for row in rows - ] - - -async def _memberships( - pool: asyncpg.Pool, account_id: int, peer_id: int -) -> list[MembershipView]: - rows = await pool.fetch(_MEMBERSHIPS, account_id, peer_id) - ids = list({row["chat_id"] for row in rows}) - known = { - chat.chat_id: chat for chat in await chats.get_chats(pool, account_id, ids) - } - views: list[MembershipView] = [] - for row in rows: - chat = known.get(row["chat_id"]) - views.append( - MembershipView( - **dict(row), - chat_title=chat.title if chat else None, - has_avatar=bool(chat and chat.has_avatar), - ) - ) - return views - - -async def get_overview( - pool: asyncpg.Pool, account_id: int, peer_id: int -) -> PeerOverview: - direct = await pool.fetchrow(_DIRECT_CHAT, account_id, peer_id) - return PeerOverview( - peer=await get_peer(pool, account_id, peer_id), - changes=profile_changes(await get_peer_history(pool, account_id, peer_id)), - avatars=await avatar_history(pool, account_id, peer_id), - memberships=await _memberships(pool, account_id, peer_id), - dm=DirectChatView(**dict(direct)) if direct else None, - ) - - -async def get_groups(pool: asyncpg.Pool, account_id: int, peer_id: int) -> PeerGroups: - async with pool.acquire() as conn, conn.transaction(): - await conn.execute("SET LOCAL jit = off") - rows = await conn.fetch(_SENT, account_id, peer_id) - sent = {row["chat_id"]: row for row in rows if row["chat_id"] < 0} - found = await chats.get_chats(pool, account_id, list(sent)) - groups = [ - SharedGroupView( - chat_id=chat.chat_id, - title=chat.title, - has_avatar=chat.has_avatar, - message_count=sent[chat.chat_id]["message_count"], - last_date=sent[chat.chat_id]["last_date"], - last_message_id=sent[chat.chat_id]["last_message_id"], - ) - for chat in found - ] - groups.sort(key=lambda group: group.last_date, reverse=True) - return PeerGroups( - groups=groups, deleted_count=sum(row["deleted_count"] for row in rows) - ) - - async def get_stories( pool: asyncpg.Pool, account_id: int, page: Page, *, peer_id: int | None = None ) -> list[StoryView]: diff --git a/backend/src/utils/read/people.py b/backend/src/utils/read/people.py new file mode 100644 index 0000000..551b52a --- /dev/null +++ b/backend/src/utils/read/people.py @@ -0,0 +1,194 @@ +from collections.abc import Callable +from datetime import datetime +from itertools import pairwise + +import asyncpg + +from utils.read import chats +from utils.read.avatars import avatar_history +from utils.read.models import ( + DirectChatView, + MembershipView, + PeerGroups, + PeerHistoryView, + PeerOverview, + ProfileChange, + SharedGroupView, +) +from utils.read.peers import HISTORY_LIMIT, get_peer, get_peer_history + +_CHAT_TITLES = """ +SELECT ts, title, previous, actor_id, message_id +FROM ( + SELECT ts, title, + CASE WHEN event = 'title' THEN actor_id END AS actor_id, + CASE WHEN event = 'title' THEN message_id END AS message_id, + lag(title) OVER (ORDER BY ts, message_id) AS previous + FROM chat_history + WHERE account_id = $1 AND chat_id = $2 AND event IN ('meta', 'title') + AND title IS NOT NULL +) titles +WHERE previous IS DISTINCT FROM title AND previous IS NOT NULL +ORDER BY ts DESC +LIMIT $3 +""" + +_MEMBERSHIPS = """ +SELECT m.chat_id, m.message_id, m.event, m.actor_id, m.ts, + COALESCE(NULLIF(trim(concat_ws(' ', a.first_name, a.last_name)), ''), + a.username) AS actor_name +FROM memberships m +LEFT JOIN peers a ON a.account_id = m.account_id AND a.peer_id = m.actor_id +WHERE m.account_id = $1 AND m.user_id = $2 +ORDER BY m.ts DESC +LIMIT 200 +""" + +_DIRECT_CHAT = """ +SELECT $2::bigint AS chat_id, COALESCE(cs.message_count, 0) AS message_count, + cs.last_date +FROM (SELECT 1) one +LEFT JOIN chat_stats cs ON cs.account_id = $1 AND cs.chat_id = $2 +WHERE cs.chat_id IS NOT NULL + OR EXISTS (SELECT 1 FROM dialogs WHERE account_id = $1 AND chat_id = $2) +""" + +_SENT = """ +SELECT chat_id, count(*) AS message_count, max(date) AS last_date, + max(message_id) AS last_message_id, + count(*) FILTER (WHERE deleted_at IS NOT NULL) AS deleted_count +FROM messages +WHERE account_id = $1 AND sender_id = $2 +GROUP BY chat_id +""" + +type Point = tuple[datetime, str | None] + + +def _full_name(state: PeerHistoryView) -> str | None: + parts = [part for part in (state.first_name, state.last_name) if part] + return " ".join(parts) or None + + +def _timeline( + states: list[PeerHistoryView], value: Callable[[PeerHistoryView], str | None] +) -> list[Point]: + points: list[Point] = [] + for state in states: + current = value(state) + if not points or points[-1][1] != current: + points.append((state.observed_at, current)) + return points + + +def _without_gaps(points: list[Point]) -> list[Point]: + kept: list[Point] = [] + for at, current in points: + gap = len(kept) > 1 and kept[-1][1] is None and kept[-2][1] == current + if gap: + kept.pop() + elif not kept or kept[-1][1] != current: + kept.append((at, current)) + return kept + + +def _known(points: list[Point]) -> list[Point]: + return _without_gaps([point for point in points if point[1] is not None]) + + +def _changes(field: str, points: list[Point]) -> list[ProfileChange]: + return [ + ProfileChange(field=field, old=old, new=new, changed_at=at) + for (_, old), (at, new) in pairwise(points) + ] + + +def profile_changes(states: list[PeerHistoryView]) -> list[ProfileChange]: + ordered = sorted(states, key=lambda state: state.observed_at) + alive = [state for state in ordered if not state.is_deleted_account] + deleted = _timeline( + ordered, lambda state: "deleted" if state.is_deleted_account else None + ) + changes = [ + *_changes("name", _without_gaps(_timeline(alive, _full_name))), + *_changes( + "username", _without_gaps(_timeline(alive, lambda state: state.username)) + ), + *_changes("phone", _known(_timeline(alive, lambda state: state.phone))), + *[change for change in _changes("deleted", deleted) if change.new is not None], + ] + return sorted(changes, key=lambda change: change.changed_at, reverse=True) + + +async def get_chat_changes( + pool: asyncpg.Pool, account_id: int, chat_id: int +) -> list[ProfileChange]: + rows = await pool.fetch(_CHAT_TITLES, account_id, chat_id, HISTORY_LIMIT) + return [ + ProfileChange( + field="title", + old=row["previous"], + new=row["title"], + changed_at=row["ts"], + actor_id=row["actor_id"], + message_id=row["message_id"], + ) + for row in rows + ] + + +async def _memberships( + pool: asyncpg.Pool, account_id: int, peer_id: int +) -> list[MembershipView]: + rows = await pool.fetch(_MEMBERSHIPS, account_id, peer_id) + ids = list({row["chat_id"] for row in rows}) + known = { + chat.chat_id: chat for chat in await chats.get_chats(pool, account_id, ids) + } + views: list[MembershipView] = [] + for row in rows: + chat = known.get(row["chat_id"]) + views.append( + MembershipView( + **dict(row), + chat_title=chat.title if chat else None, + has_avatar=bool(chat and chat.has_avatar), + ) + ) + return views + + +async def get_overview( + pool: asyncpg.Pool, account_id: int, peer_id: int +) -> PeerOverview: + direct = await pool.fetchrow(_DIRECT_CHAT, account_id, peer_id) + return PeerOverview( + peer=await get_peer(pool, account_id, peer_id), + changes=profile_changes(await get_peer_history(pool, account_id, peer_id)), + avatars=await avatar_history(pool, account_id, peer_id), + memberships=await _memberships(pool, account_id, peer_id), + dm=DirectChatView(**dict(direct)) if direct else None, + ) + + +async def get_groups(pool: asyncpg.Pool, account_id: int, peer_id: int) -> PeerGroups: + async with pool.acquire() as conn, conn.transaction(): + await conn.execute("SET LOCAL jit = off") + rows = await conn.fetch(_SENT, account_id, peer_id) + sent = {row["chat_id"]: row for row in rows if row["chat_id"] < 0} + found = await chats.get_chats(pool, account_id, list(sent)) + groups = [ + SharedGroupView( + chat_id=chat.chat_id, + title=chat.title, + has_avatar=chat.has_avatar, + message_count=sent[chat.chat_id]["message_count"], + last_date=sent[chat.chat_id]["last_date"], + last_message_id=sent[chat.chat_id]["last_message_id"], + ) + for chat in found + ] + groups.sort(key=lambda group: group.last_date, reverse=True) + return PeerGroups( + groups=groups, deleted_count=sum(row["deleted_count"] for row in rows) + )