Merge pull request #2924 from dubinc/tb-migration

Tinybird migration

Author: steven-tey

apps/web/app/(ee)/api/cron/cleanup/link-retention/route.ts

@@ -2,7 +2,7 @@ import { handleAndReturnErrorResponse } from "@/lib/api/errors";
 import { qstash } from "@/lib/cron";
 import { verifyQstashSignature } from "@/lib/cron/verify-qstash";
 import { verifyVercelSignature } from "@/lib/cron/verify-vercel";
-import { recordLinkTB, transformLinkTB } from "@/lib/tinybird";
+import { recordLink } from "@/lib/tinybird";
 import { prisma } from "@dub/prisma";
 import { APP_DOMAIN_WITH_NGROK } from "@dub/utils";
 import { Domain } from "@prisma/client";
@@ -125,12 +125,7 @@ async function deleteOldLinks(
 
     // // Record the links deletion in Tinybird
     // // not 100% sure if we need this yet, maybe we should just delete the link completely from TB to save space?
-    await recordLinkTB(
-      links.map((link) => ({
-        ...transformLinkTB(link),
-        deleted: true,
-      })),
-    );
+    await recordLink(links, { deleted: true });
 
     console.log(
       `[Link retention cleanup] Deleted ${links.length} links for ${domain.slug} that are older than ${domain.linkRetentionDays} days!`,

apps/web/app/(ee)/api/cron/framer/backfill-leads-batch/route.ts

@@ -300,6 +300,18 @@ export const POST = withWorkspace(async ({ req, workspace }) => {
           body: dataArray.map((d) => JSON.stringify(d.clickData)).join("\n"),
         },
       ),
+      // TODO: Remove after Tinybird migration
+      fetch(
+        `${process.env.TINYBIRD_API_URL}/v0/events?name=dub_click_events&wait=true`,
+        {
+          method: "POST",
+          headers: {
+            Authorization: `Bearer ${process.env.TINYBIRD_API_KEY_NEW}`,
+            "Content-Type": "application/x-ndjson",
+          },
+          body: dataArray.map((d) => JSON.stringify(d.clickData)).join("\n"),
+        },
+      ),
 
       // Record leads
       recordLeadWithTimestamp(dataArray.map((d) => d.leadEventData)),

apps/web/lib/actions/partners/revoke-program-invite.ts

@@ -2,7 +2,7 @@
 
 import { linkCache } from "@/lib/api/links/cache";
 import { getDefaultProgramIdOrThrow } from "@/lib/api/programs/get-default-program-id-or-throw";
-import { recordLinkTB, transformLinkTB } from "@/lib/tinybird";
+import { recordLink } from "@/lib/tinybird";
 import { prisma } from "@dub/prisma";
 import { waitUntil } from "@vercel/functions";
 import z from "../../zod";
@@ -72,12 +72,7 @@ export const revokeProgramInviteAction = authActionClient
         linkCache.expireMany(partnerLinks),
 
         // Record the links deletion in Tinybird
-        recordLinkTB(
-          partnerLinks.map((link) => ({
-            ...transformLinkTB(link),
-            deleted: true,
-          })),
-        ),
+        recordLink(partnerLinks, { deleted: true }),
 
         // Update totalLinks for the workspace
         prisma.project.update({

apps/web/lib/api/audit-logs/record-audit-log.ts

@@ -1,7 +1,7 @@
 import { getIP } from "@/lib/api/utils/get-ip";
-import { tb } from "@/lib/tinybird";
+import { tb, tbNew } from "@/lib/tinybird";
 import { log } from "@dub/utils";
-import { ipAddress as getIPAddress } from "@vercel/functions";
+import { ipAddress as getIPAddress, waitUntil } from "@vercel/functions";
 import { headers } from "next/headers";
 import { z } from "zod";
 import { createId } from "../create-id";
@@ -56,6 +56,7 @@ export const recordAuditLog = async (data: AuditLogInput | AuditLogInput[]) => {
     : [transformAuditLogTB(data, { headersList, ipAddress })];
 
   try {
+    waitUntil(recordAuditLogTBNew(auditLogs));
     await recordAuditLogTB(auditLogs);
   } catch (error) {
     console.error(
@@ -77,3 +78,10 @@ const recordAuditLogTB = tb.buildIngestEndpoint({
   event: auditLogSchemaTB,
   wait: true,
 });
+
+// TODO: Remove after Tinybird migration
+const recordAuditLogTBNew = tbNew.buildIngestEndpoint({
+  datasource: "dub_audit_logs",
+  event: auditLogSchemaTB,
+  wait: true,
+});

apps/web/lib/api/conversions/track-lead.ts

@@ -4,7 +4,7 @@ import { includeTags } from "@/lib/api/links/include-tags";
 import { generateRandomName } from "@/lib/names";
 import { createPartnerCommission } from "@/lib/partners/create-partner-commission";
 import { isStored, storage } from "@/lib/storage";
-import { getClickEvent, recordLead, recordLeadSync } from "@/lib/tinybird";
+import { getClickEvent, recordLead } from "@/lib/tinybird";
 import { logConversionEvent } from "@/lib/tinybird/log-conversion-events";
 import { ClickEventTB, WebhookPartner, WorkspaceProps } from "@/lib/types";
 import { redis } from "@/lib/upstash";
@@ -214,9 +214,6 @@ export const trackLead = async ({
         : leadEventPayload;
 
       await Promise.all([
-        // Use recordLeadSync which waits for the operation to complete
-        recordLeadSync(leadEventPayload),
-
         // Cache the latest lead event for 5 minutes because the ingested event is not available immediately on Tinybird
         // we're setting two keys because we want to support the use case where the customer has multiple lead events
         redis.set(`leadCache:${customer.id}`, cacheLeadEventPayload, {
@@ -235,9 +232,8 @@ export const trackLead = async ({
 
     waitUntil(
       (async () => {
-        // for async mode, record the lead event in the background
-        // for deferred mode, defer the lead event creation to a subsequent request
-        if (mode === "async") {
+        // for deferred mode, we defer the lead event creation to a subsequent request
+        if (mode !== "deferred") {
           await recordLead(createLeadEventPayload(customer.id));
         }
 

apps/web/lib/api/links/bulk-delete-links.ts

@@ -1,5 +1,5 @@
 import { storage } from "@/lib/storage";
-import { recordLinkTB, transformLinkTB } from "@/lib/tinybird";
+import { recordLink } from "@/lib/tinybird";
 import { prisma } from "@dub/prisma";
 import { R2_URL } from "@dub/utils";
 import { linkCache } from "./cache";
@@ -15,12 +15,7 @@ export async function bulkDeleteLinks(links: ExpandedLink[]) {
     linkCache.deleteMany(links),
 
     // Record the links deletion in Tinybird
-    recordLinkTB(
-      links.map((link) => ({
-        ...transformLinkTB(link),
-        deleted: true,
-      })),
-    ),
+    recordLink(links, { deleted: true }),
 
     // For links that have an image, delete the image from R2
     links

apps/web/lib/api/links/delete-link.ts

@@ -1,5 +1,5 @@
 import { storage } from "@/lib/storage";
-import { recordLinkTB, transformLinkTB } from "@/lib/tinybird";
+import { recordLink } from "@/lib/tinybird";
 import { prisma } from "@dub/prisma";
 import { R2_URL } from "@dub/utils";
 import { waitUntil } from "@vercel/functions";
@@ -30,10 +30,7 @@ export async function deleteLink(linkId: string) {
       linkCache.delete(link),
 
       // Record link in the Tinybird
-      recordLinkTB({
-        ...transformLinkTB(link),
-        deleted: true,
-      }),
+      recordLink(link, { deleted: true }),
 
       link.projectId &&
         prisma.project.update({

apps/web/lib/tinybird/client.ts

@@ -4,3 +4,9 @@ export const tb = new Tinybird({
   token: process.env.TINYBIRD_API_KEY as string,
   baseUrl: process.env.TINYBIRD_API_URL as string,
 });
+
+// TODO: Remove after Tinybird migration
+export const tbNew = new Tinybird({
+  token: process.env.TINYBIRD_API_KEY_NEW as string,
+  baseUrl: process.env.TINYBIRD_API_URL as string,
+});

apps/web/lib/tinybird/log-conversion-events.ts

@@ -1,5 +1,6 @@
+import { waitUntil } from "@vercel/functions";
 import { z } from "zod";
-import { tb } from "./client";
+import { tb, tbNew } from "./client";
 
 const schema = z.object({
   workspace_id: z.string(),
@@ -10,7 +11,18 @@ const schema = z.object({
 });
 
 // Log the conversion events for debugging purposes
-export const logConversionEvent = tb.buildIngestEndpoint({
+export const logConversionEventTB = tb.buildIngestEndpoint({
   datasource: "dub_conversion_events_log",
   event: schema,
 });
+
+// TODO: Remove after Tinybird migration
+export const logConversionEventTBNew = tbNew.buildIngestEndpoint({
+  datasource: "dub_conversion_events_log",
+  event: schema,
+});
+
+export const logConversionEvent = async (payload: any) => {
+  waitUntil(logConversionEventTBNew(payload));
+  return await logConversionEventTB(payload);
+};

apps/web/lib/tinybird/log-import-error.ts

@@ -1,7 +1,19 @@
+import { waitUntil } from "@vercel/functions";
 import { importErrorLogSchema } from "../zod/schemas/import-error-log";
-import { tb } from "./client";
+import { tb, tbNew } from "./client";
 
-export const logImportError = tb.buildIngestEndpoint({
+export const logImportErrorTB = tb.buildIngestEndpoint({
   datasource: "dub_import_error_logs",
   event: importErrorLogSchema,
 });
+
+// TODO: Remove after Tinybird migration
+export const logImportErrorTBNew = tbNew.buildIngestEndpoint({
+  datasource: "dub_import_error_logs",
+  event: importErrorLogSchema,
+});
+
+export const logImportError = async (payload: any) => {
+  waitUntil(logImportErrorTBNew(payload));
+  return await logImportErrorTB(payload);
+};