feat: add stream_user_stage_status handler and service

Author: wangeguo

Cargo.lock

@@ -29,6 +29,7 @@ clap = { version = "4.5.41", features = ["derive", "env"] }
 dotenv = "0.15.0"
 flate2 = "1.1.2"
 fs_extra = "1.3.0"
+futures = "0.3.31"
 ghrepo = "0.7.1"
 http-body-util = "0.1.3"
 indexmap = {version = "2.9.0", features = ["serde"] }
@@ -42,6 +43,7 @@ tar = "0.4.44"
 tempfile = "3.20.0"
 thiserror = "2.0.12"
 tokio = { version = "1.46.1", features = ["full"] }
+tokio-stream = "0.1.17"
 tower-http = { version = "0.6.6", features = ["cors"] }
 tracing = "0.1.41"
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }

Cargo.toml

@@ -696,6 +696,52 @@
           }
         ]
       }
+    },
+    "/v1/user/courses/{slug}/stages/{stage_slug}/status": {
+      "get": {
+        "tags": [
+          "User",
+          "Stage"
+        ],
+        "summary": "Stream the status of a specific stage for the current user.",
+        "operationId": "stream_user_stage_status",
+        "parameters": [
+          {
+            "name": "slug",
+            "in": "path",
+            "description": "The slug of course",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          },
+          {
+            "name": "stage_slug",
+            "in": "path",
+            "description": "The slug of stage",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "Successfully started streaming stage status updates"
+          },
+          "404": {
+            "description": "Course or stage not found"
+          },
+          "500": {
+            "description": "Failed to stream stage status"
+          }
+        },
+        "security": [
+          {
+            "JWTBearerAuth": []
+          }
+        ]
+      }
     }
   },
   "components": {
@@ -1107,6 +1153,23 @@
             "description": "Current progress status (in_progress, completed)"
           }
         }
+      },
+      "UserStageStatusResponse": {
+        "type": "object",
+        "required": [
+          "status",
+          "test"
+        ],
+        "properties": {
+          "status": {
+            "type": "string",
+            "description": "Current progress status (in_progress, completed)"
+          },
+          "test": {
+            "type": "string",
+            "description": "Test result status (passed, failed)"
+          }
+        }
       }
     },
     "securitySchemes": {

openapi.json

@@ -16,9 +16,15 @@ use axum::{
     Json,
     extract::{Path, State},
     http::StatusCode,
-    response::IntoResponse,
+    response::{
+        IntoResponse, Sse,
+        sse::{Event, KeepAlive},
+    },
 };
-use std::sync::Arc;
+use futures::{Stream, StreamExt};
+use std::{convert::Infallible, sync::Arc};
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::{error, info};
 
 use crate::{
     context::Context,
@@ -193,3 +199,59 @@ pub async fn complete_stage(
     let res = StageService::complete(ctx, &claims.id, &slug, &req.slug).await?;
     Ok((StatusCode::OK, Json(res)))
 }
+
+/// Stream the status of a specific stage for the current user.
+#[utoipa::path(
+    operation_id = "stream_user_stage_status",
+    get, path = "/v1/user/courses/{slug}/stages/{stage_slug}/status",
+    params(
+        ("slug" = String, description = "The slug of course"),
+        ("stage_slug" = String, description = "The slug of stage"),
+    ),
+    responses(
+        (status = 200, description = "Successfully started streaming stage status updates"),
+        (status = 404, description = "Course or stage not found"),
+        (status = 500, description = "Failed to stream stage status")
+    ),
+    security(("JWTBearerAuth" = [])),
+    tags = ["User", "Stage"]
+)]
+pub async fn stream_user_stage_status(
+    claims: Claims,
+    State(ctx): State<Arc<Context>>,
+    Path((slug, stage_slug)): Path<(String, String)>,
+) -> Sse<impl Stream<Item = axum::response::Result<Event, Infallible>>> {
+    info!(
+        "Starting to stream status updates for stage {} in course {} for user {}...",
+        stage_slug, slug, claims.id
+    );
+
+    // Create a channel for sending status updates.
+    let (sender, receiver) = tokio::sync::mpsc::channel(100);
+
+    // Spawn a background task to fetch and send status updates.
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
+        loop {
+            interval.tick().await;
+            let status =
+                StageService::get_user_stage_status(&ctx, &claims.id, &slug, &stage_slug).await;
+            if let Ok(status) = status {
+                let event = Event::default().json_data(status).unwrap_or_else(|e| {
+                    error!("Failed to serialize status update: {}", e);
+                    Event::default().data("status update error")
+                });
+                if sender.send(event).await.is_err() {
+                    break;
+                }
+            }
+        }
+    });
+
+    // Convert the receiver into a stream.
+    let stream = ReceiverStream::new(receiver);
+    let stream = stream.map(Ok);
+
+    // Return the SSE stream with keep-alive.
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}

src/handler/stage.rs

@@ -134,3 +134,12 @@ impl From<UserStageModel> for UserStageResponse {
         }
     }
 }
+
+#[derive(Debug, Serialize, Deserialize, ToSchema)]
+pub struct UserStageStatusResponse {
+    /// Current progress status (in_progress, completed)
+    pub status: String,
+
+    /// Test result status (passed, failed)
+    pub test: String,
+}

src/response/stage.rs

@@ -56,4 +56,8 @@ pub fn build() -> Router<Arc<Context>> {
         .route("/v1/user/courses/{slug}/stages", get(stage::find_user_stages))
         .route("/v1/user/courses/{slug}/stages", post(stage::complete_stage))
         .route("/v1/user/courses/{slug}/stages/{stage_slug}", get(stage::get_user_stage))
+        .route(
+            "/v1/user/courses/{slug}/stages/{stage_slug}/status",
+            get(stage::stream_user_stage_status),
+        )
 }

src/routes.rs

@@ -20,7 +20,7 @@ use crate::{
     errors::{ApiError, Result},
     model::{UserCourseModel, UserStageModel},
     repository::{CourseRepository, StageRepository},
-    response::{StageDetailResponse, StageResponse, UserStageResponse},
+    response::{StageDetailResponse, StageResponse, UserStageResponse, UserStageStatusResponse},
 };
 
 /// Service for managing stages
@@ -153,4 +153,19 @@ impl StageService {
 
         Ok(())
     }
+
+    /// Get the current status of a stage for the user.
+    pub async fn get_user_stage_status(
+        ctx: &Arc<Context>,
+        user_id: &str,
+        course_slug: &str,
+        stage_slug: &str,
+    ) -> Result<UserStageStatusResponse> {
+        // Fetch the user's stage record from the database.
+        let user_stage =
+            StageRepository::get_user_stage(&ctx.database, user_id, course_slug, stage_slug)
+                .await?;
+
+        Ok(UserStageStatusResponse { status: user_stage.status, test: "failed".into() })
+    }
 }

src/service/stage.rs

@@ -47,6 +47,7 @@ use crate::{handler, request, response};
         handler::stage::find_user_stages,
         handler::stage::complete_stage,
         handler::stage::get_user_stage,
+        handler::stage::stream_user_stage_status
     ),
     components(
         schemas(
@@ -63,6 +64,7 @@ use crate::{handler, request, response};
             request::UpdateUserCourseRequest,
             response::UserCourseResponse,
             response::UserStageResponse,
+            response::UserStageStatusResponse,
         )
     ),
     tags(