feat(web): integrate Restate client for audio pipeline invocation

apps/web/.env.sample

@@ -25,3 +25,5 @@ SUPABASE_GRAPHQL_URL=""
 SUPABASE_STORAGE_URL=""
 S3_REGION=""
 SUPABASE_SERVICE_ROLE_KEY=""
+
+RESTATE_INGRESS_URL=""

apps/web/package.json

@@ -20,6 +20,7 @@
     "@nangohq/node": "^0.69.14",
     "@netlify/vite-plugin-tanstack-start": "^1.2.1",
     "@posthog/react": "^1.5.0",
+    "@restatedev/restate-sdk-clients": "^1.9.1",
     "@sentry/tanstackstart-react": "^10.26.0",
     "@stripe/stripe-js": "^8.5.2",
     "@supabase/ssr": "^0.7.0",

apps/web/src/env.ts

@@ -19,6 +19,8 @@ export const env = createEnv({
     LOOPS_KEY: z.string().min(1),
 
     DEEPGRAM_API_KEY: z.string().min(1),
+
+    RESTATE_INGRESS_URL: z.string().min(1),
   },
 
   clientPrefix: "VITE_",

apps/web/src/functions/transcription.ts

@@ -1,10 +1,143 @@
 import { createClient } from "@deepgram/sdk";
+import * as clients from "@restatedev/restate-sdk-clients";
 import { createServerFn } from "@tanstack/react-start";
 import { z } from "zod";
 
 import { env } from "@/env";
 import { getSupabaseServerClient } from "@/functions/supabase";
 
+const PipelineStatus = z.enum([
+  "QUEUED",
+  "TRANSCRIBING",
+  "TRANSCRIBED",
+  "LLM_RUNNING",
+  "DONE",
+  "ERROR",
+]);
+
+export type PipelineStatusType = z.infer<typeof PipelineStatus>;
+
+const StatusState = z.object({
+  status: PipelineStatus,
+  transcript: z.string().optional(),
+  llmResult: z.unknown().optional(),
+  error: z.string().optional(),
+});
+
+export type StatusStateType = z.infer<typeof StatusState>;
+
+type AudioPipeline = {
+  run: (input: {
+    userId: string;
+    audioUrl: string;
+  }) => Promise<StatusStateType>;
+  getStatus: () => Promise<StatusStateType>;
+};
+
+function getRestateClient() {
+  return clients.connect({ url: env.RESTATE_INGRESS_URL });
+}
+
+export const startAudioPipeline = createServerFn({ method: "POST" })
+  .inputValidator(
+    z.object({
+      audioUrl: z.string(),
+      pipelineId: z.string().optional(),
+    }),
+  )
+  .handler(async ({ data }) => {
+    const supabase = getSupabaseServerClient();
+    const { data: userData } = await supabase.auth.getUser();
+
+    if (!userData.user) {
+      return { error: true, message: "Unauthorized" };
+    }
+
+    const pipelineId = data.pipelineId ?? crypto.randomUUID();
+
+    try {
+      const restateClient = getRestateClient();
+      const handle = await restateClient
+        .workflowClient<AudioPipeline>({ name: "AudioPipeline" }, pipelineId)
+        .workflowSubmit({ userId: userData.user.id, audioUrl: data.audioUrl });
+
+      return {
+        success: true,
+        pipelineId,
+        invocationId: handle.invocationId,
+      };
+    } catch (err) {
+      const errorMessage = err instanceof Error ? err.message : "Unknown error";
+      return { error: true, message: errorMessage };
+    }
+  });
+
+export const getAudioPipelineStatus = createServerFn({ method: "GET" })
+  .inputValidator(
+    z.object({
+      pipelineId: z.string(),
+    }),
+  )
+  .handler(async ({ data }) => {
+    const supabase = getSupabaseServerClient();
+    const { data: userData } = await supabase.auth.getUser();
+
+    if (!userData.user) {
+      return { error: true, message: "Unauthorized" };
+    }
+
+    try {
+      const restateClient = getRestateClient();
+      const status = await restateClient
+        .workflowClient<AudioPipeline>(
+          { name: "AudioPipeline" },
+          data.pipelineId,
+        )
+        .getStatus();
+
+      return {
+        success: true,
+        status,
+      };
+    } catch (err) {
+      const errorMessage = err instanceof Error ? err.message : "Unknown error";
+      return { error: true, message: errorMessage };
+    }
+  });
+
+export const getAudioPipelineResult = createServerFn({ method: "GET" })
+  .inputValidator(
+    z.object({
+      pipelineId: z.string(),
+    }),
+  )
+  .handler(async ({ data }) => {
+    const supabase = getSupabaseServerClient();
+    const { data: userData } = await supabase.auth.getUser();
+
+    if (!userData.user) {
+      return { error: true, message: "Unauthorized" };
+    }
+
+    try {
+      const restateClient = getRestateClient();
+      const workflowClient = restateClient.workflowClient<AudioPipeline>(
+        { name: "AudioPipeline" },
+        data.pipelineId,
+      );
+
+      const result = await workflowClient.workflowAttach();
+
+      return {
+        success: true,
+        result,
+      };
+    } catch (err) {
+      const errorMessage = err instanceof Error ? err.message : "Unknown error";
+      return { error: true, message: errorMessage };
+    }
+  });
+
 export const transcribeAudio = createServerFn({ method: "POST" })
   .inputValidator(
     z.object({

apps/web/src/routes/_view/app/file-transcription.tsx

@@ -1,5 +1,6 @@
+import { useQuery } from "@tanstack/react-query";
 import { createFileRoute } from "@tanstack/react-router";
-import { useMemo, useState } from "react";
+import { useEffect, useMemo, useState } from "react";
 
 import NoteEditor, { type JSONContent } from "@hypr/tiptap/editor";
 import { EMPTY_TIPTAP_DOC } from "@hypr/tiptap/shared";
@@ -10,7 +11,11 @@ import {
   TranscriptDisplay,
 } from "@/components/transcription/transcript-display";
 import { UploadArea } from "@/components/transcription/upload-area";
-import { transcribeAudio } from "@/functions/transcription";
+import {
+  getAudioPipelineStatus,
+  startAudioPipeline,
+  type StatusStateType,
+} from "@/functions/transcription";
 import { uploadAudioFile } from "@/functions/upload";
 
 export const Route = createFileRoute("/_view/app/file-transcription")({
@@ -22,16 +27,59 @@ export const Route = createFileRoute("/_view/app/file-transcription")({
 
 function Component() {
   const [file, setFile] = useState<File | null>(null);
-  const [isProcessing, setIsProcessing] = useState(false);
+  const [pipelineId, setPipelineId] = useState<string | null>(null);
   const [transcript, setTranscript] = useState<string | null>(null);
-  const [error, setError] = useState<string | null>(null);
+  const [uploadError, setUploadError] = useState<string | null>(null);
   const [noteContent, setNoteContent] = useState<JSONContent>(EMPTY_TIPTAP_DOC);
 
+  const pipelineStatusQuery = useQuery({
+    queryKey: ["audioPipelineStatus", pipelineId],
+    queryFn: async (): Promise<StatusStateType> => {
+      if (!pipelineId) {
+        throw new Error("Missing pipelineId");
+      }
+      const res = await getAudioPipelineStatus({ data: { pipelineId } });
+      if ("error" in res && res.error) {
+        throw new Error(res.message ?? "Failed to get pipeline status");
+      }
+      if (!("status" in res) || !res.status) {
+        throw new Error("Invalid response from pipeline status");
+      }
+      return res.status;
+    },
+    enabled: !!pipelineId,
+    refetchInterval: (query) => {
+      const status = query.state.data?.status;
+      const isTerminal = status === "DONE" || status === "ERROR";
+      return isTerminal ? false : 2000;
+    },
+  });
+
+  useEffect(() => {
+    const data = pipelineStatusQuery.data;
+    if (data?.status === "DONE" && data.transcript) {
+      setTranscript(data.transcript);
+    }
+  }, [pipelineStatusQuery.data]);
+
+  const isProcessing =
+    !!pipelineId &&
+    !["DONE", "ERROR"].includes(pipelineStatusQuery.data?.status ?? "");
+
+  const errorMessage =
+    uploadError ??
+    (pipelineStatusQuery.isError && pipelineStatusQuery.error instanceof Error
+      ? pipelineStatusQuery.error.message
+      : null) ??
+    (pipelineStatusQuery.data?.status === "ERROR"
+      ? pipelineStatusQuery.data.error
+      : null);
+
   const handleFileSelect = async (selectedFile: File) => {
     setFile(selectedFile);
     setTranscript(null);
-    setError(null);
-    setIsProcessing(true);
+    setUploadError(null);
+    setPipelineId(null);
 
     try {
       const reader = new FileReader();
@@ -40,8 +88,7 @@ function Component() {
       reader.onload = async () => {
         const base64Data = reader.result?.toString().split(",")[1];
         if (!base64Data) {
-          setError("Failed to read file");
-          setIsProcessing(false);
+          setUploadError("Failed to read file");
           return;
         }
 
@@ -54,52 +101,44 @@ function Component() {
         });
 
         if ("error" in uploadResult && uploadResult.error) {
-          setError(uploadResult.message || "Failed to upload file");
-          setIsProcessing(false);
+          setUploadError(uploadResult.message || "Failed to upload file");
           return;
         }
 
         if (!("url" in uploadResult)) {
-          setError("Failed to get upload URL");
-          setIsProcessing(false);
+          setUploadError("Failed to get upload URL");
           return;
         }
 
-        const transcriptionResult = await transcribeAudio({
+        const pipelineResult = await startAudioPipeline({
           data: {
             audioUrl: uploadResult.url,
-            fileName: selectedFile.name,
-            fileSize: selectedFile.size,
           },
         });
 
-        if ("error" in transcriptionResult && transcriptionResult.error) {
-          setError(transcriptionResult.message || "Failed to transcribe audio");
-          setIsProcessing(false);
+        if ("error" in pipelineResult && pipelineResult.error) {
+          setUploadError(pipelineResult.message || "Failed to start pipeline");
           return;
         }
 
-        if ("transcript" in transcriptionResult) {
-          setTranscript(transcriptionResult.transcript);
+        if ("pipelineId" in pipelineResult) {
+          setPipelineId(pipelineResult.pipelineId);
         }
-        setIsProcessing(false);
       };
 
       reader.onerror = () => {
-        setError("Failed to read file");
-        setIsProcessing(false);
+        setUploadError("Failed to read file");
       };
     } catch (err) {
-      setError(err instanceof Error ? err.message : "Unknown error");
-      setIsProcessing(false);
+      setUploadError(err instanceof Error ? err.message : "Unknown error");
     }
   };
 
   const handleRemoveFile = () => {
     setFile(null);
     setTranscript(null);
-    setError(null);
-    setIsProcessing(false);
+    setUploadError(null);
+    setPipelineId(null);
     setNoteContent(EMPTY_TIPTAP_DOC);
   };
 
@@ -128,10 +167,10 @@ function Component() {
           </div>
         </div>
 
-        {error && (
+        {errorMessage && (
           <div className="max-w-6xl mx-auto px-4 pt-8">
             <div className="border border-red-200 bg-red-50 rounded-sm p-4">
-              <p className="text-sm text-red-600">{error}</p>
+              <p className="text-sm text-red-600">{errorMessage}</p>
             </div>
           </div>
         )}