import { Flags, type ChannelMode, type ItemKind } from './constants.js'; import { concat, equalBytes, tryDecodeUtf8, toBase64Url, type Bytes } from './bytes.js'; import { IntegrityError, type IntegrityReason } from './errors.js'; import { openFrame, parseEnvelope, type Envelope } from './frame.js'; import { sha256 } from './kdf.js'; import { decodeFirstFrame, type ItemMeta } from './item.js'; import { parseLogRecord, type LogRecord, type RemovedReason } from './log.js'; /** 接收端需要的会话信息。由 ChannelSession 提供。 */ export interface ReaderContext { channelId: Uint8Array; mode: ChannelMode; /** enc 模式:取某个条目的 AES-GCM 密钥。 */ itemKey(itemId: Uint8Array): Promise; } export type ReceivedKind = ItemKind | 'raw'; export interface ReceivedItem { itemId: string; /** 帧所在日志记录的 seq(经 pushFrame 单独喂入且未提供 seq 时为空)。 */ seq: number | undefined; kind: ReceivedKind; /** 仅 enc 模式。 */ meta?: ItemMeta; snapshot: boolean; gated: boolean; /** 已收到已认证的 FINAL,且 0..FINAL 的帧全部到齐。 */ complete: boolean; /** 到第一个缺口为止已收到的内容。大文件请改用 decryptItemStream,不要在这里缓冲。 */ content: Bytes; } export type ItemEvent = /** GATED 条目的通知:字节需要另外拉取(GET /items/{id}/frames),拉取计入下载次数。 */ | { type: 'announce'; itemId: string; seq: number; flags: number; createdAt: number } /** 一个条目有新内容或已完成。小内容(单帧)只会触发一次,complete 为 true。 */ | { type: 'item'; item: ReceivedItem } /** 条目被删除、过期、淘汰、取代或下载达上限。 */ | { type: 'removed'; itemId: string; seq: number; reason: RemovedReason } /** 一帧被拒绝(篡改、乱序、模式不符等)。该条目的缓冲状态已丢弃。 */ | { type: 'rejected'; itemId: string | undefined; seq: number | undefined; reason: IntegrityReason }; interface FrameState { content: Bytes; digest: Bytes; } interface ItemState { itemFlags: number; frames: Map; finalIndex: number | undefined; maxIndex: number; meta: ItemMeta | undefined; } /** * 条目装配器:喂入日志记录或条目的帧,吐出解密、校验后的事件。 * * 校验失败**不抛异常**,而是产生 `rejected` 事件并丢弃该条目的缓冲: * 一个坏帧(或恶意服务端)不应让整个订阅挂掉。直接调用 openFrame 才会抛 IntegrityError。 */ export class ItemReader { readonly #ctx: ReaderContext; readonly #open = new Map(); /** 已完成的条目。回放时重复到达的帧直接忽略,不会让条目「复活」。 */ readonly #done = new Set(); constructor(ctx: ReaderContext) { this.#ctx = ctx; } /** 喂入一条日志记录(WebSocket 的一条二进制消息,或 parseLogRecords 的一项)。 */ async push(input: Uint8Array | LogRecord): Promise { let record: LogRecord; try { record = input instanceof Uint8Array ? parseLogRecord(input) : input; } catch (e) { if (e instanceof IntegrityError) return [{ type: 'rejected', itemId: undefined, seq: undefined, reason: e.reason }]; throw e; } switch (record.kind) { case 'announce': return [{ type: 'announce', itemId: record.itemId, seq: record.seq, flags: record.flags, createdAt: record.createdAt }]; case 'tombstone': this.#open.delete(record.itemId); this.#done.delete(record.itemId); return [{ type: 'removed', itemId: record.itemId, seq: record.seq, reason: record.reason }]; case 'frame': return this.pushFrame(record.envelope, record.seq); } } /** 喂入一个完整的封装(例如拉取 GATED 条目时按帧切出来的)。 */ async pushFrame(envelope: Uint8Array, seq?: number): Promise { let env: Envelope; try { env = parseEnvelope(envelope); } catch (e) { if (e instanceof IntegrityError) return [{ type: 'rejected', itemId: undefined, seq, reason: e.reason }]; throw e; } const itemId = toBase64Url(env.itemId); try { return await this.#accept(env, itemId, seq); } catch (e) { if (!(e instanceof IntegrityError)) throw e; this.#open.delete(itemId); return [{ type: 'rejected', itemId, seq, reason: e.reason }]; } } /** 丢弃某个条目的缓冲(例如用户取消了下载)。 */ forget(itemId: string): void { this.#open.delete(itemId); this.#done.delete(itemId); } async #accept(env: Envelope, itemId: string, seq: number | undefined): Promise { const isEncFrame = (env.flags & Flags.Enc) !== 0; // 降级防御:enc 会话只接受认证过的 enc 帧,服务端不能注入明文帧;raw 会话反之 if (isEncFrame !== (this.#ctx.mode === 'enc')) throw new IntegrityError('wrong_mode'); if (this.#done.has(itemId)) return []; const itemFlags = env.flags & (Flags.Gated | Flags.Snapshot); const gated = (env.flags & Flags.Gated) !== 0; const isFinal = (env.flags & Flags.Final) !== 0; // 非 GATED 条目必须是单帧且 FINAL(协议规则,服务端也强制) if (!gated && (env.frameIndex !== 0 || !isFinal)) throw new IntegrityError('bad_envelope'); let state = this.#open.get(itemId); if (state && state.itemFlags !== itemFlags) throw new IntegrityError('flags_changed'); if (state?.finalIndex !== undefined && env.frameIndex > state.finalIndex) throw new IntegrityError('frame_after_final'); if (isFinal && state) { if (state.maxIndex > env.frameIndex) throw new IntegrityError('final_conflict'); if (state.finalIndex !== undefined && state.finalIndex !== env.frameIndex) throw new IntegrityError('final_conflict'); } // 重复帧:内容必须逐字节相同,相同则视为重放,不产生事件 const digest = await sha256(env.payload); const dup = state?.frames.get(env.frameIndex); if (dup) { if (!equalBytes(dup.digest, digest)) throw new IntegrityError('duplicate_mismatch'); return []; } // 解密 / 取出内容 let data: Bytes; if (this.#ctx.mode === 'enc') { data = await openFrame(await this.#ctx.itemKey(env.itemId), this.#ctx.channelId, env); } else { data = env.payload; } let meta = state?.meta; let content = data; if (this.#ctx.mode === 'enc' && env.frameIndex === 0) { const first = decodeFirstFrame(data); meta = first.meta; content = first.content; } if (!state) { state = { itemFlags, frames: new Map(), finalIndex: undefined, maxIndex: -1, meta: undefined }; this.#open.set(itemId, state); } state.frames.set(env.frameIndex, { content, digest }); state.maxIndex = Math.max(state.maxIndex, env.frameIndex); if (isFinal) state.finalIndex = env.frameIndex; if (meta) state.meta = meta; // 拼接到第一个缺口为止 const parts: Uint8Array[] = []; for (let i = 0; state.frames.has(i); i++) parts.push((state.frames.get(i) as FrameState).content); const complete = state.finalIndex !== undefined && parts.length === state.finalIndex + 1; const item: ReceivedItem = { itemId, seq, kind: this.#ctx.mode === 'raw' ? 'raw' : (state.meta?.kind ?? 'other'), snapshot: (itemFlags & Flags.Snapshot) !== 0, gated, complete, content: concat(...parts), }; if (state.meta) item.meta = state.meta; if (complete) { this.#open.delete(itemId); this.#done.add(itemId); } return [{ type: 'item', item }]; } } // ───────────────────────── 便利函数 ───────────────────────── /** 内容按 UTF-8 解码,非法序列抛 TypeError。 */ export function itemText(item: Pick): string { const text = tryDecodeUtf8(item.content); if (text === null) throw new TypeError('条目内容不是合法的 UTF-8'); return text; } export function itemJson(item: Pick): unknown { return JSON.parse(itemText(item)); }