From 9b5ce9b50264427bbc2b0ca401826065884f8d4a Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Fri, 23 Oct 2026 13:43:35 +0800 Subject: [PATCH] feat(sdk): support streaming file transfers --- AGENTS.md | 1 + packages/onebot/tests/sdk-streaming.test.ts | 173 ++++++++++ packages/sdk/CHANGELOG.md | 1 + packages/sdk/STREAMING.md | 130 ++++++++ packages/sdk/VERSIONING.md | 2 +- packages/sdk/package.json | 1 + packages/sdk/src/client/api-client.ts | 45 ++- packages/sdk/src/client/file-streams.ts | 329 +++++++++++++++++++ packages/sdk/src/client/http-client.ts | 62 +++- packages/sdk/src/client/websocket-client.ts | 157 ++++++++- packages/sdk/src/internal/streaming.ts | 260 +++++++++++++++ packages/sdk/src/types/index.ts | 1 + packages/sdk/src/types/streaming.ts | 53 +++ packages/sdk/tests/file-streams.test.ts | 288 ++++++++++++++++ packages/sdk/tests/fixtures/action-types.mts | 26 ++ packages/sdk/tests/http-streaming.test.ts | 222 +++++++++++++ packages/sdk/tests/websocket-client.test.ts | 223 +++++++++++++ 17 files changed, 1960 insertions(+), 14 deletions(-) create mode 100644 packages/onebot/tests/sdk-streaming.test.ts create mode 100644 packages/sdk/STREAMING.md create mode 100644 packages/sdk/src/client/file-streams.ts create mode 100644 packages/sdk/src/internal/streaming.ts create mode 100644 packages/sdk/src/types/streaming.ts create mode 100644 packages/sdk/tests/file-streams.test.ts create mode 100644 packages/sdk/tests/http-streaming.test.ts diff --git a/AGENTS.md b/AGENTS.md index 0be9a19c..bb1c7f6e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -75,6 +75,7 @@ Use focused tests while iterating. Before handing off a code change, run the ful - **Documentation-only changes:** verify links, paths, commands and `git diff --check`; runtime tests do not validate prose. - **Shared types or actions:** run workspace typechecking as well as behavior tests. A passing Vitest run does not prove type correctness. - **SDK packaging:** keep the SDK's `NodeNext` build and explicit `.js` relative imports, including type-only exports. Its package tests compile fresh output and check native Node.js imports, consumer declarations, and complete action-map coverage. The SDK action catalog must match OneBot's executable names; `packages/onebot/tests/sdk-actions.test.ts` checks this without adding a runtime dependency on server code. SDK parameter/result contracts remain independent of server types; keep consumer examples under `packages/sdk/tests/fixtures` aligned with the server parser checks in `packages/onebot/tests/sdk-action-params.test.ts`. +- **SDK streams:** preserve the single-response client methods and use the additive interfaces in [STREAMING.md](packages/sdk/STREAMING.md) for file transfers. Test framing, cancellation, bounded buffering and incomplete transfers in the SDK; `packages/onebot/tests/sdk-streaming.test.ts` checks real adapter/handler round trips in temporary storage without QQ. - **Protocol codecs:** keep the existing Proton integration in tests so they exercise the same transformations as production. Import real implementations; do not duplicate the algorithm in the test. - **Native coverage:** `pnpm test:native` is separate from default protocol tests and requires compatible native artifacts. A missing artifact or unavailable target is a reported limitation, not permission to bypass the check. - **Live QQ behavior:** fixture tests and a successful API response do not prove delivery or client rendering. State what was tested locally and what still needs a real client. diff --git a/packages/onebot/tests/sdk-streaming.test.ts b/packages/onebot/tests/sdk-streaming.test.ts new file mode 100644 index 00000000..afaf0903 --- /dev/null +++ b/packages/onebot/tests/sdk-streaming.test.ts @@ -0,0 +1,173 @@ +import { createHash } from 'node:crypto'; +import { readFile } from 'node:fs/promises'; +import type { Server } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { Readable } from 'node:stream'; +import { afterEach, describe, expect, it } from 'vitest'; +import { + SnowLumaApiError, + SnowLumaHttpClient, + SnowLumaWebSocketClient, + type SnowLumaApiClient, +} from '../../sdk/src/index'; +import type { ApiActionContext } from '../src/api-handler'; +import { actions as uploadActions, __test as uploadTest } from '../src/actions/stream-file'; +import { actions as downloadActions } from '../src/actions/stream-download'; +import { HttpServerAdapter } from '../src/network/http-server-adapter'; +import { okResponse, type HttpServerNetwork } from '../src/types'; +import { createCompiledTestHandler, testAction } from './helpers/compiled-action-handler'; + +// Real server adapters, parsers and disk handlers, isolated by tests/setup.ts. +// No QQ session or protocol requests are involved. +describe('SDK streaming against OneBot', () => { + let adapter: HttpServerAdapter | undefined; + let ws: SnowLumaWebSocketClient | undefined; + + afterEach(async () => { + ws?.close(); + ws = undefined; + await adapter?.close(); + adapter = undefined; + + for (const key of uploadTest.uploads.keys()) { + uploadTest.cleanupUpload(key, true); + } + }); + + async function connect(transport: 'http' | 'ws'): Promise { + const api = createCompiledTestHandler({} as ApiActionContext, [ + ...uploadActions, + ...downloadActions, + testAction('sdk_ping', () => okResponse({ pong: true })), + ]); + const config: HttpServerNetwork = { + name: 'sdk-streaming', + host: '127.0.0.1', + port: 0, + path: '/', + enabled: true, + enableWebSocket: true, + accessToken: 'sdk-test', + messageFormat: 'array', + reportSelfMessage: false, + }; + adapter = new HttpServerAdapter(config.name, config, { + uin: '10001', + api, + buildLifecycleEvent: () => ({}), + buildHeartbeatEvent: () => ({}), + }); + await adapter.open(); + const { port } = (adapter as unknown as { server: Server }).server.address() as AddressInfo; + + if (transport === 'http') { + return new SnowLumaHttpClient({ + baseUrl: `http://127.0.0.1:${port}/`, + accessToken: 'sdk-test', + }); + } + + ws = new SnowLumaWebSocketClient({ + url: `ws://127.0.0.1:${port}/api`, + accessToken: 'sdk-test', + }); + await ws.connect(); + return ws; + } + + it.each(['http', 'ws'] as const)( + '%s uploads and downloads binary data incrementally', + async (transport) => { + const client = await connect(transport); + const bytes = Uint8Array.from({ length: 210007 }, (_, index) => index % 256); + const sha256 = createHash('sha256').update(bytes).digest('hex'); + const result = await client.uploadFileStream( + Readable.from([ + bytes.subarray(0, 17), + bytes.subarray(17, 100009), + bytes.subarray(100009), + ]), + { + fileSize: bytes.length, + chunkSize: 16384, + filename: '二进制.bin', + expectedSha256: sha256, + fileRetention: 0, + }, + ); + expect(result.sha256).toBe(sha256); + expect(result.total_chunks).toBe(Math.ceil(bytes.length / 16384)); + expect(await readFile(result.file_path)).toEqual(Buffer.from(bytes)); + + for (const action of [ + 'download_file_stream', + 'download_file_image_stream', + 'download_file_record_stream', + ] as const) { + const stream = client.downloadFileStream( + { file: result.file_path, chunk_size: 8192 }, + { action }, + ); + const output: Uint8Array[] = []; + + while (true) { + const next = await stream.next(); + + if (next.done) { + expect(next.value).toMatchObject({ + total_bytes: bytes.length, + total_chunks: output.length, + }); + break; + } + + expect(next.value.byteLength).toBeLessThanOrEqual(8192); + output.push(next.value); + } + + expect(output.length).toBeGreaterThan(1); + expect(Buffer.concat(output)).toEqual(Buffer.from(bytes)); + } + + await expect(client.raw('sdk_ping')).resolves.toEqual({ pong: true }); + }, + ); + + it.each(['http', 'ws'] as const)('%s handles empty files', async (transport) => { + const client = await connect(transport); + const result = await client.uploadFileStream(new Uint8Array(), { fileRetention: 0 }); + expect(result.file_size).toBe(0); + const stream = client.downloadFileStream({ file: result.file_path }); + expect(await stream.next()).toMatchObject({ + done: true, + value: { total_bytes: 0, total_chunks: 0 }, + }); + }); + + it.each(['http', 'ws'] as const)( + '%s preserves a server error after partial download output', + async (transport) => { + const client = await connect(transport); + const received: unknown[] = []; + + const run = async () => { + for await (const frame of client.stream('test_download_stream', { error: true })) { + received.push(frame); + } + }; + + await expect(run()).rejects.toBeInstanceOf(SnowLumaApiError); + expect(received).toHaveLength(10); + }, + ); + + it.each(['http', 'ws'] as const)( + '%s propagates server hash validation failure', + async (transport) => { + const client = await connect(transport); + await expect( + client.uploadFileStream(new Uint8Array([1]), { expectedSha256: '0'.repeat(64) }), + ).rejects.toThrow('sha256 mismatch'); + }, + ); +}); diff --git a/packages/sdk/CHANGELOG.md b/packages/sdk/CHANGELOG.md index 5f4c65e3..d009de0f 100644 --- a/packages/sdk/CHANGELOG.md +++ b/packages/sdk/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- Add HTTP and WebSocket `requestStream()` / `stream()`, incremental `uploadFileStream()` / `downloadFileStream()` helpers, cancellation, bounded response buffering, and transfer validation. Existing single-response methods remain unchanged; see [STREAMING.md](STREAMING.md). - Add parameter and result types for the remaining 57 actions and three aliases, covering every registered action name while preserving existing client methods and type mappings. - Describe album pagination and media, Qzone, flash transfers, stickers, group settings, and stream frame payloads; retain numeric-string inputs and signed message IDs. - Fix native Node.js ESM imports for the package root and all supported subpaths without requiring a bundler. diff --git a/packages/sdk/STREAMING.md b/packages/sdk/STREAMING.md new file mode 100644 index 00000000..a2317005 --- /dev/null +++ b/packages/sdk/STREAMING.md @@ -0,0 +1,130 @@ +# Streaming files + +Both `SnowLumaHttpClient` and `SnowLumaWebSocketClient` support incremental uploads +and downloads. Existing `request()`, `call()`, `raw()`, `rawResponse()` and +`downloadFile()` keep their single-response behavior. Use the new methods below +when a response contains multiple frames. + +## Upload + +```ts +import { createReadStream } from 'node:fs'; +import { stat } from 'node:fs/promises'; +import { SnowLumaHttpClient } from '@snowluma/sdk'; + +const client = new SnowLumaHttpClient({ baseUrl: 'http://127.0.0.1:3000/' }); +const controller = new AbortController(); +const path = './photo.png'; +const file = await client.uploadFileStream( + createReadStream(path, { signal: controller.signal }), + { + fileSize: (await stat(path)).size, + filename: 'photo.png', + signal: controller.signal, + onProgress: ({ uploadedBytes, fileSize }) => { + console.log(`${uploadedBytes}/${fileSize}`); + }, + }, +); +console.log(file.file_path, file.file_size, file.sha256); +``` + +`uploadFileStream(source, options?)` accepts `Uint8Array` (including Node.js +`Buffer`), `Blob`, `ReadableStream` or `AsyncIterable`. +Byte arrays and blobs provide their own size. Stream sources require `fileSize`: +the server needs the total chunk count before accepting the first chunk. The SDK +rejects input whose actual length differs from the declared size. + +The default chunk size is 64 KiB. `chunkSize` accepts 1 through 1048576 bytes, +keeping each encoded request below the standard server HTTP body limit. Input is +read incrementally; chunks are acknowledged sequentially, and `onProgress` is +awaited before continuing. Empty files are supported. Completion is a separate +request and returns `UploadFileStreamComplete`, including the **server's** local +file path; it does not write a local client file or send a QQ message. + +Optional settings are `filename`, `expectedSha256`, `fileRetention` (milliseconds; +zero disables server expiry), and `streamId`. The default server retention is five +minutes. A generated ID identifies each upload; an explicit ID must be fresh, not +an existing upload. The helper does not resume or automatically retry transfers. +The server validates `expectedSha256` when provided; the SDK checks the completed +metadata and returned digest against that expectation. + +`signal` cancels requests and closes readable sources. For custom async iterables, +pass the same signal to the source and ensure pending reads and iterator cleanup +can finish on cancellation (as the `createReadStream` example does). Cancellation +cannot force an arbitrary producer to resolve its own promises. A failed or +cancelled upload is not finalized; partial server uploads expire after the +server's idle timeout (currently ten minutes). The helper does not issue a reset +request or delete a completed server file automatically. + +## Download + +```ts +import { createWriteStream } from 'node:fs'; +import { pipeline } from 'node:stream/promises'; + +await pipeline( + client.downloadFileStream( + { file: file.file_path, chunk_size: 64 * 1024 }, + { onInfo: (info) => console.log(info.file_name, info.file_size) }, + ), + createWriteStream('./downloaded.png'), +); +``` + +`downloadFileStream(params, options?)` yields `Uint8Array` chunks and returns +`DownloadFileStreamComplete` as the iterator's final value. `for await` and +`pipeline()` consume chunks and verify completion, but discard that return value; +call `iterator.next()` manually if you need `total_bytes` and `total_chunks` from +the final result. `onInfo` runs before any file bytes are yielded. + +Parameters accept a server-side `file` or `file_id`, plus optional `chunk_size`. +The `action` option selects `download_file_stream` (default), +`download_file_image_stream`, or `download_file_record_stream`. + +The helper decodes base64 and checks chunk order, chunk lengths and final counts. +A zero initial file size can mean unknown size; final byte and chunk totals still +must match. Missing metadata, malformed frames, truncation and server failures +throw; bytes already consumed are not rolled back. For durable output, write to +a temporary file and rename it after the iteration succeeds. + +## Individual frames + +```ts +for await (const frame of client.stream('test_download_stream', {})) { + console.log(frame.type, frame.data_type); +} + +for await (const response of client.requestStream('test_download_stream', {})) { + console.log(response.status, response.retcode, response.data); +} +``` + +- `requestStream()` yields response envelopes, including a final failed envelope. +- `stream()` yields typed payloads and throws `SnowLumaApiError` (or its auth + subtype) for failed envelopes, including errors after earlier chunks. +- Both support the five streaming actions: the three downloads, + `test_download_stream`, and `upload_file_stream`. Each upload request returns + just one acknowledgement, including `type: 'stream'` chunk ACKs. Low-level + callers must finalize in a separate request. Reset retains the server's + existing failed-envelope behavior. +- `AbortSignal`, `echo`, and `timeoutMs` work as on ordinary requests. Timeout + covers an active streaming request (including WebSocket connection setup); + uploads apply it to each chunk/finalization request. Zero disables timeout. + Source reads and application callbacks are not timed by the request timeout. +- Stopping iteration releases client resources. HTTP also cancels its response + body. WebSocket stops collecting that request without closing the shared socket; + the protocol has no per-request download cancellation message, so the server + may continue sending remaining frames. Use unique echoes for transfers; do not + reuse an aborted transfer's echo while late frames may still arrive. +- WebSocket disconnection fails active transfers; reconnect does not replay them. +- `maxBufferBytes` defaults to 32 MiB and bounds each HTTP frame or the queued + WebSocket payloads. `maxBufferedFrames` defaults to 256 for WebSocket. An + overflow throws instead of dropping frames or buffering the entire file. + HTTP reads follow consumption; WebSocket has no per-request backpressure, so + prefer HTTP for large downloads to a slow destination. + +All new types are exported from `@snowluma/sdk` and `@snowluma/sdk/types`. Custom +`SnowLumaApiClient` subclasses can opt in by implementing `requestStream()`; +request-only subclasses remain compatible and receive an explicit unsupported +error if streaming is attempted. diff --git a/packages/sdk/VERSIONING.md b/packages/sdk/VERSIONING.md index e2075b16..aa72c72d 100644 --- a/packages/sdk/VERSIONING.md +++ b/packages/sdk/VERSIONING.md @@ -43,6 +43,6 @@ If the package version already exists on npm, the workflow skips publishing that This local catalog is not a capability check against a connected server. Every known action name has a parameter and result entry in `SnowLumaActionMap`, including aliases. Existing broad JSON contracts are retained for compatibility. New parameter types describe accepted input forms, including numeric strings and primitive coercions; server-side range and state validation still applies. Custom actions and dynamically selected names remain supported through `raw()` and `rawResponse()`. Adding an action type is a minor change unless it changes an existing method signature or return type. -Stream result types describe individual frame payloads, including intermediate and terminal frames. These declarations do not add streaming transport support to the single-response HTTP and WebSocket clients; applications consuming a full download must use a transport that reads every frame. Stream reset/error envelopes retain their existing error behavior. Album counts and IDs returned as decimal strings stay strings, and ordinary message IDs remain signed numbers. +Stream result types describe individual frame payloads, including intermediate and terminal frames. Both built-in clients support `requestStream()` (envelopes) and `stream()` (payloads with API error handling), plus incremental file upload/download helpers; see [STREAMING.md](STREAMING.md). Existing `request()`, `call()`, `raw()`, `rawResponse()` and `downloadFile()` keep their single-response contracts. Custom subclasses that only implement `request()` remain valid; the default `requestStream()` explicitly reports unsupported streaming. Stream reset/error envelopes retain their existing error behavior. Album counts and IDs returned as decimal strings stay strings, and ordinary message IDs remain signed numbers. The package consumer tests check complete action-map coverage, inferred results, compatibility aliases, and invalid input examples with both NodeNext and bundler resolution. The same parameter examples are checked against the server's real action parsers without starting QQ or executing actions. diff --git a/packages/sdk/package.json b/packages/sdk/package.json index fbadf024..2d83736f 100644 --- a/packages/sdk/package.json +++ b/packages/sdk/package.json @@ -71,6 +71,7 @@ "dist/types", "README.md", "CHANGELOG.md", + "STREAMING.md", "VERSIONING.md" ], "scripts": { diff --git a/packages/sdk/src/client/api-client.ts b/packages/sdk/src/client/api-client.ts index f2bf8dfe..3c18a2dd 100644 --- a/packages/sdk/src/client/api-client.ts +++ b/packages/sdk/src/client/api-client.ts @@ -1,10 +1,13 @@ -import { createSnowLumaApiError } from '../errors.js'; +import { createSnowLumaApiError, SnowLumaConnectionError } from '../errors.js'; import { normalizeMessage } from '../messages/index.js'; +import { downloadFileStream, uploadFileStream } from './file-streams.js'; import type { ActionParams, ActionResult, ApiResponse, DomainParams, + DownloadFileStreamOptions, + DownloadFileStreamParams, DownloadFileParams, ForwardMessageParams, GetForwardMsgParams, @@ -19,6 +22,11 @@ import type { SendMsgParams, SetGroupReactionParams, SnowLumaAction, + SnowLumaStreamAction, + StreamApiResponse, + StreamRequestOptions, + UploadFileSource, + UploadFileStreamOptions, } from '../types/index.js'; /** @@ -35,6 +43,41 @@ export abstract class SnowLumaApiClient { options?: RequestOptions, ): Promise>>; + /** Returns every stream envelope, including a terminal failed response. + * Custom request-only subclasses remain valid and can opt into streaming. */ + requestStream( + _action: TAction, + _params: ActionParams, + _options?: StreamRequestOptions, + ): AsyncIterableIterator>> { + throw new SnowLumaConnectionError('This client transport does not implement streaming'); + } + + /** Returns every stream payload; failed responses throw SnowLumaApiError. */ + async *stream( + action: TAction, + params: ActionParams, + options?: StreamRequestOptions, + ): AsyncGenerator, void, void> { + for await (const response of this.requestStream(action, params, options)) { + if (response.status !== 'ok' || response.retcode !== 0) { + throw createSnowLumaApiError(response); + } + + yield response.data; + } + } + + /** Uploads bytes incrementally and returns the server's completed file metadata. */ + uploadFileStream(source: UploadFileSource, options?: UploadFileStreamOptions) { + return uploadFileStream(this, source, options); + } + + /** Yields decoded byte chunks; the iterator's return value is completion metadata. */ + downloadFileStream(params: DownloadFileStreamParams, options?: DownloadFileStreamOptions) { + return downloadFileStream(this, params, options); + } + /** Sends a OneBot action and returns data, throwing SnowLumaApiError on failed responses. */ async call( action: TAction, diff --git a/packages/sdk/src/client/file-streams.ts b/packages/sdk/src/client/file-streams.ts new file mode 100644 index 00000000..628c7d66 --- /dev/null +++ b/packages/sdk/src/client/file-streams.ts @@ -0,0 +1,329 @@ +import { SnowLumaParseError } from '../errors.js'; +import { raceWithAbort, throwIfAborted } from '../internal/abort.js'; +import { readBytes } from '../internal/streaming.js'; +import type { + DownloadFileStreamComplete, + DownloadFileStreamInfo, + DownloadFileStreamOptions, + DownloadFileStreamParams, + UploadFileSource, + UploadFileStreamComplete, + UploadFileStreamOptions, + UploadFileStreamParams, + UploadFileStreamResult, +} from '../types/index.js'; +import type { SnowLumaApiClient } from './api-client.js'; + +function encodeBase64(bytes: Uint8Array): string { + const parts: string[] = []; + + for (let i = 0; i < bytes.length; i += 0x8000) { + parts.push(String.fromCharCode(...bytes.subarray(i, i + 0x8000))); + } + + return btoa(parts.join('')); +} + +function decodeBase64(value: string): Uint8Array { + try { + const binary = atob(value); + + if (btoa(binary) !== value) { + throw new Error('non-canonical base64'); + } + + return Uint8Array.from(binary, (char) => char.charCodeAt(0)); + } catch (cause) { + throw new SnowLumaParseError('Invalid base64 in download chunk', { cause }); + } +} + +function isSize(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0; +} + +async function* sourceBytes( + source: UploadFileSource, + signal?: AbortSignal, +): AsyncGenerator { + throwIfAborted(signal); + + if (source instanceof Uint8Array) { + yield source; + return; + } + + if (source instanceof Blob) { + yield* readBytes(source.stream(), signal); + return; + } + + if ('getReader' in source) { + yield* readBytes(source, signal); + return; + } + + const iterator = source[Symbol.asyncIterator](); + let ended = false; + + try { + while (true) { + throwIfAborted(signal); + const next = await raceWithAbort(iterator.next(), signal); + + if (next.done) { + ended = true; + return; + } + + yield next.value; + } + } finally { + if (!ended) { + await iterator.return?.(); + } + } +} + +async function* uploadChunks( + source: UploadFileSource, + fileSize: number, + chunkSize: number, + signal?: AbortSignal, +): AsyncGenerator { + let buffer = new Uint8Array(Math.min(chunkSize, fileSize)); + let used = 0; + let received = 0; + + for await (const bytes of sourceBytes(source, signal)) { + if (!(bytes instanceof Uint8Array)) { + throw new TypeError('Upload source must yield Uint8Array chunks'); + } + + received += bytes.byteLength; + + if (received > fileSize) { + throw new RangeError('Upload source is longer than fileSize'); + } + + for (let offset = 0; offset < bytes.length;) { + const count = Math.min(buffer.length - used, bytes.length - offset); + buffer.set(bytes.subarray(offset, offset + count), used); + used += count; + offset += count; + + if (used === buffer.length) { + yield buffer; + buffer = new Uint8Array(chunkSize); + used = 0; + } + } + } + + if (received !== fileSize) { + throw new RangeError('Upload source is shorter than fileSize'); + } + + if (used > 0 || fileSize === 0) { + yield buffer.subarray(0, used); + } +} + +async function uploadResponse( + client: SnowLumaApiClient, + params: UploadFileStreamParams, + options: UploadFileStreamOptions, +): Promise { + let result: UploadFileStreamResult | undefined; + + for await (const frame of client.stream('upload_file_stream', params, options)) { + if (result !== undefined) { + throw new SnowLumaParseError('Upload request returned multiple acknowledgements'); + } + + result = frame; + } + + if (!result) { + throw new SnowLumaParseError('Upload request returned no acknowledgement'); + } + + return result; +} + +export async function uploadFileStream( + client: SnowLumaApiClient, + source: UploadFileSource, + options: UploadFileStreamOptions = {}, +): Promise { + throwIfAborted(options.signal); + const knownSize = + source instanceof Uint8Array + ? source.byteLength + : source instanceof Blob + ? source.size + : undefined; + const fileSize = options.fileSize ?? knownSize; + const chunkSize = options.chunkSize ?? 64 * 1024; + const streamId = options.streamId ?? globalThis.crypto.randomUUID(); + + if (!isSize(fileSize)) { + throw new RangeError('fileSize must be a non-negative safe integer for streaming sources'); + } + + if (knownSize !== undefined && knownSize !== fileSize) { + throw new RangeError('fileSize does not match the upload source'); + } + + if (!Number.isSafeInteger(chunkSize) || chunkSize < 1 || chunkSize > 1024 * 1024) { + throw new RangeError('chunkSize must be an integer between 1 and 1048576'); + } + + if (!/^[A-Za-z0-9_-]{1,128}$/.test(streamId)) { + throw new TypeError('streamId must contain 1–128 letters, digits, underscores or hyphens'); + } + + const totalChunks = Math.max(1, Math.ceil(fileSize / chunkSize)); + let uploadedChunks = 0; + let uploadedBytes = 0; + + for await (const bytes of uploadChunks(source, fileSize, chunkSize, options.signal)) { + throwIfAborted(options.signal); + const response = await uploadResponse( + client, + { + stream_id: streamId, + total_chunks: totalChunks, + file_size: fileSize, + filename: options.filename, + expected_sha256: options.expectedSha256, + file_retention: options.fileRetention, + chunk_index: uploadedChunks, + chunk_data: encodeBase64(bytes), + }, + options, + ); + uploadedChunks++; + uploadedBytes += bytes.byteLength; + + if ( + response.type !== 'stream' || + response.status !== 'chunk_received' || + response.stream_id !== streamId || + response.total_chunks !== totalChunks || + response.received_chunks !== uploadedChunks + ) { + throw new SnowLumaParseError('Upload acknowledgement does not match the sent chunk'); + } + + await options.onProgress?.({ + streamId, + uploadedBytes, + fileSize, + uploadedChunks, + totalChunks, + }); + } + + throwIfAborted(options.signal); + // A chunk request returns early in the server handler. Finalization must + // be a separate request, after the source has ended and every ACK arrived. + const result = await uploadResponse( + client, + { stream_id: streamId, is_complete: true }, + options, + ); + + if ( + result.type !== 'response' || + result.status !== 'file_complete' || + result.stream_id !== streamId || + result.received_chunks !== totalChunks || + result.total_chunks !== totalChunks || + result.file_size !== fileSize || + typeof result.file_path !== 'string' || + !result.file_path || + typeof result.file_name !== 'string' || + typeof result.sha256 !== 'string' || + !/^[a-f0-9]{64}$/i.test(result.sha256) || + (options.expectedSha256 !== undefined && + result.sha256.toLowerCase() !== options.expectedSha256.toLowerCase()) + ) { + throw new SnowLumaParseError('Invalid upload completion metadata'); + } + + return result; +} + +export async function* downloadFileStream( + client: SnowLumaApiClient, + params: DownloadFileStreamParams, + options: DownloadFileStreamOptions = {}, +): AsyncGenerator { + let info: DownloadFileStreamInfo | undefined; + let chunks = 0; + let bytes = 0; + + for await (const frame of client.stream( + options.action ?? 'download_file_stream', + params, + options, + )) { + throwIfAborted(options.signal); + + if (frame.type === 'stream' && frame.data_type === 'file_info') { + if ( + info || + typeof frame.file_name !== 'string' || + !isSize(frame.file_size) || + !isSize(frame.chunk_size) || + frame.chunk_size === 0 + ) { + throw new SnowLumaParseError('Invalid or duplicate download file info'); + } + + info = frame; + await options.onInfo?.(info); + } else if (frame.type === 'stream' && frame.data_type === 'file_chunk') { + if ( + !info || + frame.index !== chunks || + typeof frame.data !== 'string' || + frame.base64_size !== frame.data.length || + typeof frame.progress !== 'number' || + !Number.isFinite(frame.progress) || + frame.progress < 0 || + frame.progress > 100 + ) { + throw new SnowLumaParseError('Invalid or out-of-order download chunk'); + } + + const data = decodeBase64(frame.data); + bytes += data.byteLength; + chunks++; + + if (frame.size !== data.byteLength || (info.file_size > 0 && bytes > info.file_size)) { + throw new SnowLumaParseError('Download chunk size does not match its metadata'); + } + + yield data; + } else if (frame.type === 'response' && frame.data_type === 'file_complete') { + if ( + !info || + frame.file_name !== info.file_name || + frame.total_chunks !== chunks || + frame.total_bytes !== bytes || + (info.file_size > 0 && info.file_size !== bytes) || + typeof frame.message !== 'string' + ) { + throw new SnowLumaParseError('Download completion does not match received bytes'); + } + + return frame; + } else { + throw new SnowLumaParseError('Unexpected download frame'); + } + } + + throw new SnowLumaParseError('Download ended before file completion'); +} diff --git a/packages/sdk/src/client/http-client.ts b/packages/sdk/src/client/http-client.ts index 522c07ba..4ba68a73 100644 --- a/packages/sdk/src/client/http-client.ts +++ b/packages/sdk/src/client/http-client.ts @@ -6,8 +6,9 @@ import { SnowLumaTransportError, } from '../errors.js'; import { SnowLumaApiClient, toJsonObject } from './api-client.js'; -import { createRequestAbortState, isAbortError } from '../internal/abort.js'; +import { createRequestAbortState, isAbortError, throwIfAborted } from '../internal/abort.js'; import { assertApiResponse } from '../internal/json.js'; +import { isStreamTerminal, readHttpFrames, streamLimits } from '../internal/streaming.js'; import type { ActionParams, ActionResult, @@ -16,6 +17,9 @@ import type { RequestOptions, SnowLumaAction, SnowLumaClientOptions, + SnowLumaStreamAction, + StreamApiResponse, + StreamRequestOptions, } from '../types/index.js'; /** Options for the HTTP transport client. */ @@ -109,6 +113,62 @@ export class SnowLumaHttpClient extends SnowLumaApiClient { } } + override async *requestStream( + action: TAction, + params: ActionParams, + options?: StreamRequestOptions, + ): AsyncGenerator>, void, void> { + const { maxBufferBytes } = streamLimits(options); + const timeoutMs = options?.timeoutMs ?? this.requestTimeoutMs; + const abort = createRequestAbortState(timeoutMs, options?.signal); + + try { + throwIfAborted(abort.signal); + const response = await this.fetchImpl(this.baseUrl, { + method: 'POST', + headers: this.createHeaders(), + body: JSON.stringify({ action, params, echo: options?.echo }), + signal: abort.signal, + }); + + if (!response.body) { + throw new SnowLumaParseError('SnowLuma HTTP stream has no response body'); + } + + for await (const frame of readHttpFrames(response.body, maxBufferBytes, abort.signal)) { + throwIfAborted(abort.signal); + const terminal = isStreamTerminal(action, frame); + yield frame as StreamApiResponse>; + + if (terminal) { + return; + } + } + + throw new SnowLumaParseError('SnowLuma HTTP stream ended before its terminal frame'); + } catch (error) { + if (abort.isTimedOut()) { + throw new SnowLumaTimeoutError( + `SnowLuma stream timed out after ${timeoutMs}ms`, + timeoutMs, + { cause: error }, + ); + } + + if (abort.isExternallyAborted()) { + throw new SnowLumaAbortError('SnowLuma stream aborted', { cause: error }); + } + + if (error instanceof SnowLumaTransportError) { + throw error; + } + + throw new SnowLumaConnectionError('SnowLuma HTTP stream failed', { cause: error }); + } finally { + abort.cleanup(); + } + } + private createHeaders(): Record { const headers: Record = { 'Content-Type': 'application/json', diff --git a/packages/sdk/src/client/websocket-client.ts b/packages/sdk/src/client/websocket-client.ts index e9d6a193..86945a25 100644 --- a/packages/sdk/src/client/websocket-client.ts +++ b/packages/sdk/src/client/websocket-client.ts @@ -24,7 +24,13 @@ import { type EventPredicate, } from '../events/index.js'; import { SnowLumaApiClient, toJsonObject } from './api-client.js'; -import { raceWithAbort, throwIfAborted } from '../internal/abort.js'; +import { createRequestAbortState, raceWithAbort, throwIfAborted } from '../internal/abort.js'; +import { + isStreamTerminal, + parseStreamResponse, + StreamQueue, + streamLimits, +} from '../internal/streaming.js'; import { appendAccessToken, assertApiResponse, @@ -56,6 +62,9 @@ import type { RequestOptions, SnowLumaAction, SnowLumaEvent, + SnowLumaStreamAction, + StreamApiResponse, + StreamRequestOptions, WsRole, } from '../types/index.js'; import type { @@ -81,7 +90,8 @@ interface SocketConnection { } interface PendingRequest { - resolve(response: ApiResponse): void; + stream?: boolean; + resolve(response: ApiResponse, bytes: number): void; reject(error: Error): void; } @@ -434,21 +444,141 @@ export class SnowLumaWebSocketClient extends SnowLumaApiClient { }); } + override async *requestStream( + action: TAction, + params: ActionParams, + options?: StreamRequestOptions, + ): AsyncGenerator>, void, void> { + const queue = new StreamQueue>>( + streamLimits(options), + ); + const timeoutMs = options?.timeoutMs ?? this.requestTimeoutMs; + const abort = createRequestAbortState(timeoutMs, options?.signal); + const echo = options?.echo ?? createEcho(); + const key = echoKey(echo); + + const cleanup = (): void => { + // A completed stream can still be drained while a new request uses + // this echo. Its iterator cleanup must not remove the new request. + if (this.pending.get(key) === pending) { + this.pending.delete(key); + } + + abort.signal.removeEventListener('abort', onAbort); + abort.cleanup(); + }; + + const pending: PendingRequest = { + stream: true, + resolve: (response, bytes) => { + try { + queue.push(response as StreamApiResponse>, bytes); + + if (isStreamTerminal(action, response)) { + queue.finish(); + cleanup(); + } + } catch (error) { + pending.reject( + error instanceof Error ? error : new SnowLumaConnectionError(String(error)), + ); + } + }, + reject: (error) => { + queue.fail(error); + cleanup(); + }, + }; + + const onAbort = (): void => { + pending.reject( + abort.isTimedOut() + ? new SnowLumaTimeoutError( + `SnowLuma stream timed out after ${timeoutMs}ms`, + timeoutMs, + ) + : new SnowLumaAbortError('SnowLuma stream aborted', { + cause: options?.signal?.reason, + }), + ); + }; + + try { + throwIfAborted(abort.signal); + await raceWithAbort(this.connect(), abort.signal); + throwIfAborted(abort.signal); + const socket = this.connection?.socket; + + if (!socket || socket.readyState !== WEBSOCKET_OPEN) { + throw new SnowLumaConnectionError('SnowLuma WebSocket is not open'); + } + + if (this.pending.has(key)) { + throw new SnowLumaConnectionError( + 'A WebSocket request with this echo is already pending', + ); + } + + this.pending.set(key, pending); + abort.signal.addEventListener('abort', onAbort, { once: true }); + + try { + socket.send(JSON.stringify({ action, params, echo })); + } catch (cause) { + throw new SnowLumaConnectionError('Failed to send SnowLuma WebSocket stream', { + cause, + }); + } + + while (true) { + throwIfAborted(options?.signal); + const next = await queue.next(); + + if (next.done) { + return; + } + + yield next.value; + } + } catch (error) { + if (abort.isTimedOut() && !(error instanceof SnowLumaTimeoutError)) { + throw new SnowLumaTimeoutError( + `SnowLuma stream timed out after ${timeoutMs}ms`, + timeoutMs, + { cause: error }, + ); + } + + throw error; + } finally { + cleanup(); + } + } + private handleMessage(args: unknown[]): void { const raw = unwrapMessagePayload(args[0]); this.emitter.emit('raw', raw); let parsed: unknown; + let text: string; try { - parsed = JSON.parse(rawToString(raw)); + text = rawToString(raw); + parsed = JSON.parse(text); } catch (error) { - this.emitter.emit( - 'error', - new SnowLumaParseError('SnowLuma WebSocket returned non-JSON message', { - cause: error, - }), - ); + const failure = new SnowLumaParseError('SnowLuma WebSocket returned non-JSON message', { + cause: error, + }); + + // Without a readable echo, a lost stream frame cannot be attributed + // safely. Fail active streams instead of reporting partial success. + for (const pending of this.pending.values()) { + if (pending.stream) { + pending.reject(failure); + } + } + + this.emitter.emit('error', failure); return; } @@ -460,7 +590,9 @@ export class SnowLumaWebSocketClient extends SnowLumaApiClient { let response: ApiResponse; try { - response = assertApiResponse(parsed); + response = pending.stream + ? parseStreamResponse(parsed) + : assertApiResponse(parsed); } catch (error) { pending.reject( new SnowLumaParseError('SnowLuma WebSocket response is invalid', { @@ -470,7 +602,10 @@ export class SnowLumaWebSocketClient extends SnowLumaApiClient { return; } - pending.resolve(response); + pending.resolve( + response, + pending.stream ? new TextEncoder().encode(text).byteLength : 0, + ); this.emitter.emit('response', response); return; } diff --git a/packages/sdk/src/internal/streaming.ts b/packages/sdk/src/internal/streaming.ts new file mode 100644 index 00000000..177aa5d5 --- /dev/null +++ b/packages/sdk/src/internal/streaming.ts @@ -0,0 +1,260 @@ +import { SnowLumaConnectionError, SnowLumaParseError } from '../errors.js'; +import type { + ApiResponse, + JsonValue, + SnowLumaStreamAction, + StreamApiResponse, + StreamRequestOptions, +} from '../types/index.js'; +import { raceWithAbort, throwIfAborted } from './abort.js'; +import { assertApiResponse, isJsonObject } from './json.js'; + +export function streamLimits(options?: StreamRequestOptions) { + const maxBufferBytes = options?.maxBufferBytes ?? 32 * 1024 * 1024; + const maxBufferedFrames = options?.maxBufferedFrames ?? 256; + + for (const [name, value] of Object.entries({ maxBufferBytes, maxBufferedFrames })) { + if (!Number.isSafeInteger(value) || value < 1) { + throw new RangeError(`${name} must be a positive safe integer`); + } + } + + return { maxBufferBytes, maxBufferedFrames }; +} + +export function parseStreamResponse(value: unknown): StreamApiResponse { + if (!isJsonObject(value) || !Number.isSafeInteger(value.retcode) || !('data' in value)) { + throw new SnowLumaParseError('Invalid SnowLuma stream response envelope'); + } + + let response: ApiResponse; + + try { + response = assertApiResponse(value); + } catch (cause) { + throw new SnowLumaParseError('Invalid SnowLuma stream response envelope', { cause }); + } + + if (response.status === 'ok' && response.retcode === 0) { + if ( + value.stream !== 'stream-action' || + !isJsonObject(response.data) || + (response.data.type !== 'stream' && response.data.type !== 'response') + ) { + throw new SnowLumaParseError('Invalid SnowLuma stream frame or missing stream marker'); + } + } + + return { + ...response, + stream: value.stream === 'stream-action' ? value.stream : undefined, + message: typeof value.message === 'string' ? value.message : undefined, + }; +} + +export function isStreamTerminal(action: SnowLumaStreamAction, response: ApiResponse): boolean { + // Each upload request has exactly one ACK, even though its data.type is + // "stream". Only downloads keep the same request open for multiple frames. + return ( + action === 'upload_file_stream' || + response.status !== 'ok' || + response.retcode !== 0 || + (isJsonObject(response.data) && response.data.type === 'response') + ); +} + +async function cancelReader( + reader: ReadableStreamDefaultReader, + failure?: { error: unknown }, +): Promise { + try { + await reader.cancel(); + } catch (error) { + // Retain the original failure if the source also rejects cancellation. + if (failure) { + throw new AggregateError([failure.error, error], 'Stream cleanup failed'); + } + + throw error; + } +} + +/** Cancel on early exit and always release the Web Stream's reader lock. */ +export async function* readBytes( + body: ReadableStream, + signal?: AbortSignal, +): AsyncGenerator { + const reader = body.getReader(); + let ended = false; + let failure: { error: unknown } | undefined; + + try { + while (true) { + throwIfAborted(signal); + const next = await raceWithAbort(reader.read(), signal); + + if (next.done) { + ended = true; + return; + } + + yield next.value; + } + } catch (error) { + failure = { error }; + throw error; + } finally { + try { + if (!ended) { + await cancelReader(reader, failure); + } + } finally { + reader.releaseLock(); + } + } +} + +/** HTTP frames are JSON objects separated by CRLF CRLF, not SSE or NDJSON. */ +export async function* readHttpFrames( + body: ReadableStream, + maxBytes: number, + signal: AbortSignal, +): AsyncGenerator> { + const decoder = new TextDecoder('utf-8', { fatal: true }); + const encoder = new TextEncoder(); + let buffer = ''; + let bufferedBytes = 0; + let searchFrom = 0; + + const parse = (text: string) => { + try { + return parseStreamResponse(JSON.parse(text)); + } catch (cause) { + if (cause instanceof SnowLumaParseError) { + throw cause; + } + + throw new SnowLumaParseError('Invalid JSON in SnowLuma HTTP stream', { cause }); + } + }; + + const decode = (bytes?: Uint8Array) => { + try { + return decoder.decode(bytes, { stream: bytes !== undefined }); + } catch (cause) { + throw new SnowLumaParseError('Invalid UTF-8 in SnowLuma HTTP stream', { cause }); + } + }; + + for await (const bytes of readBytes(body, signal)) { + const text = decode(bytes); + buffer += text; + bufferedBytes += encoder.encode(text).byteLength; + let end: number; + + while ((end = buffer.indexOf('\r\n\r\n', searchFrom)) !== -1) { + const frame = buffer.slice(0, end); + const size = encoder.encode(frame).byteLength; + + if (size > maxBytes) { + throw new SnowLumaParseError('SnowLuma HTTP stream frame exceeds maxBufferBytes'); + } + + buffer = buffer.slice(end + 4); + bufferedBytes -= size + 4; + searchFrom = 0; + + if (frame.trim()) { + yield parse(frame); + } + } + + if (bufferedBytes > maxBytes + 3) { + throw new SnowLumaParseError('SnowLuma HTTP stream frame exceeds maxBufferBytes'); + } + + searchFrom = Math.max(0, buffer.length - 3); + } + + buffer += decode(); + + if (encoder.encode(buffer).byteLength > maxBytes) { + throw new SnowLumaParseError('SnowLuma HTTP stream frame exceeds maxBufferBytes'); + } + + // Ordinary preflight errors are single JSON responses without a delimiter. + if (buffer.trim()) { + yield parse(buffer); + } +} + +/** WebSocket has no per-request backpressure; bound its queue and fail on overflow. */ +export class StreamQueue { + private readonly items: { value: T; size: number }[] = []; + private bytes = 0; + private ended = false; + private error?: Error; + private waiting?: { + resolve: (result: IteratorResult) => void; + reject: (error: Error) => void; + }; + + constructor(private readonly limits: ReturnType) {} + + push(value: T, size: number): void { + if (this.ended) { + return; + } + + if ( + this.bytes + size > this.limits.maxBufferBytes || + this.items.length >= this.limits.maxBufferedFrames + ) { + throw new SnowLumaConnectionError('SnowLuma WebSocket stream buffer limit exceeded'); + } + + if (this.waiting) { + this.waiting.resolve({ value, done: false }); + this.waiting = undefined; + } else { + this.items.push({ value, size }); + this.bytes += size; + } + } + + finish(): void { + this.ended = true; + this.waiting?.resolve({ value: undefined, done: true }); + this.waiting = undefined; + } + + fail(error: Error): void { + this.error = error; + this.items.length = 0; + this.bytes = 0; + this.waiting?.reject(error); + this.waiting = undefined; + this.ended = true; + } + + next(): Promise> { + if (this.error) { + return Promise.reject(this.error); + } + + const item = this.items.shift(); + + if (item) { + this.bytes -= item.size; + return Promise.resolve({ value: item.value, done: false }); + } + + if (this.ended) { + return Promise.resolve({ value: undefined, done: true }); + } + + return new Promise((resolve, reject) => { + this.waiting = { resolve, reject }; + }); + } +} diff --git a/packages/sdk/src/types/index.ts b/packages/sdk/src/types/index.ts index 71c4107d..0a4796b0 100644 --- a/packages/sdk/src/types/index.ts +++ b/packages/sdk/src/types/index.ts @@ -4,3 +4,4 @@ export * from './results.js'; export * from './params.js'; export * from './actions.js'; export * from './events.js'; +export * from './streaming.js'; diff --git a/packages/sdk/src/types/streaming.ts b/packages/sdk/src/types/streaming.ts new file mode 100644 index 00000000..43b57859 --- /dev/null +++ b/packages/sdk/src/types/streaming.ts @@ -0,0 +1,53 @@ +import type { ApiResponse, RequestOptions } from './json.js'; +import type { DownloadFileStreamInfo } from './results.js'; + +export type DownloadStreamAction = + 'download_file_stream' | 'download_file_image_stream' | 'download_file_record_stream'; + +export type SnowLumaStreamAction = + DownloadStreamAction | 'upload_file_stream' | 'test_download_stream'; + +/** Authentication/validation failures can arrive without the stream marker. */ +export interface StreamApiResponse extends ApiResponse { + stream?: 'stream-action'; + message?: string; +} + +export interface StreamRequestOptions extends RequestOptions { + /** Maximum HTTP frame or queued WebSocket payload bytes. Default: 32 MiB. */ + maxBufferBytes?: number; + /** Maximum queued WebSocket frames. Default: 256. */ + maxBufferedFrames?: number; +} + +export type UploadFileSource = + Uint8Array | Blob | ReadableStream | AsyncIterable; + +export interface UploadProgress { + streamId: string; + uploadedBytes: number; + fileSize: number; + uploadedChunks: number; + totalChunks: number; +} + +export interface UploadFileStreamOptions extends RequestOptions { + /** Required for streams/async iterables; inferred for Uint8Array and Blob. */ + fileSize?: number; + /** Decoded bytes per chunk, 1–1048576. Default: 64 KiB. */ + chunkSize?: number; + /** Must identify a fresh upload. Generated automatically when omitted. */ + streamId?: string; + filename?: string; + expectedSha256?: string; + /** Server-side retention in milliseconds; zero disables expiry. */ + fileRetention?: number; + /** Called after each acknowledged chunk; awaited before reading more input. */ + onProgress?: (progress: UploadProgress) => void | Promise; +} + +export interface DownloadFileStreamOptions extends StreamRequestOptions { + action?: DownloadStreamAction; + /** Called once before yielding file bytes. */ + onInfo?: (info: DownloadFileStreamInfo) => void | Promise; +} diff --git a/packages/sdk/tests/file-streams.test.ts b/packages/sdk/tests/file-streams.test.ts new file mode 100644 index 00000000..bf61b9a2 --- /dev/null +++ b/packages/sdk/tests/file-streams.test.ts @@ -0,0 +1,288 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + SnowLumaAbortError, + SnowLumaApiError, + SnowLumaHttpClient, + SnowLumaParseError, +} from '../src/index.js'; +import type { JsonObject } from '../src/index.js'; + +const wrap = (data: unknown) => ({ status: 'ok', retcode: 0, stream: 'stream-action', data }); +const wire = (data: unknown) => JSON.stringify(wrap(data)) + '\r\n\r\n'; +const info = { + type: 'stream', + data_type: 'file_info', + file_name: 'test', + file_size: 2, + chunk_size: 2, +}; +const chunk = { + type: 'stream', + data_type: 'file_chunk', + index: 0, + data: 'YWI=', + size: 2, + base64_size: 4, + progress: 100, +}; +const complete = { + type: 'response', + data_type: 'file_complete', + file_name: 'test', + total_chunks: 1, + total_bytes: 2, + message: 'done', +}; + +async function download(frames: unknown[], onInfo?: (value: unknown) => void) { + const client = new SnowLumaHttpClient({ + fetch: async () => new Response(frames.map(wire).join('')), + }); + const stream = client.downloadFileStream({ file: 'test' }, { onInfo }); + const chunks: Uint8Array[] = []; + + while (true) { + const result = await stream.next(); + + if (result.done) { + return { chunks, complete: result.value }; + } + + chunks.push(result.value); + } +} + +describe('file download helper', () => { + it('yields bytes, exposes metadata and returns verified completion', async () => { + const onInfo = vi.fn(); + expect(await download([info, chunk, complete], onInfo)).toEqual({ + chunks: [new Uint8Array([97, 98])], + complete, + }); + expect(onInfo).toHaveBeenCalledExactlyOnceWith(info); + }); + + it('accepts an unknown initial size and verifies the final totals', async () => { + expect( + (await download([{ ...info, file_size: 0 }, chunk, complete])).complete.total_bytes, + ).toBe(2); + }); + + it('supports empty files', async () => { + expect( + ( + await download([ + { ...info, file_size: 0 }, + { ...complete, total_chunks: 0, total_bytes: 0 }, + ]) + ).chunks, + ).toEqual([]); + }); + + it.each([ + ['missing info', [chunk, complete]], + ['duplicate info', [info, info, chunk, complete]], + ['missing chunk', [info, complete]], + ['out-of-order index', [info, { ...chunk, index: 1 }, complete]], + ['invalid base64', [info, { ...chunk, data: '????' }, complete]], + ['non-canonical base64', [info, { ...chunk, data: 'YWL=' }, complete]], + ['wrong chunk size', [info, { ...chunk, size: 3 }, complete]], + ['wrong base64 size', [info, { ...chunk, base64_size: 5 }, complete]], + ['wrong final size', [info, chunk, { ...complete, total_bytes: 3 }]], + ['wrong final count', [info, chunk, { ...complete, total_chunks: 2 }]], + ['wrong info size', [{ ...info, file_size: 3 }, chunk, complete]], + ['wrong final name', [info, chunk, { ...complete, file_name: 'other' }]], + ['unexpected frame', [info, { type: 'stream', data_type: 'other' }, complete]], + ])('rejects %s', async (_name, frames) => { + await expect(download(frames as unknown[])).rejects.toBeInstanceOf(SnowLumaParseError); + }); +}); + +function uploader(transform?: (data: JsonObject, params: JsonObject) => JsonObject) { + const calls: JsonObject[] = []; + const fetch = vi.fn(async (_input: string | URL | Request, init?: RequestInit) => { + const { params } = JSON.parse(String(init?.body)) as { params: JsonObject }; + calls.push(params); + const first = calls[0]; + const result: JsonObject = params.is_complete + ? { + type: 'response', + status: 'file_complete', + stream_id: first.stream_id, + received_chunks: first.total_chunks, + total_chunks: first.total_chunks, + file_size: first.file_size, + file_path: '/remote/file', + file_name: 'test', + sha256: 'a'.repeat(64), + } + : { + type: 'stream', + status: 'chunk_received', + stream_id: params.stream_id, + received_chunks: Number(params.chunk_index) + 1, + total_chunks: params.total_chunks, + }; + return new Response(wire(transform ? transform(result, params) : result)); + }); + return { client: new SnowLumaHttpClient({ fetch }), calls, fetch }; +} + +describe('file upload helper', () => { + it('cancels a stalled readable source before any request is sent', async () => { + const { client, fetch } = uploader(); + const cancel = vi.fn(); + const source = new ReadableStream({ cancel }); + const controller = new AbortController(); + const upload = client.uploadFileStream(source, { fileSize: 1, signal: controller.signal }); + await vi.waitFor(() => expect(source.locked).toBe(true)); + controller.abort(); + await expect(upload).rejects.toBeInstanceOf(SnowLumaAbortError); + expect(cancel).toHaveBeenCalledOnce(); + expect(source.locked).toBe(false); + expect(fetch).not.toHaveBeenCalled(); + }); + + it('does not send later chunks or finalize after cancellation from a progress callback', async () => { + const { client, calls } = uploader(); + const controller = new AbortController(); + await expect( + client.uploadFileStream(new Uint8Array([1, 2]), { + chunkSize: 1, + signal: controller.signal, + onProgress: () => controller.abort(), + }), + ).rejects.toBeInstanceOf(SnowLumaAbortError); + expect(calls).toHaveLength(1); + }); + + it('re-chunks async input, awaits progress and separately finalizes after all ACKs', async () => { + const { client, calls } = uploader(); + const progress = vi.fn(); + let readSecond = false; + const source = (async function* () { + yield new Uint8Array([1, 2, 3, 4]); + expect(progress).toHaveBeenCalledTimes(2); + readSecond = true; + yield new Uint8Array([5]); + })(); + const result = await client.uploadFileStream(source, { + fileSize: 5, + chunkSize: 2, + filename: 'input', + streamId: 'test', + onProgress: progress, + }); + expect(readSecond).toBe(true); + expect(calls.map((params) => params.chunk_data)).toEqual([ + 'AQI=', + 'AwQ=', + 'BQ==', + undefined, + ]); + expect(calls[0]).toMatchObject({ total_chunks: 3, file_size: 5, filename: 'input' }); + expect(calls[3]).toEqual({ stream_id: 'test', is_complete: true }); + expect(progress).toHaveBeenLastCalledWith({ + streamId: 'test', + uploadedChunks: 3, + totalChunks: 3, + uploadedBytes: 5, + fileSize: 5, + }); + expect(result.file_path).toBe('/remote/file'); + }); + + it.each([new Uint8Array(), new Blob([]), new Blob(['ab'])])( + 'infers size for byte and Blob sources: %s', + async (source) => { + const { client, calls } = uploader(); + const result = await client.uploadFileStream(source); + expect(result.file_size).toBe(source instanceof Blob ? source.size : source.length); + expect(calls).toHaveLength(2); + expect(calls[0].total_chunks).toBe(1); + }, + ); + + it.each([0, 4])( + 'does not finalize a streaming source with mismatched size %i', + async (fileSize) => { + const { client, calls } = uploader(); + let closed = false; + const source = (async function* () { + try { + yield new Uint8Array([1, 2]); + } finally { + closed = true; + } + })(); + await expect( + client.uploadFileStream(source, { fileSize, chunkSize: 1 }), + ).rejects.toThrow('fileSize'); + expect(closed).toBe(true); + expect(calls.some((call) => call.is_complete)).toBe(false); + }, + ); + + it('requires a declared size and validates options before making requests', async () => { + const { client, fetch } = uploader(); + const source = (async function* () { + yield new Uint8Array([1]); + })(); + await expect(client.uploadFileStream(source)).rejects.toThrow('fileSize'); + await expect(client.uploadFileStream(new Uint8Array([1]), { fileSize: 2 })).rejects.toThrow( + 'fileSize', + ); + await expect( + client.uploadFileStream(new Uint8Array([1]), { chunkSize: 0 }), + ).rejects.toThrow('chunkSize'); + await expect( + client.uploadFileStream(new Uint8Array([1]), { streamId: '../bad' }), + ).rejects.toThrow('streamId'); + expect(fetch).not.toHaveBeenCalled(); + }); + + it('rejects incorrect ACKs and cancels a readable source on failure', async () => { + const { client, calls } = uploader((data) => ({ ...data, received_chunks: 99 })); + const cancel = vi.fn(); + const source = new ReadableStream({ + start(c) { + c.enqueue(new Uint8Array([1, 2])); + }, + cancel, + }); + await expect(client.uploadFileStream(source, { fileSize: 2 })).rejects.toBeInstanceOf( + SnowLumaParseError, + ); + expect(calls).toHaveLength(1); + expect(cancel).toHaveBeenCalledOnce(); + expect(source.locked).toBe(false); + }); + + it('stops after a failed chunk and preserves the server error', async () => { + const fetch = vi.fn( + async () => + new Response( + JSON.stringify({ + status: 'failed', + retcode: 1500, + data: { type: 'error' }, + wording: 'disk full', + }), + ), + ); + const client = new SnowLumaHttpClient({ fetch }); + await expect( + client.uploadFileStream(new Uint8Array(10), { chunkSize: 2 }), + ).rejects.toBeInstanceOf(SnowLumaApiError); + expect(fetch).toHaveBeenCalledOnce(); + }); + + it('rejects inconsistent final metadata', async () => { + const { client } = uploader((data, params) => + params.is_complete ? { ...data, file_size: 99 } : data, + ); + await expect(client.uploadFileStream(new Uint8Array([1]))).rejects.toThrow( + 'completion metadata', + ); + }); +}); diff --git a/packages/sdk/tests/fixtures/action-types.mts b/packages/sdk/tests/fixtures/action-types.mts index f7cd9190..89d4ea07 100644 --- a/packages/sdk/tests/fixtures/action-types.mts +++ b/packages/sdk/tests/fixtures/action-types.mts @@ -1,12 +1,15 @@ import type { ActionParams, ActionResult, + DownloadFileStreamComplete, JsonObject, JsonValue, SnowLumaActionMap, SnowLumaHttpClient, SnowLumaKnownAction, SnowLumaWebSocketClient, + StreamApiResponse, + UploadFileStreamComplete, } from '@snowluma/sdk'; import { actionParams, actionResults, type CoveredAction } from './action-contracts.js'; @@ -40,6 +43,29 @@ export type QuickAlias = AssertTrue< declare const http: SnowLumaHttpClient; declare const ws: SnowLumaWebSocketClient; +export const uploadedFile: Promise = http.uploadFileStream( + new Blob(['file']), +); +export const download: AsyncGenerator = + ws.downloadFileStream({ file_id: 'media' }, { action: 'download_file_image_stream' }); +export const rawFrames: AsyncIterableIterator< + StreamApiResponse> +> = http.requestStream('test_download_stream', {}); + +for await (const item of ws.stream('download_file_stream', { file: '/remote/file' })) { + if (item.type === 'stream' && item.data_type === 'file_chunk') { + const encoded: string = item.data; + void encoded; + } +} + +// @ts-expect-error single-response actions use request()/call() +http.stream('get_status', {}); +// @ts-expect-error uploads accept bytes rather than text chunks +http.uploadFileStream('text'); +// @ts-expect-error downloads require a file or file ID +ws.downloadFileStream({}); + const feeds = await http.raw('get_qzone_feeds', { page_num: '2', count: 10 }); const more: boolean = feeds.has_more; const author: number = feeds.feeds[0].uin; diff --git a/packages/sdk/tests/http-streaming.test.ts b/packages/sdk/tests/http-streaming.test.ts new file mode 100644 index 00000000..9c86df7a --- /dev/null +++ b/packages/sdk/tests/http-streaming.test.ts @@ -0,0 +1,222 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + SnowLumaAbortError, + SnowLumaApiError, + SnowLumaAuthError, + SnowLumaHttpClient, + SnowLumaParseError, + SnowLumaTimeoutError, +} from '../src/index.js'; + +const data = { type: 'stream', data_type: 'data_chunk', data: '中文🙂' }; +const complete = { type: 'response', data_type: 'data_complete', data: 'done' }; +const frame = (payload: unknown) => ({ + status: 'ok', + retcode: 0, + stream: 'stream-action', + data: payload, + echo: 'stream', +}); +const wire = (payload: unknown) => JSON.stringify(payload) + '\r\n\r\n'; + +async function collect(stream: AsyncIterable): Promise { + const results: T[] = []; + + for await (const result of stream) { + results.push(result); + } + + return results; +} + +function clientFor(body: BodyInit | null) { + return new SnowLumaHttpClient({ fetch: async () => new Response(body) }); +} + +afterEach(() => vi.useRealTimers()); + +describe('HTTP streaming', () => { + it('does not send an already-aborted request', async () => { + const fetch = vi.fn(); + const client = new SnowLumaHttpClient({ fetch }); + await expect( + client.stream('test_download_stream', {}, { signal: AbortSignal.abort() }).next(), + ).rejects.toBeInstanceOf(SnowLumaAbortError); + expect(fetch).not.toHaveBeenCalled(); + }); + + it('honors cancellation between frames buffered in the same HTTP read', async () => { + const controller = new AbortController(); + const client = clientFor(wire(frame(data)) + wire(frame(complete))); + const stream = client.stream('test_download_stream', {}, { signal: controller.signal }); + await stream.next(); + controller.abort(); + await expect(stream.next()).rejects.toBeInstanceOf(SnowLumaAbortError); + }); + + it.each(['one byte', 'one read'] as const)( + 'decodes split UTF-8 and delimiters: %s', + async (mode) => { + const bytes = new TextEncoder().encode( + wire(frame(data)) + wire(frame(data)) + wire(frame(complete)), + ); + let offset = 0; + const body = new ReadableStream({ + pull(controller) { + if (offset === bytes.length) { + controller.close(); + return; + } + + const end = mode === 'one byte' ? offset + 1 : bytes.length; + controller.enqueue(bytes.subarray(offset, end)); + offset = end; + }, + }); + const fetch = vi.fn(async () => new Response(body)); + const client = new SnowLumaHttpClient({ fetch, accessToken: 'test-token' }); + const results = await collect( + client.requestStream('test_download_stream', {}, { echo: 'stream' }), + ); + expect(results.map((result) => result.data)).toEqual([data, data, complete]); + expect( + results.every( + (result) => result.stream === 'stream-action' && result.echo === 'stream', + ), + ).toBe(true); + expect(fetch).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ + headers: expect.objectContaining({ Authorization: 'Bearer test-token' }), + body: JSON.stringify({ + action: 'test_download_stream', + params: {}, + echo: 'stream', + }), + }), + ); + expect(body.locked).toBe(false); + }, + ); + + it('delivers a frame before EOF and cancels the reader when iteration stops', async () => { + const cancel = vi.fn(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(wire(frame(data)))); + }, + cancel, + }); + const stream = clientFor(body).stream('test_download_stream', {}); + expect(await stream.next()).toMatchObject({ value: data, done: false }); + await stream.return(); + expect(cancel).toHaveBeenCalledOnce(); + expect(body.locked).toBe(false); + }); + + it.each([ + ['', 'terminal'], + [wire(frame(data)), 'terminal'], + [wire({ status: 'ok', retcode: 0, data }), 'marker'], + [wire({ status: 'ok', data }), 'envelope'], + [wire(frame({ type: 'error' })), 'frame'], + ['{bad json}\r\n\r\n', 'JSON'], + ])('rejects incomplete or invalid responses: %s', async (body, message) => { + await expect(collect(clientFor(body).stream('test_download_stream', {}))).rejects.toThrow( + message, + ); + }); + + it('rejects malformed UTF-8', async () => { + await expect( + collect(clientFor(new Uint8Array([0xff])).stream('test_download_stream', {})), + ).rejects.toBeInstanceOf(SnowLumaParseError); + }); + + it('bounds individual frames while allowing multiple frames in one large read', async () => { + const text = wire(frame(data)).repeat(10) + wire(frame(complete)); + await expect( + collect(clientFor(text).stream('test_download_stream', {}, { maxBufferBytes: 200 })), + ).resolves.toHaveLength(11); + await expect( + collect(clientFor(text).stream('test_download_stream', {}, { maxBufferBytes: 10 })), + ).rejects.toThrow('maxBufferBytes'); + await expect( + collect( + clientFor('x'.repeat(500)).stream( + 'test_download_stream', + {}, + { maxBufferBytes: 10 }, + ), + ), + ).rejects.toThrow('maxBufferBytes'); + }); + + it('returns raw failures and throws typed API errors from stream()', async () => { + const failure = { status: 'failed', retcode: 1401, data: null, wording: 'denied' }; + expect( + await collect( + clientFor(JSON.stringify(failure)).requestStream('test_download_stream', {}), + ), + ).toMatchObject([failure]); + await expect( + collect(clientFor(JSON.stringify(failure)).stream('test_download_stream', {})), + ).rejects.toBeInstanceOf(SnowLumaAuthError); + const stream = clientFor(wire(frame(data)) + wire({ ...failure, retcode: 1500 })).stream( + 'test_download_stream', + {}, + ); + expect((await stream.next()).value).toEqual(data); + await expect(stream.next()).rejects.toBeInstanceOf(SnowLumaApiError); + }); + + it.each(['abort', 'timeout'] as const)('releases a pending body read on %s', async (reason) => { + vi.useFakeTimers(); + const cancel = vi.fn(); + const body = new ReadableStream({ cancel }); + const controller = new AbortController(); + const remove = vi.spyOn(controller.signal, 'removeEventListener'); + const stream = clientFor(body).stream( + 'test_download_stream', + {}, + { signal: controller.signal, timeoutMs: 100 }, + ); + const next = expect(stream.next()).rejects.toBeInstanceOf( + reason === 'abort' ? SnowLumaAbortError : SnowLumaTimeoutError, + ); + await vi.advanceTimersByTimeAsync(0); + + if (reason === 'abort') { + controller.abort(); + } else { + await vi.advanceTimersByTimeAsync(100); + } + + await next; + expect(cancel).toHaveBeenCalledOnce(); + expect(remove).toHaveBeenCalled(); + expect(body.locked).toBe(false); + expect(vi.getTimerCount()).toBe(0); + }); + + it('ends an upload ACK without waiting for a response-type frame or HTTP EOF', async () => { + const ack = { + type: 'stream', + status: 'chunk_received', + stream_id: 'id', + received_chunks: 1, + total_chunks: 2, + }; + const cancel = vi.fn(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(wire(frame(ack)))); + }, + cancel, + }); + expect( + await collect(clientFor(body).stream('upload_file_stream', { stream_id: 'id' })), + ).toEqual([ack]); + expect(cancel).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/sdk/tests/websocket-client.test.ts b/packages/sdk/tests/websocket-client.test.ts index 2ac20a0c..2c3dcdeb 100644 --- a/packages/sdk/tests/websocket-client.test.ts +++ b/packages/sdk/tests/websocket-client.test.ts @@ -1,6 +1,10 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { SnowLumaAbortError, + SnowLumaApiError, + SnowLumaConnectionError, + SnowLumaParseError, + SnowLumaTimeoutError, SnowLumaWebSocketClient, type OneBotPrivateMessageEvent, } from '../src'; @@ -74,6 +78,225 @@ const PRIVATE_EVENT: OneBotPrivateMessageEvent = { sender: { user_id: 10001, nickname: 'tester' }, }; +describe('WebSocket streaming', () => { + afterEach(() => vi.useRealTimers()); + + async function connected() { + const client = new SnowLumaWebSocketClient({ + webSocket: MockWebSocket, + requestTimeoutMs: 1000, + }); + const opening = client.connect(); + const socket = MockWebSocket.instances.at(-1)!; + socket.open(); + await opening; + return { client, socket }; + } + + const data = { type: 'stream', data_type: 'data_chunk', data: 'part' }; + const complete = { type: 'response', data_type: 'data_complete', data: 'done' }; + const frame = (echo: string, payload: unknown) => + JSON.stringify({ + status: 'ok', + retcode: 0, + stream: 'stream-action', + echo, + data: payload, + }); + + it('routes interleaved streams, normal responses and events independently', async () => { + const { client, socket } = await connected(); + const event = vi.fn(); + client.onEvent(event); + const first = client.stream('test_download_stream', {}, { echo: 'a' }); + const second = client.stream('test_download_stream', {}, { echo: 'b' }); + const a = first.next(); + const b = second.next(); + const status = client.getStatus({ echo: 'ordinary' }); + await vi.waitFor(() => expect(socket.sent).toHaveLength(3)); + socket.message(frame('b', data)); + socket.message(frame('a', data)); + socket.message(JSON.stringify(PRIVATE_EVENT)); + socket.message( + JSON.stringify({ status: 'ok', retcode: 0, echo: 'ordinary', data: { online: true } }), + ); + await expect(status).resolves.toEqual({ online: true }); + expect((await a).value).toEqual(data); + expect((await b).value).toEqual(data); + socket.message(frame('a', complete)); + socket.message(frame('b', complete)); + expect((await first.next()).value).toEqual(complete); + expect((await second.next()).value).toEqual(complete); + expect((await first.next()).done).toBe(true); + expect((await second.next()).done).toBe(true); + await vi.waitFor(() => expect(event).toHaveBeenCalledOnce()); + client.close(); + }); + + it('shares echo collision checks with single-response calls and preserves later requests', async () => { + const { client, socket } = await connected(); + const stream = client.stream('test_download_stream', {}, { echo: 'same' }); + const first = stream.next(); + await vi.waitFor(() => expect(socket.sent).toHaveLength(1)); + await expect(client.getStatus({ echo: 'same' })).rejects.toThrow('already pending'); + await expect( + client.stream('test_download_stream', {}, { echo: 'same' }).next(), + ).rejects.toThrow('already pending'); + socket.message(frame('same', data)); + await first; + socket.message(frame('same', complete)); + const ordinary = client.getStatus({ echo: 'same' }); + await vi.waitFor(() => expect(socket.sent).toHaveLength(2)); + await stream.return(); + socket.message( + JSON.stringify({ status: 'ok', retcode: 0, data: { online: true }, echo: 'same' }), + ); + await expect(ordinary).resolves.toEqual({ online: true }); + client.close(); + }); + + it.each([ + 'abort', + 'timeout', + 'close', + 'invalid frame', + 'invalid JSON', + 'send failure', + 'return', + 'complete', + ] as const)('releases stream timers and listeners after %s', async (outcome) => { + vi.useFakeTimers(); + const { client, socket } = await connected(); + const controller = new AbortController(); + const remove = vi.spyOn(controller.signal, 'removeEventListener'); + + if (outcome === 'send failure') { + vi.spyOn(socket, 'send').mockImplementationOnce(() => { + throw new Error('send'); + }); + } + + const stream = client.stream( + 'test_download_stream', + {}, + { echo: 's', signal: controller.signal }, + ); + const next = stream.next(); + const settled = next.then( + () => undefined, + (error: unknown) => error, + ); + await vi.advanceTimersByTimeAsync(0); + + switch (outcome) { + case 'abort': + controller.abort(); + break; + case 'timeout': + await vi.advanceTimersByTimeAsync(1000); + break; + case 'close': + socket.close(); + break; + case 'invalid frame': + socket.message(JSON.stringify({ echo: 's' })); + break; + case 'invalid JSON': + socket.message('bad json'); + break; + case 'return': + socket.message(frame('s', data)); + await next; + await stream.return(); + break; + case 'complete': + socket.message(frame('s', complete)); + await next; + await stream.next(); + break; + } + + const error = await settled; + const expected = { + abort: SnowLumaAbortError, + timeout: SnowLumaTimeoutError, + close: SnowLumaConnectionError, + 'send failure': SnowLumaConnectionError, + 'invalid frame': SnowLumaParseError, + 'invalid JSON': SnowLumaParseError, + }; + + if (outcome in expected) { + expect(error).toBeInstanceOf(expected[outcome as keyof typeof expected]); + } else { + expect(error).toBeUndefined(); + } + + expect(vi.getTimerCount()).toBe(0); + expect(remove).toHaveBeenCalled(); + client.close(); + }); + + it.each(['frames', 'bytes'] as const)( + 'fails a slow consumer when queued %s exceed its limit', + async (limit) => { + const { client, socket } = await connected(); + const stream = client.stream( + 'test_download_stream', + {}, + { + echo: 's', + maxBufferedFrames: limit === 'frames' ? 1 : 256, + maxBufferBytes: limit === 'bytes' ? 180 : 10000, + }, + ); + const next = stream.next(); + await vi.waitFor(() => expect(socket.sent).toHaveLength(1)); + socket.message(frame('s', data)); + await next; + socket.message(frame('s', data)); + socket.message(frame('s', data)); + await expect(stream.next()).rejects.toThrow('buffer limit'); + client.close(); + }, + ); + + it('throws a terminal API error after previously delivered frames', async () => { + const { client, socket } = await connected(); + const stream = client.stream('test_download_stream', {}, { echo: 's' }); + const next = stream.next(); + await vi.waitFor(() => expect(socket.sent).toHaveLength(1)); + socket.message(frame('s', data)); + await next; + socket.message( + JSON.stringify({ + status: 'failed', + retcode: 1500, + echo: 's', + data: { type: 'error' }, + wording: 'failed', + }), + ); + await expect(stream.next()).rejects.toBeInstanceOf(SnowLumaApiError); + client.close(); + }); + + it('times out while opening without sending a request later', async () => { + vi.useFakeTimers(); + const client = new SnowLumaWebSocketClient({ webSocket: MockWebSocket }); + const result = expect( + client.stream('test_download_stream', {}, { timeoutMs: 20 }).next(), + ).rejects.toBeInstanceOf(SnowLumaTimeoutError); + const socket = MockWebSocket.instances.at(-1)!; + await vi.advanceTimersByTimeAsync(20); + await result; + socket.open(); + expect(socket.sent).toHaveLength(0); + expect(vi.getTimerCount()).toBe(0); + client.close(); + }); +}); + describe('SnowLumaWebSocketClient', () => { afterEach(() => { vi.useRealTimers();