use redis pipeline

apps/web/app/api/cron/domains/transfer/route.ts

@@ -1,12 +1,13 @@
 import { handleAndReturnErrorResponse } from "@/lib/api/errors";
+import { linkCache } from "@/lib/api/links/cache";
 import { qstash } from "@/lib/cron";
 import { verifyQstashSignature } from "@/lib/cron/verify-qstash";
 import { prisma } from "@/lib/prisma";
 import { recordLink } from "@/lib/tinybird";
 import z from "@/lib/zod";
 import { APP_DOMAIN_WITH_NGROK, log } from "@dub/utils";
 import { NextResponse } from "next/server";
-import { sendDomainTransferredEmail, updateLinksInRedis } from "./utils";
+import { sendDomainTransferredEmail } from "./utils";
 
 const schema = z.object({
   currentWorkspaceId: z.string(),
@@ -60,7 +61,10 @@ export async function POST(req: Request) {
           where: { linkId: { in: linkIds } },
         }),
 
-        updateLinksInRedis({ links, newWorkspaceId, domain }),
+        // Update links in redis
+        linkCache.mset(
+          links.map((link) => ({ ...link, projectId: newWorkspaceId })),
+        ),
 
         // Remove the webhooks associated with the links
         prisma.linkWebhook.deleteMany({

apps/web/app/api/cron/domains/transfer/utils.ts

@@ -1,42 +1,7 @@
 import { prisma } from "@/lib/prisma";
-import { formatRedisLink, redis } from "@/lib/upstash";
-import { Link } from "@prisma/client";
 import { sendEmail } from "emails";
 import DomainTransferred from "emails/domain-transferred";
 
-// Update links in redis
-export const updateLinksInRedis = async ({
-  newWorkspaceId,
-  domain,
-  links,
-}: {
-  newWorkspaceId: string;
-  domain: string;
-  links: Link[];
-}) => {
-  const pipeline = redis.pipeline();
-
-  const formatedLinks = await Promise.all(
-    links.map(async (link) => {
-      return {
-        ...(await formatRedisLink(link)),
-        projectId: newWorkspaceId,
-        key: link.key.toLowerCase(),
-      };
-    }),
-  );
-
-  formatedLinks.map((formatedLink) => {
-    const { key, ...rest } = formatedLink;
-
-    pipeline.hset(domain.toLowerCase(), {
-      [formatedLink.key]: rest,
-    });
-  });
-
-  await pipeline.exec();
-};
-
 // Send email to the owner after the domain transfer is completed
 export const sendDomainTransferredEmail = async ({
   domain,

apps/web/app/api/domains/[domain]/route.ts

@@ -6,11 +6,11 @@ import {
 } from "@/lib/api/domains";
 import { getDomainOrThrow } from "@/lib/api/domains/get-domain-or-throw";
 import { DubApiError } from "@/lib/api/errors";
+import { linkCache } from "@/lib/api/links/cache";
 import { parseRequestBody } from "@/lib/api/utils";
 import { withWorkspace } from "@/lib/auth";
 import { prisma } from "@/lib/prisma";
 import { recordLink } from "@/lib/tinybird";
-import { redis } from "@/lib/upstash";
 import {
   DomainSchema,
   updateDomainBodySchema,
@@ -95,12 +95,8 @@ export const PATCH = withWorkspace(
     waitUntil(
       (async () => {
         if (domainUpdated) {
-          await Promise.all([
-            // remove old domain from Vercel
-            removeDomainFromVercel(domain),
-            // rename redis key
-            redis.rename(domain.toLowerCase(), newDomain.toLowerCase()),
-          ]);
+          // remove old domain from Vercel
+          await removeDomainFromVercel(domain);
 
           const allLinks = await prisma.link.findMany({
             where: {
@@ -111,6 +107,12 @@ export const PATCH = withWorkspace(
             },
           });
 
+          // rename redis keys
+          await linkCache.rename({
+            links: allLinks,
+            oldDomain: domain,
+          });
+
           // update all links in Tinybird
           recordLink(
             allLinks.map((link) => ({

apps/web/lib/api/domains/delete-domain-links.ts

@@ -1,9 +1,9 @@
 import { prisma } from "@/lib/prisma";
-import { redis } from "@/lib/upstash";
 import { R2_URL } from "@dub/utils";
 import { waitUntil } from "@vercel/functions";
 import { storage } from "../../storage";
 import { recordLink } from "../../tinybird";
+import { linkCache } from "../links/cache";
 import { removeDomainFromVercel } from "./remove-domain-vercel";
 
 /* Delete a domain and all links & images associated with it */
@@ -50,7 +50,8 @@ export async function deleteDomainAndLinks(domain: string) {
     (async () => {
       await Promise.allSettled([
         // delete all links from redis
-        redis.del(domain.toLowerCase()),
+        linkCache.deleteMany(allLinks),
+
         // record deletes in tinybird for domain & links
         recordLink([
           ...allLinks.map((link) => ({

apps/web/lib/api/links/cache.ts

@@ -19,14 +19,13 @@ class LinkCache {
       })),
     );
 
-    redisLinks.map(({ key, domain, ...redisLink }) => {
+    redisLinks.map(({ domain, key, ...redisLink }) =>
       pipeline.set(`${domain}:${key}`, JSON.stringify(redisLink), {
         ex: CACHE_EXPIRATION,
-        nx: true,
-      });
-    });
+      }),
+    );
 
-    await pipeline.exec();
+    return await pipeline.exec();
   }
 
   async set({
@@ -42,7 +41,6 @@ class LinkCache {
 
     return await redis.set(cacheKey, JSON.stringify(link), {
       ex: CACHE_EXPIRATION,
-      nx: true,
     });
   }
 
@@ -71,6 +69,25 @@ class LinkCache {
 
     return await pipeline.exec();
   }
+
+  async rename({
+    oldDomain,
+    links,
+  }: {
+    oldDomain: string;
+    links: Pick<LinkProps, "domain" | "key">[];
+  }) {
+    const pipeline = redis.pipeline();
+
+    links.forEach(({ domain, key }) => {
+      const oldCacheKey = `${oldDomain}:${key}`.toLowerCase();
+      const newCacheKey = `${domain}:${key}`.toLowerCase();
+
+      pipeline.rename(oldCacheKey, newCacheKey);
+    });
+
+    return await pipeline.exec();
+  }
 }
 
 export const linkCache = new LinkCache();

apps/web/lib/api/workspaces.ts

@@ -5,7 +5,6 @@ import { storage } from "@/lib/storage";
 import { cancelSubscription } from "@/lib/stripe";
 import { recordLink } from "@/lib/tinybird";
 import { WorkspaceProps } from "@/lib/types";
-import { redis } from "@/lib/upstash";
 import {
   DUB_DOMAINS_ARRAY,
   LEGAL_USER_ID,
@@ -55,35 +54,15 @@ export async function deleteWorkspace(
     }),
   ]);
 
-  const response = await prisma.projectUsers.deleteMany({
-    where: {
-      projectId: workspace.id,
-    },
-  });
-
   waitUntil(
     (async () => {
-      const linksByDomain: Record<string, string[]> = {};
-      defaultDomainLinks.forEach(async (link) => {
-        const { domain, key } = link;
-
-        if (!linksByDomain[domain]) {
-          linksByDomain[domain] = [];
-        }
-        linksByDomain[domain].push(key.toLowerCase());
-      });
-
-      const pipeline = redis.pipeline();
-
-      Object.entries(linksByDomain).forEach(([domain, links]) => {
-        pipeline.hdel(domain.toLowerCase(), ...links);
-      });
-
-      // delete all domains, links, and uploaded images associated with the workspace
       await Promise.allSettled([
+        // remove default domain links from redis
+        linkCache.deleteMany(defaultDomainLinks),
+
+        // delete all domains
         ...customDomains.map(({ slug }) => deleteDomainAndLinks(slug)),
-        // delete all default domain links from redis
-        pipeline.exec(),
+
         // record deletes in Tinybird for default domain links
         recordLink(
           defaultDomainLinks.map((link) => ({
@@ -97,6 +76,7 @@ export async function deleteWorkspace(
             deleted: true,
           })),
         ),
+
         // remove all images from R2
         ...defaultDomainLinks.map(({ id, image }) =>
           image && image.startsWith(`${R2_URL}/images/${id}`)
@@ -136,8 +116,6 @@ export async function deleteWorkspace(
       ]);
     })(),
   );
-
-  return response;
 }
 
 export async function deleteWorkspaceAdmin(