feat(backend): person overview with real profile changes, shared groups, joins and leaves, deleted count and group title changes

backend/src/api/routers/peers.py

@@ -5,7 +5,13 @@ from dishka.integrations.fastapi import DishkaRoute, FromDishka
 from fastapi import APIRouter, HTTPException, Query
 
 from utils.read import peers
-from utils.read.models import PeerHistoryView, PeerView
+from utils.read.models import (
+    PeerGroups,
+    PeerHistoryView,
+    PeerOverview,
+    PeerView,
+    ProfileChange,
+)
 
 router = APIRouter(prefix="/api", tags=["peers"], route_class=DishkaRoute)
 
@@ -35,3 +41,24 @@ async def peer_history(
     pool: FromDishka[asyncpg.Pool], peer_id: int, account_id: AccountId
 ) -> list[PeerHistoryView]:
     return await peers.get_peer_history(pool, account_id, peer_id)
+
+
+@router.get("/peers/{peer_id}/overview")
+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)
+
+
+@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)
+
+
+@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)

backend/src/utils/read/models.py

@@ -474,6 +474,54 @@ class PeerHistoryView(BaseModel):
     is_deleted_account: bool
 
 
+class ProfileChange(BaseModel):
+    field: str
+    old: str | None
+    new: str | None
+    changed_at: datetime
+    actor_id: int | None = None
+    message_id: int | None = None
+
+
+class MembershipView(BaseModel):
+    chat_id: int
+    chat_title: str | None
+    has_avatar: bool
+    message_id: int
+    event: str
+    actor_id: int | None
+    actor_name: str | None
+    ts: datetime
+
+
+class DirectChatView(BaseModel):
+    chat_id: int
+    message_count: int
+    last_date: datetime | None
+
+
+class PeerOverview(BaseModel):
+    peer: PeerView | None
+    changes: list[ProfileChange]
+    avatars: list[AvatarHistoryView]
+    memberships: list[MembershipView]
+    dm: DirectChatView | None
+
+
+class SharedGroupView(BaseModel):
+    chat_id: int
+    title: str | None
+    has_avatar: bool
+    message_count: int
+    last_date: datetime
+    last_message_id: int
+
+
+class PeerGroups(BaseModel):
+    groups: list[SharedGroupView]
+    deleted_count: int
+
+
 class ChatLinkView(BaseModel):
     message_id: int
     date: datetime | None

backend/src/utils/read/peers.py

@@ -1,6 +1,25 @@
+from collections.abc import Callable
+from datetime import datetime
+from itertools import pairwise
+
 import asyncpg
 
-from utils.read.models import Page, PeerHistoryView, PeerView, StoryView
+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,
+)
+
+HISTORY_LIMIT = 500
 
 _PEER_COLS = (
     "peer_id, first_name, last_name, username, phone, photo_unique_id, "
@@ -9,6 +28,70 @@ _PEER_COLS = (
     "AND a.owner_id = peers.peer_id) AS has_avatar"
 )
 
+_HISTORY = """
+SELECT observed_at, first_name, last_name, username, phone, photo_unique_id,
+       is_deleted_account
+FROM (
+    SELECT observed_at, first_name, last_name, username, phone, photo_unique_id,
+           is_deleted_account,
+           lag((first_name, last_name, username, phone, photo_unique_id,
+                is_deleted_account)) OVER (ORDER BY observed_at) AS previous
+    FROM peer_history
+    WHERE account_id = $1 AND peer_id = $2
+) states
+WHERE previous IS DISTINCT FROM
+      (first_name, last_name, username, phone, photo_unique_id, is_deleted_account)
+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
@@ -39,17 +122,139 @@ async def get_peers(
 async def get_peer_history(
     pool: asyncpg.Pool, account_id: int, peer_id: int
 ) -> list[PeerHistoryView]:
-    rows = await pool.fetch(
+    rows = await pool.fetch(_HISTORY, account_id, peer_id, HISTORY_LIMIT)
-        "SELECT observed_at, first_name, last_name, username, phone, "
-        "photo_unique_id, is_deleted_account FROM peer_history "
-        "WHERE account_id = $1 AND peer_id = $2 "
-        "ORDER BY observed_at DESC LIMIT 500",
-        account_id,
-        peer_id,
-    )
     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]: