From ebd98a3571bb56484b10165ed237475d29ec9bd5 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 10 May 2021 11:31:31 -0400 Subject: [PATCH] fixed type issues --- build.rs | 17 +++ protos/storyteller-protos | 2 +- src/clients/twitter_client.rs | 85 ++++++++--- src/main.rs | 6 +- src/proto_codegen/storyteller.pubsub.rs | 192 ++++++++++++++++++++++++ src/protos.rs | 63 ++++++++ 6 files changed, 339 insertions(+), 26 deletions(-) create mode 100644 build.rs create mode 100644 src/proto_codegen/storyteller.pubsub.rs create mode 100644 src/protos.rs diff --git a/build.rs b/build.rs new file mode 100644 index 0000000000..2b28014080 --- /dev/null +++ b/build.rs @@ -0,0 +1,17 @@ +extern crate anyhow; // TODO: No! +extern crate prost_build; + +use std::env; + +fn main() -> anyhow::Result<()> { + // NB: Control prost output dir with env var + // Typically these land in `target/`, but I want to source control these. + env::set_var("OUT_DIR", "src/proto_codegen"); + + prost_build::compile_protos( + &[ + "protos/storyteller-protos/protos/pubsub-events/pubsub-events.proto" + ], + &["protos/"])?; + Ok(()) +} diff --git a/protos/storyteller-protos b/protos/storyteller-protos index c61a069a6f..67b750d956 160000 --- a/protos/storyteller-protos +++ b/protos/storyteller-protos @@ -1 +1 @@ -Subproject commit c61a069a6ffcd6fdc15684e3086c48373a670fd0 +Subproject commit 67b750d956c9e62051e67acab850a3cbc4398ad9 diff --git a/src/clients/twitter_client.rs b/src/clients/twitter_client.rs index 6645e20b9b..18796da830 100644 --- a/src/clients/twitter_client.rs +++ b/src/clients/twitter_client.rs @@ -1,22 +1,43 @@ use crate::AnyhowResult; -use log::{info, debug, warn}; -use futures::TryStreamExt; +use anyhow::anyhow; use egg_mode::stream::StreamMessage; -use std::time::Duration; -use std::thread; use egg_mode::tweet::Tweet; +use futures::TryStreamExt; +use log::{info, debug, warn}; +use std::thread; +use std::time::Duration; +use crate::clients::redis_client::RedisClient; +use futures::executor::block_on; const VOCODES_USER_ID : u64 = 1297106371238932481; +/// The client logic pub struct TwitterClient { access_token: egg_mode::Token, + redis_client: RedisClient, +} + +/// The parts of tweets we care about +#[derive(Clone, Debug)] +pub struct TweetDetails { + pub tweet_text: String, + + pub has_mention: bool, + pub is_retweet: bool, + pub retweeted_text: Option, + + pub user_id : Option, + pub username : Option, + pub display_name : Option, + pub profile_image_url : Option, } impl TwitterClient { - pub fn new(access_token: egg_mode::Token) -> Self { + pub fn new(access_token: egg_mode::Token, redis_client: RedisClient) -> Self { Self { access_token, + redis_client, } } @@ -37,19 +58,43 @@ impl TwitterClient { .track(&["vocodes"]) .start(&self.access_token) .try_for_each(|m| { + // NB: Odd implementation is due to future type juggling. + // I should fix this, but I'm too lazy and have more work to do. if let StreamMessage::Tweet(tweet) = m { - self.handle_tweet(tweet); + self.handle_tweet_mapping_errors(Some(tweet)) } else { info!("Other (non-Tweet) message: {:?}", &m); + self.handle_tweet_mapping_errors(None) } - futures::future::ok(()) }); - Ok(stream.await?) + match stream.await { + Ok(_) => Ok(()), + Err(_) => Err(anyhow!("Stream error")), // NB: loss of error due to shenanigans + } } - fn handle_tweet(&self, tweet: Tweet) -> AnyhowResult<()> { - info!("Tweet: {:?}", &tweet); + // NB: This is some nonsense to satisfy type requirements. + // I'm too lazy to figure out why try_for_each demands egg_mode's Error. + async fn handle_tweet_mapping_errors(&self, tweet: Option) + -> Result<(), egg_mode::error::Error> + { + // NB: This is horrible. + self.handle_tweet(tweet) + .await + .map_err(|e| { + warn!("Error during stream: {:?}", e); + egg_mode::error::Error::BadUrl + }) + } + + async fn handle_tweet(&self, tweet: Option) -> AnyhowResult<()> { + let tweet = match tweet { + None => return Ok(()), + Some(tweet) => tweet, + }; + + debug!("Tweet: {:?}", &tweet); let mut user_id : Option = None; let mut username : Option = None; @@ -101,22 +146,14 @@ impl TwitterClient { tweet_text, }; - info!("\n\nTweet Details: {:?}", tweet_details); + info!("Tweet Details: {:?}", tweet_details); + self.publish_tweet(tweet_details).await?; Ok(()) } + + async fn publish_tweet(&self, tweet_details: TweetDetails) -> AnyhowResult<()> { + Ok(()) + } } -#[derive(Clone, Debug)] -pub struct TweetDetails { - tweet_text: String, - - has_mention: bool, - is_retweet: bool, - retweeted_text: Option, - - user_id : Option, - username : Option, - display_name : Option, - profile_image_url : Option, -} diff --git a/src/main.rs b/src/main.rs index df067ec7e6..3f74b98d25 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,7 @@ extern crate serde_derive; mod clients; +mod protos; mod secrets; use futures::TryStreamExt; @@ -54,7 +55,10 @@ async fn main() -> anyhow::Result<()> info!("Streaming..."); - let twitter_client = TwitterClient::new(twitter_access_token); + let twitter_client = TwitterClient::new( + twitter_access_token, + redis_client, + ); twitter_client.main_loop().await; Ok(()) diff --git a/src/proto_codegen/storyteller.pubsub.rs b/src/proto_codegen/storyteller.pubsub.rs new file mode 100644 index 0000000000..65fbfa61f2 --- /dev/null +++ b/src/proto_codegen/storyteller.pubsub.rs @@ -0,0 +1,192 @@ +/// Ingestion platforms are "dumb" and shouldn't have much logic in them. +/// (Granted, we might eventually grow to send them back messages.) +/// +/// The "enricher" is the smarts. It does denylisting, rate limiting, +/// service calls, transforms, etc. +/// +/// What if we need to scale? Several ideas: Sharding (redis db indices), +/// brokers with read-once semantics (our own intelligence - UUIDs with +/// transactional semantics in MySQL). +/// +/// Events coming from multiple sources +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct PubsubEventPayloadV1 { + /// Source of the event. + #[prost(enumeration="pubsub_event_payload_v1::IngestionSourceType", optional, tag="1")] + pub ingestion_source_type: ::core::option::Option, + #[prost(enumeration="pubsub_event_payload_v1::IngestionPayloadType", optional, tag="2")] + pub ingestion_payload_type: ::core::option::Option, + /// A binary-encoded proto whose type is based on IngestionSourceType + #[prost(bytes="vec", optional, tag="3")] + pub ingestion_source_data: ::core::option::Option<::prost::alloc::vec::Vec>, + /// A binary-encoded proto whose type is based on IngestionPayloadType + #[prost(bytes="vec", optional, tag="4")] + pub ingestion_payload_data: ::core::option::Option<::prost::alloc::vec::Vec>, + /// An optional debug message. + #[prost(string, optional, tag="5")] + pub debug_message: ::core::option::Option<::prost::alloc::string::String>, +} +/// Nested message and enum types in `PubsubEventPayloadV1`. +pub mod pubsub_event_payload_v1 { + /// TODO: This should all be merged in the future, but I need to settle on proto2 vs proto3 + /// And I also need to figure out package names, etc. + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum IngestionSourceType { + IstDoNotUse = 0, + IstAdminPanel = 1, + IstVocodesWebsite = 2, + IstTwitch = 3, + IstTwitter = 4, + IstDiscord = 5, + IstYoutube = 6, + IstReddit = 7, + IstTiktok = 8, + } + /// These will differ from the "enriched" types. + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum IngestionPayloadType { + IptDoNotUse = 0, + /// Twitch-specific events. + /// + /// This can encode lots of things. + TwitchMessage = 100, + TwitchSubscribed = 101, + /// Twitter-specific events. + TwitterFollow = 200, + TwitterRetweet = 201, + TwitterMention = 202, + /// Discord-specific events. + DiscordMessage = 300, + DiscordJoined = 301, + /// YouTube-specific events. + YoutubeSubscribed = 400, + /// Reddit-specific events. + RedditMessage = 500, + } +} +// ============================== SOURCE PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitchMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(int64, optional, tag="1")] + pub user_id: ::core::option::Option, + #[prost(string, optional, tag="2")] + pub username: ::core::option::Option<::prost::alloc::string::String>, + #[prost(bool, optional, tag="3")] + pub user_is_mod: ::core::option::Option, + #[prost(bool, optional, tag="4")] + pub user_is_subscribed: ::core::option::Option, + /// Optional. Name of the channel, prefixed with hash. eg. `#vocodes`. + #[prost(string, optional, tag="5")] + pub channel: ::core::option::Option<::prost::alloc::string::String>, +} +/// Optional. Fields describing the user that sent the message. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterMetadata { + // TODO: This is a u64. We need to be careful with math to assure non-overflow. + //optional int64 user_id = 1; + + #[prost(string, optional, tag="2")] + pub username: ::core::option::Option<::prost::alloc::string::String>, + #[prost(string, optional, tag="3")] + pub display_name: ::core::option::Option<::prost::alloc::string::String>, + #[prost(string, optional, tag="4")] + pub avatar_url: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionDiscordMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionYoutubeMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionRedditMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTikTokMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +// ============================== TWITCH PAYLOAD PROTOS ============================== // + +/// These are raw messages from Twitch that we send to the PubSub enricher. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitchMessage { + /// TODO: DEPRECATE THESE + #[prost(int64, optional, tag="1")] + pub user_id: ::core::option::Option, + #[prost(string, optional, tag="2")] + pub username: ::core::option::Option<::prost::alloc::string::String>, + #[prost(bool, optional, tag="3")] + pub is_mod: ::core::option::Option, + #[prost(bool, optional, tag="4")] + pub is_subscribed: ::core::option::Option, + /// TODO: DEPRECATE THESE + /// Name of the channel, prefixed with hash. eg. `#vocodes`. + #[prost(string, optional, tag="5")] + pub channel: ::core::option::Option<::prost::alloc::string::String>, + /// Actual message. Doesn't need to be trimmed. + #[prost(string, optional, tag="6")] + pub message_contents: ::core::option::Option<::prost::alloc::string::String>, +} +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitchSubscribed { +} +// ============================== TWITTER PAYLOAD PROTOS ============================== // + +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterFollow { +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterRetweet { + /// Intentionally blank for now + #[prost(string, optional, tag="1")] + pub original_text: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterMention { + /// Intentionally blank for now + #[prost(string, optional, tag="1")] + pub text: ::core::option::Option<::prost::alloc::string::String>, +} +// ============================== DISCORD PAYLOAD PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionDiscordMessage { + /// Raw message. Needs to be trimmed + #[prost(string, optional, tag="6")] + pub message_contents: ::core::option::Option<::prost::alloc::string::String>, +} +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionDiscordJoined { +} +// ============================== YOUTUBE PAYLOAD PROTOS ============================== // + +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionYoutubeSubscribed { +} +// ============================== REDDIT PAYLOAD PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionRedditMessage { + /// Raw message. Needs to be trimmed + #[prost(string, optional, tag="6")] + pub message_contents: ::core::option::Option<::prost::alloc::string::String>, +} diff --git a/src/protos.rs b/src/protos.rs new file mode 100644 index 0000000000..96cc94b27e --- /dev/null +++ b/src/protos.rs @@ -0,0 +1,63 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use log::{info, warn, debug}; +use crate::clients::twitter_client::TweetDetails; + +// Include the protos +pub mod protos { + include!("proto_codegen/storyteller.pubsub.rs"); +} + +// Binary encode a proto. +pub fn binary_encode_proto(proto: impl prost::Message) -> AnyhowResult> { + let mut buffer : Vec = Vec::with_capacity(proto.encoded_len()); + let encode_result = proto.encode(&mut buffer); + + match encode_result { + Err(e) => { + Err(anyhow!("Inner proto encode result: {:?}", e)) + } + Ok(_) => { + Ok(buffer) + } + } +} + +pub fn mention_to_proto<'a>(tweet_details: TweetDetails) -> AnyhowResult +{ + let mut payload_proto = protos::PubsubEventPayloadV1::default(); + + payload_proto.ingestion_source_type = + Some(protos::pubsub_event_payload_v1::IngestionSourceType::IstTwitter as i32); + + let binary_twitter_metadata = { + let mut twitter_metadata = protos::IngestionTwitterMetadata::default(); + // TODO: Don't want to overflow. Represent as bytes in the meantime + // twitter_metadata.user_id = Some(tweet_details.user_id as i64); + twitter_metadata.username = tweet_details.username.clone(); + twitter_metadata.display_name = tweet_details.display_name.clone(); + twitter_metadata.avatar_url = tweet_details.profile_image_url.clone(); + + debug!("Twitter Metadata Proto: {:?}", twitter_metadata); + + binary_encode_proto(twitter_metadata) + }?; + + payload_proto.ingestion_source_data = Some(binary_twitter_metadata); + + payload_proto.ingestion_payload_type = + Some(protos::pubsub_event_payload_v1::IngestionPayloadType::TwitterMention as i32); + + let binary_twitter_mention = { + let mut twitter_mention = protos::IngestionTwitterMention::default(); + twitter_mention.text = Some(tweet_details.tweet_text.clone()); + + debug!("Twitter Mention Proto: {:?}", twitter_mention); + + binary_encode_proto(twitter_mention) + }?; + + payload_proto.ingestion_payload_data = Some(binary_twitter_mention); + + Ok(payload_proto) +}