feat: auto-follow basic group → supergroup target migration

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,

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<string>();
   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,26 +195,78 @@ 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);
         }
       }
     });
 
     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<void> {
+    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<unknown> {
+    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<string | 'healthy' | null> {
+    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));
+}

src/queue.ts

@@ -5,6 +5,12 @@ type Task = () => Promise<void>;
 
 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);

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([]);
+  });
+});

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(