telegram and whatsapp channel integrations, validate bot tokens before saving

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-27 08:53:01 +00:00
parent bc4c20929c
commit b698e3d962
15 changed files with 2571 additions and 40 deletions
@@ -0,0 +1,56 @@
+import TelegramBot from 'node-telegram-bot-api';
+import { getServerIntegration } from 'officerdb';
+import { handleTelegramMessage } from './handler';
+
+let bot: TelegramBot | null = null;
+let cachedUsername: string | null = null;
+
+export async function startTelegramBot(token: string): Promise<void> {
+  if (bot) {
+    await stopTelegramBot();
+  }
+
+  bot = new TelegramBot(token, { polling: true });
+
+  bot.on('message', (msg) => {
+    handleTelegramMessage(msg).catch((err) => {
+      console.error('[telegram] Unhandled error in message handler:', err);
+    });
+  });
+
+  const me = await bot.getMe();
+  cachedUsername = me.username ?? null;
+  console.log(`[telegram] Bot logged in as @${cachedUsername}`);
+}
+
+export async function stopTelegramBot(): Promise<void> {
+  if (bot) {
+    await bot.stopPolling();
+    bot = null;
+    cachedUsername = null;
+    console.log('[telegram] Bot stopped');
+  }
+}
+
+export function isTelegramBotRunning(): boolean {
+  return bot !== null && bot.isPolling();
+}
+
+export function getTelegramBotUsername(): string | null {
+  return cachedUsername;
+}
+
+export function getTelegramBot(): TelegramBot | null {
+  return bot;
+}
+
+export async function startTelegramBotIfConfigured(): Promise<void> {
+  const integration = await getServerIntegration('telegram');
+  if (!integration?.enabled) return;
+
+  const config = integration.config as Record<string, unknown>;
+  const botToken = config.botToken as string | undefined;
+  if (!botToken) return;
+
+  await startTelegramBot(botToken);
+}
@@ -0,0 +1,51 @@
+const MAX_LENGTH = 4096;
+
+export function chunkMessage(text: string): string[] {
+  if (text.length <= MAX_LENGTH) return [text];
+
+  const chunks: string[] = [];
+  const paragraphs = text.split('\n\n');
+
+  let current = '';
+
+  for (const paragraph of paragraphs) {
+    if (paragraph.length > MAX_LENGTH) {
+      // Flush current chunk
+      if (current) {
+        chunks.push(current.trim());
+        current = '';
+      }
+      // Split long paragraph on newlines
+      const lines = paragraph.split('\n');
+      for (const line of lines) {
+        if (line.length > MAX_LENGTH) {
+          // Flush current
+          if (current) {
+            chunks.push(current.trim());
+            current = '';
+          }
+          // Hard-split long line
+          for (let i = 0; i < line.length; i += MAX_LENGTH) {
+            chunks.push(line.slice(i, i + MAX_LENGTH));
+          }
+        } else if (current.length + 1 + line.length > MAX_LENGTH) {
+          chunks.push(current.trim());
+          current = line;
+        } else {
+          current += (current ? '\n' : '') + line;
+        }
+      }
+    } else if (current.length + 2 + paragraph.length > MAX_LENGTH) {
+      chunks.push(current.trim());
+      current = paragraph;
+    } else {
+      current += (current ? '\n\n' : '') + paragraph;
+    }
+  }
+
+  if (current.trim()) {
+    chunks.push(current.trim());
+  }
+
+  return chunks;
+}
@@ -0,0 +1,288 @@
+import type TelegramBot from 'node-telegram-bot-api';
+import { findUserByIntegrationConfig, readConfigValue } from 'officerdb';
+import { sendAndAwait, getSessionModel, setSessionModel } from '../send-and-await';
+import { consumePairingCode } from '../pairing';
+import { chunkMessage } from './chunker';
+import { getTelegramBot } from './bot';
+import { listPiModels } from '@@/api/pi/list-models';
+import { enqueue } from '@@/queue/engine';
+import { readJob } from '@@/queue/storage';
+import { openEmailDb } from '@@/api/email/email-db';
+import type { ModelInfo } from '@@/api/pi/types';
+
+const PAIRING_CODE_PATTERN = /^[A-Z0-9]{6}$/;
+const TYPING_INTERVAL_MS = 5_000;
+
+type SendFn = (text: string) => Promise<unknown>;
+
+type AccessPolicy = { allowedModels: string[] };
+const ACCESS_POLICY_KEY = 'pi-access-policy';
+
+async function getVisibleModels(): Promise<ModelInfo[]> {
+  const allModels = await listPiModels();
+  const policy = await readConfigValue<AccessPolicy>(ACCESS_POLICY_KEY, { allowedModels: [] });
+  const allowed = policy.allowedModels;
+
+  if (allowed.length === 0) return allModels;
+
+  const allowedSet = new Set(allowed);
+  const allowedProviderSet = new Set(allowed.map((key) => key.split(':')[0]));
+
+  return allModels.filter((m) => {
+    const key = `${m.provider}:${m.id}`;
+    const isExplicitlyAllowed = allowedSet.has(key);
+    const isFromNewProvider = !allowedProviderSet.has(m.provider);
+    return isExplicitlyAllowed || isFromNewProvider;
+  });
+}
+
+type CommandContext = {
+  content: string;
+  send: SendFn;
+  userId: number;
+  email: string;
+  telegramId: string;
+};
+
+async function handleEmailSync(ctx: CommandContext): Promise<void> {
+  const { send, email } = ctx;
+
+  let countBefore = 0;
+  try {
+    const db = openEmailDb(email);
+    const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
+    countBefore = row.count;
+    db.close();
+  } catch {
+    // DB might not exist yet
+  }
+
+  await send('Syncing emails...');
+
+  const job = await enqueue({ lane: 'google-api', type: 'gmail-sync', userId: email });
+
+  const poll = async (): Promise<'completed' | 'failed' | 'cancelled'> => {
+    for (let i = 0; i < 120; i++) {
+      await new Promise((r) => setTimeout(r, 3000));
+      const current = await readJob(job.id);
+      if (!current) return 'failed';
+      if (current.status === 'completed' || current.status === 'failed' || current.status === 'cancelled') {
+        return current.status;
+      }
+    }
+    return 'failed';
+  };
+
+  const status = await poll();
+
+  if (status !== 'completed') {
+    await send('Email sync failed. Check the queue dashboard for details.');
+    return;
+  }
+
+  try {
+    const db = openEmailDb(email);
+    const row = db.query('SELECT COUNT(*) as count FROM emails WHERE deleted = 0').get() as { count: number };
+    const countAfter = row.count;
+    const newCount = countAfter - countBefore;
+
+    if (newCount <= 0) {
+      db.close();
+      await send('Sync complete — no new emails.');
+      return;
+    }
+
+    const newest = db
+      .query('SELECT from_name, from_address, subject FROM emails WHERE deleted = 0 ORDER BY date DESC LIMIT ?')
+      .all(Math.min(newCount, 20)) as Array<{ from_name: string | null; from_address: string; subject: string }>;
+    db.close();
+
+    const lines = newest.map((e) => {
+      const sender = e.from_name || e.from_address;
+      return `- *${sender}*: ${e.subject}`;
+    });
+
+    let text = `Sync complete — *${newCount}* new email${newCount !== 1 ? 's' : ''}`;
+    if (newCount > 20) text += ` (showing latest 20)`;
+    text += `:\n\n${lines.join('\n')}`;
+
+    const chunks = chunkMessage(text);
+    for (const chunk of chunks) {
+      await send(chunk);
+    }
+  } catch (err) {
+    await send(`Sync complete but failed to read results: ${err instanceof Error ? err.message : String(err)}`);
+  }
+}
+
+async function handleCommand(ctx: CommandContext): Promise<boolean> {
+  const { content, send, telegramId } = ctx;
+  const lower = content.toLowerCase();
+
+  const helpSections: Record<string, string> = {
+    models:
+      '*Models:*\n' +
+      '`!model` — show current model\n' +
+      '`!model <id>` — switch model\n' +
+      '`!models` — list available models',
+    email: '*Email:*\n' + '`!email sync` — sync Gmail and show new emails',
+  };
+
+  if (lower === '!help' || lower.startsWith('!help ')) {
+    const topic = content.slice('!help'.length).trim().toLowerCase();
+    if (topic && topic in helpSections) {
+      await send(helpSections[topic]!);
+      return true;
+    }
+    if (topic) {
+      await send(
+        `Unknown topic: \`${topic}\`\nAvailable: ${Object.keys(helpSections)
+          .map((k) => `\`${k}\``)
+          .join(', ')}`,
+      );
+      return true;
+    }
+    const full = Object.values(helpSections).join('\n\n');
+    await send(full + '\n\n`!help <topic>` — show commands for a topic');
+    return true;
+  }
+
+  if (lower === '!models') {
+    const models = await getVisibleModels();
+    if (models.length === 0) {
+      await send('No models available.');
+      return true;
+    }
+    const current = getSessionModel('telegram', telegramId);
+    const grouped = new Map<string, string[]>();
+    for (const m of models) {
+      const list = grouped.get(m.provider) ?? [];
+      list.push(m.id === current ? `*${m.id}* (current)` : m.id);
+      grouped.set(m.provider, list);
+    }
+    let text = '*Available models:*\n';
+    for (const [provider, ids] of grouped) {
+      text += `\n_${provider}_\n${ids.map((id) => `  ${id}`).join('\n')}\n`;
+    }
+    text += '\nUse `!model <id>` to switch.';
+    await send(text);
+    return true;
+  }
+
+  if (lower === '!model') {
+    const current = getSessionModel('telegram', telegramId);
+    await send(
+      current
+        ? `Current model: *${current}*`
+        : 'No active session yet — the default model will be used on your next message.',
+    );
+    return true;
+  }
+
+  if (lower.startsWith('!model ')) {
+    const requested = content.slice('!model '.length).trim();
+    if (!requested) {
+      const current = getSessionModel('telegram', telegramId);
+      await send(current ? `Current model: *${current}*` : 'No active session yet.');
+      return true;
+    }
+    const models = await getVisibleModels();
+    const match = models.find((m) => m.id === requested || m.name === requested);
+    if (!match) {
+      await send(`Model not found: \`${requested}\`\nUse \`!models\` to see available models.`);
+      return true;
+    }
+    setSessionModel('telegram', telegramId, match.id);
+    await send(`Model switched to *${match.id}*. The new model will be used on your next message.`);
+    return true;
+  }
+
+  if (lower === '!email sync') {
+    await handleEmailSync(ctx);
+    return true;
+  }
+
+  return false;
+}
+
+export async function handleTelegramMessage(msg: TelegramBot.Message): Promise<void> {
+  const bot = getTelegramBot();
+  if (!bot) return;
+
+  // Ignore non-private chats, bot messages, non-text
+  if (msg.chat.type !== 'private') return;
+  if (msg.from?.is_bot) return;
+  if (!msg.text) return;
+
+  const chatId = msg.chat.id;
+  const telegramId = String(msg.from!.id);
+  const content = msg.text.trim();
+  if (!content) return;
+
+  const send: SendFn = (text: string) => bot.sendMessage(chatId, text);
+
+  // Look up linked Officer user
+  const linked = await findUserByIntegrationConfig('telegram', 'telegramId', telegramId);
+
+  if (!linked) {
+    if (PAIRING_CODE_PATTERN.test(content.toUpperCase())) {
+      const result = await consumePairingCode(content.toUpperCase(), telegramId);
+      if (result) {
+        await send('Account linked! You can now chat with me.');
+        return;
+      }
+      await send('Invalid or expired pairing code. Please generate a new one from Officer Settings.');
+      return;
+    }
+
+    await send(
+      "I don't recognize your Telegram account. To link it:\n" +
+        '1. Go to Officer Settings → Integrations → Telegram\n' +
+        '2. Click "Link Telegram" to get a pairing code\n' +
+        '3. Send the 6-character code to me here',
+    );
+    return;
+  }
+
+  // Handle commands
+  if (content.startsWith('!')) {
+    const handled = await handleCommand({
+      content,
+      send,
+      userId: linked.user.id,
+      email: linked.user.email,
+      telegramId,
+    });
+    if (handled) return;
+  }
+
+  // Typing indicator
+  const sendTyping = () => {
+    bot.sendChatAction(chatId, 'typing').catch(() => {});
+  };
+  const typingInterval = setInterval(sendTyping, TYPING_INTERVAL_MS);
+  sendTyping();
+
+  try {
+    const result = await sendAndAwait({
+      userId: linked.user.id,
+      email: linked.user.email,
+      username: linked.user.username ?? linked.user.email.split('@')[0]!,
+      prompt: content,
+      context: 'telegram',
+      contextId: telegramId,
+    });
+
+    clearInterval(typingInterval);
+
+    const signature = `\`${result.model}\`\n`;
+    const chunks = chunkMessage(result.text);
+    for (let i = 0; i < chunks.length; i++) {
+      await send(i === 0 ? signature + chunks[i]! : chunks[i]!);
+    }
+  } catch (err) {
+    clearInterval(typingInterval);
+    console.error('[telegram] Error handling message:', err);
+    await send('Sorry, something went wrong processing your message.').catch(() => {});
+  }
+}