fix password issue

this is paging me
fal handle webhook errors (#1272 )
2026-11-11 16:36:47 -05:00 · 2026-03-31 14:54:09 -04:00 · 2026-03-31 14:37:41 -04:00 · 2026-03-31 10:50:52 -04:00 · 2026-03-31 06:22:27 -04:00 · 2026-03-31 05:01:12 -04:00
66 changed files with 1322 additions and 213 deletions
@@ -9620,6 +9620,10 @@ dependencies = [
 "workspace-build-acceleration",
 ]

+[[package]]
+name = "rootly_config"
+version = "0.0.1"
+
 [[package]]
 name = "rsa"
 version = "0.9.6"
@@ -10195,12 +10199,16 @@ dependencies = [
 "jobs_common",
 "log",
 "mysql_queries",
+ "pager",
 "reqwest 0.12.23",
+ "rootly_client",
+ "rootly_config",
 "seedance2pro_client",
 "serde",
 "serde_derive",
 "serde_json",
 "server_environment",
+ "shared_env_var_config",
 "sqlx",
 "tokens",
 "tokio",
@@ -10701,6 +10709,13 @@ dependencies = [
 "keccak",
 ]

+[[package]]
+name = "shared_env_var_config"
+version = "0.0.1"
+dependencies = [
+ "easyenv",
+]
+
 [[package]]
 name = "shlex"
 version = "1.3.0"
@@ -11278,6 +11293,7 @@ dependencies = [
 "resend-rs",
 "reusable_types",
 "rootly_client",
+ "rootly_config",
 "rust-s3",
 "rustls 0.23.36",
 "seedance2pro_client",
@@ -11285,6 +11301,7 @@ dependencies = [
 "serde_derive",
 "serde_json",
 "server_environment",
+ "shared_env_var_config",
 "speculoos",
 "sqlx",
 "strum",
@@ -13360,10 +13377,14 @@ dependencies = [
 "jobs_common",
 "log",
 "mysql_queries",
+ "pager",
+ "rootly_client",
+ "rootly_config",
 "serde",
 "serde_derive",
 "serde_json",
 "server_environment",
+ "shared_env_var_config",
 "sqlx",
 "tempdir",
 "tokio",
@@ -101,6 +101,10 @@ members = [
  # Schema (buckets): S3/R2 topology
  "crates/schema/buckets/bucket_paths",

+  # Schema (config): configuration constants for external services
+  "crates/schema/config/rootly_config",
+  "crates/schema/config/shared_env_var_config",
+
  # Schema (database): queries, database layouts, Redis keys, etc.
  "crates/schema/database/elasticsearch_schema",
  "crates/schema/database/migration",
@@ -242,6 +246,8 @@ file_to_base64_url = { path = "crates/lib/files/file_to_base64_url" }
 filesys = { path = "crates/lib/files/filesys" }
 grok_client = { path = "crates/api_clients/grok_client" }
 rootly_client = { path = "crates/api_clients/rootly_client" }
+rootly_config = { path = "crates/schema/config/rootly_config" }
+shared_env_var_config = { path = "crates/schema/config/shared_env_var_config" }
 http_headers = { path = "crates/lib/http_headers" }
 images = { path = "crates/lib/files/images" }
 jobs_common = { path = "crates/lib/jobs_common" }
@@ -0,0 +1,27 @@
+use serde_derive::{Deserialize, Serialize};
+use utoipa::ToSchema;
+
+#[derive(Deserialize, ToSchema)]
+pub struct LoginRequest {
+  pub username_or_email: String,
+  pub password: String,
+}
+
+#[derive(Serialize, ToSchema)]
+pub struct LoginSuccessResponse {
+  pub success: bool,
+
+  /// A signed session that can be sent as a header, bypassing cookies.
+  /// This is useful for API clients that don't support cookies or Google
+  /// browsers killing cross-domain cookies.
+  pub signed_session: String,
+}
+
+#[derive(Copy, Clone, Debug, Serialize, ToSchema)]
+pub enum LoginErrorType {
+  /// Account was created without a password and the user needs to create one
+  AccountNeedsPassword,
+  /// Invalid login credentials were supplied
+  InvalidCredentials,
+  ServerError,
+}
@@ -1,4 +1,5 @@
 pub mod change_password;
 pub mod edit_email;
 pub mod edit_username;
+pub mod login;
 pub mod session_info;
@@ -177,6 +177,22 @@ impl PagerClient {
          parts.push(format!("HTTP Status Code: {}", status_code));
        }

+        if let Some(user_token) = &notification.user_token {
+          parts.push(format!("User Token: {}", user_token));
+        }
+
+        if let Some(media_file_token) = &notification.media_file_token {
+          parts.push(format!("Media File Token: {}", media_file_token));
+        }
+
+        if let Some(inference_job_token) = &notification.inference_job_token {
+          parts.push(format!("Inference Job Token: {}", inference_job_token));
+        }
+
+        if let Some(third_party_id) = &notification.third_party_id {
+          parts.push(format!("Third Party ID: {}", third_party_id));
+        }
+
        if let Some(h) = &self.hostname {
          parts.push(format!("Hostname: {}", h));
        }
@@ -59,6 +59,10 @@ mod tests {
      http_status_code: None,
      is_from_error,
      urgency: None,
+      user_token: None,
+      media_file_token: None,
+      inference_job_token: None,
+      third_party_id: None,
    }
  }

@@ -214,6 +218,10 @@ mod tests {
        http_status_code,
        is_from_error,
        urgency: None,
+        user_token: None,
+        media_file_token: None,
+        inference_job_token: None,
+        third_party_id: None,
      }
    }

@@ -30,6 +30,18 @@ pub struct NotificationDetails {

  /// Urgency level for the notification.
  pub urgency: Option<NotificationUrgency>,
+
+  /// User token associated with the event, if any.
+  pub user_token: Option<String>,
+
+  /// Media file token associated with the event, if any.
+  pub media_file_token: Option<String>,
+
+  /// Inference job token associated with the event, if any.
+  pub inference_job_token: Option<String>,
+
+  /// Third-party identifier associated with the event, if any.
+  pub third_party_id: Option<String>,
 }

 impl NotificationDetails {
@@ -48,6 +60,10 @@ impl NotificationDetails {
      http_status_code: None,
      is_from_error: false,
      urgency: None,
+      user_token: None,
+      media_file_token: None,
+      inference_job_token: None,
+      third_party_id: None,
    }
  }

@@ -62,6 +78,10 @@ impl NotificationDetails {
      http_status_code: None,
      is_from_error: false,
      urgency: None,
+      user_token: None,
+      media_file_token: None,
+      inference_job_token: None,
+      third_party_id: None,
    }
  }

@@ -123,6 +143,10 @@ impl NotificationDetails {
      http_status_code: None,
      is_from_error: true,
      urgency: None,
+      user_token: None,
+      media_file_token: None,
+      inference_job_token: None,
+      third_party_id: None,
    }
  }

@@ -13,6 +13,10 @@ pub struct NotificationDetailsBuilder {
  http_status_code: Option<u16>,
  is_from_error: bool,
  urgency: Option<NotificationUrgency>,
+  user_token: Option<String>,
+  media_file_token: Option<String>,
+  inference_job_token: Option<String>,
+  third_party_id: Option<String>,
 }

 impl NotificationDetailsBuilder {
@@ -27,6 +31,10 @@ impl NotificationDetailsBuilder {
      http_status_code: None,
      is_from_error: false,
      urgency: None,
+      user_token: None,
+      media_file_token: None,
+      inference_job_token: None,
+      third_party_id: None,
    }
  }

@@ -46,6 +54,10 @@ impl NotificationDetailsBuilder {
      http_status_code: None,
      is_from_error: true,
      urgency: None,
+      user_token: None,
+      media_file_token: None,
+      inference_job_token: None,
+      third_party_id: None,
    }
  }
  
@@ -79,6 +91,26 @@ impl NotificationDetailsBuilder {
    self
  }

+  pub fn set_user_token(mut self, user_token: Option<String>) -> Self {
+    self.user_token = user_token;
+    self
+  }
+
+  pub fn set_media_file_token(mut self, media_file_token: Option<String>) -> Self {
+    self.media_file_token = media_file_token;
+    self
+  }
+
+  pub fn set_inference_job_token(mut self, inference_job_token: Option<String>) -> Self {
+    self.inference_job_token = inference_job_token;
+    self
+  }
+
+  pub fn set_third_party_id(mut self, third_party_id: Option<String>) -> Self {
+    self.third_party_id = third_party_id;
+    self
+  }
+
  pub fn build(self) -> NotificationDetails {
    NotificationDetails {
      title: self.title,
@@ -89,6 +121,10 @@ impl NotificationDetailsBuilder {
      http_status_code: self.http_status_code,
      is_from_error: self.is_from_error,
      urgency: self.urgency,
+      user_token: self.user_token,
+      media_file_token: self.media_file_token,
+      inference_job_token: self.inference_job_token,
+      third_party_id: self.third_party_id,
    }
  }
 }
@@ -1,16 +1,28 @@
-use errors::AnyhowResult;
+use crate::constants::PASSWORD_HASH_SENTINEL_VALUE;
+use crate::errors::password_confirm_error::PasswordConfirmError;

-pub fn bcrypt_confirm_password(password: String, bcrypt_hash: &str) -> AnyhowResult<bool> {
-  let verified = bcrypt::verify(&password, &bcrypt_hash)?;
+pub fn bcrypt_confirm_password(password: String, bcrypt_hash: &str) -> Result<bool, PasswordConfirmError> {
+  if bcrypt_hash.is_empty() {
+    return Err(PasswordConfirmError::HashNotProvided);
+  }
+
+  // The database stores the sentinel "*" in a BINARY(60) column, which pads with null bytes.
+  // This reproduces the real-world value: "*" followed by 59 null bytes.
+  if bcrypt_hash.trim_end_matches('\0') == PASSWORD_HASH_SENTINEL_VALUE {
+    return Err(PasswordConfirmError::HashIsSentinelValue);
+  }
+
+  let verified = bcrypt::verify(&password, bcrypt_hash)?;
  Ok(verified)
 }

 #[cfg(test)]
 mod tests {
  use crate::bcrypt_confirm_password::bcrypt_confirm_password;
+  use crate::errors::password_confirm_error::PasswordConfirmError;

  #[test]
-  fn test_confirmation() {
+  fn test_several_password_hash_pairs() {
    // NB: This is to make sure we don't migrate the algorithm to a state where it won't work anymore.
    // NB: Be mindful that these are expensive to compute and don't add too many test cases.
    let hash_to_passwords = [
@@ -25,4 +37,30 @@ mod tests {
      assert!(valid);
    }
  }
-}
+
+  #[test]
+  fn test_empty_hash_returns_error() {
+    let result = bcrypt_confirm_password("password".to_string(), "");
+    assert!(matches!(result, Err(PasswordConfirmError::HashNotProvided)));
+  }
+
+  #[test]
+  fn test_sentinel_value_returns_error() {
+    let result = bcrypt_confirm_password("password".to_string(), "*");
+    assert!(matches!(result, Err(PasswordConfirmError::HashIsSentinelValue)));
+  }
+
+  #[test]
+  fn test_sentinel_value_with_null_bytes_returns_bcrypt_error() {
+    // The database stores the sentinel "*" in a BINARY(60) column, which pads with null bytes.
+    // This reproduces the real-world value: "*" followed by 59 null bytes.
+    // It's currently paging me (until I fix it):
+    // [2026-03-31T16:03:00Z WARN  storyteller_web::http_server::endpoints::users::login_handler] Login hash comparison error: BcryptVerifyError(InvalidHash("*\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0"))
+    let mut hash_with_nulls = String::from("*");
+    for _ in 0..59 {
+      hash_with_nulls.push('\0');
+    }
+    let result = bcrypt_confirm_password("password".to_string(), &hash_with_nulls);
+    assert!(matches!(result, Err(PasswordConfirmError::HashIsSentinelValue)));
+  }
+}
@@ -1,6 +1,6 @@
-use errors::AnyhowResult;
+use crate::errors::password_hash_error::PasswordHashError;

-pub fn bcrypt_hash_password(password: String) -> AnyhowResult<String> {
+pub fn bcrypt_hash_password(password: String) -> Result<String, PasswordHashError> {
  let hash = bcrypt::hash(&password, bcrypt::DEFAULT_COST)?;
  Ok(hash)
 }
@@ -0,0 +1,3 @@
+/// Special database sentinel value indicating that a password is not set.
+/// This is used to support "no-password" signups, since the password hash database column is itself non-nullable.
+pub const PASSWORD_HASH_SENTINEL_VALUE: &str = "*";
@@ -0,0 +1,3 @@
+pub mod password_error;
+pub mod password_confirm_error;
+pub mod password_hash_error;
@@ -0,0 +1,30 @@
+use std::fmt;
+
+#[derive(Debug)]
+pub enum PasswordConfirmError {
+  /// The password hash is the empty string.
+  HashNotProvided,
+  /// The password hash is the special database sentinel value "*",
+  /// which is used to indicate that the password is not set.
+  HashIsSentinelValue,
+  /// An error with verifying the password.
+  BcryptVerifyError(bcrypt::BcryptError),
+}
+
+impl fmt::Display for PasswordConfirmError {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    match self {
+      Self::HashNotProvided => write!(f, "password hash not provided"),
+      Self::HashIsSentinelValue => write!(f, "password hash is sentinel value"),
+      Self::BcryptVerifyError(e) => write!(f, "bcrypt verify error: {}", e),
+    }
+  }
+}
+
+impl std::error::Error for PasswordConfirmError {}
+
+impl From<bcrypt::BcryptError> for PasswordConfirmError {
+  fn from(err: bcrypt::BcryptError) -> Self {
+    Self::BcryptVerifyError(err)
+  }
+}
@@ -0,0 +1,33 @@
+use std::fmt;
+
+use crate::errors::password_confirm_error::PasswordConfirmError;
+use crate::errors::password_hash_error::PasswordHashError;
+
+#[derive(Debug)]
+pub enum PasswordError {
+  CheckError(PasswordConfirmError),
+  HashError(PasswordHashError),
+}
+
+impl fmt::Display for PasswordError {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    match self {
+      Self::CheckError(e) => write!(f, "{}", e),
+      Self::HashError(e) => write!(f, "{}", e),
+    }
+  }
+}
+
+impl std::error::Error for PasswordError {}
+
+impl From<PasswordConfirmError> for PasswordError {
+  fn from(err: PasswordConfirmError) -> Self {
+    Self::CheckError(err)
+  }
+}
+
+impl From<PasswordHashError> for PasswordError {
+  fn from(err: PasswordHashError) -> Self {
+    Self::HashError(err)
+  }
+}
@@ -0,0 +1,23 @@
+use std::fmt;
+
+#[derive(Debug)]
+pub enum PasswordHashError {
+  /// An error with hashing the password.
+  BcryptHashError(bcrypt::BcryptError),
+}
+
+impl fmt::Display for PasswordHashError {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    match self {
+      Self::BcryptHashError(e) => write!(f, "bcrypt hash error: {}", e),
+    }
+  }
+}
+
+impl std::error::Error for PasswordHashError {}
+
+impl From<bcrypt::BcryptError> for PasswordHashError {
+  fn from(err: bcrypt::BcryptError) -> Self {
+    Self::BcryptHashError(err)
+  }
+}
@@ -23,3 +23,5 @@
 //  https://stytch.com/blog/argon2-vs-bcrypt-vs-scrypt/
 pub mod bcrypt_hash_password;
 pub mod bcrypt_confirm_password;
+pub mod constants;
+pub mod errors;
@@ -0,0 +1,9 @@
+[package]
+name = "rootly_config"
+edition = "2024"
+version = "0.0.1"
+publish = false
+
+[lib]
+name = "rootly_config"
+path = "src/lib.rs"
@@ -0,0 +1,2 @@
+pub mod services;
+pub mod urgencies;
@@ -0,0 +1,3 @@
+pub const ROOTLY_SERVICE_ID_STORYTELLER_WEB: &str = "9ebebb09-0c56-4c8a-8f57-d5f0f85f3f16";
+pub const ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB: &str = "2a78f884-c42c-4745-b826-b7a572780ea0";
+pub const ROOTLY_SERVICE_ID_VIDEO_THUMBNAIL_JOB: &str = "9d22997c-4474-47cb-af7b-53119eed1a88";
@@ -0,0 +1,3 @@
+pub const ROOTLY_URGENCY_ID_HIGH: &str = "62fde143-1258-4639-9ee6-1400ebf7308d";
+pub const ROOTLY_URGENCY_ID_MEDIUM: &str = "7366ba5e-f6ea-4a4e-a1b4-ae43d512e1e2";
+pub const ROOTLY_URGENCY_ID_LOW: &str = "4db3d3ed-f4ed-4818-82f5-7746da404bd2";
@@ -0,0 +1,12 @@
+[package]
+name = "shared_env_var_config"
+edition = "2024"
+version = "0.0.1"
+publish = false
+
+[lib]
+name = "shared_env_var_config"
+path = "src/lib.rs"
+
+[dependencies]
+easyenv.workspace = true
@@ -0,0 +1 @@
+pub mod paging;
@@ -0,0 +1,29 @@
+// ----- Environment Variable Names -----
+
+const ENV_ENABLE_PAGING: &str = "ENABLE_PAGING";
+const ENV_ENABLE_PAGING_FOR_500S: &str = "ENABLE_PAGING_FOR_500S";
+const ENV_ROOTLY_API_KEY: &str = "ROOTLY_API_KEY";
+const ENV_ROOTLY_NOTIFICATION_TARGET_TYPE: &str = "ROOTLY_NOTIFICATION_TARGET_TYPE";
+const ENV_ROOTLY_NOTIFICATION_TARGET_ID: &str = "ROOTLY_NOTIFICATION_TARGET_ID";
+
+// ----- Read Environment Variables -----
+
+pub fn env_enable_paging_default_false() -> bool {
+  easyenv::get_env_bool_or_default(ENV_ENABLE_PAGING, false)
+}
+
+pub fn env_enable_paging_for_500s_default_false() -> bool {
+  easyenv::get_env_bool_or_default(ENV_ENABLE_PAGING_FOR_500S, false)
+}
+
+pub fn env_optional_rootly_api_key() -> Option<String> {
+  easyenv::get_env_string_optional(ENV_ROOTLY_API_KEY)
+}
+
+pub fn env_optional_rootly_notification_target_type() -> Option<String> {
+  easyenv::get_env_string_optional(ENV_ROOTLY_NOTIFICATION_TARGET_TYPE)
+}
+
+pub fn env_optional_rootly_notification_target_id() -> Option<String> {
+  easyenv::get_env_string_optional(ENV_ROOTLY_NOTIFICATION_TARGET_ID)
+}
@@ -32,8 +32,12 @@ errors.workspace = true
 hashing = { path = "../../../lib/files/hashing" }
 jobs_common = { path = "../../../lib/jobs_common" }
 mysql_queries = { path = "../../../schema/database/mysql_queries" }
+pager = { path = "../../../lib/pager" }
+rootly_client.workspace = true
+rootly_config.workspace = true
 seedance2pro_client = { path = "../../../api_clients/seedance2pro_client" }
 server_environment = { path = "../../../lib/server_environment" }
+shared_env_var_config.workspace = true
 tokens.workspace = true

 # External
@@ -7,6 +7,8 @@ use actix_web::{web, HttpRequest, HttpResponse};
 use log::error;

 use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response;
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;

 use crate::http_server::http_server_shared_state::HttpServerSharedState;

@@ -46,7 +48,7 @@ impl std::fmt::Display for HealthCheckError {
 }

 pub async fn get_health_check_handler(
-  _http_request: HttpRequest,
+  http_request: HttpRequest,
  server_state: web::Data<Arc<HttpServerSharedState>>,
 ) -> Result<HttpResponse, HealthCheckError> {
  let job_stats = server_state
@@ -76,6 +78,30 @@ pub async fn get_health_check_handler(
  let is_healthy =
    job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold;

+  if !is_healthy {
+    let notification = NotificationDetailsBuilder::from_title(
+          format!("Health check unhealthy on {}", server_state.hostname))
+        .set_description(Some(format!(
+          "Health check returned unhealthy.\n\n\
+             Hostname: {}\n\
+             Consecutive failure count: {}\n\
+             Total failure count: {}\n\
+             Total success count: {}",
+          server_state.hostname,
+          job_stats.consecutive_failure_count,
+          job_stats.total_failure_count,
+          job_stats.total_success_count,
+        )))
+        .set_urgency(Some(NotificationUrgency::High))
+        .set_http_method(Some(http_request.method().to_string()))
+        .set_http_path(Some(http_request.path().to_string()))
+        .build();
+
+    if let Err(err) = server_state.pager.enqueue_page(notification) {
+      error!("Failed to enqueue health check alert: {:?}", err);
+    }
+  }
+
  let response = HealthCheckResponse {
    success: true,
    is_healthy,
@@ -1,7 +1,10 @@
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;

 #[derive(Clone)]
 pub struct HttpServerSharedState {
  pub job_stats: JobStats,
  pub consecutive_failure_unhealthy_threshold: u64,
+  pub pager: Pager,
+  pub hostname: String,
 }
@@ -8,6 +8,7 @@ use log::info;
 use bootstrap::bootstrap::ContainerEnvironment;
 use errors::AnyhowResult;
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;

 use crate::http_server::endpoints::health_check_handler::get_health_check_handler;
 use crate::http_server::http_server_shared_state::HttpServerSharedState;
@@ -18,12 +19,13 @@ const DEFAULT_NUM_WORKERS: usize = 2;
 pub struct CreateServerArgs {
  pub container_environment: ContainerEnvironment,
  pub job_stats: JobStats,
+  pub pager: Pager,
 }

 pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
  let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS);
  let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?;
-  let _hostname = args.container_environment.hostname.clone();
+  let hostname = args.container_environment.hostname.clone();

  let server_state = HttpServerSharedState {
    job_stats: args.job_stats.clone(),
@@ -31,6 +33,8 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
      "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD",
      3,
    )?,
+    pager: args.pager,
+    hostname,
  };

  let server_state_arc = web::Data::new(Arc::new(server_state));
@@ -1,6 +1,7 @@
 use cloud_storage::bucket_client::BucketClient;
 use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;
 use seedance2pro_client::creds::seedance2pro_session::Seedance2ProSession;
 use server_environment::ServerEnvironment;
 use sqlx::MySqlPool;
@@ -28,4 +29,7 @@ pub struct JobDependencies {

  /// Set to `true` from another thread to trigger graceful shutdown.
  pub application_shutdown: RelaxedAtomicBool,
+
+  /// Pager client for sending alerts.
+  pub pager: Pager,
 }
@@ -0,0 +1,32 @@
+use log::error;
+use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_jobs::PendingSeedance2ProJob;
+use pager::client::pager::Pager;
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
+
+/// Enqueue a pager alert for the error, then return it as `Err`.
+pub fn alert_pager_and_return_err<T>(
+  pager: &Pager,
+  title: &str,
+  err: anyhow::Error,
+  job: Option<&PendingSeedance2ProJob>,
+) -> anyhow::Result<T> {
+  let mut builder = NotificationDetailsBuilder::from_error(&err)
+      .set_title(title.to_string())
+      .set_urgency(Some(NotificationUrgency::Medium));
+
+  if let Some(job) = job {
+    builder = builder
+        .set_inference_job_token(Some(job.job_token.to_string()))
+        .set_third_party_id(Some(job.order_id.clone()))
+        .set_user_token(job.maybe_creator_user_token.as_ref().map(|t| t.to_string()));
+  }
+
+  let notification = builder.build();
+
+  if let Err(pager_err) = pager.enqueue_page(notification) {
+    error!("Failed to enqueue pager alert: {:?}", pager_err);
+  }
+
+  Err(err)
+}
@@ -1,22 +1,47 @@
 use std::collections::HashMap;
-use std::time::Duration;
+use std::time::{Duration, Instant};

 use log::{error, info, warn};
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
 use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_jobs::list_pending_seedance2pro_jobs;
 use seedance2pro_client::requests::poll_orders::poll_orders::{poll_orders, OrderStatus, PollOrdersArgs, TaskStatus};

+use crate::jobs::alert_on_error::alert_pager_and_return_err;
 use crate::jobs::process_page_batch::process_page_batch;
 use crate::job_dependencies::JobDependencies;

+const POLL_ALERT_THRESHOLD: Duration = Duration::from_secs(600);
+
 pub async fn main_loop(job_dependencies: JobDependencies) {
  while !job_dependencies.application_shutdown.get() {
+    let start = Instant::now();
    let result = run_poll_iteration(&job_dependencies).await;
+    let elapsed = start.elapsed();

    if let Err(err) = result {
      error!("Error in poll iteration: {:?}", err);
+      let _ = alert_pager_and_return_err::<()>(&job_dependencies.pager, "Seedance2Pro poll iteration error", err, None);
      let _ = job_dependencies.job_stats.increment_failure_count();
    }

+    if elapsed > POLL_ALERT_THRESHOLD {
+      warn!("Poll iteration took {:.1}s (threshold: {}s)", elapsed.as_secs_f64(), POLL_ALERT_THRESHOLD.as_secs());
+
+      let notification = NotificationDetailsBuilder::from_title(
+            "Seedance2Pro poll iteration slow".to_string())
+          .set_description(Some(format!(
+            "Poll iteration took {:.1} seconds, exceeding the 10-minute threshold.",
+            elapsed.as_secs_f64(),
+          )))
+          .set_urgency(Some(NotificationUrgency::Medium))
+          .build();
+
+      if let Err(pager_err) = job_dependencies.pager.enqueue_page(notification) {
+        error!("Failed to enqueue slow iteration alert: {:?}", pager_err);
+      }
+    }
+
    tokio::time::sleep(Duration::from_millis(job_dependencies.poll_interval_millis)).await;
  }

@@ -25,7 +50,13 @@ pub async fn main_loop(job_dependencies: JobDependencies) {

 async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
  // 1. Query all non-terminal Seedance2Pro jobs from DB.
-  let pending_jobs = list_pending_seedance2pro_jobs(&deps.mysql_pool).await?;
+  let pending_jobs = match list_pending_seedance2pro_jobs(&deps.mysql_pool).await {
+    Ok(jobs) => jobs,
+    Err(err) => {
+      error!("Failed to list pending seedance2pro jobs: {:?}", err);
+      return alert_pager_and_return_err(&deps.pager, "Seedance2Pro DB query failed", err.into(), None);
+    }
+  };

  if pending_jobs.is_empty() {
    info!("No pending Seedance2Pro jobs.");
@@ -52,6 +83,11 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
  let mut pages_in_current_batch: u32 = 0;

  loop {
+    if deps.application_shutdown.get() {
+      info!("Shutdown requested during pagination. Stopping early.");
+      break;
+    }
+
    total_page_number += 1;
    pages_in_current_batch += 1;

@@ -60,16 +96,22 @@ async fn run_poll_iteration(deps: &JobDependencies) -> anyhow::Result<()> {
      total_page_number, cursor, total_orders_seen
    );

-    let response = poll_orders(PollOrdersArgs {
+    let response = match poll_orders(PollOrdersArgs {
      session: &deps.seedance2pro_session,
      cursor,
      host_override: None,
-    })
-      .await
-      .map_err(|err| {
+    }).await {
+      Ok(r) => r,
+      Err(err) => {
        warn!("Error polling seedance2pro orders: {:?}", err);
-        anyhow::anyhow!("poll_orders failed: {:?}", err)
-      })?;
+        return alert_pager_and_return_err(
+          &deps.pager,
+          "Seedance2Pro API poll failed",
+          anyhow::anyhow!("poll_orders failed: {:?}", err),
+          None,
+        );
+      }
+    };

    let page_count = response.orders.len() as u32;
    info!("Done polling page {}. Got {} orders on this page.", total_page_number, page_count);
@@ -1,3 +1,4 @@
+pub mod alert_on_error;
 pub mod main_loop;
 pub mod process_job;
 pub mod process_page_batch;
@@ -4,6 +4,8 @@ use enums::by_table::generic_inference_jobs::frontend_failure_category::Frontend
 use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
 use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedance2pro_jobs::PendingSeedance2ProJob;
 use mysql_queries::queries::wallets::refund::try_to_refund_ledger_entry::{try_to_refund_ledger_entry, WalletRefundOutcome};
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
 use seedance2pro_client::requests::poll_orders::failure_type::FailureType;
 use seedance2pro_client::requests::poll_orders::poll_orders::OrderStatus;

@@ -82,6 +84,19 @@ pub async fn process_failed_job(
            job.job_token.as_str(),
            err,
          );
+
+          let notification = NotificationDetailsBuilder::from_error(&err)
+              .set_title("Seedance2Pro refund failed".to_string())
+              .set_inference_job_token(Some(job.job_token.to_string()))
+              .set_third_party_id(Some(job.order_id.to_string()))
+              .set_user_token(job.maybe_creator_user_token.as_ref().map(|t| t.to_string()))
+              .set_urgency(Some(NotificationUrgency::Medium))
+              .build();
+
+          if let Err(pager_err) = deps.pager.enqueue_page(notification) {
+            error!("Failed to enqueue pager alert: {:?}", pager_err);
+          }
+
          let _ = transaction.rollback().await;
          return;
        }
@@ -124,5 +139,14 @@ pub async fn process_failed_job(
      job.job_token.as_str(),
      err
    );
+
+    let notification = NotificationDetailsBuilder::from_error(&err)
+        .set_title("Seedance2Pro mark job failed error".to_string())
+        .set_urgency(Some(NotificationUrgency::Medium))
+        .build();
+
+    if let Err(pager_err) = deps.pager.enqueue_page(notification) {
+      error!("Failed to enqueue pager alert: {:?}", pager_err);
+    }
  }
 }
@@ -13,6 +13,8 @@ use mysql_queries::queries::generic_inference::seedance2pro::list_pending_seedan
 use mysql_queries::queries::media_files::create::insert_builder::media_file_insert_builder::MediaFileInsertBuilder;
 use mysql_queries::queries::generic_inference::web::mark_generic_inference_job_successfully_done_by_token::mark_generic_inference_job_successfully_done_by_token;
 use seedance2pro_client::requests::poll_orders::poll_orders::OrderStatus;
+
+use crate::jobs::alert_on_error::alert_pager_and_return_err;
 use crate::job_dependencies::JobDependencies;

 const PREFIX: &str = "artcraft_";
@@ -47,13 +49,29 @@ pub async fn process_successful_job(
  );

  // Download the video bytes.
-  let video_bytes: Vec<u8> = reqwest::get(video_url)
-    .await
-    .map_err(|err| anyhow!("reqwest error downloading video: {:?}", err))?
-    .bytes()
-    .await
-    .map_err(|err| anyhow!("error reading video bytes: {:?}", err))?
-    .to_vec();
+  let video_bytes: Vec<u8> = match reqwest::get(video_url).await {
+    Ok(resp) => match resp.bytes().await {
+      Ok(bytes) => bytes.to_vec(),
+      Err(err) => {
+        error!("Error reading video bytes for order {}: {:?}", order.order_id, err);
+        return alert_pager_and_return_err(
+          &deps.pager,
+          "Seedance2Pro video download failed",
+          anyhow!("error reading video bytes: {:?}", err),
+          Some(job),
+        );
+      }
+    },
+    Err(err) => {
+      error!("Error downloading video for order {}: {:?}", order.order_id, err);
+      return alert_pager_and_return_err(
+        &deps.pager,
+        "Seedance2Pro video download failed",
+        anyhow!("reqwest error downloading video: {:?}", err),
+        Some(job),
+      );
+    }
+  };

  info!(
    "Downloaded {} bytes for order {}",
@@ -76,11 +94,19 @@ pub async fn process_successful_job(
  );

  // Upload to public bucket.
-  deps
+  if let Err(err) = deps
    .public_bucket_client
    .upload_file_with_content_type_process(object_path, &video_bytes, "video/mp4")
    .await
-    .map_err(|err| anyhow!("error uploading video to bucket: {:?}", err))?;
+  {
+    error!("Error uploading video for order {}: {:?}", order.order_id, err);
+    return alert_pager_and_return_err(
+      &deps.pager,
+      "Seedance2Pro bucket upload failed",
+      anyhow!("error uploading video to bucket: {:?}", err),
+      Some(job),
+    );
+  }

  info!(
    "Uploaded video for order {}. Creating media file record.",
@@ -92,7 +118,7 @@ pub async fn process_successful_job(
  let maybe_frame_height = order.results.first().map(|r| r.height);

  // Insert media file record.
-  let media_file_token = MediaFileInsertBuilder::new()
+  let media_file_token = match MediaFileInsertBuilder::new()
    .maybe_creator_user(job.maybe_creator_user_token.as_ref())
    .maybe_creator_anonymous_visitor(job.maybe_creator_anonymous_visitor_token.as_ref())
    .creator_ip_address(&job.creator_ip_address)
@@ -110,7 +136,18 @@ pub async fn process_successful_job(
    .public_bucket_directory_hash(&bucket_path)
    .insert_pool(&deps.mysql_pool)
    .await
-    .map_err(|err| anyhow!("error inserting media file record: {:?}", err))?;
+  {
+    Ok(token) => token,
+    Err(err) => {
+      error!("Error inserting media file record for order {}: {:?}", order.order_id, err);
+      return alert_pager_and_return_err(
+        &deps.pager,
+        "Seedance2Pro media file insert failed",
+        anyhow!("error inserting media file record: {:?}", err),
+        Some(job),
+      );
+    }
+  };

  info!(
    "Created media file {} for order {}. Marking job {} complete.",
@@ -120,23 +157,22 @@ pub async fn process_successful_job(
  );

  // Mark inference job as successfully completed.
-  mark_generic_inference_job_successfully_done_by_token(
+  if let Err(err) = mark_generic_inference_job_successfully_done_by_token(
    &deps.mysql_pool,
    &job.job_token,
    Some(InferenceResultType::MediaFile),
    Some(media_file_token.as_str()),
    None,
    None,
-  )
-    .await
-    .map_err(|err| {
-      error!(
-        "Error marking job {} done: {:?}",
-        job.job_token.as_str(),
-        err
-      );
-      anyhow!("error marking job done: {:?}", err)
-    })?;
+  ).await {
+    error!("Error marking job {} done: {:?}", job.job_token.as_str(), err);
+    return alert_pager_and_return_err(
+      &deps.pager,
+      "Seedance2Pro job completion update failed",
+      anyhow!("error marking job done: {:?}", err),
+      Some(job),
+    );
+  }

  info!("Job {} completed successfully.", job.job_token.as_str());

@@ -28,10 +28,12 @@ use server_environment::ServerEnvironment;
 use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
 use crate::jobs::main_loop::main_loop;
 use crate::job_dependencies::JobDependencies;
+use crate::startup::build_pager::build_pager;

 pub mod http_server;
 pub mod job_dependencies;
 pub mod jobs;
+pub mod startup;

 // Bucket config
 const ENV_ACCESS_KEY: &str = "ACCESS_KEY";
@@ -41,7 +43,7 @@ const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME";
 const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL";
 const ENV_SEEDANCE2PRO_COOKIES : &str = "SEEDANCE2PRO_COOKIES";

-#[actix_web::main]
+#[tokio::main]
 async fn main() -> AnyhowResult<()> {

  let container_environment = bootstrap(BootstrapArgs {
@@ -113,12 +115,27 @@ async fn main() -> AnyhowResult<()> {
    info!("Batch page count: {}", count);
  }

+  let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname);
+
+  info!("Spawning pager worker.");
+
+  // NB: The pager worker uses Condvar::wait() which is a blocking syscall.
+  // It must run on a dedicated OS thread, not a tokio task, to avoid blocking
+  // the tokio runtime.
+  std::thread::spawn(move || {
+    let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime");
+    rt.block_on(pager_worker.run());
+  });
+
  let application_shutdown = RelaxedAtomicBool::new(false);
  let job_stats = JobStats::new();

+  let pager_for_shutdown = pager.clone();
+
  let create_server_args = CreateServerArgs {
    container_environment: container_environment.clone(),
    job_stats: job_stats.clone(),
+    pager: pager.clone(),
  };

  let job_dependencies = JobDependencies {
@@ -130,8 +147,10 @@ async fn main() -> AnyhowResult<()> {
    poll_interval_millis,
    maybe_pages_per_batch,
    application_shutdown: application_shutdown.clone(),
+    pager,
  };

+  // HTTP server runs on a separate OS thread with its own actix System.
  std::thread::spawn(move || {
    let actix_runtime = actix_web::rt::System::new();
    let http_server_handle = launch_http_server(create_server_args);
@@ -139,11 +158,30 @@ async fn main() -> AnyhowResult<()> {
    actix_runtime.block_on(http_server_handle)
      .expect("HTTP server should not exit.");

-    warn!("Server thread is shut down.");
-    application_shutdown.set(true);
+    warn!("HTTP server thread is shut down.");
+  });
+
+  // Listen for SIGTERM / Ctrl-C to trigger graceful shutdown.
+  let application_shutdown_for_signal = application_shutdown.clone();
+
+  tokio::spawn(async move {
+    match tokio::signal::ctrl_c().await {
+      Ok(()) => {
+        info!("Received shutdown signal. Shutting down...");
+        application_shutdown_for_signal.set(true);
+      }
+      Err(err) => {
+        warn!("Error listening for shutdown signal: {:?}", err);
+      }
+    }
  });

  main_loop(job_dependencies).await;

+  info!("Shutting down pager worker...");
+  pager_for_shutdown.shutdown_worker();
+
+  info!("Seedance2Pro job exiting.");
+
  Ok(())
 }
@@ -0,0 +1,65 @@
+use log::{info, warn};
+use pager::client::pager::Pager;
+use pager::client::pager_builder::PagerBuilder;
+use pager::worker::pager_worker::PagerWorker;
+use rootly_client::creds::rootly_api_key::RootlyApiKey;
+use rootly_config::services::ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB;
+use rootly_config::urgencies::{ROOTLY_URGENCY_ID_HIGH, ROOTLY_URGENCY_ID_LOW, ROOTLY_URGENCY_ID_MEDIUM};
+use shared_env_var_config::paging::{env_enable_paging_default_false, env_optional_rootly_api_key, env_optional_rootly_notification_target_id, env_optional_rootly_notification_target_type};
+
+pub fn build_pager(
+  server_environment: server_environment::ServerEnvironment,
+  hostname: &str,
+) -> (Pager, PagerWorker) {
+  let is_paging_enabled = env_enable_paging_default_false();
+
+  info!("Paging enabled: {}", is_paging_enabled);
+
+  let environment = if server_environment.is_deployed_in_production() {
+    "production"
+  } else {
+    "development"
+  };
+
+  let builder = PagerBuilder::new()
+      .application_name("seedance2-pro-job".to_string())
+      .environment(environment.to_string())
+      .hostname(hostname.to_string())
+      .service_id(ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB.to_string());
+
+  // If paging is globally disabled, use a NoOp pager regardless of API key.
+  if !is_paging_enabled {
+    warn!("ENABLE_PAGING is false. Pager will be NoOp.");
+    return builder.build_with_worker();
+  }
+
+  let maybe_api_key = env_optional_rootly_api_key();
+
+  match maybe_api_key {
+    Some(api_key) => {
+      info!("Rootly API key found. Configuring pager with Rootly backend.");
+      build_rootly_pager(builder, api_key)
+    }
+    None => {
+      warn!("ROOTLY_API_KEY not set. Pager will not send real pages.");
+      builder.build_with_worker()
+    }
+  }
+}
+
+fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWorker) {
+  let mut rootly_builder = builder
+      .rootly(RootlyApiKey::new(api_key))
+      .urgency_id_high(ROOTLY_URGENCY_ID_HIGH.to_string())
+      .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string())
+      .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string());
+
+  let target_type = env_optional_rootly_notification_target_type();
+  let target_id = env_optional_rootly_notification_target_id();
+
+  if let (Some(t_type), Some(t_id)) = (target_type, target_id) {
+    rootly_builder = rootly_builder.notification_target(t_type, t_id);
+  }
+
+  rootly_builder.build_with_worker()
+}
@@ -0,0 +1 @@
+pub mod build_pager;
@@ -31,6 +31,10 @@ easyenv.workspace = true
 errors.workspace = true
 jobs_common.workspace = true
 mysql_queries.workspace = true
+pager = { path = "../../../lib/pager" }
+rootly_client.workspace = true
+rootly_config.workspace = true
+shared_env_var_config.workspace = true
 server_environment = { path = "../../../lib/server_environment" }

 # External
@@ -7,6 +7,8 @@ use actix_web::{web, HttpRequest, HttpResponse};
 use log::error;

 use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response;
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;

 use crate::http_server::http_server_shared_state::HttpServerSharedState;

@@ -46,7 +48,7 @@ impl std::fmt::Display for HealthCheckError {
 }

 pub async fn get_health_check_handler(
-  _http_request: HttpRequest,
+  http_request: HttpRequest,
  server_state: web::Data<Arc<HttpServerSharedState>>,
 ) -> Result<HttpResponse, HealthCheckError> {
  let job_stats = server_state
@@ -76,6 +78,30 @@ pub async fn get_health_check_handler(
  let is_healthy =
    job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold;

+  if !is_healthy {
+    let notification = NotificationDetailsBuilder::from_title(
+          format!("Health check unhealthy on {}", server_state.hostname))
+        .set_description(Some(format!(
+          "Health check returned unhealthy.\n\n\
+             Hostname: {}\n\
+             Consecutive failure count: {}\n\
+             Total failure count: {}\n\
+             Total success count: {}",
+          server_state.hostname,
+          job_stats.consecutive_failure_count,
+          job_stats.total_failure_count,
+          job_stats.total_success_count,
+        )))
+        .set_urgency(Some(NotificationUrgency::High))
+        .set_http_method(Some(http_request.method().to_string()))
+        .set_http_path(Some(http_request.path().to_string()))
+        .build();
+
+    if let Err(err) = server_state.pager.enqueue_page(notification) {
+      error!("Failed to enqueue health check alert: {:?}", err);
+    }
+  }
+
  let response = HealthCheckResponse {
    success: true,
    is_healthy,
@@ -1,7 +1,10 @@
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;

 #[derive(Clone)]
 pub struct HttpServerSharedState {
  pub job_stats: JobStats,
  pub consecutive_failure_unhealthy_threshold: u64,
+  pub pager: Pager,
+  pub hostname: String,
 }
@@ -8,6 +8,7 @@ use log::info;
 use bootstrap::bootstrap::ContainerEnvironment;
 use errors::AnyhowResult;
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;

 use crate::http_server::endpoints::health_check_handler::get_health_check_handler;
 use crate::http_server::http_server_shared_state::HttpServerSharedState;
@@ -18,12 +19,13 @@ const DEFAULT_NUM_WORKERS: usize = 2;
 pub struct CreateServerArgs {
  pub container_environment: ContainerEnvironment,
  pub job_stats: JobStats,
+  pub pager: Pager,
 }

 pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
  let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS);
  let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?;
-  let _hostname = args.container_environment.hostname.clone();
+  let hostname = args.container_environment.hostname.clone();

  let server_state = HttpServerSharedState {
    job_stats: args.job_stats.clone(),
@@ -31,6 +33,8 @@ pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server> {
      "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD",
      3,
    )?,
+    pager: args.pager,
+    hostname,
  };

  let server_state_arc = web::Data::new(Arc::new(server_state));
@@ -0,0 +1,22 @@
+use log::error;
+use pager::client::pager::Pager;
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
+
+/// Enqueue a pager alert for the error, then return it as `Err`.
+pub fn alert_pager_and_return_err<T>(
+  pager: &Pager,
+  title: &str,
+  err: anyhow::Error,
+) -> anyhow::Result<T> {
+  let notification = NotificationDetailsBuilder::from_error(&err)
+      .set_title(title.to_string())
+      .set_urgency(Some(NotificationUrgency::Medium))
+      .build();
+
+  if let Err(pager_err) = pager.enqueue_page(notification) {
+    error!("Failed to enqueue pager alert: {:?}", pager_err);
+  }
+
+  Err(err)
+}
@@ -7,6 +7,7 @@ use mysql_queries::queries::media_files::thumbnails::list_video_media_files_with
  ListVideoMediaFilesWithoutThumbnailsArgs,
 };

+use crate::job::alert_on_error::alert_pager_and_return_err;
 use crate::job_dependencies::JobDependencies;
 use crate::job::process_single_media_file::process_single_media_file;

@@ -21,6 +22,7 @@ pub async fn main_loop(deps: JobDependencies) {
      }
      Err(err) => {
        error!("Error in video thumbnail batch cycle: {:?}", err);
+        let _ = alert_pager_and_return_err::<()>(&deps.pager, "Video thumbnail batch cycle error", err);
        let _ = deps.job_stats.increment_failure_count();

        // Wait before retrying after a failure.
@@ -1,2 +1,3 @@
+pub mod alert_on_error;
 pub mod main_loop;
 pub mod process_single_media_file;
@@ -1,6 +1,6 @@
 use std::path::PathBuf;

-use log::info;
+use log::{error, info};
 use tempdir::TempDir;

 use bucket_paths::legacy::typified_paths::public::media_files::bucket_file_path::MediaFileBucketPath;
@@ -10,6 +10,7 @@ use ffmpeg_utils::ffmpeg::ffmpeg_video_gif_preview::{ffmpeg_video_gif_preview, F
 use mysql_queries::queries::media_files::thumbnails::list_video_media_files_without_thumbnails_for_job::VideoMediaFileWithoutThumbnail;
 use mysql_queries::queries::media_files::thumbnails::update_video_media_file_with_thumbnail::update_video_media_file_with_thumbnail;

+use crate::job::alert_on_error::alert_pager_and_return_err;
 use crate::job_dependencies::JobDependencies;

 /// A downloaded video file alongside its owning temp directory.
@@ -32,7 +33,13 @@ pub async fn process_single_media_file(
    media_file.created_at,
  );

-  let downloaded = download_video(deps, media_file).await?;
+  let downloaded = match download_video(deps, media_file).await {
+    Ok(d) => d,
+    Err(err) => {
+      error!("Failed to download video for {}: {:?}", media_file.token.as_str(), err);
+      return alert_pager_and_return_err(&deps.pager, "Video download failed", err);
+    }
+  };

  info!(
    "Downloaded video to {:?}. Generating thumbnails for {:?}.",
@@ -46,53 +53,62 @@ pub async fn process_single_media_file(
  // Generate jpg thumbnail
  let jpg_path = downloaded.temp_dir.path().join("thumbnail.jpg");

-  ffmpeg_video_first_frame_to_jpg_thumbnail(
+  if let Err(err) = ffmpeg_video_first_frame_to_jpg_thumbnail(
    FfmpegVideoFirstFrameToJpgThumbnailArgs {
      input_video_path: &downloaded.file_path,
      output_jpg_path: &jpg_path,
    },
-  )?;
+  ) {
+    error!("Failed to generate JPG thumbnail for {}: {:?}", media_file.token.as_str(), err);
+    return alert_pager_and_return_err(&deps.pager, "JPG thumbnail generation failed", err.into());
+  }

  info!("Generated JPG thumbnail for {}", media_file.token.as_str());

  let jpg_object_path = format!("{video_object_path}{VIDEO_STATIC_JPG_THUMBNAIL_SUFFIX}");
-  
-  deps
-      .public_bucket_client
-      .upload_filename(&jpg_object_path, &jpg_path)
-      .await?;
+
+  if let Err(err) = deps.public_bucket_client.upload_filename(&jpg_object_path, &jpg_path).await {
+    error!("Failed to upload JPG thumbnail for {}: {:?}", media_file.token.as_str(), err);
+    return alert_pager_and_return_err(&deps.pager, "JPG thumbnail upload failed", err);
+  }

  info!("Uploaded JPG thumbnail to {}", jpg_object_path);

  // Generate gif thumbnail
  let gif_path = downloaded.temp_dir.path().join("thumbnail.gif");

-  ffmpeg_video_gif_preview(
+  if let Err(err) = ffmpeg_video_gif_preview(
    FfmpegVideoGifPreviewArgs {
      input_video_path: &downloaded.file_path,
      output_gif_path: &gif_path,
    },
-  )?;
+  ) {
+    error!("Failed to generate GIF preview for {}: {:?}", media_file.token.as_str(), err);
+    return alert_pager_and_return_err(&deps.pager, "GIF preview generation failed", err.into());
+  }

  info!("Generated GIF preview for {}", media_file.token.as_str());

  let gif_object_path = format!("{video_object_path}{VIDEO_ANIMATED_GIF_THUMBNAIL_SUFFIX}");

-  deps
-    .public_bucket_client
-    .upload_filename(&gif_object_path, &gif_path)
-    .await?;
+  if let Err(err) = deps.public_bucket_client.upload_filename(&gif_object_path, &gif_path).await {
+    error!("Failed to upload GIF preview for {}: {:?}", media_file.token.as_str(), err);
+    return alert_pager_and_return_err(&deps.pager, "GIF preview upload failed", err);
+  }

  info!("Uploaded GIF preview to {}", gif_object_path);

  info!("Marking thumbnail job for {:?} done", media_file.token);

  // Mark the media file as having a thumbnail in the database.
-  update_video_media_file_with_thumbnail(
+  if let Err(err) = update_video_media_file_with_thumbnail(
    &media_file.token,
    CURRENT_VIDEO_THUMBNAIL_VERSION,
    &deps.mysql_pool,
-  ).await?;
+  ).await {
+    error!("Failed to update thumbnail version for {}: {:?}", media_file.token.as_str(), err);
+    return alert_pager_and_return_err(&deps.pager, "Thumbnail DB update failed", err.into());
+  }

  info!(
    "Updated thumbnail version for media file {:?} (id: {}, created at: {})",
@@ -3,6 +3,7 @@ use std::path::PathBuf;
 use cloud_storage::bucket_client::BucketClient;
 use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
 use jobs_common::job_stats::JobStats;
+use pager::client::pager::Pager;
 use server_environment::ServerEnvironment;
 use sqlx::MySqlPool;

@@ -45,4 +46,7 @@ pub struct JobDependencies {

  /// Set to `true` from another thread to trigger graceful shutdown.
  pub application_shutdown: RelaxedAtomicBool,
+
+  /// Pager client for sending alerts.
+  pub pager: Pager,
 }
@@ -28,10 +28,12 @@ use server_environment::ServerEnvironment;
 use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs};
 use crate::job::main_loop::main_loop;
 use crate::job_dependencies::{JobDependencies, ShardInfo};
+use crate::startup::build_pager::build_pager;

 pub mod http_server;
 pub mod job;
 pub mod job_dependencies;
+pub mod startup;

 // Bucket config
 const ENV_ACCESS_KEY: &str = "ACCESS_KEY";
@@ -40,7 +42,7 @@ const ENV_REGION_NAME: &str = "REGION_NAME";
 const ENV_PUBLIC_BUCKET_NAME: &str = "PUBLIC_BUCKET_NAME";
 const ENV_S3_ENDPOINT: &str = "S3_COMPATIBLE_ENDPOINT_URL";

-#[actix_web::main]
+#[tokio::main]
 async fn main() -> AnyhowResult<()> {

  let container_environment = bootstrap(BootstrapArgs {
@@ -147,12 +149,27 @@ async fn main() -> AnyhowResult<()> {
    }
  };

+  let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname);
+
+  info!("Spawning pager worker.");
+
+  // NB: The pager worker uses Condvar::wait() which is a blocking syscall.
+  // It must run on a dedicated OS thread, not a tokio task, to avoid blocking
+  // the tokio runtime.
+  std::thread::spawn(move || {
+    let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime");
+    rt.block_on(pager_worker.run());
+  });
+
  let application_shutdown = RelaxedAtomicBool::new(false);
  let job_stats = JobStats::new();

+  let pager_for_shutdown = pager.clone();
+
  let create_server_args = CreateServerArgs {
    container_environment: container_environment.clone(),
    job_stats: job_stats.clone(),
+    pager: pager.clone(),
  };

  let job_dependencies = JobDependencies {
@@ -168,8 +185,10 @@ async fn main() -> AnyhowResult<()> {
    temp_dir,
    shard_info,
    application_shutdown: application_shutdown.clone(),
+    pager,
  };

+  // HTTP server runs on a separate OS thread with its own actix System.
  std::thread::spawn(move || {
    let actix_runtime = actix_web::rt::System::new();
    let http_server_handle = launch_http_server(create_server_args);
@@ -177,11 +196,30 @@ async fn main() -> AnyhowResult<()> {
    actix_runtime.block_on(http_server_handle)
      .expect("HTTP server should not exit.");

-    warn!("Server thread is shut down.");
-    application_shutdown.set(true);
+    warn!("HTTP server thread is shut down.");
+  });
+
+  // Listen for SIGTERM / Ctrl-C to trigger graceful shutdown.
+  let application_shutdown_for_signal = application_shutdown.clone();
+
+  tokio::spawn(async move {
+    match tokio::signal::ctrl_c().await {
+      Ok(()) => {
+        info!("Received shutdown signal. Shutting down...");
+        application_shutdown_for_signal.set(true);
+      }
+      Err(err) => {
+        warn!("Error listening for shutdown signal: {:?}", err);
+      }
+    }
  });

  main_loop(job_dependencies).await;

+  info!("Shutting down pager worker...");
+  pager_for_shutdown.shutdown_worker();
+
+  info!("Video thumbnail job exiting.");
+
  Ok(())
 }
@@ -0,0 +1,65 @@
+use log::{info, warn};
+use pager::client::pager::Pager;
+use pager::client::pager_builder::PagerBuilder;
+use pager::worker::pager_worker::PagerWorker;
+use rootly_client::creds::rootly_api_key::RootlyApiKey;
+use rootly_config::services::ROOTLY_SERVICE_ID_VIDEO_THUMBNAIL_JOB;
+use rootly_config::urgencies::{ROOTLY_URGENCY_ID_HIGH, ROOTLY_URGENCY_ID_LOW, ROOTLY_URGENCY_ID_MEDIUM};
+use shared_env_var_config::paging::{env_enable_paging_default_false, env_optional_rootly_api_key, env_optional_rootly_notification_target_id, env_optional_rootly_notification_target_type};
+
+pub fn build_pager(
+  server_environment: server_environment::ServerEnvironment,
+  hostname: &str,
+) -> (Pager, PagerWorker) {
+  let is_paging_enabled = env_enable_paging_default_false();
+
+  info!("Paging enabled: {}", is_paging_enabled);
+
+  let environment = if server_environment.is_deployed_in_production() {
+    "production"
+  } else {
+    "development"
+  };
+
+  let builder = PagerBuilder::new()
+      .application_name("video-thumbnail-job".to_string())
+      .environment(environment.to_string())
+      .hostname(hostname.to_string())
+      .service_id(ROOTLY_SERVICE_ID_VIDEO_THUMBNAIL_JOB.to_string());
+
+  // If paging is globally disabled, use a NoOp pager regardless of API key.
+  if !is_paging_enabled {
+    warn!("ENABLE_PAGING is false. Pager will be NoOp.");
+    return builder.build_with_worker();
+  }
+
+  let maybe_api_key = env_optional_rootly_api_key();
+
+  match maybe_api_key {
+    Some(api_key) => {
+      info!("Rootly API key found. Configuring pager with Rootly backend.");
+      build_rootly_pager(builder, api_key)
+    }
+    None => {
+      warn!("ROOTLY_API_KEY not set. Pager will not send real pages.");
+      builder.build_with_worker()
+    }
+  }
+}
+
+fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWorker) {
+  let mut rootly_builder = builder
+      .rootly(RootlyApiKey::new(api_key))
+      .urgency_id_high(ROOTLY_URGENCY_ID_HIGH.to_string())
+      .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string())
+      .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string());
+
+  let target_type = env_optional_rootly_notification_target_type();
+  let target_id = env_optional_rootly_notification_target_id();
+
+  if let (Some(t_type), Some(t_id)) = (target_type, target_id) {
+    rootly_builder = rootly_builder.notification_target(t_type, t_id);
+  }
+
+  rootly_builder.build_with_worker()
+}
@@ -0,0 +1 @@
+pub mod build_pager;
@@ -68,6 +68,8 @@ mysql_queries.workspace = true
 pager.workspace = true
 password = { path = "../../../lib/password" }
 rootly_client.workspace = true
+rootly_config.workspace = true
+shared_env_var_config.workspace = true
 primitives = { path = "../../../lib/data/primitives" }
 prompt_engineering = { path = "../../../lib/deprecated/machine_learning/prompt_engineering" }
 redis_caching = { path = "../../../lib/caching/redis_caching" }
@@ -176,7 +176,8 @@ use crate::http_server::endpoints::users::edit_email_handler::*;
 use crate::http_server::endpoints::users::edit_username_handler::*;
 use crate::http_server::endpoints::users::get_profile_handler::*;
 use crate::http_server::endpoints::users::google_sso::google_sso_handler::*;
-use crate::http_server::endpoints::users::login_handler::*;
+use artcraft_api_defs::users::login::{LoginRequest, LoginSuccessResponse, LoginErrorType};
+use crate::http_server::endpoints::users::login_handler::LoginErrorResponse;
 use crate::http_server::endpoints::users::logout_handler::*;
 use crate::http_server::endpoints::users::session_info_handler::*;
 use crate::http_server::endpoints::users::session_token_info_handler::*;
@@ -1,8 +1,3 @@
-// NB: Incrementally getting rid of build warnings...
-#![forbid(unused_imports)]
-#![forbid(unused_mut)]
-#![forbid(unused_variables)]
-
 use std::fmt;
 use std::fmt::Formatter;

@@ -21,25 +16,11 @@ use sqlx::MySqlPool;
 use tokens::tokens::user_sessions::UserSessionToken;
 use utoipa::ToSchema;

+use artcraft_api_defs::users::login::{LoginErrorType, LoginRequest, LoginSuccessResponse};
+use password::errors::password_confirm_error::PasswordConfirmError;
 use crate::http_server::session::lookup::user_session_feature_flags::UserSessionFeatureFlags;
 use crate::util::enroll_in_studio::enroll_in_studio;

-#[derive(Deserialize, ToSchema)]
-pub struct LoginRequest {
-  pub username_or_email: String,
-  pub password: String,
-}
-
-#[derive(Serialize, ToSchema)]
-pub struct LoginSuccessResponse {
-  pub success: bool,
-
-  /// A signed session that can be sent as a header, bypassing cookies.
-  /// This is useful for API clients that don't support cookies or Google
-  /// browsers killing cross-domain cookies.
-  pub signed_session: String,
-}
-
 #[derive(Serialize, Debug, ToSchema)]
 pub struct LoginErrorResponse {
  pub success: bool,
@@ -47,29 +28,6 @@ pub struct LoginErrorResponse {
  pub error_message: String,
 }

-#[derive(Copy, Clone, Debug, Serialize, ToSchema)]
-pub enum LoginErrorType {
-  InvalidCredentials,
-  ServerError,
-}
-
-impl LoginErrorResponse {
-  fn invalid_credentials() -> Self {
-    Self {
-      success: false,
-      error_type: LoginErrorType::InvalidCredentials,
-      error_message: "invalid credentials".to_string()
-    }
-  }
-  fn server_error() -> Self {
-    Self {
-      success: false,
-      error_type: LoginErrorType::ServerError,
-      error_message: "server error".to_string()
-    }
-  }
-}
-
 // NB: Not using DeriveMore since Clion doesn't understand it.
 impl fmt::Display for LoginErrorResponse {
  fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
@@ -81,6 +39,7 @@ impl ResponseError for LoginErrorResponse {
  fn status_code(&self) -> StatusCode {
    match self.error_type {
      LoginErrorType::InvalidCredentials => StatusCode::UNAUTHORIZED,
+      LoginErrorType::AccountNeedsPassword => StatusCode::UNAUTHORIZED,
      LoginErrorType::ServerError=> StatusCode::INTERNAL_SERVER_ERROR,
    }
  }
@@ -90,6 +49,32 @@ impl ResponseError for LoginErrorResponse {
  }
 }

+impl LoginErrorResponse {
+  fn invalid_credentials() -> Self {
+    Self {
+      success: false,
+      error_type: LoginErrorType::InvalidCredentials,
+      error_message: "invalid credentials".to_string()
+    }
+  }
+  
+  fn server_error() -> Self {
+    Self {
+      success: false,
+      error_type: LoginErrorType::ServerError,
+      error_message: "server error".to_string()
+    }
+  }
+
+  fn account_without_password() -> Self {
+    Self {
+      success: false,
+      error_type: LoginErrorType::AccountNeedsPassword,
+      error_message: "account was created without a password; please try password reset".to_string()
+    }
+  }
+}
+
 #[utoipa::path(
  post,
  tag = "Users",
@@ -146,6 +131,10 @@ pub async fn login_handler(
  };

  match bcrypt_confirm_password(request.password.clone(), &actual_hash) {
+    Err(PasswordConfirmError::HashIsSentinelValue) => {
+      warn!("Attempting to log in with a passwordless account");
+      return Err(LoginErrorResponse::account_without_password());
+    }
    Err(e) => {
      warn!("Login hash comparison error: {:?}", e);
      return Err(LoginErrorResponse::server_error());
@@ -1,11 +1,8 @@
 use std::fmt;
 use std::sync::Arc;

-use crate::http_server::endpoints::webhooks::handle_image_payload::handle_image_payload;
-use crate::http_server::endpoints::webhooks::handle_images_payload::handle_images_payload;
-use crate::http_server::endpoints::webhooks::handle_model_glb_payload::handle_model_glb_payload;
-use crate::http_server::endpoints::webhooks::handle_model_mesh_payload::handle_model_mesh_payload;
-use crate::http_server::endpoints::webhooks::handle_video_payload::handle_video_payload;
+use crate::http_server::endpoints::webhooks::process_failure::handle_failed_fal_webhook::handle_failed_fal_webhook;
+use crate::http_server::endpoints::webhooks::process_success::handle_successful_fal_webhook::handle_sucessful_fal_webhook;
 use crate::state::server_state::ServerState;
 use actix_web::error::ResponseError;
 use actix_web::http::StatusCode;
@@ -14,33 +11,22 @@ use actix_web::{web, HttpRequest, HttpResponse};
 use anyhow::Error;
 use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
 use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
-use log::{info, warn};
-use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id::get_inference_job_by_fal_id;
-use mysql_queries::queries::generic_inference::fal::mark_fal_generic_inference_job_successfully_done::{mark_fal_generic_inference_job_successfully_done, MarkJobArgs};
+use log::{error, info, warn};
+use pager::notification::notification_details_builder::NotificationDetailsBuilder;
+use pager::notification::notification_urgency::NotificationUrgency;
 use serde_json::Value;
 use utoipa::ToSchema;

-// 1. tauri --> hit endpoint to enqueue
-//
-// 2. webhook 
-//  - upload media file
-//  - update job record with media token + status
-//
-// 3. tauri --> storyteller jobs endpoint polls
-//  - alert frontend of completion
-//
-// 4. javascript polls tauri  (backend removal)
-
 // TODO(bt, 2025-06-03): Handle webhook crypto authentication
 #[derive(Debug, Deserialize, ToSchema)]
 pub struct FalWebhookRequest {
  pub status: FalWebhookStatus,
-  
+
  pub request_id: Option<String>,
  pub gateway_request_id: Option<String>,

  pub error: Option<String>,
-  
+
  /// Payload of the webhook, if any.
  pub payload: Option<Value>,
 }
@@ -116,75 +102,55 @@ pub async fn fal_webhook_handler(
 ) -> Result<Json<SimpleGenericJsonSuccess>, FalWebhookError> {

  info!("Received FAL webhook body: {:?}", request);
-  
+
  let request_id = request.request_id
      .as_deref()
      .ok_or_else(|| FalWebhookError::BadInput("Missing request_id".to_string()))?;
-  
+
+  info!("FAL webhook request_id: {} (status: {:?})", request_id, request.status);
+
+  // TODO(bt): Longer term, we should just use `fal_client` to parse webhooks and add lots of integration tests
+  //  across dozens of real messages.
  let payload = request.payload
      .as_ref()
-      .ok_or_else(|| FalWebhookError::BadInput("Missing payload".to_string()))?;
-  
-  let db_result = get_inference_job_by_fal_id(
-    request_id,
-    &server_state.mysql_pool,
-  ).await;
-  
-  let job = match db_result {
-    Ok(Some(record)) => record,
-    Ok(None) => {
-      warn!("Could not find job record by fal request id: {:?} ; request payload = {:?}", request_id, request);
-      return Err(FalWebhookError::NotFound)
-    },
-    Err(err) => {
-      warn!("Error querying job record: {:?}", err);
-      return Err(FalWebhookError::ServerError);
+      .ok_or_else(|| {
+        warn!("FAL webhook missing payload for request_id: {}", request_id);
+        FalWebhookError::BadInput("Missing payload".to_string())
+      })?;
+
+  let result = match request.status {
+    FalWebhookStatus::Ok => {
+      handle_sucessful_fal_webhook(&server_state, request_id, payload).await
+    }
+    FalWebhookStatus::Error => {
+      handle_failed_fal_webhook(
+        &server_state,
+        request_id,
+        payload,
+        request.error.as_deref(),
+      ).await
    }
  };

-  info!("Fal webhook job record: {:?}", job);
+  if let Err(ref err) = result {
+    error!("FAL webhook error for request_id {}: {:?}", request_id, err);

-  let mut maybe_media_token = None;
-  let mut maybe_batch_token = None;
+    let notification = NotificationDetailsBuilder::from_error(err)
+        .set_title("FAL webhook processing failed".to_string())
+        .set_description(Some(format!(
+          "FAL webhook failed for request_id: {}\n\nError: {:?}",
+          request_id, err,
+        )))
+        .set_third_party_id(Some(request_id.to_string()))
+        .set_urgency(Some(NotificationUrgency::High))
+        .set_http_method(Some(http_request.method().to_string()))
+        .set_http_path(Some(http_request.path().to_string()))
+        .build();

-  if let Some(payload_obj) = payload.as_object() {
-    if payload_obj.contains_key("image") {
-      info!("Handling image payload for job: {:?}", job.job_token);
-      let token = handle_image_payload(payload_obj, &job, &server_state).await?;
-      maybe_media_token = Some(token);
-    } else if payload_obj.contains_key("images") {
-      (maybe_media_token, maybe_batch_token) = handle_images_payload(payload_obj, &job, &server_state).await?;
-    } else if payload_obj.contains_key("video") {
-      info!("Handling video payload for job: {:?}", job.job_token);
-      let token = handle_video_payload(payload_obj, &job, &server_state).await?;
-      maybe_media_token = Some(token);
-    } else if payload_obj.contains_key("model_glb") {
-      info!("Handling model_glb payload for job: {:?}", job.job_token);
-      let token = handle_model_glb_payload(payload_obj, &job, &server_state).await?;
-      maybe_media_token = Some(token);
-    } else if payload_obj.contains_key("model_mesh") {
-      info!("Handling model_mesh payload for job: {:?}", job.job_token);
-      let token = handle_model_mesh_payload(payload_obj, &job, &server_state).await?;
-      maybe_media_token = Some(token);
+    if let Err(pager_err) = server_state.pager.enqueue_page(notification) {
+      error!("Failed to enqueue FAL webhook pager alert: {:?}", pager_err);
    }
  }
-  
-  if let Some(media_token) = maybe_media_token {
-    info!("Media file token: {:?}", media_token);
-    // TODO: Update job metadata.
-    mark_fal_generic_inference_job_successfully_done(MarkJobArgs {
-      job_token: &job.job_token,
-      media_file_token: &media_token,
-      maybe_batch_token: maybe_batch_token.as_ref(),
-      mysql_executor: &server_state.mysql_pool,
-      phantom: Default::default(),
-    }).await.map_err(|err| {
-      warn!("Error marking job as successfully done: {:?}", err);
-      FalWebhookError::ServerError
-    })?;
-  } else {
-    warn!("No media token found in payload for job: {:?}", job.job_token);
-  }

-  Ok(SimpleGenericJsonSuccess::wrapped(true))
+  result
 }
@@ -1,6 +1,3 @@
 pub mod fal_webhook_handler;
-pub mod handle_image_payload;
-pub mod handle_images_payload;
-pub mod handle_model_mesh_payload;
-pub mod handle_video_payload;
-pub mod handle_model_glb_payload;
+pub (super) mod process_failure;
+pub (super) mod process_success;
@@ -0,0 +1,142 @@
+use serde_json::Value;
+
+/// A single error detail item from a FAL webhook error payload.
+///
+/// See: https://fal.ai/docs/documentation/model-apis/errors
+#[derive(Debug, Clone)]
+pub struct FalErrorDetail {
+  /// Human-readable error message. Should not be parsed for logic.
+  pub msg: Option<String>,
+
+  /// Machine-readable error type (e.g. "image_too_small", "image_too_large").
+  pub error_type: Option<String>,
+}
+
+/// Attempt to parse the "detail" array from a FAL webhook payload.
+///
+/// The payload may contain a "detail" key with an array of error objects,
+/// each having optional "msg" and "type" fields.
+///
+/// Returns an empty vec if the "detail" key is missing or not an array.
+pub fn parse_fal_error_details(payload: &Value) -> Vec<FalErrorDetail> {
+  let detail_array = match payload.get("detail").and_then(|v| v.as_array()) {
+    Some(arr) => arr,
+    None => return Vec::new(),
+  };
+
+  detail_array
+    .iter()
+    .map(|item| FalErrorDetail {
+      msg: item.get("msg").and_then(|v| v.as_str()).map(|s| s.to_string()),
+      error_type: item.get("type").and_then(|v| v.as_str()).map(|s| s.to_string()),
+    })
+    .collect()
+}
+
+/// Build a human-readable summary of the error details for logging and alerting.
+pub fn summarize_fal_error_details(details: &[FalErrorDetail]) -> String {
+  if details.is_empty() {
+    return "No error details provided.".to_string();
+  }
+
+  details
+    .iter()
+    .enumerate()
+    .map(|(i, detail)| {
+      let error_type = detail.error_type.as_deref().unwrap_or("unknown");
+      let msg = detail.msg.as_deref().unwrap_or("no message");
+      format!("[{}] type={}, msg={}", i + 1, error_type, msg)
+    })
+    .collect::<Vec<_>>()
+    .join("\n")
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use serde_json::json;
+
+  #[test]
+  fn parse_typical_error_detail() {
+    let payload = json!({
+      "detail": [
+        {
+          "loc": ["body", "image_url"],
+          "msg": "Image too small",
+          "type": "image_too_small",
+          "url": "https://docs.fal.ai/errors/#image_too_small",
+          "ctx": { "min_height": 512, "min_width": 512 }
+        }
+      ]
+    });
+
+    let details = parse_fal_error_details(&payload);
+    assert_eq!(details.len(), 1);
+    assert_eq!(details[0].msg.as_deref(), Some("Image too small"));
+    assert_eq!(details[0].error_type.as_deref(), Some("image_too_small"));
+  }
+
+  #[test]
+  fn parse_multiple_error_details() {
+    let payload = json!({
+      "detail": [
+        { "msg": "First error", "type": "err_one" },
+        { "msg": "Second error", "type": "err_two" }
+      ]
+    });
+
+    let details = parse_fal_error_details(&payload);
+    assert_eq!(details.len(), 2);
+    assert_eq!(details[0].error_type.as_deref(), Some("err_one"));
+    assert_eq!(details[1].error_type.as_deref(), Some("err_two"));
+  }
+
+  #[test]
+  fn parse_missing_detail_field() {
+    let payload = json!({ "images": [] });
+    let details = parse_fal_error_details(&payload);
+    assert!(details.is_empty());
+  }
+
+  #[test]
+  fn parse_detail_not_array() {
+    let payload = json!({ "detail": "some string" });
+    let details = parse_fal_error_details(&payload);
+    assert!(details.is_empty());
+  }
+
+  #[test]
+  fn parse_detail_with_missing_fields() {
+    let payload = json!({
+      "detail": [
+        { "msg": "Only message" },
+        { "type": "only_type" },
+        {}
+      ]
+    });
+
+    let details = parse_fal_error_details(&payload);
+    assert_eq!(details.len(), 3);
+    assert_eq!(details[0].msg.as_deref(), Some("Only message"));
+    assert_eq!(details[0].error_type, None);
+    assert_eq!(details[1].msg, None);
+    assert_eq!(details[1].error_type.as_deref(), Some("only_type"));
+    assert_eq!(details[2].msg, None);
+    assert_eq!(details[2].error_type, None);
+  }
+
+  #[test]
+  fn summarize_empty() {
+    assert_eq!(summarize_fal_error_details(&[]), "No error details provided.");
+  }
+
+  #[test]
+  fn summarize_single() {
+    let details = vec![FalErrorDetail {
+      msg: Some("Image too small".to_string()),
+      error_type: Some("image_too_small".to_string()),
+    }];
+    let summary = summarize_fal_error_details(&details);
+    assert_eq!(summary, "[1] type=image_too_small, msg=Image too small");
+  }
+}
@@ -0,0 +1,122 @@
+use actix_web::web::Json;
+use log::{error, info, warn};
+use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id::get_inference_job_by_fal_id;
+use mysql_queries::queries::generic_inference::job::mark_job_failed_by_token::{mark_job_failed_by_token, MarkJobFailedByTokenArgs};
+use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
+use serde_json::Value;
+use enums::by_table::generic_inference_jobs::frontend_failure_category::FrontendFailureCategory;
+use crate::http_server::endpoints::webhooks::fal_webhook_handler::FalWebhookError;
+use crate::http_server::endpoints::webhooks::process_failure::fal_error_detail::{FalErrorDetail, parse_fal_error_details, summarize_fal_error_details};
+use crate::state::server_state::ServerState;
+
+/// Map FAL error type strings to frontend failure categories.
+///
+/// Returns `GenerationFailed` as the default if no error details match a known type.
+fn guess_failure_category(details: &[FalErrorDetail]) -> FrontendFailureCategory {
+  for detail in details {
+    if let Some(error_type) = &detail.error_type {
+      match error_type.as_str() {
+        "content_policy_violation" => return FrontendFailureCategory::RuleBansUserContent,
+        "face_detection_error" => return FrontendFailureCategory::FaceNotDetected,
+        // These all map to GenerationFailed for now, but are listed explicitly
+        // so we can refine them later.
+        "no_media_generated"
+        | "image_too_small"
+        | "image_too_large"
+        | "image_load_error"
+        | "file_download_error"
+        | "file_too_large" => return FrontendFailureCategory::GenerationFailed,
+        _ => {}
+      }
+    }
+  }
+  FrontendFailureCategory::GenerationFailed
+}
+
+/// Handle a FAL webhook with status ERROR.
+///
+/// Parses the error details from the payload, looks up the job, and marks it as failed.
+pub async fn handle_failed_fal_webhook(
+  server_state: &ServerState,
+  request_id: &str,
+  payload: &Value,
+  maybe_top_level_error: Option<&str>,
+) -> Result<Json<SimpleGenericJsonSuccess>, FalWebhookError> {
+
+  // Parse the "detail" array from the payload.
+  let error_details = parse_fal_error_details(payload);
+  let error_summary = summarize_fal_error_details(&error_details);
+
+  info!(
+    "FAL webhook ERROR for request_id {}: top_level_error={:?}, details=[{}]",
+    request_id,
+    maybe_top_level_error,
+    error_summary,
+  );
+
+  // Look up the job record.
+  let job = match get_inference_job_by_fal_id(request_id, &server_state.mysql_pool).await {
+    Ok(Some(record)) => record,
+    Ok(None) => {
+      warn!("Could not find job record by fal request_id: {}", request_id);
+      return Err(FalWebhookError::NotFound);
+    }
+    Err(err) => {
+      error!("Error querying job record for request_id {}: {:?}", request_id, err);
+      return Err(FalWebhookError::ServerError);
+    }
+  };
+
+  // Build a failure reason from the error details or the top-level error string.
+  let public_failure_reason = if !error_details.is_empty() {
+    error_details.iter()
+        .filter_map(|d| d.msg.as_deref())
+        .collect::<Vec<_>>()
+        .join("; ")
+  } else if let Some(top_level) = maybe_top_level_error {
+    top_level.to_string()
+  } else {
+    "Unknown FAL error".to_string()
+  };
+
+  let internal_failure_reason = format!(
+    "FAL request_id={}, top_level_error={:?}, details=[{}]",
+    request_id,
+    maybe_top_level_error,
+    error_summary,
+  );
+
+  let failure_category = guess_failure_category(&error_details);
+
+  info!(
+    "Marking job {} as failed for request_id {}. Category: {:?}, Reason: {}",
+    job.job_token.as_str(),
+    request_id,
+    failure_category,
+    public_failure_reason,
+  );
+
+  if let Err(err) = mark_job_failed_by_token(MarkJobFailedByTokenArgs {
+    pool: &server_state.mysql_pool,
+    job_token: &job.job_token,
+    maybe_public_failure_reason: Some(&public_failure_reason),
+    internal_debugging_failure_reason: &internal_failure_reason,
+    maybe_frontend_failure_category: Some(failure_category),
+  }).await {
+    error!(
+      "Error marking job {} as failed for request_id {}: {:?}",
+      job.job_token.as_str(),
+      request_id,
+      err,
+    );
+    return Err(FalWebhookError::ServerError);
+  }
+
+  info!(
+    "Job {} marked as failed for request_id {}.",
+    job.job_token.as_str(),
+    request_id,
+  );
+
+  Ok(SimpleGenericJsonSuccess::wrapped(true))
+}
@@ -0,0 +1,2 @@
+pub mod fal_error_detail;
+pub mod handle_failed_fal_webhook;
@@ -0,0 +1,84 @@
+use actix_web::web::Json;
+use log::{info, warn};
+use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id::get_inference_job_by_fal_id;
+use mysql_queries::queries::generic_inference::fal::mark_fal_generic_inference_job_successfully_done::{mark_fal_generic_inference_job_successfully_done, MarkJobArgs};
+use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess;
+use serde_json::Value;
+
+use crate::http_server::endpoints::webhooks::fal_webhook_handler::FalWebhookError;
+use crate::state::server_state::ServerState;
+
+use super::process_image_payload::process_image_payload;
+use super::process_images_payload::process_images_payload;
+use super::process_model_glb_payload::process_model_glb_payload;
+use super::process_model_mesh_payload::process_model_mesh_payload;
+use super::process_video_payload::process_video_payload;
+
+pub async fn handle_sucessful_fal_webhook(
+  server_state: &ServerState,
+  request_id: &str,
+  payload: &Value,
+) -> Result<Json<SimpleGenericJsonSuccess>, FalWebhookError> {
+
+  let db_result = get_inference_job_by_fal_id(
+    request_id,
+    &server_state.mysql_pool,
+  ).await;
+
+  let job = match db_result {
+    Ok(Some(record)) => record,
+    Ok(None) => {
+      warn!("Could not find job record by fal request_id: {}", request_id);
+      return Err(FalWebhookError::NotFound)
+    },
+    Err(err) => {
+      warn!("Error querying job record for request_id {}: {:?}", request_id, err);
+      return Err(FalWebhookError::ServerError);
+    }
+  };
+
+  info!("Fal webhook job record for request_id {}: {:?}", request_id, job);
+
+  let mut maybe_media_token = None;
+  let mut maybe_batch_token = None;
+
+  if let Some(payload_obj) = payload.as_object() {
+    if payload_obj.contains_key("image") {
+      info!("Handling image payload for request_id {} / job {:?}", request_id, job.job_token);
+      let token = process_image_payload(payload_obj, &job, server_state).await?;
+      maybe_media_token = Some(token);
+    } else if payload_obj.contains_key("images") {
+      (maybe_media_token, maybe_batch_token) = process_images_payload(payload_obj, &job, server_state).await?;
+    } else if payload_obj.contains_key("video") {
+      info!("Handling video payload for request_id {} / job {:?}", request_id, job.job_token);
+      let token = process_video_payload(payload_obj, &job, server_state).await?;
+      maybe_media_token = Some(token);
+    } else if payload_obj.contains_key("model_glb") {
+      info!("Handling model_glb payload for request_id {} / job {:?}", request_id, job.job_token);
+      let token = process_model_glb_payload(payload_obj, &job, server_state).await?;
+      maybe_media_token = Some(token);
+    } else if payload_obj.contains_key("model_mesh") {
+      info!("Handling model_mesh payload for request_id {} / job {:?}", request_id, job.job_token);
+      let token = process_model_mesh_payload(payload_obj, &job, server_state).await?;
+      maybe_media_token = Some(token);
+    }
+  }
+
+  if let Some(media_token) = maybe_media_token {
+    info!("Media file token for request_id {}: {:?}", request_id, media_token);
+    mark_fal_generic_inference_job_successfully_done(MarkJobArgs {
+      job_token: &job.job_token,
+      media_file_token: &media_token,
+      maybe_batch_token: maybe_batch_token.as_ref(),
+      mysql_executor: &server_state.mysql_pool,
+      phantom: Default::default(),
+    }).await.map_err(|err| {
+      warn!("Error marking job as successfully done for request_id {}: {:?}", request_id, err);
+      FalWebhookError::ServerError
+    })?;
+  } else {
+    warn!("No media token found in payload for request_id {} / job {:?}", request_id, job.job_token);
+  }
+
+  Ok(SimpleGenericJsonSuccess::wrapped(true))
+}
@@ -0,0 +1,6 @@
+pub mod handle_successful_fal_webhook;
+mod process_image_payload;
+mod process_images_payload;
+mod process_model_glb_payload;
+mod process_model_mesh_payload;
+mod process_video_payload;
@@ -29,7 +29,7 @@ pub struct FalWebhookImage {
  pub url: Option<String>,
 }

-pub async fn handle_image_payload(
+pub async fn process_image_payload(
  payload: &Map<String, Value>,
  job: &FalJobDetails,
  server_state: &ServerState,
@@ -30,7 +30,7 @@ pub struct FalWebhookImage {
  pub url: Option<String>,
 }

-pub async fn handle_images_payload(
+pub async fn process_images_payload(
  payload: &Map<String, Value>,
  job: &FalJobDetails,
  server_state: &ServerState,
@@ -63,7 +63,7 @@ pub struct FalMediaPayload {
  pub url: Option<String>,
 }

-pub async fn handle_model_glb_payload(
+pub async fn process_model_glb_payload(
  payload: &Map<String, Value>,
  job: &FalJobDetails,
  server_state: &ServerState,
@@ -25,7 +25,7 @@ pub struct FalWebhookModelMesh {
  pub url: Option<String>,
 }

-pub async fn handle_model_mesh_payload(
+pub async fn process_model_mesh_payload(
  payload: &Map<String, Value>,
  job: &FalJobDetails,
  server_state: &ServerState,
@@ -31,7 +31,7 @@ pub struct FalWebhookVideo {
  pub url: Option<String>,
 }

-pub async fn handle_video_payload(
+pub async fn process_video_payload(
  payload: &Map<String, Value>,
  job: &FalJobDetails,
  server_state: &ServerState,
@@ -1,17 +1,12 @@
+use crate::state::flags::paging_flags::PagingFlags;
 use log::{info, warn};
 use pager::client::pager::Pager;
 use pager::client::pager_builder::PagerBuilder;
 use pager::worker::pager_worker::PagerWorker;
 use rootly_client::creds::rootly_api_key::RootlyApiKey;
-
-use crate::state::flags::paging_flags::PagingFlags;
-
-/// `storyteller-web` rootly service ID
-const ROOTLY_SERVICE_ID: &str = "9ebebb09-0c56-4c8a-8f57-d5f0f85f3f16";
-
-const ROOTLY_URGENCY_ID_HIGH: &str = "62fde143-1258-4639-9ee6-1400ebf7308d";
-const ROOTLY_URGENCY_ID_MEDIUM: &str = "7366ba5e-f6ea-4a4e-a1b4-ae43d512e1e2";
-const ROOTLY_URGENCY_ID_LOW: &str = "4db3d3ed-f4ed-4818-82f5-7746da404bd2";
+use rootly_config::services::ROOTLY_SERVICE_ID_STORYTELLER_WEB;
+use rootly_config::urgencies::{ROOTLY_URGENCY_ID_HIGH, ROOTLY_URGENCY_ID_LOW, ROOTLY_URGENCY_ID_MEDIUM};
+use shared_env_var_config::paging::{env_optional_rootly_api_key, env_optional_rootly_notification_target_id, env_optional_rootly_notification_target_type};

 pub fn build_pager(
  server_environment: server_environment::ServerEnvironment,
@@ -31,7 +26,7 @@ pub fn build_pager(
      .application_name("storyteller-web".to_string())
      .environment(environment.to_string())
      .hostname(hostname.to_string())
-      .service_id(ROOTLY_SERVICE_ID.to_string());
+      .service_id(ROOTLY_SERVICE_ID_STORYTELLER_WEB.to_string());
  
  // If paging is globally disabled, use a NoOp pager regardless of API key.
  if !paging_flags.is_paging_enabled {
@@ -40,7 +35,7 @@ pub fn build_pager(
    return (pager, worker, paging_flags);
  }

-  let maybe_api_key = easyenv::get_env_string_optional("ROOTLY_API_KEY");
+  let maybe_api_key = env_optional_rootly_api_key();

  let (pager, worker) = match maybe_api_key {
    Some(api_key) => {
@@ -63,8 +58,8 @@ fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWo
      .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string())
      .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string());

-  let target_type = easyenv::get_env_string_optional("ROOTLY_NOTIFICATION_TARGET_TYPE");
-  let target_id = easyenv::get_env_string_optional("ROOTLY_NOTIFICATION_TARGET_ID");
+  let target_type = env_optional_rootly_notification_target_type();
+  let target_id = env_optional_rootly_notification_target_id();

  if let (Some(t_type), Some(t_id)) = (target_type, target_id) {
    rootly_builder = rootly_builder.notification_target(t_type, t_id);
@@ -1,3 +1,5 @@
+use shared_env_var_config::paging::{env_enable_paging_default_false, env_enable_paging_for_500s_default_false};
+
 /// Flags controlling the pager system.
 #[derive(Clone, Debug)]
 pub struct PagingFlags {
@@ -14,8 +16,8 @@ pub struct PagingFlags {
 impl PagingFlags {
  pub fn from_env() -> Self {
    Self {
-      is_paging_enabled: easyenv::get_env_bool_or_default("ENABLE_PAGING", false),
-      is_paging_for_500s_enabled: easyenv::get_env_bool_or_default("ENABLE_PAGING_FOR_500S", false),
+      is_paging_enabled: env_enable_paging_default_false(),
+      is_paging_for_500s_enabled: env_enable_paging_for_500s_default_false(),
    }
  }
 }
@@ -19,6 +19,11 @@ pub async fn enroll_in_studio(
      .map(|ff| ff.clone())
      .unwrap_or_else(|| UserSessionFeatureFlags::empty());

+  if user_feature_flags.has_flag(UserFeatureFlag::Studio)
+      && user_feature_flags.has_flag(UserFeatureFlag::VideoStyleTransfer) {
+    return Ok(());
+  }
+
  user_feature_flags.add_flags([
    UserFeatureFlag::Studio,
    UserFeatureFlag::VideoStyleTransfer,
Author	SHA1	Message	Date
Brandon Thomas	6e8875b07c	fix password issue	2026-03-31 14:54:09 -04:00
Brandon Thomas	e391bbff04	this is paging me	2026-03-31 14:37:41 -04:00
Brandon Thomas	12d3f82fe2	fal handle webhook errors (#1272 )	2026-03-31 10:50:52 -04:00
Brandon Thomas	4489c90125	fal webhook and seedance job alerting with additonal context (#1267 )	2026-03-31 06:22:27 -04:00
Brandon Thomas	fd54477f73	page on seedance jobs (#1266 ) also fix threading on video-thumbnail-job	2026-03-31 05:01:12 -04:00
Brandon Thomas	c093603b3d	config crates for sharing between services (#1265 )	2026-03-31 04:01:34 -04:00
Brandon Thomas	a9fb97d720	video-thumbnail-job alerts (#1264 )	2026-03-31 03:23:58 -04:00
Brandon Thomas	3d12be2469	don't write db if user already has flags	2026-03-31 02:29:23 -04:00
Brandon Thomas	e9938f6c93	fix login paging with sentinel value passwords (#1263 )	2026-03-31 02:18:56 -04:00