use std::{collections::HashMap, sync::Arc}; use axum::{ Json, extract::{Multipart, Path, Query, State}, http::HeaderMap, }; use chrono::{DateTime, NaiveDate, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{FromRow, PgPool}; use utoipa::ToSchema; use utoipa_axum::{router::OpenApiRouter, routes}; use uuid::Uuid; use crate::{ AppState, error::{ApiError, CommonApiErrors, ErrorBody}, }; type SharedState = Arc; const MAX_AUDIO_UPLOAD_BYTES: usize = 15 * 1024 * 1024; const MAX_SEGMENT_DURATION_MS: i32 = 600_000; const SHORT_RECORDING_DURATION_MS: i32 = 60_000; const SHORT_TRANSCRIPT_CHARS: usize = 600; pub fn router() -> OpenApiRouter { OpenApiRouter::new() .routes(routes!(get_active_feedback_session)) .routes(routes!(ensure_feedback_session)) .routes(routes!(update_feedback_session)) .routes(routes!(create_lesson_session)) .routes(routes!(upload_audio_segment)) .routes(routes!(finish_lesson_session)) .routes(routes!(mark_lesson_applied)) .routes(routes!(retry_audio_segment)) .routes(routes!(generate_feedback_content)) } #[derive(Debug, Deserialize)] struct ActiveSessionQuery { profile_id: Option, } #[derive(Debug, Deserialize, ToSchema)] #[schema(example = json!({ "profile_id": "11111111-1111-4111-8111-111111111111", "feedback_date": "2026-07-16", "content": "" }))] struct FeedbackSessionInput { profile_id: Option, feedback_date: NaiveDate, #[serde(default)] content: String, } #[derive(Debug, Deserialize, ToSchema)] struct FeedbackSessionUpdateInput { feedback_date: NaiveDate, content: String, } #[derive(Debug, FromRow)] struct FeedbackSessionRow { id: Uuid, profile_id: Option, feedback_date: NaiveDate, content: String, status: String, created_at: DateTime, updated_at: DateTime, } #[derive(Debug, FromRow)] struct LessonRow { id: Uuid, sequence: i32, status: String, duration_ms: i32, applied_directly: bool, included_in_generation: bool, started_at: DateTime, ended_at: Option>, } #[derive(Debug, FromRow)] struct SegmentRow { id: Uuid, lesson_session_id: Uuid, sequence: i32, duration_ms: i32, status: String, error_message: Option, } #[derive(Debug, Serialize, ToSchema)] struct FeedbackSessionResponse { id: Uuid, profile_id: Option, feedback_date: NaiveDate, content: String, status: String, lesson_count: usize, total_duration_ms: i64, processing_segment_count: usize, failed_segment_count: usize, needs_generation: bool, lessons: Vec, created_at: DateTime, updated_at: DateTime, } #[derive(Debug, Serialize, ToSchema)] struct LessonSummaryResponse { id: Uuid, sequence: i32, status: String, duration_ms: i32, applied_directly: bool, included_in_generation: bool, segments: Vec, started_at: DateTime, ended_at: Option>, } #[derive(Debug, Serialize, ToSchema)] struct AudioSegmentResponse { id: Uuid, sequence: i32, duration_ms: i32, status: String, error_message: Option, } #[derive(Debug, Serialize, ToSchema)] struct LessonCreatedResponse { id: Uuid, sequence: i32, status: String, started_at: DateTime, } #[derive(Debug, Serialize, ToSchema)] struct FinishedLessonResponse { id: Uuid, status: String, duration_ms: i32, transcript: String, can_apply_directly: bool, } #[derive(Debug, Deserialize, ToSchema)] struct GenerateFeedbackInput { #[schema(example = "学生本周构图更加稳定。")] existing_content: String, } #[derive(Debug, Deserialize, ToSchema)] struct MarkLessonAppliedInput { content: String, } #[derive(Debug, Serialize, ToSchema)] struct GenerateFeedbackResponse { content: String, method: String, } #[derive(Debug, ToSchema)] #[allow(dead_code)] struct AudioSegmentUpload { sequence: i32, duration_ms: i32, audio_format: String, audio: String, } #[utoipa::path( get, path = "/api/v1/feedback-sessions/active", tag = "语音反馈", summary = "读取进行中的反馈批次", description = "读取当前用户指定学生下尚未保存的反馈批次;profile_id 留空时读取不关联学生的批次。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("profile_id" = Option, Query, description = "学生档案 ID;留空表示不关联学生") ), responses( (status = 200, description = "进行中的反馈批次;不存在时返回 null", body = Option), CommonApiErrors ) )] async fn get_active_feedback_session( State(state): State, headers: HeaderMap, Query(query): Query, ) -> Result>, ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; let session = sqlx::query_as::<_, FeedbackSessionRow>( "SELECT id, profile_id, feedback_date, content, status, created_at, updated_at \ FROM feedback_sessions \ WHERE owner_id = $1 AND profile_id IS NOT DISTINCT FROM $2 AND status = 'active' \ ORDER BY updated_at DESC LIMIT 1", ) .bind(owner_id) .bind(query.profile_id) .fetch_optional(pool(&state)?) .await?; match session { Some(session) => Ok(Json(Some( load_session_response(pool(&state)?, session).await?, ))), None => Ok(Json(None)), } } #[utoipa::path( post, path = "/api/v1/feedback-sessions", tag = "语音反馈", summary = "创建或恢复反馈批次", description = "为学生创建一个反馈批次;已有未保存批次时直接恢复,并同步最新反馈日期。", params(("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer ")), request_body(content = FeedbackSessionInput, content_type = "application/json"), responses( (status = 200, description = "创建或恢复后的反馈批次", body = FeedbackSessionResponse), (status = 404, description = "学生档案不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn ensure_feedback_session( State(state): State, headers: HeaderMap, Json(input): Json, ) -> Result, ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; let pool = pool(&state)?; validate_draft_content(&input.content)?; if let Some(profile_id) = input.profile_id { assert_profile_owner(pool, owner_id, profile_id).await?; } let existing = sqlx::query_as::<_, FeedbackSessionRow>( "UPDATE feedback_sessions SET feedback_date = $3, content = $4, updated_at = now() \ WHERE id = ( \ SELECT id FROM feedback_sessions \ WHERE owner_id = $1 AND profile_id IS NOT DISTINCT FROM $2 AND status = 'active' \ ORDER BY updated_at DESC LIMIT 1 \ ) \ RETURNING id, profile_id, feedback_date, content, status, created_at, updated_at", ) .bind(owner_id) .bind(input.profile_id) .bind(input.feedback_date) .bind(&input.content) .fetch_optional(pool) .await?; let session = match existing { Some(session) => session, None => sqlx::query_as::<_, FeedbackSessionRow>( "INSERT INTO feedback_sessions (id, owner_id, profile_id, feedback_date, content) \ VALUES ($1, $2, $3, $4, $5) \ RETURNING id, profile_id, feedback_date, content, status, created_at, updated_at", ) .bind(Uuid::new_v4()) .bind(owner_id) .bind(input.profile_id) .bind(input.feedback_date) .bind(&input.content) .fetch_one(pool) .await?, }; Ok(Json(load_session_response(pool, session).await?)) } #[utoipa::path( put, path = "/api/v1/feedback-sessions/{session_id}", tag = "语音反馈", summary = "自动保存反馈草稿", description = "在用户编辑反馈正文或日期时自动保存进行中的反馈批次。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("session_id" = Uuid, Path, description = "反馈批次 ID") ), request_body(content = FeedbackSessionUpdateInput, content_type = "application/json"), responses( (status = 200, description = "更新后的反馈批次", body = FeedbackSessionResponse), (status = 404, description = "反馈批次不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn update_feedback_session( State(state): State, headers: HeaderMap, Path(session_id): Path, Json(input): Json, ) -> Result, ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; validate_draft_content(&input.content)?; let session = sqlx::query_as::<_, FeedbackSessionRow>( "UPDATE feedback_sessions SET feedback_date = $3, content = $4, updated_at = now() \ WHERE id = $1 AND owner_id = $2 AND status = 'active' \ RETURNING id, profile_id, feedback_date, content, status, created_at, updated_at", ) .bind(session_id) .bind(owner_id) .bind(input.feedback_date) .bind(&input.content) .fetch_optional(pool(&state)?) .await? .ok_or(ApiError::NotFound)?; Ok(Json(load_session_response(pool(&state)?, session).await?)) } #[utoipa::path( post, path = "/api/v1/feedback-sessions/{session_id}/lessons", tag = "语音反馈", summary = "开始一节课堂录音", description = "在反馈批次中自动创建下一节课堂录音。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("session_id" = Uuid, Path, description = "反馈批次 ID") ), responses( (status = 201, description = "课堂录音已创建", body = LessonCreatedResponse), (status = 404, description = "反馈批次不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn create_lesson_session( State(state): State, headers: HeaderMap, Path(session_id): Path, ) -> Result<(axum::http::StatusCode, Json), ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; assert_session_owner(pool(&state)?, owner_id, session_id).await?; sqlx::query( "DELETE FROM lesson_sessions \ WHERE feedback_session_id = $1 AND status IN ('recording', 'processing')", ) .bind(session_id) .execute(pool(&state)?) .await?; let id = Uuid::new_v4(); let lesson = sqlx::query_as::<_, (Uuid, i32, String, DateTime)>( "INSERT INTO lesson_sessions (id, feedback_session_id, sequence) \ SELECT $1, $2, COALESCE(MAX(sequence), 0) + 1 \ FROM lesson_sessions WHERE feedback_session_id = $2 \ RETURNING id, sequence, status, started_at", ) .bind(id) .bind(session_id) .fetch_one(pool(&state)?) .await?; touch_session(pool(&state)?, session_id).await?; Ok(( axum::http::StatusCode::CREATED, Json(LessonCreatedResponse { id: lesson.0, sequence: lesson.1, status: lesson.2, started_at: lesson.3, }), )) } #[utoipa::path( post, path = "/api/v1/feedback-lessons/{lesson_id}/segments", tag = "语音反馈", summary = "上传并转录录音片段", description = "上传单个录音片段并加入后端转录队列。接口保存音频后立即返回,页面通过反馈批次状态无感获取转录结果。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("lesson_id" = Uuid, Path, description = "课堂录音 ID") ), request_body(content = AudioSegmentUpload, content_type = "multipart/form-data"), responses( (status = 201, description = "片段已保存并进入转录队列", body = AudioSegmentResponse), (status = 404, description = "课堂录音不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn upload_audio_segment( State(state): State, headers: HeaderMap, Path(lesson_id): Path, mut multipart: Multipart, ) -> Result<(axum::http::StatusCode, Json), ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; let session_id = assert_lesson_owner(pool(&state)?, owner_id, lesson_id).await?; let mut sequence = None; let mut duration_ms = None; let mut audio_format = None; let mut audio = None; while let Some(field) = multipart .next_field() .await .map_err(|error| ApiError::BadRequest(format!("invalid multipart body: {error}")))? { let name = field.name().unwrap_or_default().to_owned(); match name.as_str() { "sequence" => sequence = Some(parse_multipart_i32(field, "sequence").await?), "duration_ms" => duration_ms = Some(parse_multipart_i32(field, "duration_ms").await?), "audio_format" => { audio_format = Some( field .text() .await .map_err(|_| ApiError::BadRequest("audio_format is invalid".to_owned()))?, ) } "audio" => { let bytes = field .bytes() .await .map_err(|_| ApiError::BadRequest("audio file is invalid".to_owned()))?; if bytes.len() > MAX_AUDIO_UPLOAD_BYTES { return Err(ApiError::BadRequest("audio file exceeds 15 MB".to_owned())); } audio = Some(bytes.to_vec()); } _ => {} } } let sequence = sequence .filter(|value| *value > 0) .ok_or_else(|| ApiError::BadRequest("sequence must be a positive integer".to_owned()))?; let duration_ms = duration_ms .filter(|value| (1..=MAX_SEGMENT_DURATION_MS).contains(value)) .ok_or_else(|| { ApiError::BadRequest("duration_ms must be between 1 and 600000".to_owned()) })?; let audio_format = audio_format .map(|value| value.trim().to_ascii_lowercase()) .filter(|value| matches!(value.as_str(), "mp3" | "aac" | "wav" | "m4a" | "amr")) .ok_or_else(|| ApiError::BadRequest("unsupported audio_format".to_owned()))?; let audio = audio .filter(|bytes| !bytes.is_empty()) .ok_or_else(|| ApiError::BadRequest("audio file cannot be empty".to_owned()))?; let file_id = Uuid::new_v4(); let directory = state .audio_storage_dir .join(owner_id.to_string()) .join(lesson_id.to_string()); tokio::fs::create_dir_all(&directory) .await .map_err(internal_io_error)?; let audio_path = directory.join(format!("{file_id}.{audio_format}")); tokio::fs::write(&audio_path, &audio) .await .map_err(internal_io_error)?; let segment_id = sqlx::query_scalar::<_, Uuid>( "INSERT INTO audio_segments \ (id, lesson_session_id, sequence, duration_ms, audio_format, audio_path, status) \ VALUES ($1, $2, $3, $4, $5, $6, 'processing') \ ON CONFLICT (lesson_session_id, sequence) DO UPDATE SET \ duration_ms = EXCLUDED.duration_ms, audio_format = EXCLUDED.audio_format, \ audio_path = EXCLUDED.audio_path, status = 'processing', transcript = NULL, \ error_message = NULL, provider_request_id = NULL, processing_started_at = NULL, \ updated_at = now() \ RETURNING id", ) .bind(file_id) .bind(lesson_id) .bind(sequence) .bind(duration_ms) .bind(&audio_format) .bind(audio_path.to_string_lossy().as_ref()) .fetch_one(pool(&state)?) .await?; sqlx::query( "UPDATE lesson_sessions SET status = 'processing', updated_at = now() WHERE id = $1", ) .bind(lesson_id) .execute(pool(&state)?) .await?; touch_session(pool(&state)?, session_id).await?; Ok(( axum::http::StatusCode::CREATED, Json(AudioSegmentResponse { id: segment_id, sequence, duration_ms, status: "processing".to_owned(), error_message: None, }), )) } #[utoipa::path( post, path = "/api/v1/feedback-lessons/{lesson_id}/finish", tag = "语音反馈", summary = "结束一节课堂录音", description = "汇总该节课的片段状态和转录文本;短录音会标记为可直接加入反馈。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("lesson_id" = Uuid, Path, description = "课堂录音 ID") ), responses( (status = 200, description = "课堂录音汇总结果", body = FinishedLessonResponse), (status = 404, description = "课堂录音不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn finish_lesson_session( State(state): State, headers: HeaderMap, Path(lesson_id): Path, ) -> Result, ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; let session_id = assert_lesson_owner(pool(&state)?, owner_id, lesson_id).await?; let segments = sqlx::query_as::<_, (String, i32, Option)>( "SELECT status, duration_ms, transcript FROM audio_segments \ WHERE lesson_session_id = $1 ORDER BY sequence", ) .bind(lesson_id) .fetch_all(pool(&state)?) .await?; let duration_ms = segments.iter().map(|segment| segment.1).sum::(); let transcript = segments .iter() .filter_map(|segment| segment.2.as_deref()) .collect::>() .join("\n"); let status = if segments.is_empty() || segments.iter().any(|segment| segment.0 == "failed") { "failed" } else if segments.iter().any(|segment| segment.0 == "processing") { "processing" } else { "ready" }; sqlx::query( "UPDATE lesson_sessions SET status = $2, duration_ms = $3, ended_at = now(), updated_at = now() \ WHERE id = $1", ) .bind(lesson_id) .bind(status) .bind(duration_ms) .execute(pool(&state)?) .await?; touch_session(pool(&state)?, session_id).await?; Ok(Json(FinishedLessonResponse { id: lesson_id, status: status.to_owned(), duration_ms, can_apply_directly: status == "ready" && duration_ms <= SHORT_RECORDING_DURATION_MS && transcript.chars().count() <= SHORT_TRANSCRIPT_CHARS, transcript, })) } #[utoipa::path( post, path = "/api/v1/feedback-lessons/{lesson_id}/mark-applied", tag = "语音反馈", summary = "标记短录音已加入反馈", description = "短录音文字直接加入反馈内容后,标记该节课无需再次汇总。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("lesson_id" = Uuid, Path, description = "课堂录音 ID") ), request_body(content = MarkLessonAppliedInput, content_type = "application/json"), responses( (status = 204, description = "已标记"), (status = 404, description = "课堂录音不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn mark_lesson_applied( State(state): State, headers: HeaderMap, Path(lesson_id): Path, Json(input): Json, ) -> Result { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; validate_draft_content(&input.content)?; let pool = pool(&state)?; let session_id = assert_lesson_owner(pool, owner_id, lesson_id).await?; let mut transaction = pool.begin().await?; let result = sqlx::query( "UPDATE lesson_sessions SET applied_directly = TRUE, updated_at = now() \ WHERE id = $1 AND status = 'ready'", ) .bind(lesson_id) .execute(&mut *transaction) .await?; if result.rows_affected() == 0 { return Err(ApiError::BadRequest("lesson is not ready".to_owned())); } sqlx::query("UPDATE feedback_sessions SET content = $2, updated_at = now() WHERE id = $1") .bind(session_id) .bind(&input.content) .execute(&mut *transaction) .await?; transaction.commit().await?; Ok(axum::http::StatusCode::NO_CONTENT) } #[utoipa::path( post, path = "/api/v1/audio-segments/{segment_id}/retry", tag = "语音反馈", summary = "重试失败的录音转录", description = "将后端保留的失败音频重新加入转录队列,接口立即返回。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("segment_id" = Uuid, Path, description = "录音片段 ID") ), responses( (status = 200, description = "片段已重新进入转录队列", body = AudioSegmentResponse), (status = 404, description = "录音片段不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn retry_audio_segment( State(state): State, headers: HeaderMap, Path(segment_id): Path, ) -> Result, ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; let segment = sqlx::query_as::<_, (i32, i32, String, Uuid)>( "SELECT s.sequence, s.duration_ms, s.audio_path, l.id \ FROM audio_segments s \ JOIN lesson_sessions l ON l.id = s.lesson_session_id \ JOIN feedback_sessions f ON f.id = l.feedback_session_id \ WHERE s.id = $1 AND f.owner_id = $2 AND f.status = 'active'", ) .bind(segment_id) .bind(owner_id) .fetch_optional(pool(&state)?) .await? .ok_or(ApiError::NotFound)?; tokio::fs::metadata(&segment.2) .await .map_err(internal_io_error)?; sqlx::query( "UPDATE audio_segments SET status = 'processing', error_message = NULL, \ processing_started_at = NULL, updated_at = now() WHERE id = $1", ) .bind(segment_id) .execute(pool(&state)?) .await?; sqlx::query( "UPDATE lesson_sessions SET status = 'processing', updated_at = now() WHERE id = $1", ) .bind(segment.3) .execute(pool(&state)?) .await?; let session_id = sqlx::query_scalar::<_, Uuid>( "SELECT feedback_session_id FROM lesson_sessions WHERE id = $1", ) .bind(segment.3) .fetch_one(pool(&state)?) .await?; touch_session(pool(&state)?, session_id).await?; Ok(Json(AudioSegmentResponse { id: segment_id, sequence: segment.0, duration_ms: segment.1, status: "processing".to_owned(), error_message: None, })) } #[utoipa::path( post, path = "/api/v1/feedback-sessions/{session_id}/generate", tag = "语音反馈", summary = "生成汇总反馈", description = "将尚未处理的长录音或多节课堂转录与教师已有内容合并。未配置大模型时使用本地抽取式汇总。", params( ("Authorization" = String, Header, description = "Bearer 会话令牌", example = "Bearer "), ("session_id" = Uuid, Path, description = "反馈批次 ID") ), request_body(content = GenerateFeedbackInput, content_type = "application/json"), responses( (status = 200, description = "汇总后的反馈正文", body = GenerateFeedbackResponse), (status = 404, description = "反馈批次不存在或不属于当前用户", body = ErrorBody), CommonApiErrors ) )] async fn generate_feedback_content( State(state): State, headers: HeaderMap, Path(session_id): Path, Json(input): Json, ) -> Result, ApiError> { let owner_id = state .auth .authenticate(state.pool.as_ref(), &headers) .await? .user_id; validate_draft_content(&input.existing_content)?; assert_session_owner(pool(&state)?, owner_id, session_id).await?; let pending_status = sqlx::query_as::<_, (bool, bool)>( "SELECT \ EXISTS(SELECT 1 FROM audio_segments s JOIN lesson_sessions l ON l.id = s.lesson_session_id \ WHERE l.feedback_session_id = $1 AND s.status = 'processing'), \ EXISTS(SELECT 1 FROM audio_segments s JOIN lesson_sessions l ON l.id = s.lesson_session_id \ WHERE l.feedback_session_id = $1 AND s.status = 'failed')", ) .bind(session_id) .fetch_one(pool(&state)?) .await?; if pending_status.0 { return Err(ApiError::BadRequest( "录音仍在后台转录中,请稍后再生成反馈".to_owned(), )); } if pending_status.1 { return Err(ApiError::BadRequest( "部分录音转录失败,请重试后再生成反馈".to_owned(), )); } let lessons = sqlx::query_as::<_, (Uuid, String)>( "SELECT l.id, string_agg(s.transcript, E'\\n' ORDER BY s.sequence) \ FROM lesson_sessions l \ JOIN audio_segments s ON s.lesson_session_id = l.id \ WHERE l.feedback_session_id = $1 AND l.status = 'ready' \ AND l.applied_directly = FALSE AND l.included_in_generation = FALSE \ AND s.status = 'ready' AND s.transcript IS NOT NULL \ GROUP BY l.id, l.sequence ORDER BY l.sequence", ) .bind(session_id) .fetch_all(pool(&state)?) .await?; if lessons.is_empty() { return Err(ApiError::BadRequest("没有待汇总的录音内容".to_owned())); } let transcripts = lessons .iter() .map(|lesson| lesson.1.clone()) .collect::>(); let summary = state .summary .summarize(&input.existing_content, &transcripts) .await .map_err(ApiError::ServiceUnavailable)?; let lesson_ids = lessons.iter().map(|lesson| lesson.0).collect::>(); let pool = pool(&state)?; let mut transaction = pool.begin().await?; sqlx::query( "UPDATE lesson_sessions SET included_in_generation = TRUE, updated_at = now() \ WHERE id = ANY($1)", ) .bind(&lesson_ids) .execute(&mut *transaction) .await?; sqlx::query("UPDATE feedback_sessions SET content = $2, updated_at = now() WHERE id = $1") .bind(session_id) .bind(&summary.content) .execute(&mut *transaction) .await?; transaction.commit().await?; Ok(Json(GenerateFeedbackResponse { content: summary.content, method: summary.method.to_owned(), })) } async fn load_session_response( pool: &PgPool, session: FeedbackSessionRow, ) -> Result { let lessons = sqlx::query_as::<_, LessonRow>( "SELECT id, sequence, status, duration_ms, applied_directly, included_in_generation, \ started_at, ended_at FROM lesson_sessions \ WHERE feedback_session_id = $1 ORDER BY sequence", ) .bind(session.id) .fetch_all(pool) .await?; let segments = sqlx::query_as::<_, SegmentRow>( "SELECT s.id, s.lesson_session_id, s.sequence, s.duration_ms, s.status, s.error_message \ FROM audio_segments s JOIN lesson_sessions l ON l.id = s.lesson_session_id \ WHERE l.feedback_session_id = $1 ORDER BY l.sequence, s.sequence", ) .bind(session.id) .fetch_all(pool) .await?; let processing_segment_count = segments .iter() .filter(|segment| segment.status == "processing") .count() + lessons .iter() .filter(|lesson| matches!(lesson.status.as_str(), "recording" | "processing")) .count(); let failed_segment_count = segments .iter() .filter(|segment| segment.status == "failed") .count(); let total_duration_ms = lessons .iter() .map(|lesson| i64::from(lesson.duration_ms)) .sum(); let needs_generation = lessons.iter().any(|lesson| { lesson.status == "ready" && !lesson.applied_directly && !lesson.included_in_generation }); let mut segments_by_lesson: HashMap> = HashMap::new(); for segment in segments { segments_by_lesson .entry(segment.lesson_session_id) .or_default() .push(AudioSegmentResponse { id: segment.id, sequence: segment.sequence, duration_ms: segment.duration_ms, status: segment.status, error_message: segment.error_message, }); } let lesson_count = lessons.len(); let lesson_responses = lessons .into_iter() .map(|lesson| LessonSummaryResponse { id: lesson.id, sequence: lesson.sequence, status: lesson.status, duration_ms: lesson.duration_ms, applied_directly: lesson.applied_directly, included_in_generation: lesson.included_in_generation, segments: segments_by_lesson.remove(&lesson.id).unwrap_or_default(), started_at: lesson.started_at, ended_at: lesson.ended_at, }) .collect(); Ok(FeedbackSessionResponse { id: session.id, profile_id: session.profile_id, feedback_date: session.feedback_date, content: session.content, status: session.status, lesson_count, total_duration_ms, processing_segment_count, failed_segment_count, needs_generation, lessons: lesson_responses, created_at: session.created_at, updated_at: session.updated_at, }) } async fn parse_multipart_i32( field: axum::extract::multipart::Field<'_>, name: &str, ) -> Result { field .text() .await .map_err(|_| ApiError::BadRequest(format!("{name} is invalid")))? .parse() .map_err(|_| ApiError::BadRequest(format!("{name} must be an integer"))) } fn pool(state: &AppState) -> Result<&PgPool, ApiError> { state.pool.as_ref().ok_or(ApiError::DatabaseUnavailable) } async fn assert_profile_owner( pool: &PgPool, owner_id: Uuid, profile_id: Uuid, ) -> Result<(), ApiError> { let exists = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM student_profiles WHERE id = $1 AND owner_id = $2)", ) .bind(profile_id) .bind(owner_id) .fetch_one(pool) .await?; if exists { Ok(()) } else { Err(ApiError::NotFound) } } async fn assert_session_owner( pool: &PgPool, owner_id: Uuid, session_id: Uuid, ) -> Result<(), ApiError> { let exists = sqlx::query_scalar::<_, bool>( "SELECT EXISTS(SELECT 1 FROM feedback_sessions WHERE id = $1 AND owner_id = $2 AND status = 'active')", ) .bind(session_id) .bind(owner_id) .fetch_one(pool) .await?; if exists { Ok(()) } else { Err(ApiError::NotFound) } } async fn assert_lesson_owner( pool: &PgPool, owner_id: Uuid, lesson_id: Uuid, ) -> Result { sqlx::query_scalar::<_, Uuid>( "SELECT f.id FROM lesson_sessions l \ JOIN feedback_sessions f ON f.id = l.feedback_session_id \ WHERE l.id = $1 AND f.owner_id = $2 AND f.status = 'active'", ) .bind(lesson_id) .bind(owner_id) .fetch_optional(pool) .await? .ok_or(ApiError::NotFound) } async fn touch_session(pool: &PgPool, session_id: Uuid) -> Result<(), ApiError> { sqlx::query("UPDATE feedback_sessions SET updated_at = now() WHERE id = $1") .bind(session_id) .execute(pool) .await?; Ok(()) } fn internal_io_error(error: std::io::Error) -> ApiError { tracing::error!(?error, "audio file operation failed"); ApiError::Internal } fn validate_draft_content(content: &str) -> Result<(), ApiError> { if content.chars().count() > 2000 { return Err(ApiError::BadRequest( "feedback content cannot exceed 2000 characters".to_owned(), )); } Ok(()) }