kcode_tg_kennedy_bot/

transport_extensions.rs

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_cold_private_attachment(
        &self,
        telegram_user_id: i64,
        attachment: Attachment,
    ) -> Result<Value, Error> {
        let input = outbound_file(attachment, self.state.max_voice_bytes, None, None, false)?;
        send_private_attachment(self.state.clone(), telegram_user_id, input).await
    }

    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.clone();
    let expected_conversation_id = input.expected_conversation_id.clone();
    if let Some(conversation_id) = conversation_id.as_deref() {
        validate_conversation_id(conversation_id)?;
    } else if expected_conversation_id.is_some() {
        return Err(ApiError::bad(
            "expectedConversationId requires conversationId.",
        ));
    }
    if let Some(expected) = expected_conversation_id.as_deref() {
        validate_conversation_id(expected)?;
    }

    let (chat_id, current_conversation_id) = private_delivery_target(&state, telegram_user_id)?;
    if conversation_id.is_some() && current_conversation_id != 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.",
        )
    })?;

    if let Some(conversation_id) = conversation_id.as_deref() {
        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,
                    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 session-bound direct-message attachment"
        );
    } else {
        tracing::info!(
            %telegram_user_id,
            duration_ms=started.elapsed().as_millis(),
            "Telegram cold direct-message attachment"
        );
    }
    let mut response = json!({
        "telegramUserId":telegram_user_id,
        "kind":kind.map(|kind| kind.as_str()).unwrap_or("document"),
        "fileName":file_name,
        "mimeType":mime_type,
        "telegramMessageId":i64::from(sent.id.0),
    });
    if let Some(conversation_id) = conversation_id {
        response["conversationId"] = json!(conversation_id);
    }
    Ok(response)
}

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,
    expected_batch_size: usize,
    delivery_label: &str,
) -> Result<RelayEvent, ApiError> {
    if complete {
        let event = fetch_event(db, event_id)?;
        let changed = db
            .execute(
                "UPDATE telegram_events SET status='complete',completed_at=?1
                 WHERE COALESCE(batch_id,id)=?2 AND status<>'complete'
                   AND conversation_id=?3",
                params![
                    Utc::now().to_rfc3339(),
                    event_batch_id(&event),
                    conversation_id
                ],
            )
            .map_err(ApiError::internal)?;
        if changed != expected_batch_size {
            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, expected_batch_size) = {
        let db = state.db.lock().map_err(ApiError::internal)?;
        let event = outbound_event(&db, &event_id, conversation_id)?;
        let expected_batch_size = active_event_batch_size(&db, &event)?;
        (event, expected_batch_size)
    };

    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,
        expected_batch_size,
        "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, expected_batch_size) = {
        let db = state.db.lock().map_err(ApiError::internal)?;
        let event = outbound_event(&db, &event_id, conversation_id)?;
        let expected_batch_size = active_event_batch_size(&db, &event)?;
        (event, expected_batch_size)
    };

    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,
        expected_batch_size,
        "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());
    }

    #[test]
    fn outbound_completion_rejects_a_batch_invalidated_during_delivery() {
        let database = Connection::open_in_memory().unwrap();
        database.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
        apply_migrations(&database).unwrap();
        let conversation_id = "019f5ca7-020f-7b63-be2f-82785fb68c03";
        let now = Utc::now().to_rfc3339();
        for (id, update_id, message_id) in [("first", 1_i64, 1_i64), ("second", 2_i64, 2_i64)] {
            database
                .execute(
                    "INSERT INTO telegram_events(
                         id,update_id,message_id,telegram_user_id,chat_id,display_name,
                         kind,text,status,conversation_id,processing_started_at,created_at,
                         session_kind,batch_id,batch_ready_at
                     ) VALUES(?1,?2,?3,42,42,'David','text',?1,'processing',?4,
                              ?5,?5,'private','batch',?5)",
                    params![id, update_id, message_id, conversation_id, now],
                )
                .unwrap();
        }
        let event = fetch_event(&database, "second").unwrap();
        let expected_batch_size = active_event_batch_size(&database, &event).unwrap();
        assert_eq!(expected_batch_size, 2);

        database
            .execute(
                "UPDATE telegram_events SET status='complete' WHERE batch_id='batch'",
                [],
            )
            .unwrap();
        let error = reconcile_outbound_event(
            &database,
            "second",
            conversation_id,
            true,
            expected_batch_size,
            "file",
        )
        .unwrap_err();
        assert_eq!(error.code, "state_conflict");
    }

    #[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_bound_and_cold_delivery_keep_distinct_binding_semantics() {
        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
        );

        let service = Service {
            state: state.clone(),
        };
        let cold = service
            .send_cold_private_attachment(
                42,
                Attachment {
                    bytes: b"cold".to_vec(),
                    file_name: Some("cold.txt".into()),
                    media_type: Some("text/plain".into()),
                    kind: None,
                    caption: None,
                },
            )
            .await
            .unwrap();
        assert!(cold.get("conversationId").is_none());
        assert_eq!(cold["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();
    }
}