♻️(devexp) refactor minio webhook setup

Instead of relying on make commands to set-up the minio webhook,
use a compose service, as we did for the createbucket one.

Aligned with the dev stack, and run by default when starting
for the first time the stack.

Makefile

@@ -45,7 +45,6 @@ COMPOSE_RUN         = $(COMPOSE) run --rm
 COMPOSE_RUN_APP     = $(COMPOSE_RUN) app-dev
 COMPOSE_RUN_CROWDIN = $(COMPOSE_RUN) crowdin crowdin
 WAIT_DB             = @$(COMPOSE_RUN) dockerize -wait tcp://$(DB_HOST):$(DB_PORT) -timeout 60s
-WAIT_MINIO          = @$(COMPOSE_RUN) dockerize -wait tcp://minio:9000 -timeout 60s
 
 # -- Backend
 MANAGE              = $(COMPOSE_RUN_APP) python manage.py
@@ -54,21 +53,6 @@ MAIL_NPM            = $(COMPOSE_RUN) -w /app/src/mail node npm
 # -- Frontend
 PATH_FRONT          = ./src/frontend
 
-# -- MinIO / Webhook
-MINIO_ALIAS         ?= meet
-MINIO_ENDPOINT      ?= http://127.0.0.1:9000
-MINIO_ACCESS_KEY    ?= meet
-MINIO_SECRET_KEY    ?= password
-MINIO_BUCKET        ?= meet-media-storage
-MINIO_WEBHOOK_NAME  ?= recording
-
-MINIO_WEBHOOK_ENDPOINT ?= http://app-dev:8000/api/v1.0/recordings/storage-hook/
-
-MINIO_QUEUE_DIR     ?= /data/minio/events
-MINIO_QUEUE_LIMIT   ?= 100000
-
-STORAGE_EVENT_TOKEN ?= password
-
 # ==============================================================================
 # RULES
 
@@ -153,55 +137,6 @@ stop: ## stop the development server using Docker
 	@$(COMPOSE) stop
 .PHONY: stop
 
-# -- MinIO webhook (configuration & events)
-minio-wait: ## wait for minio to be ready
-	@echo "$(BOLD)Waiting for MinIO$(RESET)"
-	$(WAIT_MINIO)
-.PHONY: minio-wait
-
-minio-queue-dir: minio-wait ## ensure queue dir exists in MinIO container
-	@echo "$(BOLD)Ensuring MinIO queue dir$(RESET)"
-	@$(COMPOSE) exec minio sh -lc 'mkdir -p $(MINIO_QUEUE_DIR) && chmod -R 777 $(dir $(MINIO_QUEUE_DIR)) || true'
-.PHONY: minio-queue-dir
-
-minio-alias: minio-wait ## set mc alias to MinIO
-	@echo "$(BOLD)Setting mc alias $(MINIO_ALIAS) -> $(MINIO_ENDPOINT)$(RESET)"
-	@mc alias set $(MINIO_ALIAS) $(MINIO_ENDPOINT) $(MINIO_ACCESS_KEY) $(MINIO_SECRET_KEY) >/dev/null
-.PHONY: minio-alias
-
-minio-webhook-config: minio-alias minio-queue-dir ## configure webhook on MinIO
-	@echo "$(BOLD)Configuring MinIO webhook $(MINIO_WEBHOOK_NAME)$(RESET)"
-	@mc admin config set $(MINIO_ALIAS) notify_webhook:$(MINIO_WEBHOOK_NAME) \
-	  endpoint="$(MINIO_WEBHOOK_ENDPOINT)" \
-	  auth_token="Bearer $(STORAGE_EVENT_TOKEN)" \
-	  queue_dir="$(MINIO_QUEUE_DIR)" \
-	  queue_limit="$(MINIO_QUEUE_LIMIT)"
-.PHONY: minio-webhook-config
-
-minio-restart: minio-alias ## restart MinIO after config change
-	@echo "$(BOLD)Restarting MinIO service$(RESET)"
-	@mc admin service restart $(MINIO_ALIAS)
-.PHONY: minio-restart
-
-minio-events-reset: minio-alias ## remove all bucket notifications
-	@echo "$(BOLD)Removing existing bucket events on $(MINIO_BUCKET)$(RESET)"
-	@mc event remove --force $(MINIO_ALIAS)/$(MINIO_BUCKET) >/dev/null || true
-.PHONY: minio-events-reset
-
-minio-event-add: minio-alias ## add put event -> webhook
-	@echo "$(BOLD)Adding put event -> webhook $(MINIO_WEBHOOK_NAME)$(RESET)"
-	@mc event add $(MINIO_ALIAS)/$(MINIO_BUCKET) arn:minio:sqs::$(MINIO_WEBHOOK_NAME):webhook --event put
-	@mc event list $(MINIO_ALIAS)/$(MINIO_BUCKET)
-.PHONY: minio-event-add
-
-minio-webhook-setup: ## full setup: alias, config, restart, reset events, add event
-minio-webhook-setup: \
-	minio-webhook-config \
-	minio-restart \
-	minio-events-reset \
-	minio-event-add
-.PHONY: minio-webhook-setup
-
 # -- Front
 
 frontend-development-install: ## install the frontend locally

compose.yml

@@ -46,6 +46,21 @@ services:
       /usr/bin/mc mb meet/meet-media-storage && \
       exit 0;"
 
+  createwebhook:
+    image: minio/mc
+    depends_on:
+      minio:
+        condition: service_healthy
+        restart: true
+    entrypoint: >
+      sh -c "
+      /usr/bin/mc alias set meet http://minio:9000 meet password &&
+      /usr/bin/mc admin config set meet notify_webhook:meet-webhook endpoint='http://app-dev:8000/api/v1.0/recordings/storage-hook/' auth_token='Bearer password' &&
+      /usr/bin/mc admin service restart meet --wait --json &&
+      sleep 15 &&
+      /usr/bin/mc event add meet/meet-media-storage arn:minio:sqs::meet-webhook:webhook --event put &&
+      exit 0;"
+
   app-dev:
     build:
       context: .
@@ -72,6 +87,7 @@ services:
         - nginx
         - livekit
         - createbuckets
+        - createwebhook
     extra_hosts:
       - "127.0.0.1.nip.io:host-gateway"
   

docker/dinum-frontend/Dockerfile

@@ -42,7 +42,7 @@ COPY ./docker/dinum-frontend/fonts/ \
 FROM nginxinc/nginx-unprivileged:alpine3.21 AS frontend-production
 
 USER root
-RUN apk update && apk upgrade libssl3 libcrypto3 libxml2>=2.12.7-r2 libxslt>=1.1.39-r2
+RUN apk update && apk upgrade libssl3 libcrypto3 libxml2>=2.12.7-r2 libxslt>=1.1.39-r2 libexpat>=2.7.2-r0
 
 USER nginx
 

env.d/development/summary.dist

@@ -21,8 +21,3 @@ WEBHOOK_URL="https://configure-your-url.com"
 
 POSTHOG_API_KEY="your-posthog-key"
 POSTHOG_ENABLED="False"
-
-LANGFUSE_SECRET_KEY="your-secret-key"
-LANGFUSE_PUBLIC_KEY="your-public-key"
-LANGFUSE_HOST="https://cloud.langfuse.com"
-LANFUSE_ENABLED="False"

src/frontend/Dockerfile

@@ -38,7 +38,7 @@ RUN npm run build
 FROM nginxinc/nginx-unprivileged:alpine3.21 AS frontend-production
 
 USER root
-RUN apk update && apk upgrade libssl3 libcrypto3 libxml2>=2.12.7-r2 libxslt>=1.1.39-r2
+RUN apk update && apk upgrade libssl3 libcrypto3 libxml2>=2.12.7-r2 libxslt>=1.1.39-r2 libexpat>=2.7.2-r0
 
 USER nginx
 

src/frontend/src/features/rooms/livekit/hooks/useConnectionObserver.ts

@@ -25,6 +25,18 @@ export const useConnectionObserver = () => {
       posthog.capture('reconnect-event')
     }
 
+    const handleReconnected = () => {
+      posthog.capture('reconnected-event')
+    }
+
+    const handleSignalingConnect = () => {
+      posthog.capture('signaling-connect-event')
+    }
+
+    const handleSignalingReconnect = () => {
+      posthog.capture('signaling-reconnect-event')
+    }
+
     const handleDisconnect = (
       disconnectReason: DisconnectReason | undefined
     ) => {
@@ -43,13 +55,19 @@ export const useConnectionObserver = () => {
     }
 
     room.on(RoomEvent.Connected, handleConnection)
+    room.on(RoomEvent.SignalConnected, handleSignalingConnect)
     room.on(RoomEvent.Disconnected, handleDisconnect)
     room.on(RoomEvent.Reconnecting, handleReconnect)
+    room.on(RoomEvent.Reconnected, handleReconnected)
+    room.on(RoomEvent.SignalReconnecting, handleSignalingReconnect)
 
     return () => {
       room.off(RoomEvent.Connected, handleConnection)
+      room.off(RoomEvent.SignalConnected, handleSignalingConnect)
       room.off(RoomEvent.Disconnected, handleDisconnect)
       room.off(RoomEvent.Reconnecting, handleReconnect)
+      room.off(RoomEvent.Reconnected, handleReconnected)
+      room.off(RoomEvent.SignalReconnecting, handleSignalingReconnect)
     }
   }, [room, isAnalyticsEnabled])
 

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

@@ -212,7 +212,8 @@ celeryTranscribe:
     - "worker"
     - "--pool=solo"
     - "--loglevel=info"
-    - "-Q transcribe-queue"
+    - "-Q"
+    - "transcribe-queue"
 
 celerySummarize:
   replicas: 1
@@ -248,7 +249,8 @@ celerySummarize:
     - "worker"
     - "--pool=solo"
     - "--loglevel=info"
-    - "-Q summarize-queue"
+    - "-Q"
+    - "summarize-queue"
 
 agents:
   replicas: 1

src/helm/extra/templates/minio.yaml

@@ -82,9 +82,15 @@ spec:
         volumeMounts:
         - mountPath: /data
           name: data
+        - mountPath: /etc/ssl/certs/mkcert-ca.pem
+          name: mkcert
+          subPath: rootCA.pem
       volumes:
       - name: data
         emptyDir:
+      - name: mkcert
+        secret:
+          secretName: mkcert
 ---
 apiVersion: batch/v1
 kind: Job
@@ -105,3 +111,26 @@ spec:
             exit 0
       restartPolicy: Never
   backoffLimit: 1
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: minio-webhook
+spec:
+  template:
+    spec:
+      containers:
+      - name: mc
+        image: minio/mc
+        command:
+          - /bin/sh
+          - -c
+          - |
+            /usr/bin/mc alias set meet http://minio:9000 meet password && \
+            /usr/bin/mc admin config set meet notify_webhook:meet-webhook endpoint="https://meet.127.0.0.1.nip.io/api/v1.0/recordings/storage-hook/" auth_token="Bearer password" && \
+            /usr/bin/mc admin service restart meet --wait --json && \
+            sleep 15 && \
+            /usr/bin/mc event add meet/meet-media-storage arn:minio:sqs::meet-webhook:webhook --event put && \
+            exit 0
+      restartPolicy: Never
+  backoffLimit: 1

src/summary/pyproject.toml

@@ -15,7 +15,6 @@ dependencies = [
     "posthog==6.0.3",
     "requests==2.32.4",
     "sentry-sdk[fastapi, celery]==2.30.0",
-    "langfuse==3.4.0",
 ]
 
 [project.optional-dependencies]

src/summary/summary/core/celery_worker.py

@@ -9,19 +9,18 @@ import time
 from pathlib import Path
 from typing import Optional
 
+import openai
 import sentry_sdk
 from celery import Celery, signals
 from celery.utils.log import get_task_logger
 from minio import Minio
 from mutagen import File
-from openai import OpenAI
 from requests import Session, exceptions
 from requests.adapters import HTTPAdapter
 from urllib3.util import Retry
 
 from summary.core.analytics import MetadataManager, get_analytics
 from summary.core.config import get_settings
-from summary.core.observability import Observability
 from summary.core.prompt import (
     PROMPT_SYSTEM_CLEANING,
     PROMPT_SYSTEM_NEXT_STEP,
@@ -47,14 +46,6 @@ celery = Celery(
 
 celery.config_from_object("summary.core.celery_config")
 
-obs = Observability(
-    is_enabled=settings.langfuse_is_enabled,
-    langfuse_host=settings.langfuse_host,
-    langfuse_public_key=settings.langfuse_public_key,
-    langfuse_secret_key=settings.langfuse_secret_key,
-)
-logger.info("Observability enabled: %s", obs.is_enabled)
-
 if settings.sentry_dsn and settings.sentry_is_enabled:
 
     @signals.celeryd_init.connect
@@ -120,10 +111,9 @@ class LLMService:
 
     def __init__(self):
         """Init the LLMService once."""
-        self._client = OpenAI(
+        self._client = openai.OpenAI(
             base_url=settings.llm_base_url, api_key=settings.llm_api_key
         )
-        self.gen_ctx = obs.generation
 
     def call(self, system_prompt: str, user_prompt: str):
         """Call the LLM service.
@@ -144,14 +134,6 @@ class LLMService:
             logger.error("LLM call failed: %s", e)
             raise LLMException("LLM call failed.") from e
 
-    def call_llm_gen(self, name, system, user):
-        """Call the LLM service within a generation context."""
-        with self.gen_ctx(
-            name=name,
-            model=settings.llm_model,
-        ):
-            return self.call(system, user)
-
 
 def format_segments(transcription_data):
     """Format transcription segments from WhisperX into a readable conversation format.
@@ -218,9 +200,9 @@ def task_failure_handler(task_id, exception=None, **kwargs):
     bind=True,
     autoretry_for=[exceptions.HTTPError],
     max_retries=settings.celery_max_retries,
+    queue=settings.transcribe_queue,
 )
-@obs.observe(name="process-audio", capture_input=True, capture_output=False)
-def process_audio_transcribe_summarize_v2(  # noqa: PLR0915
+def process_audio_transcribe_summarize_v2(
     self,
     filename: str,
     email: str,
@@ -241,21 +223,6 @@ def process_audio_transcribe_summarize_v2(  # noqa: PLR0915
     logger.info("Notification received")
     logger.debug("filename: %s", filename)
 
-    try:
-        obs.update_current_trace(
-            user_id=sub or email,
-            tags=["celery", "transcription", "whisperx"],
-            metadata={
-                "filename": filename,
-                "room": room,
-                "recording_date": recording_date,
-                "recording_time": recording_time,
-            },
-        )
-    except Exception as e:
-        logger.warning("Langfuse update trace failed: %s", e)
-        pass
-
     task_id = self.request.id
 
     minio_client = Minio(
@@ -270,6 +237,7 @@ def process_audio_transcribe_summarize_v2(  # noqa: PLR0915
     audio_file_stream = minio_client.get_object(
         settings.aws_storage_bucket_name, object_name=filename
     )
+
     temp_file_path = save_audio_stream(audio_file_stream)
 
     logger.info("Recording successfully downloaded")
@@ -290,7 +258,7 @@ def process_audio_transcribe_summarize_v2(  # noqa: PLR0915
         raise AudioValidationError(error_msg)
 
     logger.info("Initiating WhisperX client")
-    whisperx_client = OpenAI(
+    whisperx_client = openai.OpenAI(
         api_key=settings.whisperx_api_key,
         base_url=settings.whisperx_base_url,
         max_retries=settings.whisperx_max_retries,
@@ -299,25 +267,20 @@ def process_audio_transcribe_summarize_v2(  # noqa: PLR0915
     try:
         logger.info("Querying transcription …")
         transcription_start_time = time.time()
-        with obs.span(
-            name="whisperx.transcribe",
-            input={
-                "model": settings.whisperx_asr_model,
-                "audio_seconds": round(audio_file.info.length, 2),
-                "endpoint": settings.whisperx_base_url,
-            },
-        ):
-            with open(temp_file_path, "rb") as audio_file_rb:
-                transcription = whisperx_client.audio.transcriptions.create(
-                    model=settings.whisperx_asr_model,
-                    file=audio_file_rb,
-                )
-        metadata_manager.track(
-            task_id,
-            {"transcription_time": round(time.time() - transcription_start_time, 2)},
-        )
-        logger.info("Transcription received.")
-        logger.debug("Transcription: \n %s", transcription)
+        with open(temp_file_path, "rb") as audio_file:
+            transcription = whisperx_client.audio.transcriptions.create(
+                model=settings.whisperx_asr_model, file=audio_file
+            )
+            metadata_manager.track(
+                task_id,
+                {
+                    "transcription_time": round(
+                        time.time() - transcription_start_time, 2
+                    )
+                },
+            )
+            logger.info("Transcription received.")
+            logger.debug("Transcription: \n %s", transcription)
     finally:
         if os.path.exists(temp_file_path):
             os.remove(temp_file_path)
@@ -375,7 +338,6 @@ def process_audio_transcribe_summarize_v2(  # noqa: PLR0915
     max_retries=settings.celery_max_retries,
     queue=settings.summarize_queue,
 )
-@obs.observe(name="summarize-transcription", capture_input=False, capture_output=False)
 def summarize_transcription(self, transcript: str, email: str, sub: str, title: str):
     """Generate a summary from the provided transcription text.
 
@@ -390,21 +352,11 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title:
 
     llm_service = LLMService()
 
-    try:
-        obs.update_current_trace(
-            user_id=sub or email,
-            tags=["celery", "summarization"],
-            metadata={"title": title},
-        )
-    except Exception as e:
-        logger.warning("Langfuse update trace failed: %s", e)
-        pass
-
-    tldr = llm_service.call_llm_gen("tldr", PROMPT_SYSTEM_TLDR, transcript)
+    tldr = llm_service.call(PROMPT_SYSTEM_TLDR, transcript)
 
     logger.info("TLDR generated")
 
-    parts = llm_service.call_llm_gen("plan", PROMPT_SYSTEM_PLAN, transcript)
+    parts = llm_service.call(PROMPT_SYSTEM_PLAN, transcript)
     logger.info("Plan generated")
 
     parts = parts.split("\n")
@@ -415,22 +367,16 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title:
     for part in parts:
         prompt_user_part = PROMPT_USER_PART.format(part=part, transcript=transcript)
         logger.info("Summarizing part: %s", part)
-        parts_summarized.append(
-            llm_service.call_llm_gen("part", PROMPT_SYSTEM_PART, prompt_user_part)
-        )
+        parts_summarized.append(llm_service.call(PROMPT_SYSTEM_PART, prompt_user_part))
 
     logger.info("Parts summarized")
 
     raw_summary = "\n\n".join(parts_summarized)
 
-    next_steps = llm_service.call_llm_gen(
-        "next_steps", PROMPT_SYSTEM_NEXT_STEP, transcript
-    )
+    next_steps = llm_service.call(PROMPT_SYSTEM_NEXT_STEP, transcript)
     logger.info("Next steps generated")
 
-    cleaned_summary = llm_service.call_llm_gen(
-        "cleaning", PROMPT_SYSTEM_CLEANING, raw_summary
-    )
+    cleaned_summary = llm_service.call(PROMPT_SYSTEM_CLEANING, raw_summary)
     logger.info("Summary cleaned")
 
     summary = tldr + "\n\n" + cleaned_summary + "\n\n" + next_steps
@@ -450,8 +396,3 @@ def summarize_transcription(self, transcript: str, email: str, sub: str, title:
 
     logger.info("Webhook submitted successfully. Status: %s", response.status_code)
     logger.debug("Response body: %s", response.text)
-    try:
-        obs.flush()
-    except Exception as e:
-        logger.warning("Langfuse flush failed: %s", e)
-        pass

src/summary/summary/core/config.py

@@ -4,7 +4,6 @@ from functools import lru_cache
 from typing import Annotated, List, Optional
 
 from fastapi import Depends
-from pydantic import SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -76,12 +75,6 @@ class Settings(BaseSettings):
     task_tracker_redis_url: str = "redis://redis/0"
     task_tracker_prefix: str = "task_metadata:"
 
-    # Langfuse
-    langfuse_is_enabled: bool = True
-    langfuse_host: Optional[str] = None
-    langfuse_public_key: Optional[str] = None
-    langfuse_secret_key: Optional[SecretStr] = None
-
 
 @lru_cache
 def get_settings():

src/summary/summary/core/observability.py

@@ -1,101 +0,0 @@
-"""Wrapper around Langfuse observability."""
-
-from __future__ import annotations
-
-import logging
-from contextlib import nullcontext
-from typing import Any, Callable, ContextManager
-
-logger = logging.getLogger(__name__)
-
-try:
-    from langfuse import Langfuse as _Langfuse
-    from langfuse import observe as _lf_observe
-except Exception as e:
-    logger.debug("Langfuse import failed: %s", e)
-    _Langfuse = None
-    _lf_observe = None
-
-
-class Observability:
-    """Wrapper around Langfuse observability."""
-
-    def __init__(
-        self, is_enabled, langfuse_host, langfuse_public_key, langfuse_secret_key
-    ) -> None:
-        """Initialize the Observability instance."""
-        self._client = None
-        if hasattr(langfuse_secret_key, "get_secret_value"):
-            langfuse_secret_key = langfuse_secret_key.get_secret_value()
-
-        self._enabled = bool(
-            is_enabled and langfuse_host and langfuse_public_key and langfuse_secret_key
-        )
-
-        if not self._enabled or _Langfuse is None:
-            self._enabled = False
-            return
-
-        try:
-            self._client = _Langfuse(
-                public_key=langfuse_public_key,
-                secret_key=langfuse_secret_key,
-                host=langfuse_host,
-            )
-        except Exception as e:
-            logger.warning("Langfuse init failed: %s", e)
-            self._enabled = False
-            self._client = None
-
-    def observe(
-        self, **decorator_kwargs
-    ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
-        """Decorator to observe a function with Langfuse. If disabled, returns a no-op decorator."""  # noqa: E501
-        if self._enabled and self._client and _lf_observe is not None:
-            return _lf_observe(**decorator_kwargs)
-
-        def _noop(fn):
-            return fn
-
-        return _noop
-
-    def span(self, name: str, **kwargs) -> ContextManager[Any]:
-        """Context manager to create a span with Langfuse."""
-        if self._enabled and self._client:
-            start_span = getattr(self._client, "start_as_current_span", None)
-            if callable(start_span):
-                return start_span(name=name, **kwargs)
-        return nullcontext()
-
-    def generation(self, **kwargs) -> ContextManager[Any]:
-        """Context manager to create a generation with Langfuse."""
-        if self._enabled and self._client:
-            start_gen = getattr(self._client, "start_as_current_generation", None)
-            if callable(start_gen):
-                return start_gen(**kwargs)
-        return nullcontext()
-
-    def update_current_trace(self, **kwargs) -> None:
-        """Update the current trace with additional metadata."""
-        if not (self._enabled and self._client):
-            return
-        try:
-            self._client.update_current_trace(**kwargs)
-        except Exception as e:
-            logger.warning("Langfuse update_current_trace failed: %s", e)
-            pass
-
-    def flush(self) -> None:
-        """Flush any buffered data to Langfuse."""
-        if not (self._enabled and self._client):
-            return
-        try:
-            self._client.flush()
-        except Exception as e:
-            logger.warning("Langfuse flush failed: %s", e)
-            pass
-
-    @property
-    def is_enabled(self) -> bool:
-        """Check if observability is enabled."""
-        return bool(self._enabled and self._client)