queues and broadcasts

large outbound flows need durable jobs, bounded retry and recipient accounting.

why a queue

restarts should not lose progress — needs a durable storage, not the default in-memory one.
429s should slow the campaign, not break the bot for everyone.
403 recipients should be skipped and removed from future sends.
operators need pause, resume, cancel and progress.
workers need leases so two processes do not send the same delivery.

typed jobs

createBroadcast(bot.api, options) returns a Broadcast client. register one or more delivery types with .type(name, action), then .start(name, items) to enqueue a job — the local worker loop (autoRun, on by default) starts draining it immediately.

broadcast.ts

import { createBot } from "yaebal";
import { createBroadcast } from "@yaebal/broadcast";

const bot = createBot(process.env.BOT_TOKEN!);
const users: { chatId: number }[] = [];

// the default MemoryBroadcastStorage doesn't survive a restart — see "storage"
// below before running this against a real audience.
const broadcasts = createBroadcast(bot.api, {
  rateLimit: { limit: 25, windowMs: 1000 },
  retry: { attempts: 5, baseDelayMs: 1000, retryAfterPaddingMs: 250 },
}).type("text", async (chatId: number, text: string) => {
  await bot.api.call("sendMessage", { chat_id: chatId, text });
});

const job = await broadcasts.start("text", users.map((u) => [u.chatId, "hello"] as const));
await job.wait();

broadcast-playground.ts

import { createBot } from "yaebal";

const subscribers = new Set<number>([1001, 1002]);
const bot = createBot(process.env.BOT_TOKEN!);

bot.command("join", async (ctx) => {
  subscribers.add(ctx.chat!.id);
  await ctx.reply("subscribed");
});

bot.command("broadcast", async (ctx) => {
  const text = ctx.args.join(" ") || "release is live";

  for (const chatId of subscribers) {
    await ctx.api.call("sendMessage", { chat_id: chatId, text });
  }

  await ctx.reply(`queued ${subscribers.size} deliveries`);
});

bot.start();

storage

the built-in MemoryBroadcastStorage is the default and is fine for a single short-lived process. anything that must survive a restart or run behind more than one worker needs a real BroadcastStorage implementation — jobs, deliveries and an optional event log, backed by whatever database you already run.

storage.ts

import type { BroadcastStorage } from "@yaebal/broadcast";
import { createBroadcast } from "@yaebal/broadcast";

// unlike session (which plugs into any @yaebal/sklad adapter), broadcast
// persistence is its own interface — implement it against your database once,
// and every job/delivery/event survives a restart and can be shared across
// worker processes (each claim carries a lease so two workers can't double-send).
declare const myDurableStorage: BroadcastStorage;

const broadcasts = createBroadcast(undefined as never, {
  storage: myDurableStorage,
  leaseMs: 30_000,
  workerId: process.env.HOSTNAME ?? "worker-1",
});

record	contains
job	type, status, totals, metadata, timestamps
delivery	args, attempts, due time, lock, status, error/result
event	optional audit trail for operators and metrics — see `onEvent` below

pause, resume, cancel

start() resolves to a job handle — hand it to an admin command or a panel button.

controls.ts

import { createBroadcast } from "@yaebal/broadcast";

declare const broadcasts: ReturnType<typeof createBroadcast>;

const job = await broadcasts.start("text", [] as const);

await job.pause();
await job.resume();
await job.cancel();

const snapshot = await job.snapshot();
snapshot.sent; // BroadcastSnapshot — result totals plus the underlying job state
snapshot.job.status;

audit events

onEvent fires for every state transition — job lifecycle, each delivery attempt, rate-limit backoff, storage errors — typed as a discriminated union on event.type.

events.ts

import { createBroadcast } from "@yaebal/broadcast";

const broadcasts = createBroadcast(undefined as never, {
  onEvent(event) {
    if (event.type === "delivery_failed") {
      console.warn("broadcast delivery failed", event.delivery.args[0], event.error);
    }
    if (event.type === "job_completed") {
      console.log("broadcast", event.job.id, "done:", event.job.status);
    }
  },
});

shutdown

stop accepting work, let in-flight sends finish or release their leases, then exit.

shutdown.ts

declare const broadcasts: { stop(options?: { drain?: boolean }): Promise<void> };
declare const bot: { stop(): Promise<void> };

process.once("SIGTERM", async () => {
  // drain: true waits for in-flight sends (or their leases) instead of
  // abandoning them mid-delivery.
  await broadcasts.stop({ drain: true });
  await bot.stop();
});

operator checklist

rate limit below telegram's ceiling unless paid broadcast is explicitly enabled.
store original audience query or snapshot for auditability.
surface skipped recipients separately from failed transient sends.
make every job id visible in logs, metrics and panel actions.
test first with @yaebal/test and a tiny audience.