withRateLimit

withRateLimit throttles sends per derived key against a store. If the count exceeds max within the window, it throws a NotifyRpcRateLimitedError with retryAfterMs so queue workers can back off precisely.

0 / 3sends this window

window: 6s

click Send to start

import { withRateLimit } from '@betternotify/core/middlewares';
import { inMemoryRateLimitStore } from '@betternotify/core/stores';

const welcome = rpc
  .email()
  .input(schema)
  .subject(fn)
  .template(adapter)
  .use(withRateLimit({
    store: inMemoryRateLimitStore(),
    key: ({ args }) => String(args.to),
    max: 5,
    window: 60_000,
  }));

Options

Prop

Type

Dynamic keys

The key function receives the full send context, so you can rate-limit by recipient, route, tenant, or any combination:

withRateLimit({
  store,
  key: ({ args, route }) => `${route}:${args.to}`,
  max: 3,
  window: 60_000,
})

Error handling

When the limit is exceeded, withRateLimit throws NotifyRpcRateLimitedError:

import { NotifyRpcRateLimitedError } from '@betternotify/core';

try {
  await mail.welcome.send({ to: 'ada@example.com', input });
} catch (err) {
  if (err instanceof NotifyRpcRateLimitedError) {
    console.log(err.key);          // the rate-limit key
    console.log(err.retryAfterMs); // ms until the window resets
  }
}

Retrying rate-limited sends

NotifyRpcRateLimitedError carries retryAfterMs — the number of milliseconds until the current window resets. This makes it straightforward to delay and retry instead of dropping the send.

Manual retry

import { NotifyRpcRateLimitedError } from '@betternotify/core';

const sendWithRetry = async (args: Parameters<typeof mail.welcome.send>[0]) => {
  const [err, result] = await handlePromise(mail.welcome.send(args));
  if (err instanceof NotifyRpcRateLimitedError) {
    await new Promise((r) => setTimeout(r, err.retryAfterMs));
    return mail.welcome.send(args);
  }
  if (err) throw err;
  return result;
};

With a queue worker

When using BullMQ (or any job queue), catch the error in the worker and schedule the job to retry after the window resets. The queue handles the delay — your code just provides the timing:

import { NotifyRpcRateLimitedError } from '@betternotify/core';

worker.on('failed', (job, err) => {
  if (err instanceof NotifyRpcRateLimitedError && job) {
    job.moveToDelayed(Date.now() + err.retryAfterMs);
  }
});

retryAfterMs is a precise value derived from the store, not a guess. Fixed-window counters reset at the window boundary; sliding-window counters compute the exact gap until the oldest entry expires.