diff --git a/crates/api_clients/fal_client/src/webhook_api/hydrate_webhook_contents.rs b/crates/api_clients/fal_client/src/webhook_api/hydrate_webhook_contents.rs index 4d773e97fa..1bfc5992bd 100644 --- a/crates/api_clients/fal_client/src/webhook_api/hydrate_webhook_contents.rs +++ b/crates/api_clients/fal_client/src/webhook_api/hydrate_webhook_contents.rs @@ -1,7 +1,7 @@ use serde_json::Value; use crate::webhook_api::raw::webhook_error_type::WebhookErrorType; -use crate::webhook_api::hydrated::hydrated_webhook_contents::{ErrorData, PayloadErrorData, SuccessData, HydratedWebhookContents}; +use crate::webhook_api::hydrated::hydrated_webhook_contents::{ErrorData, PayloadErrorData, WebhookSuccessData, HydratedWebhookContents}; use crate::webhook_api::raw::raw_webhook_payload::{RawWebhookPayload, RawWebhookStatus}; use crate::webhook_api::hydrated::success_case_extractors::extract_contents_from_payload; @@ -36,7 +36,7 @@ pub fn hydrate_webhook_contents(webhook: &RawWebhookPayload) -> HydratedWebhookC let extracted_contents = extract_contents_from_payload(&value); - HydratedWebhookContents::Success(SuccessData { + HydratedWebhookContents::Success(WebhookSuccessData { payload: value, extracted_contents, }) diff --git a/crates/api_clients/fal_client/src/webhook_api/hydrated/hydrated_webhook_contents.rs b/crates/api_clients/fal_client/src/webhook_api/hydrated/hydrated_webhook_contents.rs index 35e9f31812..3c50c9f26b 100644 --- a/crates/api_clients/fal_client/src/webhook_api/hydrated/hydrated_webhook_contents.rs +++ b/crates/api_clients/fal_client/src/webhook_api/hydrated/hydrated_webhook_contents.rs @@ -7,7 +7,7 @@ use crate::webhook_api::raw::webhook_error_type::WebhookErrorType; #[derive(Debug)] pub enum HydratedWebhookContents { /// The webhook reported success and has a payload. - Success(SuccessData), + Success(WebhookSuccessData), /// The webhook reported an error (status=ERROR) with optional detail info. Error(ErrorData), @@ -20,7 +20,7 @@ pub enum HydratedWebhookContents { } #[derive(Debug)] -pub struct SuccessData { +pub struct WebhookSuccessData { /// The success data is polymorphic, so we're returning a JSON `Value` for now. /// This will allow for downstream handlers to parse the payload as needed. /// This is the entire raw success payload. diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs index 0eb118fa9a..29774e7f1c 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/fal_webhook_handler.rs @@ -51,12 +51,12 @@ pub async fn fal_webhook_handler( info!("FAL webhook request_id: {} (status: {:?})", request_id, webhook_payload.status); // Step 3: Hydrate the webhook contents. - let inner_payload = hydrate_webhook_contents(&webhook_payload); + let hydrated_contents = hydrate_webhook_contents(&webhook_payload); // Step 4 & 5: Branch on the inner payload type. - let result = match inner_payload { + let result = match hydrated_contents { HydratedWebhookContents::Success(success_data) => { - handle_successful_fal_webhook(&server_state, request_id, &success_data.payload).await + handle_successful_fal_webhook(&server_state, request_id, &success_data).await } HydratedWebhookContents::Error(error_data) => { handle_failed_fal_webhook( diff --git a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/process_success/handle_successful_fal_webhook.rs b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/process_success/handle_successful_fal_webhook.rs index 1f65511b49..2da3b88a2c 100644 --- a/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/process_success/handle_successful_fal_webhook.rs +++ b/crates/service/web/storyteller_web/src/http_server/endpoints/webhooks/process_success/handle_successful_fal_webhook.rs @@ -1,9 +1,9 @@ use actix_web::web::Json; +use fal_client::webhook_api::hydrated::hydrated_webhook_contents::WebhookSuccessData; use http_server_common::response::response_success_helpers::SimpleGenericJsonSuccess; 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 serde_json::Value; use crate::http_server::common_responses::advanced_common_web_error::AdvancedCommonWebError; use crate::state::server_state::ServerState; @@ -17,7 +17,7 @@ use super::process_video_payload::process_video_payload; pub async fn handle_successful_fal_webhook( server_state: &ServerState, request_id: &str, - payload: &Value, + success_data: &WebhookSuccessData, ) -> Result, AdvancedCommonWebError> { let db_result = get_inference_job_by_fal_id( @@ -42,25 +42,62 @@ pub async fn handle_successful_fal_webhook( 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); + // The raw payload is still available for processors that need arbitrary JSON + // access beyond the typed extracted_contents. + let payload = &success_data.payload; + + if let Some(ref extracted) = success_data.extracted_contents { + if let Some(ref _image_data) = extracted.image { + if let Some(payload_obj) = payload.as_object() { + info!("Handling image payload for request_id {} / job {:?}", request_id, job.job_token); + let token = process_image_payload(payload_obj, &job, server_state).await?; + if maybe_media_token.is_none() { + maybe_media_token = Some(token); + } + } + } + + if let Some(ref _images_data) = extracted.images { + if let Some(payload_obj) = payload.as_object() { + info!("Handling images payload for request_id {} / job {:?}", request_id, job.job_token); + let (media_token, batch_token) = process_images_payload(payload_obj, &job, server_state).await?; + if maybe_media_token.is_none() { + maybe_media_token = media_token; + } + if maybe_batch_token.is_none() { + maybe_batch_token = batch_token; + } + } + } + + if let Some(ref _video_data) = extracted.video { + if let Some(payload_obj) = payload.as_object() { + info!("Handling video payload for request_id {} / job {:?}", request_id, job.job_token); + let token = process_video_payload(payload_obj, &job, server_state).await?; + if maybe_media_token.is_none() { + maybe_media_token = Some(token); + } + } + } + + if let Some(ref _model_glb_data) = extracted.model_glb { + if let Some(payload_obj) = payload.as_object() { + 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?; + if maybe_media_token.is_none() { + maybe_media_token = Some(token); + } + } + } + + if let Some(ref _model_mesh_data) = extracted.model_mesh { + if let Some(payload_obj) = payload.as_object() { + 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?; + if maybe_media_token.is_none() { + maybe_media_token = Some(token); + } + } } }