diff --git a/backend/src/utils/jobs.py b/backend/src/utils/jobs.py index c80876b..5c31db3 100644 --- a/backend/src/utils/jobs.py +++ b/backend/src/utils/jobs.py @@ -1,4 +1,5 @@ import json +from datetime import timedelta from typing import Any import asyncpg @@ -7,15 +8,22 @@ JOBS_CHANGED_CHANNEL = "jobs_changed" async def enqueue( - pool: asyncpg.Pool, account_id: int, kind: str, params: dict[str, Any] + pool: asyncpg.Pool, + account_id: int, + kind: str, + params: dict[str, Any], + *, + cooldown: timedelta | None = None, ) -> 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", + "AND params = $3::jsonb AND (status IN ('pending', 'running') " + "OR created_at > now() - $4::interval) LIMIT 1", account_id, kind, params_json, + cooldown, ) if existing is not None: return existing diff --git a/backend/src/watcher/registry.py b/backend/src/watcher/registry.py index 8e01163..49befa8 100644 --- a/backend/src/watcher/registry.py +++ b/backend/src/watcher/registry.py @@ -1,6 +1,7 @@ import asyncio import contextlib from dataclasses import dataclass +from datetime import timedelta import asyncpg from pyrogram import Client, enums @@ -30,6 +31,7 @@ TOKEN_MISSING = "Токен не найден на диске" # noqa: S105 INVALID_TOKEN_ERRORS = ("ACCESS_TOKEN_INVALID", "ACCESS_TOKEN_EXPIRED") BACKLOG_ATTEMPTS = 3 BACKLOG_RETRY_PAUSE = 60 +ENRICH_COOLDOWN = timedelta(hours=6) @dataclass @@ -70,14 +72,24 @@ def _handlers(pool: asyncpg.Pool) -> tuple: ) if message.chat.type is not enums.ChatType.PRIVATE: await enqueue( - pool, ctx.account_id, "enrich_chat", {"chat_id": meta.chat_id} + pool, + ctx.account_id, + "enrich_chat", + {"chat_id": meta.chat_id}, + cooldown=ENRICH_COOLDOWN, ) async def on_member(client: Client, update) -> None: # noqa: ANN001 ctx = getattr(client, "capture", None) if ctx is None or update.chat is None: return - await enqueue(pool, ctx.account_id, "enrich_chat", {"chat_id": update.chat.id}) + await enqueue( + pool, + ctx.account_id, + "enrich_chat", + {"chat_id": update.chat.id}, + cooldown=ENRICH_COOLDOWN, + ) return ( MessageHandler(on_message),