okooooooo/CodePilot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: Telegram 图片接收 — offset 安全、caption 兜底、拒绝通知

Browse Source 
P1a: caption-only 消息丢失
- poll loop 文本提取统一为 m.text ?? m.caption ?? ''
- 修复 image_enabled=false 时带 caption 的图片消息被跳过的回归

P1b: offset 越序风险
- markUpdateProcessed 改为连续水位推进（contiguous walk），
  不再直接跳到 updateId+1，避免越过尚未 flush 的相册 ID
- flushMediaGroup 成功入队后预注册所有 buffered ID 到 recentUpdateIds，
  保证 ack 时水位能连续推过整个相册

P2: 图片拒绝通知
- telegram-media: 统一返回 MediaDownloadResult { attachment, rejected, rejectedMessage }
  reject code 覆盖 too_large | unsupported_type | download_failed
  所有超限检查点（file_size/content-length/buffer）均返回可读文案
- telegram-adapter: processSingleImageMessage/flushMediaGroup 收集拒绝结果后
  直接通过 this.send() 发送通知到 Telegram，禁止静默丢弃

P3: 清理 lint warning
- 移除未使用的 escapeHtml import、QueueEntry interface
- 移除 downloadFileById 未使用的 expectedSize 参数

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
7418andClaude Opus 4.6 committed 2026-03-02 12:53:55 +08:00
1 parent 570d26de3a
commit 887870c343
2 files changed
+122 -79
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/lib/bridge/adapters/telegram-adapter.ts
+79 -50
View File
@@ -14,7 +14,7 @@ import type {
} from '../types';
import type { FileAttachment } from '@/types';
import { BaseChannelAdapter, registerAdapterFactory } from '../channel-adapter';
import { callTelegramApi, escapeHtml } from './telegram-utils';
import { callTelegramApi } from './telegram-utils';
import {
isImageEnabled,
downloadPhoto,
@@ -22,7 +22,7 @@ import {
isSupportedImageMime,
inferMimeType,
} from './telegram-media';
import type { TelegramPhotoSize, TelegramDocument } from './telegram-media';
import type { TelegramPhotoSize, TelegramDocument, MediaDownloadResult } from './telegram-media';
import { getChannelOffset, setChannelOffset, insertAuditLog } from '../../db';
import { getSetting } from '../../db';
@@ -57,14 +57,6 @@ interface TelegramUpdate {
};
}
/**
* Async queue entry — resolved by the poll loop, consumed by consumeOne().
*/
interface QueueEntry {
message: InboundMessage;
resolve: () => void;
}
/** Media group debounce buffer entry for album messages. */
interface MediaGroupBufferEntry {
updates: TelegramUpdate[];
@@ -384,15 +376,22 @@ export class TelegramAdapter extends BaseChannelAdapter {
/**
* Mark an update as safely processed (enqueued or intentionally skipped).
* Advances committedOffset and prunes the dedup set when it exceeds capacity.
*
* Uses contiguous watermark advancement: committedOffset only advances when
* there are no gaps (e.g., media-group updates still buffered) below it.
* This prevents offset from jumping past un-flushed album messages.
*/
private markUpdateProcessed(updateId: number): void {
if (updateId >= this.committedOffset) {
this.committedOffset = updateId + 1;
}
this.recentUpdateIds.add(updateId);
// Walk committedOffset forward contiguously — only advance while
// the current position has been confirmed as processed.
while (this.recentUpdateIds.has(this.committedOffset)) {
this.committedOffset++;
}
// Prune dedup set when it exceeds capacity
if (this.recentUpdateIds.size > DEDUP_SET_MAX) {
// Remove oldest entries (Set iterates in insertion order)
const excess = this.recentUpdateIds.size - DEDUP_SET_MAX;
let removed = 0;
for (const id of this.recentUpdateIds) {
@@ -510,6 +509,9 @@ export class TelegramAdapter extends BaseChannelAdapter {
const hasDocImage = m.document && this.isDocumentImage(m.document);
const hasMedia = hasPhoto || hasDocImage;
// Unified text extraction: text for regular messages, caption for media messages
const messageText = m.text ?? m.caption ?? '';
if (hasMedia && isImageEnabled()) {
if (m.media_group_id) {
// Album message — buffer for debounce, advance fetchOffset immediately
@@ -519,8 +521,9 @@ export class TelegramAdapter extends BaseChannelAdapter {
// Single image message — process immediately
await this.processSingleImageMessage(update, chatId, userId, displayName);
}
} else if (m.text) {
// Pure text message — original logic
} else if (messageText) {
// Text/caption message (covers: pure text, image_enabled=false + caption,
// unsupported document + caption)
const msg: InboundMessage = {
messageId: String(m.message_id),
address: {
@@ -529,7 +532,7 @@ export class TelegramAdapter extends BaseChannelAdapter {
userId,
displayName,
},
text: m.text,
text: messageText,
timestamp: m.date * 1000,
raw: update,
updateId: update.update_id,
@@ -542,7 +545,7 @@ export class TelegramAdapter extends BaseChannelAdapter {
chatId,
direction: 'inbound',
messageId: String(m.message_id),
summary: m.text.slice(0, 200),
summary: messageText.slice(0, 200),
});
} catch { /* best effort */ }
@@ -586,6 +589,7 @@ export class TelegramAdapter extends BaseChannelAdapter {
/**
* Process a single image message (no media_group_id).
* Downloads the image and enqueues a message with attachments.
* Sends rejection notifications directly to Telegram on failure.
*/
private async processSingleImageMessage(
update: TelegramUpdate,
@@ -595,45 +599,44 @@ export class TelegramAdapter extends BaseChannelAdapter {
): Promise<void> {
const m = update.message!;
const token = this.botToken;
const address = { channelType: 'telegram' as const, chatId, userId, displayName };
if (!token) {
this.markUpdateProcessed(update.update_id);
return;
}
const attachments: FileAttachment[] = [];
let rejectedReason: string | undefined;
const rejections: MediaDownloadResult[] = [];
if (m.photo && m.photo.length > 0) {
const att = await downloadPhoto(token, m.photo, String(m.message_id));
if (att) attachments.push(att);
const result = await downloadPhoto(token, m.photo, String(m.message_id));
if (result.attachment) {
attachments.push(result.attachment);
} else if (result.rejected && result.rejected !== 'unsupported_type') {
rejections.push(result);
}
} 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;
} else if (result.rejected && result.rejected !== 'unsupported_type') {
rejections.push(result);
}
}
// Send rejection notification directly to user
if (rejections.length > 0) {
const notice = rejections.map(r => r.rejectedMessage || 'Image processing failed').join('\n');
this.send({ address, text: notice, parseMode: 'plain' }).catch(() => {});
}
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);
}
// Nothing usable (all images failed, no text) — mark processed
this.markUpdateProcessed(update.update_id);
return;
}
@@ -654,8 +657,8 @@ export class TelegramAdapter extends BaseChannelAdapter {
const msg: InboundMessage = {
messageId: String(m.message_id),
address: { channelType: 'telegram', chatId, userId, displayName },
text: text,
address,
text,
timestamp: m.date * 1000,
raw: update,
updateId: update.update_id,
@@ -706,6 +709,13 @@ export class TelegramAdapter extends BaseChannelAdapter {
if (!entry) return;
this.mediaGroupBuffers.delete(mediaGroupId);
const address = {
channelType: 'telegram' as const,
chatId: entry.chatId,
userId: entry.userId,
displayName: entry.displayName,
};
const token = this.botToken;
if (!token) {
// Can't download — mark all as processed
@@ -717,6 +727,7 @@ export class TelegramAdapter extends BaseChannelAdapter {
}
const attachments: FileAttachment[] = [];
const rejections: MediaDownloadResult[] = [];
let caption = '';
let firstMessageId = '';
let firstDate = 0;
@@ -734,14 +745,31 @@ export class TelegramAdapter extends BaseChannelAdapter {
}
if (m.photo && m.photo.length > 0) {
const att = await downloadPhoto(token, m.photo, String(m.message_id));
if (att) attachments.push(att);
const result = await downloadPhoto(token, m.photo, String(m.message_id));
if (result.attachment) {
attachments.push(result.attachment);
} else if (result.rejected && result.rejected !== 'unsupported_type') {
rejections.push(result);
}
} 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);
if (result.attachment) {
attachments.push(result.attachment);
} else if (result.rejected && result.rejected !== 'unsupported_type') {
rejections.push(result);
}
}
}
// Send rejection notification if any images failed
if (rejections.length > 0) {
const reasons = rejections.map(r => r.rejectedMessage || 'Image processing failed').join('\n');
const notice = rejections.length === 1
? reasons
: `${rejections.length} image(s) failed:\n${reasons}`;
this.send({ address, text: notice, parseMode: 'plain' }).catch(() => {});
}
const text = caption;
const hasContent = attachments.length > 0 || text.trim();
@@ -771,14 +799,15 @@ export class TelegramAdapter extends BaseChannelAdapter {
// Use the max updateId so acknowledgeUpdate advances offset past all buffered updates
const maxUpdateId = Math.max(...entry.updateIds);
// Pre-register all buffered IDs in recentUpdateIds so the contiguous
// watermark walk can advance past them when bridge-manager acks maxUpdateId.
for (const uid of entry.updateIds) {
this.recentUpdateIds.add(uid);
}
const msg: InboundMessage = {
messageId: firstMessageId,
address: {
channelType: 'telegram',
chatId: entry.chatId,
userId: entry.userId,
displayName: entry.displayName,
},
address,
text,
timestamp: firstDate * 1000,
updateId: maxUpdateId,
src/lib/bridge/adapters/telegram-media.ts
+43 -29
View File
@@ -46,6 +46,19 @@ export interface TelegramDocument {
file_size?: number;
}
// ── Download Result ──────────────────────────────────────────
export type MediaRejectCode = 'too_large' | 'unsupported_type' | 'download_failed';
/** Unified result for all media download attempts. */
export interface MediaDownloadResult {
attachment: FileAttachment | null;
/** Rejection code — set when attachment is null and failure is user-actionable. */
rejected?: MediaRejectCode;
/** Human-readable rejection message for display in Telegram. */
rejectedMessage?: string;
}
// ── Public API ───────────────────────────────────────────────
/**
@@ -127,33 +140,30 @@ export function selectOptimalPhoto(photos: TelegramPhotoSize[]): TelegramPhotoSi
}
/**
* Download a photo from Telegram's photo[] array and return a FileAttachment.
* Download a photo from Telegram's photo[] array.
*
* 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<FileAttachment | null> {
): Promise<MediaDownloadResult> {
const selected = selectOptimalPhoto(photos);
return downloadFileById(botToken, selected.file_id, messageId, selected.file_size);
return downloadFileById(botToken, selected.file_id, messageId);
}
/**
* 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 }> {
): Promise<MediaDownloadResult> {
// Check MIME type
const mime = doc.mime_type || inferMimeType(doc.file_name || '');
if (!mime || !isSupportedImageMime(mime)) {
@@ -163,16 +173,14 @@ export async function downloadDocumentImage(
// 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.`,
rejected: 'too_large',
rejectedMessage: formatSizeError(doc.file_size, maxSize),
};
}
const attachment = await downloadFileById(botToken, doc.file_id, messageId, doc.file_size);
return { attachment };
return downloadFileById(botToken, doc.file_id, messageId);
}
// ── Internal ─────────────────────────────────────────────────
@@ -186,8 +194,7 @@ async function downloadFileById(
botToken: string,
fileId: string,
messageId: string,
expectedSize?: number,
): Promise<FileAttachment | null> {
): Promise<MediaDownloadResult> {
const maxSize = getMaxImageSize();
for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
@@ -208,7 +215,7 @@ async function downloadFileById(
await sleep(1000 * Math.pow(2, attempt - 1));
continue;
}
return null;
return { attachment: null, rejected: 'download_failed', rejectedMessage: 'Failed to get file info from Telegram.' };
}
const filePath: string = getFileData.result.file_path;
@@ -217,7 +224,7 @@ async function downloadFileById(
// Pre-check size from API response
if (fileSize && fileSize > maxSize) {
console.warn(`[telegram-media] File too large: ${fileSize} bytes (max ${maxSize})`);
return null;
return { attachment: null, rejected: 'too_large', rejectedMessage: formatSizeError(fileSize, maxSize) };
}
// Step 2: Download the file
@@ -232,21 +239,21 @@ async function downloadFileById(
await sleep(1000 * Math.pow(2, attempt - 1));
continue;
}
return null;
return { attachment: null, rejected: 'download_failed', rejectedMessage: 'Failed to download image from Telegram.' };
}
// 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;
return { attachment: null, rejected: 'too_large', rejectedMessage: formatSizeError(parseInt(contentLength, 10), maxSize) };
}
// 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;
return { attachment: null, rejected: 'too_large', rejectedMessage: formatSizeError(buffer.length, maxSize) };
}
// Step 4: Determine MIME type
@@ -256,15 +263,15 @@ async function downloadFileById(
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: {
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);
@@ -273,11 +280,18 @@ async function downloadFileById(
await sleep(1000 * Math.pow(2, attempt - 1));
continue;
}
return null;
return { attachment: null, rejected: 'download_failed', rejectedMessage: 'Image download failed after retries.' };
}
}
return null;
return { attachment: null, rejected: 'download_failed', rejectedMessage: 'Image download failed after retries.' };
}
/** Format a human-readable size-exceeded error message. */
function formatSizeError(actualBytes: number, limitBytes: number): string {
const actualMB = (actualBytes / (1024 * 1024)).toFixed(1);
const limitMB = (limitBytes / (1024 * 1024)).toFixed(0);
return `Image too large (${actualMB} MB, limit ${limitMB} MB). Please send as a photo instead of a file.`;
}
function sleep(ms: number): Promise<void> {