diff --git a/src/lib/bridge/adapters/telegram-adapter.ts b/src/lib/bridge/adapters/telegram-adapter.ts index 2499029..936ba11 100644 --- a/src/lib/bridge/adapters/telegram-adapter.ts +++ b/src/lib/bridge/adapters/telegram-adapter.ts @@ -12,8 +12,17 @@ import type { OutboundMessage, SendResult, } from '../types'; +import type { FileAttachment } from '@/types'; import { BaseChannelAdapter, registerAdapterFactory } from '../channel-adapter'; import { callTelegramApi, escapeHtml } from './telegram-utils'; +import { + isImageEnabled, + downloadPhoto, + downloadDocumentImage, + isSupportedImageMime, + inferMimeType, +} from './telegram-media'; +import type { TelegramPhotoSize, TelegramDocument } from './telegram-media'; import { getChannelOffset, setChannelOffset, insertAuditLog } from '../../db'; import { getSetting } from '../../db'; @@ -34,6 +43,10 @@ interface TelegramUpdate { chat: { id: number; first_name?: string; title?: string; username?: string }; from?: { id: number; first_name: string; username?: string }; text?: string; + caption?: string; + photo?: TelegramPhotoSize[]; + document?: TelegramDocument; + media_group_id?: string; date: number; }; callback_query?: { @@ -52,6 +65,19 @@ interface QueueEntry { resolve: () => void; } +/** Media group debounce buffer entry for album messages. */ +interface MediaGroupBufferEntry { + updates: TelegramUpdate[]; + updateIds: number[]; + timer: ReturnType; + chatId: string; + userId: string; + displayName: string; +} + +/** Debounce window for media group messages (ms). */ +const MEDIA_GROUP_DEBOUNCE_MS = 500; + export class TelegramAdapter extends BaseChannelAdapter { readonly channelType: ChannelType = 'telegram'; @@ -60,6 +86,7 @@ export class TelegramAdapter extends BaseChannelAdapter { private queue: InboundMessage[] = []; private waiters: Array<(msg: InboundMessage | null) => void> = []; private typingIntervals = new Map>(); + private mediaGroupBuffers = new Map(); /** Committed offset — the highest update_id that has been safely enqueued or skipped. */ private committedOffset = 0; @@ -120,6 +147,12 @@ export class TelegramAdapter extends BaseChannelAdapter { } this.typingIntervals.clear(); + // Clear media group debounce timers + for (const [, entry] of this.mediaGroupBuffers) { + clearTimeout(entry.timer); + } + this.mediaGroupBuffers.clear(); + console.log('[telegram-adapter] Stopped'); } @@ -461,10 +494,11 @@ export class TelegramAdapter extends BaseChannelAdapter { // Answer callback to dismiss the loading state this.answerCallback(cb.id).catch(() => {}); - } else if (update.message?.text) { + } else if (update.message) { const m = update.message; const chatId = String(m.chat.id); const userId = m.from ? String(m.from.id) : chatId; + const displayName = m.from?.username || m.from?.first_name || chatId; if (!this.isAuthorized(userId, chatId)) { console.warn('[telegram-adapter] Unauthorized message from userId:', userId, 'chatId:', chatId); @@ -472,32 +506,51 @@ export class TelegramAdapter extends BaseChannelAdapter { continue; } - const msg: InboundMessage = { - messageId: String(m.message_id), - address: { - channelType: 'telegram', - chatId, - userId, - displayName: m.from?.username || m.from?.first_name || chatId, - }, - text: m.text!, - timestamp: m.date * 1000, - raw: update, - updateId: update.update_id, - }; + const hasPhoto = m.photo && m.photo.length > 0; + const hasDocImage = m.document && this.isDocumentImage(m.document); + const hasMedia = hasPhoto || hasDocImage; - // Audit log - try { - insertAuditLog({ - channelType: 'telegram', - chatId, - direction: 'inbound', + if (hasMedia && isImageEnabled()) { + if (m.media_group_id) { + // Album message — buffer for debounce, advance fetchOffset immediately + this.bufferMediaGroup(m.media_group_id, update, chatId, userId, displayName); + // Don't markUpdateProcessed yet — offset will be committed on flush + } else { + // Single image message — process immediately + await this.processSingleImageMessage(update, chatId, userId, displayName); + } + } else if (m.text) { + // Pure text message — original logic + const msg: InboundMessage = { messageId: String(m.message_id), - summary: m.text!.slice(0, 200), - }); - } catch { /* best effort */ } + address: { + channelType: 'telegram', + chatId, + userId, + displayName, + }, + text: m.text, + timestamp: m.date * 1000, + raw: update, + updateId: update.update_id, + }; - this.enqueue(msg); + // Audit log + try { + insertAuditLog({ + channelType: 'telegram', + chatId, + direction: 'inbound', + messageId: String(m.message_id), + summary: m.text.slice(0, 200), + }); + } catch { /* best effort */ } + + this.enqueue(msg); + } else { + // Unhandled message type (sticker, voice, etc.) — skip + this.markUpdateProcessed(update.update_id); + } } else { // Unhandled update type — still safe to advance past it this.markUpdateProcessed(update.update_id); @@ -517,6 +570,223 @@ export class TelegramAdapter extends BaseChannelAdapter { } } } + + /** + * Check if a Telegram document is a supported image type. + */ + private isDocumentImage(doc: TelegramDocument): boolean { + if (doc.mime_type && isSupportedImageMime(doc.mime_type)) return true; + if (doc.file_name) { + const mime = inferMimeType(doc.file_name); + if (mime && isSupportedImageMime(mime)) return true; + } + return false; + } + + /** + * Process a single image message (no media_group_id). + * Downloads the image and enqueues a message with attachments. + */ + private async processSingleImageMessage( + update: TelegramUpdate, + chatId: string, + userId: string, + displayName: string, + ): Promise { + const m = update.message!; + const token = this.botToken; + if (!token) { + this.markUpdateProcessed(update.update_id); + return; + } + + const attachments: FileAttachment[] = []; + let rejectedReason: string | undefined; + + if (m.photo && m.photo.length > 0) { + const att = await downloadPhoto(token, m.photo, String(m.message_id)); + if (att) attachments.push(att); + } else if (m.document) { + const result = await downloadDocumentImage(token, m.document, String(m.message_id)); + if (result.attachment) { + attachments.push(result.attachment); + } else if (result.rejected) { + rejectedReason = result.rejected; + } + } + + const text = m.caption || m.text || ''; + const hasContent = attachments.length > 0 || text.trim(); + + if (!hasContent) { + // Nothing usable — if rejected, we'll still mark processed + if (rejectedReason && rejectedReason !== 'unsupported_type') { + // Enqueue a rejection notification as text + const msg: InboundMessage = { + messageId: String(m.message_id), + address: { channelType: 'telegram', chatId, userId, displayName }, + text: `[Image rejected] ${rejectedReason}`, + timestamp: m.date * 1000, + raw: update, + updateId: update.update_id, + }; + this.enqueue(msg); + } else { + this.markUpdateProcessed(update.update_id); + } + return; + } + + const summary = attachments.length > 0 + ? `[${attachments.length} image(s)] ${text.slice(0, 150)}` + : text.slice(0, 200); + + // Audit log + try { + insertAuditLog({ + channelType: 'telegram', + chatId, + direction: 'inbound', + messageId: String(m.message_id), + summary, + }); + } catch { /* best effort */ } + + const msg: InboundMessage = { + messageId: String(m.message_id), + address: { channelType: 'telegram', chatId, userId, displayName }, + text: text, + timestamp: m.date * 1000, + raw: update, + updateId: update.update_id, + attachments: attachments.length > 0 ? attachments : undefined, + }; + + this.enqueue(msg); + } + + /** + * Buffer a media group update for debounced processing. + * Resets the 500ms timer on each new update in the same group. + */ + private bufferMediaGroup( + mediaGroupId: string, + update: TelegramUpdate, + chatId: string, + userId: string, + displayName: string, + ): void { + const existing = this.mediaGroupBuffers.get(mediaGroupId); + + if (existing) { + // Add to existing buffer, reset timer + clearTimeout(existing.timer); + existing.updates.push(update); + existing.updateIds.push(update.update_id); + existing.timer = setTimeout(() => this.flushMediaGroup(mediaGroupId), MEDIA_GROUP_DEBOUNCE_MS); + } else { + // New buffer + const timer = setTimeout(() => this.flushMediaGroup(mediaGroupId), MEDIA_GROUP_DEBOUNCE_MS); + this.mediaGroupBuffers.set(mediaGroupId, { + updates: [update], + updateIds: [update.update_id], + timer, + chatId, + userId, + displayName, + }); + } + } + + /** + * Flush a media group buffer — download all images and enqueue a single message. + */ + private async flushMediaGroup(mediaGroupId: string): Promise { + const entry = this.mediaGroupBuffers.get(mediaGroupId); + if (!entry) return; + this.mediaGroupBuffers.delete(mediaGroupId); + + const token = this.botToken; + if (!token) { + // Can't download — mark all as processed + for (const uid of entry.updateIds) { + this.markUpdateProcessed(uid); + } + this.persistCommittedOffset(); + return; + } + + const attachments: FileAttachment[] = []; + let caption = ''; + let firstMessageId = ''; + let firstDate = 0; + + // Download all images in the group + for (const update of entry.updates) { + const m = update.message!; + if (!firstMessageId) { + firstMessageId = String(m.message_id); + firstDate = m.date; + } + // Use caption from whichever update has it (Telegram only sends caption on one) + if (m.caption && !caption) { + caption = m.caption; + } + + if (m.photo && m.photo.length > 0) { + const att = await downloadPhoto(token, m.photo, String(m.message_id)); + if (att) attachments.push(att); + } else if (m.document && this.isDocumentImage(m.document)) { + const result = await downloadDocumentImage(token, m.document, String(m.message_id)); + if (result.attachment) attachments.push(result.attachment); + } + } + + const text = caption; + const hasContent = attachments.length > 0 || text.trim(); + + if (!hasContent) { + // All downloads failed and no caption — mark all processed + for (const uid of entry.updateIds) { + this.markUpdateProcessed(uid); + } + this.persistCommittedOffset(); + return; + } + + const summary = attachments.length > 0 + ? `[Album: ${attachments.length} image(s)] ${text.slice(0, 150)}` + : text.slice(0, 200); + + try { + insertAuditLog({ + channelType: 'telegram', + chatId: entry.chatId, + direction: 'inbound', + messageId: firstMessageId, + summary, + }); + } catch { /* best effort */ } + + // Use the max updateId so acknowledgeUpdate advances offset past all buffered updates + const maxUpdateId = Math.max(...entry.updateIds); + + const msg: InboundMessage = { + messageId: firstMessageId, + address: { + channelType: 'telegram', + chatId: entry.chatId, + userId: entry.userId, + displayName: entry.displayName, + }, + text, + timestamp: firstDate * 1000, + updateId: maxUpdateId, + attachments: attachments.length > 0 ? attachments : undefined, + }; + + this.enqueue(msg); + } } // Self-register so bridge-manager can create TelegramAdapter via the registry. diff --git a/src/lib/bridge/adapters/telegram-media.ts b/src/lib/bridge/adapters/telegram-media.ts new file mode 100644 index 0000000..97a03d3 --- /dev/null +++ b/src/lib/bridge/adapters/telegram-media.ts @@ -0,0 +1,285 @@ +/** + * Telegram Media — download and process images from Telegram messages. + * + * Handles photo[] size selection, file download via Bot API, base64 conversion, + * and document-type image validation. Produces FileAttachment objects that plug + * directly into the existing streamClaude vision pipeline. + */ + +import type { FileAttachment } from '@/types'; +import { getSetting } from '../../db'; + +const TELEGRAM_API = 'https://api.telegram.org'; + +/** Claude vision optimal long-edge size (px). */ +const OPTIMAL_LONG_EDGE = 1568; + +/** Default max image size in bytes (20 MB). */ +const DEFAULT_MAX_IMAGE_SIZE = 20 * 1024 * 1024; + +/** Max retry attempts for download. */ +const MAX_RETRIES = 3; + +/** Supported image MIME types for Claude vision. */ +const SUPPORTED_IMAGE_MIMES = new Set([ + 'image/jpeg', + 'image/png', + 'image/gif', + 'image/webp', +]); + +// ── Telegram Photo Types ───────────────────────────────────── + +export interface TelegramPhotoSize { + file_id: string; + file_unique_id: string; + width: number; + height: number; + file_size?: number; +} + +export interface TelegramDocument { + file_id: string; + file_unique_id: string; + file_name?: string; + mime_type?: string; + file_size?: number; +} + +// ── Public API ─────────────────────────────────────────────── + +/** + * Check whether the Telegram image feature is enabled. + */ +export function isImageEnabled(): boolean { + const setting = getSetting('bridge_telegram_image_enabled'); + // Default to true if not explicitly set to 'false' + return setting !== 'false'; +} + +/** + * Get the configured max image size in bytes. + */ +function getMaxImageSize(): number { + const setting = getSetting('bridge_telegram_max_image_size'); + if (setting) { + const parsed = parseInt(setting, 10); + if (!isNaN(parsed) && parsed > 0) return parsed; + } + return DEFAULT_MAX_IMAGE_SIZE; +} + +/** + * Check if a MIME type is a supported image format. + */ +export function isSupportedImageMime(mime: string): boolean { + return SUPPORTED_IMAGE_MIMES.has(mime.toLowerCase()); +} + +/** + * Infer MIME type from a file path/name extension. + * Returns undefined if the extension is not a recognized image type. + */ +export function inferMimeType(filePath: string): string | undefined { + const ext = filePath.split('.').pop()?.toLowerCase(); + switch (ext) { + case 'jpg': + case 'jpeg': + return 'image/jpeg'; + case 'png': + return 'image/png'; + case 'gif': + return 'image/gif'; + case 'webp': + return 'image/webp'; + default: + return undefined; + } +} + +/** + * Select the optimal photo size from Telegram's photo[] array. + * + * Strategy: sort by long edge ascending, pick the smallest version whose + * long edge >= OPTIMAL_LONG_EDGE (1568px, Claude vision optimal). If none + * are large enough, take the largest available. + */ +export function selectOptimalPhoto(photos: TelegramPhotoSize[]): TelegramPhotoSize { + if (photos.length === 1) return photos[0]; + + // Sort by long edge ascending + const sorted = [...photos].sort((a, b) => { + const aLong = Math.max(a.width, a.height); + const bLong = Math.max(b.width, b.height); + return aLong - bLong; + }); + + // Find smallest version with long edge >= optimal + for (const photo of sorted) { + const longEdge = Math.max(photo.width, photo.height); + if (longEdge >= OPTIMAL_LONG_EDGE) { + return photo; + } + } + + // None large enough — take the largest + return sorted[sorted.length - 1]; +} + +/** + * Download a photo from Telegram's photo[] array and return a FileAttachment. + * + * Selects the optimal size, calls getFile API, downloads the binary, + * and converts to base64. + * + * Returns null if download fails after retries. + */ +export async function downloadPhoto( + botToken: string, + photos: TelegramPhotoSize[], + messageId: string, +): Promise { + const selected = selectOptimalPhoto(photos); + return downloadFileById(botToken, selected.file_id, messageId, selected.file_size); +} + +/** + * Download a document-type image from Telegram. + * + * Pre-checks file_size against the max limit before initiating download. + * Returns null with a reason string if the document is too large or not a supported image. + */ +export async function downloadDocumentImage( + botToken: string, + doc: TelegramDocument, + messageId: string, +): Promise<{ attachment: FileAttachment | null; rejected?: string }> { + // Check MIME type + const mime = doc.mime_type || inferMimeType(doc.file_name || ''); + if (!mime || !isSupportedImageMime(mime)) { + return { attachment: null, rejected: 'unsupported_type' }; + } + + // Pre-check file size before downloading + const maxSize = getMaxImageSize(); + if (doc.file_size && doc.file_size > maxSize) { + const sizeMB = (doc.file_size / (1024 * 1024)).toFixed(1); + const limitMB = (maxSize / (1024 * 1024)).toFixed(0); + return { + attachment: null, + rejected: `Image file too large (${sizeMB} MB). Max allowed: ${limitMB} MB. Please send as a photo instead of a file.`, + }; + } + + const attachment = await downloadFileById(botToken, doc.file_id, messageId, doc.file_size); + return { attachment }; +} + +// ── Internal ───────────────────────────────────────────────── + +/** + * Download a file by its Telegram file_id. + * Calls getFile → download URL → binary → base64 FileAttachment. + * Retries up to MAX_RETRIES with exponential backoff. + */ +async function downloadFileById( + botToken: string, + fileId: string, + messageId: string, + expectedSize?: number, +): Promise { + const maxSize = getMaxImageSize(); + + for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) { + try { + // Step 1: Get file path from Telegram + const getFileUrl = `${TELEGRAM_API}/bot${botToken}/getFile`; + const getFileRes = await fetch(getFileUrl, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ file_id: fileId }), + signal: AbortSignal.timeout(15_000), + }); + + const getFileData = await getFileRes.json(); + if (!getFileData.ok || !getFileData.result?.file_path) { + console.warn(`[telegram-media] getFile failed for ${fileId}:`, getFileData.description); + if (attempt < MAX_RETRIES) { + await sleep(1000 * Math.pow(2, attempt - 1)); + continue; + } + return null; + } + + const filePath: string = getFileData.result.file_path; + const fileSize: number | undefined = getFileData.result.file_size; + + // Pre-check size from API response + if (fileSize && fileSize > maxSize) { + console.warn(`[telegram-media] File too large: ${fileSize} bytes (max ${maxSize})`); + return null; + } + + // Step 2: Download the file + const downloadUrl = `${TELEGRAM_API}/file/bot${botToken}/${filePath}`; + const downloadRes = await fetch(downloadUrl, { + signal: AbortSignal.timeout(60_000), + }); + + if (!downloadRes.ok) { + console.warn(`[telegram-media] Download failed: HTTP ${downloadRes.status}`); + if (attempt < MAX_RETRIES) { + await sleep(1000 * Math.pow(2, attempt - 1)); + continue; + } + return null; + } + + // Check Content-Length header + const contentLength = downloadRes.headers.get('content-length'); + if (contentLength && parseInt(contentLength, 10) > maxSize) { + console.warn(`[telegram-media] Content-Length exceeds max: ${contentLength}`); + return null; + } + + // Step 3: Read buffer and validate actual size + const buffer = Buffer.from(await downloadRes.arrayBuffer()); + if (buffer.length > maxSize) { + console.warn(`[telegram-media] Downloaded buffer too large: ${buffer.length} bytes`); + return null; + } + + // Step 4: Determine MIME type + const mime = inferMimeType(filePath) || 'image/jpeg'; + + // Step 5: Convert to base64 and build FileAttachment + const base64 = buffer.toString('base64'); + const fileName = filePath.split('/').pop() || `image_${messageId}`; + + const attachment: FileAttachment = { + id: `tg-${messageId}-${fileId.slice(0, 8)}`, + name: fileName, + type: mime, + size: buffer.length, + data: base64, + }; + + return attachment; + } catch (err) { + const errMsg = err instanceof Error ? err.message : String(err); + console.warn(`[telegram-media] Download attempt ${attempt}/${MAX_RETRIES} failed:`, errMsg); + + if (attempt < MAX_RETRIES) { + await sleep(1000 * Math.pow(2, attempt - 1)); + continue; + } + return null; + } + } + + return null; +} + +function sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); +} diff --git a/src/lib/bridge/bridge-manager.ts b/src/lib/bridge/bridge-manager.ts index 0ff198d..df2f4e6 100644 --- a/src/lib/bridge/bridge-manager.ts +++ b/src/lib/bridge/bridge-manager.ts @@ -328,7 +328,8 @@ async function handleMessage( } const rawText = msg.text.trim(); - if (!rawText) { ack(); return; } + const hasAttachments = msg.attachments && msg.attachments.length > 0; + if (!rawText && !hasAttachments) { ack(); return; } // Check for IM commands (before sanitization — commands are validated individually) if (rawText.startsWith('/')) { @@ -350,7 +351,7 @@ async function handleMessage( }); } - if (!text) { ack(); return; } + if (!text && !hasAttachments) { ack(); return; } // Regular message — route to conversation engine const binding = router.resolve(msg.address); @@ -366,7 +367,10 @@ async function handleMessage( try { // Pass permission callback so requests are forwarded to IM immediately // during streaming (the stream blocks until permission is resolved). - const result = await engine.processMessage(binding, text, async (perm) => { + // Use text or empty string for image-only messages (prompt is still required by streamClaude) + const promptText = text || (hasAttachments ? 'Describe this image.' : ''); + + const result = await engine.processMessage(binding, promptText, async (perm) => { await broker.forwardPermissionRequest( adapter, msg.address, @@ -376,7 +380,7 @@ async function handleMessage( binding.codepilotSessionId, perm.suggestions, ); - }, taskAbort.signal); + }, taskAbort.signal, hasAttachments ? msg.attachments : undefined); // Send response text if (result.responseText) { diff --git a/src/lib/bridge/conversation-engine.ts b/src/lib/bridge/conversation-engine.ts index 705e46d..796abb3 100644 --- a/src/lib/bridge/conversation-engine.ts +++ b/src/lib/bridge/conversation-engine.ts @@ -7,7 +7,7 @@ */ import type { ChannelBinding } from './types'; -import type { SSEEvent, TokenUsage, MessageContentBlock } from '@/types'; +import type { SSEEvent, TokenUsage, MessageContentBlock, FileAttachment } from '@/types'; import { streamClaude } from '../claude-client'; import { addMessage, @@ -61,6 +61,7 @@ export async function processMessage( text: string, onPermissionRequest?: OnPermissionRequest, abortSignal?: AbortSignal, + files?: FileAttachment[], ): Promise { const sessionId = binding.codepilotSessionId; @@ -86,8 +87,11 @@ export async function processMessage( }, 60_000); try { - // Save user message - addMessage(sessionId, 'user', text); + // Save user message (with image indicator if attachments present) + const userMsgText = files && files.length > 0 + ? `[${files.length} image(s) attached] ${text}` + : text; + addMessage(sessionId, 'user', userMsgText); // Resolve provider const session = getSession(sessionId); @@ -139,6 +143,7 @@ export async function processMessage( permissionMode, provider: resolvedProvider, conversationHistory: historyMsgs, + files, onRuntimeStatusChange: (status: string) => { try { setSessionRuntimeStatus(sessionId, status); } catch { /* best effort */ } }, diff --git a/src/lib/bridge/types.ts b/src/lib/bridge/types.ts index ceb2a43..b321e94 100644 --- a/src/lib/bridge/types.ts +++ b/src/lib/bridge/types.ts @@ -49,6 +49,8 @@ export interface InboundMessage { raw?: unknown; /** Adapter-specific update ID for deferred offset acknowledgement */ updateId?: number; + /** File attachments (images, documents) from the IM channel */ + attachments?: import('@/types').FileAttachment[]; } /** Outbound message to send to an IM channel */