feat(channels): complete Telegram Managed DM flow and realtime sync (#288)

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<Record<string, boolean>>({});
   const [fieldValues, setFieldValues] = useState<Record<string, Record<string, string>>>({});
   const [error, setError] = useState<string | null>(null);
+  const managedDmPollControllers = useRef<Record<string, AbortController>>({});
 
   const runBusy = useCallback(async (key: string, task: () => Promise<void>) => {
     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,19 +242,20 @@ const TelegramConfig = ({ definition }: TelegramConfigProps) => {
         }
       });
     },
-    [dispatch, fieldValues, runBusy]
+    [dispatch, fieldValues, runBusy, startManagedDmPolling]
   );
 
   const handleDisconnect = useCallback(
     (authMode: ChannelAuthMode) => {
       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 (

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,22 +67,42 @@ 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(<TelegramConfig definition={telegramDef} />);
 
     const connectButtons = screen.getAllByText('Connect');
     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();
   });
 });

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