Vakyartha documentation

619 // Words sent with a photo or file arrive as its caption.
620 let text = msg["text"]
621 .as_str()
622 .or_else(|| msg["caption"].as_str())
623 .map(String::from);
624 let photo_file_id = msg["photo"]
625 .as_array()
626 .and_then(|sizes| sizes.last())
627 .and_then(|largest| largest["file_id"].as_str())
628 .map(String::from);
629 let document =
630 msg["document"]["file_id"]
631 .as_str()
632 .map(|file_id| TelegramDocument {
633 file_id: file_id.to_string(),
634 file_name: msg["document"]["file_name"]
635 .as_str()
636 .unwrap_or("file")
637 .to_string(),
638 mime_type: msg["document"]["mime_type"]
639 .as_str()
640 .unwrap_or("application/octet-stream")
641 .to_string(),
642 });
643 let voice = msg["voice"]["file_id"]
644 .as_str()
645 .map(|file_id| TelegramDocument {
646 file_id: file_id.to_string(),
647 file_name: "voice.ogg".into(),
648 mime_type: "audio/ogg".into(),
649 });
650 let routable = chat_id.is_some()
651 && (text.is_some()
652 || photo_file_id.is_some()
653 || document.is_some()
654 || voice.is_some());
655 match (chat_id, routable) {
656 (Some(chat_id), true) => out.push(TelegramUpdate {
657 update_id,
658 chat_id,
659 sender_id: sender_id.unwrap_or(chat_id),
660 text: text.unwrap_or_default(),
661 photo_file_id,
662 document,
663 voice,
664 callback: None,
665 }),
666 _ => out.push(TelegramUpdate {
667 update_id,
668 chat_id: 0,
669 sender_id: 0,
670 text: String::new(),
671 photo_file_id: None,
672 document: None,
673 voice: None,
674 callback: None,
675 }),
676 }
677 }
678 }
679 Ok(out)
680 }
681
682 async fn send_message(&self, chat_id: i64, reply: &GatewayReply) -> Result<(), String> {
683 // Validate the packet against the same schema/surface boundary used
684 // by Slack and Discord. Telegram still keeps the plain reply text as
685 // its lossless fallback when the packet is absent or incompatible.
686 let rendered = reply.delivery.as_ref().and_then(|packet| {
687 let body = serde_json::json!({"delivery": packet});
688 super::prepared_packet(body, "telegram")
689 .ok()
690 .map(|packet| packet.chunks)
691 });
692 let (chunks, parse_html) = match rendered {
693 Some(chunks) if !chunks.is_empty() => (chunks, true),
694 _ => (vec![reply.text.clone()], false),
695 };
696 for chunk in chunks {
697 let url = format!("{}/bot{}/sendMessage", self.api_base, self.bot_token);
698 let mut body = serde_json::json!({
699 "chat_id": chat_id,
700 "text": chunk,
701 "link_preview_options": { "is_disabled": true },
702 });
703 if parse_html {
704 body["parse_mode"] = serde_json::Value::String("HTML".into());
705 }
706 let resp = http().post(&url).json(&body).send().await;
707 match resp {
708 Ok(r) if r.status().is_success() => {}
709 Ok(r) => {
710 let status = r.status();
711 // Converter edge-case guard: resend that chunk as plain
712 // text so a formatting bug degrades to ugly, not lost.
713 if parse_html && status.as_u16() == 400 {
714 eprintln!(
715 "[telegram] HTML rejected ({status}); falling back to plain text — \
716 chunk head: {}",
717 chunk.chars().take(80).collect::<String>()
718 );
719 let fallback = serde_json::json!({
720 "chat_id": chat_id,
721 "text": crate::channels::strip_tags(&chunk),
722 });
723 let r2 = http().post(&url).json(&fallback).send().await;
724 match r2 {
725 Ok(r2) if r2.status().is_success() => continue,
726 Ok(r2) => {
727 return Err(format!(
728 "sendMessage returned {} (plain retry: {})",
729 status,
730 r2.status()
731 ));
732 }
733 Err(e) => {
734 return Err(telegram_http_error("sendMessage retry", &e));
735 }
736 }
737 }
738 return Err(format!("sendMessage returned {status}"));
739 }
740 Err(e) => return Err(telegram_http_error("sendMessage", &e)),
741 }
742 }
743 Ok(())
744 }
745
746 /// Sends a file the turn produced, under this bot's own token, with the
747 /// change summary as its caption (Telegram allows 1024 characters).
748 async fn send_document(&self, chat_id: i64, file: &ReturnedFile) -> Result<(), String> {
749 use base64::Engine as _;
750 let bytes = base64::engine::general_purpose::STANDARD
751 .decode(file.data.as_bytes())
752 .map_err(|e| format!("bad file encoding: {e}"))?;
753 let part = reqwest::multipart::Part::bytes(bytes)
754 .file_name(file.name.clone())
755 .mime_str(&file.mime)
756 .map_err(|e| format!("document mime: {e}"))?;
757 let caption: String = file.caption.chars().take(1024).collect();
758 let body = reqwest::multipart::Form::new()
759 .text("chat_id", chat_id.to_string())
760 .text("caption", caption)
761 .part("document", part);
762 let sent = http()
763 .post(format!(
764 "{}/bot{}/sendDocument",
765 self.api_base, self.bot_token
766 ))
767 .multipart(body)
768 .send()
769 .await
770 .map_err(|e| telegram_http_error("sendDocument", &e))?;
771 if !sent.status().is_success() {
772 return Err(format!("sendDocument returned {}", sent.status()));
773 }
774 Ok(())
775 }
776
777 async fn send_voice(
778 &self,
779 chat_id: i64,
780 text: &str,
781 session_id: Option<&str>,
782 ) -> Result<(), String> {
783 let response = http()
784 .post(format!("{}/voice/speak", self.gateway_url))
785 .bearer_auth(&self.gateway_token)
786 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
787 .send()
788 .await
789 .map_err(|e| format!("voice speak request: {e}"))?;
790 if !response.status().is_success() {
791 return Err(format!("voice speak returned {}", response.status()));
792 }
793 let audio = response
794 .bytes()
795 .await
796 .map_err(|e| format!("voice speak body: {e}"))?;
797 if audio.is_empty() {
798 return Err("voice speak returned empty audio".into());
799 }
800 // Telegram's native voice method only accepts an OGG container with
801 // Opus audio. The provider-neutral synthesis endpoint deliberately
802 // returns WAV, so negotiate the transport codec here instead of
803 // silently uploading an invalid WAV as `sendVoice`.
804 let audio = wav_to_telegram_opus(audio.to_vec()).await?;
805 let part = reqwest::multipart::Part::bytes(audio)
806 .file_name("reply.ogg")
807 .mime_str("audio/ogg")
808 .map_err(|e| format!("voice mime: {e}"))?;
809 let body = reqwest::multipart::Form::new()
810 .text("chat_id", chat_id.to_string())
811 .part("voice", part);
812 let sent = http()
813 .post(format!("{}/bot{}/sendVoice", self.api_base, self.bot_token))
814 .multipart(body)
815 .send()
816 .await
817 .map_err(|e| format!("sendVoice: {e}"))?;
818 if !sent.status().is_success() {
819 return Err(format!("sendVoice returned {}", sent.status()));
820 }
821 Ok(())
822 }
823
824 /// getFile → two-step download of a Telegram-hosted file, returned as
825 /// raw bytes plus the `file_path` Telegram reported (its extension is
826 /// how photo mime type gets inferred below). 1 MiB bot-API cap.
827 async fn download_file(&self, file_id: &str) -> Result<(String, Vec<u8>), String> {
828 #[derive(serde::Deserialize)]
829 struct FileResp {
830 ok: bool,
831 result: FileMeta,
832 }
833 #[derive(serde::Deserialize)]
834 struct FileMeta {
835 file_path: Option<String>,
836 }
837 let meta: FileResp = http()
838 .get(format!("{}/bot{}/getFile", self.api_base, self.bot_token))
839 .query(&[("file_id", file_id)])
840 .send()
841 .await
842 .map_err(|e| telegram_http_error("getFile", &e))?
843 .json()
844 .await
845 .map_err(|e| telegram_http_error("getFile body", &e))?;
846 if !meta.ok {
847 return Err("getFile not ok".into());
848 }
849 let path = meta.result.file_path.ok_or("getFile missing file_path")?;
850 let bytes = http()
851 .get(format!(
852 "{}/file/bot{}/{}",
853 self.api_base, self.bot_token, path
854 ))
855 .send()
856 .await
857 .map_err(|e| telegram_http_error("download", &e))?
858 .error_for_status()
859 .map_err(|e| telegram_http_error("download status", &e))?
860 .bytes()
861 .await
862 .map_err(|e| telegram_http_error("download body", &e))?;
863 Ok((path, bytes.to_vec()))
864 }
865
866 /// Largest-photo variant, downscaled by Telegram on the sender side —
867 /// well inside vision budgets. Returned as (mime, base64).
868 async fn fetch_photo_base64(&self, file_id: &str) -> Result<(String, String), String> {
869 let (path, bytes) = self.download_file(file_id).await?;
870 let mime = if path.ends_with(".jpg") || path.ends_with(".jpeg") {
871 "image/jpeg"
872 } else if path.ends_with(".webp") {
873 "image/webp"
874 } else {
875 "image/png"
876 };
877 use base64::Engine as _;
878 Ok((
879 mime.to_string(),
880 base64::engine::general_purpose::STANDARD.encode(bytes),
881 ))
882 }
883
884 /// A non-photo attachment (code, logs, CSVs, Office files, ...).
885 /// `Ok(None)` means the file was over `DOCUMENT_MAX_BYTES` and was not
886 /// received; the caller tells the sender rather than truncating it.
887 async fn fetch_document_base64(
888 &self,
889 doc: &TelegramDocument,
890 ) -> Result<Option<String>, String> {
891 let (_, bytes) = self.download_file(&doc.file_id).await?;
892 if bytes.len() > Self::DOCUMENT_MAX_BYTES {
893 return Ok(None);
894 }
895 use base64::Engine as _;
896 Ok(Some(
897 base64::engine::general_purpose::STANDARD.encode(bytes),
898 ))
899 }
900
901 /// Run until the process is killed. Transient poll/send failures back
902 /// off and retry; they never drop the update stream position.
903 /// Non-acking ownership probe: `timeout=0, offset=-1` returns at most
904 /// the LAST update and acknowledges nothing, so probing is safe before
905 /// the real loop decides where to start.
906 async fn probe_ownership(&self) -> Result<(), PollBlock> {
907 let url = format!("{}/bot{}/getUpdates", self.api_base, self.bot_token);
908 let resp = http()
909 .get(&url)
910 .query(&[("timeout", "0"), ("offset", "-1")])
911 .send()
912 .await
913 .map_err(|_| PollBlock::Transient)?;
914 match resp.status().as_u16() {
915 200 => Ok(()),
916 409 => Err(PollBlock::Conflict),
917 _ => Err(PollBlock::Transient),
918 }
919 }
920
921 /// Hot-standby: while a rival owns the bot, wait quietly and take over
922 /// the moment it disappears. Logs entry once, then every 10th attempt.
923 async fn await_ownership(&self) {
924 let id = identity();
925 eprintln!(
926 "[telegram] bot token is owned by ANOTHER getUpdates consumer;\n[telegram] {id} standing by as hot standby (auto-takeover on rival exit)"
927 );
928 let mut attempt: u32 = 0;
929 loop {
930 tokio::time::sleep(std::time::Duration::from_secs(standby_backoff_secs(
931 attempt,
932 )))
933 .await;
934 attempt += 1;
935 match self.probe_ownership().await {
936 Ok(()) => {
937 eprintln!("[telegram] {id} took over polling (rival gone)");
938 return;
939 }
940 Err(PollBlock::Conflict) => {
941 if attempt.is_multiple_of(10) {
942 eprintln!(
943 "[telegram] still owned elsewhere ({attempt} probes); standing by"
944 );
945 }
946 }
947 Err(PollBlock::Transient) => {}
948 }
949 }
950 }
951
952 pub async fn run(&self) -> Result<(), String> {
953 // Local mutual exclusion first: two bridges on one host must fail
954 // fast with the holder's identity instead of flapping 409s.
955 let _instance_lock = match &self.locks_dir {
956 Some(dir) => Some(InstanceLock::acquire(dir, &self.bot_token)?),
957 None => None,
958 };
959
960 // Ownership probe: if another machine/session holds the long-poll,
961 // become hot standby instead of hammering 409s forever.
962 match self.probe_ownership().await {
963 Err(PollBlock::Conflict) => self.await_ownership().await,
964 Err(PollBlock::Transient) | Ok(()) => {}
965 }
966
967 let mut offset: i64 = 0;
968 let mut failures: u32 = 0;
969 let watch = (!self.token_env.is_empty())
970 .then(|| super::CredentialWatch::new(&self.token_env, &self.bot_token));
971 loop {
972 // Revocation and rotation are facts about the credential
973 // store, noticed here within one poll cycle, rather than
974 // something an API handler orchestrates by restarting this
975 // process.
976 if let Some(watch) = &watch {
977 match watch.check() {
978 super::CredentialState::Unchanged => {}
979 super::CredentialState::Rotated => {
980 eprintln!(
981 "[telegram] credential rotated; exiting so the service manager \
982 restarts this bridge with the new one"
983 );
984 return Ok(());
985 }
986 super::CredentialState::Revoked => {
987 eprintln!(
988 "[telegram] credential revoked; this bridge is stopping and will \
989 not poll again until a token is set"
990 );
991 return Ok(());
992 }
993 }
994 }
995 match self.tick(offset).await {
996 Ok(next) => {
997 offset = next;
998 failures = 0;
999 }
1000 Err(e) => {
1001 failures += 1;
1002 match classify_poll_error(&e) {
1003 PollBlock::Conflict => {
1004 // A rival appeared mid-run: hand over gracefully
1005 // and stand by for auto-takeover.
1006 eprintln!(
1007 "[telegram] lost ownership to another getUpdates consumer; entering hot standby ({})",
1008 identity()
1009 );
1010 self.await_ownership().await;
1011 }
1012 PollBlock::Transient => {
1013 // Never give up (docs/design/31-network-resilience.md):
1014 // outages, DHCP switches, and sleep/wake are all
1015 // ordinary transients. Backoff doubles to a 30s
1016 // cap — the same ceiling as hot standby — and
1017 // resets on first success. The offset cursor
1018 // makes every recovery gap-free.
1019 if failures == 1 || failures.is_multiple_of(10) {
1020 eprintln!("[telegram] poll failed ({failures} consecutive): {e}");
1021 }
1022 tokio::time::sleep(std::time::Duration::from_secs(
1023 standby_backoff_secs(failures),
1024 ))
1025 .await;
1026 }
1027 }
1028 }
1029 }
1030 }
1031 }
1032}
1033
1034#[cfg(test)]
1035#[allow(clippy::unwrap_used, clippy::expect_used)]
1036mod tests {
1037 use super::*;
1038
1039 #[test]
1040 fn approve_button_maps_to_a_yes_verdict_with_the_request_id() {
1041 assert_eq!(
1042 callback_data_to_verdict_text("approve:ab12cd34"),
1043 ("approve", Some("yes ab12cd34".into()))
1044 );
1045 }
1046
1047 #[test]
1048 fn deny_button_maps_to_a_no_verdict_with_the_request_id() {
1049 assert_eq!(
1050 callback_data_to_verdict_text("deny:ab12cd34"),
1051 ("deny", Some("no ab12cd34".into()))
1052 );
1053 }
1054
1055 #[test]
1056 fn unknown_callback_data_is_rejected_not_guessed() {
1057 assert_eq!(
1058 callback_data_to_verdict_text("something-else"),
1059 ("something-else", None)
1060 );
1061 assert_eq!(callback_data_to_verdict_text(""), ("", None));
1062 }
1063
1064 #[test]
1065 fn classify_maps_conflict_vs_transient() {
1066 assert_eq!(
1067 classify_poll_error("getUpdates returned 409 Conflict"),
1068 PollBlock::Conflict
1069 );
1070 assert_eq!(
1071 classify_poll_error("getUpdates returned 502 Bad Gateway"),
1072 PollBlock::Transient
1073 );
1074 assert_eq!(
1075 classify_poll_error("getUpdates: error sending request"),
1076 PollBlock::Transient
1077 );
1078 }
1079
1080 #[tokio::test]
1081 async fn telegram_http_errors_never_include_bot_token_or_url() {
1082 let error = reqwest::Client::new()
1083 .get("http://127.0.0.1:1/botsecret-token/getUpdates")
1084 .send()
1085 .await
1086 .unwrap_err();
1087 let rendered = telegram_http_error("getUpdates", &error);
1088 assert!(!rendered.contains("secret-token"));
1089 assert!(!rendered.contains("127.0.0.1"));
1090 assert!(!rendered.contains("/bot"));
1091 }
1092
1093 #[tokio::test]
1094 async fn opus_conversion_rejects_invalid_wav_without_claiming_delivery() {
1095 let result = wav_to_telegram_opus(b"not-a-wav".to_vec()).await;
1096 assert!(result.is_err());
1097 let error = result.unwrap_err();
1098 assert!(error.contains("ffmpeg") || error.contains("conversion"));
1099 }
1100
1101 #[test]
1102 fn gateway_reply_keeps_session_for_voice_receipt() {
1103 let value = serde_json::json!({
1104 "text": "reply",
1105 "session_id": "session-42",
1106 "delivery": {"status": "delivered"}
1107 });
1108 assert_eq!(value["session_id"].as_str(), Some("session-42"));
1109 assert_eq!(value["text"].as_str(), Some("reply"));
1110 }
1111
1112 #[test]
1113 fn standby_backoff_doubles_and_caps_at_thirty() {
1114 assert_eq!(standby_backoff_secs(0), 1);
1115 assert_eq!(standby_backoff_secs(1), 2);
1116 assert_eq!(standby_backoff_secs(5), 30);
1117 assert_eq!(standby_backoff_secs(50), 30);
1118 }
1119
1120 #[test]
1121 fn instance_lock_is_exclusive_per_token_and_released_on_drop() {
1122 let dir = tempfile::tempdir().unwrap();
1123 let a = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-A");
1124 assert!(a.is_ok(), "first holder acquires");
1125 let b = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-A");
1126 assert!(b.is_err(), "second holder on SAME token rejected");
1127 let c = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-B");
1128 assert!(c.is_ok(), "different token = different lock file");
1129 drop(a);
1130 let d = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-A");
1131 assert!(d.is_ok(), "flock releases when holder drops");
1132 }
1133}
1134

619 // Words sent with a photo or file arrive as its caption.
620 let text = msg["text"]
621 .as_str()
622 .or_else(|| msg["caption"].as_str())
623 .map(String::from);
624 let photo_file_id = msg["photo"]
625 .as_array()
626 .and_then(|sizes| sizes.last())
627 .and_then(|largest| largest["file_id"].as_str())
628 .map(String::from);
629 let document =
630 msg["document"]["file_id"]
631 .as_str()
632 .map(|file_id| TelegramDocument {
633 file_id: file_id.to_string(),
634 file_name: msg["document"]["file_name"]
635 .as_str()
636 .unwrap_or("file")
637 .to_string(),
638 mime_type: msg["document"]["mime_type"]
639 .as_str()
640 .unwrap_or("application/octet-stream")
641 .to_string(),
642 });
643 let voice = msg["voice"]["file_id"]
644 .as_str()
645 .map(|file_id| TelegramDocument {
646 file_id: file_id.to_string(),
647 file_name: "voice.ogg".into(),
648 mime_type: "audio/ogg".into(),
649 });
650 let routable = chat_id.is_some()
651 && (text.is_some()
652 || photo_file_id.is_some()
653 || document.is_some()
654 || voice.is_some());
655 match (chat_id, routable) {
656 (Some(chat_id), true) => out.push(TelegramUpdate {
657 update_id,
658 chat_id,
659 sender_id: sender_id.unwrap_or(chat_id),
660 text: text.unwrap_or_default(),
661 photo_file_id,
662 document,
663 voice,
664 callback: None,
665 }),
666 _ => out.push(TelegramUpdate {
667 update_id,
668 chat_id: 0,
669 sender_id: 0,
670 text: String::new(),
671 photo_file_id: None,
672 document: None,
673 voice: None,
674 callback: None,
675 }),
676 }
677 }
678 }
679 Ok(out)
680 }
681
682 async fn send_message(&self, chat_id: i64, reply: &GatewayReply) -> Result<(), String> {
683 // Validate the packet against the same schema/surface boundary used
684 // by Slack and Discord. Telegram still keeps the plain reply text as
685 // its lossless fallback when the packet is absent or incompatible.
686 let rendered = reply.delivery.as_ref().and_then(|packet| {
687 let body = serde_json::json!({"delivery": packet});
688 super::prepared_packet(body, "telegram")
689 .ok()
690 .map(|packet| packet.chunks)
691 });
692 let (chunks, parse_html) = match rendered {
693 Some(chunks) if !chunks.is_empty() => (chunks, true),
694 _ => (vec![reply.text.clone()], false),
695 };
696 for chunk in chunks {
697 let url = format!("{}/bot{}/sendMessage", self.api_base, self.bot_token);
698 let mut body = serde_json::json!({
699 "chat_id": chat_id,
700 "text": chunk,
701 "link_preview_options": { "is_disabled": true },
702 });
703 if parse_html {
704 body["parse_mode"] = serde_json::Value::String("HTML".into());
705 }
706 let resp = http().post(&url).json(&body).send().await;
707 match resp {
708 Ok(r) if r.status().is_success() => {}
709 Ok(r) => {
710 let status = r.status();
711 // Converter edge-case guard: resend that chunk as plain
712 // text so a formatting bug degrades to ugly, not lost.
713 if parse_html && status.as_u16() == 400 {
714 eprintln!(
715 "[telegram] HTML rejected ({status}); falling back to plain text — \
716 chunk head: {}",
717 chunk.chars().take(80).collect::<String>()
718 );
719 let fallback = serde_json::json!({
720 "chat_id": chat_id,
721 "text": crate::channels::strip_tags(&chunk),
722 });
723 let r2 = http().post(&url).json(&fallback).send().await;
724 match r2 {
725 Ok(r2) if r2.status().is_success() => continue,
726 Ok(r2) => {
727 return Err(format!(
728 "sendMessage returned {} (plain retry: {})",
729 status,
730 r2.status()
731 ));
732 }
733 Err(e) => {
734 return Err(telegram_http_error("sendMessage retry", &e));
735 }
736 }
737 }
738 return Err(format!("sendMessage returned {status}"));
739 }
740 Err(e) => return Err(telegram_http_error("sendMessage", &e)),
741 }
742 }
743 Ok(())
744 }
745
746 /// Sends a file the turn produced, under this bot's own token, with the
747 /// change summary as its caption (Telegram allows 1024 characters).
748 async fn send_document(&self, chat_id: i64, file: &ReturnedFile) -> Result<(), String> {
749 use base64::Engine as _;
750 let bytes = base64::engine::general_purpose::STANDARD
751 .decode(file.data.as_bytes())
752 .map_err(|e| format!("bad file encoding: {e}"))?;
753 let part = reqwest::multipart::Part::bytes(bytes)
754 .file_name(file.name.clone())
755 .mime_str(&file.mime)
756 .map_err(|e| format!("document mime: {e}"))?;
757 let caption: String = file.caption.chars().take(1024).collect();
758 let body = reqwest::multipart::Form::new()
759 .text("chat_id", chat_id.to_string())
760 .text("caption", caption)
761 .part("document", part);
762 let sent = http()
763 .post(format!(
764 "{}/bot{}/sendDocument",
765 self.api_base, self.bot_token
766 ))
767 .multipart(body)
768 .send()
769 .await
770 .map_err(|e| telegram_http_error("sendDocument", &e))?;
771 if !sent.status().is_success() {
772 return Err(format!("sendDocument returned {}", sent.status()));
773 }
774 Ok(())
775 }
776
777 async fn send_voice(
778 &self,
779 chat_id: i64,
780 text: &str,
781 session_id: Option<&str>,
782 ) -> Result<(), String> {
783 let response = http()
784 .post(format!("{}/voice/speak", self.gateway_url))
785 .bearer_auth(&self.gateway_token)
786 .json(&serde_json::json!({"text": text, "format": "wav", "session_id": session_id}))
787 .send()
788 .await
789 .map_err(|e| format!("voice speak request: {e}"))?;
790 if !response.status().is_success() {
791 return Err(format!("voice speak returned {}", response.status()));
792 }
793 let audio = response
794 .bytes()
795 .await
796 .map_err(|e| format!("voice speak body: {e}"))?;
797 if audio.is_empty() {
798 return Err("voice speak returned empty audio".into());
799 }
800 // Telegram's native voice method only accepts an OGG container with
801 // Opus audio. The provider-neutral synthesis endpoint deliberately
802 // returns WAV, so negotiate the transport codec here instead of
803 // silently uploading an invalid WAV as `sendVoice`.
804 let audio = wav_to_telegram_opus(audio.to_vec()).await?;
805 let part = reqwest::multipart::Part::bytes(audio)
806 .file_name("reply.ogg")
807 .mime_str("audio/ogg")
808 .map_err(|e| format!("voice mime: {e}"))?;
809 let body = reqwest::multipart::Form::new()
810 .text("chat_id", chat_id.to_string())
811 .part("voice", part);
812 let sent = http()
813 .post(format!("{}/bot{}/sendVoice", self.api_base, self.bot_token))
814 .multipart(body)
815 .send()
816 .await
817 .map_err(|e| format!("sendVoice: {e}"))?;
818 if !sent.status().is_success() {
819 return Err(format!("sendVoice returned {}", sent.status()));
820 }
821 Ok(())
822 }
823
824 /// getFile → two-step download of a Telegram-hosted file, returned as
825 /// raw bytes plus the `file_path` Telegram reported (its extension is
826 /// how photo mime type gets inferred below). 1 MiB bot-API cap.
827 async fn download_file(&self, file_id: &str) -> Result<(String, Vec<u8>), String> {
828 #[derive(serde::Deserialize)]
829 struct FileResp {
830 ok: bool,
831 result: FileMeta,
832 }
833 #[derive(serde::Deserialize)]
834 struct FileMeta {
835 file_path: Option<String>,
836 }
837 let meta: FileResp = http()
838 .get(format!("{}/bot{}/getFile", self.api_base, self.bot_token))
839 .query(&[("file_id", file_id)])
840 .send()
841 .await
842 .map_err(|e| telegram_http_error("getFile", &e))?
843 .json()
844 .await
845 .map_err(|e| telegram_http_error("getFile body", &e))?;
846 if !meta.ok {
847 return Err("getFile not ok".into());
848 }
849 let path = meta.result.file_path.ok_or("getFile missing file_path")?;
850 let bytes = http()
851 .get(format!(
852 "{}/file/bot{}/{}",
853 self.api_base, self.bot_token, path
854 ))
855 .send()
856 .await
857 .map_err(|e| telegram_http_error("download", &e))?
858 .error_for_status()
859 .map_err(|e| telegram_http_error("download status", &e))?
860 .bytes()
861 .await
862 .map_err(|e| telegram_http_error("download body", &e))?;
863 Ok((path, bytes.to_vec()))
864 }
865
866 /// Largest-photo variant, downscaled by Telegram on the sender side —
867 /// well inside vision budgets. Returned as (mime, base64).
868 async fn fetch_photo_base64(&self, file_id: &str) -> Result<(String, String), String> {
869 let (path, bytes) = self.download_file(file_id).await?;
870 let mime = if path.ends_with(".jpg") || path.ends_with(".jpeg") {
871 "image/jpeg"
872 } else if path.ends_with(".webp") {
873 "image/webp"
874 } else {
875 "image/png"
876 };
877 use base64::Engine as _;
878 Ok((
879 mime.to_string(),
880 base64::engine::general_purpose::STANDARD.encode(bytes),
881 ))
882 }
883
884 /// A non-photo attachment (code, logs, CSVs, Office files, ...).
885 /// `Ok(None)` means the file was over `DOCUMENT_MAX_BYTES` and was not
886 /// received; the caller tells the sender rather than truncating it.
887 async fn fetch_document_base64(
888 &self,
889 doc: &TelegramDocument,
890 ) -> Result<Option<String>, String> {
891 let (_, bytes) = self.download_file(&doc.file_id).await?;
892 if bytes.len() > Self::DOCUMENT_MAX_BYTES {
893 return Ok(None);
894 }
895 use base64::Engine as _;
896 Ok(Some(
897 base64::engine::general_purpose::STANDARD.encode(bytes),
898 ))
899 }
900
901 /// Run until the process is killed. Transient poll/send failures back
902 /// off and retry; they never drop the update stream position.
903 /// Non-acking ownership probe: `timeout=0, offset=-1` returns at most
904 /// the LAST update and acknowledges nothing, so probing is safe before
905 /// the real loop decides where to start.
906 async fn probe_ownership(&self) -> Result<(), PollBlock> {
907 let url = format!("{}/bot{}/getUpdates", self.api_base, self.bot_token);
908 let resp = http()
909 .get(&url)
910 .query(&[("timeout", "0"), ("offset", "-1")])
911 .send()
912 .await
913 .map_err(|_| PollBlock::Transient)?;
914 match resp.status().as_u16() {
915 200 => Ok(()),
916 409 => Err(PollBlock::Conflict),
917 _ => Err(PollBlock::Transient),
918 }
919 }
920
921 /// Hot-standby: while a rival owns the bot, wait quietly and take over
922 /// the moment it disappears. Logs entry once, then every 10th attempt.
923 async fn await_ownership(&self) {
924 let id = identity();
925 eprintln!(
926 "[telegram] bot token is owned by ANOTHER getUpdates consumer;\n[telegram] {id} standing by as hot standby (auto-takeover on rival exit)"
927 );
928 let mut attempt: u32 = 0;
929 loop {
930 tokio::time::sleep(std::time::Duration::from_secs(standby_backoff_secs(
931 attempt,
932 )))
933 .await;
934 attempt += 1;
935 match self.probe_ownership().await {
936 Ok(()) => {
937 eprintln!("[telegram] {id} took over polling (rival gone)");
938 return;
939 }
940 Err(PollBlock::Conflict) => {
941 if attempt.is_multiple_of(10) {
942 eprintln!(
943 "[telegram] still owned elsewhere ({attempt} probes); standing by"
944 );
945 }
946 }
947 Err(PollBlock::Transient) => {}
948 }
949 }
950 }
951
952 pub async fn run(&self) -> Result<(), String> {
953 // Local mutual exclusion first: two bridges on one host must fail
954 // fast with the holder's identity instead of flapping 409s.
955 let _instance_lock = match &self.locks_dir {
956 Some(dir) => Some(InstanceLock::acquire(dir, &self.bot_token)?),
957 None => None,
958 };
959
960 // Ownership probe: if another machine/session holds the long-poll,
961 // become hot standby instead of hammering 409s forever.
962 match self.probe_ownership().await {
963 Err(PollBlock::Conflict) => self.await_ownership().await,
964 Err(PollBlock::Transient) | Ok(()) => {}
965 }
966
967 let mut offset: i64 = 0;
968 let mut failures: u32 = 0;
969 let watch = (!self.token_env.is_empty())
970 .then(|| super::CredentialWatch::new(&self.token_env, &self.bot_token));
971 loop {
972 // Revocation and rotation are facts about the credential
973 // store, noticed here within one poll cycle, rather than
974 // something an API handler orchestrates by restarting this
975 // process.
976 if let Some(watch) = &watch {
977 match watch.check() {
978 super::CredentialState::Unchanged => {}
979 super::CredentialState::Rotated => {
980 eprintln!(
981 "[telegram] credential rotated; exiting so the service manager \
982 restarts this bridge with the new one"
983 );
984 return Ok(());
985 }
986 super::CredentialState::Revoked => {
987 eprintln!(
988 "[telegram] credential revoked; this bridge is stopping and will \
989 not poll again until a token is set"
990 );
991 return Ok(());
992 }
993 }
994 }
995 match self.tick(offset).await {
996 Ok(next) => {
997 offset = next;
998 failures = 0;
999 }
1000 Err(e) => {
1001 failures += 1;
1002 match classify_poll_error(&e) {
1003 PollBlock::Conflict => {
1004 // A rival appeared mid-run: hand over gracefully
1005 // and stand by for auto-takeover.
1006 eprintln!(
1007 "[telegram] lost ownership to another getUpdates consumer; entering hot standby ({})",
1008 identity()
1009 );
1010 self.await_ownership().await;
1011 }
1012 PollBlock::Transient => {
1013 // Never give up (docs/design/31-network-resilience.md):
1014 // outages, DHCP switches, and sleep/wake are all
1015 // ordinary transients. Backoff doubles to a 30s
1016 // cap — the same ceiling as hot standby — and
1017 // resets on first success. The offset cursor
1018 // makes every recovery gap-free.
1019 if failures == 1 || failures.is_multiple_of(10) {
1020 eprintln!("[telegram] poll failed ({failures} consecutive): {e}");
1021 }
1022 tokio::time::sleep(std::time::Duration::from_secs(
1023 standby_backoff_secs(failures),
1024 ))
1025 .await;
1026 }
1027 }
1028 }
1029 }
1030 }
1031 }
1032}
1033
1034#[cfg(test)]
1035#[allow(clippy::unwrap_used, clippy::expect_used)]
1036mod tests {
1037 use super::*;
1038
1039 #[test]
1040 fn approve_button_maps_to_a_yes_verdict_with_the_request_id() {
1041 assert_eq!(
1042 callback_data_to_verdict_text("approve:ab12cd34"),
1043 ("approve", Some("yes ab12cd34".into()))
1044 );
1045 }
1046
1047 #[test]
1048 fn deny_button_maps_to_a_no_verdict_with_the_request_id() {
1049 assert_eq!(
1050 callback_data_to_verdict_text("deny:ab12cd34"),
1051 ("deny", Some("no ab12cd34".into()))
1052 );
1053 }
1054
1055 #[test]
1056 fn unknown_callback_data_is_rejected_not_guessed() {
1057 assert_eq!(
1058 callback_data_to_verdict_text("something-else"),
1059 ("something-else", None)
1060 );
1061 assert_eq!(callback_data_to_verdict_text(""), ("", None));
1062 }
1063
1064 #[test]
1065 fn classify_maps_conflict_vs_transient() {
1066 assert_eq!(
1067 classify_poll_error("getUpdates returned 409 Conflict"),
1068 PollBlock::Conflict
1069 );
1070 assert_eq!(
1071 classify_poll_error("getUpdates returned 502 Bad Gateway"),
1072 PollBlock::Transient
1073 );
1074 assert_eq!(
1075 classify_poll_error("getUpdates: error sending request"),
1076 PollBlock::Transient
1077 );
1078 }
1079
1080 #[tokio::test]
1081 async fn telegram_http_errors_never_include_bot_token_or_url() {
1082 let error = reqwest::Client::new()
1083 .get("http://127.0.0.1:1/botsecret-token/getUpdates")
1084 .send()
1085 .await
1086 .unwrap_err();
1087 let rendered = telegram_http_error("getUpdates", &error);
1088 assert!(!rendered.contains("secret-token"));
1089 assert!(!rendered.contains("127.0.0.1"));
1090 assert!(!rendered.contains("/bot"));
1091 }
1092
1093 #[tokio::test]
1094 async fn opus_conversion_rejects_invalid_wav_without_claiming_delivery() {
1095 let result = wav_to_telegram_opus(b"not-a-wav".to_vec()).await;
1096 assert!(result.is_err());
1097 let error = result.unwrap_err();
1098 assert!(error.contains("ffmpeg") || error.contains("conversion"));
1099 }
1100
1101 #[test]
1102 fn gateway_reply_keeps_session_for_voice_receipt() {
1103 let value = serde_json::json!({
1104 "text": "reply",
1105 "session_id": "session-42",
1106 "delivery": {"status": "delivered"}
1107 });
1108 assert_eq!(value["session_id"].as_str(), Some("session-42"));
1109 assert_eq!(value["text"].as_str(), Some("reply"));
1110 }
1111
1112 #[test]
1113 fn standby_backoff_doubles_and_caps_at_thirty() {
1114 assert_eq!(standby_backoff_secs(0), 1);
1115 assert_eq!(standby_backoff_secs(1), 2);
1116 assert_eq!(standby_backoff_secs(5), 30);
1117 assert_eq!(standby_backoff_secs(50), 30);
1118 }
1119
1120 #[test]
1121 fn instance_lock_is_exclusive_per_token_and_released_on_drop() {
1122 let dir = tempfile::tempdir().unwrap();
1123 let a = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-A");
1124 assert!(a.is_ok(), "first holder acquires");
1125 let b = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-A");
1126 assert!(b.is_err(), "second holder on SAME token rejected");
1127 let c = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-B");
1128 assert!(c.is_ok(), "different token = different lock file");
1129 drop(a);
1130 let d = InstanceLock::acquire(&dir.path().to_path_buf(), "tok-A");
1131 assert!(d.is_ok(), "flock releases when holder drops");
1132 }
1133}
1134

Indexing the workspace…