wip unit test parser classes

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

bin/_config.sh

@@ -5,7 +5,7 @@ set -eo pipefail
 REPO_DIR="$(cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd)"
 UNSET_USER=0
 
-COMPOSE_FILE="${REPO_DIR}/compose.yml"
+COMPOSE_FILE="${REPO_DIR}/docker-compose.yml"
 COMPOSE_PROJECT="meet"
 
 

docker-compose.yml

@@ -1,3 +1,4 @@
+version: '3.8'
 
 services:
   postgresql:

src/backend/core/admin.py

@@ -79,16 +79,23 @@ class RoomAdmin(admin.ModelAdmin):
     inlines = (ResourceAccessInline,)
 
 
-class RecordingAccessInline(admin.TabularInline):
-    """Inline admin class for recording accesses."""
-
-    model = models.RecordingAccess
-    extra = 0
-
-
 @admin.register(models.Recording)
 class RecordingAdmin(admin.ModelAdmin):
     """Recording admin interface declaration."""
 
-    inlines = (RecordingAccessInline,)
     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/analytics.py

@@ -0,0 +1,58 @@
+"""
+Meet analytics class.
+"""
+
+import uuid
+
+from django.conf import settings
+
+from june import analytics as jAnalytics
+
+
+class Analytics:
+    """Analytics integration
+
+    This class wraps the June analytics code to avoid coupling our code directly
+    with this third-party library. By doing so, we create a generic interface
+    for analytics that can be easily modified or replaced in the future.
+    """
+
+    def __init__(self):
+        key = getattr(settings, "ANALYTICS_KEY", None)
+
+        if key is not None:
+            jAnalytics.write_key = key
+
+        self._enabled = key is not None
+
+    def _is_anonymous_user(self, user):
+        """Check if the user is anonymous."""
+        return user is None or user.is_anonymous
+
+    def identify(self, user, **kwargs):
+        """Identify a user"""
+
+        if self._is_anonymous_user(user) or not self._enabled:
+            return
+
+        traits = kwargs.pop("traits", {})
+        traits.update({"email": user.email_anonymized})
+
+        jAnalytics.identify(user_id=user.sub, traits=traits, **kwargs)
+
+    def track(self, user, **kwargs):
+        """Track an event"""
+
+        if not self._enabled:
+            return
+
+        event_data = {}
+        if self._is_anonymous_user(user):
+            event_data["anonymous_id"] = str(uuid.uuid4())
+        else:
+            event_data["user_id"] = user.sub
+
+        jAnalytics.track(**event_data, **kwargs)
+
+
+analytics = Analytics()

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,11 +67,30 @@ class RoomPermissions(permissions.BasePermission):
         return obj.is_administrator(user)
 
 
-class ResourceAccessPermission(IsAuthenticated):
+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.
     """
 
+    def has_permission(self, request, view):
+        """Only allow authenticated users."""
+        return request.user.is_authenticated
+
     def has_object_permission(self, request, view, obj):
         """
         Check that the logged-in user is administrator of the linked room.
@@ -81,9 +102,21 @@ class ResourceAccessPermission(IsAuthenticated):
         return obj.resource.is_administrator(user)
 
 
-class HasAbilityPermission(IsAuthenticated):
-    """Permission class for access objects."""
+class IsRecordingEnabled(permissions.BasePermission):
+    """Check if the recording feature is enabled."""
 
-    def has_object_permission(self, request, view, obj):
-        """Check permission for a given object."""
-        return obj.get_abilities(request.user).get(view.action, False)
+    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/serializers.py

@@ -87,15 +87,6 @@ class NestedResourceAccessSerializer(ResourceAccessSerializer):
     user = UserSerializer(read_only=True)
 
 
-class ListRoomSerializer(serializers.ModelSerializer):
-    """Serialize Room model for a list API endpoint."""
-
-    class Meta:
-        model = models.Room
-        fields = ["id", "name", "slug", "is_public"]
-        read_only_fields = ["id", "slug"]
-
-
 class RoomSerializer(serializers.ModelSerializer):
     """Serialize Room model for the API."""
 
@@ -145,14 +136,3 @@ class RoomSerializer(serializers.ModelSerializer):
         output["is_administrable"] = is_admin
 
         return output
-
-
-class RecordingSerializer(serializers.ModelSerializer):
-    """Serialize Recording for the API."""
-
-    room = ListRoomSerializer(read_only=True)
-
-    class Meta:
-        model = models.Recording
-        fields = ["id", "room", "created_at", "updated_at", "status"]
-        read_only_fields = fields

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,16 +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):
     """
@@ -201,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
@@ -233,6 +342,14 @@ class RoomViewSet(
             role=models.RoleChoices.OWNER,
         )
 
+        analytics.track(
+            user=self.request.user,
+            event="Create Room",
+            properties={
+                "slug": room.slug,
+            },
+        )
+
 
 class ResourceAccessListModelMixin:
     """List mixin for resource access API."""
@@ -279,25 +396,53 @@ class ResourceAccessViewSet(
     serializer_class = serializers.ResourceAccessSerializer
 
 
-class RecordingViewSet(
-    mixins.DestroyModelMixin,
-    mixins.ListModelMixin,
-    viewsets.GenericViewSet,
-):
+class RecordingViewSet(viewsets.GenericViewSet, mixins.ListModelMixin):
     """
-    API endpoints to access and perform actions on recordings.
+    API endpoints to access and perform actions on recording.
     """
 
-    pagination_class = Pagination
-    permission_classes = [permissions.HasAbilityPermission]
-    queryset = models.Recording.objects.all()
-    serializer_class = serializers.RecordingSerializer
+    @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."""
 
-    def get_queryset(self):
-        """Restrict recordings to the user's ones."""
-        user = self.request.user
-        return (
-            super()
-            .get_queryset()
-            .filter(Q(accesses__user=user) | Q(accesses__team__in=user.get_teams()))
+        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/authentication/backends.py

@@ -10,6 +10,8 @@ from mozilla_django_oidc.auth import (
 
 from core.models import User
 
+from ..analytics import analytics
+
 
 class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
     """Custom OpenID Connect (OIDC) Authentication Backend.
@@ -79,6 +81,7 @@ class OIDCAuthenticationBackend(MozillaOIDCAuthenticationBackend):
             else:
                 user = None
 
+        analytics.identify(user=user)
         return user
 
     def create_user(self, claims):

src/backend/core/authentication/views.py

@@ -22,6 +22,8 @@ from mozilla_django_oidc.views import (
     OIDCLogoutView as MozillaOIDCOIDCLogoutView,
 )
 
+from ..analytics import analytics
+
 
 class OIDCLogoutView(MozillaOIDCOIDCLogoutView):
     """Custom logout view for handling OpenID Connect (OIDC) logout flow.
@@ -98,6 +100,10 @@ class OIDCLogoutView(MozillaOIDCOIDCLogoutView):
 
         logout_url = self.redirect_url
 
+        analytics.track(
+            user=request.user,
+            event="Signed Out",
+        )
         if request.user.is_authenticated:
             logout_url = self.construct_oidc_logout_url(request)
 

src/backend/core/factories.py

@@ -73,40 +73,10 @@ class RecordingFactory(factory.django.DjangoModelFactory):
     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
-
-    @factory.post_generation
-    def users(self, create, extracted, **kwargs):
-        """Add users to recording from a given list of users with or without roles."""
-        if create and extracted:
-            for item in extracted:
-                if isinstance(item, models.User):
-                    UserRecordingAccessFactory(recording=self, user=item)
-                else:
-                    UserRecordingAccessFactory(
-                        recording=self, user=item[0], role=item[1]
-                    )
-
-
-class UserRecordingAccessFactory(factory.django.DjangoModelFactory):
-    """Create fake recording user accesses for testing."""
-
-    class Meta:
-        model = models.RecordingAccess
-
-    recording = factory.SubFactory(RecordingFactory)
-    user = factory.SubFactory(UserFactory)
-    role = factory.fuzzy.FuzzyChoice(models.RoleChoices.values)
-
-
-class TeamRecordingAccessFactory(factory.django.DjangoModelFactory):
-    """Create fake recording team accesses for testing."""
-
-    class Meta:
-        model = models.RecordingAccess
-
-    recording = factory.SubFactory(RecordingFactory)
-    team = factory.Sequence(lambda n: f"team{n}")
-    role = factory.fuzzy.FuzzyChoice(models.RoleChoices.values)

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/0005_recording_recordingaccess_and_more.py

@@ -1,67 +0,0 @@
-# Generated by Django 5.1.1 on 2024-11-06 14:31
-
-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.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=[('initiated', 'Initiated'), ('active', 'Active'), ('stopped', 'Stopped'), ('saved', 'Saved'), ('aborted', 'Aborted'), ('failed_to_start', 'Failed to Start'), ('failed_to_stop', 'Failed to Stop')], default='initiated', max_length=20)),
-                ('worker_id', models.CharField(blank=True, help_text='Enter an identifier for the worker recording.This ID is retained even when the worker stops, allowing for easy tracking.', max_length=255, null=True, verbose_name='Worker ID')),
-                ('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',),
-            },
-        ),
-        migrations.CreateModel(
-            name='RecordingAccess',
-            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')),
-                ('team', models.CharField(blank=True, max_length=100)),
-                ('role', models.CharField(choices=[('member', 'Member'), ('administrator', 'Administrator'), ('owner', 'Owner')], default='member', max_length=20)),
-                ('recording', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='accesses', to='core.recording')),
-                ('user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
-            ],
-            options={
-                'verbose_name': 'Recording/user relation',
-                'verbose_name_plural': 'Recording/user relations',
-                'db_table': 'meet_recording_access',
-                'ordering': ('-created_at',),
-            },
-        ),
-        migrations.AddConstraint(
-            model_name='recording',
-            constraint=models.UniqueConstraint(condition=models.Q(('status__in', ['active', 'initiated'])), fields=('room',), name='unique_initiated_or_active_recording_per_room'),
-        ),
-        migrations.AddConstraint(
-            model_name='recordingaccess',
-            constraint=models.UniqueConstraint(condition=models.Q(('user__isnull', False)), fields=('user', 'recording'), name='unique_recording_user', violation_error_message='This user is already in this recording.'),
-        ),
-        migrations.AddConstraint(
-            model_name='recordingaccess',
-            constraint=models.UniqueConstraint(condition=models.Q(('team__gt', '')), fields=('team', 'recording'), name='unique_recording_team', violation_error_message='This team is already in this recording.'),
-        ),
-        migrations.AddConstraint(
-            model_name='recordingaccess',
-            constraint=models.CheckConstraint(condition=models.Q(models.Q(('team', ''), ('user__isnull', False)), models.Q(('team__gt', ''), ('user__isnull', True)), _connector='OR'), name='check_recording_access_either_user_or_team', violation_error_message='Either user or team must be set, not both.'),
-        ),
-    ]

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

@@ -4,7 +4,6 @@ Declare and configure the models for the Meet core application
 
 import uuid
 from logging import getLogger
-from typing import List
 
 from django.conf import settings
 from django.contrib.auth import models as auth_models
@@ -40,7 +39,7 @@ class RoleChoices(models.TextChoices):
 
 
 class RecordingStatusChoices(models.TextChoices):
-    """Enumeration of possible states for a recording operation."""
+    """Recording status choices."""
 
     INITIATED = "initiated", _("Initiated")
     ACTIVE = "active", _("Active")
@@ -51,25 +50,27 @@ class RecordingStatusChoices(models.TextChoices):
     FAILED_TO_STOP = "failed_to_stop", _("Failed to Stop")
 
     @classmethod
-    def is_final(cls, status):
-        """Determine if the recording status represents a final state.
-
-        A final status indicates the recording flow has completed, either
-        successfully or unsuccessfully.
-        """
-
-        return status in {
+    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_unsuccessful(cls, status):
-        """Determine if the recording status represents an unsuccessful state."""
-        return status in {cls.ABORTED, cls.FAILED_TO_START, cls.FAILED_TO_STOP}
+    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):
@@ -205,38 +206,10 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
         return f"***@{self.email.split('@')[1]}"
 
 
-def get_resource_roles(resource: models.Model, user: User) -> List[str]:
-    """
-    Get all roles assigned to a user for a specific resource, including team-based roles.
-
-    Args:
-        resource: The resource to check permissions for
-        user: The user to get roles for
-
-    Returns:
-        List of role strings assigned to the user
-    """
-    if not user.is_authenticated:
-        return []
-
-    # Use pre-annotated roles if available from viewset optimization
-    if hasattr(resource, "user_roles"):
-        return resource.user_roles or []
-
-    try:
-        return list(
-            resource.accesses.filter_user(user)
-            .values_list("role", flat=True)
-            .distinct()
-        )
-    except (IndexError, models.ObjectDoesNotExist):
-        return []
-
-
 class Resource(BaseModel):
     """Model to define access control"""
 
-    is_public = models.BooleanField(default=settings.RESOURCE_DEFAULT_IS_PUBLIC)
+    is_public = models.BooleanField(default=settings.DEFAULT_ROOM_IS_PUBLIC)
     users = models.ManyToManyField(
         User,
         through="ResourceAccess",
@@ -389,78 +362,19 @@ class Room(Resource):
         super().clean_fields(exclude=exclude)
 
 
-class BaseAccessManager(models.Manager):
-    """Base manager for handling resource access control."""
-
-    def filter_user(self, user):
-        """Filter accesses for a given user, including both direct and team-based access."""
-        return self.filter(models.Q(user=user) | models.Q(team__in=user.get_teams()))
-
-
-class BaseAccess(BaseModel):
-    """Base model for accesses to handle resources."""
-
-    user = models.ForeignKey(
-        User,
-        on_delete=models.CASCADE,
-        null=True,
-        blank=True,
-    )
-    team = models.CharField(max_length=100, blank=True)
-    role = models.CharField(
-        max_length=20, choices=RoleChoices.choices, default=RoleChoices.MEMBER
-    )
-
-    objects = BaseAccessManager()
-
-    class Meta:
-        abstract = True
-
-    def _get_abilities(self, resource, user):
-        """
-        Compute and return abilities for a given user taking into account
-        the current state of the object.
-        """
-
-        roles = get_resource_roles(resource, user)
-
-        is_owner = RoleChoices.OWNER in roles
-        has_privileges = is_owner or RoleChoices.ADMIN in roles
-
-        # Default values for unprivileged users
-        set_role_to = set()
-        can_delete = False
-
-        # Special handling when modifying an owner's access
-        if self.role == RoleChoices.OWNER:
-            # Prevent orphaning the resource
-            can_delete = (
-                is_owner
-                and resource.accesses.filter(role=RoleChoices.OWNER).count() > 1
-            )
-            if can_delete:
-                set_role_to = {RoleChoices.ADMIN, RoleChoices.OWNER, RoleChoices.MEMBER}
-        elif has_privileges:
-            can_delete = True
-            set_role_to = {RoleChoices.ADMIN, RoleChoices.MEMBER}
-            if is_owner:
-                set_role_to.add(RoleChoices.OWNER)
-
-        # Remove the current role as we don't want to propose it as an option
-        set_role_to.discard(self.role)
-
-        return {
-            "destroy": can_delete,
-            "update": bool(set_role_to),
-            "partial_update": bool(set_role_to),
-            "retrieve": bool(roles),
-            "set_role_to": sorted(r.value for r in set_role_to),
-        }
-
-
+# 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,
@@ -472,14 +386,20 @@ class Recording(BaseModel):
         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."
+            "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."
         ),
     )
 
@@ -491,102 +411,32 @@ class Recording(BaseModel):
         constraints = [
             models.UniqueConstraint(
                 fields=["room"],
-                condition=models.Q(
-                    status__in=[
-                        RecordingStatusChoices.ACTIVE,
-                        RecordingStatusChoices.INITIATED,
-                    ]
-                ),
+                condition=models.Q(status=RecordingStatusChoices.ACTIVE)
+                | models.Q(status=RecordingStatusChoices.INITIATED),
                 name="unique_initiated_or_active_recording_per_room",
             )
         ]
 
     def __str__(self):
-        return f"Recording {self.id} ({self.status})"
+        return str(self.id)
 
     def get_abilities(self, user):
         """Compute and return abilities for a given user on the recording."""
-
-        roles = set(get_resource_roles(self, user))
-
-        is_owner_or_admin = bool(
-            roles.intersection({RoleChoices.OWNER, RoleChoices.ADMIN})
-        )
-
-        is_final_status = RecordingStatusChoices.is_final(self.status)
+        is_creator = user == self.creator
+        is_final_status = RecordingStatusChoices.is_final_status(self.status)
 
         return {
-            "destroy": is_owner_or_admin and is_final_status,
+            "destroy": is_creator and is_final_status,
             "partial_update": False,
-            "retrieve": is_owner_or_admin,
-            "stop": is_owner_or_admin and not is_final_status,
+            "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_unsuccessful = RecordingStatusChoices.is_unsuccessful(self.status)
+        is_in_error = RecordingStatusChoices.is_error_status(self.status)
         is_already_saved = self.status == RecordingStatusChoices.SAVED
 
-        return not is_unsuccessful and not is_already_saved
-
-
-class RecordingAccess(BaseAccess):
-    """Relation model to give access to a recording for a user or a team with a role.
-
-    Recording Status Flow:
-    1. INITIATED: Initial state when recording is requested
-    2. ACTIVE: Recording is currently in progress
-    3. STOPPED: Recording has been stopped by user/system
-    4. SAVED: Recording has been successfully processed and stored
-
-    Error States:
-    - FAILED_TO_START: Worker failed to initialize recording
-    - FAILED_TO_STOP: Worker failed during stop operation
-    - ABORTED: Recording was terminated before completion
-
-    Warning: Worker failures may lead to database inconsistency between the actual
-    recording state and its status in the database.
-    """
-
-    recording = models.ForeignKey(
-        Recording,
-        on_delete=models.CASCADE,
-        related_name="accesses",
-    )
-
-    class Meta:
-        db_table = "meet_recording_access"
-        ordering = ("-created_at",)
-        verbose_name = _("Recording/user relation")
-        verbose_name_plural = _("Recording/user relations")
-        constraints = [
-            models.UniqueConstraint(
-                fields=["user", "recording"],
-                condition=models.Q(user__isnull=False),  # Exclude null users
-                name="unique_recording_user",
-                violation_error_message=_("This user is already in this recording."),
-            ),
-            models.UniqueConstraint(
-                fields=["team", "recording"],
-                condition=models.Q(team__gt=""),  # Exclude empty string teams
-                name="unique_recording_team",
-                violation_error_message=_("This team is already in this recording."),
-            ),
-            models.CheckConstraint(
-                condition=models.Q(user__isnull=False, team="")
-                | models.Q(user__isnull=True, team__gt=""),
-                name="check_recording_access_either_user_or_team",
-                violation_error_message=_("Either user or team must be set, not both."),
-            ),
-        ]
-
-    def __str__(self):
-        return f"{self.user!s} is {self.role:s} in {self.recording!s}"
-
-    def get_abilities(self, user):
-        """
-        Compute and return abilities for a given user on the recording access.
-        """
-        return self._get_abilities(self.recording, user)
+        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/test_api_recordings_delete.py

@@ -1,114 +0,0 @@
-"""
-Test recordings API endpoints in the Meet core app: delete.
-"""
-
-import pytest
-from rest_framework.test import APIClient
-
-from ...factories import RecordingFactory, UserFactory, UserRecordingAccessFactory
-from ...models import Recording
-
-pytestmark = pytest.mark.django_db
-
-
-def test_api_recordings_delete_anonymous():
-    """Anonymous users should not be allowed to destroy a recording."""
-    recording = RecordingFactory()
-    client = APIClient()
-
-    response = client.delete(
-        f"/api/v1.0/recordings/{recording.id!s}/",
-    )
-
-    assert response.status_code == 401
-    assert Recording.objects.count() == 1
-
-
-def test_api_recordings_delete_authenticated():
-    """
-    Authenticated users should not be allowed to delete a recording
-    from which they are not related.
-    """
-    recording = RecordingFactory()
-    user = UserFactory()
-
-    client = APIClient()
-    client.force_login(user)
-
-    response = client.delete(
-        f"/api/v1.0/recordings/{recording.id!s}/",
-    )
-
-    assert response.status_code == 404
-    assert Recording.objects.count() == 1
-
-
-def test_api_recordings_delete_members():
-    """
-    Authenticated users should not be allowed to delete a recording
-    from which they are only a member.
-    """
-
-    user = UserFactory()
-    access = UserRecordingAccessFactory(role="member", user=user)
-
-    client = APIClient()
-    client.force_login(user)
-
-    response = client.delete(
-        f"/api/v1.0/recordings/{access.recording.id}/",
-    )
-
-    assert response.status_code == 403
-    assert Recording.objects.count() == 1
-
-
-@pytest.mark.parametrize(
-    "role",
-    ["owner", "administrator"],
-)
-def test_api_recordings_delete_active(role):
-    """
-    Authenticated users cannot delete active recordings, even with deletion privileges.
-    """
-
-    user = UserFactory()
-
-    recording = RecordingFactory(status="active")
-    access = UserRecordingAccessFactory(role=role, user=user, recording=recording)
-
-    client = APIClient()
-    client.force_login(user)
-
-    response = client.delete(
-        f"/api/v1.0/recordings/{access.recording.id}/",
-    )
-
-    assert response.status_code == 403
-    assert Recording.objects.count() == 1
-
-
-@pytest.mark.parametrize(
-    "role",
-    ["owner", "administrator"],
-)
-def test_api_recordings_delete_final(role):
-    """
-    Authenticated users should not be allowed to delete an active recording
-    from which they are an admin or owner.
-    """
-
-    user = UserFactory()
-
-    recording = RecordingFactory(status="saved")
-    access = UserRecordingAccessFactory(role=role, user=user, recording=recording)
-
-    client = APIClient()
-    client.force_login(user)
-
-    response = client.delete(
-        f"/api/v1.0/recordings/{access.recording.id}/",
-    )
-
-    assert response.status_code == 204
-    assert Recording.objects.count() == 0

src/backend/core/tests/recording/test_api_recordings_list.py

@@ -1,180 +0,0 @@
-"""
-Test recordings API endpoints in the Meet core app: list.
-"""
-
-import operator
-from unittest import mock
-
-import pytest
-from rest_framework.pagination import PageNumberPagination
-from rest_framework.test import APIClient
-
-from core import factories
-
-pytestmark = pytest.mark.django_db
-
-
-def test_api_recordings_list_anonymous():
-    """Anonymous users should not be able to list recordings."""
-    factories.RecordingFactory()
-    response = APIClient().get("/api/v1.0/recordings/")
-
-    assert response.status_code == 401
-
-
-@pytest.mark.parametrize(
-    "role",
-    ["administrator", "member", "owner"],
-)
-def test_api_recordings_list_authenticated_direct(role):
-    """
-    Authenticated users listing recordings, should only see the recordings
-    to which they are related.
-    """
-    user = factories.UserFactory()
-    client = APIClient()
-    client.force_login(user)
-
-    other_user = factories.UserFactory()
-
-    access = factories.UserRecordingAccessFactory(role=role, user=user)
-    factories.UserRecordingAccessFactory(user=other_user)
-
-    recording = access.recording
-    room = recording.room
-
-    response = client.get(
-        "/api/v1.0/recordings/",
-    )
-
-    assert response.status_code == 200
-    results = response.json()["results"]
-    assert len(results) == 1
-    expected_ids = {
-        str(recording.id),
-    }
-    result_ids = {result["id"] for result in results}
-    assert expected_ids == result_ids
-    assert results[0] == {
-        "id": str(recording.id),
-        "created_at": recording.created_at.isoformat().replace("+00:00", "Z"),
-        "room": {
-            "id": str(room.id),
-            "is_public": room.is_public,
-            "name": room.name,
-            "slug": room.slug,
-        },
-        "status": "initiated",
-        "updated_at": recording.updated_at.isoformat().replace("+00:00", "Z"),
-    }
-
-
-def test_api_recording_list_authenticated_via_team(mock_user_get_teams):
-    """
-    Authenticated users should be able to list recordings they are a
-    owner/administrator/member of via a team.
-    """
-    user = factories.UserFactory()
-
-    client = APIClient()
-    client.force_login(user)
-
-    mock_user_get_teams.return_value = ["team1", "team2", "unknown"]
-
-    recordings_team1 = [
-        access.recording
-        for access in factories.TeamRecordingAccessFactory.create_batch(2, team="team1")
-    ]
-    recordings_team2 = [
-        access.recording
-        for access in factories.TeamRecordingAccessFactory.create_batch(3, team="team2")
-    ]
-
-    expected_ids = {
-        str(recording.id) for recording in recordings_team1 + recordings_team2
-    }
-
-    response = client.get("/api/v1.0/recordings/")
-
-    assert response.status_code == 200
-    results = response.json()["results"]
-    assert len(results) == 5
-    results_id = {result["id"] for result in results}
-    assert expected_ids == results_id
-
-
-@mock.patch.object(PageNumberPagination, "get_page_size", return_value=2)
-def test_api_recordings_list_pagination(_mock_page_size):
-    """Pagination should work as expected."""
-    user = factories.UserFactory()
-    client = APIClient()
-    client.force_login(user)
-
-    recording_ids = [
-        str(access.recording_id)
-        for access in factories.UserRecordingAccessFactory.create_batch(3, user=user)
-    ]
-
-    response = client.get("/api/v1.0/recordings/")
-
-    assert response.status_code == 200
-    content = response.json()
-    assert content["count"] == 3
-    assert content["next"] == "http://testserver/api/v1.0/recordings/?page=2"
-    assert content["previous"] is None
-
-    assert len(content["results"]) == 2
-    for item in content["results"]:
-        recording_ids.remove(item["id"])
-
-    # Get page 2
-    response = client.get(
-        "/api/v1.0/recordings/?page=2",
-    )
-
-    assert response.status_code == 200
-    content = response.json()
-
-    assert content["count"] == 3
-    assert content["next"] is None
-    assert content["previous"], "http://testserver/api/v1.0/recordings/"
-
-    assert len(content["results"]) == 1
-    recording_ids.remove(content["results"][0]["id"])
-    assert recording_ids == []
-
-
-def test_api_recordings_list_authenticated_distinct():
-    """A recording for a room with several related users should only be listed once."""
-    user = factories.UserFactory()
-    other_user = factories.UserFactory()
-    client = APIClient()
-    client.force_login(user)
-
-    recording = factories.RecordingFactory(users=[user, other_user])
-
-    response = client.get("/api/v1.0/recordings/")
-
-    assert response.status_code == 200
-    content = response.json()
-    assert len(content["results"]) == 1
-    assert content["results"][0]["id"] == str(recording.id)
-
-
-def test_api_recordings_list_ordering_default():
-    """Recordings should be ordered by descending "updated_at" by default"""
-    user = factories.UserFactory()
-    client = APIClient()
-    client.force_login(user)
-
-    factories.RecordingFactory.create_batch(5, users=[user])
-
-    response = client.get("/api/v1.0/recordings/")
-
-    assert response.status_code == 200
-    results = response.json()["results"]
-    assert len(results) == 5
-
-    # Check that results are sorted by descending "updated_at" as expected
-    for i in range(4):
-        assert operator.ge(results[i]["updated_at"], results[i + 1]["updated_at"])

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_analytics.py

@@ -0,0 +1,132 @@
+"""
+Test for the Analytics class.
+"""
+
+# pylint: disable=W0212
+
+from unittest.mock import patch
+
+from django.contrib.auth.models import AnonymousUser
+from django.test.utils import override_settings
+
+import pytest
+
+from core.analytics import Analytics
+from core.factories import UserFactory
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture(name="mock_june_analytics")
+def _mock_june_analytics():
+    with patch("core.analytics.jAnalytics") as mock:
+        yield mock
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+def test_analytics_init_enabled(mock_june_analytics):
+    """Should enable analytics and set the write key correctly when ANALYTICS_KEY is set."""
+    analytics = Analytics()
+    assert analytics._enabled is True
+    assert mock_june_analytics.write_key == "test_key"
+
+
+@override_settings(ANALYTICS_KEY=None)
+def test_analytics_init_disabled():
+    """Should disable analytics when ANALYTICS_KEY is not set."""
+    analytics = Analytics()
+    assert analytics._enabled is False
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+def test_analytics_identify_user(mock_june_analytics):
+    """Should identify a user with the correct traits when analytics is enabled."""
+    user = UserFactory(sub="12345", email="user@example.com")
+    analytics = Analytics()
+    analytics.identify(user)
+    mock_june_analytics.identify.assert_called_once_with(
+        user_id="12345", traits={"email": "***@example.com"}
+    )
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+def test_analytics_identify_user_with_traits(mock_june_analytics):
+    """Should identify a user with additional traits when analytics is enabled."""
+    user = UserFactory(sub="12345", email="user@example.com")
+    analytics = Analytics()
+    analytics.identify(user, traits={"email": "user@example.com", "foo": "foo"})
+    mock_june_analytics.identify.assert_called_once_with(
+        user_id="12345", traits={"email": "***@example.com", "foo": "foo"}
+    )
+
+
+@override_settings(ANALYTICS_KEY=None)
+def test_analytics_identify_not_enabled(mock_june_analytics):
+    """Should not call identify when analytics is not enabled."""
+    user = UserFactory(sub="12345", email="user@example.com")
+    analytics = Analytics()
+    analytics.identify(user)
+    mock_june_analytics.identify.assert_not_called()
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+def test_analytics_identify_no_user(mock_june_analytics):
+    """Should not call identify when the user is None."""
+    analytics = Analytics()
+    analytics.identify(None)
+    mock_june_analytics.identify.assert_not_called()
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+def test_analytics_identify_anonymous_user(mock_june_analytics):
+    """Should not call identify when the user is anonymous."""
+    user = AnonymousUser()
+    analytics = Analytics()
+    analytics.identify(user)
+    mock_june_analytics.identify.assert_not_called()
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+def test_analytics_track_event(mock_june_analytics):
+    """Should track an event with the correct user and event details when analytics is enabled."""
+    user = UserFactory(sub="12345")
+    analytics = Analytics()
+    analytics.track(user, event="test_event", foo="foo")
+    mock_june_analytics.track.assert_called_once_with(
+        user_id="12345", event="test_event", foo="foo"
+    )
+
+
+@override_settings(ANALYTICS_KEY=None)
+def test_analytics_track_event_not_enabled(mock_june_analytics):
+    """Should not call track when analytics is not enabled."""
+    user = UserFactory(sub="12345")
+    analytics = Analytics()
+    analytics.track(user, event="test_event", foo="foo")
+
+    mock_june_analytics.track.assert_not_called()
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+@patch("uuid.uuid4", return_value="test_uuid4")
+def test_analytics_track_event_no_user(mock_uuid4, mock_june_analytics):
+    """Should track an event with a random anonymous user ID when the user is None."""
+    analytics = Analytics()
+    analytics.track(None, event="test_event", foo="foo")
+    mock_june_analytics.track.assert_called_once_with(
+        anonymous_id="test_uuid4", event="test_event", foo="foo"
+    )
+    mock_uuid4.assert_called_once()
+
+
+@override_settings(ANALYTICS_KEY="test_key")
+@patch("uuid.uuid4", return_value="test_uuid4")
+def test_analytics_track_event_anonymous_user(mock_uuid4, mock_june_analytics):
+    """Should track an event with a random anonymous user ID when the user is anonymous."""
+    user = AnonymousUser()
+    analytics = Analytics()
+    analytics.track(user, event="test_event", foo="foo")
+    mock_june_analytics.track.assert_called_once_with(
+        anonymous_id="test_uuid4", event="test_event", foo="foo"
+    )
+    mock_uuid4.assert_called_once()

src/backend/core/tests/test_model_recording.py

@@ -6,12 +6,7 @@ from django.core.exceptions import ValidationError
 
 import pytest
 
-from core.factories import (
-    RecordingFactory,
-    RoomFactory,
-    UserFactory,
-    UserRecordingAccessFactory,
-)
+from core.factories import RecordingFactory, RoomFactory, UserFactory
 from core.models import Recording, RecordingStatusChoices
 
 pytestmark = pytest.mark.django_db
@@ -20,7 +15,7 @@ pytestmark = pytest.mark.django_db
 def test_models_recording_str():
     """The str representation should be the recording ID."""
     recording = RecordingFactory()
-    assert str(recording) == f"Recording {recording.id} (initiated)"
+    assert str(recording) == str(recording.id)
 
 
 def test_models_recording_ordering():
@@ -31,6 +26,14 @@ def test_models_recording_ordering():
     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()
@@ -66,36 +69,25 @@ def test_models_recording_multiple_finished_allowed():
 
 
 # Test get_abilities method
-
-
-@pytest.mark.parametrize(
-    "role",
-    ["owner", "administrator"],
-)
-def test_models_recording_get_abilities_privileges_active(role):
-    """Test abilities for active recording and privileged user."""
-
-    user = UserFactory()
-    access = UserRecordingAccessFactory(role=role, user=user)
-    access.recording.status = RecordingStatusChoices.ACTIVE
-    abilities = access.recording.get_abilities(user)
+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,  # Privileged users can always retrieve
+        "retrieve": True,  # Creator can always retrieve
         "stop": True,  # Not final status, so can stop
         "update": False,
     }
 
 
-def test_models_recording_get_abilities_member_active():
-    """Test abilities for a user who is unprivileged."""
-
-    user = UserFactory()
-    access = UserRecordingAccessFactory(role="member", user=user)
-    access.recording.status = RecordingStatusChoices.ACTIVE
-    abilities = access.recording.get_abilities(user)
+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,
@@ -106,47 +98,21 @@ def test_models_recording_get_abilities_member_active():
     }
 
 
-@pytest.mark.parametrize(
-    "role",
-    ["owner", "administrator"],
-)
-def test_models_recording_get_abilities_privileges_final(role):
-    """Test abilities for active recording and privileged user."""
-
-    user = UserFactory()
-    access = UserRecordingAccessFactory(role=role, user=user)
-    access.recording.status = RecordingStatusChoices.SAVED
-    abilities = access.recording.get_abilities(user)
+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,
+        "destroy": True,  # Final status, creator can destroy
         "partial_update": False,
-        "retrieve": True,  # Privileged users can always retrieve
-        "stop": False,  # In final status, so can not stop
-        "update": False,
-    }
-
-
-def test_models_recording_get_abilities_member_final():
-    """Test abilities for a user who is unprivileged."""
-
-    user = UserFactory()
-    access = UserRecordingAccessFactory(role="member", user=user)
-    access.recording.status = RecordingStatusChoices.SAVED
-    abilities = access.recording.get_abilities(user)
-
-    assert abilities == {
-        "destroy": False,
-        "partial_update": False,
-        "retrieve": False,
-        "stop": 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)
@@ -185,6 +151,12 @@ def test_models_recording_worker_id_optional():
     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()
@@ -193,6 +165,12 @@ def test_models_recording_invalid_status():
         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()
@@ -201,6 +179,14 @@ def test_models_recording_room_deletion():
     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

src/backend/core/tests/test_models_recording_accesses.py

@@ -1,312 +0,0 @@
-"""
-Unit tests for the RecordingAccess model
-"""
-
-from django.contrib.auth.models import AnonymousUser
-from django.core.exceptions import ValidationError
-
-import pytest
-
-from core import factories
-
-pytestmark = pytest.mark.django_db
-
-
-def test_models_recording_accesses_str():
-    """
-    The str representation should include user email, recording ID and role.
-    """
-    user = factories.UserFactory(email="david.bowman@example.com")
-    access = factories.UserRecordingAccessFactory(
-        role="member",
-        user=user,
-    )
-    assert (
-        str(access)
-        == f"david.bowman@example.com is member in Recording {access.recording.id} (initiated)"
-    )
-
-
-def test_models_recording_accesses_unique_user():
-    """Recording accesses should be unique for a given couple of user and recording."""
-    access = factories.UserRecordingAccessFactory()
-
-    with pytest.raises(
-        ValidationError,
-        match="This user is already in this recording.",
-    ):
-        factories.UserRecordingAccessFactory(
-            user=access.user, recording=access.recording
-        )
-
-
-def test_models_recording_accesses_several_empty_teams():
-    """A recording can have several recording accesses with an empty team."""
-    access = factories.UserRecordingAccessFactory()
-    factories.UserRecordingAccessFactory(recording=access.recording)
-
-
-def test_models_recording_accesses_unique_team():
-    """Recording accesses should be unique for a given couple of team and recording."""
-    access = factories.TeamRecordingAccessFactory()
-
-    with pytest.raises(
-        ValidationError,
-        match="This team is already in this recording.",
-    ):
-        factories.TeamRecordingAccessFactory(
-            team=access.team, recording=access.recording
-        )
-
-
-def test_models_recording_accesses_several_null_users():
-    """A recording can have several recording accesses with a null user."""
-    access = factories.TeamRecordingAccessFactory()
-    factories.TeamRecordingAccessFactory(recording=access.recording)
-
-
-def test_models_recording_accesses_user_and_team_set():
-    """User and team can't both be set on a recording access."""
-    with pytest.raises(
-        ValidationError,
-        match="Either user or team must be set, not both.",
-    ):
-        factories.UserRecordingAccessFactory(team="my-team")
-
-
-def test_models_recording_accesses_user_and_team_empty():
-    """User and team can't both be empty on a recording access."""
-    with pytest.raises(
-        ValidationError,
-        match="Either user or team must be set, not both.",
-    ):
-        factories.UserRecordingAccessFactory(user=None)
-
-
-# get_abilities
-
-
-def test_models_recording_access_get_abilities_anonymous():
-    """Check abilities returned for an anonymous user."""
-    access = factories.UserRecordingAccessFactory()
-    abilities = access.get_abilities(AnonymousUser())
-    assert abilities == {
-        "destroy": False,
-        "retrieve": False,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }
-
-
-def test_models_recording_access_get_abilities_authenticated():
-    """Check abilities returned for an authenticated user."""
-    access = factories.UserRecordingAccessFactory()
-    user = factories.UserFactory()
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": False,
-        "retrieve": False,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }
-
-
-# - for owner
-
-
-def test_models_recording_access_get_abilities_for_owner_of_self_allowed():
-    """
-    Check abilities of self access for the owner of a recording when
-    there is more than one owner left.
-    """
-    access = factories.UserRecordingAccessFactory(role="owner")
-    factories.UserRecordingAccessFactory(recording=access.recording, role="owner")
-    abilities = access.get_abilities(access.user)
-    assert abilities == {
-        "destroy": True,
-        "retrieve": True,
-        "update": True,
-        "partial_update": True,
-        "set_role_to": ["administrator", "member"],
-    }
-
-
-def test_models_recording_access_get_abilities_for_owner_of_self_last():
-    """
-    Check abilities of self access for the owner of a recording when there is only one owner left.
-    """
-    access = factories.UserRecordingAccessFactory(role="owner")
-    abilities = access.get_abilities(access.user)
-    assert abilities == {
-        "destroy": False,
-        "retrieve": True,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }
-
-
-def test_models_recording_access_get_abilities_for_owner_of_owner():
-    """Check abilities of owner access for the owner of a recording."""
-    access = factories.UserRecordingAccessFactory(role="owner")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="owner"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": True,
-        "retrieve": True,
-        "update": True,
-        "partial_update": True,
-        "set_role_to": ["administrator", "member"],
-    }
-
-
-def test_models_recording_access_get_abilities_for_owner_of_administrator():
-    """Check abilities of administrator access for the owner of a recording."""
-    access = factories.UserRecordingAccessFactory(role="administrator")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="owner"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": True,
-        "retrieve": True,
-        "update": True,
-        "partial_update": True,
-        "set_role_to": ["member", "owner"],
-    }
-
-
-def test_models_recording_access_get_abilities_for_owner_of_member():
-    """Check abilities of member access for the owner of a recording."""
-    access = factories.UserRecordingAccessFactory(role="member")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="owner"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": True,
-        "retrieve": True,
-        "update": True,
-        "partial_update": True,
-        "set_role_to": ["administrator", "owner"],
-    }
-
-
-# - for administrator
-
-
-def test_models_recording_access_get_abilities_for_administrator_of_owner():
-    """Check abilities of owner access for the administrator of a recording."""
-    access = factories.UserRecordingAccessFactory(role="owner")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="administrator"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": False,
-        "retrieve": True,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }
-
-
-def test_models_recording_access_get_abilities_for_administrator_of_administrator():
-    """Check abilities of administrator access for the administrator of a recording."""
-    access = factories.UserRecordingAccessFactory(role="administrator")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="administrator"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": True,
-        "retrieve": True,
-        "update": True,
-        "partial_update": True,
-        "set_role_to": ["member"],
-    }
-
-
-def test_models_recording_access_get_abilities_for_administrator_of_member():
-    """Check abilities of member access for the administrator of a recording."""
-    access = factories.UserRecordingAccessFactory(role="member")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="administrator"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": True,
-        "retrieve": True,
-        "update": True,
-        "partial_update": True,
-        "set_role_to": ["administrator"],
-    }
-
-
-# - for member
-
-
-def test_models_recording_access_get_abilities_for_member_of_owner():
-    """Check abilities of owner access for the member of a recording."""
-    access = factories.UserRecordingAccessFactory(role="owner")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="member"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": False,
-        "retrieve": True,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }
-
-
-def test_models_recording_access_get_abilities_for_member_of_administrator():
-    """Check abilities of administrator access for the member of a recording."""
-    access = factories.UserRecordingAccessFactory(role="administrator")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="member"
-    ).user
-    abilities = access.get_abilities(user)
-    assert abilities == {
-        "destroy": False,
-        "retrieve": True,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }
-
-
-def test_models_recording_access_get_abilities_for_member_of_member_user(
-    django_assert_num_queries,
-):
-    """Check abilities of member access for the member of a recording."""
-    access = factories.UserRecordingAccessFactory(role="member")
-    factories.UserRecordingAccessFactory(recording=access.recording)  # another one
-    user = factories.UserRecordingAccessFactory(
-        recording=access.recording, role="member"
-    ).user
-
-    with django_assert_num_queries(1):
-        abilities = access.get_abilities(user)
-
-    assert abilities == {
-        "destroy": False,
-        "retrieve": True,
-        "update": False,
-        "partial_update": False,
-        "set_role_to": [],
-    }

src/backend/core/urls.py

@@ -12,10 +12,10 @@ from core.authentication.urls import urlpatterns as oidc_urls
 router = DefaultRouter()
 router.register("users", viewsets.UserViewSet, basename="users")
 router.register("rooms", viewsets.RoomViewSet, basename="rooms")
-router.register("recordings", viewsets.RecordingViewSet, basename="recordings")
 router.register(
     "resource-accesses", viewsets.ResourceAccessViewSet, basename="resource_accesses"
 )
+router.register("recording", viewsets.RecordingViewSet, basename="recordings")
 
 urlpatterns = [
     path(

src/backend/meet/settings.py

@@ -395,12 +395,37 @@ class Base(Configuration):
         ),
         "url": values.Value(environ_name="LIVEKIT_API_URL", environ_prefix=None),
     }
-    RESOURCE_DEFAULT_IS_PUBLIC = values.BooleanValue(
-        True, environ_name="RESOURCE_DEFAULT_IS_PUBLIC", environ_prefix=None
+    DEFAULT_ROOM_IS_PUBLIC = values.BooleanValue(
+        True, environ_name="DEFAULT_ROOM_IS_PUBLIC", environ_prefix=None
     )
     ALLOW_UNREGISTERED_ROOMS = values.BooleanValue(
         True, environ_name="ALLOW_UNREGISTERED_ROOMS", environ_prefix=None
     )
+    ANALYTICS_KEY = values.Value(
+        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
@@ -521,6 +546,8 @@ class Test(Base):
 
     CELERY_TASK_ALWAYS_EAGER = values.BooleanValue(True)
 
+    ANALYTICS_KEY = None
+
     def __init__(self):
         # pylint: disable=invalid-name
         self.INSTALLED_APPS += ["drf_spectacular_sidecar"]

src/backend/pyproject.toml

@@ -42,6 +42,7 @@ dependencies = [
     "dockerflow==2024.4.2",
     "easy_thumbnails==2.10",
     "factory_boy==3.3.1",
+    "freezegun==1.5.1",
     "gunicorn==23.0.0",
     "jsonschema==4.23.0",
     "june-analytics-python==2.3.0",
@@ -57,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]
@@ -69,7 +71,6 @@ dependencies = [
 dev = [
     "django-extensions==3.2.3",
     "drf-spectacular-sidecar==2024.7.1",
-    "freezegun==1.5.1",
     "ipdb==0.13.13",
     "ipython==8.27.0",
     "pyfakefs==5.6.0",

src/frontend/Dockerfile

@@ -1,4 +1,4 @@
-FROM node:20-alpine AS frontend-deps
+FROM node:20-alpine as frontend-deps
 
 WORKDIR /home/frontend/
 
@@ -11,11 +11,11 @@ COPY .dockerignore ./.dockerignore
 COPY ./src/frontend/ .
 
 ### ---- Front-end builder image ----
-FROM frontend-deps AS meet
+FROM frontend-deps as meet
 
 WORKDIR /home/frontend
 
-FROM frontend-deps AS meet-dev
+FROM frontend-deps as meet-dev
 
 WORKDIR /home/frontend
 
@@ -25,14 +25,14 @@ CMD [ "npm", "run", "dev"]
 
 # Tilt will rebuild Meet target so, we dissociate meet and meet-builder
 # to avoid rebuilding the app at every changes.
-FROM meet AS meet-builder
+FROM meet as meet-builder
 
 WORKDIR /home/frontend
 
 RUN npm run build
 
 # ---- Front-end image ----
-FROM nginxinc/nginx-unprivileged:1.26-alpine AS frontend-production
+FROM nginxinc/nginx-unprivileged:1.26-alpine as frontend-production
 
 # Un-privileged user running the application
 ARG DOCKER_USER

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/components/Join.tsx

@@ -19,7 +19,7 @@ export const Join = ({
           micLabel={t('join.audioinput.label')}
           camLabel={t('join.videoinput.label')}
           joinLabel={t('join.joinLabel')}
-          userLabel={t('join.usernameLabel')}
+          userLabel={t('join.userLabel')}
         />
       </CenteredContent>
     </Screen>

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/frontend/src/locales/de/rooms.json

@@ -22,6 +22,7 @@
     "joinMeeting": "",
     "toggleOff": "",
     "toggleOn": "",
+    "userLabel": "",
     "usernameHint": "",
     "usernameLabel": ""
   },

src/frontend/src/locales/en/rooms.json

@@ -22,6 +22,7 @@
     "joinMeeting": "Join meeting",
     "toggleOff": "Click to turn off",
     "toggleOn": "Click to turn on",
+    "userLabel": "",
     "usernameHint": "Shown to other participants",
     "usernameLabel": "Your name"
   },

src/frontend/src/locales/fr/rooms.json

@@ -22,6 +22,7 @@
     "joinMeeting": "Rejoindre la réjoindre",
     "toggleOff": "Cliquez pour désactiver",
     "toggleOn": "Cliquez pour activer",
+    "userLabel": "",
     "usernameHint": "Affiché aux autres participants",
     "usernameLabel": "Votre nom"
   },

src/helm/env.d/dev/values.meet.yaml.gotmpl

@@ -47,6 +47,7 @@ backend:
     {{- end }}
     {{- end }}
     LIVEKIT_API_URL: https://livekit.127.0.0.1.nip.io/
+    ANALYTICS_KEY: xwhoIMCZ8PBRjQ2t
     ALLOW_UNREGISTERED_ROOMS: False
     FRONTEND_SILENCE_LIVEKIT_DEBUG: False
     FRONTEND_SUPPORT: "{'id': '58ea6697-8eba-4492-bc59-ad6562585041'}"

src/helm/env.d/production/values.meet.yaml.gotmpl

@@ -94,6 +94,7 @@ backend:
         name: backend
         key: LIVEKIT_API_KEY
     LIVEKIT_API_URL: https://livekit-preprod.beta.numerique.gouv.fr
+    ANALYTICS_KEY: mwuxTKi8o2xzWH54
     ALLOW_UNREGISTERED_ROOMS: False
     FRONTEND_SILENCE_LIVEKIT_DEBUG: False
     FRONTEND_ANALYTICS: "{'id': 'phc_RPYko028Oqtj0c9exLIWwrlrjLxSdxT0ntW0Lam4iom', 'host': 'https://product.visio.numerique.gouv.fr'}"

src/helm/env.d/staging/values.meet.yaml.gotmpl

@@ -93,6 +93,7 @@ backend:
         name: backend
         key: LIVEKIT_API_KEY
     LIVEKIT_API_URL: https://livekit-staging.beta.numerique.gouv.fr
+    ANALYTICS_KEY: Roi1k6IAc2DEqHB0
     ALLOW_UNREGISTERED_ROOMS: False
     FRONTEND_ANALYTICS: "{'id': 'phc_RPYko028Oqtj0c9exLIWwrlrjLxSdxT0ntW0Lam4iom', 'host': 'https://product.visio-staging.beta.numerique.gouv.fr'}"
     FRONTEND_SUPPORT: "{'id': '58ea6697-8eba-4492-bc59-ad6562585041'}"

src/helm/extra/templates/s3.yaml

@@ -5,4 +5,4 @@ metadata:
   namespace: {{ .Release.Namespace | quote }}
 spec:
   provider: data
-  versioned: true
+  versioned: true