feat: migrate dashboard reindexing to qstash logic for async handling (#3701)

Author: sahil485

packages/fern-dashboard/src/app/actions/toggleAskAi.ts

@@ -75,5 +75,5 @@ export async function getToggleStatus({
 }): Promise<string> {
   const faiClient = getFaiClient({ token: process.env.FERN_TOKEN ?? "" });
   const response = await faiClient.settings.getToggleStatus({ domain });
-  return response.status || "unknown";
+  return response.status || "failed";
 }

packages/fern-dashboard/src/components/settings/ToggleAskAiButton.tsx

@@ -99,8 +99,6 @@ export function ToggleAskAiButton({
     pollIntervalRef.current = setInterval(() => {
       void pollJobStatus();
     }, 7000);
-
-    void pollJobStatus();
   }, [pollJobStatus]);
 
   useEffect(() => {

packages/fern-docs/bundle/src/app/[host]/[domain]/api/fern-docs/revalidate/route.ts

@@ -39,7 +39,7 @@ import {
   queueTurbopufferReindex,
 } from "@/server/queue-reindex";
 
-export const maxDuration = 600; // 10 minutes timeout
+export const maxDuration = 800; // 13 minutes timeout
 
 export async function GET(
   req: NextRequest,
@@ -94,7 +94,7 @@ export async function GET(
           // reindex unless explicitly disabled
           req.nextUrl.searchParams.get("reindex") !== "false"
         ) {
-          reindexPromise = reindex(docs, host, domain)
+          reindexPromise = reindex(docs, host, domain, maxDuration)
             .then((services) => {
               controller.enqueue(
                 `reindex-queued:services=${services.join(",")}\n`
@@ -358,7 +358,8 @@ export async function GET(
 async function reindex(
   docs: DocsV2Read.LoadDocsForUrlResponse,
   host: string,
-  domain: string
+  domain: string,
+  maxDuration: number
 ) {
   const { basePath } = docs.baseUrl;
 
@@ -371,7 +372,12 @@ async function reindex(
   ).ask_ai_enabled;
 
   if (isAskAiEnabled) {
-    await queueTurbopufferReindex(host, withoutStaging(domain), basePath);
+    await queueTurbopufferReindex(
+      host,
+      withoutStaging(domain),
+      basePath,
+      maxDuration
+    );
     return ["algolia", "turbopuffer"];
   }
   return ["algolia"];

packages/fern-docs/bundle/src/app/[host]/[domain]/api/fern-docs/search/v2/reindex/turbopuffer/start/route.ts

@@ -1,31 +1,13 @@
 import { NextRequest, NextResponse } from "next/server";
 
-import { createOpenAI } from "@ai-sdk/openai";
-
-import { createCachedDocsLoader } from "@fern-api/docs-loader";
-import {
-  fdrEnvironment,
-  fernToken_admin,
-  getFaiOrigin,
-  openaiApiKey,
-  turbopufferApiKey,
-} from "@fern-api/docs-server/env-variables";
 import { isLocal } from "@fern-api/docs-server/isLocal";
 import { isSelfHosted } from "@fern-api/docs-server/isSelfHosted";
-import { postToSlack } from "@fern-api/docs-server/slack";
-import { Gate, withBasicTokenAnonymous } from "@fern-api/docs-server/withRbac";
+import { loadWithUrl } from "@fern-api/docs-server/loadWithUrl";
 import { getDocsDomainEdge } from "@fern-api/docs-server/xfernhost/edge";
-import { slugToHref, withoutStaging } from "@fern-api/docs-utils";
-import { FernAIClient } from "@fern-api/fai-sdk";
-import { getAuthEdgeConfig, getEdgeFlags } from "@fern-docs/edge-config";
-import {
-  getFernDocsIndexName,
-  getTurbopufferNamespace,
-  getTurbopufferVectorizer,
-  turbopufferUpsertTask,
-} from "@fern-docs/search-ask-fern";
+import { withoutStaging } from "@fern-api/docs-utils";
 
-import { JobManager, createJobResponse } from "@/jobs";
+import { createJobResponse } from "@/jobs";
+import { queueTurbopufferStartReindex } from "@/server/queue-reindex";
 
 export const maxDuration = 800; // 13 minutes
 
@@ -42,111 +24,22 @@ export async function GET(req: NextRequest): Promise<NextResponse> {
   const deleteExisting =
     req.nextUrl.searchParams.get("deleteExisting") === "true";
 
-  const fernDocsIndexName = getFernDocsIndexName();
-  const namespace = getTurbopufferNamespace(domain, fernDocsIndexName);
-
-  const job_id = await JobManager.createJob(domain);
-
-  JobManager.executeJob(domain, async () => {
-    const openai = createOpenAI({ apiKey: openaiApiKey() });
-    const embeddingModel = openai.embedding("text-embedding-3-large");
-
-    try {
-      const loader = await createCachedDocsLoader(host, domain);
-      const metadata = await loader.getMetadata();
-      if (metadata == null) {
-        throw new Error("Documentation not found");
-      }
-      if (metadata.isPreview) {
-        return {
-          message: "Preview sites are not indexed",
-          added: 0,
-          domain,
-          namespace,
-        };
-      }
-
-      const [authEdgeConfig, edgeFlags] = await Promise.all([
-        getAuthEdgeConfig(domain),
-        getEdgeFlags(domain),
-      ]);
-
-      console.log("Starting turbopuffer upsert for namespace: ", namespace);
-
-      const numInserted = await turbopufferUpsertTask({
-        apiKey: turbopufferApiKey(),
-        namespace,
-        payload: {
-          environment: fdrEnvironment(),
-          fernToken: fernToken_admin(),
-          domain: withoutStaging(domain),
-          ...edgeFlags,
-        },
-        vectorizer: getTurbopufferVectorizer(embeddingModel),
-        authed: (node) => {
-          if (authEdgeConfig == null) {
-            return false;
-          }
-          return (
-            withBasicTokenAnonymous(authEdgeConfig, slugToHref(node.slug)) ===
-            Gate.DENY
-          );
-        },
-        deleteExisting,
-      });
-
-      const faiClient = new FernAIClient({
-        baseUrl: getFaiOrigin(),
-      });
-
-      console.log("Syncing index to query index for domain: ", domain);
-
-      const syncResponse = await faiClient.index.syncIndexToQueryIndex(domain, {
-        index_name: fernDocsIndexName,
-      });
-
-      console.log("Synced finished for domain: ", domain);
-
-      const pollJobStatus = async (jobId: string): Promise<void> => {
-        while (true) {
-          const statusResponse = await faiClient.index.getJobStatus(jobId);
-          const { status, success, error } = statusResponse;
-
-          if (status === "completed") {
-            if (success === false) {
-              throw new Error(`Sync job failed: ${error || "Unknown error"}`);
-            }
-            break;
-          } else if (status === "failed") {
-            throw new Error(`Sync job failed: ${error || "Unknown error"}`);
-          }
-
-          await new Promise((resolve) => setTimeout(resolve, 15000));
-        }
-      };
-
-      await pollJobStatus(syncResponse.job_id);
-
-      return {
-        added: numInserted,
-        domain,
-        namespace,
-        message: "Turbopuffer reindex completed successfully",
-      };
-    } catch (error) {
-      console.error(`[turbopuffer-start] ${JSON.stringify(error)}`);
-
-      postToSlack(
-        "#search-notifs",
-        `:rotating_light: [TURBOPUFFER-START] Failed to reindex for ${domain} with the following error: ${String(error)}`,
-        "turbopuffer-reindex-start"
-      );
-
-      throw error;
-    }
-  }).catch((error) => {
-    console.error(`Job ${job_id} execution failed:`, error);
-  });
+  const docs = await loadWithUrl(domain);
+  const { basePath } = docs.baseUrl;
+
+  const messageId = await queueTurbopufferStartReindex(
+    host,
+    withoutStaging(domain),
+    basePath,
+    deleteExisting,
+    maxDuration
+  );
+
+  if (!messageId) {
+    return NextResponse.json("Failed to queue turbopuffer reindex", {
+      status: 400,
+    });
+  }
 
-  return createJobResponse("Turbopuffer reindex job started", job_id);
+  return createJobResponse(messageId, "in_progress");
 }

packages/fern-docs/bundle/src/app/[host]/[domain]/api/fern-docs/search/v2/reindex/turbopuffer/status/route.ts

@@ -2,9 +2,9 @@ import { NextRequest, NextResponse } from "next/server";
 
 import { isLocal } from "@fern-api/docs-server/isLocal";
 import { isSelfHosted } from "@fern-api/docs-server/isSelfHosted";
-import { getDocsDomainEdge } from "@fern-api/docs-server/xfernhost/edge";
 
-import { JobManager, createJobStatusResponse } from "@/jobs";
+import { createJobStatusResponse } from "@/jobs";
+import { getMessageStatus } from "@/server/queue";
 
 export async function GET(req: NextRequest): Promise<NextResponse> {
   if (isLocal() || isSelfHosted()) {
@@ -23,7 +23,6 @@ export async function GET(req: NextRequest): Promise<NextResponse> {
     );
   }
 
-  const domain = getDocsDomainEdge(req);
-  const job = await JobManager.getJobStatus(domain);
-  return createJobStatusResponse(job);
+  const status = await getMessageStatus(jobId);
+  return createJobStatusResponse(jobId, status);
 }

packages/fern-docs/bundle/src/jobs.ts

@@ -1,7 +1,6 @@
 import { NextResponse } from "next/server";
 
 import { kv } from "@vercel/kv";
-import { randomUUID } from "crypto";
 
 interface JobStatus {
   id: string;
@@ -13,80 +12,47 @@ interface JobStatus {
   result?: any;
 }
 
-export class JobManager {
-  static async createJob(domain: string): Promise<string> {
-    const job: JobStatus = {
-      id: randomUUID(),
-      status: "pending",
-      created_at: Date.now(),
-    };
-
-    await kv.hset(domain, { [`tpuf_job`]: job });
-    return job.id;
-  }
-
-  static async executeJob<T>(
-    domain: string,
-    task: () => Promise<T>
-  ): Promise<void> {
-    const job = await kv.hget<JobStatus>(domain, `tpuf_job`);
-    if (!job) {
-      throw new Error("Job not found");
-    }
-
-    job.status = "in_progress";
-    job.started_at = Date.now();
-    await kv.hset(domain, { [`tpuf_job`]: job });
-
-    try {
-      const result = await task();
-      job.status = "completed";
-      job.completed_at = Date.now();
-      job.result = result;
-    } catch (error) {
-      job.status = "failed";
-      job.completed_at = Date.now();
-      job.error = error instanceof Error ? error.message : String(error);
-    }
-
-    await kv.hset(domain, { [`tpuf_job`]: job });
-  }
-
-  static async getJobStatus(domain: string): Promise<JobStatus | undefined> {
-    const job = await kv.hget<JobStatus>(domain, `tpuf_job`);
-    return job ?? undefined;
-  }
+export async function createJob(
+  domain: string,
+  messageId: string
+): Promise<string> {
+  const job: JobStatus = {
+    id: messageId,
+    status: "in_progress",
+    created_at: Date.now(),
+  };
+
+  await kv.hset(domain, { [`tpuf_job`]: job });
+  return job.id;
+}
 
-  static async getJob(domain: string): Promise<JobStatus | undefined> {
-    const job = await kv.hget<JobStatus>(domain, `tpuf_job`);
-    return job ?? undefined;
-  }
+export async function getJobStatus(
+  domain: string
+): Promise<JobStatus | undefined> {
+  const job = await kv.hget<JobStatus>(domain, `tpuf_job`);
+  return job ?? undefined;
 }
 
 export function createJobResponse(
-  message: string = "Job created successfully",
-  job_id: string
+  job_id: string,
+  status: "completed" | "failed" | "in_progress"
 ) {
   return NextResponse.json({
-    message,
     job_id,
+    status,
   });
 }
 
-export function createJobStatusResponse(job: JobStatus | undefined) {
-  if (!job) {
+export function createJobStatusResponse(
+  jobId: string,
+  status: "completed" | "failed" | "in_progress"
+) {
+  if (!status) {
     return NextResponse.json({ error: "Job not found" }, { status: 404 });
   }
 
   return NextResponse.json({
-    id: job.id,
-    status: job.status,
-    completed: job.status === "completed",
-    failed: job.status === "failed",
-    created_at: job.created_at,
-    started_at: job.started_at,
-    completed_at: job.completed_at,
-    error: job.error,
-    result: job.result,
+    id: jobId,
+    status,
   });
 }