Add rate limiting configuration for email sending

- Introduced `getResendRateLimitConfig` function to retrieve and calculate email sending rate limits.
- Added `setResendRateLimit` mutation to allow dynamic updates of the email sending rate limit.
- Updated schema to include `resendOptions` for storing rate limit configurations.
- Enhanced shared types with `vResendOptions` for validation of rate limit settings.
- Refactored rate limiter initialization to support dynamic configurations.

Author: sethconve

src/component/_generated/component.ts

@@ -154,6 +154,13 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
         string,
         Name
       >;
+      setResendRateLimit: FunctionReference<
+        "mutation",
+        "internal",
+        { rateLimitEmailsPerSecond: number },
+        null,
+        Name
+      >;
       updateManualEmail: FunctionReference<
         "mutation",
         "internal",

src/component/lib.ts

@@ -8,7 +8,8 @@ import {
   type ActionCtx,
 } from "./_generated/server.js";
 import { Workpool } from "@convex-dev/workpool";
-import { RateLimiter } from "@convex-dev/rate-limiter";
+import { RateLimiter, SECOND } from "@convex-dev/rate-limiter";
+import type { RateLimitConfig } from "@convex-dev/rate-limiter";
 import { api, components, internal } from "./_generated/api.js";
 import { internalMutation } from "./_generated/server.js";
 import { type Id, type Doc } from "./_generated/dataModel.js";
@@ -33,6 +34,8 @@ const BATCH_SIZE = 100;
 const EMAIL_POOL_SIZE = 4;
 const CALLBACK_POOL_SIZE = 4;
 const RESEND_ONE_CALL_EVERY_MS = 600; // Half the stated limit, but it keeps us sane.
+const DEFAULT_EMAILS_PER_SECOND = SECOND / RESEND_ONE_CALL_EVERY_MS;
+const MIN_EMAILS_PER_SECOND = 0.1;
 const FINALIZED_EMAIL_RETENTION_MS = 1000 * 60 * 60 * 24 * 7; // 7 days
 const FINALIZED_EPOCH = Number.MAX_SAFE_INTEGER;
 const ABANDONED_EMAIL_RETENTION_MS = 1000 * 60 * 60 * 24 * 30; // 30 days
@@ -79,14 +82,7 @@ const callbackPool = new Workpool(components.callbackWorkpool, {
 });
 
 // We rate limit our calls to the Resend API.
-// FUTURE -- make this rate configurable if an account ups its sending rate with Resend.
-const resendApiRateLimiter = new RateLimiter(components.rateLimiter, {
-  resendApi: {
-    kind: "fixed window",
-    period: RESEND_ONE_CALL_EVERY_MS,
-    rate: 1,
-  },
-});
+const resendApiRateLimiter = new RateLimiter(components.rateLimiter);
 
 // Enqueue an email to be send.  A background job will grab batches
 // of emails and enqueue them to be sent by the workpool.
@@ -190,6 +186,23 @@ export const sendEmail = mutation({
   },
 });
 
+async function getResendRateLimitConfig(
+  ctx: MutationCtx,
+): Promise<RateLimitConfig> {
+  const record = await ctx.db.query("resendOptions").unique();
+  const raw = record?.options.rateLimitEmailsPerSecond;
+  const emailsPerSecond =
+    typeof raw === "number" && Number.isFinite(raw) && raw > 0
+      ? Math.max(raw, MIN_EMAILS_PER_SECOND)
+      : Math.max(DEFAULT_EMAILS_PER_SECOND, MIN_EMAILS_PER_SECOND);
+  return {
+    kind: "token bucket",
+    rate: emailsPerSecond,
+    period: SECOND,
+    capacity: Math.max(emailsPerSecond, 1),
+  };
+}
+
 export const createManualEmail = mutation({
   args: {
     from: v.string(),
@@ -271,6 +284,26 @@ export const cancelEmail = mutation({
   },
 });
 
+export const setResendRateLimit = mutation({
+  args: { rateLimitEmailsPerSecond: v.number() },
+  returns: v.null(),
+  handler: async (ctx, { rateLimitEmailsPerSecond }) => {
+    if (
+      !Number.isFinite(rateLimitEmailsPerSecond) ||
+      rateLimitEmailsPerSecond <= 0
+    ) {
+      throw new Error("rateLimitEmailsPerSecond must be a positive number");
+    }
+    const existing = await ctx.db.query("resendOptions").unique();
+    const options = { rateLimitEmailsPerSecond };
+    if (existing) {
+      await ctx.db.patch(existing._id, { options });
+    } else {
+      await ctx.db.insert("resendOptions", { options });
+    }
+  },
+});
+
 // Get the status of an email.
 export const getStatus = query({
   args: {
@@ -684,9 +717,11 @@ async function createResendBatchPayload(
 }
 
 const FIXED_WINDOW_DELAY = 100;
-async function getDelay(ctx: RunMutationCtx & RunQueryCtx): Promise<number> {
+async function getDelay(ctx: MutationCtx): Promise<number> {
+  const config = await getResendRateLimitConfig(ctx);
   const limit = await resendApiRateLimiter.limit(ctx, "resendApi", {
     reserve: true,
+    config,
   });
   //console.log(`RL: ${limit.ok} ${limit.retryAfter}`);
   const jitter = Math.random() * FIXED_WINDOW_DELAY;

src/component/schema.ts

@@ -1,6 +1,6 @@
 import { defineSchema, defineTable } from "convex/server";
 import { v } from "convex/values";
-import { vOptions, vStatus } from "./shared.js";
+import { vOptions, vResendOptions, vStatus } from "./shared.js";
 
 export default defineSchema({
   content: defineTable({
@@ -15,6 +15,9 @@ export default defineSchema({
   lastOptions: defineTable({
     options: vOptions,
   }),
+  resendOptions: defineTable({
+    options: vResendOptions,
+  }),
   deliveryEvents: defineTable({
     emailId: v.id("emails"),
     resendId: v.string(),

src/component/shared.ts

@@ -41,6 +41,10 @@ export const vOptions = v.object({
 
 export type RuntimeConfig = Infer<typeof vOptions>;
 
+export const vResendOptions = v.object({
+  rateLimitEmailsPerSecond: v.number(),
+});
+
 const commonFields = {
   broadcast_id: v.optional(v.string()),
   created_at: v.string(),