diff --git a/src/filter.ts b/src/filter.ts index ef28048..da22850 100644 --- a/src/filter.ts +++ b/src/filter.ts @@ -78,6 +78,19 @@ export function toInputPeer(peer: string): string | number { return Number.isNaN(id) ? peer : id; } +// Swap one stored peer identifier for another across a peer list, preserving +// order and dropping duplicates. Used when a basic group is upgraded to a +// supergroup: its id changes (e.g. -4187363166 → -1004187363166) and the group +// config must follow without introducing a duplicate target. +export function replacePeer(peers: string[], from: string, to: string): string[] { + const out: string[] = []; + for (const peer of peers) { + const next = peer === from ? to : peer; + if (!out.includes(next)) out.push(next); + } + return out; +} + export function matchesPeer( chat: { id: number | string; username?: string | null }, peer: string, diff --git a/src/forwarder.ts b/src/forwarder.ts index 09a4de7..f9ef484 100644 --- a/src/forwarder.ts +++ b/src/forwarder.ts @@ -2,9 +2,16 @@ import type { Message } from '@mtcute/core'; import { Dispatcher } from '@mtcute/dispatcher'; import type { TelegramClient } from '@mtcute/node'; -import { matchesContentType, matchesKeywords, matchesPeer, toInputPeer } from './filter.js'; +import { saveConfig } from './config.js'; +import { + matchesContentType, + matchesKeywords, + matchesPeer, + replacePeer, + toInputPeer, +} from './filter.js'; import { logger as defaultLogger, type Logger } from './logger.js'; -import { type ForwardOutcome, RateLimitedQueue } from './queue.js'; +import { type ForwardOutcome, RateLimitedQueue, SkippedForward } from './queue.js'; import type { AppConfig, ForwardGroup } from './types.js'; interface ForwarderStats { @@ -18,14 +25,24 @@ interface ForwarderOptions { // When true, log every match but never actually forward — for verifying // filters safely without sending anything or risking FloodWait. dryRun?: boolean; + // Persist config after an automatic target migration (basic group → supergroup + // rewrites the stored target id). Injectable so tests don't touch the real + // config file; defaults to saving ~/.telegram-forwarder/config.json. + persistConfig?: (config: AppConfig) => void; } export class Forwarder { private readonly client: TelegramClient; + private readonly config: AppConfig; private readonly groups: ForwardGroup[]; private readonly queue: RateLimitedQueue; private readonly log: Logger; private readonly dryRun: boolean; + private readonly persistConfig: (config: AppConfig) => void; + // Targets retired this session: an invalid-peer error we could not recover via + // migration. Skipped silently on subsequent messages so a permanently-broken + // peer does not spam the log every time the source posts. + private readonly deadTargets = new Set(); private dp: Dispatcher | null = null; private readonly stats: ForwarderStats = { forwarded: 0, skipped: 0, failed: 0 }; @@ -51,10 +68,12 @@ export class Forwarder { }; constructor(client: TelegramClient, config: AppConfig, opts: ForwarderOptions = {}) { - const { logger = defaultLogger, dryRun = false } = opts; + const { logger = defaultLogger, dryRun = false, persistConfig = saveConfig } = opts; this.client = client; + this.config = config; this.groups = config.groups.filter((g) => g.enabled); this.dryRun = dryRun; + this.persistConfig = persistConfig; this.log = logger.withTag('forwarder'); this.queue = new RateLimitedQueue(config.rateLimit, { logger, @@ -176,19 +195,16 @@ export class Forwarder { 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(async () => { - await this.client.forwardMessages({ - messages: [msg], - toChatId: toInputPeer(targetPeer), - noAuthor: group.noAuthor, - }); - }, label); + this.queue.enqueue(() => this.forwardOne(msg, group, targetPeer), label); } } }); @@ -196,6 +212,61 @@ export class Forwarder { this.log.info(`Dispatcher started. Monitoring ${this.groups.length} group(s).`); } + // Forward one message to one target, transparently following a basic-group → + // supergroup upgrade. After a migration the old (chat-range) id is dead and + // Telegram answers with CHANNEL_INVALID/PEER_ID_INVALID. We then probe the + // target with getChat: a `migratedTo` means follow it (rewrite config + retry + // on the new id); a still-healthy chat means the error was not about this + // target (rethrow); an unreachable chat means retire it for this session. + // 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 { + try { + await this.forward(msg, group, targetPeer); + } catch (err) { + if (!isInvalidTargetError(err)) throw err; + + const migration = await this.probeTarget(targetPeer); + if (migration === 'healthy') throw err; + if (migration === null) { + this.deadTargets.add(targetPeer); + throw new SkippedForward( + `Target "${targetPeer}" is unreachable and not a known migration — ` + + `retiring it this session. Run "group edit" to fix "${group.name}".`, + ); + } + + group.targetPeers = replacePeer(group.targetPeers, targetPeer, migration); + this.persistConfig(this.config); + this.log.info( + `Target "${targetPeer}" was upgraded to a supergroup — now forwarding ` + + `"${group.name}" to "${migration}" (config updated).`, + ); + await this.forward(msg, group, migration); + } + } + + private forward(msg: Message, group: ForwardGroup, targetPeer: string): Promise { + return this.client.forwardMessages({ + messages: [msg], + toChatId: toInputPeer(targetPeer), + noAuthor: group.noAuthor, + }); + } + + // Classify a target that just failed with an invalid-peer error: + // string — the new supergroup id it was migrated to (follow it) + // 'healthy' — the target still resolves and was not migrated (error was elsewhere) + // null — the target could not be fetched at all (retire it) + private async probeTarget(targetPeer: string): Promise { + try { + const chat = await this.client.getChat(toInputPeer(targetPeer)); + return chat.migratedToId === null ? 'healthy' : String(chat.migratedToId); + } catch { + return null; + } + } + // Number of forwards still queued (not yet sent). Useful when reporting // what gets dropped on shutdown. get pending(): number { @@ -248,3 +319,16 @@ function isTransientNetworkError(err: Error): boolean { if (code && TRANSIENT_NETWORK_CODES.includes(code)) return true; return TRANSIENT_NETWORK_CODES.some((c) => err.message.includes(c)); } + +// Telegram errors meaning "the destination peer reference is no longer valid". +// For a forward target this is the signature of a basic group upgraded to a +// supergroup (its id changed) — or a chat we can no longer reach. Matched on +// both the RpcError `errorMessage` and a plain message string. +const INVALID_TARGET_ERRORS = ['CHANNEL_INVALID', 'PEER_ID_INVALID', 'CHAT_ID_INVALID']; + +export function isInvalidTargetError(err: unknown): boolean { + if (!(err instanceof Error)) return false; + const msg = + ('errorMessage' in err ? (err as { errorMessage?: string }).errorMessage : null) ?? err.message; + return INVALID_TARGET_ERRORS.some((code) => msg.includes(code)); +} diff --git a/src/queue.ts b/src/queue.ts index 2ba8f3a..3888d1c 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -5,6 +5,12 @@ type Task = () => Promise; export type ForwardOutcome = 'sent' | 'skipped' | 'failed'; +// Thrown by a task to tell the queue the forward was deliberately skipped rather +// than failed: the queue records it as 'skipped' and surfaces the message as a +// warning, not an error. Used when a target is retired mid-session (e.g. an +// unreachable peer that is not a recoverable migration). +export class SkippedForward extends Error {} + interface QueueDeps { logger?: Logger; // Called once per task with its terminal outcome so callers can keep stats. @@ -57,6 +63,12 @@ export class RateLimitedQueue { this.log.success(label); this.onOutcome?.('sent', label); } catch (err: unknown) { + if (err instanceof SkippedForward) { + this.log.warn(err.message); + this.onOutcome?.('skipped', label); + return; + } + if (isDeletedMessage(err)) { this.log.warn(`Message was deleted before forwarding — skipping (${label})`); this.onOutcome?.('skipped', label); diff --git a/test/filter.test.ts b/test/filter.test.ts index cef30c4..dd99fd5 100644 --- a/test/filter.test.ts +++ b/test/filter.test.ts @@ -1,6 +1,12 @@ import { describe, expect, it } from 'vitest'; -import { matchesContentType, matchesKeywords, matchesPeer, toInputPeer } from '../src/filter.js'; +import { + matchesContentType, + matchesKeywords, + matchesPeer, + replacePeer, + toInputPeer, +} from '../src/filter.js'; const makeMsg = (mediaType?: string, flags?: { isAnimation?: boolean; isRound?: boolean }) => ({ media: mediaType @@ -158,3 +164,24 @@ describe('toInputPeer', () => { expect(toInputPeer('12345')).toBe(12345); }); }); + +describe('replacePeer', () => { + it('swaps the matching id and leaves the others untouched', () => { + expect(replacePeer(['-4187363166', '@other'], '-4187363166', '-1004187363166')).toEqual([ + '-1004187363166', + '@other', + ]); + }); + + it('is a no-op when the id to replace is absent', () => { + expect(replacePeer(['@a', '@b'], '-1', '-2')).toEqual(['@a', '@b']); + }); + + it('drops the duplicate when the new id is already in the list', () => { + expect(replacePeer(['-100', '@x'], '@x', '-100')).toEqual(['-100']); + }); + + it('returns an empty list unchanged', () => { + expect(replacePeer([], '-1', '-2')).toEqual([]); + }); +}); diff --git a/test/forwarder.test.ts b/test/forwarder.test.ts index bb9692d..a0acb5a 100644 --- a/test/forwarder.test.ts +++ b/test/forwarder.test.ts @@ -1,7 +1,7 @@ import type { TelegramClient } from '@mtcute/node'; import { describe, expect, it, vi } from 'vitest'; -import { Forwarder } from '../src/forwarder.js'; +import { Forwarder, isInvalidTargetError } from '../src/forwarder.js'; import type { Logger } from '../src/logger.js'; import type { AppConfig, ForwardGroup } from '../src/types.js'; @@ -153,6 +153,30 @@ describe('Forwarder', () => { }); }); +describe('isInvalidTargetError', () => { + it('matches invalid-peer RpcErrors by errorMessage', () => { + const rpc = Object.assign(new Error('Telegram API error 400'), { + errorMessage: 'CHANNEL_INVALID', + }); + expect(isInvalidTargetError(rpc)).toBe(true); + }); + + it('matches the same codes in a plain message string', () => { + expect(isInvalidTargetError(new Error('Telegram API error 400: PEER_ID_INVALID'))).toBe(true); + expect(isInvalidTargetError(new Error('CHAT_ID_INVALID'))).toBe(true); + }); + + it('does not swallow FloodWait or deleted-message errors', () => { + expect(isInvalidTargetError(new Error('FLOOD_WAIT_60'))).toBe(false); + expect(isInvalidTargetError(new Error('MESSAGE_ID_INVALID'))).toBe(false); + }); + + it('ignores non-Error values', () => { + expect(isInvalidTargetError('CHANNEL_INVALID')).toBe(false); + expect(isInvalidTargetError(null)).toBe(false); + }); +}); + describe('Forwarder.checkPeers', () => { it('does not scan dialogs when every peer resolves from cache', async () => { const { client, log, forwarder } = makeForwarder(