use crate::{errors::AppError, extractors::JwtClaims, parsers, state::AppState}; use api_types::{ requests::{ CompleteJobRequest, ConfigurePipelineRequest, EnqueueJobRequest, FailJobRequest, ManagePluginRequest, }, responses::{ BatchProgressResponse, JobListResponse, JobResponse, PipelineResponse, PluginResponse, }, }; use application::processing::{ CompleteJobCommand, ConfigurePipelineCommand, EnqueueJobCommand, FailJobCommand, ListJobsQuery, ManagePluginCommand, PipelineStepConfig, PluginAction, ReportBatchProgressQuery, StartJobCommand, }; use axum::{ Json, extract::{Path, Query, State}, http::StatusCode, }; use domain::{ errors::DomainError, value_objects::{MetadataValue, StructuredData, SystemId}, }; fn hashmap_to_structured( map: &std::collections::HashMap, ) -> StructuredData { let mut sd = StructuredData::new(); for (k, v) in map { sd.insert(k.clone(), MetadataValue::from(v.clone())); } sd } #[derive(Debug, serde::Deserialize)] pub struct ListJobsParams { pub status: Option, pub limit: Option, pub offset: Option, } #[utoipa::path( get, path = "/api/v1/jobs", security(("bearer_token" = [])), params( ("status" = Option, Query, description = "Status filter"), ("limit" = Option, Query, description = "Page size"), ("offset" = Option, Query, description = "Page offset") ), responses( (status = 200, description = "Job list", body = JobListResponse), (status = 401, description = "Unauthorized") ) )] pub async fn list_jobs( State(state): State, claims: JwtClaims, Query(params): Query, ) -> Result, AppError> { super::require_admin(&claims)?; let query = ListJobsQuery { status: params.status, limit: params.limit.unwrap_or(20).min(100), offset: params.offset.unwrap_or(0), }; let result = state.processing.list_jobs.execute(query).await?; let jobs = result.jobs.iter().map(JobResponse::from_domain).collect(); Ok(Json(JobListResponse { jobs, total: result.total, })) } #[utoipa::path( post, path = "/api/v1/jobs", request_body = EnqueueJobRequest, security(("bearer_token" = [])), responses( (status = 201, description = "Job enqueued", body = JobResponse), (status = 401, description = "Unauthorized") ) )] pub async fn enqueue_job( State(state): State, claims: JwtClaims, Json(req): Json, ) -> Result<(StatusCode, Json), AppError> { super::require_admin(&claims)?; let payload = req .payload .as_ref() .map(hashmap_to_structured) .unwrap_or_default(); let cmd = EnqueueJobCommand { job_type: parsers::job_type(&req.job_type), priority: req.priority.unwrap_or(0), payload, target_asset_id: req.target_asset_id.map(SystemId::from_uuid), batch_id: req.batch_id.map(SystemId::from_uuid), }; let job = state.processing.enqueue_job.execute(cmd).await?; Ok((StatusCode::CREATED, Json(JobResponse::from_domain(&job)))) } #[utoipa::path( post, path = "/api/v1/jobs/{id}/start", security(("bearer_token" = [])), params(("id" = uuid::Uuid, Path, description = "Job ID")), responses( (status = 200, description = "Job started", body = JobResponse), (status = 404, description = "Not found") ) )] pub async fn start_job( State(state): State, claims: JwtClaims, Path((job_id,)): Path<(uuid::Uuid,)>, ) -> Result, AppError> { super::require_admin(&claims)?; let cmd = StartJobCommand { job_id: SystemId::from_uuid(job_id), }; let job = state.processing.start_job.execute(cmd).await?; Ok(Json(JobResponse::from_domain(&job))) } #[utoipa::path( post, path = "/api/v1/jobs/{id}/complete", request_body = CompleteJobRequest, security(("bearer_token" = [])), params(("id" = uuid::Uuid, Path, description = "Job ID")), responses( (status = 200, description = "Job completed", body = JobResponse), (status = 404, description = "Not found") ) )] pub async fn complete_job( State(state): State, claims: JwtClaims, Path((job_id,)): Path<(uuid::Uuid,)>, Json(req): Json, ) -> Result, AppError> { super::require_admin(&claims)?; let cmd = CompleteJobCommand { job_id: SystemId::from_uuid(job_id), result: hashmap_to_structured(&req.result), }; let job = state.processing.complete_job.execute(cmd).await?; Ok(Json(JobResponse::from_domain(&job))) } #[utoipa::path( post, path = "/api/v1/jobs/{id}/fail", request_body = FailJobRequest, security(("bearer_token" = [])), params(("id" = uuid::Uuid, Path, description = "Job ID")), responses( (status = 200, description = "Job failed", body = JobResponse), (status = 404, description = "Not found") ) )] pub async fn fail_job( State(state): State, claims: JwtClaims, Path((job_id,)): Path<(uuid::Uuid,)>, Json(req): Json, ) -> Result, AppError> { super::require_admin(&claims)?; let cmd = FailJobCommand { job_id: SystemId::from_uuid(job_id), error: req.error, }; let job = state.processing.fail_job.execute(cmd).await?; Ok(Json(JobResponse::from_domain(&job))) } #[utoipa::path( get, path = "/api/v1/jobs/batches/{id}", security(("bearer_token" = [])), params(("id" = uuid::Uuid, Path, description = "Batch ID")), responses( (status = 200, description = "Batch progress", body = BatchProgressResponse), (status = 404, description = "Not found") ) )] pub async fn batch_progress( State(state): State, claims: JwtClaims, Path((batch_id,)): Path<(uuid::Uuid,)>, ) -> Result, AppError> { super::require_admin(&claims)?; let query = ReportBatchProgressQuery { batch_id: SystemId::from_uuid(batch_id), }; let progress = state.processing.batch_progress.execute(query).await?; Ok(Json(BatchProgressResponse::from_domain(&progress))) } #[utoipa::path( post, path = "/api/v1/plugins", request_body = ManagePluginRequest, security(("bearer_token" = [])), responses( (status = 201, description = "Plugin managed", body = PluginResponse), (status = 401, description = "Unauthorized") ) )] pub async fn manage_plugin( State(state): State, claims: JwtClaims, Json(req): Json, ) -> Result<(StatusCode, Json), AppError> { super::require_admin(&claims)?; let action = match req.action.as_str() { "create" => { let name = req.name.ok_or_else(|| { AppError::from(DomainError::Validation("name required for create".into())) })?; let pt = req.plugin_type.as_deref().unwrap_or("media_processor"); let plugin_type = parsers::plugin_type(pt)?; let config = req .config .as_ref() .map(hashmap_to_structured) .unwrap_or_default(); PluginAction::Create { name, plugin_type, config, } } "enable" => PluginAction::Enable, "disable" => PluginAction::Disable, other => { return Err(AppError::from(DomainError::Validation(format!( "Invalid plugin action: {other}. Use create, enable, or disable" )))); } }; let cmd = ManagePluginCommand { plugin_id: req.plugin_id.map(SystemId::from_uuid), action, }; let plugin = state.processing.manage_plugin.execute(cmd).await?; Ok(( StatusCode::CREATED, Json(PluginResponse::from_domain(&plugin)), )) } #[utoipa::path( post, path = "/api/v1/pipelines", request_body = ConfigurePipelineRequest, security(("bearer_token" = [])), responses( (status = 201, description = "Pipeline configured", body = PipelineResponse), (status = 401, description = "Unauthorized") ) )] pub async fn configure_pipeline( State(state): State, claims: JwtClaims, Json(req): Json, ) -> Result<(StatusCode, Json), AppError> { super::require_admin(&claims)?; let steps = req .steps .iter() .map(|s| PipelineStepConfig { plugin_id: SystemId::from_uuid(s.plugin_id), config: hashmap_to_structured(&s.config), }) .collect(); let cmd = ConfigurePipelineCommand { trigger_event: req.trigger_event, steps, }; let pipeline = state.processing.configure_pipeline.execute(cmd).await?; Ok(( StatusCode::CREATED, Json(PipelineResponse::from_domain(&pipeline)), )) }