feat(restate): add storage cleanup service for orphaned audio files (#2189)

- Add StorageCleanup service with cleanupOldFiles handler
- Lists all files in audio-files bucket and deletes those older than cutoff
- Add listFiles and listAllFiles functions to supabase.ts
- Add logging to existing deletion catch block in stt-file.ts
- Register new service in index.ts

This addresses the issue where audio files could become orphaned in Supabase
storage when:
- Upload succeeds but pipeline start fails
- Deletion silently fails
- Restate workflow crashes before finally block
- Provider callback never arrives

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: yujonglee <[email protected]>

Author: devin-ai-integration[bot]

apps/restate/src/index.ts

@@ -2,13 +2,14 @@ import * as restate from "@restatedev/restate-sdk-cloudflare-workers/fetch";
 
 import { type Env, envSchema } from "./env";
 import { rateLimiter } from "./services/rate-limit";
+import { storageCleanup } from "./services/storage-cleanup";
 import { sttFile } from "./services/stt-file";
 
 export default {
   fetch(request: Request, _env: Env, _ctx: ExecutionContext) {
     const env = envSchema.parse(_env);
     return restate.createEndpointHandler({
-      services: [rateLimiter, sttFile],
+      services: [rateLimiter, sttFile, storageCleanup],
       ...(env.RESTATE_IDENTITY_KEY
         ? { identityKeys: [env.RESTATE_IDENTITY_KEY] }
         : {}),

apps/restate/src/services/storage-cleanup.ts

@@ -0,0 +1,75 @@
+import * as restate from "@restatedev/restate-sdk-cloudflare-workers/fetch";
+import { serde } from "@restatedev/restate-sdk-zod";
+import { z } from "zod";
+
+import { type Env } from "../env";
+import { deleteFile, listAllFiles } from "../supabase";
+
+const CleanupInput = z.object({
+  cutoffHours: z.number().min(1).default(24),
+});
+
+export type CleanupInputType = z.infer<typeof CleanupInput>;
+
+const CleanupResult = z.object({
+  deletedCount: z.number(),
+  failedCount: z.number(),
+  totalScanned: z.number(),
+  errors: z.array(z.string()),
+});
+
+export type CleanupResultType = z.infer<typeof CleanupResult>;
+
+export const storageCleanup = restate.service({
+  name: "StorageCleanup",
+  handlers: {
+    cleanupOldFiles: restate.handlers.handler(
+      { input: serde.zod(CleanupInput) },
+      async (
+        ctx: restate.Context,
+        input: CleanupInputType,
+      ): Promise<CleanupResultType> => {
+        const env = ctx.request().extraArgs[0] as Env;
+        const cutoffMs = input.cutoffHours * 60 * 60 * 1000;
+        const cutoffDate = new Date(Date.now() - cutoffMs);
+
+        const files = await ctx.run("list-files", () => listAllFiles(env));
+
+        let deletedCount = 0;
+        let failedCount = 0;
+        const errors: string[] = [];
+
+        for (const file of files) {
+          const fileDate = new Date(file.created_at);
+          if (fileDate < cutoffDate) {
+            const filePath = file.name;
+            try {
+              await ctx.run(`delete-${filePath}`, () =>
+                deleteFile(env, filePath),
+              );
+              deletedCount++;
+            } catch (err) {
+              failedCount++;
+              const errorMsg =
+                err instanceof Error ? err.message : "Unknown error";
+              errors.push(`Failed to delete ${filePath}: ${errorMsg}`);
+              if (errors.length >= 10) {
+                errors.push("... (truncated, too many errors)");
+                break;
+              }
+            }
+          }
+        }
+
+        return {
+          deletedCount,
+          failedCount,
+          totalScanned: files.length,
+          errors,
+        };
+      },
+    ),
+  },
+});
+
+export type StorageCleanup = typeof storageCleanup;

apps/restate/src/services/stt-file.ts

@@ -87,9 +87,16 @@ export const sttFile = restate.workflow({
           ctx.set("error", error);
           throw err;
         } finally {
-          await ctx.run("cleanup", () =>
-            deleteFile(env, input.fileId).catch(() => {}),
-          );
+          await ctx.run("cleanup", async () => {
+            try {
+              await deleteFile(env, input.fileId);
+            } catch (err) {
+              console.error("Failed to delete audio file from storage", {
+                fileId: input.fileId,
+                error: err instanceof Error ? err.message : String(err),
+              });
+            }
+          });
         }
       },
     ),

apps/restate/src/supabase.ts

@@ -79,3 +79,63 @@ export async function deleteFile(
     throw new Error(`Failed to delete file: ${response.status} ${body}`);
   }
 }
+
+export interface StorageFile {
+  name: string;
+  id: string;
+  created_at: string;
+  updated_at: string;
+  metadata: Record<string, unknown>;
+}
+
+export async function listFiles(
+  env: SupabaseEnv,
+  prefix: string = "",
+  limit: number = 100,
+  offset: number = 0,
+): Promise<StorageFile[]> {
+  const { url, serviceRoleKey } = getSupabaseConfig(env);
+
+  const response = await fetch(`${url}/storage/v1/object/list/audio-files`, {
+    method: "POST",
+    headers: {
+      Authorization: `Bearer ${serviceRoleKey}`,
+      apikey: serviceRoleKey,
+      "Content-Type": "application/json",
+    },
+    body: JSON.stringify({
+      prefix,
+      limit,
+      offset,
+    }),
+  });
+
+  if (!response.ok) {
+    const body = await response.text();
+    throw new Error(`Failed to list files: ${response.status} ${body}`);
+  }
+
+  return (await response.json()) as StorageFile[];
+}
+
+export async function listAllFiles(env: SupabaseEnv): Promise<StorageFile[]> {
+  const allFiles: StorageFile[] = [];
+  let offset = 0;
+  const limit = 100;
+
+  while (true) {
+    const files = await listFiles(env, "", limit, offset);
+    if (files.length === 0) break;
+
+    for (const file of files) {
+      if (file.id) {
+        allFiles.push(file);
+      }
+    }
+
+    if (files.length < limit) break;
+    offset += limit;
+  }
+
+  return allFiles;
+}