use std::{
collections::HashSet,
path::PathBuf,
sync::Arc,
time::{Duration, Instant},
};
use anyhow::Context;
use futures::StreamExt;
use kcode_telegram_text_delivery::send_telegram_message;
use kcode_telegram_transport_state::{
AcceptedGroupMessage, AcceptedMessage, AdmittedGroup, Event, GroupAdmission,
GroupSecuritySnapshot, Identity, Media as StoredMedia, MembershipObservation, MessageContent,
MessagePreparation, MessageRevision, ReplyDelivery, ResetDelivery, SentMessage, StateError,
TransportState,
};
use serde::Serialize;
use serde_json::{Value, json};
use teloxide::{
net::Download,
prelude::*,
requests::Request,
types::{
AllowedUpdate, ChatMemberKind, Message, MessageEntityKind, MessageKind, Update, UpdateKind,
},
};
use zeroize::Zeroize;
mod inbound;
mod native_media;
mod telegram_requests;
mod transport_extensions;
mod update_dispatch;
const UNAUTHORIZED_MESSAGE: &str =
"Sorry, this Kennedy bot is private and your Telegram handle is not whitelisted.";
const TELEGRAM_POLL_TIMEOUT_SECONDS: u32 = 90;
const TELEGRAM_HTTP_TIMEOUT_SECONDS: u64 = 120;
pub struct BotToken(String);
impl BotToken {
pub fn new(value: String) -> anyhow::Result<Self> {
anyhow::ensure!(
!value.trim().is_empty(),
"Telegram bot token must not be empty"
);
Ok(Self(value))
}
fn expose(&self) -> &str {
&self.0
}
}
impl std::fmt::Debug for BotToken {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("BotToken([REDACTED])")
}
}
impl Drop for BotToken {
fn drop(&mut self) {
self.0.zeroize();
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct IdentityObservation {
pub telegram_user_id: i64,
pub username: Option<String>,
pub display_name: String,
}
#[derive(Clone, Debug, Default)]
pub struct WhitelistSnapshot {
pub telegram_user_ids: HashSet<i64>,
}
impl WhitelistSnapshot {
pub fn contains(&self, telegram_user_id: i64) -> bool {
self.telegram_user_ids.contains(&telegram_user_id)
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum AddUserOutcome {
Forbidden,
Whitelisted {
handle: String,
telegram_user_id: Option<i64>,
},
}
pub trait IdentitySink: Send + Sync {
fn observe_identity(&self, observation: &IdentityObservation) -> anyhow::Result<()>;
fn whitelist(&self) -> anyhow::Result<WhitelistSnapshot>;
fn request_add_user(
&self,
requested_by_telegram_user_id: i64,
handle: &str,
) -> anyhow::Result<AddUserOutcome>;
fn observe_group(&self, group_id: &str) -> anyhow::Result<()>;
}
pub struct Config {
pub database: PathBuf,
pub bot_token: Option<BotToken>,
pub identity_sink: Arc<dyn IdentitySink>,
pub max_voice_bytes: usize,
}
#[derive(Clone)]
pub struct Service {
state: AppState,
}
pub struct Runtime {
service: Service,
}
#[derive(Clone)]
struct AppState {
transport: TransportState,
identity_sink: Arc<dyn IdentitySink>,
bot: Option<Bot>,
max_voice_bytes: usize,
bot_user_id: Option<i64>,
bot_username: Option<String>,
}
#[derive(Debug)]
pub struct Error {
code: &'static str,
message: String,
}
type ApiError = Error;
impl Error {
fn new(code: &'static str, message: impl Into<String>) -> Self {
Self {
code,
message: message.into(),
}
}
fn bad(message: impl Into<String>) -> Self {
Self::new("invalid_request", message)
}
fn unavailable() -> Self {
Self::new(
"telegram_unavailable",
"The Telegram bot token is not configured.",
)
}
fn internal(error: impl std::fmt::Display) -> Self {
tracing::warn!(error=%error, "Telegram relay request failed");
Self::new(
"internal_error",
"An unexpected Telegram relay error occurred.",
)
}
fn state(error: StateError) -> Self {
if error.code() == "internal_error" {
Self::internal(error)
} else {
Self::new(error.code(), error.message())
}
}
pub fn code(&self) -> &'static str {
self.code
}
pub fn message(&self) -> &str {
&self.message
}
}
impl std::fmt::Display for Error {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for Error {}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PrivateSession {
pub telegram_user_id: i64,
pub current_conversation_id: Option<String>,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Status {
pub service: &'static str,
pub status: &'static str,
pub telegram: &'static str,
pub inbound_media_kinds: &'static [&'static str],
pub outbound_media_kinds: &'static [&'static str],
pub max_media_bytes: usize,
}
#[derive(Clone, Debug)]
pub struct Media {
pub bytes: Vec<u8>,
pub media_type: String,
}
#[derive(Clone, Debug)]
pub struct MediaMetadata {
pub size_bytes: u64,
pub media_type: String,
}
#[derive(Clone, Debug)]
pub struct Attachment {
pub bytes: Vec<u8>,
pub file_name: Option<String>,
pub media_type: Option<String>,
pub kind: Option<String>,
pub caption: Option<String>,
}
pub async fn open(config: Config) -> anyhow::Result<Runtime> {
if config.max_voice_bytes == 0 {
anyhow::bail!("telegram max_voice_bytes must be greater than zero");
}
let transport =
TransportState::open(&config.database).context("opening Telegram transport state")?;
let bot = match config.bot_token.as_ref() {
Some(token) => {
let client = teloxide::net::default_reqwest_settings()
.timeout(Duration::from_secs(TELEGRAM_HTTP_TIMEOUT_SECONDS))
.build()
.context("building Telegram HTTP client")?;
Some(Bot::with_client(token.expose(), client))
}
None => None,
};
let (bot_user_id, bot_username) = if let Some(bot) = bot.as_ref() {
let me = telegram_requests::retry_request("get_me", || bot.get_me().send())
.await
.map_err(|error| {
anyhow::anyhow!(
"validating Telegram bot token failed ({})",
telegram_requests::request_error_class(&error)
)
})?;
(
Some(i64::try_from(me.id.0).context("Telegram bot ID exceeds SQLite range")?),
me.username.clone(),
)
} else {
(None, None)
};
let service = Service {
state: AppState {
transport,
identity_sink: config.identity_sink,
bot: bot.clone(),
max_voice_bytes: config.max_voice_bytes,
bot_user_id,
bot_username,
},
};
tracing::info!(enabled = bot.is_some(), "Telegram transport ready");
Ok(Runtime { service })
}
impl Runtime {
pub fn service(&self) -> Service {
self.service.clone()
}
pub async fn run(self) -> anyhow::Result<()> {
let Some(bot) = self.service.state.bot.clone() else {
return std::future::pending::<anyhow::Result<()>>().await;
};
inbound::poll_telegram(bot, self.service.state)
.await
.context("polling Telegram")
}
}
impl Service {
pub fn status(&self) -> Status {
Status {
service: "kcode-tg-kennedy-bot",
status: "ok",
telegram: if self.state.bot.is_some() {
"ready"
} else {
"disabled"
},
inbound_media_kinds: &native_media::INBOUND_MEDIA_KINDS,
outbound_media_kinds: &native_media::OUTBOUND_MEDIA_KINDS,
max_media_bytes: self.state.max_voice_bytes,
}
}
pub async fn list_private_sessions(&self) -> Result<Vec<PrivateSession>, Error> {
self.state
.transport
.list_private_sessions()
.map_err(ApiError::state)
.map(|sessions| {
sessions
.into_iter()
.map(|session| PrivateSession {
telegram_user_id: session.telegram_user_id,
current_conversation_id: session.current_conversation_id,
})
.collect()
})
}
pub async fn send_private_message(
&self,
telegram_user_id: i64,
conversation_id: String,
expected_conversation_id: Option<String>,
text: String,
) -> Result<Value, Error> {
send_private_message(
self.state.clone(),
telegram_user_id,
conversation_id,
expected_conversation_id,
text,
)
.await
}
pub async fn send_cold_private_message(
&self,
telegram_user_id: i64,
text: String,
) -> Result<Value, Error> {
send_cold_private_message(self.state.clone(), telegram_user_id, text).await
}
pub async fn send_group_message(&self, group_id: String, text: String) -> Result<Value, Error> {
send_group_message(self.state.clone(), group_id, text).await
}
pub async fn list_group_ingress(&self) -> Result<Value, Error> {
let batches = self
.state
.transport
.list_group_ingress()
.map_err(ApiError::state)?;
Ok(json!({"batches":batches}))
}
pub async fn complete_group_ingress(&self, batch_id: String) -> Result<Value, Error> {
self.state
.transport
.complete_group_ingress(&batch_id)
.map_err(ApiError::state)?;
Ok(json!({"id":batch_id,"status":"complete"}))
}
pub async fn list_group_session_updates(&self) -> Result<Value, Error> {
let updates = self
.state
.transport
.list_group_session_updates()
.map_err(ApiError::state)?;
Ok(json!({"updates":updates}))
}
pub async fn detach_group_session(
&self,
conversation_id: String,
group_id: String,
telegram_user_id: i64,
) -> Result<Value, Error> {
self.state
.transport
.detach_group_session(&conversation_id, &group_id, telegram_user_id)
.map_err(ApiError::state)?;
Ok(json!({
"conversationId":conversation_id,
"groupId":group_id,
"telegramUserId":telegram_user_id,
"status":"detached",
}))
}
pub async fn acknowledge_group_session_context(
&self,
conversation_id: String,
through_message_id: i64,
) -> Result<Value, Error> {
self.state
.transport
.acknowledge_group_session_context(&conversation_id, through_message_id)
.map_err(ApiError::state)?;
Ok(json!({
"conversationId":conversation_id,
"throughMessageId":through_message_id,
}))
}
pub async fn complete_silent_group_reset(
&self,
conversation_id: String,
) -> Result<Value, Error> {
self.state
.transport
.complete_silent_group_reset(&conversation_id)
.map_err(ApiError::state)?;
Ok(json!({"conversationId":conversation_id,"status":"complete"}))
}
pub async fn save_group_message_preparation(
&self,
chat_id: i64,
message_id: i64,
text: String,
model: Option<String>,
format: Option<String>,
truncated: bool,
) -> Result<Value, Error> {
self.state
.transport
.save_group_message_preparation(
chat_id,
message_id,
MessagePreparation {
text: text.clone(),
model,
format,
truncated,
},
)
.map_err(ApiError::state)?;
Ok(json!({"chatId":chat_id,"messageId":message_id,"text":text}))
}
pub async fn list_events(&self) -> Result<Value, Error> {
let events = self
.state
.transport
.pending_events()
.map_err(ApiError::state)?;
Ok(json!({"events":events}))
}
pub async fn bind_event(
&self,
event_id: String,
conversation_id: String,
expected_conversation_id: Option<String>,
) -> Result<Value, Error> {
let event = self
.state
.transport
.bind_event(
&event_id,
&conversation_id,
expected_conversation_id.as_deref(),
)
.map_err(ApiError::state)?;
serde_json::to_value(event).map_err(ApiError::internal)
}
pub async fn save_transcription(
&self,
event_id: String,
text: String,
transcription_model: String,
) -> Result<Value, Error> {
let event = self
.state
.transport
.save_transcription(&event_id, &text, &transcription_model)
.map_err(ApiError::state)?;
serde_json::to_value(event).map_err(ApiError::internal)
}
pub async fn reply_event(
&self,
event_id: String,
conversation_id: String,
text: String,
context_warning: Option<String>,
) -> Result<Value, Error> {
let event = reply_event(
self.state.clone(),
event_id,
conversation_id,
text,
context_warning,
)
.await?;
serde_json::to_value(event).map_err(ApiError::internal)
}
pub async fn abort_event(
&self,
event_id: String,
conversation_id: Option<String>,
message: String,
) -> Result<Value, Error> {
let event = abort_event(self.state.clone(), event_id, conversation_id, message).await?;
serde_json::to_value(event).map_err(ApiError::internal)
}
pub async fn interrupt_event(
&self,
event_id: String,
conversation_id: String,
) -> Result<Value, Error> {
let event = self
.state
.transport
.interrupt_event(&event_id, &conversation_id)
.map_err(ApiError::state)?;
serde_json::to_value(event).map_err(ApiError::internal)
}
pub async fn complete_reset(
&self,
event_id: String,
message: Option<String>,
) -> Result<Value, Error> {
let event = complete_reset(self.state.clone(), event_id, message).await?;
serde_json::to_value(event).map_err(ApiError::internal)
}
pub fn event_media(&self, event_id: &str) -> Result<Media, Error> {
self.state
.transport
.event_media(event_id)
.map(convert_media)
.map_err(ApiError::state)
}
pub fn event_media_metadata(&self, event_id: &str) -> Result<MediaMetadata, Error> {
self.state
.transport
.event_media_metadata(event_id)
.map(|media| MediaMetadata {
size_bytes: media.size_bytes,
media_type: media.media_type,
})
.map_err(ApiError::state)
}
pub fn group_message_media(&self, chat_id: i64, message_id: i64) -> Result<Media, Error> {
self.state
.transport
.group_message_media(chat_id, message_id)
.map(convert_media)
.map_err(ApiError::state)
}
pub fn group_message_media_metadata(
&self,
chat_id: i64,
message_id: i64,
) -> Result<MediaMetadata, Error> {
self.state
.transport
.group_message_media_metadata(chat_id, message_id)
.map(|media| MediaMetadata {
size_bytes: media.size_bytes,
media_type: media.media_type,
})
.map_err(ApiError::state)
}
}
pub fn migrate_storage(database: &std::path::Path) -> anyhow::Result<()> {
kcode_telegram_transport_state::migrate_storage(database)
}
fn convert_media(media: StoredMedia) -> Media {
Media {
bytes: media.bytes,
media_type: media.media_type,
}
}
fn normalize_username(value: &str) -> String {
value.trim().trim_start_matches('@').to_ascii_lowercase()
}
fn nonempty_verbatim(value: &str) -> Option<&str> {
(!value.trim().is_empty()).then_some(value)
}
fn validate_opaque_group_id(group_id: &str) -> Result<&str, ApiError> {
let group_id = group_id.trim();
if group_id.is_empty() || group_id.len() > 200 || group_id.chars().any(char::is_control) {
return Err(ApiError::bad("groupId is not a valid opaque group ID."));
}
Ok(group_id)
}
async fn send_telegram_text(
bot: &Bot,
chat_id: i64,
text: &str,
reply_to_message_id: Option<i64>,
) -> Result<Vec<Message>, ApiError> {
kcode_telegram_text_delivery::send_telegram_text(bot, chat_id, text, reply_to_message_id)
.await
.map_err(|error| {
tracing::warn!(
%chat_id,
error_class = telegram_requests::request_error_class(&error),
"Telegram reply failed"
);
ApiError::new("telegram_send_failed", "Telegram did not accept the reply.")
})
}
fn sent_messages(messages: Vec<Message>) -> Vec<SentMessage> {
messages
.into_iter()
.map(|message| SentMessage {
message_id: i64::from(message.id.0),
text: message.text().unwrap_or("").to_owned(),
sent_at: message.date.to_rfc3339(),
})
.collect()
}
async fn send_cold_private_message(
state: AppState,
telegram_user_id: i64,
text: String,
) -> Result<Value, ApiError> {
let started = Instant::now();
let text = nonempty_verbatim(&text).ok_or_else(|| ApiError::bad("text must not be empty."))?;
let delivery = state
.transport
.cold_private_delivery(telegram_user_id)
.map_err(ApiError::state)?;
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let sent = send_telegram_text(bot, delivery.chat_id(), text, None).await?;
let message_ids = sent
.iter()
.map(|message| i64::from(message.id.0))
.collect::<Vec<_>>();
tracing::info!(
%telegram_user_id,
duration_ms=started.elapsed().as_millis(),
"Telegram cold direct message"
);
Ok(json!({"telegramUserId":telegram_user_id,"messageIds":message_ids}))
}
async fn send_private_message(
state: AppState,
telegram_user_id: i64,
conversation_id: String,
expected_conversation_id: Option<String>,
text: String,
) -> Result<Value, ApiError> {
let started = Instant::now();
let text = nonempty_verbatim(&text).ok_or_else(|| ApiError::bad("text must not be empty."))?;
let delivery = state
.transport
.private_delivery(
telegram_user_id,
conversation_id.clone(),
expected_conversation_id,
)
.map_err(ApiError::state)?;
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let sent = send_telegram_text(bot, delivery.chat_id(), text, None).await?;
let message_ids = sent
.iter()
.map(|message| i64::from(message.id.0))
.collect::<Vec<_>>();
delivery.record_accepted().map_err(ApiError::state)?;
tracing::info!(
%telegram_user_id,
%conversation_id,
duration_ms=started.elapsed().as_millis(),
"Telegram cold direct message"
);
Ok(json!({
"telegramUserId":telegram_user_id,
"conversationId":conversation_id,
"messageIds":message_ids,
}))
}
async fn validated_group_delivery(
state: &AppState,
group_id: &str,
) -> Result<AdmittedGroup, ApiError> {
let group_id = validate_opaque_group_id(group_id)?;
let candidate = state
.transport
.group_delivery_candidate(group_id)
.map_err(ApiError::state)?;
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let snapshot = inbound::group_security_snapshot(bot, state, candidate.chat_id())
.await
.map_err(|error| {
tracing::warn!(
%group_id,
error_class = telegram_requests::anyhow_error_class(&error),
"Telegram group delivery authorization refresh failed"
);
ApiError::new(
"group_validation_failed",
"Telegram group membership could not be revalidated.",
)
})?;
candidate
.authorize(snapshot)
.map_err(ApiError::state)?
.ok_or_else(|| {
ApiError::new(
"group_not_allowed",
"Kennedy may send only when she is an administrator and every historical group member is whitelisted.",
)
})
}
async fn send_group_message(
state: AppState,
group_id: String,
text: String,
) -> Result<Value, ApiError> {
let started = Instant::now();
let group_id = validate_opaque_group_id(&group_id)?.to_owned();
let text = nonempty_verbatim(&text).ok_or_else(|| ApiError::bad("text must not be empty."))?;
let delivery = validated_group_delivery(&state, &group_id).await?;
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let sent = send_telegram_text(bot, delivery.chat_id(), text, None).await?;
let message_ids = sent
.iter()
.map(|message| i64::from(message.id.0))
.collect::<Vec<_>>();
tracing::info!(
%group_id,
duration_ms=started.elapsed().as_millis(),
"Telegram cold group message"
);
Ok(json!({"groupId":group_id,"messageIds":message_ids}))
}
async fn reply_event(
state: AppState,
id: String,
conversation_id: String,
text: String,
context_warning: Option<String>,
) -> Result<Event, ApiError> {
let started = Instant::now();
let text = nonempty_verbatim(&text).ok_or_else(|| ApiError::bad("text must not be empty."))?;
let delivery = match state
.transport
.reply_delivery(&id, &conversation_id)
.map_err(ApiError::state)?
{
ReplyDelivery::Complete(event) => return Ok(event),
ReplyDelivery::Pending(delivery) => delivery,
};
let event = delivery.event();
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let group_reply = (event.session_kind == "group").then_some(event.message_id);
let mut sent = send_telegram_text(bot, event.chat_id, text, group_reply).await?;
if let Some(warning) = context_warning.as_deref().and_then(nonempty_verbatim) {
sent.extend(send_telegram_text(bot, event.chat_id, warning, None).await?);
}
let event = delivery
.record_text(sent_messages(sent), true)
.map_err(ApiError::state)?;
tracing::info!(event_id=%id, duration_ms=started.elapsed().as_millis(), "Telegram reply");
Ok(event)
}
async fn abort_event(
state: AppState,
id: String,
conversation_id: Option<String>,
message: String,
) -> Result<Event, ApiError> {
let message =
nonempty_verbatim(&message).ok_or_else(|| ApiError::bad("message must not be empty."))?;
let outcome = state
.transport
.abort_event(&id, conversation_id.as_deref())
.map_err(ApiError::state)?;
let event = outcome.event().clone();
if !outcome.newly_aborted() {
return Ok(event);
}
let sent = if let Some(bot) = state.bot.as_ref() {
let group_reply = (event.session_kind == "group").then_some(event.message_id);
match send_telegram_text(bot, event.chat_id, message, group_reply).await {
Ok(sent) => sent_messages(sent),
Err(error) => {
tracing::warn!(event_id=%id, error=%error.message, "Telegram timeout notice could not be delivered");
Vec::new()
}
}
} else {
Vec::new()
};
outcome.record_notice(sent).map_err(ApiError::state)?;
tracing::warn!(event_id=%id, "Telegram response aborted at its hard timeout");
Ok(event)
}
async fn complete_reset(
state: AppState,
id: String,
message: Option<String>,
) -> Result<Event, ApiError> {
let delivery = match state
.transport
.reset_delivery(&id)
.map_err(ApiError::state)?
{
ResetDelivery::Complete(event) => return Ok(event),
ResetDelivery::Pending(delivery) => delivery,
};
let event = delivery.event();
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let message = message.as_deref().and_then(nonempty_verbatim).unwrap_or(
"Conversation reset. Your previous Telegram session has been queued for memory ingress.",
);
let group_reply = (event.session_kind == "group").then_some(event.message_id);
let sent = send_telegram_text(bot, event.chat_id, message, group_reply).await?;
delivery
.record_reset(sent_messages(sent))
.map_err(ApiError::state)
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Default)]
struct TestIdentities;
impl IdentitySink for TestIdentities {
fn observe_identity(&self, _observation: &IdentityObservation) -> anyhow::Result<()> {
Ok(())
}
fn whitelist(&self) -> anyhow::Result<WhitelistSnapshot> {
Ok(WhitelistSnapshot::default())
}
fn request_add_user(
&self,
_requested_by_telegram_user_id: i64,
_handle: &str,
) -> anyhow::Result<AddUserOutcome> {
Ok(AddUserOutcome::Forbidden)
}
fn observe_group(&self, _group_id: &str) -> anyhow::Result<()> {
Ok(())
}
}
#[test]
fn token_debug_is_redacted() {
let token = BotToken::new("123:secret".into()).unwrap();
assert_eq!(format!("{token:?}"), "BotToken([REDACTED])");
assert!(BotToken::new(" ".into()).is_err());
}
#[test]
fn disabled_status_preserves_the_public_capability_projection() {
let service = Service {
state: AppState {
transport: TransportState::open_in_memory().unwrap(),
identity_sink: Arc::new(TestIdentities),
bot: None,
max_voice_bytes: 4096,
bot_user_id: None,
bot_username: None,
},
};
let status = service.status();
assert_eq!(status.service, "kcode-tg-kennedy-bot");
assert_eq!(status.telegram, "disabled");
assert_eq!(status.max_media_bytes, 4096);
assert!(status.inbound_media_kinds.contains(&"document"));
assert!(status.outbound_media_kinds.contains(&"photo"));
}
#[test]
fn verbatim_text_validation_does_not_trim_delivery_content() {
assert_eq!(nonempty_verbatim(" hello\n"), Some(" hello\n"));
assert_eq!(nonempty_verbatim(" \n\t"), None);
}
}