firehose publishing

debug/mysql/last_firehose_events.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+# ONLY FOR LOCAL DEV
+mysql -u root \
+  -proot \
+  -h localhost \
+  -D storyteller \
+  -e "select * from firehose_entries\G"

debug/mysql/last_w2l_inference_results.sh

@@ -0,0 +1,45 @@
+#!/bin/bash
+# ONLY FOR LOCAL DEV
+
+mysql -u root \
+  -proot \
+  -h localhost \
+  -D storyteller \
+  -e "select * from w2l_results\G"
+
+
+echo '==============================================='
+
+read -r -d '' query << EOF
+
+    SELECT
+      w2l_results.token as w2l_result_token,
+      w2l_results.maybe_tts_inference_result_token,
+      w2l_templates.token as maybe_w2l_template_token,
+      w2l_templates.template_type,
+      users.token as maybe_creator_user_token,
+      users.username as maybe_creator_username,
+      users.display_name as maybe_creator_display_name,
+      w2l_results.file_size_bytes,
+      w2l_results.frame_width,
+      w2l_results.frame_height,
+      w2l_results.duration_millis,
+      w2l_results.created_at,
+      w2l_results.updated_at
+    FROM
+      w2l_results
+      LEFT OUTER JOIN w2l_templates ON w2l_results.maybe_w2l_template_token = w2l_templates.token
+      LEFT OUTER JOIN users ON w2l_results.maybe_creator_user_token = users.token
+    WHERE
+      w2l_results.deleted_at IS NULL
+      AND users.username = 'echelon'\G
+EOF
+
+echo $query
+
+mysql -u root \
+  -proot \
+  -h localhost \
+  -D storyteller \
+  -e "${query}"
+

migrations/2021-05-15-071534_users/up.sql

@@ -5,7 +5,7 @@ CREATE TABLE users (
   -- Not used for anything except replication.
   id BIGINT(20) NOT NULL AUTO_INCREMENT,
 
-  -- Effective "primary key"
+  -- Visible "primary key"
   token VARCHAR(32) NOT NULL,
 
   -- Username is a lookup key; display_name allows the user to add custom case.

migrations/2021-05-30-010200_firehose/down.sql

@@ -0,0 +1,4 @@
+# noinspection SqlResolveForFile
+# noinspection SqlNoDataSourceInspectionForFile
+
+drop table firehose_entries;

migrations/2021-05-30-010200_firehose/up.sql

@@ -0,0 +1,51 @@
+# noinspection SqlResolveForFile
+# noinspection SqlNoDataSourceInspectionForFile
+
+CREATE TABLE firehose_entries (
+    -- Not used for anything except replication.
+    id BIGINT(20) NOT NULL AUTO_INCREMENT,
+
+    -- Visible "primary key"
+    token VARCHAR(32) NOT NULL,
+
+    -- The type of the event
+    event_type VARCHAR(32) NOT NULL,
+
+    -- The target user
+    maybe_target_user_token VARCHAR(32) DEFAULT NULL,
+
+    -- The target "entity", which varies by event type.
+    maybe_target_entity_token VARCHAR(32) DEFAULT NULL,
+
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+    -- Settings
+    -- DO NOT REORDER.
+    -- event_type ENUM(
+    --    'not-set',
+    --    'user_sign_up',
+    --    'tts_model_upload_started',
+    --    'tts_model_upload_completed',
+    --    'tts_inference_started',
+    --    'tts_inference_completed',
+    --    'w2l_template_upload_started',
+    --    'w2l_template_upload_completed',
+    --    'w2l_inference_started',
+    --    'w2l_inference_completed',
+    --    'twitter_mention',
+    --    'twitter_retweet',
+    --    'discord_join',
+    --    'discord_message',
+    --    'twitch_subscribe',
+    --    'twitch_follow'
+    -- ) NOT NULL DEFAULT 'not-set',
+
+    -- INDICES --
+    PRIMARY KEY (id),
+    UNIQUE KEY (token),
+    KEY fk_maybe_target_user_token (maybe_target_user_token),
+    KEY fk_maybe_target_entity_token (maybe_target_entity_token),
+    KEY index_event_type (event_type)
+
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

src/common_queries/firehose_publisher.rs

@@ -0,0 +1,81 @@
+use anyhow::anyhow;
+use crate::util::anyhow_result::AnyhowResult;
+use crate::util::random_prefix_crockford_token::random_prefix_crockford_token;
+use log::{warn,info};
+use sqlx::{MySqlPool};
+use std::sync::Arc;
+use sqlx::error::Error::Database;
+
+enum FirehoseEvent {
+  UserSignUp,
+}
+
+impl FirehoseEvent {
+  pub fn to_db_value(&self) -> &'static str {
+    match self {
+      FirehoseEvent::UserSignUp => "user_sign_up",
+    }
+  }
+}
+
+#[derive(Clone)]
+pub struct FirehosePublisher {
+  pub mysql_pool: MySqlPool,
+}
+
+impl FirehosePublisher {
+
+  pub async fn publish_user_sign_up(&self, user_token: &str) -> AnyhowResult<()> {
+    let token = random_prefix_crockford_token("EV", 32)?;
+    let event_type = FirehoseEvent::UserSignUp;
+
+    let query_result = sqlx::query!(
+        r#"
+INSERT INTO firehose_entries
+SET
+  token = ?,
+  event_type = ?,
+  maybe_target_user_token = ?,
+  maybe_target_entity_token = ?
+        "#,
+      token,
+      event_type.to_db_value(),
+      user_token,
+      user_token,
+    )
+      .execute(&self.mysql_pool)
+      .await;
+
+    let record_id = match query_result {
+      Ok(res) => {
+        res.last_insert_id()
+      },
+      Err(err) => {
+        warn!("Insert record DB error: {:?}", err);
+
+        // NB: SQLSTATE[23000]: Integrity constraint violation
+        // NB: MySQL Error Code 1062: Duplicate key insertion (this is harder to access)
+        match err {
+          Database(err) => {
+            let maybe_code = err.code().map(|c| c.into_owned());
+            /*match maybe_code.as_deref() {
+              Some("23000") => {
+                if err.message().contains("username") {
+                  return Err(UsernameTaken);
+                } else if err.message().contains("email_address") {
+                  return Err(EmailTaken);
+                }
+              }
+              _ => {},
+            }*/
+          },
+          _ => {},
+        }
+        return Err(anyhow!("Error inserting record"));
+      }
+    };
+
+
+    Ok(())
+  }
+}

src/common_queries/mod.rs

@@ -1,4 +1,5 @@
 pub mod badges;
+pub mod firehose_publisher;
 pub mod list_w2l_inference_results;
 pub mod list_w2l_templates;
 pub mod sessions;

src/http_server/endpoints/users/create_account.rs

@@ -215,6 +215,13 @@ VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )
 
   info!("new user session created");
 
+  server_state.firehose_publisher.publish_user_sign_up(&user_token)
+    .await
+    .map_err(|e| {
+      warn!("error publishing event: {:?}", e);
+      CreateAccountError::ServerError
+    })?;
+
   let session_cookie = match server_state.cookie_manager.create_cookie(&session_token) {
     Ok(cookie) => cookie,
     Err(_) => return Err(ServerError),

src/main.rs

@@ -53,6 +53,7 @@ use crate::buckets::bucket_client::BucketClient;
 use crate::http_server::endpoints::users::list_user_w2l_templates::list_user_w2l_templates_handler;
 use crate::http_server::endpoints::users::list_user_w2l_inference_results::list_user_w2l_inference_results_handler;
 use crate::http_server::endpoints::w2l::get_w2l_result::get_w2l_inference_result_handler;
+use crate::common_queries::firehose_publisher::FirehosePublisher;
 
 const DEFAULT_BIND_ADDRESS : &'static str = "0.0.0.0:12345";
 
@@ -104,6 +105,10 @@ async fn main() -> AnyhowResult<()> {
     .connect(&db_connection_string)
     .await?;
 
+  let firehose_publisher = FirehosePublisher {
+    mysql_pool: pool.clone(), // NB: Pool is clone/sync/send-safe
+  };
+
   info!("Reading env vars and setting up utils...");
 
   let bind_address = easyenv::get_env_string_or_default("BIND_ADDRESS", DEFAULT_BIND_ADDRESS);
@@ -153,6 +158,7 @@ async fn main() -> AnyhowResult<()> {
     },
     hostname: server_hostname,
     mysql_pool: pool,
+    firehose_publisher,
     cookie_manager,
     session_checker,
     private_bucket_client,

src/server_state.rs

@@ -2,6 +2,7 @@ use crate::http_server::web_utils::cookie_manager::CookieManager;
 use crate::http_server::web_utils::session_checker::SessionChecker;
 use sqlx::MySqlPool;
 use crate::buckets::bucket_client::BucketClient;
+use crate::common_queries::firehose_publisher::FirehosePublisher;
 
 /// State that is injected into every endpoint.
 #[derive(Clone)]
@@ -17,6 +18,8 @@ pub struct ServerState {
 
   pub session_checker: SessionChecker,
 
+  pub firehose_publisher: FirehosePublisher,
+
   pub private_bucket_client: BucketClient,
   pub public_bucket_client: BucketClient,