Bulk Emails

Fan a large recipient list out with enqueueMany, then bound the send rate and concurrency with per-task limits.

Sending to a large list is an enqueue-throughput problem: stage every message as its own job in as few round-trips as possible, then let per-task limits pace the actual sends so you stay within your provider's quota.

The Node SDK has no implicit task-batching (the Python SDK's batch= collector is Python-specific). The idiomatic Node approach is batch enqueue (enqueueMany) plus per-task maxConcurrent / rateLimit — one job per message, with the worker doing the pacing.

tasks.ts

One job per recipient keeps retries and failures isolated — a bad address never blocks the rest. The limits cap how fast and how many run at once.

import { Queue } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "bulk-email.db" });

queue.task("sendEmail", async (to: string, subject: string, body: string) => {
  await provider.send(to, subject, body);
}, {
  rateLimit: "600/m",     // stay under the provider's quota
  maxConcurrent: 20,      // at most 20 in flight at once
  maxRetries: 5,
  retryBackoff: { baseMs: 2_000, maxMs: 300_000 },
});

send.ts

enqueueMany stages a whole chunk in a single storage round-trip. Chunk a very large list so each call stays a reasonable size.

import { queue } from "./tasks";

function* chunks<T>(items: T[], size: number) {
  for (let i = 0; i < items.length; i += size) {
    yield items.slice(i, i + size);
  }
}

export function sendCampaign(recipients: string[], subject: string, body: string) {
  const ids: string[] = [];
  for (const chunk of chunks(recipients, 1_000)) {
    ids.push(...queue.enqueueMany("sendEmail", chunk.map((to) => ({
      args: [to, subject, body],
    }))));
  }
  return ids; // one job id per recipient
}

Give each entry a uniqueKey (say `email:${to}:${campaignId}`) if a retried send loop must not double-send: batch entries dedup exactly like a single enqueue, resolving to the already-enqueued job's id.

Monitoring the send

Watch the backlog drain and catch addresses that exhausted their retries.

console.log(await queue.statsByQueue("default")); // { pending, running, completed, failed, dead, cancelled }

for (const dead of await queue.deadLetters(50)) {
  console.warn(`gave up on ${dead.taskName}: ${dead.error}`);
}

Running it

bash

node worker.ts   # registers sendEmail, runWorker()

bash

node -e "const list=['a@example.com','b@example.com']; import('./send.ts').then(s => s.sendCampaign(list, 'Launch', '<html>'))"

Key patterns

Pattern	Where
One round-trip per chunk	`queue.enqueueMany`
Provider quota	`task(..., { rateLimit })`
Concurrency cap	`task(..., { maxConcurrent })`
Isolated failures	one job per recipient
Drain + failure visibility	`statsByQueue` + `deadLetters`

tasks.ts

One job per recipient keeps retries and failures isolated — a bad address never blocks the rest. The limits cap how fast and how many run at once.

import { Queue } from "@byteveda/flexiq";

export const queue = new Queue({ dbPath: "bulk-email.db" });

queue.task("sendEmail", async (to: string, subject: string, body: string) => {
  await provider.send(to, subject, body);
}, {
  rateLimit: "600/m",     // stay under the provider's quota
  maxConcurrent: 20,      // at most 20 in flight at once
  maxRetries: 5,
  retryBackoff: { baseMs: 2_000, maxMs: 300_000 },
});

send.ts

enqueueMany stages a whole chunk in a single storage round-trip. Chunk a very large list so each call stays a reasonable size.

import { queue } from "./tasks";

function* chunks<T>(items: T[], size: number) {
  for (let i = 0; i < items.length; i += size) {
    yield items.slice(i, i + size);
  }
}

export function sendCampaign(recipients: string[], subject: string, body: string) {
  const ids: string[] = [];
  for (const chunk of chunks(recipients, 1_000)) {
    ids.push(...queue.enqueueMany("sendEmail", chunk.map((to) => ({
      args: [to, subject, body],
    }))));
  }
  return ids; // one job id per recipient
}

console.log(await queue.statsByQueue("default")); // { pending, running, completed, failed, dead, cancelled } for (const dead of await queue.deadLetters(50)) { console.warn(`gave up on ${dead.taskName}: ${dead.error}`); }

Pattern

Where

One round-trip per chunk

queue.enqueueMany

Provider quota

task(..., { rateLimit })

Concurrency cap

task(..., { maxConcurrent })

Isolated failures

one job per recipient

Drain + failure visibility

statsByQueue + deadLetters