fix fal timeout (#113)

* fix logos

* fix fal timeout

Author: ctate

lib/steps/condition.ts

@@ -26,3 +26,4 @@ export async function conditionStep(
     Promise.resolve(evaluateCondition(input))
   );
 }
+conditionStep.maxRetries = 0;

lib/steps/database-query.ts

@@ -139,3 +139,4 @@ export async function databaseQueryStep(
   "use step";
   return withStepLogging(input, () => databaseQuery(input));
 }
+databaseQueryStep.maxRetries = 0;

lib/steps/http-request.ts

@@ -101,3 +101,4 @@ export async function httpRequestStep(
   "use step";
   return withStepLogging(input, () => httpRequest(input));
 }
+httpRequestStep.maxRetries = 0;

lib/steps/trigger.ts

@@ -54,3 +54,4 @@ export async function triggerStep(input: TriggerInput): Promise<TriggerResult> {
   // Normal trigger execution with logging
   return withStepLogging(input, () => Promise.resolve(executeTrigger(input)));
 }
+triggerStep.maxRetries = 0;

plugins/ai-gateway/steps/generate-image.ts

@@ -87,5 +87,6 @@ export async function generateImageStep(
 
   return withStepLogging(input, () => stepHandler(input, credentials));
 }
+generateImageStep.maxRetries = 0;
 
 export const _integrationType = "ai-gateway";

plugins/ai-gateway/steps/generate-text.ts

@@ -143,5 +143,6 @@ export async function generateTextStep(
 
   return withStepLogging(input, () => stepHandler(input, credentials));
 }
+generateTextStep.maxRetries = 0;
 
 export const _integrationType = "ai-gateway";

plugins/blob/steps/list.ts

@@ -119,6 +119,7 @@ export async function listBlobsStep(
 
   return withStepLogging(input, () => stepHandler(input, credentials));
 }
+listBlobsStep.maxRetries = 0;
 
 export const _integrationType = "blob";
 

plugins/blob/steps/put.ts

@@ -123,6 +123,7 @@ export async function putBlobStep(input: PutBlobInput): Promise<PutBlobResult> {
 
   return withStepLogging(input, () => stepHandler(input, credentials));
 }
+putBlobStep.maxRetries = 0;
 
 export const _integrationType = "blob";
 

plugins/fal/steps/generate-image.ts

@@ -6,6 +6,20 @@ import { getErrorMessage } from "@/lib/utils";
 import type { FalCredentials } from "../credentials";
 
 const FAL_API_URL = "https://queue.fal.run";
+const POLL_INTERVAL_MS = 1000;
+const MAX_POLL_ATTEMPTS = 600; // 10 minutes max
+
+type FalQueueResponse = {
+  status: "IN_QUEUE" | "IN_PROGRESS" | "COMPLETED";
+  request_id: string;
+  response_url: string;
+  status_url: string;
+};
+
+type FalStatusResponse = {
+  status: "IN_QUEUE" | "IN_PROGRESS" | "COMPLETED";
+  response_url?: string;
+};
 
 type FalImageResponse = {
   images?: Array<{
@@ -35,6 +49,51 @@ export type FalGenerateImageInput = StepInput &
     integrationId?: string;
   };
 
+/**
+ * Poll fal.ai queue until the request is completed
+ */
+async function pollForResult(
+  statusUrl: string,
+  responseUrl: string,
+  apiKey: string
+): Promise<FalImageResponse> {
+  for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt++) {
+    const statusResponse = await fetch(statusUrl, {
+      method: "GET",
+      headers: {
+        Authorization: `Key ${apiKey}`,
+      },
+    });
+
+    if (!statusResponse.ok) {
+      throw new Error(`Failed to check status: HTTP ${statusResponse.status}`);
+    }
+
+    const status = (await statusResponse.json()) as FalStatusResponse;
+
+    if (status.status === "COMPLETED") {
+      // Fetch the actual result
+      const resultResponse = await fetch(responseUrl, {
+        method: "GET",
+        headers: {
+          Authorization: `Key ${apiKey}`,
+        },
+      });
+
+      if (!resultResponse.ok) {
+        throw new Error(`Failed to fetch result: HTTP ${resultResponse.status}`);
+      }
+
+      return (await resultResponse.json()) as FalImageResponse;
+    }
+
+    // Wait before polling again
+    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
+  }
+
+  throw new Error("Request timed out waiting for fal.ai to complete");
+}
+
 /**
  * Core logic - portable between app and export
  */
@@ -68,7 +127,20 @@ async function stepHandler(
       throw new Error(`HTTP ${response.status}: ${errorText}`);
     }
 
-    const result = (await response.json()) as FalImageResponse;
+    const queueResponse = (await response.json()) as FalQueueResponse;
+
+    // If the response is queued, poll for the result
+    let result: FalImageResponse;
+    if (queueResponse.status === "IN_QUEUE" || queueResponse.status === "IN_PROGRESS") {
+      result = await pollForResult(
+        queueResponse.status_url,
+        queueResponse.response_url,
+        apiKey
+      );
+    } else {
+      // Immediate response (shouldn't happen with queue endpoint, but handle it)
+      result = queueResponse as unknown as FalImageResponse;
+    }
 
     if (result.error) {
       throw new Error(result.error);
@@ -103,5 +175,6 @@ export async function falGenerateImageStep(
 
   return withStepLogging(input, () => stepHandler(input, credentials));
 }
+falGenerateImageStep.maxRetries = 0;
 
 export const _integrationType = "fal";

plugins/fal/steps/generate-video.ts

@@ -6,6 +6,20 @@ import { getErrorMessage } from "@/lib/utils";
 import type { FalCredentials } from "../credentials";
 
 const FAL_API_URL = "https://queue.fal.run";
+const POLL_INTERVAL_MS = 2000;
+const MAX_POLL_ATTEMPTS = 300; // 10 minutes max for video
+
+type FalQueueResponse = {
+  status: "IN_QUEUE" | "IN_PROGRESS" | "COMPLETED";
+  request_id: string;
+  response_url: string;
+  status_url: string;
+};
+
+type FalStatusResponse = {
+  status: "IN_QUEUE" | "IN_PROGRESS" | "COMPLETED";
+  response_url?: string;
+};
 
 type FalVideoResponse = {
   video?: {
@@ -29,6 +43,49 @@ export type FalGenerateVideoInput = StepInput &
     integrationId?: string;
   };
 
+/**
+ * Poll fal.ai queue until the request is completed
+ */
+async function pollForResult(
+  statusUrl: string,
+  responseUrl: string,
+  apiKey: string
+): Promise<FalVideoResponse> {
+  for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt++) {
+    const statusResponse = await fetch(statusUrl, {
+      method: "GET",
+      headers: {
+        Authorization: `Key ${apiKey}`,
+      },
+    });
+
+    if (!statusResponse.ok) {
+      throw new Error(`Failed to check status: HTTP ${statusResponse.status}`);
+    }
+
+    const status = (await statusResponse.json()) as FalStatusResponse;
+
+    if (status.status === "COMPLETED") {
+      const resultResponse = await fetch(responseUrl, {
+        method: "GET",
+        headers: {
+          Authorization: `Key ${apiKey}`,
+        },
+      });
+
+      if (!resultResponse.ok) {
+        throw new Error(`Failed to fetch result: HTTP ${resultResponse.status}`);
+      }
+
+      return (await resultResponse.json()) as FalVideoResponse;
+    }
+
+    await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
+  }
+
+  throw new Error("Request timed out waiting for fal.ai to complete");
+}
+
 /**
  * Core logic - portable between app and export
  */
@@ -45,7 +102,6 @@ async function stepHandler(
   try {
     const model = input.model || "fal-ai/minimax-video";
 
-    // Build request body based on whether it's text-to-video or image-to-video
     const requestBody: Record<string, unknown> = {
       prompt: input.prompt,
     };
@@ -68,7 +124,18 @@ async function stepHandler(
       throw new Error(`HTTP ${response.status}: ${errorText}`);
     }
 
-    const result = (await response.json()) as FalVideoResponse;
+    const queueResponse = (await response.json()) as FalQueueResponse;
+
+    let result: FalVideoResponse;
+    if (queueResponse.status === "IN_QUEUE" || queueResponse.status === "IN_PROGRESS") {
+      result = await pollForResult(
+        queueResponse.status_url,
+        queueResponse.response_url,
+        apiKey
+      );
+    } else {
+      result = queueResponse as unknown as FalVideoResponse;
+    }
 
     if (result.error) {
       throw new Error(result.error);
@@ -100,5 +167,6 @@ export async function falGenerateVideoStep(
 
   return withStepLogging(input, () => stepHandler(input, credentials));
 }
+falGenerateVideoStep.maxRetries = 0;
 
 export const _integrationType = "fal";