Add support for async events export

steven-tey · steven-tey · commit 1e180d0a9030 · 2026-01-14T14:10:43.000-08:00
diff --git a/apps/web/app/(ee)/api/cron/events/export/fetch-events-batch.ts b/apps/web/app/(ee)/api/cron/events/export/fetch-events-batch.ts
@@ -0,0 +1,26 @@
+import { getEvents } from "@/lib/analytics/get-events";
+import { EventsFilters } from "@/lib/analytics/types";
+
+export async function* fetchEventsBatch(
+  filters: Omit<EventsFilters, "page" | "limit">,
+  pageSize: number = 1000,
+) {
+  let page = 1;
+  let hasMore = true;
+
+  while (hasMore) {
+    const events = await getEvents({
+      ...filters,
+      page,
+      limit: pageSize,
+    });
+
+    if (events.length > 0) {
+      yield { events };
+      page++;
+      hasMore = events.length === pageSize;
+    } else {
+      hasMore = false;
+    }
+  }
+}
diff --git a/apps/web/app/(ee)/api/cron/events/export/route.ts b/apps/web/app/(ee)/api/cron/events/export/route.ts
@@ -0,0 +1,143 @@
+import {
+  eventsExportColumnAccessors,
+  eventsExportColumnNames,
+} from "@/lib/analytics/events-export-helpers";
+import { convertToCSV } from "@/lib/analytics/utils/convert-to-csv";
+import { createDownloadableExport } from "@/lib/api/create-downloadable-export";
+import { handleAndReturnErrorResponse } from "@/lib/api/errors";
+import { generateExportFilename } from "@/lib/api/utils/generate-export-filename";
+import { generateRandomString } from "@/lib/api/utils/generate-random-string";
+import { verifyQstashSignature } from "@/lib/cron/verify-qstash";
+import { eventsQuerySchema } from "@/lib/zod/schemas/analytics";
+import { sendEmail } from "@dub/email";
+import ExportReady from "@dub/email/templates/export-ready";
+import { prisma } from "@dub/prisma";
+import { capitalize, log } from "@dub/utils";
+import * as z from "zod/v4";
+import { logAndRespond } from "../../utils";
+import { fetchEventsBatch } from "./fetch-events-batch";
+
+const payloadSchema = eventsQuerySchema.extend({
+  columns: z
+    .string()
+    .transform((c) => c.split(","))
+    .pipe(z.string().array()),
+  workspaceId: z.string(),
+  userId: z.string(),
+  linkId: z.string().optional(),
+  folderIds: z.array(z.string()).optional(),
+  folderId: z.string().optional(),
+  dataAvailableFrom: z.string().optional(),
+});
+
+// POST /api/cron/events/export - QStash worker for processing large event exports
+export async function POST(req: Request) {
+  try {
+    const rawBody = await req.text();
+
+    await verifyQstashSignature({
+      req,
+      rawBody,
+    });
+
+    const { columns, workspaceId, userId, ...filters } = payloadSchema.parse(
+      JSON.parse(rawBody),
+    );
+
+    const user = await prisma.user.findUnique({
+      where: {
+        id: userId,
+      },
+      select: {
+        email: true,
+      },
+    });
+
+    if (!user) {
+      return logAndRespond(`User ${userId} not found. Skipping the export.`);
+    }
+
+    if (!user.email) {
+      return logAndRespond(`User ${userId} has no email. Skipping the export.`);
+    }
+
+    const workspace = await prisma.project.findUnique({
+      where: {
+        id: workspaceId,
+      },
+      select: {
+        id: true,
+        name: true,
+        createdAt: true,
+      },
+    });
+
+    if (!workspace) {
+      return logAndRespond(
+        `Workspace ${workspaceId} not found. Skipping the export.`,
+      );
+    }
+
+    const { linkId, folderIds, folderId, dataAvailableFrom, ...eventFilters } =
+      filters;
+
+    // Fetch events in batches and build CSV
+    const allEvents: Record<string, any>[] = [];
+
+    const eventsFilters = {
+      ...eventFilters,
+      ...(linkId && { linkId }),
+      workspaceId,
+      folderIds,
+      folderId: folderId || "",
+      dataAvailableFrom: dataAvailableFrom
+        ? new Date(dataAvailableFrom)
+        : workspace.createdAt,
+    };
+
+    for await (const { events } of fetchEventsBatch(eventsFilters)) {
+      const formattedEvents = events.map((row) =>
+        Object.fromEntries(
+          columns.map((c) => [
+            eventsExportColumnNames?.[c] ?? capitalize(c),
+            eventsExportColumnAccessors[c]?.(row) ?? row?.[c],
+          ]),
+        ),
+      );
+      allEvents.push(...formattedEvents);
+    }
+
+    const csvData = convertToCSV(allEvents);
+
+    const { downloadUrl } = await createDownloadableExport({
+      fileKey: `exports/events/${generateRandomString(16)}.csv`,
+      fileName: generateExportFilename("events"),
+      body: csvData,
+      contentType: "text/csv",
+    });
+
+    await sendEmail({
+      to: user.email,
+      subject: "Your events export is ready",
+      react: ExportReady({
+        email: user.email,
+        exportType: "events",
+        downloadUrl,
+        workspace: {
+          name: workspace.name,
+        },
+      }),
+    });
+
+    return logAndRespond(
+      `Export (${allEvents.length} events) generated and email sent to user.`,
+    );
+  } catch (error) {
+    await log({
+      message: `Error exporting events: ${error.message}`,
+      type: "cron",
+    });
+
+    return handleAndReturnErrorResponse(error);
+  }
+}
diff --git a/apps/web/app/(ee)/api/events/export/route.ts b/apps/web/app/(ee)/api/events/export/route.ts
@@ -1,3 +1,8 @@
+import {
+  eventsExportColumnAccessors,
+  eventsExportColumnNames,
+} from "@/lib/analytics/events-export-helpers";
+import { getAnalytics } from "@/lib/analytics/get-analytics";
 import { getEvents } from "@/lib/analytics/get-events";
 import { getFolderIdsToFilter } from "@/lib/analytics/get-folder-ids-to-filter";
 import { convertToCSV } from "@/lib/analytics/utils";
@@ -6,41 +11,14 @@ import { getLinkOrThrow } from "@/lib/api/links/get-link-or-throw";
 import { throwIfClicksUsageExceeded } from "@/lib/api/links/usage-checks";
 import { assertValidDateRangeForPlan } from "@/lib/api/utils/assert-valid-date-range-for-plan";
 import { withWorkspace } from "@/lib/auth";
+import { qstash } from "@/lib/cron";
 import { verifyFolderAccess } from "@/lib/folder/permissions";
-import { ClickEvent, LeadEvent, SaleEvent } from "@/lib/types";
 import { eventsQuerySchema } from "@/lib/zod/schemas/analytics";
-import { COUNTRIES, capitalize } from "@dub/utils";
+import { APP_DOMAIN_WITH_NGROK, capitalize } from "@dub/utils";
+import { NextResponse } from "next/server";
 import * as z from "zod/v4";
 
-type Row = ClickEvent | LeadEvent | SaleEvent;
-
-const columnNames: Record<string, string> = {
-  trigger: "Event",
-  url: "Destination URL",
-  os: "OS",
-  referer: "Referrer",
-  refererUrl: "Referrer URL",
-  timestamp: "Date",
-  invoiceId: "Invoice ID",
-  saleAmount: "Sale Amount",
-  clickId: "Click ID",
-};
-
-const columnAccessors = {
-  trigger: (r: Row) => r.click.trigger,
-  event: (r: LeadEvent | SaleEvent) => r.eventName,
-  url: (r: ClickEvent) => r.click.url,
-  link: (r: Row) => r.domain + (r.key === "_root" ? "" : `/${r.key}`),
-  country: (r: Row) =>
-    r.country ? COUNTRIES[r.country] ?? r.country : r.country,
-  referer: (r: ClickEvent) => r.click.referer,
-  refererUrl: (r: ClickEvent) => r.click.refererUrl,
-  customer: (r: LeadEvent | SaleEvent) =>
-    r.customer.name + (r.customer.email ? ` <${r.customer.email}>` : ""),
-  invoiceId: (r: SaleEvent) => r.sale.invoiceId,
-  saleAmount: (r: SaleEvent) => "$" + (r.sale.amount / 100).toFixed(2),
-  clickId: (r: ClickEvent) => r.click.id,
-};
+const MAX_EVENTS_TO_EXPORT = 1000;
 
 // GET /api/events/export – get export data for analytics
 export const GET = withWorkspace(
@@ -93,20 +71,57 @@ export const GET = withWorkspace(
           userId: session.user.id,
         });
 
+    // Count events using getAnalytics with groupBy: "count"
+    const countResponse = await getAnalytics({
+      ...parsedParams,
+      groupBy: "count",
+      ...(link && { linkId: link.id }),
+      folderIds,
+      workspaceId: workspace.id,
+      dataAvailableFrom: workspace.createdAt,
+    });
+
+    // Extract the count based on event type
+    // getAnalytics with groupBy: "count" returns an object like { clicks: 123 } or { leads: 45 } or { sales: 10, saleAmount: 5000 }
+    const eventsCount =
+      typeof countResponse === "object" && countResponse !== null
+        ? (countResponse[event as keyof typeof countResponse] as number) ?? 0
+        : typeof countResponse === "number"
+          ? countResponse
+          : 0;
+
+    // Process the export in the background if the number of events is greater than MAX_EVENTS_TO_EXPORT
+    if (eventsCount > MAX_EVENTS_TO_EXPORT) {
+      await qstash.publishJSON({
+        url: `${APP_DOMAIN_WITH_NGROK}/api/cron/events/export`,
+        body: {
+          ...searchParams,
+          workspaceId: workspace.id,
+          userId: session.user.id,
+          ...(link && { linkId: link.id }),
+          folderIds: folderIds ? folderIds : undefined,
+          folderId: folderId || "",
+          dataAvailableFrom: workspace.createdAt.toISOString(),
+        },
+      });
+
+      return NextResponse.json({}, { status: 202 });
+    }
+
     const response = await getEvents({
       ...parsedParams,
       ...(link && { linkId: link.id }),
       workspaceId: workspace.id,
-      limit: 100000,
+      limit: MAX_EVENTS_TO_EXPORT,
       folderIds,
       folderId: folderId || "",
     });
 
     const data = response.map((row) =>
       Object.fromEntries(
         columns.map((c) => [
-          columnNames?.[c] ?? capitalize(c),
-          columnAccessors[c]?.(row) ?? row?.[c],
+          eventsExportColumnNames?.[c] ?? capitalize(c),
+          eventsExportColumnAccessors[c]?.(row) ?? row?.[c],
         ]),
       ),
     );
diff --git a/apps/web/lib/analytics/events-export-helpers.ts b/apps/web/lib/analytics/events-export-helpers.ts
@@ -0,0 +1,32 @@
+import { ClickEvent, LeadEvent, SaleEvent } from "@/lib/types";
+import { COUNTRIES } from "@dub/utils";
+
+export type Row = ClickEvent | LeadEvent | SaleEvent;
+
+export const eventsExportColumnNames: Record<string, string> = {
+  trigger: "Event",
+  url: "Destination URL",
+  os: "OS",
+  referer: "Referrer",
+  refererUrl: "Referrer URL",
+  timestamp: "Date",
+  invoiceId: "Invoice ID",
+  saleAmount: "Sale Amount",
+  clickId: "Click ID",
+};
+
+export const eventsExportColumnAccessors = {
+  trigger: (r: Row) => r.click.trigger,
+  event: (r: LeadEvent | SaleEvent) => r.eventName,
+  url: (r: ClickEvent) => r.click.url,
+  link: (r: Row) => r.domain + (r.key === "_root" ? "" : `/${r.key}`),
+  country: (r: Row) =>
+    r.country ? COUNTRIES[r.country] ?? r.country : r.country,
+  referer: (r: ClickEvent) => r.click.referer,
+  refererUrl: (r: ClickEvent) => r.click.refererUrl,
+  customer: (r: LeadEvent | SaleEvent) =>
+    r.customer.name + (r.customer.email ? ` <${r.customer.email}>` : ""),
+  invoiceId: (r: SaleEvent) => r.sale.invoiceId,
+  saleAmount: (r: SaleEvent) => "$" + (r.sale.amount / 100).toFixed(2),
+  clickId: (r: ClickEvent) => r.click.id,
+};
diff --git a/apps/web/ui/analytics/events/events-export-button.tsx b/apps/web/ui/analytics/events/events-export-button.tsx
@@ -1,5 +1,6 @@
 import useWorkspace from "@/lib/swr/use-workspace";
 import { Button, Download, TooltipContent } from "@dub/ui";
+import { useSession } from "next-auth/react";
 import { Dispatch, SetStateAction, useContext } from "react";
 import { toast } from "sonner";
 import { AnalyticsContext } from "../analytics-provider";
@@ -13,6 +14,7 @@ export function EventsExportButton({
   const { exportQueryString } = useContext(EventsContext);
   const { eventsApiPath } = useContext(AnalyticsContext);
   const { slug, plan } = useWorkspace();
+  const { data: session } = useSession();
 
   const needsHigherPlan = plan === "free" || plan === "pro";
 
@@ -31,12 +33,25 @@ export function EventsExportButton({
       throw new Error(response.statusText);
     }
 
+    if (response.status === 202) {
+      setOpenPopover(false);
+      return {
+        isAsync: true,
+        message: `Your export is being processed and we'll send you an email (${session?.user?.email}) when it's ready to download.`,
+      };
+    }
+
     const blob = await response.blob();
     const url = window.URL.createObjectURL(blob);
     const a = document.createElement("a");
     a.href = url;
     a.download = `Dub Events Export - ${new Date().toISOString()}.csv`;
     a.click();
+    setOpenPopover(false);
+    return {
+      isAsync: false,
+      message: "Exported successfully",
+    };
   }
 
   return (
@@ -57,10 +72,9 @@ export function EventsExportButton({
       onClick={() => {
         toast.promise(exportData(), {
           loading: "Exporting file...",
-          success: "Exported successfully",
+          success: (data) => data.message,
           error: (error) => error,
         });
-        setOpenPopover(false);
       }}
     />
   );
diff --git a/packages/email/src/templates/export-ready.tsx b/packages/email/src/templates/export-ready.tsx
@@ -24,7 +24,7 @@ export default function ExportReady({
 }: {
   email: string;
   downloadUrl: string;
-  exportType: "partners" | "commissions" | "links";
+  exportType: "partners" | "commissions" | "links" | "events";
   expiresInDays?: number;
   program?: {
     name: string;
@@ -48,8 +48,9 @@ export default function ExportReady({
               Your {exportType} export is ready
             </Heading>
             <Text className="text-sm leading-6 text-black">
-              Your export of {exportType} from <strong>{contextName}</strong>{" "}
-              has been completed and is ready to download.
+              Your {exportType} export from your <strong>{contextName}</strong>{" "}
+              {program ? "program" : "workspace"} has been completed and is
+              ready to download.
             </Text>
             <Section className="my-8">
               <Link
