From fca64a6cca53f28ceaaf6e45a7bc5807b94014ba Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Sun, 30 May 2021 02:12:27 -0400 Subject: [PATCH] firehose publishing --- debug/mysql/last_firehose_events.sh | 7 ++ debug/mysql/last_w2l_inference_results.sh | 45 +++++++++++ migrations/2021-05-15-071534_users/up.sql | 2 +- .../2021-05-30-010200_firehose/down.sql | 4 + migrations/2021-05-30-010200_firehose/up.sql | 51 ++++++++++++ src/common_queries/firehose_publisher.rs | 81 +++++++++++++++++++ src/common_queries/mod.rs | 1 + .../endpoints/users/create_account.rs | 7 ++ src/main.rs | 6 ++ src/server_state.rs | 3 + 10 files changed, 206 insertions(+), 1 deletion(-) create mode 100755 debug/mysql/last_firehose_events.sh create mode 100755 debug/mysql/last_w2l_inference_results.sh create mode 100644 migrations/2021-05-30-010200_firehose/down.sql create mode 100644 migrations/2021-05-30-010200_firehose/up.sql create mode 100644 src/common_queries/firehose_publisher.rs diff --git a/debug/mysql/last_firehose_events.sh b/debug/mysql/last_firehose_events.sh new file mode 100755 index 0000000000..3e4077969b --- /dev/null +++ b/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" diff --git a/debug/mysql/last_w2l_inference_results.sh b/debug/mysql/last_w2l_inference_results.sh new file mode 100755 index 0000000000..ebc1e305b8 --- /dev/null +++ b/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}" + diff --git a/migrations/2021-05-15-071534_users/up.sql b/migrations/2021-05-15-071534_users/up.sql index ecbcaa6e81..1eae07bce6 100644 --- a/migrations/2021-05-15-071534_users/up.sql +++ b/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. diff --git a/migrations/2021-05-30-010200_firehose/down.sql b/migrations/2021-05-30-010200_firehose/down.sql new file mode 100644 index 0000000000..6bc740efcd --- /dev/null +++ b/migrations/2021-05-30-010200_firehose/down.sql @@ -0,0 +1,4 @@ +# noinspection SqlResolveForFile +# noinspection SqlNoDataSourceInspectionForFile + +drop table firehose_entries; diff --git a/migrations/2021-05-30-010200_firehose/up.sql b/migrations/2021-05-30-010200_firehose/up.sql new file mode 100644 index 0000000000..374c8a34ee --- /dev/null +++ b/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; diff --git a/src/common_queries/firehose_publisher.rs b/src/common_queries/firehose_publisher.rs new file mode 100644 index 0000000000..3fecd0a7da --- /dev/null +++ b/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(()) + } +} diff --git a/src/common_queries/mod.rs b/src/common_queries/mod.rs index 0d32c80f74..bdefa2998f 100644 --- a/src/common_queries/mod.rs +++ b/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; diff --git a/src/http_server/endpoints/users/create_account.rs b/src/http_server/endpoints/users/create_account.rs index 3209066cb4..00d96e7a44 100644 --- a/src/http_server/endpoints/users/create_account.rs +++ b/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), diff --git a/src/main.rs b/src/main.rs index dbbc954d98..f743dd510e 100644 --- a/src/main.rs +++ b/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, diff --git a/src/server_state.rs b/src/server_state.rs index 01824d453e..97d0be1178 100644 --- a/src/server_state.rs +++ b/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,