pubsub-enricher: initial commit / app skeleton

.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,29 @@
+[package]
+name = "pubsub-enricher"
+version = "0.0.0"
+edition = "2018"
+authors = [
+  "Brandon Thomas <bt@brand.io>",
+  "Brandon Thomas <echelon@gmail.com>"
+]
+
+[[bin]]
+name = "pubsub-enricher"
+path = "src/main.rs"
+
+[dependencies]
+anyhow = "1.0.38"
+easyenv = "0.0.4"
+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"] }
+
+[build-dependencies]
+anyhow = "1.0.38"
+prost-build = "0.7.0"

README.md

@@ -0,0 +1,12 @@
+pubsub-enricher
+===============
+Listen to pubsub firehose, enrich events, then republish.
+
+Protos
+------
+
+Protos are defined in a git submodule:
+
+```
+git submodule add git@github.com:storytold/storyteller-protos.git
+```

build.rs

@@ -0,0 +1,12 @@
+extern crate anyhow; // TODO: No!
+extern crate prost_build;
+
+fn main() -> anyhow::Result<()> {
+  prost_build::compile_protos(
+    &[
+      // TODO: This needs to respect glob
+      "protos/storyteller-protos/protos/twitch-gateway/twitch-gateway.proto"
+    ],
+    &["protos/"])?;
+  Ok(())
+}

src/main.rs

@@ -0,0 +1,53 @@
+#[macro_use]
+extern crate serde_derive;
+
+use anyhow::{Context, Error};
+use anyhow::anyhow;
+
+use redis::aio::Connection;
+use redis::{AsyncCommands, RedisResult};
+
+pub type AnyhowResult<T> = anyhow::Result<T>;
+
+use std::thread;
+use std::time::Duration;
+
+mod protos;
+mod redis_client;
+mod secrets;
+
+use crate::secrets::Secrets;
+use crate::redis_client::RedisClient;
+
+const ENV_SUBSCRIBE_TOPIC : &'static str = "SUBSCRIBE_TOPIC";
+const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "twitch";
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+  easyenv::init_env_logger(None);
+
+  let redis_publish_topic = easyenv::get_env_string_or_default(
+    ENV_SUBSCRIBE_TOPIC, ENV_SUBSCRIBE_TOPIC_DEFAULT);
+
+  let secrets = Secrets::from_file("secrets.toml")?;
+
+  let mut redis_client = RedisClient::new(&secrets.redis);
+  redis_client.connect().await?;
+
+  loop {
+    /*match twitch_client.main_loop().await {
+      Ok(_) => {
+        println!("Early exit? Restarting...");
+        thread::sleep(Duration::from_secs(5));
+      },
+      Err(e) => {
+        println!("There was an error: {:?}", e);
+        thread::sleep(Duration::from_secs(5));
+        println!("Restarting client...");
+      }
+    }*/
+    thread::sleep(Duration::from_secs(5));
+  }
+
+  Ok(())
+}

src/protos.rs

@@ -0,0 +1,4 @@
+// Include the protos
+pub mod protos {
+  include!(concat!(env!("OUT_DIR"), "/storyteller.twitchgateway.rs"));
+}

src/redis_client.rs

@@ -0,0 +1,102 @@
+use crate::AnyhowResult;
+use crate::secrets::RedisSecrets;
+
+use anyhow::anyhow;
+use std::thread;
+use std::time::Duration;
+use redis::AsyncCommands;
+use redis::aio::Connection;
+use redis;
+
+pub struct RedisClient {
+  secrets: RedisSecrets,
+
+  /// This manages our connection.
+  /// It's meant to be absent when not connected.
+  /// TODO: Probably need to reset this when we disconnect.
+  connection: Option<Connection>,
+
+  connection_failure_count: u32,
+}
+
+impl RedisClient {
+  pub fn new(secrets: &RedisSecrets) -> Self {
+    Self {
+      secrets: secrets.clone(),
+      connection: None,
+      connection_failure_count: 0,
+    }
+  }
+
+  pub async fn connect(&mut self) -> AnyhowResult<()> {
+    let client = redis::Client::open(self.secrets.connection_url())?;
+    let mut connection = client.get_async_connection().await?;
+
+    self.connection = Some(connection);
+    self.connection_failure_count = 0;
+
+    Ok(())
+  }
+
+  pub async fn failure_notify_maybe_reconnect(&mut self) -> AnyhowResult<()> {
+    self.connection_failure_count += 1;
+
+    if self.connection_failure_count > 5 {
+      println!("Attempting Redis Reconnect in 3 sec...");
+      thread::sleep(Duration::from_secs(3));
+      self.connect().await?;
+      self.connection_failure_count = 0;
+    } else {
+      println!("Not ready to reconnect. Fail Count = {}", self.connection_failure_count);
+    }
+    Ok(())
+  }
+
+  pub async fn publish(&mut self, channel: &str, message: &str) -> AnyhowResult<u32> {
+    let connection = match &mut self.connection {
+      None => {
+        return Err(anyhow!("Not connected"));
+      },
+      Some(connection) => connection,
+    };
+
+    // TODO: Smart reconnect
+    /*let result = connection.publish(channel, message).await?;
+    let result = match result {
+      Ok(r) => r,
+      Err(err) => {
+        match err.kind() {
+          IoError |  TryAgain | ClusterDown | MasterDown => {},
+          _ => return Err(anyhow!("Redis error: {:?}", err)),
+        }
+      }
+    };*/
+
+    let result = connection.publish(channel, message).await?;
+    Ok(result)
+  }
+
+  pub async fn publish_bytes(&mut self, channel: &str, message: &[u8]) -> AnyhowResult<u32> {
+    let connection = match &mut self.connection {
+      None => {
+        return Err(anyhow!("Not connected"));
+      },
+      Some(connection) => connection,
+    };
+
+    // TODO: Smart reconnect
+    /*let result = connection.publish(channel, message).await?;
+    let result = match result {
+      Ok(r) => r,
+      Err(err) => {
+        match err.kind() {
+          IoError |  TryAgain | ClusterDown | MasterDown => {},
+          _ => return Err(anyhow!("Redis error: {:?}", err)),
+        }
+      }
+    };*/
+
+    let result = connection.publish(channel, message).await?;
+    Ok(result)
+  }
+}

src/secrets.rs

@@ -0,0 +1,64 @@
+use crate::AnyhowResult;
+use std::io::Read;
+
+#[derive(Deserialize)]
+pub struct Secrets {
+  pub redis: RedisSecrets,
+  pub twitch: TwitchSecrets,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct RedisSecrets {
+  pub username: String,
+  pub password: String,
+  pub host: String,
+  pub port: u32,
+}
+
+#[derive(Deserialize, Clone)]
+pub struct TwitchSecrets {
+  /// Twitch username
+  pub username: String,
+
+  /// From the creator dashboard of Twitch
+  pub stream_key: String,
+
+  /// Identifies our application uniquely; cannot be reassigned.
+  pub application_client_id: String,
+
+  /// Application secret (generated)
+  pub application_client_secret: String,
+
+  /// OAuth access token generated with URL handshake
+  /// NB: This changes frequently and must be regenerated.
+  /// Go to this URL:
+  ///
+  /// https://id.twitch.tv/oauth2/authorize?response_type=token
+  ///   &client_id=___APPLICATION_CLIENT_ID___
+  ///   &redirect_uri=http://localhost&scope=viewing_activity_read%20user_read%20channel_read%20chat:read
+  ///   &state=randomUUID
+  ///
+  /// Which will redirect to 'localhost' per the redirect_uri.
+  /// The oauth token will be "access_token" prepended with "oauth:" (which isn't included)
+  pub oauth_access_token: String,
+
+  /// Channels to join and monitor.
+  pub watch_channels: Vec<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 connection_url(&self) -> String {
+    format!("rediss://{}:{}@{}:{}", self.username, self.password, self.host, self.port)
+  }
+}
+