diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index 24cd98f9..364c8be3 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -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) + } +} diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 76c6fa82..91cc70b0 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -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. diff --git a/internal/channels/telegram/send.go b/internal/channels/telegram/send.go index 88e53a38..1d621302 100644 --- a/internal/channels/telegram/send.go +++ b/internal/channels/telegram/send.go @@ -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. diff --git a/internal/http/auth_test.go b/internal/http/auth_test.go index 34b66cc8..c5c07552 100644 --- a/internal/http/auth_test.go +++ b/internal/http/auth_test.go @@ -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) diff --git a/internal/store/pairing_store.go b/internal/store/pairing_store.go index 2427b4ad..c10a81ce 100644 --- a/internal/store/pairing_store.go +++ b/internal/store/pairing_store.go @@ -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 } diff --git a/internal/store/pg/pairing.go b/internal/store/pg/pairing.go index 1aeca1c3..b14fa31d 100644 --- a/internal/store/pg/pairing.go +++ b/internal/store/pg/pairing.go @@ -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) diff --git a/internal/store/sqlitestore/pairing.go b/internal/store/sqlitestore/pairing.go index 0019f51c..8be7b592 100644 --- a/internal/store/sqlitestore/pairing.go +++ b/internal/store/sqlitestore/pairing.go @@ -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 {