fixed type issues

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(())
+}

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<String>,
+
+  pub user_id : Option<u64>,
+  pub username : Option<String>,
+  pub display_name : Option<String>,
+  pub profile_image_url : Option<String>,
 }
 
 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<Tweet>)
+    -> 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<Tweet>) -> AnyhowResult<()> {
+    let tweet = match tweet {
+      None => return Ok(()),
+      Some(tweet) => tweet,
+    };
+
+    debug!("Tweet: {:?}", &tweet);
 
     let mut user_id : Option<u64> = None;
     let mut username : Option<String> = 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<String>,
-
-  user_id : Option<u64>,
-  username : Option<String>,
-  display_name : Option<String>,
-  profile_image_url : Option<String>,
-}

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(())

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<i32>,
+    #[prost(enumeration="pubsub_event_payload_v1::IngestionPayloadType", optional, tag="2")]
+    pub ingestion_payload_type: ::core::option::Option<i32>,
+    /// 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<u8>>,
+    /// 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<u8>>,
+    /// 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<i64>,
+    #[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<bool>,
+    #[prost(bool, optional, tag="4")]
+    pub user_is_subscribed: ::core::option::Option<bool>,
+    /// 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<i64>,
+    #[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<bool>,
+    #[prost(bool, optional, tag="4")]
+    pub is_subscribed: ::core::option::Option<bool>,
+    /// 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>,
+}

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<Vec<u8>> {
+  let mut buffer : Vec<u8> = 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<protos::PubsubEventPayloadV1>
+{
+  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)
+}