Enhancements to Classroom Progress Updates, RAG Ingestion, and Protobuf Contracts

docker-compose.yml

@@ -79,6 +79,7 @@ services:
       - Cloudinary__CloudName=${CLOUDINARY_CLOUD_NAME}
       - Cloudinary__ApiKey=${CLOUDINARY_API_KEY}
       - Cloudinary__ApiSecret=${CLOUDINARY_API_SECRET}
+      - Pdflayer__ApiKey=e4a39484d4c94acacd0da61ba5970b09
     ports:
       - 7001:80
       - 5001:5001
@@ -608,7 +609,7 @@ services:
       - DEEPSEEK_API_KEY=${DEEPSEEK_API_KEY}
       - RESOURCE_GRPC_CERT_PATH=${RESOURCE_GRPC_CERT_PATH}
     ports:
-      - '7010:80' 
+      - '7010:80'
       - '5010:5010' # gRPC
     # depends_on:
     #   - qdrant

src/BuildingBlocks/EventBus.Messages/Subscription/SubscriptionCancelledEvent.cs

@@ -0,0 +1,9 @@
+using Contracts.Domains;
+
+namespace EventBus.Messages.Subscription;
+
+public record SubscriptionCancelledEvent : DomainEvent
+{
+    public List<int> LicenseAssignmentIds { get; set; } = new();
+}
+

src/BuildingBlocks/Infrastructure/Abstractions/Services/PdfLayer/PdfService.cs

@@ -1,14 +1,12 @@
 using Contracts.Abstractions.Services;
 using Microsoft.Extensions.Configuration;
-using System;
-using System.Threading.Tasks;
 
 namespace Infrastructure.Abstractions.Services.PdfLayer
 {
     public class PdfService(HttpClient httpClient, IConfiguration configuration) : IPdfService
     {
         private readonly HttpClient _httpClient = httpClient;
-        private readonly string _apiKey = configuration["Pdflayer:ApiKey"] ?? throw new ArgumentNullException("pdflayer:ApiKey");
+        private readonly string _apiKey = configuration["Pdflayer:ApiKey"] ?? throw new ArgumentNullException("Pdflayer:ApiKey");
 
         public async Task<byte[]> ConvertHtmlToPdfAsync(string htmlContent)
         {

src/BuildingBlocks/Shared/Protos/Classroom/certificates.proto

@@ -22,7 +22,7 @@ service GrpcCertificate {
 //    };
 //  }
 
-  rpc GetCertificateById (GetCertificateRequest) returns (GrpcCertificateResponse) {
+  rpc GetCertificateById (GetCertificateRequest) returns (GrpcCertificateDetail) {
     option (google.api.http) = {
       get: "/api/certificates/{id}"
     };
@@ -91,6 +91,21 @@ message GrpcCertificateModel {
   google.protobuf.Timestamp completedAt = 11;
 }
 
+message GrpcCertificateDetail {
+  int32 id = 1;
+  string userId = 2;
+  google.protobuf.Int32Value courseEnrollmentId = 3;
+  google.protobuf.Int32Value curriculumEnrollmentId = 4;
+  string certificateType = 5;
+  google.protobuf.Timestamp issueDate = 6;
+  string verificationCode = 7;
+  string certificateUrl = 8;
+  google.protobuf.StringValue userName = 9;
+  google.protobuf.StringValue title = 10;
+  google.protobuf.Timestamp completedAt = 11;
+  repeated string lessons = 12;
+}
+
 enum CertificateType {
   UNKNOWN = 0;
   COURSE = 1;

src/BuildingBlocks/Shared/Protos/Classroom/classrooms.proto

@@ -255,6 +255,18 @@ message GetClassroomLearningSnapshotRequest {
 	google.protobuf.Int32Value days_back = 3;
 }
 
+message GrpcStudentProgressSummary {
+	string student_id = 1;
+
+	double assessment_completion_rate = 2; 
+	int32 total_assessments = 3;
+	int32 completed_assessments = 4;
+
+	double content_completion_rate = 5;
+	int32 total_sections = 6;
+	int32 completed_sections = 7;
+}
+
 message GrpcClassroomLearningSnapshotResponse {
 	GrpcClassroomBasicInfo classroom = 1;
 	repeated GrpcStudentLearningData students = 2;
@@ -265,6 +277,8 @@ message GrpcClassroomLearningSnapshotResponse {
 	repeated GrpcSectionProgressData section_progress = 7;
 	repeated GrpcTopicCatalogItem topics_catalog = 8;
 	GrpcAnalysisPeriod analysis_period = 9;
+
+	repeated GrpcStudentProgressSummary student_progress_summaries = 10;
 }
 
 message GrpcClassroomBasicInfo {

src/BuildingBlocks/Shared/Protos/Order/organization.proto

@@ -305,4 +305,5 @@ message CurriculumSubscriptionInfo {
   int32 subscriptionId = 1;
   string startDate = 2;
   string endDate = 3;
+  string planName = 4;
 }

src/BuildingBlocks/Shared/Protos/Order/organization_subscription_order.proto

@@ -33,6 +33,13 @@ service GrpcOrganizationSubscriptionOrderService {
         };
     }
 
+    rpc CancelOrganizationSubscriptionOrder (CancelOrganizationSubscriptionOrderRequest) returns (google.protobuf.Empty) {
+        option (google.api.http) = {
+            post: "/api/organization-subscription-orders/{id}/cancel"
+            body: "*"
+        };
+    }
+
     rpc GetOrganizationSubscriptionOrderById (GetOrganizationSubscriptionOrderRequest) returns (GrpcOrganizationSubscriptionOrderDetail) {
         option (google.api.http) = {
             get: "/api/organization-subscription-orders/{id}"
@@ -80,6 +87,10 @@ message DeleteOrganizationSubscriptionOrderRequest {
     int32 id = 1;
 }
 
+message CancelOrganizationSubscriptionOrderRequest {
+    int32 id = 1;
+}
+
 message GetOrganizationSubscriptionOrderRequest {
     int32 id = 1;
 }

src/Services/AIService/app/api/http/dependencies.py

@@ -9,6 +9,7 @@
 
 from app.core.llm.client import LLMClient
 from app.core.rag.ingestion_pipeline import IngestionPipeline
+from app.core.rag.ingestion_service import IngestionService
 from app.core.rag.document_processor import DocumentProcessor
 from app.core.embedding.pipeline import EmbeddingPipeline, get_embedding_pipeline
 from app.core.graph.builder import GraphBuilder
@@ -285,11 +286,38 @@ def get_classroom_snapshot_updater() -> ClassroomSnapshotUpdater:
     )
 
 
+@lru_cache(maxsize=1)
+def get_ingestion_service() -> Optional[IngestionService]:
+    """
+    Provide ingestion service for RAG indexing with debouncing.
+    Returns None if RAG features are disabled or unavailable.
+    """
+    try:
+        ingestion_pipeline = get_ingestion_pipeline()
+        if not ingestion_pipeline:
+            return None
+
+        classroom_repository = get_classroom_repository()
+
+        return IngestionService(
+            ingestion_pipeline=ingestion_pipeline,
+            classroom_repository=classroom_repository,
+            debounce_seconds=getattr(settings, 'RAG_INGESTION_DEBOUNCE_SECONDS', 300),
+            ingestion_ttl_hours=getattr(settings, 'RAG_INGESTION_TTL_HOURS', 24),
+        )
+    except Exception as e:
+        logger = logging.getLogger(__name__)
+        logger.warning(f"Failed to initialize ingestion service: {e}. Continuing without RAG ingestion.")
+        return None
+
+
 @lru_cache(maxsize=1)
 def get_classroom_snapshot_event_handler() -> ClassroomSnapshotEventHandler:
+    ingestion_service = get_ingestion_service()
     return ClassroomSnapshotEventHandler(
         snapshot_store=get_classroom_snapshot_store(),
         snapshot_updater=get_classroom_snapshot_updater(),
+        ingestion_service=ingestion_service,
     )
 
 
@@ -383,6 +411,7 @@ def get_teacher_service() -> TeacherService:
         classroom_snapshot_store=get_classroom_snapshot_store(),
         classroom_snapshot_updater=get_classroom_snapshot_updater(),
         direct_grading_pipeline=direct_grading_pipeline,
+        ingestion_service=get_ingestion_service(),
     )
 
 

src/Services/AIService/app/api/http/routers/teacher.py

@@ -44,6 +44,16 @@ class BuildGraphRequest(BaseModel):
     force_rebuild: bool = False
 
 
+class TriggerProgressEventRequest(BaseModel):
+    """Request model for triggering test progress event"""
+    classroom_id: int
+    student_id: str
+    course_enrollment_id: int = 1
+    course_id: int = 1
+    progress_percentage: int = 50
+    status: str = "InProgress"
+
+
 @router.post("/student-analysis", response_model=InterventionResponse)
 async def student_analysis(
     request: StudentAnalysisRequest,
@@ -132,3 +142,136 @@ async def build_graph(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+@router.post("/test/trigger-progress-event")
+async def trigger_progress_event(
+    request: TriggerProgressEventRequest,
+    direct: bool = False,  # Query parameter: ?direct=true
+):
+    """
+    Test endpoint to trigger ClassroomStudentProgressUpdatedEvent.
+
+    This can work in two modes:
+    1. RabbitMQ mode (default): Publishes event to RabbitMQ
+    2. Direct mode (direct=true): Directly calls event handler (useful when RabbitMQ unavailable)
+
+    The event will trigger:
+    1. Snapshot refresh
+    2. RAG ingestion (with debouncing - waits 5 minutes before ingesting)
+
+    Use this to test the event-driven ingestion flow.
+    """
+    from app.core.snapshot.events import ClassroomEvent
+    from app.api.http.dependencies import get_classroom_snapshot_event_handler
+
+    # Create event payload matching C# event structure
+    event_data = {
+        "StudentId": request.student_id,
+        "ClassroomId": request.classroom_id,
+        "CourseEnrollmentId": request.course_enrollment_id,
+        "CourseId": request.course_id,
+        "ProgressPercentage": request.progress_percentage,
+        "Status": request.status,
+    }
+
+    if direct:
+        # Direct mode: Call event handler directly (bypass RabbitMQ)
+        try:
+            event_handler = get_classroom_snapshot_event_handler()
+            classroom_event = ClassroomEvent(
+                type="STUDENT_PROGRESS_UPDATED",
+                classroom_id=request.classroom_id,
+                student_id=request.student_id,
+                payload={
+                    "course_enrollment_id": request.course_enrollment_id,
+                    "course_id": request.course_id,
+                    "progress_percentage": request.progress_percentage,
+                    "status": request.status,
+                },
+            )
+            await event_handler.handle_event(classroom_event)
+
+            logger.info(
+                f"[TeacherRouter] Test progress event handled directly for classroom {request.classroom_id}, "
+                f"student {request.student_id}"
+            )
+
+            return {
+                "success": True,
+                "message": "Event handled directly (bypassed RabbitMQ)",
+                "event_data": event_data,
+                "note": "Event was processed directly. RAG ingestion will be scheduled with 5-minute debounce.",
+            }
+        except Exception as e:
+            logger.error("[TeacherRouter] Error handling event directly: %s", e, exc_info=True)
+            raise HTTPException(status_code=500, detail=f"Failed to handle event: {str(e)}")
+    else:
+        # RabbitMQ mode: Publish to RabbitMQ
+        try:
+            import aio_pika
+            import json
+            from app.infrastructure.config.settings import settings
+
+            exchange_name = "EventBus.Messages:ClassroomStudentProgressUpdatedEvent"
+            routing_key = "EventBus.Messages:ClassroomStudentProgressUpdatedEvent"
+
+            # Connect and publish
+            connection = await aio_pika.connect_robust(settings.RABBITMQ_URL)
+
+            try:
+                async with connection:
+                    channel = await connection.channel()
+
+                    # Declare exchange
+                    try:
+                        exchange = await channel.declare_exchange(
+                            exchange_name,
+                            aio_pika.ExchangeType.TOPIC,
+                            durable=True,
+                        )
+                    except Exception:
+                        exchange = await channel.declare_exchange(
+                            exchange_name,
+                            aio_pika.ExchangeType.FANOUT,
+                            durable=True,
+                        )
+
+                    # Publish message
+                    message_body = json.dumps(event_data).encode("utf-8")
+                    message = aio_pika.Message(
+                        body=message_body,
+                        content_type="application/json",
+                        delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+                    )
+
+                    await exchange.publish(
+                        message,
+                        routing_key=routing_key,
+                    )
+
+                    logger.info(
+                        f"[TeacherRouter] Test progress event published to RabbitMQ for classroom {request.classroom_id}, "
+                        f"student {request.student_id}"
+                    )
+
+                    return {
+                        "success": True,
+                        "message": "Event published to RabbitMQ successfully",
+                        "event_data": event_data,
+                        "note": "Event will be consumed by ClassroomProgressEventConsumer. "
+                               "RAG ingestion will be scheduled with 5-minute debounce.",
+                    }
+            finally:
+                await connection.close()
+
+        except Exception as e:
+            logger.warning(
+                f"[TeacherRouter] Failed to publish to RabbitMQ: {e}. "
+                f"Hint: Use ?direct=true to test without RabbitMQ"
+            )
+            raise HTTPException(
+                status_code=503,
+                detail=f"Failed to publish event to RabbitMQ: {str(e)}. "
+                       f"Use ?direct=true query parameter to test without RabbitMQ."
+            )
+
+

src/Services/AIService/app/core/agent/plan_solve_agent.py

@@ -86,17 +86,9 @@ async def _generate_plan(self, question: str) -> List[str]:
         """Generate action plan"""
         prompt = self.PLANNER_PROMPT.format(question=question)
         messages: List[LLMMessage] = [{"role": "user", "content": prompt}]
-
-        # Log request data before calling LLM (full prompt, no truncate)
-        logger.info(
-            f"[Plan-Solve] Calling LLM to generate plan | "
-            f"question={question}, prompt_length={len(prompt)}, use_remote={self.use_remote} | "
-            f"full_prompt={prompt}"
-        )
-
         response = await self.llm.generate(messages, use_remote=self.use_remote)
         response_text = response.content if hasattr(response, 'content') else str(response)
-        
+
         # Parse Python list
         try:
             if "```python" in response_text:
@@ -107,6 +99,7 @@ async def _generate_plan(self, question: str) -> List[str]:
                 plan_str = response_text.strip()
 
             plan = ast.literal_eval(plan_str)
+
             return plan if isinstance(plan, list) else []
         except Exception as e:
             logger.warning(f"[Plan-Solve] Failed to parse plan: {e}")
@@ -128,15 +121,6 @@ async def _execute_step(
             current_step=current_step
         )
         messages: List[LLMMessage] = [{"role": "user", "content": prompt}]
-
-        # Log request data before calling LLM (full prompt, no truncate)
-        logger.info(
-            f"[Plan-Solve] Calling LLM to execute step | "
-            f"current_step={current_step}, question={question}, "
-            f"prompt_length={len(prompt)}, use_remote={self.use_remote} | "
-            f"full_prompt={prompt}"
-        )
-
         response = await self.llm.generate(messages, use_remote=self.use_remote)
         return response.content if hasattr(response, 'content') else str(response)