fix(telegram): handle group-to-supergroup migration seamlessly

When a Telegram group upgrades to a supergroup, the chat ID changes and
all existing references become stale. This caused send failures (400),
orphaned sessions, and required manual re-pairing.

Add dual-path migration handling:
- Proactive: intercept inbound MigrateToChatID before isServiceMessage
- Reactive: detect 400 + MigrateToChatID on send, migrate DB, retry

DB migration updates in a single transaction (scoped by tenant + channel):
- paired_devices: sender_id, chat_id
- sessions: session_key, user_id
- channel_contacts: sender_id
- channel_pending_messages: history_key

Also invalidates in-memory caches (approvedGroups, pairingReplySent,
groupHistory) and handles media sends via migration retry in Send().
This commit is contained in:
viettranx committed 2026-04-05 13:12:33 +07:00
1 parent 1dd5635f2b
commit 473679d28d
7 files changed
+203 -1

No files matched your search

+33
View File
@@ -465,3 +465,36 @@ func resolveThreadIDForSend(threadID int) int {
}
return threadID
}
// migrateGroupChat handles a Telegram group→supergroup migration by updating
// all DB references (paired_devices, sessions, channel_contacts) and invalidating
// in-memory caches. Safe to call multiple times (idempotent).
func (c *Channel) migrateGroupChat(ctx context.Context, oldChatID, newChatID int64) {
oldStr := fmt.Sprintf("%d", oldChatID)
newStr := fmt.Sprintf("%d", newChatID)
slog.Info("telegram: migrating group chat",
"old_chat_id", oldStr, "new_chat_id", newStr, "channel", c.Name())
// Update DB (paired_devices, sessions, channel_contacts).
if c.pairingService != nil {
if err := c.pairingService.MigrateGroupChatID(ctx, c.Name(), oldStr, newStr); err != nil {
slog.Error("telegram: failed to migrate group chat in DB",
"old_chat_id", oldStr, "new_chat_id", newStr, "error", err)
return
}
}
// Invalidate approvedGroups cache.
c.approvedGroups.Delete(oldStr)
c.approvedGroups.Store(newStr, true)
// Clear pairing reply debounce for old group sender.
oldGroupSender := fmt.Sprintf("group:%d", oldChatID)
c.pairingReplySent.Delete(oldGroupSender)
// Clear in-memory pending history for old key (will rebuild from DB on next access).
if c.groupHistory != nil {
c.groupHistory.Clear(oldStr)
}
}
+9
View File
@@ -22,6 +22,15 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
return
}
// Proactive migration detection: group upgraded to supergroup.
// Must run BEFORE isServiceMessage() — migration messages have no text/media.
if message.MigrateToChatID != 0 {
slog.Info("telegram: group migrated to supergroup (inbound)",
"old_chat_id", message.Chat.ID, "new_chat_id", message.MigrateToChatID)
c.migrateGroupChat(ctx, message.Chat.ID, message.MigrateToChatID)
return
}
// Skip service messages (member added/removed, title changed, etc.).
// These have no text/caption and no meaningful media — processing them
// pollutes mention gate and history with "[empty message]" entries.
+32 -1
View File
@@ -28,6 +28,16 @@ var (
htmlTagRe = regexp.MustCompile(`<[^>]*>`)
)
// extractMigrateChatID checks if a Telegram API error contains a group→supergroup
// migration indicator and returns the new chat ID, or 0 if not a migration error.
func extractMigrateChatID(err error) int64 {
var apiErr *telegoapi.Error
if errors.As(err, &apiErr) && apiErr.Parameters != nil && apiErr.Parameters.MigrateToChatID != 0 {
return apiErr.Parameters.MigrateToChatID
}
return 0
}
const (
sendMaxRetries = 3
sendRetryDelay = 2 * time.Second
@@ -229,7 +239,17 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error {
c.placeholders.Delete(localKey)
_ = c.deleteMessage(ctx, chatID, pID.(int))
}
return c.sendMediaMessage(ctx, chatID, msg, replyToMsgID, threadID)
err := c.sendMediaMessage(ctx, chatID, msg, replyToMsgID, threadID)
// Migration fallback: group upgraded to supergroup — retry with new chat ID.
if err != nil {
if newChatID := extractMigrateChatID(err); newChatID != 0 {
slog.Info("telegram: group migrated to supergroup (media send-path)",
"old_chat_id", chatID, "new_chat_id", newChatID)
c.migrateGroupChat(ctx, chatID, newChatID)
return c.sendMediaMessage(ctx, newChatID, msg, replyToMsgID, threadID)
}
}
return err
}
// Text-only message
@@ -407,6 +427,17 @@ func (c *Channel) sendHTMLWithDepth(ctx context.Context, chatID int64, htmlConte
})
if err != nil {
// Case 0: Group migrated to supergroup — update DB and retry with new chat ID.
if newChatID := extractMigrateChatID(err); newChatID != 0 {
slog.Info("telegram: group migrated to supergroup (send-path)",
"old_chat_id", chatID, "new_chat_id", newChatID)
c.migrateGroupChat(ctx, chatID, newChatID)
if depth < maxSplitDepth {
return c.sendHTMLWithDepth(ctx, newChatID, htmlContent, replyTo, threadID, depth+1)
}
return fmt.Errorf("migration retry depth exceeded: %w", err)
}
errStr := err.Error()
// Case 1: Message too long. Split into smaller chunks and send individually.
+3
View File
@@ -139,6 +139,9 @@ func (m *mockPairingStore) IsPaired(_ context.Context, senderID, channel string)
}
func (m *mockPairingStore) ListPending(context.Context) []store.PairingRequestData { return nil }
func (m *mockPairingStore) ListPaired(context.Context) []store.PairedDeviceData { return nil }
func (m *mockPairingStore) MigrateGroupChatID(context.Context, string, string, string) error {
return nil
}
func TestResolveAuth_GatewayToken(t *testing.T) {
setupTestCache(t, nil)
+4
View File
@@ -33,4 +33,8 @@ type PairingStore interface {
IsPaired(ctx context.Context, senderID, channel string) (bool, error)
ListPending(ctx context.Context) []PairingRequestData
ListPaired(ctx context.Context) []PairedDeviceData
// MigrateGroupChatID updates all references from oldChatID to newChatID
// across paired_devices, sessions, and channel_contacts within a transaction.
// Scoped by tenant_id and channel. Idempotent (safe to call multiple times).
MigrateGroupChatID(ctx context.Context, channel, oldChatID, newChatID string) error
}
+61
View File
@@ -234,6 +234,67 @@ func (s *PGPairingStore) ListPaired(ctx context.Context) []store.PairedDeviceDat
return result
}
func (s *PGPairingStore) MigrateGroupChatID(ctx context.Context, channel, oldChatID, newChatID string) error {
tid := tenantIDForInsert(ctx)
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("begin migrate tx: %w", err)
}
defer tx.Rollback()
// 1. paired_devices: update sender_id and chat_id
if _, err := tx.ExecContext(ctx,
`UPDATE paired_devices
SET sender_id = REPLACE(sender_id, $1, $2),
chat_id = REPLACE(chat_id, $1, $2)
WHERE sender_id LIKE '%' || $1 || '%'
AND channel = $3
AND tenant_id = $4`,
oldChatID, newChatID, channel, tid,
); err != nil {
return fmt.Errorf("migrate paired_devices: %w", err)
}
// 2. sessions: update session_key and user_id
if _, err := tx.ExecContext(ctx,
`UPDATE sessions
SET session_key = REPLACE(session_key, ':' || $1, ':' || $2),
user_id = REPLACE(user_id, ':' || $1, ':' || $2)
WHERE session_key LIKE '%:telegram:%:' || $1 || '%'
AND tenant_id = $3`,
oldChatID, newChatID, tid,
); err != nil {
return fmt.Errorf("migrate sessions: %w", err)
}
// 3. channel_contacts: update sender_id
if _, err := tx.ExecContext(ctx,
`UPDATE channel_contacts
SET sender_id = REPLACE(sender_id, $1, $2)
WHERE sender_id LIKE '%' || $1 || '%'
AND channel_type = 'telegram'
AND tenant_id = $3`,
oldChatID, newChatID, tid,
); err != nil {
return fmt.Errorf("migrate channel_contacts: %w", err)
}
// 4. channel_pending_messages: update history_key
if _, err := tx.ExecContext(ctx,
`UPDATE channel_pending_messages
SET history_key = REPLACE(history_key, $1, $2)
WHERE history_key LIKE '%' || $1 || '%'
AND channel_name = $3
AND tenant_id = $4`,
oldChatID, newChatID, channel, tid,
); err != nil {
return fmt.Errorf("migrate channel_pending_messages: %w", err)
}
return tx.Commit()
}
func generatePairingCode() string {
b := make([]byte, codeLength)
rand.Read(b)
+61
View File
@@ -235,6 +235,67 @@ func (s *SQLitePairingStore) ListPaired(ctx context.Context) []store.PairedDevic
return result
}
func (s *SQLitePairingStore) MigrateGroupChatID(ctx context.Context, channel, oldChatID, newChatID string) error {
tid := tenantIDForInsert(ctx)
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return fmt.Errorf("begin migrate tx: %w", err)
}
defer tx.Rollback()
// 1. paired_devices: update sender_id and chat_id
if _, err := tx.ExecContext(ctx,
`UPDATE paired_devices
SET sender_id = REPLACE(sender_id, ?, ?),
chat_id = REPLACE(chat_id, ?, ?)
WHERE sender_id LIKE '%' || ? || '%'
AND channel = ?
AND tenant_id = ?`,
oldChatID, newChatID, oldChatID, newChatID, oldChatID, channel, tid,
); err != nil {
return fmt.Errorf("migrate paired_devices: %w", err)
}
// 2. sessions: update session_key and user_id
if _, err := tx.ExecContext(ctx,
`UPDATE sessions
SET session_key = REPLACE(session_key, ':' || ?, ':' || ?),
user_id = REPLACE(user_id, ':' || ?, ':' || ?)
WHERE session_key LIKE '%:telegram:%:' || ? || '%'
AND tenant_id = ?`,
oldChatID, newChatID, oldChatID, newChatID, oldChatID, tid,
); err != nil {
return fmt.Errorf("migrate sessions: %w", err)
}
// 3. channel_contacts: update sender_id
if _, err := tx.ExecContext(ctx,
`UPDATE channel_contacts
SET sender_id = REPLACE(sender_id, ?, ?)
WHERE sender_id LIKE '%' || ? || '%'
AND channel_type = 'telegram'
AND tenant_id = ?`,
oldChatID, newChatID, oldChatID, tid,
); err != nil {
return fmt.Errorf("migrate channel_contacts: %w", err)
}
// 4. channel_pending_messages: update history_key
if _, err := tx.ExecContext(ctx,
`UPDATE channel_pending_messages
SET history_key = REPLACE(history_key, ?, ?)
WHERE history_key LIKE '%' || ? || '%'
AND channel_name = ?
AND tenant_id = ?`,
oldChatID, newChatID, oldChatID, channel, tid,
); err != nil {
return fmt.Errorf("migrate channel_pending_messages: %w", err)
}
return tx.Commit()
}
// parseTimeToMillis parses a Go time.Time string (possibly with monotonic clock suffix)
// from SQLite and returns Unix milliseconds. Falls back to 0 on parse failure.
func parseTimeToMillis(s string) int64 {