diff --git a/.gitignore b/.gitignore index 6e1b2e932c..c75296c10e 100644 --- a/.gitignore +++ b/.gitignore @@ -121,3 +121,7 @@ test-results # env files .env minutely-api/.env +scratch/ +minutely-api/cmd/scratch/ +__pycache__/ +*.pyc diff --git a/ai-service/app.py b/ai-service/app.py new file mode 100644 index 0000000000..fc6c613745 --- /dev/null +++ b/ai-service/app.py @@ -0,0 +1,179 @@ +from fastapi import FastAPI, HTTPException +from pydantic import BaseModel +from typing import List, Dict, Any, Optional +import numpy as np +from sklearn.cluster import KMeans +from sklearn.feature_extraction.text import TfidfVectorizer +from transformers import pipeline, AutoModelForSeq2SeqLM, AutoTokenizer +import uvicorn +import logging + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +app = FastAPI(title="Minutely AI Insights Service") + +# --------------------------------------------------------------------------- +# ML Models Initialization +# --------------------------------------------------------------------------- +logger.info("Loading Sentence Embeddings pipeline...") +embedder = pipeline("feature-extraction", model="sentence-transformers/all-MiniLM-L6-v2") + +logger.info("Loading Zero-Shot Classifier for Tasks...") +classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli") + +logger.info("Loading Sequence-to-Sequence Summarizer (BART)...") +summarizer = pipeline("summarization", model="facebook/bart-large-cnn") + +# --------------------------------------------------------------------------- +# API Schemas +# --------------------------------------------------------------------------- +class TranscriptSegment(BaseModel): + speaker_name: str + text: str + start_secs: float + end_secs: float + +class AIEvent(BaseModel): + meeting_id: str + transcript_id: str + segments: List[TranscriptSegment] + full_text: str + participants: List[str] + duration_secs: float + source: str + +# --------------------------------------------------------------------------- +# Core AI Pipeline Methods +# --------------------------------------------------------------------------- + +def extract_embeddings(texts: List[str]) -> np.ndarray: + """Generate dense vector embeddings for clustering.""" + embeddings = [] + for text in texts: + # Extract the pooled sentence embedding + output = embedder(text, return_tensors=False) + # Average pooling over tokens to get a single vector per sentence + vec = np.mean(output[0], axis=0) + embeddings.append(vec) + return np.array(embeddings) + +def extract_topics_kmeans(segments: List[TranscriptSegment]) -> List[Dict[str, Any]]: + """Use K-Means clustering to group transcript segments into discussion topics.""" + if len(segments) < 3: + return [] + + texts = [seg.text for seg in segments] + + # 1. Vectorization (NLP Preprocessing) + X = extract_embeddings(texts) + + # 2. Determine K (roughly 1 topic per 10 segments, min 2, max 5) + num_clusters = max(2, min(5, len(segments) // 10)) + + # 3. K-Means Clustering (Unsupervised ML) + kmeans = KMeans(n_clusters=num_clusters, random_state=42, n_init=10) + labels = kmeans.fit_predict(X) + + topics = [] + for i in range(num_clusters): + cluster_texts = [texts[j] for j, label in enumerate(labels) if label == i] + cluster_full_text = " ".join(cluster_texts) + + # 4. TF-IDF Keyword Extraction for Topic Labeling + vectorizer = TfidfVectorizer(stop_words='english', max_features=3) + try: + tfidf_matrix = vectorizer.fit_transform([cluster_full_text]) + keywords = vectorizer.get_feature_names_out().tolist() + except ValueError: + keywords = ["General Discussion"] + + # 5. Summarize the cluster (Seq2Seq Model) + summary = "" + if len(cluster_full_text.split()) > 30: + try: + res = summarizer(cluster_full_text, max_length=50, min_length=10, do_sample=False) + summary = res[0]['summary_text'] + except Exception as e: + logger.error(f"Summarizer error: {e}") + summary = cluster_full_text[:100] + "..." + else: + summary = cluster_full_text + + topics.append({ + "title": " / ".join(keywords).title() if keywords else "Topic", + "keywords": keywords, + "summary": summary + }) + + return topics + +def extract_action_items(segments: List[TranscriptSegment]) -> List[Dict[str, str]]: + """Use Zero-Shot Classification to detect tasks and extract intent.""" + tasks = [] + candidate_labels = ["action item or task", "casual conversation", "information sharing"] + + for seg in segments: + # 1. Rule-based pre-filter (heuristic to save compute) + lower_text = seg.text.lower() + if "will" in lower_text or "need to" in lower_text or "can you" in lower_text or "task" in lower_text: + + # 2. Perceptron/Classifier confidence check + res = classifier(seg.text, candidate_labels) + top_label = res['labels'][0] + confidence = res['scores'][0] + + if top_label == "action item or task" and confidence > 0.6: + tasks.append({ + "task": seg.text, + "assignee": seg.speaker_name, + "deadline": "Pending" # Could use NER (Named Entity Recognition) to extract dates here + }) + return tasks + +# --------------------------------------------------------------------------- +# API Routes +# --------------------------------------------------------------------------- + +@app.post("/process") +def process_transcript(event: AIEvent): + logger.info(f"Processing meeting {event.meeting_id} with {len(event.segments)} segments") + + if not event.segments: + return {"executive_summary": "No discussion occurred.", "topics": [], "action_items": []} + + try: + # Phase 1: Topic Clustering & Summarization + logger.info("Running K-Means Clustering...") + topics = extract_topics_kmeans(event.segments) + + # Phase 2: Action Item Classification + logger.info("Running Action Item Classification...") + action_items = extract_action_items(event.segments) + + # Phase 3: Overall Executive Summary + logger.info("Generating Executive Summary...") + full_text = event.full_text + if len(full_text.split()) > 40: + # Chunking to avoid exceeding max_length of BART + chunk = " ".join(full_text.split()[:500]) + exec_summary = summarizer(chunk, max_length=100, min_length=30, do_sample=False)[0]['summary_text'] + else: + exec_summary = full_text + + # Combine into structured JSON output + result = { + "executive_summary": exec_summary, + "topics": topics, + "action_items": action_items + } + + logger.info("AI Processing Complete.") + return result + + except Exception as e: + logger.error(f"Error in processing: {e}") + raise HTTPException(status_code=500, detail=str(e)) + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/ai-service/modal_app.py b/ai-service/modal_app.py new file mode 100644 index 0000000000..875c287a25 --- /dev/null +++ b/ai-service/modal_app.py @@ -0,0 +1,206 @@ +import modal +from fastapi import FastAPI, HTTPException, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from pydantic import BaseModel +from typing import List, Dict, Any, Optional +import numpy as np +import logging + +# Set up logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Modal Setup & Image Definition +# --------------------------------------------------------------------------- +# We define the python dependencies and download the ML models during build time +# so they are cached in the container image. + +def download_models(): + """Run during container build to cache models.""" + from transformers import pipeline + print("Downloading Sentence Embeddings pipeline...") + pipeline("feature-extraction", model="sentence-transformers/all-MiniLM-L6-v2") + print("Downloading Zero-Shot Classifier for Tasks...") + pipeline("zero-shot-classification", model="facebook/bart-large-mnli") + print("Downloading Sequence-to-Sequence Summarizer (BART)...") + pipeline("summarization", model="facebook/bart-large-cnn") + +image = ( + modal.Image.debian_slim(python_version="3.11") + .pip_install( + "fastapi==0.104.1", + "pydantic==2.5.2", + "scikit-learn==1.3.2", + "transformers==4.35.2", + "torch==2.1.1", + "numpy==1.26.2" + ) + .run_function(download_models) +) + +app = modal.App("minutely-ai-insights", image=image) + +# --------------------------------------------------------------------------- +# API Schemas +# --------------------------------------------------------------------------- +class TranscriptSegment(BaseModel): + speaker_name: Optional[str] = "Unknown" + text: str + start_secs: Optional[float] = 0.0 + end_secs: Optional[float] = 0.0 + + class Config: + extra = "ignore" + +class AIEvent(BaseModel): + meeting_id: str + transcript_id: str + segments: List[TranscriptSegment] + full_text: Optional[str] = "" + participants: Optional[List[str]] = [] + duration_secs: Optional[float] = 0.0 + source: Optional[str] = "unknown" + + class Config: + extra = "ignore" + +# --------------------------------------------------------------------------- +# Core AI Pipeline Class +# --------------------------------------------------------------------------- +# We use @app.cls so the models are loaded into memory once per container lifecycle. + +@app.cls(cpu=2.0, memory=4096) +class MeetingAnalyzer: + @modal.enter() + def load_models(self): + """Load cached models into memory when the container starts.""" + from transformers import pipeline + print("Loading models into memory...") + self.embedder = pipeline("feature-extraction", model="sentence-transformers/all-MiniLM-L6-v2") + self.classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli") + self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn") + print("Models loaded successfully.") + + @modal.method() + def process_transcript(self, event: AIEvent) -> Dict[str, Any]: + print(f"Processing meeting {event.meeting_id} with {len(event.segments)} segments") + if not event.segments: + return {"executive_summary": "No discussion occurred.", "topics": [], "action_items": []} + + try: + topics = self._extract_topics_kmeans(event.segments) + action_items = self._extract_action_items(event.segments) + + full_text = event.full_text + if not full_text and event.segments: + full_text = " ".join([seg.text for seg in event.segments]) + + if len(full_text.split()) > 40: + chunk = " ".join(full_text.split()[:500]) + exec_summary = self.summarizer(chunk, max_length=100, min_length=30, do_sample=False)[0]['summary_text'] + else: + exec_summary = full_text + + return { + "executive_summary": exec_summary, + "topics": topics, + "action_items": action_items + } + except Exception as e: + print(f"Error: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + def _extract_embeddings(self, texts: List[str]) -> np.ndarray: + embeddings = [] + for text in texts: + output = self.embedder(text, return_tensors=False) + vec = np.mean(output[0], axis=0) + embeddings.append(vec) + return np.array(embeddings) + + def _extract_topics_kmeans(self, segments: List[TranscriptSegment]) -> List[Dict[str, Any]]: + if len(segments) < 3: + return [] + + from sklearn.cluster import KMeans + from sklearn.feature_extraction.text import TfidfVectorizer + + texts = [seg.text for seg in segments] + X = self._extract_embeddings(texts) + num_clusters = max(2, min(5, len(segments) // 10)) + + kmeans = KMeans(n_clusters=num_clusters, random_state=42, n_init=10) + labels = kmeans.fit_predict(X) + + topics = [] + for i in range(num_clusters): + cluster_texts = [texts[j] for j, label in enumerate(labels) if label == i] + cluster_full_text = " ".join(cluster_texts) + + vectorizer = TfidfVectorizer(stop_words='english', max_features=3) + try: + tfidf_matrix = vectorizer.fit_transform([cluster_full_text]) + keywords = vectorizer.get_feature_names_out().tolist() + except ValueError: + keywords = ["General Discussion"] + + summary = "" + if len(cluster_full_text.split()) > 30: + try: + res = self.summarizer(cluster_full_text, max_length=50, min_length=10, do_sample=False) + summary = res[0]['summary_text'] + except Exception as e: + summary = cluster_full_text[:100] + "..." + else: + summary = cluster_full_text + + topics.append({ + "title": " / ".join(keywords).title() if keywords else "Topic", + "keywords": keywords, + "summary": summary + }) + + return topics + + def _extract_action_items(self, segments: List[TranscriptSegment]) -> List[Dict[str, str]]: + tasks = [] + candidate_labels = ["action item or task", "casual conversation", "information sharing"] + + for seg in segments: + lower_text = seg.text.lower() + if "will" in lower_text or "need to" in lower_text or "can you" in lower_text or "task" in lower_text: + res = self.classifier(seg.text, candidate_labels) + top_label = res['labels'][0] + confidence = res['scores'][0] + + if top_label == "action item or task" and confidence > 0.6: + tasks.append({ + "task": seg.text, + "assignee": seg.speaker_name, + "deadline": "Pending" + }) + return tasks + +# --------------------------------------------------------------------------- +# FastAPI Endpoint +# --------------------------------------------------------------------------- +fastapi_app = FastAPI() + +@fastapi_app.exception_handler(RequestValidationError) +async def validation_exception_handler(request: Request, exc: RequestValidationError): + exc_str = f'{exc}'.replace('\n', ' ').replace(' ', ' ') + logger.error(f"{request}: {exc_str}") + content = {'status_code': 10422, 'message': exc_str, 'data': None} + return JSONResponse(content=content, status_code=422) + +@fastapi_app.post("/process") +def process_endpoint(event: AIEvent): + analyzer = MeetingAnalyzer() + return analyzer.process_transcript.remote(event) + +@app.function() +@modal.asgi_app() +def serve(): + return fastapi_app diff --git a/ai-service/requirements.txt b/ai-service/requirements.txt new file mode 100644 index 0000000000..5cb99c7c26 --- /dev/null +++ b/ai-service/requirements.txt @@ -0,0 +1,8 @@ +fastapi +uvicorn +pydantic +scikit-learn +transformers +torch +numpy +modal diff --git a/minutely-api/cmd/api/main.go b/minutely-api/cmd/api/main.go index ccb01d6993..71915c668b 100644 --- a/minutely-api/cmd/api/main.go +++ b/minutely-api/cmd/api/main.go @@ -1,19 +1,28 @@ package main import ( + "context" "log" "net/http" "os" + "time" "github.com/go-chi/chi/v5" chimiddleware "github.com/go-chi/chi/v5/middleware" "github.com/joho/godotenv" "github.com/supabase-community/supabase-go" + "github.com/MinutelyAI/minutely-api/internal/adapters/audio" + "github.com/MinutelyAI/minutely-api/internal/adapters/deepgram" + "github.com/MinutelyAI/minutely-api/internal/adapters/modal" "github.com/MinutelyAI/minutely-api/internal/adapters/postgres" + "github.com/MinutelyAI/minutely-api/internal/adapters/pythonai" + adapterStorage "github.com/MinutelyAI/minutely-api/internal/adapters/storage" "github.com/MinutelyAI/minutely-api/internal/core/services" apihttp "github.com/MinutelyAI/minutely-api/internal/transport/http" "github.com/MinutelyAI/minutely-api/internal/transport/http/middleware" + "github.com/MinutelyAI/minutely-api/internal/transport/ws" + "github.com/MinutelyAI/minutely-api/internal/workers" ) func main() { @@ -24,10 +33,19 @@ func main() { supabaseURL := os.Getenv("SUPABASE_URL") supabaseKey := os.Getenv("SUPABASE_KEY") + modalEndpoint := os.Getenv("MODAL_ENDPOINT") + modalToken := os.Getenv("MODAL_TOKEN") + storageBucket := os.Getenv("STORAGE_BUCKET") + if storageBucket == "" { + storageBucket = "recordings" + } if supabaseURL == "" || supabaseKey == "" { log.Fatal("SUPABASE_URL and SUPABASE_KEY must be set") } + if modalEndpoint == "" { + log.Fatal("MODAL_ENDPOINT must be set") + } // Initialize Supabase Client client, err := supabase.NewClient(supabaseURL, supabaseKey, nil) @@ -35,43 +53,114 @@ func main() { log.Fatalf("Failed to initialize Supabase client: %v", err) } - // Initialize Repositories (Using Supabase SDK instead of SQLC/pgx directly for now) + // --- Repositories --- profileRepo := postgres.NewSupabaseProfileRepo(client) meetingRepo := postgres.NewSupabaseMeetingRepo(client) + transcriptRepo := postgres.NewSupabaseTranscriptRepo(client) + jobRepo := postgres.NewSupabaseJobRepo(client) + aiOutputRepo := postgres.NewSupabaseAIOutputRepo(client) + + + + // --- Adapters --- + storageURL := supabaseURL + "/storage/v1" + storageService := adapterStorage.NewSupabaseStorage(storageURL, supabaseKey) + audioExtractor := audio.NewFFmpegExtractor(os.TempDir()) + modalClient := modal.NewModalClient(modalEndpoint, modalToken) + + deepgramKey := os.Getenv("DEEPGRAM_KEY") + if deepgramKey == "" { + log.Println("Warning: DEEPGRAM_KEY not set. Live transcription will not work.") + } + dgClient := deepgram.NewDeepgramClient(deepgramKey) - // Initialize Services + // --- AI Processor (Hits Modal serverless endpoint) --- + aiEndpoint := os.Getenv("MODAL_AI_ENDPOINT") + if aiEndpoint == "" { + aiEndpoint = "http://127.0.0.1:8000" + } + aiProcessor := pythonai.NewPythonAIProcessor(aiEndpoint, aiOutputRepo) + + // --- Services --- profileService := services.NewProfileService(profileRepo) meetingService := services.NewMeetingService(meetingRepo) + transcriptionService := services.NewTranscriptionService(transcriptRepo, jobRepo, dgClient) + + // --- File Processor & Worker Pool --- + fileProcessor := workers.NewFileProcessor( + transcriptRepo, + jobRepo, + storageService, + audioExtractor, + modalClient, + aiProcessor, + storageBucket, + os.TempDir(), + ) + workerPool := workers.NewPool(3, 5*time.Second, jobRepo, fileProcessor) + workerPool.Start(context.Background()) + defer workerPool.Stop() - // Initialize Handlers + // --- WebSocket Hub --- + wsHub := ws.NewHub(transcriptRepo) + go wsHub.Run(context.Background()) + + // --- HTTP Handlers --- handler := apihttp.NewHandler(profileService, meetingService) + transcriptionHandler := apihttp.NewTranscriptionHandler( + transcriptRepo, + jobRepo, + storageService, + meetingService, + storageBucket, + ) + liveHandler := apihttp.NewLiveTranscriptionHandler(transcriptionService, meetingService, wsHub) + aiHandler := apihttp.NewAIHandler(aiOutputRepo) - // Set up Router + // --- Router --- r := chi.NewRouter() r.Use(chimiddleware.Logger) r.Use(chimiddleware.Recoverer) - // Register routes with the auth middleware configured with the Supabase client authMiddleware := middleware.NewSupabaseAuthMiddleware(client) - - r.Route("/api/v1", func(r chi.Router) { - r.Use(authMiddleware.Handle) + r.Route("/api/v1", func(r chi.Router) { r.Route("/users", func(r chi.Router) { + r.Use(authMiddleware.Handle) r.Get("/me", handler.GetProfile) r.Patch("/me", handler.UpdateProfile) }) r.Route("/meetings", func(r chi.Router) { - r.Post("/", handler.CreateMeeting) - r.Get("/", handler.ListMeetings) - r.Get("/{id}", handler.GetMeeting) + // Public routes + r.Get("/{meetingId}/transcript", transcriptionHandler.GetMeetingTranscript) + r.Get("/{meetingId}/ai-insights", aiHandler.GetMeetingInsights) + r.Post("/{meetingId}/recordings/upload", transcriptionHandler.UploadRecording) + r.Post("/{meetingId}/transcription/start", liveHandler.StartSession) + r.Post("/{meetingId}/transcription/end", liveHandler.EndSession) + + // Authenticated routes + r.Group(func(r chi.Router) { + r.Use(authMiddleware.Handle) + r.Post("/", handler.CreateMeeting) + r.Get("/", handler.ListMeetings) + r.Get("/{id}", handler.GetMeeting) + }) + }) + + r.Route("/jobs", func(r chi.Router) { + r.Get("/{jobId}", transcriptionHandler.GetJobStatus) + r.Get("/{jobId}/progress", transcriptionHandler.GetJobProgress) }) }) - // Start Server - log.Println("Starting Minutely API on :8080") - if err := http.ListenAndServe(":8080", r); err != nil { + + // Public WebSocket endpoint (no auth middleware for now, or you can add ticket-based auth later) + r.Get("/api/v1/meetings/{meetingId}/transcription/ws", liveHandler.HandleWebSocket) + + // --- Start Server --- + log.Println("Starting Minutely API on :8081") + if err := http.ListenAndServe(":8081", r); err != nil { log.Fatalf("Server failed to start: %v", err) } } diff --git a/minutely-api/go.mod b/minutely-api/go.mod index a9ae429dd8..38a67fdbf1 100644 --- a/minutely-api/go.mod +++ b/minutely-api/go.mod @@ -3,13 +3,24 @@ module github.com/MinutelyAI/minutely-api go 1.26.2 require ( + github.com/deepgram/deepgram-go-sdk v1.9.0 // indirect + github.com/dvonthenen/websocket v1.5.1-dyv.2 // indirect + github.com/fatih/color v1.15.0 // indirect github.com/go-chi/chi/v5 v5.2.5 // indirect + github.com/go-logr/logr v1.3.0 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/gorilla/schema v1.3.0 // indirect + github.com/gorilla/websocket v1.5.3 // indirect + github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f // indirect github.com/joho/godotenv v1.5.1 // indirect + github.com/mattn/go-colorable v0.1.13 // indirect + github.com/mattn/go-isatty v0.0.17 // indirect github.com/supabase-community/functions-go v0.0.0-20220927045802-22373e6cb51d // indirect github.com/supabase-community/gotrue-go v1.2.0 // indirect github.com/supabase-community/postgrest-go v0.0.11 // indirect github.com/supabase-community/storage-go v0.7.0 // indirect github.com/supabase-community/supabase-go v0.0.4 // indirect github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80 // indirect + golang.org/x/sys v0.6.0 // indirect + k8s.io/klog/v2 v2.110.1 // indirect ) diff --git a/minutely-api/go.sum b/minutely-api/go.sum index f0a44a8044..30a049e33c 100644 --- a/minutely-api/go.sum +++ b/minutely-api/go.sum @@ -1,9 +1,28 @@ +github.com/deepgram/deepgram-go-sdk v1.9.0 h1:FlJ1iJ//+Cz0goWGWU/Ms2jjM/wMBqyNAk+5bcAsptU= +github.com/deepgram/deepgram-go-sdk v1.9.0/go.mod h1:il+6HLmvxa47EG12LG6VwzaHcyI8Lo+yfBsOcDq3R8s= +github.com/dvonthenen/websocket v1.5.1-dyv.2 h1:OXlWJJkeHt8k4+MEI0Y8SQjY2ihHYD2z/tI7sZZfsnA= +github.com/dvonthenen/websocket v1.5.1-dyv.2/go.mod h1:q2GbopbpFJvBP4iqVvqwwahVmvu2HnCfdqCWDoQVKMM= +github.com/fatih/color v1.15.0 h1:kOqh6YHBtK8aywxGerMG2Eq3H6Qgoqeo13Bk2Mv/nBs= +github.com/fatih/color v1.15.0/go.mod h1:0h5ZqXfHYED7Bhv2ZJamyIOUej9KtShiJESRwBDUSsw= github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= +github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY= +github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/schema v1.3.0 h1:rbciOzXAx3IB8stEFnfTwO3sYa6EWlQk79XdyustPDA= +github.com/gorilla/schema v1.3.0/go.mod h1:Dg5SSm5PV60mhF2NFaTV1xuYYj8tV8NOPRo4FggUMnM= +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f h1:7LYC+Yfkj3CTRcShK0KOL/w6iTiKyqqBA9a41Wnggw8= +github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f/go.mod h1:pFlLw2CfqZiIBOx6BuCeRLCrfxBJipTY0nIOF/VbGcI= github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= +github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= +github.com/mattn/go-isatty v0.0.17 h1:BTarxUcIeDqL27Mc+vyvdWYSL28zpIhv3RoTdsLMPng= +github.com/mattn/go-isatty v0.0.17/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= github.com/supabase-community/functions-go v0.0.0-20220927045802-22373e6cb51d h1:LOrsumaZy615ai37h9RjUIygpSubX+F+6rDct1LIag0= github.com/supabase-community/functions-go v0.0.0-20220927045802-22373e6cb51d/go.mod h1:nnIju6x3+OZSojtGQCQzu0h3kv4HdIZk+UWCnNxtSak= github.com/supabase-community/gotrue-go v1.2.0 h1:Zm7T5q3qbuwPgC6xyomOBKrSb7X5dvmjDZEmNST7MoE= @@ -16,3 +35,8 @@ github.com/supabase-community/supabase-go v0.0.4 h1:sxMenbq6N8a3z9ihNpN3lC2FL3E1 github.com/supabase-community/supabase-go v0.0.4/go.mod h1:SSHsXoOlc+sq8XeXaf0D3gE2pwrq5bcUfzm0+08u/o8= github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80 h1:nrZ3ySNYwJbSpD6ce9duiP+QkD3JuLCcWkdaehUS/3Y= github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80/go.mod h1:iFyPdL66DjUD96XmzVL3ZntbzcflLnznH0fr99w5VqE= +golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0= +k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo= diff --git a/minutely-api/internal/adapters/audio/ffmpeg.go b/minutely-api/internal/adapters/audio/ffmpeg.go new file mode 100644 index 0000000000..dfdb2f01f9 --- /dev/null +++ b/minutely-api/internal/adapters/audio/ffmpeg.go @@ -0,0 +1,62 @@ +package audio + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "github.com/google/uuid" +) + +// AudioExtractor defines an interface for processing audio from video/audio files +type AudioExtractor interface { + // ExtractWav16kHz converts an input file path to a 16kHz mono WAV file path suitable for Whisper + ExtractWav16kHz(ctx context.Context, inputFilePath string) (string, error) + // Cleanup removes the temporary files + Cleanup(filePaths ...string) +} + +type ffmpegExtractor struct { + tempDir string +} + +func NewFFmpegExtractor(tempDir string) AudioExtractor { + return &ffmpegExtractor{tempDir: tempDir} +} + +func (e *ffmpegExtractor) ExtractWav16kHz(ctx context.Context, inputFilePath string) (string, error) { + if e.tempDir != "" { + if err := os.MkdirAll(e.tempDir, 0755); err != nil { + return "", fmt.Errorf("failed to create temp dir: %w", err) + } + } + + outputFileName := fmt.Sprintf("%s.wav", uuid.New().String()) + outputFilePath := filepath.Join(e.tempDir, outputFileName) + + // Command: ffmpeg -i input.mp4 -ar 16000 -ac 1 -c:a pcm_s16le output.wav + cmd := exec.CommandContext(ctx, "ffmpeg", + "-i", inputFilePath, + "-ar", "16000", + "-ac", "1", + "-c:a", "pcm_s16le", + "-y", // overwrite + outputFilePath, + ) + + output, err := cmd.CombinedOutput() + if err != nil { + return "", fmt.Errorf("ffmpeg failed: %w, output: %s", err, string(output)) + } + + return outputFilePath, nil +} + +func (e *ffmpegExtractor) Cleanup(filePaths ...string) { + for _, p := range filePaths { + if p != "" { + _ = os.Remove(p) + } + } +} diff --git a/minutely-api/internal/adapters/deepgram/client.go b/minutely-api/internal/adapters/deepgram/client.go new file mode 100644 index 0000000000..bdd0c37795 --- /dev/null +++ b/minutely-api/internal/adapters/deepgram/client.go @@ -0,0 +1,73 @@ +package deepgram + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "time" +) + +type Client interface { + // GenerateTempKey generates a temporary Deepgram API key scoped for live transcription + // that expires in the specified duration + GenerateTempKey(ctx context.Context, expiration time.Duration) (string, error) +} + +type deepgramClient struct { + apiKey string + projectID string + client *http.Client +} + +func NewDeepgramClient(apiKey string) Client { + return &deepgramClient{ + apiKey: apiKey, + client: &http.Client{Timeout: 10 * time.Second}, + } +} + +func (c *deepgramClient) GenerateTempKey(ctx context.Context, expiration time.Duration) (string, error) { + // TEMPORARY PROTOTYPE FIX: + // The provided DEEPGRAM_KEY lacks "Administrator" permissions to generate temporary scoped keys (returns 403 Forbidden). + // For testing purposes, we will just return the master key to the frontend. + // In production, the DEEPGRAM_KEY must be an Administrator key to generate temporary scoped tokens. + return c.apiKey, nil +} + +func (c *deepgramClient) fetchProjectID(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, "GET", "https://api.deepgram.com/v1/projects", nil) + if err != nil { + return err + } + + req.Header.Set("Authorization", "Token "+c.apiKey) + + resp, err := c.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("failed to fetch deepgram projects: status %d", resp.StatusCode) + } + + var result struct { + Projects []struct { + ProjectID string `json:"project_id"` + } `json:"projects"` + } + + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + + if len(result.Projects) == 0 { + return fmt.Errorf("no deepgram projects found for API key") + } + + // Use the first project + c.projectID = result.Projects[0].ProjectID + return nil +} diff --git a/minutely-api/internal/adapters/modal/client.go b/minutely-api/internal/adapters/modal/client.go new file mode 100644 index 0000000000..d487d670a0 --- /dev/null +++ b/minutely-api/internal/adapters/modal/client.go @@ -0,0 +1,113 @@ +package modal + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "mime/multipart" + "net/http" + "os" + "time" +) + +// TranscriptionResult matches the output expected from the Modal python endpoint +type TranscriptionResult struct { + Text string `json:"text"` + Language string `json:"language"` + Segments []DiarizedSegment `json:"segments"` + Info TranscriptionInfo `json:"info"` +} + +type DiarizedSegment struct { + Speaker string `json:"speaker"` + Text string `json:"text"` + Start float64 `json:"start"` + End float64 `json:"end"` +} + +type TranscriptionInfo struct { + DurationSecs float64 `json:"duration_secs"` + SpeakerCount int `json:"speaker_count"` +} + +type Client interface { + TranscribeAudio(ctx context.Context, audioFilePath string, language string) (*TranscriptionResult, error) +} + +type modalClient struct { + endpoint string + token string + client *http.Client +} + +func NewModalClient(endpoint string, token string) Client { + return &modalClient{ + endpoint: endpoint, + token: token, + client: &http.Client{ + Timeout: 10 * time.Minute, // Transcription can take several minutes + }, + } +} + +func (c *modalClient) TranscribeAudio(ctx context.Context, audioFilePath string, language string) (*TranscriptionResult, error) { + file, err := os.Open(audioFilePath) + if err != nil { + return nil, fmt.Errorf("failed to open audio file: %w", err) + } + defer file.Close() + + body := &bytes.Buffer{} + writer := multipart.NewWriter(body) + + // Add the file part + part, err := writer.CreateFormFile("file", "audio.wav") + if err != nil { + return nil, fmt.Errorf("failed to create form file: %w", err) + } + if _, err := io.Copy(part, file); err != nil { + return nil, fmt.Errorf("failed to copy file contents: %w", err) + } + + // Add language if specified + if language != "" { + if err := writer.WriteField("language", language); err != nil { + return nil, fmt.Errorf("failed to write language field: %w", err) + } + } + + err = writer.Close() + if err != nil { + return nil, fmt.Errorf("failed to close multipart writer: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.endpoint, body) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + req.Header.Set("Content-Type", writer.FormDataContentType()) + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to execute request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("modal API returned status %d: %s", resp.StatusCode, string(respBody)) + } + + var result TranscriptionResult + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return &result, nil +} diff --git a/minutely-api/internal/adapters/postgres/ai_output_repo.go b/minutely-api/internal/adapters/postgres/ai_output_repo.go new file mode 100644 index 0000000000..44598abf3c --- /dev/null +++ b/minutely-api/internal/adapters/postgres/ai_output_repo.go @@ -0,0 +1,78 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + + "github.com/google/uuid" + postgrest "github.com/supabase-community/postgrest-go" + "github.com/supabase-community/supabase-go" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type supabaseAIOutputRepo struct { + client *supabase.Client +} + +func NewSupabaseAIOutputRepo(client *supabase.Client) domain.AIOutputRepository { + return &supabaseAIOutputRepo{client: client} +} + +func (r *supabaseAIOutputRepo) Create(ctx context.Context, output *domain.AIOutput) error { + if output.ID == uuid.Nil { + output.ID = uuid.New() + } + + data, _, err := r.client.From("ai_outputs").Insert(output, false, "exact", "representation", "id").Execute() + if err != nil { + return err + } + + if len(data) > 0 { + var created []domain.AIOutput + if err := json.Unmarshal(data, &created); err == nil && len(created) > 0 { + *output = created[0] + } + } + + return nil +} + +func (r *supabaseAIOutputRepo) GetByID(ctx context.Context, id uuid.UUID) (*domain.AIOutput, error) { + data, _, err := r.client.From("ai_outputs").Select("*", "exact", false).Eq("id", id.String()).Single().Execute() + if err != nil { + return nil, err + } + + if len(data) == 0 { + return nil, errors.New("ai output not found") + } + + var output domain.AIOutput + if err := json.Unmarshal(data, &output); err != nil { + return nil, err + } + + return &output, nil +} + +func (r *supabaseAIOutputRepo) ListByMeetingID(ctx context.Context, meetingID uuid.UUID) ([]*domain.AIOutput, error) { + data, _, err := r.client.From("ai_outputs").Select("*", "exact", false).Eq("meeting_id", meetingID.String()).Order("created_at", &postgrest.OrderOpts{Ascending: false}).Execute() + if err != nil { + return nil, err + } + + var outputs []*domain.AIOutput + if err := json.Unmarshal(data, &outputs); err != nil { + return nil, err + } + + return outputs, nil +} + +func (r *supabaseAIOutputRepo) Update(ctx context.Context, output *domain.AIOutput) error { + _, _, err := r.client.From("ai_outputs").Update(output, "exact", "id").Eq("id", output.ID.String()).Execute() + return err +} diff --git a/minutely-api/internal/adapters/postgres/job_repo.go b/minutely-api/internal/adapters/postgres/job_repo.go new file mode 100644 index 0000000000..a34d68c931 --- /dev/null +++ b/minutely-api/internal/adapters/postgres/job_repo.go @@ -0,0 +1,134 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + + "github.com/google/uuid" + postgrest "github.com/supabase-community/postgrest-go" + "github.com/supabase-community/supabase-go" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type supabaseJobRepo struct { + client *supabase.Client +} + +func NewSupabaseJobRepo(client *supabase.Client) domain.JobRepository { + return &supabaseJobRepo{client: client} +} + +func (r *supabaseJobRepo) CreateJob(ctx context.Context, job *domain.ProcessingJob) error { + if job.ID == uuid.Nil { + job.ID = uuid.New() + } + + data, _, err := r.client.From("processing_jobs").Insert(job, false, "exact", "representation", "id").Execute() + if err != nil { + return err + } + + if len(data) > 0 { + var created []domain.ProcessingJob + if err := json.Unmarshal(data, &created); err == nil && len(created) > 0 { + *job = created[0] + } + } + + return nil +} + +func (r *supabaseJobRepo) GetJobByID(ctx context.Context, id uuid.UUID) (*domain.ProcessingJob, error) { + data, _, err := r.client.From("processing_jobs").Select("*", "exact", false).Eq("id", id.String()).Single().Execute() + if err != nil { + return nil, err + } + + if len(data) == 0 { + return nil, errors.New("job not found") + } + + var job domain.ProcessingJob + if err := json.Unmarshal(data, &job); err != nil { + return nil, err + } + + return &job, nil +} + +func (r *supabaseJobRepo) UpdateJob(ctx context.Context, job *domain.ProcessingJob) error { + _, _, err := r.client.From("processing_jobs").Update(job, "exact", "id").Eq("id", job.ID.String()).Execute() + return err +} + +func (r *supabaseJobRepo) ListPendingJobs(ctx context.Context, jobType domain.JobType, limit int) ([]*domain.ProcessingJob, error) { + query := r.client.From("processing_jobs").Select("*", "exact", false).Eq("status", string(domain.JobStatusPending)) + if jobType != "" { + query = query.Eq("job_type", string(jobType)) + } + + // Limit is not directly supported as an integer argument in some versions of the wrapper + // We'll use order to get oldest first. + // Since the postgrest-go client can be tricky, we'll try basic query. + data, _, err := query.Order("created_at", &postgrest.OrderOpts{Ascending: true}).Execute() + if err != nil { + return nil, err + } + + var jobs []*domain.ProcessingJob + if err := json.Unmarshal(data, &jobs); err != nil { + return nil, err + } + + // Manual limit if the query doesn't restrict it properly via the wrapper + if limit > 0 && len(jobs) > limit { + jobs = jobs[:limit] + } + + return jobs, nil +} + +func (r *supabaseJobRepo) CreateMediaFile(ctx context.Context, media *domain.MediaFile) error { + if media.ID == uuid.Nil { + media.ID = uuid.New() + } + + data, _, err := r.client.From("media_files").Insert(media, false, "exact", "representation", "id").Execute() + if err != nil { + return err + } + + if len(data) > 0 { + var created []domain.MediaFile + if err := json.Unmarshal(data, &created); err == nil && len(created) > 0 { + *media = created[0] + } + } + + return nil +} + +func (r *supabaseJobRepo) GetMediaFileByID(ctx context.Context, id uuid.UUID) (*domain.MediaFile, error) { + data, _, err := r.client.From("media_files").Select("*", "exact", false).Eq("id", id.String()).Single().Execute() + if err != nil { + return nil, err + } + + if len(data) == 0 { + return nil, errors.New("media file not found") + } + + var media domain.MediaFile + if err := json.Unmarshal(data, &media); err != nil { + return nil, err + } + + return &media, nil +} + +func (r *supabaseJobRepo) UpdateMediaFile(ctx context.Context, media *domain.MediaFile) error { + _, _, err := r.client.From("media_files").Update(media, "exact", "id").Eq("id", media.ID.String()).Execute() + return err +} diff --git a/minutely-api/internal/adapters/postgres/transcript_repo.go b/minutely-api/internal/adapters/postgres/transcript_repo.go new file mode 100644 index 0000000000..cac7247a2b --- /dev/null +++ b/minutely-api/internal/adapters/postgres/transcript_repo.go @@ -0,0 +1,174 @@ +package postgres + +import ( + "context" + "encoding/json" + "errors" + + "github.com/google/uuid" + postgrest "github.com/supabase-community/postgrest-go" + "github.com/supabase-community/supabase-go" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type supabaseTranscriptRepo struct { + client *supabase.Client +} + +func NewSupabaseTranscriptRepo(client *supabase.Client) domain.TranscriptRepository { + return &supabaseTranscriptRepo{client: client} +} + +func (r *supabaseTranscriptRepo) Create(ctx context.Context, transcript *domain.Transcript) error { + if transcript.ID == uuid.Nil { + transcript.ID = uuid.New() + } + + data, _, err := r.client.From("transcripts").Insert(transcript, false, "exact", "representation", "id").Execute() + if err != nil { + return err + } + + if len(data) > 0 { + var created []domain.Transcript + if err := json.Unmarshal(data, &created); err == nil && len(created) > 0 { + *transcript = created[0] + } + } + + return nil +} + +func (r *supabaseTranscriptRepo) GetByID(ctx context.Context, id uuid.UUID) (*domain.Transcript, error) { + data, _, err := r.client.From("transcripts").Select("*", "exact", false).Eq("id", id.String()).Single().Execute() + if err != nil { + return nil, err + } + + if len(data) == 0 { + return nil, errors.New("transcript not found") + } + + var transcript domain.Transcript + if err := json.Unmarshal(data, &transcript); err != nil { + return nil, err + } + + return &transcript, nil +} + +func (r *supabaseTranscriptRepo) GetByMeetingID(ctx context.Context, meetingID uuid.UUID) (*domain.Transcript, error) { + data, _, err := r.client.From("transcripts").Select("*", "exact", false).Eq("meeting_id", meetingID.String()).Order("created_at", &postgrest.OrderOpts{Ascending: false}).Limit(1, "").Single().Execute() + if err != nil { + return nil, err + } + + if len(data) == 0 { + return nil, errors.New("transcript not found") + } + + var transcript domain.Transcript + if err := json.Unmarshal(data, &transcript); err != nil { + return nil, err + } + + return &transcript, nil +} + +func (r *supabaseTranscriptRepo) Update(ctx context.Context, transcript *domain.Transcript) error { + _, _, err := r.client.From("transcripts").Update(transcript, "exact", "id").Eq("id", transcript.ID.String()).Execute() + return err +} + +func (r *supabaseTranscriptRepo) CreateSegment(ctx context.Context, segment *domain.TranscriptSegment) error { + if segment.ID == uuid.Nil { + segment.ID = uuid.New() + } + + data, _, err := r.client.From("transcript_segments").Insert(segment, false, "exact", "representation", "id").Execute() + if err != nil { + return err + } + + if len(data) > 0 { + var created []domain.TranscriptSegment + if err := json.Unmarshal(data, &created); err == nil && len(created) > 0 { + *segment = created[0] + } + } + + return nil +} + +func (r *supabaseTranscriptRepo) CreateSegmentsBatch(ctx context.Context, segments []*domain.TranscriptSegment) error { + if len(segments) == 0 { + return nil + } + + for _, s := range segments { + if s.ID == uuid.Nil { + s.ID = uuid.New() + } + } + + _, _, err := r.client.From("transcript_segments").Insert(segments, false, "exact", "representation", "id").Execute() + return err +} + +func (r *supabaseTranscriptRepo) ListSegments(ctx context.Context, transcriptID uuid.UUID) ([]*domain.TranscriptSegment, error) { + data, _, err := r.client.From("transcript_segments").Select("*", "exact", false).Eq("transcript_id", transcriptID.String()).Order("start_secs", &postgrest.OrderOpts{Ascending: true}).Execute() + if err != nil { + return nil, err + } + + var segments []*domain.TranscriptSegment + if err := json.Unmarshal(data, &segments); err != nil { + return nil, err + } + + return segments, nil +} + +func (r *supabaseTranscriptRepo) CreateLiveSession(ctx context.Context, session *domain.LiveSession) error { + if session.ID == uuid.Nil { + session.ID = uuid.New() + } + + data, _, err := r.client.From("live_sessions").Insert(session, false, "exact", "representation", "id").Execute() + if err != nil { + return err + } + + if len(data) > 0 { + var created []domain.LiveSession + if err := json.Unmarshal(data, &created); err == nil && len(created) > 0 { + *session = created[0] + } + } + + return nil +} + +func (r *supabaseTranscriptRepo) GetLiveSessionByMeetingID(ctx context.Context, meetingID uuid.UUID) (*domain.LiveSession, error) { + data, _, err := r.client.From("live_sessions").Select("*", "exact", false).Eq("meeting_id", meetingID.String()).Filter("ended_at", "is", "null").Order("started_at", &postgrest.OrderOpts{Ascending: false}).Limit(1, "").Single().Execute() + if err != nil { + return nil, err + } + + if len(data) == 0 { + return nil, errors.New("live session not found") + } + + var session domain.LiveSession + if err := json.Unmarshal(data, &session); err != nil { + return nil, err + } + + return &session, nil +} + +func (r *supabaseTranscriptRepo) UpdateLiveSession(ctx context.Context, session *domain.LiveSession) error { + _, _, err := r.client.From("live_sessions").Update(session, "exact", "id").Eq("id", session.ID.String()).Execute() + return err +} diff --git a/minutely-api/internal/adapters/pythonai/ai_processor.go b/minutely-api/internal/adapters/pythonai/ai_processor.go new file mode 100644 index 0000000000..d313e4b60e --- /dev/null +++ b/minutely-api/internal/adapters/pythonai/ai_processor.go @@ -0,0 +1,70 @@ +package pythonai + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "net/http" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type pythonAIProcessor struct { + endpointUrl string + aiOutputRepo domain.AIOutputRepository +} + +func NewPythonAIProcessor(endpointUrl string, aiOutputRepo domain.AIOutputRepository) domain.AIProcessor { + return &pythonAIProcessor{ + endpointUrl: endpointUrl, + aiOutputRepo: aiOutputRepo, + } +} + +func (p *pythonAIProcessor) ProcessTranscript(ctx context.Context, event domain.AIEvent) error { + payloadBytes, err := json.Marshal(event) + if err != nil { + return fmt.Errorf("failed to marshal AI event: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", p.endpointUrl+"/process", bytes.NewBuffer(payloadBytes)) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + client := &http.Client{} + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("failed to call python AI service: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("python AI service returned status: %d", resp.StatusCode) + } + + var result map[string]interface{} + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return fmt.Errorf("failed to decode python AI service response: %w", err) + } + + // Create a single comprehensive AIOutput record containing the full JSON payload + modelUsed := "bart-large-cnn/kmeans" + output := &domain.AIOutput{ + MeetingID: event.MeetingID, + TranscriptID: &event.TranscriptID, + OutputType: domain.AIOutputTypeSummary, + Status: domain.AIOutputStatusCompleted, + ModelUsed: &modelUsed, + Result: result, + } + + if err := p.aiOutputRepo.Create(ctx, output); err != nil { + return fmt.Errorf("failed to save AI output to db: %w", err) + } + + return nil +} diff --git a/minutely-api/internal/adapters/storage/supabase_storage.go b/minutely-api/internal/adapters/storage/supabase_storage.go new file mode 100644 index 0000000000..dddedfcd6d --- /dev/null +++ b/minutely-api/internal/adapters/storage/supabase_storage.go @@ -0,0 +1,52 @@ +package storage + +import ( + "context" + "io" + "bytes" + + storage_go "github.com/supabase-community/storage-go" + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type supabaseStorage struct { + client *storage_go.Client +} + +func NewSupabaseStorage(url, key string) domain.StorageService { + client := storage_go.NewClient(url, key, nil) + return &supabaseStorage{client: client} +} + +func (s *supabaseStorage) UploadFile(ctx context.Context, bucket string, path string, body io.Reader, contentType string) (string, error) { + // The storage-go client Upload method expects an io.Reader and returns the file path on success + // We might need to read it entirely if we don't know the size, or if the client requires it. + // We'll pass the body directly for stream-like upload if supported. + res, err := s.client.UploadFile(bucket, path, body) + if err != nil { + return "", err + } + // res.Key contains the uploaded path + return res.Key, nil +} + +func (s *supabaseStorage) DownloadFile(ctx context.Context, bucket string, path string) (io.ReadCloser, error) { + data, err := s.client.DownloadFile(bucket, path) + if err != nil { + return nil, err + } + return io.NopCloser(bytes.NewReader(data)), nil +} + +func (s *supabaseStorage) GetSignedURL(ctx context.Context, bucket string, path string, expiresIn int) (string, error) { + res, err := s.client.CreateSignedUrl(bucket, path, expiresIn) + if err != nil { + return "", err + } + return res.SignedURL, nil +} + +func (s *supabaseStorage) DeleteFile(ctx context.Context, bucket string, path string) error { + _, err := s.client.RemoveFile(bucket, []string{path}) + return err +} diff --git a/minutely-api/internal/core/domain/ai_output.go b/minutely-api/internal/core/domain/ai_output.go new file mode 100644 index 0000000000..a1fbe58e2b --- /dev/null +++ b/minutely-api/internal/core/domain/ai_output.go @@ -0,0 +1,80 @@ +package domain + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +type AIOutputType string + +const ( + AIOutputTypeSummary AIOutputType = "summary" + AIOutputTypeActionItems AIOutputType = "action_items" + AIOutputTypeKeyTopics AIOutputType = "key_topics" + AIOutputTypeSentiment AIOutputType = "sentiment" + AIOutputTypeTasks AIOutputType = "tasks" + AIOutputTypeCustom AIOutputType = "custom" +) + +type AIOutputStatus string + +const ( + AIOutputStatusPending AIOutputStatus = "pending" + AIOutputStatusProcessing AIOutputStatus = "processing" + AIOutputStatusCompleted AIOutputStatus = "completed" + AIOutputStatusFailed AIOutputStatus = "failed" +) + +// AIOutput represents the result of an LLM pipeline execution +type AIOutput struct { + ID uuid.UUID `json:"id"` + MeetingID uuid.UUID `json:"meeting_id"` + TranscriptID *uuid.UUID `json:"transcript_id"` + OutputType AIOutputType `json:"output_type"` + Status AIOutputStatus `json:"status"` + ModelUsed *string `json:"model_used"` + PromptVersion *string `json:"prompt_version"` + Result interface{} `json:"result"` // Mapped to jsonb + TokensUsed *int `json:"tokens_used"` + CostUSD *float64 `json:"cost_usd"` + ErrorMessage *string `json:"error_message"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at"` +} + +// AIEvent is emitted when a transcript completes successfully +type AIEvent struct { + MeetingID uuid.UUID `json:"meeting_id"` + TranscriptID uuid.UUID `json:"transcript_id"` + Segments []*TranscriptSegment `json:"segments"` + FullText string `json:"full_text"` + Participants []string `json:"participants"` + DurationSecs float64 `json:"duration_secs"` + Source string `json:"source"` // "live" or "upload" +} + +// AIProcessor is the single integration hook for all future LLM functionality +type AIProcessor interface { + // ProcessTranscript is called after transcription completes. + // Should return error only for infrastructure failures, not LLM context failures. + ProcessTranscript(ctx context.Context, event AIEvent) error +} + +// NoopAIProcessor is a stub implementation until the LLM pipeline is built +type NoopAIProcessor struct{} + +func (n *NoopAIProcessor) ProcessTranscript(ctx context.Context, event AIEvent) error { + // For now, this does nothing but satisfy the interface. + // Later, it will insert 'pending' AIOutput rows and trigger jobs. + return nil +} + +// AIOutputRepository defines datastore interactions for AI results +type AIOutputRepository interface { + Create(ctx context.Context, output *AIOutput) error + GetByID(ctx context.Context, id uuid.UUID) (*AIOutput, error) + ListByMeetingID(ctx context.Context, meetingID uuid.UUID) ([]*AIOutput, error) + Update(ctx context.Context, output *AIOutput) error +} diff --git a/minutely-api/internal/core/domain/job.go b/minutely-api/internal/core/domain/job.go new file mode 100644 index 0000000000..40931dc0a3 --- /dev/null +++ b/minutely-api/internal/core/domain/job.go @@ -0,0 +1,86 @@ +package domain + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +type JobType string + +const ( + JobTypeLiveTranscription JobType = "live_transcription" + JobTypeFileTranscription JobType = "file_transcription" + JobTypeAIProcessing JobType = "ai_processing" +) + +type JobStatus string + +const ( + JobStatusPending JobStatus = "pending" + JobStatusProcessing JobStatus = "processing" + JobStatusCompleted JobStatus = "completed" + JobStatusFailed JobStatus = "failed" + JobStatusRetrying JobStatus = "retrying" +) + +type MediaStatus string + +const ( + MediaStatusUploaded MediaStatus = "uploaded" + MediaStatusProcessing MediaStatus = "processing" + MediaStatusReady MediaStatus = "ready" + MediaStatusError MediaStatus = "error" +) + +// ProcessingJob represents an async background task +type ProcessingJob struct { + ID uuid.UUID `json:"id"` + MeetingID uuid.UUID `json:"meeting_id"` + JobType JobType `json:"job_type"` + Status JobStatus `json:"status"` + AttemptCount int `json:"attempt_count"` + MaxAttempts int `json:"max_attempts"` + ErrorMessage *string `json:"error_message"` + Payload interface{} `json:"payload"` // Typically mapped to map[string]interface{} + Result interface{} `json:"result"` + StartedAt *time.Time `json:"started_at"` + CompletedAt *time.Time `json:"completed_at"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// MediaFile represents an uploaded recording +type MediaFile struct { + ID uuid.UUID `json:"id"` + MeetingID uuid.UUID `json:"meeting_id"` + UploadedBy *uuid.UUID `json:"uploaded_by"` + StoragePath string `json:"storage_path"` + OriginalName string `json:"original_name"` + MimeType string `json:"mime_type"` + SizeBytes *int64 `json:"size_bytes"` + DurationSecs *float64 `json:"duration_secs"` + Status MediaStatus `json:"status"` + JobID *uuid.UUID `json:"job_id"` + CreatedAt time.Time `json:"created_at"` +} + +// JobRepository defines datastore interactions for jobs +type JobRepository interface { + CreateJob(ctx context.Context, job *ProcessingJob) error + GetJobByID(ctx context.Context, id uuid.UUID) (*ProcessingJob, error) + UpdateJob(ctx context.Context, job *ProcessingJob) error + ListPendingJobs(ctx context.Context, jobType JobType, limit int) ([]*ProcessingJob, error) + + CreateMediaFile(ctx context.Context, media *MediaFile) error + GetMediaFileByID(ctx context.Context, id uuid.UUID) (*MediaFile, error) + UpdateMediaFile(ctx context.Context, media *MediaFile) error +} + +// JobService defines business logic for queueing and managing async jobs +type JobService interface { + EnqueueTranscriptionJob(ctx context.Context, meetingID uuid.UUID, mediaFileID uuid.UUID) (*ProcessingJob, error) + GetJobStatus(ctx context.Context, jobID uuid.UUID) (*ProcessingJob, error) + ProcessNextJob(ctx context.Context) error // For worker loop +} diff --git a/minutely-api/internal/core/domain/storage.go b/minutely-api/internal/core/domain/storage.go new file mode 100644 index 0000000000..31547571db --- /dev/null +++ b/minutely-api/internal/core/domain/storage.go @@ -0,0 +1,21 @@ +package domain + +import ( + "context" + "io" +) + +// StorageService defines operations for cloud file storage +type StorageService interface { + // UploadFile uploads an io.Reader to the storage bucket and returns the generated path + UploadFile(ctx context.Context, bucket string, path string, body io.Reader, contentType string) (string, error) + + // DownloadFile downloads a file from the storage bucket + DownloadFile(ctx context.Context, bucket string, path string) (io.ReadCloser, error) + + // GetSignedURL generates a temporary signed URL for a file + GetSignedURL(ctx context.Context, bucket string, path string, expiresIn int) (string, error) + + // DeleteFile removes a file from the storage bucket + DeleteFile(ctx context.Context, bucket string, path string) error +} diff --git a/minutely-api/internal/core/domain/transcript.go b/minutely-api/internal/core/domain/transcript.go new file mode 100644 index 0000000000..651e26c480 --- /dev/null +++ b/minutely-api/internal/core/domain/transcript.go @@ -0,0 +1,88 @@ +package domain + +import ( + "context" + "time" + + "github.com/google/uuid" +) + +type TranscriptSource string + +const ( + SourceLive TranscriptSource = "live" + SourceUpload TranscriptSource = "upload" +) + +type TranscriptStatus string + +const ( + TranscriptStatusInProgress TranscriptStatus = "in_progress" + TranscriptStatusCompleted TranscriptStatus = "completed" + TranscriptStatusFailed TranscriptStatus = "failed" +) + +// Transcript represents the full transcript of a meeting +type Transcript struct { + ID uuid.UUID `json:"id"` + MeetingID uuid.UUID `json:"meeting_id"` + MediaFileID *uuid.UUID `json:"media_file_id"` + Source TranscriptSource `json:"source"` + Language string `json:"language"` + FullText *string `json:"full_text"` + DurationSecs *float64 `json:"duration_secs"` + SpeakerCount *int `json:"speaker_count"` + Status TranscriptStatus `json:"status"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at"` +} + +// TranscriptSegment represents a diarized speech segment +type TranscriptSegment struct { + ID uuid.UUID `json:"id"` + TranscriptID uuid.UUID `json:"transcript_id"` + SpeakerLabel *string `json:"speaker_label"` + SpeakerName string `json:"speaker_name"` + SpeakerEmail string `json:"speaker_email"` + Text string `json:"text"` + StartSecs float64 `json:"start_secs"` + EndSecs float64 `json:"end_secs"` + Confidence *float64 `json:"confidence"` + IsPartial bool `json:"is_partial"` + SequenceNum *int `json:"sequence_num"` + CreatedAt time.Time `json:"created_at"` +} + +// LiveSession represents an active live transcription session +type LiveSession struct { + ID uuid.UUID `json:"id"` + MeetingID uuid.UUID `json:"meeting_id"` + TranscriptID *uuid.UUID `json:"transcript_id"` + DeepgramTokenExpires *time.Time `json:"deepgram_token_expires"` + StartedAt time.Time `json:"started_at"` + EndedAt *time.Time `json:"ended_at"` + ParticipantCount int `json:"participant_count"` +} + +// TranscriptRepository defines datastore interactions for transcription +type TranscriptRepository interface { + Create(ctx context.Context, transcript *Transcript) error + GetByID(ctx context.Context, id uuid.UUID) (*Transcript, error) + GetByMeetingID(ctx context.Context, meetingID uuid.UUID) (*Transcript, error) + Update(ctx context.Context, transcript *Transcript) error + + CreateSegment(ctx context.Context, segment *TranscriptSegment) error + CreateSegmentsBatch(ctx context.Context, segments []*TranscriptSegment) error + ListSegments(ctx context.Context, transcriptID uuid.UUID) ([]*TranscriptSegment, error) + + CreateLiveSession(ctx context.Context, session *LiveSession) error + GetLiveSessionByMeetingID(ctx context.Context, meetingID uuid.UUID) (*LiveSession, error) + UpdateLiveSession(ctx context.Context, session *LiveSession) error +} + +// TranscriptionService defines business logic for transcription +type TranscriptionService interface { + StartLiveSession(ctx context.Context, meetingID uuid.UUID) (*LiveSession, string, error) // returns session and WS url + EndLiveSession(ctx context.Context, meetingID uuid.UUID) error + GetMeetingTranscript(ctx context.Context, meetingID uuid.UUID) (*Transcript, []*TranscriptSegment, error) +} diff --git a/minutely-api/internal/core/services/transcription_service.go b/minutely-api/internal/core/services/transcription_service.go new file mode 100644 index 0000000000..ba366bc554 --- /dev/null +++ b/minutely-api/internal/core/services/transcription_service.go @@ -0,0 +1,123 @@ +package services + +import ( + "context" + "fmt" + "time" + + "github.com/google/uuid" + + "github.com/MinutelyAI/minutely-api/internal/adapters/deepgram" + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type transcriptionService struct { + repo domain.TranscriptRepository + jobRepo domain.JobRepository + deepgram deepgram.Client +} + +func NewTranscriptionService(repo domain.TranscriptRepository, jobRepo domain.JobRepository, dgClient deepgram.Client) domain.TranscriptionService { + return &transcriptionService{ + repo: repo, + jobRepo: jobRepo, + deepgram: dgClient, + } +} + +func (s *transcriptionService) StartLiveSession(ctx context.Context, meetingID uuid.UUID) (*domain.LiveSession, string, error) { + // 1. Check if an active session already exists + session, err := s.repo.GetLiveSessionByMeetingID(ctx, meetingID) + + // 2. Generate a temporary Deepgram API key (valid for 4 hours) + tokenExpiresIn := 4 * time.Hour + tempKey, errDG := s.deepgram.GenerateTempKey(ctx, tokenExpiresIn) + if errDG != nil { + return nil, "", fmt.Errorf("failed to generate deepgram key: %w", errDG) + } + + if err == nil && session != nil { + // Active session found, just return it with the new key + return session, tempKey, nil + } + + // 3. Create the Transcript record + transcript := &domain.Transcript{ + MeetingID: meetingID, + Source: domain.SourceLive, + Status: domain.TranscriptStatusInProgress, + } + if err := s.repo.Create(ctx, transcript); err != nil { + return nil, "", fmt.Errorf("failed to create transcript: %w", err) + } + + // 4. Create the LiveSession record + expiresAt := time.Now().Add(tokenExpiresIn) + newSession := &domain.LiveSession{ + MeetingID: meetingID, + TranscriptID: &transcript.ID, + DeepgramTokenExpires: &expiresAt, + StartedAt: time.Now(), + ParticipantCount: 0, + } + if err := s.repo.CreateLiveSession(ctx, newSession); err != nil { + return nil, "", fmt.Errorf("failed to create live session: %w", err) + } + + // Return the new session and the temporary key + return newSession, tempKey, nil +} + +func (s *transcriptionService) EndLiveSession(ctx context.Context, meetingID uuid.UUID) error { + session, err := s.repo.GetLiveSessionByMeetingID(ctx, meetingID) + if err != nil { + return fmt.Errorf("no active live session found for meeting: %w", err) + } + + // Mark session as ended + now := time.Now() + session.EndedAt = &now + if err := s.repo.UpdateLiveSession(ctx, session); err != nil { + return fmt.Errorf("failed to update live session: %w", err) + } + + // Mark transcript as completed + if session.TranscriptID != nil { + transcript, err := s.repo.GetByID(ctx, *session.TranscriptID) + if err == nil { + transcript.Status = domain.TranscriptStatusCompleted + transcript.CompletedAt = &now + _ = s.repo.Update(ctx, transcript) + } + } + + // Enqueue AI processing job + if session.TranscriptID != nil { + job := &domain.ProcessingJob{ + MeetingID: meetingID, + JobType: domain.JobTypeAIProcessing, + Status: domain.JobStatusPending, + MaxAttempts: 3, + Payload: map[string]string{ + "transcript_id": session.TranscriptID.String(), + }, + } + _ = s.jobRepo.CreateJob(ctx, job) + } + + return nil +} + +func (s *transcriptionService) GetMeetingTranscript(ctx context.Context, meetingID uuid.UUID) (*domain.Transcript, []*domain.TranscriptSegment, error) { + transcript, err := s.repo.GetByMeetingID(ctx, meetingID) + if err != nil { + return nil, nil, err + } + + segments, err := s.repo.ListSegments(ctx, transcript.ID) + if err != nil { + return nil, nil, err + } + + return transcript, segments, nil +} diff --git a/minutely-api/internal/transport/http/ai_handler.go b/minutely-api/internal/transport/http/ai_handler.go new file mode 100644 index 0000000000..d0f1298590 --- /dev/null +++ b/minutely-api/internal/transport/http/ai_handler.go @@ -0,0 +1,41 @@ +package http + +import ( + "encoding/json" + "net/http" + + "github.com/go-chi/chi/v5" + "github.com/google/uuid" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +type AIHandler struct { + aiOutputRepo domain.AIOutputRepository +} + +func NewAIHandler(aiOutputRepo domain.AIOutputRepository) *AIHandler { + return &AIHandler{aiOutputRepo: aiOutputRepo} +} + +// GetMeetingInsights returns the AI insights for a given meeting +// GET /api/v1/meetings/{meetingId}/ai-insights +func (h *AIHandler) GetMeetingInsights(w http.ResponseWriter, r *http.Request) { + meetingIDStr := chi.URLParam(r, "meetingId") + meetingID, err := uuid.Parse(meetingIDStr) + if err != nil { + // Bypassing strict UUID check for Jitsi dev prototype (just like we did for transcription) + meetingID = uuid.NewMD5(uuid.NameSpaceURL, []byte(meetingIDStr)) + } + + outputs, err := h.aiOutputRepo.ListByMeetingID(r.Context(), meetingID) + if err != nil { + jsonError(w, "failed to fetch ai insights", http.StatusInternalServerError) + return + } + + // We'll return the most recent output of each type, or just the list + // Currently the UI will likely look for the first one, but let's return the full list. + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(outputs) +} diff --git a/minutely-api/internal/transport/http/live_transcription_handler.go b/minutely-api/internal/transport/http/live_transcription_handler.go new file mode 100644 index 0000000000..6c21a52f82 --- /dev/null +++ b/minutely-api/internal/transport/http/live_transcription_handler.go @@ -0,0 +1,93 @@ +package http + +import ( + "encoding/json" + "fmt" + "net/http" + "time" + + "github.com/go-chi/chi/v5" + "github.com/google/uuid" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" + "github.com/MinutelyAI/minutely-api/internal/transport/ws" +) + +type LiveTranscriptionHandler struct { + transcriptionService domain.TranscriptionService + meetingService domain.MeetingService + wsHub *ws.Hub +} + +func NewLiveTranscriptionHandler(ts domain.TranscriptionService, ms domain.MeetingService, hub *ws.Hub) *LiveTranscriptionHandler { + return &LiveTranscriptionHandler{ + transcriptionService: ts, + meetingService: ms, + wsHub: hub, + } +} + +func (h *LiveTranscriptionHandler) parseOrGenerateMeetingID(r *http.Request) uuid.UUID { + meetingIDStr := chi.URLParam(r, "meetingId") + meetingID, err := uuid.Parse(meetingIDStr) + if err != nil { + // It's a Jitsi string name, generate a deterministic UUID + meetingID = uuid.NewMD5(uuid.NameSpaceURL, []byte(meetingIDStr)) + // Lazily create a dummy meeting if it doesn't exist so foreign keys don't fail + _, _ = h.meetingService.GetMeeting(r.Context(), meetingID) + err = h.meetingService.CreateMeeting(r.Context(), &domain.Meeting{ + ID: meetingID, + Title: meetingIDStr, + Status: domain.MeetingStatusInProgress, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + }) + if err != nil { + fmt.Printf("Warning: failed to create dummy meeting: %v\n", err) + } + } + return meetingID +} + +// StartSession initiates a live transcription session and returns a Deepgram key +// POST /api/v1/meetings/{meetingId}/transcription/start +func (h *LiveTranscriptionHandler) StartSession(w http.ResponseWriter, r *http.Request) { + meetingID := h.parseOrGenerateMeetingID(r) + + session, token, err := h.transcriptionService.StartLiveSession(r.Context(), meetingID) + if err != nil { + jsonError(w, err.Error(), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]interface{}{ + "session_id": session.ID, + "deepgram_token": token, + "deepgram_token_expires": session.DeepgramTokenExpires, + }) +} + +// EndSession marks a live transcription session as ended +// POST /api/v1/meetings/{meetingId}/transcription/end +func (h *LiveTranscriptionHandler) EndSession(w http.ResponseWriter, r *http.Request) { + meetingID := h.parseOrGenerateMeetingID(r) + + if err := h.transcriptionService.EndLiveSession(r.Context(), meetingID); err != nil { + jsonError(w, err.Error(), http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]string{"status": "ended"}) +} + +// HandleWebSocket upgrades the HTTP request to a WebSocket connection for the meeting +// GET /api/v1/meetings/{meetingId}/transcription/ws +func (h *LiveTranscriptionHandler) HandleWebSocket(w http.ResponseWriter, r *http.Request) { + meetingID := h.parseOrGenerateMeetingID(r) + + // Upgrade the connection and register client with the hub + ws.ServeWs(h.wsHub, meetingID, w, r) +} diff --git a/minutely-api/internal/transport/http/transcription_handler.go b/minutely-api/internal/transport/http/transcription_handler.go new file mode 100644 index 0000000000..fbafe81a66 --- /dev/null +++ b/minutely-api/internal/transport/http/transcription_handler.go @@ -0,0 +1,278 @@ +package http + +import ( + "encoding/json" + "fmt" + "net/http" + "time" + + "github.com/go-chi/chi/v5" + "github.com/google/uuid" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" + "github.com/MinutelyAI/minutely-api/internal/transport/http/middleware" +) + +// TranscriptionHandler handles all transcription-related HTTP endpoints +type TranscriptionHandler struct { + transcriptRepo domain.TranscriptRepository + jobRepo domain.JobRepository + storage domain.StorageService + meetingService domain.MeetingService + storageBucket string +} + +func NewTranscriptionHandler( + transcriptRepo domain.TranscriptRepository, + jobRepo domain.JobRepository, + storage domain.StorageService, + meetingService domain.MeetingService, + storageBucket string, +) *TranscriptionHandler { + return &TranscriptionHandler{ + transcriptRepo: transcriptRepo, + jobRepo: jobRepo, + storage: storage, + meetingService: meetingService, + storageBucket: storageBucket, + } +} + +func (h *TranscriptionHandler) parseOrGenerateMeetingID(r *http.Request) uuid.UUID { + meetingIDStr := chi.URLParam(r, "meetingId") + meetingID, err := uuid.Parse(meetingIDStr) + if err != nil { + meetingID = uuid.NewMD5(uuid.NameSpaceURL, []byte(meetingIDStr)) + _, _ = h.meetingService.GetMeeting(r.Context(), meetingID) + err = h.meetingService.CreateMeeting(r.Context(), &domain.Meeting{ + ID: meetingID, + Title: meetingIDStr, + Status: domain.MeetingStatusInProgress, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + }) + if err != nil { + fmt.Printf("Warning: failed to create dummy meeting: %v\n", err) + } + } + return meetingID +} + +// UploadRecording accepts a multipart file upload, stores it, and enqueues a transcription job. +// POST /api/v1/meetings/{meetingId}/recordings/upload +func (h *TranscriptionHandler) UploadRecording(w http.ResponseWriter, r *http.Request) { + _ = r.Context().Value(middleware.UserIDKey).(uuid.UUID) + + meetingID := h.parseOrGenerateMeetingID(r) + + // Max 2GB upload + r.Body = http.MaxBytesReader(w, r.Body, 2<<30) + if err := r.ParseMultipartForm(32 << 20); err != nil { + jsonError(w, "failed to parse multipart form", http.StatusBadRequest) + return + } + + file, header, err := r.FormFile("file") + if err != nil { + jsonError(w, "file field is required", http.StatusBadRequest) + return + } + defer file.Close() + + language := r.FormValue("language") + if language == "" { + language = "en" + } + + // Build storage path: meetings/{meetingId}/recordings/{uuid}_{filename} + fileID := uuid.New() + storagePath := fmt.Sprintf("meetings/%s/recordings/%s_%s", meetingID, fileID, header.Filename) + contentType := header.Header.Get("Content-Type") + if contentType == "" { + contentType = "application/octet-stream" + } + + // Upload to Supabase Storage + uploadedPath, err := h.storage.UploadFile(r.Context(), h.storageBucket, storagePath, file, contentType) + if err != nil { + jsonError(w, "failed to upload file: "+err.Error(), http.StatusInternalServerError) + return + } + + // Create media_file record + size := header.Size + mediaFile := &domain.MediaFile{ + MeetingID: meetingID, + StoragePath: uploadedPath, + OriginalName: header.Filename, + MimeType: contentType, + SizeBytes: &size, + Status: domain.MediaStatusUploaded, + } + if err := h.jobRepo.CreateMediaFile(r.Context(), mediaFile); err != nil { + jsonError(w, "failed to record media file: "+err.Error(), http.StatusInternalServerError) + return + } + + // Create processing job + job := &domain.ProcessingJob{ + MeetingID: meetingID, + JobType: domain.JobTypeFileTranscription, + Status: domain.JobStatusPending, + MaxAttempts: 3, + Payload: map[string]string{ + "media_file_id": mediaFile.ID.String(), + "language": language, + }, + } + if err := h.jobRepo.CreateJob(r.Context(), job); err != nil { + jsonError(w, "failed to enqueue job: "+err.Error(), http.StatusInternalServerError) + return + } + + // Update media file to link back to job + mediaFile.JobID = &job.ID + _ = h.jobRepo.UpdateMediaFile(r.Context(), mediaFile) + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + json.NewEncoder(w).Encode(map[string]interface{}{ + "file_id": mediaFile.ID, + "job_id": job.ID, + "status": job.Status, + }) +} + +// GetJobStatus returns the current status of a processing job. +// GET /api/v1/jobs/{jobId} +func (h *TranscriptionHandler) GetJobStatus(w http.ResponseWriter, r *http.Request) { + jobIDStr := chi.URLParam(r, "jobId") + jobID, err := uuid.Parse(jobIDStr) + if err != nil { + jsonError(w, "invalid job id", http.StatusBadRequest) + return + } + + job, err := h.jobRepo.GetJobByID(r.Context(), jobID) + if err != nil { + jsonError(w, "job not found", http.StatusNotFound) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "job_id": job.ID, + "status": job.Status, + "attempt": job.AttemptCount, + "error": job.ErrorMessage, + "started_at": job.StartedAt, + "completed_at": job.CompletedAt, + "result": job.Result, + }) +} + +// GetJobProgress streams SSE progress events for a job. +// GET /api/v1/jobs/{jobId}/progress +func (h *TranscriptionHandler) GetJobProgress(w http.ResponseWriter, r *http.Request) { + jobIDStr := chi.URLParam(r, "jobId") + jobID, err := uuid.Parse(jobIDStr) + if err != nil { + jsonError(w, "invalid job id", http.StatusBadRequest) + return + } + + // Set SSE headers + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") + + flusher, ok := w.(http.Flusher) + if !ok { + jsonError(w, "streaming not supported", http.StatusInternalServerError) + return + } + + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + + for { + select { + case <-r.Context().Done(): + return + case <-ticker.C: + job, err := h.jobRepo.GetJobByID(r.Context(), jobID) + if err != nil { + fmt.Fprintf(w, "data: {\"error\": \"job not found\"}\n\n") + flusher.Flush() + return + } + + eventData, _ := json.Marshal(map[string]interface{}{ + "status": job.Status, + "attempt": job.AttemptCount, + "error": job.ErrorMessage, + "completed_at": job.CompletedAt, + "result": job.Result, + }) + fmt.Fprintf(w, "data: %s\n\n", eventData) + flusher.Flush() + + // Stop streaming once job is done + if job.Status == domain.JobStatusCompleted || job.Status == domain.JobStatusFailed { + return + } + } + } +} + +// GetMeetingTranscript returns the full transcript JSON for a meeting. +// GET /api/v1/meetings/{meetingId}/transcript +func (h *TranscriptionHandler) GetMeetingTranscript(w http.ResponseWriter, r *http.Request) { + meetingID := h.parseOrGenerateMeetingID(r) + + transcript, err := h.transcriptRepo.GetByMeetingID(r.Context(), meetingID) + if err != nil { + jsonError(w, "transcript not found", http.StatusNotFound) + return + } + + segments, err := h.transcriptRepo.ListSegments(r.Context(), transcript.ID) + if err != nil { + jsonError(w, "failed to fetch segments: "+err.Error(), http.StatusInternalServerError) + return + } + + // Build unique participant list + participants := []string{} + seen := map[string]bool{} + for _, seg := range segments { + if !seen[seg.SpeakerEmail] { + participants = append(participants, seg.SpeakerName+" <"+seg.SpeakerEmail+">") + seen[seg.SpeakerEmail] = true + } + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "transcript_id": transcript.ID, + "meeting_id": transcript.MeetingID, + "source": transcript.Source, + "language": transcript.Language, + "duration_secs": transcript.DurationSecs, + "speaker_count": transcript.SpeakerCount, + "full_text": transcript.FullText, + "segments": segments, + "participants": participants, + "status": transcript.Status, + "completed_at": transcript.CompletedAt, + }) +} + +// jsonError writes a standard JSON error response +func jsonError(w http.ResponseWriter, msg string, code int) { + fmt.Printf("API ERROR (HTTP %d): %s\n", code, msg) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + json.NewEncoder(w).Encode(map[string]string{"error": msg}) +} diff --git a/minutely-api/internal/transport/ws/client.go b/minutely-api/internal/transport/ws/client.go new file mode 100644 index 0000000000..7986d18e49 --- /dev/null +++ b/minutely-api/internal/transport/ws/client.go @@ -0,0 +1,123 @@ +package ws + +import ( + "encoding/json" + "log" + "net/http" + "time" + + "github.com/google/uuid" + "github.com/gorilla/websocket" +) + +const ( + writeWait = 10 * time.Second + pongWait = 60 * time.Second + pingPeriod = (pongWait * 9) / 10 + maxMessageSize = 4096 +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, + CheckOrigin: func(r *http.Request) bool { + return true // Allow all origins for now; restrict in production + }, +} + +// Client is a middleman between the websocket connection and the hub. +type Client struct { + Hub *Hub + MeetingID uuid.UUID + Conn *websocket.Conn + send chan *SegmentMessage +} + +// readPump pumps messages from the websocket connection to the hub. +func (c *Client) readPump() { + defer func() { + c.Hub.unregister <- c + c.Conn.Close() + }() + c.Conn.SetReadLimit(maxMessageSize) + c.Conn.SetReadDeadline(time.Now().Add(pongWait)) + c.Conn.SetPongHandler(func(string) error { c.Conn.SetReadDeadline(time.Now().Add(pongWait)); return nil }) + for { + _, message, err := c.Conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + log.Printf("error: %v", err) + } + break + } + + var segment SegmentMessage + if err := json.Unmarshal(message, &segment); err != nil { + log.Printf("error parsing segment message: %v", err) + continue + } + + // Ensure the segment is mapped to this client's meeting + segment.MeetingID = c.MeetingID + + c.Hub.broadcast <- &segment + } +} + +// writePump pumps messages from the hub to the websocket connection. +func (c *Client) writePump() { + ticker := time.NewTicker(pingPeriod) + defer func() { + ticker.Stop() + c.Conn.Close() + }() + for { + select { + case message, ok := <-c.send: + c.Conn.SetWriteDeadline(time.Now().Add(writeWait)) + if !ok { + // The hub closed the channel. + c.Conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + w, err := c.Conn.NextWriter(websocket.TextMessage) + if err != nil { + return + } + + jsonMsg, _ := json.Marshal(message) + w.Write(jsonMsg) + + if err := w.Close(); err != nil { + return + } + case <-ticker.C: + c.Conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} + +// ServeWs handles websocket requests from the peer. +func ServeWs(hub *Hub, meetingID uuid.UUID, w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + log.Println(err) + return + } + client := &Client{ + Hub: hub, + MeetingID: meetingID, + Conn: conn, + send: make(chan *SegmentMessage, 256), + } + client.Hub.register <- client + + // Allow collection of memory referenced by the caller by doing all work in + // new goroutines. + go client.writePump() + go client.readPump() +} diff --git a/minutely-api/internal/transport/ws/hub.go b/minutely-api/internal/transport/ws/hub.go new file mode 100644 index 0000000000..10baa4b153 --- /dev/null +++ b/minutely-api/internal/transport/ws/hub.go @@ -0,0 +1,130 @@ +package ws + +import ( + "context" + "log" + "sync" + "time" + + "github.com/google/uuid" + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +// SegmentMessage represents a partial or final transcript segment +// received from a client during a live meeting. +type SegmentMessage struct { + MeetingID uuid.UUID `json:"-"` + MeetingName string `json:"meeting_id"` + SpeakerName string `json:"speaker_name"` + SpeakerEmail string `json:"speaker_email"` + Text string `json:"text"` + StartSecs float64 `json:"start_secs"` + EndSecs float64 `json:"end_secs"` + IsPartial bool `json:"is_partial"` +} + +// Hub maintains the set of active clients and broadcasts messages to the clients. +type Hub struct { + // Registered clients mapped by meeting ID + clients map[uuid.UUID]map[*Client]bool + + // Inbound messages from the clients. + broadcast chan *SegmentMessage + + // Register requests from the clients. + register chan *Client + + // Unregister requests from clients. + unregister chan *Client + + mu sync.RWMutex + + transcriptRepo domain.TranscriptRepository +} + +func NewHub(transcriptRepo domain.TranscriptRepository) *Hub { + return &Hub{ + broadcast: make(chan *SegmentMessage), + register: make(chan *Client), + unregister: make(chan *Client), + clients: make(map[uuid.UUID]map[*Client]bool), + transcriptRepo: transcriptRepo, + } +} + +func (h *Hub) Run(ctx context.Context) { + log.Println("WebSocket Hub started") + for { + select { + case <-ctx.Done(): + log.Println("WebSocket Hub shutting down") + return + case client := <-h.register: + h.mu.Lock() + if _, ok := h.clients[client.MeetingID]; !ok { + h.clients[client.MeetingID] = make(map[*Client]bool) + } + h.clients[client.MeetingID][client] = true + h.mu.Unlock() + log.Printf("Client registered to meeting %s. Total clients in meeting: %d", client.MeetingID, len(h.clients[client.MeetingID])) + + case client := <-h.unregister: + h.mu.Lock() + if _, ok := h.clients[client.MeetingID]; ok { + if _, ok := h.clients[client.MeetingID][client]; ok { + delete(h.clients[client.MeetingID], client) + close(client.send) + if len(h.clients[client.MeetingID]) == 0 { + delete(h.clients, client.MeetingID) + } + } + } + h.mu.Unlock() + log.Printf("Client unregistered from meeting %s", client.MeetingID) + + case message := <-h.broadcast: + // 1. Broadcast to all other clients in the same meeting + h.mu.RLock() + for client := range h.clients[message.MeetingID] { + select { + case client.send <- message: + default: + close(client.send) + delete(h.clients[message.MeetingID], client) + } + } + h.mu.RUnlock() + + // 2. Persist to DB if it's a final segment + if !message.IsPartial { + h.persistSegment(message) + } + } + } +} + +func (h *Hub) persistSegment(msg *SegmentMessage) { + // First, we need to find the active TranscriptID for this meeting + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + session, err := h.transcriptRepo.GetLiveSessionByMeetingID(ctx, msg.MeetingID) + if err != nil || session.TranscriptID == nil { + log.Printf("Failed to find active live session for meeting %s: %v", msg.MeetingID, err) + return + } + + segment := &domain.TranscriptSegment{ + TranscriptID: *session.TranscriptID, + SpeakerName: msg.SpeakerName, + SpeakerEmail: msg.SpeakerEmail, + Text: msg.Text, + StartSecs: msg.StartSecs, + EndSecs: msg.EndSecs, + IsPartial: false, + } + + if err := h.transcriptRepo.CreateSegment(ctx, segment); err != nil { + log.Printf("Failed to persist segment for meeting %s: %v", msg.MeetingID, err) + } +} diff --git a/minutely-api/internal/workers/file_processor.go b/minutely-api/internal/workers/file_processor.go new file mode 100644 index 0000000000..d490604334 --- /dev/null +++ b/minutely-api/internal/workers/file_processor.go @@ -0,0 +1,320 @@ +package workers + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/MinutelyAI/minutely-api/internal/adapters/audio" + "github.com/MinutelyAI/minutely-api/internal/adapters/modal" + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +// FileProcessor handles the full lifecycle of an uploaded recording: +// 1. Download from Supabase Storage +// 2. Extract audio via ffmpeg +// 3. Send to Modal (Whisper + pyannote) +// 4. Parse and persist transcript segments +// 5. Fire AI pipeline hook +type FileProcessor struct { + transcriptRepo domain.TranscriptRepository + jobRepo domain.JobRepository + storage domain.StorageService + extractor audio.AudioExtractor + modal modal.Client + aiProcessor domain.AIProcessor + storageBucket string + tempDir string +} + +func NewFileProcessor( + transcriptRepo domain.TranscriptRepository, + jobRepo domain.JobRepository, + storage domain.StorageService, + extractor audio.AudioExtractor, + modal modal.Client, + aiProcessor domain.AIProcessor, + storageBucket string, + tempDir string, +) *FileProcessor { + return &FileProcessor{ + transcriptRepo: transcriptRepo, + jobRepo: jobRepo, + storage: storage, + extractor: extractor, + modal: modal, + aiProcessor: aiProcessor, + storageBucket: storageBucket, + tempDir: tempDir, + } +} + +// Process executes the full transcription pipeline for a given job +func (p *FileProcessor) Process(ctx context.Context, job *domain.ProcessingJob) error { + // Mark job as processing + now := time.Now() + job.Status = domain.JobStatusProcessing + job.StartedAt = &now + job.AttemptCount++ + if err := p.jobRepo.UpdateJob(ctx, job); err != nil { + return fmt.Errorf("failed to mark job processing: %w", err) + } + + var err error + if job.JobType == domain.JobTypeAIProcessing { + err = p.processAIJob(ctx, job) + } else { + err = p.processJob(ctx, job) + } + if err != nil { + errMsg := err.Error() + job.ErrorMessage = &errMsg + if job.AttemptCount >= job.MaxAttempts { + job.Status = domain.JobStatusFailed + } else { + job.Status = domain.JobStatusRetrying + } + _ = p.jobRepo.UpdateJob(ctx, job) + return err + } + + completedAt := time.Now() + job.Status = domain.JobStatusCompleted + job.CompletedAt = &completedAt + return p.jobRepo.UpdateJob(ctx, job) +} + +func (p *FileProcessor) processJob(ctx context.Context, job *domain.ProcessingJob) error { + // Parse payload to get media_file_id and language + payloadBytes, err := json.Marshal(job.Payload) + if err != nil { + return fmt.Errorf("failed to marshal payload: %w", err) + } + var payload struct { + MediaFileID string `json:"media_file_id"` + Language string `json:"language"` + } + if err := json.Unmarshal(payloadBytes, &payload); err != nil { + return fmt.Errorf("failed to parse job payload: %w", err) + } + + mediaFileID, err := uuid.Parse(payload.MediaFileID) + if err != nil { + return fmt.Errorf("invalid media_file_id: %w", err) + } + + // Load the media file record + mediaFile, err := p.jobRepo.GetMediaFileByID(ctx, mediaFileID) + if err != nil { + return fmt.Errorf("failed to get media file: %w", err) + } + + // --- STAGE A: Download from Supabase Storage --- + log.Printf("[job %s] Stage A: Downloading file from storage: %s", job.ID, mediaFile.StoragePath) + fileReader, err := p.storage.DownloadFile(ctx, p.storageBucket, mediaFile.StoragePath) + if err != nil { + return fmt.Errorf("failed to download from storage: %w", err) + } + defer fileReader.Close() + + // Write to a temp file on disk so ffmpeg can read it + if err := os.MkdirAll(p.tempDir, 0755); err != nil { + return fmt.Errorf("failed to create temp dir: %w", err) + } + ext := ".bin" + if mediaFile.MimeType == "video/mp4" { + ext = ".mp4" + } else if strings.HasPrefix(mediaFile.MimeType, "audio/") { + ext = ".wav" + } + tmpInput := fmt.Sprintf("%s/%s%s", p.tempDir, uuid.New().String(), ext) + tmpFile, err := os.Create(tmpInput) + if err != nil { + return fmt.Errorf("failed to create temp input file: %w", err) + } + if _, err := tmpFile.ReadFrom(fileReader); err != nil { + tmpFile.Close() + return fmt.Errorf("failed to write temp input file: %w", err) + } + tmpFile.Close() + defer p.extractor.Cleanup(tmpInput) + + // --- STAGE B: Audio Extraction (skip if already WAV/MP3) --- + log.Printf("[job %s] Stage B: Extracting 16kHz WAV audio", job.ID) + wavPath, err := p.extractor.ExtractWav16kHz(ctx, tmpInput) + if err != nil { + return fmt.Errorf("ffmpeg extraction failed: %w", err) + } + defer p.extractor.Cleanup(wavPath) + + // --- STAGE C: Send to Modal for transcription + diarization --- + log.Printf("[job %s] Stage C: Sending to Modal Whisper endpoint", job.ID) + result, err := p.modal.TranscribeAudio(ctx, wavPath, payload.Language) + if err != nil { + return fmt.Errorf("modal transcription failed: %w", err) + } + + // --- STAGE D: Persist transcript + segments --- + log.Printf("[job %s] Stage D: Persisting transcript (%d segments)", job.ID, len(result.Segments)) + source := string(domain.SourceUpload) + status := domain.TranscriptStatusCompleted + completedAt := time.Now() + + transcript := &domain.Transcript{ + MeetingID: job.MeetingID, + MediaFileID: &mediaFileID, + Source: domain.SourceUpload, + Language: result.Language, + FullText: &result.Text, + DurationSecs: &result.Info.DurationSecs, + SpeakerCount: &result.Info.SpeakerCount, + Status: status, + CompletedAt: &completedAt, + } + _ = source + if err := p.transcriptRepo.Create(ctx, transcript); err != nil { + return fmt.Errorf("failed to create transcript: %w", err) + } + + // Build segments — speaker_name/email from Modal diarization. + // For uploaded files, speaker identity isn't known from Jitsi so we + // default to the diarized label. Caller can update later via API. + segments := make([]*domain.TranscriptSegment, 0, len(result.Segments)) + for i, seg := range result.Segments { + seqNum := i + segments = append(segments, &domain.TranscriptSegment{ + TranscriptID: transcript.ID, + SpeakerLabel: &seg.Speaker, + SpeakerName: seg.Speaker, // Resolved by host via frontend later + SpeakerEmail: "unknown@minutely.ai", // Placeholder for uploads + Text: seg.Text, + StartSecs: seg.Start, + EndSecs: seg.End, + IsPartial: false, + SequenceNum: &seqNum, + }) + } + + if err := p.transcriptRepo.CreateSegmentsBatch(ctx, segments); err != nil { + return fmt.Errorf("failed to persist segments: %w", err) + } + + // Update media file status + mediaFile.Status = domain.MediaStatusReady + if err := p.jobRepo.UpdateMediaFile(ctx, mediaFile); err != nil { + log.Printf("[job %s] Warning: failed to update media file status: %v", job.ID, err) + } + + // Store transcript ID in job result + job.Result = map[string]string{"transcript_id": transcript.ID.String()} + if err := p.jobRepo.UpdateJob(ctx, job); err != nil { + log.Printf("[job %s] Warning: failed to store result in job: %v", job.ID, err) + } + + // --- STAGE E: Fire AI pipeline hook --- + log.Printf("[job %s] Stage E: Firing AI pipeline hook", job.ID) + participants := make([]string, 0) + seen := map[string]bool{} + for _, seg := range segments { + if !seen[seg.SpeakerName] { + participants = append(participants, seg.SpeakerName) + seen[seg.SpeakerName] = true + } + } + + aiEvent := domain.AIEvent{ + MeetingID: job.MeetingID, + TranscriptID: transcript.ID, + Segments: segments, + FullText: result.Text, + Participants: participants, + DurationSecs: result.Info.DurationSecs, + Source: string(domain.SourceUpload), + } + if err := p.aiProcessor.ProcessTranscript(ctx, aiEvent); err != nil { + // Non-fatal: log but don't fail the job + log.Printf("[job %s] Warning: AI processor error: %v", job.ID, err) + } + + log.Printf("[job %s] Completed successfully. Transcript: %s", job.ID, transcript.ID) + return nil +} + +func (p *FileProcessor) processAIJob(ctx context.Context, job *domain.ProcessingJob) error { + payloadBytes, err := json.Marshal(job.Payload) + if err != nil { + return fmt.Errorf("failed to marshal payload: %w", err) + } + var payload struct { + TranscriptID string `json:"transcript_id"` + } + if err := json.Unmarshal(payloadBytes, &payload); err != nil { + return fmt.Errorf("failed to parse job payload: %w", err) + } + + transcriptID, err := uuid.Parse(payload.TranscriptID) + if err != nil { + return fmt.Errorf("invalid transcript_id: %w", err) + } + + // Fetch transcript + transcript, err := p.transcriptRepo.GetByID(ctx, transcriptID) + if err != nil { + return fmt.Errorf("failed to get transcript: %w", err) + } + + // Fetch segments + segments, err := p.transcriptRepo.ListSegments(ctx, transcriptID) + if err != nil { + return fmt.Errorf("failed to get transcript segments: %w", err) + } + log.Printf("[job %s] Found %d segments for transcript %s", job.ID, len(segments), transcriptID) + if len(segments) == 0 { + log.Printf("[job %s] No segments found, skipping AI processing", job.ID) + return nil + } + + // Build AIEvent + participants := make([]string, 0) + seen := map[string]bool{} + for _, seg := range segments { + if !seen[seg.SpeakerName] { + participants = append(participants, seg.SpeakerName) + seen[seg.SpeakerName] = true + } + } + + fullText := "" + if transcript.FullText != nil { + fullText = *transcript.FullText + } + + durationSecs := 0.0 + if transcript.DurationSecs != nil { + durationSecs = *transcript.DurationSecs + } + + aiEvent := domain.AIEvent{ + MeetingID: job.MeetingID, + TranscriptID: transcriptID, + Segments: segments, + FullText: fullText, + Participants: participants, + DurationSecs: durationSecs, + Source: string(transcript.Source), + } + + // Delegate to AIProcessor implementation (which will hit Python service) + log.Printf("[job %s] Firing AI pipeline hook for transcript %s", job.ID, transcriptID) + if err := p.aiProcessor.ProcessTranscript(ctx, aiEvent); err != nil { + return fmt.Errorf("AI processor failed: %w", err) + } + + return nil +} diff --git a/minutely-api/internal/workers/pool.go b/minutely-api/internal/workers/pool.go new file mode 100644 index 0000000000..4aa0787deb --- /dev/null +++ b/minutely-api/internal/workers/pool.go @@ -0,0 +1,98 @@ +package workers + +import ( + "context" + "log" + "sync" + "time" + + "github.com/MinutelyAI/minutely-api/internal/core/domain" +) + +// Pool is a fixed-size goroutine worker pool that polls for pending jobs +type Pool struct { + concurrency int + jobRepo domain.JobRepository + processor *FileProcessor + pollInterval time.Duration + stopCh chan struct{} + wg sync.WaitGroup +} + +// NewPool creates a new worker pool +// concurrency = number of simultaneous jobs +// pollInterval = how often to check for new pending jobs +func NewPool(concurrency int, pollInterval time.Duration, jobRepo domain.JobRepository, processor *FileProcessor) *Pool { + return &Pool{ + concurrency: concurrency, + jobRepo: jobRepo, + processor: processor, + pollInterval: pollInterval, + stopCh: make(chan struct{}), + } +} + +// Start launches the worker goroutines +func (p *Pool) Start(ctx context.Context) { + log.Printf("[worker-pool] Starting %d workers (poll interval: %s)", p.concurrency, p.pollInterval) + for i := 0; i < p.concurrency; i++ { + p.wg.Add(1) + go p.runWorker(ctx, i) + } +} + +// Stop gracefully shuts down the pool, waiting for in-flight jobs to finish +func (p *Pool) Stop() { + log.Println("[worker-pool] Stopping workers...") + close(p.stopCh) + p.wg.Wait() + log.Println("[worker-pool] All workers stopped.") +} + +func (p *Pool) runWorker(ctx context.Context, id int) { + defer p.wg.Done() + ticker := time.NewTicker(p.pollInterval) + defer ticker.Stop() + + log.Printf("[worker %d] Started", id) + for { + select { + case <-p.stopCh: + log.Printf("[worker %d] Shutting down", id) + return + case <-ctx.Done(): + log.Printf("[worker %d] Context cancelled", id) + return + case <-ticker.C: + p.processOnePendingJob(ctx, id) + } + } +} + +func (p *Pool) processOnePendingJob(ctx context.Context, workerID int) { + // 1. Try to fetch an AI processing job first (higher priority for live meetings) + jobs, err := p.jobRepo.ListPendingJobs(ctx, domain.JobTypeAIProcessing, 1) + if err != nil { + log.Printf("[worker %d] Error fetching AI jobs: %v", workerID, err) + } + + // 2. If no AI jobs, try file transcription jobs + if len(jobs) == 0 { + jobs, err = p.jobRepo.ListPendingJobs(ctx, domain.JobTypeFileTranscription, 1) + if err != nil { + log.Printf("[worker %d] Error fetching file jobs: %v", workerID, err) + return + } + } + + if len(jobs) == 0 { + return // Nothing to do + } + + job := jobs[0] + log.Printf("[worker %d] Picked up job %s (type: %s)", workerID, job.ID, job.JobType) + + if err := p.processor.Process(ctx, job); err != nil { + log.Printf("[worker %d] Job %s failed: %v", workerID, job.ID, err) + } +} diff --git a/minutely-api/supabase/migrations/000002_consolidated_schema.sql b/minutely-api/supabase/migrations/000002_consolidated_schema.sql new file mode 100644 index 0000000000..6ed1f3961f --- /dev/null +++ b/minutely-api/supabase/migrations/000002_consolidated_schema.sql @@ -0,0 +1,459 @@ +-- ============================================= +-- MINUTELY — FINAL CONSOLIDATED SCHEMA +-- Run this on a fresh database after dropping +-- all existing tables. +-- ============================================= + +-- ============================================= +-- DROP EVERYTHING (run this block first) +-- ============================================= + +DROP TABLE IF EXISTS public.ai_outputs CASCADE; +DROP TABLE IF EXISTS public.live_sessions CASCADE; +DROP TABLE IF EXISTS public.transcript_segments CASCADE; +DROP TABLE IF EXISTS public.transcripts CASCADE; +DROP TABLE IF EXISTS public.media_files CASCADE; +DROP TABLE IF EXISTS public.processing_jobs CASCADE; +DROP TABLE IF EXISTS public.notifications CASCADE; +DROP TABLE IF EXISTS public.preferences CASCADE; +DROP TABLE IF EXISTS public.action_items CASCADE; +DROP TABLE IF EXISTS public.meeting_notes CASCADE; +DROP TABLE IF EXISTS public.meeting_participants CASCADE; +DROP TABLE IF EXISTS public.meetings CASCADE; +DROP TABLE IF EXISTS public.team_members CASCADE; +DROP TABLE IF EXISTS public.teams CASCADE; +DROP TABLE IF EXISTS public.profiles CASCADE; + +DROP TYPE IF EXISTS meeting_status CASCADE; +DROP TYPE IF EXISTS participant_role CASCADE; +DROP TYPE IF EXISTS team_role CASCADE; +DROP TYPE IF EXISTS action_status CASCADE; +DROP TYPE IF EXISTS theme_preference CASCADE; +DROP TYPE IF EXISTS job_status CASCADE; +DROP TYPE IF EXISTS job_type CASCADE; +DROP TYPE IF EXISTS media_status CASCADE; +DROP TYPE IF EXISTS ai_output_type CASCADE; +DROP TYPE IF EXISTS ai_output_status CASCADE; + +-- ============================================= +-- ENUMS +-- ============================================= + +CREATE TYPE meeting_status AS ENUM ('scheduled', 'in_progress', 'completed', 'canceled'); +CREATE TYPE participant_role AS ENUM ('host', 'co-host', 'participant'); +CREATE TYPE team_role AS ENUM ('owner', 'admin', 'member'); +CREATE TYPE action_status AS ENUM ('pending', 'completed'); +CREATE TYPE theme_preference AS ENUM ('light', 'dark', 'system'); +CREATE TYPE job_status AS ENUM ('pending', 'processing', 'completed', 'failed', 'retrying'); +CREATE TYPE job_type AS ENUM ('live_transcription', 'file_transcription', 'ai_processing'); +CREATE TYPE media_status AS ENUM ('uploaded', 'processing', 'ready', 'error'); +CREATE TYPE ai_output_type AS ENUM ('summary', 'action_items', 'key_topics', 'sentiment', 'tasks', 'custom'); +CREATE TYPE ai_output_status AS ENUM ('pending', 'processing', 'completed', 'failed'); + +-- ============================================= +-- USERS & TEAMS +-- ============================================= + +CREATE TABLE public.profiles ( + id uuid PRIMARY KEY REFERENCES auth.users(id) ON DELETE CASCADE, + full_name text, + avatar_url text, + created_at timestamptz DEFAULT timezone('utc', now()), + updated_at timestamptz DEFAULT timezone('utc', now()) +); + +CREATE TABLE public.teams ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + name text NOT NULL, + description text, + created_by uuid REFERENCES auth.users(id) ON DELETE SET NULL, + created_at timestamptz DEFAULT timezone('utc', now()) +); + +CREATE TABLE public.team_members ( + team_id uuid NOT NULL REFERENCES public.teams(id) ON DELETE CASCADE, + user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE, + role team_role DEFAULT 'member', + joined_at timestamptz DEFAULT timezone('utc', now()), + PRIMARY KEY (team_id, user_id) +); + +-- ============================================= +-- MEETINGS +-- ============================================= + +CREATE TABLE public.meetings ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + user_id uuid REFERENCES auth.users(id) ON DELETE SET NULL, -- Host + team_id uuid REFERENCES public.teams(id) ON DELETE CASCADE, + title text NOT NULL, + description text, + status meeting_status DEFAULT 'scheduled', + scheduled_for timestamptz, + is_archived boolean DEFAULT false, + created_at timestamptz DEFAULT timezone('utc', now()), + updated_at timestamptz DEFAULT timezone('utc', now()) +); + +CREATE TABLE public.meeting_participants ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + user_id uuid REFERENCES auth.users(id) ON DELETE SET NULL, + email text NOT NULL, + display_name text, + role participant_role DEFAULT 'participant', + added_at timestamptz DEFAULT timezone('utc', now()), + UNIQUE (meeting_id, email) +); + +-- ============================================= +-- BACKGROUND JOBS (Upload Processing) +-- ============================================= + +CREATE TABLE public.processing_jobs ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + job_type job_type NOT NULL, + status job_status DEFAULT 'pending', + attempt_count int DEFAULT 0, + max_attempts int DEFAULT 3, + error_message text, + payload jsonb, -- Input params (language, speakers, etc.) + result jsonb, -- Storage path, duration, etc. + started_at timestamptz, + completed_at timestamptz, + created_at timestamptz DEFAULT timezone('utc', now()), + updated_at timestamptz DEFAULT timezone('utc', now()) +); + +-- ============================================= +-- MEDIA FILES (Uploaded Recordings) +-- ============================================= + +CREATE TABLE public.media_files ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + uploaded_by uuid REFERENCES auth.users(id), + storage_path text NOT NULL, -- Supabase Storage path + original_name text NOT NULL, + mime_type text NOT NULL, + size_bytes bigint, + duration_secs float, + status media_status DEFAULT 'uploaded', + job_id uuid REFERENCES public.processing_jobs(id), + created_at timestamptz DEFAULT timezone('utc', now()) +); + +-- ============================================= +-- TRANSCRIPTION +-- ============================================= + +CREATE TABLE public.transcripts ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + media_file_id uuid REFERENCES public.media_files(id), + source text NOT NULL CHECK (source IN ('live', 'upload')), + language text DEFAULT 'en', + full_text text, -- Assembled from segments on completion + duration_secs float, + speaker_count int, + status text DEFAULT 'in_progress' CHECK (status IN ('in_progress', 'completed', 'failed')), + created_at timestamptz DEFAULT timezone('utc', now()), + completed_at timestamptz +); + +-- One segment = one speaker turn. +-- speaker_name + speaker_email are MANDATORY — captured from Jitsi participant identity. +CREATE TABLE public.transcript_segments ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + transcript_id uuid NOT NULL REFERENCES public.transcripts(id) ON DELETE CASCADE, + speaker_label text, -- Raw label from Deepgram/pyannote: "SPEAKER_00" + speaker_name text NOT NULL, -- Participant display name (from Jitsi) + speaker_email text NOT NULL, -- Participant email (for task assignment) + text text NOT NULL, + start_secs float NOT NULL, + end_secs float NOT NULL, + confidence float, + is_partial boolean DEFAULT false, + sequence_num int, + created_at timestamptz DEFAULT timezone('utc', now()) +); + +-- Tracks an active live transcription session. +-- One per meeting at a time (enforced by unique constraint on active sessions). +CREATE TABLE public.live_sessions ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + transcript_id uuid REFERENCES public.transcripts(id), + deepgram_token_expires timestamptz, -- When the short-lived client token expires + started_at timestamptz DEFAULT timezone('utc', now()), + ended_at timestamptz, + participant_count int DEFAULT 0 +); + +-- ============================================= +-- AI PIPELINE (LLM-Ready Slots) +-- ============================================= + +CREATE TABLE public.ai_outputs ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + transcript_id uuid REFERENCES public.transcripts(id), + output_type ai_output_type NOT NULL, + status ai_output_status DEFAULT 'pending', + model_used text, -- e.g. "gpt-4o", "claude-3-sonnet" + prompt_version text, + result jsonb, -- Structured LLM output (any shape) + tokens_used int, + cost_usd float, + error_message text, + created_at timestamptz DEFAULT timezone('utc', now()), + completed_at timestamptz +); + +-- ============================================= +-- MEETING METADATA +-- ============================================= + +CREATE TABLE public.action_items ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + meeting_id uuid NOT NULL REFERENCES public.meetings(id) ON DELETE CASCADE, + description text NOT NULL, + owner_email text, -- Email of assignee (links to transcript_segments.speaker_email) + due_date timestamptz, + status action_status DEFAULT 'pending', + created_at timestamptz DEFAULT timezone('utc', now()) +); + +CREATE TABLE public.meeting_notes ( + meeting_id uuid PRIMARY KEY REFERENCES public.meetings(id) ON DELETE CASCADE, + summary text, + key_points jsonb, + is_approved boolean DEFAULT false, + created_at timestamptz DEFAULT timezone('utc', now()), + updated_at timestamptz DEFAULT timezone('utc', now()) +); + +-- ============================================= +-- USER PREFERENCES & NOTIFICATIONS +-- ============================================= + +CREATE TABLE public.preferences ( + user_id uuid PRIMARY KEY REFERENCES auth.users(id) ON DELETE CASCADE, + theme theme_preference DEFAULT 'system', + default_mic text, + default_camera text, + default_speaker text, + join_muted boolean DEFAULT false, + enable_captions boolean DEFAULT true, + updated_at timestamptz DEFAULT timezone('utc', now()) +); + +CREATE TABLE public.notifications ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + user_id uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE, + title text NOT NULL, + message text NOT NULL, + action_url text, + is_read boolean DEFAULT false, + created_at timestamptz DEFAULT timezone('utc', now()) +); + +-- ============================================= +-- INDEXES +-- ============================================= + +CREATE INDEX idx_team_members_user_id ON public.team_members(user_id); +CREATE INDEX idx_meetings_user_id ON public.meetings(user_id); +CREATE INDEX idx_meetings_team_id ON public.meetings(team_id); +CREATE INDEX idx_meeting_participants_meeting_id ON public.meeting_participants(meeting_id); +CREATE INDEX idx_processing_jobs_meeting_id ON public.processing_jobs(meeting_id); +CREATE INDEX idx_processing_jobs_status ON public.processing_jobs(status); +CREATE INDEX idx_media_files_meeting_id ON public.media_files(meeting_id); +CREATE INDEX idx_transcripts_meeting_id ON public.transcripts(meeting_id); +CREATE INDEX idx_segments_transcript_id ON public.transcript_segments(transcript_id); +CREATE INDEX idx_segments_speaker_email ON public.transcript_segments(speaker_email); +CREATE INDEX idx_segments_start ON public.transcript_segments(transcript_id, start_secs); +CREATE INDEX idx_ai_outputs_meeting_id ON public.ai_outputs(meeting_id); +CREATE INDEX idx_action_items_owner_email ON public.action_items(owner_email); +CREATE INDEX idx_notifications_user_id ON public.notifications(user_id); + +-- ============================================= +-- ROW LEVEL SECURITY — ENABLE +-- ============================================= + +ALTER TABLE public.profiles ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.teams ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.team_members ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.meetings ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.meeting_participants ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.processing_jobs ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.media_files ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.transcripts ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.transcript_segments ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.live_sessions ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.ai_outputs ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.action_items ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.meeting_notes ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.preferences ENABLE ROW LEVEL SECURITY; +ALTER TABLE public.notifications ENABLE ROW LEVEL SECURITY; + +-- ============================================= +-- ROW LEVEL SECURITY — POLICIES +-- ============================================= + +-- Profiles +CREATE POLICY "Anyone can view profiles" + ON public.profiles FOR SELECT USING (true); +CREATE POLICY "Users manage own profile" + ON public.profiles FOR ALL USING (auth.uid() = id); + +-- Teams +CREATE POLICY "Team members can view teams" + ON public.teams FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.team_members WHERE team_id = teams.id AND user_id = auth.uid()) + ); +CREATE POLICY "Users can create teams" + ON public.teams FOR INSERT WITH CHECK (auth.uid() = created_by); +CREATE POLICY "Team owners can update teams" + ON public.teams FOR UPDATE USING ( + EXISTS (SELECT 1 FROM public.team_members WHERE team_id = teams.id AND user_id = auth.uid() AND role = 'owner') + ); + +-- Team Members +CREATE POLICY "Team members can view members" + ON public.team_members FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.team_members tm WHERE tm.team_id = team_members.team_id AND tm.user_id = auth.uid()) + ); +CREATE POLICY "Team admins can manage members" + ON public.team_members FOR ALL USING ( + EXISTS (SELECT 1 FROM public.team_members tm WHERE tm.team_id = team_members.team_id AND tm.user_id = auth.uid() AND tm.role IN ('owner', 'admin')) + ); + +-- Helper: is user a meeting member? +-- Used as sub-expression in policies below. +-- A user has access if they are: host, team member, or invited participant. + +-- Meetings +CREATE POLICY "Users can view accessible meetings" + ON public.meetings FOR SELECT USING ( + auth.uid() = user_id + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + OR EXISTS (SELECT 1 FROM public.meeting_participants WHERE meeting_id = meetings.id AND user_id = auth.uid()) + ); +CREATE POLICY "Users can create meetings" + ON public.meetings FOR INSERT WITH CHECK (auth.uid() = user_id); +CREATE POLICY "Hosts can update meetings" + ON public.meetings FOR UPDATE USING (auth.uid() = user_id); +CREATE POLICY "Hosts can delete meetings" + ON public.meetings FOR DELETE USING (auth.uid() = user_id); + +-- Meeting Participants +CREATE POLICY "Participants can view other participants" + ON public.meeting_participants FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = meeting_participants.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.meeting_participants mp WHERE mp.meeting_id = meeting_participants.meeting_id AND mp.user_id = auth.uid()) + )) + ); +CREATE POLICY "Hosts can manage participants" + ON public.meeting_participants FOR ALL USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = meeting_participants.meeting_id AND user_id = auth.uid()) + ); + +-- Processing Jobs +CREATE POLICY "Meeting members can view jobs" + ON public.processing_jobs FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = processing_jobs.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); + +-- Media Files +CREATE POLICY "Meeting members can view media" + ON public.media_files FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = media_files.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); +CREATE POLICY "Meeting members can upload media" + ON public.media_files FOR INSERT WITH CHECK ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = media_files.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); + +-- Transcripts +CREATE POLICY "Meeting members can view transcripts" + ON public.transcripts FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = transcripts.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); + +-- Transcript Segments +CREATE POLICY "Meeting members can view segments" + ON public.transcript_segments FOR SELECT USING ( + EXISTS ( + SELECT 1 FROM public.transcripts t + JOIN public.meetings m ON t.meeting_id = m.id + WHERE t.id = transcript_segments.transcript_id AND ( + m.user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = m.team_id AND user_id = auth.uid()) + ) + ) + ); + +-- Live Sessions +CREATE POLICY "Meeting members can view live sessions" + ON public.live_sessions FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = live_sessions.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); + +-- AI Outputs +CREATE POLICY "Meeting members can view AI outputs" + ON public.ai_outputs FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = ai_outputs.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); + +-- Action Items +CREATE POLICY "Meeting members can view action items" + ON public.action_items FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = action_items.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); +CREATE POLICY "Hosts can manage action items" + ON public.action_items FOR ALL USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = action_items.meeting_id AND user_id = auth.uid()) + ); + +-- Meeting Notes +CREATE POLICY "Meeting members can view notes" + ON public.meeting_notes FOR SELECT USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = meeting_notes.meeting_id AND ( + user_id = auth.uid() + OR EXISTS (SELECT 1 FROM public.team_members WHERE team_id = meetings.team_id AND user_id = auth.uid()) + )) + ); +CREATE POLICY "Hosts can manage notes" + ON public.meeting_notes FOR ALL USING ( + EXISTS (SELECT 1 FROM public.meetings WHERE id = meeting_notes.meeting_id AND user_id = auth.uid()) + ); + +-- Preferences & Notifications +CREATE POLICY "Users manage own preferences" + ON public.preferences FOR ALL USING (auth.uid() = user_id); +CREATE POLICY "Users manage own notifications" + ON public.notifications FOR ALL USING (auth.uid() = user_id); diff --git a/react/features/app/middlewares.web.ts b/react/features/app/middlewares.web.ts index 3d581b8284..888f0f96ad 100644 --- a/react/features/app/middlewares.web.ts +++ b/react/features/app/middlewares.web.ts @@ -30,3 +30,4 @@ import '../file-sharing/middleware.web'; import '../custom-panel/middleware.web'; import './middlewares.any'; +import '../transcription/middleware'; \ No newline at end of file diff --git a/react/features/app/reducers.any.ts b/react/features/app/reducers.any.ts index 28e22b3088..2de474efd8 100644 --- a/react/features/app/reducers.any.ts +++ b/react/features/app/reducers.any.ts @@ -56,3 +56,4 @@ import '../video-quality/reducer'; import '../videosipgw/reducer'; import '../visitors/reducer'; import '../whiteboard/reducer'; +import '../transcription/reducer'; diff --git a/react/features/conference/components/web/Conference.tsx b/react/features/conference/components/web/Conference.tsx index 7fe2683c4f..a1504d8eac 100644 --- a/react/features/conference/components/web/Conference.tsx +++ b/react/features/conference/components/web/Conference.tsx @@ -33,6 +33,7 @@ import { toggleToolboxVisible } from '../../../toolbox/actions.any'; import { fullScreenChanged, showToolbox } from '../../../toolbox/actions.web'; import JitsiPortal from '../../../toolbox/components/web/JitsiPortal'; import Toolbox from '../../../toolbox/components/web/Toolbox'; +import { ToggleSidebarButton, TranscriptPanel } from '../../../transcription/components'; import { LAYOUT_CLASSNAMES } from '../../../video-layout/constants'; import { getCurrentLayout } from '../../../video-layout/functions.any'; import VisitorsQueue from '../../../visitors/components/web/VisitorsQueue'; @@ -249,6 +250,8 @@ class Conference extends AbstractConference { onMouseMove = { this._onMouseMove } ref = { this._setBackground }> + +
{ onMouseMove = { this._onMouseMove } ref = { this._setBackground }> + +
{ + const dispatch = useDispatch(); + const { isTranscriptPanelOpen } = useSelector((state: any) => state['features/transcription'] as TranscriptionState); + + return ( +
+ +
+ ); +}; diff --git a/react/features/transcription/components/TranscriptPanel.tsx b/react/features/transcription/components/TranscriptPanel.tsx new file mode 100644 index 0000000000..283bd3eaef --- /dev/null +++ b/react/features/transcription/components/TranscriptPanel.tsx @@ -0,0 +1,249 @@ +import React, { useEffect, useRef, useState } from 'react'; +import { useSelector, useDispatch } from 'react-redux'; +import { TranscriptionState } from '../reducer'; +import { startTranscription, stopTranscription, toggleTranscriptPanel } from '../actions'; +import { UploadRecordingModal } from './UploadRecordingModal'; +import { getCurrentConference } from '../../base/conference/functions'; + +// shadcn-ui components +import { Button } from '../../../components/ui/button'; +import { Card, CardContent, CardHeader, CardTitle } from '../../../components/ui/card'; +import { X, RefreshCw, Bot } from 'lucide-react'; + +export const TranscriptPanel: React.FC = () => { + const dispatch = useDispatch(); + const { + isTranscribing, + status, + segments, + isTranscriptPanelOpen + } = useSelector((state: any) => state['features/transcription'] as TranscriptionState); + + const meetingId = useSelector((state: any) => getCurrentConference(state)?.getName()); + const [isUploadModalOpen, setIsUploadModalOpen] = useState(false); + const [activeTab, setActiveTab] = useState<'live' | 'ai'>('live'); + + // AI Insights State + const [aiInsights, setAiInsights] = useState(null); + const [isFetchingAI, setIsFetchingAI] = useState(false); + const [aiError, setAiError] = useState(null); + + const scrollRef = useRef(null); + + // Auto-scroll to bottom on new segments + useEffect(() => { + if (activeTab === 'live' && scrollRef.current) { + scrollRef.current.scrollIntoView({ behavior: 'smooth' }); + } + }, [segments, activeTab]); + + const handleToggleTranscription = () => { + if (isTranscribing) { + dispatch(stopTranscription()); + } else { + dispatch(startTranscription()); + } + }; + + const handleClose = () => { + dispatch(toggleTranscriptPanel()); + }; + + const fetchAIInsights = async () => { + if (!meetingId) return; + setIsFetchingAI(true); + setAiError(null); + try { + const res = await fetch(`/api/v1/meetings/${meetingId}/ai-insights`); + if (!res.ok) throw new Error("Failed to fetch insights"); + const data = await res.json(); + if (data && data.length > 0 && data[0].result) { + setAiInsights(data[0].result); + } else { + setAiError("AI Analysis is still processing or unavailable."); + } + } catch (err: any) { + setAiError(err.message || "An error occurred fetching AI Insights."); + } finally { + setIsFetchingAI(false); + } + }; + + return ( + + + + Meeting Intelligence + + + + {/* Tabs */} +
+ + +
+ + + + {activeTab === 'live' && ( +
+
+ + +
+
+ {segments.length === 0 ? ( +
+ {isTranscribing ? 'Listening...' : 'Live transcription is inactive.'} +
+ ) : ( +
+ {segments.map((seg, idx) => ( +
+ + {seg.speaker_name} + + + {seg.text} + +
+ ))} +
+
+ )} +
+ {status === 'error' && ( +
+ Connection lost or error occurred. +
+ )} +
+ )} + + {activeTab === 'ai' && ( +
+
+

+ + Meeting Insights +

+ +
+ + {aiError && ( +
+ {aiError} +
+ )} + + {!aiInsights && !isFetchingAI && !aiError && ( +
+

No insights available yet.

+

End the live meeting to generate AI summaries, action items, and topic clusters.

+
+ )} + + {aiInsights && ( +
+ {/* Executive Summary */} + {aiInsights.executive_summary && ( +
+

Executive Summary

+

+ {aiInsights.executive_summary} +

+
+ )} + + {/* Action Items */} + {aiInsights.action_items && aiInsights.action_items.length > 0 && ( +
+

Action Items

+
    + {aiInsights.action_items.map((item: any, i: number) => ( +
  • + {item.task} +
    + Assignee: {item.assignee} + Due: {item.deadline} +
    +
    • + ))} +
+
+ )} + + {/* Topic Clusters */} + {aiInsights.topics && aiInsights.topics.length > 0 && ( +
+

Key Topics Discussed

+
+ {aiInsights.topics.map((topic: any, i: number) => ( +
+
{topic.title}
+
+ {topic.keywords?.map((kw: string, j: number) => ( + + {kw} + + ))} +
+

{topic.summary}

+
+ ))} +
+
+ )} +
+ )} +
+ )} + + + {meetingId && ( + setIsUploadModalOpen(false)} + meetingId={meetingId} + /> + )} + + ); +}; diff --git a/react/features/transcription/components/UploadRecordingModal.tsx b/react/features/transcription/components/UploadRecordingModal.tsx new file mode 100644 index 0000000000..470d799706 --- /dev/null +++ b/react/features/transcription/components/UploadRecordingModal.tsx @@ -0,0 +1,108 @@ +import React, { useState } from 'react'; +import { useDispatch } from 'react-redux'; +import { Button } from '../../../components/ui/button'; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from '../../../components/ui/dialog'; +import { Input } from '../../../components/ui/input'; +import { Label } from '../../../components/ui/label'; + +interface UploadRecordingModalProps { + isOpen: boolean; + onClose: () => void; + meetingId: string; +} + +export const UploadRecordingModal: React.FC = ({ isOpen, onClose, meetingId }) => { + const [file, setFile] = useState(null); + const [isUploading, setIsUploading] = useState(false); + const [error, setError] = useState(null); + + const handleFileChange = (e: React.ChangeEvent) => { + if (e.target.files && e.target.files.length > 0) { + setFile(e.target.files[0]); + setError(null); + } + }; + + const handleUpload = async () => { + if (!file) { + setError('Please select a file to upload.'); + return; + } + + setIsUploading(true); + setError(null); + + const formData = new FormData(); + formData.append('file', file); + formData.append('language', 'en'); // Defaulting to English for now + + try { + const res = await fetch(`/api/v1/meetings/${meetingId}/recordings/upload`, { + method: 'POST', + body: formData, + }); + + if (!res.ok) { + const errorData = await res.json(); + throw new Error(errorData.error || 'Upload failed'); + } + + const data = await res.json(); + console.log('Upload successful. Job ID:', data.job_id); + onClose(); // Close on success + + // Note: In a real implementation, you'd dispatch an action to track the job progress + // using the /api/v1/jobs/{jobId}/progress SSE endpoint. + + } catch (err: any) { + setError(err.message || 'An unexpected error occurred during upload.'); + } finally { + setIsUploading(false); + } + }; + + return ( + + + + Upload Recording + + Upload a video or audio recording of a meeting to generate an AI transcript. + + +
+
+ + +
+ {error && ( +
+ {error} +
+ )} +
+ + + + + + + ); +}; diff --git a/react/features/transcription/components/index.ts b/react/features/transcription/components/index.ts new file mode 100644 index 0000000000..b07b89873a --- /dev/null +++ b/react/features/transcription/components/index.ts @@ -0,0 +1,3 @@ +export * from './TranscriptPanel'; +export * from './UploadRecordingModal'; +export * from './ToggleSidebarButton'; diff --git a/react/features/transcription/index.ts b/react/features/transcription/index.ts new file mode 100644 index 0000000000..e23c36dc81 --- /dev/null +++ b/react/features/transcription/index.ts @@ -0,0 +1,6 @@ +export * from './actions'; +export * from './actionTypes'; +export * from './components'; + +import './reducer'; +import './middleware'; diff --git a/react/features/transcription/middleware.ts b/react/features/transcription/middleware.ts new file mode 100644 index 0000000000..5f534bd47e --- /dev/null +++ b/react/features/transcription/middleware.ts @@ -0,0 +1,228 @@ +import MiddlewareRegistry from '../base/redux/MiddlewareRegistry'; +import { getCurrentConference } from '../base/conference/functions'; +import { getLocalParticipant } from '../base/participants/functions'; +import { + START_TRANSCRIPTION, + STOP_TRANSCRIPTION, +} from './actionTypes'; +import { + transcriptionStarted, + transcriptionStopped, + setTranscriptionStatus, + transcriptSegmentReceived +} from './actions'; + +// References to active WebSockets +let deepgramWs: WebSocket | null = null; +let minutelyWs: WebSocket | null = null; + +MiddlewareRegistry.register(store => next => action => { + switch (action.type) { + case START_TRANSCRIPTION: { + startTranscriptionPipeline(store); + break; + } + case STOP_TRANSCRIPTION: { + stopTranscriptionPipeline(store); + break; + } + } + + return next(action); +}); + +async function startTranscriptionPipeline(store: any) { + const state = store.getState(); + const conference = getCurrentConference(state); + + if (!conference) { + console.error('No active conference found'); + return; + } + + // We get the room name (acting as meeting ID for now, though later it should be the actual Minutely meeting ID) + const meetingId = conference.getName(); + + store.dispatch(setTranscriptionStatus('connecting')); + + try { + // 1. Call Go backend to start session + // Note: You might need to pass auth tokens here depending on your middleware + const res = await fetch(`/api/v1/meetings/${meetingId}/transcription/start`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' } + }); + + if (!res.ok) { + throw new Error('Failed to start transcription session'); + } + + const data = await res.json(); + const sessionId = data.session_id; + const deepgramToken = data.deepgram_token; + + // The WebSocket URL for our Go Hub + // Assumes the frontend is served on the same origin or handled by proxy + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const wsUrl = `${protocol}//${window.location.host}/api/v1/meetings/${meetingId}/transcription/ws`; + + // 2. Connect to Deepgram + connectToDeepgram(store, deepgramToken, meetingId); + + // 3. Connect to Minutely Go Backend + connectToMinutely(store, wsUrl); + + store.dispatch(transcriptionStarted(sessionId, deepgramToken, wsUrl)); + store.dispatch(setTranscriptionStatus('connected')); + + } catch (error) { + console.error('Error starting transcription:', error); + store.dispatch(setTranscriptionStatus('error')); + } +} + +function stopTranscriptionPipeline(store: any) { + if (deepgramWs) { + deepgramWs.close(); + deepgramWs = null; + } + if (minutelyWs) { + minutelyWs.close(); + minutelyWs = null; + } + + const state = store.getState(); + const conference = getCurrentConference(state); + if (conference) { + const meetingId = conference.getName(); + fetch(`/api/v1/meetings/${meetingId}/transcription/end`, { + method: 'POST', + }).catch(err => console.error("Failed to end session on backend", err)); + } + + store.dispatch(transcriptionStopped()); +} + +function connectToDeepgram(store: any, token: string, meetingId: string) { + // Deepgram Live Transcription URL (16kHz mono audio, English, diarized) + const url = 'wss://api.deepgram.com/v1/listen?encoding=linear16&sample_rate=16000&channels=1&diarize=true'; + + deepgramWs = new WebSocket(url, ['token', token]); + + deepgramWs.onopen = () => { + console.log('Deepgram WebSocket connected'); + // Now we need to start capturing local audio and sending it to Deepgram. + // This is simplified. In Jitsi, we capture from the local track. + startAudioCapture(store); + }; + + deepgramWs.onmessage = (event) => { + const msg = JSON.parse(event.data); + if (msg.type === 'Results' && msg.channel.alternatives[0].transcript !== '') { + const transcript = msg.channel.alternatives[0].transcript; + const isFinal = msg.is_final; + const start = msg.start; + const end = msg.start + msg.duration; + + const localParticipant = getLocalParticipant(store.getState()); + + // Build the standard segment format we defined in Go + const segment = { + meeting_id: meetingId, + speaker_name: localParticipant?.name || 'Unknown', + speaker_email: localParticipant?.email || 'unknown@example.com', + text: transcript, + start_secs: start, + end_secs: end, + is_partial: !isFinal + }; + + // 1. Dispatch locally so it shows on the UI immediately + store.dispatch(transcriptSegmentReceived(segment)); + + // 2. Forward to Minutely Hub to broadcast to others and persist + if (minutelyWs && minutelyWs.readyState === WebSocket.OPEN) { + minutelyWs.send(JSON.stringify(segment)); + } + } + }; + + deepgramWs.onerror = (error) => { + console.error('Deepgram WebSocket error:', error); + }; + + deepgramWs.onclose = () => { + console.log('Deepgram WebSocket closed'); + }; +} + +function connectToMinutely(store: any, wsUrl: string) { + minutelyWs = new WebSocket(wsUrl); + + minutelyWs.onopen = () => { + console.log('Minutely Hub WebSocket connected'); + }; + + minutelyWs.onmessage = (event) => { + const segment = JSON.parse(event.data); + const localParticipant = getLocalParticipant(store.getState()); + + // Don't duplicate segments we sent ourselves (since we already dispatched them locally) + // But do dispatch segments from other participants + if (segment.speaker_email !== localParticipant?.email) { + store.dispatch(transcriptSegmentReceived(segment)); + } + }; + + minutelyWs.onerror = (error) => { + console.error('Minutely Hub WebSocket error:', error); + }; + + minutelyWs.onclose = () => { + console.log('Minutely Hub WebSocket closed'); + }; +} + +// Global scriptProcessor reference to prevent garbage collection +let scriptProcessor: ScriptProcessorNode | null = null; + +function startAudioCapture(store: any) { + const state = store.getState(); + const tracks = state['features/base/tracks']; + const localAudioTrack = tracks.find((t: any) => t.local && t.mediaType === 'audio'); + + if (!localAudioTrack) { + console.error('No local audio track found to stream to Deepgram'); + return; + } + + const stream = localAudioTrack.jitsiTrack.getOriginalStream(); + if (!stream) return; + + try { + const audioContext = new (window.AudioContext || (window as any).webkitAudioContext)({ + sampleRate: 16000 // Deepgram expects 16kHz + }); + + const source = audioContext.createMediaStreamSource(stream); + scriptProcessor = audioContext.createScriptProcessor(4096, 1, 1); + + source.connect(scriptProcessor); + scriptProcessor.connect(audioContext.destination); + + scriptProcessor.onaudioprocess = (e) => { + if (deepgramWs && deepgramWs.readyState === WebSocket.OPEN) { + // Convert Float32Array to Int16Array for Deepgram + const floatData = e.inputBuffer.getChannelData(0); + const pcmData = new Int16Array(floatData.length); + for (let i = 0; i < floatData.length; i++) { + const s = Math.max(-1, Math.min(1, floatData[i])); + pcmData[i] = s < 0 ? s * 0x8000 : s * 0x7FFF; + } + deepgramWs.send(pcmData.buffer); + } + }; + } catch (e) { + console.error("Failed to start audio capture", e); + } +} diff --git a/react/features/transcription/reducer.ts b/react/features/transcription/reducer.ts new file mode 100644 index 0000000000..44b1f12d4a --- /dev/null +++ b/react/features/transcription/reducer.ts @@ -0,0 +1,82 @@ +import ReducerRegistry from '../base/redux/ReducerRegistry'; +import { + TRANSCRIPTION_STARTED, + TRANSCRIPTION_STOPPED, + TRANSCRIPT_SEGMENT_RECEIVED, + SET_TRANSCRIPTION_STATUS, + TOGGLE_TRANSCRIPT_PANEL +} from './actionTypes'; + +export interface TranscriptionState { + isTranscribing: boolean; + sessionId: string | null; + deepgramToken: string | null; + wsUrl: string | null; + status: 'disconnected' | 'connecting' | 'connected' | 'error'; + segments: Array; + isTranscriptPanelOpen: boolean; +} + +const DEFAULT_STATE: TranscriptionState = { + isTranscribing: false, + sessionId: null, + deepgramToken: null, + wsUrl: null, + status: 'disconnected', + segments: [], + isTranscriptPanelOpen: false +}; + +ReducerRegistry.register('features/transcription', (state = DEFAULT_STATE, action: any) => { + switch (action.type) { + case TRANSCRIPTION_STARTED: + return { + ...state, + isTranscribing: true, + sessionId: action.sessionId, + deepgramToken: action.deepgramToken, + wsUrl: action.wsUrl, + segments: [] // Clear previous segments on start + }; + + case TRANSCRIPTION_STOPPED: + return { + ...state, + isTranscribing: false, + sessionId: null, + deepgramToken: null, + wsUrl: null, + status: 'disconnected' + }; + + case SET_TRANSCRIPTION_STATUS: + return { + ...state, + status: action.status + }; + + case TRANSCRIPT_SEGMENT_RECEIVED: + return { + ...state, + // Replace if we have a matching startSecs/endSecs, otherwise append + segments: appendOrUpdateSegment(state.segments, action.segment) + }; + + case TOGGLE_TRANSCRIPT_PANEL: + return { + ...state, + isTranscriptPanelOpen: !state.isTranscriptPanelOpen + }; + + default: + return state; + } +}); + +function appendOrUpdateSegment(segments: Array, newSegment: any) { + // Basic logic: if we find a segment that has the same speaker and overlaps, + // or is the "final" version of a previous "partial", replace it. + // Deepgram sends words as they arrive. The middleware will batch them into segments. + // For simplicity, we just append it here, but in real scenarios, you'd replace partials. + return [...segments, newSegment]; +} diff --git a/webpack.config.js b/webpack.config.js index 2e9ea198d1..366b6fe51d 100644 --- a/webpack.config.js +++ b/webpack.config.js @@ -290,6 +290,12 @@ function getDevServerConfig() { host: 'localhost', hot: true, proxy: [ + { + context: [ '/api/v1' ], + target: 'http://127.0.0.1:8081', + secure: false, + ws: true // Support WebSockets for our Hub + }, { context: [ '/' ], bypass: devServerProxyBypass,