Use bot MTProto and Bot API for Telegram backup operations.

Backup restore runs via the bot token without a user scraping session;
backup store tries bot copyMessage first and falls back to user MTProto
when needed.

Author: mhdzumair

backend/src/bot/api.rs

@@ -193,16 +193,43 @@ impl BotApi {
         from_chat_id: i64,
         message_id: i64,
     ) -> Result<Value, BotApiError> {
-        let data = self
-            .post(
-                "copyMessage",
-                json!({
-                    "chat_id": chat_id,
-                    "from_chat_id": from_chat_id,
-                    "message_id": message_id,
-                }),
-            )
-            .await?;
+        self.copy_message_with_caption(chat_id, from_chat_id, message_id, None)
+            .await
+    }
+
+    pub async fn copy_message_with_caption(
+        &self,
+        chat_id: i64,
+        from_chat_id: i64,
+        message_id: i64,
+        caption: Option<&str>,
+    ) -> Result<Value, BotApiError> {
+        self.copy_message_with_caption_json(
+            json!(chat_id),
+            json!(from_chat_id),
+            message_id,
+            caption,
+        )
+        .await
+    }
+
+    pub async fn copy_message_with_caption_json(
+        &self,
+        chat_id: Value,
+        from_chat_id: Value,
+        message_id: i64,
+        caption: Option<&str>,
+    ) -> Result<Value, BotApiError> {
+        let mut body = json!({
+            "chat_id": chat_id,
+            "from_chat_id": from_chat_id,
+            "message_id": message_id,
+        });
+        if let Some(text) = caption {
+            body["caption"] = json!(text);
+            body["parse_mode"] = json!("Markdown");
+        }
+        let data = self.post("copyMessage", body).await?;
         data.get("result")
             .cloned()
             .ok_or_else(|| BotApiError::Parse("missing result".into()))

backend/src/jobs/handlers/telegram_backup_ops.rs

@@ -1,8 +1,13 @@
 //! Admin jobs: copy Telegram streams to the backup channel or restore DB rows from it.
 
 use async_trait::async_trait;
+use std::collections::HashMap;
+use std::sync::Arc;
 use tracing::info;
 
+use grammers_client::Client;
+use grammers_session::types::PeerRef;
+
 use crate::{
     db::{
         telegram::{TelegramStreamBackupRow, list_streams_for_backup_store},
@@ -12,9 +17,10 @@ use crate::{
         error::JobError,
         handler::{JobCtx, JobHandler},
     },
+    scrapers::telegram_clients::is_auth_key_duplicated,
     services::telegram_backup::{
-        BackupBatchMetrics, resolve_mtproto_client, restore_stream_from_backup_message,
-        store_stream_to_backup,
+        BackupBatchMetrics, resolve_bot_mtproto_client, resolve_mtproto_client,
+        restore_stream_from_backup_message, store_stream_to_backup,
     },
     services::telegram_peer,
 };
@@ -71,10 +77,42 @@ fn default_capture_file_id() -> bool {
     true
 }
 
+async fn load_user_session(
+    ctx: &JobCtx,
+    preferred: Option<UserId>,
+) -> (Option<Arc<Client>>, Option<HashMap<i64, PeerRef>>) {
+    for attempt in 0..2 {
+        let session = resolve_mtproto_client(&ctx.state, preferred).await.ok();
+        let Some((user_id, client)) = session else {
+            return (None, None);
+        };
+
+        let (dialog_peers, dialog_error) =
+            telegram_peer::load_dialog_peer_map(client.as_ref()).await;
+        if let Some(err) = dialog_error.as_deref()
+            && attempt == 0
+            && is_auth_key_duplicated(err)
+        {
+            tracing::warn!(
+                "telegram_backup_store: AUTH_KEY_DUPLICATED for user {} — recycling client",
+                user_id.0
+            );
+            ctx.state.telegram_clients.invalidate(user_id).await;
+            telegram_peer::invalidate_dialog_peer_cache(user_id).await;
+            continue;
+        }
+
+        return (Some(client), Some(dialog_peers));
+    }
+
+    (None, None)
+}
+
 async fn run_backup_store_batch(
     ctx: &JobCtx,
     args: &TelegramBackupStoreArgs,
-    client: &grammers_client::Client,
+    user_client: Option<&Client>,
+    user_dialog_peers: Option<&HashMap<i64, PeerRef>>,
     after_id: i32,
 ) -> Result<(BackupBatchMetrics, i32), JobError> {
     let batch_size = args.batch_size.clamp(1, 200);
@@ -86,7 +124,6 @@ async fn run_backup_store_batch(
         return Ok((BackupBatchMetrics::default(), after_id));
     }
 
-    let (dialog_peers, _) = telegram_peer::load_dialog_peer_map(client).await;
     let mut metrics = BackupBatchMetrics::default();
     let mut last_id = after_id;
 
@@ -96,7 +133,15 @@ async fn run_backup_store_batch(
         }
         last_id = row.id;
         metrics.processed += 1;
-        match store_one(ctx, client, &dialog_peers, &row, args.capture_file_id).await {
+        match store_one(
+            ctx,
+            user_client,
+            user_dialog_peers,
+            &row,
+            args.capture_file_id,
+        )
+        .await
+        {
             Ok(true) => metrics.stored += 1,
             Ok(false) => metrics.skipped += 1,
             Err(e) => {
@@ -111,12 +156,19 @@ async fn run_backup_store_batch(
 
 async fn store_one(
     ctx: &JobCtx,
-    client: &grammers_client::Client,
-    dialog_peers: &std::collections::HashMap<i64, grammers_session::types::PeerRef>,
+    user_client: Option<&Client>,
+    user_dialog_peers: Option<&HashMap<i64, PeerRef>>,
     row: &TelegramStreamBackupRow,
     capture_file_id: bool,
 ) -> Result<bool, String> {
-    store_stream_to_backup(&ctx.state, client, dialog_peers, row, capture_file_id).await?;
+    store_stream_to_backup(
+        &ctx.state,
+        user_client,
+        user_dialog_peers,
+        row,
+        capture_file_id,
+    )
+    .await?;
     Ok(true)
 }
 
@@ -127,9 +179,11 @@ impl JobHandler for TelegramBackupStore {
     type Args = TelegramBackupStoreArgs;
 
     async fn run(&self, args: Self::Args, ctx: JobCtx) -> Result<(), JobError> {
-        if !ctx.state.telegram_clients.api_configured() {
+        if ctx.state.config.telegram_bot_token.is_none()
+            && !ctx.state.telegram_clients.api_configured()
+        {
             return Err(JobError::other(
-                "Telegram API credentials are not configured",
+                "Configure TELEGRAM_BOT_TOKEN or Telegram API credentials with a scraping session",
             ));
         }
         if ctx
@@ -146,9 +200,7 @@ impl JobHandler for TelegramBackupStore {
         }
 
         let preferred = args.mediafusion_user_id.map(UserId);
-        let (_user_id, client) = resolve_mtproto_client(&ctx.state, preferred)
-            .await
-            .map_err(JobError::other)?;
+        let (user_client, user_dialog_peers) = load_user_session(&ctx, preferred).await;
 
         let mut after_id = args.after_id;
         let mut totals = BackupBatchMetrics::default();
@@ -159,8 +211,14 @@ impl JobHandler for TelegramBackupStore {
                 return Err(JobError::Cancelled);
             }
 
-            let (batch, last_id) =
-                run_backup_store_batch(&ctx, &args, client.as_ref(), after_id).await?;
+            let (batch, last_id) = run_backup_store_batch(
+                &ctx,
+                &args,
+                user_client.as_deref(),
+                user_dialog_peers.as_ref(),
+                after_id,
+            )
+            .await?;
             if batch.processed == 0 {
                 break;
             }
@@ -197,6 +255,9 @@ impl JobHandler for TelegramBackupRestore {
     type Args = TelegramBackupRestoreArgs;
 
     async fn run(&self, args: Self::Args, ctx: JobCtx) -> Result<(), JobError> {
+        if ctx.state.config.telegram_bot_token.is_none() {
+            return Err(JobError::other("TELEGRAM_BOT_TOKEN is not configured"));
+        }
         if !ctx.state.telegram_clients.api_configured() {
             return Err(JobError::other(
                 "Telegram API credentials are not configured",
@@ -211,21 +272,21 @@ impl JobHandler for TelegramBackupRestore {
             .filter(|s| !s.is_empty())
             .ok_or_else(|| JobError::other("TELEGRAM_BACKUP_CHANNEL_ID is not configured"))?;
 
-        let preferred = args.mediafusion_user_id.map(UserId);
-        let (_user_id, client) = resolve_mtproto_client(&ctx.state, preferred)
+        let _ = args.mediafusion_user_id;
+        let client = resolve_bot_mtproto_client(&ctx.state)
             .await
             .map_err(JobError::other)?;
 
-        let (dialog_peers, _) = telegram_peer::load_dialog_peer_map(&client).await;
+        let (dialog_peers, _) = telegram_peer::load_dialog_peer_map(client.as_ref()).await;
         let (_, backup_peer_ref) = telegram_peer::resolve_channel_peer(
-            &client,
+            client.as_ref(),
             backup_channel,
             &dialog_peers,
         )
         .await
         .ok_or_else(|| {
             JobError::other(format!(
-                "backup channel {backup_channel} is not accessible with the scraping session"
+                "backup channel {backup_channel} is not accessible to the bot — add the bot as admin with read history"
             ))
         })?;
 

backend/src/scrapers/telegram_clients.rs

@@ -1,4 +1,4 @@
-//! Per-user Telegram MTProto client pool.
+//! Per-user Telegram MTProto client pool and shared bot MTProto client.
 
 use std::sync::Arc;
 use std::time::Duration;
@@ -7,6 +7,8 @@ use grammers_client::Client;
 use grammers_client::sender::SenderPool;
 use grammers_session::storages::MemorySession;
 use moka::future::Cache;
+use std::sync::OnceLock;
+use tokio::sync::Mutex;
 use tokio::task::JoinHandle;
 
 use crate::{
@@ -30,6 +32,8 @@ pub fn is_auth_key_duplicated(err: &str) -> bool {
     err.contains("AUTH_KEY_DUPLICATED")
 }
 
+static BOT_CLIENT: OnceLock<Mutex<Option<Arc<CachedClient>>>> = OnceLock::new();
+
 pub struct TelegramClientPool {
     cache: Cache<UserId, Arc<CachedClient>>,
     config: AppConfig,
@@ -80,6 +84,54 @@ impl TelegramClientPool {
         }
         self.cache.invalidate(&user_id).await;
     }
+
+    /// MTProto client signed in with `TELEGRAM_BOT_TOKEN` for backup-channel operations.
+    pub async fn get_bot_client(&self) -> Option<Arc<Client>> {
+        if !self.api_configured() {
+            return None;
+        }
+        let bot_token = self.config.telegram_bot_token.as_deref()?;
+        let api_id = self.config.telegram_api_id?;
+        let api_hash = self.config.telegram_api_hash.as_deref()?;
+
+        let slot = BOT_CLIENT.get_or_init(|| Mutex::new(None));
+        let mut guard = slot.lock().await;
+        if let Some(cached) = guard.as_ref() {
+            return Some(Arc::clone(&cached.client));
+        }
+
+        match build_bot_client(api_id, api_hash, bot_token).await {
+            Ok(cached) => {
+                let client = Arc::clone(&cached.client);
+                *guard = Some(cached);
+                Some(client)
+            }
+            Err(e) => {
+                tracing::warn!("telegram: bot MTProto client init failed: {e}");
+                None
+            }
+        }
+    }
+}
+
+async fn build_bot_client(
+    api_id: i32,
+    api_hash: &str,
+    bot_token: &str,
+) -> Result<Arc<CachedClient>, Box<dyn std::error::Error + Send + Sync>> {
+    let session = Arc::new(MemorySession::default());
+    let pool = SenderPool::new(Arc::clone(&session) as Arc<_>, api_id);
+    let runner = pool.runner;
+    let handle = pool.handle;
+    let runner_task = tokio::spawn(async move {
+        runner.run().await;
+    });
+    let client = Client::new(handle);
+    client.bot_sign_in(bot_token, api_hash).await?;
+    Ok(Arc::new(CachedClient {
+        client: Arc::new(client),
+        runner: runner_task,
+    }))
 }
 
 async fn build_client(