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

.env-template

@@ -0,0 +1,11 @@
+# Redis connection
+REDIS_USERNAME=
+REDIS_PASSWORD=
+REDIS_HOST=
+REDIS_PORT=
+REDIS_USES_TLS=
+
+# Redis pubsub
+#SUBSCRIBE_TOPIC=firehose
+#PUBLISH_TOPIC=unreal
+#REDIS_MAX_RETRY_COUNT=3

.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: pubsub-enricher
+
+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,37 @@
+[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.5"
+futures = "0.3.14"
+lazy_static = "1.4.0"
+log = "0.4.14"
+prost = "0.7.0"
+regex = "1.4.6"
+rustls = "0.18"
+serde = "1.0.123"
+serde_derive = "1.0.123"
+tokio = { version = "0.2", features = ["full"] }
+toml = "0.5.8"
+
+google_maps = "1.0.3"
+
+redis = { version = "0.17.0", features = ["tls", "async-std-tls-comp", "tokio-tls-comp", "connection-manager"] }
+
+[build-dependencies]
+anyhow = "1.0.38"
+prost-build = "0.7.0"
+
+[dev-dependencies]
+expectest = "0.12.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
+
+# 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/pubsub-enricher'
+LABEL org.opencontainers.image.documentation='https://github.com/storytold/pubsub-enricher'
+LABEL org.opencontainers.image.source='https://github.com/storytold/pubsub-enricher'
+
+# 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/pubsub-enricher /
+COPY --from=build /tmp/target/release/pubsub-enricher /
+
+# 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 pubsub-enricher
+
+EXPOSE 8080
+CMD LD_LIBRARY_PATH=/usr/lib /pubsub-enricher
+

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,18 @@
+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/storyteller-protos/protos/unreal/unreal.proto",
+  ],
+    &["protos/"])?;
+  Ok(())
+}

src/clients/mod.rs

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

src/clients/redis_client.rs

@@ -0,0 +1,68 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use crate::protos::protos;
+use crate::secrets::RedisSecrets;
+use log::{info, warn, debug};
+use prost::Message;
+use redis::aio::{Connection, PubSub, ConnectionManager};
+use redis::{AsyncCommands, Msg, PubSubCommands};
+use redis;
+use std::thread;
+use std::time::Duration;
+use tokio::stream::{Stream, StreamExt};
+
+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/redis_subscriber.rs

@@ -0,0 +1,101 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use crate::dispatcher::Dispatcher;
+use crate::protos::protos;
+use crate::secrets::RedisSecrets;
+use log::{info, warn};
+use prost::Message;
+use redis::aio::PubSub;
+use std::thread;
+use std::time::Duration;
+use tokio::stream::{Stream, StreamExt};
+
+pub struct RedisSubscribeClient {
+  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<PubSub>,
+
+  connection_failure_count: u32,
+
+  /// Event dispatcher that handles routing and passing to handlers.
+  dispatcher: Dispatcher,
+}
+
+impl RedisSubscribeClient {
+  pub fn new(secrets: &RedisSecrets, dispatcher: Dispatcher) -> Self {
+    Self {
+      secrets: secrets.clone(),
+      connection: None,
+      connection_failure_count: 0,
+      dispatcher,
+    }
+  }
+
+  pub async fn connect(&mut self) -> AnyhowResult<()> {
+    let client = redis::Client::open(self.secrets.connection_url())?;
+    let mut connection = client.get_async_connection().await?.into_pubsub();
+
+    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 subscribe(&mut self, topic: &str) -> AnyhowResult<()> {
+    let mut connection = match &mut self.connection {
+      None => {
+        return Err(anyhow!("Not connected"));
+      },
+      Some(connection) => connection,
+    };
+
+    connection.subscribe(topic).await?;
+    Ok(())
+  }
+
+  pub async fn start_stream(&mut self) -> AnyhowResult<()> {
+    let connection = match &mut self.connection {
+      None => {
+        return Err(anyhow!("Not connected"));
+      },
+      Some(connection) => connection,
+    };
+
+    loop {
+      for message in connection.on_message().next().await {
+        let bytes = message.get_payload_bytes();
+
+        let mut message_proto = match protos::PubsubEventPayloadV1::decode(bytes) {
+          Ok(m) => m,
+          Err(e) => {
+            warn!("Error decoding proto: {:?}", e);
+            continue;
+          },
+        };
+
+        self.dispatcher.handle_pubsub_event(message_proto);
+      }
+
+      thread::sleep(Duration::from_millis(250));
+    }
+
+    Ok(())
+  }
+}

src/dispatcher.rs

@@ -0,0 +1,76 @@
+use crate::protos::protos;
+use lazy_static::lazy_static;
+use anyhow::anyhow;
+use log::{info, warn, debug};
+use regex::Regex;
+use std::collections::HashMap;
+
+use crate::AnyhowResult;
+use crate::handlers::coordinate_and_geocode_handler::CoordinateAndGeocodeHandler;
+use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand;
+use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource};
+
+pub trait TextCommandHandler {
+  fn handle_text_command(&self,
+                         command: &FirstPassParsedCommand,
+                         event: &InboundEvent,
+                         event_source: &InboundEventSource) -> AnyhowResult<()>;
+}
+
+pub struct Dispatcher {
+  text_command_handlers: HashMap<String, Box<dyn TextCommandHandler>>,
+}
+
+impl Dispatcher {
+  pub fn new() -> Self {
+    Self {
+      text_command_handlers: HashMap::new(),
+    }
+  }
+
+  pub fn add_text_command_handler(&mut self, command: &str, handler: Box<dyn TextCommandHandler>) {
+    self.text_command_handlers.insert(command.to_string(), handler);
+  }
+
+  pub fn handle_pubsub_event(&self, message: protos::PubsubEventPayloadV1) -> AnyhowResult<()> {
+    info!("Handling Proto: {:?}", message);
+
+    let event_source = InboundEventSource::parse_from_payload(&message)?;
+    let event = InboundEvent::parse_from_payload(&message)?;
+
+    info!("Source: {:?}", event_source);
+    info!("Event: {:?}", event);
+
+    match event {
+      InboundEvent::TwitchMessage(ref message) => {
+        let command = FirstPassParsedCommand::try_parse(&message.message_contents())?;
+        self.handle_text_command(&command, &event, &event_source)?;
+      }
+    };
+
+    // TODO... other event types, including non-text commands.
+
+    Ok(())
+  }
+
+  /// Handle text-based commands.
+  fn handle_text_command(&self,
+                         command: &FirstPassParsedCommand,
+                         event: &InboundEvent,
+                         event_source: &InboundEventSource)
+    -> AnyhowResult<()>
+  {
+    info!("Handle command: {}", &command.command);
+
+    let handler = match self.text_command_handlers.get(&command.command) {
+      Some(h) => h,
+      None => {
+        debug!("No handler for command: {}", &command.command);
+        return Ok(());
+      }
+    };
+
+    info!("Dispatching handler...");
+    handler.handle_text_command(command, event, event_source)
+  }
+}

src/handlers/coordinate_and_geocode_handler.rs

@@ -0,0 +1,206 @@
+use crate::AnyhowResult;
+use crate::clients::redis_client::RedisClient;
+use crate::dispatcher::TextCommandHandler;
+use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand;
+use futures::executor::block_on;
+use lazy_static::lazy_static;
+use log::{info, warn, debug};
+use prost::Message;
+use regex::Regex;
+use std::sync::{RwLock, Arc, Mutex};
+use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource};
+use crate::protos::populate_source::populate_source;
+use crate::protos::protos;
+use crate::protos::binary_encode_proto::binary_encode_proto;
+
+// TODO: maybe separate text command handling and data source concerns
+//  like this? Though maybe it's too early to optimize this.
+pub struct CoordinateAndGeocodeHandler {
+  redis_client: Arc<Mutex<RedisClient>>,
+}
+
+impl CoordinateAndGeocodeHandler {
+  pub fn new(redis_client: Arc<Mutex<RedisClient>>) -> Self {
+    Self {
+      redis_client,
+    }
+  }
+
+  fn handle_lat_long(&self,
+                     lat_long: LatLong,
+                     event: &InboundEvent,
+                     event_source: &InboundEventSource) -> AnyhowResult<()>
+  {
+    let mut unreal_proto = protos::UnrealEventPayloadV1::default();
+
+    // Cesium
+    let mut cesium_proto = protos::CesiumWarpRequest::default();
+    cesium_proto.latitude = lat_long.latitude;
+    cesium_proto.longitude = lat_long.longitude;
+
+    debug!("Payload Proto: {:?}", cesium_proto);
+
+    unreal_proto.payload_type = protos::unreal_event_payload_v1::PayloadType::CesiumWarp as i32;
+    unreal_proto.payload_data = binary_encode_proto(cesium_proto)?;
+    unreal_proto.debug_message = "Hello from Rust!".to_string();
+
+    populate_source(&mut unreal_proto, event_source)?;
+
+    let final_binary = binary_encode_proto(unreal_proto)?;
+
+    match self.redis_client.lock() {
+      Ok(mut redis_client) => {
+        debug!("Publishing to Redis...");
+        let future = redis_client.publish_bytes("unreal", &final_binary);
+        block_on(future);
+        debug!("Published to redis.");
+      },
+      Err(_) => {},
+    }
+
+    Ok(())
+  }
+}
+
+impl TextCommandHandler for CoordinateAndGeocodeHandler {
+  fn handle_text_command(&self,
+                         command: &FirstPassParsedCommand,
+                         event: &InboundEvent,
+                         event_source: &InboundEventSource)
+    -> AnyhowResult<()>
+  {
+    let maybe_lat_long = parse_lat_long(&command.unparsed_arguments);
+
+    if let Some(lat_long) = maybe_lat_long {
+      self.handle_lat_long(lat_long, event, event_source);
+    }
+
+    // TODO: Geocoding.
+
+    Ok(())
+  }
+}
+
+#[derive(Clone,Debug,PartialEq)]
+pub struct LatLong {
+  pub latitude: f64,
+  pub longitude: f64,
+}
+
+/// Command payload contains the following example instruction:
+///   * "34.0659° N, 84.6769° W"
+///   * "33.753746, -84.386330"
+pub fn parse_lat_long(input: &str) -> Option<LatLong> {
+  const NUMERAL : &'static str = r"(\-?\d+\.?\d*)\s*°?\s*([NSEW])?";
+  lazy_static! {
+    static ref COORDINATE_REGEX: Regex =
+      Regex::new(&format!(r"^({})\s*,?\s*({})$", NUMERAL, NUMERAL)).expect("Regex should work");
+  }
+
+  let captures = match COORDINATE_REGEX.captures(&input) {
+    None => return None,
+    Some(caps) => caps,
+  };
+
+  let maybe_lat = captures.get(2)
+    .map(|m| m.as_str())
+    .and_then(|s| s.parse::<f64>().ok());
+
+  let maybe_lat_dir = captures.get(3)
+    .map(|m| m.as_str());
+
+  let maybe_long = captures.get(5)
+    .map(|m| m.as_str())
+    .and_then(|s| s.parse::<f64>().ok());
+
+  let maybe_long_dir = captures.get(6)
+    .map(|m| m.as_str());
+
+  let mut latitude = match maybe_lat {
+    None => return None,
+    Some(lat) => lat,
+  };
+
+  let mut longitude = match maybe_long {
+    None => return None,
+    Some(long) => long,
+  };
+
+  match maybe_lat_dir {
+    Some("S") => latitude *= -1.0,
+    _ => {},
+  }
+
+  match maybe_long_dir {
+    Some("W") => longitude *= -1.0,
+    _ => {},
+  }
+
+  Some(LatLong {
+    latitude,
+    longitude
+  })
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use expectest::prelude::*;
+
+  #[test]
+  fn parse_lat_long_invalid() {
+    expect!(parse_lat_long("foo bar baz")).to(be_none());
+    expect!(parse_lat_long("1 2 3")).to(be_none());
+  }
+
+  #[test]
+  fn parse_lat_long_raw_coordinates() {
+    expect!(parse_lat_long("33.753746, -84.386330")).to(
+      be_eq(Some(LatLong {
+        latitude: 33.753746,
+        longitude: -84.386330,
+      })));
+
+    expect!(parse_lat_long("-123, -456")).to(
+      be_eq(Some(LatLong {
+        latitude: -123.0,
+        longitude: -456.0,
+      })));
+
+    expect!(parse_lat_long("-99,100")).to(
+      be_eq(Some(LatLong {
+        latitude: -99.0,
+        longitude: 100.0,
+      })));
+  }
+
+  #[test]
+  fn parse_lat_long_directions() {
+    expect!(parse_lat_long("34.0659 N, 84.6769 W")).to(
+      be_eq(Some(LatLong {
+        latitude: 34.0659,
+        longitude: -84.6769,
+      })));
+
+    expect!(parse_lat_long("123 S, 456 E")).to(
+      be_eq(Some(LatLong {
+        latitude: -123.0,
+        longitude: 456.0,
+      })));
+  }
+
+  #[test]
+  fn parse_lat_long_directions_degrees() {
+    expect!(parse_lat_long("34.0659°  N, 84.6769° W")).to(
+      be_eq(Some(LatLong {
+        latitude: 34.0659,
+        longitude: -84.6769,
+      })));
+
+    expect!(parse_lat_long("123°S, 777° E")).to(
+      be_eq(Some(LatLong {
+        latitude: -123.0,
+        longitude: 777.0,
+      })));
+  }
+}

src/handlers/mod.rs

@@ -0,0 +1,3 @@
+pub mod coordinate_and_geocode_handler;
+pub mod spawn_handler;
+pub mod vocode_handler;

src/handlers/spawn_handler.rs

@@ -0,0 +1,141 @@
+use crate::AnyhowResult;
+use crate::clients::redis_client::RedisClient;
+use crate::dispatcher::TextCommandHandler;
+use crate::protos::binary_encode_proto::binary_encode_proto;
+use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource};
+use crate::protos::protos;
+use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand;
+use futures::executor::block_on;
+use lazy_static::lazy_static;
+use log::{info, warn, debug};
+use prost::Message;
+use regex::Regex;
+use std::sync::{RwLock, Arc, Mutex};
+use crate::protos::populate_source::populate_source;
+
+// TODO: maybe separate text command handling and data source concerns
+//  like this? Though maybe it's too early to optimize this.
+pub struct SpawnHandler {
+  redis_client: Arc<Mutex<RedisClient>>,
+}
+
+impl SpawnHandler {
+  pub fn new(redis_client: Arc<Mutex<RedisClient>>) -> Self {
+    Self {
+      redis_client,
+    }
+  }
+
+  fn handle(&self,
+            spawn: SpawnCommand,
+            event: &InboundEvent,
+            event_source: &InboundEventSource)
+            -> AnyhowResult<()> {
+
+    let mut unreal_proto = protos::UnrealEventPayloadV1::default();
+
+    // Payload
+    let mut spawn_proto = protos::SpawnCreatureRequest::default();
+    spawn_proto.name_slug = spawn.spawn_slug;
+
+    debug!("Payload Proto: {:?}", spawn_proto);
+
+    unreal_proto.payload_type = protos::unreal_event_payload_v1::PayloadType::CesiumWarp as i32;
+    unreal_proto.payload_data = binary_encode_proto(spawn_proto)?;
+    unreal_proto.debug_message = "Hello from Rust!".to_string();
+
+    populate_source(&mut unreal_proto, event_source)?;
+
+    let final_binary = binary_encode_proto(unreal_proto)?;
+
+    match self.redis_client.lock() {
+      Ok(mut redis_client) => {
+        debug!("Publishing to Redis...");
+        let future = redis_client.publish_bytes("unreal", &final_binary);
+        block_on(future);
+        debug!("Published to redis");
+      },
+      Err(_) => {},
+    }
+
+    Ok(())
+  }
+}
+
+impl TextCommandHandler for SpawnHandler {
+  fn handle_text_command(&self,
+                         command: &FirstPassParsedCommand,
+                         event: &InboundEvent,
+                         event_source: &InboundEventSource)
+                         -> AnyhowResult<()>
+  {
+    let maybe_spawn_command = parse_spawn(&command.unparsed_arguments);
+
+    debug!("Maybe spawn command: {:?}", maybe_spawn_command);
+
+    if let Some(spawn) = maybe_spawn_command {
+      self.handle(spawn, event, event_source);
+    }
+    Ok(())
+  }
+}
+
+#[derive(Clone,Debug,PartialEq)]
+pub struct SpawnCommand {
+  pub spawn_slug: String,
+  // TODO: Args in the future.
+}
+
+pub fn parse_spawn(input: &str) -> Option<SpawnCommand> {
+  lazy_static! {
+    static ref REGEX: Regex =
+      Regex::new(r"^\s*([\w\-]+)\s*$").expect("Regex should work");
+  }
+
+  let captures = match REGEX.captures(&input) {
+    None => return None,
+    Some(caps) => caps,
+  };
+
+  let maybe_capture = captures.get(1)
+    .map(|m| m.as_str());
+
+
+  let capture = match maybe_capture {
+    None => return None,
+    Some(cap) => cap,
+  };
+
+  Some(SpawnCommand {
+    spawn_slug: capture.to_string(),
+  })
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use expectest::prelude::*;
+
+  #[test]
+  fn failure_cases() {
+    expect!(parse_spawn("too many words")).to(be_none());
+    expect!(parse_spawn("")).to(be_none());
+    expect!(parse_spawn("     ")).to(be_none());
+  }
+
+  #[test]
+  fn single_word() {
+    expect!(parse_spawn("goomba")).to(
+      be_eq(Some(SpawnCommand {
+        spawn_slug: "goomba".to_string(),
+      })));
+  }
+
+  #[test]
+  fn hyphenated_word() {
+    expect!(parse_spawn("like-like")).to(
+      be_eq(Some(SpawnCommand {
+        spawn_slug: "like-like".to_string(),
+      })));
+  }
+}

src/handlers/vocode_handler.rs

@@ -0,0 +1,152 @@
+use crate::AnyhowResult;
+use crate::clients::redis_client::RedisClient;
+use crate::dispatcher::TextCommandHandler;
+use crate::protos::binary_encode_proto::binary_encode_proto;
+use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource};
+use crate::protos::protos;
+use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand;
+use futures::executor::block_on;
+use lazy_static::lazy_static;
+use log::{info, warn, debug};
+use prost::Message;
+use regex::Regex;
+use std::sync::{RwLock, Arc, Mutex};
+use crate::protos::populate_source::populate_source;
+
+// TODO: maybe separate text command handling and data source concerns
+//  like this? Though maybe it's too early to optimize this.
+pub struct TtsHandler {
+  redis_client: Arc<Mutex<RedisClient>>,
+}
+
+impl TtsHandler {
+  pub fn new(redis_client: Arc<Mutex<RedisClient>>) -> Self {
+    Self {
+      redis_client,
+    }
+  }
+
+  fn handle(&self,
+            tts: TtsCommand,
+            event: &InboundEvent,
+            event_source: &InboundEventSource)
+            -> AnyhowResult<()> {
+
+    let mut unreal_proto = protos::UnrealEventPayloadV1::default();
+
+    // Payload
+    let mut tts_proto = protos::VocodesTtsRequest::default();
+    tts_proto.voice_slug = tts.speaker_slug;
+    tts_proto.text = tts.utterance;
+
+    debug!("Payload Proto: {:?}", tts_proto);
+
+    unreal_proto.payload_type = protos::unreal_event_payload_v1::PayloadType::VocodesTts as i32;
+    unreal_proto.payload_data = binary_encode_proto(tts_proto)?;
+    unreal_proto.debug_message = "Hello from Rust!".to_string();
+
+    populate_source(&mut unreal_proto, event_source)?;
+
+    let final_binary = binary_encode_proto(unreal_proto)?;
+
+    match self.redis_client.lock() {
+      Ok(mut redis_client) => {
+        debug!("Publishing to Redis...");
+        let future = redis_client.publish_bytes("unreal", &final_binary);
+        block_on(future);
+        debug!("Published to redis");
+      },
+      Err(_) => {},
+    }
+
+    Ok(())
+  }
+}
+
+impl TextCommandHandler for TtsHandler {
+  fn handle_text_command(&self,
+                         command: &FirstPassParsedCommand,
+                         event: &InboundEvent,
+                         event_source: &InboundEventSource)
+                         -> AnyhowResult<()>
+  {
+    let maybe_tts_command = parse_tts(&command.unparsed_arguments);
+
+    debug!("Maybe tts command: {:?}", maybe_tts_command);
+
+    if let Some(tts) = maybe_tts_command {
+      self.handle(tts, event, event_source);
+    }
+    Ok(())
+  }
+}
+
+#[derive(Clone,Debug,PartialEq)]
+pub struct TtsCommand {
+  pub speaker_slug: String,
+  pub utterance: String,
+}
+
+pub fn parse_tts(input: &str) -> Option<TtsCommand> {
+  lazy_static! {
+    static ref REGEX: Regex =
+      Regex::new(r"^\s*([\w\-]+)\s+(.*)\s*$").expect("Regex should work");
+  }
+
+  let captures = match REGEX.captures(&input) {
+    None => return None,
+    Some(caps) => caps,
+  };
+
+  let maybe_speaker = captures.get(1)
+    .map(|m| m.as_str());
+
+  let maybe_utterance = captures.get(2)
+    .map(|m| m.as_str());
+
+  let speaker = match maybe_speaker {
+    None => return None,
+    Some(cap) => cap,
+  };
+
+  let utterance = match maybe_utterance {
+    None => return None,
+    Some(cap) => cap,
+  };
+
+  Some(TtsCommand {
+    speaker_slug: speaker.to_lowercase(),
+    utterance: utterance.to_string(),
+  })
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use expectest::prelude::*;
+
+  #[test]
+  fn failure_cases() {
+    expect!(parse_tts("onewordthatstretcheson")).to(be_none());
+    expect!(parse_tts("")).to(be_none());
+    expect!(parse_tts("     ")).to(be_none());
+    expect!(parse_tts(" !!! nope    ")).to(be_none());
+  }
+
+  #[test]
+  fn speaker_and_sentence() {
+    expect!(parse_tts("sonic gotta go fast")).to(
+      be_eq(Some(TtsCommand {
+        speaker_slug: "sonic".to_string(),
+        utterance: "gotta go fast".to_string()
+      })));
+  }
+
+  fn speaker_lowercase() {
+    expect!(parse_tts("LINK EXCUSE ME PRINCESS")).to(
+      be_eq(Some(TtsCommand {
+        speaker_slug: "link".to_string(),
+        utterance: "EXCUSE ME PRINCESS".to_string()
+      })));
+  }
+}

src/main.rs

@@ -0,0 +1,121 @@
+#![allow(dead_code)]
+#![allow(unused_mut)]
+#![allow(unused_imports)]
+#![warn(unused_must_use)]
+//#![allow(warnings)]
+
+#[macro_use]
+extern crate lazy_static;
+#[macro_use]
+extern crate serde_derive;
+
+mod clients;
+mod dispatcher;
+mod handlers;
+mod protos;
+mod secrets;
+mod text_chat_parsers;
+
+use anyhow::anyhow;
+use anyhow::{Context, Error};
+use log::{warn, info};
+use redis::aio::Connection;
+use redis::{AsyncCommands, RedisResult};
+use std::thread;
+use std::time::Duration;
+
+use crate::clients::redis_client::RedisClient;
+use crate::clients::redis_subscriber::RedisSubscribeClient;
+use crate::dispatcher::Dispatcher;
+use crate::handlers::coordinate_and_geocode_handler::CoordinateAndGeocodeHandler;
+use crate::secrets::{Secrets, RedisSecrets};
+use std::sync::{Arc, Mutex};
+use crate::handlers::spawn_handler::SpawnHandler;
+use crate::handlers::vocode_handler::TtsHandler;
+
+pub type AnyhowResult<T> = anyhow::Result<T>;
+
+const ENV_SUBSCRIBE_TOPIC : &'static str = "SUBSCRIBE_TOPIC";
+const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "firehose";
+
+const ENV_PUBLISH_TOPIC : &'static str = "PUBLISH_TOPIC";
+const ENV_PUBLISH_TOPIC_DEFAULT : &'static str = "unreal";
+
+const ENV_REDIS_MAX_RETRY_COUNT : &'static str = "REDIS_MAX_RETRY_COUNT";
+const ENV_REDIS_MAX_RETRY_COUNT_DEFAULT : u32 = 3;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+  easyenv::init_all_with_default_logging(None);
+
+  info!("Reading env config...");
+
+  let redis_subscribe_topic = easyenv::get_env_string_or_default(
+    ENV_SUBSCRIBE_TOPIC, ENV_SUBSCRIBE_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 mut dispatcher = Dispatcher::new();
+
+  let mut redis_client = RedisClient::new(
+    &redis_secrets,
+    redis_max_retry_count
+  );
+
+  info!("Connecting to Redis...");
+
+  redis_client.connect().await?;
+
+  let redis_client = Arc::new(Mutex::new(redis_client));
+
+  dispatcher.add_text_command_handler(
+    "goto",
+    Box::new(CoordinateAndGeocodeHandler::new(redis_client.clone())));
+  dispatcher.add_text_command_handler(
+    "spawn",
+    Box::new(SpawnHandler::new(redis_client.clone())));
+  dispatcher.add_text_command_handler(
+    "tts",
+    Box::new(TtsHandler::new(redis_client.clone())));
+
+  let mut redis_pubsub_client = RedisSubscribeClient::new(
+    &redis_secrets,
+    dispatcher
+  );
+
+  info!("Initial connect to PubSub Redis...");
+
+  redis_pubsub_client.connect().await?;
+
+  loop {
+    info!("Main loop iter... (redis pubsub connect, subscribe, start stream...)");
+    redis_pubsub_client.connect().await?;
+    redis_pubsub_client.subscribe(&redis_subscribe_topic).await?;
+    redis_pubsub_client.start_stream().await?;
+    thread::sleep(Duration::from_secs(5));
+  }
+
+  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),
+  ))
+}
+

src/proto_codegen/storyteller.pubsub.rs

@@ -0,0 +1,181 @@
+/// 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>,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct IngestionTwitterMetadata {
+    /// 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 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 {
+}
+/// Intentionally blank for now
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct IngestionTwitterRetweet {
+}
+/// Intentionally blank for now
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct IngestionTwitterMention {
+}
+// ============================== 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/proto_codegen/storyteller.unreal.rs

@@ -0,0 +1,187 @@
+/// The payloads delivered to unreal engine over the pubsub bus.
+/// These are the "enriched" events.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct UnrealEventPayloadV1 {
+    /// Source of the event.
+    #[prost(enumeration="unreal_event_payload_v1::SourceType", tag="1")]
+    pub source_type: i32,
+    /// Type of the payload.
+    #[prost(enumeration="unreal_event_payload_v1::PayloadType", tag="2")]
+    pub payload_type: i32,
+    /// A binary-encoded proto whose type is based on SourceType
+    #[prost(bytes="vec", tag="3")]
+    pub source_data: ::prost::alloc::vec::Vec<u8>,
+    /// A binary-encoded proto whose type is based on PayloadType
+    #[prost(bytes="vec", tag="4")]
+    pub payload_data: ::prost::alloc::vec::Vec<u8>,
+    /// An optional debug message.
+    #[prost(string, tag="5")]
+    pub debug_message: ::prost::alloc::string::String,
+}
+/// Nested message and enum types in `UnrealEventPayloadV1`.
+pub mod unreal_event_payload_v1 {
+    /// If people Retweet us or mention us, we should show that in the engine!
+    #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+    #[repr(i32)]
+    pub enum SourceType {
+        StDoNotUse = 0,
+        StAdminPanel = 1,
+        StVocodesWebsite = 2,
+        StTwitch = 3,
+        StTwitter = 4,
+        StDiscord = 5,
+        StYoutube = 6,
+        StReddit = 7,
+        StTiktok = 8,
+    }
+    #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+    #[repr(i32)]
+    pub enum PayloadType {
+        PtDoNotUse = 0,
+        /// Game events. These can come from many sources (control panel, Twitch, Discord)
+        VocodesTts = 1,
+        LevelWarp = 2,
+        CesiumWarp = 3,
+        CesiumTimeChange = 4,
+        ///CHANGE_MUSIC = 6;
+        SpawnCreature = 5,
+        /// Twitter-specific events.
+        TwitterFollow = 200,
+        TwitterRetweet = 201,
+        TwitterMention = 202,
+        /// Discord-specific events.
+        DiscordJoined = 300,
+        /// YouTube-specific events.
+        YoutubeSubscribed = 400,
+    }
+}
+// ============================== SOURCE PROTOS ============================== //
+
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TwitchMetadata {
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(int64, tag="1")]
+    pub user_id: i64,
+    #[prost(string, tag="2")]
+    pub username: ::prost::alloc::string::String,
+    #[prost(bool, tag="3")]
+    pub user_is_mod: bool,
+    #[prost(bool, tag="4")]
+    pub user_is_subscribed: bool,
+    /// Optional. Name of the channel, prefixed with hash. eg. `#vocodes`.
+    #[prost(string, tag="5")]
+    pub channel: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TwitterMetadata {
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(string, tag="1")]
+    pub username: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct DiscordMetadata {
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(string, tag="1")]
+    pub username: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct YoutubeMetadata {
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(string, tag="1")]
+    pub username: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct RedditMetadata {
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(string, tag="1")]
+    pub username: ::prost::alloc::string::String,
+}
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TikTokMetadata {
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(string, tag="1")]
+    pub username: ::prost::alloc::string::String,
+}
+// ============================== GAME EVENT PROTOS ============================== //
+
+/// Change time in Cesium.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct CesiumTimeChangeRequest {
+    /// Required
+    #[prost(int64, tag="1")]
+    pub hour: i64,
+}
+/// Warp within a Cesium geospatial world.
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct CesiumWarpRequest {
+    /// Required
+    #[prost(double, tag="1")]
+    pub latitude: f64,
+    /// Required
+    #[prost(double, tag="2")]
+    pub longitude: f64,
+    /// TODO: Remove these.
+    /// Optional. Fields describing the user that sent the message.
+    #[prost(int64, tag="3")]
+    pub twitch_user_id: i64,
+    #[prost(string, tag="4")]
+    pub twitch_username: ::prost::alloc::string::String,
+    #[prost(bool, tag="5")]
+    pub twitch_user_is_mod: bool,
+    #[prost(bool, tag="6")]
+    pub twitch_user_is_subscribed: bool,
+    /// TODO: Remove these.
+    /// Optional. Name of the channel, prefixed with hash. eg. `#vocodes`.
+    #[prost(string, tag="7")]
+    pub twitch_channel: ::prost::alloc::string::String,
+}
+/// Change the level
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct LevelWarpRequest {
+    /// Required. Level to warp to.
+    #[prost(string, tag="1")]
+    pub level_slug: ::prost::alloc::string::String,
+}
+/// Spawn a creature
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct SpawnCreatureRequest {
+    /// Required. Name of the creature.
+    #[prost(string, tag="1")]
+    pub name_slug: ::prost::alloc::string::String,
+}
+/// Do text to speech
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct VocodesTtsRequest {
+    /// Required. Speaker to use.
+    #[prost(string, tag="1")]
+    pub voice_slug: ::prost::alloc::string::String,
+    /// Required. Text to speak.
+    #[prost(string, tag="2")]
+    pub text: ::prost::alloc::string::String,
+}
+// ============================== TWITTER EVENT PROTOS ============================== //
+
+/// Intentionally blank
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TwitterFollow {
+}
+/// Intentionally blank
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TwitterRetweet {
+}
+/// Intentionally blank
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct TwitterMention {
+}
+// ============================== DISCORD EVENT PROTOS ============================== //
+
+/// Intentionally blank
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct DiscordJoined {
+}
+// ============================== YOUTUBE EVENT PROTOS ============================== //
+
+/// Intentionally blank
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct YoutubeSubscribed {
+}

src/protos/binary_encode_proto.rs

@@ -0,0 +1,17 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+
+// 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)
+    }
+  }
+}

src/protos/inbound_proto_utils.rs

@@ -0,0 +1,70 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use crate::protos::protos::IngestionTwitchMessage;
+use crate::protos::protos::IngestionTwitchMetadata;
+use crate::protos::protos::PubsubEventPayloadV1;
+use crate::protos::protos::pubsub_event_payload_v1::IngestionPayloadType;
+use crate::protos::protos::pubsub_event_payload_v1::IngestionSourceType;
+use crate::protos::protos;
+use prost::Message;
+
+// TODO: add some type safety w/ generics and type bounds
+
+/// Sources of events
+#[derive(Debug, Clone)]
+pub enum InboundEventSource {
+  Twitch(IngestionTwitchMetadata),
+}
+
+/// Events
+#[derive(Debug, Clone)]
+pub enum InboundEvent {
+  TwitchMessage(IngestionTwitchMessage),
+}
+
+impl InboundEventSource {
+  pub fn parse_from_payload(event_payload: &PubsubEventPayloadV1)
+    -> AnyhowResult<Self>
+  {
+    match Self::parse_type_proto(event_payload) {
+      Some(IngestionSourceType::IstTwitch) => {
+        IngestionTwitchMetadata::decode(event_payload.ingestion_source_data())
+          .map(|m| Self::Twitch(m))
+          .map_err(|_| anyhow!("error decoding source proto"))
+      },
+      None => Err(anyhow!("invalid source type")),
+      Some(_) => Err(anyhow!("source type is not yet handled")),
+    }
+  }
+
+  fn parse_type_proto(event_payload: &PubsubEventPayloadV1)
+    -> Option<IngestionSourceType>
+  {
+    event_payload.ingestion_source_type
+      .and_then(|num| IngestionSourceType::from_i32(num))
+  }
+}
+
+impl InboundEvent {
+  pub fn parse_from_payload(event_payload: &PubsubEventPayloadV1)
+    -> AnyhowResult<Self>
+  {
+    match Self::parse_type_proto(event_payload) {
+      Some(IngestionPayloadType::TwitchMessage) => {
+        IngestionTwitchMessage::decode(event_payload.ingestion_payload_data())
+          .map(|m| Self::TwitchMessage(m))
+          .map_err(|_| anyhow!("error decoding payload proto"))
+      },
+      None => Err(anyhow!("invalid payload type")),
+      Some(_) => Err(anyhow!("payload type is not yet handled")),
+    }
+  }
+
+  fn parse_type_proto(event_payload: &PubsubEventPayloadV1)
+    -> Option<IngestionPayloadType>
+  {
+    event_payload.ingestion_payload_type
+      .and_then(|num| IngestionPayloadType::from_i32(num))
+  }
+}
+

src/protos/mod.rs

@@ -0,0 +1,10 @@
+pub mod binary_encode_proto;
+pub mod inbound_proto_utils;
+pub mod outbound_proto_utils;
+pub mod populate_source;
+
+// Include the protos
+pub mod protos {
+  include!("../proto_codegen/storyteller.pubsub.rs");
+  include!("../proto_codegen/storyteller.unreal.rs");
+}

src/protos/outbound_proto_utils.rs

@@ -0,0 +1,16 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use crate::protos::protos::IngestionTwitchMessage;
+use crate::protos::protos::IngestionTwitchMetadata;
+use crate::protos::protos::PubsubEventPayloadV1;
+use crate::protos::protos::pubsub_event_payload_v1::IngestionPayloadType;
+use crate::protos::protos::pubsub_event_payload_v1::IngestionSourceType;
+use crate::protos::protos;
+use prost::Message;
+
+pub enum OutboundPayload {
+}
+
+pub enum OutboundSourceInfo {
+  Twitch(),
+}

src/protos/populate_source.rs

@@ -0,0 +1,31 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use crate::protos::binary_encode_proto::binary_encode_proto;
+use crate::protos::protos::UnrealEventPayloadV1;
+use crate::protos::protos;
+use log::{info, warn, debug};
+use crate::protos::inbound_proto_utils::InboundEventSource;
+
+pub fn populate_source(unreal_proto: &mut UnrealEventPayloadV1, event_source: &InboundEventSource) -> AnyhowResult<()>
+{
+  match event_source {
+    InboundEventSource::Twitch(ref twitch_source) => {
+      let mut twitch_metadata = protos::TwitchMetadata::default();
+      twitch_metadata.username = twitch_source.username.clone().unwrap_or("".to_string());
+      twitch_metadata.user_id = twitch_source.user_id.clone().unwrap_or(0);
+      twitch_metadata.user_is_mod = twitch_source.user_is_mod.unwrap_or(false);
+      twitch_metadata.user_is_subscribed = twitch_source.user_is_subscribed.unwrap_or(false);
+      twitch_metadata.channel = twitch_source.channel.clone().unwrap_or("".to_string());
+
+      debug!("Source Proto: {:?}", twitch_metadata);
+
+      unreal_proto.source_type = protos::unreal_event_payload_v1::SourceType::StTwitch as i32;
+      unreal_proto.source_data = binary_encode_proto(twitch_metadata)?;
+    },
+    _ => {
+      return Err(anyhow!("Unknown event source proto."));
+    },
+  }
+
+  Ok(())
+}

src/secrets.rs

@@ -0,0 +1,94 @@
+use crate::AnyhowResult;
+use std::io::Read;
+
+#[derive(Deserialize)]
+pub struct Secrets {
+  pub redis: RedisSecrets,
+  pub twitch: TwitchSecrets,
+  pub google: GoogleSecrets,
+}
+
+#[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 GoogleSecrets {
+  pub api_key: String,
+}
+
+#[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 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)
+  }
+}
+

src/text_chat_parsers/first_pass_command_parser.rs

@@ -0,0 +1,87 @@
+use anyhow::anyhow;
+use crate::AnyhowResult;
+use lazy_static::lazy_static;
+use log::debug;
+use regex::Regex;
+
+pub struct FirstPassParsedCommand {
+  /// The first "word" / "verb" of a sentence.
+  /// This is all-lowercase, no spaces, `[\w\-]+`
+  pub command: String,
+
+  /// The remaining portion, stripped of leading/trailing spaces.
+  pub unparsed_arguments: String,
+}
+
+impl FirstPassParsedCommand {
+
+  pub fn try_parse(message: &str) -> AnyhowResult<Self> {
+    lazy_static! {
+      static ref COMMAND_REGEX : Regex = Regex::new(r"^\s*([\w\-]+)\s+(.*)\s*$").expect("should parse");
+    }
+
+    let captures = match COMMAND_REGEX.captures(&message) {
+      None => return Err(anyhow!("message could not parse")),
+      Some(caps) => caps,
+    };
+
+    let command = match captures.get(1) {
+      None => return Err(anyhow!("could not get first capture group")),
+      Some(cmd) => cmd.as_str().trim().to_lowercase(),
+    };
+
+    debug!("Command: {}", &command);
+
+    let unparsed_payload = match captures.get(2) {
+      None => return Err(anyhow!("could not get second capture group")),
+      Some(p) => p.as_str().trim().to_string(),
+    };
+
+    debug!("Unparsed payload: {}", &unparsed_payload);
+
+    Ok(Self {
+      command: command,
+      unparsed_arguments: unparsed_payload,
+    })
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+
+  #[test]
+  fn failure_cases() {
+    assert!(FirstPassParsedCommand::try_parse("").is_err());
+    assert!(FirstPassParsedCommand::try_parse("     ").is_err());
+    assert!(FirstPassParsedCommand::try_parse("foo").is_err());
+    assert!(FirstPassParsedCommand::try_parse("foo-bar").is_err());
+  }
+
+  #[test]
+  fn parse_two_words() {
+    let r = FirstPassParsedCommand::try_parse("foo bar");
+    assert!(r.is_ok());
+    let r = r.expect("Is okay");
+    assert_eq!(&r.command, "foo");
+    assert_eq!(&r.unparsed_arguments, "bar");
+  }
+
+  #[test]
+  fn parse_two_words_spaces() {
+    let r = FirstPassParsedCommand::try_parse("  foo \t   bar \t \n ");
+    assert!(r.is_ok());
+    let r = r.expect("Is okay");
+    assert_eq!(&r.command, "foo");
+    assert_eq!(&r.unparsed_arguments, "bar");
+  }
+
+  #[test]
+  fn parse_hyphenated_command() {
+    let r = FirstPassParsedCommand::try_parse("foo-bar baz");
+    assert!(r.is_ok());
+    let r = r.expect("Is okay");
+    assert_eq!(&r.command, "foo-bar");
+    assert_eq!(&r.unparsed_arguments, "baz");
+  }
+}

src/text_chat_parsers/mod.rs

@@ -0,0 +1 @@
+pub mod first_pass_command_parser;