egress recorder endpoints added and configed

.env.prod

@@ -25,3 +25,9 @@ captcha_public_key="6LdgCjseAAAAAIwg41-kyyulmwDtqD2Gk3THIwy2"
 captcha_private_key="6LdgCjseAAAAAPHMsIHuQgYAGTJ7_QlhqG4G0NyS"
 ONLINE_CLASS_FRONTEND_DOMAIN="meet.imamjavad.online"
 FCM_API_KEY=""
+
+ONLINE_CLASS_WEB_EGRESS_ENABLED=True
+ONLINE_CLASS_WEB_EGRESS_SERVICE_URL=https://habibmeet.nwhco.ir
+ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN=9b7f2b3e6c1a4d8f9e2c7a1b5d3f6e8c
+ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN=3f9c7e1a8d2b6c4f5a1e9d7b2c8f6a3e
+ONLINE_CLASS_WEB_EGRESS_TIMEOUT=20

apps/course/migrations/0014_courselivesession_web_egress_fields.py

@@ -0,0 +1,78 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("course", "0013_courselivesession_auto_close_fields"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="courselivesession",
+            name="web_egress_error",
+            field=models.TextField(
+                blank=True,
+                help_text="Last WebEgress error message, if any.",
+                null=True,
+                verbose_name="Web Egress Error",
+            ),
+        ),
+        migrations.AddField(
+            model_name="courselivesession",
+            name="web_egress_id",
+            field=models.CharField(
+                blank=True,
+                help_text="Active LiveKit WebEgress identifier for this session.",
+                max_length=255,
+                null=True,
+                verbose_name="Web Egress ID",
+            ),
+        ),
+        migrations.AddField(
+            model_name="courselivesession",
+            name="web_egress_started_at",
+            field=models.DateTimeField(
+                blank=True,
+                help_text="Timestamp when WebEgress recording started.",
+                null=True,
+                verbose_name="Web Egress Started At",
+            ),
+        ),
+        migrations.AddField(
+            model_name="courselivesession",
+            name="web_egress_status",
+            field=models.CharField(
+                choices=[
+                    ("idle", "Idle"),
+                    ("starting", "Starting"),
+                    ("recording", "Recording"),
+                    ("stopping", "Stopping"),
+                    ("processing", "Processing"),
+                    ("completed", "Completed"),
+                    ("failed", "Failed"),
+                ],
+                default="idle",
+                help_text="Current WebEgress recording status for this session.",
+                max_length=20,
+                verbose_name="Web Egress Status",
+            ),
+        ),
+        migrations.AddField(
+            model_name="courselivesession",
+            name="web_egress_stopped_at",
+            field=models.DateTimeField(
+                blank=True,
+                help_text="Timestamp when WebEgress recording stopped.",
+                null=True,
+                verbose_name="Web Egress Stopped At",
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="courselivesession",
+            index=models.Index(
+                fields=["web_egress_status"],
+                name="course_cour_web_egr_5d55cb_idx",
+            ),
+        ),
+    ]

apps/course/models/live_session.py

@@ -6,6 +6,24 @@ from apps.account.models import User
 
 
 class CourseLiveSession(models.Model):
+    WEB_EGRESS_STATUS_IDLE = "idle"
+    WEB_EGRESS_STATUS_STARTING = "starting"
+    WEB_EGRESS_STATUS_RECORDING = "recording"
+    WEB_EGRESS_STATUS_STOPPING = "stopping"
+    WEB_EGRESS_STATUS_PROCESSING = "processing"
+    WEB_EGRESS_STATUS_COMPLETED = "completed"
+    WEB_EGRESS_STATUS_FAILED = "failed"
+
+    WEB_EGRESS_STATUS_CHOICES = (
+        (WEB_EGRESS_STATUS_IDLE, _("Idle")),
+        (WEB_EGRESS_STATUS_STARTING, _("Starting")),
+        (WEB_EGRESS_STATUS_RECORDING, _("Recording")),
+        (WEB_EGRESS_STATUS_STOPPING, _("Stopping")),
+        (WEB_EGRESS_STATUS_PROCESSING, _("Processing")),
+        (WEB_EGRESS_STATUS_COMPLETED, _("Completed")),
+        (WEB_EGRESS_STATUS_FAILED, _("Failed")),
+    )
+
     course = models.ForeignKey(
         Course,
         on_delete=models.CASCADE,
@@ -55,6 +73,38 @@ class CourseLiveSession(models.Model):
         null=True,
         blank=True,
     )
+    web_egress_id = models.CharField(
+        max_length=255,
+        verbose_name=_("Web Egress ID"),
+        help_text=_("Active LiveKit WebEgress identifier for this session."),
+        null=True,
+        blank=True,
+    )
+    web_egress_status = models.CharField(
+        max_length=20,
+        choices=WEB_EGRESS_STATUS_CHOICES,
+        default=WEB_EGRESS_STATUS_IDLE,
+        verbose_name=_("Web Egress Status"),
+        help_text=_("Current WebEgress recording status for this session."),
+    )
+    web_egress_started_at = models.DateTimeField(
+        verbose_name=_("Web Egress Started At"),
+        help_text=_("Timestamp when WebEgress recording started."),
+        null=True,
+        blank=True,
+    )
+    web_egress_stopped_at = models.DateTimeField(
+        verbose_name=_("Web Egress Stopped At"),
+        help_text=_("Timestamp when WebEgress recording stopped."),
+        null=True,
+        blank=True,
+    )
+    web_egress_error = models.TextField(
+        verbose_name=_("Web Egress Error"),
+        help_text=_("Last WebEgress error message, if any."),
+        null=True,
+        blank=True,
+    )
     recorded_file = models.FileField(
         upload_to="live_session_recordings/",
         verbose_name=_("Recorded File"),
@@ -77,6 +127,7 @@ class CourseLiveSession(models.Model):
             models.Index(fields=["course", "created_at"]),
             models.Index(fields=["room_id"]),
             models.Index(fields=["ended_at", "auto_close_after_moderator_exit_at"]),
+            models.Index(fields=["web_egress_status"]),
         ]
 
 

apps/course/services/__init__.py

@@ -1,3 +1,8 @@
 from .plugnmeet import PlugNMeetClient, PlugNMeetError
 
-__all__ = ['PlugNMeetClient', 'PlugNMeetError']
+__all__ = [
+    'PlugNMeetClient',
+    'PlugNMeetError',
+    'WebEgressClient',
+    'WebEgressError',
+]

apps/course/services/web_egress.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+import requests
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+
+
+class WebEgressError(Exception):
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: Optional[int] = None,
+        response_data: Optional[Dict[str, Any]] = None,
+    ):
+        super().__init__(message)
+        self.status_code = status_code
+        self.response_data = response_data or {}
+
+
+class WebEgressClient:
+    def __init__(
+        self,
+        *,
+        base_url: Optional[str] = None,
+        service_token: Optional[str] = None,
+        timeout: Optional[float] = None,
+    ):
+        self.base_url = (
+            base_url
+            or getattr(settings, "ONLINE_CLASS_WEB_EGRESS_SERVICE_URL", "")
+        ).rstrip("/")
+        self.service_token = (
+            service_token
+            or getattr(settings, "ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN", "")
+        )
+        self.timeout = timeout or getattr(
+            settings, "ONLINE_CLASS_WEB_EGRESS_TIMEOUT", 20.0
+        )
+
+        if not self.base_url:
+            raise ImproperlyConfigured(
+                "ONLINE_CLASS_WEB_EGRESS_SERVICE_URL must be configured."
+            )
+
+    def start_recording(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return self._request("POST", "/api/web-egress/start/", json=payload)
+
+    def stop_recording(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return self._request("POST", "/api/web-egress/stop/", json=payload)
+
+    def get_status(self, *, room_id: str, session_id: int) -> Dict[str, Any]:
+        return self._request(
+            "GET",
+            "/api/web-egress/status/",
+            params={"room_id": room_id, "session_id": session_id},
+        )
+
+    def _request(self, method: str, path: str, **kwargs) -> Dict[str, Any]:
+        headers = kwargs.pop("headers", {})
+        headers["Content-Type"] = "application/json"
+        if self.service_token:
+            headers["Authorization"] = f"Bearer {self.service_token}"
+
+        try:
+            response = requests.request(
+                method,
+                f"{self.base_url}{path}",
+                headers=headers,
+                timeout=self.timeout,
+                **kwargs,
+            )
+        except requests.RequestException as exc:
+            raise WebEgressError("Failed to reach WebEgress service.") from exc
+
+        data = self._safe_json(response)
+        if response.status_code >= 400:
+            message = (
+                (data or {}).get("message")
+                or (data or {}).get("detail")
+                or response.text
+                or "WebEgress service request failed."
+            )
+            raise WebEgressError(
+                message,
+                status_code=response.status_code,
+                response_data=data,
+            )
+
+        if data is None:
+            raise WebEgressError("WebEgress service returned an invalid response.")
+
+        if isinstance(data, dict) and data.get("status") is False:
+            raise WebEgressError(
+                data.get("message") or data.get("detail") or "WebEgress failed.",
+                status_code=response.status_code,
+                response_data=data,
+            )
+
+        return data
+
+    @staticmethod
+    def _safe_json(response: requests.Response) -> Optional[Dict[str, Any]]:
+        try:
+            return response.json()
+        except ValueError:
+            return None

apps/course/urls.py

@@ -31,6 +31,8 @@ urlpatterns = [
     path('online/token/validate/', views.CourseOnlineClassTokenValidateAPIView.as_view(), name='course-online-token-validate'),
     re_path(r'(?P<slug>[\w-]+)/online/room/create/$', views.CourseLiveSessionRoomCreateAPIView.as_view(), name='course-live-session-room-create'),
     path('online/room/token/', views.CourseLiveSessionTokenAPIView.as_view(), name='course-live-session-token'),
+    path('online/room/recording/', views.CourseLiveSessionRecordingAPIView.as_view(), name='course-live-session-recording'),
+    path('online/room/recording/web-egress/callback/', views.CourseLiveSessionWebEgressCallbackAPIView.as_view(), name='course-live-session-web-egress-callback'),
     path('online/room/set-recording-title/', views.CourseLiveSessionSetRecordingTitleAPIView.as_view(), name='course-live-session-set-recording-title'),
     path('live-sessions/<str:room_slug>/recorded-file/', views.CourseLiveSessionRecordedFileAPIView.as_view(), name='course-live-session-recorded-file'),
     

apps/course/views/live_session.py

@@ -1,4 +1,5 @@
 import logging
+from datetime import timedelta
 
 from django.core.exceptions import ImproperlyConfigured
 from django.shortcuts import get_object_or_404
@@ -16,6 +17,7 @@ import jwt
 from apps.course.models import Course, CourseLiveSession, Participant, LiveSessionRecording, LiveSessionUser
 from apps.course.serializers import LiveSessionRoomCreateSerializer, LiveSessionTokenSerializer, LiveSessionRecordedFileSerializer, LiveSessionRecordingSerializer
 from apps.course.services.plugnmeet import PlugNMeetClient, PlugNMeetError
+from apps.course.services.web_egress import WebEgressClient, WebEgressError
 from utils.exceptions import AppAPIException
 from django.conf import settings
 logger = logging.getLogger(__name__)
@@ -53,6 +55,89 @@ def get_plugnmeet_webhook_url() -> str:
     return f"{base}/api/courses/plugnmeet/webhook/"
 
 
+def get_online_class_frontend_base() -> str:
+    base = getattr(
+        settings,
+        "ONLINE_CLASS_FRONTEND_DOMAIN",
+        getattr(settings, "SITE_DOMAIN", ""),
+    ).rstrip("/")
+    if base and not base.startswith(("http://", "https://")):
+        base = f"https://{base}"
+    return base
+
+
+def get_web_egress_callback_url() -> str:
+    base = getattr(settings, "SITE_DOMAIN", "").rstrip("/")
+    if not base:
+        raise ImproperlyConfigured(
+            "SITE_DOMAIN must be configured for WebEgress callback delivery."
+        )
+    return f"{base}/api/courses/online/room/recording/web-egress/callback/"
+
+
+def decode_plugnmeet_access_token(raw_token: str) -> dict:
+    try:
+        return jwt.decode(
+            raw_token,
+            settings.PLUGNMEET_API_SECRET,
+            algorithms=["HS256"],
+        )
+    except jwt.PyJWTError as exc:
+        raise AppAPIException(
+            {"message": "Invalid or expired meeting access token."},
+            status_code=status.HTTP_401_UNAUTHORIZED,
+        ) from exc
+
+
+def extract_meeting_token_from_request(request) -> str:
+    auth_header = request.headers.get("Authorization", "").strip()
+    if auth_header.lower().startswith("bearer "):
+        return auth_header[7:].strip()
+    if auth_header.lower().startswith("token "):
+        return auth_header[6:].strip()
+    if auth_header:
+        return auth_header
+    token = request.data.get("access_token") or request.query_params.get("access_token")
+    if token:
+        return token
+    raise AppAPIException(
+        {"message": "Meeting access token is required."},
+        status_code=status.HTTP_401_UNAUTHORIZED,
+    )
+
+
+def get_session_from_meeting_token(request, *, require_admin: bool = False):
+    meeting_token = extract_meeting_token_from_request(request)
+    claims = decode_plugnmeet_access_token(meeting_token)
+
+    room_id = claims.get("room_id")
+    if not room_id:
+        raise AppAPIException(
+            {"message": "Meeting token is missing room information."},
+            status_code=status.HTTP_400_BAD_REQUEST,
+        )
+
+    session = get_object_or_404(CourseLiveSession, room_id=room_id)
+    if require_admin and not claims.get("is_admin"):
+        raise AppAPIException(
+            {"message": "Only moderators can control recording."},
+            status_code=status.HTTP_403_FORBIDDEN,
+        )
+
+    return session, claims
+
+
+def is_web_egress_recording_status(status_value: str) -> bool:
+    return status_value in {
+        CourseLiveSession.WEB_EGRESS_STATUS_STARTING,
+        CourseLiveSession.WEB_EGRESS_STATUS_RECORDING,
+    }
+
+
+def build_web_egress_title(session: CourseLiveSession) -> str:
+    return session.recording_title or f"{session.subject} - Recording"
+
+
 class CourseLiveSessionRoomCreateAPIView(GenericAPIView):
     permission_classes = [IsAuthenticated]
     authentication_classes = [TokenAuthentication]
@@ -656,6 +741,420 @@ class CourseLiveSessionTokenAPIView(GenericAPIView):
         return None
 
 
+class CourseLiveSessionRecordingAPIView(GenericAPIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+
+    @swagger_auto_schema(
+        operation_description="Get or control WebEgress recording state for the current room.",
+        tags=["Imam-Javad - Course"],
+        responses={200: openapi.Response(description="Recording state returned successfully")},
+    )
+    def get(self, request, *args, **kwargs):
+        session, _ = get_session_from_meeting_token(request, require_admin=False)
+        self._refresh_status_from_service(session)
+        return Response(self._build_status_payload(session), status=status.HTTP_200_OK)
+
+    @swagger_auto_schema(
+        operation_description="Start or stop WebEgress recording for the current room.",
+        tags=["Imam-Javad - Course"],
+        request_body=openapi.Schema(
+            type=openapi.TYPE_OBJECT,
+            required=["action"],
+            properties={
+                "action": openapi.Schema(
+                    type=openapi.TYPE_STRING,
+                    enum=["start", "stop"],
+                    description="Recording control action.",
+                ),
+            },
+        ),
+        responses={200: openapi.Response(description="Recording action completed successfully")},
+    )
+    def post(self, request, *args, **kwargs):
+        if not getattr(settings, "ONLINE_CLASS_WEB_EGRESS_ENABLED", False):
+            raise AppAPIException(
+                {"message": "WebEgress recording is not enabled."},
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            )
+
+        session, claims = get_session_from_meeting_token(request, require_admin=True)
+        action = str(request.data.get("action") or "").strip().lower()
+        if action not in {"start", "stop"}:
+            raise AppAPIException(
+                {"message": "Invalid recording action."},
+                status_code=status.HTTP_400_BAD_REQUEST,
+            )
+
+        if action == "start":
+            payload = self._start_recording(session, claims)
+        else:
+            payload = self._stop_recording(session)
+
+        return Response(payload, status=status.HTTP_200_OK)
+
+    def _start_recording(self, session: CourseLiveSession, claims: dict) -> dict:
+        if is_web_egress_recording_status(session.web_egress_status):
+            return self._build_status_payload(
+                session,
+                message="Recording is already active or being processed.",
+            )
+
+        client = WebEgressClient()
+        now = timezone.now()
+        session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_STARTING
+        session.web_egress_started_at = now
+        session.web_egress_stopped_at = None
+        session.web_egress_error = None
+        session.save(
+            update_fields=[
+                "web_egress_status",
+                "web_egress_started_at",
+                "web_egress_stopped_at",
+                "web_egress_error",
+                "updated_at",
+            ]
+        )
+
+        try:
+            service_response = client.start_recording(
+                {
+                    "session_id": session.id,
+                    "course_id": session.course_id,
+                    "room_id": session.room_id,
+                    "room_title": session.subject,
+                    "recording_title": build_web_egress_title(session),
+                    "join_url": self._build_web_egress_join_url(session),
+                    "callback_url": get_web_egress_callback_url(),
+                    "callback_token": getattr(
+                        settings,
+                        "ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN",
+                        "",
+                    ),
+                    "requested_by": {
+                        "user_id": str(claims.get("user_id") or claims.get("sub") or ""),
+                        "name": claims.get("name") or "",
+                    },
+                }
+            )
+        except (ImproperlyConfigured, WebEgressError) as exc:
+            session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_FAILED
+            session.web_egress_error = str(exc)
+            session.save(
+                update_fields=[
+                    "web_egress_status",
+                    "web_egress_error",
+                    "updated_at",
+                ]
+            )
+            raise AppAPIException(
+                {"message": str(exc)},
+                status_code=status.HTTP_502_BAD_GATEWAY,
+            )
+
+        session.web_egress_id = service_response.get("egress_id") or session.web_egress_id
+        session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_RECORDING
+        session.web_egress_error = None
+        session.save(
+            update_fields=[
+                "web_egress_id",
+                "web_egress_status",
+                "web_egress_error",
+                "updated_at",
+            ]
+        )
+
+        return self._build_status_payload(
+            session,
+            message=service_response.get("message") or "Recording started successfully.",
+        )
+
+    def _stop_recording(self, session: CourseLiveSession) -> dict:
+        if not is_web_egress_recording_status(session.web_egress_status):
+            return self._build_status_payload(
+                session,
+                message="Recording is not active.",
+            )
+
+        client = WebEgressClient()
+        try:
+            service_response = client.stop_recording(
+                {
+                    "session_id": session.id,
+                    "room_id": session.room_id,
+                    "egress_id": session.web_egress_id,
+                }
+            )
+        except (ImproperlyConfigured, WebEgressError) as exc:
+            session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_FAILED
+            session.web_egress_error = str(exc)
+            session.save(
+                update_fields=[
+                    "web_egress_status",
+                    "web_egress_error",
+                    "updated_at",
+                ]
+            )
+            raise AppAPIException(
+                {"message": str(exc)},
+                status_code=status.HTTP_502_BAD_GATEWAY,
+            )
+
+        session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_STOPPING
+        session.web_egress_stopped_at = timezone.now()
+        session.web_egress_error = None
+        session.save(
+            update_fields=[
+                "web_egress_status",
+                "web_egress_stopped_at",
+                "web_egress_error",
+                "updated_at",
+            ]
+        )
+
+        return self._build_status_payload(
+            session,
+            message=service_response.get("message") or "Recording stop requested successfully.",
+        )
+
+    def _refresh_status_from_service(self, session: CourseLiveSession) -> None:
+        if not getattr(settings, "ONLINE_CLASS_WEB_EGRESS_ENABLED", False):
+            return
+        if not session.web_egress_id and session.web_egress_status in {
+            CourseLiveSession.WEB_EGRESS_STATUS_IDLE,
+            CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED,
+            CourseLiveSession.WEB_EGRESS_STATUS_FAILED,
+        }:
+            return
+
+        try:
+            client = WebEgressClient()
+            service_response = client.get_status(
+                room_id=session.room_id,
+                session_id=session.id,
+            )
+        except (ImproperlyConfigured, WebEgressError) as exc:
+            logger.warning(
+                "[WebEgress] Failed to refresh status for session=%s room_id=%s error=%s",
+                session.id,
+                session.room_id,
+                str(exc),
+            )
+            return
+
+        normalized_status = self._normalize_service_status(service_response)
+        update_fields = ["updated_at"]
+
+        if service_response.get("egress_id") is not None and session.web_egress_id != service_response.get("egress_id"):
+            session.web_egress_id = service_response.get("egress_id")
+            update_fields.append("web_egress_id")
+
+        if normalized_status and session.web_egress_status != normalized_status:
+            session.web_egress_status = normalized_status
+            update_fields.append("web_egress_status")
+
+        error_message = service_response.get("error") or service_response.get("message")
+        if normalized_status == CourseLiveSession.WEB_EGRESS_STATUS_FAILED:
+            session.web_egress_error = error_message or session.web_egress_error
+            update_fields.append("web_egress_error")
+
+        if normalized_status == CourseLiveSession.WEB_EGRESS_STATUS_RECORDING and not session.web_egress_started_at:
+            session.web_egress_started_at = timezone.now()
+            update_fields.append("web_egress_started_at")
+
+        if normalized_status in {
+            CourseLiveSession.WEB_EGRESS_STATUS_STOPPING,
+            CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING,
+            CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED,
+            CourseLiveSession.WEB_EGRESS_STATUS_FAILED,
+        } and not session.web_egress_stopped_at:
+            session.web_egress_stopped_at = timezone.now()
+            update_fields.append("web_egress_stopped_at")
+
+        if len(update_fields) > 1:
+            session.save(update_fields=update_fields)
+
+    def _normalize_service_status(self, payload: dict) -> str:
+        raw = str(payload.get("state") or payload.get("status") or "").strip().lower()
+        if raw in {"recording", "active", "started"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_RECORDING
+        if raw in {"starting", "pending"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_STARTING
+        if raw in {"stopping"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_STOPPING
+        if raw in {"processing", "finalizing", "uploading"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING
+        if raw in {"completed", "done", "finished"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED
+        if raw in {"failed", "error"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_FAILED
+        if payload.get("is_recording") is True:
+            return CourseLiveSession.WEB_EGRESS_STATUS_RECORDING
+        if payload.get("is_recording") is False:
+            return CourseLiveSession.WEB_EGRESS_STATUS_IDLE
+        return ""
+
+    def _build_web_egress_join_url(self, session: CourseLiveSession) -> str:
+        client = PlugNMeetClient()
+        join_response = client.get_join_token(
+            {
+                "room_id": session.room_id,
+                "user_info": {
+                    "user_id": f"web-egress-{session.id}",
+                    "name": "Session Recorder",
+                    "is_admin": True,
+                    "is_hidden": True,
+                },
+            }
+        )
+        access_token = join_response.get("token")
+        if not access_token:
+            raise WebEgressError("PlugNMeet did not return a recorder access token.")
+
+        frontend_base = get_online_class_frontend_base()
+        return (
+            f"{frontend_base}/?access_token={access_token}"
+            f"&web_egress=1&room_id={session.room_id}"
+        )
+
+    def _build_status_payload(self, session: CourseLiveSession, message: str | None = None) -> dict:
+        return {
+            "success": True,
+            "room_id": session.room_id,
+            "session_id": session.id,
+            "status": session.web_egress_status,
+            "is_recording": is_web_egress_recording_status(session.web_egress_status),
+            "egress_id": session.web_egress_id,
+            "started_at": session.web_egress_started_at.isoformat()
+            if session.web_egress_started_at
+            else None,
+            "stopped_at": session.web_egress_stopped_at.isoformat()
+            if session.web_egress_stopped_at
+            else None,
+            "error": session.web_egress_error,
+            "message": message,
+        }
+
+
+class CourseLiveSessionWebEgressCallbackAPIView(GenericAPIView):
+    permission_classes = [AllowAny]
+    authentication_classes = []
+
+    @swagger_auto_schema(
+        operation_description="Receive final WebEgress recording callback and save the final file.",
+        tags=["Imam-Javad - Course"],
+        responses={201: openapi.Response(description="Recording callback processed successfully")},
+    )
+    def post(self, request, *args, **kwargs):
+        self._verify_callback_token(request)
+
+        room_id = request.data.get("room_id")
+        session_id = request.data.get("session_id")
+        status_value = str(request.data.get("status") or "completed").strip().lower()
+        egress_id = request.data.get("egress_id")
+        error_message = request.data.get("error") or ""
+
+        if not room_id and not session_id:
+            raise AppAPIException(
+                {"message": "room_id or session_id is required."},
+                status_code=status.HTTP_400_BAD_REQUEST,
+            )
+
+        session = self._get_target_session(room_id=room_id, session_id=session_id)
+
+        uploaded_file = request.FILES.get("file")
+        duration_seconds = request.data.get("file_time_seconds")
+        recording = None
+
+        if uploaded_file:
+            file_duration = None
+            if duration_seconds not in (None, ""):
+                try:
+                    file_duration = timedelta(seconds=float(duration_seconds))
+                except (TypeError, ValueError):
+                    file_duration = None
+
+            recording = LiveSessionRecording.objects.create(
+                session=session,
+                file=uploaded_file,
+                title=request.data.get("title") or build_web_egress_title(session),
+                recording_type=request.data.get("recording_type") or "video",
+                file_time=file_duration,
+            )
+
+        session.web_egress_id = egress_id or session.web_egress_id
+        session.web_egress_stopped_at = session.web_egress_stopped_at or timezone.now()
+        session.web_egress_error = error_message or None
+        session.web_egress_status = self._normalize_callback_status(
+            status_value=status_value,
+            has_file=bool(uploaded_file),
+        )
+        session.save(
+            update_fields=[
+                "web_egress_id",
+                "web_egress_stopped_at",
+                "web_egress_error",
+                "web_egress_status",
+                "updated_at",
+            ]
+        )
+
+        response_status = status.HTTP_201_CREATED if recording else status.HTTP_200_OK
+        return Response(
+            {
+                "success": True,
+                "session_id": session.id,
+                "room_id": session.room_id,
+                "recording_id": recording.id if recording else None,
+                "status": session.web_egress_status,
+            },
+            status=response_status,
+        )
+
+    def _verify_callback_token(self, request) -> None:
+        expected = getattr(settings, "ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN", "")
+        if not expected:
+            raise AppAPIException(
+                {"message": "WebEgress callback token is not configured."},
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+
+        received = (
+            request.headers.get("X-Web-Egress-Token")
+            or request.data.get("callback_token")
+            or ""
+        )
+        if received != expected:
+            raise AppAPIException(
+                {"message": "Invalid WebEgress callback token."},
+                status_code=status.HTTP_403_FORBIDDEN,
+            )
+
+    def _get_target_session(self, *, room_id=None, session_id=None) -> CourseLiveSession:
+        queryset = CourseLiveSession.objects.order_by("-started_at", "-id")
+        if session_id:
+            try:
+                return queryset.get(id=int(session_id))
+            except (ValueError, CourseLiveSession.DoesNotExist):
+                pass
+        if room_id:
+            return get_object_or_404(queryset, room_id=room_id)
+        raise AppAPIException(
+            {"message": "Unable to resolve target live session."},
+            status_code=status.HTTP_404_NOT_FOUND,
+        )
+
+    @staticmethod
+    def _normalize_callback_status(*, status_value: str, has_file: bool) -> str:
+        if has_file:
+            return CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED
+        if status_value in {"failed", "error"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_FAILED
+        if status_value in {"processing", "uploading", "stopping"}:
+            return CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING
+        return CourseLiveSession.WEB_EGRESS_STATUS_COMPLETED
+
+
 class CourseLiveSessionRecordedFileAPIView(GenericAPIView):
     permission_classes = [AllowAny]
     authentication_classes = [TokenAuthentication]

apps/course/views/webhook.py

@@ -210,7 +210,18 @@ class PlugNMeetWebhookAPIView(APIView):
             session = CourseLiveSession.objects.get(room_id=room_id, ended_at__isnull=True)
             now = timezone.now()
             session.ended_at = now
-            session.save(update_fields=['ended_at', 'updated_at'])
+            update_fields = ['ended_at', 'updated_at']
+            if session.web_egress_status in {
+                CourseLiveSession.WEB_EGRESS_STATUS_STARTING,
+                CourseLiveSession.WEB_EGRESS_STATUS_RECORDING,
+                CourseLiveSession.WEB_EGRESS_STATUS_STOPPING,
+            }:
+                session.web_egress_status = CourseLiveSession.WEB_EGRESS_STATUS_PROCESSING
+                update_fields.append('web_egress_status')
+                if not session.web_egress_stopped_at:
+                    session.web_egress_stopped_at = now
+                    update_fields.append('web_egress_stopped_at')
+            session.save(update_fields=update_fields)
             
             # Close active user sessions
             updated_count = LiveSessionUser.objects.filter(

config/settings/base.py

@@ -314,6 +314,23 @@ SITE_DOMAIN = "https://imamjavad.online"
 DOVODI_DOMAIN = "https://dovodi.newhorizonco.uk"
 ONLINE_CLASS_FRONTEND_DOMAIN = env('ONLINE_CLASS_FRONTEND_DOMAIN', default=SITE_DOMAIN)
 ONLINE_CLASS_TOKEN_TTL = env.int('ONLINE_CLASS_TOKEN_TTL', default=3000)
+ONLINE_CLASS_WEB_EGRESS_ENABLED = env.bool('ONLINE_CLASS_WEB_EGRESS_ENABLED', default=False)
+ONLINE_CLASS_WEB_EGRESS_SERVICE_URL = env(
+    'ONLINE_CLASS_WEB_EGRESS_SERVICE_URL',
+    default='',
+)
+ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN = env(
+    'ONLINE_CLASS_WEB_EGRESS_SERVICE_TOKEN',
+    default='',
+)
+ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN = env(
+    'ONLINE_CLASS_WEB_EGRESS_CALLBACK_TOKEN',
+    default='',
+)
+ONLINE_CLASS_WEB_EGRESS_TIMEOUT = env.float(
+    'ONLINE_CLASS_WEB_EGRESS_TIMEOUT',
+    default=20.0,
+)
 PLUGNMEET_SERVER_URL ='https://meet.imamjavad.online'
 PLUGNMEET_API_KEY ='habibmeet_api_key_2024'
 PLUGNMEET_API_SECRET ='habibmeet_secret_zumyyYWqv7KR2kUqvYdq4z4sXg7XTBD2ljT6_2024'