feat: mcp and backfill fixes

backend/src/userbot/modules/jobs/context.py

@@ -24,3 +24,6 @@ class JobContext:
     async def report_progress(self, progress: dict[str, Any]) -> None:
         self.job.progress = progress
         await repository.report_progress(self.pool, self.job_id, progress)
+
+    async def is_canceled(self) -> bool:
+        return await repository.is_canceled(self.pool, self.job_id)

backend/src/userbot/modules/jobs/handlers/backfill.py

@@ -1,8 +1,12 @@
 from pyrogram.errors import PeerIdInvalid
 
 from userbot.modules.capture import capture_message
+from userbot.modules.capture.chat_meta import meta_from_chat
 from userbot.modules.jobs.context import JobContext
 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.policy.models import CaptureToggles
 
 SAVE_EVERY = 100
@@ -25,14 +29,24 @@ async def backfill(ctx: JobContext) -> None:
     max_id = (ctx.job.cursor or {}).get("max_id", 0)
     processed = ctx.job.progress.get("processed", 0)
     kwargs = {"max_id": max_id} if max_id else {}
+    self_id = client.me.id if client.me else None
     try:
         async for message in client.get_chat_history(chat_id, **kwargs):
             await capture_message(client, message, capture, toggles)
+            if should_transcribe_on_backfill(message, self_id) and message.chat:
+                meta = meta_from_chat(message.chat, capture.contacts.ids)
+                already = await stt_repo.is_transcribed(
+                    capture.pool, capture.account_id, chat_id, message.id
+                )
+                if capture.resolve(meta).stt and not already:
+                    await safe_transcribe(client, capture, chat_id, message.id)
             processed += 1
             if processed % SAVE_EVERY == 0:
                 next_max = message.id - 1
                 await ctx.save_cursor({"max_id": next_max})
                 await ctx.report_progress({"processed": processed, "max_id": next_max})
+                if await ctx.is_canceled():
+                    return
     except PeerIdInvalid:
         await ctx.report_progress({"processed": processed, "error": "peer_id_invalid"})
         return

backend/src/userbot/modules/jobs/repository.py

@@ -79,13 +79,18 @@ async def finish(
 ) -> None:
     await pool.execute(
         "UPDATE jobs SET status = $2, error = $3, finished_at = now(), "
-        "updated_at = now() WHERE id = $1",
+        "updated_at = now() WHERE id = $1 AND status = 'running'",
         job_id,
         status.value,
         error,
     )
 
 
+async def is_canceled(pool: asyncpg.Pool, job_id: int) -> bool:
+    status = await pool.fetchval("SELECT status FROM jobs WHERE id = $1", job_id)
+    return status == JobStatus.CANCELED.value
+
+
 async def get_job(pool: asyncpg.Pool, job_id: int) -> Job | None:
     row = await pool.fetchrow("SELECT * FROM jobs WHERE id = $1", job_id)
     return _row_to_job(row) if row else None

backend/src/userbot/modules/stt/__init__.py

@@ -1,3 +1,7 @@
-from userbot.modules.stt.service import is_transcribable, transcribe_message
+from userbot.modules.stt.service import (
+    is_transcribable,
+    should_transcribe_on_backfill,
+    transcribe_message,
+)
 
-__all__ = ["is_transcribable", "transcribe_message"]
+__all__ = ["is_transcribable", "should_transcribe_on_backfill", "transcribe_message"]

backend/src/userbot/modules/stt/repository.py

@@ -9,6 +9,11 @@ UPDATE media SET extracted_text = $4
 WHERE account_id = $1 AND chat_id = $2 AND message_id = $3
 """
 
+_IS_TRANSCRIBED = """
+SELECT extracted_text IS NOT NULL FROM media
+WHERE account_id = $1 AND chat_id = $2 AND message_id = $3
+"""
+
 _VOICE_READS_BOX = """
 SELECT md.chat_id, md.message_id, m.sender_id,
     md.extracted_text IS NULL AS untranscribed
@@ -34,6 +39,12 @@ async def set_extracted_text(
     await pool.execute(_SET_EXTRACTED_TEXT, account_id, chat_id, message_id, text)
 
 
+async def is_transcribed(
+    pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int
+) -> bool:
+    return bool(await pool.fetchval(_IS_TRANSCRIBED, account_id, chat_id, message_id))
+
+
 async def voice_reads(
     pool: asyncpg.Pool,
     account_id: int,

backend/src/userbot/modules/stt/service.py

@@ -4,6 +4,7 @@ from pyrogram.types import Message
 from userbot.modules.capture.context import CaptureContext
 from userbot.modules.media import self_destruct_ttl
 from userbot.modules.stt import repository
+from utils.logging import logger
 
 
 def is_transcribable(message: Message) -> bool:
@@ -12,6 +13,19 @@ def is_transcribable(message: Message) -> bool:
     return message.voice is not None or message.video_note is not None
 
 
+def should_transcribe_on_backfill(message: Message, self_id: int | None) -> bool:
+    if not is_transcribable(message):
+        return False
+    if message.outgoing:
+        return True
+    sender = message.from_user.id if message.from_user else None
+    if sender is None and message.sender_chat is not None:
+        sender = message.sender_chat.id
+    if sender == self_id:
+        return True
+    return not message.unread_media
+
+
 async def transcribe_message(
     client: Client, ctx: CaptureContext, chat_id: int, message_id: int
 ) -> None:
@@ -21,7 +35,18 @@ async def transcribe_message(
     result = await client.invoke(
         raw.functions.messages.TranscribeAudio(peer=peer, msg_id=message_id)
     )
-    if not result.pending and result.text:
+    if result.pending:
+        logger.info(
+            f"[yellow]STT pending {chat_id}/{message_id} "
+            f"(trial_remains={result.trial_remains_num})[/]"
+        )
+        return
+    if result.text:
         await repository.set_extracted_text(
             ctx.pool, ctx.account_id, chat_id, message_id, result.text
         )
+    else:
+        logger.info(
+            f"[yellow]STT empty {chat_id}/{message_id} "
+            f"(trial_remains={result.trial_remains_num})[/]"
+        )