From 1fa4f18a9b4dc7d1270c3a06bf0cc84a240d706d Mon Sep 17 00:00:00 2001 From: h Date: Tue, 1 Sep 2026 21:38:22 +0200 Subject: [PATCH] perf(*): bound list queries, batch fetches, harden media downloads --- .../a6d2f8b41c9e_read_path_indexes.py | 49 ++++++++++ backend/src/api/routers/backfill.py | 15 ++- backend/src/api/routers/chats.py | 9 ++ backend/src/api/routers/discover.py | 3 +- backend/src/api/routers/profile.py | 10 +- .../src/userbot/modules/capture/repository.py | 2 +- backend/src/userbot/modules/client.py | 1 + backend/src/userbot/modules/download.py | 2 +- backend/src/userbot/modules/jobs/consumer.py | 3 + .../userbot/modules/jobs/handlers/backfill.py | 20 +++- .../modules/jobs/handlers/enrich_chat.py | 26 ++++-- .../modules/jobs/handlers/fetch_media.py | 8 +- .../src/userbot/modules/jobs/repository.py | 10 ++ .../src/userbot/modules/media/downloader.py | 71 ++++++++++++-- backend/src/userbot/runner.py | 16 +--- backend/src/utils/jobs.py | 12 ++- backend/src/utils/read/accounts.py | 11 ++- backend/src/utils/read/analytics.py | 4 +- backend/src/utils/read/avatars.py | 2 +- backend/src/utils/read/chats.py | 23 ++++- backend/src/utils/read/discover.py | 3 +- backend/src/utils/read/media.py | 11 ++- backend/src/utils/read/peers.py | 3 +- backend/src/utils/read/presence.py | 6 +- backend/src/utils/read/profile.py | 24 +++-- frontend/src/lib/api/custom-emoji.ts | 19 +++- frontend/src/lib/api/endpoints.ts | 20 +++- frontend/src/lib/api/media.ts | 25 ++++- frontend/src/lib/api/stories.ts | 19 +++- .../src/lib/components/MessageList.svelte | 26 ++++++ .../lib/components/policy/PolicyEditor.svelte | 10 +- .../components/profile/ChatCalendar.svelte | 93 +++++++++---------- .../lib/components/shares/SharesPanel.svelte | 11 ++- .../stories/AllStoriesArchive.svelte | 10 +- frontend/src/lib/stores/chats.svelte.ts | 34 ++++++- 35 files changed, 473 insertions(+), 138 deletions(-) create mode 100644 backend/migrations/versions/a6d2f8b41c9e_read_path_indexes.py diff --git a/backend/migrations/versions/a6d2f8b41c9e_read_path_indexes.py b/backend/migrations/versions/a6d2f8b41c9e_read_path_indexes.py new file mode 100644 index 0000000..5d22a65 --- /dev/null +++ b/backend/migrations/versions/a6d2f8b41c9e_read_path_indexes.py @@ -0,0 +1,49 @@ +"""read path indexes + +Revision ID: a6d2f8b41c9e +Revises: f2b8d3c9a51e +Create Date: 2026-09-01 12:00:00.000000 + +""" + +from collections.abc import Sequence + +from alembic import op + +revision: str = "a6d2f8b41c9e" +down_revision: str | None = "f2b8d3c9a51e" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute("CREATE INDEX ix_jobs_account_id ON jobs (account_id, id DESC)") + op.execute( + "CREATE INDEX ix_messages_deleted ON messages " + "(account_id, deleted_at DESC) WHERE deleted_at IS NOT NULL" + ) + op.execute( + "CREATE INDEX ix_messages_pinned ON messages " + "(account_id, chat_id, date DESC, message_id DESC) " + "WHERE raw->>'service' LIKE '%PINNED_MESSAGE%'" + ) + op.execute("CREATE INDEX ix_watches_account ON watches (account_id, id DESC)") + op.execute("CREATE INDEX ix_alerts_account_ts ON alerts (account_id, ts DESC)") + op.execute( + "CREATE INDEX ix_stories_account_date ON stories " + "(account_id, date DESC NULLS LAST)" + ) + op.execute( + "CREATE INDEX ix_annotations_account_msg ON annotations " + "(account_id, chat_id, message_id)" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS ix_annotations_account_msg") + op.execute("DROP INDEX IF EXISTS ix_stories_account_date") + op.execute("DROP INDEX IF EXISTS ix_alerts_account_ts") + op.execute("DROP INDEX IF EXISTS ix_watches_account") + op.execute("DROP INDEX IF EXISTS ix_messages_pinned") + op.execute("DROP INDEX IF EXISTS ix_messages_deleted") + op.execute("DROP INDEX IF EXISTS ix_jobs_account_id") diff --git a/backend/src/api/routers/backfill.py b/backend/src/api/routers/backfill.py index 5cf8306..d5b11fc 100644 --- a/backend/src/api/routers/backfill.py +++ b/backend/src/api/routers/backfill.py @@ -8,6 +8,7 @@ from fastapi import APIRouter, HTTPException, Query from pydantic import BaseModel from utils.jobs import enqueue +from utils.read.models import DEFAULT_LIMIT, Page router = APIRouter(prefix="/api", tags=["backfill"], route_class=DishkaRoute) @@ -130,16 +131,26 @@ async def list_jobs( pool: FromDishka[asyncpg.Pool], account_id: Annotated[int, Query()], status: Annotated[str | None, Query()] = None, + limit: Annotated[int, Query()] = DEFAULT_LIMIT, + offset: Annotated[int, Query()] = 0, ) -> list[JobView]: + page = Page(limit=limit, offset=offset) if status is None: rows = await pool.fetch( - "SELECT * FROM jobs WHERE account_id = $1 ORDER BY id DESC", account_id + "SELECT * FROM jobs WHERE account_id = $1 " + "ORDER BY id DESC LIMIT $2 OFFSET $3", + account_id, + page.capped_limit, + page.offset, ) else: rows = await pool.fetch( - "SELECT * FROM jobs WHERE account_id = $1 AND status = $2 ORDER BY id DESC", + "SELECT * FROM jobs WHERE account_id = $1 AND status = $2 " + "ORDER BY id DESC LIMIT $3 OFFSET $4", account_id, status, + page.capped_limit, + page.offset, ) return [_to_view(row) for row in rows] diff --git a/backend/src/api/routers/chats.py b/backend/src/api/routers/chats.py index d34ae37..d3cf3a6 100644 --- a/backend/src/api/routers/chats.py +++ b/backend/src/api/routers/chats.py @@ -10,6 +10,7 @@ from utils.policy import repository from utils.read import chats from utils.read.models import ( DEFAULT_LIMIT, + MAX_LIMIT, ChatListItem, MessageVersionView, MessageView, @@ -49,6 +50,14 @@ async def list_chats( ) +@router.get("/chats/batch") +async def get_chats_batch( + pool: FromDishka[asyncpg.Pool], account_id: AccountId, ids: Annotated[str, Query()] +) -> list[ChatListItem]: + parsed = [int(part) for part in ids.split(",") if part.strip()] + return await chats.get_chats(pool, account_id, parsed[:MAX_LIMIT]) + + @router.get("/chats/{chat_id}") async def get_chat( pool: FromDishka[asyncpg.Pool], chat_id: int, account_id: AccountId diff --git a/backend/src/api/routers/discover.py b/backend/src/api/routers/discover.py index 72e3ba4..b2c94ce 100644 --- a/backend/src/api/routers/discover.py +++ b/backend/src/api/routers/discover.py @@ -13,7 +13,7 @@ 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 +from utils.read.models import MAX_LIMIT, DiscoverItem router = APIRouter(prefix="/api", tags=["discover"], route_class=DishkaRoute) @@ -68,6 +68,7 @@ async def discover_peers( ) -> list[DiscoverItem]: if not query.strip(): return [] + limit = min(limit, MAX_LIMIT) items = await discover.search(pool, account_id, query, limit) if not remote: return items diff --git a/backend/src/api/routers/profile.py b/backend/src/api/routers/profile.py index fcd15aa..57c5618 100644 --- a/backend/src/api/routers/profile.py +++ b/backend/src/api/routers/profile.py @@ -52,9 +52,15 @@ async def chat_links( @router.get("/calendar") async def chat_calendar( - pool: FromDishka[asyncpg.Pool], chat_id: int, account_id: AccountId + pool: FromDishka[asyncpg.Pool], + chat_id: int, + account_id: AccountId, + date_from: Annotated[datetime | None, Query()] = None, + date_to: Annotated[datetime | None, Query()] = None, ) -> list[DayCount]: - return await profile.daily_counts(pool, account_id, chat_id) + return await profile.daily_counts( + pool, account_id, chat_id, date_from=date_from, date_to=date_to + ) @router.get("/message-at") diff --git a/backend/src/userbot/modules/capture/repository.py b/backend/src/userbot/modules/capture/repository.py index 4023805..7ed42bd 100644 --- a/backend/src/userbot/modules/capture/repository.py +++ b/backend/src/userbot/modules/capture/repository.py @@ -122,7 +122,7 @@ async def max_message_id( pool: asyncpg.Pool, account_id: int, chat_id: int ) -> int | None: return await pool.fetchval( - "SELECT max(message_id) FROM messages WHERE account_id = $1 AND chat_id = $2", + "SELECT last_message_id FROM chat_stats WHERE account_id = $1 AND chat_id = $2", account_id, chat_id, ) diff --git a/backend/src/userbot/modules/client.py b/backend/src/userbot/modules/client.py index c3a665e..779c3fc 100644 --- a/backend/src/userbot/modules/client.py +++ b/backend/src/userbot/modules/client.py @@ -31,6 +31,7 @@ class PyroClient(Client): lang_code="en", system_lang_code="en-US", client_platform=enums.ClientPlatform.DESKTOP, + max_concurrent_transmissions=4, ) self.capture: CaptureContext | None = None diff --git a/backend/src/userbot/modules/download.py b/backend/src/userbot/modules/download.py index c4cfc46..bcf383a 100644 --- a/backend/src/userbot/modules/download.py +++ b/backend/src/userbot/modules/download.py @@ -50,7 +50,7 @@ async def _reset_media_sessions(client: Client) -> None: async def download_bytes(client: Client, target: Downloadable) -> bytes | None: try: buffer = await client.download_media(target, in_memory=True) - except TimeoutError: + except (TimeoutError, OSError): await _reset_media_sessions(client) buffer = await client.download_media(target, in_memory=True) return buffer.getvalue() if isinstance(buffer, BytesIO) else None diff --git a/backend/src/userbot/modules/jobs/consumer.py b/backend/src/userbot/modules/jobs/consumer.py index 55d298a..fcd99da 100644 --- a/backend/src/userbot/modules/jobs/consumer.py +++ b/backend/src/userbot/modules/jobs/consumer.py @@ -69,6 +69,9 @@ class JobConsumer: requeued = await repository.requeue_running(self.pool, self.account_id) if requeued: logger.info(f"[yellow]Requeued {requeued} stale running job(s).[/]") + pruned = await repository.prune_finished(self.pool, self.account_id) + if pruned: + logger.info(f"[yellow]Pruned {pruned} finished job(s).[/]") conn = await asyncpg.connect(dsn=env.db.connection_url) await conn.add_listener(JOBS_CHANGED_CHANNEL, lambda *_: self._wake.set()) logger.info(f"[green]Job consumer running for account {self.account_id}.[/]") diff --git a/backend/src/userbot/modules/jobs/handlers/backfill.py b/backend/src/userbot/modules/jobs/handlers/backfill.py index 84a442c..7a8abc5 100644 --- a/backend/src/userbot/modules/jobs/handlers/backfill.py +++ b/backend/src/userbot/modules/jobs/handlers/backfill.py @@ -1,5 +1,5 @@ from pyrogram import Client -from pyrogram.errors import PeerIdInvalid +from pyrogram.errors import FloodPremiumWait, FloodWait, PeerIdInvalid from pyrogram.types import Message from userbot.modules.capture import capture_message @@ -11,6 +11,7 @@ from userbot.modules.jobs.registry import register from userbot.modules.stt import repository as stt_repo from userbot.modules.stt import should_transcribe_on_backfill from userbot.modules.stt.gate import safe_transcribe +from utils.logging import logger from utils.policy.models import CaptureToggles SAVE_EVERY = 100 @@ -61,21 +62,30 @@ async def backfill(ctx: JobContext) -> None: min_id = await resolve_min_id(ctx, chat_id) await ctx.save_cursor({"max_id": max_id, "min_id": min_id}) processed = ctx.job.progress.get("processed", 0) + errors = ctx.job.progress.get("errors", 0) self_id = client.me.id if client.me else None try: async for message in client.get_chat_history( chat_id, max_id=max_id, min_id=min_id ): - await capture_message(client, message, capture, toggles) - await maybe_transcribe(client, capture, chat_id, message, self_id) + try: + await capture_message(client, message, capture, toggles) + await maybe_transcribe(client, capture, chat_id, message, self_id) + except (FloodWait, FloodPremiumWait, PeerIdInvalid): + raise + except Exception: + errors += 1 + logger.exception(f"Backfill capture failed for {chat_id}/{message.id}") processed += 1 if processed % SAVE_EVERY == 0: next_max = message.id - 1 await ctx.save_cursor({"max_id": next_max, "min_id": min_id}) - await ctx.report_progress({"processed": processed, "max_id": next_max}) + await ctx.report_progress( + {"processed": processed, "errors": errors, "max_id": next_max} + ) if await ctx.is_canceled(): return except PeerIdInvalid: await ctx.report_progress({"processed": processed, "error": "peer_id_invalid"}) return - await ctx.report_progress({"processed": processed, "done": True}) + await ctx.report_progress({"processed": processed, "errors": errors, "done": True}) diff --git a/backend/src/userbot/modules/jobs/handlers/enrich_chat.py b/backend/src/userbot/modules/jobs/handlers/enrich_chat.py index ba16601..2cbaf82 100644 --- a/backend/src/userbot/modules/jobs/handlers/enrich_chat.py +++ b/backend/src/userbot/modules/jobs/handlers/enrich_chat.py @@ -65,16 +65,28 @@ async def _enrich_members(client: Client, ctx: CaptureContext, chat_id: int) -> return +async def _fetch_users(client: Client, ids: list[int]) -> list[User]: + try: + users = await client.get_users(ids) + except BadRequest: + users = [] + for sender_id in ids: + try: + users.append(await client.get_users(sender_id)) + except BadRequest: + continue + if isinstance(users, User): + return [users] + return [user for user in users if isinstance(user, User)] + + async def _enrich_senders(client: Client, ctx: CaptureContext, chat_id: int) -> None: rows = await ctx.pool.fetch(_MISSING_SENDERS, ctx.account_id, chat_id) ids = [row["sender_id"] for row in rows] - for sender_id in ids: - try: - user = await client.get_users(sender_id) - except BadRequest: - continue - if isinstance(user, User): - await _save_user(ctx, user) + if not ids: + return + for user in await _fetch_users(client, ids): + await _save_user(ctx, user) @register("enrich_chat") diff --git a/backend/src/userbot/modules/jobs/handlers/fetch_media.py b/backend/src/userbot/modules/jobs/handlers/fetch_media.py index c283864..6bb5ef3 100644 --- a/backend/src/userbot/modules/jobs/handlers/fetch_media.py +++ b/backend/src/userbot/modules/jobs/handlers/fetch_media.py @@ -18,6 +18,10 @@ async def fetch_media(ctx: JobContext) -> None: if isinstance(message, list): message = message[0] if message else None if message is None or message.empty: - return + msg = f"message {chat_id}/{message_id} is unavailable" + raise RuntimeError(msg) toggles = CaptureToggles(media=True, self_destruct_media=True) - await capture_media(client, message, capture, chat_id, message_id, toggles) + ok = await capture_media(client, message, capture, chat_id, message_id, toggles) + if not ok: + msg = f"media download failed for {chat_id}/{message_id}" + raise RuntimeError(msg) diff --git a/backend/src/userbot/modules/jobs/repository.py b/backend/src/userbot/modules/jobs/repository.py index 6593e7a..72b8eae 100644 --- a/backend/src/userbot/modules/jobs/repository.py +++ b/backend/src/userbot/modules/jobs/repository.py @@ -48,6 +48,16 @@ async def requeue_running(pool: asyncpg.Pool, account_id: int) -> int: return int(result.split()[-1]) +async def prune_finished(pool: asyncpg.Pool, account_id: int) -> int: + result = await pool.execute( + "DELETE FROM jobs WHERE account_id = $1 " + "AND status IN ('done', 'canceled', 'failed') " + "AND finished_at < now() - interval '30 days'", + account_id, + ) + return int(result.split()[-1]) + + async def save_cursor(pool: asyncpg.Pool, job_id: int, cursor: dict[str, Any]) -> None: await pool.execute( "UPDATE jobs SET cursor = $2::jsonb, updated_at = now() WHERE id = $1", diff --git a/backend/src/userbot/modules/media/downloader.py b/backend/src/userbot/modules/media/downloader.py index eacc32c..d488fe1 100644 --- a/backend/src/userbot/modules/media/downloader.py +++ b/backend/src/userbot/modules/media/downloader.py @@ -1,14 +1,23 @@ from typing import Any from pyrogram import Client +from pyrogram.errors import ( + FileIdInvalid, + FileReferenceExpired, + FileReferenceInvalid, + FloodPremiumWait, + FloodWait, +) from pyrogram.types import Message from userbot.modules.capture import repository from userbot.modules.capture.context import CaptureContext -from userbot.modules.download import download_bytes +from userbot.modules.download import Downloadable, download_bytes from utils.logging import logger from utils.policy.models import CaptureToggles +STALE_FILE_ID = (FileIdInvalid, FileReferenceExpired, FileReferenceInvalid) + _MEDIA_ATTRS = ( "photo", "video", @@ -48,6 +57,48 @@ def media_unique_id(message: Message) -> str | None: return getattr(obj, "file_unique_id", None) if obj is not None else None +def _download_target(message: Message) -> Downloadable | None: + kind, obj = media_object(message) + if obj is None: + return None + return message if getattr(message, kind or "", None) is obj else obj + + +async def _fresh_target( + client: Client, chat_id: int, message_id: int +) -> Downloadable | None: + message = await client.get_messages(chat_id, message_id) + if not isinstance(message, Message) or message.empty: + return None + return _download_target(message) + + +async def _download( + client: Client, target: Downloadable | None, chat_id: int, message_id: int +) -> bytes | None: + if target is None: + return None + try: + return await download_bytes(client, target) + except TimeoutError: + logger.warning( + f"[yellow]Media download timed out for {chat_id}/{message_id}.[/]" + ) + return None + except STALE_FILE_ID: + fresh = await _fresh_target(client, chat_id, message_id) + if fresh is None: + return None + try: + return await download_bytes(client, fresh) + except (TimeoutError, *STALE_FILE_ID): + logger.warning( + f"[yellow]Media download failed after refetch " + f"for {chat_id}/{message_id}.[/]" + ) + return None + + async def capture_media( # noqa: PLR0913 client: Client, message: Message, @@ -55,10 +106,10 @@ async def capture_media( # noqa: PLR0913 chat_id: int, message_id: int, toggles: CaptureToggles, -) -> None: +) -> bool: kind, obj = media_object(message) if obj is None: - return + return True unique_id = getattr(obj, "file_unique_id", None) ttl = getattr(obj, "ttl_seconds", None) want = toggles.self_destruct_media if ttl else toggles.media @@ -66,6 +117,7 @@ async def capture_media( # noqa: PLR0913 mime = getattr(obj, "mime_type", None) storage_key: str | None = None downloaded = False + flood: FloodWait | FloodPremiumWait | None = None if want: existing = await repository.current_media( ctx.pool, ctx.account_id, chat_id, message_id @@ -80,13 +132,11 @@ async def capture_media( # noqa: PLR0913 file_size = existing["file_size"] downloaded = True else: - target = message if getattr(message, kind or "", None) is obj else obj + target = _download_target(message) try: - data = await download_bytes(client, target) - except TimeoutError: - logger.warning( - f"[yellow]Media download timed out for {chat_id}/{message_id}.[/]" - ) + data = await _download(client, target, chat_id, message_id) + except (FloodWait, FloodPremiumWait) as exc: + flood = exc data = None if data is not None: storage_key = ctx.storage.put(data) @@ -105,3 +155,6 @@ async def capture_media( # noqa: PLR0913 unique_id, downloaded=downloaded, ) + if flood is not None: + raise flood + return downloaded or not want diff --git a/backend/src/userbot/runner.py b/backend/src/userbot/runner.py index 434ba8b..bf45952 100644 --- a/backend/src/userbot/runner.py +++ b/backend/src/userbot/runner.py @@ -49,18 +49,6 @@ async def _cancel(task: asyncio.Task) -> None: await task -async def _enqueue_once(pool: asyncpg.Pool, account_id: int, kind: str) -> None: - existing = await pool.fetchval( - "SELECT 1 FROM jobs WHERE account_id = $1 AND kind = $2 " - "AND status IN ('pending', 'running') LIMIT 1", - account_id, - kind, - ) - if existing is None: - await enqueue(pool, account_id, kind, {}) - logger.info(f"[green]Queued {kind}.[/]") - - class AccountRegistry: def __init__(self, pool: asyncpg.Pool, storage: ContentAddressedStorage) -> None: self._pool = pool @@ -121,8 +109,8 @@ class AccountRegistry: client, asyncio.create_task(consumer.run()), device_model ) logger.info(f"[green]Client started:[/] {me.full_name} ({me.id})") - await _enqueue_once(self._pool, account_id, "sync_dialogs") - await _enqueue_once(self._pool, account_id, "sync_contacts") + await enqueue(self._pool, account_id, "sync_dialogs", {}) + await enqueue(self._pool, account_id, "sync_contacts", {}) async def _stop(self, session_name: str) -> None: account = self._running.pop(session_name, None) diff --git a/backend/src/utils/jobs.py b/backend/src/utils/jobs.py index cfdf54f..c80876b 100644 --- a/backend/src/utils/jobs.py +++ b/backend/src/utils/jobs.py @@ -9,12 +9,22 @@ JOBS_CHANGED_CHANNEL = "jobs_changed" async def enqueue( pool: asyncpg.Pool, account_id: int, kind: str, params: dict[str, Any] ) -> int: + params_json = json.dumps(params) + existing = await pool.fetchval( + "SELECT id FROM jobs WHERE account_id = $1 AND kind = $2 " + "AND params = $3::jsonb AND status IN ('pending', 'running') LIMIT 1", + account_id, + kind, + params_json, + ) + if existing is not None: + return existing job_id = await pool.fetchval( "INSERT INTO jobs (account_id, kind, params) " "VALUES ($1, $2, $3::jsonb) RETURNING id", account_id, kind, - json.dumps(params), + params_json, ) await pool.execute(f"NOTIFY {JOBS_CHANGED_CHANNEL}") return job_id diff --git a/backend/src/utils/read/accounts.py b/backend/src/utils/read/accounts.py index c8682fa..d473759 100644 --- a/backend/src/utils/read/accounts.py +++ b/backend/src/utils/read/accounts.py @@ -30,10 +30,19 @@ RETURNING {_ACCOUNT_COLS} """ # noqa: S608 +_self_ids: dict[int, int] = {} + + async def self_user_id(pool: asyncpg.Pool, account_id: int) -> int | None: - return await pool.fetchval( + cached = _self_ids.get(account_id) + if cached is not None: + return cached + tg_user_id = await pool.fetchval( "SELECT tg_user_id FROM accounts WHERE account_id = $1", account_id ) + if tg_user_id is not None: + _self_ids[account_id] = tg_user_id + return tg_user_id async def list_accounts(pool: asyncpg.Pool) -> list[AccountView]: diff --git a/backend/src/utils/read/analytics.py b/backend/src/utils/read/analytics.py index 2a615c4..b128e6f 100644 --- a/backend/src/utils/read/analytics.py +++ b/backend/src/utils/read/analytics.py @@ -33,7 +33,7 @@ async def message_volume( async def response_stats( - pool: asyncpg.Pool, account_id: int, chat_id: int + pool: asyncpg.Pool, account_id: int, chat_id: int, *, days: int = 365 ) -> ResponseStats: self_id = await self_user_id(pool, account_id) rows = await pool.fetch( @@ -43,6 +43,7 @@ async def response_stats( "lag(date) OVER w AS prev_date " "FROM messages " "WHERE account_id = $1 AND chat_id = $2 AND sender_id IS NOT NULL " + "AND date >= $4 " "WINDOW w AS (ORDER BY date, message_id)), " "resp AS (" "SELECT (sender_id = $3) AS is_mine, " @@ -55,6 +56,7 @@ async def response_stats( account_id, chat_id, self_id, + datetime.now(UTC) - timedelta(days=days), ) stats = ResponseStats( mine_median_seconds=None, mine_count=0, their_median_seconds=None, their_count=0 diff --git a/backend/src/utils/read/avatars.py b/backend/src/utils/read/avatars.py index dfc9fb9..03dfd5b 100644 --- a/backend/src/utils/read/avatars.py +++ b/backend/src/utils/read/avatars.py @@ -21,7 +21,7 @@ WHERE account_id = $1 AND owner_id = $2 AND unique_id = $3 _AVATAR_HISTORY = """ SELECT unique_id, first_seen_at, downloaded FROM avatars WHERE account_id = $1 AND owner_id = $2 -ORDER BY first_seen_at DESC +ORDER BY first_seen_at DESC LIMIT 200 """ diff --git a/backend/src/utils/read/chats.py b/backend/src/utils/read/chats.py index c04d8d1..d54ecf4 100644 --- a/backend/src/utils/read/chats.py +++ b/backend/src/utils/read/chats.py @@ -62,6 +62,18 @@ _ONE_ID = """ WHERE account_id = $1 AND scope_type = 'chat' AND scope_id = $2 """ +_MANY_IDS = """ + SELECT chat_id FROM chat_stats + WHERE account_id = $1 AND chat_id = ANY($2::bigint[]) + UNION + SELECT chat_id FROM dialogs + WHERE account_id = $1 AND chat_id = ANY($2::bigint[]) + UNION + SELECT scope_id AS chat_id FROM capture_policy + WHERE account_id = $1 AND scope_type = 'chat' + AND scope_id = ANY($2::bigint[]) +""" + _CHAT_ROWS = """ WITH ids AS ({ids}) SELECT ids.chat_id, @@ -171,6 +183,15 @@ async def get_chat( return _chat_item(row) if row is not None else None +async def get_chats( + pool: asyncpg.Pool, account_id: int, ids: list[int] +) -> list[ChatListItem]: + if not ids: + return [] + rows = await pool.fetch(_CHAT_ROWS.format(ids=_MANY_IDS), account_id, ids) + return [_chat_item(row) for row in rows] + + async def get_chat_history( # noqa: PLR0913 pool: asyncpg.Pool, account_id: int, @@ -308,7 +329,7 @@ async def get_message_versions( rows = await pool.fetch( "SELECT observed_at, edit_date, text FROM message_versions " "WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 " - "ORDER BY observed_at", + "ORDER BY observed_at LIMIT 500", account_id, chat_id, message_id, diff --git a/backend/src/utils/read/discover.py b/backend/src/utils/read/discover.py index ff34214..5aac8ec 100644 --- a/backend/src/utils/read/discover.py +++ b/backend/src/utils/read/discover.py @@ -54,9 +54,8 @@ WITH chat_meta AS ( bool_or(is_broadcast) AS is_broadcast FROM hits GROUP BY chat_id ), counts AS ( - SELECT chat_id, count(*) AS message_count FROM messages + SELECT chat_id, message_count FROM chat_stats 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, diff --git a/backend/src/utils/read/media.py b/backend/src/utils/read/media.py index d8231ad..1995497 100644 --- a/backend/src/utils/read/media.py +++ b/backend/src/utils/read/media.py @@ -1,7 +1,8 @@ +import json + import asyncpg from utils.files import media_file_name -from utils.read.message_view import load_raw from utils.read.models import MediaVersionView, MediaView MEDIA_COLS = ( @@ -40,15 +41,15 @@ async def _web_page_media_stub( pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int ) -> MediaView | None: row = await pool.fetchrow( - "SELECT date, raw FROM messages " + "SELECT date, raw->'web_page' AS web_page FROM messages " "WHERE account_id = $1 AND chat_id = $2 AND message_id = $3", account_id, chat_id, message_id, ) - if row is None: + if row is None or row["web_page"] is None: return None - web_page = load_raw(row["raw"]).get("web_page") + web_page = json.loads(row["web_page"]) if not isinstance(web_page, dict): return None kind = next( @@ -105,7 +106,7 @@ async def get_media_versions( rows = await pool.fetch( f"SELECT {_VERSION_COLS} FROM media_versions " # noqa: S608 "WHERE account_id = $1 AND chat_id = $2 AND message_id = $3 " - "ORDER BY observed_at", + "ORDER BY observed_at LIMIT 500", account_id, chat_id, message_id, diff --git a/backend/src/utils/read/peers.py b/backend/src/utils/read/peers.py index 662a8da..014b0f5 100644 --- a/backend/src/utils/read/peers.py +++ b/backend/src/utils/read/peers.py @@ -42,7 +42,8 @@ async def get_peer_history( rows = await pool.fetch( "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", + "WHERE account_id = $1 AND peer_id = $2 " + "ORDER BY observed_at DESC LIMIT 500", account_id, peer_id, ) diff --git a/backend/src/utils/read/presence.py b/backend/src/utils/read/presence.py index 5e61233..db1351e 100644 --- a/backend/src/utils/read/presence.py +++ b/backend/src/utils/read/presence.py @@ -1,9 +1,11 @@ -from datetime import datetime +from datetime import UTC, datetime, timedelta import asyncpg from utils.read.models import Page, PresenceHourly, PresenceSample +HOURLY_DEFAULT_DAYS = 90 + async def presence_history( # noqa: PLR0913 pool: asyncpg.Pool, @@ -54,6 +56,8 @@ async def presence_hourly( date_from: datetime | None = None, date_to: datetime | None = None, ) -> list[PresenceHourly]: + if date_from is None: + date_from = datetime.now(UTC) - timedelta(days=HOURLY_DEFAULT_DAYS) params: list[object] = [account_id, peer_id] where = "account_id = $1 AND peer_id = $2" if date_from is not None: diff --git a/backend/src/utils/read/profile.py b/backend/src/utils/read/profile.py index 5f54bff..8663625 100644 --- a/backend/src/utils/read/profile.py +++ b/backend/src/utils/read/profile.py @@ -42,17 +42,27 @@ async def chat_links( async def daily_counts( - pool: asyncpg.Pool, account_id: int, chat_id: int + pool: asyncpg.Pool, + account_id: int, + chat_id: int, + *, + date_from: datetime | None = None, + date_to: datetime | None = None, ) -> list[DayCount]: self_id = await self_user_id(pool, account_id) + params: list[object] = [account_id, chat_id, self_id] + where = "account_id = $1 AND chat_id = $2" + if date_from is not None: + params.append(date_from) + where += f" AND date >= ${len(params)}" + if date_to is not None: + params.append(date_to) + where += f" AND date < ${len(params)}" rows = await pool.fetch( - "SELECT date_trunc('day', date) AS day, count(*) AS count, " + "SELECT date_trunc('day', date) AS day, count(*) AS count, " # noqa: S608 "count(*) FILTER (WHERE sender_id = $3) AS outgoing FROM messages " - "WHERE account_id = $1 AND chat_id = $2 " - "GROUP BY day ORDER BY day", - account_id, - chat_id, - self_id, + f"WHERE {where} GROUP BY day ORDER BY day", + *params, ) return [DayCount(**dict(row)) for row in rows] diff --git a/frontend/src/lib/api/custom-emoji.ts b/frontend/src/lib/api/custom-emoji.ts index 863cf37..d04e5b7 100644 --- a/frontend/src/lib/api/custom-emoji.ts +++ b/frontend/src/lib/api/custom-emoji.ts @@ -9,10 +9,27 @@ export interface CustomEmojiAsset { url: string; } +const MAX_CACHED = 500; + const ready = new Map(); const missing = new Set(); const inflight = new Map>(); +function remember(key: string, asset: CustomEmojiAsset) { + ready.set(key, asset); + while (ready.size > MAX_CACHED) { + const oldest = ready.keys().next(); + if (oldest.done) { + return; + } + const stale = ready.get(oldest.value); + ready.delete(oldest.value); + if (stale) { + URL.revokeObjectURL(stale.url); + } + } +} + function authHeaders(): Record { return auth.token ? { Authorization: `Bearer ${auth.token}` } : {}; } @@ -34,7 +51,7 @@ async function fetchEmoji( if (response.ok) { const blob = await response.blob(); const asset = { url: URL.createObjectURL(blob), mime: blob.type }; - ready.set(key, asset); + remember(key, asset); return asset; } if (response.status === 409 && retry) { diff --git a/frontend/src/lib/api/endpoints.ts b/frontend/src/lib/api/endpoints.ts index 3e1f727..18b91d5 100644 --- a/frontend/src/lib/api/endpoints.ts +++ b/frontend/src/lib/api/endpoints.ts @@ -303,8 +303,14 @@ export function getMessageLinks( }); } -export function getChatCalendar(chatId: number): Promise { - return request(`/chats/${chatId}/calendar`, { account: true }); +export function getChatCalendar( + chatId: number, + range: { date_from?: string; date_to?: string } = {} +): Promise { + return request(`/chats/${chatId}/calendar`, { + account: true, + query: { ...range }, + }); } export function getMessageAt(chatId: number, date: string): Promise { @@ -334,6 +340,16 @@ export function getPeers(ids: number[]): Promise { }); } +export function getChatsBatch(ids: number[]): Promise { + if (ids.length === 0) { + return Promise.resolve([]); + } + return request("/chats/batch", { + account: true, + query: { ids: ids.join(",") }, + }); +} + export function enrichChat(chatId: number): Promise<{ job_id: number }> { return request<{ job_id: number }>(`/chats/${chatId}/enrich`, { method: "POST", diff --git a/frontend/src/lib/api/media.ts b/frontend/src/lib/api/media.ts index 4d2d34e..a315035 100644 --- a/frontend/src/lib/api/media.ts +++ b/frontend/src/lib/api/media.ts @@ -102,9 +102,30 @@ export function visualKind(kind: string): VisualKind { return "other"; } +const MAX_CACHED = 200; + const ready = new Map(); const inflight = new Map>(); +function remember( + cache: Map, + key: string, + item: InlineMedia +) { + cache.set(key, item); + while (cache.size > MAX_CACHED) { + const oldest = cache.keys().next(); + if (oldest.done) { + return; + } + const stale = cache.get(oldest.value); + cache.delete(oldest.value); + if (stale?.state === "ready") { + URL.revokeObjectURL(stale.url); + } + } +} + function cacheKey(account: number, chatId: number, messageId: number): string { return `${account}:${chatId}:${messageId}`; } @@ -183,7 +204,7 @@ export function loadMediaItem(media: MediaRef): Promise { const promise = resolveById(media) .then((result) => { if (result.state === "ready") { - byId.set(key, result); + remember(byId, key, result); } return result; }) @@ -251,7 +272,7 @@ export function loadInlineMedia( const promise = resolve(chatId, messageId) .then((result) => { if (result.state === "ready") { - ready.set(key, result); + remember(ready, key, result); } return result; }) diff --git a/frontend/src/lib/api/stories.ts b/frontend/src/lib/api/stories.ts index bf82d4d..4dea521 100644 --- a/frontend/src/lib/api/stories.ts +++ b/frontend/src/lib/api/stories.ts @@ -3,10 +3,27 @@ import { auth } from "$lib/stores/auth.svelte"; const BASE = import.meta.env.VITE_API_BASE ?? "/api"; +const MAX_CACHED = 100; + const ready = new Map(); const missing = new Set(); const inflight = new Map>(); +function remember(key: string, url: string) { + ready.set(key, url); + while (ready.size > MAX_CACHED) { + const oldest = ready.keys().next(); + if (oldest.done) { + return; + } + const stale = ready.get(oldest.value); + ready.delete(oldest.value); + if (stale) { + URL.revokeObjectURL(stale); + } + } +} + function authHeaders(): Record { return auth.token ? { Authorization: `Bearer ${auth.token}` } : {}; } @@ -21,7 +38,7 @@ async function fetchStoryMedia( const response = await fetch(url, { headers: authHeaders() }); if (response.ok) { const objectUrl = URL.createObjectURL(await response.blob()); - ready.set(key, objectUrl); + remember(key, objectUrl); return objectUrl; } missing.add(key); diff --git a/frontend/src/lib/components/MessageList.svelte b/frontend/src/lib/components/MessageList.svelte index 4768c37..14ac230 100644 --- a/frontend/src/lib/components/MessageList.svelte +++ b/frontend/src/lib/components/MessageList.svelte @@ -30,6 +30,7 @@ const SCROLL_THRESHOLD = 160; const STICK_OFFSET = 9; const IDLE_DELAY = 1500; + const MAX_MESSAGES = 360; let messages = $state([]); let loading = $state(true); @@ -176,6 +177,27 @@ } } + function trimTail() { + if (messages.length <= MAX_MESSAGES) { + return; + } + messages = messages.slice(0, MAX_MESSAGES); + hasNewer = true; + } + + async function trimHead() { + if (messages.length <= MAX_MESSAGES || container === null) { + return; + } + const el = container; + const prevHeight = el.scrollHeight; + const prevTop = el.scrollTop; + messages = messages.slice(messages.length - MAX_MESSAGES); + hasMore = true; + await tick(); + el.scrollTop = prevTop + (el.scrollHeight - prevHeight); + } + async function loadOlder() { if ( loadingOlder || @@ -204,6 +226,7 @@ ensurePeers(fresh); await tick(); el.scrollTop = prevTop + (el.scrollHeight - prevHeight); + trimTail(); } } finally { loadingOlder = false; @@ -230,6 +253,7 @@ messages = [...messages, ...fresh]; ensurePeers(fresh); await tick(); + await trimHead(); } } finally { loadingNewer = false; @@ -321,6 +345,7 @@ const stick = isNearBottom(); messages = [...messages, message]; ensurePeers([message]); + await trimHead(); if (stick) { await tick(); scrollToBottom(); @@ -391,6 +416,7 @@ ...appended, ]; ensurePeers(fresh); + await trimHead(); if (appended.length > 0 && stick) { await tick(); scrollToBottom(); diff --git a/frontend/src/lib/components/policy/PolicyEditor.svelte b/frontend/src/lib/components/policy/PolicyEditor.svelte index 9b32311..33ad1b7 100644 --- a/frontend/src/lib/components/policy/PolicyEditor.svelte +++ b/frontend/src/lib/components/policy/PolicyEditor.svelte @@ -92,11 +92,11 @@ }); $effect(() => { - for (const policy of chatPolicies) { - if (policy.scope_id !== null) { - chats.ensure(policy.scope_id); - } - } + chats.ensureMany( + chatPolicies + .map((policy) => policy.scope_id) + .filter((id): id is number => id !== null) + ); }); function folderTitle(id: number | null): string { diff --git a/frontend/src/lib/components/profile/ChatCalendar.svelte b/frontend/src/lib/components/profile/ChatCalendar.svelte index 95060b4..9dd5a70 100644 --- a/frontend/src/lib/components/profile/ChatCalendar.svelte +++ b/frontend/src/lib/components/profile/ChatCalendar.svelte @@ -2,7 +2,6 @@ import { goto } from "$app/navigation"; import { getChatCalendar, getMessageAt } from "$lib/api/endpoints"; import type { DayCount } from "$lib/api/types"; - import EmptyState from "$lib/components/ui/EmptyState.svelte"; import Icon from "$lib/components/ui/Icon.svelte"; import Spinner from "$lib/components/ui/Spinner.svelte"; import { accounts } from "$lib/stores/accounts.svelte"; @@ -46,6 +45,13 @@ let days = $state([]); let loading = $state(false); let cursor = $state(null); + const cache = new Map(); + + const now = new Date(); + const currentMonth: YearMonth = { + year: now.getUTCFullYear(), + month: now.getUTCMonth(), + }; const byKey = $derived( new Map(days.map((day) => [day.day.slice(0, 10), day.count])) @@ -54,33 +60,27 @@ days.reduce((max, day) => Math.max(max, day.count), 0) ); - const minMonth = $derived.by(() => { - if (days.length === 0) { - return null; - } - const date = new Date(days[0].day); - return { year: date.getUTCFullYear(), month: date.getUTCMonth() }; - }); - const maxMonth = $derived.by(() => { - if (days.length === 0) { - return null; - } - const date = new Date(days.at(-1)?.day ?? days[0].day); - return { year: date.getUTCFullYear(), month: date.getUTCMonth() }; - }); - - const view = $derived(cursor ?? maxMonth); + const view = $derived(cursor ?? currentMonth); function index(value: YearMonth): number { return value.year * 12 + value.month; } - const canPrev = $derived( - Boolean(view && minMonth && index(view) > index(minMonth)) - ); - const canNext = $derived( - Boolean(view && maxMonth && index(view) < index(maxMonth)) - ); + const canNext = $derived(index(view) < index(currentMonth)); + + function monthKey(value: YearMonth): string { + return `${value.year}-${value.month}`; + } + + function monthRange(value: YearMonth): { + date_from: string; + date_to: string; + } { + return { + date_from: new Date(Date.UTC(value.year, value.month, 1)).toISOString(), + date_to: new Date(Date.UTC(value.year, value.month + 1, 1)).toISOString(), + }; + } function level(count: number): number { if (count <= 0 || maxCount <= 0) { @@ -92,35 +92,17 @@ ); } - function clamp(value: YearMonth): YearMonth { - if (minMonth && index(value) < index(minMonth)) { - return minMonth; - } - if (maxMonth && index(value) > index(maxMonth)) { - return maxMonth; - } - return value; - } - function shift(months: number) { - if (!view) { - return; - } - const total = index(view) + months; - cursor = clamp({ year: Math.floor(total / 12), month: total % 12 }); + const total = Math.min(index(view) + months, index(currentMonth)); + cursor = { year: Math.floor(total / 12), month: total % 12 }; } const lead = $derived( - view - ? (new Date(Date.UTC(view.year, view.month, 1)).getUTCDay() + 6) % 7 - : 0 + (new Date(Date.UTC(view.year, view.month, 1)).getUTCDay() + 6) % 7 ); const blanks = $derived(Array.from({ length: lead }, (_, i) => i)); const cells = $derived.by(() => { - if (!view) { - return []; - } const count = new Date(Date.UTC(view.year, view.month + 1, 0)).getUTCDate(); const result: DayCell[] = []; for (let day = 1; day <= count; day++) { @@ -133,17 +115,30 @@ return result; }); + $effect(() => { + const _id = chatId; + const _account = accounts.selectedId; + cache.clear(); + cursor = null; + days = []; + }); + $effect(() => { const _id = chatId; if (accounts.selectedId === null) { return; } + const target = view; + const cached = cache.get(monthKey(target)); + if (cached !== undefined) { + days = cached; + return; + } let active = true; loading = true; - days = []; - cursor = null; - getChatCalendar(chatId) + getChatCalendar(chatId, monthRange(target)) .then((result) => { + cache.set(monthKey(target), result); if (active) { days = result; } @@ -175,8 +170,6 @@ {#if loading && days.length === 0}
-{:else if !view} - {:else}
@@ -184,7 +177,6 @@ type="button" class="step" onclick={() => shift(-12)} - disabled={!canPrev} aria-label="Предыдущий год" > « @@ -193,7 +185,6 @@ type="button" class="step" onclick={() => shift(-1)} - disabled={!canPrev} aria-label="Предыдущий месяц" > diff --git a/frontend/src/lib/components/shares/SharesPanel.svelte b/frontend/src/lib/components/shares/SharesPanel.svelte index 651f97c..5c7db9c 100644 --- a/frontend/src/lib/components/shares/SharesPanel.svelte +++ b/frontend/src/lib/components/shares/SharesPanel.svelte @@ -35,6 +35,7 @@ let items = $state([]); let loading = $state(false); let filter = $state("active"); + let fetchedAll = $state(false); let expanded = $state(null); let token = 0; @@ -44,16 +45,20 @@ : items ); const activeCount = $derived( - items.filter((item) => shareState(item) === "active").length + fetchedAll + ? items.filter((item) => shareState(item) === "active").length + : items.length ); async function load() { const current = ++token; + const activeOnly = filter === "active"; loading = true; try { - const rows = await listShares(); + const rows = await listShares(activeOnly); if (current === token) { items = rows; + fetchedAll = !activeOnly; } } catch { if (current === token) { @@ -118,7 +123,7 @@ let loadedKey = ""; $effect(() => { - const key = `${accounts.selectedId}:${shareUi.revision}`; + const key = `${accounts.selectedId}:${shareUi.revision}:${filter}`; if (accounts.selectedId !== null && key !== loadedKey) { loadedKey = key; load(); diff --git a/frontend/src/lib/components/stories/AllStoriesArchive.svelte b/frontend/src/lib/components/stories/AllStoriesArchive.svelte index 89f5184..e83fde8 100644 --- a/frontend/src/lib/components/stories/AllStoriesArchive.svelte +++ b/frontend/src/lib/components/stories/AllStoriesArchive.svelte @@ -2,7 +2,7 @@ import { untrack } from "svelte"; import { enqueueStoriesBackfill, - getChat, + getChatsBatch, getPeers, getStories, } from "$lib/api/endpoints"; @@ -63,17 +63,13 @@ const chatIds = [...byPeer.keys()].filter((id) => id < 0); const [peers, fetched] = await Promise.all([ getPeers(peerIds), - Promise.all(chatIds.map((id) => getChat(id))), + getChatsBatch(chatIds), ]); if (current !== token) { return; } const peerById = new Map(peers.map((peer) => [peer.peer_id, peer])); - const chatById = new Map( - fetched - .filter((item) => item !== null) - .map((item) => [item.chat_id, item]) - ); + const chatById = new Map(fetched.map((item) => [item.chat_id, item])); groups = [...byPeer.entries()].map(([peerId, stories]) => { if (peerId > 0) { const peer = peerById.get(peerId) ?? null; diff --git a/frontend/src/lib/stores/chats.svelte.ts b/frontend/src/lib/stores/chats.svelte.ts index eb4c450..332cb99 100644 --- a/frontend/src/lib/stores/chats.svelte.ts +++ b/frontend/src/lib/stores/chats.svelte.ts @@ -1,4 +1,10 @@ -import { enrichChat, getChat, getJob, listChats } from "$lib/api/endpoints"; +import { + enrichChat, + getChat, + getChatsBatch, + getJob, + listChats, +} from "$lib/api/endpoints"; import type { Chat, LiveEvent } from "$lib/api/types"; import { folderContains } from "$lib/format/folders"; import { accounts } from "$lib/stores/accounts.svelte"; @@ -245,6 +251,32 @@ function createChats() { }) .catch(() => resolving.delete(id)); }, + ensureMany(ids: number[]) { + syncAccount(); + if (account === null) { + return; + } + const missing = [ + ...new Set(ids.filter((id) => !(resolving.has(id) || known(id)))), + ]; + if (missing.length === 0) { + return; + } + for (const id of missing) { + resolving.add(id); + } + getChatsBatch(missing) + .then((fetched) => { + for (const chat of fetched) { + extra[chat.chat_id] = chat; + } + }) + .catch(() => { + for (const id of missing) { + resolving.delete(id); + } + }); + }, load(folderId: number | null = folders.selectedId) { return load(folderId, false); },