aves-tyto 0.5.0

use std::sync::Arc;

use slack_morphism::prelude::{
    SlackClient, SlackClientEventsListenerEnvironment, SlackClientEventsUserState, SlackClientHyperHttpsConnector,
    SlackClientSocketModeConfig, SlackClientSocketModeListener, SlackEventCallbackBody, SlackEventId,
    SlackInteractionEvent, SlackMessageEvent, SlackPushEventCallback, SlackSocketModeListenerCallbacks, SlackTeamId,
    UserCallbackResult,
};
use tokio::sync::mpsc;
use tracing::error;

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

use super::connection::{SlackConnection, connect};
use super::helpers::{interaction_to_message, normalize_update};

struct CallbackState {
    tx: mpsc::Sender<InboundEvent>,
    interaction_tx: mpsc::Sender<MessagingInteraction>,
}

pub(super) async fn listen(
    connection: Arc<SlackConnection>,
    tx: mpsc::Sender<InboundEvent>,
    interaction_tx: mpsc::Sender<MessagingInteraction>,
) -> Result<MessagingTask, MessagingSubscriptionError> {
    let delivery = tx.clone();
    let interaction_delivery = interaction_tx.clone();
    let state = CallbackState { tx, interaction_tx };
    let environment =
        Arc::new(SlackClientEventsListenerEnvironment::new(connection.client.clone()).with_user_state(state));

    let config = SlackClientSocketModeConfig::new();
    let callbacks = SlackSocketModeListenerCallbacks::<SlackClientHyperHttpsConnector>::new()
        .with_push_events(on_push_event)
        .with_interaction_events(on_interaction);
    let listener = SlackClientSocketModeListener::new(&config, environment, callbacks);

    connect(listener.listen_for(&connection.token)).await?;

    Ok(tokio::spawn(async move {
        tokio::select! {
            _ = delivery.closed() => Ok(()),
            _ = interaction_delivery.closed() => Ok(()),
            _ = listener.start() => Err(MessagingSubscriptionError::StreamClosed {
              listener_type: ListenerType::Slack,
              message: "Slack Socket Mode connection closed".to_owned()
            }),
        }
    }))
}

async fn on_push_event(
    push_event: SlackPushEventCallback,
    _client: Arc<SlackClient<SlackClientHyperHttpsConnector>>,
    state_storage: SlackClientEventsUserState,
) -> UserCallbackResult<()> {
    let raw_payload = serde_json::to_string(&push_event).unwrap_or_default();
    let team_id = push_event.team_id.clone();
    let event_id = push_event.event_id.clone();
    if let SlackEventCallbackBody::Message(message) = push_event.event {
        let state_guard = state_storage.read().await;
        if let Some(state) = state_guard.get_user_state::<CallbackState>() {
            handle_message_event(message, &team_id, &event_id, &raw_payload, &state.tx).await;
        } else {
            tracing::error!("Slack callback state missing from user state storage");
        }
    }
    Ok(())
}

async fn on_interaction(
    interaction: SlackInteractionEvent,
    _client: Arc<SlackClient<SlackClientHyperHttpsConnector>>,
    state_storage: SlackClientEventsUserState,
) -> UserCallbackResult<()> {
    let mapped = interaction_to_message(&interaction);
    if let Some(interaction) = mapped {
        let state_guard = state_storage.read().await;
        if let Some(state) = state_guard.get_user_state::<CallbackState>()
            && state.interaction_tx.try_send(interaction).is_err()
        {
            error!("Slack interaction queue is unavailable");
        }
    }
    Ok(())
}

async fn handle_message_event(
    msg_event: SlackMessageEvent,
    team_id: &SlackTeamId,
    event_id: &SlackEventId,
    raw_payload: &str,
    tx: &mpsc::Sender<InboundEvent>,
) {
    let channel_id = match &msg_event.origin.channel {
        Some(ch) => ch.clone(),
        None => {
            tracing::debug!("Slack message event has no channel ID; skipping");
            return;
        }
    };

    if msg_event.subtype.is_some() {
        tracing::debug!("Slack message event has subtype; skipping");
        return;
    }

    let event = InboundEvent {
        raw_payload: raw_payload.as_bytes().to_vec(),
        normalized: normalize_update(&msg_event, channel_id, team_id, event_id),
    };
    if tx.send(event).await.is_err() {
        error!("Slack message mpsc receiver closed; listener will exit");
    }
}

#[cfg(test)]
mod tests {
    use slack_morphism::prelude::{
        SlackChannelId, SlackMessageContent, SlackMessageEvent, SlackMessageEventType, SlackMessageOrigin,
        SlackMessageSender, SlackTs,
    };
    use tokio::sync::mpsc;

    use crate::channels::MessagingSubscription;

    use super::*;

    const TARGET_CHANNEL: &str = "C12345678";
    use super::super::filter::SlackFilter;

    async fn handle_message_event(
        message: SlackMessageEvent,
        team: &SlackTeamId,
        event: &SlackEventId,
        policy: &SlackFilter,
        tx: &mpsc::Sender<InboundEvent>,
    ) {
        let (raw_tx, mut raw_rx) = mpsc::channel(8);
        let (normalized_tx, normalized_rx) = mpsc::channel(8);
        let (filtered_tx, mut filtered_rx) = mpsc::channel(8);
        let task = super::super::filter::subscribe(
            MessagingSubscription {
                subscribed_channels: policy.subscribed_channels.clone(),
                allowed_senders: policy.allowed_senders.clone(),
            },
            normalized_rx,
            filtered_tx,
        )
        .unwrap();
        let serialized = serde_json::to_string(&message).unwrap();
        super::handle_message_event(message, team, event, &serialized, &raw_tx).await;
        if let Ok(event) = raw_rx.try_recv()
            && let Some(normalized) = event.normalized
        {
            normalized_tx.send(normalized).await.unwrap();
        }
        drop(raw_tx);
        drop(normalized_tx);
        task.await.unwrap().unwrap();
        if let Ok(normalized) = filtered_rx.try_recv() {
            tx.send(InboundEvent {
                raw_payload: Vec::new(),
                normalized: Some(normalized),
            })
            .await
            .unwrap();
        }
    }
    const OTHER_CHANNEL: &str = "C99999999";
    const TEAM_ID: &str = "T0001";
    const EVENT_ID: &str = "Ev0001";

    fn make_team_id() -> SlackTeamId {
        SlackTeamId::new(TEAM_ID.into())
    }

    fn make_event_id() -> SlackEventId {
        SlackEventId::new(EVENT_ID.into())
    }

    fn make_ts() -> SlackTs {
        SlackTs::new("1234567890.000001".into())
    }

    /// Build a minimal `SlackMessageEvent` for the target channel with text content.
    fn make_text_event(channel: &str, text: &str) -> SlackMessageEvent {
        let origin = SlackMessageOrigin::new(make_ts()).with_channel(SlackChannelId::new(channel.into()));
        let sender = SlackMessageSender::new();
        let content = SlackMessageContent::new().with_text(text.into());
        SlackMessageEvent::new(origin, sender).with_content(content)
    }

    /// Build a minimal `SlackMessageEvent` with no channel ID in origin.
    fn make_event_no_channel(text: &str) -> SlackMessageEvent {
        let origin = SlackMessageOrigin::new(make_ts());
        let sender = SlackMessageSender::new();
        let content = SlackMessageContent::new().with_text(text.into());
        SlackMessageEvent::new(origin, sender).with_content(content)
    }

    /// Build a `SlackMessageEvent` with a subtype (e.g. bot_message).
    fn make_event_with_subtype(channel: &str, text: &str) -> SlackMessageEvent {
        let origin = SlackMessageOrigin::new(make_ts()).with_channel(SlackChannelId::new(channel.into()));
        let sender = SlackMessageSender::new();
        let content = SlackMessageContent::new().with_text(text.into());
        SlackMessageEvent::new(origin, sender)
            .with_content(content)
            .with_subtype(SlackMessageEventType::BotMessage)
    }

    /// Build a `SlackMessageEvent` with no text content (attachment-only).
    fn make_event_no_text(channel: &str) -> SlackMessageEvent {
        let origin = SlackMessageOrigin::new(make_ts()).with_channel(SlackChannelId::new(channel.into()));
        let sender = SlackMessageSender::new();
        // No content set — simulates attachment-only message.
        SlackMessageEvent::new(origin, sender)
    }

    // ── handle_message_event tests ────────────────────────────────────────────

    #[tokio::test]
    async fn message_matching_channel_forwarded() {
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_text_event(TARGET_CHANNEL, "hello slack");

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        let msg = rx.try_recv().expect("expected a forwarded message");
        let msg = msg.normalized.expect("normalized");
        assert_eq!(msg.body, "hello slack");
        assert_eq!(msg.metadata["channel_type"], "slack");
    }

    #[tokio::test]
    async fn message_wrong_channel_dropped() {
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_text_event(OTHER_CHANNEL, "wrong channel");

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        assert!(rx.try_recv().is_err(), "expected no message for wrong channel");
    }

    #[tokio::test]
    async fn message_no_channel_id_dropped() {
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_event_no_channel("no channel");

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        assert!(rx.try_recv().is_err(), "expected no message when channel ID absent");
    }

    #[tokio::test]
    async fn message_with_subtype_dropped() {
        // Non-human subtypes (bot_message, channel_join, etc.) must be filtered.
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_event_with_subtype(TARGET_CHANNEL, "bot says hi");

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        assert!(rx.try_recv().is_err(), "expected no message for bot_message subtype");
    }

    #[tokio::test]
    async fn message_no_text_dropped() {
        // Attachment-only messages (no text content) must be filtered.
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_event_no_text(TARGET_CHANNEL);

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        assert!(rx.try_recv().is_err(), "expected no message for attachment-only event");
    }

    #[tokio::test]
    async fn message_empty_text_dropped() {
        // Empty text string must be treated as no body.
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_text_event(TARGET_CHANNEL, "");

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        assert!(rx.try_recv().is_err(), "expected no message for empty text");
    }

    #[tokio::test]
    async fn message_metadata_contains_channel_type() {
        let (tx, mut rx) = mpsc::channel(10);
        let event = make_text_event(TARGET_CHANNEL, "metadata check");

        handle_message_event(
            event,
            &make_team_id(),
            &make_event_id(),
            &SlackFilter::new(Some(vec![TARGET_CHANNEL.to_owned()])),
            &tx,
        )
        .await;

        let msg = rx
            .try_recv()
            .expect("expected a forwarded message")
            .normalized
            .expect("normalized");
        assert_eq!(msg.metadata["channel_type"], "slack");
        let channel_id_meta = msg.metadata["channel_id"].as_str().unwrap_or("");
        assert_eq!(channel_id_meta, TARGET_CHANNEL);
    }
}