wip unit test parser classes

sorry for this absurd commit, mybad quentin … I'll rework my commits' history before merging.

Dockerfile

@@ -1,7 +1,7 @@
 # Django Meet
 
 # ---- base image to inherit from ----
-FROM python:3.12.6-alpine3.20 as base
+FROM python:3.12.6-alpine3.20 AS base
 
 # Upgrade pip to its latest release to speed up dependencies installation
 RUN python -m pip install --upgrade pip setuptools
@@ -11,7 +11,7 @@ RUN apk update && \
   apk upgrade
 
 # ---- Back-end builder image ----
-FROM base as back-builder
+FROM base AS back-builder
 
 WORKDIR /builder
 
@@ -23,7 +23,7 @@ RUN mkdir /install && \
 
 
 # ---- mails ----
-FROM node:20 as mail-builder
+FROM node:20 AS mail-builder
 
 COPY ./src/mail /mail/app
 
@@ -34,7 +34,7 @@ RUN yarn install --frozen-lockfile && \
 
 
 # ---- static link collector ----
-FROM base as link-collector
+FROM base AS link-collector
 ARG MEET_STATIC_ROOT=/data/static
 
 RUN apk add \
@@ -58,7 +58,7 @@ RUN DJANGO_CONFIGURATION=Build DJANGO_JWT_PRIVATE_SIGNING_KEY=Dummy \
 RUN rdfind -makesymlinks true -followsymlinks true -makeresultsfile false ${MEET_STATIC_ROOT}
 
 # ---- Core application image ----
-FROM base as core
+FROM base AS core
 
 ENV PYTHONUNBUFFERED=1
 
@@ -93,7 +93,7 @@ WORKDIR /app
 ENTRYPOINT [ "/usr/local/bin/entrypoint" ]
 
 # ---- Development image ----
-FROM core as backend-development
+FROM core AS backend-development
 
 # Switch back to the root user to install development dependencies
 USER root:root
@@ -119,7 +119,7 @@ ENV DB_HOST=postgresql \
 CMD ["python", "manage.py", "runserver", "0.0.0.0:8000"]
 
 # ---- Production image ----
-FROM core as backend-production
+FROM core AS backend-production
 
 ARG MEET_STATIC_ROOT=/data/static
 

src/backend/.pylintrc

@@ -447,10 +447,10 @@ max-bool-expr=5
 max-branches=12
 
 # Maximum number of locals for function / method body
-max-locals=15
+max-locals=20
 
 # Maximum number of parents for a class (see R0901).
-max-parents=7
+max-parents=10
 
 # Maximum number of public methods for a class (see R0904).
 max-public-methods=20

src/backend/core/admin.py

@@ -77,3 +77,25 @@ class RoomAdmin(admin.ModelAdmin):
     """Room admin interface declaration."""
 
     inlines = (ResourceAccessInline,)
+
+
+@admin.register(models.Recording)
+class RecordingAdmin(admin.ModelAdmin):
+    """Recording admin interface declaration."""
+
+    list_display = ("id", "status", "room", "created_at", "worker_id")
+    list_select_related = ("creator",)
+    readonly_fields = (
+        "id",
+        "created_at",
+        "updated_at",
+        "worker_id",
+        "room",
+        "creator",
+        "mode",
+    )
+
+    fieldsets = (
+        (None, {"fields": ("status", "room", "creator", "worker_id", "mode")}),
+        ("Timestamps", {"fields": ("created_at", "updated_at")}),
+    )

src/backend/core/api/permissions.py

@@ -1,5 +1,7 @@
 """Permission handlers for the Meet core app."""
 
+from django.conf import settings
+
 from rest_framework import permissions
 
 from ..models import RoleChoices
@@ -65,6 +67,21 @@ class RoomPermissions(permissions.BasePermission):
         return obj.is_administrator(user)
 
 
+class IsRoomOwnerOrAdministrator(permissions.BasePermission):
+    """Temporary"""
+
+    message = "You must be an admin or owner to start a recording."
+
+    def has_permission(self, request, view):
+        # Get the room object
+        room = view.get_object()
+
+        if not room.is_owner(request.user) and not room.is_administrator(request.user):
+            return False
+
+        return True
+
+
 class ResourceAccessPermission(permissions.BasePermission):
     """
     Permissions for a room that can only be updated by room administrators.
@@ -83,3 +100,23 @@ class ResourceAccessPermission(permissions.BasePermission):
             return obj.user == user
 
         return obj.resource.is_administrator(user)
+
+
+class IsRecordingEnabled(permissions.BasePermission):
+    """Check if the recording feature is enabled."""
+
+    message = "Access denied, recording is disabled."
+
+    def has_permission(self, request, view):
+        """Determine if access is allowed based on settings."""
+        return settings.RECORDING_ENABLE
+
+
+class IsStorageEventEnabled(permissions.BasePermission):
+    """Check if the storage event feature is enabled."""
+
+    message = "Access denied, storage event is disabled."
+
+    def has_permission(self, request, view):
+        """Determine if access is allowed based on settings."""
+        return settings.RECORDING_STORAGE_EVENT_ENABLE

src/backend/core/api/viewsets.py

@@ -1,6 +1,7 @@
 """API endpoints"""
 
 import uuid
+from logging import getLogger
 
 from django.conf import settings
 from django.db.models import Q
@@ -14,17 +15,38 @@ from rest_framework import (
     pagination,
     viewsets,
 )
+from rest_framework import (
+    exceptions as drf_exceptions,
+)
 from rest_framework import (
     response as drf_response,
 )
+from rest_framework import (
+    status as drf_status,
+)
 
 from core import models, utils
+from core.recording.event import (
+    InvalidBucketError,
+    InvalidFileTypeError,
+    ParsingEventDataError,
+    StorageEventAuthentication,
+    get_parser,
+)
+from core.recording.worker import (
+    RecordingStartError,
+    RecordingStopError,
+    WorkerServiceMediator,
+    worker_service_provider,
+)
 
 from ..analytics import analytics
 from . import permissions, serializers
 
 # pylint: disable=too-many-ancestors
 
+logger = getLogger(__name__)
+
 
 class NestedGenericViewSet(viewsets.GenericViewSet):
     """
@@ -186,13 +208,6 @@ class RoomViewSet(
         """
         try:
             instance = self.get_object()
-
-            analytics.track(
-                user=self.request.user,
-                event="Get Room",
-                properties={"slug": instance.slug},
-            )
-
         except Http404:
             if not settings.ALLOW_UNREGISTERED_ROOMS:
                 raise
@@ -209,10 +224,96 @@ class RoomViewSet(
                 },
             }
         else:
+            analytics.track(
+                user=self.request.user,
+                event="Get Room",
+                properties={"slug": instance.slug},
+            )
             data = self.get_serializer(instance).data
 
         return drf_response.Response(data)
 
+    @decorators.action(
+        detail=True,
+        methods=["post"],
+        url_path="start-recording",
+        permission_classes=[
+            permissions.IsRoomOwnerOrAdministrator,
+            permissions.IsRecordingEnabled,
+        ],
+    )
+    def start_room_recording(self, request, pk=None):  # pylint: disable=unused-argument
+        """Start recording a room."""
+
+        room = self.get_object()
+        mode = request.data.get("mode")
+
+        if mode is None:
+            return drf_response.Response(
+                {"error": "Recording mode is required."},
+                status=drf_status.HTTP_400_BAD_REQUEST,
+            )
+
+        # May raise exception if an active or initiated recording already exist for the room
+        recording = models.Recording.objects.create(
+            creator=request.user, room=room, mode=mode
+        )
+
+        worker_service = worker_service_provider.create(mode=recording.mode)
+        worker_manager = WorkerServiceMediator(worker_service=worker_service)
+
+        try:
+            worker_manager.start(recording)
+        except RecordingStartError:
+            return drf_response.Response(
+                {"error": f"Recording failed to start for room {room.slug}"},
+                status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+
+        return drf_response.Response(
+            {"message": f"Recording successfully started for room {room.slug}"},
+            status=drf_status.HTTP_201_CREATED,
+        )
+
+    @decorators.action(
+        detail=True,
+        methods=["post"],
+        url_path="stop-recording",
+        permission_classes=[
+            permissions.IsRoomOwnerOrAdministrator,
+            permissions.IsRecordingEnabled,
+        ],
+    )
+    def stop_room_recording(self, request, pk=None):  # pylint: disable=unused-argument
+        """Stop room recording."""
+
+        room = self.get_object()
+
+        # fixme - do I need to be the creator of the recording to stop it?
+        try:
+            recording = models.Recording.objects.get(
+                room=room, status=models.RecordingStatusChoices.ACTIVE
+            )
+        except models.Recording.DoesNotExist as e:
+            raise drf_exceptions.NotFound(
+                "No active recording found for this room."
+            ) from e
+
+        worker_service = worker_service_provider.create(mode=recording.mode)
+        worker_manager = WorkerServiceMediator(worker_service=worker_service)
+
+        try:
+            worker_manager.stop(recording)
+        except RecordingStopError:
+            return drf_response.Response(
+                {"error": f"Recording failed to stop for room {room.slug}"},
+                status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+
+        return drf_response.Response(
+            {"message": f"Recording stopped for room {room.slug}."}
+        )
+
     def list(self, request, *args, **kwargs):
         """Limit listed rooms to the ones related to the authenticated user."""
         user = self.request.user
@@ -293,3 +394,55 @@ class ResourceAccessViewSet(
     permission_classes = [permissions.ResourceAccessPermission]
     queryset = models.ResourceAccess.objects.all()
     serializer_class = serializers.ResourceAccessSerializer
+
+
+class RecordingViewSet(viewsets.GenericViewSet, mixins.ListModelMixin):
+    """
+    API endpoints to access and perform actions on recording.
+    """
+
+    @decorators.action(
+        detail=False,
+        methods=["post"],
+        url_path="storage-hook",
+        authentication_classes=[StorageEventAuthentication],
+        permission_classes=[permissions.IsStorageEventEnabled],
+    )
+    def on_storage_event_received(self, request, pk=None):  # pylint: disable=unused-argument
+        """Handle incoming storage hook events for recordings."""
+
+        parser = get_parser()
+
+        try:
+            recording_id = parser.get_recording_id(request.data)
+
+        except ParsingEventDataError as e:
+            raise drf_exceptions.PermissionDenied(f"Invalid request data: {e}") from e
+
+        except InvalidBucketError as e:
+            raise drf_exceptions.PermissionDenied("Invalid bucket specified") from e
+
+        except InvalidFileTypeError as e:
+            return drf_response.Response(
+                {"message": f"Ignore this file type, {e}"},
+            )
+
+        try:
+            recording = models.Recording.objects.get(id=recording_id)
+        except models.Recording.DoesNotExist as e:
+            raise drf_exceptions.NotFound("No recording found for this event.") from e
+
+        if not recording.is_savable():
+            raise drf_exceptions.PermissionDenied(
+                f"Recording with ID {recording_id} cannot be saved because it is either,"
+                f"in an error state or has already been saved."
+            )
+
+        recording.status = models.RecordingStatusChoices.SAVED
+        recording.save()
+
+        # todo - trigger post-processing
+
+        return drf_response.Response(
+            {"message": "Event processed."},
+        )

src/backend/core/factories.py

@@ -65,3 +65,18 @@ class RoomFactory(ResourceFactory):
 
     name = factory.Faker("catch_phrase")
     slug = factory.LazyAttribute(lambda o: slugify(o.name))
+
+
+class RecordingFactory(factory.django.DjangoModelFactory):
+    """Create fake recording for testing."""
+
+    class Meta:
+        model = models.Recording
+
+    creator = factory.SubFactory(UserFactory)
+    room = factory.SubFactory(RoomFactory)
+    status = models.RecordingStatusChoices.INITIATED
+    mode = factory.fuzzy.FuzzyChoice(
+        [choice[0] for choice in models.RecordingModeChoices.choices]
+    )
+    worker_id = None

src/backend/core/migrations/0005_alter_user_language_recording.py

@@ -0,0 +1,41 @@
+# Generated by Django 5.1.1 on 2024-10-24 18:51
+
+import django.db.models.deletion
+import uuid
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0004_alter_user_language'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='user',
+            name='language',
+            field=models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language'),
+        ),
+        migrations.CreateModel(
+            name='Recording',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
+                ('status', models.CharField(choices=[('active', 'Active'), ('stopped', 'Stopped'), ('aborted', 'Aborted'), ('saved', 'Saved'), ('failed_to_start', 'Failed to Start'), ('failed_to_stop', 'Failed to Stop')], default='active', max_length=20)),
+                ('mode', models.CharField(choices=[('screen_recording', 'SCREEN_RECORDING'), ('transcript', 'TRANSCRIPT')], default='screen_recording', help_text='Defines the type of recording being performed.', max_length=20, verbose_name='Recording mode')),
+                ('worker_id', models.CharField(blank=True, help_text='Enter an identifier for the worker recording. This ID is retained even whenthe worker stops, allowing for easy tracking and management of recorded data.', max_length=255, null=True, verbose_name='Worker ID')),
+                ('creator', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recordings', to=settings.AUTH_USER_MODEL, verbose_name='Creator')),
+                ('room', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recordings', to='core.room', verbose_name='Room')),
+            ],
+            options={
+                'verbose_name': 'Recording',
+                'verbose_name_plural': 'Recordings',
+                'db_table': 'meet_recording',
+                'ordering': ('-created_at',),
+                'constraints': [models.UniqueConstraint(condition=models.Q(('status', 'active')), fields=('room',), name='unique_active_recording_per_room')],
+            },
+        ),
+    ]

src/backend/core/migrations/0006_remove_recording_unique_active_recording_per_room_and_more.py

@@ -0,0 +1,36 @@
+# Generated by Django 5.1.1 on 2024-10-31 21:31
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0005_alter_user_language_recording'),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name='recording',
+            name='unique_active_recording_per_room',
+        ),
+        migrations.AlterField(
+            model_name='recording',
+            name='mode',
+            field=models.CharField(choices=[('screen_recording', 'SCREEN_RECORDING'), ('transcript', 'TRANSCRIPT')], help_text='Defines the kind of worker being called.', max_length=20, verbose_name='Worker kind'),
+        ),
+        migrations.AlterField(
+            model_name='recording',
+            name='status',
+            field=models.CharField(choices=[('active', 'Active'), ('initiated', 'Initiated'), ('stopped', 'Stopped'), ('aborted', 'Aborted'), ('saved', 'Saved'), ('failed_to_start', 'Failed to Start'), ('failed_to_stop', 'Failed to Stop')], default='initiated', max_length=20),
+        ),
+        migrations.AlterField(
+            model_name='user',
+            name='language',
+            field=models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language'),
+        ),
+        migrations.AddConstraint(
+            model_name='recording',
+            constraint=models.UniqueConstraint(condition=models.Q(('status', 'active'), ('status', 'initiated'), _connector='OR'), fields=('room',), name='unique_initiated_or_active_recording_per_room'),
+        ),
+    ]

src/backend/core/models.py

@@ -38,6 +38,41 @@ class RoleChoices(models.TextChoices):
         return role == cls.OWNER
 
 
+class RecordingStatusChoices(models.TextChoices):
+    """Recording status choices."""
+
+    INITIATED = "initiated", _("Initiated")
+    ACTIVE = "active", _("Active")
+    STOPPED = "stopped", _("Stopped")
+    SAVED = "saved", _("Saved")
+    ABORTED = "aborted", _("Aborted")
+    FAILED_TO_START = "failed_to_start", _("Failed to Start")
+    FAILED_TO_STOP = "failed_to_stop", _("Failed to Stop")
+
+    @classmethod
+    def is_final_status(cls, status):
+        """Check if the status is a final status."""
+        return status in [
+            cls.STOPPED,
+            cls.SAVED,
+            cls.ABORTED,
+            cls.FAILED_TO_START,
+            cls.FAILED_TO_STOP,
+        ]
+
+    @classmethod
+    def is_error_status(cls, status):
+        """Check if the status is an error status."""
+        return status in [cls.ABORTED, cls.FAILED_TO_START, cls.FAILED_TO_STOP]
+
+
+class RecordingModeChoices(models.TextChoices):
+    """Recording mode choices."""
+
+    SCREEN_RECORDING = "screen_recording", _("SCREEN_RECORDING")
+    TRANSCRIPT = "transcript", _("TRANSCRIPT")
+
+
 class BaseModel(models.Model):
     """
     Serves as an abstract base model for other models, ensuring that records are validated
@@ -325,3 +360,83 @@ class Room(Resource):
         else:
             raise ValidationError({"name": f'Room name "{self.name:s}" is reserved.'})
         super().clean_fields(exclude=exclude)
+
+
+# todo - discuss how the path could changed, and we could loose track of file
+# todo - discuss having constraint to avoid transitioning from a final status to another
+# todo - discuss persisting extra metadata on the event sent by the storage
+# todo - discuss inheriting from ResourceAccess model
+class Recording(BaseModel):
+    """Model for recordings that take place in a room"""
+
+    creator = models.ForeignKey(
+        User,
+        on_delete=models.CASCADE,
+        related_name="recordings",
+        verbose_name=_("Creator"),
+    )
+    room = models.ForeignKey(
+        Room,
+        on_delete=models.CASCADE,
+        related_name="recordings",
+        verbose_name=_("Room"),
+    )
+    status = models.CharField(
+        max_length=20,
+        choices=RecordingStatusChoices.choices,
+        default=RecordingStatusChoices.INITIATED,
+    )
+    mode = models.CharField(
+        max_length=20,
+        choices=RecordingModeChoices.choices,
+        verbose_name=_("Worker kind"),
+        help_text=_("Defines the kind of worker being called."),
+    )
+    worker_id = models.CharField(
+        max_length=255,
+        null=True,
+        blank=True,
+        verbose_name=_("Worker ID"),
+        help_text=_(
+            "Enter an identifier for the worker recording. This ID is retained even when"
+            "the worker stops, allowing for easy tracking and management of recorded data."
+        ),
+    )
+
+    class Meta:
+        db_table = "meet_recording"
+        ordering = ("-created_at",)
+        verbose_name = _("Recording")
+        verbose_name_plural = _("Recordings")
+        constraints = [
+            models.UniqueConstraint(
+                fields=["room"],
+                condition=models.Q(status=RecordingStatusChoices.ACTIVE)
+                | models.Q(status=RecordingStatusChoices.INITIATED),
+                name="unique_initiated_or_active_recording_per_room",
+            )
+        ]
+
+    def __str__(self):
+        return str(self.id)
+
+    def get_abilities(self, user):
+        """Compute and return abilities for a given user on the recording."""
+        is_creator = user == self.creator
+        is_final_status = RecordingStatusChoices.is_final_status(self.status)
+
+        return {
+            "destroy": is_creator and is_final_status,
+            "partial_update": False,
+            "retrieve": is_creator,
+            "stop": is_creator and not is_final_status,
+            "update": False,
+        }
+
+    def is_savable(self) -> bool:
+        """Determine if the recording can be saved based on its current status."""
+
+        is_in_error = RecordingStatusChoices.is_error_status(self.status)
+        is_already_saved = self.status == RecordingStatusChoices.SAVED
+
+        return not is_in_error and not is_already_saved

src/backend/core/recording/event/__init__.py

@@ -0,0 +1,5 @@
+"""Meet event parser classes and exceptions."""
+
+from .authentication import *
+from .exceptions import *
+from .parsers import *

src/backend/core/recording/event/authentication.py

@@ -0,0 +1,65 @@
+"""Authentication class for storage event token validation."""
+
+import logging
+import secrets
+
+from django.conf import settings
+from django.utils.translation import gettext_lazy as _
+
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+
+logger = logging.getLogger(__name__)
+
+
+class StorageEventAuthentication(BaseAuthentication):
+    """Authenticate requests using a Bearer token for storage event integration.
+
+    This class validates Bearer tokens for storage events that don't map to database users.
+    It's designed for S3-compatible storage integrations and similar use cases.
+
+    Events are submitted when a webhook is configured on some bucket's events.
+    """
+
+    AUTH_HEADER = "Authorization"
+    TOKEN_TYPE = "Bearer"  # noqa S105
+
+    def authenticate(self, request):
+        """Validate the Bearer token from the Authorization header."""
+
+        required_token = settings.RECORDING_STORAGE_EVENT_TOKEN
+        if not required_token:
+            return None
+
+        auth_header = request.headers.get(self.AUTH_HEADER)
+
+        if not auth_header:
+            logger.warning(
+                "Authentication failed: Missing Authorization header (ip: %s)",
+                request.META.get("REMOTE_ADDR"),
+            )
+            raise AuthenticationFailed(_("Authorization header is required"))
+
+        auth_parts = auth_header.split(" ")
+        if len(auth_parts) != 2 or auth_parts[0] != self.TOKEN_TYPE:
+            logger.warning(
+                "Authentication failed: Invalid authorization header (ip: %s)",
+                request.META.get("REMOTE_ADDR"),
+            )
+            raise AuthenticationFailed(_("Invalid authorization header."))
+
+        token = auth_parts[1]
+
+        # Use constant-time comparison to prevent timing attacks
+        if not secrets.compare_digest(token.encode(), required_token.encode()):
+            logger.warning(
+                "Authentication failed: Invalid token (ip: %s)",
+                request.META.get("REMOTE_ADDR"),
+            )
+            raise AuthenticationFailed(_("Invalid token"))
+
+        return None
+
+    def authenticate_header(self, request):
+        """Return the WWW-Authenticate header value."""
+        return f"{self.TOKEN_TYPE} realm='Storage event API'"

src/backend/core/recording/event/exceptions.py

@@ -0,0 +1,17 @@
+"""Storage parsers specific exceptions."""
+
+
+class ParsingEventDataError(Exception):
+    """Raised when the request data is malformed, incomplete, or missing."""
+
+
+class InvalidBucketError(Exception):
+    """Raised when the bucket name in the request does not match the expected one."""
+
+
+class InvalidFileTypeError(Exception):
+    """Raised when the file type in the request is not supported."""
+
+
+class InvalidFilepathError(Exception):
+    """Raised when the filepath in the request is invalid."""

src/backend/core/recording/event/parsers.py

@@ -0,0 +1,149 @@
+"""Meet storage event parser classes."""
+
+import logging
+import re
+from dataclasses import dataclass
+from functools import lru_cache
+from typing import Any, Dict, Optional, Protocol
+
+from django.conf import settings
+from django.utils.module_loading import import_string
+
+from .exceptions import (
+    InvalidBucketError,
+    InvalidFilepathError,
+    InvalidFileTypeError,
+    ParsingEventDataError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class StorageEvent:
+    """Represents a storage event with relevant metadata.
+
+    Attributes:
+        filepath: Identifier for the affected recording
+        filetype: Type of storage event
+        bucket_name: When the event occurred
+        metadata: Additional event data
+    """
+
+    filepath: str
+    filetype: str
+    bucket_name: str
+    metadata: Optional[Dict[str, Any]]
+
+    def __post_init__(self):
+        if self.filepath is None:
+            raise TypeError("filepath cannot be None")
+        if self.filetype is None:
+            raise TypeError("filetype cannot be None")
+        if self.bucket_name is None:
+            raise TypeError("bucket_name cannot be None")
+
+
+class EventParser(Protocol):
+    """Interface for parsing storage events."""
+
+    def __init__(self, bucket_name, allowed_filetypes=None):
+        """Initialize parser with bucket name and optional allowed filetypes."""
+
+    def parse(self, data: Dict) -> StorageEvent:
+        """Extract storage event data from raw dictionary input."""
+
+    def validate(self, data: StorageEvent) -> None:
+        """Verify storage event data meets all requirements."""
+
+    def get_recording_id(self, data: Dict) -> str:
+        """Extract recording ID from event dictionary."""
+
+
+@lru_cache(maxsize=1)
+def get_parser() -> EventParser:
+    """Return cached instance of configured event parser.
+
+    Uses function memoization instead of a factory class since the only
+    varying parameter is the parser class from settings. A factory class
+    would add unnecessary complexity when a cached function provides the
+    same singleton behavior with simpler code.
+    """
+
+    event_parser_cls = import_string(settings.RECORDING_EVENT_PARSER_CLASS)
+    return event_parser_cls(bucket_name=settings.AWS_STORAGE_BUCKET_NAME)
+
+
+class MinioParser:
+    """Handle parsing and validation of Minio storage events."""
+
+    def __init__(self, bucket_name: str, allowed_filetypes=None):
+        """Initialize parser with target bucket name and accepted filetypes."""
+
+        if not bucket_name:
+            raise ValueError("Bucket name cannot be None or empty")
+
+        self._bucket_name = bucket_name
+        self._allowed_filetypes = allowed_filetypes or {"audio/ogg", "video/mp4"}
+
+        # pylint: disable=line-too-long
+        self._filepath_regex = re.compile(
+            r"(?P<url_encoded_folder_path>(?:[^%]+%2F)*)?(?P<recording_id>[0-9a-fA-F\-]{36})\.(?P<extension>[a-zA-Z0-9]+)"
+        )
+
+    @staticmethod
+    def parse(data):
+        """Convert raw Minio event dictionary to StorageEvent object."""
+
+        if not data:
+            raise ParsingEventDataError("Received empty data.")
+
+        try:
+            record = data["Records"][0]
+            s3 = record["s3"]
+            bucket_name = s3["bucket"]["name"]
+            file_object = s3["object"]
+            filepath = file_object["key"]
+            filetype = file_object["contentType"]
+        except (KeyError, IndexError) as e:
+            raise ParsingEventDataError(f"Missing or malformed key: {e}.") from e
+        try:
+            return StorageEvent(
+                filepath=filepath,
+                filetype=filetype,
+                bucket_name=bucket_name,
+                metadata=None,
+            )
+        except TypeError as e:
+            raise ParsingEventDataError(f"Missing essential data fields: {e}") from e
+
+    def validate(self, event_data: StorageEvent) -> str:
+        """Verify StorageEvent matches bucket, filetype and filepath requirements."""
+
+        if event_data.bucket_name != self._bucket_name:
+            raise InvalidBucketError(
+                f"Invalid bucket: expected {self._bucket_name}, got {event_data.bucket_name}"
+            )
+
+        if not event_data.filetype in self._allowed_filetypes:
+            raise InvalidFileTypeError(
+                f"Invalid file type, expected {self._allowed_filetypes},"
+                f"got '{event_data.filetype}'"
+            )
+
+        match = self._filepath_regex.match(event_data.filepath)
+        if not match:
+            raise InvalidFilepathError(
+                f"Invalid filepath structure: {event_data.filepath}"
+            )
+
+        recording_id = match.group("recording_id")
+        return recording_id
+
+    def get_recording_id(self, data):
+        """Extract recording ID from Minio event through parsing and validation."""
+
+        event_data = self.parse(data)
+        recording_id = self.validate(event_data)
+
+        return recording_id

src/backend/core/recording/worker/__init__.py

@@ -0,0 +1,13 @@
+"""Meet worker services classes and exceptions."""
+
+from .exceptions import *
+from .factories import WorkerServiceFactory
+from .mediator import WorkerServiceMediator
+from .services import AudioCompositeEgressService, VideoCompositeEgressService
+
+# Expose the worker service factory instance at the module level.
+# This provides a centralized access point and abstracts away the service creation
+# details from the code using the services.
+worker_service_provider = WorkerServiceFactory()
+worker_service_provider.register("transcript", AudioCompositeEgressService)
+worker_service_provider.register("screen_recording", VideoCompositeEgressService)

src/backend/core/recording/worker/exceptions.py

@@ -0,0 +1,21 @@
+"""Recording and worker services specific exceptions."""
+
+
+class WorkerRequestError(Exception):
+    """Raised when there is an issue with the worker request"""
+
+
+class WorkerConnectionError(Exception):
+    """Raised when there is an issue connecting to the worker."""
+
+
+class WorkerResponseError(Exception):
+    """Raised when the worker's response is not as expected."""
+
+
+class RecordingStartError(Exception):
+    """Raised when there is an error starting the recording."""
+
+
+class RecordingStopError(Exception):
+    """Raised when there is an error stopping the recording."""

src/backend/core/recording/worker/factories.py

@@ -0,0 +1,90 @@
+"""Factory, configurations and Protocol to create worker services"""
+
+import logging
+from dataclasses import dataclass
+from functools import lru_cache
+from typing import Any, ClassVar, Dict, Optional, Protocol
+
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class WorkerServiceConfig:
+    """Declare Worker Service common configurations"""
+
+    output_folder: str
+    server_configurations: Dict[str, Any]
+    verify_ssl: Optional[bool]
+    bucket_args: Optional[dict]
+
+    @classmethod
+    @lru_cache
+    def from_settings(cls) -> "WorkerServiceConfig":
+        """Load configuration from Django settings with caching for efficiency."""
+
+        logger.debug("Loading WorkerServiceConfig from settings.")
+        return cls(
+            output_folder=settings.RECORDING_OUTPUT_FOLDER,
+            server_configurations=settings.LIVEKIT_CONFIGURATION,
+            verify_ssl=settings.RECORDING_VERIFY_SSL,
+            bucket_args={
+                "endpoint": settings.AWS_S3_ENDPOINT_URL,
+                "access_key": settings.AWS_S3_ACCESS_KEY_ID,
+                "secret": settings.AWS_S3_SECRET_ACCESS_KEY,
+                "region": settings.AWS_S3_REGION_NAME,
+                "bucket": settings.AWS_STORAGE_BUCKET_NAME,
+                "force_path_style": True,
+            },
+        )
+
+
+class WorkerService(Protocol):
+    """Define the interface for interacting with a worker service."""
+
+    hrid: ClassVar[str]
+
+    def __init__(self, config: WorkerServiceConfig):
+        """Initialize the service with the given configuration."""
+
+    def start(self, room_id: str, recording_id: str) -> str:
+        """Start a recording for a specified room."""
+
+    def stop(self, worker_id: str) -> str:
+        """Stop recording for a specified worker."""
+
+
+class WorkerServiceFactory:
+    """Factory to instantiate worker services based on a specified mode
+
+    This factory currently uses a common configuration (`_default_config`) to initialize all
+    workers. In the future, if workers require different configurations, consider refactoring
+    to a builder pattern. With a builder pattern, specific builders could be registered per
+    WorkerService type, allowing each builder to handle instantiation of the worker with its
+    unique configuration requirements.
+    """
+
+    def __init__(self):
+        """Initialize the WorkerServiceFactory with a default configuration and worker registry."""
+
+        self._worker_service_registry = {}
+        self._default_config = WorkerServiceConfig.from_settings()
+
+    def register(self, mode, worker_service: WorkerService):
+        """Register a worker service for a specific mode."""
+
+        if mode in self._worker_service_registry:
+            raise KeyError(f"Worker service for mode '{mode}' is already registered.")
+
+        self._worker_service_registry[mode] = worker_service
+
+    def create(self, mode: str) -> WorkerService:
+        """Instantiate a worker service for the specified mode."""
+
+        worker_service_cls = self._worker_service_registry.get(mode)
+
+        if not worker_service_cls:
+            raise ValueError(f"Unknown worker service for mode: {mode}.")
+
+        return worker_service_cls(config=self._default_config)

src/backend/core/recording/worker/mediator.py

@@ -0,0 +1,102 @@
+"""Mediator between the worker service and recording instances in the Django ORM."""
+
+import logging
+
+from core.models import Recording, RecordingStatusChoices
+
+from .exceptions import (
+    RecordingStartError,
+    RecordingStopError,
+    WorkerConnectionError,
+    WorkerRequestError,
+    WorkerResponseError,
+)
+from .factories import WorkerService
+
+logger = logging.getLogger(__name__)
+
+
+class WorkerServiceMediator:
+    """Mediate interactions between a worker service and a recording instance.
+
+    This class avoids direct coupling between the worker service and the Django ORM.
+    It is responsible for updating the recording instance based on the worker service's
+    status and responses. It also encapsulates worker-related errors into more
+    user-friendly higher-level exceptions.
+
+    This class follows the Mediator design pattern to centralize and coordinate
+    the communication between the worker service and the recording instances. It's not only
+    a facade for the worker service, because it adds functionalities to the worker service.
+    """
+
+    def __init__(self, worker_service: WorkerService):
+        """Initialize the WorkerServiceMediator with the provided worker service."""
+
+        self._worker_service = worker_service
+
+    def start(self, recording: Recording):
+        """Start the recording process using the worker service.
+
+        Args:
+            recording (Recording): The recording instance to start.
+
+        Raises:
+            RecordingStartError: If there is an error starting the recording.
+        """
+
+        # FIXME - no manipulations of room_name should be required
+        room_name = f"{recording.room.id!s}".replace("-", "")
+
+        if recording.status != RecordingStatusChoices.INITIATED:
+            logger.error("Cannot start recording in %s status.", recording.status)
+            raise RecordingStartError()
+
+        try:
+            worker_id = self._worker_service.start(room_name, recording.id)
+        except (WorkerRequestError, WorkerConnectionError, WorkerResponseError) as e:
+            logger.exception(
+                "Failed to start recording for room %s: %s", recording.room.slug, e
+            )
+            recording.status = RecordingStatusChoices.FAILED_TO_START
+            raise RecordingStartError() from e
+        else:
+            recording.worker_id = worker_id
+            recording.status = RecordingStatusChoices.ACTIVE
+        finally:
+            recording.save()
+
+        logger.info(
+            "Worker started for room %s (worker ID: %s, mode: %s)",
+            recording.room,
+            recording.worker_id,
+            recording.mode,
+        )
+
+    def stop(self, recording: Recording):
+        """Stop the recording process using the worker service.
+
+        Args:
+            recording (Recording): The recording instance to stop.
+
+        Raises:
+            RecordingStopError: If there is an error stopping the recording.
+        """
+
+        if recording.status != RecordingStatusChoices.ACTIVE:
+            logger.error("Cannot stop recording in %s status.", recording.status)
+            raise RecordingStopError()
+
+        try:
+            response = self._worker_service.stop(worker_id=recording.worker_id)
+        except (WorkerConnectionError, WorkerResponseError) as e:
+            logger.exception(
+                "Failed to stop recording for room %s: %s", recording.room.slug, e
+            )
+            recording.status = RecordingStatusChoices.FAILED_TO_STOP
+            raise RecordingStopError() from e
+        else:
+            recording.status = RecordingStatusChoices[response]
+        finally:
+            recording.save()
+
+        logger.info("Worker stopped for room %s", recording.room)

src/backend/core/recording/worker/services.py

@@ -0,0 +1,141 @@
+"""Worker services in charge of recording a room."""
+
+# pylint: disable=no-member
+
+import aiohttp
+from asgiref.sync import async_to_sync
+from livekit import api as livekit_api
+from livekit.api.egress_service import EgressService
+
+from .exceptions import WorkerConnectionError, WorkerResponseError
+from .factories import WorkerServiceConfig
+
+
+class BaseEgressService:
+    """Base egress defining common method to manage and interact with LiveKit egress processes."""
+
+    def __init__(self, config: WorkerServiceConfig):
+        self._config = config
+        self._s3 = livekit_api.S3Upload(**config.bucket_args)
+
+    def _get_filepath(self, filename, extension):
+        """Construct the file path for a given filename and extension.
+        Unsecure method, doesn't handle paths robustly and securely.
+        """
+        return f"{self._config.output_folder}/{filename}.{extension}"
+
+    @async_to_sync
+    async def _handle_request(self, request, method_name: str):
+        """Handle making a request to the LiveKit API and returns the response."""
+
+        # Use HTTP connector for local development with Tilt,
+        # where cluster communications are unsecure
+        connector = aiohttp.TCPConnector(ssl=self._config.verify_ssl)
+
+        async with aiohttp.ClientSession(connector=connector) as session:
+            client = EgressService(session, **self._config.server_configurations)
+            method = getattr(client, method_name)
+            try:
+                response = await method(request)
+            except livekit_api.TwirpError as e:
+                raise WorkerConnectionError(
+                    f"LiveKit client connection error, {e.message}."
+                ) from e
+
+            return response
+
+    def stop(self, worker_id):
+        """Stop an ongoing egress worker.
+
+        The StopEgressRequest is shared among all types of egress,
+        so a single implementation in the base class should be sufficient.
+        """
+
+        request = livekit_api.StopEgressRequest(
+            egress_id=worker_id,
+        )
+
+        response = self._handle_request(request, "stop_egress")
+
+        if not response.status:
+            raise WorkerResponseError(
+                "LiveKit response is missing the recording status."
+            )
+
+        # To avoid exposing EgressStatus values and coupling with LiveKit outside of this class,
+        # the response status is mapped to simpler "ABORTED", "STOPPED" or "FAILED_TO_STOP" strings.
+        if response.status == livekit_api.EgressStatus.EGRESS_ABORTED:
+            return "ABORTED"
+
+        # todo - verify if it's functional
+        if response.status == livekit_api.EgressStatus.EGRESS_ENDING:
+            return "STOPPED"
+
+        return "FAILED_TO_STOP"
+
+    def start(self, room_name, recording_id):
+        """Start the egress process for a recording (not implemented in the base class).
+
+        Each derived class must implement this method, providing the necessary parameters for
+        its specific egress type (e.g. audio_only, streaming output).
+        """
+        raise NotImplementedError("Subclass must implement this method.")
+
+
+class VideoCompositeEgressService(BaseEgressService):
+    """Record multiple participant video and audio tracks into a single output '.mp4' file."""
+
+    hrid = "video-recording-composite-livekit-egress"
+
+    def start(self, room_name, recording_id):
+        """Start the video composite egress process for a recording."""
+
+        file_type = livekit_api.EncodedFileType.MP4
+        filepath = self._get_filepath(filename=recording_id, extension="mp4")
+
+        file_output = livekit_api.EncodedFileOutput(
+            file_type=file_type,
+            filepath=filepath,
+            s3=self._s3,
+        )
+
+        request = livekit_api.RoomCompositeEgressRequest(
+            room_name=room_name,
+            file_outputs=[file_output],
+        )
+
+        response = self._handle_request(request, "start_room_composite_egress")
+
+        if not response.egress_id:
+            raise WorkerResponseError("Egress ID not found in the response.")
+
+        return response.egress_id
+
+
+class AudioCompositeEgressService(BaseEgressService):
+    """Record multiple participant audio tracks into a single output '.ogg' file."""
+
+    hrid = "audio-recording-composite-livekit-egress"
+
+    def start(self, room_name, recording_id):
+        """Start the audio composite egress process for a recording."""
+
+        file_type = livekit_api.EncodedFileType.OGG
+        filepath = self._get_filepath(filename=recording_id, extension="ogg")
+
+        file_output = livekit_api.EncodedFileOutput(
+            file_type=file_type,
+            filepath=filepath,
+            s3=self._s3,
+        )
+
+        request = livekit_api.RoomCompositeEgressRequest(
+            room_name=room_name, file_outputs=[file_output], audio_only=True
+        )
+
+        response = self._handle_request(request, "start_room_composite_egress")
+
+        if not response.egress_id:
+            raise WorkerResponseError("Egress ID not found in the response.")
+
+        return response.egress_id

src/backend/core/tests/recording/event/test_authentication.py

@@ -0,0 +1,107 @@
+"""
+Test event authentication.
+"""
+
+# pylint: disable=E1128
+
+from django.test import RequestFactory, override_settings
+
+import pytest
+from rest_framework.exceptions import AuthenticationFailed
+
+from core.recording.event.authentication import (
+    StorageEventAuthentication,
+)
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_successful_authentication():
+    """Test successful authentication with valid token."""
+
+    request = RequestFactory().get("/")
+    request.headers = {"Authorization": "Bearer valid-test-token"}
+
+    result = StorageEventAuthentication().authenticate(request)
+    assert (
+        result is None
+    )  # Authentication successful but returns None as per implementation
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN=None)
+@pytest.mark.parametrize("add_header", [True, False])
+def test_disabled_authentication(add_header):
+    """Test behavior when token is not configured in settings."""
+    # mock_settings.RECORDING_STORAGE_EVENT_TOKEN = ''
+    request = RequestFactory().get("/")
+    if add_header:
+        request.headers = {"Authorization": "Bearer some-token"}
+
+    result = StorageEventAuthentication().authenticate(request)
+    assert result is None
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_missing_auth_header():
+    """Test failure when Authorization header is missing."""
+    request = RequestFactory().get("/")
+    request.headers = {}
+
+    with pytest.raises(AuthenticationFailed, match="Authorization header is required"):
+        StorageEventAuthentication().authenticate(request)
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_invalid_auth_header_format():
+    """Test failure when Authorization header has invalid format."""
+    request = RequestFactory().get("/")
+    request.headers = {"Authorization": "InvalidFormat"}
+
+    with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
+        StorageEventAuthentication().authenticate(request)
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_invalid_token_type():
+    """Test failure when token type is not Bearer."""
+    request = RequestFactory().get("/")
+    request.headers = {"Authorization": "Basic some-token"}
+
+    with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
+        StorageEventAuthentication().authenticate(request)
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_invalid_token():
+    """Test failure when token is invalid."""
+    request = RequestFactory().get("/")
+    request.headers = {"Authorization": "Bearer wrong-token"}
+
+    with pytest.raises(AuthenticationFailed, match="Invalid token"):
+        StorageEventAuthentication().authenticate(request)
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_malformed_auth_header():
+    """Test failure when Authorization header is malformed."""
+    request = RequestFactory().get("/")
+    request.headers = {"Authorization": "Bearer"}  # Missing token part
+
+    with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
+        StorageEventAuthentication().authenticate(request)
+
+
+def test_authenticate_header():
+    """Test the WWW-Authenticate header value."""
+    request = RequestFactory().get("/")
+    header = StorageEventAuthentication().authenticate_header(request)
+    assert header == "Bearer realm='Storage event API'"
+
+
+@override_settings(RECORDING_STORAGE_EVENT_TOKEN="valid-test-token")
+def test_multiple_spaces_in_auth_header():
+    """Test failure when Authorization header contains multiple spaces."""
+    request = RequestFactory().get("/")
+    request.headers = {"Authorization": "Bearer   extra-spaces-token"}
+
+    with pytest.raises(AuthenticationFailed, match="Invalid authorization header"):
+        StorageEventAuthentication().authenticate(request)

src/backend/core/tests/recording/event/test_parsers.py

@@ -0,0 +1,308 @@
+"""
+Test event parsers.
+"""
+
+# pylint: disable=W0212,W0621,W0613
+
+from unittest import mock
+
+from django.conf import settings
+from django.test import override_settings
+
+import pytest
+
+from core.recording.event.exceptions import (
+    InvalidBucketError,
+    InvalidFilepathError,
+    InvalidFileTypeError,
+    ParsingEventDataError,
+)
+from core.recording.event.parsers import (
+    MinioParser,
+    StorageEvent,
+    get_parser,
+)
+
+
+@pytest.fixture
+def valid_minio_event():
+    """Mock a valid Minio event."""
+    return {
+        "Records": [
+            {
+                "s3": {
+                    "bucket": {"name": "test-bucket"},
+                    "object": {
+                        "key": "recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+                        "contentType": "audio/ogg",
+                    },
+                }
+            }
+        ]
+    }
+
+
+@pytest.fixture
+def minio_parser():
+    """Mock a Minio parser."""
+    return MinioParser(bucket_name="test-bucket")
+
+
+def test_parse_valid_event(minio_parser, valid_minio_event):
+    """Test parsing a valid Minio event."""
+    event = minio_parser.parse(valid_minio_event)
+    assert isinstance(event, StorageEvent)
+    assert event.filepath == "recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg"
+    assert event.filetype == "audio/ogg"
+    assert event.bucket_name == "test-bucket"
+    assert event.metadata is None
+
+
+def test_parse_empty_data(minio_parser):
+    """Test parsing empty event data raises error."""
+    with pytest.raises(ParsingEventDataError, match="Received empty data."):
+        minio_parser.parse({})
+
+
+def test_parse_missing_keys(minio_parser):
+    """Test parsing event with missing key."""
+
+    invalid_minio_event = {
+        "Records": [
+            {
+                "s3": {
+                    "bucket": {"name": None},
+                    # Missing 'object' key
+                }
+            }
+        ]
+    }
+
+    with pytest.raises(ParsingEventDataError, match="Missing or malformed key"):
+        minio_parser.parse(invalid_minio_event)
+
+
+def test_parse_none_key(minio_parser):
+    """Test parsing event with None field."""
+
+    invalid_minio_event = {
+        "Records": [
+            {
+                "s3": {
+                    "bucket": {"name": "test-bucket"},
+                    "object": {
+                        "key": "recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+                        "contentType": None,  # 'contentType' should not be None
+                    },
+                }
+            }
+        ]
+    }
+
+    with pytest.raises(ParsingEventDataError, match="Missing essential data fields"):
+        minio_parser.parse(invalid_minio_event)
+
+
+def test_validate_invalid_bucket(minio_parser):
+    """Test validation with wrong bucket name."""
+    event = StorageEvent(
+        filepath="recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+        filetype="audio/ogg",
+        bucket_name="wrong-bucket",
+        metadata=None,
+    )
+    with pytest.raises(InvalidBucketError):
+        minio_parser.validate(event)
+
+
+def test_validate_invalid_filetype(minio_parser):
+    """Test validation with unsupported file type."""
+    event = StorageEvent(
+        filepath="recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.txt",
+        filetype="text/plain",  # Not included in the default allowed filetypes
+        bucket_name="test-bucket",
+        metadata=None,
+    )
+    with pytest.raises(InvalidFileTypeError):
+        minio_parser.validate(event)
+
+
+@pytest.mark.parametrize(
+    "invalid_filepath",
+    [
+        "invalid_filepath",
+        "recording/46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+        "recording%2F46d1a1212426484d8fb309b5d886f7a8.ogg",
+    ],
+)
+def test_validate_invalid_filepath(invalid_filepath, minio_parser):
+    """Test validation with malformed filepath."""
+    event = StorageEvent(
+        filepath=invalid_filepath,
+        filetype="audio/ogg",
+        bucket_name="test-bucket",
+        metadata=None,
+    )
+    with pytest.raises(InvalidFilepathError):
+        minio_parser.validate(event)
+
+
+def test_validate_valid_event(minio_parser):
+    """Test validation with valid event data."""
+    event = StorageEvent(
+        filepath="recording%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+        filetype="audio/ogg",
+        bucket_name="test-bucket",
+        metadata=None,
+    )
+    recording_id = minio_parser.validate(event)
+    assert recording_id == "46d1a121-2426-484d-8fb3-09b5d886f7a8"
+
+
+def test_get_recording_id_success(minio_parser, valid_minio_event):
+    """Test successful extraction of recording ID."""
+    recording_id = minio_parser.get_recording_id(valid_minio_event)
+    assert recording_id == "46d1a121-2426-484d-8fb3-09b5d886f7a8"
+
+
+def test_validate_filepath_with_folder(minio_parser):
+    """Test validation of filepath with folder structure."""
+    event = StorageEvent(
+        filepath="parent_folder%2Ffolder%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+        filetype="audio/ogg",
+        bucket_name="test-bucket",
+        metadata=None,
+    )
+    recording_id = minio_parser.validate(event)
+    assert recording_id == "46d1a121-2426-484d-8fb3-09b5d886f7a8"
+
+
+def test_parse_with_video_type(minio_parser):
+    """Test parsing event with video file type."""
+    video_event = {
+        "Records": [
+            {
+                "s3": {
+                    "bucket": {"name": "test-bucket"},
+                    "object": {
+                        "key": "46d1a121-2426-484d-8fb3-09b5d886f7a8.mp4",
+                        "contentType": "video/mp4",
+                    },
+                }
+            }
+        ]
+    }
+    event = minio_parser.parse(video_event)
+    assert event.filetype == "video/mp4"
+    assert event.filepath.endswith(".mp4")
+
+
+def test_empty_allowed_filetypes():
+    """Test MinioParser with empty allowed_filetypes."""
+    empty_types = set()
+    parser = MinioParser(bucket_name="test-bucket", allowed_filetypes=empty_types)
+    assert parser._allowed_filetypes == {"audio/ogg", "video/mp4"}
+
+
+def test_custom_allowed_filetypes():
+    """Test MinioParser with empty allowed_filetypes."""
+    custom_types = {"audio/mp3", "video/mov"}
+    parser = MinioParser(bucket_name="test-bucket", allowed_filetypes=custom_types)
+    assert parser._allowed_filetypes == {"audio/mp3", "video/mov"}
+
+
+def test_validate_custom_filetypes():
+    """Test validation of filepath with folder structure."""
+
+    parser = MinioParser(bucket_name="test-bucket", allowed_filetypes={"audio/mp3"})
+
+    event = StorageEvent(
+        filepath="parent_folder%2Ffolder%2F46d1a121-2426-484d-8fb3-09b5d886f7a8.ogg",
+        filetype="audio/mp3",
+        bucket_name="test-bucket",
+        metadata=None,
+    )
+    parser.validate(event)
+
+
+def test_constructor_none_bucket():
+    """Test MinioParser constructor with None bucket name."""
+    with pytest.raises(ValueError, match="Bucket name cannot be None or empty"):
+        MinioParser(bucket_name=None)
+
+
+def test_constructor_empty_bucket():
+    """Test MinioParser constructor with empty bucket name."""
+    with pytest.raises(ValueError, match="Bucket name cannot be None or empty"):
+        MinioParser(bucket_name="")
+
+
+@pytest.fixture
+def clear_lru_cache():
+    """Fixture to clear the LRU cache between tests."""
+    get_parser.cache_clear()
+    yield
+    get_parser.cache_clear()
+
+
+@override_settings(AWS_STORAGE_BUCKET_NAME="test-bucket")
+def test_returns_correct_instance(clear_lru_cache):
+    """Test if get_parser returns the correct parser instance."""
+    parser = get_parser()
+    assert isinstance(parser, MinioParser)
+    assert parser._bucket_name == "test-bucket"
+
+
+@override_settings(AWS_STORAGE_BUCKET_NAME="test-bucket")
+def test_caching_behavior(clear_lru_cache):
+    """Test if the function properly caches the parser instance."""
+    parser1 = get_parser()
+    parser2 = get_parser()
+    assert parser1 is parser2  # Check object identity
+
+
+@override_settings(AWS_STORAGE_BUCKET_NAME="different-bucket")
+def test_different_settings_new_instance():
+    """Test if changing settings creates a new instance."""
+    parser = get_parser()
+    assert parser._bucket_name == "different-bucket"
+
+
+def test_import_error_handling(clear_lru_cache):
+    """Test handling of import errors for invalid parser class."""
+    with override_settings(RECORDING_EVENT_PARSER_CLASS="invalid.parser.path"):
+        with pytest.raises(ImportError):
+            get_parser()
+
+
+@mock.patch("core.recording.event.parsers.import_string")
+def test_parser_instantiation_called_once(mock_import_string, clear_lru_cache):
+    """Test that parser class is instantiated only once due to caching."""
+    mock_parser_cls = type(
+        "MockParser",
+        (),
+        {
+            "__init__": lambda self, bucket_name: setattr(
+                self, "_bucket_name", bucket_name
+            )
+        },
+    )
+    mock_import_string.return_value = mock_parser_cls
+
+    # First call
+    parser1 = get_parser()
+    # Second call
+    parser2 = get_parser()
+
+    # Verify import_string was called only once
+    mock_import_string.assert_called_once_with(settings.RECORDING_EVENT_PARSER_CLASS)
+    assert parser1 is parser2
+
+
+def test_cache_clear_behavior(clear_lru_cache):
+    """Test that cache clearing creates new instance."""
+    parser1 = get_parser()
+    get_parser.cache_clear()
+    parser2 = get_parser()
+
+    assert parser1 is not parser2  # Should be different instances after cache clear

src/backend/core/tests/recording/worker/test_factories.py

@@ -0,0 +1,154 @@
+"""
+Test worker service factories.
+"""
+
+# pylint: disable=W0212,W0621,W0613
+
+from dataclasses import FrozenInstanceError
+
+from django.test import override_settings
+
+import pytest
+
+from core.recording.worker.factories import WorkerServiceConfig, WorkerServiceFactory
+
+
+@pytest.fixture(autouse=True)
+def clear_lru_cache():
+    """Clear the lru_cache before and after each test"""
+    WorkerServiceConfig.from_settings.cache_clear()
+    yield
+    WorkerServiceConfig.from_settings.cache_clear()
+
+
+@pytest.fixture
+def test_settings():
+    """Fixture to provide test Django settings"""
+    mocked_settings = {
+        "RECORDING_OUTPUT_FOLDER": "/test/output",
+        "LIVEKIT_CONFIGURATION": {"server": "test.example.com"},
+        "RECORDING_VERIFY_SSL": True,
+        "AWS_S3_ENDPOINT_URL": "https://s3.test.com",
+        "AWS_S3_ACCESS_KEY_ID": "test_key",
+        "AWS_S3_SECRET_ACCESS_KEY": "test_secret",
+        "AWS_S3_REGION_NAME": "test-region",
+        "AWS_STORAGE_BUCKET_NAME": "test-bucket",
+    }
+
+    # Use override_settings to properly patch Django settings
+    with override_settings(**mocked_settings):
+        yield test_settings
+
+
+@pytest.fixture
+def config(test_settings):
+    """Fixture to provide a WorkerServiceConfig instance"""
+    return WorkerServiceConfig.from_settings()
+
+
+# Tests
+def test_config_initialization(config):
+    """Test that WorkerServiceConfig is properly initialized from settings"""
+    assert config.output_folder == "/test/output"
+    assert config.server_configurations == {"server": "test.example.com"}
+    assert config.verify_ssl is True
+    assert config.bucket_args == {
+        "endpoint": "https://s3.test.com",
+        "access_key": "test_key",
+        "secret": "test_secret",
+        "region": "test-region",
+        "bucket": "test-bucket",
+        "force_path_style": True,
+    }
+
+
+def test_config_immutability(config):
+    """Test that config instances are immutable after creation"""
+    with pytest.raises(FrozenInstanceError):
+        config.output_folder = "new/path"
+
+
+@override_settings(
+    RECORDING_OUTPUT_FOLDER="/test/output",
+    LIVEKIT_CONFIGURATION={"server": "test.example.com"},
+    RECORDING_VERIFY_SSL=True,
+    AWS_S3_ENDPOINT_URL="https://s3.test.com",
+    AWS_S3_ACCESS_KEY_ID="test_key",
+    AWS_S3_SECRET_ACCESS_KEY="test_secret",
+    AWS_S3_REGION_NAME="test-region",
+    AWS_STORAGE_BUCKET_NAME="test-bucket",
+)
+def test_config_caching():
+    """Test that from_settings method caches its result"""
+    # Clear cache before testing caching behavior
+    WorkerServiceConfig.from_settings.cache_clear()
+
+    config1 = WorkerServiceConfig.from_settings()
+    config2 = WorkerServiceConfig.from_settings()
+    assert config1 is config2
+
+
+@pytest.fixture
+def mock_worker_service():
+    """Fixture to provide a mock WorkerService implementation"""
+
+    class TestWorkerService:
+        """Mock WorkerService"""
+
+        hrid = "test-worker"
+
+        def __init__(self, config):
+            """Mock init"""
+            self.config = config
+
+        def start(self, room_id, recording_id):
+            """Mock start method"""
+            return f"worker_{room_id}_{recording_id}"
+
+        def stop(self, worker_id):
+            """Mock stop method"""
+            return f"stopped_{worker_id}"
+
+    return TestWorkerService
+
+
+@pytest.fixture
+def factory():
+    """Fixture to provide a clean WorkerServiceFactory instance"""
+    return WorkerServiceFactory()
+
+
+def test_factory_initialization(factory, config):
+    """Test that factory is properly initialized"""
+    assert factory._worker_service_registry == {}
+    assert isinstance(factory._default_config, WorkerServiceConfig)
+    assert factory._default_config == config
+
+
+def test_register_worker_service(factory, mock_worker_service):
+    """Test registering a new worker service"""
+    factory.register("test", mock_worker_service)
+    assert factory._worker_service_registry["test"] == mock_worker_service
+
+
+def test_register_duplicate_worker_service(factory, mock_worker_service):
+    """Test that registering a duplicate worker service raises an error"""
+    factory.register("test", mock_worker_service)
+    with pytest.raises(KeyError) as exc_info:
+        factory.register("test", mock_worker_service)
+    assert "already registered" in str(exc_info.value)
+
+
+def test_create_worker_service(factory, mock_worker_service):
+    """Test creating a worker service instance"""
+    factory.register("test", mock_worker_service)
+    worker = factory.create("test")
+    assert isinstance(worker, mock_worker_service)
+    assert worker.hrid == "test-worker"
+
+
+def test_create_unknown_worker_service(factory):
+    """Test that creating an unknown worker service raises an error"""
+    with pytest.raises(ValueError) as exc_info:
+        factory.create("unknown")
+    assert "Unknown worker service" in str(exc_info.value)

src/backend/core/tests/recording/worker/test_mediator.py

@@ -0,0 +1,161 @@
+"""Test WorkerServiceMediator class."""
+
+# pylint: disable=W0621,W0613
+
+from unittest.mock import Mock
+
+import pytest
+
+from core.factories import RecordingFactory
+from core.models import RecordingStatusChoices
+from core.recording.worker.exceptions import (
+    RecordingStartError,
+    RecordingStopError,
+    WorkerConnectionError,
+    WorkerRequestError,
+    WorkerResponseError,
+)
+from core.recording.worker.factories import WorkerService
+from core.recording.worker.mediator import WorkerServiceMediator
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def mock_worker_service():
+    """Fixture for mock worker service"""
+    return Mock(spec=WorkerService)
+
+
+@pytest.fixture
+def mediator(mock_worker_service):
+    """Fixture for WorkerServiceMediator"""
+    return WorkerServiceMediator(mock_worker_service)
+
+
+def test_start_recording_success(mediator, mock_worker_service):
+    """Test successful recording start"""
+    # Setup
+    worker_id = "test-worker-123"
+    mock_worker_service.start.return_value = worker_id
+
+    mock_recording = RecordingFactory(
+        status=RecordingStatusChoices.INITIATED, worker_id=None
+    )
+    mediator.start(mock_recording)
+
+    # Verify worker service call
+    expected_room_name = str(mock_recording.room.id).replace("-", "")
+    mock_worker_service.start.assert_called_once_with(
+        expected_room_name, mock_recording.id
+    )
+
+    # Verify recording updates
+    mock_recording.refresh_from_db()
+    assert mock_recording.worker_id == worker_id
+    assert mock_recording.status == RecordingStatusChoices.ACTIVE
+
+
+@pytest.mark.parametrize(
+    "error_class", [WorkerRequestError, WorkerConnectionError, WorkerResponseError]
+)
+def test_mediator_start_recording_worker_errors(
+    mediator, mock_worker_service, error_class
+):
+    """Test handling of various worker errors during start"""
+    # Setup
+    mock_worker_service.start.side_effect = error_class("Test error")
+    mock_recording = RecordingFactory(
+        status=RecordingStatusChoices.INITIATED, worker_id=None
+    )
+
+    # Execute and verify
+    with pytest.raises(RecordingStartError):
+        mediator.start(mock_recording)
+
+    # Verify recording updates
+    mock_recording.refresh_from_db()
+    assert mock_recording.status == RecordingStatusChoices.FAILED_TO_START
+    assert mock_recording.worker_id is None
+
+
+@pytest.mark.parametrize(
+    "status",
+    [
+        RecordingStatusChoices.ACTIVE,
+        RecordingStatusChoices.FAILED_TO_START,
+        RecordingStatusChoices.FAILED_TO_STOP,
+        RecordingStatusChoices.STOPPED,
+        RecordingStatusChoices.SAVED,
+        RecordingStatusChoices.ABORTED,
+    ],
+)
+def test_mediator_start_recording_from_forbidden_status(
+    mediator, mock_worker_service, status
+):
+    """Test handling of various worker errors during start"""
+    # Setup
+    mock_recording = RecordingFactory(status=status)
+
+    # Execute and verify
+    with pytest.raises(RecordingStartError):
+        mediator.start(mock_recording)
+
+    # Verify recording was not updated
+    mock_recording.refresh_from_db()
+    assert mock_recording.status == status
+
+
+def test_mediator_stop_recording_success(mediator, mock_worker_service):
+    """Test successful recording stop"""
+    # Setup
+    mock_recording = RecordingFactory(
+        status=RecordingStatusChoices.ACTIVE, worker_id="test-worker-123"
+    )
+    mock_worker_service.stop.return_value = "STOPPED"
+
+    # Execute
+    mediator.stop(mock_recording)
+
+    # Verify worker service call
+    mock_worker_service.stop.assert_called_once_with(worker_id=mock_recording.worker_id)
+
+    # Verify recording updates
+    mock_recording.refresh_from_db()
+    assert mock_recording.status == RecordingStatusChoices.STOPPED
+
+
+def test_mediator_stop_recording_aborted(mediator, mock_worker_service):
+    """Test recording stop when worker returns ABORTED"""
+    # Setup
+    mock_recording = RecordingFactory(
+        status=RecordingStatusChoices.ACTIVE, worker_id="test-worker-123"
+    )
+    mock_worker_service.stop.return_value = "ABORTED"
+
+    # Execute
+    mediator.stop(mock_recording)
+
+    # Verify recording updates
+    mock_recording.refresh_from_db()
+    assert mock_recording.status == RecordingStatusChoices.ABORTED
+
+
+@pytest.mark.parametrize("error_class", [WorkerConnectionError, WorkerResponseError])
+def test_mediator_stop_recording_worker_errors(
+    mediator, mock_worker_service, error_class
+):
+    """Test handling of worker errors during stop"""
+    # Setup
+    mock_recording = RecordingFactory(
+        status=RecordingStatusChoices.ACTIVE, worker_id="test-worker-123"
+    )
+    mock_worker_service.stop.side_effect = error_class("Test error")
+
+    # Execute and verify
+    with pytest.raises(RecordingStopError):
+        mediator.stop(mock_recording)
+
+    # Verify recording updates
+    mock_recording.refresh_from_db()
+    assert mock_recording.status == RecordingStatusChoices.FAILED_TO_STOP

src/backend/core/tests/recording/worker/test_services.py

@@ -0,0 +1,368 @@
+"""
+Test worker service classes.
+"""
+
+# pylint: disable=W0212,W0621,W0613,E1101
+
+from unittest.mock import AsyncMock, Mock, patch
+
+import aiohttp
+import pytest
+
+from core.recording.worker.exceptions import WorkerConnectionError, WorkerResponseError
+from core.recording.worker.factories import WorkerServiceConfig
+from core.recording.worker.services import (
+    AudioCompositeEgressService,
+    BaseEgressService,
+    VideoCompositeEgressService,
+    livekit_api,
+)
+
+
+@pytest.fixture
+def config():
+    """Fixture to provide a WorkerServiceConfig instance"""
+    return WorkerServiceConfig(
+        output_folder="/test/output",
+        server_configurations={
+            "host": "test.livekit.io",
+            "api_key": "test_key",
+            "api_secret": "test_secret",
+        },
+        verify_ssl=True,
+        bucket_args={
+            "endpoint": "https://s3.test.com",
+            "access_key": "test_key",
+            "secret": "test_secret",
+            "region": "test-region",
+            "bucket": "test-bucket",
+            "force_path_style": True,
+        },
+    )
+
+
+@pytest.fixture
+def mock_s3_upload():
+    """Fixture for mocked S3Upload"""
+    with patch("core.recording.worker.services.livekit_api.S3Upload") as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_egress_service():
+    """Fixture for mocked EgressService"""
+    with patch("core.recording.worker.services.EgressService") as mock:
+        yield mock
+
+
+@pytest.fixture
+def service(config, mock_s3_upload):
+    """Fixture for BaseEgressService instance"""
+    return BaseEgressService(config)
+
+
+@pytest.fixture
+def mock_client_session():
+    """Fixture for mocked aiohttp.ClientSession"""
+    with patch("aiohttp.ClientSession") as mock:
+        mock.return_value.__aenter__ = AsyncMock()
+        mock.return_value.__aexit__ = AsyncMock()
+        yield mock
+
+
+@pytest.fixture
+def mock_tcp_connector():
+    """Fixture for TCPConnector"""
+    with patch("aiohttp.TCPConnector") as mock_connector:
+        mock_connector_instance = Mock()
+        mock_connector.return_value = mock_connector_instance
+        yield mock_connector
+
+
+@pytest.fixture
+def video_service(config):
+    """Fixture for VideoCompositeEgressService"""
+    service = VideoCompositeEgressService(config)
+    service._handle_request = Mock()  # Mock the request handler
+    return service
+
+
+@pytest.fixture
+def audio_service(config):
+    """Fixture for AudioCompositeEgressService"""
+    service = AudioCompositeEgressService(config)
+    service._handle_request = Mock()  # Mock the request handler
+    return service
+
+
+def test_base_egress_initialization(config, mock_s3_upload):
+    """Test service initialization"""
+
+    service = BaseEgressService(config)
+    assert service._config == config
+
+    mock_s3_upload.assert_called_once_with(
+        endpoint="https://s3.test.com",
+        access_key="test_key",
+        secret="test_secret",
+        region="test-region",
+        bucket="test-bucket",
+        force_path_style=True,
+    )
+
+
+@pytest.mark.parametrize(
+    "filename,extension,expected",
+    [
+        ("test", "mp4", "/test/output/test.mp4"),
+        ("recording123", "ogg", "/test/output/recording123.ogg"),
+        ("live_stream", "m3u8", "/test/output/live_stream.m3u8"),
+    ],
+)
+def test_base_egress_filepath_construction(service, filename, extension, expected):
+    """Test filepath construction with various inputs"""
+    result = service._get_filepath(filename, extension)
+    assert result == expected
+    assert result.startswith(service._config.output_folder)
+    assert result.endswith(f"{filename}.{extension}")
+
+
+def test_base_egress_handle_request_success(
+    config, service, mock_client_session, mock_egress_service, mock_tcp_connector
+):
+    """Test successful request handling"""
+    # Setup mock response
+    mock_response = Mock()
+    mock_method = AsyncMock(return_value=mock_response)
+    mock_egress_instance = Mock()
+    mock_egress_instance.test_method = mock_method
+    mock_egress_service.return_value = mock_egress_instance
+
+    # Create test request
+    test_request = Mock()
+
+    response = service._handle_request(test_request, "test_method")
+
+    mock_client_session.assert_called_once_with(
+        connector=mock_tcp_connector.return_value
+    )
+
+    # Verify EgressService initialization
+    mock_egress_service.assert_called_once_with(
+        mock_client_session.return_value.__aenter__.return_value,
+        **service._config.server_configurations,
+    )
+
+    # Verify method call and response
+    mock_method.assert_called_once_with(test_request)
+    assert response == mock_response
+
+
+def test_base_egress_handle_request_connection_error(service, mock_egress_service):
+    """Test handling of connection errors"""
+    # Setup mock error
+    mock_method = AsyncMock(
+        side_effect=livekit_api.TwirpError(msg="Connection failed", code=500)
+    )
+    mock_egress_instance = Mock()
+    mock_egress_instance.test_method = mock_method
+    mock_egress_service.return_value = mock_egress_instance
+
+    # Create test request
+    test_request = Mock()
+
+    # Verify error handling
+    with pytest.raises(WorkerConnectionError) as exc:
+        service._handle_request(test_request, "test_method")
+
+    assert "LiveKit client connection error" in str(exc.value)
+    assert "Connection failed" in str(exc.value)
+
+
+@pytest.mark.parametrize(
+    "response_status,expected_result",
+    [
+        (livekit_api.EgressStatus.EGRESS_ABORTED, "ABORTED"),
+        (livekit_api.EgressStatus.EGRESS_COMPLETE, "FAILED_TO_STOP"),
+        (livekit_api.EgressStatus.EGRESS_ENDING, "STOPPED"),
+        (livekit_api.EgressStatus.EGRESS_FAILED, "FAILED_TO_STOP"),
+    ],
+)
+def test_base_egress_stop_with_status(service, response_status, expected_result):
+    """Test stop method with different response statuses"""
+    # Mock _handle_request
+    mock_response = Mock(status=response_status)
+    service._handle_request = Mock(return_value=mock_response)
+
+    # Execute stop
+    result = service.stop("test_worker_id")
+
+    # Verify request and response handling
+    service._handle_request.assert_called_once_with(
+        livekit_api.StopEgressRequest(egress_id="test_worker_id"), "stop_egress"
+    )
+    assert result == expected_result
+
+
+def test_base_egress_stop_missing_status(service):
+    """Test stop method when response is missing status"""
+    # Mock _handle_request with missing status
+    mock_response = Mock(status=None)
+    service._handle_request = Mock(return_value=mock_response)
+
+    # Verify error handling
+    with pytest.raises(WorkerResponseError) as exc:
+        service.stop("test_worker_id")
+
+    assert "missing the recording status" in str(exc.value)
+
+
+def test_base_egress_start_not_implemented(service):
+    """Test that start method raises NotImplementedError"""
+    with pytest.raises(NotImplementedError) as exc:
+        service.start("test_room", "test_recording")
+    assert "Subclass must implement this method" in str(exc.value)
+
+
+@pytest.mark.parametrize("verify_ssl", [True, False])
+def test_base_egress_ssl_verification_config(verify_ssl):
+    """Test SSL verification configuration"""
+    config = WorkerServiceConfig(
+        output_folder="/test/output",
+        server_configurations={
+            "host": "test.livekit.io",
+            "api_key": "test_key",
+            "api_secret": "test_secret",
+        },
+        verify_ssl=verify_ssl,
+        bucket_args={
+            "endpoint": "https://s3.test.com",
+            "access_key": "test_key",
+            "secret": "test_secret",
+            "region": "test-region",
+            "bucket": "test-bucket",
+            "force_path_style": True,
+        },
+    )
+
+    service = BaseEgressService(config)
+
+    # Mock ClientSession to capture connector configuration
+    with patch("aiohttp.ClientSession") as mock_session:
+        mock_session.return_value.__aenter__ = AsyncMock()
+        mock_session.return_value.__aexit__ = AsyncMock()
+
+        # Trigger request to verify connector configuration
+        service._handle_request(Mock(), "test_method")
+
+        # Verify SSL configuration
+        connector = mock_session.call_args[1]["connector"]
+        assert isinstance(connector, aiohttp.TCPConnector)
+        assert connector._ssl == verify_ssl
+
+
+def test_video_composite_egress_hrid(video_service):
+    """Test HRID is correct"""
+    assert video_service.hrid == "video-recording-composite-livekit-egress"
+
+
+def test_video_composite_egress_start_success(video_service):
+    """Test successful start of video composite egress"""
+    # Setup mock response
+    egress_id = "test-egress-123"
+    video_service._handle_request.return_value = Mock(egress_id=egress_id)
+
+    # Test parameters
+    room_name = "test-room"
+    recording_id = "rec-123"
+
+    # Call start
+    result = video_service.start(room_name, recording_id)
+
+    # Verify result
+    assert result == egress_id
+
+    # Verify request construction
+    video_service._handle_request.assert_called_once()
+    request = video_service._handle_request.call_args[0][0]
+    method = video_service._handle_request.call_args[0][1]
+
+    # Verify request properties
+    assert isinstance(request, livekit_api.RoomCompositeEgressRequest)
+    assert request.room_name == room_name
+    assert len(request.file_outputs) == 1
+
+    assert not request.audio_only  # Video service shouldn't set audio_only
+
+    # Verify file output configuration
+    file_output = request.file_outputs[0]
+    assert file_output.file_type == livekit_api.EncodedFileType.MP4
+    assert file_output.filepath == f"/test/output/{recording_id}.mp4"
+    assert file_output.s3.bucket == "test-bucket"
+
+    # Verify method name
+    assert method == "start_room_composite_egress"
+
+
+def test_video_composite_egress_start_missing_egress_id(video_service):
+    """Test handling of missing egress ID in response"""
+    # Setup mock response without egress_id
+    video_service._handle_request.return_value = Mock(egress_id=None)
+
+    with pytest.raises(WorkerResponseError) as exc_info:
+        video_service.start("test-room", "rec-123")
+
+    assert "Egress ID not found" in str(exc_info.value)
+
+
+def test_audio_composite_egress_hrid(audio_service):
+    """Test HRID is correct"""
+    assert audio_service.hrid == "audio-recording-composite-livekit-egress"
+
+
+def test_audio_composite_egress_start_success(audio_service):
+    """Test successful start of audio composite egress"""
+    # Setup mock response
+    egress_id = "test-egress-123"
+    audio_service._handle_request.return_value = Mock(egress_id=egress_id)
+
+    # Test parameters
+    room_name = "test-room"
+    recording_id = "rec-123"
+
+    # Call start
+    result = audio_service.start(room_name, recording_id)
+
+    # Verify result
+    assert result == egress_id
+
+    # Verify request construction
+    audio_service._handle_request.assert_called_once()
+    request = audio_service._handle_request.call_args[0][0]
+    method = audio_service._handle_request.call_args[0][1]
+
+    # Verify request properties
+    assert isinstance(request, livekit_api.RoomCompositeEgressRequest)
+    assert request.room_name == room_name
+    assert len(request.file_outputs) == 1
+    assert request.audio_only is True  # Audio service should set audio_only
+
+    # Verify file output configuration
+    file_output = request.file_outputs[0]
+    assert file_output.file_type == livekit_api.EncodedFileType.OGG
+    assert file_output.filepath == f"/test/output/{recording_id}.ogg"
+    assert file_output.s3.bucket == "test-bucket"
+
+    # Verify method name
+    assert method == "start_room_composite_egress"
+
+
+def test_audio_composite_egress_start_missing_egress_id(audio_service):
+    """Test handling of missing egress ID in response"""
+    # Setup mock response without egress_id
+    audio_service._handle_request.return_value = Mock(egress_id=None)
+
+    with pytest.raises(WorkerResponseError) as exc_info:
+        audio_service.start("test-room", "rec-123")
+
+    assert "Egress ID not found" in str(exc_info.value)

src/backend/core/tests/rooms/test_api_room_start_recording.py

@@ -0,0 +1,178 @@
+"""
+Test rooms API endpoints in the Meet core app: start recording.
+"""
+
+# pylint: disable=W0621,W0613
+
+from unittest import mock
+
+from django.test.utils import override_settings
+
+import pytest
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from ...factories import RoomFactory, UserFactory
+from ...models import Recording
+from ...recording.worker.exceptions import RecordingStartError
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def mock_worker_service():
+    """Mock worker service."""
+    return mock.Mock()
+
+
+@pytest.fixture
+def mock_worker_service_provider(mock_worker_service):
+    """Mock worker service factory."""
+    with mock.patch(
+        "core.api.viewsets.worker_service_provider.create",
+        return_value=mock_worker_service,
+    ) as mock_worker_service_provider:
+        yield mock_worker_service_provider
+
+
+@pytest.fixture
+def mock_worker_manager(mock_worker_service):
+    """Mock worker service mediator."""
+    with mock.patch("core.api.viewsets.WorkerServiceMediator") as mock_mediator_class:
+        mock_mediator = mock.Mock()
+        mock_mediator_class.return_value = mock_mediator
+        yield mock_mediator
+
+
+def test_start_recording_anonymous():
+    """Anonymous users should not be allowed to start room recordings."""
+    room = RoomFactory()
+    client = APIClient()
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/", {"mode": "standard"}
+    )
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+    assert Recording.objects.count() == 0
+
+
+def test_start_recording_non_owner_and_non_administrator():
+    """Non-owner and Non-Administrator users should not be allowed to start room recordings."""
+    room = RoomFactory()
+    user = UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/", {"mode": "standard"}
+    )
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert Recording.objects.count() == 0
+
+
+@override_settings(RECORDING_ENABLE=False)
+def test_start_recording_recording_disabled():
+    """Should fail if recording is disabled for the room."""
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/", {"mode": "standard"}
+    )
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json() == {"detail": "Access denied, recording is disabled."}
+    assert Recording.objects.count() == 0
+
+
+@override_settings(RECORDING_ENABLE=True)
+def test_start_recording_missing_mode():
+    """Should fail if recording mode is not provided."""
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/start-recording/", {})
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json() == {"error": "Recording mode is required."}
+    assert Recording.objects.count() == 0
+
+
+@override_settings(RECORDING_ENABLE=True)
+def test_start_recording_worker_error(
+    mock_worker_service_provider, mock_worker_manager
+):
+    """Should handle worker service errors appropriately."""
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    # Configure mock mediator to raise error
+    mock_start = mock.Mock()
+    mock_start.side_effect = RecordingStartError("Failed to connect to worker")
+
+    mock_worker_manager.start = mock_start
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/", {"mode": "screen_recording"}
+    )
+
+    mock_worker_service_provider.assert_called_once_with(mode="screen_recording")
+
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+    assert response.json() == {
+        "error": f"Recording failed to start for room {room.slug}"
+    }
+
+    # Recording object should be created even if worker fails
+    assert Recording.objects.count() == 1
+    recording = Recording.objects.first()
+    assert recording.room == room
+    assert recording.mode == "screen_recording"
+    assert recording.creator == user
+
+
+@override_settings(RECORDING_ENABLE=True)
+def test_start_recording_success(mock_worker_service_provider, mock_worker_manager):
+    """Should successfully start recording when everything is configured correctly."""
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    mock_start = mock.Mock()
+    mock_worker_manager.start = mock_start
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/", {"mode": "screen_recording"}
+    )
+
+    mock_worker_service_provider.assert_called_once_with(mode="screen_recording")
+
+    assert response.status_code == status.HTTP_201_CREATED
+    assert response.json() == {
+        "message": f"Recording successfully started for room {room.slug}"
+    }
+
+    # Verify the mediator was called with the recording
+    recording = Recording.objects.first()
+    mock_start.assert_called_once_with(recording)

src/backend/core/tests/rooms/test_api_room_stop_recording.py

@@ -0,0 +1,169 @@
+"""
+Test rooms API endpoints in the Meet core app: stop recording.
+"""
+
+# pylint: disable=W0621,W0613
+
+from unittest import mock
+
+from django.test.utils import override_settings
+
+import pytest
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from ...factories import RecordingFactory, RoomFactory, UserFactory
+from ...models import Recording, RecordingStatusChoices
+from ...recording.worker.exceptions import RecordingStopError
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def mock_worker_service():
+    """Mock worker service."""
+    return mock.Mock()
+
+
+@pytest.fixture
+def mock_worker_service_provider(mock_worker_service):
+    """Mock worker service factory."""
+    with mock.patch(
+        "core.api.viewsets.worker_service_provider.create",
+        return_value=mock_worker_service,
+    ) as mock_worker_service_provider:
+        yield mock_worker_service_provider
+
+
+@pytest.fixture
+def mock_worker_manager(mock_worker_service):
+    """Mock worker service mediator."""
+    with mock.patch("core.api.viewsets.WorkerServiceMediator") as mock_mediator_class:
+        mock_mediator = mock.Mock()
+        mock_mediator_class.return_value = mock_mediator
+        yield mock_mediator
+
+
+def test_stop_recording_anonymous():
+    """Anonymous users should not be allowed to stop room recordings."""
+    room = RoomFactory()
+    RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
+    client = APIClient()
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED
+    # Verify recording status hasn't changed
+    assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
+
+
+def test_stop_recording_non_owner_and_non_administrator():
+    """Non-owner and Non-Administrator users should not be allowed to stop room recordings."""
+    room = RoomFactory()
+    user = UserFactory()
+    RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    # Verify recording status hasn't changed
+    assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
+
+
+@override_settings(RECORDING_ENABLE=False)
+def test_stop_recording_recording_disabled():
+    """Should fail if recording is disabled for the room."""
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+    assert response.json() == {"detail": "Access denied, recording is disabled."}
+    # Verify no recording exists
+    assert Recording.objects.count() == 0
+
+
+@override_settings(RECORDING_ENABLE=True)
+def test_stop_recording_no_active_recording():
+    """Should fail when there is no active recording for the room."""
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+    assert response.json() == {"detail": "No active recording found for this room."}
+
+
+@override_settings(RECORDING_ENABLE=True)
+def test_stop_recording_worker_error(mock_worker_service_provider, mock_worker_manager):
+    """Should handle worker service errors appropriately."""
+    room = RoomFactory()
+    user = UserFactory()
+    recording = RecordingFactory(
+        room=room, status=RecordingStatusChoices.ACTIVE, mode="screen_recording"
+    )
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    # Configure mock mediator to raise error
+    mock_stop = mock.Mock()
+    mock_stop.side_effect = RecordingStopError("Failed to connect to worker")
+    mock_worker_manager.stop = mock_stop
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    mock_worker_service_provider.assert_called_once_with(mode="screen_recording")
+    mock_stop.assert_called_once_with(recording)
+
+    assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+    assert response.json() == {
+        "error": f"Recording failed to stop for room {room.slug}"
+    }
+    # Verify recording status hasn't changed
+    assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
+
+
+@override_settings(RECORDING_ENABLE=True)
+def test_stop_recording_success(mock_worker_service_provider, mock_worker_manager):
+    """Should successfully stop recording when everything is configured correctly."""
+    room = RoomFactory()
+    user = UserFactory()
+    recording = RecordingFactory(
+        room=room, status=RecordingStatusChoices.ACTIVE, mode="screen_recording"
+    )
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    mock_stop = mock.Mock()
+    mock_worker_manager.stop = mock_stop
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    mock_worker_service_provider.assert_called_once_with(mode="screen_recording")
+    mock_stop.assert_called_once_with(recording)
+
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == {"message": f"Recording stopped for room {room.slug}."}
+
+    # Verify the recording still exists
+    assert Recording.objects.count() == 1

src/backend/core/tests/test_model_recording.py

@@ -0,0 +1,198 @@
+"""
+Unit tests for the Recording model
+"""
+
+from django.core.exceptions import ValidationError
+
+import pytest
+
+from core.factories import RecordingFactory, RoomFactory, UserFactory
+from core.models import Recording, RecordingStatusChoices
+
+pytestmark = pytest.mark.django_db
+
+
+def test_models_recording_str():
+    """The str representation should be the recording ID."""
+    recording = RecordingFactory()
+    assert str(recording) == str(recording.id)
+
+
+def test_models_recording_ordering():
+    """Recordings should be returned ordered by created_at in descending order."""
+    RecordingFactory.create_batch(3)
+    recordings = Recording.objects.all()
+    assert recordings[0].created_at >= recordings[1].created_at
+    assert recordings[1].created_at >= recordings[2].created_at
+
+
+def test_models_recording_creator_relationship():
+    """It should maintain proper relationship with creator."""
+    user = UserFactory()
+    recording = RecordingFactory(creator=user)
+    assert recording.creator == user
+    assert recording in user.recordings.all()
+
+
+def test_models_recording_room_relationship():
+    """It should maintain proper relationship with room."""
+    room = RoomFactory()
+    recording = RecordingFactory(room=room)
+    assert recording.room == room
+    assert recording in room.recordings.all()
+
+
+def test_models_recording_default_status():
+    """A new recording should have INITIATED status by default."""
+    recording = RecordingFactory()
+    assert recording.status == RecordingStatusChoices.INITIATED
+
+
+def test_models_recording_unique_initiated_or_active_per_room():
+    """Only one initiated or active recording should be allowed per room."""
+    room = RoomFactory()
+    RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
+
+    with pytest.raises(ValidationError):
+        RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
+
+    with pytest.raises(ValidationError):
+        RecordingFactory(room=room, status=RecordingStatusChoices.INITIATED)
+
+
+def test_models_recording_multiple_finished_allowed():
+    """Multiple finished recordings should be allowed in the same room."""
+    room = RoomFactory()
+    RecordingFactory(room=room, status=RecordingStatusChoices.SAVED)
+    RecordingFactory(room=room, status=RecordingStatusChoices.SAVED)
+    assert room.recordings.count() == 2
+
+
+# Test get_abilities method
+def test_models_recording_get_abilities_creator():
+    """Test abilities for the recording creator."""
+    recording = RecordingFactory(status=RecordingStatusChoices.ACTIVE)
+    abilities = recording.get_abilities(recording.creator)
+
+    assert abilities == {
+        "destroy": False,  # Not final status
+        "partial_update": False,
+        "retrieve": True,  # Creator can always retrieve
+        "stop": True,  # Not final status, so can stop
+        "update": False,
+    }
+
+
+def test_models_recording_get_abilities_non_creator():
+    """Test abilities for a user who isn't the creator."""
+    recording = RecordingFactory()
+    other_user = UserFactory()
+    abilities = recording.get_abilities(other_user)
+
+    assert abilities == {
+        "destroy": False,
+        "partial_update": False,
+        "retrieve": False,
+        "stop": False,
+        "update": False,
+    }
+
+
+def test_models_recording_get_abilities_final_status():
+    """Test abilities when recording is in final status."""
+    recording = RecordingFactory(status=RecordingStatusChoices.SAVED)
+    abilities = recording.get_abilities(recording.creator)
+
+    assert abilities == {
+        "destroy": True,  # Final status, creator can destroy
+        "partial_update": False,
+        "retrieve": True,
+        "stop": False,  # Can't stop when in final status
+        "update": False,
+    }
+
+
+# Test is_savable method
+def test_models_recording_is_savable_normal():
+    """Test is_savable for normal recording status."""
+    recording = RecordingFactory(status=RecordingStatusChoices.ACTIVE)
+    assert recording.is_savable() is True
+
+
+@pytest.mark.parametrize(
+    "status",
+    [
+        RecordingStatusChoices.FAILED_TO_STOP,
+        RecordingStatusChoices.FAILED_TO_START,
+        RecordingStatusChoices.ABORTED,
+    ],
+)
+def test_models_recording_is_savable_error(status):
+    """Test is_savable for error status."""
+    recording = RecordingFactory(status=status)
+    assert recording.is_savable() is False
+
+
+def test_models_recording_is_savable_already_saved():
+    """Test is_savable for already saved recording."""
+    recording = RecordingFactory(status=RecordingStatusChoices.SAVED)
+    assert recording.is_savable() is False
+
+
+# Test few corner cases
+
+
+def test_models_recording_worker_id_optional():
+    """Worker ID should be optional."""
+    recording = RecordingFactory(worker_id=None)
+    assert recording.worker_id is None
+
+    recording = RecordingFactory(worker_id="worker-123")
+    assert recording.worker_id == "worker-123"
+
+
+def test_models_recording_mode_required():
+    """Recording mode should be required."""
+    with pytest.raises(ValidationError):
+        RecordingFactory(mode=None)
+
+
+def test_models_recording_invalid_status():
+    """Test that setting an invalid status raises an error."""
+    recording = RecordingFactory()
+    recording.status = "INVALID_STATUS"
+    with pytest.raises(ValidationError):
+        recording.save()
+
+
+def test_models_recording_invalid_mode():
+    """Test that setting an invalid mode raises an error."""
+    with pytest.raises(ValidationError):
+        RecordingFactory(mode="INVALID_MODE")
+
+
+def test_models_recording_room_deletion():
+    """Test that deleting a room cascades to its recordings."""
+    room = RoomFactory()
+    recording = RecordingFactory(room=room)
+    room.delete()
+    assert not Recording.objects.filter(id=recording.id).exists()
+
+
+def test_models_recording_creator_deletion():
+    """Test that deleting a creator cascades to their recordings."""
+    creator = UserFactory()
+    recording = RecordingFactory(creator=creator)
+    creator.delete()
+    assert not Recording.objects.filter(id=recording.id).exists()
+
+
+def test_models_recording_worker_id_very_long():
+    """Test worker_id with maximum length."""
+    long_id = "w" * 255
+    recording = RecordingFactory(worker_id=long_id)
+    assert recording.worker_id == long_id
+
+    too_long_id = "w" * 256
+    with pytest.raises(ValidationError):
+        RecordingFactory(worker_id=too_long_id)

src/backend/core/urls.py

@@ -15,6 +15,7 @@ router.register("rooms", viewsets.RoomViewSet, basename="rooms")
 router.register(
     "resource-accesses", viewsets.ResourceAccessViewSet, basename="resource_accesses"
 )
+router.register("recording", viewsets.RecordingViewSet, basename="recordings")
 
 urlpatterns = [
     path(

src/backend/meet/settings.py

@@ -119,6 +119,25 @@ class Base(Configuration):
         },
     }
 
+    # Media
+    AWS_S3_ENDPOINT_URL = values.Value(
+        environ_name="AWS_S3_ENDPOINT_URL", environ_prefix=None
+    )
+    AWS_S3_ACCESS_KEY_ID = values.Value(
+        environ_name="AWS_S3_ACCESS_KEY_ID", environ_prefix=None
+    )
+    AWS_S3_SECRET_ACCESS_KEY = values.Value(
+        environ_name="AWS_S3_SECRET_ACCESS_KEY", environ_prefix=None
+    )
+    AWS_S3_REGION_NAME = values.Value(
+        environ_name="AWS_S3_REGION_NAME", environ_prefix=None
+    )
+    AWS_STORAGE_BUCKET_NAME = values.Value(
+        "meet-media-storage",
+        environ_name="AWS_STORAGE_BUCKET_NAME",
+        environ_prefix=None,
+    )
+
     # Internationalization
     # https://docs.djangoproject.com/en/3.1/topics/i18n/
 
@@ -386,6 +405,28 @@ class Base(Configuration):
         None, environ_name="ANALYTICS_KEY", environ_prefix=None
     )
 
+    # Recording settings
+    RECORDING_ENABLE = values.BooleanValue(
+        False, environ_name="RECORDING_ENABLE", environ_prefix=None
+    )
+    RECORDING_OUTPUT_FOLDER = values.Value(
+        "recordings", environ_name="RECORDING_OUTPUT_FOLDER", environ_prefix=None
+    )
+    RECORDING_VERIFY_SSL = values.BooleanValue(
+        True, environ_name="RECORDING_VERIFY_SSL", environ_prefix=None
+    )
+    RECORDING_STORAGE_EVENT_ENABLE = values.BooleanValue(
+        False, environ_name="RECORDING_STORAGE_EVENT_ENABLE", environ_prefix=None
+    )
+    RECORDING_STORAGE_EVENT_TOKEN = values.Value(
+        None, environ_name="RECORDING_STORAGE_HOOK_TOKEN", environ_prefix=None
+    )
+    RECORDING_EVENT_PARSER_CLASS = values.Value(
+        "core.recording.event.parsers.MinioParser",
+        environ_name="RECORDING_EVENT_PARSER_CLASS",
+        environ_prefix=None,
+    )
+
     # pylint: disable=invalid-name
     @property
     def ENVIRONMENT(self):

src/backend/pyproject.toml

@@ -58,6 +58,7 @@ dependencies = [
     "whitenoise==6.7.0",
     "mozilla-django-oidc==4.0.1",
     "livekit-api==0.7.0",
+    "aiohttp==3.10.10",
 ]
 
 [project.urls]

src/frontend/src/features/rooms/api/startRecording.ts

@@ -0,0 +1,24 @@
+import { useMutation, UseMutationOptions } from '@tanstack/react-query'
+import { fetchApi } from '@/api/fetchApi'
+import { ApiError } from '@/api/ApiError'
+import { ApiRoom } from './ApiRoom'
+
+export interface StartRecordingParams {
+  slug: string
+}
+
+const startRecording = ({ slug }: StartRecordingParams): Promise<ApiRoom> => {
+  return fetchApi(`rooms/${slug}/start-recording/`, {
+    method: 'POST',
+    body: JSON.stringify({}),
+  })
+}
+
+export function useStartRecording(
+  options?: UseMutationOptions<ApiRoom, ApiError, StartRecordingParams>
+) {
+  return useMutation<ApiRoom, ApiError, StartRecordingParams>({
+    mutationFn: startRecording,
+    onSuccess: options?.onSuccess,
+  })
+}

src/frontend/src/features/rooms/api/stopRecording.ts

@@ -0,0 +1,24 @@
+import { useMutation, UseMutationOptions } from '@tanstack/react-query'
+import { fetchApi } from '@/api/fetchApi'
+import { ApiError } from '@/api/ApiError'
+import { ApiRoom } from './ApiRoom'
+
+export interface StopRecordingParams {
+  slug: string
+}
+
+const stopRecording = ({ slug }: StopRecordingParams): Promise<ApiRoom> => {
+  return fetchApi(`rooms/${slug}/stop-recording/`, {
+    method: 'POST',
+    body: JSON.stringify({}),
+  })
+}
+
+export function useStopRecording(
+  options?: UseMutationOptions<ApiRoom, ApiError, StopRecordingParams>
+) {
+  return useMutation<ApiRoom, ApiError, StopRecordingParams>({
+    mutationFn: stopRecording,
+    onSuccess: options?.onSuccess,
+  })
+}

src/frontend/src/features/rooms/livekit/prefabs/ControlBar.tsx

@@ -18,6 +18,10 @@ import { SelectToggleDevice } from '../components/controls/SelectToggleDevice'
 import { LeaveButton } from '../components/controls/LeaveButton'
 import { ScreenShareToggle } from '../components/controls/ScreenShareToggle'
 import { css } from '@/styled-system/css'
+import { Button } from '@/primitives'
+import { useStartRecording } from '@/features/rooms/api/startRecording.ts'
+import { useStopRecording } from '@/features/rooms/api/stopRecording.ts'
+import { useParams } from 'wouter'
 
 /** @public */
 export type ControlBarControls = {
@@ -66,6 +70,12 @@ export function ControlBar({
 }: ControlBarProps) {
   const [isChatOpen, setIsChatOpen] = React.useState(false)
   const layoutContext = useMaybeLayoutContext()
+
+  const { roomId } = useParams()
+
+  const { mutateAsync: startRecording } = useStartRecording()
+  const { mutateAsync: stopRecording } = useStopRecording()
+
   React.useEffect(() => {
     if (layoutContext?.widget.state?.showChat !== undefined) {
       setIsChatOpen(layoutContext?.widget.state?.showChat)
@@ -151,6 +161,12 @@ export function ControlBar({
       <OptionsButton />
       <LeaveButton />
       <StartMediaButton />
+      <Button onPress={async () => await startRecording({ slug: roomId })}>
+        start
+      </Button>
+      <Button onPress={async () => await stopRecording({ slug: roomId })}>
+        stop
+      </Button>
     </div>
   )
 }

src/helm/helmfile.yaml

@@ -59,6 +59,13 @@ releases:
           buckets:
             - name: meet-media-storage
               versioning: true
+      - ingress:
+          enabled: true
+          hostname: minio-console.127.0.0.1.nip.io
+          servicePort: 9001
+          annotations:
+            nginx.ingress.kubernetes.io/proxy-body-size: "0"
+            kubernetes.io/ingress.class: nginx
 
   - name: redis
     installed: {{ eq .Environment.Name "dev" | toYaml }}