mirror of
https://github.com/hectorqin/reader.git
synced 2026-11-11 05:20:54 +08:00
feat(search): stream source results over HTTP
This commit is contained in:
hectorqin committed
1 parent
171297dc39
commit
5f8554eadd
15 files changed
+418
-149
No files matched your search
+18
-2
@@ -835,11 +835,27 @@ Content-Type: application/json
|
||||
|
||||
```text
|
||||
GET /api/v1/sources/:id/browse?ref=<optional>&cursor=<optional>&limit=<1..200>
|
||||
GET /api/v1/sources/:id/search?q=<query>&cursor=<optional>&limit=<1..200>
|
||||
POST /api/v1/sources/:id/search
|
||||
GET /api/v1/sources/:id/entries?ref=<entry-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":"..."}`）同时支持取消插件内部搜索任务。
|
||||
|
||||
### 凭据与来源启停
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<string, unknown> | 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<string, unknown>;
|
||||
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'))!;
|
||||
|
||||
@@ -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<SourceHost, Map<string, AbortController>>();
|
||||
|
||||
/** 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<string>(), refs = new Set<string>();
|
||||
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 });
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<void>(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<void>(resolve => { entered = resolve; });
|
||||
const aborted = new Promise<void>(resolve => { cancelled = resolve; });
|
||||
const cleanup = new Promise<void>(resolve => { closed = resolve; });
|
||||
const cursors: Array<string | undefined> = [];
|
||||
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');
|
||||
});
|
||||
+35
-8
@@ -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<Session> | null = null;
|
||||
private sessionGeneration = 0;
|
||||
private readonly streams = new Set<AbortController>();
|
||||
private sessionWrites: Promise<void> = 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<void> {
|
||||
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<void> { await this.call('/api/v1/plugins', 'POST', { folder, trusted: true }); }
|
||||
async enablePlugin(id: string, enabled: boolean): Promise<void> { await this.call(`/api/v1/plugins/${encodeURIComponent(id)}`, 'PATCH', { enabled }); }
|
||||
async uninstallPlugin(id: string): Promise<void> { await this.call(`/api/v1/plugins/${encodeURIComponent(id)}`, 'DELETE'); }
|
||||
async sourceCatalog(id: string, query: { ref?: string; query?: string; cursor?: string; filters?: Record<string, string>; sessionId?: string; resultLimit?: number } = {}, options: RequestOptions = {}): Promise<SourcePage> {
|
||||
async sourceCatalog(id: string, query: { ref?: string; cursor?: string } = {}, options: RequestOptions = {}): Promise<SourcePage> {
|
||||
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<string, string>; resultLimit?: number }, options: RequestOptions = {}): AsyncGenerator<SourcePage> {
|
||||
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<void> {
|
||||
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<import('../core/platform.ts').HttpResponse> {
|
||||
const generation = this.sessionGeneration;
|
||||
const assertCurrent = (): void => this.assertSession(generation);
|
||||
const attempt = async (token: string | null) => {
|
||||
assertCurrent();
|
||||
const headers: Record<string, string> = { accept: options.binary ? '*/*' : 'application/json' };
|
||||
const headers: Record<string, string> = { 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) {
|
||||
|
||||
@@ -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<Uint8Array>, 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();
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
@@ -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<Uint8Array>;
|
||||
/** Raw text, used to build a useful error when JSON parsing fails. */
|
||||
text?: string;
|
||||
}
|
||||
|
||||
@@ -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<string>();
|
||||
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<void>(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) => <Button className="primary" disabled={this.busy || option.available === false}
|
||||
onClick={() => void this.run(() => this.acquire(entry, option.id))}>{option.label}</Button>)}</div>
|
||||
</article>)}
|
||||
{this.searchState !== 'idle' && this.searchState !== 'limited' && (this.page?.nextCursor || (this.searchSession && this.searchState === 'stopped')) && !this.searchRun && <div className="catalog-pagination"><Button disabled={this.busy} onClick={() => void this.search(true)}>{this.page?.batch || this.searchSession ? '继续搜索' : '加载更多结果'}</Button></div>}
|
||||
{this.searchState !== 'idle' && this.searchState !== 'limited' && (this.page?.nextCursor || (this.searchSession && (this.searchState === 'stopped' || this.searchState === 'error'))) && !this.searchRun && <div className="catalog-pagination"><Button disabled={this.busy} onClick={() => void this.search(true)}>{this.page?.batch || !this.page ? '继续搜索' : '加载更多结果'}</Button></div>}
|
||||
{this.searchState === 'idle' && (this.path.length > 1 || this.page?.nextCursor) && <nav className="catalog-pagination" aria-label="搜索结果翻页">
|
||||
{this.path.length > 1 && <Button disabled={this.busy} onClick={() => void this.run(async () => { const previous = this.path[this.path.length - 2]!; await this.catalog(previous, false); this.path.pop(); })}>上一页</Button>}
|
||||
{this.page?.nextCursor && <Button disabled={this.busy} onClick={() => void this.run(() => this.catalog({ ...this.path.at(-1), cursor: this.page!.nextCursor! }))}>下一页</Button>}
|
||||
|
||||
@@ -40,7 +40,9 @@ it('uses provider-defined search filter keys and retains filters across paginati
|
||||
else if (request.url.endsWith('/sources')) json = { sources: [{ ...descriptor, id: 's1', name: '第三方', sourceType: 'custom', enabled: true, descriptor }] };
|
||||
else if (request.url.endsWith('/subscriptions')) json = { subscriptions: [] };
|
||||
else if (request.url.endsWith('/search-filters')) json = [{ key: 'region', label: '地区', type: 'select', options: [{ value: '', label: '全部' }, { value: 'asia', label: '亚洲' }] }];
|
||||
else if (request.url.includes('/search?')) json = { items: [], nextCursor: 'next' };
|
||||
else if (request.url.endsWith('/search')) return { status: 200, headers: {}, stream: new ReadableStream({ start(controller) {
|
||||
controller.enqueue(new TextEncoder().encode('event: results\ndata: {"items":[],"nextCursor":"next"}\n\nevent: done\ndata: {}\n\n')); controller.close();
|
||||
} }) };
|
||||
return { status: 200, headers: {}, json };
|
||||
});
|
||||
const screen = new SourcesScreen({ api, admin: false, onOpen() {}, onBack() {}, onSignedOut() {} }); screens.push(screen); document.body.append(screen.element); await screen.show();
|
||||
@@ -51,8 +53,8 @@ it('uses provider-defined search filter keys and retains filters across paginati
|
||||
const select = screen.element.querySelector<HTMLSelectElement>('.sources-search select')!; select.value = 'asia'; select.dispatchEvent(new Event('change', { bubbles: true }));
|
||||
const input = screen.element.querySelector<HTMLInputElement>('input[type=search]')!; input.value = '测试'; input.dispatchEvent(new Event('input', { bubbles: true })); click('搜索');
|
||||
await vi.waitFor(() => expect(screen.element.textContent).toContain('加载更多结果')); click('加载更多结果');
|
||||
await vi.waitFor(() => expect(transport.requests.filter(request => request.url.includes('/search?'))).toHaveLength(2));
|
||||
for (const request of transport.requests.filter(request => request.url.includes('/search?'))) expect(new URL(request.url, 'http://test').searchParams.get('filters')).toBe('{"region":"asia"}');
|
||||
await vi.waitFor(() => expect(transport.requests.filter(request => request.url.endsWith('/search'))).toHaveLength(2));
|
||||
for (const request of transport.requests.filter(request => request.url.endsWith('/search'))) expect(JSON.parse(bodyText(request)).filters).toEqual({ region: 'asia' });
|
||||
});
|
||||
|
||||
it('keeps tab drafts and refresh selection, follows action navigation and uses source instance endpoints', async () => {
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import { afterEach, expect, it, vi } from 'vitest';
|
||||
import { eventStream } from '../src/api/event-stream.ts';
|
||||
import { ReaderApi } from '../src/api/client.ts';
|
||||
import { FetchTransport } from '../src/core/fetch-transport.ts';
|
||||
import type { Session } from '../src/api/types.ts';
|
||||
import { FakeTransport, makePlatform, bodyText } from './helpers/env.ts';
|
||||
const encoder = new TextEncoder();
|
||||
const event = (name: string, value: unknown) => `event: ${name}\ndata: ${JSON.stringify(value)}\n\n`;
|
||||
const result = { items: [{ ref: 'one', title: '中文书籍' }], batch: { completed: 1, total: 2 }, nextCursor: 'next' };
|
||||
const query = { query: '中文', sessionId: 'stream-session-id', filters: { group: '小说' } };
|
||||
function body(text: string, width = 999999) { const bytes = encoder.encode(text); let offset = 0; return new ReadableStream<Uint8Array>({ pull(controller) {
|
||||
if (offset >= bytes.length) { controller.close(); return; }
|
||||
controller.enqueue(bytes.slice(offset, offset += width));
|
||||
} }); }
|
||||
const session = (id: string): Session => ({ user: { id, username: id, displayName: id, role: 'member', createdAt: 0 }, accessToken: 'token-' + id, refreshToken: 'refresh-' + id, accessTokenExpiresAt: 9e15, refreshTokenExpiresAt: 9e15 });
|
||||
async function setup() {
|
||||
const transport = new FakeTransport(); const api = new ReaderApi(makePlatform(transport), { async load() { return session('a'); }, async save() {}, async clear() {} });
|
||||
await api.restore(); return { transport, api };
|
||||
}
|
||||
afterEach(() => vi.unstubAllGlobals());
|
||||
|
||||
it('decodes byte-split UTF-8, CRLF, comments and multiline data', async () => {
|
||||
const received = [];
|
||||
for await (const value of eventStream(body(': heartbeat\r\nevent: results\r\ndata: {"title":\r\ndata: "中文书籍"}\r\n\r\n', 1))) received.push(value);
|
||||
expect(received).toEqual([{ event: 'results', data: '{"title":\n"中文书籍"}' }]);
|
||||
});
|
||||
|
||||
it('returns the first event before HTTP closes and sends one authenticated POST', async () => {
|
||||
const { api, transport } = await setup(); let controller!: ReadableStreamDefaultController<Uint8Array>;
|
||||
transport.respondWith(() => ({ status: 200, headers: {}, stream: new ReadableStream({ start(value) { controller = value; } }) }));
|
||||
const stream = api.searchSource('id', query); const first = stream.next();
|
||||
await vi.waitFor(() => expect(controller).toBeDefined());
|
||||
controller.enqueue(encoder.encode(event('results', result)));
|
||||
expect((await first).value).toEqual(result); expect(transport.requests).toHaveLength(1);
|
||||
expect(transport.requests[0]).toMatchObject({ method: 'POST', url: '/api/v1/sources/id/search', headers: { authorization: 'Bearer token-a', accept: 'text/event-stream' } });
|
||||
expect(JSON.parse(bodyText(transport.requests[0]))).toEqual(query);
|
||||
controller.enqueue(encoder.encode(event('done', { reason: 'complete' })));
|
||||
expect((await stream.next()).done).toBe(true);
|
||||
});
|
||||
|
||||
it('refreshes before opening a stream but never replays a partially consumed stream', async () => {
|
||||
const { api, transport } = await setup();
|
||||
transport.respondWith(request => {
|
||||
if (request.url.endsWith('/refresh')) return { status: 200, headers: {}, json: { ...session('a'), accessToken: 'rotated' } };
|
||||
if (request.headers.authorization === 'Bearer token-a') return { status: 401, headers: {}, json: { error: { code: 'TOKEN_EXPIRED' } } };
|
||||
return { status: 200, headers: {}, stream: body(event('results', result) + event('error', { code: 'SITE_ERROR', message: '失败', status: 502 })) };
|
||||
});
|
||||
const stream = api.searchSource('id', query);
|
||||
expect((await stream.next()).value).toEqual(result);
|
||||
await expect(stream.next()).rejects.toMatchObject({ code: 'SITE_ERROR' });
|
||||
expect(transport.requests).toHaveLength(3); expect(api.currentSession()?.accessToken).toBe('rotated');
|
||||
});
|
||||
|
||||
it('reports unexpected EOF instead of silently completing', async () => {
|
||||
const { api, transport } = await setup(); transport.respondWith(() => ({ status: 200, headers: {}, stream: body(event('results', result), 3) }));
|
||||
const stream = api.searchSource('id', query); expect((await stream.next()).value).toEqual(result);
|
||||
await expect(stream.next()).rejects.toMatchObject({ code: 'STREAM_INTERRUPTED' });
|
||||
});
|
||||
|
||||
it('aborts an idle stream when the user stops or changes server', async () => {
|
||||
for (const changeServer of [false, true]) {
|
||||
const { api, transport } = await setup(); let cancelled = false;
|
||||
transport.respondWith(() => ({ status: 200, headers: {}, stream: new ReadableStream({ cancel() { cancelled = true; } }) }));
|
||||
const stop = new AbortController(), stream = api.searchSource('id', query, { signal: stop.signal });
|
||||
const pending = stream.next(); const rejected = changeServer
|
||||
? expect(pending).rejects.toMatchObject({ code: 'ACCOUNT_CHANGED' })
|
||||
: expect(pending).rejects.toMatchObject({ name: 'AbortError' });
|
||||
await vi.waitFor(() => expect(transport.requests).toHaveLength(1));
|
||||
if (changeServer) api.setBaseUrl('http://other'); else stop.abort();
|
||||
await rejected; expect(cancelled).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
it('fetch transport leaves successful streams unbuffered and rejects unexpected content types', async () => {
|
||||
const transport = new FetchTransport(() => 'http://reader.test');
|
||||
const source = new ReadableStream<Uint8Array>();
|
||||
vi.stubGlobal('fetch', vi.fn().mockResolvedValueOnce(new Response(source, { headers: { 'content-type': 'text/event-stream; charset=utf-8' } })).mockResolvedValueOnce(new Response('{}', { headers: { 'content-type': 'application/json' } })));
|
||||
const request = { url: '/search', method: 'POST', headers: {}, stream: true };
|
||||
const response = await transport.send(request); expect(response.stream).toBe(source); await response.stream!.cancel();
|
||||
await expect(transport.send(request)).rejects.toMatchObject({ code: 'INVALID_STREAM' });
|
||||
});
|
||||
|
||||
it('rejects oversized unfinished events and cancels the reader', async () => {
|
||||
let cancelled = false;
|
||||
const stream = new ReadableStream<Uint8Array>({ start(controller) { controller.enqueue(encoder.encode('data: ' + 'x'.repeat(2 * 1024 * 1024))); }, cancel() { cancelled = true; } });
|
||||
await expect(eventStream(stream).next()).rejects.toMatchObject({ code: 'INVALID_STREAM' }); expect(cancelled).toBe(true);
|
||||
});
|
||||
@@ -33,105 +33,76 @@ async function setup(admin = true) {
|
||||
else if (request.url.endsWith('/sources')) json = { sources: [source], source };
|
||||
else if (request.url.endsWith('/plugins')) json = { plugins: [{ pluginId: 'remote', name: '远程插件', enabled: true, builtin: false }] };
|
||||
else if (request.url.endsWith('/subscriptions')) json = { subscriptions: [{ bookId: 'book1', title: '追更小说', enabled: false, intervalMinutes: 60, newChapters: 2, lastSuccessAt: 1000, nextCheckAt: 0 }] };
|
||||
else if (request.url.includes('/browse?') || request.url.includes('/search?')) json = { items: [{ ref: 'book-ref', title: '来源小说', options: [{ id: 'epub', label: '下载 EPUB' }] }] };
|
||||
else if (request.url.includes('/browse?') || request.url.endsWith('/search')) json = { items: [{ ref: 'book-ref', title: '来源小说', options: [{ id: 'epub', label: '下载 EPUB' }] }] };
|
||||
else if (request.url.endsWith('/acquire')) json = { kind: 'ready', publicationId: 'book1' };
|
||||
else if (request.url.endsWith('/books/book1')) json = { book: { id: 'book1', title: '来源小说' } };
|
||||
if (request.url.endsWith('/search')) return { status: 200, headers: {}, stream: new ReadableStream({ start(controller) {
|
||||
controller.enqueue(new TextEncoder().encode('event: results\ndata: ' + JSON.stringify(json) + '\n\nevent: done\ndata: {}\n\n')); controller.close();
|
||||
} }) };
|
||||
return { status: 200, headers: {}, json };
|
||||
});
|
||||
await screen.show();
|
||||
return { api, screen, transport, onOpen };
|
||||
}
|
||||
|
||||
it('separates partial failures from results and distinguishes unavailable from empty searches', async () => {
|
||||
const { screen, api } = await setup();
|
||||
await chooseSource(screen.element);
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog');
|
||||
async function* pages(...values: Array<SourcePage | Promise<SourcePage>>): AsyncGenerator<SourcePage> { for (const value of values) yield await value; }
|
||||
|
||||
it('separates partial failures, preserves single-source pagination and clears old results on a new search', async () => {
|
||||
const { screen, api } = await setup(); await chooseSource(screen.element);
|
||||
const search = vi.spyOn(api, 'searchSource');
|
||||
const errors = [{ source: '测试来源', code: 'HTTP_ERROR', message: '站点返回 HTTP 503' }];
|
||||
catalog.mockResolvedValueOnce({ items: [{ ref: 'ok', title: '成功结果' }], errors, title: '本批来源' });
|
||||
search.mockImplementationOnce(() => pages({ items: [{ ref: 'ok', title: '成功结果' }], errors }));
|
||||
input(screen.element, '搜索书籍', '小说'); await click(screen.element, '搜索');
|
||||
expect(screen.element.querySelector('.catalog-summary')?.textContent).toContain('已找到 1 本书');
|
||||
expect(screen.element.querySelector('details')?.open).toBe(false);
|
||||
expect(screen.element.textContent).toContain('成功结果');
|
||||
catalog.mockResolvedValueOnce({ items: [], errors, nextCursor: 'more' });
|
||||
search.mockImplementationOnce(() => pages({ items: [], errors, nextCursor: 'more' }));
|
||||
await click(screen.element, '搜索');
|
||||
expect(screen.element.querySelector('details')?.open).toBe(true);
|
||||
expect(screen.element.textContent).toContain('部分来源搜索失败');
|
||||
expect(screen.element.textContent).toContain('加载更多结果');
|
||||
expect(screen.element.textContent).not.toContain('没有找到匹配书籍');
|
||||
catalog.mockResolvedValueOnce({ items: [] });
|
||||
await click(screen.element, '加载更多结果');
|
||||
expect(catalog).toHaveBeenLastCalledWith('source-1', { query: '小说', filters: {}, cursor: 'more' }, {});
|
||||
search.mockImplementationOnce(() => pages({ items: [] })); await click(screen.element, '加载更多结果');
|
||||
expect(search.mock.calls.at(-1)?.[1]).toMatchObject({ query: '小说', cursor: 'more', filters: {} });
|
||||
expect(screen.element.querySelector('details')).not.toBeNull();
|
||||
catalog.mockResolvedValueOnce({ items: [] });
|
||||
search.mockImplementationOnce(() => pages({ items: [] }));
|
||||
input(screen.element, '搜索书籍', '新关键词'); await click(screen.element, '搜索');
|
||||
expect(screen.element.querySelector('details')).toBeNull();
|
||||
expect(screen.element.textContent).toContain('没有找到匹配书籍');
|
||||
expect(screen.element.querySelector('.catalog-pagination')).toBeNull();
|
||||
});
|
||||
|
||||
it('merges batches by ref, stops in flight, ignores stale replies and resumes from the pending cursor', async () => {
|
||||
const { screen, api } = await setup();
|
||||
await chooseSource(screen.element);
|
||||
const descriptor = { ...opds, capabilities: ['browse', 'search', 'search.cancel'] };
|
||||
vi.spyOn(api, 'sourceTypes').mockResolvedValue([descriptor]);
|
||||
vi.spyOn(api, 'sources').mockResolvedValue([{ ...source, descriptor }]);
|
||||
await screen.show(); await chooseSource(screen.element);
|
||||
it('merges events from one stream, aborts on stop, ignores late data and resumes the last received cursor', async () => {
|
||||
const { screen, api } = await setup(); await chooseSource(screen.element);
|
||||
const cancel = vi.spyOn(api, 'cancelSourceSearch').mockResolvedValue();
|
||||
let resolve!: (page: SourcePage) => void;
|
||||
const first = { ref: 'first', title: '第一批结果' };
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog')
|
||||
.mockResolvedValueOnce({ items: [first, first], batch: { completed: 3, total: 7 }, nextCursor: 'second' })
|
||||
.mockImplementationOnce(() => new Promise<SourcePage>(done => { resolve = done; }));
|
||||
input(screen.element, '搜索书籍', '斗罗');
|
||||
button(screen.element, '搜索').click();
|
||||
await vi.waitFor(() => expect(catalog).toHaveBeenCalledTimes(2));
|
||||
expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(1);
|
||||
const signal = catalog.mock.calls[1]![2]!.signal!;
|
||||
button(screen.element, '停止搜索').click();
|
||||
await vi.waitFor(() => expect(screen.element.textContent).toContain('已停止搜索'));
|
||||
expect(signal.aborted).toBe(true);
|
||||
expect(screen.element.textContent).toContain('第一批结果');
|
||||
resolve({ items: [{ ref: 'late', title: '迟到结果' }], batch: { completed: 1, total: 1 } });
|
||||
await new Promise(resolveTick => setTimeout(resolveTick, 10));
|
||||
const search = vi.spyOn(api, 'searchSource').mockImplementationOnce(() => pages(
|
||||
{ items: [first, first], batch: { completed: 3, total: 7 }, nextCursor: 'second' },
|
||||
new Promise<SourcePage>(done => { resolve = done; })));
|
||||
input(screen.element, '搜索书籍', '斗罗'); button(screen.element, '搜索').click();
|
||||
await vi.waitFor(() => expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(1));
|
||||
expect(search).toHaveBeenCalledTimes(1);
|
||||
const signal = search.mock.calls[0]![2]!.signal!;
|
||||
await click(screen.element, '停止搜索'); expect(signal.aborted).toBe(true); expect(cancel).toHaveBeenCalledTimes(1);
|
||||
resolve({ items: [{ ref: 'late', title: '迟到结果' }] }); await new Promise(done => setTimeout(done, 10));
|
||||
expect(screen.element.textContent).not.toContain('迟到结果');
|
||||
expect(catalog).toHaveBeenCalledTimes(2);
|
||||
catalog.mockResolvedValueOnce({ items: [first, { ref: 'second', title: '第二批结果' }], batch: { completed: 6, total: 7 }, nextCursor: 'third' })
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'third', title: '第三批结果' }], batch: { completed: 7, total: 7 } });
|
||||
search.mockImplementationOnce(() => pages(
|
||||
{ items: [first, { ref: 'second', title: '第二批结果' }], batch: { completed: 6, total: 7 }, nextCursor: 'third' },
|
||||
{ items: [{ ref: 'third', title: '第三批结果' }], batch: { completed: 7, total: 7 } }));
|
||||
await click(screen.element, '继续搜索');
|
||||
expect(catalog.mock.calls[2]![1]!.cursor).toBe('second');
|
||||
expect(search).toHaveBeenCalledTimes(2); expect(search.mock.calls[1]![1].cursor).toBe('second');
|
||||
expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(3);
|
||||
expect(screen.element.textContent).toContain('搜索完成');
|
||||
expect(screen.element.textContent).toContain('7 / 7');
|
||||
catalog.mockResolvedValueOnce({ items: [{ ref: 'new', title: '新搜索结果' }] });
|
||||
input(screen.element, '搜索书籍', '新关键词'); await click(screen.element, '搜索');
|
||||
expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(1);
|
||||
expect(screen.element.textContent).not.toContain('第一批结果');
|
||||
});
|
||||
|
||||
it('continues into later available sources after a partial batch failure', async () => {
|
||||
it('keeps earlier results when a stream fails and resumes after the last event', async () => {
|
||||
const { screen, api } = await setup(); await chooseSource(screen.element);
|
||||
const descriptor = { ...opds, capabilities: ['search'] };
|
||||
vi.spyOn(api, 'sourceTypes').mockResolvedValue([descriptor]);
|
||||
vi.spyOn(api, 'sources').mockResolvedValue([{ ...source, descriptor }]);
|
||||
await screen.show(); await chooseSource(screen.element);
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog')
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'one', title: '第一来源结果' }], errors: [{ source: '失效来源', code: 'HTTP_ERROR', message: '站点暂不可用' }], batch: { completed: 2, total: 5 }, nextCursor: 'remaining' })
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'two', title: '后续来源结果' }], batch: { completed: 5, total: 5 } });
|
||||
input(screen.element, '搜索书籍', '继续测试'); await click(screen.element, '搜索');
|
||||
expect(catalog).toHaveBeenCalledTimes(2);
|
||||
expect(screen.element.textContent).toContain('后续来源结果');
|
||||
expect(screen.element.textContent).toContain('搜索完成');
|
||||
expect(screen.element.textContent).toContain('站点暂不可用');
|
||||
});
|
||||
|
||||
it('does not auto-page a single-source catalog and stops loops in malformed batch responses', async () => {
|
||||
const { screen, api } = await setup(); await chooseSource(screen.element);
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog').mockResolvedValueOnce({ items: [], nextCursor: 'page2' });
|
||||
input(screen.element, '搜索书籍', 'query'); await click(screen.element, '搜索');
|
||||
expect(catalog).toHaveBeenCalledTimes(1); expect(button(screen.element, '加载更多结果')).toBeDefined();
|
||||
catalog.mockResolvedValue({ items: [], batch: { completed: 3, total: 9 }, nextCursor: 'same' });
|
||||
await click(screen.element, '搜索');
|
||||
expect(screen.element.textContent).toContain('搜索进度没有推进');
|
||||
expect(catalog).toHaveBeenCalledTimes(3);
|
||||
const search = vi.spyOn(api, 'searchSource').mockImplementationOnce(async function* () {
|
||||
yield { items: [{ ref: 'one', title: '保留结果' }], batch: { completed: 1, total: 2 }, nextCursor: 'resume' };
|
||||
throw new Error('连接中断');
|
||||
});
|
||||
input(screen.element, '搜索书籍', '查询'); await click(screen.element, '搜索');
|
||||
expect(screen.element.textContent).toContain('搜索中断'); expect(screen.element.textContent).toContain('保留结果');
|
||||
search.mockImplementationOnce(() => pages({ items: [{ ref: 'two', title: '恢复结果' }], batch: { completed: 2, total: 2 } }));
|
||||
await click(screen.element, '继续搜索'); expect(search.mock.calls[1]![1].cursor).toBe('resume');
|
||||
expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(2);
|
||||
});
|
||||
function button(root: HTMLElement, label: string) {
|
||||
const found = [...root.querySelectorAll('button')].find((element) => element.textContent?.trim() === label || element.getAttribute('aria-label') === label);
|
||||
@@ -288,47 +259,41 @@ async function sessionSetup() {
|
||||
return value;
|
||||
}
|
||||
|
||||
it('polls rolling heartbeats and drains completed-source buffers through the last cursor', async () => {
|
||||
it('consumes heartbeats and buffered results through one HTTP stream', async () => {
|
||||
const { screen, api } = await sessionSetup();
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog')
|
||||
.mockResolvedValueOnce({ items: [], batch: { completed: 0, total: 2 }, nextCursor: 'heartbeat' })
|
||||
.mockResolvedValueOnce({ items: [], batch: { completed: 0, total: 2 }, nextCursor: 'second' })
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'one', title: '结果一' }], batch: { completed: 2, total: 2 }, nextCursor: 'buffered' })
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'two', title: '结果二' }], batch: { completed: 2, total: 2 } });
|
||||
const search = vi.spyOn(api, 'searchSource').mockImplementationOnce(() => pages(
|
||||
{ items: [], batch: { completed: 0, total: 2 }, nextCursor: 'heartbeat' },
|
||||
{ items: [], batch: { completed: 0, total: 2 }, nextCursor: 'second' },
|
||||
{ items: [{ ref: 'one', title: '结果一' }], batch: { completed: 2, total: 2 }, nextCursor: 'buffered' },
|
||||
{ items: [{ ref: 'two', title: '结果二' }], batch: { completed: 2, total: 2 } }));
|
||||
input(screen.element, '搜索书籍', '查询'); await click(screen.element, '搜索');
|
||||
expect(catalog).toHaveBeenCalledTimes(4);
|
||||
expect(new Set(catalog.mock.calls.map(call => call[1]?.sessionId)).size).toBe(1);
|
||||
expect(catalog.mock.calls[0]?.[1]?.resultLimit).toBe(10000);
|
||||
expect(search).toHaveBeenCalledTimes(1); expect(search.mock.calls[0]![1].resultLimit).toBe(10000);
|
||||
expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(2);
|
||||
expect(screen.element.textContent).toContain('搜索完成');
|
||||
});
|
||||
|
||||
it('stops before the first response and waits for cancellation before resuming the session', async () => {
|
||||
const { screen, api } = await sessionSetup();
|
||||
let stopped!: () => void;
|
||||
it('stops before the first event and waits for cancellation before resuming', async () => {
|
||||
const { screen, api } = await sessionSetup(); let stopped!: () => void;
|
||||
const cancel = vi.spyOn(api, 'cancelSourceSearch').mockImplementationOnce(() => new Promise<void>(resolve => { stopped = resolve; }));
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog').mockImplementationOnce(() => new Promise(() => {}))
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'resumed', title: '恢复结果' }], batch: { completed: 2, total: 2 } });
|
||||
const search = vi.spyOn(api, 'searchSource').mockImplementationOnce(() => pages(new Promise(() => {})))
|
||||
.mockImplementationOnce(() => pages({ items: [{ ref: 'resumed', title: '恢复结果' }], batch: { completed: 2, total: 2 } }));
|
||||
input(screen.element, '搜索书籍', '查询'); button(screen.element, '搜索').click();
|
||||
await vi.waitFor(() => expect(catalog).toHaveBeenCalledTimes(1));
|
||||
const id = catalog.mock.calls[0]![1]!.sessionId;
|
||||
await vi.waitFor(() => expect(search).toHaveBeenCalledTimes(1));
|
||||
const id = search.mock.calls[0]![1].sessionId;
|
||||
await click(screen.element, '停止搜索'); expect(cancel).toHaveBeenCalledWith('source-1', id);
|
||||
button(screen.element, '继续搜索').click(); await new Promise(resolve => setTimeout(resolve, 10));
|
||||
expect(catalog).toHaveBeenCalledTimes(1);
|
||||
button(screen.element, '继续搜索').click(); await new Promise(resolve => setTimeout(resolve, 10)); expect(search).toHaveBeenCalledTimes(1);
|
||||
stopped(); await vi.waitFor(() => expect(screen.element.textContent).toContain('搜索完成'));
|
||||
expect(catalog.mock.calls[1]![1]!.sessionId).toBe(id);
|
||||
expect(screen.element.textContent).toContain('恢复结果');
|
||||
expect(search.mock.calls[1]![1].sessionId).toBe(id);
|
||||
});
|
||||
|
||||
it('drains buffered limited results then cancels and hides continue', async () => {
|
||||
it('drains limited result events then cancels and hides continue', async () => {
|
||||
const { screen, api } = await sessionSetup();
|
||||
const cancel = vi.spyOn(api, 'cancelSourceSearch').mockResolvedValue();
|
||||
const catalog = vi.spyOn(api, 'sourceCatalog')
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'one', title: '上限结果一' }], batch: { completed: 1, total: 8 }, limitReached: true, nextCursor: 'buffer' })
|
||||
.mockResolvedValueOnce({ items: [{ ref: 'two', title: '上限结果二' }], batch: { completed: 1, total: 8 }, limitReached: true });
|
||||
const search = vi.spyOn(api, 'searchSource').mockImplementationOnce(() => pages(
|
||||
{ items: [{ ref: 'one', title: '结果一' }], batch: { completed: 1, total: 8 }, limitReached: true, nextCursor: 'buffer' },
|
||||
{ items: [{ ref: 'two', title: '结果二' }], batch: { completed: 1, total: 8 }, limitReached: true }));
|
||||
input(screen.element, '搜索书籍', '查询'); await click(screen.element, '搜索');
|
||||
expect(catalog).toHaveBeenCalledTimes(2); expect(cancel).toHaveBeenCalledTimes(1);
|
||||
expect(search).toHaveBeenCalledTimes(1); expect(cancel).toHaveBeenCalledTimes(1);
|
||||
expect(screen.element.querySelectorAll('.catalog-book')).toHaveLength(2);
|
||||
expect(screen.element.textContent).toContain('已达到结果上限');
|
||||
expect(screen.element.querySelector('.catalog-pagination')).toBeNull();
|
||||
expect(screen.element.textContent).toContain('已达到结果上限'); expect(screen.element.querySelector('.catalog-pagination')).toBeNull();
|
||||
});
|
||||
@@ -26,12 +26,13 @@ try {
|
||||
{ key: 'source', label: '书源', type: 'select', options: [{ value: '', label: '全部书源' }, { value: 'one', label: '示例书源' }] },
|
||||
];
|
||||
else if (path.endsWith('/demo/search')) {
|
||||
requests.push(Object.fromEntries(url.searchParams));
|
||||
const last = !!url.searchParams.get('cursor');
|
||||
requests.push(route.request().postDataJSON());
|
||||
const last = true;
|
||||
json = { title: '本批：艾途小说 / 得奇小说网 / 就爱文学', batch: { completed: last ? 6 : 3, total: 6 },
|
||||
items: mode === 'partial' ? [{ ref: 'book', title: '斗破苍穹', authors: ['天蚕土豆'], description: '【示例书源】这里是斗气的世界，没有花哨艳丽的魔法，有的仅仅是繁衍到巅峰的斗气。', options: [{ id: 'read', label: '加入书架' }] }] : [],
|
||||
...(mode !== 'empty' ? { errors } : {}), ...(!last ? { nextCursor: 'batch2' } : {}) };
|
||||
}
|
||||
if (path.endsWith('/demo/search')) return route.fulfill({ contentType: 'text/event-stream', body: 'event: results\ndata: ' + JSON.stringify(json) + '\n\nevent: done\ndata: {}\n\n' });
|
||||
if (json !== undefined) return route.fulfill({ contentType: 'application/json', body: JSON.stringify(json) });
|
||||
return route.continue();
|
||||
});
|
||||
@@ -55,7 +56,8 @@ try {
|
||||
mode = 'partial'; await page.getByRole('button', { name: '搜索', exact: true }).click();
|
||||
await page.locator('.search-progress').filter({ hasText: '搜索完成' }).waitFor();
|
||||
assert.equal(await page.locator('.catalog-book').count(), 1, 'repeated entries merge by ref');
|
||||
assert.equal(requests.at(-1).cursor, 'batch2');
|
||||
assert.equal(requests.at(-1).query, '斗破苍穹');
|
||||
assert.equal(requests.at(-1).filters.group, 'fiction');
|
||||
assert.equal(await page.locator('.catalog-errors').evaluate(el => el.open), false);
|
||||
await page.locator('.catalog-errors summary').focus(); await page.keyboard.press('Enter');
|
||||
assert.equal(await page.locator('.catalog-errors').evaluate(el => el.open), true);
|
||||
|
||||
Reference in new issue
Block a user