From 01ef98abb6935ab8b1fde1dfda3da8bd7913bcd0 Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Tue, 30 Jun 2026 02:27:45 +0800 Subject: [PATCH] feat(onebot): Stream API transport plumbing (#163 phase 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Foundation for NapCat-compatible Stream API actions (chunked file upload/download): a single request answered by many response frames. - streaming.ts: StreamSink + frame wrappers (wrapStreamFrame / wrapStreamTerminal) matching NapCat's wire — each frame an OB11 envelope with stream:'stream-action', data.type ∈ stream/response/error, and message mirrored into wording. - ApiHandler: registerStreamAction / isStreamAction; handle() threads a sink; processStreamRequest drives WS multi-frame dispatch with optional liveness-based abort; dead processRequest removed. - action-kit: defineStreamAction (sink-aware run, registers as a stream action) reusing defineAction's parse/describe. - adapters: HTTP server streams chunked \r\n\r\n-delimited frames (no Content-Length, no false application/json), aborting on client disconnect; WS server/client use a backpressure-aware safeSendAsync and abort a stream once the socket closes. No stream actions are registered yet — those land in the next phases. Verified by two adversarial review passes. 480 onebot tests green. --- packages/onebot/src/action-kit.ts | 58 ++++++ packages/onebot/src/api-handler.ts | 90 +++++++--- .../onebot/src/network/http-server-adapter.ts | 45 ++++- packages/onebot/src/network/utils.ts | 13 ++ .../onebot/src/network/ws-client-adapter.ts | 11 +- .../onebot/src/network/ws-server-adapter.ts | 17 +- packages/onebot/src/streaming.ts | 75 ++++++++ packages/onebot/src/types.ts | 7 + .../tests/http-server-content-type.test.ts | 1 + .../tests/network-bootstrap-meta.test.ts | 2 +- packages/onebot/tests/network-manager.test.ts | 2 +- packages/onebot/tests/streaming.test.ts | 167 ++++++++++++++++++ .../tests/ws-client-stale-close.test.ts | 2 +- 13 files changed, 452 insertions(+), 38 deletions(-) create mode 100644 packages/onebot/src/streaming.ts create mode 100644 packages/onebot/tests/streaming.test.ts diff --git a/packages/onebot/src/action-kit.ts b/packages/onebot/src/action-kit.ts index 7bd506b7..98546bd4 100644 --- a/packages/onebot/src/action-kit.ts +++ b/packages/onebot/src/action-kit.ts @@ -21,6 +21,7 @@ import { RETCODE, failedResponse } from './types'; import type { ApiResponse, JsonObject, JsonValue } from './types'; import type { ApiActionContext, ApiHandler } from './api-handler'; +import { type StreamSink, NOOP_SINK } from './streaming'; // ─────────────────────────── result currency ─────────────────────────── @@ -445,6 +446,63 @@ export function defineAction(def: ActionDef): ActionSpec { }; } +// ─────────────────────────── defineStreamAction ─────────────────────────── +// A Stream API action (#163): same param contract + doc machinery as +// `defineAction`, but `run` additionally receives a `StreamSink` to push +// intermediate frames, and it registers via `registerStreamAction` so the +// network adapter streams its output. The handler's return value is still the +// terminal `ApiResponse` (the final frame). Reuses `defineAction` for +// parse/describe so the two stay in lock-step. + +interface StreamActionDef { + name: string | readonly [string, ...string[]]; + summary?: string; + returns?: string; + returnsSchema?: JsonSchema; + readOnly?: boolean; + params: S; + rules?: (r: RuleBuilders>) => readonly CrossFieldRule[]; + run: (p: InferParams, ctx: ApiActionContext, raw: JsonObject, sink: StreamSink) => Promise | ApiResponse; +} + +export interface StreamActionSpec { + readonly names: string[]; + readonly params: S; + parse(raw: JsonObject): CoerceResult>; + describe(): ActionDoc; + register(h: ApiHandler, ctx: ApiActionContext): void; +} + +export function defineStreamAction(def: StreamActionDef): StreamActionSpec { + // Borrow defineAction's parse + describe (run is never invoked — register + // below installs the sink-aware handler instead). + const base = defineAction({ + name: def.name, + summary: def.summary, + returns: def.returns, + returnsSchema: def.returnsSchema, + readOnly: def.readOnly, + params: def.params, + rules: def.rules, + run: () => failedResponse(RETCODE.INTERNAL_ERROR, 'stream action dispatched without a sink'), + }); + + return { + names: base.names, + params: def.params, + parse: base.parse, + describe: base.describe, + register: (h, ctx) => { + const handler = async (params: JsonObject, sink?: StreamSink): Promise => { + const r = base.parse(params); + if (!r.ok) return failedResponse(RETCODE.BAD_REQUEST, wording(r)); + return def.run(r.value, ctx, params, sink ?? NOOP_SINK); + }; + for (const name of base.names) h.registerStreamAction(name, handler); + }, + }; +} + // ─────────────────────────── presets ─────────────────────────── // `group_id` appears in 61/142 actions, `user_id` in 27 — these collapse the // dominant shape. The injected fields are surfaced in `describe()` so docs diff --git a/packages/onebot/src/api-handler.ts b/packages/onebot/src/api-handler.ts index 1e958e16..52219035 100644 --- a/packages/onebot/src/api-handler.ts +++ b/packages/onebot/src/api-handler.ts @@ -14,6 +14,7 @@ import { register as registerRequest } from './actions/request'; import type { ForwardPreviewMeta } from './modules/message-actions'; import type { JsonObject, JsonValue, MessageMeta } from './types'; import { RETCODE, failedResponse } from './types'; +import { type StreamSink, wrapStreamFrame, wrapStreamTerminal } from './streaming'; const moduleLog = createLogger('Bridge.Action'); @@ -80,7 +81,7 @@ export interface ApiActionContext { fetchPttText: (messageId: number) => Promise<{ text: string }>; } -type ActionHandler = (params: JsonObject) => Promise; +type ActionHandler = (params: JsonObject, sink?: StreamSink) => Promise; /** A handled-action record handed to debug observers. */ export interface ActionRecord { @@ -93,6 +94,9 @@ export type ActionObserver = (rec: ActionRecord) => void; export class ApiHandler { private readonly handlers = new Map(); + /** Names that answer with a multi-frame Stream API response (#163). The + * network adapter consults this BEFORE dispatch to pick streaming output. */ + private readonly streamActions = new Set(); private readonly log: Logger; /** Debug-stream taps — notified after every handled action. Attached * on-demand (ref-counted) by the WebUI debug stream. */ @@ -122,7 +126,19 @@ export class ApiHandler { this.handlers.set(action, handler); } - async handle(action: string, params: JsonObject): Promise { + /** Register a Stream API action — dispatched exactly like a normal action, + * but flagged so adapters stream its frames (the handler receives a sink). */ + registerStreamAction(action: string, handler: ActionHandler): void { + this.handlers.set(action, handler); + this.streamActions.add(action); + } + + /** Whether `action` answers with a multi-frame Stream API response. */ + isStreamAction(action: string): boolean { + return this.streamActions.has(action); + } + + async handle(action: string, params: JsonObject, sink?: StreamSink): Promise { const handler = this.handlers.get(action); if (!handler) { this.log.debug('unknown action %s', action); @@ -134,15 +150,16 @@ export class ApiHandler { // the wrap + id allocation when trace is actually live, so the default // path stays allocation-free. if (getLogLevel() !== 'trace') { - return this.runAction(action, handler, params); + return this.runAction(action, handler, params, sink); } - return runWithRequestId(nextRequestId(), () => this.runAction(action, handler, params)); + return runWithRequestId(nextRequestId(), () => this.runAction(action, handler, params, sink)); } private async runAction( action: string, handler: ActionHandler, params: JsonObject, + sink?: StreamSink, ): Promise { // Terse breadcrumb to the log file (debug, always persisted): lets the // operator grep "what did the bot get asked to do" in post-mortems. @@ -154,7 +171,7 @@ export class ApiHandler { const startedAt = Date.now(); let response: import('./types').ApiResponse; try { - response = await handler(params); + response = await handler(params, sink); this.log.trace(() => [`${action} ⇒ ${response.status} (${Date.now() - startedAt}ms)`]); } catch (error) { // Action failures are almost always param-shape problems coming @@ -185,33 +202,52 @@ export class ApiHandler { } } - async processRequest(rawRequest: string): Promise { - if (!rawRequest.trim()) { - return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request')); - } + /** WS dispatch supporting Stream API multi-frame responses. A normal action + * emits exactly one frame; a stream action emits each intermediate frame + * then the terminal frame — every frame carries the request's echo. `emit` + * writes one JSON string per frame; awaiting it lets the transport apply + * backpressure. `isAlive`, when supplied, is checked before each stream + * frame — returning false aborts the action (e.g. the client disconnected), + * so a dead client can't make a download keep pumping frames into the void. */ + async processStreamRequest( + rawRequest: string, + emit: (json: string) => void | Promise, + isAlive?: () => boolean, + ): Promise { + const bad = (): void => { void emit(JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request'))); }; + if (!rawRequest.trim()) { bad(); return; } + let action: string; + let params: JsonObject; + let echo: JsonValue | undefined; try { const parsed = JSON.parse(rawRequest) as unknown; - if (!isJsonObject(parsed)) { - return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request')); - } - - const action = asString(parsed.action); - if (!action) { - return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request')); - } - - const params = isJsonObject(parsed.params) ? parsed.params : {}; - const echo = parsed.echo; - const response = await this.handle(action, params); - if (echo !== undefined) { - response.echo = toJsonValue(echo); - } - - return JSON.stringify(response); + if (!isJsonObject(parsed)) { bad(); return; } + const a = asString(parsed.action); + if (!a) { bad(); return; } + action = a; + params = isJsonObject(parsed.params) ? parsed.params : {}; + echo = parsed.echo !== undefined ? toJsonValue(parsed.echo) : undefined; } catch { - return JSON.stringify(failedResponse(RETCODE.BAD_REQUEST, 'bad request')); + bad(); + return; } + + if (!this.isStreamAction(action)) { + const response = await this.handle(action, params); + if (echo !== undefined) response.echo = echo; + await emit(JSON.stringify(response)); + return; + } + + const sink: StreamSink = { + send: async (frame) => { + if (isAlive && !isAlive()) throw new Error('stream transport closed'); + await emit(JSON.stringify(wrapStreamFrame(frame, echo))); + }, + }; + const response = await this.handle(action, params, sink); + await emit(JSON.stringify(wrapStreamTerminal(response, echo))); } } diff --git a/packages/onebot/src/network/http-server-adapter.ts b/packages/onebot/src/network/http-server-adapter.ts index e683d087..b617df8e 100644 --- a/packages/onebot/src/network/http-server-adapter.ts +++ b/packages/onebot/src/network/http-server-adapter.ts @@ -5,8 +5,10 @@ import { type Server, type ServerResponse, } from 'http'; +import type { ApiHandler } from '../api-handler'; import type { DispatchPayload } from '../event-filter'; -import type { HttpServerNetwork, JsonObject, JsonValue } from '../types'; +import type { ApiResponse, HttpServerNetwork, JsonObject, JsonValue } from '../types'; +import { type StreamSink, wrapStreamFrame, wrapStreamTerminal } from '../streaming'; import { IOneBotNetworkAdapter, type AdapterStatus, type NetworkAdapterContext } from './adapter'; import { isAuthorized, normalizePath } from './utils'; @@ -181,6 +183,13 @@ export class HttpServerAdapter extends IOneBotNetworkAdapter return; } + // Stream API (#163): answer with chunked multi-frame output — each frame + // a JSON envelope delimited by `\r\n\r\n`, terminated by the final frame. + if (this.ctx.api.isStreamAction(action)) { + await streamHttpResponse(res, this.ctx.api, action, params as JsonObject, echo as JsonValue | undefined); + return; + } + const response = await this.ctx.api.handle(action, params as JsonObject); if (echo !== undefined) { response.echo = echo as JsonValue; @@ -212,7 +221,41 @@ function readRequestBody(req: IncomingMessage, maxBytes = 2 * 1024 * 1024): Prom } function writeJson(res: ServerResponse, statusCode: number, data: unknown): void { + // A streaming response may already have flushed headers (and even ended); a + // late error must not double-send and trip ERR_HTTP_HEADERS_SENT. + if (res.headersSent || res.writableEnded) return; res.statusCode = statusCode; res.setHeader('Content-Type', 'application/json; charset=utf-8'); res.end(JSON.stringify(data)); } + +/** Stream a multi-frame Stream API response (#163): chunked transfer (no + * Content-Length → Node frames it), each frame a JSON envelope delimited by + * `\r\n\r\n`, the action's terminal response written last. Matches NapCat: + * the body is `\r\n\r\n`-joined JSON objects, NOT a single JSON document, so + * no `application/json` Content-Type is claimed. Each `res.write` is awaited + * (flush callback) for backpressure, and the sink aborts the action once the + * client disconnects so a big download stops pumping into a dead socket. */ +async function streamHttpResponse( + res: ServerResponse, + api: ApiHandler, + action: string, + params: JsonObject, + echo: JsonValue | undefined, +): Promise { + res.statusCode = 200; + const writeFrame = (frame: ApiResponse): Promise => + new Promise((resolve) => { + if (res.writableEnded || res.destroyed) { resolve(); return; } + res.write(JSON.stringify(frame) + '\r\n\r\n', () => resolve()); + }); + const sink: StreamSink = { + send: async (frame) => { + if (res.writableEnded || res.destroyed) throw new Error('stream client disconnected'); + await writeFrame(wrapStreamFrame(frame, echo)); + }, + }; + const response = await api.handle(action, params, sink); + await writeFrame(wrapStreamTerminal(response, echo)); + if (!res.writableEnded && !res.destroyed) res.end(); +} diff --git a/packages/onebot/src/network/utils.ts b/packages/onebot/src/network/utils.ts index c65c7c87..64458964 100644 --- a/packages/onebot/src/network/utils.ts +++ b/packages/onebot/src/network/utils.ts @@ -31,6 +31,19 @@ export function safeSend(socket: WebSocket, payload: string, onError?: (err: Err }); } +/** Backpressure-aware send: resolves only once `ws` has flushed `payload` to + * the socket (the `send` callback fires post-write), so a streaming producer + * that awaits this won't outrun a slow client and balloon the send buffer. + * Never rejects — a closed socket or send error resolves quietly (the stream + * sink handles liveness/abort separately so a dead client can't crash the + * per-message handler with an unhandled rejection). */ +export function safeSendAsync(socket: WebSocket, payload: string): Promise { + return new Promise((resolve) => { + if (socket.readyState !== 1 /* WebSocket.OPEN */) { resolve(); return; } + socket.send(payload, () => resolve()); + }); +} + export function safeClose(socket: WebSocket, code = 1000, reason = 'normal'): void { if (socket.readyState === 3 /* CLOSED */ || socket.readyState === 2 /* CLOSING */) return; socket.close(code, reason); diff --git a/packages/onebot/src/network/ws-client-adapter.ts b/packages/onebot/src/network/ws-client-adapter.ts index c68f0049..59682996 100644 --- a/packages/onebot/src/network/ws-client-adapter.ts +++ b/packages/onebot/src/network/ws-client-adapter.ts @@ -8,7 +8,7 @@ import { } from '../event-filter'; import type { JsonObject, WsClientNetwork, WsRole } from '../types'; import { IOneBotNetworkAdapter, type AdapterStatus, type NetworkAdapterContext } from './adapter'; -import { rawDataToString, safeClose, safeSend } from './utils'; +import { rawDataToString, safeClose, safeSend, safeSendAsync } from './utils'; const moduleLog = createLogger('OneBot.WS-Client'); const DEFAULT_RECONNECT_INTERVAL_MS = 5000; @@ -146,8 +146,13 @@ export class WsClientAdapter extends IOneBotNetworkAdapter { if (this.role !== 'Api' && this.role !== 'Universal') return; const text = rawDataToString(raw); if (!text) return; - const response = await this.ctx.api.processRequest(text); - safeSend(socket, response); + // Stream API (#163): one frame for a normal action, N for a streaming one. + // Async send = backpressure; liveness check aborts on disconnect. + await this.ctx.api.processStreamRequest( + text, + (frame) => safeSendAsync(socket, frame), + () => socket.readyState === 1, + ); } private sendBootstrapMetaEvents(socket: WebSocket): void { diff --git a/packages/onebot/src/network/ws-server-adapter.ts b/packages/onebot/src/network/ws-server-adapter.ts index 1904b32f..b2f2056f 100644 --- a/packages/onebot/src/network/ws-server-adapter.ts +++ b/packages/onebot/src/network/ws-server-adapter.ts @@ -9,7 +9,7 @@ import { } from '../event-filter'; import type { JsonObject, WsRole, WsServerNetwork } from '../types'; import { IOneBotNetworkAdapter, type AdapterStatus, type NetworkAdapterContext } from './adapter'; -import { isAuthorized, normalizePath, parseRequestPath, rawDataToString, safeClose, safeSend } from './utils'; +import { isAuthorized, normalizePath, parseRequestPath, rawDataToString, safeClose, safeSend, safeSendAsync } from './utils'; const moduleLog = createLogger('OneBot.WS-Server'); @@ -119,7 +119,9 @@ export class WsServerAdapter extends IOneBotNetworkAdapter { this.connections.set(socket, conn); socket.on('message', (raw: Buffer) => { - void this.handleApiMessage(socket, role, raw); + void this.handleApiMessage(socket, role, raw).catch((err) => { + this.log.warn('[%s] handleApiMessage threw: %s', this.name, err instanceof Error ? (err.stack ?? err.message) : String(err)); + }); }); socket.on('close', () => this.connections.delete(socket)); socket.on('error', (err: Error) => { @@ -135,8 +137,15 @@ export class WsServerAdapter extends IOneBotNetworkAdapter { if (role !== 'Api' && role !== 'Universal') return; const text = rawDataToString(raw); if (!text) return; - const response = await this.ctx.api.processRequest(text); - safeSend(socket, response); + // Stream API (#163) emits multiple frames per request; processStreamRequest + // sends one frame for a normal action and N for a streaming one. The async + // send applies backpressure (awaits flush); the liveness check aborts a + // streaming action once the client goes away. + await this.ctx.api.processStreamRequest( + text, + (frame) => safeSendAsync(socket, frame), + () => socket.readyState === 1, + ); } private sendBootstrapMetaEvents(socket: WebSocket): void { diff --git a/packages/onebot/src/streaming.ts b/packages/onebot/src/streaming.ts new file mode 100644 index 00000000..80d03aa3 --- /dev/null +++ b/packages/onebot/src/streaming.ts @@ -0,0 +1,75 @@ +// streaming — Stream API plumbing (#163, NapCat-compatible). +// +// A handful of OneBot actions ("Stream API") answer a single request with +// MANY response frames instead of one: an upload acks each chunk, a download +// pushes the file out chunk-by-chunk. SnowLuma's normal dispatch is strictly +// one-request → one-response, so streaming needs a small seam: +// +// • a `StreamSink` the action calls to emit intermediate frames, and +// • per-adapter wiring that writes each frame on the wire (HTTP chunked / +// repeated WS messages) instead of closing after the first. +// +// Wire format mirrors NapCat exactly so its client scripts work unchanged: +// every frame is an OB11 envelope carrying `stream:'stream-action'`, and the +// frame `data` is a StreamPacket whose `type` is one of stream/response/error. + +import type { ApiResponse, JsonValue } from './types'; + +/** Marks an OB11 envelope as one frame of a streaming response. */ +export const STREAM_MARK = 'stream-action'; + +/** Frame `type` discriminator carried inside each frame's `data`. */ +export enum StreamStatus { + /** An intermediate chunk frame (transfer in progress). */ + Stream = 'stream', + /** The terminal frame — the stream finished successfully. */ + Response = 'response', + /** A stream was reset / aborted by the caller. */ + Reset = 'reset', + /** The terminal frame — the stream failed. */ + Error = 'error', +} + +/** A sink an action uses to push intermediate stream frames. Each `send` + * carries the raw frame DATA (a StreamPacket: `{ type, data_type, ... }`); + * the network adapter wraps it in an OB11 envelope + the stream marker and + * flushes it on the wire. The action's eventual return value is the terminal + * frame, so do NOT `send` the final result through the sink. */ +export interface StreamSink { + send(frame: JsonValue): Promise; +} + +/** Fallback sink for when a stream action is invoked over a transport that + * can't stream (or by a plain client): intermediate frames are dropped, the + * terminal response is still returned normally. */ +export const NOOP_SINK: StreamSink = { send: async () => { /* dropped */ } }; + +/** Wrap a raw stream-frame `data` payload into an OB11 envelope carrying the + * stream marker (mirrors NapCat's `OB11Response.ok(data, echo, true)`). The + * empty `message`/`wording` pair matches NapCat — clients reading either key + * on an intermediate frame get `''`, never `undefined`. */ +export function wrapStreamFrame(data: JsonValue, echo: JsonValue | undefined): ApiResponse { + const frame: ApiResponse = { status: 'ok', retcode: 0, data, message: '', wording: '', stream: STREAM_MARK }; + if (echo !== undefined) frame.echo = echo; + return frame; +} + +/** Turn an action's terminal `ApiResponse` into the final streaming frame: + * attach the stream marker + echo, mirror `wording` into `message` (NapCat + * sets both), and — on failure — normalise `data` to an error packet + * (`{ type:'error' }`) so NapCat clients see the error `type` while the + * human-readable reason stays in `message`/`wording`. + * + * Contract: on success the action's `data` MUST carry `type:'response'` (the + * client treats that frame as the stream terminator). Stream actions are + * responsible for that, mirroring NapCat's `BaseDownloadStream`. */ +export function wrapStreamTerminal(response: ApiResponse, echo: JsonValue | undefined): ApiResponse { + const frame: ApiResponse = { ...response, stream: STREAM_MARK }; + if (echo !== undefined) frame.echo = echo; + if (response.status === 'failed') { + frame.data = { type: StreamStatus.Error, data_type: 'error' }; + } + frame.message = frame.wording ?? ''; + if (frame.wording === undefined) frame.wording = ''; + return frame; +} diff --git a/packages/onebot/src/types.ts b/packages/onebot/src/types.ts index 2859ef5f..57e99feb 100644 --- a/packages/onebot/src/types.ts +++ b/packages/onebot/src/types.ts @@ -7,6 +7,13 @@ export interface ApiResponse { data: JsonValue; echo?: JsonValue; wording?: string; + /** Mirror of `wording` carried on Stream API frames — NapCat puts the + * message text in BOTH `message` and `wording`, so clients reading either + * key work. Only set on stream frames. */ + message?: string; + /** Only set on Stream API frames — `'stream-action'`. Marks an envelope as + * one frame of a multi-frame streaming response (NapCat Stream API wire). */ + stream?: string; } export interface OneBotRequest { diff --git a/packages/onebot/tests/http-server-content-type.test.ts b/packages/onebot/tests/http-server-content-type.test.ts index 49207745..f3d350ad 100644 --- a/packages/onebot/tests/http-server-content-type.test.ts +++ b/packages/onebot/tests/http-server-content-type.test.ts @@ -18,6 +18,7 @@ function fakeCtx(handleFn: (action: string, params: Record) => const data = handleFn(action, params); return { status: 'ok', retcode: 0, data }; }), + isStreamAction: () => false, } as any, buildLifecycleEvent: vi.fn(() => ({})), buildHeartbeatEvent: vi.fn(() => ({})), diff --git a/packages/onebot/tests/network-bootstrap-meta.test.ts b/packages/onebot/tests/network-bootstrap-meta.test.ts index 9befb0c6..cb26808c 100644 --- a/packages/onebot/tests/network-bootstrap-meta.test.ts +++ b/packages/onebot/tests/network-bootstrap-meta.test.ts @@ -68,7 +68,7 @@ import type { JsonObject, WsClientNetwork, WsServerNetwork } from '../src/types' function ctx(): NetworkAdapterContext { return { uin: '10001', - api: { processRequest: async () => '' } as never, + api: { processStreamRequest: async () => {} } as never, buildLifecycleEvent: (sub) => ({ post_type: 'meta_event', meta_event_type: 'lifecycle', sub_type: sub }), buildHeartbeatEvent: () => ({ post_type: 'meta_event', meta_event_type: 'heartbeat', interval: 5000 }), }; diff --git a/packages/onebot/tests/network-manager.test.ts b/packages/onebot/tests/network-manager.test.ts index a7de011f..758ed382 100644 --- a/packages/onebot/tests/network-manager.test.ts +++ b/packages/onebot/tests/network-manager.test.ts @@ -52,7 +52,7 @@ class FakeAdapter extends IOneBotNetworkAdapter { const NULL_CTX: NetworkAdapterContext = { uin: '10001', - api: { handle: async () => ({ status: 'failed', retcode: 0, data: null }), processRequest: async () => '' } as never, + api: { handle: async () => ({ status: 'failed', retcode: 0, data: null }), processStreamRequest: async () => {} } as never, buildLifecycleEvent: () => ({}), buildHeartbeatEvent: () => ({}), }; diff --git a/packages/onebot/tests/streaming.test.ts b/packages/onebot/tests/streaming.test.ts new file mode 100644 index 00000000..6f4818bc --- /dev/null +++ b/packages/onebot/tests/streaming.test.ts @@ -0,0 +1,167 @@ +// Phase 1 — Stream API plumbing (#163). Verifies the dispatch seam: +// defineStreamAction registers a streaming handler, processStreamRequest +// emits N intermediate frames + a terminal frame (all echo-tagged, marked +// `stream:'stream-action'`), normal actions stay single-frame, and failures +// terminate with an error frame. Wire format mirrors NapCat's Stream API. +import { describe, it, expect } from 'vitest'; +import { ApiHandler, type ApiActionContext } from '../src/api-handler'; +import { defineAction, defineStreamAction, f } from '../src/action-kit'; +import { okResponse, failedResponse } from '../src/types'; +import { wrapStreamFrame, wrapStreamTerminal, STREAM_MARK } from '../src/streaming'; + +function newHandler(): ApiHandler { + // The real register* run lazily (handlers capture ctx, never call it at + // registration), so an empty ctx is enough to exercise the dispatch seam. + return new ApiHandler({} as ApiActionContext); +} + +const parse = (frames: string[]) => frames.map((f) => JSON.parse(f) as Record); + +describe('Stream API plumbing', () => { + it('streams N intermediate frames + a terminal frame, all echo-tagged', async () => { + const h = newHandler(); + defineStreamAction({ + name: 'demo_stream', + params: { n: f.uint() }, + run: async (p, _ctx, _raw, sink) => { + for (let i = 0; i < p.n; i++) await sink.send({ type: 'stream', index: i }); + return okResponse({ type: 'response', total: p.n }); + }, + }).register(h, {} as ApiActionContext); + + const frames: string[] = []; + await h.processStreamRequest( + JSON.stringify({ action: 'demo_stream', params: { n: 2 }, echo: 'e1' }), + (j) => frames.push(j), + ); + + const got = parse(frames); + expect(got).toHaveLength(3); + expect(got[0]).toMatchObject({ status: 'ok', retcode: 0, stream: STREAM_MARK, echo: 'e1', data: { type: 'stream', index: 0 } }); + expect(got[1]).toMatchObject({ stream: STREAM_MARK, echo: 'e1', data: { type: 'stream', index: 1 } }); + expect(got[2]).toMatchObject({ status: 'ok', stream: STREAM_MARK, echo: 'e1', data: { type: 'response', total: 2 } }); + }); + + it('marks the action as a stream action; normal actions are not', () => { + const h = newHandler(); + defineStreamAction({ name: 'demo_stream', params: {}, run: () => okResponse({ type: 'response' }) }).register(h, {} as ApiActionContext); + defineAction({ name: 'demo_normal', params: {}, run: () => okResponse({ ok: true }) }).register(h, {} as ApiActionContext); + expect(h.isStreamAction('demo_stream')).toBe(true); + expect(h.isStreamAction('demo_normal')).toBe(false); + expect(h.isStreamAction('nope')).toBe(false); + }); + + it('a normal action through processStreamRequest emits exactly one un-marked frame', async () => { + const h = newHandler(); + defineAction({ name: 'demo_normal', params: {}, run: () => okResponse({ ok: true }) }).register(h, {} as ApiActionContext); + + const frames: string[] = []; + await h.processStreamRequest(JSON.stringify({ action: 'demo_normal', params: {}, echo: 'e2' }), (j) => frames.push(j)); + + expect(frames).toHaveLength(1); + const r = parse(frames)[0]; + expect(r).toMatchObject({ status: 'ok', echo: 'e2', data: { ok: true } }); + expect(r.stream).toBeUndefined(); + }); + + it('a throwing stream action terminates with a single error frame', async () => { + const h = newHandler(); + defineStreamAction({ + name: 'demo_boom', + params: {}, + run: async () => { throw new Error('boom'); }, + }).register(h, {} as ApiActionContext); + + const frames: string[] = []; + await h.processStreamRequest(JSON.stringify({ action: 'demo_boom', params: {}, echo: 'e3' }), (j) => frames.push(j)); + + expect(frames).toHaveLength(1); + expect(parse(frames)[0]).toMatchObject({ + status: 'failed', stream: STREAM_MARK, echo: 'e3', wording: 'boom', data: { type: 'error', data_type: 'error' }, + }); + }); + + it('emits already-sent frames before the terminal error when a stream throws mid-way', async () => { + const h = newHandler(); + defineStreamAction({ + name: 'demo_partial', + params: {}, + run: async (_p, _ctx, _raw, sink) => { + await sink.send({ type: 'stream', index: 0 }); + throw new Error('mid'); + }, + }).register(h, {} as ApiActionContext); + + const frames: string[] = []; + await h.processStreamRequest(JSON.stringify({ action: 'demo_partial', params: {} }), (j) => frames.push(j)); + + const got = parse(frames); + expect(got).toHaveLength(2); + expect(got[0]).toMatchObject({ stream: STREAM_MARK, data: { type: 'stream', index: 0 } }); + expect(got[1]).toMatchObject({ status: 'failed', data: { type: 'error' } }); + }); + + it('stream param validation fails terminate with an error frame', async () => { + const h = newHandler(); + defineStreamAction({ name: 'demo_needs', params: { n: f.uint() }, run: () => okResponse({ type: 'response' }) }).register(h, {} as ApiActionContext); + + const frames: string[] = []; + await h.processStreamRequest(JSON.stringify({ action: 'demo_needs', params: {} }), (j) => frames.push(j)); + expect(frames).toHaveLength(1); + expect(parse(frames)[0]).toMatchObject({ status: 'failed', stream: STREAM_MARK, data: { type: 'error' } }); + }); + + it('aborts a stream when isAlive turns false (client disconnect)', async () => { + const h = newHandler(); + let sent = 0; + defineStreamAction({ + name: 'demo_abort', + params: {}, + run: async (_p, _ctx, _raw, sink) => { + for (let i = 0; i < 5; i++) { await sink.send({ type: 'stream', index: i }); sent++; } + return okResponse({ type: 'response' }); + }, + }).register(h, {} as ApiActionContext); + + const frames: string[] = []; + let alive = true; + await h.processStreamRequest( + JSON.stringify({ action: 'demo_abort', params: {} }), + (j) => { frames.push(j); if (frames.length >= 2) alive = false; }, + () => alive, + ); + + expect(sent).toBeLessThan(5); // loop aborted once the client "left" + const got = parse(frames); + expect(got[got.length - 1]).toMatchObject({ status: 'failed', data: { type: 'error' } }); + }); + + it('wrapStreamFrame / wrapStreamTerminal shape (NapCat message+wording mirror)', () => { + expect(wrapStreamFrame({ type: 'stream', x: 1 }, 'e')).toEqual({ + status: 'ok', retcode: 0, data: { type: 'stream', x: 1 }, message: '', wording: '', stream: STREAM_MARK, echo: 'e', + }); + // no echo → no echo key + expect(wrapStreamFrame({ type: 'stream' }, undefined)).toEqual({ + status: 'ok', retcode: 0, data: { type: 'stream' }, message: '', wording: '', stream: STREAM_MARK, + }); + // failed terminal → data normalised to an error packet; reason in BOTH message + wording + expect(wrapStreamTerminal(failedResponse(1404, 'nope'), 'e')).toMatchObject({ + status: 'failed', retcode: 1404, message: 'nope', wording: 'nope', stream: STREAM_MARK, echo: 'e', + data: { type: 'error', data_type: 'error' }, + }); + // ok terminal → data passes through, message/wording empty + expect(wrapStreamTerminal(okResponse({ type: 'response', a: 1 }), undefined)).toMatchObject({ + status: 'ok', stream: STREAM_MARK, message: '', data: { type: 'response', a: 1 }, + }); + }); + + it('malformed stream requests emit one bad-request frame', async () => { + const h = newHandler(); + const frames: string[] = []; + await h.processStreamRequest('not json', (j) => frames.push(j)); + await h.processStreamRequest('', (j) => frames.push(j)); + await h.processStreamRequest(JSON.stringify({ params: {} }), (j) => frames.push(j)); // no action + expect(frames).toHaveLength(3); + for (const f of parse(frames)) expect(f).toMatchObject({ status: 'failed' }); + }); +}); diff --git a/packages/onebot/tests/ws-client-stale-close.test.ts b/packages/onebot/tests/ws-client-stale-close.test.ts index 18d2108d..49407ca2 100644 --- a/packages/onebot/tests/ws-client-stale-close.test.ts +++ b/packages/onebot/tests/ws-client-stale-close.test.ts @@ -70,7 +70,7 @@ import type { WsClientNetwork } from '../src/types'; function ctx(): NetworkAdapterContext { return { uin: '10001', - api: { processRequest: async () => '' } as never, + api: { processStreamRequest: async () => {} } as never, buildLifecycleEvent: () => ({}), buildHeartbeatEvent: () => ({}), };