perf(core): improve async handling and caching performance

Convert delayed_delete_file to async and use asyncio.sleep.
Add pre‑semaphore cache hit to serve cached files without acquiring the download semaphore.
Set SQLite busy_timeout to reduce lock errors.
Introduce a diagnostics counter for pending _persist_media_file_id_async tasks.
Refactor flag extraction to accept pre‑generated HTML and compute html body once.
Batch reply enrichment in RSS generator to minimize API calls.
Store cached messages directly in tg_cache and add fallback for legacy double‑pickle format.

BREAKING CHANGE: delayed_delete_file is now async and must be awaited.

api_server.py

@@ -190,13 +190,9 @@ async def find_file_id_in_message(message: Message, file_unique_id: str) -> Unio
     return None
 
 
-def delayed_delete_file(file_path: str, delay: int = 300) -> None:
-    """
-    Delete temporary file after a delay to ensure complete file delivery.
-    Delay is set to 5 minutes by default.
-    NOTE: Runs in a background thread via Starlette BackgroundTask.
-    """
-    time.sleep(delay)
+async def delayed_delete_file(file_path: str, delay: int = 300) -> None:
+    """Delete a temporary file after a delay. Runs as an async background task."""
+    await asyncio.sleep(delay)
     try:
         os.remove(file_path)
         logger.info(f"Deleted temporary file {file_path} after delay of {delay} seconds")
@@ -955,6 +951,29 @@ async def get_media(channel: str, post_id: int, file_unique_id: str, request: Re
             
         try: # Wrap the download and prepare call
             import time as _time
+
+            # Pre-semaphore cache check: serve already-cached files without acquiring the semaphore
+            base_cache_dir = os.path.abspath("./data/cache")
+            channel_dir = os.path.join(base_cache_dir, str(channel))
+            post_dir = os.path.join(channel_dir, str(post_id))
+            cache_path = os.path.join(post_dir, file_unique_id)
+            if os.path.exists(cache_path) and os.path.getsize(cache_path) > 0:
+                # File is already in cache — skip semaphore and serve directly
+                logger.info(f"pre_semaphore_cache_hit: {channel}/{post_id}/{file_unique_id}")
+                # Fire-and-forget timestamp update with error handling to avoid silent failures
+                async def _update_access(_ch, _pid, _fid):
+                    try:
+                        await asyncio.to_thread(
+                            update_media_file_access_sync,
+                            DB_PATH, str(_ch), _pid, _fid,
+                            datetime.now().timestamp()
+                        )
+                    except Exception as _e:
+                        logger.warning(f"Failed to update access time for {_ch}/{_pid}/{_fid}: {_e}")
+                asyncio.create_task(_update_access(channel, post_id, file_unique_id))
+                return await prepare_file_response(cache_path, request=request,
+                                                   media_key=(str(channel), post_id, file_unique_id))
+
             _sem_wait_start = _time.monotonic()
             async with HTTP_DOWNLOAD_SEMAPHORE:  # limit concurrent live HTTP downloads
                 _sem_wait = _time.monotonic() - _sem_wait_start

file_io.py

@@ -17,6 +17,7 @@ def _open_db(db_path: str) -> sqlite3.Connection:
     """Open a SQLite connection with WAL journal mode enabled."""
     conn = sqlite3.connect(db_path)
     conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA busy_timeout = 5000")  # Wait up to 5 seconds on lock
     return conn
 
 

post_parser.py

@@ -28,6 +28,9 @@ Config = get_settings()
 
 logger = logging.getLogger(__name__)
 
+# Module-level counter for in-flight persist tasks (used only in diagnostics)
+_persist_pending_count = 0
+
 
 #tests
 #http://127.0.0.1:8000/post/html/DragorWW_space/114 — video
@@ -325,11 +328,15 @@ class PostParser:
             return result
         return None
 
-    def _extract_flags(self, message: Message) -> List[str]: #Tests: tests/postparser_extract_flags.py
+    def _extract_flags(self, message: Message, html_body: Optional[str] = None) -> List[str]: #Tests: tests/postparser_extract_flags.py
         # Use raw text/caption for some checks before HTML processing
         message_text_str = str(message.text or message.caption or '')
         # Use HTML body for checks involving formatted text or links within HTML
-        message_body_html = self._generate_html_body(message)
+        # If html_body is provided (pre-computed), use it directly to avoid redundant generation
+        if html_body is None:
+            message_body_html = self._generate_html_body(message)
+        else:
+            message_body_html = html_body
         flags = []
 
         # Add "fwd" flag for forwarded messages
@@ -491,6 +498,10 @@ class PostParser:
         return return_html
 
     def process_message(self, message: Message) -> Dict[Any, Any]:
+        # Compute html body once — avoids triple _generate_html_body calls
+        html_body = self._generate_html_body(message)
+        flags = self._extract_flags(message, html_body=html_body)
+        footer = self.generate_html_footer(message, flags_list=flags)
         result = {
             'channel': self.get_channel_username(message),
             'message_id': message.id,
@@ -499,10 +510,10 @@ class PostParser:
             'text': message.text or message.caption or '',
             'html': {
                 'title': self._generate_title(message),
-                'body': self._generate_html_body(message),
-                'footer': self.generate_html_footer(message)
+                'body': html_body,
+                'footer': footer
             },
-            'flags': self._extract_flags(message),
+            'flags': flags,
             'author': self._get_author_info(message),
             'views': message.views,
             'reactions': self._extract_reactions(message),
@@ -992,16 +1003,21 @@ class PostParser:
                     try:
                         loop = asyncio.get_running_loop()
                         if loop.is_running():
+                            global _persist_pending_count
                             task = loop.create_task(
                                 self._persist_media_file_id_async(channel_username, message.id, file_unique_id, added_ts)
                             )
-                            # Count pending fire-and-forget tasks to surface backlog issues
-                            all_tasks = asyncio.all_tasks(loop)
-                            persist_tasks = [t for t in all_tasks if '_persist_media_file_id_async' in str(t.get_coro())]
-                            task_count = len(persist_tasks)
-                            if task_count > 5:
-                                logger.warning(f"persist_task_queue: {task_count} pending _persist_media_file_id_async tasks (msg {message.id})")
-                            logger.debug(f"persist_task_created: queued for {channel_username}/{message.id}/{file_unique_id}, total pending: {task_count}")
+                            _persist_pending_count += 1
+
+                            def _on_task_done(t):
+                                global _persist_pending_count
+                                _persist_pending_count -= 1
+
+                            task.add_done_callback(_on_task_done)
+
+                            if _persist_pending_count > 5:
+                                logger.warning(f"persist_task_queue: {_persist_pending_count} pending _persist_media_file_id_async tasks (msg {message.id})")
+                            logger.debug(f"persist_task_created: queued for {channel_username}/{message.id}/{file_unique_id}, total pending: {_persist_pending_count}")
                         else:
                             # Fallback sync path (should not occur during normal FastAPI runtime)
                             upsert_media_file_id_sync(DB_PATH, channel_username, message.id, file_unique_id, added_ts)

rss_generator.py

@@ -477,17 +477,47 @@ async def generate_channel_rss(channel: str | int,
 
 async def _reply_enrichment(client: Client, messages: list[Message]) -> list[Message]:
     """
-    Enrich messages with reply to messages
+    Enrich messages with reply-to messages.
+
+    Instead of one API call per message, replies are batched: all message IDs
+    that need enrichment are grouped by chat_id and fetched in a single
+    client.get_messages() call per chat_id.
     """
+    # Collect messages that need reply enrichment, grouped by chat_id
+    chat_messages: dict[int, list[Message]] = {}
     for message in messages:
         if message.reply_to_message_id and message.chat:
-            full_message = await client.get_messages(message.chat.id, message.id)
-            if isinstance(full_message, list):
-                if full_message and full_message[0].reply_to_message:
-                    message.reply_to_message = full_message[0].reply_to_message
-            else:
-                if full_message and full_message.reply_to_message:
-                    message.reply_to_message = full_message.reply_to_message
+            chat_id = message.chat.id
+            if chat_id not in chat_messages:
+                chat_messages[chat_id] = []
+            chat_messages[chat_id].append(message)
+
+    if not chat_messages:
+        # No messages with replies — return unchanged
+        return messages
+
+    # Build a lookup: {(chat_id, message_id): full_message} using one batch call per chat
+    reply_lookup: dict[tuple[int, int], Message] = {}
+    for chat_id, chat_msgs in chat_messages.items():
+        ids_to_fetch = [m.id for m in chat_msgs]
+        try:
+            fetched = await client.get_messages(chat_id, ids_to_fetch)
+            # get_messages may return a single Message or a list
+            if not isinstance(fetched, list):
+                fetched = [fetched]
+            for fm in fetched:
+                if fm and not getattr(fm, 'empty', False):
+                    reply_lookup[(chat_id, fm.id)] = fm
+        except Exception as e:
+            logger.error(f"reply_enrichment_batch_error: chat_id {chat_id}, ids {ids_to_fetch}, error {str(e)}")
+
+    # Apply reply_to_message from lookup
+    for message in messages:
+        if message.reply_to_message_id and message.chat:
+            key = (message.chat.id, message.id)
+            full_message = reply_lookup.get(key)
+            if full_message and full_message.reply_to_message:
+                message.reply_to_message = full_message.reply_to_message
 
     return messages
 

tg_cache.py

@@ -42,11 +42,11 @@ def _save_history_to_cache(channel_id: Union[str, int], messages: List[Message],
     try:
         cache_file = _get_history_cache_file_path(channel_id)
         
-        # Create cache metadata
+        # Create cache metadata — store messages directly (no inner pickle.dumps)
         cache_data = {
             'timestamp': time.time(),
             'limit': limit,
-            'messages': pickle.dumps(messages)
+            'messages': messages
         }
         
         with open(cache_file, 'wb') as f:
@@ -94,8 +94,12 @@ def _get_history_from_cache(channel_id: Union[str, int], limit: int, max_age_hou
             logger.info(f"history_cache_limit_mismatch: channel {channel_id}, cached limit {cached_limit}, requested limit {limit}")
             return None
         
-        # Restore message list from pickle
-        messages = pickle.loads(cache_data['messages'])
+        # Restore message list; handle old cache files that used double-pickle (bytes = old format)
+        raw = cache_data['messages']
+        if isinstance(raw, bytes):
+            messages = pickle.loads(raw)
+        else:
+            messages = raw
         logger.info(f"history_cache_hit: channel {channel_id}, limit {limit}, messages {len(messages)}, age {cache_age:.1f}s")
         return messages