diff --git a/README.md b/README.md index 6da8e9b1..40866081 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,23 @@ MTProto 服务的 `host`/`port` 是实际监听地址；Bridge 的 `serverHost`/ token: your-qqnt-bridge-token # 省略时读取环境变量 QQNT_BRIDGE_TOKEN ``` +### 图片存储 + +在 bridge 配置中启用 `imageStorage.enabled`，服务器会在接收群聊、私聊和图文多图消息后主动保存图片，无需客户端在线或打开会话。默认关闭，不保存视频、头像和普通附件，也不回填历史消息；未保存且已过期的图片无法恢复。 + +```yaml +imageStorage: + enabled: true + path: data/bridge-images + adminUserIds: [] +``` + +图片存储查询和删除需要现有 SSO 登录服务及显式管理员白名单。先在 WebUI 的「Your account」页面登录；通过现有 `GET /sso/me`（请求头 `Authorization: Bearer <登录令牌>`）读取自己的用户 `id`，由维护者将该数值加入 `imageStorage.adminUserIds`，例如 `[1]`。普通注册用户不会自动获得权限。未启用 SSO、未登录、令牌失效或白名单为空时一律拒绝管理请求；这不影响新图自动保存、Telegram 账号读取自己的副本及既有匿名账号发现。 + +原图及平台提供的独立预览图分别保存，按各自的格式和字节范围通过现有 MTProto 媒体下载读取，不公开新的 HTTP 文件地址。图片数仅计算原图，容量包含原图和预览图的实际文件字节数，删除会同时清理两种副本。关闭开关只停止新下载，已有副本仍可读取和管理。下载失败不会阻断消息接收，待处理和失败的新图任务会在账号重新连接或服务重启后重试。 + +WebUI 的「图片存储」页面可按账号、会话及消息日期筛选，以 UTC 自然日统计图片数和实际文件字节数，按会话或日期分页汇总。删除仅清理指定范围内的服务端副本，不删除平台消息或原图；同一消息重复推送不会重新生成主动删除的副本。删除失败会保留可重试记录。数据库与 `imageStorage.path` 目录应一起备份，修改目录不会自动迁移旧文件。 + ### Linux 生产部署 ```sh diff --git a/app.yml b/app.yml index c488444a..1635af9c 100644 --- a/app.yml +++ b/app.yml @@ -15,6 +15,10 @@ name: '@mtproto-relay/bridge' config: serverHost: 127.0.0.1 + imageStorage: + enabled: false + path: data/bridge-images + adminUserIds: [] groupFilesMiniApp: enabled: true path: /group-files diff --git a/packages/bridge/client/index.ts b/packages/bridge/client/index.ts index acfafaa2..be20dbd5 100644 --- a/packages/bridge/client/index.ts +++ b/packages/bridge/client/index.ts @@ -1,10 +1,12 @@ import Page0 from './accounts.js' import Page1 from './stickers.js' import Page2 from './bots.js' +import StoragePage from './storage.js' import './style.css' import type { ClientModule } from 'cordis-webui-solidjs/client' export default { pages: [ { path: "/platform-accounts", title: "Platform accounts", icon: "○", group: "Workspace", component: Page0 }, { path: "/sticker-packs", title: "Sticker collections", icon: "✦", group: "Workspace", component: Page1 }, { path: "/bots", title: "Your bots", icon: "◇", group: "Workspace", component: Page2 }, + { path: "/image-storage", title: "Image storage", icon: "▣", group: "Workspace", component: StoragePage }, ] } satisfies ClientModule diff --git a/packages/bridge/client/storage.tsx b/packages/bridge/client/storage.tsx new file mode 100644 index 00000000..7553d590 --- /dev/null +++ b/packages/bridge/client/storage.tsx @@ -0,0 +1,421 @@ +/** @jsxImportSource solid-js */ +import { + batch, + createResource, + createSignal, + For, + onCleanup, + Show, +} from 'solid-js' +import type { + ImageStorageDashboardData, + ImageStorageFilter, + ImageStorageGroup, + StickerPackDashboardData, +} from '../src/dashboard-types.js' +import { useRpc, type PageProps } from 'cordis-webui-solidjs/client' +import { sessionToken } from 'cordis-webui-solidjs/session' +import { + ActionError, + EmptyState, + formatBytes, + LiveContent, + Modal, + PageHeader, + useAction, +} from 'cordis-webui-solidjs/components' + +export default function StoragePage(props: PageProps) { + const rpc = useRpc( + props.entryId, + ), + action = useAction() + const [account, setAccount] = createSignal(''), + [conversation, setConversation] = createSignal(''), + [startDate, setStartDate] = createSignal(''), + [endDate, setEndDate] = createSignal(''), + [groupBy, setGroupBy] = createSignal<'conversation' | 'date'>( + 'conversation', + ), + [page, setPage] = createSignal(0), + [token, setToken] = createSignal(sessionToken()), + [deletion, setDeletion] = createSignal<{ + filter: ImageStorageFilter + scope: string + count: number + bytes: number + }>() + const syncToken = () => { + const current = sessionToken() + if (current !== token()) + batch(() => { + setDeletion(undefined) + setToken(current) + }) + } + window.addEventListener('storage', syncToken) + window.addEventListener('focus', syncToken) + onCleanup(() => { + window.removeEventListener('storage', syncToken) + window.removeEventListener('focus', syncToken) + }) + const invalidDates = () => + !!(startDate() && endDate() && startDate() > endDate()), + filter = (): ImageStorageFilter => ({ + platformSessionId: account() || undefined, + conversationId: conversation().trim() || undefined, + startDate: startDate() || undefined, + endDate: endDate() || undefined, + }) + const [result, { refetch }] = createResource( + () => + rpc.ready && !invalidDates() + ? { + query: { + ...filter(), + groupBy: groupBy(), + page: page() + 1, + pageSize: 25, + }, + token: token(), + } + : false, + ({ query, token }) => rpc.data.getImageStorage(query, token ?? undefined), + ) + const snapshot = () => (result.error ? undefined : result()), + busy = () => !rpc.ready || action.busy() || result.loading, + accountName = (id: string) => + rpc.data.stickerAccounts?.find((item) => item.platformSessionId === id) + ?.displayName ?? id, + scope = (value: ImageStorageFilter) => + [ + value.platformSessionId + ? 'Account: ' + + accountName(value.platformSessionId) + + ' (' + + value.platformSessionId + + ')' + : 'All accounts', + value.conversationId + ? 'Conversation: ' + value.conversationId + : 'All group and private conversations', + (value.startDate ?? 'Beginning') + + ' to ' + + (value.endDate ?? 'present') + + ' (UTC, inclusive)', + ].join('\n') + const requestDeletion = (group?: ImageStorageGroup) => { + const current = snapshot() + if (!current || busy() || invalidDates()) return + const selected: ImageStorageFilter = { + ...filter(), + ...(group && { platformSessionId: group.platformSessionId }), + ...(group?.conversationId && { conversationId: group.conversationId }), + ...(group?.date && { startDate: group.date, endDate: group.date }), + } + action.clear() + setDeletion({ + filter: selected, + scope: scope(selected), + count: group?.count ?? current.count, + bytes: group?.bytes ?? current.bytes, + }) + } + const change = (update: () => void) => + batch(() => { + update() + setPage(0) + }) + return ( + <> + { + if (sessionToken() === token()) void refetch() + else syncToken() + }} + > + Refresh + + } + /> + + +
+

+ New image storage:{' '} + + {rpc.data.imageStorageEnabled ? 'Enabled' : 'Disabled (default)'} + + . Change this in the bridge configuration. Existing copies remain + available when disabled. +

+
+ + + + + +
+ + +

Loading image storage…

+ + +
+

+ {snapshot()!.count.toLocaleString()} images ·{' '} + {formatBytes(snapshot()!.bytes)} ( + {snapshot()!.bytes.toLocaleString()} bytes) + + {' '} + · {snapshot()!.pendingCount} pending + +

+ +
+

+ Only server copies are deleted. Messages and original platform + images are not affected. Pending downloads in the selected range + are cancelled. +

+ + } + > +
+ + + + + + + + + + + + + {(group) => ( + + + + + + + + )} + + +
Account + {groupBy() === 'date' ? 'Date (UTC)' : 'Conversation'} + ImagesStored sizeActions
{accountName(group.platformSessionId)} + {group.date ?? + group.conversationTitle ?? + group.conversationId} + +
+ + {group.conversationId} + + + 		{group.count.toLocaleString()} + {formatBytes(group.bytes)} + + +
+
+ + + +
+ + + {(selected) => ( + { + if (!action.busy()) setDeletion(undefined) + }} + > +
+

{selected.scope}

+

+ {selected.count.toLocaleString()} images ·{' '} + {formatBytes(selected.bytes)} ({selected.bytes.toLocaleString()}{' '} + bytes). Pending downloads in this range will also be cancelled. +

+

+ Messages and original platform images will not be deleted. + Server copies cannot be restored without downloading them again. +

+ +
+ + +
+
+ + )} + + + ) +} diff --git a/packages/bridge/client/style.css b/packages/bridge/client/style.css index d97e4139..acfb6ec5 100644 --- a/packages/bridge/client/style.css +++ b/packages/bridge/client/style.css @@ -163,6 +163,15 @@ font-size: 12px; overflow-wrap: anywhere; } + .image-storage-filters { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(min(100%, 190px), 1fr)); + gap: 16px; + } + .image-storage-scope { + white-space: pre-line; + overflow-wrap: anywhere; + } .sticker-controls { margin-bottom: 24px; } diff --git a/packages/bridge/package.json b/packages/bridge/package.json index ae2bfa8b..db95f744 100644 --- a/packages/bridge/package.json +++ b/packages/bridge/package.json @@ -40,6 +40,12 @@ "title": "Your bots", "icon": "bot", "group": "Accounts" + }, + { + "path": "/image-storage", + "title": "Image storage", + "icon": "database", + "group": "Workspace" } ] } diff --git a/packages/bridge/src/dashboard-types.ts b/packages/bridge/src/dashboard-types.ts index 6f57b779..88ddf6a8 100644 --- a/packages/bridge/src/dashboard-types.ts +++ b/packages/bridge/src/dashboard-types.ts @@ -96,4 +96,44 @@ export interface BotDashboardData { refreshBots(): Promise } -export type BridgeDashboardData = PlatformAccountDashboardData & StickerPackDashboardData & BotDashboardData +export interface ImageStorageFilter { + platformSessionId?: string + conversationId?: string + startDate?: string + endDate?: string +} + +export interface ImageStorageQuery extends ImageStorageFilter { + groupBy?: 'conversation' | 'date' + /** 从 1 开始，默认 1。 */ + page?: number + /** 默认 50，最大 100。 */ + pageSize?: number +} + +export interface ImageStorageGroup { + platformSessionId: string + conversationId?: string + conversationTitle?: string + date?: string + count: number + bytes: number +} + +export interface ImageStorageResult { + count: number + bytes: number + pendingCount: number + groups: ImageStorageGroup[] + totalGroups: number + page: number + pageSize: number +} + +export interface ImageStorageDashboardData { + imageStorageEnabled: boolean + getImageStorage(query: ImageStorageQuery, token?: string): Promise + deleteImageStorage(filter: ImageStorageFilter, token?: string): Promise<{ count: number, bytes: number }> +} + +export type BridgeDashboardData = PlatformAccountDashboardData & StickerPackDashboardData & BotDashboardData & ImageStorageDashboardData diff --git a/packages/bridge/src/dialogs.ts b/packages/bridge/src/dialogs.ts index 5109871c..c684b16f 100644 --- a/packages/bridge/src/dialogs.ts +++ b/packages/bridge/src/dialogs.ts @@ -292,6 +292,7 @@ export class DialogRpc { private readonly _dialogFolders?: DialogFolderStore, private readonly _systemPeers?: SystemPeerService, private readonly _messageProjection?: MessageProjectionPipeline, + private readonly _imageStorage?: import('./image-storage.js').ImageStorage, ) { this._actions = new PlatformMessageActions(_platform, _session) // 有 `MessageStore` 时淘汰旧条目并由数据库兜底；无存储模式不可淘汰，避免重用已分配的 ID。 @@ -2505,7 +2506,9 @@ export class DialogRpc { async getFile(req: tl.upload.RawGetFileRequest): Promise { const offset = safeOffset(req.offset) - if (offset < 0 || req.limit <= 0) throw new RpcError(400, 'OFFSET_INVALID') + if (offset < 0 || !Number.isSafeInteger(req.limit) || req.limit <= 0 || req.limit > 1024 * 1024) { + throw new RpcError(400, 'OFFSET_INVALID') + } if (req.location._ === 'inputStickerSetThumb') { const bytes = await this._stickers?.getSetThumb(req.location.stickerset, offset, req.limit) if (!bytes) throw new RpcError(400, 'LOCATION_INVALID') @@ -2564,6 +2567,20 @@ export class DialogRpc { } } } + const archivedSource = this._imageStorage + ? await this._store?.getMedia(this._session.platformSessionId, req.location.id.toNumber()) + : undefined + if (archivedSource) { + const isPreview = shouldServeMediaPreview(req.location, archivedSource.media) + const media = isPreview ? previewMedia(archivedSource.media) : archivedSource.media + const bytes = await this._imageStorage!.read( + this._session, archivedSource.messageId, archivedSource.media, offset, req.limit, isPreview, + ) + if (bytes) return { + _: 'upload.file', type: storageFileType(media.mimeType ?? 'application/octet-stream'), + mtime: archivedSource.timestamp, bytes, + } + } const transient = this._messageProjection?.resolveMedia( this._session, req.location.id.toNumber(), @@ -2589,7 +2606,7 @@ export class DialogRpc { bytes: await readSourceRange(staged.media.source, offset, req.limit), } } - const stored = await this._store.getMedia(this._session.platformSessionId, req.location.id.toNumber()) + const stored = archivedSource ?? await this._store.getMedia(this._session.platformSessionId, req.location.id.toNumber()) if (!stored) throw new RpcError(400, 'FILE_ID_INVALID') const media = shouldServeMediaPreview(req.location, stored.media) ? previewMedia(stored.media) @@ -2626,11 +2643,20 @@ export class DialogRpc { if (decodeBridgeMediaReference(location.fileReference) !== mediaId) { throw new RpcError(400, 'FILE_REFERENCE_INVALID') } + const archivedSource = this._imageStorage + ? await this._store?.getMedia(this._session.platformSessionId, mediaId) + : undefined + if (archivedSource && await this._imageStorage!.has( + this._session, archivedSource.messageId, archivedSource.media, shouldServeMediaPreview(location, archivedSource.media), + )) { + // 沿用 patched client 已有的直链失败回退，不暴露本地文件 URL。 + throw new RpcError(400, 'MEDIA_DIRECT_URL_UNAVAILABLE') + } if (!this._platform.resolveMediaUrl) { throw new RpcError(400, 'MEDIA_DIRECT_URL_UNAVAILABLE') } const transient = this._messageProjection?.resolveMedia(this._session, mediaId) - const stored = transient ? undefined : await this._store?.getMedia(this._session.platformSessionId, mediaId) + const stored = transient ? undefined : archivedSource ?? await this._store?.getMedia(this._session.platformSessionId, mediaId) if (!transient && !stored) throw new RpcError(400, 'FILE_ID_INVALID') const source = transient?.media ?? stored!.media const media = shouldServeMediaPreview(location, source) diff --git a/packages/bridge/src/image-storage-auth.e2e.test.ts b/packages/bridge/src/image-storage-auth.e2e.test.ts new file mode 100644 index 00000000..eeb30e36 --- /dev/null +++ b/packages/bridge/src/image-storage-auth.e2e.test.ts @@ -0,0 +1,110 @@ +import SSO from '@cordisjs/plugin-sso' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { startApp, waitForWebuiRoute } from '../../test-suite/src/harness.js' + +const disposals: Array<() => Promise> = [] +afterEach(async () => { for (const dispose of disposals.splice(0).reverse()) await dispose() }) + +async function fixture(adminUserIds: number[] = [1]) { + const root = await mkdtemp(join(tmpdir(), 'crossgram-image-auth-')) + disposals.push(() => rm(root, { recursive: true, force: true })) + const app = await startApp({ bridgeConfig: { + uploadPath: join(root, 'uploads'), imageStorage: { path: join(root, 'images'), adminUserIds }, + } }) + disposals.push(app.stop) + const entry = await waitForWebuiRoute(app.ctx, '/platform-accounts') + const id = 'a'.repeat(64) + await app.ctx.database.create('mtproto_image_storage', { + id, platformId: 'private-account', platformSessionId: 'private-session', messageId: 1, + platformMediaId: 'private-image', isPreview: false, conversationId: 'secret-group', conversationTitle: '私密群', + date: '2026-10-15', timestamp: 1792022400, media: {}, state: 'pending', bytes: 0, + }) + const frames: any[] = [] + const socket = new WebSocket(app.ctx.server.baseUrl.replace('http:', 'ws:') + '/api') + disposals.push(async () => { socket.close() }) + socket.addEventListener('message', event => frames.push(JSON.parse(String(event.data)))) + await vi.waitFor(() => expect(frames[0]?.type).toBe('entry:init')) + let serial = 0 + const call = async (method: string, token?: unknown, filter: unknown = {}) => { + const sn = ++serial + socket.send(JSON.stringify({ type: 'rpc:request', body: { sn, entryId: entry.id, method, args: [filter, token] } })) + await vi.waitFor(() => expect(frames.some(frame => frame.type === 'rpc:response' && frame.body.sn === sn)).toBe(true)) + return frames.find(frame => frame.type === 'rpc:response' && frame.body.sn === sn).body + } + const pending = async () => expect(await app.ctx.database.get('mtproto_image_storage', { id })) + .toMatchObject([{ state: 'pending' }]) + const sso = async () => { + const fiber = app.ctx.plugin(SSO) + disposals.push(() => fiber.dispose()) + await fiber + await app.ctx.database.prepared() + const admin = await app.ctx.sso.createUser('image-storage-auth-admin') + const member = await app.ctx.sso.createUser('image-storage-auth-member') + expect(admin.user.id).toBe(1) + return { + admin: await app.ctx.sso.createSession(admin.user.id, admin.identityId), + member: await app.ctx.sso.createSession(member.user.id, member.identityId), + } + } + return { ...app, entry, id, frames, socket, call, pending, sso } +} + +describe('图片存储 WebSocket 管理权限 e2e', () => { + it('无SSO时拒绝新RPC，匿名账号发现保持原行为且不广播归档清单', async () => { + const app = await fixture() + for (const method of ['getImageStorage', 'deleteImageStorage']) { + expect(await app.call(method)).toMatchObject({ ok: false, message: expect.stringContaining('SSO') }) + } + await app.pending() + app.socket.send(JSON.stringify({ type: 'entry:subscribe', body: { id: app.entry.id } })) + await vi.waitFor(() => expect(app.frames.some(frame => frame.type === 'entry:snapshot')).toBe(true)) + const snapshot = app.frames.find(frame => frame.type === 'entry:snapshot') + expect(snapshot.body.data.accounts).toBeInstanceOf(Array) + expect(JSON.stringify(app.frames)).not.toContain('secret-group') + expect(JSON.stringify(app.frames)).not.toContain('私密群') + }) + + it('每次query/delete都拒绝匿名、伪造token、普通SSO用户和已撤销管理员会话', async () => { + const app = await fixture() + const tokens = await app.sso() + for (const method of ['getImageStorage', 'deleteImageStorage']) { + expect(await app.call(method)).toMatchObject({ ok: false, message: expect.stringContaining('Your account') }) + expect(await app.call(method, { id: 1 })).toMatchObject({ ok: false, message: expect.stringContaining('Your account') }) + expect(await app.call(method, 'forged-token')).toMatchObject({ ok: false, message: expect.stringContaining('登录已失效') }) + expect(await app.call(method, tokens.member)).toMatchObject({ ok: false, message: expect.stringContaining('没有图片存储管理权限') }) + await app.pending() + } + expect(JSON.stringify(app.frames)).not.toContain('secret-group') + expect(await app.call('getImageStorage', tokens.admin)).toMatchObject({ ok: true, value: { + count: 0, bytes: 0, pendingCount: 1, groups: [{ conversationId: 'secret-group' }], + } }) + await app.ctx.sso.destroySession(tokens.admin) + for (const method of ['getImageStorage', 'deleteImageStorage']) { + expect(await app.call(method, tokens.admin)).toMatchObject({ ok: false, message: expect.stringContaining('登录已失效') }) + } + await app.pending() + const identity = (await app.ctx.sso.getIdentities(1))[0] + const expired = await app.ctx.sso.createSession(1, identity.id) + await app.ctx.database.set('sso.session', { token: expired }, { expiresAt: new Date(0) }) + for (const method of ['getImageStorage', 'deleteImageStorage']) { + expect(await app.call(method, expired)).toMatchObject({ ok: false, message: expect.stringContaining('登录已失效') }) + } + await app.pending() + const fresh = await app.ctx.sso.createSession(1, identity.id) + expect(await app.call('deleteImageStorage', fresh, { conversationId: 'secret-group' })) + .toMatchObject({ ok: true, value: { count: 0, bytes: 0 } }) + expect(await app.ctx.database.get('mtproto_image_storage', { id: app.id })).toMatchObject([{ state: 'deleted' }]) + }) + + it('空管理员白名单在已登录时仍拒绝管理，不把普通注册用户当管理员', async () => { + const app = await fixture([]) + const tokens = await app.sso() + for (const method of ['getImageStorage', 'deleteImageStorage']) { + expect(await app.call(method, tokens.admin)).toMatchObject({ ok: false, message: expect.stringContaining('imageStorage.adminUserIds') }) + } + await app.pending() + }) +}) diff --git a/packages/bridge/src/image-storage.e2e.test.ts b/packages/bridge/src/image-storage.e2e.test.ts new file mode 100644 index 00000000..0475ce5f --- /dev/null +++ b/packages/bridge/src/image-storage.e2e.test.ts @@ -0,0 +1,134 @@ +import { Context } from 'cordis' +import Database from '@cordisjs/plugin-database' +import SQLiteDriver from '@cordisjs/plugin-database-sqlite' +import Long from 'long' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it, vi } from 'vitest' +import { defineModels } from './models.js' +import { MessageStore } from './message-store.js' +import { ImageStorage } from './image-storage.js' +import { DialogRpc } from './dialogs.js' +import { PlatformRegistry, PlatformSubscriptionManager } from './platform-manager.js' +import type { IMConversation, IMEvent, IMMedia, IMMessage, IMPlatform, PlatformSession } from './platform.js' + +const session: PlatformSession = { platformId: 'images', platformSessionId: 'image-account', userId: 'self', credentials: {}, metadata: {} } +const conversation: IMConversation = { id: 'group', kind: 'group', title: '图片群' } + +describe('图片存储接收与下载 e2e', () => { + it('没有客户端也主动归档；媒体行重建、源过期和关闭存储后仍可分块读取，账号不能越权', async () => { + const ctx = new Context() + const fibers = [ctx.plugin(Database), ctx.plugin(SQLiteDriver, { path: ':memory:' })] + await Promise.all(fibers) + await new Promise(resolve => setTimeout(resolve, 25)) + defineModels(ctx) + await ctx.database.prepared() + const path = await mkdtemp(join(tmpdir(), 'crossgram-image-e2e-')) + const data = new Uint8Array(Array.from({ length: 14000 }, (_, index) => index % 251)) + const preview = new Uint8Array(Array.from({ length: 200 }, (_, index) => (index * 7) % 251)) + let receive!: (event: IMEvent) => void | Promise + let expired = false + const download = vi.fn(async function* (_session: PlatformSession, media: IMMedia) { + if (expired) throw new Error('QQ 图片源已过期') + const bytes = media.id.endsWith(':preview') ? preview : data + yield bytes.subarray(0, 5000) + yield bytes.subarray(5000) + }) + const resolveMediaUrl = vi.fn(async () => ({ url: 'https://example.com/expired', expiresAt: Date.now() + 60000, supportsRange: true })) + const platform: IMPlatform = { + capabilities: { + history: false, + send: { text: true, images: true, files: false, mixed: true, maxTextLength: 4096, maxMedia: 10 }, + conversations: { groups: true, channels: false, subchannels: false }, + }, + async subscribe(_session, handler) { receive = handler; return () => {} }, + async sendMessage() { throw new Error('此测试不发送消息') }, + downloadMedia: download, + resolveMediaUrl, + } + const store = new MessageStore(ctx.database) + const errors = vi.fn() + const storage = new ImageStorage(ctx.database, { enabled: true, path }, errors) + let restored: ImageStorage | undefined + ctx.on('im-platform/event-committed', async (account, committed) => { + if (committed.event.type !== 'message' && committed.event.type !== 'message-edit') return + if (!('result' in committed) || !('message' in committed.result)) return + await storage.enqueue(account, committed.event.conversation, committed.event.message, Number(committed.result.message.id)) + }) + const subscriptions = new PlatformSubscriptionManager(ctx.database, new PlatformRegistry([['images', platform]]), store, undefined, undefined, undefined, ctx) + try { + await storage.activate(session, platform) + await subscriptions.ensure(session) + const message: IMMessage = { + id: 'message-one', conversationId: conversation.id, senderId: 'alice', timestamp: 1792022400, + content: { parts: [{ type: 'text', text: '第一张' }, { type: 'media', media: { + id: 'original-media', kind: 'image', size: data.length, mimeType: 'image/png', width: 2000, height: 1000, + locator: { path: '../../not-a-local-file' }, + preview: { mimeType: 'image/webp', size: preview.length, width: 100, height: 50, locator: { key: 'preview' } }, + } }, { type: 'text', text: '第二张' }, { type: 'media', media: { + id: 'second-media', kind: 'image', size: data.length, locator: {}, + } }] }, + } + await receive({ type: 'message', conversation, message }) + await expect.poll(async () => { + const result = await storage.query({}) + return [result.count, result.pendingCount] + }).toEqual([2, 0]) + expect(await storage.query({})).toMatchObject({ count: 2, bytes: data.length * 2 + preview.length }) + expect(download).toHaveBeenCalledTimes(3) + expired = true + const [storedMessage] = await ctx.database.get('mtproto_im_message', { platformSessionId: session.platformSessionId }) + const before = await ctx.database.get('mtproto_im_media', { messageId: storedMessage.id }) + await ctx.database.remove('mtproto_im_media', { messageId: storedMessage.id }) + await store.ingest(session, conversation, { + ...message, content: { parts: [...message.content.parts, { type: 'text', text: '更新文字' }] }, + }) + const after = await ctx.database.get('mtproto_im_media', { messageId: storedMessage.id }) + expect(after[0].id).not.toBe(before[0].id) + await storage.close() + restored = new ImageStorage(ctx.database, { enabled: false, path }, errors) + const rpc = new DialogRpc( + platform, session, new MessageStore(ctx.database), undefined, undefined, 1, + undefined, undefined, undefined, undefined, undefined, undefined, undefined, undefined, + undefined, undefined, undefined, undefined, undefined, undefined, restored, + ) + const id = Long.fromNumber(after[0].id) + const location = { _: 'inputPhotoFileLocation' as const, id, accessHash: id, + fileReference: new TextEncoder().encode(`bridge-media:${id.toString()}`), thumbSize: 'x' } + await expect(rpc.getFileUrl(location)).rejects.toThrow('MEDIA_DIRECT_URL_UNAVAILABLE') + const thumb = { ...location, thumbSize: 'm' } + await expect(rpc.getFileUrl(thumb)).rejects.toThrow('MEDIA_DIRECT_URL_UNAVAILABLE') + const thumbnail = await rpc.getFile({ _: 'upload.getFile', location: thumb, offset: 0, limit: 4096 }) + expect(thumbnail).toMatchObject({ _: 'upload.file', type: { _: 'storage.fileWebp' }, bytes: preview }) + const documentThumb = await rpc.getFile({ _: 'upload.getFile', + location: { ...thumb, _: 'inputDocumentFileLocation' }, offset: 0, limit: 4096, + }) + expect(documentThumb).toMatchObject({ _: 'upload.file', type: { _: 'storage.fileWebp' }, bytes: preview }) + const previewTail = await rpc.getFile({ _: 'upload.getFile', location: thumb, offset: 180, limit: 4096 }) + expect(previewTail).toMatchObject({ _: 'upload.file', bytes: preview.subarray(180) }) + expect(resolveMediaUrl).not.toHaveBeenCalled() + const chunks: Uint8Array[] = [] + for (const offset of [0, 4096, 8192, 12288]) { + const file = await rpc.getFile({ _: 'upload.getFile', location, offset, limit: 4096 }) + if (file._ !== 'upload.file') throw new Error('应返回本地图片') + expect(file.type).toEqual({ _: 'storage.filePng' }) + chunks.push(file.bytes) + } + expect(Buffer.concat(chunks)).toEqual(Buffer.from(data)) + expect(download).toHaveBeenCalledTimes(3) + expect(await restored.read({ ...session, platformSessionId: 'unauthorized' }, storedMessage.id, message.content.parts[1].type === 'media' ? message.content.parts[1].media : {} as never, 0, 100)).toBeUndefined() + expect(await restored.delete({ conversationId: conversation.id })).toEqual({ count: 2, bytes: data.length * 2 + preview.length }) + expect(await ctx.database.get('mtproto_image_storage', { state: 'deleted' })).toHaveLength(3) + await expect(rpc.getFile({ _: 'upload.getFile', location: thumb, offset: 0, limit: 4096 })).rejects.toThrow() + await expect(rpc.getFile({ _: 'upload.getFile', location, offset: 0, limit: 4096 })).rejects.toThrow() + expect(errors).not.toHaveBeenCalled() + } finally { + await subscriptions.stop() + await restored?.close() + await storage.close() + for (const fiber of fibers.reverse()) await fiber.dispose() + await rm(path, { recursive: true, force: true }) + } + }) +}) diff --git a/packages/bridge/src/image-storage.test.ts b/packages/bridge/src/image-storage.test.ts new file mode 100644 index 00000000..e5785988 --- /dev/null +++ b/packages/bridge/src/image-storage.test.ts @@ -0,0 +1,360 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import Database from '@cordisjs/plugin-database' +import SQLiteDriver from '@cordisjs/plugin-database-sqlite' +import { mkdtemp, readdir, rm, mkdir, writeFile, stat } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { defineModels } from './models.js' +import { ImageStorage } from './image-storage.js' +import { QQNTClient } from '../../platform-crossgram/src/client.js' +import type { IMConversation, IMMedia, IMMessage, IMPlatform, PlatformSession } from './platform.js' + +const session: PlatformSession = { platformSessionId: 'account-one', platformId: 'test', userId: 'self', credentials: {}, metadata: {} } +const group: IMConversation = { id: 'group/../one', kind: 'group', title: '群一' } +const media: IMMedia = { id: '../../image', kind: 'image', size: 99999, mimeType: 'image/png', locator: { key: 'image' } } +const bytes = new Uint8Array([1, 2, 3, 4, 5, 6, 7]) +const disposals: Array<() => Promise> = [] +afterEach(async () => { for (const dispose of disposals.splice(0).reverse()) await dispose() }) + +function message(id = 'one', timestamp = Date.parse('2026-10-15T00:00:00Z') / 1000, parts = [media]): IMMessage { + return { id, conversationId: group.id, senderId: 'sender', timestamp, + content: { parts: [{ type: 'text', text: '图文' }, ...parts.map(item => ({ type: 'media' as const, media: item }))] } } +} + +async function setup(enabled = true) { + const ctx = new Context() + const fibers = [ctx.plugin(Database), ctx.plugin(SQLiteDriver, { path: ':memory:' })] + await Promise.all(fibers) + await new Promise(resolve => setTimeout(resolve, 25)) + defineModels(ctx) + await ctx.database.prepared() + const path = await mkdtemp(join(tmpdir(), 'crossgram-images-')) + disposals.push(async () => { + for (const fiber of fibers.reverse()) await Promise.resolve((fiber as any).dispose?.()) + await rm(path, { recursive: true, force: true }) + }) + const errors = vi.fn() + const storage = new ImageStorage(ctx.database, { enabled, path }, errors) + disposals.push(() => storage.close()) + const download = vi.fn(async function* (_session: PlatformSession, _media: IMMedia, _options?: { signal?: AbortSignal }) { + yield bytes.subarray(0, 3) + yield bytes.subarray(3) + }) + const platform = { downloadMedia: download } as unknown as IMPlatform + return { ctx, path, storage, errors, platform, download } +} + +async function ready(storage: ImageStorage, count = 1) { + await expect.poll(async () => (await storage.query({})).count).toBe(count) +} + +describe('ImageStorage', () => { + it('默认关闭，不保存新图', async () => { + const { ctx, path, platform } = await setup() + const storage = new ImageStorage(ctx.database, { path }) + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + expect(storage.enabled).toBe(false) + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 0, totalGroups: 0 }) + expect(await readdir(path)).toEqual([]) + await storage.close() + }) + + it('主动保存图文多图、忽略附件，按实际字节统计并保留账号和UTC日期隔离', async () => { + const { storage, platform, path, download } = await setup() + await storage.activate(session, platform) + const other = { ...session, platformSessionId: 'account-two' } + await storage.activate(other, platform) + const yesterday = Date.parse('2026-10-14T23:59:59Z') / 1000 + await storage.enqueue(session, group, message('one', yesterday, [media, { ...media, id: 'second' }, { ...media, id: 'file', kind: 'file' }]), 1) + await storage.enqueue(other, { ...group, id: 'private', kind: 'direct', title: '私聊' }, message(), 2) + await ready(storage, 3) + expect(download).toHaveBeenCalledTimes(3) + expect(await storage.query({ platformSessionId: session.platformSessionId, startDate: '2026-10-14', endDate: '2026-10-14' })) + .toMatchObject({ count: 2, bytes: bytes.length * 2, totalGroups: 1, pendingCount: 0, + groups: [{ conversationId: group.id, conversationTitle: group.title, count: 2, bytes: bytes.length * 2 }] }) + expect(await storage.query({ groupBy: 'date', pageSize: 1, page: 2 })) + .toMatchObject({ count: 3, bytes: bytes.length * 3, totalGroups: 2, groups: [{ date: '2026-10-15', platformSessionId: other.platformSessionId }] }) + expect((await readdir(path)).every(name => /^[a-f0-9]{64}$/.test(name))).toBe(true) + if (process.platform !== 'win32') { + for (const account of await readdir(path)) { + expect((await stat(join(path, account))).mode & 0o777).toBe(0o700) + for (const file of await readdir(join(path, account))) { + expect((await stat(join(path, account, file))).mode & 0o777).toBe(0o600) + } + } + } + }) + + it('重复事件幂等，关闭开关和重建实例后仍能读取原生范围', async () => { + const { ctx, storage, path, platform, download } = await setup() + await storage.activate(session, platform) + await Promise.all(Array.from({ length: 5 }, () => storage.enqueue(session, group, message(), 1))) + await ready(storage) + await storage.close() + const disabled = new ImageStorage(ctx.database, { path }) + expect(await disabled.read(session, 1, media, 2, 3)).toEqual(bytes.subarray(2, 5)) + expect(await disabled.read(session, 1, media, 100, 3)).toEqual(new Uint8Array()) + expect(await disabled.read({ ...session, platformSessionId: 'other' }, 1, media, 0, 3)).toBeUndefined() + expect(await disabled.has(session, 1, media)).toBe(true) + expect(download).toHaveBeenCalledTimes(1) + await disabled.close() + }) + + it('迟到的独立预览可补存，统计不重复计图，主动删除后原图和预览均不能复活', async () => { + const { storage, platform, download, ctx } = await setup() + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await ready(storage) + const withPreview = { ...media, preview: { mimeType: 'image/webp', size: 999, width: 100, height: 50, locator: { key: 'preview' } } } + await storage.enqueue(session, group, message('one', undefined, [withPreview]), 1) + await expect.poll(async () => (await storage.query({})).pendingCount).toBe(0) + expect(await storage.query({})).toMatchObject({ count: 1, bytes: bytes.length * 2 }) + expect(await storage.read(session, 1, withPreview, 0, 100, true)).toEqual(bytes) + expect(await storage.delete({})).toEqual({ count: 1, bytes: bytes.length * 2 }) + await storage.enqueue(session, group, message('one', undefined, [withPreview]), 1) + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 0 }) + expect(await ctx.database.get('mtproto_image_storage', { state: 'deleted' })).toHaveLength(2) + expect(download).toHaveBeenCalledTimes(2) + }) + + it('下载失败不留半文件，重启后恢复待下载的新图片', async () => { + const { ctx, storage, path, platform, download, errors } = await setup() + download.mockImplementationOnce(async function* () { yield bytes; throw new Error('过期') }) + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await expect.poll(() => errors.mock.calls.length).toBeGreaterThan(0) + await storage.close() + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 1 }) + const accounts = await readdir(path) + expect(await readdir(join(path, accounts[0]))).toEqual([]) + const restored = new ImageStorage(ctx.database, { enabled: true, path }) + disposals.push(() => restored.close()) + await restored.activate(session, platform) + await ready(restored) + expect(await restored.read(session, 1, media, 0, 100)).toEqual(bytes) + }) + + it('队列并发有界且无需客户端触发，删除待下载项不能被旧任务或重复推送恢复', async () => { + const { storage, platform, download, ctx } = await setup() + let active = 0 + let maximum = 0 + const releases: Array<() => void> = [] + download.mockImplementation(async function* () { + active++ + maximum = Math.max(maximum, active) + try { await new Promise(resolve => releases.push(resolve)); yield bytes } finally { active-- } + }) + await storage.activate(session, platform) + await storage.enqueue(session, group, message('many', undefined, Array.from({ length: 5 }, (_, i) => ({ ...media, id: String(i) }))), 1) + await expect.poll(() => releases.length).toBe(2) + const result = await storage.delete({ platformSessionId: session.platformSessionId }) + expect(result).toEqual({ count: 0, bytes: 0 }) + for (const release of releases) release() + await storage.close() + expect(maximum).toBe(2) + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 0 }) + expect(await ctx.database.get('mtproto_image_storage', { state: 'deleted' })).toHaveLength(5) + }) + + it('停止时中断流式下载、不留半文件，待下载状态可在重启后恢复', async () => { + const { ctx, storage, platform, path, download } = await setup() + download.mockImplementationOnce(async function* (_session, _media, options) { + yield bytes.subarray(0, 2) + await new Promise(resolve => { + if (options?.signal?.aborted) resolve() + else options?.signal?.addEventListener('abort', () => resolve(), { once: true }) + }) + throw new Error('已中断') + }) + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await expect.poll(() => download.mock.calls.length).toBe(1) + await storage.close() + expect(await storage.query({})).toMatchObject({ pendingCount: 1, count: 0, bytes: 0 }) + for (const account of await readdir(path)) expect(await readdir(join(path, account))).toEqual([]) + const restored = new ImageStorage(ctx.database, { enabled: true, path }) + disposals.push(() => restored.close()) + await restored.activate(session, platform) + await ready(restored) + }) + + it('账号停用取消进行中的下载，重新激活后恢复而不记录正常取消为错误', async () => { + const { storage, platform, download, errors } = await setup() + let signal: AbortSignal | undefined + download.mockImplementationOnce(async function* (_session, _media, options) { + signal = options?.signal + await new Promise(resolve => signal?.addEventListener('abort', () => resolve(), { once: true })) + throw new Error('已取消') + }) + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await expect.poll(() => signal !== undefined).toBe(true) + storage.deactivate(session) + expect(signal?.aborted).toBe(true) + await storage.activate(session, platform) + await ready(storage) + expect(errors).not.toHaveBeenCalled() + expect(download).toHaveBeenCalledTimes(2) + }) + + it('超时终止下载且不计入已保存容量', async () => { + const { storage, platform, download, errors } = await setup() + download.mockImplementationOnce(async function* (_session, _media, options) { + yield bytes.subarray(0, 2) + await new Promise(resolve => { + if (options?.signal?.aborted) resolve() + else options?.signal?.addEventListener('abort', () => resolve(), { once: true }) + }) + throw new Error('下载超时') + }) + await storage.activate(session, platform) + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + try { + await storage.enqueue(session, group, message(), 1) + await vi.waitFor(() => expect(download).toHaveBeenCalledTimes(1)) + await vi.advanceTimersByTimeAsync(120001) + } finally { + vi.useRealTimers() + } + await expect.poll(() => errors.mock.calls.length).toBeGreaterThan(0) + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 1 }) + }) + + it('真实QQ直链查询永挂时，归档超时释放两槽并可关闭，迟到失败不产生未处理拒绝', async () => { + const { ctx, storage, path, platform } = await setup() + const lookup = Promise.withResolvers() + const fetch = vi.fn(() => lookup.promise) + const qq = new QQNTClient({ fetch }) + const qqPlatform = { downloadMedia: (_session: PlatformSession, item: IMMedia, options?: { signal?: AbortSignal }) => + qq.downloadFile(item.locator as Parameters[0], options) } as unknown as IMPlatform + const images = Array.from({ length: 3 }, (_, index) => ({ ...media, id: `qq-${index}`, locator: { + messageId: 'msg', chatType: 2, peerUid: 'g', fileUuid: `distinct-${index}`, + } })) + await storage.activate(session, qqPlatform) + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + try { + await storage.enqueue(session, group, message('qq-stalled', undefined, images), 1) + await vi.waitFor(() => expect(fetch).toHaveBeenCalledTimes(2)) + await vi.advanceTimersByTimeAsync(120001) + await vi.waitFor(() => expect(fetch).toHaveBeenCalledTimes(3)) + await vi.advanceTimersByTimeAsync(120001) + } finally { + vi.useRealTimers() + } + try { + await expect.poll(async () => (await ctx.database.get('mtproto_image_storage', { state: 'failed' })).length).toBe(3) + let timer: ReturnType | undefined + try { + await Promise.race([storage.close(), new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error('永挂QQ查询阻塞了归档关闭')), 1000) + })]) + } finally { + clearTimeout(timer) + } + for (const account of await readdir(path)) expect(await readdir(join(path, account))).toEqual([]) + expect(await ctx.database.get('mtproto_image_storage', { state: 'pending' })).toHaveLength(0) + } finally { + lookup.reject(new Error('迟到的QQ直链查询失败')) + await new Promise(resolve => setImmediate(resolve)) + } + const restored = new ImageStorage(ctx.database, { enabled: true, path }) + disposals.push(() => restored.close()) + await restored.activate(session, platform) + await ready(restored, 3) + }) + + it('按会话或日期删除，只清除指定副本并保留墓碑', async () => { + const { storage, platform, download } = await setup() + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await storage.enqueue(session, { ...group, id: 'other' }, message('two'), 2) + await ready(storage, 2) + expect(await storage.delete({ conversationId: group.id, startDate: '2026-10-15', endDate: '2026-10-15' })) + .toEqual({ count: 1, bytes: bytes.length }) + await storage.enqueue(session, group, message(), 1) + expect(await storage.query({})).toMatchObject({ count: 1, bytes: bytes.length }) + expect(await storage.read(session, 1, media, 0, 10)).toBeUndefined() + expect(await storage.read(session, 2, media, 0, 10)).toEqual(bytes) + expect(download).toHaveBeenCalledTimes(2) + }) + + it('文件缺失不返回成功；删除磁盘错误保留可重试状态和容量', async () => { + const { ctx, storage, path, platform } = await setup() + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await ready(storage) + const [account] = await readdir(path) + const [file] = await readdir(join(path, account)) + const location = join(path, account, file) + await rm(location) + await mkdir(location) + await writeFile(join(location, 'blocked'), 'x') + await expect(storage.delete({})).rejects.toThrow() + expect(await ctx.database.get('mtproto_image_storage', {})).toMatchObject([{ state: 'deleting', bytes: bytes.length }]) + await rm(location, { recursive: true }) + expect(await storage.delete({})).toEqual({ count: 1, bytes: bytes.length }) + await storage.enqueue(session, group, message(), 2) + await ready(storage) + const [remaining] = await readdir(join(path, account)) + await rm(join(path, account, remaining)) + expect(await storage.read(session, 2, media, 0, 10)).toBeUndefined() + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 1 }) + }) + + it('未启动下载的任务持久化，重启后按有界并发排空队列', async () => { + const { ctx, storage, path, platform, download } = await setup() + await storage.enqueue(session, group, message('queued', undefined, Array.from({ length: 7 }, (_, i) => ({ ...media, id: String(i) }))), 1) + expect(await storage.query({})).toMatchObject({ pendingCount: 7, count: 0 }) + await storage.close() + const restored = new ImageStorage(ctx.database, { enabled: true, path }) + disposals.push(() => restored.close()) + await restored.activate(session, platform) + await ready(restored, 7) + expect(await restored.query({})).toMatchObject({ count: 7, pendingCount: 0, bytes: bytes.length * 7 }) + expect(download).toHaveBeenCalledTimes(7) + }) + + it('原子改名后尚未提交元数据的完整文件可恢复，不再依赖上游', async () => { + const { ctx, storage, path, platform, download } = await setup() + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await ready(storage) + await storage.close() + await ctx.database.set('mtproto_image_storage', {}, { state: 'pending', bytes: 0 }) + download.mockImplementation(async function* () { throw new Error('上游已过期') }) + const restored = new ImageStorage(ctx.database, { enabled: true, path }) + disposals.push(() => restored.close()) + await restored.activate(session, platform) + await ready(restored) + expect(await restored.read(session, 1, media, 0, 100)).toEqual(bytes) + expect(download).toHaveBeenCalledTimes(1) + }) + + it('空下载不记成功，磁盘写入失败不会留下成功记录', async () => { + const { ctx, storage, platform, download, errors, path } = await setup() + download.mockImplementationOnce(async function* () {}) + await storage.activate(session, platform) + await storage.enqueue(session, group, message(), 1) + await expect.poll(() => errors.mock.calls.length).toBe(1) + expect(await storage.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 1 }) + await storage.close() + await rm(path, { recursive: true }) + await writeFile(path, '目录不可用') + const blocked = new ImageStorage(ctx.database, { enabled: true, path }, errors) + disposals.push(() => blocked.close()) + await blocked.activate(session, platform) + await expect.poll(() => errors.mock.calls.length).toBeGreaterThan(1) + expect(await blocked.query({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 1 }) + }) + + it('校验日期、分页、筛选、路径和读取范围', async () => { + const { ctx, storage } = await setup() + for (const query of [null, { startDate: '2026-02-30' }, { endDate: 'x' }, { startDate: '2026-10-16', endDate: '2026-10-15' }, { page: 0 }, { pageSize: 1000 }, { conversationId: {} }, { platformSessionId: '' }, { groupBy: 'all' }]) { + await expect(storage.query(query as any)).rejects.toThrow() + } + expect(() => new ImageStorage(ctx.database, { path: '' })).toThrow() + await expect(storage.read(session, 1, media, -1, 2)).rejects.toThrow() + }) +}) diff --git a/packages/bridge/src/image-storage.ts b/packages/bridge/src/image-storage.ts new file mode 100644 index 00000000..f6a1006a --- /dev/null +++ b/packages/bridge/src/image-storage.ts @@ -0,0 +1,319 @@ +import { $, type Database, type Query } from '@cordisjs/plugin-database' +import { createHash } from 'node:crypto' +import { createWriteStream } from 'node:fs' +import { mkdir, open, rename, rm, stat } from 'node:fs/promises' +import { resolve } from 'node:path' +import { Readable } from 'node:stream' +import { pipeline } from 'node:stream/promises' +import type { ImageStorageFilter, ImageStorageQuery, ImageStorageResult } from './dashboard-types.js' +import type { ImageStorageRow } from './models.js' +import type { IMConversation, IMMedia, IMMessage, IMPlatform, JsonValue, PlatformSession } from './platform.js' + +const TABLE = 'mtproto_image_storage' +const DOWNLOAD_CONCURRENCY = 2 +const DOWNLOAD_TIMEOUT_MS = 120_000 + +/** 文件身份独立于可重建的媒体行；删除后保留墓碑，重复推送不能重新归档。 */ +export class ImageStorage { + readonly enabled: boolean + private readonly _root: string + private readonly _sessions = new Map() + private readonly _downloads = new Map }>() + // ponytail: 仅串行化元数据与原子提交，下载不持锁；吞吐受限时再改为按账号锁。 + private _tail: Promise = Promise.resolve() + private _pumping = false + private _pumpAgain = false + private _closed = false + + constructor( + private readonly _database: Database, + config: { enabled?: boolean, path?: string } = {}, + private readonly _onError: (error: unknown) => void = () => {}, + ) { + this.enabled = config.enabled ?? false + const path = config.path ?? 'data/bridge-images' + if (typeof path !== 'string' || !path.trim() || path.includes('\0')) throw new Error('图片存储目录无效。') + this._root = resolve(path) + } + + async activate(session: PlatformSession, platform: IMPlatform): Promise { + if (!this.enabled || this._closed) return + this._sessions.set(session.platformSessionId, { session, platform }) + await this._exclusive(() => this._database.set(TABLE, { + platformId: session.platformId, platformSessionId: session.platformSessionId, state: 'failed', + }, { state: 'pending' })) + this._kick() + } + + deactivate(session: PlatformSession): void { + this._sessions.delete(session.platformSessionId) + for (const job of this._downloads.values()) { + if (job.sessionId === session.platformSessionId) job.abort.abort('deactivated') + } + } + + async enqueue(session: PlatformSession, conversation: IMConversation, message: IMMessage, messageId: number): Promise { + if (!this.enabled || this._closed) return + const images = message.content.parts.flatMap(part => part.type === 'media' && part.media.kind === 'image' + ? [part.media] : []) + if (!images.length) return + if (!Number.isSafeInteger(message.timestamp) || message.timestamp < 0) throw new Error('图片消息时间无效。') + const date = new Date(message.timestamp * 1000).toISOString().slice(0, 10) + await this._exclusive(async () => { + for (const media of images) { + const [original] = await this._database.get(TABLE, { + id: this._key(session.platformSessionId, messageId, media.id), + }) + if (original?.state === 'deleted' || original?.state === 'deleting') continue + const variants = [{ isPreview: false, item: media }] + if (media.preview?.locator != null) variants.push({ isPreview: true, item: { + id: `${media.id}:preview`, kind: 'image', mimeType: media.preview.mimeType, + size: media.preview.size, width: media.preview.width, height: media.preview.height, + locator: media.preview.locator, + } }) + for (const { isPreview, item } of variants) { + const id = this._key(session.platformSessionId, messageId, media.id, isPreview) + const [existing] = await this._database.get(TABLE, { id }) + if (existing) continue + await this._database.create(TABLE, { + id, platformId: session.platformId, platformSessionId: session.platformSessionId, + messageId, platformMediaId: media.id, isPreview, conversationId: conversation.id, + conversationTitle: conversation.title, date, timestamp: message.timestamp, + media: JSON.parse(JSON.stringify(item)) as JsonValue, state: 'pending', bytes: 0, + }) + } + } + }) + this._kick() + } + + async read(session: PlatformSession, messageId: number, media: IMMedia, offset: number, limit: number, isPreview = false): Promise { + if (!Number.isSafeInteger(offset) || offset < 0 || !Number.isSafeInteger(limit) || limit <= 0 || limit > 1024 * 1024) { + throw new Error('图片读取范围无效。') + } + const row = await this._ready(session, messageId, media, isPreview) + if (!row) return + let file + try { + file = await open(this._path(row), 'r') + const info = await file.stat() + if (!info.isFile() || info.size !== row.bytes) throw new Error('图片副本不完整。') + const bytes = new Uint8Array(Math.min(limit, Math.max(0, info.size - offset))) + const { bytesRead } = await file.read(bytes, 0, bytes.length, offset) + return bytes.subarray(0, bytesRead) + } catch (error) { + await this._exclusive(() => this._database.set(TABLE, { id: row.id, state: 'ready' }, { + state: 'failed', bytes: (error as NodeJS.ErrnoException).code === 'ENOENT' ? 0 : row.bytes, + })) + this._onError(error) + return + } finally { + await file?.close() + } + } + + async has(session: PlatformSession, messageId: number, media: IMMedia, isPreview = false): Promise { + return (await this.read(session, messageId, media, 0, 1, isPreview)) !== undefined + } + + async query(input: ImageStorageQuery): Promise { + const filter = this._filter(input) + const page = input.page ?? 1 + const pageSize = input.pageSize ?? 50 + if (!Number.isSafeInteger(page) || page < 1 || page > 1_000_000 + || !Number.isSafeInteger(pageSize) || pageSize < 1 || pageSize > 100 + || (input.groupBy !== undefined && input.groupBy !== 'date' && input.groupBy !== 'conversation')) { + throw new Error('图片存储分页或分组参数无效。') + } + const field = input.groupBy === 'date' ? 'date' : 'conversationId' + const aggregate = () => this._database.select(TABLE, filter).groupBy(['platformSessionId', field], row => ({ + conversationTitle: $.max(row.conversationTitle), + count: $.sum($.if($.and($.gt(row.bytes, 0), $.eq(row.isPreview, false)), 1, 0)), bytes: $.sum(row.bytes), + })) + const [totals, totalGroups, rows] = await Promise.all([ + this._database.select(TABLE, filter).groupBy([], row => ({ + count: $.sum($.if($.and($.gt(row.bytes, 0), $.eq(row.isPreview, false)), 1, 0)), bytes: $.sum(row.bytes), + pendingCount: $.sum($.if($.in(row.state, ['pending', 'failed']), 1, 0)), + })).execute(), + aggregate().execute(row => $.sum(1)), + aggregate().orderBy('platformSessionId').orderBy(field).offset((page - 1) * pageSize).limit(pageSize).execute(), + ]) + return { + count: Number(totals[0]?.count ?? 0), bytes: Number(totals[0]?.bytes ?? 0), + pendingCount: Number(totals[0]?.pendingCount ?? 0), totalGroups: Number(totalGroups ?? 0), page, pageSize, + groups: rows.map(row => ({ + platformSessionId: row.platformSessionId, + ...(field === 'date' ? { date: row.date } : { conversationId: row.conversationId, conversationTitle: row.conversationTitle }), + count: Number(row.count), bytes: Number(row.bytes), + })), + } + } + + async delete(input: ImageStorageFilter): Promise<{ count: number, bytes: number }> { + const filter = this._filter(input) + return this._exclusive(async () => { + let count = 0 + let bytes = 0 + // 逐批删，不把整个归档目录清单读进内存。 + while (true) { + const rows = await this._database.select(TABLE, filter).orderBy('id').limit(100).execute() + if (!rows.length) return { count, bytes } + for (const row of rows) { + await this._database.set(TABLE, { id: row.id }, { state: 'deleting' }) + this._downloads.get(row.id)?.abort.abort() + await rm(this._path(row), { force: true }) + await rm(`${this._path(row)}.part`, { force: true }) + await this._database.set(TABLE, { id: row.id }, { state: 'deleted', bytes: 0 }) + if (row.bytes > 0 && !row.isPreview) count++ + bytes += row.bytes + } + } + }) + } + + async close(): Promise { + this._closed = true + for (const job of this._downloads.values()) job.abort.abort() + await Promise.all([...this._downloads.values()].map(job => job.done)) + await this._tail + } + + private _filter(input: ImageStorageFilter): Query { + if (!input || typeof input !== 'object' || Array.isArray(input)) throw new Error('图片存储筛选参数无效。') + const filter: Query = { state: { $ne: 'deleted' } } + for (const field of ['platformSessionId', 'conversationId'] as const) { + const value = input[field] + if (value === undefined) continue + if (typeof value !== 'string' || !value || value.length > 4096 || value.includes('\0')) throw new Error('图片存储账号或会话无效。') + filter[field] = value + } + for (const field of ['startDate', 'endDate'] as const) { + const value = input[field] + if (value === undefined) continue + if (typeof value !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(value) + || !Number.isFinite(Date.parse(value)) || new Date(value).toISOString().slice(0, 10) !== value) { + throw new Error('日期必须是有效的 YYYY-MM-DD（UTC）。') + } + } + if (input.startDate && input.endDate && input.startDate > input.endDate) throw new Error('起始日期不能晚于结束日期。') + if (input.startDate || input.endDate) filter.date = { + ...(input.startDate ? { $gte: input.startDate } : {}), ...(input.endDate ? { $lte: input.endDate } : {}), + } + return filter + } + + private _key(sessionId: string, messageId: number, mediaId: string, isPreview = false): string { + const identity = isPreview ? [sessionId, messageId, mediaId, 'preview'] : [sessionId, messageId, mediaId] + return createHash('sha256').update(JSON.stringify(identity)).digest('hex') + } + + private _path(row: ImageStorageRow): string { + if (!/^[a-f0-9]{64}$/.test(row.id)) throw new Error('图片存储文件标识无效。') + const account = createHash('sha256').update(row.platformSessionId).digest('hex') + return resolve(this._root, account, row.id) + } + + private async _ready(session: PlatformSession, messageId: number, media: IMMedia, isPreview: boolean): Promise { + if (media.kind !== 'image') return + const [row] = await this._database.get(TABLE, { + id: this._key(session.platformSessionId, messageId, media.id, isPreview), + platformSessionId: session.platformSessionId, platformId: session.platformId, state: 'ready', + }) + return row + } + + private _exclusive(callback: () => Promise): Promise { + const next = this._tail.then(callback) + this._tail = next.catch(() => {}) + return next + } + + private _kick(): void { + if (this._closed || !this.enabled) return + if (this._pumping) { + this._pumpAgain = true + return + } + this._pumping = true + void this._pump().catch(this._onError).finally(() => { + this._pumping = false + if (this._pumpAgain) { + this._pumpAgain = false + this._kick() + } + }) + } + + private async _pump(): Promise { + for (const { session, platform } of this._sessions.values()) { + const slots = DOWNLOAD_CONCURRENCY - this._downloads.size + if (slots <= 0 || this._closed) return + if (!platform.downloadMedia) continue + const rows = await this._database.select(TABLE, { + platformId: session.platformId, platformSessionId: session.platformSessionId, + state: 'pending', id: { $nin: [...this._downloads.keys()] }, + }).orderBy('id').limit(slots).execute() + for (const row of rows) { + if (this._closed) return + const active = this._sessions.get(session.platformSessionId) + if (active?.session !== session || active.platform !== platform) break + const abort = new AbortController() + const done = this._download(row, session, platform, abort).catch(this._onError).finally(() => { + this._downloads.delete(row.id) + this._kick() + }) + this._downloads.set(row.id, { sessionId: session.platformSessionId, abort, done }) + } + } + } + + private async _download(row: ImageStorageRow, session: PlatformSession, platform: IMPlatform, abort: AbortController): Promise { + const path = this._path(row) + const temp = `${path}.part` + let renamed = false + const timer = setTimeout(() => abort.abort(new Error('图片下载超时。')), DOWNLOAD_TIMEOUT_MS) + timer.unref?.() + try { + await mkdir(resolve(path, '..'), { recursive: true, mode: 0o700 }) + await rm(temp, { force: true }) + // 原子改名后、元数据提交前崩溃的文件已经完整，可直接恢复，无需再次回源。 + const recovered = await stat(path).catch((error: NodeJS.ErrnoException) => { + if (error.code === 'ENOENT') return undefined + throw error + }) + if (recovered?.isFile() && recovered.size > 0 && (!row.bytes || recovered.size === row.bytes)) { + await this._exclusive(async () => { + if (!abort.signal.aborted && !this._closed) await this._database.set(TABLE, { + id: row.id, state: 'pending', + }, { state: 'ready', bytes: recovered.size }) + }) + return + } + await pipeline( + Readable.from(platform.downloadMedia!(session, row.media as unknown as IMMedia, { signal: abort.signal })), + createWriteStream(temp, { flags: 'wx', mode: 0o600 }), { signal: abort.signal }, + ) + const { size } = await stat(temp) + if (!size) throw new Error('图片下载结果为空。') + await this._exclusive(async () => { + const [current] = await this._database.get(TABLE, { id: row.id, state: 'pending' }) + if (!current || abort.signal.aborted || this._closed) return + await rename(temp, path) + renamed = true + await this._database.set(TABLE, { id: row.id, state: 'pending' }, { state: 'ready', bytes: size }) + renamed = false + }) + } catch (error) { + if (renamed) await rm(path, { force: true }) + if (!this._closed) { + await this._exclusive(() => this._database.set(TABLE, { id: row.id, state: 'pending' }, { + state: abort.signal.reason === 'deactivated' ? 'pending' : 'failed', + })) + if (abort.signal.reason !== 'deactivated') this._onError(error) + } + } finally { + clearTimeout(timer) + await rm(temp, { force: true }) + } + } +} diff --git a/packages/bridge/src/index.ts b/packages/bridge/src/index.ts index f29f4e35..4383c5dd 100644 --- a/packages/bridge/src/index.ts +++ b/packages/bridge/src/index.ts @@ -39,7 +39,6 @@ import { DraftStore } from './draft-store.js' import { NotificationSettingsStore } from './notification-settings.js' import { makeCrossGramServerConfig, makePlatformAccountView, makeUnavailableAccountView, - type PlatformAccountDashboardData, } from './account-dashboard.js' import { AuthTransferStore } from './auth-transfer.js' import { @@ -49,7 +48,7 @@ import { BlockedPeerStore, type BlockedContentMode } from './blocked-peers.js' import { DialogFolderStore } from './dialog-folders.js' import { collectStickerDashboard, setStickerPackAssignment, - type StickerDashboardPack, type StickerDashboardSourceAccount, type StickerPackDashboardData, + type StickerDashboardPack, type StickerDashboardSourceAccount, } from './sticker-dashboard.js' import { CallRegistry, type VoiceMediaStartProvider, type VoiceWorkerClient } from './voice/call-registry.js' import type { VoiceCallMediaProvider } from './voice/media.js' @@ -58,7 +57,6 @@ import { VoiceWorkerSocketClient } from './voice/voice-worker-client.js' import { VoiceRpc } from './voice/voice-rpc.js' import { SpeechPipeline } from './speech.js' import { SystemPeerCallbackError, SystemPeerService } from './system-peer.js' -import type { BotDashboardData } from './bot-dashboard.js' import { createRequestResolver, RequestInboxSystemPeerProvider } from './request-inbox.js' import { ActiveSessionStore, createAuthorizationReservationQueue, registerActiveSessionRpc, @@ -67,6 +65,8 @@ import { MtprotoBridgeService, type BridgeSessionState } from './bridge-service. import { BridgeManagementError, BridgeManagementService } from './management-service.js' import { registerGroupFilesMiniApp } from './group-files-miniapp.js' import { MessageProjectionPipeline } from './message-projection.js' +import { ImageStorage } from './image-storage.js' +import type { BridgeDashboardData } from './dashboard-types.js' export * from './platform.js' export { defineModels } from './models.js' @@ -77,6 +77,7 @@ export { export * from './message-projection.js' export * from './detached-message-projection.js' export * from './message-store.js' +export * from './image-storage.js' export * from './message-actions.js' export * from './platform-manager.js' export * from './upload-manager.js' @@ -130,6 +131,8 @@ export interface BridgeConfig { tokenTtlSeconds?: number } uploadPath?: string + /** 新消息图片的持久副本；关闭后已有副本仍可读取和管理。 */ + imageStorage?: { enabled?: boolean, path?: string, adminUserIds?: number[] } autoMuteGroupChats?: boolean blockedContentMode?: BlockedContentMode recalledMessageMode?: 'hide' | 'show' @@ -178,6 +181,17 @@ export const Config = z.object({ tokenTtlSeconds: 600, }), uploadPath: z.string().default('data/bridge-uploads'), + imageStorage: z.object({ + enabled: z.boolean().default(false).i18n({ + 'en-US': enUS['imageStorage.enabled'], 'zh-CN': zhCN['imageStorage.enabled'], + }), + path: z.string().default('data/bridge-images').i18n({ + 'en-US': enUS['imageStorage.path'], 'zh-CN': zhCN['imageStorage.path'], + }), + adminUserIds: z.array(z.natural().min(1)).default([]).i18n({ + 'en-US': enUS['imageStorage.adminUserIds'], 'zh-CN': zhCN['imageStorage.adminUserIds'], + }), + }).default({ enabled: false, path: 'data/bridge-images', adminUserIds: [] }), autoMuteGroupChats: z.boolean().default(true), blockedContentMode: z.union([ z.const('show'), z.const('hide-user'), z.const('hide-related'), @@ -238,6 +252,19 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { authKeyId => ctx.mtproto.finishAuthKeyRevocation(authKeyId), ) const store = new MessageStore(ctx.database, undefined, ctx.updateStore, historyTrace, messageProjection) + const imageStorage = new ImageStorage(ctx.database, config.imageStorage, error => bridgeLogger.error( + '图片存储操作失败：%s', String(error).replace(/https?:\/\/\S+/gi, '[已隐藏 URL]'), + )) + const stopImageStorage = platforms.onCommittedEvent((session, committed) => { + if (committed.event.type !== 'message' && committed.event.type !== 'message-edit') return + const result = committed as Extract + return imageStorage.enqueue(session, committed.event.conversation, committed.event.message, result.result.message.id) + .catch(error => bridgeLogger.error('图片归档排队失败：%s', String(error))) + }) + ctx.effect(() => () => { + stopImageStorage() + return imageStorage.close() + }, 'mtproto-bridge.image-storage') const drafts = new DraftStore(ctx.database) const notificationSettings = new NotificationSettingsStore( ctx.database, config.autoMuteGroupChats ?? true, @@ -413,12 +440,16 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { ctx.effect(() => unregisterRequestInbox, 'mtproto-bridge.request-inbox') platforms.onSessionChange((event, binding) => { if (event === 'deactivate') { + imageStorage.deactivate(binding.session) const key = `${binding.session.platformId}\u0000${binding.session.platformSessionId}` stickerRpcs.delete(key) reactionRpcs.delete(key) stickerProviders.releaseSession(binding.session.platformSessionId) return } + void imageStorage.activate(binding.session, binding.platform).catch(error => { + bridgeLogger.error('图片存储恢复失败：%s', String(error)) + }) void systemPeers.bootstrap(binding.session).catch((error) => { bridgeLogger.warn('system peer bootstrap failed: %s', String(error)) }) @@ -437,6 +468,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { updates.requestRecovery(rpcContext) }, messageProjection, + imageStorage, ) new MtprotoBridgeService(ctx, requireBridgeSession, dcId) registerGroupFilesMiniApp(ctx, platforms, requireBridgeSession, config.groupFilesMiniApp) @@ -528,7 +560,28 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { const provisionedAccounts = new Map() const accountErrors = new Map() let publishedStickerPacks: StickerDashboardPack[] = [] - const dashboard: PlatformAccountDashboardData & StickerPackDashboardData & BotDashboardData = { + const requireImageStorageAdmin = async (token: unknown) => { + const adminUserIds = config.imageStorage?.adminUserIds ?? [] + if (!adminUserIds.length) throw new Error('请先配置 imageStorage.adminUserIds，指定可管理图片存储的 SSO 用户 ID。') + const sso = ctx.get('sso') + if (!sso) throw new Error('图片存储管理需要启用 SSO 登录服务。') + if (typeof token !== 'string' || !token || token.length > 4096) { + throw new Error('请先在 Your account 页面登录后管理图片存储。') + } + const user = await sso.validateSession(token) + if (!user) throw new Error('登录已失效，请在 Your account 页面重新登录。') + if (!adminUserIds.includes(user.id)) throw new Error('当前用户没有图片存储管理权限。') + } + const dashboard: BridgeDashboardData = { + imageStorageEnabled: imageStorage.enabled, + async getImageStorage(query, token) { + await requireImageStorageAdmin(token) + return imageStorage.query(query) + }, + async deleteImageStorage(filter, token) { + await requireImageStorageAdmin(token) + return imageStorage.delete(filter) + }, accounts: [], serverConfig: makeCrossGramServerConfig( config.serverHost ?? '127.0.0.1', config.serverPort ?? 4430, ctx.mtproto.rsaKey.publicKeyPem, @@ -578,7 +631,7 @@ export function apply(ctx: Context, config: BridgeConfig = {}): void { baseUrl: import.meta.url, source: '../client/index.ts', manifest: '../dist/manifest.json', - routes: ['/platform-accounts', '/sticker-packs', '/bots'], + routes: ['/platform-accounts', '/sticker-packs', '/bots', '/image-storage'], }, dashboard) const currentAccounts = (now = Date.now()) => registry.ids.sort().map((platformId) => { @@ -1487,6 +1540,7 @@ export function createSessionResolver( rpc: ServerRpcContext, ) => void | Promise, messageProjection?: MessageProjectionPipeline, + imageStorage?: ImageStorage, ) { const projection = messageProjection ?? new MessageProjectionPipeline(ctx) const loading = new Map>() @@ -1570,6 +1624,7 @@ export function createSessionResolver( dialogFolders, systemPeers, projection, + imageStorage, ) return state } diff --git a/packages/bridge/src/locales/en-US.yml b/packages/bridge/src/locales/en-US.yml index 644a9034..b780915e 100644 --- a/packages/bridge/src/locales/en-US.yml +++ b/packages/bridge/src/locales/en-US.yml @@ -5,6 +5,10 @@ altEndpoints: Alternate MTProto endpoints for the same data center, as host:port apiPrefix: HTTP prefix used for platform account assets. groupFilesMiniApp: Telegram attachment-menu Mini App used to browse platform-native group files. uploadPath: Directory used to store uploads received from Telegram clients. +imageStorage: Persistent server copies of new message images; existing copies remain readable and manageable when disabled. +imageStorage.enabled: Save images from new group and private messages automatically; disabled by default, without historical backfill. +imageStorage.path: Image copy directory, default data/bridge-images; back up together with the database. +imageStorage.adminUserIds: SSO user IDs allowed to manage image copies across all accounts; empty by default, denying all management requests. autoMuteGroupChats: Mute group chats by default; per-chat Telegram settings can override this. blockedContentMode: "Blocked-user visibility: show keeps Telegram's default group behavior; hide-user hides their messages and reactions; hide-related also hides messages replying to or mentioning them." recalledMessageMode: "QQ recalled-message visibility: show keeps the message with a recalled marker; hide removes it from visible history." diff --git a/packages/bridge/src/locales/zh-CN.yml b/packages/bridge/src/locales/zh-CN.yml index 6050eded..89b8ef1a 100644 --- a/packages/bridge/src/locales/zh-CN.yml +++ b/packages/bridge/src/locales/zh-CN.yml @@ -5,6 +5,10 @@ altEndpoints: 在同一数据中心中、紧随主地址向 Telegram 客户端 apiPrefix: 平台账户资源使用的 HTTP 路径前缀。 groupFilesMiniApp: 用于浏览平台原生群文件的 Telegram 附件菜单小程序。 uploadPath: 用于存储 Telegram 客户端上传文件的目录。 +imageStorage: 新消息图片的服务端持久存储；关闭后已有副本仍可读取和管理。 +imageStorage.enabled: 主动保存群聊和私聊新消息中的图片，默认关闭，不回填历史图片。 +imageStorage.path: 图片副本目录，默认 data/bridge-images；应与数据库一起备份。 +imageStorage.adminUserIds: 允许管理所有账号图片副本的 SSO 用户 ID 白名单；默认空列表，拒绝所有管理请求。 autoMuteGroupChats: 默认关闭群聊通知；可在 Telegram 中为单个群聊单独开启。 blockedContentMode: 拉黑用户内容显示策略：show 保留 Telegram 默认群聊显示；hide-user 隐藏其消息和 reaction；hide-related 还会隐藏回复或 @ 该用户的消息。 recalledMessageMode: QQ 撤回消息显示策略：show 保留消息并标记为已撤回；hide 直接隐藏消息。 diff --git a/packages/bridge/src/message-store.ts b/packages/bridge/src/message-store.ts index e97395f9..39848539 100644 --- a/packages/bridge/src/message-store.ts +++ b/packages/bridge/src/message-store.ts @@ -68,6 +68,7 @@ export interface ProjectedMessage { } export interface StoredMedia { + messageId: number media: import('./platform.js').IMMedia timestamp: number } @@ -1291,6 +1292,7 @@ export class MessageStore { }) if (!message) return return { + messageId: message.id, media: { id: row.platformMediaId, kind: row.kind, diff --git a/packages/bridge/src/models.ts b/packages/bridge/src/models.ts index 83244864..fe589884 100644 --- a/packages/bridge/src/models.ts +++ b/packages/bridge/src/models.ts @@ -152,6 +152,22 @@ export interface IMMediaRow { locator: JsonValue } +export interface ImageStorageRow { + id: string + platformId: string + platformSessionId: string + messageId: number + platformMediaId: string + isPreview: boolean + conversationId: string + conversationTitle: string + date: string + timestamp: number + media: JsonValue + state: 'pending' | 'ready' | 'failed' | 'deleting' | 'deleted' + bytes: number +} + export interface TlMessagePartRow { id: number platformSessionId: string @@ -301,6 +317,7 @@ declare module '@cordisjs/plugin-database' { mtproto_im_message_alias: IMMessageAliasRow mtproto_message_mention: MessageMentionRow mtproto_im_media: IMMediaRow + mtproto_image_storage: ImageStorageRow mtproto_tl_message_part: TlMessagePartRow mtproto_id_counter: IdCounterRow mtproto_message_id_epoch: MessageIdEpochRow @@ -420,6 +437,20 @@ export function defineModels(ctx: Context): void { indexes: ['messageId'], }) + ctx.model.extend('mtproto_image_storage', { + id: 'string', platformId: 'string', platformSessionId: 'string', messageId: 'unsigned', + platformMediaId: 'text', isPreview: { type: 'boolean', initial: false }, + conversationId: 'text', conversationTitle: 'text', date: 'string', + timestamp: 'integer', media: 'json', state: 'string', + bytes: { type: 'decimal', precision: 20, scale: 0 }, + }, { + primary: 'id', + indexes: [ + ['platformSessionId', 'messageId'], ['platformSessionId', 'state'], + ['platformSessionId', 'conversationId', 'date'], ['date', 'state'], + ], + }) + ctx.model.extend('mtproto_tl_message_part', { id: 'unsigned', platformSessionId: 'string', conversationId: 'unsigned', messageId: 'unsigned', mediaId: { type: 'unsigned', nullable: true }, scope: 'string', diff --git a/packages/cordis-webui-solidjs/src/bridge.browser.e2e.test.ts b/packages/cordis-webui-solidjs/src/bridge.browser.e2e.test.ts index 1d502a38..5c0481b7 100644 --- a/packages/cordis-webui-solidjs/src/bridge.browser.e2e.test.ts +++ b/packages/cordis-webui-solidjs/src/bridge.browser.e2e.test.ts @@ -131,6 +131,13 @@ describe('Crossgram accounts, stickers and bots in the Solid shell', () => { sourcePlugin: 'test-provider', })), botUpdatedAt: Date.now(), + imageStorageEnabled: false, + async getImageStorage() { + return { count: 0, bytes: 0, pendingCount: 0, groups: [], totalGroups: 0, page: 1, pageSize: 25 } + }, + async deleteImageStorage() { + return { count: 0, bytes: 0 } + }, async refresh() { entry.mutate((value) => { value.accounts = value.accounts.map((account) => ({ ...account })) diff --git a/packages/cordis-webui-solidjs/src/startup.browser.e2e.test.ts b/packages/cordis-webui-solidjs/src/startup.browser.e2e.test.ts index afef3f72..404b7267 100644 --- a/packages/cordis-webui-solidjs/src/startup.browser.e2e.test.ts +++ b/packages/cordis-webui-solidjs/src/startup.browser.e2e.test.ts @@ -29,7 +29,7 @@ describe('actual Cordis CLI startup', () => { JSON.stringify( pathToFileURL(join(root, 'packages/bridge/src/index.ts')).href, ) + - ",manifest:'../dist/manifest.json',routes:['/platform-accounts','/sticker-packs','/bots']},{accounts:[],serverConfig:{name:'CrossGram',host:'example.test',port:4430,dcs:[],rsa_key:'PUBLIC'},stickerAccounts:[],stickerPacks:[],bots:[],refresh:async()=>{}}); await entry.ready; await writeFile(" + + ",manifest:'../dist/manifest.json',routes:['/platform-accounts','/sticker-packs','/bots','/image-storage']},{accounts:[],serverConfig:{name:'CrossGram',host:'example.test',port:4430,dcs:[],rsa_key:'PUBLIC'},stickerAccounts:[],stickerPacks:[],bots:[],refresh:async()=>{}}); await entry.ready; await writeFile(" + JSON.stringify(marker) + ',JSON.stringify({baseUrl:ctx.server.baseUrl,buildId:ctx.webui.buildId,module:entry.toJSON()}));}', 'utf8', @@ -91,6 +91,7 @@ describe('actual Cordis CLI startup', () => { '/platform-accounts', '/sticker-packs', '/bots', + '/image-storage', ]) browser = await chromium.launch({ executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH, diff --git a/packages/cordis-webui-solidjs/src/storage.browser.e2e.test.ts b/packages/cordis-webui-solidjs/src/storage.browser.e2e.test.ts new file mode 100644 index 00000000..3aee2212 --- /dev/null +++ b/packages/cordis-webui-solidjs/src/storage.browser.e2e.test.ts @@ -0,0 +1,504 @@ +import { Context } from 'cordis' +import Server from '@cordisjs/plugin-server' +import { chromium } from 'playwright' +import { describe, expect, it, vi } from 'vitest' +import type { + ImageStorageDashboardData, + ImageStorageFilter, + ImageStorageQuery, + ImageStorageGroup, + StickerDashboardAccount, +} from '../../bridge/src/dashboard-types.js' +import { openPage, waitForConnection } from './browser-test-utils.js' +import SolidWebUI from './index.js' + +describe('Image storage management in the Solid shell', () => { + it('filters and paginates UTC summaries, confirms scoped deletion, and handles busy, error and empty states on a phone', async () => { + const ctx = new Context(), + fibers = [ + ctx.plugin(Server, { host: '127.0.0.1', port: 0 }), + ctx.plugin(SolidWebUI, { uiPath: '/console' }), + ] + let browser: Awaited> | undefined + try { + await Promise.all(fibers) + await vi.waitFor(() => + expect(fibers.every((fiber) => fiber.state === 2)).toBe(true), + ) + const ui = ctx.get('webui') as unknown as SolidWebUI, + queries: ImageStorageQuery[] = [], + deletions: ImageStorageFilter[] = [], + queryTokens: (string | undefined)[] = [], + deleteTokens: (string | undefined)[] = [], + adminToken = 'image-storage-admin-session', + accounts: StickerDashboardAccount[] = [ + { + platformId: 'qq', + platformSessionId: 'first', + platformKind: 'QQ', + displayName: 'First account', + userId: '123', + }, + { + platformId: 'qq', + platformSessionId: 'second', + platformKind: 'QQ', + displayName: 'Second account', + userId: '456', + }, + ] + let records = [ + ...Array.from({ length: 26 }, (_, id) => ({ + platformSessionId: 'first', + conversationId: 'group-' + id, + conversationTitle: 'Group ' + id, + date: id === 0 ? '2026-10-01' : '2026-10-02', + count: 1, + bytes: 1024, + })), + { + platformSessionId: 'second', + conversationId: 'private-1', + conversationTitle: 'Private conversation', + date: '2026-10-02', + count: 2, + bytes: 2048, + }, + ], + failQuery = false, + failDelete = false, + adminConfigured = false, + releaseDelete: (() => void) | undefined + const matches = ( + row: (typeof records)[number], + filter: ImageStorageFilter, + ) => + (!filter.platformSessionId || + row.platformSessionId === filter.platformSessionId) && + (!filter.conversationId || + row.conversationId === filter.conversationId) && + (!filter.startDate || row.date >= filter.startDate) && + (!filter.endDate || row.date <= filter.endDate) + const authorize = (token?: string) => { + if (!token) + throw new Error( + 'Sign in through Your account to manage image storage.', + ) + if (token !== adminToken) + throw new Error('You do not have permission to manage image storage.') + if (!adminConfigured) + throw new Error( + 'Configure imageStorage.adminUserIds before managing image storage.', + ) + } + const data: ImageStorageDashboardData & { + stickerAccounts: StickerDashboardAccount[] + } = { + imageStorageEnabled: false, + stickerAccounts: accounts, + async getImageStorage(query, token) { + queryTokens.push(token) + authorize(token) + queries.push(query) + if (failQuery) throw new Error('Storage query unavailable') + const rows = records.filter((row) => matches(row, query)), + groups = new Map() + for (const row of rows) { + const key = + row.platformSessionId + + '/' + + (query.groupBy === 'date' ? row.date : row.conversationId), + group = groups.get(key) ?? { + platformSessionId: row.platformSessionId, + ...(query.groupBy === 'date' + ? { date: row.date } + : { + conversationId: row.conversationId, + conversationTitle: row.conversationTitle, + }), + count: 0, + bytes: 0, + } + group.count += row.count + group.bytes += row.bytes + groups.set(key, group) + } + const page = query.page ?? 1, + pageSize = query.pageSize ?? 25 + return { + count: rows.reduce((sum, row) => sum + row.count, 0), + bytes: rows.reduce((sum, row) => sum + row.bytes, 0), + pendingCount: 0, + groups: [...groups.values()].slice( + (page - 1) * pageSize, + page * pageSize, + ), + totalGroups: groups.size, + page, + pageSize, + } + }, + async deleteImageStorage(filter, token) { + deleteTokens.push(token) + authorize(token) + deletions.push(filter) + if (failDelete) throw new Error('Storage deletion failed') + await new Promise((resolve) => { + releaseDelete = resolve + }) + const deleted = records.filter((row) => matches(row, filter)) + records = records.filter((row) => !matches(row, filter)) + return { + count: deleted.reduce((sum, row) => sum + row.count, 0), + bytes: deleted.reduce((sum, row) => sum + row.bytes, 0), + } + }, + } + const entry = ui.addEntry( + { + baseUrl: new URL('../../bridge/src/index.ts', import.meta.url).href, + manifest: '../dist/manifest.json', + routes: ['/image-storage'], + }, + data, + ) + await entry.ready + browser = await chromium.launch({ + executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH, + }) + const context = await browser.newContext({ + viewport: { width: 390, height: 844 }, + }), + page = await context.newPage(), + errors: string[] = [] + page.on('pageerror', (error) => errors.push(error.message)) + await page.goto(ctx.server.baseUrl + '/console/') + await waitForConnection(page) + await openPage(page, 'Image storage') + await page + .getByRole('alert') + .getByText( + 'Error: Sign in through Your account to manage image storage.', + { exact: true }, + ) + .waitFor() + expect(queryTokens[0]).toBeFalsy() + expect(queries).toEqual([]) + await page.evaluate(() => + localStorage.setItem('cordis:webui-sso:token', 'member-session'), + ) + await page.getByRole('button', { name: 'Refresh', exact: true }).click() + await page + .getByRole('alert') + .getByText( + 'Error: You do not have permission to manage image storage.', + { exact: true }, + ) + .waitFor() + await page.evaluate( + (token) => localStorage.setItem('cordis:webui-sso:token', token), + adminToken, + ) + await page.getByRole('button', { name: 'Refresh', exact: true }).click() + await page + .getByRole('alert') + .getByText( + 'Error: Configure imageStorage.adminUserIds before managing image storage.', + { exact: true }, + ) + .waitFor() + adminConfigured = true + await page.getByRole('button', { name: 'Refresh', exact: true }).click() + await page.getByText('28 images · 28 KiB').waitFor() + expect(queryTokens.at(-1)).toBe(adminToken) + expect( + await page.getByText('Disabled (default)', { exact: true }).isVisible(), + ).toBe(true) + expect(await page.locator('tbody tr').count()).toBe(25) + expect(queries.at(-1)).toEqual({ + groupBy: 'conversation', + page: 1, + pageSize: 25, + }) + const otherTab = await page.context().newPage() + await otherTab.goto(ctx.server.baseUrl + '/console/') + await otherTab.evaluate(() => + localStorage.removeItem('cordis:webui-sso:token'), + ) + await page + .getByRole('alert') + .getByText( + 'Error: Sign in through Your account to manage image storage.', + { exact: true }, + ) + .waitFor() + expect(await page.locator('tbody tr').count()).toBe(0) + expect( + await page + .getByRole('button', { name: 'Delete filtered copies', exact: true }) + .count(), + ).toBe(0) + await otherTab.evaluate(() => + localStorage.setItem('cordis:webui-sso:token', 'member-session'), + ) + await page + .getByRole('alert') + .getByText( + 'Error: You do not have permission to manage image storage.', + { exact: true }, + ) + .waitFor() + await otherTab.evaluate( + (token) => localStorage.setItem('cordis:webui-sso:token', token), + adminToken, + ) + await page.getByText('28 images · 28 KiB').waitFor() + await otherTab.close() + await page.getByRole('button', { name: 'Next page', exact: true }).click() + await page.getByText('Page 2 of 2 · 27 groups', { exact: true }).waitFor() + expect(await page.locator('tbody tr').count()).toBe(2) + expect(queries.at(-1)?.page).toBe(2) + await page.getByLabel('Group by', { exact: true }).selectOption('date') + await page.getByText('Page 1 of 1 · 3 groups', { exact: true }).waitFor() + expect(queries.at(-1)).toMatchObject({ groupBy: 'date', page: 1 }) + await page.getByLabel('Account', { exact: true }).selectOption('second') + await page.getByText('2 images · 2 KiB').waitFor() + await page + .getByLabel('Conversation ID (group or private)', { exact: true }) + .fill('private-1') + await page + .getByLabel('Conversation ID (group or private)', { exact: true }) + .press('Tab') + await expect.poll(() => queries.at(-1)?.conversationId).toBe('private-1') + await page + .getByLabel('Start date (UTC)', { exact: true }) + .fill('2026-10-02') + await expect.poll(() => queries.at(-1)?.startDate).toBe('2026-10-02') + await page + .getByLabel('End date (UTC, inclusive)', { exact: true }) + .fill('2026-10-02') + await expect.poll(() => queries.at(-1)?.endDate).toBe('2026-10-02') + const selected = { + platformSessionId: 'second', + conversationId: 'private-1', + startDate: '2026-10-02', + endDate: '2026-10-02', + } + expect(queries.at(-1)).toEqual({ + ...selected, + groupBy: 'date', + page: 1, + pageSize: 25, + }) + await page + .getByRole('button', { name: 'Delete filtered copies', exact: true }) + .click() + const dialog = page.getByRole('dialog', { + name: 'Delete server image copies?', + }) + expect(await dialog.textContent()).toContain( + 'Account: Second account (second)', + ) + expect(await dialog.textContent()).toContain('Conversation: private-1') + expect(await dialog.textContent()).toContain( + '2026-10-02 to 2026-10-02 (UTC, inclusive)', + ) + expect(await dialog.textContent()).toContain( + '2 images · 2 KiB (2,048 bytes)', + ) + await dialog.getByRole('button', { name: 'Cancel', exact: true }).click() + expect(deletions).toEqual([]) + expect(deleteTokens).toEqual([]) + await page + .getByRole('button', { + name: 'Delete copies for Second account / 2026-10-02', + exact: true, + }) + .click() + await page.evaluate(() => + localStorage.removeItem('cordis:webui-sso:token'), + ) + await dialog + .getByRole('button', { name: 'Delete copies', exact: true }) + .click() + await dialog + .getByRole('alert') + .getByText('Sign in through Your account to manage image storage.', { + exact: true, + }) + .waitFor() + expect(deletions).toEqual([]) + expect(deleteTokens[0]).toBeFalsy() + await page.evaluate( + (token) => localStorage.setItem('cordis:webui-sso:token', token), + adminToken, + ) + failDelete = true + await dialog + .getByRole('button', { name: 'Delete copies', exact: true }) + .click() + await dialog + .getByRole('alert') + .getByText('Storage deletion failed', { exact: true }) + .waitFor() + expect(deletions).toEqual([selected]) + failDelete = false + await dialog + .getByRole('button', { name: 'Delete copies', exact: true }) + .click() + await expect.poll(() => !!releaseDelete).toBe(true) + expect( + await dialog + .getByRole('button', { name: 'Cancel', exact: true }) + .isDisabled(), + ).toBe(true) + expect( + await page.getByLabel('Account', { exact: true }).isDisabled(), + ).toBe(true) + releaseDelete!() + await page + .getByRole('heading', { name: 'No stored images in this range' }) + .waitFor() + expect(await page.getByText('0 images · 0 B').isVisible()).toBe(true) + expect( + await page + .getByRole('button', { name: 'Delete filtered copies', exact: true }) + .isDisabled(), + ).toBe(true) + expect(records).toHaveLength(26) + expect( + records.some( + (row) => + row.conversationId === 'group-0' && row.date === '2026-10-01', + ), + ).toBe(true) + + // 改回其他账号后仍能看到未选范围的副本，并验证失败后可重新查询。 + await page.getByLabel('Account', { exact: true }).selectOption('first') + await expect.poll(() => queries.at(-1)?.platformSessionId).toBe('first') + await page + .getByLabel('Conversation ID (group or private)', { exact: true }) + .fill('') + await page + .getByLabel('Conversation ID (group or private)', { exact: true }) + .press('Tab') + await page.getByText('25 images · 25 KiB').waitFor() + failQuery = true + await page.getByRole('button', { name: 'Refresh', exact: true }).click() + await page + .getByRole('alert') + .getByText('Error: Storage query unavailable', { exact: true }) + .waitFor() + expect( + await page + .getByRole('button', { name: 'Delete filtered copies', exact: true }) + .count(), + ).toBe(0) + failQuery = false + await page.getByRole('button', { name: 'Refresh', exact: true }).click() + await page.getByText('25 images · 25 KiB').waitFor() + expect( + await page.evaluate( + () => document.documentElement.scrollWidth <= innerWidth, + ), + ).toBe(true) + await page + .getByLabel('Group by', { exact: true }) + .selectOption('conversation') + await page.getByText('Page 1 of 1 · 25 groups', { exact: true }).waitFor() + await page + .getByRole('button', { + name: 'Delete copies for First account / Group 1', + exact: true, + }) + .click() + expect(await dialog.textContent()).toContain('Conversation: group-1') + expect(await dialog.textContent()).toContain( + '1 images · 1 KiB (1,024 bytes)', + ) + releaseDelete = undefined + await dialog + .getByRole('button', { name: 'Delete copies', exact: true }) + .click() + await expect.poll(() => !!releaseDelete).toBe(true) + releaseDelete!() + await page.getByText('24 images · 24 KiB').waitFor() + expect(deletions.at(-1)).toEqual({ + platformSessionId: 'first', + conversationId: 'group-1', + startDate: '2026-10-02', + endDate: '2026-10-02', + }) + await page + .getByRole('button', { name: 'Delete filtered copies', exact: true }) + .click() + expect(await dialog.textContent()).toContain( + 'All group and private conversations', + ) + expect(await dialog.textContent()).toContain( + '24 images · 24 KiB (24,576 bytes)', + ) + releaseDelete = undefined + await dialog + .getByRole('button', { name: 'Delete copies', exact: true }) + .click() + await expect.poll(() => !!releaseDelete).toBe(true) + releaseDelete!() + await page + .getByRole('heading', { name: 'No stored images in this range' }) + .waitFor() + expect(deletions.at(-1)).toEqual({ + platformSessionId: 'first', + startDate: '2026-10-02', + endDate: '2026-10-02', + }) + expect(records).toHaveLength(1) + expect(records[0]).toMatchObject({ + platformSessionId: 'first', + conversationId: 'group-0', + date: '2026-10-01', + }) + // 原图已删而预览删除失败时，仅剩字节占用也必须能重试范围删除。 + records.push({ + platformSessionId: 'first', conversationId: 'preview-retry', + conversationTitle: 'Preview retry', date: '2026-10-02', count: 0, bytes: 200, + }) + await page.getByRole('button', { name: 'Refresh', exact: true }).click() + await page.getByText('0 images · 200 B').waitFor() + const retryDelete = page.getByRole('button', { name: 'Delete filtered copies', exact: true }) + expect(await retryDelete.isEnabled()).toBe(true) + await retryDelete.click() + expect(await dialog.textContent()).toContain('0 images · 200 B (200 bytes)') + releaseDelete = undefined + await dialog.getByRole('button', { name: 'Delete copies', exact: true }).click() + await expect.poll(() => !!releaseDelete).toBe(true) + releaseDelete!() + await page.getByRole('heading', { name: 'No stored images in this range' }).waitFor() + expect(records).toHaveLength(1) + const beforeInvalidRange = queries.length + await page + .getByLabel('Start date (UTC)', { exact: true }) + .fill('2026-10-03') + await page + .getByRole('alert') + .getByText('Start date must be on or before end date.', { exact: true }) + .waitFor() + expect(queries).toHaveLength(beforeInvalidRange) + expect( + await page + .getByRole('button', { name: 'Refresh', exact: true }) + .isDisabled(), + ).toBe(true) + expect(queryTokens).toContain('member-session') + expect(queryTokens.at(-1)).toBe(adminToken) + expect(deleteTokens.slice(1).every((token) => token === adminToken)).toBe( + true, + ) + expect(errors).toEqual([]) + entry.dispose() + } finally { + await browser?.close() + for (const fiber of fibers.reverse()) await fiber.dispose() + } + }, 90_000) +}) diff --git a/packages/platform-crossgram/src/client.test.ts b/packages/platform-crossgram/src/client.test.ts index 94a0ad06..5c031c2d 100644 --- a/packages/platform-crossgram/src/client.test.ts +++ b/packages/platform-crossgram/src/client.test.ts @@ -1199,6 +1199,103 @@ describe('QQNTClient streaming transport', () => { expect(second.toString()).toBe('efgh') }) + describe('共享直链刷新等待取消', () => { + const locator = { + messageId: 'image', elementId: 'element', chatType: 2 as const, peerUid: 'group', + kind: 'image' as const, fileName: 'photo.jpg', fileUuid: 'image-uuid', + } + + it('已取消的下载不发起直链请求', async () => { + const fetch = vi.fn() + const client = new QQNTClient({ fetch }) + const controller = new AbortController() + const reason = new Error('归档已取消') + controller.abort(reason) + + await expect(collect(client.downloadFile(locator, { signal: controller.signal }))).rejects.toBe(reason) + expect(fetch).not.toHaveBeenCalled() + }) + + it('直链请求永不返回时仍立即取消下载并移除监听', async () => { + const fetch = vi.fn(() => new Promise(() => {})) + const client = new QQNTClient({ fetch }) + const controller = new AbortController() + const add = vi.spyOn(controller.signal, 'addEventListener') + const remove = vi.spyOn(controller.signal, 'removeEventListener') + const reason = new Error('图片归档超时') + const download = collect(client.downloadFile(locator, { signal: controller.signal })) + const rejected = expect(download).rejects.toBe(reason) + + controller.abort(reason) + await rejected + + expect(fetch).toHaveBeenCalledTimes(1) + expect(add).toHaveBeenCalledTimes(1) + expect(remove).toHaveBeenCalledExactlyOnceWith('abort', add.mock.calls[0]![1]) + }, 1_000) + + it('取消一个调用不取消其他调用共享的请求', async () => { + const lookup = Promise.withResolvers() + const fetch = vi.fn(() => lookup.promise) + const client = new QQNTClient({ endpoint: 'http://bridge.invalid/v1', token: 'bridge-token', fetch }) + const controller = new AbortController() + const other = new AbortController() + const first = collect(client.downloadFile(locator, { signal: controller.signal })) + const second = client.resolveFileUrl(locator, other.signal) + const reason = new Error('首个调用超时') + const rejected = expect(first).rejects.toBe(reason) + + controller.abort(reason) + await rejected + const direct = { url: 'https://cdn.qq.example/image', expiresAt: Date.now() + 60_000 } + lookup.resolve(Response.json(direct)) + + await expect(second).resolves.toEqual(direct) + await expect(client.resolveFileUrl(locator)).resolves.toEqual(direct) + expect(fetch).toHaveBeenCalledExactlyOnceWith('http://bridge.invalid/v1/files/direct-url', { + method: 'POST', + headers: { authorization: 'Bearer bridge-token', 'content-type': 'application/json' }, + body: JSON.stringify(locator), + }) + }, 1_000) + + it.each(['成功', '失败'] as const)('直链请求%s后移除调用方监听', async (outcome) => { + const lookup = Promise.withResolvers() + const client = new QQNTClient({ fetch: vi.fn(() => lookup.promise) }) + const controller = new AbortController() + const add = vi.spyOn(controller.signal, 'addEventListener') + const remove = vi.spyOn(controller.signal, 'removeEventListener') + const pending = client.resolveFileUrl(locator, controller.signal) + const direct = { url: 'https://cdn.qq.example/image', expiresAt: Date.now() + 60_000 } + const error = new Error('直链请求失败') + + if (outcome === '成功') { + lookup.resolve(Response.json(direct)) + await expect(pending).resolves.toEqual(direct) + } else { + lookup.reject(error) + await expect(pending).rejects.toBe(error) + } + + expect(add).toHaveBeenCalledTimes(1) + expect(remove).toHaveBeenCalledExactlyOnceWith('abort', add.mock.calls[0]![1]) + }) + + it('调用取消后共享请求再失败不会产生未处理拒绝', async () => { + const lookup = Promise.withResolvers() + const client = new QQNTClient({ fetch: vi.fn(() => lookup.promise) }) + const controller = new AbortController() + const pending = client.resolveFileUrl(locator, controller.signal) + const reason = new Error('图片归档超时') + const rejected = expect(pending).rejects.toBe(reason) + + controller.abort(reason) + await rejected + lookup.reject(new Error('直链请求稍后失败')) + await new Promise((resolve) => setImmediate(resolve)) + }, 1_000) + }) + it('reuses a file direct URL until the bridge-provided expiry and then refreshes it', async () => { let now = 1_800_000_000_000 vi.spyOn(Date, 'now').mockImplementation(() => now) diff --git a/packages/platform-crossgram/src/client.ts b/packages/platform-crossgram/src/client.ts index 957a9d58..39bf28bf 100644 --- a/packages/platform-crossgram/src/client.ts +++ b/packages/platform-crossgram/src/client.ts @@ -1091,9 +1091,20 @@ export class QQNTClient { return value }).finally(() => this.directUrlRefreshes.delete(key)) if (!active) this.directUrlRefreshes.set(key, refresh) - const resolved = await refresh - if (signal?.aborted) throw signal.reason ?? new Error('download aborted') - return resolved + if (!signal) return refresh + let abort!: () => void + try { + const resolved = await new Promise((resolve, reject) => { + abort = () => reject(signal.reason ?? new Error('download aborted')) + signal.addEventListener('abort', abort, { once: true }) + if (signal.aborted) abort() + refresh.then(resolve, reject) + }) + if (signal.aborted) throw signal.reason ?? new Error('download aborted') + return resolved + } finally { + signal.removeEventListener('abort', abort) + } } /** 请求缓存直链；若 QQ 返回密钥过期的可重试错误码，则重新解析直链并重试一次。 */ diff --git a/packages/test-suite/src/config-schema.test.ts b/packages/test-suite/src/config-schema.test.ts index 3f1ba2ad..07834722 100644 --- a/packages/test-suite/src/config-schema.test.ts +++ b/packages/test-suite/src/config-schema.test.ts @@ -20,7 +20,7 @@ import { Config as memoryUpdateStoreConfig } from '../../update-store-memory/src const cases = [ ['bridge', bridgeConfig, [ - 'dcId', 'serverHost', 'serverPort', 'altEndpoints', 'apiPrefix', 'groupFilesMiniApp', 'uploadPath', 'autoMuteGroupChats', + 'dcId', 'serverHost', 'serverPort', 'altEndpoints', 'apiPrefix', 'groupFilesMiniApp', 'uploadPath', 'imageStorage', 'autoMuteGroupChats', 'blockedContentMode', 'recalledMessageMode', 'voiceWorkerSocketPath', 'voiceWorkerTimeoutMs', 'voiceDirectIce', 'voiceTurnHost', 'voiceTurnPort', 'voiceTurnSharedSecret', 'voiceTurnTtlSeconds', ]], @@ -93,6 +93,7 @@ describe('plugin config schemas', () => { tokenTtlSeconds: 600, }, uploadPath: 'data/bridge-uploads', + imageStorage: { enabled: false, path: 'data/bridge-images', adminUserIds: [] }, autoMuteGroupChats: true, blockedContentMode: 'hide-user', recalledMessageMode: 'show', @@ -140,6 +141,28 @@ describe('plugin config schemas', () => { expect(memoryUpdateStoreConfig({})).toEqual({ retention: 1_000 }) }) + it('图片存储字段可编辑，默认关闭且支持显式启用和自定义目录', () => { + const json = bridgeConfig.toJSON() + const root = json.refs[json.uid] + const storage = json.refs[root.dict.imageStorage as unknown as number] + expect(storage.type).toBe('object') + expect(Object.keys(storage.dict)).toEqual(['enabled', 'path', 'adminUserIds']) + for (const uid of Object.values(storage.dict)) { + expect(json.refs[uid as unknown as number].meta.description).toBeTruthy() + } + expect(bridgeConfig({ imageStorage: {} }).imageStorage) + .toEqual({ enabled: false, path: 'data/bridge-images', adminUserIds: [] }) + expect(bridgeConfig({ imageStorage: { enabled: true } }).imageStorage) + .toEqual({ enabled: true, path: 'data/bridge-images', adminUserIds: [] }) + expect(bridgeConfig({ imageStorage: { path: 'data/custom-images' } }).imageStorage) + .toEqual({ enabled: false, path: 'data/custom-images', adminUserIds: [] }) + expect(() => bridgeConfig({ imageStorage: { enabled: 'true' } } as any)).toThrow(/enabled/) + expect(() => bridgeConfig({ imageStorage: { path: 123 } } as any)).toThrow(/path/) + expect(bridgeConfig({ imageStorage: { adminUserIds: [1, 42] } }).imageStorage.adminUserIds).toEqual([1, 42]) + expect(() => bridgeConfig({ imageStorage: { adminUserIds: [0] } })).toThrow(/adminUserIds/) + expect(() => bridgeConfig({ imageStorage: { adminUserIds: ['1'] } } as any)).toThrow(/adminUserIds/) + }) + it('validates and normalizes alternate bridge endpoints', () => { expect(bridgeConfig({ altEndpoints: [ 'bridge-backup.example:8443', diff --git a/packages/test-suite/src/image-storage.e2e.test.ts b/packages/test-suite/src/image-storage.e2e.test.ts new file mode 100644 index 00000000..55ff9834 --- /dev/null +++ b/packages/test-suite/src/image-storage.e2e.test.ts @@ -0,0 +1,371 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { mkdtemp, readFile, readdir, rm, stat } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { randomUUID } from 'node:crypto' +import { join } from 'node:path' +import { pathToFileURL } from 'node:url' +import Long from 'long' +import SSO from '@cordisjs/plugin-sso' +import type { tl } from '@mtcute/core' +import { __tlWriterMap, addPublicKey } from '@mtcute/core/utils.js' +import { MemoryStorage, TelegramClient } from '@mtcute/node' +import { NodeCryptoProvider } from '@mtcute/node/utils.js' +import { generateWriterCodeForTlEntries, parseTlToEntries } from '@mtcute/tl-utils' +import { CROSSGRAM_API_SCHEMA, generateRsaKeyPair } from '@mtproto-relay/mtproto' +import * as bridge from '@mtproto-relay/bridge' +import type { ImageStorageDashboardData } from '../../bridge/src/dashboard-types.js' +import { makePlatformPlugin, startApp, waitForPlatformLogin, waitForWebuiRoute } from './harness.js' + +// mtcute 将 Long 指纹转为未补零的十六进制查找公钥，测试密钥避开前导零指纹。 +let rsaKey = generateRsaKeyPair() +while (rsaKey.fingerprint.startsWith('0')) rsaKey = generateRsaKeyPair() + +const PLATFORM_ID = 'image-storage-e2e' +const GROUP: bridge.IMConversation = { id: 'group-a', kind: 'group', title: '归档群 A' } +const OTHER_GROUP: bridge.IMConversation = { id: 'group-b', kind: 'group', title: '归档群 B' } +const DAY = '2026-10-14' +const MIDNIGHT = Date.parse('2026-10-15T00:00:00Z') / 1000 +const PNG = Buffer.from('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=', 'base64') +const BYTES = Buffer.concat([PNG, Buffer.from(Array.from({ length: 12_288 }, (_, index) => index % 251))]) +const writerMap = Object.assign(Object.create(null), __tlWriterMap, new Function(`${generateWriterCodeForTlEntries( + parseTlToEntries(CROSSGRAM_API_SCHEMA), + { variableName: 'm', includePrelude: true, includeStaticSizes: true }, +)};return m`)()) +const disposals: Array<() => Promise> = [] + +afterEach(async () => { + for (const dispose of disposals.splice(0).reverse()) await dispose() +}) + +async function step(name: string, work: Promise): Promise { + let timer: ReturnType | undefined + try { + return await Promise.race([work, new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`图片归档 e2e 等待超时：${name}`)), 10_000) + timer.unref?.() + })]) + } finally { + if (timer) clearTimeout(timer) + } +} + +function image(id: string): bridge.IMMessagePart { + return { type: 'media', media: { + id, kind: 'image', mimeType: 'image/png', name: `${id}.png`, + size: 1, width: 1, height: 1, locator: { id }, + } } +} + +function sourcePlatform() { + const instanceId = randomUUID() + let handler: ((event: bridge.IMEvent) => void | Promise) | undefined + let sequence = 0 + let expired = false + let historyCalls = 0 + const downloads: string[] = [] + const urls: string[] = [] + const conversations = new Map() + const platform: bridge.IMPlatform = { + capabilities: { + history: true, + send: { text: false, images: false, files: false, mixed: false, maxTextLength: 0, maxMedia: 0 }, + conversations: { groups: true, channels: false, subchannels: false }, + }, + async getAccount() { return { credentials: {}, user: { id: 'self', firstName: '归档测试账号' } } }, + async subscribe(_session, next) { + handler = next + return () => { handler = undefined } + }, + async getDialogs() { + return { dialogs: [...conversations.values()].map(({ conversation, messages }) => ({ + conversation, unreadCount: 0, lastMessage: messages.at(-1), + })) } + }, + async getHistory(_session, conversation) { + historyCalls++ + return { messages: [...(conversations.get(conversation.id)?.messages ?? [])].reverse() } + }, + async getUser(_session, id) { return { id, firstName: id } }, + async sendMessage(): Promise { throw new Error('测试平台不支持发送') }, + async *downloadMedia(_session, media, options) { + downloads.push(media.id) + if (expired) throw new Error('源图片已过期') + const offset = options?.offset ?? 0 + const bytes = BYTES.subarray(offset, offset + (options?.limit ?? BYTES.length)) + yield bytes.subarray(0, 31) + yield bytes.subarray(31) + }, + async resolveMediaUrl(_session, media) { + urls.push(media.id) + if (expired) throw new Error('源直链已过期') + return { url: `https://images.example.test/${media.id}`, expiresAt: Date.now() + 60_000, supportsRange: true } + }, + } + return { + platform, downloads, urls, + get historyCalls() { return historyCalls }, + expire() { expired = true }, + async push(parts: bridge.IMMessagePart[] = [image('photo')], conversation = GROUP, timestamp = MIDNIGHT - 1) { + if (!handler) throw new Error('平台尚未订阅') + const message: bridge.IMMessage = { + id: `incoming-${instanceId}-${++sequence}`, conversationId: conversation.id, + senderId: 'sender', timestamp, content: { parts }, + } + const state = conversations.get(conversation.id) ?? { conversation, messages: [] } + state.messages.push(message) + conversations.set(conversation.id, state) + await handler({ type: 'message', conversation, message }) + return message + }, + } +} + +async function fixture(options: { enabled?: boolean, root?: string } = {}) { + const root = options.root ?? await mkdtemp(join(tmpdir(), 'bridge-image-storage-e2e-')) + if (!options.root) disposals.push(() => rm(root, { recursive: true, force: true })) + const source = sourcePlatform() + const app = await startApp({ + rsaKey, + databasePath: pathToFileURL(join(root, 'bridge.db')).href, + authKeyStorePath: join(root, 'auth-keys.json'), + bridgeConfig: { uploadPath: join(root, 'uploads'), imageStorage: { + path: join(root, 'images'), adminUserIds: [1], + ...(options.enabled === undefined ? {} : { enabled: options.enabled }), + } }, + platform: { id: PLATFORM_ID, adapter: source.platform }, + }) + const sso = app.ctx.plugin(SSO) + let stopped = false + const stop = async () => { + if (stopped) return + stopped = true + await sso.dispose() + await app.stop() + } + disposals.push(stop) + await sso + await app.ctx.database.prepared() + let [identity] = await app.ctx.sso.getIdentities(1) + if (!identity) { + const created = await app.ctx.sso.createUser('image-storage-e2e') + expect(created.user.id).toBe(1) + identity = (await app.ctx.sso.getIdentity(created.identityId))! + } + const token = await app.ctx.sso.createSession(1, identity.id) + expect(await app.ctx.sso.validateSession(token)).toMatchObject({ id: 1 }) + const login = await waitForPlatformLogin(app.ctx, PLATFORM_ID) + await vi.waitFor(() => expect(app.ctx.imPlatform.sessions).toHaveLength(1)) + const entry = await waitForWebuiRoute(app.ctx, '/platform-accounts') + const data = entry.data as ImageStorageDashboardData + const dashboard: ImageStorageDashboardData = { + imageStorageEnabled: data.imageStorageEnabled, + getImageStorage: query => data.getImageStorage(query, token), + deleteImageStorage: filter => data.deleteImageStorage(filter, token), + } + return { ...app, stop, root, source, login, dashboard } +} + +type Fixture = Awaited> + +async function ready(f: Fixture, count: number) { + await vi.waitFor(async () => { + expect(await f.dashboard.getImageStorage({})).toMatchObject({ count, pendingCount: 0 }) + }, { timeout: 5_000, interval: 10 }) +} + +async function connectClient(f: Fixture, platformId = PLATFORM_ID) { + const login = await waitForPlatformLogin(f.ctx, platformId) + addPublicKey(new NodeCryptoProvider(), f.rsaKey.publicKeyPem, false) + const dc = { id: 1, ipAddress: '127.0.0.1', port: f.port } + const client = new TelegramClient({ + apiId: 1, apiHash: 'crossgram-image-storage-e2e', storage: new MemoryStorage(), + defaultDcs: { main: dc, media: dc }, updates: {}, logLevel: 0, writerMap, + initConnectionOptions: { + deviceModel: 'Crossgram image storage E2E', systemVersion: 'test', appVersion: '0.1.0', + systemLangCode: 'zh', langPack: '', langCode: 'zh', + }, + }) + disposals.push(() => client.destroy()) + await step(`MTProto 账号登录 ${platformId}`, client.start({ + phone: `+${login.auth.virtualPhone}`, code: () => bridge.generateLoginCode(login.auth.totpSecret), + codeSentCallback: () => {}, + })) + return client +} + +async function photoLocation(client: TelegramClient, title = GROUP.title, caption?: string): Promise { + const dialogs = await client.call({ + _: 'messages.getDialogs', offsetDate: 0, offsetId: 0, offsetPeer: { _: 'inputPeerEmpty' }, + limit: 100, hash: Long.ZERO, + }) + if (dialogs._ === 'messages.dialogsNotModified') throw new Error('群会话未返回') + const chat = dialogs.chats.find(chat => 'title' in chat && chat.title === title) + if (!chat || chat._ !== 'channel') throw new Error(`找不到归档群：${title}`) + const history = await client.call({ + _: 'messages.getHistory', peer: { _: 'inputPeerChannel', channelId: chat.id, accessHash: chat.accessHash! }, + offsetId: 0, offsetDate: 0, addOffset: 0, limit: 100, maxId: 0, minId: 0, hash: Long.ZERO, + }) + if (history._ === 'messages.messagesNotModified') throw new Error('图片历史未返回') + const message = history.messages.find(message => message._ === 'message' && message.media?._ === 'messageMediaPhoto' + && (caption === undefined || message.message === caption)) + if (message?._ !== 'message' || message.media?._ !== 'messageMediaPhoto' || message.media.photo?._ !== 'photo') { + throw new Error('归档图片没有可下载的 photo') + } + const photo = message.media.photo + return { _: 'inputPhotoFileLocation', id: photo.id, accessHash: photo.accessHash, fileReference: photo.fileReference, thumbSize: 'x' } +} + +async function download(client: TelegramClient, location: tl.RawInputPhotoFileLocation, offset = 0, limit = 32_768) { + const result = await client.call({ _: 'upload.getFile', location, offset, limit, precise: true }) + if (result._ !== 'upload.file') throw new Error('图片被重定向而不是返回文件') + return Buffer.from(result.bytes) +} + +function directUrl(client: TelegramClient, location: tl.RawInputPhotoFileLocation) { + return client.call({ _: 'crossgram.getFileUrl', location } as unknown as tl.RpcMethod) as Promise +} + +describe('服务端图片归档 e2e', () => { + it('默认关闭时不主动下载平台推送的图片', async () => { + const f = await fixture() + await f.source.push() + expect(f.dashboard.imageStorageEnabled).toBe(false) + expect(await f.dashboard.getImageStorage({})).toMatchObject({ count: 0, bytes: 0, pendingCount: 0 }) + expect(f.source.downloads).toEqual([]) + expect(await f.ctx.database.get('mtproto_image_storage', {})).toEqual([]) + }) + + it('没有客户端和历史请求时也将图文中的所有图片落盘而不保存普通附件', async () => { + const f = await fixture({ enabled: true }) + await f.source.push([ + { type: 'text', text: '图文开头' }, image('first'), + { type: 'text', text: '图文结尾' }, image('second'), + { type: 'media', media: { id: 'attachment', kind: 'file', mimeType: 'application/pdf', locator: { id: 'attachment' } } }, + ]) + await ready(f, 2) + expect(f.source.downloads.sort()).toEqual(['first', 'second']) + expect(f.source.historyCalls).toBe(0) + const files = await readdir(join(f.root, 'images'), { recursive: true }) + const saved = [] + for (const file of files) { + const path = join(f.root, 'images', file) + if ((await stat(path)).isFile()) saved.push(await readFile(path)) + } + expect(saved).toEqual([BYTES, BYTES]) + }) + + it('按消息的 UTC 日期和来源群统计实际字节而非平台声明的大小', async () => { + const f = await fixture({ enabled: true }) + await f.source.push([image('before-midnight')], GROUP, MIDNIGHT - 1) + await f.source.push([image('at-midnight')], GROUP, MIDNIGHT) + await f.source.push([image('other-group')], OTHER_GROUP, MIDNIGHT - 1) + await ready(f, 3) + expect(await f.dashboard.getImageStorage({ + platformSessionId: f.login.session.id, conversationId: GROUP.id, startDate: DAY, endDate: DAY, groupBy: 'date', + })).toMatchObject({ + count: 1, bytes: BYTES.length, totalGroups: 1, + groups: [{ platformSessionId: f.login.session.id, date: DAY, count: 1, bytes: BYTES.length }], + }) + expect(await f.dashboard.getImageStorage({ groupBy: 'conversation' })).toMatchObject({ + count: 3, bytes: BYTES.length * 3, totalGroups: 2, + groups: [ + { conversationId: GROUP.id, conversationTitle: GROUP.title, count: 2, bytes: BYTES.length * 2 }, + { conversationId: OTHER_GROUP.id, conversationTitle: OTHER_GROUP.title, count: 1, bytes: BYTES.length }, + ], + }) + }) + + it.each([ + { name: '完整文件', offset: 0, limit: 32_768 }, + { name: '中间分块', offset: 4096, limit: 4096 }, + { name: '文件尾部', offset: 12_288, limit: 4096 }, + { name: '越过文件末尾', offset: 16_384, limit: 4096 }, + ])('源下载过期后通过真实 upload.getFile 读取归档的$name', async ({ offset, limit }) => { + const f = await fixture({ enabled: true }) + await f.source.push() + await ready(f, 1) + f.source.expire() + const client = await connectClient(f) + const location = await photoLocation(client) + expect(await download(client, location, offset, limit)).toEqual(BYTES.subarray(offset, offset + limit)) + expect(f.source.downloads).toEqual(['photo']) + }) + + it('已有归档时拒绝 QQ 直链并让客户端回退到 MTProto 文件下载', async () => { + const f = await fixture({ enabled: true }) + await f.source.push() + await ready(f, 1) + const client = await connectClient(f) + const location = await photoLocation(client) + await expect(directUrl(client, location)).rejects.toMatchObject({ text: 'MEDIA_DIRECT_URL_UNAVAILABLE' }) + expect(f.source.urls).toEqual([]) + expect(await download(client, location)).toEqual(BYTES) + }) + + it('另一个已认证账号不能用原账号的 photo 引用读取归档', async () => { + const f = await fixture({ enabled: true }) + await f.source.push() + await ready(f, 1) + const owner = await connectClient(f) + const location = await photoLocation(owner) + const otherSource = sourcePlatform() + const other = f.ctx.plugin(makePlatformPlugin('image-storage-other', otherSource.platform)) + await other + disposals.push(async () => { await other.dispose() }) + const foreign = await connectClient(f, 'image-storage-other') + await expect(download(foreign, location)).rejects.toMatchObject({ text: 'FILE_ID_INVALID' }) + await expect(directUrl(foreign, location)).rejects.toMatchObject({ text: 'FILE_ID_INVALID' }) + expect(otherSource.downloads).toEqual([]) + }) + + it('重启并关闭新图片存储后仍能读取旧副本且不下载新推送', async () => { + const original = await fixture({ enabled: true }) + await original.source.push() + await ready(original, 1) + const originalClient = await connectClient(original) + const location = await photoLocation(originalClient) + await originalClient.destroy() + await original.stop() + const restarted = await fixture({ enabled: false, root: original.root }) + restarted.source.expire() + await restarted.source.push([image('new-photo')]) + const client = await connectClient(restarted) + expect(await download(client, location)).toEqual(BYTES) + expect(restarted.dashboard.imageStorageEnabled).toBe(false) + expect(await restarted.dashboard.getImageStorage({})).toMatchObject({ count: 1, bytes: BYTES.length }) + expect(restarted.source.downloads).toEqual([]) + }) + + it('只删除目标账号、群和日期的副本且保留消息及平台源下载回退', async () => { + const f = await fixture({ enabled: true }) + await f.source.push([{ type: 'text', text: 'target-caption' }, image('target')], GROUP, MIDNIGHT - 1) + await f.source.push([image('later')], GROUP, MIDNIGHT) + await f.source.push([image('other-group')], OTHER_GROUP, MIDNIGHT - 1) + const otherSource = sourcePlatform() + const other = f.ctx.plugin(makePlatformPlugin('image-storage-other', otherSource.platform)) + await other + disposals.push(async () => { await other.dispose() }) + await waitForPlatformLogin(f.ctx, 'image-storage-other') + await vi.waitFor(() => expect(f.ctx.imPlatform.sessions).toHaveLength(2)) + await otherSource.push([image('other-account')], GROUP, MIDNIGHT - 1) + await ready(f, 4) + const client = await connectClient(f) + const location = await photoLocation(client, GROUP.title, 'target-caption') + const filter = { platformSessionId: f.login.session.id, conversationId: GROUP.id, startDate: DAY, endDate: DAY } + expect(await f.dashboard.deleteImageStorage(filter)).toEqual({ count: 1, bytes: BYTES.length }) + expect(await f.dashboard.getImageStorage(filter)).toMatchObject({ count: 0, bytes: 0 }) + expect(await f.dashboard.getImageStorage({})).toMatchObject({ count: 3, bytes: BYTES.length * 3 }) + expect((await f.ctx.database.get('mtproto_image_storage', { state: 'ready' })).map(row => row.platformMediaId).sort()) + .toEqual(['later', 'other-account', 'other-group']) + const files = await readdir(join(f.root, 'images'), { recursive: true }) + let diskBytes = 0 + for (const file of files) { + const info = await stat(join(f.root, 'images', file)) + if (info.isFile()) diskBytes += info.size + } + expect(diskBytes).toBe(BYTES.length * 3) + expect(await download(client, location)).toEqual(BYTES) + expect(f.source.downloads.filter(id => id === 'target')).toHaveLength(2) + expect(JSON.parse((await directUrl(client, location)).data)).toMatchObject({ url: 'https://images.example.test/target' }) + }) +})