feat: forward albums as a single grouped batch

AGENTS.md

@@ -96,3 +96,8 @@ knip on every PR.
   (spam, mass-forwarding to unrelated targets, etc.).
 - **Deleted-message and FloodWait handling lives in `queue.ts`.** Errors are
   classified by inspecting RpcError shape/message; preserve that when editing.
+- **Album handling spans `client.ts` and `forwarder.ts`.** `createClient`
+  enables `messageGroupingInterval`, which means messages belonging to a media
+  group are dispatched ONLY as `message_group` updates — `onNewMessage` never
+  fires for them. Any new message-handling code must subscribe to both updates
+  (see `Forwarder.start`), or albums will be silently dropped.

README.md

@@ -9,6 +9,8 @@ A CLI tool that monitors Telegram channels and automatically forwards specific c
 - Pick source and target channels from a list — no manual ID entry
 - Filter by content type — photos, videos, text, stickers, and more
 - Filter by keyword — only forward (or skip) messages whose text/caption matches
+- Albums (media groups) arrive as albums — parts are batched into a single forward that keeps the
+  grouped layout, and the caption counts for the whole album when matching keywords
 - Handles Telegram FloodWait automatically with reactive backoff
 - Dry-run mode shows what would be forwarded without sending anything
 - Verbose mode traces why each message is or isn't forwarded; optional file logging

src/client.ts

@@ -2,11 +2,24 @@ import { TelegramClient } from '@mtcute/node';
 
 import type { AppConfig } from './types.js';
 
+// Buffer window for collecting the parts of an album (media group) into a
+// single message_group update. Telegram delivers an album as separate messages
+// sharing a groupedId, usually within tens of milliseconds; 250 ms is the value
+// recommended by mtcute. Trade-off: every album is delayed by exactly this long
+// before it can be forwarded (single messages are unaffected), while a window
+// too short to catch a slow album splits it into two smaller albums — a
+// cosmetic issue, never a lost message.
+//
+// NOTE: with this enabled, album messages are dispatched ONLY as message_group
+// updates — onNewMessage never fires for them (see Forwarder.start).
+const MESSAGE_GROUPING_INTERVAL_MS = 250;
+
 export function createClient(config: AppConfig): TelegramClient {
   return new TelegramClient({
     apiId: config.apiId,
     apiHash: config.apiHash,
     // Passing a string path → @mtcute/node automatically uses SQLite (better-sqlite3)
     storage: config.sessionPath,
+    updates: { messageGroupingInterval: MESSAGE_GROUPING_INTERVAL_MS },
   });
 }

src/filter.ts

@@ -50,25 +50,34 @@ export function matchesContentType(msg: MessageLike, types: ContentType[]): bool
   }
 }
 
-// Decide whether a message's text passes a group's keyword filters. Matching is
-// case-insensitive substring. An empty (or absent) list means "no constraint":
-// no exclude list blocks nothing, no include list lets any text through.
-// Exclude takes precedence — a message hit by both is dropped. A media-only
-// message has empty text, so it can never satisfy a non-empty include list.
+// Decide whether a logical unit — a single message, or every message of one
+// album — passes a group's keyword filters. Matching is case-insensitive
+// substring. An empty (or absent) list means "no constraint": no exclude list
+// blocks nothing, no include list lets any text through.
+//
+// The unit passes as a whole or not at all: an album's caption usually lives
+// on just one of its messages, but it speaks for the entire post. Exclude
+// takes precedence — one excluded text drops the whole unit, even when another
+// message satisfies an include keyword. A media-only unit has no text, so it
+// can never satisfy a non-empty include list.
 export function matchesKeywords(
-  msg: MessageLike,
+  messages: MessageLike[],
   include: string[] = [],
   exclude: string[] = [],
 ): boolean {
   if (include.length === 0 && exclude.length === 0) return true;
 
-  const text = (msg.text ?? '').toLowerCase();
+  const texts = messages.map((msg) => (msg.text ?? '').toLowerCase());
 
-  if (exclude.some((kw) => text.includes(kw.toLowerCase()))) return false;
-  if (include.length > 0) return include.some((kw) => text.includes(kw.toLowerCase()));
+  if (texts.some((text) => hasAnyKeyword(text, exclude))) return false;
+  if (include.length > 0) return texts.some((text) => hasAnyKeyword(text, include));
   return true;
 }
 
+function hasAnyKeyword(text: string, keywords: string[]): boolean {
+  return keywords.some((kw) => text.includes(kw.toLowerCase()));
+}
+
 // mtcute reads a bare string as a username/phone; numeric (Bot-API marked) IDs
 // such as -1001234567890 must be passed as numbers, otherwise it tries to
 // resolve them as @usernames and fails with "Peer with username ... not found".

src/forwarder.ts

@@ -1,5 +1,5 @@
 import type { Message } from '@mtcute/core';
-import { Dispatcher } from '@mtcute/dispatcher';
+import { Dispatcher, type MessageContext } from '@mtcute/dispatcher';
 import type { TelegramClient } from '@mtcute/node';
 
 import { saveConfig } from './config.js';
@@ -11,7 +11,12 @@ import {
   toInputPeer,
 } from './filter.js';
 import { logger as defaultLogger, type Logger } from './logger.js';
-import { type ForwardOutcome, RateLimitedQueue, SkippedForward } from './queue.js';
+import {
+  type ForwardOutcome,
+  isDeletedMessageError,
+  RateLimitedQueue,
+  SkippedForward,
+} from './queue.js';
 import type { AppConfig, ForwardGroup } from './types.js';
 
 interface ForwarderStats {
@@ -187,51 +192,81 @@ export class Forwarder {
     const dp = Dispatcher.for(this.client);
     this.dp = dp;
 
-    dp.onNewMessage(async (upd) => {
-      const chat = upd.chat as { id: number; username?: string | null };
+    // A single message arrives via onNewMessage as a one-element unit. An
+    // album's messages are buffered by the client (messageGroupingInterval in
+    // client.ts) and arrive ONLY as one message_group update — onNewMessage
+    // never fires for them, so BOTH handlers are needed to see every message
+    // exactly once. Either way ctx.messages holds the whole unit, and the unit
+    // is forwarded as one batch so an album reaches the target as an album.
+    const handleUpdate = (ctx: MessageContext): void => this.handleMessages(ctx.messages);
+    dp.onNewMessage(handleUpdate);
+    dp.onMessageGroup(handleUpdate);
 
-      for (const group of this.groups) {
-        const isSourceMatch = group.sourcePeers.some((peer) => matchesPeer(chat, peer));
-        if (!isSourceMatch) continue;
+    this.log.info(`Dispatcher started. Monitoring ${this.groups.length} group(s).`);
+  }
 
-        const mediaType = upd.media?.type ?? 'text';
-        if (!matchesContentType(upd, group.contentTypes)) {
-          this.log.debug(
-            `msg ${upd.id} from ${describeChat(chat)} — skipped: type '${mediaType}' ` +
-              `not in [${group.contentTypes.join(', ')}] for "${group.name}"`,
-          );
-          continue;
-        }
+  // Match one unit (a single message or a whole album) against the enabled
+  // groups and enqueue a forward per matching target. Matching must never
+  // throw: a handler error would surface as an opaque client error and
+  // silently drop the unit, so anything unexpected is caught and logged.
+  private handleMessages(messages: MessageContext[]): void {
+    try {
+      this.matchAndEnqueue(messages);
+    } catch (err) {
+      const ids = messages.map((msg) => msg.id).join(', ');
+      const reason = err instanceof Error ? err.message : String(err);
+      this.log.error(`Failed to process msg(s) ${ids}: ${reason}`);
+    }
+  }
 
-        if (!matchesKeywords(upd, group.includeKeywords, group.excludeKeywords)) {
-          this.log.debug(
-            `msg ${upd.id} from ${describeChat(chat)} — skipped: keyword filter for "${group.name}"`,
-          );
-          continue;
-        }
+  private matchAndEnqueue(messages: MessageContext[]): void {
+    const first = messages[0];
+    if (!first) return;
+    const chat = first.chat as { id: number; username?: string | null };
 
-        const msg = upd as unknown as Message;
-        for (const targetPeer of group.targetPeers) {
-          // A target retired earlier this session is skipped silently.
-          if (this.deadTargets.has(targetPeer)) continue;
-
-          const label = `[${group.name}] msg ${upd.id} (${mediaType}) → ${targetPeer}`;
-          if (this.dryRun) {
-            this.log.info(`[dry-run] would forward ${label}`);
-            continue;
-          }
-          this.log.debug(`Enqueued ${label}`);
-          this.queue.enqueue(() => this.forwardOne(msg, group, targetPeer), label);
-        }
+    for (const group of this.groups) {
+      const isSourceMatch = group.sourcePeers.some((peer) => matchesPeer(chat, peer));
+      if (!isSourceMatch) continue;
+
+      if (!matchesKeywords(messages, group.includeKeywords, group.excludeKeywords)) {
+        this.log.debug(
+          `${describeBatch(messages)} from ${describeChat(chat)} — skipped: keyword filter ` +
+            `for "${group.name}"`,
+        );
+        continue;
       }
-    });
 
-    this.log.info(`Dispatcher started. Monitoring ${this.groups.length} group(s).`);
+      // Content types are matched per message: forwarding the matching subset
+      // of an album in one call still reaches the target as one (smaller) album.
+      const matching = messages.filter((msg) => matchesContentType(msg, group.contentTypes));
+      if (matching.length === 0) {
+        this.log.debug(
+          `${describeBatch(messages)} from ${describeChat(chat)} — skipped: no type in ` +
+            `[${group.contentTypes.join(', ')}] for "${group.name}"`,
+        );
+        continue;
+      }
+
+      const batch = matching as unknown as Message[];
+      for (const targetPeer of group.targetPeers) {
+        // A target retired earlier this session is skipped silently.
+        if (this.deadTargets.has(targetPeer)) continue;
+
+        const label = `[${group.name}] ${describeBatch(matching)} → ${targetPeer}`;
+        if (this.dryRun) {
+          this.log.info(`[dry-run] would forward ${label}`);
+          continue;
+        }
+        this.log.debug(`Enqueued ${label}`);
+        this.queue.enqueue(() => this.forwardBatch(batch, group, targetPeer), label);
+      }
+    }
   }
 
-  // Forward one message to one target, transparently recovering from the two ways
-  // an invalid-peer error (CHANNEL_INVALID/PEER_ID_INVALID) can arise. We probe
-  // the target with getChat to tell them apart:
+  // Forward one unit (a single message or an album batch) to one target,
+  // transparently recovering from the two ways an invalid-peer error
+  // (CHANNEL_INVALID/PEER_ID_INVALID) can arise. We probe the target with
+  // getChat to tell them apart:
   //
   //   • the TARGET was a basic group upgraded to a supergroup (its id changed) —
   //     `migratedTo` means follow it (rewrite config + retry on the new id); an
@@ -242,9 +277,13 @@ export class Forwarder {
   //
   // Other errors (FloodWait, deleted message, transient) are rethrown so the
   // queue's own retry logic handles them.
-  private async forwardOne(msg: Message, group: ForwardGroup, targetPeer: string): Promise<void> {
+  private async forwardBatch(
+    msgs: Message[],
+    group: ForwardGroup,
+    targetPeer: string,
+  ): Promise<void> {
     try {
-      await this.forward(msg, group, targetPeer);
+      await this.forwardSurviving(msgs, group, targetPeer);
     } catch (err) {
       if (!isInvalidPeerError(err)) throw err;
 
@@ -254,7 +293,7 @@ export class Forwarder {
       // source. Re-warm the cache once and retry before giving up.
       if (migration === 'healthy') {
         if (await this.recoverSource(group)) {
-          await this.forward(msg, group, targetPeer);
+          await this.forwardSurviving(msgs, group, targetPeer);
           return;
         }
         throw err;
@@ -274,13 +313,48 @@ export class Forwarder {
         `Target "${targetPeer}" was upgraded to a supergroup — now forwarding ` +
           `"${group.name}" to "${migration}" (config updated).`,
       );
-      await this.forward(msg, group, migration);
+      await this.forwardSurviving(msgs, group, migration);
+    }
+  }
+
+  // Forward a batch, surviving the deletion race this tool exists to win: when
+  // part of an album is deleted between matching and sending, the batched
+  // forward fails with MESSAGE_ID_INVALID for the WHOLE album. Rather than
+  // dropping the parts that still exist, re-fetch the batch and retry with the
+  // survivors; only when nothing survives does the deleted-message error
+  // propagate for the queue to record as a skip. A single message needs no
+  // recovery — deleted is deleted.
+  private async forwardSurviving(
+    msgs: Message[],
+    group: ForwardGroup,
+    targetPeer: string,
+  ): Promise<void> {
+    try {
+      await this.forward(msgs, group, targetPeer);
+    } catch (err) {
+      if (msgs.length <= 1 || !isDeletedMessageError(err)) throw err;
+
+      const survivors = await this.fetchSurviving(msgs);
+      if (survivors.length === 0) throw err;
+      this.log.warn(
+        `${msgs.length - survivors.length} of ${msgs.length} album message(s) were deleted ` +
+          `before forwarding — forwarding the ${survivors.length} remaining.`,
+      );
+      await this.forward(survivors, group, targetPeer);
     }
   }
 
-  private forward(msg: Message, group: ForwardGroup, targetPeer: string): Promise<unknown> {
+  // Re-fetch a batch from the source and keep only the messages that still
+  // exist (getMessages returns null in place of a deleted message).
+  private async fetchSurviving(msgs: Message[]): Promise<Message[]> {
+    const ids = msgs.map((msg) => msg.id);
+    const fetched = await this.client.getMessages(msgs[0]!.chat.id, ids);
+    return fetched.filter((msg): msg is Message => msg !== null);
+  }
+
+  private forward(msgs: Message[], group: ForwardGroup, targetPeer: string): Promise<unknown> {
     return this.client.forwardMessages({
-      messages: [msg],
+      messages: msgs,
       toChatId: toInputPeer(targetPeer),
       noAuthor: group.noAuthor,
     });
@@ -349,6 +423,19 @@ function describeChat(chat: { id: number; username?: string | null }): string {
   return chat.username ? `@${chat.username}` : String(chat.id);
 }
 
+// Compact description of a unit for queue labels and logs: a lone message
+// keeps the familiar "msg 123 (photo)" form, an album shows its id range and
+// the media types it contains.
+function describeBatch(
+  messages: readonly { id: number; media?: { type: string } | null }[],
+): string {
+  const first = messages[0]!;
+  if (messages.length === 1) return `msg ${first.id} (${first.media?.type ?? 'text'})`;
+  const last = messages[messages.length - 1]!;
+  const types = [...new Set(messages.map((msg) => msg.media?.type ?? 'text'))].join(', ');
+  return `album ${first.id}-${last.id} (${messages.length} messages: ${types})`;
+}
+
 // Socket-level errors that just mean the connection dropped and will be
 // re-established — not something the user needs to act on. Matched by both the
 // Node `code` (when present) and the message text, since mtcute may rewrap the

src/queue.ts

@@ -69,7 +69,7 @@ export class RateLimitedQueue {
         return;
       }
 
-      if (isDeletedMessage(err)) {
+      if (isDeletedMessageError(err)) {
         this.log.warn(`Message was deleted before forwarding — skipping (${label})`);
         this.onOutcome?.('skipped', label);
         return;
@@ -96,7 +96,10 @@ export class RateLimitedQueue {
   }
 }
 
-function isDeletedMessage(err: unknown): boolean {
+// "This message no longer exists" in its various shapes. Exported because the
+// forwarder uses the same classification to salvage the surviving parts of an
+// album whose batched forward failed on a deleted member.
+export function isDeletedMessageError(err: unknown): boolean {
   if (!(err instanceof Error)) return false;
   const msg =
     ('errorMessage' in err ? (err as { errorMessage: string }).errorMessage : null) ?? err.message;