diff --git a/docs/api.md b/docs/api.md index 3022434..060b52e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -835,11 +835,27 @@ Content-Type: application/json ```text GET /api/v1/sources/:id/browse?ref=&cursor=&limit=<1..200> -GET /api/v1/sources/:id/search?q=&cursor=&limit=<1..200> +POST /api/v1/sources/:id/search GET /api/v1/sources/:id/entries?ref= ``` -目录响应形如 `{ "items": [...], "navigation": [...], "nextCursor": "...", "title": "..." }`；可选字段可省略。`ref` 和 `cursor` 是来源拥有的不透明值，客户端只保存并原样回传。条目返回 `ref`、标题、作者、封面和 `options`；不要从 `ref` 推断 URL 或拼接下一页地址。引用和查询参数必须为非空字符串且不超过 16,384 字符；路径中的 `publicationRef` 需要 URL 编码，路由允许最大 16,384 字符的参数。 +浏览仍返回普通 JSON 目录响应。搜索使用 Streamable HTTP：请求体为 `{ "query": "...", "sessionId": "...", "cursor": "...", "filters": {}, "resultLimit": 10000 }`，响应为 `text/event-stream`，持续发送 `results`、最终 `done` 或 `error` 事件。`results.data` 是 CatalogPage，客户端边收边合并；断开连接会取消在途搜索。`ref` 和 `cursor` 是来源拥有的不透明值，客户端只保存并原样回传。条目返回 `ref`、标题、作者、封面和 `options`；不要从 `ref` 推断 URL 或拼接下一页地址。引用和查询参数必须为非空字符串且不超过 16,384 字符；路径中的 `publicationRef` 需要 URL 编码，路由允许最大 16,384 字符的参数。 + +搜索请求必须携带 Bearer 凭据。`sessionId` 必填，由 16–80 个字母、数字、`_` 或 `-` 组成；继续同一次搜索时保留该值并回传最后收到的 `nextCursor`，新关键词或筛选条件使用新的值。可选 `limit` 为每页 1–200 条，`resultLimit` 为 1–10000 条，默认上限 10000 条。旧 GET 搜索接口已移除。 + +```text +event: results +data: {"items":[{"ref":"book-1","title":"示例书籍"}],"nextCursor":"opaque-cursor","batch":{"completed":1,"total":2}} + +event: results +data: {"items":[],"batch":{"completed":2,"total":2}} + +event: done +data: {"reason":"complete"} + +``` + +服务端在同一响应中自动消费后续分页。`done.reason` 为 `complete` 或 `limit`；错误事件形如 `{"code":"SEARCH_FAILED","message":"搜索中断，已保留收到的结果。","status":500}`。响应开始前的鉴权或参数错误使用普通 HTTP 错误响应，响应开始后的错误通过 `error` 事件返回。没有收到 `done` 或 `error` 就结束连接属于异常中断，客户端保留已收到的结果，不自动重放搜索。停止搜索时中止流；`POST /api/v1/sources/:id/search/cancel`（body 为 `{"sessionId":"..."}`）同时支持取消插件内部搜索任务。 ### 凭据与来源启停 diff --git a/docs/source-plugins.md b/docs/source-plugins.md index a75eff5..87f1194 100644 --- a/docs/source-plugins.md +++ b/docs/source-plugins.md @@ -384,7 +384,7 @@ POST /sources 创建来源实例 PATCH /sources/:id 仅修改 enabled PUT /sources/:id/credentials/:key 设置当前用户凭据 GET /sources/:id/browse 浏览目录 -GET /sources/:id/search 搜索 +POST /sources/:id/search Streamable HTTP 搜索（SSE results/done/error 事件） GET /sources/:id/entries?ref=... 详情 POST /sources/:id/acquire 获取内容，body 传 entryRef GET /sources/:id/publications/:publicationRef/manifest diff --git a/server/src/http/routes/sources.ts b/server/src/http/routes/sources.ts index abfbf95..d405280 100644 --- a/server/src/http/routes/sources.ts +++ b/server/src/http/routes/sources.ts @@ -5,6 +5,7 @@ import { authenticate, currentUser, requireAdmin } from '../auth.ts'; import { badRequest } from '../../lib/errors.ts'; import { SourceHost } from '../../services/source-host.ts'; import { withSignal } from '../request-signal.ts'; +import { searchResponse } from '../search-stream.ts'; function textBody(body: unknown, name: string): string { const value = (body as Record | null)?.[name]; @@ -23,10 +24,10 @@ function searchSessionId(value?: string): string | undefined { if (value !== undefined && !/^[a-zA-Z0-9_-]{16,80}$/.test(value)) throw badRequest('invalid search session id'); return value; } -function searchResultLimit(value?: string): number | undefined { +function searchResultLimit(value: unknown, max = 10000): number | undefined { if (value === undefined) return undefined; - const limit = Number(value); - if (!Number.isInteger(limit) || limit < 1 || limit > 10000) throw badRequest('resultLimit must be an integer from 1 to 10000'); + const limit = value; + if (typeof limit !== 'number' || !Number.isInteger(limit) || limit < 1 || limit > max) throw badRequest(`${max === 200 ? 'limit' : 'resultLimit'} must be an integer from 1 to ${max}`); return limit; } @@ -94,14 +95,18 @@ export function registerSourceRoutes(app: FastifyInstance, ctx: AppContext): voi ref: parameter(q.ref, 'ref'), cursor: parameter(q.cursor, 'cursor'), limit: pageLimit(q.limit), }, signal)); }); - app.get('/api/v1/sources/:id/search', { preHandler: auth }, async (request, reply) => { - const user = currentUser(request); const { id } = request.params as { id: string }; - const q = request.query as { q?: string; cursor?: string; limit?: string; filters?: string; sessionId?: string; resultLimit?: string }; - const query = parameter(q.q, 'q', true)!; - return withSignal(request, reply, (signal) => host.search(user.id, id, { - query, cursor: parameter(q.cursor, 'cursor'), limit: pageLimit(q.limit), filters: searchFilters(q.filters), - sessionId: searchSessionId(q.sessionId), resultLimit: searchResultLimit(q.resultLimit), - }, signal)); + app.post('/api/v1/sources/:id/search', { preHandler: auth }, async (request, reply) => { + const user = currentUser(request), { id } = request.params as { id: string }; + const body = (request.body ?? {}) as Record; + const query = parameter(body.query, 'query', true)!; + const sessionId = searchSessionId(parameter(body.sessionId, 'sessionId', true))!; + const search = { query, sessionId, cursor: parameter(body.cursor, 'cursor'), + limit: searchResultLimit(body.limit, 200), resultLimit: searchResultLimit(body.resultLimit), + filters: searchFilters(body.filters === undefined ? undefined : JSON.stringify(body.filters)) }; + reply.header('content-type', 'text/event-stream; charset=utf-8'); + reply.header('cache-control', 'no-cache, no-transform'); + reply.header('x-accel-buffering', 'no'); + return reply.send(searchResponse(request, reply, host, user.id, id, search)); }); app.post('/api/v1/sources/:id/search/cancel', { preHandler: auth }, async request => { const sessionId = searchSessionId(textBody(request.body, 'sessionId'))!; diff --git a/server/src/http/search-stream.ts b/server/src/http/search-stream.ts new file mode 100644 index 0000000..7628e6a --- /dev/null +++ b/server/src/http/search-stream.ts @@ -0,0 +1,64 @@ +import { Readable } from 'node:stream'; +import type { FastifyReply, FastifyRequest } from 'fastify'; +import { AppError } from '../lib/errors.ts'; +import type { CatalogPage, SearchRequest } from '../sources/types.ts'; +import type { SourceHost } from '../services/source-host.ts'; + +const frame = (event: string, data: unknown) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; +const connections = new WeakMap>(); + +/** One authenticated POST response carries every result page of a search. */ +export function searchResponse(request: FastifyRequest, reply: FastifyReply, host: SourceHost, userId: string, sourceId: string, search: SearchRequest): Readable { + const controller = new AbortController(), abort = () => controller.abort(); + let active = connections.get(host); + if (!active) { active = new Map(); connections.set(host, active); } + const key = JSON.stringify([userId, sourceId, search.sessionId]); + active.get(key)?.abort(); active.set(key, controller); + request.raw.once('aborted', abort); reply.raw.once('close', abort); + async function* events() { + let cursor = search.cursor, previous: CatalogPage['batch'], complete = false; + const seen = new Set(), refs = new Set(); + const limit = search.resultLimit ?? 10000; + try { + yield ': search connected\n\n'; + do { + controller.signal.throwIfAborted(); + const key = cursor ?? ''; + if (seen.has(key)) throw new AppError(502, 'INVALID_CURSOR', '来源返回了重复的搜索游标'); + seen.add(key); + const page = await host.search(userId, sourceId, { ...search, cursor }, controller.signal); + if (page.batch && previous && (page.batch.total !== previous.total || page.batch.completed < previous.completed)) { + throw new AppError(502, 'INVALID_PROGRESS', '来源搜索进度异常'); + } + previous = page.batch; + const items = page.items.filter(entry => { + if (refs.has(entry.ref) || refs.size >= limit) return false; + refs.add(entry.ref); return true; + }); + const capped = refs.size >= limit; + const result = { ...page, items, ...(capped ? { nextCursor: undefined, limitReached: true } : {}) }; + yield frame('results', result); + // Stream ordinary provider pagination as well; the browser receives one + // continuous result stream regardless of whether the provider is batched. + cursor = !capped ? page.nextCursor : undefined; + if (!cursor) { complete = true; yield frame('done', { reason: capped || page.limitReached ? 'limit' : page.nextCursor ? 'page' : 'complete' }); } + } while (cursor && !controller.signal.aborted); + } catch (error) { + if (!controller.signal.aborted) yield frame('error', { + code: error instanceof AppError ? error.code : 'SEARCH_FAILED', + message: error instanceof AppError ? error.message : '搜索中断，已保留收到的结果。', + status: error instanceof AppError ? error.statusCode : 500, + }); + } finally { + controller.abort(); + request.raw.removeListener('aborted', abort); reply.raw.removeListener('close', abort); + if (active!.get(key) === controller) { + active!.delete(key); + if (search.sessionId && (!complete || previous)) await host.cancelSearch(userId, sourceId, search.sessionId).catch(error => { + request.log.warn({ err: error, sourceId }, 'search session cancellation failed'); + }); + } + } + } + return Readable.from(events(), { objectMode: false, highWaterMark: 64 * 1024 }); +} diff --git a/server/src/services/source-host.ts b/server/src/services/source-host.ts index 63c12d9..ac5c755 100644 --- a/server/src/services/source-host.ts +++ b/server/src/services/source-host.ts @@ -141,7 +141,7 @@ export class SourceHost { async cancelSearch(userId: string, id: string, sessionId: string) { // Cancellation must still get through when normal request slots are occupied. const row = this.instance(id), provider = this.registry.get(row.plugin_id, row.source_type)?.provider; - if (!provider?.cancelSearch) throw badRequest('source has no search sessions', 'SOURCE_UNSUPPORTED'); + if (!provider?.cancelSearch) return; await provider.cancelSearch(this.context(row, userId, AbortSignal.timeout(5000)), sessionId); } async searchFilters(userId: string, id: string, signal?: AbortSignal) { diff --git a/server/test/sources-api.test.ts b/server/test/sources-api.test.ts index 59cadf9..a63daaf 100644 --- a/server/test/sources-api.test.ts +++ b/server/test/sources-api.test.ts @@ -19,6 +19,12 @@ import { UploadService } from '../src/services/uploads.ts'; import type { CatalogPage, SourceProvider } from '../src/sources/types.ts'; interface Session { token: string; id: string } +function streamEvents(body: string): Array<{ event: string; data: any }> { + return body.split('\n\n').filter(frame => frame.startsWith('event:')).map(frame => ({ + event: /^event: (.+)/m.exec(frame)![1]!, data: JSON.parse(/^data: (.+)/m.exec(frame)![1]!), + })); +} +const searchPayload = { query: 'book', sessionId: 'test-search-session' }; function auth(session: Session) { return { authorization: `Bearer ${session.token}` }; } @@ -179,10 +185,11 @@ test('source query validation rejects repeated text parameters and invalid page const response = await h.app.inject({ method: 'GET', url: `/api/v1/sources/local/browse?${query}`, headers: auth(h.admin) }); assert.equal(response.statusCode, 400, `${query}: ${response.body}`); } - for (const query of ['q=first&q=second', 'q=', 'q=valid&cursor=one&cursor=two']) { - const response = await h.app.inject({ method: 'GET', url: `/api/v1/sources/local/search?${query}`, headers: auth(h.admin) }); - assert.equal(response.statusCode, 400, `${query}: ${response.body}`); + for (const payload of [{ query: ['one', 'two'] }, { query: '' }, { query: 'valid', cursor: ['one', 'two'] }, { query: 'valid', resultLimit: true }, { query: 'valid', limit: 201 }]) { + const response = await h.app.inject({ method: 'POST', url: '/api/v1/sources/local/search', headers: auth(h.admin), payload: { ...searchPayload, ...payload } }); + assert.equal(response.statusCode, 400, response.body); } + assert.equal((await h.app.inject({ method: 'GET', url: '/api/v1/sources/local/search?q=book', headers: auth(h.admin) })).statusCode, 404); }); test('local source acquisition restores the existing publication to the requesting users shelf', async (t) => { @@ -427,14 +434,14 @@ test('search sessions pass generic options and cancellation remains available wi assert.deepEqual(request.filters, { group: 'chosen' }); entered++; if (entered === 2) ready(); await new Promise(resolve => { held.push(resolve); }); - return { items: [], batch: { completed: 0, total: 2 }, nextCursor: 'opaque-heartbeat' }; + return { items: [], batch: { completed: 2, total: 2 } }; }, async cancelSearch(ctx, sessionId) { stopped.push({ user: ctx.userId, source: ctx.instance.id, session: sessionId }); } }; h.ctx.sources!.registry.register({ pluginId: 'reader.sessions', provider }); await h.ctx.sources!.create({ id: 'session-test', pluginId: 'reader.sessions', sourceType: 'sessions', name: 'session test', config: {} }); - const url = '/api/v1/sources/session-test/search?q=test&sessionId=search-session-id&resultLimit=3&filters=' + encodeURIComponent(JSON.stringify({ group: 'chosen' })); - const requests = [h.app.inject({ method: 'GET', url, headers: auth(member) }), h.app.inject({ method: 'GET', url, headers: auth(h.admin) })]; + const url = '/api/v1/sources/session-test/search', payload = { query: 'test', sessionId: 'search-session-id', resultLimit: 3, filters: { group: 'chosen' } }; + const requests = [h.app.inject({ method: 'POST', url, payload, headers: auth(member) }), h.app.inject({ method: 'POST', url, payload, headers: auth(h.admin) })]; try { await full; const response = await h.app.inject({ method: 'POST', url: '/api/v1/sources/session-test/search/cancel', headers: auth(member), payload: { sessionId: 'search-session-id' } }); @@ -442,7 +449,61 @@ test('search sessions pass generic options and cancellation remains available wi assert.deepEqual(stopped, [{ user: member.id, source: 'session-test', session: 'search-session-id' }]); assert.equal((await h.app.inject({ method: 'POST', url: '/api/v1/sources/session-test/search/cancel', payload: { sessionId: 'search-session-id' } })).statusCode, 401); } finally { held.forEach(release => release()); await Promise.all(requests); } - for (const query of ['sessionId=bad', 'sessionId=search-session-id&sessionId=other-session-id', 'resultLimit=0', 'resultLimit=10001', 'resultLimit=1.5']) { - assert.equal((await h.app.inject({ method: 'GET', url: '/api/v1/sources/session-test/search?q=book&' + query, headers: auth(member) })).statusCode, 400, query); + for (const invalid of [{ sessionId: 'bad' }, { sessionId: ['search-session-id', 'other-session-id'] }, { resultLimit: 0 }, { resultLimit: 10001 }, { resultLimit: 1.5 }]) { + assert.equal((await h.app.inject({ method: 'POST', url, payload: { ...payload, ...invalid }, headers: auth(member) })).statusCode, 400); } }); + +test('Streamable HTTP sends results before completion and disconnect cancels pending work', { timeout: 10000 }, async t => { + const h = await harness(); t.after(() => h.close()); + let entered!: () => void, cancelled!: () => void, closed!: () => void; + const waiting = new Promise(resolve => { entered = resolve; }); + const aborted = new Promise(resolve => { cancelled = resolve; }); + const cleanup = new Promise(resolve => { closed = resolve; }); + const cursors: Array = []; + const provider: SourceProvider = { + descriptor: { id: 'streaming', label: 'Stream', version: '1', capabilities: ['search', 'search.cancel', 'search.session', 'detail'] }, + async detail(_ctx, ref) { return { ref, title: ref }; }, + async acquire() { return { kind: 'action-required', action: { type: 'external', label: 'Test' } }; }, + async search(ctx, request) { + cursors.push(request.cursor); + if (!request.cursor) return { items: [{ ref: 'first', title: '首条中文结果' }], batch: { completed: 1, total: 2 }, nextCursor: 'resume-here' }; + entered(); + return new Promise((_, reject) => { ctx.signal.addEventListener('abort', () => { cancelled(); reject(ctx.signal.reason); }, { once: true }); }); + }, + async cancelSearch() { closed(); } + }; + h.ctx.sources!.registry.register({ pluginId: 'reader.streaming', provider }); + await h.ctx.sources!.create({ id: 'stream-test', pluginId: 'reader.streaming', sourceType: 'streaming', name: 'stream test', config: {} }); + const base = await h.app.listen({ port: 0, host: '127.0.0.1' }); + const controller = new AbortController(); t.after(() => controller.abort()); + const response = await fetch(base + '/api/v1/sources/stream-test/search', { method: 'POST', headers: { ...auth(h.admin), 'content-type': 'application/json', accept: 'text/event-stream' }, body: JSON.stringify(searchPayload), signal: controller.signal }); + assert.match(response.headers.get('content-type')!, /text\/event-stream/); assert.equal(response.headers.get('x-accel-buffering'), 'no'); + const reader = response.body!.getReader(), decoder = new TextDecoder(); let text = ''; + while (!text.includes('首条中文结果')) { const part = await reader.read(); assert.equal(part.done, false); text += decoder.decode(part.value, { stream: true }); } + await waiting; assert.deepEqual(cursors, [undefined, 'resume-here']); assert.doesNotMatch(text, /event: done/); + controller.abort(); await Promise.all([aborted, cleanup]); await reader.cancel().catch(() => {}); +}); + +test('stream errors retain earlier events and caps prevent extra provider calls', async t => { + const h = await harness(); t.after(() => h.close()); let calls = 0; + const provider: SourceProvider = { + descriptor: { id: 'stream-error', label: 'Stream', version: '1', capabilities: ['search', 'detail'] }, + async detail(_ctx, ref) { return { ref, title: ref }; }, + async acquire() { return { kind: 'action-required', action: { type: 'external', label: 'Test' } }; }, + async search(_ctx, request) { + calls++; + if (request.cursor) return { items: [], batch: { completed: 1, total: 2 }, nextCursor: 'same' }; + return { items: [{ ref: 'one', title: 'one' }, { ref: 'one', title: 'duplicate' }, { ref: 'two', title: 'two' }], batch: { completed: 1, total: 2 }, nextCursor: 'same' }; + } + }; + h.ctx.sources!.registry.register({ pluginId: 'reader.stream-error', provider }); + await h.ctx.sources!.create({ id: 'stream-error', pluginId: 'reader.stream-error', sourceType: 'stream-error', name: 'stream error', config: {} }); + const response = await h.app.inject({ method: 'POST', url: '/api/v1/sources/stream-error/search', headers: auth(h.admin), payload: searchPayload }); + const events = streamEvents(response.body); assert.equal(events[0]!.data.items.length, 2); + assert.equal(events.at(-1)!.event, 'error'); assert.equal(events.at(-1)!.data.code, 'INVALID_CURSOR'); + calls = 0; + const limited = await h.app.inject({ method: 'POST', url: '/api/v1/sources/stream-error/search', headers: auth(h.admin), payload: { ...searchPayload, resultLimit: 1 } }); + const result = streamEvents(limited.body); assert.equal(calls, 1); assert.equal(result[0]!.data.items.length, 1); + assert.equal(result[0]!.data.limitReached, true); assert.equal(result.at(-1)!.data.reason, 'limit'); +}); diff --git a/web/src/api/client.ts b/web/src/api/client.ts index 63bac75..fcb12c8 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -1,3 +1,4 @@ +import { eventStream } from './event-stream.ts'; import type { ExtensionField, ExtensionPage } from './sources.ts'; import { ApiError, errorForStatus, parseErrorBody } from './errors.ts'; import type { @@ -127,6 +128,7 @@ export class ReaderApi { private session: Session | null = null; private refreshPromise: Promise | null = null; private sessionGeneration = 0; + private readonly streams = new Set(); private sessionWrites: Promise = Promise.resolve(); private readonly listeners = new Set<(session: Session | null) => void>(); @@ -144,6 +146,7 @@ export class ReaderApi { setBaseUrl(url: string): void { const next = url.replace(/\/+$/, ''); if (next !== this.currentBaseUrl) { + for (const stream of this.streams) stream.abort(); this.sessionGeneration += 1; this.refreshPromise = null; } @@ -169,6 +172,7 @@ export class ReaderApi { private async setSession(session: Session | null, rotation = false): Promise { if (!rotation) { + for (const stream of this.streams) stream.abort(); this.sessionGeneration += 1; this.refreshPromise = null; } @@ -260,15 +264,35 @@ export class ReaderApi { async installPlugin(folder: string): Promise { await this.call('/api/v1/plugins', 'POST', { folder, trusted: true }); } async enablePlugin(id: string, enabled: boolean): Promise { await this.call(`/api/v1/plugins/${encodeURIComponent(id)}`, 'PATCH', { enabled }); } async uninstallPlugin(id: string): Promise { await this.call(`/api/v1/plugins/${encodeURIComponent(id)}`, 'DELETE'); } - async sourceCatalog(id: string, query: { ref?: string; query?: string; cursor?: string; filters?: Record; sessionId?: string; resultLimit?: number } = {}, options: RequestOptions = {}): Promise { + async sourceCatalog(id: string, query: { ref?: string; cursor?: string } = {}, options: RequestOptions = {}): Promise { const params = new URLSearchParams(); if (query.ref) params.set('ref', query.ref); - if (query.query) params.set('q', query.query); if (query.cursor) params.set('cursor', query.cursor); - if (query.filters) params.set('filters', JSON.stringify(query.filters)); - if (query.sessionId) params.set('sessionId', query.sessionId); - if (query.resultLimit !== undefined) params.set('resultLimit', String(query.resultLimit)); - return this.get(`/api/v1/sources/${encodeURIComponent(id)}/${query.query ? 'search' : 'browse'}?${params}`, options); + return this.get(`/api/v1/sources/${encodeURIComponent(id)}/browse?${params}`, options); + } + async *searchSource(id: string, query: { query: string; sessionId: string; cursor?: string; filters?: Record; resultLimit?: number }, options: RequestOptions = {}): AsyncGenerator { + const generation = this.sessionGeneration, controller = new AbortController(); + const signal = options.signal ? AbortSignal.any([options.signal, controller.signal]) : controller.signal; + this.streams.add(controller); + try { + const response = await this.request(`/api/v1/sources/${encodeURIComponent(id)}/search`, 'POST', query, { ...options, signal, stream: true }); + if (!response.stream) throw new ApiError('server', '搜索结果流缺失', 'INVALID_STREAM'); + for await (const event of eventStream(response.stream, signal)) { + this.assertSession(generation); + let data: unknown; + try { data = JSON.parse(event.data); } catch { throw new ApiError('server', '搜索事件格式错误', 'INVALID_STREAM'); } + if (event.event === 'done') return; + if (event.event === 'error') { + const error = data as { message?: string; code?: string; status?: number }; + throw new ApiError(errorForStatus(error.status ?? 500), error.message ?? '搜索失败', error.code ?? 'SEARCH_FAILED', error.status ?? 500); + } + if (event.event !== 'results' || !data || typeof data !== 'object' || !Array.isArray((data as SourcePage).items)) { + throw new ApiError('server', '无效的搜索事件', 'INVALID_STREAM'); + } + yield data as SourcePage; + } + throw new ApiError('offline', '搜索连接中断，已保留收到的结果，可继续搜索。', 'STREAM_INTERRUPTED'); + } finally { controller.abort(); this.streams.delete(controller); } } async cancelSourceSearch(id: string, sessionId: string): Promise { await this.call(`/api/v1/sources/${encodeURIComponent(id)}/search/cancel`, 'POST', { sessionId }); @@ -638,13 +662,13 @@ export class ReaderApi { path: string, method: string, body: unknown, - options: RequestOptions & { binary?: boolean } = {}, + options: RequestOptions & { binary?: boolean; stream?: boolean } = {}, ): Promise { const generation = this.sessionGeneration; const assertCurrent = (): void => this.assertSession(generation); const attempt = async (token: string | null) => { assertCurrent(); - const headers: Record = { accept: options.binary ? '*/*' : 'application/json' }; + const headers: Record = { accept: options.stream ? 'text/event-stream' : options.binary ? '*/*' : 'application/json' }; if (body !== undefined) headers['content-type'] = 'application/json'; if (token) headers.authorization = `Bearer ${token}`; return this.platform.transport.send({ @@ -653,6 +677,7 @@ export class ReaderApi { headers, ...(body !== undefined ? { body: JSON.stringify(body) } : {}), ...(options.binary ? { binary: true } : {}), + ...(options.stream ? { stream: true } : {}), ...(options.signal ? { signal: options.signal } : {}), }); }; @@ -660,6 +685,7 @@ export class ReaderApi { const token = this.session?.accessToken ?? null; try { const response = await attempt(token); + if (generation !== this.sessionGeneration) await response.stream?.cancel().catch(() => {}); assertCurrent(); return response; } catch (err) { @@ -682,6 +708,7 @@ export class ReaderApi { assertCurrent(); try { const response = await attempt(refreshed.accessToken); + if (generation !== this.sessionGeneration) await response.stream?.cancel().catch(() => {}); assertCurrent(); return response; } catch (retryErr) { diff --git a/web/src/api/event-stream.ts b/web/src/api/event-stream.ts new file mode 100644 index 0000000..6aca3a9 --- /dev/null +++ b/web/src/api/event-stream.ts @@ -0,0 +1,38 @@ +import { ApiError } from './errors.ts'; + +/** Decode SSE across arbitrary byte boundaries, including split UTF-8 and CRLF. */ +export async function* eventStream(body: ReadableStream, signal?: AbortSignal): AsyncGenerator<{ event: string; data: string }> { + const reader = body.getReader(), decoder = new TextDecoder(); + let buffer = '', event = '', data: string[] = [], size = 0; + const abort = () => { void reader.cancel().catch(() => {}); }; + signal?.addEventListener('abort', abort, { once: true }); + try { + while (true) { + signal?.throwIfAborted(); + const chunk = await reader.read(); signal?.throwIfAborted(); + buffer += decoder.decode(chunk.value, { stream: !chunk.done }); + let end; + while ((end = buffer.search(/[\r\n]/)) !== -1) { + if (buffer[end] === '\r' && end === buffer.length - 1 && !chunk.done) break; + const line = buffer.slice(0, end); + buffer = buffer.slice(end + (buffer.slice(end, end + 2) === '\r\n' ? 2 : 1)); + size += line.length; + if (size > 2 * 1024 * 1024) throw new ApiError('server', '搜索事件过大', 'INVALID_STREAM'); + if (!line) { + if (data.length) yield { event: event || 'message', data: data.join('\n') }; + event = ''; data = []; size = 0; + } else if (!line.startsWith(':')) { + const colon = line.indexOf(':'), field = colon < 0 ? line : line.slice(0, colon); + const value = colon < 0 ? '' : line.slice(colon + 1).replace(/^ /, ''); + if (field === 'event') event = value; + if (field === 'data') data.push(value); + } + } + if (buffer.length + size > 2 * 1024 * 1024) throw new ApiError('server', '搜索事件过大', 'INVALID_STREAM'); + if (chunk.done) break; + } + } finally { + signal?.removeEventListener('abort', abort); + await reader.cancel().catch(() => {}); reader.releaseLock(); + } +} diff --git a/web/src/core/fetch-transport.ts b/web/src/core/fetch-transport.ts index 8b072e1..149a911 100644 --- a/web/src/core/fetch-transport.ts +++ b/web/src/core/fetch-transport.ts @@ -43,6 +43,14 @@ export class FetchTransport implements Transport { headers[key.toLowerCase()] = value; }); + if (request.stream) { + if (!response.ok) throw await this.toError(response); + if (!response.body || !/^text\/event-stream(?:;|$)/i.test(headers['content-type'] ?? '')) { + await response.body?.cancel(); + throw new ApiError('server', '服务器没有返回搜索结果流', 'INVALID_STREAM'); + } + return { status: response.status, headers, stream: response.body }; + } if (request.binary) { if (!response.ok) throw await this.toError(response); const buffer = new Uint8Array(await response.arrayBuffer()); diff --git a/web/src/core/platform.ts b/web/src/core/platform.ts index 61601e9..1eec5fa 100644 --- a/web/src/core/platform.ts +++ b/web/src/core/platform.ts @@ -23,6 +23,8 @@ export interface HttpRequest { body?: string | FormData; /** Binary response mode: used for book bodies and covers. */ binary?: boolean; + /** Return the response body without buffering it (Streamable HTTP). */ + stream?: boolean; /** * Called with `loaded / total` while the body is being sent. * @@ -44,6 +46,7 @@ export interface HttpResponse { json?: unknown; /** Present for `binary` requests. */ bytes?: Uint8Array; + stream?: ReadableStream; /** Raw text, used to build a useful error when JSON parsing fails. */ text?: string; } diff --git a/web/src/ui/sources-screen.tsx b/web/src/ui/sources-screen.tsx index f14b599..1c076f0 100644 --- a/web/src/ui/sources-screen.tsx +++ b/web/src/ui/sources-screen.tsx @@ -131,27 +131,20 @@ export class SourcesScreen { const source = this.selected, run = new AbortController(); const request = append ? this.searchRequest : { query: this.query.trim(), filters: { ...this.filterValues } }; if (!request?.query) return; - let cursor = append ? this.page?.nextCursor : undefined; + const cursor = append ? this.page?.nextCursor : undefined; if (append && (this.searchState === 'limited' || (!cursor && !this.searchSession))) return; if (!append) { this.page = null; this.path = []; this.acquired = null; } - if (!append) this.searchSession = source.descriptor?.capabilities.includes('search.session') ? { sourceId: source.id, id: searchId() } : null; + if (!append) this.searchSession = { sourceId: source.id, id: searchId() }; const session = this.searchSession; this.searchRequest = request; this.searchRun = run; this.searchState = 'searching'; this.message = ''; this.draw(); const current = () => !this.disposed && this.searchRun === run; - const seen = new Set(); try { await this.searchStop; if (!current()) return; - do { - const key = cursor ?? ''; - if (seen.has(key)) throw new Error('来源返回了重复的搜索游标，请重新搜索。'); - seen.add(key); - // Legacy providers may terminate their whole worker on abort. Only declared - // cooperative search providers receive the signal; stale replies are always ignored. - const page = await this.options.api.sourceCatalog(source.id, { ...request, ...(cursor ? { cursor } : {}), ...(session ? { sessionId: session.id, resultLimit: 10000 } : {}) }, - source.descriptor?.capabilities.includes('search.cancel') ? { signal: run.signal } : {}); + for await (const page of this.options.api.searchSource(source.id, + { ...request, ...(cursor ? { cursor } : {}), sessionId: session!.id, resultLimit: 10000 }, { signal: run.signal })) { if (!current()) return; - if (page.batch && this.page?.batch && (page.batch.completed < this.page.batch.completed || (!session && page.batch.completed === this.page.batch.completed))) throw new Error('来源搜索进度没有推进，请重新搜索。'); + if (page.batch && this.page?.batch && (page.batch.completed < this.page.batch.completed)) throw new Error('来源搜索进度没有推进，请重新搜索。'); const entries = new Map((this.page?.items ?? []).map(entry => [entry.ref, entry])); for (const entry of page.items) { if (entries.has(entry.ref) || entries.size < 10000) entries.set(entry.ref, entry); } const errors = new Map((this.page?.errors ?? []).map(error => [error.source + '\0' + error.code, error])); @@ -162,9 +155,7 @@ export class SourcesScreen { this.message = '已达到单次搜索结果上限，请缩小搜索范围后重新搜索。'; this.searchState = 'limited'; delete this.page.nextCursor; this.cancelSession(); break; } - cursor = page.batch && (session || page.batch.completed < page.batch.total) ? page.nextCursor : undefined; - if (cursor) await new Promise(resolve => setTimeout(resolve, 0)); - } while (cursor && current()); + } if (current() && this.searchState === 'searching') this.searchState = 'complete'; } catch (error) { if (!current()) return; @@ -278,7 +269,7 @@ export class SourcesScreen { {(entry.options?.length ? entry.options : [{ id: '', label: '加入书架' }]).map((option) => )} )} - {this.searchState !== 'idle' && this.searchState !== 'limited' && (this.page?.nextCursor || (this.searchSession && this.searchState === 'stopped')) && !this.searchRun &&
} + {this.searchState !== 'idle' && this.searchState !== 'limited' && (this.page?.nextCursor || (this.searchSession && (this.searchState === 'stopped' || this.searchState === 'error'))) && !this.searchRun &&
} {this.searchState === 'idle' && (this.path.length > 1 || this.page?.nextCursor) &&