wip backend meet

We are in the process of moving meet to summary v2 routes and tasks.
This commits removes code from summary and moves some code from summary to meet,
which should have the responsability of this code.

src/backend/core/analytics.py

@@ -0,0 +1,52 @@
+"""Analytics module."""
+
+import logging
+from enum import StrEnum
+
+from django.conf import settings
+
+import posthog
+
+from core.models import User
+
+logger = logging.getLogger(__name__)
+
+
+class EventName(StrEnum):
+    """Analytics event names."""
+
+    TRANSCRIPT_GENERATION_SUCCESS = "transcript_generation_success"
+    TRANSCRIPT_GENERATION_FAILURE = "transcript_generation_failure"
+    SUMMARY_GENERATION_SUCCESS = "summary_generation_success"
+    SUMMARY_GENERATION_FAILURE = "summary_generation_failure"
+
+
+def capture_event(event_name: EventName, *, user: User, properties=None) -> None:
+    """
+    Capture an analytics event with user properties.
+    """
+    if not settings.POSTHOG_ENABLED:
+        return
+
+    properties = properties or {}
+    properties["$set"] = {
+        "name": user.full_name,
+        "email": user.email,
+        "sub": user.sub,
+    }
+    posthog.capture(event_name, distinct_id=user.id, properties=properties)
+
+
+def is_feature_enabled(feature_name: str, distinct_id: str) -> bool:
+    """Check if a feature flag is enabled for a user."""
+    if not settings.POSTHOG_ENABLED:
+        return False
+
+    try:
+        return posthog.feature_enabled(feature_name, distinct_id)
+    except Exception as e:  # noqa: BLE001
+        logger.error("Error checking feature flag %s: %s", feature_name, e)
+        return False
+
+
+__all__ = ["EventName", "capture_event", "is_feature_enabled"]

src/backend/core/api/permissions.py

@@ -136,3 +136,15 @@ class FilePermission(IsAuthenticated):
             raise Http404
 
         return obj.get_abilities(request.user).get(view.action, False)
+
+
+class TranscribeWebhookPermission(permissions.BasePermission):
+    """
+    Permissions applying to the summary webhook endpoint.
+    """
+
+    def has_permission(self, request, view):
+        return request.method == "POST"
+
+    def has_object_permission(self, request, view, obj):
+        return False

src/backend/core/api/viewsets.py

@@ -17,6 +17,7 @@ from django.utils.text import slugify
 from django.utils.translation import gettext_lazy as _
 
 from django_filters import rest_framework as django_filters
+from pydantic import ValidationError
 from rest_framework import (
     decorators,
     filters,
@@ -34,7 +35,7 @@ from rest_framework import (
     status as drf_status,
 )
 
-from core import enums, models, utils
+from core import analytics, enums, models, utils
 from core.api.filters import ListFileFilter
 from core.enums import MEDIA_STORAGE_URL_PATTERN
 from core.recording.enums import FileExtension
@@ -80,6 +81,10 @@ from core.services.subtitle import SubtitleException, SubtitleService
 from core.tasks.file import process_file_deletion
 
 from ..authentication.livekit import LiveKitTokenAuthentication
+from ..authentication.webhooks import AiWebhookAuthentication
+from ..models import AiJobStatusChoices, AiRecordingJob
+from ..tasks.ai_job import handle_summary_received, handle_transcript_received
+from ..transcription import webhook_schemas
 from . import permissions, serializers, throttling
 from .feature_flag import FeatureFlag
 
@@ -915,15 +920,9 @@ class RecordingViewSet(
 
         # Attempt to notify external services about the recording
         # This is a non-blocking operation - failures are logged but don't interrupt the flow
-        notification_succeeded = notification_service.notify_external_services(
-            recording
-        )
+        notification_service.notify_external_services(recording)
 
-        recording.status = (
-            models.RecordingStatusChoices.NOTIFICATION_SUCCEEDED
-            if notification_succeeded
-            else models.RecordingStatusChoices.SAVED
-        )
+        recording.status = models.RecordingStatusChoices.SAVED
         recording.save()
 
         return drf_response.Response(
@@ -1332,3 +1331,92 @@ class FileViewSet(
         request = utils.generate_s3_authorization_headers(f"{url_params.get('key'):s}")
 
         return drf_response.Response("authorized", headers=request.headers, status=200)
+
+
+class AiJobViewSet(
+    viewsets.GenericViewSet,
+):
+    """AI jobs API."""
+
+    permission_classes = []
+    serializer_class = None
+
+    def get_queryset(self):
+        """Restrict AI jobs to current user except webhook endpoint."""
+
+        raise NotImplementedError()
+
+    @decorators.action(
+        detail=False,
+        methods=["post"],
+        url_path="webhook",
+        authentication_classes=[AiWebhookAuthentication],
+        permission_classes=[permissions.TranscribeWebhookPermission],
+    )
+    def on_ai_event(self, request):
+        """Handle incoming hook events for recordings."""
+        logger.debug("Received transcribe webhook event: %s", request.data)
+
+        try:
+            payload = webhook_schemas.webhook_payload_adapter.validate_python(
+                request.data
+            )
+        except ValidationError as exc:
+            logger.error("Invalid webhook payload: %s", exc)
+            raise drf_exceptions.ValidationError(detail=exc) from exc
+
+        ai_recording_job = AiRecordingJob.objects.filter(
+            remote_job_id=payload.job_id
+        ).first()
+
+        if not ai_recording_job:
+            logger.warning("No AI recording job found for job ID: %s", payload.job_id)
+            return drf_response.Response(
+                {"message": "No AI recording job found for job ID, ignoring."},
+            )
+
+        if ai_recording_job.status == AiJobStatusChoices.SUCCESS:
+            logger.warning(
+                "AI recording job already in success state for job ID: %s",
+                payload.job_id,
+            )
+            return drf_response.Response(
+                {"message": "AI recording job already in success state, ignoring."},
+            )
+
+        if isinstance(payload, webhook_schemas.TranscribeWebhookSuccessPayload):
+            handle_transcript_received.apply_async(
+                args=[payload.job_id, payload.transcription_data_url]
+            )
+        elif isinstance(payload, webhook_schemas.SummarizeWebhookSuccessPayload):
+            handle_summary_received.apply_async(
+                args=[payload.job_id, payload.summary_data_url]
+            )
+        elif isinstance(
+            payload,
+            (
+                webhook_schemas.SummarizeWebhookFailurePayload,
+                webhook_schemas.TranscribeWebhookFailurePayload,
+            ),
+        ):
+            ai_recording_job.status = AiJobStatusChoices.FAILED
+            ai_recording_job.save()
+            analytics.capture_event(
+                analytics.EventName.TRANSCRIPT_GENERATION_FAILURE
+                if isinstance(payload, webhook_schemas.TranscribeWebhookFailurePayload)
+                else analytics.EventName.SUMMARY_GENERATION_FAILURE,
+                user=ai_recording_job.user,
+                properties={
+                    "generation_time_seconds": (
+                        timezone.now() - ai_recording_job.created_at
+                    ).total_seconds(),
+                    "ai_recording_job_id": ai_recording_job.id,
+                    "recording_id": ai_recording_job.recording.id,
+                },
+            )
+        else:
+            raise NotImplementedError()
+
+        return drf_response.Response(
+            {"message": "Event processed."},
+        )

src/backend/core/authentication/webhooks.py

@@ -0,0 +1,34 @@
+"""Webhooks authentication."""
+
+import logging
+
+from django.conf import settings
+from django.contrib.auth.models import AnonymousUser
+
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.exceptions import AuthenticationFailed
+
+logger = logging.getLogger(__name__)
+
+
+class AiWebhookAuthentication(BaseAuthentication):
+    """
+    Custom authentication class for AI webhook requests.
+    Validates the API key in the Authorization header.
+    """
+
+    def authenticate(self, request):
+        """
+        Authenticate the request and return a two-tuple of (user, token).
+        """
+
+        authorization_header: str = request.headers.get("Authorization") or ""
+        if authorization_header.removeprefix("Bearer ") != settings.AI_WEBHOOK_API_KEY:
+            logger.warning(
+                "Authentication failed: Bad Authorization header (ip: %s)",
+                request.META.get("REMOTE_ADDR"),
+            )
+            raise AuthenticationFailed()
+
+        # No users are associated with the transcribe webhooks
+        return AnonymousUser(), None

src/backend/core/migrations/0019_airecordingjob.py

@@ -0,0 +1,36 @@
+# Generated by Django 5.2.14 on 2026-05-14 12:40
+
+import django.db.models.deletion
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0018_rename_active_application_is_active'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AiRecordingJob',
+            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')),
+                ('remote_job_id', models.CharField(blank=True, max_length=255, null=True, unique=True)),
+                ('type', models.CharField(choices=[('transcript', 'Transcript'), ('summary', 'Summary')], max_length=25)),
+                ('status', models.CharField(choices=[('pending', 'Pending'), ('success', 'Success'), ('failed', 'Failed')], max_length=25)),
+                ('language', models.CharField(choices=[('fr', 'fr'), ('en', 'en'), ('de', 'de'), ('nl', 'nl')], default='fr', max_length=2)),
+                ('docs_app_id', models.CharField(blank=True, max_length=255, null=True)),
+                ('recording', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='ai_jobs', to='core.recording')),
+            ],
+            options={
+                'verbose_name': 'AiJob',
+                'verbose_name_plural': 'AiJobs',
+                'db_table': 'ai_job',
+                'ordering': ('created_at',),
+                'indexes': [models.Index(fields=['recording', 'type', '-created_at'], name='ai_job_recordi_ca452e_idx')],
+            },
+        ),
+    ]

src/backend/core/models.py

@@ -590,6 +590,16 @@ class Recording(BaseModel):
         verbose_name=_("Recording options"),
         help_text=_("Recording options"),
     )
+    started_at = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text=_("Recording start timestamp as recorded by livekit."),
+    )
+    ended_at = models.DateTimeField(
+        null=True,
+        blank=True,
+        help_text=_("Recording end timestamp as recorded by livekit."),
+    )
 
     class Meta:
         db_table = "meet_recording"
@@ -736,6 +746,66 @@ class RecordingAccess(BaseAccess):
         return self._get_abilities(self.recording, user)
 
 
+class AiJobStatusChoices(models.TextChoices):
+    """Possible states of a file."""
+
+    PENDING = "pending", _("Pending")
+    SUCCESS = "success", _("Success")
+    FAILED = "failed", _("Failed")
+
+
+class AiJobTypeChoices(models.TextChoices):
+    """Possible types of Ai Jobs."""
+
+    TRANSCRIPT = "transcript", _("Transcript")
+    SUMMARIZE = "summary", _("Summary")
+
+
+class AiRecordingJob(BaseModel):
+    """
+    A job that is run to process an audio file.
+    """
+
+    remote_job_id = models.CharField(max_length=255, unique=True, null=True, blank=True)
+    type = models.CharField(
+        max_length=25,
+        choices=AiJobTypeChoices.choices,
+    )
+    recording = models.ForeignKey(
+        Recording, on_delete=models.CASCADE, related_name="ai_jobs"
+    )
+    status = models.CharField(
+        max_length=25,
+        choices=AiJobStatusChoices.choices,
+    )
+    language = models.CharField(
+        max_length=2,
+        choices=(("fr", "fr"), ("en", "en"), ("de", "de"), ("nl", "nl")),
+        default="fr",
+    )
+    docs_app_id = models.CharField(max_length=255, null=True, blank=True)
+
+    class Meta:
+        db_table = "ai_job"
+        verbose_name = _("AiJob")
+        verbose_name_plural = _("AiJobs")
+        ordering = ("created_at",)
+        indexes = [
+            models.Index(fields=["recording", "type", "-created_at"]),
+        ]
+
+    def __str__(self):
+        return f"{self.recording.id} - {self.type} - {self.status}"
+
+    @property
+    def user(self):
+        return (
+            RecordingAccess.objects.select_related("user")
+            .filter(role=RoleChoices.OWNER, recording_id=self.recording.id)
+            .first()
+        ).user
+
+
 class ApplicationScope(models.TextChoices):
     """Available permission scopes for application operations."""
 

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

@@ -12,11 +12,11 @@ from django.utils.translation import get_language, override
 from django.utils.translation import gettext_lazy as _
 
 import aiohttp
-import requests
 from asgiref.sync import async_to_sync
 from livekit import api as livekit_api
 
 from core import models, utils
+from core.tasks.ai_job import call_transcribe_service
 
 logger = logging.getLogger(__name__)
 
@@ -45,22 +45,17 @@ class NotificationService:
         """Process a recording based on its mode."""
 
         if recording.mode == models.RecordingModeChoices.TRANSCRIPT:
-            return self._notify_summary_service(recording)
-
-        if recording.mode == models.RecordingModeChoices.SCREEN_RECORDING:
-            summary_success = True
+            self._notify_summary_service(recording)
+        elif recording.mode == models.RecordingModeChoices.SCREEN_RECORDING:
             if recording.options.get("transcribe", False):
-                summary_success = self._notify_summary_service(recording)
-
-            email_success = self._notify_user_by_email(recording)
-            return email_success and summary_success
-
-        logger.error(
-            "Unknown recording mode %s for recording %s",
-            recording.mode,
-            recording.id,
-        )
-        return False
+                self._notify_summary_service(recording)
+            self._notify_user_by_email(recording)
+        else:
+            logger.error(
+                "Unknown recording mode %s for recording %s",
+                recording.mode,
+                recording.id,
+            )
 
     @staticmethod
     def _notify_user_by_email(recording) -> bool:
@@ -187,71 +182,17 @@ class NotificationService:
             or not settings.SUMMARY_SERVICE_API_TOKEN
         ):
             logger.error("Summary service not configured")
-            return False
-
-        owner_access = (
-            models.RecordingAccess.objects.select_related("user")
-            .filter(
-                role=models.RoleChoices.OWNER,
-                recording_id=recording.id,
-            )
-            .first()
-        )
-
-        if settings.METADATA_COLLECTOR_ENABLED and recording.options.get(
-            "collect_metadata", False
-        ):
-            output_folder = settings.METADATA_COLLECTOR_OUTPUT_FOLDER
-            metadata_filename = f"{output_folder}/{recording.id}-metadata.json"
-        else:
-            metadata_filename = None
-
-        if not owner_access:
-            logger.error("No owner found for recording %s", recording.id)
-            return False
+            return
 
         started_at, ended_at = async_to_sync(
             NotificationService._get_recording_timestamps
         )(recording.worker_id)
 
-        payload = {
-            "owner_id": str(owner_access.user.id),
-            "recording_filename": recording.key,
-            "metadata_filename": metadata_filename,
-            "email": owner_access.user.email,
-            "sub": owner_access.user.sub,
-            "room": recording.room.name,
-            "language": recording.options.get("language"),
-            "owner_timezone": str(owner_access.user.timezone),
-            "download_link": f"{get_recording_download_base_url()}/{recording.id}",
-            "context_language": owner_access.user.language,
-            "recording_start_at": (started_at.isoformat() if started_at else None),
-            "recording_end_at": (ended_at.isoformat() if ended_at else None),
-        }
+        recording.started_at = started_at
+        recording.ended_at = ended_at
+        recording.save()
 
-        headers = {
-            "Content-Type": "application/json",
-            "Authorization": f"Bearer {settings.SUMMARY_SERVICE_API_TOKEN}",
-        }
-
-        try:
-            response = requests.post(
-                settings.SUMMARY_SERVICE_ENDPOINT,
-                json=payload,
-                headers=headers,
-                timeout=30,
-            )
-            response.raise_for_status()
-        except requests.RequestException as exc:
-            logger.exception(
-                "Summary service error for recording %s. URL: %s. Exception: %s",
-                recording.id,
-                settings.SUMMARY_SERVICE_ENDPOINT,
-                exc,
-            )
-            return False
-
-        return True
+        call_transcribe_service.apply_async(args=[recording.id])
 
 
 notification_service = NotificationService()

src/backend/core/tasks/ai_job.py

@@ -0,0 +1,319 @@
+import logging
+from datetime import datetime
+from urllib.parse import urljoin
+
+from django.conf import settings
+from django.utils import timezone
+
+import requests
+
+from core import analytics, models
+from core.models import (
+    AiJobStatusChoices,
+    AiJobTypeChoices,
+    AiRecordingJob,
+    Recording,
+)
+from core.tasks._task import task
+from core.transcription.locales import get_locale
+from core.transcription.transcript_formatter import TranscriptFormatter
+from core.transcription.webhook_schemas import WhisperXResponse
+from core.utils import generate_download_s3_file_url
+
+logger = logging.getLogger(__name__)
+
+
+@task
+def call_transcribe_service(recording_id):
+    """
+    Call the transcribe service for a given recording.
+    """
+    try:
+        recording = Recording.objects.get(id=recording_id)
+    except Recording.DoesNotExist:
+        logger.error("Recoding %s does not exist", recording_id)
+        return None
+
+    owner_access = (
+        models.RecordingAccess.objects.select_related("user")
+        .filter(
+            role=models.RoleChoices.OWNER,
+            recording_id=recording.id,
+        )
+        .first()
+    )
+
+    if not owner_access:
+        logger.error("No owner found for recording %s", recording.id)
+        return False
+
+    metadata = None
+    if (
+        settings.METADATA_COLLECTOR_ENABLED
+        and recording.options.get("collect_metadata", False)
+        and recording.started_at
+        and recording.ended_at
+    ):
+        output_folder = settings.METADATA_COLLECTOR_OUTPUT_FOLDER
+        metadata_filename = f"{output_folder}/{recording.id}-metadata.json"
+        metadata = {
+            "cloud_storage_url": generate_download_s3_file_url(metadata_filename),
+            "start_at": recording.started_at,
+            "end_at": recording.ended_at,
+        }
+
+    language = (
+        recording.options.get("language") or settings.TRANSCRIPTION_DEFAULT_LANGUAGE
+    )
+    ai_transcribe_job = AiRecordingJob.objects.create(
+        remote_job_id=None,
+        recording=recording,
+        type=AiJobTypeChoices.TRANSCRIPT,
+        status=AiJobStatusChoices.PENDING,
+        language=language,
+    )
+
+    try:
+        response = requests.post(
+            settings.AI_SERVICE_URL + "async-jobs/transcribe/",
+            json={
+                "user_sub": owner_access.user.sub,
+                "language": language,
+                "cloud_storage_url": generate_download_s3_file_url(
+                    recording.key, expires_in=60 * 60 * 24, override_domain=False
+                ),
+                "metadata": metadata,
+            },
+            headers={
+                "Authorization": f"Bearer {settings.AI_SERVICE_API_KEY}",
+            },
+            timeout=10,
+        )
+        response.raise_for_status()
+    except Exception as e:
+        logger.error(
+            "Creating transcription job failed for recording %s: %s", recording_id, e
+        )
+        ai_transcribe_job.status = AiJobStatusChoices.FAILED
+        ai_transcribe_job.save()
+        raise e
+
+    data = response.json()
+
+    ai_transcribe_job.remote_job_id = data["job_id"]
+    ai_transcribe_job.save()
+
+    recording.status = models.RecordingStatusChoices.NOTIFICATION_SUCCEEDED
+    recording.save()
+
+    logger.info("Transcription job created for recording %s", recording_id)
+    return ai_transcribe_job.id
+
+
+def format_transcript(  # noqa: PLR0913
+    transcription,
+    *,
+    context_language: str | None,
+    language: str,
+    room: str | None,
+    recording_datetime: datetime | None,
+    owner_timezone: str | None,
+    download_link: str | None,
+) -> tuple[str, str]:
+    """Format a transcription into readable content with a title.
+
+    Resolves the locale from context_language / language, then uses
+    TranscriptFormatter to produce markdown content and a title.
+
+    Returns a (content, title) tuple.
+    """
+    locale = get_locale(context_language, language)
+    formatter = TranscriptFormatter(locale)
+
+    return formatter.format(
+        transcription,
+        room=room,
+        recording_datetime=recording_datetime,
+        owner_timezone=owner_timezone,
+        download_link=download_link,
+    )
+
+
+@task
+def handle_transcript_received(remote_job_id, url):
+    """
+    Store the transcript and call the summarize service for a given recording.
+    """
+    ai_transcript_job = AiRecordingJob.objects.filter(
+        remote_job_id=remote_job_id, type=AiJobTypeChoices.TRANSCRIPT
+    ).first()
+    if not ai_transcript_job:
+        logger.warning("No AI recording job found for job ID: %s", remote_job_id)
+        return
+
+    user = ai_transcript_job.user
+    recording = ai_transcript_job.recording
+
+    response = requests.get(url, timeout=(10, 20))
+    response.raise_for_status()
+    transcript = WhisperXResponse(**response.json())
+
+    # Format output
+    content, title = format_transcript(
+        transcript,
+        context_language=user.language,
+        language=ai_transcript_job.language,
+        room=recording.room.name,
+        recording_datetime=recording.started_at or recording.created_at,
+        owner_timezone=user.timezone,
+        download_link=urljoin(settings.RECORDING_DOWNLOAD_BASE_URL, recording.id),
+    )
+
+    create_document_in_docs(
+        title=title, content=content, email=user.email, sub=user.sub
+    )
+
+    ai_transcript_job.status = AiJobStatusChoices.SUCCESS
+    ai_transcript_job.save()
+
+    analytics.capture_event(
+        analytics.EventName.TRANSCRIPT_GENERATION_SUCCESS,
+        user=ai_transcript_job.user,
+        properties={
+            "generation_time_seconds": (
+                timezone.now() - ai_transcript_job.created_at
+            ).total_seconds(),
+            "ai_recording_job_id": ai_transcript_job.id,
+            "language": ai_transcript_job.language,
+            "recording_id": ai_transcript_job.recording.id,
+            "transcript_size": len(response.content),
+        },
+    )
+
+    # LLM Summarization
+    if analytics.is_feature_enabled("summary-enabled", distinct_id=user.sub):
+        ai_summary_job = AiRecordingJob.objects.create(
+            remote_job_id=None,
+            file=recording,
+            type=AiJobTypeChoices.SUMMARIZE,
+            status=AiJobStatusChoices.PENDING,
+            language=ai_transcript_job.language,
+        )
+
+        try:
+            summary_response = requests.post(
+                settings.AI_SERVICE_URL + "async-jobs/summarize/",
+                json={
+                    "user_sub": ai_summary_job.user.sub,
+                    "language": ai_transcript_job.language,
+                    "content": content,
+                },
+                headers={
+                    "Authorization": f"Bearer {settings.AI_SERVICE_API_KEY}",
+                },
+                timeout=10,
+            )
+            summary_response.raise_for_status()
+        except Exception as e:
+            logger.error(
+                "Creating summary job failed for recording %s: %s", recording.id, e
+            )
+            ai_summary_job.status = AiJobStatusChoices.FAILED
+            ai_summary_job.save()
+            raise e
+
+        ai_summary_job.remote_job_id = summary_response.json()["job_id"]
+        ai_summary_job.save()
+
+        logger.info("Summary job created for recording %s", recording.id)
+
+
+@task
+def handle_summary_received(remote_job_id, url):
+    """
+    Store the summary of a given file.
+    """
+    ai_summary_job = AiRecordingJob.objects.filter(
+        remote_job_id=remote_job_id, type=AiJobTypeChoices.SUMMARIZE
+    ).first()
+    if not ai_summary_job:
+        logger.warning("No AI file job found for job ID: %s", remote_job_id)
+        return
+
+    recording = ai_summary_job.recording
+
+    logger.info("Storing summary for recording %s & url %s", recording.id, url)
+    response = requests.get(url, timeout=(10, 20))
+    response.raise_for_status()
+
+    user = ai_summary_job.user
+
+    # We dynamically recompute the title for the document since we don't have access to the transcript
+    _, title = format_transcript(
+        None,
+        context_language=user.language,
+        language=ai_summary_job.language,
+        room=recording.room.name,
+        recording_datetime=recording.started_at or recording.created_at,
+        owner_timezone=user.timezone,
+        download_link=urljoin(settings.RECORDING_DOWNLOAD_BASE_URL, recording.id),
+    )
+
+    create_document_in_docs(
+        title=get_locale(user.language).summary_title_template.format(title=title),
+        content=response.text,
+        email=user.email,
+        sub=user.sub,
+    )
+
+    logger.info("Summary created in docs for recording %s & url %s", recording.id, url)
+    ai_summary_job.status = AiJobStatusChoices.SUCCESS
+    ai_summary_job.save()
+
+    analytics.capture_event(
+        analytics.EventName.TRANSCRIPT_GENERATION_SUCCESS,
+        user=ai_summary_job.user,
+        properties={
+            "generation_time_seconds": (
+                timezone.now() - ai_summary_job.created_at
+            ).total_seconds(),
+            "ai_recording_job_id": ai_summary_job.id,
+            "language": ai_summary_job.language,
+            "recording_id": ai_summary_job.recording.id,
+            "transcript_size": len(response.content),
+        },
+    )
+
+
+def create_document_in_docs(*, title: str, content: str, email: str, sub: str) -> str:
+    """
+    Create a document in Docs for a given file.
+    """
+
+    response = requests.post(
+        urljoin(settings.DOCS_BASE_URL, "/api/v1.0/documents/create-for-owner/"),
+        json={
+            "title": title,
+            "content": content,
+            "email": email,
+            "sub": sub,
+        },
+        headers={
+            "Authorization": f"Bearer {settings.DOCS_SERVER_TO_SERVER_API_KEY}",
+        },
+        timeout=20,
+    )
+
+    if response.status_code != 201:
+        logger.error(
+            "Failed to create document in Docs %s",
+            title,
+        )
+        response.raise_for_status()
+
+    docs_app_id = response.json()["id"]
+    logger.info(
+        "Document created in Docs => %s (in docs)",
+        docs_app_id,
+    )
+    return docs_app_id

src/backend/core/transcription/locales/__init__.py

@@ -2,9 +2,10 @@
 
 from typing import Optional
 
-from summary.core.config import get_settings
-from summary.core.locales import de, en, fr, nl
-from summary.core.locales.strings import LocaleStrings
+from django.conf import settings
+
+from core.transcription.locales import de, en, fr, nl
+from core.transcription.locales.strings import LocaleStrings
 
 _LOCALES = {"fr": fr, "en": en, "de": de, "nl": nl}
 
@@ -27,4 +28,4 @@ def get_locale(*languages: Optional[str]) -> LocaleStrings:
         if base_lang in _LOCALES:
             return _LOCALES[base_lang].STRINGS
 
-    return _LOCALES[get_settings().default_context_language].STRINGS
+    return _LOCALES[settings.TRANSCRIPTION_DEFAULT_LANGUAGE].STRINGS

src/backend/core/transcription/locales/de.py

@@ -1,6 +1,6 @@
 """German locale strings."""
 
-from summary.core.locales.strings import LocaleStrings
+from core.transcription.locales.strings import LocaleStrings
 
 STRINGS = LocaleStrings(
     empty_transcription="""
@@ -30,4 +30,5 @@ Einige Punkte, die wir Ihnen empfehlen zu überprüfen:
     document_title_template=(
         'Besprechung "{room}" am {room_recording_date} um {room_recording_time}'
     ),
+    summary_title_template="Zusammenfassung von {title}",
 )

src/backend/core/transcription/locales/en.py

@@ -1,6 +1,6 @@
 """English locale strings."""
 
-from summary.core.locales.strings import LocaleStrings
+from core.transcription.locales.strings import LocaleStrings
 
 STRINGS = LocaleStrings(
     empty_transcription="""
@@ -30,4 +30,5 @@ A few things we recommend you check:
     document_title_template=(
         'Meeting "{room}" on {room_recording_date} at {room_recording_time}'
     ),
+    summary_title_template="Summary of {title}",
 )

src/backend/core/transcription/locales/fr.py

@@ -1,6 +1,6 @@
 """French locale strings (default)."""
 
-from summary.core.locales.strings import LocaleStrings
+from core.transcription.locales.strings import LocaleStrings
 
 STRINGS = LocaleStrings(
     empty_transcription="""
@@ -30,4 +30,5 @@ Quelques points que nous vous conseillons de vérifier :
     document_title_template=(
         'Réunion "{room}" du {room_recording_date} à {room_recording_time}'
     ),
+    summary_title_template="Résumé de {title}",
 )

src/backend/core/transcription/locales/nl.py

@@ -1,6 +1,6 @@
 """Dutch locale strings."""
 
-from summary.core.locales.strings import LocaleStrings
+from core.transcription.locales.strings import LocaleStrings
 
 STRINGS = LocaleStrings(
     empty_transcription="""
@@ -30,4 +30,5 @@ Een paar punten die wij u aanraden te controleren:
     document_title_template=(
         'Vergadering "{room}" op {room_recording_date} om {room_recording_time}'
     ),
+    summary_title_template="Samenvatting van {title}",
 )

src/backend/core/transcription/locales/strings.py

@@ -13,3 +13,4 @@ class LocaleStrings:
     hallucination_replacement_text: str
     document_default_title: str
     document_title_template: str
+    summary_title_template: str

src/backend/core/transcription/transcript_formatter.py

@@ -5,10 +5,9 @@ from datetime import datetime
 from typing import Tuple
 from zoneinfo import ZoneInfo
 
-from summary.core.config import get_settings
-from summary.core.locales import LocaleStrings
+from django.conf import settings
 
-settings = get_settings()
+from core.transcription.locales.strings import LocaleStrings
 
 logger = logging.getLogger(__name__)
 
@@ -38,11 +37,11 @@ class TranscriptFormatter:
 
         return None
 
-    def format(
+    def format(  # pylint: disable=too-many-arguments,too-many-positional-arguments
         self,
         transcription,
         room: str | None = None,
-        recording_datetime: str | None = None,
+        recording_datetime: datetime | None = None,
         owner_timezone: str | None = None,
         download_link: str | None = None,
     ) -> Tuple[str, str]:
@@ -100,14 +99,14 @@ class TranscriptFormatter:
     def _generate_title(
         self,
         room: str | None = None,
-        recording_datetime: str | None = None,
+        recording_datetime: datetime | None = None,
         owner_timezone: str | None = None,
     ) -> str:
         """Generate title from context or return default."""
         if not room or not recording_datetime:
             return self._locale.document_default_title
 
-        dt = datetime.fromisoformat(recording_datetime)
+        dt = recording_datetime
         if owner_timezone:
             dt = dt.astimezone(ZoneInfo(owner_timezone))
 

src/backend/core/transcription/webhook_schemas.py

@@ -0,0 +1,163 @@
+"""Transcribe / summary Shared / Webhook models."""
+
+from typing import Annotated, Literal, Union
+
+from pydantic import BaseModel, Field, TypeAdapter
+
+
+class WordSegment(BaseModel):
+    """Word segment model for transcription tasks."""
+
+    word: str = Field(title="Word")
+    start: float | None = Field(
+        default=None, title="Start Time", description="Start time in seconds."
+    )
+    end: float | None = Field(
+        default=None, title="End Time", description="End time in seconds."
+    )
+    score: float | None = Field(
+        default=None,
+        title="Confidence Score",
+        description="Confidence score for the word segment.",
+    )
+    speaker: str | None = Field(
+        default=None,
+        title="Speaker",
+        description="Speaker identifier for the word segment.",
+    )
+
+
+class Segment(BaseModel):
+    """Segment model for transcription tasks."""
+
+    start: float | None = Field(
+        default=None, title="Start Time", description="Start time in seconds."
+    )
+    end: float | None = Field(
+        default=None, title="End Time", description="End time in seconds."
+    )
+    text: str = Field(
+        title="Segment Text", description="Transcribed text for the segment."
+    )
+    words: tuple[WordSegment, ...] | None = Field(
+        title="Word Segments", description="List of word segments within the segment."
+    )
+    speaker: str | None = Field(
+        default=None, title="Speaker", description="Speaker identifier for the segment."
+    )
+
+
+class WhisperXResponse(BaseModel):
+    """Model for WhisperX response."""
+
+    segments: tuple[Segment, ...] = Field(
+        title="Segments", description="List of transcribed segments."
+    )
+    word_segments: tuple[WordSegment, ...] = Field(
+        title="Word Segments", description="List of word segments."
+    )
+
+
+class BaseWebhook(BaseModel):
+    """Base webhook payload."""
+
+    job_id: str = Field(
+        title="Job ID",
+        description="The ID of the job document in the receiver system.",
+    )
+
+
+class TranscribeWebhookSuccessPayload(BaseWebhook):
+    """Payload for a successful transcription webhook."""
+
+    type: Literal["transcript"] = Field(default="transcript")
+    status: Literal["success"] = Field(default="success")
+    transcription_data_url: str = Field(
+        title="Transcript", description="URL to the raw transcription data."
+    )
+
+
+class TranscribeWebhookPendingPayload(BaseWebhook):
+    """Payload for a pending transcription webhook-like response."""
+
+    type: Literal["transcript"] = Field(default="transcript")
+    status: Literal["pending"] = Field(default="pending")
+
+
+class TranscribeWebhookFailurePayload(BaseWebhook):
+    """Payload for a failed transcription webhook."""
+
+    type: Literal["transcript"] = Field(default="transcript")
+    status: Literal["failure"] = Field(default="failure")
+    error_code: Literal["unknown_error"] = Field(
+        title="Error code", description="The error code."
+    )
+
+
+TranscribeWebhookPayloads = Annotated[
+    Union[
+        TranscribeWebhookSuccessPayload,
+        TranscribeWebhookPendingPayload,
+        TranscribeWebhookFailurePayload,
+    ],
+    Field(discriminator="status"),
+]
+
+
+class SummarizeWebhookSuccessPayload(BaseWebhook):
+    """Payload for a successful summarization webhook."""
+
+    type: Literal["summary"] = Field(default="summary")
+    status: Literal["success"] = Field(default="success")
+    summary_data_url: str = Field(
+        title="Summary", description="URL to the raw summary data."
+    )
+
+
+class SummarizeWebhookPendingPayload(BaseWebhook):
+    """Payload for a pending summarization webhook-like response."""
+
+    type: Literal["summary"] = Field(default="summary")
+    status: Literal["pending"] = Field(default="pending")
+
+
+class SummarizeWebhookFailurePayload(BaseWebhook):
+    """Payload for a failed summarization webhook."""
+
+    type: Literal["summary"] = Field(default="summary")
+    status: Literal["failure"] = Field(default="failure")
+    error_code: Literal["unknown_error"] = Field(
+        title="Error code", description="The error code."
+    )
+
+
+SummarizeWebhookPayloads = Annotated[
+    Union[
+        SummarizeWebhookSuccessPayload,
+        SummarizeWebhookPendingPayload,
+        SummarizeWebhookFailurePayload,
+    ],
+    Field(discriminator="status"),
+]
+
+WebhookPayloads = Annotated[
+    Union[TranscribeWebhookPayloads, SummarizeWebhookPayloads],
+    Field(discriminator="type"),
+]
+
+
+webhook_payload_adapter = TypeAdapter(WebhookPayloads)
+
+__all__ = [
+    "TranscribeWebhookSuccessPayload",
+    "TranscribeWebhookPendingPayload",
+    "TranscribeWebhookFailurePayload",
+    "SummarizeWebhookSuccessPayload",
+    "SummarizeWebhookPendingPayload",
+    "SummarizeWebhookFailurePayload",
+    "TranscribeWebhookPayloads",
+    "SummarizeWebhookPayloads",
+    "WebhookPayloads",
+    "WhisperXResponse",
+    "webhook_payload_adapter",
+]

src/backend/core/urls.py

@@ -16,6 +16,7 @@ router.register("users", viewsets.UserViewSet, basename="users")
 router.register("rooms", viewsets.RoomViewSet, basename="rooms")
 router.register("recordings", viewsets.RecordingViewSet, basename="recordings")
 router.register("files", viewsets.FileViewSet, basename="files")
+router.register("ai-jobs", viewsets.AiJobViewSet, basename="ai-jobs")
 router.register(
     "resource-accesses", viewsets.ResourceAccessViewSet, basename="resource_accesses"
 )

src/backend/core/utils.py

@@ -455,3 +455,38 @@ def generate_upload_policy(file):
     )
 
     return policy
+
+
+def generate_download_s3_file_url(
+    key, *, expires_in: int, override_domain: bool = True
+):
+    """
+    Generate a S3 signed download url for a given key.
+    """
+
+    # This settings should be used if the backend application and the frontend application
+    # can't connect to the object storage with the same domain. This is the case in the
+    # docker compose stack used in development. The frontend application will use localhost
+    # to connect to the object storage while the backend application will use the object storage
+    # service name declared in the docker compose stack.
+    # This is needed because the domain name is used to compute the signature. So it can't be
+    # changed dynamically by the frontend application.
+    if settings.AWS_S3_DOMAIN_REPLACE and override_domain:
+        s3_client = boto3.client(
+            "s3",
+            aws_access_key_id=settings.AWS_S3_ACCESS_KEY_ID,
+            aws_secret_access_key=settings.AWS_S3_SECRET_ACCESS_KEY,
+            endpoint_url=settings.AWS_S3_DOMAIN_REPLACE,
+            config=botocore.client.Config(
+                region_name=settings.AWS_S3_REGION_NAME,
+                signature_version=settings.AWS_S3_SIGNATURE_VERSION,
+            ),
+        )
+    else:
+        s3_client = default_storage.connection.meta.client
+
+    return s3_client.generate_presigned_url(
+        ClientMethod="get_object",
+        Params={"Bucket": default_storage.bucket_name, "Key": key},
+        ExpiresIn=expires_in,
+    )

src/backend/meet/settings.py

@@ -19,6 +19,7 @@ from socket import gethostbyname, gethostname
 from django.utils.translation import gettext_lazy as _
 
 import dj_database_url
+import posthog
 import sentry_sdk
 from configurations import Configuration, values
 from lasuite.configuration.values import SecretFileValue
@@ -451,6 +452,11 @@ class Base(Configuration):
     CELERY_BROKER_URL = values.Value("redis://redis:6379/0", environ_prefix=None)
     CELERY_BROKER_TRANSPORT_OPTIONS = values.DictValue({}, environ_prefix=None)
 
+    # Analytics
+    POSTHOG_ENABLED = values.BooleanValue(False, environ_prefix=None)
+    POSTHOG_API_KEY = values.Value(None, environ_prefix=None)
+    POSTHOG_API_HOST = values.Value(None, environ_prefix=None)
+
     # Session
     SESSION_ENGINE = values.Value(
         default="django.contrib.sessions.backends.cache",
@@ -744,6 +750,20 @@ class Base(Configuration):
     SUMMARY_SERVICE_API_TOKEN = SecretFileValue(
         None, environ_name="SUMMARY_SERVICE_API_TOKEN", environ_prefix=None
     )
+    DOCS_BASE_URL = values.Value(
+        "https://example.com",
+        environ_name="DOCS_BASE_URL",
+        environ_prefix=None,
+    )
+    DOCS_SERVER_TO_SERVER_API_KEY = SecretFileValue(
+        None,
+        environ_name="DOCS_SERVER_TO_SERVER_API_KEY",
+        environ_prefix=None,
+    )
+    TRANSCRIPTION_DEFAULT_LANGUAGE = values.Value(
+        default="fr", environ_name="TRANSCRIPTION_DEFAULT_LANGUAGE", environ_prefix=None
+    )
+
     SCREEN_RECORDING_BASE_URL = values.Value(
         None, environ_name="SCREEN_RECORDING_BASE_URL", environ_prefix=None
     )

src/backend/pyproject.toml

@@ -62,6 +62,7 @@ dependencies = [
     "livekit-api==1.1.0",
     "aiohttp==3.13.4",
     "urllib3==2.7.0",
+    "posthog>=7.14.2",
 ]
 
 [project.urls]

src/backend/uv.lock

@@ -148,6 +148,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
 ]
 
+[[package]]
+name = "backoff"
+version = "2.2.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/47/d7/5bbeb12c44d7c4f2fb5b56abce497eb5ed9f34d85701de869acedd602619/backoff-2.2.1.tar.gz", hash = "sha256:03f829f5bb1923180821643f8753b0502c3b682293992485b0eef2807afa5cba", size = 17001, upload-time = "2022-10-05T19:19:32.061Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" },
+]
+
 [[package]]
 name = "billiard"
 version = "4.2.4"
@@ -559,6 +568,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/1e/77/dc8c558f7593132cf8fefec57c4f60c83b16941c574ac5f619abb3ae7933/dill-0.4.1-py3-none-any.whl", hash = "sha256:1e1ce33e978ae97fcfcff5638477032b801c46c7c65cf717f95fbc2248f79a9d", size = 120019, upload-time = "2026-01-19T02:36:55.663Z" },
 ]
 
+[[package]]
+name = "distro"
+version = "1.9.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/fc/f8/98eea607f65de6527f8a2e8885fc8015d3e6f5775df186e443e0964a11c3/distro-1.9.0.tar.gz", hash = "sha256:2fa77c6fd8940f116ee1d6b94a2f90b13b5ea8d019b98bc8bafdcabcdd9bdbed", size = 60722, upload-time = "2023-12-24T09:54:32.31Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" },
+]
+
 [[package]]
 name = "dj-database-url"
 version = "3.1.2"
@@ -1204,6 +1222,7 @@ dependencies = [
     { name = "markdown" },
     { name = "mozilla-django-oidc" },
     { name = "nested-multipart-parser" },
+    { name = "posthog" },
     { name = "psycopg", extra = ["binary"] },
     { name = "pydantic" },
     { name = "pyjwt" },
@@ -1266,6 +1285,7 @@ requires-dist = [
     { name = "markdown", specifier = "==3.10.2" },
     { name = "mozilla-django-oidc", specifier = "==5.0.2" },
     { name = "nested-multipart-parser", specifier = "==1.6.0" },
+    { name = "posthog", specifier = ">=7.14.2" },
     { name = "psycopg", extras = ["binary"], specifier = "==3.3.3" },
     { name = "pydantic", specifier = "==2.12.5" },
     { name = "pyjwt", specifier = "==2.12.1" },
@@ -1506,6 +1526,21 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
 ]
 
+[[package]]
+name = "posthog"
+version = "7.14.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "backoff" },
+    { name = "distro" },
+    { name = "requests" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/ca/e6/fe25f9eaeb5b4b66aa738554ba1ef9feece8b1d5b6a9ea431782b6c6c58f/posthog-7.14.2.tar.gz", hash = "sha256:b913dc23acc301a95ca9b851c193b261932d01a66a9af91eb6e9883cd05d5b6b", size = 205633, upload-time = "2026-05-13T16:36:27.153Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c5/b8/75f9eed446c1d48405871542b36fdf1dba3333756a477bcb0d9ef70d17a4/posthog-7.14.2-py3-none-any.whl", hash = "sha256:f78b45a5ad5c72e55bf1cadebe8cf46bae2a6094e7fe01cbc7b5ec0531aab4d8", size = 240750, upload-time = "2026-05-13T16:36:25.544Z" },
+]
+
 [[package]]
 name = "pprintpp"
 version = "0.4.0"

src/summary/summary/api/main.py

@@ -2,11 +2,8 @@
 
 from fastapi import APIRouter, Depends
 
-from summary.api.route import tasks, tasks_v2
+from summary.api.route import tasks_v2
 from summary.core.security import verify_tenant_api_key
 
-api_router_v1 = APIRouter(dependencies=[Depends(verify_tenant_api_key)])
-api_router_v1.include_router(tasks.router_tasks_v1, tags=["tasks"])
-
 api_router_v2 = APIRouter(dependencies=[Depends(verify_tenant_api_key)])
 api_router_v2.include_router(tasks_v2.router_tasks_v2, tags=["tasks"])

src/summary/summary/api/route/tasks.py

@@ -1,79 +0,0 @@
-"""API routes related to application tasks."""
-
-import time
-from typing import Optional
-
-from celery.result import AsyncResult
-from fastapi import APIRouter
-from pydantic import BaseModel, field_validator
-
-from summary.core.celery_worker import (
-    process_audio_transcribe_summarize_v2,
-)
-from summary.core.config import get_settings
-
-settings = get_settings()
-
-
-class TranscribeSummarizeTaskCreation(BaseModel):
-    """Transcription and summarization parameters."""
-
-    owner_id: str
-    recording_filename: str
-    metadata_filename: Optional[str] = None
-    email: str
-    sub: str
-    version: Optional[int] = 2
-    room: Optional[str]
-    owner_timezone: Optional[str]
-    language: Optional[str]
-    download_link: Optional[str]
-    context_language: Optional[str] = None
-    recording_start_at: Optional[str] = None
-    recording_end_at: Optional[str] = None
-
-    @field_validator("language")
-    @classmethod
-    def validate_language(cls, v):
-        """Validate 'language' parameter."""
-        if v is not None and v not in settings.whisperx_allowed_languages:
-            raise ValueError(
-                f"Language '{v}' is not allowed. "
-                f"Allowed languages: {', '.join(settings.whisperx_allowed_languages)}"
-            )
-        return v
-
-
-router_tasks_v1 = APIRouter(prefix="/tasks")
-
-
-@router_tasks_v1.post("/")
-async def create_transcribe_summarize_task(request: TranscribeSummarizeTaskCreation):
-    """Create a transcription and summarization task."""
-    task = process_audio_transcribe_summarize_v2.apply_async(
-        args=[
-            request.owner_id,
-            request.recording_filename,
-            request.metadata_filename,
-            request.email,
-            request.sub,
-            time.time(),
-            request.room,
-            request.owner_timezone,
-            request.language,
-            request.download_link,
-            request.context_language,
-            request.recording_start_at,
-            request.recording_end_at,
-        ],
-        queue=settings.transcribe_queue,
-    )
-
-    return {"id": task.id, "message": "Task created"}
-
-
-@router_tasks_v1.get("/{task_id}")
-async def get_task_status(task_id: str):
-    """Check task status by ID."""
-    task = AsyncResult(task_id)
-    return {"id": task_id, "status": task.status}

src/summary/summary/core/celery_worker.py

@@ -4,7 +4,6 @@
 
 import json
 import time
-from datetime import datetime
 
 import openai
 import sentry_sdk
@@ -16,8 +15,8 @@ from summary.core.analytics import MetadataManager, get_analytics
 from summary.core.config import get_settings
 from summary.core.file_service import FileService, FileServiceException
 from summary.core.llm_service import LLMException, LLMObservability, LLMService
-from summary.core.locales import get_locale
 from summary.core.models import (
+    RecordingMetadata,
     SummarizeTaskV2Payload,
     TranscribeTaskV2Payload,
 )
@@ -39,11 +38,9 @@ from summary.core.shared_models import (
     WhisperXResponse,
     webhook_payload_adapter,
 )
-from summary.core.transcript_formatter import TranscriptFormatter
 from summary.core.user_assign import resolve_speaker_identities
 from summary.core.webhook_service import (
     call_webhook_v2,
-    submit_content,
 )
 
 settings = get_settings()
@@ -79,23 +76,17 @@ file_service = FileService()
 def transcribe_audio(
     *,
     task_id: str,
-    recording_filename: str | None = None,
     language: str,
-    cloud_storage_url=None,
+    cloud_storage_url: str,
     raises: bool = False,
 ):
     """Transcribe an audio file using WhisperX.
 
-    Downloads the audio from MinIO or a cloud storage URL, sends it to
+    Downloads the audio from a cloud storage URL, sends it to
     WhisperX for transcription, and tracks metadata throughout the process.
 
     Returns the transcription object, or None if the file could not be retrieved.
     """
-    if bool(recording_filename) == bool(cloud_storage_url):
-        raise ValueError(
-            "Either filename or cloud_storage_url must be provided, but not both."
-        )
-
     logger.info("Initiating WhisperX client")
     whisperx_client = openai.OpenAI(
         api_key=settings.whisperx_api_key.get_secret_value(),
@@ -106,7 +97,6 @@ def transcribe_audio(
     # Transcription
     try:
         with file_service.prepare_audio_file(
-            remote_object_key=recording_filename,
             cloud_storage_url=cloud_storage_url,
         ) as (audio_file, metadata):
             metadata_manager.track(task_id, {"audio_length": metadata["duration"]})
@@ -150,10 +140,8 @@ def transcribe_audio(
         )
         logger.exception(
             (
-                "Unexpected error while preparing file | filename: %s "
-                "| cloud_storage_url: %s"
+                "Unexpected error while preparing file %s "
             ),
-            recording_filename,
             redacted_cloud_storage_url,
         )
         return None
@@ -163,41 +151,31 @@ def transcribe_audio(
 
 
 def resolve_speaker_identities_and_apply_to(
-    transcription, recording_start_at, recording_end_at, metadata_filename, task_id
-):
+    *, transcription: WhisperXResponse, recording_metadata: RecordingMetadata, task_id
+) -> WhisperXResponse:
     """Assign users to detected speakers and rewrite the transcriptions.
 
     Args:
         transcription: output of meet-whisperx after transcription and diarization
-        recording_start_at: sourced from LiveKit FileInfo via the egress_ended webhook
-        recording_end_at: sourced from LiveKit FileInfo via the egress_ended webhook
-        metadata_filename: name of metadata file containing VAD information in S3
+        recording_metadata: Metadata of the recording
         task_id: current task id, for logging purposes
     """
-    recording_start_dt = (
-        datetime.fromisoformat(recording_start_at) if recording_start_at else None
-    )
-    recording_end_dt = (
-        datetime.fromisoformat(recording_end_at) if recording_end_at else None
-    )
-
     logger.debug(
         "recording_start_dt: %s ; recording_end_dt: %s",
-        recording_start_dt,
-        recording_end_dt,
+        recording_metadata.start_at,
+        recording_metadata.end_at,
     )
-    if (recording_start_dt is None) or (recording_end_dt is None):
-        logger.debug("Skipping resolve_speaker_identities")
-        return transcription
 
     logger.debug("Running resolve_speaker_identities")
     try:
-        metadata = file_service.read_json(metadata_filename)
+        metadata = file_service.read_cloud_storage_json(
+            recording_metadata.cloud_storage_url
+        )
         speaker_mapping = resolve_speaker_identities(
             metadata,
             transcription,
-            recording_start_dt,
-            recording_end_dt,
+            recording_metadata.start_at,
+            recording_metadata.end_at,
         )
         new_transcription = speaker_mapping.apply_to(transcription.model_dump())
         return new_transcription
@@ -221,34 +199,6 @@ def resolve_speaker_identities_and_apply_to(
         return transcription
 
 
-def format_transcript(
-    transcription,
-    context_language: str | None,
-    language: str,
-    room: str | None,
-    recording_datetime: str | None,
-    owner_timezone: str | None,
-    download_link: str | None,
-) -> tuple[str, str]:
-    """Format a transcription into readable content with a title.
-
-    Resolves the locale from context_language / language, then uses
-    TranscriptFormatter to produce markdown content and a title.
-
-    Returns a (content, title) tuple.
-    """
-    locale = get_locale(context_language, language)
-    formatter = TranscriptFormatter(locale)
-
-    return formatter.format(
-        transcription,
-        room=room,
-        recording_datetime=recording_datetime,
-        owner_timezone=owner_timezone,
-        download_link=download_link,
-    )
-
-
 def format_actions(llm_output: dict) -> str:
     """Format the actions from the LLM output into a markdown list.
 
@@ -267,126 +217,23 @@ def format_actions(llm_output: dict) -> str:
     return ""
 
 
-@celery.task(
-    bind=True,
-    autoretry_for=[exceptions.HTTPError],
-    max_retries=settings.celery_max_retries,
-    queue=settings.transcribe_queue,
-)
-def process_audio_transcribe_summarize_v2(
-    self,
-    owner_id: str,
-    recording_filename: str,
-    metadata_filename: str | None,
-    email: str,
-    sub: str,
-    received_at: float,
-    room: str | None,
-    owner_timezone: str | None,
-    language: str | None,
-    download_link: str | None,
-    context_language: str | None = None,
-    recording_start_at: str | None = None,
-    recording_end_at: str | None = None,
-):
-    """Process an audio file by transcribing it and generating a summary.
-
-    This Celery task orchestrates:
-    1. Audio transcription via WhisperX
-    2. Transcript formatting
-    3. Webhook submission
-    4. Conditional summarization queuing
-
-    Args:
-        self: Celery task instance (passed on with bind=True)
-        owner_id: Unique identifier of the recording owner.
-        recording_filename: Name of the audio file in MinIO storage.
-        metadata_filename: Name of the audio file in MinIO storage.
-        email: Email address of the recording owner.
-        sub: OIDC subject identifier of the recording owner.
-        received_at: Unix timestamp when the recording was received.
-        room: room name where the recording took place.
-        owner_timezone: IANA timezone of the recording owner (e.g. "Europe/Paris").
-        language: ISO 639-1 language code for transcription.
-        download_link: URL to download the original recording.
-        context_language: ISO 639-1 language code of the meeting summary context text.
-        recording_start_at: ISO 8601 timestamp of when file recording actually started
-            (from LiveKit FileInfo.started_at via the egress_ended webhook).
-        recording_end_at: ISO 8601 timestamp of when file recording ended
-            (from LiveKit FileInfo.ended_at via the egress_ended webhook).
-    """
-    logger.info(
-        "Notification received | Owner: %s | Room: %s",
-        owner_id,
-        room,
-    )
-
-    task_id = self.request.id
-
-    # Transcribe the audio
-    transcription = transcribe_audio(
-        task_id=task_id, recording_filename=recording_filename, language=language
-    )
-    if transcription is None:
-        return
-
-    # Assign speakers and rewrite transcription/diarization output
-    if settings.is_resolve_speaker_identities_enabled and (
-        metadata_filename is not None
-    ):
-        transcription = resolve_speaker_identities_and_apply_to(
-            transcription,
-            recording_start_at,
-            recording_end_at,
-            metadata_filename,
-            task_id,
-        )
-
-    # Format output
-    content, title = format_transcript(
-        transcription,
-        context_language,
-        language,
-        room,
-        recording_start_at,
-        owner_timezone,
-        download_link,
-    )
-
-    submit_content(content, title, email, sub)
-    metadata_manager.capture(task_id, settings.posthog_event_success)
-
-    # LLM Summarization
-    if (
-        analytics.is_feature_enabled("summary-enabled", distinct_id=owner_id)
-        and settings.is_summary_enabled
-    ):
-        logger.info("Queuing summary generation task.")
-        summarize_transcription.apply_async(
-            args=[owner_id, content, email, sub, title],
-            queue=settings.summarize_queue,
-        )
-    else:
-        logger.info("Summary generation not enabled for this user. Skipping.")
-
-
-@signals.task_prerun.connect(sender=process_audio_transcribe_summarize_v2)
-def task_started(task_id=None, task=None, args=None, **kwargs):
-    """Signal handler called before task execution begins."""
-    task_args = args or []
-    metadata_manager.create(task_id, task_args)
-
-
-@signals.task_retry.connect(sender=process_audio_transcribe_summarize_v2)
-def task_retry_handler(request=None, reason=None, einfo=None, **kwargs):
-    """Signal handler called when task execution retries."""
-    metadata_manager.retry(request.id)
-
-
-@signals.task_failure.connect(sender=process_audio_transcribe_summarize_v2)
-def task_failure_handler(task_id, exception=None, **kwargs):
-    """Signal handler called when task execution fails permanently."""
-    metadata_manager.capture(task_id, settings.posthog_event_failure)
+# @signals.task_prerun.connect(sender=process_audio_transcribe_summarize_v2)
+# def task_started(task_id=None, task=None, args=None, **kwargs):
+#     """Signal handler called before task execution begins."""
+#     task_args = args or []
+#     metadata_manager.create(task_id, task_args)
+#
+#
+# @signals.task_retry.connect(sender=process_audio_transcribe_summarize_v2)
+# def task_retry_handler(request=None, reason=None, einfo=None, **kwargs):
+#     """Signal handler called when task execution retries."""
+#     metadata_manager.retry(request.id)
+#
+#
+# @signals.task_failure.connect(sender=process_audio_transcribe_summarize_v2)
+# def task_failure_handler(task_id, exception=None, **kwargs):
+#     """Signal handler called when task execution fails permanently."""
+#     metadata_manager.capture(task_id, settings.posthog_event_failure)
 
 
 def summarize_transcription_internals(
@@ -468,29 +315,6 @@ def summarize_transcription_internals(
     return summary
 
 
-@celery.task(
-    bind=True,
-    autoretry_for=[LLMException, Exception],
-    max_retries=settings.celery_max_retries,
-    queue=settings.summarize_queue,
-)
-def summarize_transcription(
-    self, owner_id: str, transcript: str, email: str, sub: str, title: str
-):
-    """Generate a summary from the provided transcription text.
-
-    This Celery task performs the following operations:
-    1. Run summary internals
-    2. Sends the final summary via webhook.
-    """
-    summary = summarize_transcription_internals(
-        owner_id=owner_id, transcript=transcript, session_id=self.request.id
-    )
-    summary_title = settings.summary_title_template.format(title=title)
-
-    submit_content(summary, summary_title, email, sub)
-
-
 ##################################################################################
 # Tasks v2
 ##################################################################################
@@ -549,6 +373,17 @@ def process_audio_transcribe_v2_task(
         ).model_dump()
     )
 
+    # Assign speakers and rewrite transcription/diarization output
+    if settings.is_resolve_speaker_identities_enabled and payload.metadata is not None:
+        try:
+            transcription_res = resolve_speaker_identities_and_apply_to(
+                transcription=transcription_res,
+                recording_metadata=payload.metadata,
+                task_id=job_id,
+            )
+        except BaseException as e:
+            logger.error(f"Failed to resolve speaker identities, skipping: {e}")
+
     file_service.store_transcript(
         transcript=transcription_res,
         job_id=job_id,
@@ -561,6 +396,8 @@ def process_audio_transcribe_v2_task(
     call_webhook_v2_task.apply_async(
         args=[success_payload.model_dump(), payload.tenant_id]
     )
+    metadata_manager.capture(job_id, settings.posthog_event_success)
+
     return success_payload.model_dump()
 
 

src/summary/summary/core/config.py

@@ -1,9 +1,8 @@
 """Application configuration and settings."""
 
 import logging
-import os
 from functools import cached_property, lru_cache
-from typing import Annotated, Any, List, Literal, Mapping, Optional, Set
+from typing import Annotated, List, Literal, Mapping, Optional, Set
 
 from fastapi import Depends
 from pydantic import (
@@ -36,8 +35,6 @@ class AuthorizedTenant(BaseModel):
     )
 
 
-V1_DEFAULT_TENANT_ID = "__deprecated_meet_tenant__"
-
 
 class Settings(BaseSettings):
     """Configuration settings loaded from environment variables and .env file."""
@@ -45,14 +42,12 @@ class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env", frozen=True)
 
     app_name: str = "summary"
-    app_api_v1_str: str = "/api/v1"
     app_api_v2_str: str = "/api/v2"
 
     # Authorized Tenants
     # Using env variables to store authorized tenants for now
     # to avoid any other external dependency (DB)
     authorized_tenants: tuple[AuthorizedTenant, ...] = Field(default_factory=tuple)
-    v1_tenant_id: str = V1_DEFAULT_TENANT_ID
 
     # Audio recordings
     recording_max_duration: Optional[int] = None
@@ -72,8 +67,6 @@ class Settings(BaseSettings):
     celery_result_backend: str = "redis://redis/0"
     celery_max_retries: int = 1
 
-    transcribe_queue: str = "transcribe-queue"
-    summarize_queue: str = "summarize-queue"
     # v2 tasks
     transcribe_queue_v2: str = "transcribe-queue-v2"
     summarize_queue_v2: str = "summarize-queue-v2"
@@ -114,9 +107,6 @@ class Settings(BaseSettings):
     webhook_status_forcelist: List[int] = [502, 503, 504]
     webhook_backoff_factor: float = 0.1
 
-    # Locale
-    default_context_language: Literal["de", "en", "fr", "nl"] = "fr"
-
     # Output related settings
     summary_title_template: Optional[str] = "Résumé de {title}"
 
@@ -145,33 +135,6 @@ class Settings(BaseSettings):
     task_tracker_redis_url: str = "redis://redis/0"
     task_tracker_prefix: str = "task_metadata:"
 
-    @model_validator(mode="before")
-    @classmethod
-    def legacy_default_tenant_config(cls, data: Any) -> Any:
-        """Migrate the legacy default tenant configuration."""
-        if isinstance(data, dict):
-            api_key = os.getenv("APP_API_TOKEN")
-            webhook_api_key = os.getenv("WEBHOOK_API_TOKEN")
-            webhook_url = os.getenv("WEBHOOK_URL")
-            if api_key and webhook_api_key and webhook_url:
-                logger.warning(
-                    "Deprecated legacy app configuration detected, "
-                    "please use only the new 'authorized_tenants' field instead."
-                )
-
-                authorized_tenants = list(data.get("authorized_tenants", []))
-                authorized_tenants.append(
-                    AuthorizedTenant(
-                        id=V1_DEFAULT_TENANT_ID,
-                        api_key=SecretStr(api_key),
-                        webhook_url=webhook_url,
-                        webhook_api_key=SecretStr(webhook_api_key),
-                    )
-                )
-                data["authorized_tenants"] = tuple(authorized_tenants)
-
-        return data
-
     @model_validator(mode="after")
     def validate_authorized_tenants(self):
         """Validate authorized tenants configuration."""
@@ -190,16 +153,6 @@ class Settings(BaseSettings):
             raise ValueError("Duplicate application API api_keys are not allowed")
         return self
 
-    @model_validator(mode="after")
-    def validate_default_v1_tenant(self):
-        """Validate default v1 tenant configuration."""
-        if not any(
-            tenant.id == self.v1_tenant_id for tenant in self.authorized_tenants
-        ):
-            raise ValueError("v1 tenant is not configured in authorized tenants")
-
-        return self
-
     @cached_property
     def authorized_tenant_api_keys(self) -> frozenset[str]:
         """Return a frozenset of authorized tenant API api_keys."""

src/summary/summary/core/file_service.py

@@ -13,7 +13,6 @@ from urllib.parse import urlparse
 
 import requests
 from minio import Minio
-from minio.error import MinioException, S3Error
 
 from summary.core.config import get_settings
 from summary.core.shared_models import WhisperXResponse
@@ -144,54 +143,6 @@ class FileService:
         self._allowed_extensions = settings.recording_allowed_extensions
         self._max_duration = settings.recording_max_duration
 
-    def _download_from_minio(self, remote_object_key) -> Path:
-        """Download file from MinIO to local temporary file.
-
-        The file is downloaded to a temporary location for local manipulation
-        such as validation, conversion, or processing before being used.
-        """
-        logger.info("Download recording | object_key: %s", remote_object_key)
-
-        if not remote_object_key:
-            logger.warning("Invalid object_key '%s'", remote_object_key)
-            raise ValueError("Invalid object_key")
-
-        extension = Path(remote_object_key).suffix.lower()
-
-        if extension not in self._allowed_extensions:
-            logger.warning("Invalid file extension '%s'", extension)
-            raise ValueError(f"Invalid file extension '{extension}'")
-
-        response = None
-
-        try:
-            response = self._minio_client.get_object(
-                self._bucket_name, remote_object_key
-            )
-
-            with tempfile.NamedTemporaryFile(
-                suffix=extension, delete=False, prefix="minio_download_"
-            ) as tmp:
-                for chunk in response.stream(self._stream_chunk_size):
-                    tmp.write(chunk)
-
-                tmp.flush()
-                local_path = Path(tmp.name)
-
-                logger.info("Recording successfully downloaded")
-                logger.debug("Recording local file path: %s", local_path)
-
-                return local_path
-
-        except (MinioException, S3Error) as e:
-            raise FileServiceException(
-                "Unexpected error while downloading object."
-            ) from e
-
-        finally:
-            if response:
-                response.close()
-
     def _download_from_cloud_storage_url(self, cloud_storage_url: str) -> Path:
         """Download file from a cloud storage URL to local temporary file."""
         logger.info(
@@ -302,33 +253,19 @@ class FileService:
                 os.remove(output_path)
             raise RuntimeError("Failed to extract audio.") from e
 
-    def read_json(self, object_name: str) -> dict:
+    def read_cloud_storage_json(self, cloud_storage_url: str) -> dict:
         """Read and parse a JSON file from MinIO storage."""
-        logger.info("Reading JSON: %s", object_name)
-
-        if not object_name:
-            raise ValueError("Invalid object_name")
-
-        response = None
+        logger.info("Reading JSON: %s", cloud_storage_url)
+        local_path = self._download_from_cloud_storage_url(cloud_storage_url)
         try:
-            response = self._minio_client.get_object(self._bucket_name, object_name)
-            return json.loads(response.read())
-        except (MinioException, S3Error) as e:
-            raise FileServiceException(
-                "Unexpected error while reading JSON object."
-            ) from e
+            return json.load(local_path.open("r"))
         except (json.JSONDecodeError, UnicodeDecodeError) as e:
             raise FileServiceException("Invalid JSON content.") from e
-        finally:
-            if response:
-                response.close()
-                response.release_conn()
 
     @contextmanager
     def prepare_audio_file(
         self,
-        remote_object_key: str | None = None,
-        cloud_storage_url: str | None = None,
+        cloud_storage_url: str,
     ):
         """Download and prepare audio file for processing.
 
@@ -341,20 +278,9 @@ class FileService:
         file_handle = None
 
         try:
-            if bool(remote_object_key) == bool(cloud_storage_url):
-                raise ValueError(
-                    (
-                        "Exactly one of 'remote_object_key' or "
-                        "'cloud_storage_url' must be provided."
-                    )
-                )
-
-            if cloud_storage_url:
-                downloaded_path = self._download_from_cloud_storage_url(
-                    cloud_storage_url
-                )
-            else:
-                downloaded_path = self._download_from_minio(remote_object_key)
+            downloaded_path = self._download_from_cloud_storage_url(
+                cloud_storage_url
+            )
 
             duration = self._validate_duration(downloaded_path)
 

src/summary/summary/core/models.py

@@ -1,6 +1,6 @@
 """Models for the API & Celery tasks creation."""
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import AwareDatetime, BaseModel, Field, field_validator
 
 from summary.core.config import get_settings
 from summary.core.types import Url
@@ -14,6 +14,17 @@ class SharedV2TaskCreation(BaseModel):
     user_sub: str = Field(title="User Sub", description="The user's sub.")
 
 
+class RecordingMetadata(BaseModel):
+    """Model for recording metadata."""
+
+    cloud_storage_url: Url = Field(
+        title="Cloud Storage URL",
+        description="The URL of the metadata file for speaker assignement.",
+    )
+    start_at: AwareDatetime = Field(title="Start time of the recording to transcribe")
+    end_at: AwareDatetime = Field(title="End time of the recording to transcribe")
+
+
 class TranscribeTaskV2Request(SharedV2TaskCreation):
     """Model for creating a transcribe and summarize task (used for API request)."""
 
@@ -27,7 +38,12 @@ class TranscribeTaskV2Request(SharedV2TaskCreation):
         description="The language of the context text.",
     )
     language: str = Field(
-        title="Language", description="The language of the content to summarize."
+        title="Language", description="The language of the content to transcribe."
+    )
+    metadata: RecordingMetadata | None = Field(
+        title="Metadata",
+        description="The metadata for the transcribe task.",
+        default=None,
     )
 
     @field_validator("language")

src/summary/summary/core/shared_models.py

@@ -159,4 +159,5 @@ __all__ = [
     "SummarizeWebhookPayloads",
     "WebhookPayloads",
     "WhisperXResponse",
+    "webhook_payload_adapter",
 ]

src/summary/summary/core/webhook_service.py

@@ -46,55 +46,6 @@ def _post_with_retries(*, url, data, api_key: str | None = None):
         session.close()
 
 
-def call_webhook_v1(*, tenant_id: str, payload: dict) -> None:
-    """Call webhook with payload a payload and optional token."""
-    tenant = settings.get_authorized_tenant(tenant_id=tenant_id)
-
-    logger.debug("Submitting to %s", tenant.webhook_url)
-    logger.debug("Request payload: %s", json.dumps(payload, indent=2))
-
-    response = _post_with_retries(
-        url=tenant.webhook_url,
-        api_key=tenant.webhook_api_key.get_secret_value(),
-        data=payload,
-    )
-
-    try:
-        response_data = response.json()
-        document_id = response_data.get("id", "N/A")
-    except (json.JSONDecodeError, AttributeError):
-        document_id = "Unable to parse response"
-        response_data = response.text
-
-    logger.info(
-        "Delivery success | Document %s submitted (HTTP %s)",
-        document_id,
-        response.status_code,
-    )
-    logger.debug("Full response: %s", response_data)
-
-
-def submit_content(content: str, title: str, email: str, sub: str) -> None:
-    """Submit content to the configured webhook destination.
-
-    Builds the payload, sends it with retries, and logs the outcome.
-
-    Notes:
-        Deprecated: Use call_webhook_v2 directly instead.
-
-    Deprecated:
-        This will route content to the v1 default tenant
-    """
-    data = {
-        "title": title,
-        "content": content,
-        "email": email,
-        "sub": sub,
-    }
-
-    call_webhook_v1(payload=data, tenant_id=settings.v1_tenant_id)
-
-
 def call_webhook_v2(
     *,
     tenant_id: str,

src/summary/summary/main.py

@@ -4,7 +4,7 @@ import sentry_sdk
 from fastapi import FastAPI
 
 from summary.api import health
-from summary.api.main import api_router_v1, api_router_v2
+from summary.api.main import api_router_v2
 from summary.core.config import get_settings
 
 settings = get_settings()
@@ -17,6 +17,5 @@ app = FastAPI(
     title=settings.app_name,
 )
 
-app.include_router(api_router_v1, prefix=settings.app_api_v1_str)
 app.include_router(api_router_v2, prefix=settings.app_api_v2_str)
 app.include_router(health.router)