From adea0f4afd571b22fb99889805d04fa50b547544 Mon Sep 17 00:00:00 2001 From: Roya Date: Sat, 11 Jul 2026 01:31:28 +0800 Subject: [PATCH] feat: forward albums as a single grouped batch Telegram delivers an album (media group) as separate messages sharing a groupedId, so each part was forwarded individually and reached the target as loose messages. Enable mtcute's messageGroupingInterval so a whole album arrives as one message_group update, and forward the unit in a single forwardMessages call that keeps the grouped layout. - route onNewMessage and onMessageGroup through one handler; with grouping enabled album messages are dispatched ONLY as message_group, so both subscriptions are required to see every message - match keywords per unit: an album's caption lives on one message but speaks for the whole post (exclude on any part drops the unit) - keep content-type filtering per message; the matching subset is still batched and stays an album at the target - salvage the surviving parts when an album member is deleted between matching and sending, instead of losing the whole batch to MESSAGE_ID_INVALID - singles are unaffected; an album waits at most 250 ms (upstream recommendation) and now needs one API call instead of N rate-limited sends, finishing sooner than before --- AGENTS.md | 5 ++ README.md | 2 + src/client.ts | 13 +++ src/filter.ts | 27 ++++--- src/forwarder.ts | 177 ++++++++++++++++++++++++++++++----------- src/queue.ts | 7 +- test/filter.test.ts | 44 +++++++--- test/forwarder.test.ts | 161 ++++++++++++++++++++++++++++++++++++- 8 files changed, 366 insertions(+), 70 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index af87c2e..de69deb 100644 --- a/AGENTS.md +++ b/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. diff --git a/README.md b/README.md index 0c49905..b59d6ac 100644 --- a/README.md +++ b/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 diff --git a/src/client.ts b/src/client.ts index e8cd553..5ed0ad1 100644 --- a/src/client.ts +++ b/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 }, }); } diff --git a/src/filter.ts b/src/filter.ts index da22850..00f2973 100644 --- a/src/filter.ts +++ b/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". diff --git a/src/forwarder.ts b/src/forwarder.ts index bf95dd9..47a0d01 100644 --- a/src/forwarder.ts +++ b/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 { + private async forwardBatch( + msgs: Message[], + group: ForwardGroup, + targetPeer: string, + ): Promise { 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 { + 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 { + // 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 { + 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 { 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 diff --git a/src/queue.ts b/src/queue.ts index 3888d1c..27ab3bf 100644 --- a/src/queue.ts +++ b/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; diff --git a/test/filter.test.ts b/test/filter.test.ts index dd99fd5..56cc1f3 100644 --- a/test/filter.test.ts +++ b/test/filter.test.ts @@ -84,39 +84,57 @@ describe('matchesContentType', () => { describe('matchesKeywords', () => { it('passes everything when both lists are empty or absent', () => { - expect(matchesKeywords({ text: 'anything' })).toBe(true); - expect(matchesKeywords({ text: 'anything' }, [], [])).toBe(true); - expect(matchesKeywords({}, [], [])).toBe(true); + expect(matchesKeywords([{ text: 'anything' }])).toBe(true); + expect(matchesKeywords([{ text: 'anything' }], [], [])).toBe(true); + expect(matchesKeywords([{}], [], [])).toBe(true); }); it('keeps only messages containing an include keyword', () => { - expect(matchesKeywords({ text: 'breaking news today' }, ['breaking'])).toBe(true); - expect(matchesKeywords({ text: 'calm and quiet' }, ['breaking'])).toBe(false); + expect(matchesKeywords([{ text: 'breaking news today' }], ['breaking'])).toBe(true); + expect(matchesKeywords([{ text: 'calm and quiet' }], ['breaking'])).toBe(false); }); it('matches any one of several include keywords', () => { - expect(matchesKeywords({ text: 'market update' }, ['breaking', 'market'])).toBe(true); + expect(matchesKeywords([{ text: 'market update' }], ['breaking', 'market'])).toBe(true); }); it('drops messages containing an exclude keyword', () => { - expect(matchesKeywords({ text: 'sponsored post' }, [], ['sponsored'])).toBe(false); - expect(matchesKeywords({ text: 'real content' }, [], ['sponsored'])).toBe(true); + expect(matchesKeywords([{ text: 'sponsored post' }], [], ['sponsored'])).toBe(false); + expect(matchesKeywords([{ text: 'real content' }], [], ['sponsored'])).toBe(true); }); it('lets exclude win over include when both match', () => { - expect(matchesKeywords({ text: 'breaking ad' }, ['breaking'], ['ad'])).toBe(false); + expect(matchesKeywords([{ text: 'breaking ad' }], ['breaking'], ['ad'])).toBe(false); }); it('is case-insensitive', () => { - expect(matchesKeywords({ text: 'BREAKING' }, ['breaking'])).toBe(true); - expect(matchesKeywords({ text: 'Sponsored' }, [], ['SPONSORED'])).toBe(false); + expect(matchesKeywords([{ text: 'BREAKING' }], ['breaking'])).toBe(true); + expect(matchesKeywords([{ text: 'Sponsored' }], [], ['SPONSORED'])).toBe(false); }); it('treats a media-only message (no text) as empty', () => { // No text can satisfy a required include keyword… - expect(matchesKeywords({}, ['breaking'])).toBe(false); + expect(matchesKeywords([{}], ['breaking'])).toBe(false); // …but it is never caught by an exclude keyword. - expect(matchesKeywords({}, [], ['ad'])).toBe(true); + expect(matchesKeywords([{}], [], ['ad'])).toBe(true); + }); + + it('lets one captioned message satisfy include for the whole album', () => { + // An album's caption lives on a single message but speaks for the unit. + expect(matchesKeywords([{ text: 'breaking news' }, {}, {}], ['breaking'])).toBe(true); + expect(matchesKeywords([{}, {}, {}], ['breaking'])).toBe(false); + }); + + it('drops the whole album when any message hits an exclude keyword', () => { + expect(matchesKeywords([{ text: 'nice' }, { text: 'sponsored' }], [], ['sponsored'])).toBe( + false, + ); + }); + + it('lets exclude on one message beat include on another', () => { + expect(matchesKeywords([{ text: 'breaking' }, { text: 'ad' }], ['breaking'], ['ad'])).toBe( + false, + ); }); }); diff --git a/test/forwarder.test.ts b/test/forwarder.test.ts index b079a0b..8dca5eb 100644 --- a/test/forwarder.test.ts +++ b/test/forwarder.test.ts @@ -1,3 +1,4 @@ +import { Dispatcher } from '@mtcute/dispatcher'; import type { TelegramClient } from '@mtcute/node'; import { describe, expect, it, vi } from 'vitest'; @@ -28,6 +29,15 @@ const makeFakeClient = () => ({ }, resolvePeer: vi.fn<(peer: string | number) => Promise<{ _: string }>>(), iterDialogs: vi.fn<() => AsyncIterable>(), + forwardMessages: + vi.fn< + (params: { + messages: { id: number }[]; + toChatId: string | number; + noAuthor: boolean; + }) => Promise + >(), + getMessages: vi.fn<(chatId: number, ids: number[]) => Promise<({ id: number } | null)[]>>(), }); // An async iterable that yields nothing — enough for warmPeerCache to drain it. @@ -70,15 +80,34 @@ const makeConfig = (groups: ForwardGroup[]): AppConfig => ({ rateLimit: { delayMs: 0, jitterMs: 0, maxRetries: 0 }, }); -const makeForwarder = (config: AppConfig) => { +const makeForwarder = (config: AppConfig, opts: { dryRun?: boolean } = {}) => { const client = makeFakeClient(); const log = makeLogger(); const forwarder = new Forwarder(client as unknown as TelegramClient, config, { logger: log as unknown as Logger, + ...opts, }); return { client, log, forwarder }; }; +// A fake update message: just the fields the matching path reads. Chat matches +// the default group's '@src' source. +const makeMsg = (id: number, media: { type: string } | null = { type: 'photo' }, text = '') => ({ + id, + chat: { id: -100777, username: 'src' }, + media, + text, +}); + +// Drive the private unit-handling entrypoint directly — the same method both +// onNewMessage and onMessageGroup feed. Building real tl-shaped updates to go +// through the Dispatcher would couple these tests to mtcute internals. +const dispatch = (forwarder: Forwarder, messages: unknown[]): void => + (forwarder as unknown as { handleMessages(messages: unknown[]): void }).handleMessages(messages); + +const deletedError = () => + Object.assign(new Error('Telegram API error 400'), { errorMessage: 'MESSAGE_ID_INVALID' }); + describe('Forwarder', () => { it('monitors only the enabled groups', () => { const { client, log, forwarder } = makeForwarder( @@ -151,6 +180,136 @@ describe('Forwarder', () => { expect(() => forwarder.stop()).not.toThrow(); expect(client.onUpdate.remove).not.toHaveBeenCalled(); }); + + it('registers both the single-message and the album handler', () => { + // With messageGroupingInterval enabled (client.ts), album messages arrive + // ONLY as message_group updates — missing either handler silently drops + // a whole class of messages. + const onNewMessage = vi.spyOn(Dispatcher.prototype, 'onNewMessage'); + const onMessageGroup = vi.spyOn(Dispatcher.prototype, 'onMessageGroup'); + const { forwarder } = makeForwarder(makeConfig([makeGroup()])); + + forwarder.start(); + + expect(onNewMessage).toHaveBeenCalledTimes(1); + expect(onMessageGroup).toHaveBeenCalledTimes(1); + + forwarder.stop(); + onNewMessage.mockRestore(); + onMessageGroup.mockRestore(); + }); +}); + +describe('Forwarder message handling', () => { + it('forwards a single message on its own', async () => { + const { client, forwarder } = makeForwarder(makeConfig([makeGroup()])); + client.forwardMessages.mockResolvedValue(undefined); + + dispatch(forwarder, [makeMsg(7)]); + + await vi.waitFor(() => expect(client.forwardMessages).toHaveBeenCalledTimes(1)); + const call = client.forwardMessages.mock.calls[0]![0]; + expect(call.messages.map((msg) => msg.id)).toEqual([7]); + expect(call.toChatId).toBe('@dst'); + expect(call.noAuthor).toBe(false); + }); + + it('forwards an album as one batched call so it stays an album', async () => { + const { client, forwarder } = makeForwarder(makeConfig([makeGroup()])); + client.forwardMessages.mockResolvedValue(undefined); + + dispatch(forwarder, [makeMsg(1), makeMsg(2), makeMsg(3)]); + + await vi.waitFor(() => expect(client.forwardMessages).toHaveBeenCalledTimes(1)); + expect(client.forwardMessages.mock.calls[0]![0].messages.map((msg) => msg.id)).toEqual([ + 1, 2, 3, + ]); + }); + + it('forwards only the content-type-matching subset, still in one call', async () => { + const { client, forwarder } = makeForwarder( + makeConfig([makeGroup({ contentTypes: ['photo'] })]), + ); + client.forwardMessages.mockResolvedValue(undefined); + + dispatch(forwarder, [makeMsg(1), makeMsg(2, { type: 'video' }), makeMsg(3)]); + + await vi.waitFor(() => expect(client.forwardMessages).toHaveBeenCalledTimes(1)); + expect(client.forwardMessages.mock.calls[0]![0].messages.map((msg) => msg.id)).toEqual([1, 3]); + }); + + it('forwards a whole album when only its caption message matches include keywords', async () => { + const { client, forwarder } = makeForwarder( + makeConfig([makeGroup({ includeKeywords: ['breaking'] })]), + ); + client.forwardMessages.mockResolvedValue(undefined); + + dispatch(forwarder, [makeMsg(1, { type: 'photo' }, 'breaking news'), makeMsg(2), makeMsg(3)]); + + await vi.waitFor(() => expect(client.forwardMessages).toHaveBeenCalledTimes(1)); + expect(client.forwardMessages.mock.calls[0]![0].messages).toHaveLength(3); + }); + + it('drops the whole album when any message hits an exclude keyword', async () => { + const { client, forwarder } = makeForwarder( + makeConfig([makeGroup({ excludeKeywords: ['sponsored'] })]), + ); + + dispatch(forwarder, [makeMsg(1), makeMsg(2, { type: 'photo' }, 'sponsored post')]); + + await new Promise((resolve) => setTimeout(resolve, 10)); + expect(client.forwardMessages).not.toHaveBeenCalled(); + expect(forwarder.pending).toBe(0); + }); + + it('salvages the surviving album parts when one is deleted mid-flight', async () => { + const { client, log, forwarder } = makeForwarder(makeConfig([makeGroup()])); + client.forwardMessages.mockRejectedValueOnce(deletedError()).mockResolvedValue(undefined); + client.getMessages.mockResolvedValue([makeMsg(1), null, makeMsg(3)]); + + dispatch(forwarder, [makeMsg(1), makeMsg(2), makeMsg(3)]); + + await vi.waitFor(() => expect(client.forwardMessages).toHaveBeenCalledTimes(2)); + expect(client.getMessages).toHaveBeenCalledWith(-100777, [1, 2, 3]); + expect(client.forwardMessages.mock.calls[1]![0].messages.map((msg) => msg.id)).toEqual([1, 3]); + expect(log.warn).toHaveBeenCalledWith(expect.stringContaining('deleted before forwarding')); + }); + + it('skips a deleted single message without a refetch', async () => { + const { client, log, forwarder } = makeForwarder(makeConfig([makeGroup()])); + client.forwardMessages.mockRejectedValue(deletedError()); + + dispatch(forwarder, [makeMsg(1)]); + + await vi.waitFor(() => + expect(log.warn).toHaveBeenCalledWith(expect.stringContaining('deleted before forwarding')), + ); + expect(client.getMessages).not.toHaveBeenCalled(); + expect(client.forwardMessages).toHaveBeenCalledTimes(1); + }); + + it('gives up on an album only when every part is gone', async () => { + const { client, log, forwarder } = makeForwarder(makeConfig([makeGroup()])); + client.forwardMessages.mockRejectedValue(deletedError()); + client.getMessages.mockResolvedValue([null, null]); + + dispatch(forwarder, [makeMsg(1), makeMsg(2)]); + + await vi.waitFor(() => + expect(log.warn).toHaveBeenCalledWith(expect.stringContaining('skipping')), + ); + expect(client.forwardMessages).toHaveBeenCalledTimes(1); + }); + + it('dry-run logs the album and sends nothing', () => { + const { client, log, forwarder } = makeForwarder(makeConfig([makeGroup()]), { dryRun: true }); + + dispatch(forwarder, [makeMsg(1), makeMsg(2)]); + + expect(log.info).toHaveBeenCalledWith(expect.stringContaining('[dry-run] would forward')); + expect(log.info).toHaveBeenCalledWith(expect.stringContaining('album 1-2')); + expect(client.forwardMessages).not.toHaveBeenCalled(); + }); }); describe('isInvalidPeerError', () => {