Fase 1c: canal Telegram (telegram-worker) — codigo migrado, inerte

Envio de alertas Telegram desde el servicio (docs/legacy-telegram.md): por alerta, texto Markdown (content + firelink i18n) + ubicacion, como node-red. - types: canal 'telegram', notif.type 'telegram', campo telegramNotified - telegram/message.ts: buildTelegramMessages (firelink es/en/gl + location) - telegram/client.ts: cliente Bot API (fetch), interpretResponse (429 retry_after, 403 blocked) - telegram/rate.ts: throttle por chat (1 msg/s) + computeWaitMs puro - handlers.processTelegramJob: idempotencia (notification_sends canal telegram), modos dry-run/shadow/canary/full, throttle, 429->retry, bloqueo->sub inactiva - poller/queue/index: canal telegram integrado; worker con rate limiter global BullMQ (25/s por bot); cliente por idioma (es/gl->ES, en->EN) - config: bloque telegram (tokens es/en, server, globalPerSec, perChatMinMs) 85 tests en verde, typecheck y build OK. Sin TELEGRAM_TOKEN_* configurados el worker no se activa (todo inerte). Falta: extraer tokens de bot de flows_cred + rollout con el usuario. Los flujos conversacionales de los bots siguen en node-red.
2026-07-13 18:49:52 +02:00 · 2026-07-13 18:49:52 +02:00 · d1fc6798dc
commit d1fc6798dc
parent a1ed206985
14 changed files with 479 additions and 14 deletions
--- a/.env.example
+++ b/.env.example
@ -47,6 +47,14 @@ MATCHER_AUDIENCE=web,mobile  # tipos de suscripción que generan notifs aquí (t
 MATCHER_POLL_MS=30000        # cada cuánto sondea activefires (Mongo 3.2, sin change streams)
 MATCHER_BATCH=2000           # máx fuegos por ciclo

+# --- Telegram (fase 1c: envío de alertas por los bots) ---
+# Añade 'telegram' a OWNED_CHANNELS y MATCHER_AUDIENCE para activarlo.
+TELEGRAM_TOKEN_ES=           # SECRETO: token del bot ES (extraer de flows_cred de node-red)
+TELEGRAM_TOKEN_EN=           # SECRETO: token del bot EN
+TELEGRAM_SERVER=             # host para el firelink (por defecto = host de ROOT_URL)
+TELEGRAM_GLOBAL_PER_SEC=25   # límite global msg/s por bot (Telegram permite ~30)
+TELEGRAM_PER_CHAT_MS=1000    # mínimo ms entre mensajes al mismo chat (~1 msg/s)
+
 # --- Enlaces / mapas ---
 ROOT_URL=https://fuegos.comunes.org/
 GMAPS_KEY=                   # Meteor.settings.gmaps.key
--- a/README.md
+++ b/README.md
@ -74,6 +74,9 @@ sistema) + Redis local. En fase 3 pasa a Docker Compose.
 - [x] **1b**: matcher geoespacial (content i18n, geolib, `sealed` vía Iron —
      compatible con la web —, dedupe 500 m), ingesta por polling, modos
      `shadow`/`full`, comparador de shadow. Tests en verde.
+- [x] **1c**: telegram-worker (texto + ubicación, firelink, rate limiting
+      global 25/s + 1/s por chat, 429/bloqueo, idempotencia). Tests en verde.
+      **Código migrado; sin tokens configurados → inerte.**
 - [ ] **1a full**: envío masivo FCM (~8176 users) — con el usuario, vigilando.
 - [ ] **1b**: rollout `shadow` ≥3 días → comparar → cutover node-red → `full`.
- [ ] **1c**: envío Telegram.
+- [ ] **1c**: extraer tokens de bot → rollout `shadow`/canary → cutover node-red.
--- a/docs/legacy-telegram.md
+++ b/docs/legacy-telegram.md
@ -0,0 +1,57 @@
+# Envío Telegram viejo (fase 1c — caracterización)
+
+> Fuente: `todos-contra-el-fuego/telegram-bot/flows.json` (nodos "calc distances",
+> "send location", senders `telegram sender`), locales node-red.
+
+## Qué envía node-red por cada alerta de fuego a un chat suscrito
+
+**Dos mensajes** al `chatId` de la suscripción (subs `type: 'telegram'`):
+
+1. **Texto** (`type: "message"`, `parse_mode: "Markdown"`):
+   `content` + salto de línea + `firelink`.
+   - `content` = la misma cadena `kmnasa`/`kmvecinal` del matching
+     (ver `legacy-matching.md` §3; lleva `🔥` y el `км` cirílico).
+   - `firelink` (i18n): `[<label>](https://{{server}}/fire/{{enc}})` con
+     `enc` = `sealed` (el mismo Iron del matching).
+     - es: `[más información](https://{{server}}/fire/{{enc}})`
+     - en: `[more information](https://{{server}}/fire/{{enc}})`
+     - gl: `[máis información](https://{{server}}/fire/{{enc}})`
+2. **Ubicación** (`type: "location"`): `{ latitude: fire.lat, longitude: fire.lon }`.
+
+No hay botones/inline-keyboard en la alerta (solo `parse_mode: Markdown`).
+
+## Bots
+
+- Dos instancias node-red separadas: **ES** (`/opt/node-red-data-git`, puerto
+  1880) y **EN** (`/opt/node-red-data-git-en`, 1881). Tokens en
+  `flows_cred*.json` (cifrados con el `credentialSecret` de node-red; copiados a
+  `tcef-private-config`). El servicio los recibe por env (`TELEGRAM_TOKEN_ES` /
+  `TELEGRAM_TOKEN_EN`) — **no** están en claro aquí; hay que extraerlos/pedirlos.
+- Selección de bot: por idioma del suscriptor (es/gl → bot ES, en → bot EN).
+
+## Límites de la Bot API de Telegram
+
+- ~30 msg/s global por bot → el servicio usa **25/s** (margen).
+- ~1 msg/s por chat individual.
+- 429 devuelve `parameters.retry_after` (segundos) → respetar.
+- Bot bloqueado por el usuario (403 "bot was blocked by the user") → marcar la
+  sub inactiva y **no** reintentar.
+
+## Integración en el servicio (unificada)
+
+- El matcher (1b) genera docs `notifications` con `type: 'telegram'` para las
+  subs telegram cuando `MATCHER_AUDIENCE` incluye `telegram`. El doc lleva
+  `subsId` → el worker carga la sub para el `chatId`.
+- Un **`telegram-worker`** (canal `telegram`, como fcm/email) consume esos docs:
+  envía texto + ubicación, marca `telegramNotified`, idempotencia por
+  `notification_sends` (canal `telegram`), mismos modos
+  `dry-run/shadow/canary/full` y kill switch.
+
+## Fuera de alcance (documentado)
+
+- **Canales públicos**: si los flows publican alertas también en canales
+  públicos del proyecto (además de a las subs), esa vía se migra aparte o se deja
+  explícitamente en node-red. Config opcional `TELEGRAM_PUBLIC_CHANNELS`; por
+  defecto **no** se toca.
+- Flujos conversacionales de los bots (/start, alta/baja de subs, /lang, etc.)
+  **se quedan en node-red** — solo se migra el ENVÍO de alertas.
--- a/src/config.ts
+++ b/src/config.ts
@ -62,6 +62,18 @@ export interface Config {
    batchSize: number;
  };

+  /** Fase 1c — Telegram sending. */
+  telegram: {
+    /** Bot tokens by language family (es/gl → es, en → en). SECRET. */
+    tokens: { es?: string; en?: string };
+    /** Site host for firelink (defaults to the rootUrl host). */
+    server: string;
+    /** Global send rate cap per bot (msg/s); Telegram allows ~30, we use 25. */
+    globalPerSec: number;
+    /** Minimum ms between sends to the same chat (~1 msg/s). */
+    perChatMinMs: number;
+  };
+
  /** Base URL for building fire/subscription links (Meteor ROOT_URL). */
  rootUrl: string;
  gmaps: { key?: string; fireIconUrl?: string };
@ -85,11 +97,21 @@ const DEFAULTS = {
    pollIntervalMs: 30_000,
    batchSize: 2000,
  },
+  telegram: { globalPerSec: 25, perChatMinMs: 1000 },
  rootUrl: 'https://fuegos.comunes.org/',
  logLevel: 'info',
  serviceName: 'tcef-notifications',
 };

+/** Extract the host from a URL for the Telegram firelink (fuegos.comunes.org). */
+function hostOf(url: string): string {
+  try {
+    return new URL(url).host;
+  } catch {
+    return url.replace(/^https?:\/\//, '').replace(/\/.*$/, '');
+  }
+}
+
 function parseBool(v: string | undefined): boolean | undefined {
  if (v === undefined) return undefined;
  return v === '1' || v.toLowerCase() === 'true';
@ -176,6 +198,15 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config {
      pollIntervalMs: Number(env.MATCHER_POLL_MS ?? file.matcher?.pollIntervalMs ?? DEFAULTS.matcher.pollIntervalMs),
      batchSize: Number(env.MATCHER_BATCH ?? file.matcher?.batchSize ?? DEFAULTS.matcher.batchSize),
    },
+    telegram: {
+      tokens: {
+        es: env.TELEGRAM_TOKEN_ES ?? file.telegram?.tokens?.es,
+        en: env.TELEGRAM_TOKEN_EN ?? file.telegram?.tokens?.en,
+      },
+      server: env.TELEGRAM_SERVER ?? file.telegram?.server ?? hostOf(env.ROOT_URL ?? file.rootUrl ?? DEFAULTS.rootUrl),
+      globalPerSec: Number(env.TELEGRAM_GLOBAL_PER_SEC ?? file.telegram?.globalPerSec ?? DEFAULTS.telegram.globalPerSec),
+      perChatMinMs: Number(env.TELEGRAM_PER_CHAT_MS ?? file.telegram?.perChatMinMs ?? DEFAULTS.telegram.perChatMinMs),
+    },
    rootUrl: env.ROOT_URL ?? file.rootUrl ?? DEFAULTS.rootUrl,
    gmaps: {
      key: env.GMAPS_KEY ?? file.gmaps?.key,
--- a/src/handlers.ts
+++ b/src/handlers.ts
@ -6,7 +6,10 @@ import type { Repos } from './db.js';
 import type { FcmClient } from './fcm.js';
 import type { Mailer } from './mailer.js';
 import type { Logger } from './logger.js';
-import { buildEmail, buildPush, normalizeLang } from './messages.js';
+import { buildEmail, buildPush, normalizeLang, type Lang } from './messages.js';
+import { buildTelegramMessages } from './telegram/message.js';
+import type { TelegramClient } from './telegram/client.js';
+import type { PerChatThrottle } from './telegram/rate.js';
 import { getEmailOf } from './recipient.js';
 import { decideSend } from './mode.js';
 import { underPerUserLimit } from './limits.js';
@ -20,6 +23,11 @@ export interface Deps {
  /** Present only when the service may actually send (canary/full). */
  fcm?: FcmClient;
  mailer?: Mailer;
+  telegram?: {
+    clientFor(lang: Lang): TelegramClient | undefined;
+    throttle: PerChatThrottle;
+    server: string;
+  };
 }

 /** Transient failure that should be retried by BullMQ (backoff). */
@ -176,3 +184,84 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi
  );
  log.info({ notificationId, to: cfg.email.redirectTo ?? emailAddress }, 'email: sent');
 }
+
+/**
+ * Telegram handler. Sends the two messages node-red sent (text + location) to
+ * the subscription's chat, with per-chat throttling, 429/blocked handling and
+ * the same safeguards as the other channels (docs/legacy-telegram.md).
+ */
+export async function processTelegramJob(deps: Deps, notificationId: string): Promise<void> {
+  const { cfg, repos, log } = deps;
+  const notif = await repos.notifications.findOne({ _id: new ObjectId(notificationId) });
+  if (!notif) return;
+  if (notif.type !== 'telegram' || notif.telegramNotified === true) return;
+  if (!notif.subsId) {
+    log.warn({ notificationId }, 'telegram: notification has no subsId');
+    return;
+  }
+
+  const sub = await repos.subscriptions.findOne({ _id: notif.subsId });
+  if (!sub || typeof sub.chatId !== 'number') {
+    log.warn({ notificationId, subsId: String(notif.subsId) }, 'telegram: subscription/chatId not found');
+    return;
+  }
+  const chatId = sub.chatId;
+  const user = await repos.users.findOne({ _id: notif.userId });
+  const lang = normalizeLang(user?.lang);
+
+  const decision = decideSend(cfg, notif);
+  if (!decision.send) {
+    log.info({ notificationId, chatId, mode: cfg.mode, reason: decision.reason }, `telegram: not sending (${decision.reason})`);
+    return;
+  }
+
+  const client = deps.telegram?.clientFor(lang);
+  if (!deps.telegram || !client) throw new Error(`telegram client not configured for lang ${lang}`);
+
+  const won = await reserve(repos, notif._id, 'telegram', cfg.mode);
+  if (!won) {
+    log.debug({ notificationId }, 'telegram: already reserved/sent (idempotent skip)');
+    return;
+  }
+
+  await deps.telegram.throttle.acquire(chatId);
+  const msgs = buildTelegramMessages(notif, lang, deps.telegram.server);
+
+  // Text first, then location — same order as node-red.
+  const textRes = await client.sendMessage(chatId, msgs.text, msgs.parseMode);
+  if (!textRes.ok) return await handleTgFailure(deps, notif, sub._id, chatId, textRes);
+  const locRes = await client.sendLocation(chatId, msgs.latitude, msgs.longitude);
+  if (!locRes.ok) return await handleTgFailure(deps, notif, sub._id, chatId, locRes);
+
+  await markResult(repos, notif._id, 'telegram', 'sent');
+  await repos.notifications.updateOne(
+    { _id: notif._id },
+    { $set: { telegramNotified: true, telegramNotifiedAt: new Date() } },
+  );
+  log.info({ notificationId, chatId }, 'telegram: sent');
+}
+
+async function handleTgFailure(
+  deps: Deps,
+  notif: NotificationDoc,
+  subId: NotificationDoc['subsId'],
+  chatId: number,
+  res: { blocked?: boolean; retryAfterMs?: number; error: string },
+): Promise<void> {
+  const { repos, log } = deps;
+  if (res.blocked) {
+    // Bot blocked by the user → mark the sub inactive, do not retry.
+    await markResult(repos, notif._id, 'telegram', 'dead-token', res.error);
+    await repos.subscriptions.updateOne(
+      { _id: subId },
+      { $set: { active: false, blockedAt: new Date() } } as never,
+    );
+    log.warn({ chatId, error: res.error }, 'telegram: bot blocked, sub marked inactive');
+    return;
+  }
+  // Transient (incl. 429) → release and retry with backoff (respect retry_after).
+  await release(repos, notif._id, 'telegram');
+  const err = new RetryableError(res.error);
+  log.warn({ chatId, retryAfterMs: res.retryAfterMs, error: res.error }, 'telegram: transient failure, will retry');
+  throw err;
+}
--- a/src/index.ts
+++ b/src/index.ts
@ -7,7 +7,9 @@ import { createQueues, redisConnection, QUEUE_NAMES } from './queue.js';
 import { createFcmClient, type FcmClient } from './fcm.js';
 import { createMailer, type Mailer } from './mailer.js';
 import { createPoller } from './poller.js';
-import { processEmailJob, processFcmJob, type Deps } from './handlers.js';
+import { processEmailJob, processFcmJob, processTelegramJob, type Deps } from './handlers.js';
+import { createTelegramClient } from './telegram/client.js';
+import { PerChatThrottle } from './telegram/rate.js';
 import type { Channel, SendJob } from './types.js';

 async function main(): Promise<void> {
@ -30,22 +32,41 @@ async function main(): Promise<void> {
  const willSend = cfg.mode === 'canary' || cfg.mode === 'full';
  let fcm: FcmClient | undefined;
  let mailer: Mailer | undefined;
+  let telegram: Deps['telegram'];
  if (willSend && cfg.ownedChannels.includes('fcm')) fcm = createFcmClient(cfg);
  if (willSend && cfg.ownedChannels.includes('email')) mailer = createMailer(cfg);
+  if (willSend && cfg.ownedChannels.includes('telegram')) {
+    const clients = {
+      es: cfg.telegram.tokens.es ? createTelegramClient(cfg.telegram.tokens.es) : undefined,
+      en: cfg.telegram.tokens.en ? createTelegramClient(cfg.telegram.tokens.en) : undefined,
+    };
+    telegram = {
+      clientFor: (lang) => (lang === 'en' ? clients.en : clients.es),
+      throttle: new PerChatThrottle(cfg.telegram.perChatMinMs),
+      server: cfg.telegram.server,
+    };
+  }

-  const deps: Deps = { cfg, repos, log, fcm, mailer };
+  const deps: Deps = { cfg, repos, log, fcm, mailer, telegram };

  const workers: Worker<SendJob>[] = [];
  const handlerFor: Record<Channel, (id: string) => Promise<void>> = {
    fcm: (id) => processFcmJob(deps, id),
    email: (id) => processEmailJob(deps, id),
+    telegram: (id) => processTelegramJob(deps, id),
  };
+  const concurrencyFor: Record<Channel, number> = { fcm: 20, email: 5, telegram: 5 };

  for (const channel of cfg.ownedChannels) {
    const worker = new Worker<SendJob>(
      QUEUE_NAMES[channel],
      async (job) => handlerFor[channel](job.data.notificationId),
-      { connection, concurrency: channel === 'email' ? 5 : 20 },
+      {
+        connection,
+        concurrency: concurrencyFor[channel],
+        // Telegram: global rate cap per bot (~25 msg/s) via BullMQ limiter.
+        ...(channel === 'telegram' ? { limiter: { max: cfg.telegram.globalPerSec, duration: 1000 } } : {}),
+      },
    );
    worker.on('failed', (job, err) => {
      log.error({ channel, jobId: job?.id, err: err.message }, 'worker: job failed (dead-letter if attempts exhausted)');
--- a/src/poller.ts
+++ b/src/poller.ts
@ -9,9 +9,14 @@ import type { Channel, NotificationDoc, SendJob } from './types.js';

 /** Pending-doc filter for a channel — the CORRECT fields (old cron had a typo). */
 export function pendingFilter(channel: Channel): FilterQuery<NotificationDoc> {
-  return channel === 'fcm'
-    ? { type: 'mobile', notified: { $ne: true } }
-    : { type: 'web', emailNotified: { $ne: true } };
+  switch (channel) {
+    case 'fcm':
+      return { type: 'mobile', notified: { $ne: true } };
+    case 'email':
+      return { type: 'web', emailNotified: { $ne: true } };
+    case 'telegram':
+      return { type: 'telegram', telegramNotified: { $ne: true } };
+  }
 }

 export interface Poller {
@ -54,7 +59,7 @@ export function createPoller(
  }

  async function runOnce(): Promise<Record<Channel, number>> {
-    const result: Record<Channel, number> = { fcm: 0, email: 0 };
+    const result: Record<Channel, number> = { fcm: 0, email: 0, telegram: 0 };
    if (running) {
      log.debug('poller: previous cycle still running, skipping');
      return result;
@ -73,7 +78,7 @@ export function createPoller(
          throw err;
        }
      }
-      if (result.fcm + result.email > 0) {
+      if (result.fcm + result.email + result.telegram > 0) {
        log.info({ enqueued: result, mode: cfg.mode, owned: cfg.ownedChannels }, 'poller: cycle complete');
      }
    } finally {
--- a/src/queue.ts
+++ b/src/queue.ts
@ -5,6 +5,7 @@ import type { Channel, SendJob } from './types.js';
 export const QUEUE_NAMES: Record<Channel, string> = {
  fcm: 'tcef-fcm',
  email: 'tcef-email',
+  telegram: 'tcef-telegram',
 };

 /**
@ -34,6 +35,7 @@ export function createQueues(cfg: Config): Record<Channel, Queue<SendJob>> {
  return {
    fcm: new Queue<SendJob>(QUEUE_NAMES.fcm, opts),
    email: new Queue<SendJob>(QUEUE_NAMES.email, opts),
+    telegram: new Queue<SendJob>(QUEUE_NAMES.telegram, opts),
  };
 }

--- a/src/telegram/client.ts
+++ b/src/telegram/client.ts
@ -0,0 +1,47 @@
+/** Result of a Telegram Bot API call. */
+export type TgResult =
+  | { ok: true }
+  | { ok: false; retryAfterMs?: number; blocked?: boolean; error: string };
+
+export interface TelegramClient {
+  sendMessage(chatId: number, text: string, parseMode: 'Markdown'): Promise<TgResult>;
+  sendLocation(chatId: number, latitude: number, longitude: number): Promise<TgResult>;
+}
+
+const API = 'https://api.telegram.org';
+
+/** Interpret a Telegram API JSON response into a TgResult. Pure — unit-tested. */
+export function interpretResponse(status: number, body: unknown): TgResult {
+  const b = body as { ok?: boolean; error_code?: number; description?: string; parameters?: { retry_after?: number } };
+  if (b?.ok) return { ok: true };
+  const code = b?.error_code ?? status;
+  const description = b?.description ?? `HTTP ${status}`;
+  if (code === 429) {
+    const retry = b?.parameters?.retry_after ?? 1;
+    return { ok: false, retryAfterMs: retry * 1000, error: description };
+  }
+  // 403 = bot blocked / kicked by the user → mark sub inactive, do not retry.
+  const blocked = code === 403;
+  return { ok: false, blocked, error: description };
+}
+
+/** Real client over the Telegram Bot API (Node 22 global fetch). */
+export function createTelegramClient(token: string): TelegramClient {
+  async function call(method: string, payload: Record<string, unknown>): Promise<TgResult> {
+    try {
+      const res = await fetch(`${API}/bot${token}/${method}`, {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: JSON.stringify(payload),
+      });
+      const json = await res.json().catch(() => ({}));
+      return interpretResponse(res.status, json);
+    } catch (err) {
+      return { ok: false, error: `network: ${String(err)}` };
+    }
+  }
+  return {
+    sendMessage: (chatId, text, parseMode) => call('sendMessage', { chat_id: chatId, text, parse_mode: parseMode }),
+    sendLocation: (chatId, latitude, longitude) => call('sendLocation', { chat_id: chatId, latitude, longitude }),
+  };
+}
--- a/src/telegram/message.ts
+++ b/src/telegram/message.ts
@ -0,0 +1,35 @@
+import type { Lang } from '../messages.js';
+import type { NotificationDoc } from '../types.js';
+
+/**
+ * `firelink` i18n, verbatim from the node-red locales (docs/legacy-telegram.md).
+ * `{{server}}` = site host, `{{enc}}` = the Iron `sealed` string.
+ */
+export const FIRELINK: Partial<Record<Lang, string>> = {
+  es: '[más información](https://{{server}}/fire/{{enc}})',
+  en: '[more information](https://{{server}}/fire/{{enc}})',
+  gl: '[máis información](https://{{server}}/fire/{{enc}})',
+};
+
+export interface TelegramMessages {
+  text: string;
+  parseMode: 'Markdown';
+  latitude: number;
+  longitude: number;
+}
+
+/**
+ * Build the two messages node-red sends per alert: a Markdown text
+ * (`content` + firelink) and a location (fire lat/lon).
+ */
+export function buildTelegramMessages(notif: NotificationDoc, lang: Lang, server: string): TelegramMessages {
+  const tpl = FIRELINK[lang] ?? FIRELINK.es!;
+  const link = tpl.replaceAll('{{server}}', server).replaceAll('{{enc}}', notif.sealed);
+  const [lon, lat] = notif.geo.coordinates;
+  return {
+    text: `${notif.content}\n${link}`,
+    parseMode: 'Markdown',
+    latitude: lat,
+    longitude: lon,
+  };
+}
--- a/src/telegram/rate.ts
+++ b/src/telegram/rate.ts
@ -0,0 +1,26 @@
+/**
+ * How long to wait before sending to a chat again, to respect Telegram's
+ * ~1 msg/s per-chat limit. Pure so it can be unit-tested.
+ */
+export function computeWaitMs(lastSentMs: number | undefined, nowMs: number, minIntervalMs: number): number {
+  if (lastSentMs === undefined) return 0;
+  const elapsed = nowMs - lastSentMs;
+  return elapsed >= minIntervalMs ? 0 : minIntervalMs - elapsed;
+}
+
+/** Per-chat throttle enforcing a minimum interval between sends to the same chat. */
+export class PerChatThrottle {
+  private last = new Map<number, number>();
+
+  constructor(
+    private readonly minIntervalMs: number,
+    private readonly now: () => number = () => Date.now(),
+    private readonly sleep: (ms: number) => Promise<void> = (ms) => new Promise((r) => setTimeout(r, ms)),
+  ) {}
+
+  async acquire(chatId: number): Promise<void> {
+    const wait = computeWaitMs(this.last.get(chatId), this.now(), this.minIntervalMs);
+    if (wait > 0) await this.sleep(wait);
+    this.last.set(chatId, this.now());
+  }
+}
--- a/src/types.ts
+++ b/src/types.ts
@ -7,11 +7,13 @@ export interface NotificationDoc {
  subsId?: ObjectId;
  content: string;
  geo: { type: 'Point'; coordinates: [number, number] }; // [lng, lat]
-  type: 'mobile' | 'web';
+  type: 'mobile' | 'web' | 'telegram';
  notified?: boolean;
  notifiedAt?: Date;
  emailNotified?: boolean;
  emailNotifiedAt?: Date;
+  telegramNotified?: boolean;
+  telegramNotifiedAt?: Date;
  when: Date;
  sealed: string;
  createdAt?: Date;
@ -31,11 +33,13 @@ export interface UserDoc {
  services?: { password?: { bcrypt?: string } };
 }

-export type Channel = 'fcm' | 'email';
+export type Channel = 'fcm' | 'email' | 'telegram';

 /** Maps a notification `type` to a channel. */
 export function channelOfType(type: NotificationDoc['type']): Channel {
-  return type === 'mobile' ? 'fcm' : 'email';
+  if (type === 'mobile') return 'fcm';
+  if (type === 'telegram') return 'telegram';
+  return 'email';
 }

 /** Record in `notification_sends` — persistent idempotency (unique on notificationId+channel). */
--- a/test/poller.test.ts
+++ b/test/poller.test.ts
@ -46,7 +46,7 @@ describe('poller.runOnce', () => {

    const res = await poller.runOnce();

-    expect(res).toEqual({ fcm: 1, email: 1 });
+    expect(res).toEqual({ fcm: 1, email: 1, telegram: 0 });
    expect(added.fcm).toHaveLength(1);
    expect(added.email).toHaveLength(1);
    // claimedBy provenance set
--- a/test/telegram.test.ts
+++ b/test/telegram.test.ts
@ -0,0 +1,137 @@
+import { beforeEach, describe, expect, it } from 'vitest';
+import { ObjectId } from 'mongodb';
+import type { Repos } from '../src/db.js';
+import type { Config } from '../src/config.js';
+import { buildTelegramMessages } from '../src/telegram/message.js';
+import { computeWaitMs, PerChatThrottle } from '../src/telegram/rate.js';
+import { interpretResponse, type TelegramClient, type TgResult } from '../src/telegram/client.js';
+import { processTelegramJob, type Deps } from '../src/handlers.js';
+import type { NotificationDoc } from '../src/types.js';
+import { aNotif, aUser, freshRepos, silentLog, testCfg } from './helpers.js';
+
+describe('buildTelegramMessages', () => {
+  it('builds text (content + firelink) and location', () => {
+    const notif = aNotif({ type: 'telegram', content: '🔥 Alerta de posible fuego a 5 км de distancia (fuente NASA)', sealed: 'ENC', geo: { type: 'Point', coordinates: [-3.7, 40.4] } });
+    const m = buildTelegramMessages(notif, 'es', 'fuegos.comunes.org');
+    expect(m.text).toBe('🔥 Alerta de posible fuego a 5 км de distancia (fuente NASA)\n[más información](https://fuegos.comunes.org/fire/ENC)');
+    expect(m.parseMode).toBe('Markdown');
+    expect(m.latitude).toBe(40.4);
+    expect(m.longitude).toBe(-3.7);
+  });
+  it('uses the english firelink', () => {
+    const notif = aNotif({ type: 'telegram', sealed: 'X' });
+    expect(buildTelegramMessages(notif, 'en', 'fires.comunes.org').text).toContain('[more information](https://fires.comunes.org/fire/X)');
+  });
+});
+
+describe('rate limiting', () => {
+  it('computeWaitMs respects the per-chat interval', () => {
+    expect(computeWaitMs(undefined, 1000, 1000)).toBe(0);
+    expect(computeWaitMs(1000, 1500, 1000)).toBe(500);
+    expect(computeWaitMs(1000, 2000, 1000)).toBe(0);
+  });
+  it('PerChatThrottle waits then records', async () => {
+    let t = 0;
+    const waits: number[] = [];
+    const thr = new PerChatThrottle(1000, () => t, async (ms) => { waits.push(ms); t += ms; });
+    await thr.acquire(42); // first: no wait
+    await thr.acquire(42); // immediately after: waits 1000
+    expect(waits).toEqual([1000]);
+  });
+});
+
+describe('interpretResponse', () => {
+  it('ok', () => {
+    expect(interpretResponse(200, { ok: true })).toEqual({ ok: true });
+  });
+  it('429 with retry_after', () => {
+    const r = interpretResponse(429, { ok: false, error_code: 429, description: 'Too Many Requests', parameters: { retry_after: 7 } });
+    expect(r).toEqual({ ok: false, retryAfterMs: 7000, error: 'Too Many Requests' });
+  });
+  it('403 marks blocked', () => {
+    const r = interpretResponse(403, { ok: false, error_code: 403, description: 'bot was blocked by the user' });
+    expect(r).toMatchObject({ ok: false, blocked: true });
+  });
+});
+
+// ---- handler ----
+function fakeTgClient(result: TgResult = { ok: true }) {
+  const calls: Array<{ method: string; chatId: number }> = [];
+  const client: TelegramClient = {
+    async sendMessage(chatId) { calls.push({ method: 'message', chatId }); return result; },
+    async sendLocation(chatId) { calls.push({ method: 'location', chatId }); return result; },
+  };
+  return { client, calls };
+}
+
+let repos: Repos;
+const baseCfg: Config = testCfg({ ownedChannels: ['telegram'] });
+beforeEach(async () => { repos = await freshRepos(); });
+
+function tgDeps(clientResult: TgResult, over: Partial<Deps> = {}): { deps: Deps; calls: Array<{ method: string; chatId: number }> } {
+  const { client, calls } = fakeTgClient(clientResult);
+  const deps: Deps = {
+    cfg: baseCfg, repos, log: silentLog,
+    telegram: {
+      clientFor: () => client,
+      throttle: new PerChatThrottle(0, () => 0, async () => {}),
+      server: 'fuegos.comunes.org',
+    },
+    ...over,
+  };
+  return { deps, calls };
+}
+
+async function seedTelegramNotif(chatId = 111): Promise<NotificationDoc> {
+  const subsId = new ObjectId();
+  await repos.subscriptions.insertOne({ _id: subsId, location: { lat: 40.4, lon: -3.7 }, geo: { type: 'Point', coordinates: [-3.7, 40.4] }, distance: 40, owner: 'u1', type: 'telegram', chatId } as never);
+  await repos.users.insertOne(aUser({ _id: 'u1', lang: 'es' }));
+  const notif = aNotif({ type: 'telegram', userId: 'u1', subsId });
+  await repos.notifications.insertOne(notif);
+  return notif;
+}
+
+describe('processTelegramJob', () => {
+  it('sends text + location and marks telegramNotified (full)', async () => {
+    const notif = await seedTelegramNotif(111);
+    const { deps, calls } = tgDeps({ ok: true });
+    await processTelegramJob(deps, notif._id.toHexString());
+    expect(calls).toEqual([{ method: 'message', chatId: 111 }, { method: 'location', chatId: 111 }]);
+    const after = await repos.notifications.findOne({ _id: notif._id });
+    expect(after?.telegramNotified).toBe(true);
+    expect(await repos.sends.findOne({ notificationId: notif._id, channel: 'telegram' })).not.toBeNull();
+  });
+
+  it('dry-run does not send', async () => {
+    const notif = await seedTelegramNotif();
+    const { deps, calls } = tgDeps({ ok: true }, { cfg: { ...baseCfg, mode: 'dry-run' } });
+    await processTelegramJob(deps, notif._id.toHexString());
+    expect(calls).toHaveLength(0);
+    expect((await repos.notifications.findOne({ _id: notif._id }))?.telegramNotified).toBeUndefined();
+  });
+
+  it('is idempotent: does not resend', async () => {
+    const notif = await seedTelegramNotif();
+    const { deps, calls } = tgDeps({ ok: true });
+    await processTelegramJob(deps, notif._id.toHexString());
+    await processTelegramJob(deps, notif._id.toHexString()); // guard: already notified
+    expect(calls).toHaveLength(2); // only the first run sent (2 msgs)
+  });
+
+  it('marks the sub inactive when the bot is blocked (403), no retry', async () => {
+    const notif = await seedTelegramNotif(222);
+    const { deps } = tgDeps({ ok: false, blocked: true, error: 'blocked' });
+    await processTelegramJob(deps, notif._id.toHexString());
+    const sub = await repos.subscriptions.findOne({ chatId: 222 });
+    expect((sub as { active?: boolean }).active).toBe(false);
+    const send = await repos.sends.findOne({ notificationId: notif._id, channel: 'telegram' });
+    expect(send?.status).toBe('dead-token');
+  });
+
+  it('releases reservation and rethrows on 429 (retry)', async () => {
+    const notif = await seedTelegramNotif();
+    const { deps } = tgDeps({ ok: false, retryAfterMs: 5000, error: 'Too Many Requests' });
+    await expect(processTelegramJob(deps, notif._id.toHexString())).rejects.toThrow();
+    expect(await repos.sends.countDocuments({ notificationId: notif._id })).toBe(0);
+  });
+});