fatline-rs 0.2.2

use async_trait::async_trait;
use eyre::Result;
use futures::{Stream, StreamExt};
use futures::stream::{BoxStream, iter};

use crate::action::*;
use crate::HubService;
use crate::posts::Cast;
use crate::proto::{CastId, HubEvent, HubEventType, MessageType, SubscribeRequest};
use crate::proto::hub_event::Body;
use crate::utils::{cast_action_from_message, link_from_message, reaction_from_message};

#[async_trait]
pub trait Subscriptions {
    async fn get_new_events(&mut self) -> Result<impl Stream<Item=StreamingType>>;
}

pub enum StreamingType {
    // adds
    CastAdd(Cast),
    LinkAdd(LinkInfo),
    ReactionAdd(ReactionInfo),
    // removals
    CastRemove(CastId),
    LinkRemove(LinkInfo),
    ReactionRemove(ReactionInfo)
}

fn hub_event_to_streaming_type(event: HubEvent) -> Option<Vec<StreamingType>> {
    if event.r#type() == HubEventType::MergeMessage {
        match event.body {
            Some(Body::MergeMessageBody(body)) => {
                // not sure what to do with this information yet
                // body.deleted_messages
                body.message.and_then(|message| {
                    // get the type
                    let message_type = message.data.clone()?.r#type();
                    // if it's a link add type then figure out the state
                    if message_type == MessageType::LinkAdd
                        || message_type == MessageType::LinkRemove
                        || message_type == MessageType::LinkCompactState {
                        let actions = link_from_message(message)?;
                        Some(actions.iter().cloned().map(|action| {
                            match action {
                                LinkAction::AddFollow(follow) => {
                                    StreamingType::LinkAdd(follow)
                                }
                                LinkAction::RemoveFollow(unfollow) => {
                                    StreamingType::LinkRemove(unfollow)
                                }
                            }
                        }).collect())
                    } else if message_type == MessageType::ReactionAdd
                        || message_type == MessageType::ReactionRemove {
                        let action = reaction_from_message(message)?;
                        let streaming_type = match action {
                            ReactionAction::Add(add) => StreamingType::ReactionAdd(add),
                            ReactionAction::Remove(rem) => StreamingType::ReactionRemove(rem)
                        };
                        Some(vec![streaming_type])
                    } else if message_type == MessageType::CastAdd {
                        match cast_action_from_message(message)? {
                            CastAction::CastAdd(cast) => {
                                Some(vec![
                                    StreamingType::CastAdd(cast)
                                ])
                            },
                            CastAction::CastRemove(cast_id) => {
                                let fatline_cast_id = CastId {
                                    fid: cast_id.fid,
                                    hash: cast_id.hash
                                };
                                Some(vec![
                                    StreamingType::CastRemove(fatline_cast_id)
                                ])
                            }
                        }
                    } else {
                        None
                    }
                })
            },
            _ => None
        }
    } else {
        None
    }
}

#[async_trait]
impl Subscriptions for HubService {
    async fn get_new_events(&mut self) -> Result<BoxStream<StreamingType>> {
        let subscription = self.subscribe(SubscribeRequest {
            event_types: vec![],
            from_id: None,
            total_shards: None,
            shard_index: None,
        }).await?.into_inner();
        let mapped = subscription.filter_map(|event| async {
            match event {
                Ok(res) => {
                    hub_event_to_streaming_type(res)
                },
                Err(_) => None
            }
        }).flat_map(iter).boxed();
        Ok(mapped)
    }
}