Upload main.rs

src/main.rs

@@ -63,6 +63,11 @@ struct AppState {
     arl_store_path: String,
     /// CDN URL кэш для предзагрузки. Ключ: track_id. TTL 20 мин.
     cdn_cache: Arc<RwLock<HashMap<u64, CdnCacheEntry>>>,
+    /// Общий переиспользуемый клиент для аплоада в Telegram.
+    /// Живёт весь процесс — держит keep-alive соединение и DNS-кэш к
+    /// api.telegram.org "тёплыми", вместо пересоздания клиента (а с ним
+    /// DNS+TLS handshake с нуля) на каждый /send_audio.
+    tg_http: Arc<reqwest::Client>,
 }
 
 fn license_token_from_ud(ud: &serde_json::Value) -> String {
@@ -151,10 +156,7 @@ async fn api_client_for_arl(state: &AppState, arl: Option<String>) -> Arc<Mutex<
 }
 
 fn extract_media_url_and_format(json: &serde_json::Value, formats: &Vec<&str>) -> (String, String, usize) {
-    use std::io::Write;
-    let mut log_file = std::fs::OpenOptions::new().create(true).append(true).open("debug_dzmedia.log").unwrap();
-    let _ = writeln!(log_file, "--- MEDIA JSON RESPONSE ---");
-    let _ = writeln!(log_file, "{}", serde_json::to_string_pretty(json).unwrap_or_default());
+    tracing::trace!(target: "dzmedia::media_json", "{}", json);
 
     let data = json.get("data").and_then(|v| v.as_array());
     let Some(data_arr) = data else {
@@ -339,25 +341,23 @@ async fn media_url_for_track(
     id: u64,
     formats: &Vec<Format>,
 ) -> Result<(String, String, u64), String> {
-    use std::io::Write;
-    let mut log_file = std::fs::OpenOptions::new().create(true).append(true).open("debug_dzmedia.log").unwrap();
-    let _ = writeln!(log_file, "--- NEW REQUEST FOR ID {} ---", id);
+    tracing::trace!("media_url_for_track: new request for id {}", id);
 
     if client.license_token.is_empty() {
         if let Err(e) = client.force_renew().await {
-            let _ = writeln!(log_file, "force_renew err: {}", e);
+            tracing::warn!("media_url_for_track: force_renew err: {}", e);
             return Err(format!("renew:{e}"));
         }
     }
 
     let q_json = serde_json::json!({"sng_ids":[id],"array_default":["SNG_ID","TRACK_TOKEN","FALLBACK"]});
     let r: serde_json::Value = client.api_call("song.getListData", &q_json).await.map_err(|e| {
-        let _ = writeln!(log_file, "api_call err: {}", e);
+        tracing::warn!("media_url_for_track: api_call err: {}", e);
         e.to_string()
     })?;
 
     let item = r.pointer("/data/0").ok_or_else(|| {
-        let _ = writeln!(log_file, "No valid ID");
+        tracing::warn!("media_url_for_track: no valid id");
         "No valid ID".to_string()
     })?;
 
@@ -373,33 +373,24 @@ async fn media_url_for_track(
         })
     }).unwrap_or(0);
 
-    let _ = writeln!(log_file, "track_token: {}, fallback_id: {}, fallback_token: {}", track_token, fallback_id, fallback_token);
+    tracing::trace!("media_url_for_track: track_token={}, fallback_id={}, fallback_token={}", track_token, fallback_id, fallback_token);
 
     if fallback_id > 0 {
         let q_fb = serde_json::json!({"sng_ids":[fallback_id],"array_default":["SNG_ID","TRACK_TOKEN","FALLBACK"]});
-        let _ = writeln!(log_file, "Querying fallback ID: {}", fallback_id);
         if let Ok(r_fb) = client.api_call::<serde_json::Value, serde_json::Value>("song.getListData", &q_fb).await {
             if let Some(item_fb) = r_fb.pointer("/data/0") {
                 if let Some(real_tk) = item_fb.get("TRACK_TOKEN").and_then(|v| v.as_str()) {
-                    let _ = writeln!(log_file, "Fetched real_tk for fallback: {}", real_tk);
                     let tokens = vec![real_tk];
                     match fetch_media_url(client, formats, tokens).await {
                         Ok((url, fmt, _)) => {
-                            let _ = writeln!(log_file, "fetch_media_url SUCCESS for fallback: fmt={}", fmt);
                             return Ok((url, fmt, fallback_id));
                         },
                         Err(e) => {
-                            let _ = writeln!(log_file, "fetch_media_url FAILED for fallback: {}", e);
+                            tracing::trace!("media_url_for_track: fetch_media_url failed for fallback: {}", e);
                         }
                     }
-                } else {
-                    let _ = writeln!(log_file, "NO TRACK_TOKEN in fallback response");
                 }
-            } else {
-                let _ = writeln!(log_file, "Empty data array in fallback response");
             }
-        } else {
-            let _ = writeln!(log_file, "api_call for fallback ID FAILED");
         }
     }
 
@@ -416,20 +407,18 @@ async fn media_url_for_track(
     }
 
     if tokens.is_empty() {
-        let _ = writeln!(log_file, "empty TRACK_TOKEN and FALLBACK");
+        tracing::warn!("media_url_for_track: empty TRACK_TOKEN and FALLBACK");
         return Err("empty TRACK_TOKEN and FALLBACK".to_string());
     }
 
-    let _ = writeln!(log_file, "Falling back to old logic. Tokens: {:?}", tokens);
     let res = fetch_media_url(client, formats, tokens).await;
     match res {
         Ok((url, fmt, idx)) => {
-            let _ = writeln!(log_file, "Old logic SUCCESS: fmt={}", fmt);
             let id_to_use = *token_ids.get(idx).unwrap_or(&id);
             Ok((url, fmt, id_to_use))
         },
         Err(e) => {
-            let _ = writeln!(log_file, "Old logic FAILED: {}", e);
+            tracing::warn!("media_url_for_track: old logic failed: {}", e);
             Err(e)
         }
     }
@@ -749,7 +738,6 @@ impl From<&StreamParams> for AudioTagMeta {
 }
 
 async fn stream_info(State(state): State<AppState>, Query(q): Query<StreamParams>) -> impl IntoResponse {
-    let _ = std::fs::remove_file("debug_dzmedia.log");
     let formats = match q.format.as_deref().unwrap_or("AUTO") {
         "FLAC"     => vec![Format::FLAC, Format::MP3_320, Format::MP3_128],
         "MP3_320"  => vec![Format::MP3_320, Format::MP3_128],
@@ -841,8 +829,6 @@ async fn stream_info(State(state): State<AppState>, Query(q): Query<StreamParams
         ((total_bytes as f64) * 8.0 / (duration as f64) / 1000.0).round() as u64
     } else { 0 };
 
-    let log_data = std::fs::read_to_string("debug_dzmedia.log").unwrap_or_default();
-
     (StatusCode::OK, Json(json!({
         "id":          q.id,
         "requested":   requested,
@@ -851,7 +837,7 @@ async fn stream_info(State(state): State<AppState>, Query(q): Query<StreamParams
         "bytes":       total_bytes,
         "bitrate_kbps": bitrate_kbps,
         "mime":        mime,
-        "log":         log_data
+        "log":         ""
     }))).into_response()
 }
 
@@ -899,7 +885,7 @@ async fn fetch_cover_picture(cover_url: &str) -> Option<Picture> {
     Some(picture)
 }
 
-async fn inject_audio_metadata(mut bytes: Vec<u8>, ext: &str, meta: &AudioTagMeta) -> Vec<u8> {
+async fn inject_audio_metadata(bytes: Vec<u8>, ext: &str, meta: &AudioTagMeta) -> Vec<u8> {
     let has_text = meta.title.is_some() || meta.performer.is_some() || meta.album.is_some();
     let cover = match meta.cover_url.as_deref() {
         Some(url) => fetch_cover_picture(url).await,
@@ -910,12 +896,6 @@ async fn inject_audio_metadata(mut bytes: Vec<u8>, ext: &str, meta: &AudioTagMet
         return bytes;
     }
 
-    // Временно отключаем добавление метаданных для FLAC - возможно lofty ломает файл
-    if ext.eq_ignore_ascii_case("flac") {
-        tracing::warn!("Skipping metadata injection for FLAC to avoid corruption");
-        return bytes;
-    }
-
     let tag_type = if ext.eq_ignore_ascii_case("flac") {
         TagType::VorbisComments
     } else {
@@ -936,35 +916,55 @@ async fn inject_audio_metadata(mut bytes: Vec<u8>, ext: &str, meta: &AudioTagMet
         tag.push_picture(picture);
     }
 
-    let unique = SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map(|d| d.as_nanos())
-        .unwrap_or(0);
-    let temp_path = std::env::temp_dir().join(format!("dzmedia_tagged_{}.{}", unique, ext));
-    if let Err(e) = std::fs::write(&temp_path, &bytes) {
-        tracing::warn!("metadata temp write failed: {}", e);
-        return bytes;
-    }
+    let ext_owned = ext.to_string();
+    let bytes_for_write = bytes.clone();
 
-    let write_options = if ext.eq_ignore_ascii_case("mp3") {
-        WriteOptions::new().use_id3v23(true)
-    } else {
-        WriteOptions::new()
-    };
+    // Пишем/сохраняем/читаем temp-файл в spawn_blocking: на 20+MB FLAC
+    // это не микросекунды, а реальный синхронный I/O. Выполненный прямо
+    // в async fn, он держит tokio worker thread занятым и на это время
+    // не даёт прогрессировать вообще никаким другим задачам на этом
+    // воркере — включая коннект другого параллельного запроса в Telegram.
+    let tagged: Option<Vec<u8>> = tokio::task::spawn_blocking(move || {
+        let unique = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0);
+        let temp_path = std::env::temp_dir().join(format!("dzmedia_tagged_{}.{}", unique, ext_owned));
+
+        if let Err(e) = std::fs::write(&temp_path, &bytes_for_write) {
+            tracing::warn!("metadata temp write failed: {}", e);
+            return None;
+        }
 
-    let write_result = tag.save_to_path(&temp_path, write_options);
-    if let Err(e) = write_result {
+        let write_options = if ext_owned.eq_ignore_ascii_case("mp3") {
+            WriteOptions::new().use_id3v23(true)
+        } else {
+            WriteOptions::new()
+        };
+
+        if let Err(e) = tag.save_to_path(&temp_path, write_options) {
+            let _ = std::fs::remove_file(&temp_path);
+            tracing::warn!("metadata tagging failed: {}", e);
+            return None;
+        }
+
+        let result = match std::fs::read(&temp_path) {
+            Ok(tagged) => Some(tagged),
+            Err(e) => {
+                tracing::warn!("metadata temp read failed: {}", e);
+                None
+            }
+        };
         let _ = std::fs::remove_file(&temp_path);
-        tracing::warn!("metadata tagging failed: {}", e);
-        return bytes;
-    }
+        result
+    })
+    .await
+    .unwrap_or_else(|e| {
+        tracing::warn!("metadata blocking task failed: {}", e);
+        None
+    });
 
-    match std::fs::read(&temp_path) {
-        Ok(tagged) => bytes = tagged,
-        Err(e) => tracing::warn!("metadata temp read failed: {}", e),
-    }
-    let _ = std::fs::remove_file(&temp_path);
-    bytes
+    tagged.unwrap_or(bytes)
 }
 
 async fn send_audio(
@@ -1078,15 +1078,13 @@ async fn send_audio(
     let mime = if ext == "flac" { "audio/flac" } else { "audio/mpeg" };
     let filename = format!("track_{}.{}", q.id, ext);
 
-    // Загружаем в Telegram через отдельный клиент.
+    // Загружаем в Telegram через общий, уже прогретый клиент (см. AppState::tg_http).
+    // Раньше клиент создавался заново на каждый вызов — из-за этого каждый
+    // /send_audio платил полный DNS+TLS handshake с нуля, и на "холодных"
+    // контейнерах это не укладывалось в connect_timeout (лог показывал
+    // ровно 10.0с — таймаут коннекта, а не медленную передачу данных).
     let tg_url = format!("https://api.telegram.org/bot{}/sendAudio", bot_token);
-    let tg_http = reqwest::Client::builder()
-        .https_only(true)
-        .connect_timeout(Duration::from_secs(10))
-        .timeout(Duration::from_secs(30))
-        .tcp_keepalive(Duration::from_secs(20))
-        .build()
-        .unwrap();
+    let tg_http = state.tg_http.clone();
 
     let mut tg_json: Option<serde_json::Value> = None;
     let mut last_tg_err: Option<String> = None;
@@ -3991,6 +3989,17 @@ async fn main() {
     let port = std::env::var("PORT").unwrap_or("7860".to_string());
     let port: u16 = port.parse().unwrap_or(7860);
 
+    let tg_http = Arc::new(
+        reqwest::Client::builder()
+            .https_only(true)
+            .connect_timeout(Duration::from_secs(20))
+            .timeout(Duration::from_secs(60))
+            .tcp_keepalive(Duration::from_secs(30))
+            .pool_idle_timeout(Duration::from_secs(90))
+            .build()
+            .unwrap(),
+    );
+
     let state = AppState {
         api: Arc::new(Mutex::new(APIClient::new())),
         api_by_arl: Arc::new(RwLock::new(HashMap::new())),
@@ -3999,10 +4008,28 @@ async fn main() {
         pair_store_path: std::env::var("PAIR_STORE_PATH").ok().unwrap_or_else(|| "pair_store.json".to_string()),
         arl_store_path: std::env::var("ARL_STORE_PATH").ok().unwrap_or_else(|| "arl_store.json".to_string()),
         cdn_cache: Arc::new(RwLock::new(HashMap::new())),
+        tg_http,
     };
     arl_store_load(&state).await;
     pair_store_load(&state).await;
 
+    // Прогреваем DNS+TLS до api.telegram.org сразу при старте контейнера,
+    // а не на первом реальном /send_audio. Именно на "холодном" первом
+    // запросе после старта конекты к Telegram чаще всего не укладываются
+    // в connect_timeout.
+    if let Ok(bot_token) = std::env::var("BOT_TOKEN") {
+        if !bot_token.is_empty() {
+            let warm_client = state.tg_http.clone();
+            tokio::spawn(async move {
+                let url = format!("https://api.telegram.org/bot{}/getMe", bot_token);
+                match warm_client.get(&url).send().await {
+                    Ok(_) => tracing::info!("Telegram connection pre-warmed"),
+                    Err(e) => tracing::warn!("Telegram warm-up failed (non-fatal): {}", e),
+                }
+            });
+        }
+    }
+
     let cors = CorsLayer::new()
         .allow_origin(Any)
         .allow_methods([Method::GET, Method::POST, Method::OPTIONS, Method::HEAD])