344 lines
13 KiB
TypeScript
344 lines
13 KiB
TypeScript
import { Flags, MAX_FRAME_DATA, type ItemKind } from './constants.js';
|
|
import { concat, readU16, toBytes, tryDecodeUtf8, utf8, writeU16, toBase64Url, type Bytes } from './bytes.js';
|
|
import { IntegrityError, LimitError } from './errors.js';
|
|
|
|
// ───────────────────────── 条目元数据(protocol §7) ─────────────────────────
|
|
|
|
export interface ItemMeta {
|
|
kind: ItemKind;
|
|
name?: string;
|
|
mime?: string;
|
|
/** 内容总字节数,流式上传时可能未知。 */
|
|
size?: number;
|
|
/** 毫秒时间戳,仅供参考,排序以 seq 为准。 */
|
|
ts: number;
|
|
/** 设备名。 */
|
|
dev?: string;
|
|
/** WebDrop 频道中的 thread ID,加密元数据。 */
|
|
threadId?: string;
|
|
/** 文件所在的虚拟文件夹 ID,加密元数据。 */
|
|
folderId?: string;
|
|
}
|
|
|
|
const KINDS: readonly ItemKind[] = ['text', 'json', 'file', 'manifest', 'other'];
|
|
|
|
/** u16 meta_len ‖ meta_json。字段顺序固定,undefined 的字段省略,保证字节级确定。 */
|
|
export function encodeMeta(meta: ItemMeta): Bytes {
|
|
const o: Record<string, unknown> = { v: 1, kind: meta.kind };
|
|
if (meta.name !== undefined) o.name = meta.name;
|
|
if (meta.mime !== undefined) o.mime = meta.mime;
|
|
if (meta.size !== undefined) o.size = meta.size;
|
|
o.ts = meta.ts;
|
|
if (meta.dev !== undefined) o.dev = meta.dev;
|
|
if (meta.threadId !== undefined) o.thread_id = meta.threadId;
|
|
if (meta.folderId !== undefined) o.folder_id = meta.folderId;
|
|
|
|
const json = utf8(JSON.stringify(o));
|
|
if (json.length > 0xffff) throw new LimitError('条目元数据过大');
|
|
return concat(writeU16(json.length), json);
|
|
}
|
|
|
|
/** 解析第 0 帧的 data:u16 meta_len ‖ meta_json ‖ 内容。 */
|
|
export function decodeFirstFrame(data: Uint8Array): { meta: ItemMeta; content: Bytes } {
|
|
if (data.length < 2) throw new IntegrityError('bad_meta');
|
|
const metaLen = readU16(data, 0);
|
|
if (data.length < 2 + metaLen) throw new IntegrityError('bad_meta');
|
|
|
|
const text = tryDecodeUtf8(data.subarray(2, 2 + metaLen));
|
|
if (text === null) throw new IntegrityError('bad_meta');
|
|
|
|
let raw: unknown;
|
|
try {
|
|
raw = JSON.parse(text);
|
|
} catch {
|
|
throw new IntegrityError('bad_meta');
|
|
}
|
|
if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) throw new IntegrityError('bad_meta');
|
|
const o = raw as Record<string, unknown>;
|
|
|
|
if (o.v !== 1 || typeof o.kind !== 'string' || typeof o.ts !== 'number' || !Number.isFinite(o.ts)) {
|
|
throw new IntegrityError('bad_meta');
|
|
}
|
|
const meta: ItemMeta = {
|
|
// 未知的 kind 按 other 处理,向前兼容
|
|
kind: (KINDS as readonly string[]).includes(o.kind) ? (o.kind as ItemKind) : 'other',
|
|
ts: o.ts,
|
|
};
|
|
for (const key of ['name', 'mime', 'dev'] as const) {
|
|
const v = o[key];
|
|
if (v === undefined) continue;
|
|
if (typeof v !== 'string') throw new IntegrityError('bad_meta');
|
|
meta[key] = v;
|
|
}
|
|
for (const [wire, key] of [['thread_id', 'threadId'], ['folder_id', 'folderId']] as const) {
|
|
const v = o[wire];
|
|
if (v === undefined) continue;
|
|
if (typeof v !== 'string' || v.length > 80) throw new IntegrityError('bad_meta');
|
|
meta[key] = v;
|
|
}
|
|
if (o.size !== undefined) {
|
|
if (typeof o.size !== 'number' || !Number.isSafeInteger(o.size) || o.size < 0) throw new IntegrityError('bad_meta');
|
|
meta.size = o.size;
|
|
}
|
|
return { meta, content: data.slice(2 + metaLen) };
|
|
}
|
|
|
|
// ───────────────────────── 条目编码(加密) ─────────────────────────
|
|
|
|
export type ContentSource = Uint8Array | Blob | ReadableStream<Uint8Array> | AsyncIterable<Uint8Array>;
|
|
|
|
export interface PlanOptions {
|
|
control?: boolean;
|
|
/** 指定条目 id(重试、续传时必须复用同一个);默认随机生成。 */
|
|
itemId?: Uint8Array;
|
|
/** 作为频道的「最新快照」。快照必须是单帧,所以内容不能超过一帧。 */
|
|
snapshot?: boolean;
|
|
/**
|
|
* 是否 GATED(不实时内联推送,只能按条目拉取并计下载次数)。
|
|
* 'auto'(默认):内容放不进一帧,或设置了 maxDownloads 时为 GATED。
|
|
*/
|
|
gated?: boolean | 'auto';
|
|
/** 下载次数上限,0 为不限。设置后条目必须是 GATED;需要作为 X-WD-Max-Downloads 随第 0 帧发送。 */
|
|
maxDownloads?: number;
|
|
/** 单帧 data 的字节上限,默认 1,048,572。调小只用于测试和极弱网络。 */
|
|
maxFrameData?: number;
|
|
/** 从第几帧开始产出(断点续传)。只支持长度已知的来源(Uint8Array、Blob)。 */
|
|
fromFrame?: number;
|
|
}
|
|
|
|
export interface EncodedFrame {
|
|
index: number;
|
|
final: boolean;
|
|
/** 完整封装,可直接作为 POST /frames 的请求体。 */
|
|
bytes: Bytes;
|
|
}
|
|
|
|
export interface ItemPlan {
|
|
/** 条目 id 的 base64url 文本。 */
|
|
itemId: string;
|
|
idBytes: Bytes;
|
|
/**
|
|
* 是否 GATED。来源长度已知时在创建计划时就确定;来源是流(长度未知)时,
|
|
* 要等第 0 帧产出之后才确定,所以请在拿到第 0 帧之后再读取它和 frame0Headers。
|
|
*/
|
|
readonly gated: boolean;
|
|
/** 第 0 帧要带的附加请求头(X-WD-Max-Downloads)。 */
|
|
frame0Headers: Record<string, string>;
|
|
/** 按序产出加密后的帧。同一个计划只能迭代一次。 */
|
|
frames: AsyncGenerator<EncodedFrame, void, void>;
|
|
}
|
|
|
|
export async function collectFrames(plan: ItemPlan): Promise<EncodedFrame[]> {
|
|
const out: EncodedFrame[] = [];
|
|
for await (const f of plan.frames) out.push(f);
|
|
return out;
|
|
}
|
|
|
|
export function newItemId(): Bytes {
|
|
const id = new Uint8Array(16);
|
|
globalThis.crypto.getRandomValues(id);
|
|
return id;
|
|
}
|
|
|
|
/** 把各种来源统一成字节块的异步序列。 */
|
|
async function* chunksOf(source: ContentSource): AsyncGenerator<Uint8Array, void, void> {
|
|
if (source instanceof Uint8Array) {
|
|
yield source;
|
|
return;
|
|
}
|
|
if (typeof Blob !== 'undefined' && source instanceof Blob) {
|
|
if (typeof source.stream === 'function') {
|
|
yield* chunksOf(source.stream());
|
|
return;
|
|
}
|
|
|
|
// Older Safari releases don't expose Blob.stream(). Read bounded slices
|
|
// instead of materializing the complete file in memory.
|
|
const sliceSize = 256 * 1024;
|
|
for (let offset = 0; offset < source.size; offset += sliceSize) {
|
|
const part = source.slice(offset, Math.min(offset + sliceSize, source.size));
|
|
if (typeof part.arrayBuffer === 'function') {
|
|
yield new Uint8Array(await part.arrayBuffer());
|
|
} else {
|
|
yield await readBlobPart(part);
|
|
}
|
|
}
|
|
return;
|
|
}
|
|
if (typeof ReadableStream !== 'undefined' && source instanceof ReadableStream) {
|
|
// 不依赖 ReadableStream 的异步迭代器(旧版 Safari 没有)
|
|
const reader = source.getReader();
|
|
try {
|
|
for (;;) {
|
|
const { done, value } = await reader.read();
|
|
if (done) return;
|
|
yield value;
|
|
}
|
|
} finally {
|
|
reader.releaseLock();
|
|
}
|
|
}
|
|
yield* source as AsyncIterable<Uint8Array>;
|
|
}
|
|
|
|
function readBlobPart(blob: Blob): Promise<Uint8Array> {
|
|
return new Promise((resolve, reject) => {
|
|
const reader = new FileReader();
|
|
reader.onload = () => reader.result instanceof ArrayBuffer
|
|
? resolve(new Uint8Array(reader.result))
|
|
: reject(new TypeError('文件读取失败'));
|
|
reader.onerror = () => reject(reader.error ?? new TypeError('文件读取失败'));
|
|
reader.onabort = () => reject(new DOMException('文件读取已取消', 'AbortError'));
|
|
reader.readAsArrayBuffer(blob);
|
|
});
|
|
}
|
|
|
|
interface Chunk {
|
|
data: Bytes;
|
|
last: boolean;
|
|
}
|
|
|
|
/**
|
|
* 把任意切分的字节流重新切成 [firstCap, cap, cap, ...] 大小的块,并通过向前看一个块知道哪一块是最后一块。
|
|
* 内容恰好填满某一块时不会多出一个空块;内容为空时产出一个空的最后一块。
|
|
*/
|
|
async function* rechunk(source: ContentSource, firstCap: number, cap: number): AsyncGenerator<Chunk, void, void> {
|
|
const parts: Uint8Array[] = [];
|
|
let total = 0;
|
|
let target = firstCap;
|
|
let ended = false;
|
|
const iter = chunksOf(source)[Symbol.asyncIterator]();
|
|
|
|
const take = (n: number): Bytes => {
|
|
const out = new Uint8Array(n);
|
|
let filled = 0;
|
|
while (filled < n) {
|
|
const head = parts[0] as Uint8Array;
|
|
const need = n - filled;
|
|
if (head.length <= need) {
|
|
out.set(head, filled);
|
|
filled += head.length;
|
|
parts.shift();
|
|
} else {
|
|
out.set(head.subarray(0, need), filled);
|
|
parts[0] = head.subarray(need);
|
|
filled += need;
|
|
}
|
|
}
|
|
total -= n;
|
|
return out;
|
|
};
|
|
|
|
for (;;) {
|
|
// 需要缓冲到 target+1 字节,才能确定当前块不是最后一块
|
|
while (!ended && total <= target) {
|
|
const r = await iter.next();
|
|
if (r.done) ended = true;
|
|
else if (r.value.length > 0) {
|
|
parts.push(r.value);
|
|
total += r.value.length;
|
|
}
|
|
}
|
|
if (total > target) {
|
|
yield { data: take(target), last: false };
|
|
target = cap;
|
|
continue;
|
|
}
|
|
yield { data: take(total), last: true };
|
|
return;
|
|
}
|
|
}
|
|
|
|
function isSized(source: ContentSource): source is Uint8Array | Blob {
|
|
return source instanceof Uint8Array || (typeof Blob !== 'undefined' && source instanceof Blob);
|
|
}
|
|
|
|
/** 构造条目计划的公共部分:标志位、GATED 的取舍、校验。 */
|
|
function resolveGating(opts: PlanOptions, fitsInOneFrame: boolean | undefined): boolean {
|
|
const wantsLimit = opts.maxDownloads !== undefined;
|
|
if (opts.snapshot && (opts.gated === true || wantsLimit)) throw new RangeError('快照不能是 GATED,也不能设置下载上限');
|
|
if (opts.gated === true || wantsLimit) return true;
|
|
if (opts.gated === false) {
|
|
if (fitsInOneFrame === false) throw new LimitError('内容放不进单帧,不能关闭 GATED');
|
|
return false;
|
|
}
|
|
return fitsInOneFrame === false;
|
|
}
|
|
|
|
/**
|
|
* enc 模式:把内容加密成一个条目(protocol §7)。
|
|
* 小内容单帧、非 GATED;放不下就拆成多帧并标记 GATED。第 0 帧携带元数据。
|
|
*/
|
|
export function planItem(
|
|
seal: (p: { flags: number; itemId: Uint8Array; index: number; data: Uint8Array }) => Promise<Bytes>,
|
|
source: ContentSource,
|
|
meta: ItemMeta,
|
|
opts: PlanOptions = {},
|
|
encrypted = true,
|
|
): ItemPlan {
|
|
const idBytes: Bytes = opts.itemId ? toBytes(opts.itemId) : newItemId();
|
|
const maxData = opts.maxFrameData ?? MAX_FRAME_DATA;
|
|
if (maxData < 1 || maxData > MAX_FRAME_DATA) throw new RangeError('maxFrameData 超出范围');
|
|
|
|
const head = encodeMeta(meta);
|
|
const cap0 = maxData - head.length;
|
|
if (cap0 < 0) throw new LimitError('元数据超过单帧上限');
|
|
|
|
const size = source instanceof Uint8Array ? source.length : typeof Blob !== 'undefined' && source instanceof Blob ? source.size : undefined;
|
|
const fits = size === undefined ? undefined : size <= cap0;
|
|
let gated = resolveGating(opts, fits);
|
|
|
|
const fromFrame = opts.fromFrame ?? 0;
|
|
if (fromFrame > 0 && !isSized(source)) throw new RangeError('只有长度已知的来源才能从中间帧续传');
|
|
|
|
const frame0Headers: Record<string, string> = {};
|
|
if (opts.maxDownloads !== undefined) frame0Headers['X-WD-Max-Downloads'] = String(opts.maxDownloads);
|
|
|
|
const snapshot = opts.snapshot === true;
|
|
if (opts.control && (snapshot || gated)) throw new RangeError('控制条目必须是非快照的单帧');
|
|
|
|
async function* frames(): AsyncGenerator<EncodedFrame, void, void> {
|
|
const base = (gatedNow: boolean) => (gatedNow ? Flags.Gated : 0) | (snapshot ? Flags.Snapshot : 0) | (opts.control ? Flags.Control : 0) | (encrypted ? Flags.Enc : 0);
|
|
const emit = async (index: number, last: boolean, data: Uint8Array, gatedNow: boolean): Promise<EncodedFrame> => ({
|
|
index,
|
|
final: last,
|
|
bytes: await seal({ flags: base(gatedNow) | (last ? Flags.Final : 0), itemId: idBytes, index, data }),
|
|
});
|
|
|
|
if (fromFrame > 0) {
|
|
// 断点续传:跳过已经上传的帧。第 k(>=1)帧的内容起点 = cap0 + (k-1) * maxData
|
|
const blob = source as Uint8Array | Blob;
|
|
const total = blob instanceof Uint8Array ? blob.length : blob.size;
|
|
if (total <= cap0) throw new RangeError('内容只有一帧,没有可续传的后续帧');
|
|
const offset = cap0 + (fromFrame - 1) * maxData;
|
|
if (offset >= total) throw new RangeError('fromFrame 超出条目的帧数');
|
|
const rest = blob instanceof Uint8Array ? blob.subarray(offset) : blob.slice(offset);
|
|
let index = fromFrame;
|
|
for await (const chunk of rechunk(rest, maxData, maxData)) {
|
|
yield await emit(index++, chunk.last, chunk.data, true);
|
|
}
|
|
return;
|
|
}
|
|
|
|
let index = 0;
|
|
for await (const chunk of rechunk(source, cap0, maxData)) {
|
|
if (index === 0) {
|
|
// 第 0 帧:来源长度未知时,到这里才知道放不放得进单帧
|
|
if (chunk.last && opts.gated !== true && opts.maxDownloads === undefined) gated = false;
|
|
else if (!chunk.last) {
|
|
if (snapshot) throw new LimitError('快照必须是单帧,内容超过了一帧');
|
|
if (opts.gated === false) throw new LimitError('内容放不进单帧,不能关闭 GATED');
|
|
gated = true;
|
|
}
|
|
yield await emit(0, chunk.last, concat(head, chunk.data), gated);
|
|
} else {
|
|
yield await emit(index, chunk.last, chunk.data, true);
|
|
}
|
|
index++;
|
|
}
|
|
}
|
|
|
|
return { itemId: toBase64Url(idBytes), idBytes, get gated() { return gated; }, frame0Headers, frames: frames() };
|
|
}
|
|
|