Caching & Rate Limiting

Copyimport { Redis } from "@upstash/redis";
 
// Initiate Redis instance by connecting to REST URL
export const redis = new Redis({
  url: process.env.UPSTASH_REDIS_REST_URL || "",
  token: process.env.UPSTASH_REDIS_REST_TOKEN || "",
});
 
// This is a separate global Redis instance that we use
// for global operations (e.g. linkCache, recordClick)
// so that if this redis goes down, it won't impact other endpoints
const hasGlobalRedisConfig =
  !!process.env.UPSTASH_GLOBAL_REDIS_REST_URL &&
  !!process.env.UPSTASH_GLOBAL_REDIS_REST_TOKEN;
 
const redisConfig = {
  url: hasGlobalRedisConfig
    ? process.env.UPSTASH_GLOBAL_REDIS_REST_URL
    : process.env.UPSTASH_REDIS_REST_URL || "",
  token: hasGlobalRedisConfig
    ? process.env.UPSTASH_GLOBAL_REDIS_REST_TOKEN
    : process.env.UPSTASH_REDIS_REST_TOKEN || "",
};
 
export const redisGlobal = new Redis(redisConfig);
 
export const redisGlobalWithTimeout = new Redis({
  ...redisConfig,
  signal: () => AbortSignal.timeout(1500),
});

Copyimport { Index } from "@upstash/vector";
 
export const vectorIndex = new Index({
  url: process.env.UPSTASH_VECTOR_REST_URL ?? "",
  token: process.env.UPSTASH_VECTOR_REST_TOKEN ?? "",
});
 
export const emojiVectorIndex = new Index<{
  emoji: string;
  label: string;
}>({
  url: process.env.UPSTASH_VECTOR_EMOJI_REST_URL ?? "",
  token: process.env.UPSTASH_VECTOR_EMOJI_REST_TOKEN ?? "",
});

Copyconst linkLRUCache = new LRUCache<string, RedisLinkProps>({
  max: 10000, // max 10,000 entries
  ttl: 5000, // 5 seconds
});

Copyasync get({ domain, key }: Pick<LinkProps, "domain" | "key">): Promise<{
  cachedLink: RedisLinkProps | null;
  redisFailOver: boolean;
}> {
  // here we use linkcache:${domain}:${key} instead of this._createKey({ domain, key })
  // because the key can either be cached as case-sensitive or case-insensitive depending on the domain
  // so we should get the original key from the cache
  const cacheKey = `linkcache:${domain}:${key}`;

Copy_createKey({ domain, key }: Pick<LinkProps, "domain" | "key">) {
  const caseSensitive = isCaseSensitiveDomain(domain);
  const originalKey = caseSensitive ? decodeKey(key) : key;
  const cacheKey = `linkcache:${domain}:${originalKey}`;
 
  return caseSensitive ? cacheKey : cacheKey.toLowerCase();
}

Copy_createStaticPagesCacheKeys({
  domain,
  key,
}: Pick<LinkProps, "domain" | "key">) {
  domain = domain.toLowerCase();
  // here we set 2 cache tags to invalidate the cache:
  // 1. static:${domain}:${key} - for the specific static page
  // 2. static:${domain} - scope all links under the domain (for easy purging in PATCH /domains/:domain)
  return `static:${domain}:${key},static:${domain}`;
}

Copyprivate async _invalidateVercelRuntimeCache(cacheKey: string) {
  return vercelCache
    .get(cacheKey)
    .then((cachedLink) =>
      cachedLink ? vercelCache.delete(cacheKey) : undefined,
    );
}

Copyexport const ratelimit = (
  requests: number = 10,
  seconds:
    | `${number} ms`
    | `${number} s`
    | `${number} m`
    | `${number} h`
    | `${number} d` = "10 s",
) => {
  return new Ratelimit({
    redis: redis,
    limiter: Ratelimit.slidingWindow(requests, seconds),
    analytics: true,
    prefix: "dub",
    timeout: 1000,
  });

Copyexport async function assertRateLimit({
  policy,
  identifier,
}: {
  policy: RatelimitPolicy;
  identifier: string | string[]; // dynamic part(s) of the key, e.g. email or [email, ip]
}) {
  if (!shouldApplyRateLimit) {
    return;
  }
 
  const key = [
    policy.keyPrefix,
    ...(Array.isArray(identifier) ? identifier : [identifier]),
  ].join(":");
 
  const { success, reset } = await ratelimit(
    policy.attempts,
    policy.window,
  ).limit(key);

Copyif (!success) {
  const retryAfter = formatRetryAfter(reset);
 
  const message =
    typeof policy.message === "function"
      ? policy.message({
          retryAfter,
          attempts: policy.attempts,
          window: policy.window,
        })
      : policy.message ??
        `Too many requests. Please try again in ${retryAfter}.`;
 
  throw new DubApiError({
    code: "rate_limit_exceeded",
    message,
  });
}

Copyconst ip = ipAddress(req);
const { success } = await ratelimit(10, "10 s").limit(`unsplash:${ip}`);
if (!success) {
  return new Response("Don't DDoS me pls 🥺", { status: 429 });
}

Copyexport async function withRedisLock<T>({
  key,
  ttlSeconds,
  fn,
}: {
  key: string;
  ttlSeconds: number;
  fn: () => Promise<T>;
}): Promise<T | null> {
  const token = crypto.randomUUID();
  const acquired = await redis.set(key, token, {
    nx: true,
    ex: ttlSeconds,
  });
 
  if (!acquired) {
    return null;
  }
 
  try {
    return await fn();
  } finally {
    // Lua compare-and-delete: only the token owner may release the lock.
    await redis.eval(RELEASE_LOCK_SCRIPT, [key], [token]);
  }
}

Copyconst RELEASE_LOCK_SCRIPT = `
  if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
  else
    return 0
  end
`;

Copyconst CACHE_KEY_PREFIX = "workspace:product";
const CACHE_EXPIRATION = 60 * 60 * 24 * 30; // 30 days
 
class WorkspaceProductCache {
  async get({ slug }: { slug: string }) {
    return await redis.get<DubProduct>(this._createKey({ slug }));
  }
 
  async set({ slug, product }: { slug: string; product: DubProduct }) {
    return await redis.set(this._createKey({ slug }), product, {
      ex: CACHE_EXPIRATION,
    });
  }
 
  async delete({ slug }: { slug: string }) {
    return await redis.del(this._createKey({ slug }));
  }
 
  _createKey({ slug }: { slug: string }) {
    return `${CACHE_KEY_PREFIX}:${slug}`;
  }
}

Copyconst CACHE_KEY_PREFIX = "onboarding-step";
export const ONBOARDING_WINDOW_SECONDS = 60 * 60 * 24; // 24 hours
 
class OnboardingStepCache {
  async set({ userId, step }: { userId: string; step: OnboardingStep }) {
    return await redis.set(`${CACHE_KEY_PREFIX}:${userId}`, step, {
      ex: ONBOARDING_WINDOW_SECONDS,
    });
  }
 
  async mset({ userIds, step }: { userIds: string[]; step: OnboardingStep }) {
    const pipeline = redis.pipeline();
    userIds.forEach((userId) => {
      pipeline.set(`${CACHE_KEY_PREFIX}:${userId}`, step, {
        ex: ONBOARDING_WINDOW_SECONDS,
      });
    });
    return await pipeline.exec();
  }
 
  async get({ userId }: { userId: string }) {
    return await redis.get(`${CACHE_KEY_PREFIX}:${userId}`);
  }
}

Copyclass ReferralsEmbedToken {
  async create(props: ReferralsEmbedTokenProps) {
    const publicToken = createId({
      prefix: EMBED_PUBLIC_TOKEN_PREFIX,
    });
 
    await redis.set(publicToken, JSON.stringify(props), {
      ex: EMBED_PUBLIC_TOKEN_EXPIRY,
      nx: true,
    });
 
    return {
      publicToken,
      expires: new Date(Date.now() + EMBED_PUBLIC_TOKEN_EXPIRY * 1000),
    };
  }
 
  async get(token: string) {
    return await redis.get<ReferralsEmbedTokenProps>(token);
  }
}

Copyconst tokenCacheItemSchema = z.object({
  id: z.string().nullish(),
  expires: z.date().nullish(),
  user: z.object({
    id: z.string(),
    name: z.string().nullable(),
    email: z.string().nullable(),
    isMachine: z.boolean(),
    defaultWorkspace: z.string().nullish(),
    defaultPartnerId: z.string().nullish(),
  }),
  scopes: z.string().nullish(),
  projectId: z.string().nullish(),
  project: z
    .object({
      plan: z.string().nullish(),
      trialEndsAt: z.coerce.date().nullish(),
    })
    .nullish(),
  installationId: z.string().nullish(),
});

Copyasync expireMany({ hashedKeys }: { hashedKeys: string[] }) {
  if (hashedKeys.length === 0) {
    return;
  }
 
  const pipeline = redis.pipeline();
 
  hashedKeys.forEach((hashedKey) => {
    pipeline.expire(this._createKey({ hashedKey }), 1);
  });
 
  return await pipeline.exec();
}

Copyclass AllowedHostnamesCache {
  async mset({
    allowedHostnames,
    domains,
  }: {
    allowedHostnames: string;
    domains: string[];
  }) {
    if (domains.length === 0) {
      return;
    }
 
    const pipeline = redis.pipeline();
 
    domains.forEach((domain) => {
      pipeline.set(this._createKey({ domain }), allowedHostnames, {
        ex: CACHE_EXPIRATION,
      });
    });
 
    return await pipeline.exec();
  }

Copyconst shopifyCheckoutCacheSchema = z.object({
  clickId: z
    .string()
    .nullish()
    .transform((value) => value ?? undefined),
  workspaceId: z.string().optional(),
  order: shopifyOrderSchema.optional(),
  dispatched: z.boolean().optional(),
});
 
type ShopifyCheckoutCacheItem = z.infer<typeof shopifyCheckoutCacheSchema>;

Copyif (!checkout.order || !checkout.workspaceId || !checkout.clickId) {
  console.info(
    "Shopify order dispatch skipped: checkout incomplete",
    logContext,
  );
  return false;
}
 
if (checkout.dispatched) {
  console.info(
    "Shopify order dispatch skipped: already dispatched",
    logContext,
  );
  return false;
}
 
// Claim the checkout
const key = shopifyCheckoutCache.createKey(checkoutToken);
const claim = await redis.hsetnx(key, "dispatched", true);
const claimed = Boolean(claim);
 
if (!claimed) {
  console.info("Shopify order dispatch skipped: claim lost", logContext);
  return false;
}

Copyexport class RedisStream {
  private streamKey: string;
 
  constructor(streamKey: string) {
    this.streamKey = streamKey;
  }
 
  /**
   * Read and process a batch of items from the stream
   * This provides a consumable batch of messages with the ability to delete them upon successful completion of the handler
   */
  async processBatch<T>(
    handler: (
      records: RedisStreamEntry<T>[],
    ) => Promise<any & { processedEntryIds: string[] }>,
    options: {
      startId?: string;
      endId?: string;
      count?: number;
      deleteAfterRead?: boolean;
    } = {},
  ): Promise<any> {
    const {
      startId = "0",
      endId = "+",
      count = 1000,
      deleteAfterRead = true,
    } = options;
 
    try {
      // Read entries from the stream
      const entriesMap = await redis.xrange(
        this.streamKey,
        startId,
        endId,
        count,
      );
 
      const entries: RedisStreamEntry<T>[] = Object.entries(entriesMap).map(
        ([id, data]) => ({
          id,
          data: data as any,
        }),
      );
 
      const { processedEntryIds, ...response } = await handler(entries);
 
      // Optionally delete processed entries to prevent memory buildup
      if (deleteAfterRead) {
        try {
          if (processedEntryIds.length > 0) {
            // xdel supports deleting multiple IDs at once
            await redis.xdel(this.streamKey, processedEntryIds);
          }
        } catch (error) {
          console.warn("Failed to clean up processed stream entries:", error);
          // Don't throw - this is not critical
        }
      }
 
      return response as Response;
    } catch (error) {
      console.error(
        "Failed to read workspace usage updates from stream:",
        error,
      );
      throw error;
    }
  }

Copy// Publishes a link click event to the stream to update:
// - link clicks count + lastClicked timestamp
// - workspace usage + totalClicks
// - program enrollment totalClicks
export const publishLinkClickEvent = async ({
  linkId,
  timestamp,
  workspaceId,
  programId,
  partnerId,
}: LinkClickEvent) => {
  const payload = {
    linkId,
    timestamp,
    ...(workspaceId && { workspaceId }),
    ...(programId && partnerId && { programId, partnerId }),
  };
 
  try {
    return await redis.xadd(STREAM_KEY, "*", payload);
  } catch (error) {
    logger.error("stream.publish_failed", {
      service: "upstash",
      streamKey: STREAM_KEY,
      error: toErrorFields(error),
      correlation: {
        linkId,
        workspaceId,
        programId,
        partnerId,
      },
    });
 
    return await Promise.allSettled([
      conn.execute(
        "UPDATE Link SET clicks = clicks + 1, lastClicked = NOW() WHERE id = ?",
        [linkId],
      ),
 
      workspaceId
        ? conn.execute(
            "UPDATE Project p SET p.usage = p.usage + 1, p.totalClicks = p.totalClicks + 1 WHERE id = ?",
            [workspaceId],
          )
        : null,
 
      programId && partnerId
        ? conn.execute(
            "UPDATE ProgramEnrollment SET totalClicks = totalClicks + 1 WHERE programId = ? AND partnerId = ?",
            [programId, partnerId],
          )
        : null,
 
      logger.flush(),
    ]);
  }

Copyexport const publishWorkspaceLinksUsageEvent = async (
  event: WorkspaceLinksUsageEvent,
) => {
  const { workspaceId, linksCount, timestamp } = event;
 
  try {
    return await redis.xadd(WORKSPACE_LINKS_USAGE_UPDATES_STREAM_KEY, "*", {
      workspaceId,
      linksCount,
      timestamp,
    });
  } catch (error) {
    console.error(
      "Failed to publish workspace links usage event (falling back to database update):",
      error,
    );
    // fallback on writing directly to the database
    return await conn.execute(
      "UPDATE Project SET linksUsage = linksUsage + ?, totalLinks = totalLinks + ? WHERE id = ?",
      [linksCount, linksCount, workspaceId],
    );
  }

Copyexport const publishWorkspaceClickEvent = async (event) => {
  try {
    const parsedEvent = clickEventSchemaTB.parse({
      ...event,
      qr: Number(event.qr),
      bot: Number(event.bot),
    });
 
    const hasLinkClickedWebhooks = await clickWebhookWorkspaces.has(
      parsedEvent.workspace_id,
    );
 
    if (!hasLinkClickedWebhooks) {
      return;
    }
 
    return await redis.xadd(STREAM_KEY, "*", parsedEvent);
  } catch (error) {
    logger.error("stream.publish_failed", {
      service: "upstash",
      streamKey: STREAM_KEY,
      error: toErrorFields(error),
      correlation: {
        workspaceId: event.workspace_id,
        clickId: event.click_id,
        linkId: event.link_id,
      },
    });
 
    await logger.flush();
    throw error;
  }

Copyimport { withCron } from "@/lib/cron/with-cron";
import { syncGoogleAdsInstalledWorkspaceSet } from "@/lib/integrations/google-ads/installed-workspaces";
import { logAndRespond } from "../utils";
import { cleanupRedundantLinkWebhookEntries } from "./cleanup-redundant-link-webhook-entries";
import { syncClickWebhookWorkspaceSet } from "./sync-click-webhook-workspace-set";
 
export const dynamic = "force-dynamic";
 
/*
    GET /api/cron/sync-redis-resources
    Rebuild various Redis resources:
    - syncClickWebhookWorkspaceSet: rebuild the Redis set of workspaces with active link.clicked webhooks
    - cleanupRedundantLinkWebhookEntries: remove redundant LinkWebhook entries for webhooks that are not scoped to links (folders, workspace)
    - syncGoogleAdsInstalledWorkspaceSet: rebuild the Redis set of workspaces with Google Ads installed
*/
 
// Runs every 5 minutes (*/5 * * * *)
export const GET = withCron(async () => {
  const result = await Promise.allSettled([
    syncClickWebhookWorkspaceSet(),
    cleanupRedundantLinkWebhookEntries(),
    syncGoogleAdsInstalledWorkspaceSet(),
  ]);
 
  [
    "syncClickWebhookWorkspaceSet",
    "cleanupRedundantLinkWebhookEntries",
    "syncGoogleAdsInstalledWorkspaceSet",
  ].map((name, index) => {
    const settled = result[index];
    if (settled.status === "fulfilled") {
      console.log(`${name}: ${settled.value}`);
    } else if (settled.status === "rejected") {
      console.error(`${name}: ${settled.reason}`);
    } else {
      console.error(`${name}: unknown error`);
    }
  });
 
  return logAndRespond("Synced Redis resources.");
});

Copyimport { redis } from "@/lib/upstash";
import { EventType } from "@prisma/client";
 
const REWARD_VERSION_TTL_SECONDS = 24 * 60 * 60; // 24 hours
 
export function getRewardVersionKey({
  groupId,
  event,
}: {
  groupId: string;
  event: EventType; // Reward type
}) {
  return `reward-version:${groupId}:${event}`;
}
 
export async function incrementRewardVersion({
  groupId,
  event,
}: {
  groupId: string;
  event: EventType;
}) {
  const key = getRewardVersionKey({
    groupId,
    event,
  });
 
  const version = await redis.incr(key);
  await redis.expire(key, REWARD_VERSION_TTL_SECONDS);
  return version;
}
 
export async function isStaleRewardVersion({
  version,
  groupId,
  event,
}: {
  version: number;
  groupId: string;
  event: EventType;
}) {
  const key = getRewardVersionKey({
    groupId,
    event,
  });
 
  const currentVersion = await redis.get(key);
 
  return currentVersion != null && version < Number(currentVersion);
}