upload finished results from fal (wip)

This commit is contained in:
Brandon Thomas committed 2026-05-11 23:23:26 -04:00
1 parent 97a1e2ec57
commit 32be098943
8 files changed
+551 -23

No files matched your search

+1
View File
@@ -33,6 +33,7 @@ build_metadata.workspace = true
cookie_store.workspace = true
enums.workspace = true
errors.workspace = true
fal_client.workspace = true
filesys.workspace = true
grok_client.workspace = true
uuid_utils.workspace = true
@@ -21,6 +21,7 @@ use crate::services::sora::state::sora_task_queue::SoraTaskQueue;
use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager;
use crate::services::worldlabs::state::worldlabs_bearer_bridge::WorldlabsBearerBridge;
use crate::services::worldlabs::state::worldlabs_credential_manager::WorldlabsCredentialManager;
use crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache;
use crate::core::threads::third_party_task_polling_thread::third_party_task_polling_thread::third_party_task_polling_thread;
use crate::services::worldlabs::threads::worldlabs_marble_task_polling::worldlabs_marble_task_polling;
use errors::AnyhowResult;
@@ -40,6 +41,7 @@ pub async fn handle_tauri_startup(
grok_image_prompt_queue: GrokImagePromptQueue,
worldlabs_bearer_bridge: WorldlabsBearerBridge,
worldlabs_creds_manager: WorldlabsCredentialManager,
credential_cache: ProviderCredentialLoadingCache,
) -> AnyhowResult<()> {
set_app_log_level(
@@ -118,7 +120,12 @@ pub async fn handle_tauri_startup(
));
tauri::async_runtime::spawn(third_party_task_polling_thread(
app.clone(),
app_env_configs.clone(),
root.clone(),
task_database.clone(),
storyteller_creds_manager.clone(),
credential_cache,
));
spawn_discord_presence_thread()?;
@@ -0,0 +1,273 @@
use crate::core::events::basic_sendable_event_trait::BasicSendableEvent;
use crate::core::events::generation_events::common::{GenerationAction, GenerationServiceProvider};
use crate::core::events::generation_events::generation_complete_event::GenerationCompleteEvent;
use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs;
use crate::core::state::data_dir::app_data_root::AppDataRoot;
use crate::core::state::data_dir::trait_data_subdir::DataSubdir;
use crate::core::state::task_database::TaskDatabase;
use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager;
use artcraft_api_defs::prompts::create_prompt::CreatePromptRequest;
use artcraft_api_defs::utils::media_links_to_thumbnail_template::media_links_to_thumbnail_template;
use artcraft_client::credentials::storyteller_credential_set::StorytellerCredentialSet;
use artcraft_client::endpoints::media_files::get_media_file::get_media_file;
use artcraft_client::endpoints::media_files::upload_image_media_file_from_file::{
upload_image_media_file_from_file, UploadImageFromFileArgs,
};
use artcraft_client::endpoints::media_files::upload_video_media_file_from_file::{
upload_video_media_file_from_file, UploadVideoFromFileArgs,
};
use artcraft_client::endpoints::prompts::create_prompt::create_prompt;
use enums::common::generation_provider::GenerationProvider;
use enums::tauri::tasks::task_media_file_class::TaskMediaFileClass;
use enums::tauri::tasks::task_type::TaskType;
use fal_client::polling::poll_job_response::poll_job_response::PollJobResponse;
use fal_client::polling::poll_job_response::success_case_extractors::PollResponseExtractedContents;
use log::{error, info, warn};
use sqlite_tasks::queries::task::Task;
use sqlite_tasks::queries::update_successful_task_status_with_metadata::{
update_successful_task_status_with_metadata, UpdateSuccessfulTaskArgs,
};
use std::fs::File;
use std::io::Write;
use std::path::PathBuf;
use tauri::AppHandle;
use tokens::tokens::media_files::MediaFileToken;
use uuid_utils::uuid::generate_random_uuid;
pub async fn handle_fal_complete(
app_handle: &AppHandle,
app_env_configs: &AppEnvConfigs,
app_data_root: &AppDataRoot,
task_database: &TaskDatabase,
storyteller_creds_manager: &StorytellerCredentialManager,
task: &Task,
job_response: PollJobResponse,
) {
info!("[FalComplete] Handling completed task {}", task.id.as_str());
let result = handle_fal_complete_inner(
app_handle,
app_env_configs,
app_data_root,
task_database,
storyteller_creds_manager,
task,
job_response,
).await;
if let Err(err) = result {
error!("[FalComplete] Failed to handle task {}: {:?}", task.id.as_str(), err);
}
}
async fn handle_fal_complete_inner(
app_handle: &AppHandle,
app_env_configs: &AppEnvConfigs,
app_data_root: &AppDataRoot,
task_database: &TaskDatabase,
storyteller_creds_manager: &StorytellerCredentialManager,
task: &Task,
job_response: PollJobResponse,
) -> Result<(), Box<dyn std::error::Error>> {
let creds = storyteller_creds_manager.get_credentials()?
.ok_or("No Storyteller credentials available")?;
let extracted = job_response.extracted_contents;
// Determine what kind of media we got and collect download URLs.
let (urls, media_class, generation_action) = collect_media_urls(task, &extracted)?;
if urls.is_empty() {
warn!("[FalComplete] Task {} completed but no downloadable media found in response", task.id.as_str());
//warn!("[FalComplete] Raw body: {}", &job_response.raw_body[..job_response.raw_body.len().min(500)]);
return Ok(());
}
// Create a prompt record
let prompt_response = create_prompt(
&app_env_configs.storyteller_host,
Some(&creds),
CreatePromptRequest {
uuid_idempotency_token: generate_random_uuid(),
positive_prompt: None,
negative_prompt: None,
model_type: None,
generation_provider: Some(GenerationProvider::Fal),
maybe_generation_mode: None,
maybe_aspect_ratio: None,
maybe_resolution: None,
maybe_batch_count: None,
maybe_generate_audio: None,
maybe_duration_seconds: None,
},
).await?;
info!("[FalComplete] Created prompt: {:?}", prompt_response.prompt_token);
let mut maybe_primary_media_file_token: Option<MediaFileToken> = None;
for (i, url) in urls.iter().enumerate() {
info!("[FalComplete] Downloading result {} from: {}", i, url);
let download_path = download_file(url, app_data_root, i).await?;
info!("[FalComplete] Uploading result {} to backend...", i);
let media_token = upload_to_backend(
&creds,
app_env_configs,
&download_path,
&prompt_response.prompt_token,
media_class,
).await?;
info!("[FalComplete] Uploaded as media file: {:?}", media_token);
if maybe_primary_media_file_token.is_none() {
maybe_primary_media_file_token = Some(media_token);
}
}
// Look up CDN/thumbnail URLs for the primary media file
let mut maybe_cdn_url = None;
let mut maybe_thumbnail_url_template = None;
if let Some(media_file_token) = maybe_primary_media_file_token.as_ref() {
match get_media_file(&app_env_configs.storyteller_host, media_file_token).await {
Ok(response) => {
maybe_cdn_url = Some(response.media_file.media_links.cdn_url.to_string());
maybe_thumbnail_url_template = media_links_to_thumbnail_template(&response.media_file.media_links)
.map(|s| s.to_string());
}
Err(err) => {
error!("[FalComplete] Failed to look up media file after upload: {:?} (failing open)", err);
}
}
}
// Mark the task as completed
let updated = update_successful_task_status_with_metadata(UpdateSuccessfulTaskArgs {
db: task_database.get_connection(),
task_id: &task.id,
maybe_batch_token: None,
maybe_primary_media_file_token: maybe_primary_media_file_token.as_ref(),
maybe_primary_media_file_class: Some(media_class),
maybe_primary_media_file_cdn_url: maybe_cdn_url.as_deref(),
maybe_primary_media_file_thumbnail_url_template: maybe_thumbnail_url_template.as_deref(),
}).await?;
if updated {
let event = GenerationCompleteEvent {
action: Some(generation_action),
service: GenerationServiceProvider::Fal,
model: None,
};
event.send_infallible(app_handle);
}
info!("[FalComplete] Task {} fully handled", task.id.as_str());
Ok(())
}
// ── Helpers ──
/// Collect downloadable media URLs from the extracted response contents.
fn collect_media_urls(
task: &Task,
extracted: &Option<PollResponseExtractedContents>,
) -> Result<(Vec<String>, TaskMediaFileClass, GenerationAction), Box<dyn std::error::Error>> {
let extracted = match extracted {
Some(e) => e,
None => return Ok((vec![], TaskMediaFileClass::Image, GenerationAction::GenerateImage)),
};
// Images (batch)
if let Some(images) = &extracted.images {
let urls: Vec<String> = images.iter()
.filter_map(|img| img.url.clone())
.collect();
if !urls.is_empty() {
return Ok((urls, TaskMediaFileClass::Image, GenerationAction::GenerateImage));
}
}
// Single image (e.g. background removal)
if let Some(image) = &extracted.image {
if let Some(url) = &image.url {
return Ok((vec![url.clone()], TaskMediaFileClass::Image, GenerationAction::GenerateImage));
}
}
// Video
if let Some(video) = &extracted.video {
if let Some(url) = &video.url {
return Ok((vec![url.clone()], TaskMediaFileClass::Video, GenerationAction::GenerateVideo));
}
}
// 3D model (GLB)
if let Some(glb) = &extracted.model_glb {
if let Some(url) = &glb.url {
// TODO: We may want a TaskMediaFileClass::Mesh variant
return Ok((vec![url.clone()], TaskMediaFileClass::Image, GenerationAction::ImageTo3d));
}
}
Ok((vec![], TaskMediaFileClass::Image, GenerationAction::GenerateImage))
}
async fn download_file(
url: &str,
app_data_root: &AppDataRoot,
index: usize,
) -> Result<PathBuf, Box<dyn std::error::Error>> {
let response = reqwest::get(url).await?;
let bytes = response.bytes().await?;
let extension = url_utils::download_extension::extract_download_extension_from_url::extract_download_extension_from_url_str(url)
.map(|ext| ext.as_extension_without_period())
.unwrap_or("bin");
let tempdir = app_data_root.temp_dir().path();
let filename = format!("fal_{}_{}.{}", generate_random_uuid(), index, extension);
let download_path = tempdir.join(filename);
let mut file = File::create(&download_path)?;
file.write_all(&bytes)?;
Ok(download_path)
}
async fn upload_to_backend(
creds: &StorytellerCredentialSet,
app_env_configs: &AppEnvConfigs,
download_path: &PathBuf,
prompt_token: &tokens::tokens::prompts::PromptToken,
media_class: TaskMediaFileClass,
) -> Result<MediaFileToken, Box<dyn std::error::Error>> {
let media_token = match media_class {
TaskMediaFileClass::Video => {
let result = upload_video_media_file_from_file(UploadVideoFromFileArgs {
api_host: &app_env_configs.storyteller_host,
maybe_creds: Some(creds),
path: download_path,
maybe_prompt_token: Some(prompt_token),
}).await?;
result.media_file_token
}
_ => {
let result = upload_image_media_file_from_file(UploadImageFromFileArgs {
api_host: &app_env_configs.storyteller_host,
maybe_creds: Some(creds),
path: download_path,
is_intermediate_system_file: false,
maybe_prompt_token: Some(prompt_token),
maybe_batch_token: None,
}).await?;
result.media_file_token
}
};
Ok(media_token)
}
@@ -0,0 +1,51 @@
use crate::core::events::basic_sendable_event_trait::BasicSendableEvent;
use crate::core::events::generation_events::common::{GenerationAction, GenerationServiceProvider};
use crate::core::events::generation_events::generation_failed_event::GenerationFailedEvent;
use crate::core::state::task_database::TaskDatabase;
use enums::tauri::tasks::task_status::TaskStatus;
use enums::tauri::tasks::task_type::TaskType;
use log::{error, info};
use sqlite_tasks::queries::task::Task;
use sqlite_tasks::queries::update_task_status::{update_task_status, UpdateTaskArgs};
use tauri::AppHandle;
pub async fn handle_fal_failure(
app_handle: &AppHandle,
task_database: &TaskDatabase,
task: &Task,
reason: &str,
) {
info!("[FalPolling] Marking task {} as failed: {}", task.id.as_str(), reason);
let update_result = update_task_status(UpdateTaskArgs {
db: task_database.get_connection(),
task_id: &task.id,
status: TaskStatus::CompleteFailure,
}).await;
if let Err(err) = update_result {
error!("[FalPolling] Failed to update task status for {}: {:?}", task.id.as_str(), err);
}
let action = task_type_to_generation_action(task.task_type);
let event = GenerationFailedEvent {
action,
service: GenerationServiceProvider::Fal,
model: None,
reason: Some(reason.to_string()),
};
event.send_infallible(app_handle);
}
fn task_type_to_generation_action(task_type: TaskType) -> GenerationAction {
match task_type {
TaskType::ImageGeneration => GenerationAction::GenerateImage,
TaskType::VideoGeneration => GenerationAction::GenerateVideo,
TaskType::BackgroundRemoval => GenerationAction::RemoveBackground,
TaskType::ObjectGeneration => GenerationAction::ImageTo3d,
TaskType::GaussianGeneration => GenerationAction::GenerateGaussian,
TaskType::ImageInpaintEdit => GenerationAction::ImageInpaintEdit,
}
}
@@ -1,3 +1,3 @@
mod handle_fal_complete;
mod handle_fal_failure;
pub mod handle_fal_complete;
pub mod handle_fal_failure;
pub mod poll_fal_tasks;
@@ -0,0 +1,183 @@
use crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache;
use crate::core::providers::credentials::payload::provider_credential_payload::ProviderCredentialPayload;
use crate::core::providers::credentials::provider_credential_key::ProviderCredentialKey;
use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs;
use crate::core::state::data_dir::app_data_root::AppDataRoot;
use crate::core::state::task_database::TaskDatabase;
use crate::core::threads::third_party_task_polling_thread::handlers::fal::handle_fal_complete::handle_fal_complete;
use crate::core::threads::third_party_task_polling_thread::handlers::fal::handle_fal_failure::handle_fal_failure;
use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager;
use enums::tauri::tasks::task_status::TaskStatus;
use fal_client::creds::fal_api_key::FalApiKey;
use fal_client::polling::poll_job_response::poll_job_response::{poll_job_response, PollJobResponseArgs};
use fal_client::polling::poll_job_status::poll_job_status::{poll_job_status, FalJobStatus, PollJobStatusArgs};
use log::{error, info, warn};
use sqlite_tasks::queries::task::Task;
use sqlite_tasks::queries::update_task_status::{update_task_status, UpdateTaskArgs};
use tauri::AppHandle;
pub async fn poll_fal_tasks(
app_handle: &AppHandle,
app_env_configs: &AppEnvConfigs,
app_data_root: &AppDataRoot,
task_database: &TaskDatabase,
storyteller_creds_manager: &StorytellerCredentialManager,
credential_cache: &ProviderCredentialLoadingCache,
fal_tasks: &[&Task],
) {
let api_key = match load_fal_api_key(credential_cache) {
Some(key) => key,
None => {
warn!("[FalPolling] No FAL API key configured, skipping poll");
return;
}
};
for task in fal_tasks {
let result = poll_single_fal_task(
app_handle,
app_env_configs,
app_data_root,
task_database,
storyteller_creds_manager,
&api_key,
task,
).await;
if let Err(err) = result {
error!(
"[FalPolling] Error processing task {}: {:?}",
task.id.as_str(),
err,
);
}
}
}
async fn poll_single_fal_task(
app_handle: &AppHandle,
app_env_configs: &AppEnvConfigs,
app_data_root: &AppDataRoot,
task_database: &TaskDatabase,
storyteller_creds_manager: &StorytellerCredentialManager,
api_key: &FalApiKey,
task: &Task,
) -> Result<(), Box<dyn std::error::Error>> {
let status_url = match &task.queue_status_url {
Some(url) => url.as_str(),
None => {
warn!("[FalPolling] Task {} has no queue_status_url, marking as failed", task.id.as_str());
handle_fal_failure(
app_handle,
task_database,
task,
"No queue status URL to check",
).await;
return Ok(());
}
};
// Step 1: Check job status
info!("[FalPolling] Checking status for task {}: {}", task.id.as_str(), status_url);
let status_response = poll_job_status(PollJobStatusArgs {
status_url,
api_key,
}).await?;
match status_response.status {
FalJobStatus::InQueue | FalJobStatus::InProgress => {
info!(
"[FalPolling] Task {} still {:?} (queue_position={:?})",
task.id.as_str(),
status_response.status,
status_response.maybe_queue_position,
);
return Ok(());
}
FalJobStatus::Unknown(ref status) => {
warn!("[FalPolling] Task {} has unknown status: {}", task.id.as_str(), status);
return Ok(());
}
FalJobStatus::Completed => {
info!("[FalPolling] Task {} completed, fetching results...", task.id.as_str());
}
}
// Step 2: Fetch the response
let response_url = pick_response_url(task, &status_response.response_url);
let response_url = match response_url {
Some(url) => url,
None => {
warn!("[FalPolling] Task {} completed but has no response URL, marking as failed", task.id.as_str());
handle_fal_failure(
app_handle,
task_database,
task,
"Job completed but no response URL available",
).await;
return Ok(());
}
};
let job_response = match poll_job_response(PollJobResponseArgs {
response_url: &response_url,
api_key,
}).await {
Ok(response) => response,
Err(err) => {
error!(
"[FalPolling] Failed to fetch response for task {}: {}. Skipping for now.",
task.id.as_str(),
err,
);
return Ok(());
}
};
// Step 3: Handle completed job
handle_fal_complete(
app_handle,
app_env_configs,
app_data_root,
task_database,
storyteller_creds_manager,
task,
job_response,
).await;
Ok(())
}
// ── Helpers ──
fn load_fal_api_key(credential_cache: &ProviderCredentialLoadingCache) -> Option<FalApiKey> {
match credential_cache.get_credentials(ProviderCredentialKey::FalApiKey) {
Ok(Some(ProviderCredentialPayload::ApiKey(data))) => {
Some(FalApiKey::from_str(data.as_str()))
}
_ => None,
}
}
/// Use the task's stored response URL first; fall back to the one from the status poll.
/// Log if they differ.
fn pick_response_url(task: &Task, polled_response_url: &Option<String>) -> Option<String> {
match (&task.queue_response_url, polled_response_url) {
(Some(db_url), Some(polled_url)) => {
if db_url != polled_url {
warn!(
"[FalPolling] Task {} response URLs differ: db={}, polled={}. Using db URL.",
task.id.as_str(),
db_url,
polled_url,
);
}
Some(db_url.clone())
}
(Some(db_url), None) => Some(db_url.clone()),
(None, Some(polled_url)) => Some(polled_url.clone()),
(None, None) => None,
}
}
@@ -1,5 +1,10 @@
use crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache;
use crate::core::state::app_env_configs::app_env_configs::AppEnvConfigs;
use crate::core::state::data_dir::app_data_root::AppDataRoot;
use crate::core::state::task_database::TaskDatabase;
use crate::core::threads::third_party_task_polling_thread::handlers::fal::poll_fal_tasks::poll_fal_tasks;
use crate::core::utils::task_database_pending_statuses::TASK_DATABASE_PENDING_STATUSES;
use crate::services::storyteller::state::storyteller_credential_manager::StorytellerCredentialManager;
use enums::common::generation_provider::GenerationProvider;
use log::{error, info, warn};
use sqlite_tasks::queries::list_non_artcraft_pending_tasks::{
@@ -7,24 +12,31 @@ use sqlite_tasks::queries::list_non_artcraft_pending_tasks::{
};
use sqlite_tasks::queries::task::Task;
use std::time::Duration;
use tauri::AppHandle;
const SLEEP_NO_THIRD_PARTY_JOBS_SEEN: Duration = Duration::from_secs(10);
const SLEEP_THIRD_PARTY_JOBS_SEEN: Duration = Duration::from_secs(2);
const SLEEP_BETWEEN_FAL_POLLS: Duration = Duration::from_secs(1);
const SLEEP_ON_ERROR: Duration = Duration::from_secs(30);
/// Main loop that polls for third-party (non-Artcraft) tasks and processes them.
///
/// This thread runs for the lifetime of the application. On error, it pauses
/// and gracefully continues.
pub async fn third_party_task_polling_thread(
app_handle: AppHandle,
app_env_configs: AppEnvConfigs,
app_data_root: AppDataRoot,
task_database: TaskDatabase,
storyteller_creds_manager: StorytellerCredentialManager,
credential_cache: ProviderCredentialLoadingCache,
) -> ! {
let mut has_ever_seen_third_party_jobs = false;
loop {
let result = poll_iteration(
&app_handle,
&app_env_configs,
&app_data_root,
&task_database,
&storyteller_creds_manager,
&credential_cache,
&mut has_ever_seen_third_party_jobs,
).await;
@@ -36,7 +48,12 @@ pub async fn third_party_task_polling_thread(
}
async fn poll_iteration(
app_handle: &AppHandle,
app_env_configs: &AppEnvConfigs,
app_data_root: &AppDataRoot,
task_database: &TaskDatabase,
storyteller_creds_manager: &StorytellerCredentialManager,
credential_cache: &ProviderCredentialLoadingCache,
has_ever_seen_third_party_jobs: &mut bool,
) -> Result<(), PollError> {
let task_list = list_non_artcraft_pending_tasks(ListNonArtcraftPendingTasksArgs {
@@ -58,7 +75,6 @@ async fn poll_iteration(
*has_ever_seen_third_party_jobs = true;
// Partition tasks by provider.
let fal_tasks: Vec<&Task> = tasks.iter()
.filter(|t| t.provider == GenerationProvider::Fal)
.collect();
@@ -67,7 +83,6 @@ async fn poll_iteration(
.filter(|t| t.provider != GenerationProvider::Fal)
.collect();
// Shed non-FAL jobs for now — we'll add handlers for other providers later.
if !non_fal_tasks.is_empty() {
for task in &non_fal_tasks {
warn!(
@@ -80,25 +95,21 @@ async fn poll_iteration(
}
if fal_tasks.is_empty() {
let sleep_duration = SLEEP_THIRD_PARTY_JOBS_SEEN;
tokio::time::sleep(sleep_duration).await;
tokio::time::sleep(SLEEP_THIRD_PARTY_JOBS_SEEN).await;
return Ok(());
}
info!(
"[ThirdPartyPolling] {} FAL job(s) ready to check",
fal_tasks.len(),
);
info!("[ThirdPartyPolling] {} FAL job(s) ready to check", fal_tasks.len());
for task in &fal_tasks {
info!(
"[ThirdPartyPolling] task_id={}, provider_job_id={:?}, type={:?}, status_url={:?}",
task.id.as_str(),
task.provider_job_id,
task.task_type,
task.queue_status_url,
);
}
poll_fal_tasks(
app_handle,
app_env_configs,
app_data_root,
task_database,
storyteller_creds_manager,
credential_cache,
&fal_tasks,
).await;
tokio::time::sleep(SLEEP_BETWEEN_FAL_POLLS).await;
+2
View File
@@ -97,6 +97,7 @@ pub fn run() {
let app_preferences = load_app_preferences_or_default(&app_data_root);
let provider_credential_cache = crate::core::providers::credentials::provider_credential_loading_cache::ProviderCredentialLoadingCache::new(app_data_root.clone());
let provider_credential_cache_2 = provider_credential_cache.clone();
// NB: tauri-plugin-http stores the credentials on disk, so we can defer to that for now.
// println!("Attempting to read existing artcraft credentials...");
@@ -179,6 +180,7 @@ pub fn run() {
grok_prompt_queue_2,
worldlabs_bearer_bridge_2,
worldlabs_creds_manager_2,
provider_credential_cache_2,
).await;
if let Err(err) = result {