insert news story, tighten up timings

Cargo.lock

@@ -739,6 +739,7 @@ dependencies = [
  "actix-web-actors",
  "actix_helpers",
  "async-openai",
+ "chrono",
  "clap",
  "concurrency",
  "dotenv",

crates/lib/sqlite_queries/src/queries/by_table/mod.rs

@@ -1,3 +1,4 @@
+pub mod news_stories;
 pub mod news_story_productions;
 pub mod tts_render_tasks;
 pub mod web_scraping_targets;

crates/lib/sqlite_queries/src/queries/by_table/news_stories/insert_news_story.rs

@@ -0,0 +1,50 @@
+use chrono::{DateTime, Utc};
+use errors::{anyhow, AnyhowResult};
+use sqlx::SqlitePool;
+use tokens::tokens::news_stories::NewsStoryToken;
+
+pub struct Args <'a> {
+  pub news_story_token: &'a NewsStoryToken,
+
+  pub original_news_canonical_url: &'a str,
+
+  pub replayable_until: DateTime<Utc>,
+
+  pub sqlite_pool: &'a SqlitePool,
+}
+
+pub async fn insert_news_story(args: Args<'_>) -> AnyhowResult<()> {
+  let news_story_token = args.news_story_token.as_str();
+
+  let query = sqlx::query!(
+        r#"
+INSERT INTO news_stories (
+  news_story_token,
+  original_news_canonical_url,
+  replayable_until,
+  is_playable
+)
+VALUES (
+  ?,
+  ?,
+  ?,
+  true
+)
+        "#,
+        news_story_token,
+        args.original_news_canonical_url,
+        args.replayable_until
+    );
+
+  let query_result = query.execute(args.sqlite_pool)
+      .await;
+
+  let _record_id = match query_result {
+    Ok(res) => res.last_insert_rowid(),
+    Err(err) => {
+      return Err(anyhow!("error inserting: {:?}", err));
+    }
+  };
+
+  Ok(())
+}

crates/lib/sqlite_queries/src/queries/by_table/news_stories/mod.rs

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

crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/list_news_story_productions_ready_for_debut.rs

@@ -0,0 +1,41 @@
+use crate::queries::by_table::news_story_productions::list::news_story_production_item::NewsStoryProductionItem;
+use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
+use errors::{anyhow, AnyhowResult};
+use sqlx::SqlitePool;
+use tokens::tokens::news_stories::NewsStoryToken;
+
+pub async fn list_news_story_productions_ready_for_debut(
+  last_id: i64,
+  limit: i64,
+  sqlite_pool: &SqlitePool,
+) -> AnyhowResult<Vec<NewsStoryProductionItem>> {
+
+  let query = sqlx::query_as!(
+    NewsStoryProductionItem,
+        r#"
+SELECT
+  id,
+  news_story_token as `news_story_token: tokens::tokens::news_stories::NewsStoryToken`,
+  original_news_canonical_url,
+  overall_production_status as `overall_production_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
+  llm_rendition_status as `llm_rendition_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`,
+  llm_rendition_attempts,
+  audio_generation_status as `audio_generation_status: enums::common::sqlite::awaitable_job_status::AwaitableJobStatus`
+FROM news_story_productions
+WHERE
+  overall_production_status = "processing"
+  AND llm_rendition_status = "done"
+  AND audio_generation_status = "done"
+  AND id > ?
+ORDER BY id ASC
+LIMIT ?
+        "#,
+        last_id,
+        limit,
+    );
+
+  let records = query.fetch_all(sqlite_pool)
+      .await?;
+
+  Ok(records)
+}

crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/list/mod.rs

@@ -2,4 +2,5 @@ pub mod list_news_story_productions_awaiting_audio_final_verification;
 pub mod list_news_story_productions_awaiting_audio_generation;
 pub mod list_news_story_productions_awaiting_greenlight;
 pub mod list_news_story_productions_awaiting_llm_rendition;
+pub mod list_news_story_productions_ready_for_debut;
 pub mod news_story_production_item;

crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/mod.rs

@@ -1,4 +1,5 @@
 pub mod update_news_story_audio_finalized;
 pub mod update_news_story_audio_preprocessing_status;
 pub mod update_news_story_llm_rendition_status;
+pub mod update_news_story_post_production_complete;
 pub mod update_news_story_production_greenlit_status;

crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_audio_finalized.rs

@@ -10,20 +10,18 @@ pub struct Args <'a> {
 }
 
 pub async fn update_news_story_audio_finalized(args: Args<'_>) -> AnyhowResult<()> {
-  let mut overall_production_status = AwaitableJobStatus::Done.to_str(); // TODO: Changes if we include imagery.
+  //let mut overall_production_status = AwaitableJobStatus::Done.to_str(); // TODO: Changes if we include imagery.
   let mut audio_generation_status = AwaitableJobStatus::Done.to_str();
 
   let query = sqlx::query!(
         r#"
 UPDATE news_story_productions
 SET
-  overall_production_status = ?,
   audio_generation_status = ?,
   version = version + 1
 WHERE
   news_story_token = ?
         "#,
-        overall_production_status,
         audio_generation_status,
         args.news_story_token,
     );

crates/lib/sqlite_queries/src/queries/by_table/news_story_productions/update/update_news_story_post_production_complete.rs

@@ -0,0 +1,34 @@
+use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
+use errors::{anyhow, AnyhowResult};
+use sqlx::SqlitePool;
+use tokens::tokens::news_stories::NewsStoryToken;
+
+pub struct Args <'a> {
+  pub news_story_token: &'a NewsStoryToken,
+  pub sqlite_pool: &'a SqlitePool,
+}
+
+pub async fn update_news_story_post_production_complete(args: Args<'_>) -> AnyhowResult<()> {
+  let mut overall_production_status = AwaitableJobStatus::Done.to_str();
+
+  let query = sqlx::query!(
+        r#"
+UPDATE news_story_productions
+SET
+  overall_production_status = ?,
+  version = version + 1
+WHERE
+  news_story_token = ?
+        "#,
+        overall_production_status,
+        args.news_story_token,
+    );
+
+  let query_result = query.execute(args.sqlite_pool)
+      .await;
+
+  match query_result {
+    Ok(_) => Ok(()),
+    Err(err) => Err(anyhow!("error updating: {:?}", err)),
+  }
+}

crates/programs/aichatbot-sidecar/Cargo.toml

@@ -29,6 +29,7 @@ tokens = { path = "../../_types/tokens" }
 web_scrapers = { path = "../../lib/web_scrapers" }
 
 # External
+chrono = { version = "0.4.19", features = ["serde"] }
 clap = "2.34.0" # TODO: This is really out of date, but has conflicts when updated
 dotenv = "0.15.0"
 easyenv = "0.0.6"

crates/programs/aichatbot-sidecar/src/main.rs

@@ -42,6 +42,7 @@ use web_scrapers::sites::theguardian::theguardian_scraper::theguardian_scraper_t
 use crate::workers::audio::fakeyou_audio_create::main_loop::fakeyou_audio_create_main_loop;
 use crate::workers::audio::fakeyou_audio_download::main_loop::fakeyou_audio_download_main_loop;
 use crate::workers::news_stories::news_story_audio_final_verification::main_loop::news_story_audio_final_verification_main_loop;
+use crate::workers::news_stories::news_story_post_production_finalization::main_loop::news_story_post_production_finalization_main_loop;
 
 //#[tokio::main]
 //pub async fn main() -> AnyhowResult<()> {
@@ -124,7 +125,7 @@ pub async fn main() -> AnyhowResult<()> {
   // Thus, we launch everything else into its own thread.
   thread::spawn(move || {
     let tokio_runtime = Builder::new_multi_thread()
-        .worker_threads(8)
+        .worker_threads(16)
         .thread_name("tokio-worker")
         .thread_stack_size(3 * 1024 * 1024)
         .enable_time()
@@ -140,6 +141,7 @@ pub async fn main() -> AnyhowResult<()> {
     let job_state7 = job_state.clone();
     let job_state8 = job_state.clone();
     let job_state9 = job_state.clone();
+    let job_state10 = job_state.clone();
 
     tokio_runtime.spawn(async {
       let _r = web_index_ingestion_main_loop(job_state2).await;
@@ -173,6 +175,9 @@ pub async fn main() -> AnyhowResult<()> {
       let _r = news_story_audio_final_verification_main_loop(job_state9).await;
     });
 
+    tokio_runtime.spawn(async {
+      let _r = news_story_post_production_finalization_main_loop(job_state10).await;
+    });
 
     // TODO: Final scheduling thread
     //tokio_runtime.spawn(async {

crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_create/main_loop.rs

@@ -97,7 +97,7 @@ async fn scrape_jobs_of_status(status: TtsRenderStatus, job_state: &Arc<JobState
 
       last_id = target.id;
 
-      thread::sleep(Duration::from_secs(3));
+      thread::sleep(Duration::from_secs(1));
     }
   }
 }

crates/programs/aichatbot-sidecar/src/workers/audio/fakeyou_audio_download/main_loop.rs

@@ -91,7 +91,7 @@ async fn query_and_process_jobs(job_state: &Arc<JobState>) {
 
       last_id = target.id;
 
-      thread::sleep(Duration::from_secs(3));
+      thread::sleep(Duration::from_secs(1));
     }
   }
 }

crates/programs/aichatbot-sidecar/src/workers/news_stories/mod.rs

@@ -2,3 +2,4 @@ pub mod news_story_audio_final_verification;
 pub mod news_story_audio_preprocessing;
 pub mod news_story_greenlighting;
 pub mod news_story_llm_rendition;
+pub mod news_story_post_production_finalization;

crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_audio_final_verification/main_loop.rs

@@ -71,7 +71,7 @@ async fn query_all_batches(job_state: &Arc<JobState>) {
 
       last_id = target.id;
 
-      thread::sleep(Duration::from_secs(5)); // NB: We probably (?) need to throttle API calls to OpenAI
+      thread::sleep(Duration::from_millis(100));
     }
   }
 }

crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_greenlighting/main_loop.rs

@@ -67,7 +67,7 @@ async fn query_and_process_all_batches(job_state: &Arc<JobState>) {
       }
 
       last_id = target.id;
-      thread::sleep(Duration::from_secs(1));
+      thread::sleep(Duration::from_millis(100));
     }
   }
 }

crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/main_loop.rs

@@ -0,0 +1,78 @@
+use crate::shared_state::job_state::JobState;
+use crate::workers::news_stories::news_story_post_production_finalization::process_single_item::process_single_item;
+use enums::by_table::web_rendition_targets::rendition_status::RenditionStatus;
+use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
+use log::{debug, error, info};
+use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_awaiting_audio_final_verification::list_news_story_productions_awaiting_audio_final_verification;
+use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_awaiting_llm_rendition::list_news_story_productions_awaiting_llm_rendition;
+use sqlite_queries::queries::by_table::news_story_productions::list::list_news_story_productions_ready_for_debut::list_news_story_productions_ready_for_debut;
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+
+/// Find in-production stories with all tasks done, then promote them to stories
+pub async fn news_story_post_production_finalization_main_loop(job_state: Arc<JobState>) {
+  loop {
+    debug!("news_story_post_production main loop");
+
+    query_all_batches(&job_state).await;
+
+    debug!("news_story_post_production loop finished; waiting...");
+    thread::sleep(Duration::from_secs(1));
+  }
+}
+
+async fn query_all_batches(job_state: &Arc<JobState>) {
+  const BATCH_SIZE : i64 = 10;
+
+  let mut last_id = 0;
+  let mut failure_count = 0;
+
+  loop {
+    // NB: Protect sqlite from contention.
+    thread::sleep(Duration::from_millis(500));
+
+    debug!("news_story_post_production querying targets from id > {} ...", last_id);
+
+    let query_result = list_news_story_productions_ready_for_debut(
+      last_id, BATCH_SIZE, &job_state.sqlite_pool).await;
+
+    let targets = match query_result {
+      Ok(targets) => targets,
+      Err(err) => {
+        error!("failure querying batch: {:?}", err);
+
+        failure_count += 1;
+
+        // NB: Don't starve progress.
+        if failure_count > 2 {
+          failure_count = 0;
+          last_id += 1;
+        } else if failure_count > 3 {
+          return;
+        }
+
+        continue;
+      }
+    };
+
+    failure_count = 0;
+
+    if targets.is_empty() {
+      return; // Done with batches.
+    }
+
+    for target in targets {
+      debug!("Final post production for target: {:?}", target.original_news_canonical_url);
+
+      let result = process_single_item(&target, job_state).await;
+      if let Err(err) = result {
+        error!("Error processing target: {:?}", err);
+      }
+
+      last_id = target.id;
+
+      thread::sleep(Duration::from_millis(200));
+    }
+  }
+}

crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/mod.rs

@@ -0,0 +1,2 @@
+pub mod main_loop;
+pub mod process_single_item;

crates/programs/aichatbot-sidecar/src/workers/news_stories/news_story_post_production_finalization/process_single_item.rs

@@ -0,0 +1,39 @@
+use chrono::Duration;
+use chrono::Utc;
+use crate::shared_state::job_state::JobState;
+use enums::by_table::tts_render_tasks::tts_render_status::TtsRenderStatus;
+use errors::AnyhowResult;
+use log::{debug, info};
+use once_cell::sync::Lazy;
+use regex::Regex;
+use sqlite_queries::queries::by_table::news_stories::insert_news_story::Args as InsertArgs;
+use sqlite_queries::queries::by_table::news_stories::insert_news_story::insert_news_story;
+use sqlite_queries::queries::by_table::news_story_productions::list::news_story_production_item::NewsStoryProductionItem;
+use sqlite_queries::queries::by_table::news_story_productions::update::update_news_story_post_production_complete::Args as UpdateArgs;
+use sqlite_queries::queries::by_table::news_story_productions::update::update_news_story_post_production_complete::update_news_story_post_production_complete;
+use std::ops::Add;
+use std::sync::Arc;
+
+/// Stop playing stories after 24 hours elapses
+static STORY_FRESHNESS_THRESHOLD : Lazy<Duration> = Lazy::new(|| {
+  Duration::days(1)
+});
+
+pub async fn process_single_item(target: &NewsStoryProductionItem, job_state: &Arc<JobState>) -> AnyhowResult<()> {
+
+  let replayable_until = Utc::now().add(*STORY_FRESHNESS_THRESHOLD);
+
+  insert_news_story(InsertArgs {
+    news_story_token: &target.news_story_token,
+    original_news_canonical_url: &target.original_news_canonical_url,
+    replayable_until,
+    sqlite_pool: &job_state.sqlite_pool,
+  }).await?;
+
+  update_news_story_post_production_complete(UpdateArgs {
+    news_story_token: &target.news_story_token,
+    sqlite_pool: &job_state.sqlite_pool,
+  }).await?;
+
+  Ok(())
+}