Merge remote-tracking branch 'old_h/master' into merge-more-old-code

.env-template

@@ -0,0 +1,16 @@
+# Redis connection
+REDIS_USERNAME=
+REDIS_PASSWORD=
+REDIS_HOST=
+REDIS_PORT=
+REDIS_USES_TLS=
+
+# Redis publishing
+#REDIS_TOPIC=firehose
+#REDIS_MAX_RETRY_COUNT=3
+
+# Twitter secrets
+TWITTER_API_KEY=
+TWITTER_API_SECRET=
+TWITTER_ACCESS_KEY=
+TWITTER_ACCESS_SECRET=

.github/workflows/docker-publish.yml

@@ -0,0 +1,65 @@
+# Adapted from
+# https://github.com/actions/starter-workflows/ci/docker-publish.yml
+# If using base images, don't forget to add READ access, eg. from a package settings page like:
+# https://github.com/orgs/storytold/packages/container/docker-base-images-rust-ssl/settings/actions_access
+name: Publish Docker Container to GHCR
+on:
+  push:
+    branches:
+      # NB: Default-branch doesn't work, despite Github's documentation.
+      #- $default-branch
+      - master
+env:
+  IMAGE_NAME: twitter-ingestion-client
+
+jobs:
+  # Push image to GitHub Packages.
+  # See also https://docs.docker.com/docker-hub/builds/
+  push:
+    runs-on: ubuntu-latest
+    if: github.event_name == 'push'
+    permissions:
+      packages: write
+      contents: read
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Login to GitHub Container Registry
+        uses: docker/login-action@v1
+        with:
+          registry: ghcr.io
+          #username: ${{ github.repository_owner }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }} 
+
+      - name: Build image
+        run: docker build . --file Dockerfile --tag $IMAGE_NAME --label "runnumber=${GITHUB_RUN_ID}"
+
+      - name: Push image to GitHub Container Registry
+        run: |
+          IMAGE_ID=ghcr.io/${{ github.repository_owner }}/$IMAGE_NAME
+
+          # Change all uppercase to lowercase
+          IMAGE_ID=$(echo $IMAGE_ID | tr '[A-Z]' '[a-z]')
+
+          # Strip git ref prefix from version
+          VERSION=$(echo "${{ github.ref }}" | sed -e 's,.*/\(.*\),\1,')
+
+          # Strip "v" prefix from tag name
+          [[ "${{ github.ref }}" == "refs/tags/"* ]] && VERSION=$(echo $VERSION | sed -e 's/^v//')
+
+          # Use Docker `latest` tag convention
+          [ "$VERSION" == "$default-branch" ] && VERSION=latest
+
+          echo IMAGE_ID=$IMAGE_ID
+          echo VERSION=$VERSION
+
+          docker tag $IMAGE_NAME $IMAGE_ID:$VERSION
+          docker push $IMAGE_ID:$VERSION
+
+          SHORT_SHA=$(echo ${GITHUB_SHA} | cut -c1-12)
+
+          docker tag $IMAGE_NAME $IMAGE_ID:$SHORT_SHA
+          docker push $IMAGE_ID:$SHORT_SHA
+

.gitignore

@@ -0,0 +1,6 @@
+*~
+.env
+.idea/
+secrets.toml
+secrets.yaml
+target/

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "protos/storyteller-protos"]
+	path = protos/storyteller-protos
+	url = git@github.com:storytold/storyteller-protos.git

Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "twitter-ingestion-client"
+version = "0.0.0"
+edition = "2018"
+authors = [
+  "Brandon Thomas <bt@brand.io>",
+  "Brandon Thomas <echelon@gmail.com>"
+]
+
+[[bin]]
+name = "twitter-ingestion-client"
+path = "src/main.rs"
+
+[[bin]]
+name = "twitter-get-acccess-token"
+path = "src/main_get_access_token.rs"
+
+[dependencies]
+anyhow = "1.0.38"
+easyenv = "0.0.5"
+futures = "0.3.14"
+log = "0.4.14"
+prost = "0.7.0"
+rustls = "0.18"
+serde = "1.0.123"
+serde_derive = "1.0.123"
+tokio = { version = "0.2", features = ["full"] }
+toml = "0.5.8"
+
+redis = { version = "0.17.0", features = ["tls", "async-std-tls-comp", "tokio-tls-comp", "connection-manager"] }
+
+# Twitter
+egg-mode = "0.15.0"
+
+[build-dependencies]
+anyhow = "1.0.38"
+prost-build = "0.7.0"
+

Dockerfile

@@ -0,0 +1,49 @@
+# Custom base image
+# Make sure to add this repository so it has read acces to the base image:
+# https://github.com/orgs/storytold/packages/container/docker-base-images-rust-ssl/settings/actions_access
+FROM ghcr.io/storytold/docker-base-images-rust-ssl:latest as build
+WORKDIR /tmp
+
+COPY Cargo.lock . 
+COPY Cargo.toml .
+COPY src/ ./src
+COPY protos/ ./protos
+
+RUN $HOME/.cargo/bin/cargo fetch
+
+#RUN LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} $HOME/.cargo/bin/cargo build
+RUN LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} $HOME/.cargo/bin/cargo build --release --bin twitter-ingestion-client
+
+# Final image
+FROM ubuntu:xenial
+WORKDIR /
+
+# See: https://github.com/opencontainers/image-spec/blob/master/annotations.md
+LABEL org.opencontainers.image.authors='bt@brand.io, echelon@gmail.com'
+LABEL org.opencontainers.image.url='https://github.com/storytold/twitter-ingestion-client'
+LABEL org.opencontainers.image.documentation='https://github.com/storytold/twitter-ingestion-client'
+LABEL org.opencontainers.image.source='https://github.com/storytold/twitter-ingestion-client'
+
+# NB: Comment this out for non-debug images
+# TODO: Figure out how this is done elsewhere with just the static binaries
+RUN apt-get update \
+    && apt-get install -y \
+        curl \
+        wget
+
+#COPY --from=build /tmp/target/debug/twitter-ingestion-client /
+COPY --from=build /tmp/target/release/twitter-ingestion-client /
+
+# SSL certs are required for crypto
+COPY --from=build /etc/ssl /etc/ssl
+
+# Required dynamically linked libraries
+COPY --from=build /lib/x86_64-linux-gnu/libssl.*             /lib/x86_64-linux-gnu/
+COPY --from=build /lib/x86_64-linux-gnu/libcrypto.*          /lib/x86_64-linux-gnu/
+
+# Make sure all the links resolve
+RUN ldd twitter-ingestion-client
+
+EXPOSE 8080
+CMD LD_LIBRARY_PATH=/usr/lib /twitter-ingestion-client
+

README.md

@@ -0,0 +1,12 @@
+twitter-ingestion-client
+========================
+
+Ingest events from Twitter and send them to Redis PubSub
+
+Protos
+------
+Protos are defined in a git submodule:
+
+```
+git submodule add git@github.com:storytold/storyteller-protos.git
+```

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/mod.rs

@@ -0,0 +1,2 @@
+pub mod redis_client;
+pub mod twitter_client;

src/clients/redis_client.rs

@@ -0,0 +1,65 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use crate::secrets::RedisSecrets;
+use log::{debug, info};
+use redis::AsyncCommands;
+use redis::aio::{Connection, ConnectionManager};
+use redis;
+use std::thread;
+use std::time::Duration;
+
+pub struct RedisClient {
+  secrets: RedisSecrets,
+
+  /// This manages our connection.
+  /// It's meant to be absent when not connected.
+  connection_manager: Option<ConnectionManager>,
+
+  /// Attempts to retry a Redis command.
+  max_retry_count: u32,
+}
+
+impl RedisClient {
+  pub fn new(secrets: &RedisSecrets, retry_count: u32) -> Self {
+    Self {
+      secrets: secrets.clone(),
+      connection_manager: None,
+      max_retry_count: retry_count,
+    }
+  }
+
+  pub async fn connect(&mut self) -> AnyhowResult<()> {
+    let client = redis::Client::open(self.secrets.connection_url())?;
+
+    let connection_manager = client.get_tokio_connection_manager().await?;
+    self.connection_manager = Some(connection_manager);
+
+    Ok(())
+  }
+
+  pub async fn publish_bytes(&mut self, channel: &str, message: &[u8]) -> AnyhowResult<u32> {
+    for i in 0 .. self.max_retry_count {
+      match self.try_publish_bytes(channel, message).await {
+        Ok(r) => return Ok(r),
+        Err(_) => {
+          debug!("Redis publish_bytes failed; retrying (attempt {})", i);
+          continue;
+        },
+      }
+    }
+    Err(anyhow!("Retry count elapsed."))
+  }
+
+  async fn try_publish_bytes(&mut self, channel: &str, message: &[u8]) -> AnyhowResult<u32> {
+    let connection_manager = match &mut self.connection_manager {
+      None => {
+        self.connect().await?;
+        return Err(anyhow!("Not connected"));
+      },
+      Some(cm) => cm,
+    };
+
+    let result = connection_manager.publish(channel, message).await?;
+    Ok(result)
+  }
+}

src/clients/twitter_client.rs

@@ -0,0 +1,210 @@
+use crate::AnyhowResult;
+use anyhow::anyhow;
+use egg_mode::stream::StreamMessage;
+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;
+use crate::protos::{mention_to_proto, binary_encode_proto};
+use std::sync::{RwLock, Arc};
+
+const VOCODES_USER_ID : u64 = 1297106371238932481;
+
+/// The client logic
+pub struct TwitterClient {
+  access_token: egg_mode::Token,
+  redis_client: RedisClientWrapper,
+  redis_publish_topic: String,
+}
+
+/// 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>,
+}
+
+// NB: We're using interior mutability.
+struct RedisClientWrapper {
+  redis_client: Arc::<RwLock<RedisClient>>,
+}
+
+impl RedisClientWrapper {
+  pub fn new(redis_client: RedisClient) -> Self {
+    Self {
+      redis_client: Arc::new(RwLock::new(redis_client))
+    }
+  }
+  pub async fn publish_bytes(&self, channel: &str, message: &[u8]) -> AnyhowResult<u32> {
+    let mut redis_client = match self.redis_client.write() {
+      Ok(rc) => rc,
+      Err(e) => return Err(anyhow!("Error: {:?}", )),
+    };
+    redis_client.publish_bytes(channel, message).await
+  }
+}
+
+impl TwitterClient {
+
+  pub fn new(
+    access_token: egg_mode::Token,
+    redis_client: RedisClient,
+    redis_publish_topic: String) -> Self
+  {
+    Self {
+      access_token,
+      redis_client: RedisClientWrapper::new(redis_client),
+      redis_publish_topic,
+    }
+  }
+
+  pub async fn main_loop(&mut self) {
+    loop {
+      info!("Connecting to twitter stream...");
+      if let Err(e) = self.run_stream().await {
+        warn!("Disconnected from stream; error = {:?}", e);
+      }
+      info!("Disconnected; reconnecting after timeout...");
+      thread::sleep(Duration::from_secs(30));
+    }
+  }
+
+  pub async fn run_stream(&mut self) -> AnyhowResult<()> {
+    let stream = egg_mode::stream::filter()
+      .follow(&[VOCODES_USER_ID]) // vocodes
+      .track(&["vocodes"])
+      .start(&self.access_token)
+      //.fold(&self)
+      .try_for_each(|m: StreamMessage| {
+        // NB: Odd implementation is due to future type juggling.
+        // I should fix this, but I'm too lazy and have more work to do.
+        let future = if let StreamMessage::Tweet(tweet) = m {
+          self.handle_tweet_mapping_errors(Some(tweet))
+        } else {
+          info!("Other (non-Tweet) message: {:?}", &m);
+          self.handle_tweet_mapping_errors(None)
+        };
+
+        match block_on(future) {
+          Ok(x) => futures::future::ok(x),
+          Err(e) => futures::future::err(e),
+        }
+
+      });
+
+    match stream.await {
+      Ok(_) => Ok(()),
+      Err(_) => Err(anyhow!("Stream error")), // NB: loss of error due to shenanigans
+    }
+  }
+
+  // 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(&mut 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(&mut 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;
+    let mut display_name : Option<String> = None;
+    let mut profile_image_url : Option<String> = None;
+
+    if let Some(user) = tweet.user {
+      if user.id == VOCODES_USER_ID {
+        warn!("Tweet is by @vocodes account; skipping");
+        return Ok(());
+      }
+
+      user_id = Some(user.id);
+      username = Some(user.screen_name.clone());
+      display_name = Some(user.name.clone());
+      profile_image_url = Some(user.profile_image_url.clone());
+    }
+
+    let mut has_mention = false;
+
+    for mention in tweet.entities.user_mentions.iter() {
+      if mention.id == VOCODES_USER_ID {
+        has_mention = true;
+      }
+    }
+
+    let mut is_retweet = false;
+    let mut retweeted_text : Option<String> = None;
+
+    if let Some(retweet) = tweet.retweeted_status {
+      if let Some(original_author) = retweet.user {
+        if original_author.id == VOCODES_USER_ID {
+          is_retweet = true;
+          retweeted_text = Some(retweet.text.clone());
+        }
+      }
+    }
+
+    let tweet_text = tweet.text.clone();
+
+    let tweet_details = TweetDetails {
+      has_mention,
+      is_retweet,
+      retweeted_text,
+      user_id,
+      username,
+      display_name,
+      profile_image_url,
+      tweet_text,
+    };
+
+    info!("Tweet Details: {:?}", tweet_details);
+    self.publish_tweet(tweet_details).await?;
+
+    Ok(())
+  }
+
+  async fn publish_tweet(&mut self, tweet_details: TweetDetails) -> AnyhowResult<()> {
+    let binary_proto = if !tweet_details.is_retweet && tweet_details.has_mention {
+      // NB: Retweets can have mentions of the target user.
+      let proto = mention_to_proto(tweet_details)?;
+      binary_encode_proto(proto)?
+    } else if tweet_details.is_retweet {
+      let proto = mention_to_proto(tweet_details)?;
+      binary_encode_proto(proto)?
+    } else {
+      return Ok(());
+    };
+
+    self.redis_client.publish_bytes(
+      &self.redis_publish_topic,
+      &binary_proto
+    ).await?;
+
+    Ok(())
+  }
+}
+

src/main.rs

@@ -0,0 +1,99 @@
+#![allow(dead_code)]
+#![allow(unused_mut)]
+#![allow(unused_imports)]
+#![warn(unused_must_use)]
+//#![allow(warnings)]
+
+#[macro_use]
+extern crate serde_derive;
+
+mod clients;
+mod protos;
+mod secrets;
+
+use crate::clients::redis_client::RedisClient;
+use crate::clients::twitter_client::TwitterClient;
+use crate::secrets::RedisSecrets;
+use crate::secrets::Secrets;
+use crate::secrets::TwitterSecrets;
+use egg_mode::stream::StreamMessage;
+use futures::TryStreamExt;
+use log::{info, warn, debug};
+
+pub type AnyhowResult<T> = anyhow::Result<T>;
+
+const ENV_PUBLISH_TOPIC : &'static str = "PUBLISH_TOPIC";
+const ENV_PUBLISH_TOPIC_DEFAULT : &'static str = "firehose";
+
+const ENV_REDIS_MAX_RETRY_COUNT : &'static str = "REDIS_MAX_RETRY_COUNT";
+const ENV_REDIS_MAX_RETRY_COUNT_DEFAULT : u32 = 3;
+
+/// Monitor Twitter for interesting events against our account.
+/// With this implementation, we can capture: Mentions, Retweets, Replies.
+/// We cannot capture: Follows.
+#[tokio::main]
+async fn main() -> anyhow::Result<()>
+{
+  easyenv::init_all_with_default_logging(None);
+
+  info!("Reading env config...");
+
+  let redis_publish_topic = easyenv::get_env_string_or_default(
+    ENV_PUBLISH_TOPIC, ENV_PUBLISH_TOPIC_DEFAULT);
+
+  let redis_max_retry_count = easyenv::get_env_num(
+    ENV_REDIS_MAX_RETRY_COUNT,
+    ENV_REDIS_MAX_RETRY_COUNT_DEFAULT)?;
+
+  //let secrets = Secrets::from_file("secrets.toml")?;
+
+  let redis_secrets = get_redis_secrets()?;
+  let twitter_secrets = get_twitter_secrets()?;
+
+  let mut redis_client = RedisClient::new(
+    &redis_secrets,
+    redis_max_retry_count,
+  );
+
+  info!("Connecting to Redis...");
+
+  redis_client.connect().await?;
+
+  info!("Verifying Twitter access token...");
+  let twitter_access_token = twitter_secrets.verify_access_token().await?;
+
+  info!("Streaming from Twitter...");
+
+  let mut twitter_client = TwitterClient::new(
+    twitter_access_token,
+    redis_client,
+    redis_publish_topic,
+  );
+
+  twitter_client.main_loop().await;
+  Ok(())
+}
+
+fn get_redis_secrets() -> AnyhowResult<RedisSecrets> {
+  Ok(RedisSecrets::new(
+    &easyenv::get_env_string_or_default(
+      "REDIS_USERNAME", ""),
+    &easyenv::get_env_string_or_default(
+    "REDIS_PASSWORD", ""),
+    &easyenv::get_env_string_or_default(
+    "REDIS_HOST", ""),
+    easyenv::get_env_num::<u32>(
+    "REDIS_PORT", 6379)?,
+    easyenv::get_env_bool_or_default(
+      "REDIS_USES_TLS", false),
+  ))
+}
+
+fn get_twitter_secrets() -> AnyhowResult<TwitterSecrets> {
+  Ok(TwitterSecrets::new(
+    &easyenv::get_env_string_required("TWITTER_API_KEY")?,
+    &easyenv::get_env_string_required("TWITTER_API_SECRET")?,
+    &easyenv::get_env_string_optional("TWITTER_ACCESS_KEY"),
+    &easyenv::get_env_string_optional("TWITTER_ACCESS_SECRET"),
+  ))
+}

src/main_get_access_token.rs

@@ -0,0 +1,44 @@
+#![allow(dead_code)]
+#![allow(unused_mut)]
+#![allow(unused_imports)]
+#![warn(unused_must_use)]
+//#![allow(warnings)]
+
+#[macro_use]
+extern crate serde_derive;
+
+mod secrets;
+
+//use futures::TryStreamExt;
+use egg_mode::stream::StreamMessage;
+use crate::secrets::Secrets;
+
+pub type AnyhowResult<T> = anyhow::Result<T>;
+
+const ENV_PUBLISH_TOPIC : &'static str = "PUBLISH_TOPIC";
+const ENV_PUBLISH_TOPIC_DEFAULT : &'static str = "firehose";
+
+const ENV_REDIS_MAX_RETRY_COUNT : &'static str = "REDIS_MAX_RETRY_COUNT";
+const ENV_REDIS_MAX_RETRY_COUNT_DEFAULT : u32 = 3;
+
+/// Monitor Twitter for interesting events against our account.
+/// With this implementation, we can capture: Mentions, Retweets, Replies.
+/// We cannot capture: Follows.
+#[tokio::main]
+async fn main() -> anyhow::Result<()>
+{
+  easyenv::init_env_logger(None);
+
+  let redis_publish_topic = easyenv::get_env_string_or_default(
+    ENV_PUBLISH_TOPIC, ENV_PUBLISH_TOPIC_DEFAULT);
+
+  let redis_max_retry_count = easyenv::get_env_num(
+    ENV_REDIS_MAX_RETRY_COUNT,
+    ENV_REDIS_MAX_RETRY_COUNT_DEFAULT)?;
+
+  let secrets = Secrets::from_file("secrets.toml")?;
+
+  secrets.twitter.request_access_token().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,102 @@
+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)
+}
+
+pub fn retweet_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_retweet = {
+    let mut twitter_retweet = protos::IngestionTwitterRetweet::default();
+    twitter_retweet.original_text = tweet_details.retweeted_text.clone();
+
+    debug!("Twitter Retweet Proto: {:?}", twitter_retweet);
+
+    binary_encode_proto(twitter_retweet)
+  }?;
+
+  payload_proto.ingestion_payload_data = Some(binary_twitter_retweet);
+
+  Ok(payload_proto)
+}

src/secrets.rs

@@ -0,0 +1,155 @@
+use crate::AnyhowResult;
+use std::io::Read;
+use log::{info, warn, debug};
+use anyhow::anyhow;
+
+#[derive(Deserialize)]
+pub struct Secrets {
+  pub redis: RedisSecrets,
+  pub twitter: TwitterSecrets,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct RedisSecrets {
+  pub username: String,
+  pub password: String,
+  pub host: String,
+  pub port: u32,
+  pub uses_tls: bool,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct TwitterSecrets {
+  pub api_key: String,
+  pub api_secret_key: String,
+
+  // These have to be obtained over an OAuth-like flow
+  pub access_key: Option<String>,
+  pub access_secret: Option<String>,
+}
+
+impl Secrets {
+  pub fn from_file(filename: &str) -> AnyhowResult<Self> {
+    let mut file = std::fs::File::open(filename)?;
+    let mut buffer = String::new();
+    file.read_to_string(&mut buffer)?;
+    let secrets = toml::from_str(&buffer)?;
+    Ok(secrets)
+  }
+}
+
+impl RedisSecrets {
+  pub fn new(username: &str, password: &str, host: &str, port: u32, uses_tls: bool) -> Self {
+    Self {
+      username: username.to_string(),
+      password: password.to_string(),
+      host: host.to_string(),
+      port,
+      uses_tls,
+    }
+  }
+
+  pub fn connection_url(&self) -> String {
+    let protocol = if self.uses_tls { "rediss" } else { "redis" };
+    let mut auth = "".to_string();
+
+    if !self.username.is_empty() {
+      if !self.password.is_empty() {
+        auth = format!("{}:{}@", self.username, self.password);
+      } else {
+        auth = format!("{}@", self.username);
+      }
+    } else if !self.password.is_empty() {
+      auth = format!("default:{}@", self.password);
+    }
+
+    format!("{}://{}{}:{}", protocol, auth, self.host, self.port)
+  }
+}
+
+impl TwitterSecrets {
+  pub fn new(api_key: &str,
+             api_secret_key: &str,
+             access_key: &Option<String>,
+             access_secret: &Option<String>) -> Self {
+
+    Self {
+      api_key: api_key.to_string(),
+      api_secret_key: api_secret_key.to_string(),
+      access_key: access_key.clone(),
+      access_secret: access_secret.clone(),
+    }
+  }
+
+  /// This assumes we have a valid pair of access (key, secret).
+  pub async fn verify_access_token(&self) -> AnyhowResult<egg_mode::Token> {
+    let consumer_keypair = self.make_consumer_keypair();
+    let access_keypair = self.make_access_keypair()?;
+
+    let access_token = egg_mode::Token::Access {
+      consumer: consumer_keypair,
+      access: access_keypair,
+    };
+
+    egg_mode::auth::verify_tokens(&access_token).await?;
+
+    Ok(access_token)
+  }
+
+  /// Call this to get an access token.
+  /// This is an interactive flow that requests input from stdin.
+  pub async fn request_access_token(&self) -> AnyhowResult<()> {
+    let consumer_keypair = self.make_consumer_keypair();
+    let request_token = egg_mode::auth::request_token(&consumer_keypair, "oob")
+      .await?;
+
+    warn!("Go to the following URL, sign in, and paste the PIN that comes back:");
+    warn!("{}", egg_mode::auth::authorize_url(&request_token));
+
+    let mut pin = String::new();
+    std::io::stdin().read_line(&mut pin)?;
+
+    let token_result = egg_mode::auth::access_token(consumer_keypair, &request_token, pin)
+      .await?;
+
+    let access_token = token_result.0;
+    let user_id = token_result.1;
+    let username = token_result.2;
+
+    warn!("Ordinarily these should not be logged, but there is no log aggregator installed.");
+    warn!("In any event, it is best to do this locally before creating k8s secrets.");
+    warn!("user_id = {:?}", user_id);
+    warn!("username = {:?}", username);
+
+    match access_token {
+      egg_mode::Token::Access {
+        access: ref access,
+        ..
+      } => {
+        warn!("access_key = {:?}", access.key);
+        warn!("access_secret = {:?}", access.secret);
+      }
+      _ => unreachable!(),
+    }
+
+    Ok(())
+  }
+
+  pub fn make_consumer_keypair(&self) -> egg_mode::KeyPair {
+    egg_mode::KeyPair::new(self.api_key.to_string(), self.api_secret_key.to_string())
+  }
+
+  pub fn make_access_keypair(&self) -> AnyhowResult<egg_mode::KeyPair> {
+    let access_key = self.access_key
+      .as_ref()
+      .ok_or(anyhow!("No access key!"))?
+      .clone();
+
+    let access_secret = self.access_secret
+      .as_ref()
+      .ok_or(anyhow!("No access secret!"))?
+      .clone();
+
+    Ok(egg_mode::KeyPair::new(access_key, access_secret))
+  }
+}

src/util.rs

@@ -0,0 +1,184 @@
+
+pub struct Config {
+  pub token: egg_mode::Token,
+  pub user_id: u64,
+  pub screen_name: String,
+}
+
+impl Config {
+  pub async fn load() -> Self {
+    let a1 = Config::load_inner().await;
+    if let Some(conf) = a1 {
+      return conf;
+    }
+
+    Config::load_inner().await.unwrap()
+  }
+
+  /// This needs to be a separate function so we can retry after creating the
+  /// twitter_settings file. Idealy we would recurse, but that requires boxing
+  /// the output which doesn't seem worthwhile
+  async fn load_inner() -> Option<Self> {
+    //IMPORTANT: make an app for yourself at apps.twitter.com and get your
+    //key/secret into these files; these examples won't work without them
+    let consumer_key = include_str!("consumer_key").trim();
+    let consumer_secret = include_str!("consumer_secret").trim();
+
+    let con_token = egg_mode::KeyPair::new(consumer_key, consumer_secret);
+
+    let mut config = String::new();
+    let user_id: u64;
+    let username: String;
+    let token: egg_mode::Token;
+
+    //look at all this unwrapping! who told you it was my birthday?
+    if let Ok(mut f) = std::fs::File::open("twitter_settings") {
+      f.read_to_string(&mut config).unwrap();
+
+      let mut iter = config.split('\n');
+
+      username = iter.next().unwrap().to_string();
+      user_id = u64::from_str_radix(&iter.next().unwrap(), 10).unwrap();
+      let access_token = egg_mode::KeyPair::new(
+        iter.next().unwrap().to_string(),
+        iter.next().unwrap().to_string(),
+      );
+      token = egg_mode::Token::Access {
+        consumer: con_token,
+        access: access_token,
+      };
+
+      if let Err(err) = egg_mode::auth::verify_tokens(&token).await {
+        println!("We've hit an error using your old tokens: {:?}", err);
+        println!("We'll have to reauthenticate before continuing.");
+        std::fs::remove_file("twitter_settings").unwrap();
+      } else {
+        println!("Welcome back, {}!\n", username);
+      }
+    } else {
+      let request_token = egg_mode::auth::request_token(&con_token, "oob").await.unwrap();
+
+      println!("Go to the following URL, sign in, and give me the PIN that comes back:");
+      println!("{}", egg_mode::auth::authorize_url(&request_token));
+
+      let mut pin = String::new();
+      std::io::stdin().read_line(&mut pin).unwrap();
+      println!("");
+
+      let tok_result = egg_mode::auth::access_token(con_token, &request_token, pin)
+        .await
+        .unwrap();
+
+      token = tok_result.0;
+      user_id = tok_result.1;
+      username = tok_result.2;
+
+      match token {
+        egg_mode::Token::Access {
+          access: ref access_token,
+          ..
+        } => {
+          config.push_str(&username);
+          config.push('\n');
+          config.push_str(&format!("{}", user_id));
+          config.push('\n');
+          config.push_str(&access_token.key);
+          config.push('\n');
+          config.push_str(&access_token.secret);
+        }
+        _ => unreachable!(),
+      }
+
+      let mut f = std::fs::File::create("twitter_settings").unwrap();
+      f.write_all(config.as_bytes()).unwrap();
+
+      println!("Welcome, {}, let's get this show on the road!", username);
+    }
+
+    //TODO: Is there a better way to query whether a file exists?
+    if std::fs::metadata("twitter_settings").is_ok() {
+      Some(Config {
+        token: token,
+        user_id: user_id,
+        screen_name: username,
+      })
+    } else {
+      None
+    }
+  }
+}
+
+
+pub fn print_tweet(tweet: &egg_mode::tweet::Tweet) {
+  if let Some(ref user) = tweet.user {
+    println!(
+      "{} (@{}) posted at {}",
+      Paint::blue(&user.name),
+      Paint::bold(Paint::blue(&user.screen_name)),
+      tweet.created_at.with_timezone(&chrono::Local)
+    );
+  }
+
+  if let Some(ref screen_name) = tweet.in_reply_to_screen_name {
+    println!("➜ in reply to @{}", Paint::blue(screen_name));
+  }
+
+  if let Some(ref status) = tweet.retweeted_status {
+    println!("{}", Paint::red("Retweet ➜"));
+    print_tweet(status);
+    return;
+  } else {
+    println!("{}", Paint::green(&tweet.text));
+  }
+
+  if let Some(source) = &tweet.source {
+    println!("➜ via {} ({})", source.name, source.url);
+  }
+
+  if let Some(ref place) = tweet.place {
+    println!("➜ from: {}", place.full_name);
+  }
+
+  if let Some(ref status) = tweet.quoted_status {
+    println!("{}", Paint::red("➜ Quoting the following status:"));
+    print_tweet(status);
+  }
+
+  if !tweet.entities.hashtags.is_empty() {
+    println!("➜ Hashtags contained in the tweet:");
+    for tag in &tweet.entities.hashtags {
+      println!("  {}", tag.text);
+    }
+  }
+
+  if !tweet.entities.symbols.is_empty() {
+    println!("➜ Symbols contained in the tweet:");
+    for tag in &tweet.entities.symbols {
+      println!("  {}", tag.text);
+    }
+  }
+
+  if !tweet.entities.urls.is_empty() {
+    println!("➜ URLs contained in the tweet:");
+    for url in &tweet.entities.urls {
+      if let Some(expanded_url) = &url.expanded_url {
+        println!("  {}", expanded_url);
+      }
+    }
+  }
+
+  if !tweet.entities.user_mentions.is_empty() {
+    println!("➜ Users mentioned in the tweet:");
+    for user in &tweet.entities.user_mentions {
+      println!("  {}", Paint::bold(Paint::blue(&user.screen_name)));
+    }
+  }
+
+  if let Some(ref media) = tweet.extended_entities {
+    println!("➜ Media attached to the tweet:");
+    for info in &media.media {
+      println!("  A {:?}", info.media_type);
+    }
+  }
+}
+