diff --git a/app/src/components/channels/TelegramConfig.tsx b/app/src/components/channels/TelegramConfig.tsx index 2d0de0861..b8ea20181 100644 --- a/app/src/components/channels/TelegramConfig.tsx +++ b/app/src/components/channels/TelegramConfig.tsx @@ -1,8 +1,9 @@ import debug from 'debug'; -import { useCallback, useState } from 'react'; +import { useCallback, useEffect, useRef, useState } from 'react'; import { AUTH_MODE_LABELS } from '../../lib/channels/definitions'; import { channelConnectionsApi } from '../../services/api/channelConnectionsApi'; +import { managedDmApi } from '../../services/api/managedDmApi'; import { callCoreRpc } from '../../services/coreRpcClient'; import { disconnectChannelConnection, @@ -21,7 +22,8 @@ import ChannelFieldInput from './ChannelFieldInput'; import ChannelStatusBadge from './ChannelStatusBadge'; const log = debug('channels:telegram'); -const MANAGED_DM_FOLLOW_UP_MESSAGE = 'Managed DM setup will be enabled in a follow-up update.'; +const MANAGED_DM_CONNECTING_MESSAGE = 'Open Telegram and message the bot to complete setup.'; +const MANAGED_DM_TIMEOUT_MESSAGE = 'Managed DM verification timed out. Try connecting again.'; interface TelegramConfigProps { definition: ChannelDefinition; @@ -34,6 +36,7 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { const [busyKeys, setBusyKeys] = useState>({}); const [fieldValues, setFieldValues] = useState>>({}); const [error, setError] = useState(null); + const managedDmPollControllers = useRef>({}); const runBusy = useCallback(async (key: string, task: () => Promise) => { setBusyKeys(prev => ({ ...prev, [key]: true })); @@ -55,6 +58,85 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { })); }, []); + const stopManagedDmPolling = useCallback((key: string) => { + managedDmPollControllers.current[key]?.abort(); + delete managedDmPollControllers.current[key]; + }, []); + + useEffect(() => { + return () => { + for (const controller of Object.values(managedDmPollControllers.current)) { + controller.abort(); + } + managedDmPollControllers.current = {}; + }; + }, []); + + const startManagedDmPolling = useCallback( + (key: string, token: string) => { + stopManagedDmPolling(key); + const controller = new AbortController(); + managedDmPollControllers.current[key] = controller; + + void (async () => { + log('polling managed dm status', { key, tokenLength: token.length }); + try { + const status = await managedDmApi.pollManagedDmStatusUntilVerified(token, { + signal: controller.signal, + }); + + if (controller.signal.aborted) { + return; + } + + if (status?.verified) { + log('managed dm verified via polling', { + key, + telegramUsername: status.telegramUsername, + }); + dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: 'managed_dm', + patch: { status: 'connected', lastError: undefined, capabilities: ['dm'] }, + }) + ); + return; + } + + dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: 'managed_dm', + patch: { status: 'error', lastError: MANAGED_DM_TIMEOUT_MESSAGE }, + }) + ); + setError(MANAGED_DM_TIMEOUT_MESSAGE); + } catch (pollError) { + if (controller.signal.aborted) { + return; + } + + const msg = pollError instanceof Error ? pollError.message : String(pollError); + log('managed dm polling failed', { key, error: msg }); + dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: 'managed_dm', + patch: { status: 'error', lastError: msg }, + }) + ); + setError(msg); + } finally { + if (managedDmPollControllers.current[key] === controller) { + delete managedDmPollControllers.current[key]; + } + } + })(); + }, + [dispatch, stopManagedDmPolling] + ); + const handleConnect = useCallback( (spec: AuthModeSpec) => { const key = `telegram:${spec.mode}`; @@ -94,17 +176,35 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { if (result.status === 'pending_auth' && result.auth_action) { if (result.auth_action === 'telegram_managed_dm') { - log('managed dm connect requested before backend flow is enabled'); - dispatch( - upsertChannelConnection({ - channel: 'telegram', - authMode: spec.mode, - patch: { - status: 'disconnected', - lastError: result.message ?? MANAGED_DM_FOLLOW_UP_MESSAGE, - }, - }) - ); + try { + const initiated = await managedDmApi.initiateManagedDm(); + log('managed dm initiate success', { + key, + tokenLength: initiated.token.length, + expiresAt: initiated.expiresAt, + }); + await openUrl(initiated.deepLink); + dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: spec.mode, + patch: { status: 'connecting', lastError: MANAGED_DM_CONNECTING_MESSAGE }, + }) + ); + startManagedDmPolling(key, initiated.token); + } catch (managedDmError) { + const msg = + managedDmError instanceof Error ? managedDmError.message : String(managedDmError); + log('managed dm initiate failed', { key, error: msg }); + dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: spec.mode, + patch: { status: 'error', lastError: msg }, + }) + ); + setError(msg); + } } else if (result.auth_action.includes('oauth')) { dispatch( upsertChannelConnection({ @@ -142,7 +242,7 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { } }); }, - [dispatch, fieldValues, runBusy] + [dispatch, fieldValues, runBusy, startManagedDmPolling] ); const handleDisconnect = useCallback( @@ -150,11 +250,12 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => { const key = `telegram:${authMode}`; void runBusy(key, async () => { log('disconnecting telegram via %s', authMode); + stopManagedDmPolling(`telegram:${authMode}`); await channelConnectionsApi.disconnectChannel('telegram', authMode); dispatch(disconnectChannelConnection({ channel: 'telegram', authMode })); }); }, - [dispatch, runBusy] + [dispatch, runBusy, stopManagedDmPolling] ); return ( diff --git a/app/src/components/channels/__tests__/TelegramConfig.test.tsx b/app/src/components/channels/__tests__/TelegramConfig.test.tsx index a104f06a3..138dbe9b3 100644 --- a/app/src/components/channels/__tests__/TelegramConfig.test.tsx +++ b/app/src/components/channels/__tests__/TelegramConfig.test.tsx @@ -3,7 +3,9 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { FALLBACK_DEFINITIONS } from '../../../lib/channels/definitions'; import { channelConnectionsApi } from '../../../services/api/channelConnectionsApi'; +import { managedDmApi } from '../../../services/api/managedDmApi'; import { renderWithProviders } from '../../../test/test-utils'; +import { openUrl } from '../../../utils/openUrl'; import TelegramConfig from '../TelegramConfig'; const telegramDef = FALLBACK_DEFINITIONS.find(d => d.id === 'telegram')!; @@ -17,6 +19,16 @@ vi.mock('../../../services/api/channelConnectionsApi', () => ({ }, })); +vi.mock('../../../services/api/managedDmApi', () => ({ + managedDmApi: { + initiateManagedDm: vi.fn(), + getManagedDmStatus: vi.fn(), + pollManagedDmStatusUntilVerified: vi.fn(), + }, +})); + +vi.mock('../../../utils/openUrl', () => ({ openUrl: vi.fn() })); + afterEach(() => { vi.clearAllMocks(); }); @@ -55,12 +67,22 @@ describe('TelegramConfig', () => { }); }); - it('surfaces a follow-up message for managed dm without starting a missing rpc flow', async () => { + it('starts managed dm flow, opens the deep link, and marks the channel connected after verification', async () => { vi.mocked(channelConnectionsApi.connectChannel).mockResolvedValue({ status: 'pending_auth', auth_action: 'telegram_managed_dm', restart_required: false, }); + vi.mocked(managedDmApi.initiateManagedDm).mockResolvedValue({ + token: 'managed-dm-token', + deepLink: 'https://t.me/openhuman_bot?start=manageddm_managed-dm-token', + expiresAt: '2026-04-04T12:00:00.000Z', + }); + vi.mocked(managedDmApi.pollManagedDmStatusUntilVerified).mockResolvedValue({ + verified: true, + telegramUsername: 'telegram-user', + expiresAt: '2026-04-04T12:05:00.000Z', + }); renderWithProviders(); @@ -68,9 +90,19 @@ describe('TelegramConfig', () => { fireEvent.click(connectButtons[1]); await waitFor(() => { - expect( - screen.getByText('Managed DM setup will be enabled in a follow-up update.') - ).toBeInTheDocument(); + expect(managedDmApi.initiateManagedDm).toHaveBeenCalledTimes(1); }); + await waitFor(() => { + expect(openUrl).toHaveBeenCalledWith( + 'https://t.me/openhuman_bot?start=manageddm_managed-dm-token' + ); + }); + await waitFor(() => { + expect(managedDmApi.pollManagedDmStatusUntilVerified).toHaveBeenCalledWith( + 'managed-dm-token', + expect.objectContaining({ signal: expect.any(AbortSignal) }) + ); + }); + expect(await screen.findByText('Connected')).toBeInTheDocument(); }); }); diff --git a/app/src/services/api/__tests__/managedDmApi.test.ts b/app/src/services/api/__tests__/managedDmApi.test.ts new file mode 100644 index 000000000..f76295bae --- /dev/null +++ b/app/src/services/api/__tests__/managedDmApi.test.ts @@ -0,0 +1,40 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { managedDmApi } from '../managedDmApi'; + +vi.mock('../../apiClient', () => ({ apiClient: { post: vi.fn(), get: vi.fn() } })); + +const apiClient = vi.mocked((await import('../../apiClient')).apiClient); + +describe('managedDmApi', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('initiates managed dm through the backend api', async () => { + apiClient.post.mockResolvedValueOnce({ + data: { + token: 'dm-token', + deepLink: 'https://t.me/openhuman_bot?start=manageddm_dm-token', + expiresAt: '2026-04-04T12:00:00.000Z', + }, + }); + + await expect(managedDmApi.initiateManagedDm()).resolves.toEqual({ + token: 'dm-token', + deepLink: 'https://t.me/openhuman_bot?start=manageddm_dm-token', + expiresAt: '2026-04-04T12:00:00.000Z', + }); + }); + + it('polls until verified and returns the verified status', async () => { + apiClient.get + .mockResolvedValueOnce({ data: { verified: false, telegramUsername: null } }) + .mockResolvedValueOnce({ data: { verified: true, telegramUsername: 'telegram-user' } }); + + await expect( + managedDmApi.pollManagedDmStatusUntilVerified('dm-token', { intervalMs: 0, timeoutMs: 100 }) + ).resolves.toEqual({ verified: true, telegramUsername: 'telegram-user' }); + expect(apiClient.get).toHaveBeenCalledTimes(2); + }); +}); diff --git a/app/src/services/api/managedDmApi.ts b/app/src/services/api/managedDmApi.ts new file mode 100644 index 000000000..3d6e394bf --- /dev/null +++ b/app/src/services/api/managedDmApi.ts @@ -0,0 +1,100 @@ +import { apiClient } from '../apiClient'; + +export interface ManagedDmInitiateResponse { + token: string; + deepLink: string; + expiresAt: string; +} + +export interface ManagedDmStatusResponse { + verified: boolean; + telegramUsername?: string | null; + expiresAt?: string; +} + +interface ApiEnvelope { + success: boolean; + data: T; +} + +export interface ManagedDmPollOptions { + intervalMs?: number; + timeoutMs?: number; + signal?: AbortSignal; +} + +const DEFAULT_POLL_INTERVAL_MS = 3_000; +const DEFAULT_POLL_TIMEOUT_MS = 5 * 60 * 1_000; + +const sleep = (ms: number, signal?: AbortSignal): Promise => + new Promise(resolve => { + if (signal?.aborted) { + resolve(); + return; + } + + const timeoutId = window.setTimeout(() => { + cleanup(); + resolve(); + }, ms); + + const onAbort = () => { + cleanup(); + resolve(); + }; + + const cleanup = () => { + window.clearTimeout(timeoutId); + signal?.removeEventListener('abort', onAbort); + }; + + signal?.addEventListener('abort', onAbort, { once: true }); + }); + +export async function initiateManagedDm(): Promise { + const response = await apiClient.post>( + '/telegram/managed-dm/initiate' + ); + return response.data; +} + +export async function getManagedDmStatus(token: string): Promise { + const response = await apiClient.get>( + `/telegram/managed-dm/status/${encodeURIComponent(token)}` + ); + return response.data; +} + +export async function pollManagedDmStatusUntilVerified( + token: string, + options: ManagedDmPollOptions = {} +): Promise { + const intervalMs = options.intervalMs ?? DEFAULT_POLL_INTERVAL_MS; + const timeoutMs = options.timeoutMs ?? DEFAULT_POLL_TIMEOUT_MS; + const startedAt = Date.now(); + + while (Date.now() - startedAt < timeoutMs) { + if (options.signal?.aborted) { + return null; + } + + try { + const status = await getManagedDmStatus(token); + if (status.verified) { + return status; + } + } catch { + // Best-effort polling: keep trying until timeout or cancellation. + } + + await sleep(intervalMs, options.signal); + } + + return null; +} + +export const managedDmApi = { + initiateManagedDm, + getManagedDmStatus, + pollManagedDmStatusUntilVerified, +}; diff --git a/app/src/services/socketService.ts b/app/src/services/socketService.ts index a1a4fc8b8..fcaf24d8a 100644 --- a/app/src/services/socketService.ts +++ b/app/src/services/socketService.ts @@ -236,6 +236,25 @@ class SocketService { this.socket.on('channel:connection-updated', handleChannelConnectionUpdated); this.socket.on('channel_connection_updated', handleChannelConnectionUpdated); + this.socket.on('channel:managed-dm-verified', data => { + const obj = data as Record | null; + if (!obj || typeof obj !== 'object') return; + const token = typeof obj.token === 'string' ? obj.token : undefined; + const telegramUsername = + typeof obj.telegramUsername === 'string' ? obj.telegramUsername : undefined; + const chatId = typeof obj.chatId === 'number' ? obj.chatId : undefined; + if (!token) return; + + socketLog('Managed DM verified', { tokenLength: token.length, telegramUsername, chatId }); + store.dispatch( + upsertChannelConnection({ + channel: 'telegram', + authMode: 'managed_dm', + patch: { status: 'connected', lastError: undefined, capabilities: ['dm'] }, + }) + ); + }); + this.socket.connect(); }