@yaebal/broadcast

typed broadcast jobs for yaebal bots: queue many deliveries, keep progress in storage, retry transient failures, skip blocked users, pause/resume/cancel jobs and stream events to logs or ui.

install

terminal

pnpm add @yaebal/broadcast

fast path

use broadcast() when you want to send one text message and wait for the result. it creates a local worker, sends every delivery, then stops it.

broadcast.ts

import { Bot } from "@yaebal/core";
import { broadcast } from "@yaebal/broadcast";

const bot = new Bot(token);

const result = await broadcast(bot.api, subscriberIds, "hello everyone", {
  rateLimit: { limit: 25, windowMs: 1_000 },
  retry: { attempts: 5, fixedDelayMs: 1_000 },
  extra: { disable_notification: true },
  onError: (chatId, error) => {
    console.error("delivery failed", chatId, error);
  },
});

console.log(result.sent, "sent", result.skipped, "skipped", result.failed, "failed");

typed jobs

use Broadcast for reusable campaigns, background delivery, progress inspection and job controls. type() accumulates valid job names and tuple arguments through the chain, so start("digest", ...) only accepts the arguments of the registered handler.

typed-jobs.ts

import { Broadcast, MemoryBroadcastStorage } from "@yaebal/broadcast";

const broadcaster = new Broadcast(bot.api, {
  storage: new MemoryBroadcastStorage(),
  concurrency: 5,
  rateLimit: { limit: 25, windowMs: 1_000 },
  onEvent: (event) => console.log(event.type),
}).type("digest", (chatId: number, text: string) =>
  bot.api.sendMessage({ chat_id: chatId, text }),
);

const job = await broadcaster.start("digest", [
  [1001, "weekly digest"],
  [1002, "weekly digest"],
]);

const result = await job.wait();

controls

every started job returns a handle. handles are small references around storage, so you can also call broadcaster.pause(id), broadcaster.resume(id) and broadcaster.cancel(id) from an admin panel or command handler.

controls.ts

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

const snapshot = await job.snapshot();
const jobs = await broadcaster.listJobs();
const failed = await broadcaster.listDeliveries(job.id, { status: "failed" });

any method

text messages are just the common case. queueMethod() builds params per target for any telegram bot api method, and type() can run fully custom async logic.

method.ts

const job = await broadcaster.queueMethod(
  "sendPhoto",
  users,
  (user) => ({
    chat_id: user.chatId,
    photo: user.photo,
    caption: "new drop",
  }),
);

storage

MemoryBroadcastStorage is for tests, examples and single-process bots. production bots can implement BroadcastStorage on top of redis, postgres, sqlite or another queue/database. for multi-worker delivery, make claim(workerId, now, leaseMs) atomic.

behavior

feature	default
rate limit	`25` deliveries per second
retry budget	`5` attempts per delivery
blocked users	telegram `403` is counted as `skipped`
flood waits	telegram `429` honors `response_parameters.retry_after`
progress	`total`, `sent`, `failed`, `skipped`, `retried`, `status`
events	job lifecycle, delivery lifecycle, retry, rate-limit and storage errors

exports

export	kind	use
`broadcast`	function	send one text campaign and wait for completion
`Broadcast`	class	typed job engine with storage, worker, controls and events
`MemoryBroadcastStorage`	class	in-memory storage adapter for tests and examples
`createBroadcast`	function	factory wrapper around `new Broadcast(api, options)`
`decideRetry`	function	unit-testable retry policy helper

see the runnable examples/broadcast bot for subscribers, admin commands, progress and graceful shutdown.