aves-tyto 0.5.0

use std::sync::Arc;
use std::time::Duration;

use teloxide_core::Bot;
use teloxide_core::payloads::GetUpdatesSetters;
use teloxide_core::prelude::Requester;
use teloxide_core::types::{AllowedUpdate, UpdateKind};
use tokio::sync::mpsc;
use tokio::time::timeout;

use crate::channels::{InboundEvent, MessagingTask};
use crate::messages::ListenerType;
use crate::types::{MessagingInteraction, MessagingSubscriptionError};

use super::helpers::{REQUEST_DEADLINE_SECS, callback_interaction, normalize_update};

/// Telegram long-poll wait, in seconds.
const LONG_POLL_TIMEOUT_SECS: u32 = 30;
/// Maximum number of updates requested per poll.
const POLL_LIMIT: u8 = 100;
/// Outer deadline for a long-poll request.
const GET_UPDATES_DEADLINE_SECS: u64 = (LONG_POLL_TIMEOUT_SECS as u64) * 2;

/// Poll Telegram and forward supported updates until either receiver closes.
pub(super) async fn listen(
    bot: Arc<Bot>,
    tx: mpsc::Sender<InboundEvent>,
    interaction_tx: mpsc::Sender<MessagingInteraction>,
) -> Result<MessagingTask, MessagingSubscriptionError> {
    Ok(tokio::spawn(async move {
        let mut offset = 0;
        loop {
            let updates = tokio::select! {
                _ = tx.closed() => return Ok(()),
                _ = interaction_tx.closed() => return Ok(()),
                result = timeout(Duration::from_secs(GET_UPDATES_DEADLINE_SECS), bot.get_updates().offset(offset).timeout(LONG_POLL_TIMEOUT_SECS).limit(POLL_LIMIT).allowed_updates([AllowedUpdate::Message, AllowedUpdate::ChannelPost, AllowedUpdate::CallbackQuery])) => {
                    result.map_err(|_| MessagingSubscriptionError::Timeout { listener_type: ListenerType::Telegram, operation: "getUpdates".to_owned(), timeout: format!("{GET_UPDATES_DEADLINE_SECS}s") })?.map_err(|source| MessagingSubscriptionError::Protocol { listener_type: ListenerType::Telegram, message: format!("Telegram getUpdates error: {source}") })?
                }
            };
            for update in updates {
                offset = update.id.0 as i32 + 1;
                match &update.kind {
                    UpdateKind::Message(_) | UpdateKind::ChannelPost(_) => {
                        let raw_payload =
                            serde_json::to_string(&update).map_err(|error| MessagingSubscriptionError::Protocol {
                                listener_type: ListenerType::Telegram,
                                message: format!("Telegram update serialization failed: {error}"),
                            })?;
                        let event = InboundEvent {
                            raw_payload: raw_payload.into_bytes(),
                            normalized: normalize_update(&update),
                        };
                        if tx.send(event).await.is_err() {
                            return Ok(());
                        }
                    }
                    UpdateKind::CallbackQuery(callback) => {
                        timeout(
                            Duration::from_secs(REQUEST_DEADLINE_SECS),
                            bot.answer_callback_query(callback.id.clone()),
                        )
                        .await
                        .map_err(|_| MessagingSubscriptionError::Timeout {
                            listener_type: ListenerType::Telegram,
                            operation: "answerCallbackQuery".to_owned(),
                            timeout: format!("{}s", REQUEST_DEADLINE_SECS),
                        })?
                        .map_err(|source| MessagingSubscriptionError::Protocol {
                            listener_type: ListenerType::Telegram,
                            message: format!("Telegram callback acknowledgement failed: {source}"),
                        })?;
                        let raw_payload =
                            serde_json::to_string(callback).map_err(|error| MessagingSubscriptionError::Protocol {
                                listener_type: ListenerType::Telegram,
                                message: format!("Telegram callback serialization failed: {error}"),
                            })?;
                        if let Some(interaction) = callback_interaction(callback, raw_payload)
                            && interaction_tx.send(interaction).await.is_err()
                        {
                            return Ok(());
                        }
                    }
                    _ => {}
                }
            }
        }
    }))
}

#[cfg(test)]
mod tests {
    use tokio::sync::mpsc;

    use crate::channels::MessagingSubscription;
    use crate::types::NormalizedMessage;

    #[tokio::test]
    async fn subscription_checks_explicit_chat_ids_without_opening_a_connection() {
        for channels in [Some(Vec::new()), Some(vec!["not-a-chat-id".to_owned()])] {
            let (_, rx) = mpsc::channel::<NormalizedMessage>(1);
            let (tx, _) = mpsc::channel(1);
            assert!(
                super::super::filter::subscribe(
                    MessagingSubscription {
                        subscribed_channels: channels,
                        allowed_senders: Vec::new()
                    },
                    rx,
                    tx
                )
                .is_err()
            );
        }
        for channels in [None, Some(vec!["-100123456789".to_owned()])] {
            let (_, rx) = mpsc::channel::<NormalizedMessage>(1);
            let (tx, _) = mpsc::channel(1);
            let task = super::super::filter::subscribe(
                MessagingSubscription {
                    subscribed_channels: channels,
                    allowed_senders: Vec::new(),
                },
                rx,
                tx,
            )
            .unwrap();
            task.abort();
        }
    }
}