use teloxide::{
payloads::SendDocumentSetters,
requests::Request as TelegramRequest,
types::{InputFile, ReplyParameters},
};
use super::*;
const TELEGRAM_CAPTION_LIMIT: usize = 1_024;
const MAX_FILE_NAME_CHARACTERS: usize = 255;
const MAX_MIME_TYPE_CHARACTERS: usize = 255;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub(super) struct DetachGroupSession {
pub(super) group_id: String,
pub(super) telegram_user_id: i64,
}
pub(super) async fn detach_group_session(
state: AppState,
conversation_id: String,
input: DetachGroupSession,
) -> Result<Value, ApiError> {
validate_conversation_id(&conversation_id)?;
let group_id = input.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."));
}
let db = state.db.lock().map_err(ApiError::internal)?;
let changed = db
.execute(
"UPDATE telegram_group_sessions
SET current_conversation_id=NULL,updated_at=?1
WHERE group_id=?2 AND telegram_user_id=?3
AND current_conversation_id=?4",
params![
Utc::now().to_rfc3339(),
group_id,
input.telegram_user_id,
conversation_id
],
)
.map_err(ApiError::internal)?;
if changed != 1 {
return Err(ApiError::conflict(
"This Telegram group session is absent, detached, or bound to a newer conversation.",
));
}
reclaim_group_working_messages(&db, group_id).map_err(ApiError::internal)?;
Ok(json!({
"conversationId":conversation_id,
"groupId":group_id,
"telegramUserId":input.telegram_user_id,
"status":"detached",
}))
}
#[derive(Debug, Default)]
struct OutboundFile {
conversation_id: Option<String>,
expected_conversation_id: Option<String>,
kind: Option<String>,
explicit_file_name: Option<String>,
mime_type: Option<String>,
caption: Option<String>,
complete: bool,
bytes: Option<Vec<u8>>,
}
fn outbound_file(
attachment: Attachment,
maximum_bytes: usize,
conversation_id: Option<String>,
expected_conversation_id: Option<String>,
complete: bool,
) -> Result<OutboundFile, ApiError> {
if attachment.bytes.len() > maximum_bytes {
return Err(ApiError::bad(format!(
"The file exceeds the configured {maximum_bytes}-byte Telegram media limit."
)));
}
Ok(OutboundFile {
conversation_id,
expected_conversation_id,
kind: attachment.kind,
explicit_file_name: attachment.file_name,
mime_type: attachment.media_type,
caption: attachment.caption,
complete,
bytes: Some(attachment.bytes),
})
}
impl Service {
pub async fn send_private_attachment(
&self,
telegram_user_id: i64,
conversation_id: String,
expected_conversation_id: Option<String>,
attachment: Attachment,
) -> Result<Value, Error> {
let input = outbound_file(
attachment,
self.state.max_voice_bytes,
Some(conversation_id),
expected_conversation_id,
false,
)?;
send_private_attachment(self.state.clone(), telegram_user_id, input).await
}
pub async fn send_group_attachment(
&self,
group_id: String,
attachment: Attachment,
) -> Result<Value, Error> {
let input = outbound_file(attachment, self.state.max_voice_bytes, None, None, false)?;
send_group_attachment(self.state.clone(), group_id, input).await
}
pub async fn send_event_attachment(
&self,
event_id: String,
conversation_id: String,
attachment: Attachment,
complete: bool,
) -> Result<Value, Error> {
let is_native_media = attachment.kind.is_some();
let input = outbound_file(
attachment,
self.state.max_voice_bytes,
Some(conversation_id),
None,
complete,
)?;
if is_native_media {
send_event_media(self.state.clone(), event_id, input).await
} else {
send_event_file(self.state.clone(), event_id, input).await
}
}
}
async fn send_private_attachment(
state: AppState,
telegram_user_id: i64,
input: OutboundFile,
) -> Result<Value, ApiError> {
let started = Instant::now();
let conversation_id = input
.conversation_id
.as_deref()
.ok_or_else(|| ApiError::bad("conversationId is required."))?;
validate_conversation_id(conversation_id)?;
if let Some(expected) = input.expected_conversation_id.as_deref() {
validate_conversation_id(expected)?;
}
let (chat_id, current_conversation_id) = {
let db = state.db.lock().map_err(ApiError::internal)?;
db.query_row(
"SELECT chat_id,current_conversation_id
FROM telegram_private_sessions WHERE telegram_user_id=?1",
[telegram_user_id],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, Option<String>>(1)?)),
)
.optional()
.map_err(ApiError::internal)?
.ok_or_else(|| {
ApiError::new(
"private_session_not_found",
"This Telegram user has not opened a private chat with Kennedy.",
)
})?
};
if current_conversation_id != input.expected_conversation_id {
return Err(ApiError::conflict(
"The Telegram user's current private conversation changed before delivery.",
));
}
let kind = input
.kind
.as_deref()
.map(|value| {
native_media::NativeMediaKind::parse(value).ok_or_else(|| {
ApiError::bad(
"kind must be photo, video, animation, audio, video_note, or sticker.",
)
})
})
.transpose()?;
if input.caption.is_some() && kind.is_some_and(|kind| !kind.accepts_caption()) {
return Err(ApiError::bad(format!(
"caption is not accepted for {} media.",
kind.expect("checked as present").as_str()
)));
}
let caption = input.caption.as_deref().and_then(nonempty_verbatim);
if caption.is_some_and(|value| value.encode_utf16().count() > TELEGRAM_CAPTION_LIMIT) {
return Err(ApiError::bad(
"The Telegram attachment caption exceeds 1024 UTF-16 code units.",
));
}
let bytes = input
.bytes
.ok_or_else(|| ApiError::bad("A nonempty file part is required."))?;
if bytes.is_empty() {
return Err(ApiError::bad("A nonempty file part is required."));
}
let supplied_file_name = input.explicit_file_name.as_deref();
if let Some(file_name) = supplied_file_name {
validate_file_name(file_name)?;
}
let mime_type = input.mime_type.as_deref().unwrap_or_else(|| {
kind.map(|kind| kind.fallback_mime(supplied_file_name))
.unwrap_or("application/octet-stream")
});
validate_mime_type(mime_type)?;
let file_name = supplied_file_name
.map(ToOwned::to_owned)
.unwrap_or_else(|| {
kind.map(|kind| kind.default_file_name(telegram_user_id, mime_type))
.unwrap_or_else(|| format!("attachment-{telegram_user_id}.bin"))
});
validate_file_name(&file_name)?;
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let sent = if let Some(kind) = kind {
native_media::send_native_media(bot, chat_id, kind, &bytes, &file_name, caption, None).await
} else {
let mut request = bot.send_document(
ChatId(chat_id),
InputFile::memory(bytes).file_name(file_name.clone()),
);
if let Some(caption) = caption {
request = request.caption(caption.to_owned());
}
telegram_requests::retry_request("send_document", || request.clone().send()).await
}
.map_err(|error| {
tracing::warn!(
%telegram_user_id,
error_class = telegram_requests::request_error_class(&error),
"Telegram private attachment send failed"
);
ApiError::new(
"telegram_send_failed",
"Telegram did not accept the attachment.",
)
})?;
let changed = {
let db = state.db.lock().map_err(ApiError::internal)?;
db.execute(
"UPDATE telegram_private_sessions
SET current_conversation_id=?1,updated_at=?2
WHERE telegram_user_id=?3 AND current_conversation_id IS ?4",
params![
conversation_id,
Utc::now().to_rfc3339(),
telegram_user_id,
input.expected_conversation_id,
],
)
.map_err(ApiError::internal)?
};
if changed != 1 {
return Err(ApiError::conflict(
"Telegram accepted the attachment, but the user's current private conversation changed before it could be attached.",
));
}
tracing::info!(
%telegram_user_id,
%conversation_id,
duration_ms=started.elapsed().as_millis(),
"Telegram cold direct-message attachment"
);
Ok(json!({
"telegramUserId":telegram_user_id,
"conversationId":conversation_id,
"kind":kind.map(|kind| kind.as_str()).unwrap_or("document"),
"fileName":file_name,
"mimeType":mime_type,
"telegramMessageId":i64::from(sent.id.0),
}))
}
async fn send_group_attachment(
state: AppState,
group_id: String,
input: OutboundFile,
) -> Result<Value, ApiError> {
let started = Instant::now();
let group_id = validate_opaque_group_id(&group_id)?.to_owned();
let kind = input
.kind
.as_deref()
.map(|value| {
native_media::NativeMediaKind::parse(value).ok_or_else(|| {
ApiError::bad(
"kind must be photo, video, animation, audio, video_note, or sticker.",
)
})
})
.transpose()?;
if input.caption.is_some() && kind.is_some_and(|kind| !kind.accepts_caption()) {
return Err(ApiError::bad(format!(
"caption is not accepted for {} media.",
kind.expect("checked as present").as_str()
)));
}
let caption = input.caption.as_deref().and_then(nonempty_verbatim);
if caption.is_some_and(|value| value.encode_utf16().count() > TELEGRAM_CAPTION_LIMIT) {
return Err(ApiError::bad(
"The Telegram attachment caption exceeds 1024 UTF-16 code units.",
));
}
let bytes = input
.bytes
.ok_or_else(|| ApiError::bad("A nonempty file part is required."))?;
if bytes.is_empty() {
return Err(ApiError::bad("A nonempty file part is required."));
}
let supplied_file_name = input.explicit_file_name.as_deref();
if let Some(file_name) = supplied_file_name {
validate_file_name(file_name)?;
}
let mime_type = input.mime_type.as_deref().unwrap_or_else(|| {
kind.map(|kind| kind.fallback_mime(supplied_file_name))
.unwrap_or("application/octet-stream")
});
validate_mime_type(mime_type)?;
let file_name = supplied_file_name
.map(ToOwned::to_owned)
.unwrap_or_else(|| {
kind.map(|kind| kind.default_file_name(0, mime_type))
.unwrap_or_else(|| "attachment.bin".into())
});
validate_file_name(&file_name)?;
let chat_id = validated_group_delivery_target(&state, &group_id).await?;
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let sent = if let Some(kind) = kind {
native_media::send_native_media(bot, chat_id, kind, &bytes, &file_name, caption, None).await
} else {
let mut request = bot.send_document(
ChatId(chat_id),
InputFile::memory(bytes).file_name(file_name.clone()),
);
if let Some(caption) = caption {
request = request.caption(caption.to_owned());
}
telegram_requests::retry_request("send_document", || request.clone().send()).await
}
.map_err(|error| {
tracing::warn!(
%group_id,
error_class = telegram_requests::request_error_class(&error),
"Telegram cold group attachment send failed"
);
ApiError::new(
"telegram_send_failed",
"Telegram did not accept the attachment.",
)
})?;
tracing::info!(
%group_id,
duration_ms=started.elapsed().as_millis(),
"Telegram cold group attachment"
);
Ok(json!({
"groupId":group_id,
"kind":kind.map(|kind| kind.as_str()).unwrap_or("document"),
"fileName":file_name,
"mimeType":mime_type,
"telegramMessageId":i64::from(sent.id.0),
}))
}
fn validate_file_name(value: &str) -> Result<(), ApiError> {
if value.trim().is_empty()
|| value.chars().count() > MAX_FILE_NAME_CHARACTERS
|| value
.chars()
.any(|character| character.is_control() || matches!(character, '/' | '\\'))
{
return Err(ApiError::bad(
"fileName must be a nonempty path-free name of at most 255 characters.",
));
}
Ok(())
}
fn validate_mime_type(value: &str) -> Result<(), ApiError> {
if value.trim().is_empty()
|| value.chars().count() > MAX_MIME_TYPE_CHARACTERS
|| value.chars().any(char::is_control)
{
return Err(ApiError::bad(
"The file content type must be a nonempty value of at most 255 characters.",
));
}
Ok(())
}
fn outbound_event(
db: &Connection,
event_id: &str,
conversation_id: &str,
) -> Result<RelayEvent, ApiError> {
let event = fetch_event(db, event_id)?;
if event.status == "complete" {
return Err(ApiError::conflict(
"The Telegram event is already complete.",
));
}
if event.conversation_id.as_deref() != Some(conversation_id) {
return Err(ApiError::conflict(
"The event is not bound to this conversation.",
));
}
Ok(event)
}
fn reconcile_outbound_event(
db: &Connection,
event_id: &str,
conversation_id: &str,
complete: bool,
delivery_label: &str,
) -> Result<RelayEvent, ApiError> {
if complete {
let changed = db
.execute(
"UPDATE telegram_events SET status='complete',completed_at=?1
WHERE id=?2 AND status<>'complete' AND conversation_id=?3",
params![Utc::now().to_rfc3339(), event_id, conversation_id],
)
.map_err(ApiError::internal)?;
if changed != 1 {
return Err(ApiError::conflict(format!(
"The {delivery_label} was sent, but the event binding changed before completion."
)));
}
} else {
let current = fetch_event(db, event_id)?;
if current.status == "complete"
|| current.conversation_id.as_deref() != Some(conversation_id)
{
return Err(ApiError::conflict(format!(
"The {delivery_label} was sent, but the event binding changed before delivery could be reconciled."
)));
}
}
let event = fetch_event(db, event_id)?;
if complete && let Some(group_id) = event.group_id.as_deref() {
reclaim_group_working_messages(db, group_id).map_err(ApiError::internal)?;
}
Ok(event)
}
async fn send_event_file(
state: AppState,
event_id: String,
input: OutboundFile,
) -> Result<Value, ApiError> {
let conversation_id = input
.conversation_id
.as_deref()
.ok_or_else(|| ApiError::bad("conversationId is required."))?;
validate_conversation_id(conversation_id)?;
let file_name = input
.explicit_file_name
.as_deref()
.ok_or_else(|| ApiError::bad("The file must have a fileName."))?;
validate_file_name(file_name)?;
let mime_type = input
.mime_type
.as_deref()
.unwrap_or("application/octet-stream");
validate_mime_type(mime_type)?;
let caption = input.caption.as_deref().and_then(nonempty_verbatim);
if caption.is_some_and(|value| value.encode_utf16().count() > TELEGRAM_CAPTION_LIMIT) {
return Err(ApiError::bad(
"The Telegram file caption exceeds 1024 UTF-16 code units.",
));
}
let bytes = input
.bytes
.ok_or_else(|| ApiError::bad("A nonempty file part is required."))?;
if bytes.is_empty() {
return Err(ApiError::bad("A nonempty file part is required."));
}
let event = {
let db = state.db.lock().map_err(ApiError::internal)?;
outbound_event(&db, &event_id, conversation_id)?
};
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let mut request = bot.send_document(
ChatId(event.chat_id),
InputFile::memory(bytes.clone()).file_name(file_name.to_owned()),
);
if let Some(caption) = caption {
request = request.caption(caption.to_owned());
}
if event.session_kind == "group"
&& let Ok(message_id) = i32::try_from(event.message_id)
{
request = request.reply_parameters(
ReplyParameters::new(teloxide::types::MessageId(message_id))
.allow_sending_without_reply(),
);
}
let sent = telegram_requests::retry_request("send_document", || request.clone().send())
.await
.map_err(|error| {
tracing::warn!(
event_id = %event_id,
error_class = telegram_requests::request_error_class(&error),
"Telegram file send failed"
);
ApiError::new("telegram_send_failed", "Telegram did not accept the file.")
})?;
let db = state.db.lock().map_err(ApiError::internal)?;
if event.session_kind == "group" {
let archive_text = caption
.map(ToOwned::to_owned)
.unwrap_or_else(|| format!("[File: {file_name}]"));
db.execute(
"INSERT INTO telegram_group_messages(
chat_id,message_id,update_id,display_name,text,reply_to_message_id,
sent_by_kennedy,created_at,kind,media_bytes,mime_type,file_name,
source_conversation_id,group_id
) VALUES(?1,?2,0,'Kennedy',?3,?4,1,?5,'document',?6,?7,?8,?9,?10)
ON CONFLICT(chat_id,message_id) DO NOTHING",
params![
event.chat_id,
i64::from(sent.id.0),
archive_text,
event.message_id,
sent.date.to_rfc3339(),
bytes,
mime_type,
file_name,
conversation_id,
event.group_id
],
)
.map_err(ApiError::internal)?;
if let Some(group_id) = event.group_id.as_deref() {
queue_stale_group_session_resets(&db, event.chat_id, group_id, i64::from(sent.id.0))
.map_err(ApiError::internal)?;
}
}
let reconciled_event =
reconcile_outbound_event(&db, &event_id, conversation_id, input.complete, "file")?;
Ok(json!({
"event":reconciled_event,
"fileName":file_name,
"mimeType":mime_type,
"telegramMessageId":i64::from(sent.id.0),
"complete":input.complete,
}))
}
async fn send_event_media(
state: AppState,
event_id: String,
input: OutboundFile,
) -> Result<Value, ApiError> {
let conversation_id = input
.conversation_id
.as_deref()
.ok_or_else(|| ApiError::bad("conversationId is required."))?;
validate_conversation_id(conversation_id)?;
let kind_text = input
.kind
.as_deref()
.ok_or_else(|| ApiError::bad("kind is required."))?;
let kind = native_media::NativeMediaKind::parse(kind_text).ok_or_else(|| {
ApiError::bad("kind must be photo, video, animation, audio, video_note, or sticker.")
})?;
if input.caption.is_some() && !kind.accepts_caption() {
return Err(ApiError::bad(format!(
"caption is not accepted for {} media.",
kind.as_str()
)));
}
let caption = input.caption.as_deref().and_then(nonempty_verbatim);
if caption.is_some_and(|value| value.encode_utf16().count() > TELEGRAM_CAPTION_LIMIT) {
return Err(ApiError::bad(
"The Telegram media caption exceeds 1024 UTF-16 code units.",
));
}
let bytes = input
.bytes
.ok_or_else(|| ApiError::bad("A nonempty file part is required."))?;
if bytes.is_empty() {
return Err(ApiError::bad("A nonempty file part is required."));
}
let event = {
let db = state.db.lock().map_err(ApiError::internal)?;
outbound_event(&db, &event_id, conversation_id)?
};
let supplied_file_name = input.explicit_file_name.as_deref();
if let Some(file_name) = supplied_file_name {
validate_file_name(file_name)?;
}
let mime_type = input
.mime_type
.as_deref()
.unwrap_or_else(|| kind.fallback_mime(supplied_file_name));
validate_mime_type(mime_type)?;
let file_name = supplied_file_name
.map(ToOwned::to_owned)
.unwrap_or_else(|| kind.default_file_name(event.message_id, mime_type));
validate_file_name(&file_name)?;
let reply_parameters = if event.session_kind == "group" {
i32::try_from(event.message_id).ok().map(|message_id| {
ReplyParameters::new(teloxide::types::MessageId(message_id))
.allow_sending_without_reply()
})
} else {
None
};
let bot = state.bot.as_ref().ok_or_else(ApiError::unavailable)?;
let sent = native_media::send_native_media(
bot,
event.chat_id,
kind,
&bytes,
&file_name,
caption,
reply_parameters,
)
.await
.map_err(|error| {
tracing::warn!(
event_id = %event_id,
media_kind = kind.as_str(),
error_class = telegram_requests::request_error_class(&error),
"Telegram native-media send failed"
);
ApiError::new(
"telegram_send_failed",
"Telegram did not accept the native media.",
)
})?;
let db = state.db.lock().map_err(ApiError::internal)?;
if event.session_kind == "group" {
let duration_seconds = native_media::message_duration(&sent, kind);
db.execute(
"INSERT INTO telegram_group_messages(
chat_id,message_id,update_id,display_name,text,reply_to_message_id,
sent_by_kennedy,created_at,kind,media_bytes,mime_type,file_name,
duration_seconds,source_conversation_id,group_id
) VALUES(?1,?2,0,'Kennedy',?3,?4,1,?5,?6,?7,?8,?9,?10,?11,?12)
ON CONFLICT(chat_id,message_id) DO NOTHING",
params![
event.chat_id,
i64::from(sent.id.0),
caption.unwrap_or(""),
event.message_id,
sent.date.to_rfc3339(),
kind.as_str(),
bytes,
mime_type,
file_name,
duration_seconds,
conversation_id,
event.group_id
],
)
.map_err(ApiError::internal)?;
if let Some(group_id) = event.group_id.as_deref() {
queue_stale_group_session_resets(&db, event.chat_id, group_id, i64::from(sent.id.0))
.map_err(ApiError::internal)?;
}
}
let reconciled_event = reconcile_outbound_event(
&db,
&event_id,
conversation_id,
input.complete,
"native media",
)?;
Ok(json!({
"event":reconciled_event,
"kind":kind.as_str(),
"fileName":file_name,
"mimeType":mime_type,
"telegramMessageId":i64::from(sent.id.0),
"complete":input.complete,
}))
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Default)]
struct ExtensionIdentitySink;
impl IdentitySink for ExtensionIdentitySink {
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(())
}
}
fn extension_state(database: Connection) -> AppState {
AppState {
db: Arc::new(Mutex::new(database)),
identity_sink: Arc::new(ExtensionIdentitySink),
bot: None,
max_voice_bytes: 1024,
bot_user_id: None,
bot_username: None,
}
}
fn group_database() -> (Connection, String) {
let database = Connection::open_in_memory().unwrap();
database.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
apply_migrations(&database).unwrap();
let group = ensure_group(&database, -100, "Friends").unwrap();
let now = Utc::now().to_rfc3339();
database
.execute(
"INSERT INTO telegram_group_messages(
chat_id,message_id,update_id,display_name,text,created_at,kind,group_id
) VALUES(-100,1,1,'Participant','hello',?1,'text',?2)",
params![now, group.group_id],
)
.unwrap();
(database, group.group_id)
}
#[tokio::test]
async fn matching_detach_clears_only_the_expected_group_user_pointer() {
let (database, group_id) = group_database();
let expected = "019f5ca7-020f-7b63-be2f-82785fb68c03";
let other = "119f5ca7-020f-7b63-be2f-82785fb68c04";
let now = Utc::now().to_rfc3339();
for (user_id, conversation_id) in [(42, expected), (77, other)] {
database
.execute(
"INSERT INTO telegram_group_sessions(
group_id,telegram_user_id,current_conversation_id,updated_at,
last_context_message_id,last_invocation_message_id
) VALUES(?1,?2,?3,?4,0,0)",
params![group_id, user_id, conversation_id, now],
)
.unwrap();
}
let state = extension_state(database);
let before = list_group_session_updates(state.clone()).await.unwrap();
assert_eq!(before["updates"].as_array().unwrap().len(), 2);
let _ = detach_group_session(
state.clone(),
expected.to_owned(),
DetachGroupSession {
group_id: group_id.clone(),
telegram_user_id: 42,
},
)
.await
.unwrap();
{
let database = state.db.lock().unwrap();
assert_eq!(
database
.query_row(
"SELECT current_conversation_id FROM telegram_group_sessions
WHERE group_id=?1 AND telegram_user_id=42",
[&group_id],
|row| row.get::<_, Option<String>>(0),
)
.unwrap(),
None
);
assert_eq!(
database
.query_row(
"SELECT current_conversation_id FROM telegram_group_sessions
WHERE group_id=?1 AND telegram_user_id=77",
[&group_id],
|row| row.get::<_, Option<String>>(0),
)
.unwrap()
.as_deref(),
Some(other)
);
}
let after = list_group_session_updates(state).await.unwrap();
let conversations = after["updates"]
.as_array()
.unwrap()
.iter()
.filter_map(|update| update["conversationId"].as_str())
.collect::<Vec<_>>();
assert!(!conversations.contains(&expected));
assert!(conversations.contains(&other));
}
#[tokio::test]
async fn stale_detach_cannot_clear_a_rebound_group_session() {
let (database, group_id) = group_database();
let stale = "019f5ca7-020f-7b63-be2f-82785fb68c03";
let current = "219f5ca7-020f-7b63-be2f-82785fb68c05";
database
.execute(
"INSERT INTO telegram_group_sessions(
group_id,telegram_user_id,current_conversation_id,updated_at,
last_context_message_id,last_invocation_message_id
) VALUES(?1,42,?2,?3,0,0)",
params![group_id, current, Utc::now().to_rfc3339()],
)
.unwrap();
let state = extension_state(database);
let error = detach_group_session(
state.clone(),
stale.to_owned(),
DetachGroupSession {
group_id: group_id.clone(),
telegram_user_id: 42,
},
)
.await
.unwrap_err();
assert_eq!(error.code, "state_conflict");
assert_eq!(
state
.db
.lock()
.unwrap()
.query_row(
"SELECT current_conversation_id FROM telegram_group_sessions
WHERE group_id=?1 AND telegram_user_id=42",
[&group_id],
|row| row.get::<_, String>(0),
)
.unwrap(),
current
);
}
#[test]
fn outbound_file_names_are_bounded_and_path_free() {
assert!(validate_file_name("report.pdf").is_ok());
assert!(validate_file_name("../secret").is_err());
assert!(validate_file_name("folder\\secret").is_err());
assert!(validate_file_name("").is_err());
assert!(validate_file_name(&"a".repeat(256)).is_err());
}
#[tokio::test]
async fn native_media_rejects_inapplicable_captions() {
let state = extension_state(group_database().0);
let inapplicable = outbound_file(
Attachment {
bytes: b"media".to_vec(),
file_name: Some("note.mp4".into()),
media_type: Some("video/mp4".into()),
kind: Some("video_note".into()),
caption: Some("not allowed".into()),
},
1024,
Some("019f5ca7-020f-7b63-be2f-82785fb68c03".into()),
None,
false,
)
.unwrap();
let error = send_event_media(state, "event".into(), inapplicable)
.await
.unwrap_err();
assert_eq!(error.code, "invalid_request");
assert!(error.message.contains("caption is not accepted"));
}
#[tokio::test]
async fn private_attachment_requires_an_established_chat_and_exact_binding() {
let state = extension_state(group_database().0);
let attachment = || Attachment {
bytes: b"hello".to_vec(),
file_name: Some("note.txt".into()),
media_type: Some("text/plain".into()),
kind: None,
caption: None,
};
let missing = outbound_file(
attachment(),
1024,
Some("019f5ca7-020f-7b63-be2f-82785fb68c03".into()),
None,
false,
)
.unwrap();
let error = send_private_attachment(state.clone(), 42, missing)
.await
.unwrap_err();
assert_eq!(error.code, "private_session_not_found");
{
let database = state.db.lock().unwrap();
ensure_transport_user(&database, 42, 9001).unwrap();
database
.execute(
"UPDATE telegram_private_sessions SET current_conversation_id=?1
WHERE telegram_user_id=42",
["119f5ca7-020f-7b63-be2f-82785fb68c04"],
)
.unwrap();
}
let stale = outbound_file(
attachment(),
1024,
Some("019f5ca7-020f-7b63-be2f-82785fb68c03".into()),
Some("219f5ca7-020f-7b63-be2f-82785fb68c05".into()),
false,
)
.unwrap();
let error = send_private_attachment(state, 42, stale).await.unwrap_err();
assert_eq!(error.code, "state_conflict");
}
#[tokio::test]
async fn private_attachment_uses_telegram_and_rebinds_after_acceptance() {
async fn accept(uri: axum::http::Uri) -> Json<Value> {
assert!(uri.path().to_ascii_lowercase().ends_with("/senddocument"));
Json(json!({
"ok":true,
"result":{
"message_id":901,
"date":1629404938,
"from":{
"id":999,
"is_bot":true,
"first_name":"Kennedy",
"username":"KennedyBot"
},
"chat":{"id":9001,"first_name":"User","type":"private"},
"document":{
"file_id":"sent",
"file_unique_id":"sent-unique",
"file_name":"note.txt",
"mime_type":"text/plain",
"file_size":5
}
}
}))
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server = tokio::spawn(async move {
axum::serve(
listener,
Router::new().fallback(axum::routing::post(accept)),
)
.await
.unwrap();
});
let (database, _) = group_database();
ensure_transport_user(&database, 42, 9001).unwrap();
let mut state = extension_state(database);
state.bot =
Some(Bot::new("test-token").set_api_url(format!("http://{address}").parse().unwrap()));
let conversation_id = "019f5ca7-020f-7b63-be2f-82785fb68c03";
let request = outbound_file(
Attachment {
bytes: b"hello".to_vec(),
file_name: Some("note.txt".into()),
media_type: Some("text/plain".into()),
kind: None,
caption: None,
},
1024,
Some(conversation_id.into()),
None,
false,
)
.unwrap();
let response = send_private_attachment(state.clone(), 42, request)
.await
.unwrap();
assert_eq!(response["kind"], "document");
assert_eq!(response["telegramMessageId"], 901);
assert_eq!(
state
.db
.lock()
.unwrap()
.query_row(
"SELECT current_conversation_id FROM telegram_private_sessions
WHERE telegram_user_id=42",
[],
|row| row.get::<_, String>(0),
)
.unwrap(),
conversation_id
);
server.abort();
}
#[tokio::test]
async fn native_group_send_archives_exact_media_and_completes_after_telegram_success() {
#[derive(Clone, Default)]
struct Capture(Arc<Mutex<Vec<(String, String)>>>);
async fn accept(
State(capture): State<Capture>,
uri: axum::http::Uri,
body: axum::body::Bytes,
) -> Json<Value> {
capture.0.lock().unwrap().push((
uri.path().to_ascii_lowercase(),
String::from_utf8_lossy(&body).into(),
));
Json(json!({
"ok":true,
"result":{
"message_id":900,
"date":1629404938,
"from":{
"id":999,
"is_bot":true,
"first_name":"Kennedy",
"username":"KennedyBot"
},
"chat":{"id":-100,"title":"Friends","type":"supergroup"},
"photo":[
{
"file_id":"sent",
"file_unique_id":"sent-unique",
"width":100,
"height":100,
"file_size":5
}
],
"caption":"exact caption"
}
}))
}
let capture = Capture::default();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let server_capture = capture.clone();
let server = tokio::spawn(async move {
axum::serve(
listener,
Router::new()
.fallback(axum::routing::post(accept))
.with_state(server_capture),
)
.await
.unwrap();
});
let (database, group_id) = group_database();
let conversation_id = "019f5ca7-020f-7b63-be2f-82785fb68c03";
database
.execute(
"INSERT INTO telegram_events(
id,update_id,message_id,telegram_user_id,chat_id,display_name,
kind,text,status,conversation_id,created_at,session_kind,group_id
) VALUES(
'event',10,7,42,-100,'David','text','invoke','processing',
?1,?2,'group',?3
)",
params![conversation_id, Utc::now().to_rfc3339(), group_id],
)
.unwrap();
let mut state = extension_state(database);
state.bot =
Some(Bot::new("test-token").set_api_url(format!("http://{address}").parse().unwrap()));
let request = outbound_file(
Attachment {
bytes: b"media".to_vec(),
file_name: Some("photo.jpg".into()),
media_type: Some("image/jpeg".into()),
kind: Some("photo".into()),
caption: Some("exact caption".into()),
},
1024,
Some(conversation_id.into()),
None,
true,
)
.unwrap();
let response = send_event_media(state.clone(), "event".into(), request)
.await
.unwrap();
assert_eq!(response["kind"], "photo");
assert_eq!(response["complete"], true);
assert_eq!(response["event"]["status"], "complete");
let requests = capture.0.lock().unwrap();
assert_eq!(requests.len(), 1);
assert!(requests[0].0.ends_with("/sendphoto"));
assert!(requests[0].1.contains("\"message_id\":7"));
assert!(
requests[0]
.1
.contains("\"allow_sending_without_reply\":true")
);
drop(requests);
let archived = state
.db
.lock()
.unwrap()
.query_row(
"SELECT kind,text,media_bytes,mime_type,file_name,
reply_to_message_id,source_conversation_id,group_id
FROM telegram_group_messages
WHERE chat_id=-100 AND message_id=900",
[],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Vec<u8>>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, i64>(5)?,
row.get::<_, String>(6)?,
row.get::<_, String>(7)?,
))
},
)
.unwrap();
assert_eq!(
archived,
(
"photo".into(),
"exact caption".into(),
b"media".to_vec(),
"image/jpeg".into(),
"photo.jpg".into(),
7,
conversation_id.into(),
group_id,
)
);
server.abort();
}
}