194 lines
14 KiB
TypeScript
194 lines
14 KiB
TypeScript
/**
|
||
* 导出文档parser5：标准头先建立源树，随后每请求一项才延伸elements范围，实现先转换/加入/显示。
|
||
* 头乱序时明确回退索引解释器，再保留同一ordinal接口；不能因此声称所有文档都零预扫描。
|
||
* 范围扫描串行推进，已定位记录可独立重读；失败尾部成为明确错误记录，成功前缀不被回滚。
|
||
*/
|
||
import {readImportRange,isImportByteFailure,type ImportByteReader} from './import-byte-reader.ts';
|
||
import type {ImportLine,ImportSource} from './imports.ts';
|
||
import {createImportBudget,type ImportScheduling} from './import-budget.ts';
|
||
import {openDocumentImportSource} from './import-document.ts';
|
||
import {fault,validJson} from '../canvas-core/values.ts';
|
||
|
||
/** 已定位元素的原字节范围；只随请求积累，不在标准头准备时扫完整elements尾部。 */
|
||
type Range={start:number;end:number};
|
||
/** 单源共享byte缓存，仅减少重复读取；各scanner仍有自己的逻辑cursor。 */
|
||
type ScanCache={start:number;bytes:Uint8Array};
|
||
/** envelope内的轻量拥有树记录，不包含入口payload或renderer状态。 */
|
||
type Canvas={id:string;parent:null|{canvasId:string;entryElementId:string}};
|
||
/** 原源/指纹、可选已验证byte端口、取消与预算；指纹由上层完整性流程管理。 */
|
||
type Options=ImportScheduling&{bytes:Blob;fingerprint:string;readBytes?:ImportByteReader;signal:AbortSignal;maxRecordBytes?:number};
|
||
/** 明确支持的format1顶层字段，重复/未知字段拒绝，不隐式猜未来格式。 */
|
||
const allowed=new Set(['documentFormatVersion','projectId','canvases','elements','tombstones']);
|
||
/** JSON语法空白字节，不用于处理字符串内容。 */
|
||
const whitespace=(byte:number)=>byte===9||byte===10||byte===13||byte===32;
|
||
|
||
/** Each scan owns a private byte cursor; shared progress records completed boundaries. */
|
||
/** 从指定字节位置建立私有 cursor；仅外层 scanNext 将完整结构边界写回源级进度。
|
||
* 取消不发布业务结果，但已定位的完整 ranges 可保留用于重试，不承诺整次 read 的缓存/边界都回滚。
|
||
* options/缓存沿用内部引用；scanner 负责定位/单值解码，不验证完整文档引用关系。 */
|
||
function scanner(options:Options,signal:AbortSignal,start=0,cache:ScanCache={start:-1,bytes:new Uint8Array()}){
|
||
const {bytes}=options,budget=createImportBudget(signal,options);
|
||
let cursor=start;
|
||
/** 在读取/采纳边界检查本次scanner的signal，缓存命中也不能跳过取消。 */
|
||
const check=()=>{if(signal.aborted)throw fault('import_aborted');};
|
||
/** 共享缓存不足时精确读16KiB，不推进私有cursor，EOF返回-1。 */
|
||
async function peek(){
|
||
check();if(cursor>=bytes.size)return -1;
|
||
if(cursor<cache.start||cursor>=cache.start+cache.bytes.length){const loaded=await readImportRange(bytes,options.readBytes,cursor,cursor+16384,signal);cache.start=cursor;cache.bytes=loaded;check();}
|
||
return cache.bytes[cursor-cache.start];
|
||
}
|
||
/** 推进一个字节并返回预算可能产生的让出Promise。 */
|
||
function step(){cursor++;return budget.step();}
|
||
/** 连续消费非字符串区的合法JSON空白。 */
|
||
async function spaces(){while(whitespace(await peek()))await step();}
|
||
/** 严格匹配一个结构字节，不修补缺逗号/冒号等源损坏。 */
|
||
async function take(byte:number){await spaces();if(await peek()!==byte)throw fault('invalid_document_import');await step();}
|
||
/** 用 quoted/escaped/显式 stack 定位一个值范围，最多 256 层；不收集 payload 字节。
|
||
* 此步只找词法/括号边界，primitive 或字符串内容仍可能是非法 JSON，稍后 decode 才完整解析。
|
||
* 超大值也要扫描完才得到范围，maxRecordBytes 仅约束后续解码分配，不限制扫描总工作量。 */
|
||
async function value():Promise<Range>{
|
||
await spaces();const start=cursor,stack:number[]=[];let quoted=false,escaped=false;
|
||
while(true){
|
||
const byte=await peek();
|
||
if(byte<0){if(quoted||stack.length)throw fault('invalid_document_import');break;}
|
||
if(!quoted&&!stack.length&&(byte===44||byte===93||byte===125||whitespace(byte)))break;
|
||
if(quoted){if(escaped)escaped=false;else if(byte===92)escaped=true;else if(byte===34)quoted=false;}
|
||
else if(byte===34)quoted=true;
|
||
else if(byte===91||byte===123){stack.push(byte);if(stack.length>256)throw fault('json_nesting_limit');}
|
||
else if(byte===93||byte===125){if(stack.pop()!==(byte===93?91:123))throw fault('invalid_document_import');}
|
||
const ended=!quoted&&!stack.length&&(byte===34||byte===93||byte===125);
|
||
await step();if(ended)break;
|
||
}
|
||
if(cursor===start)throw fault('invalid_document_import');return{start,end:cursor};
|
||
}
|
||
/** 读取单 range，按指定上限控制分配，再 fatal UTF-8/JSON.parse/validJson；不重写业务 ID。
|
||
* 默认 payload 256KiB，header 调用可传独立上限；合法 JSON 不等同元素 schema 或引用关系合法。
|
||
* JSON.parse 本身为同步工作，扫描预算不抢占这一步；字节端口异常保持源级错误，不转成坏记录。 */
|
||
async function decode(range:Range,limit=options.maxRecordBytes??256*1024):Promise<any>{
|
||
check();if(range.end-range.start>limit)throw fault('record_too_large');
|
||
const data=await readImportRange(bytes,options.readBytes,range.start,range.end,signal);let text:string;
|
||
try{text=new TextDecoder('utf-8',{fatal:true}).decode(data);}catch(error){check();throw fault('invalid_utf8');}
|
||
check();try{const value=JSON.parse(text);if(!validJson(value))throw Error();return value;}catch{throw fault('invalid_json');}
|
||
}
|
||
return{peek,step,spaces,take,value,decode,position:()=>cursor};
|
||
}
|
||
|
||
/** 仅验证envelope墓碑元数据为正安全整数表，目标导入仍使用新ID/新生命周期。 */
|
||
function tombstones(value:unknown){
|
||
if(!value||typeof value!=='object'||Array.isArray(value)||Object.values(value).some(n=>!Number.isSafeInteger(n)||(n as number)<1))throw fault('invalid_document_import');
|
||
}
|
||
|
||
/** Parser 5: canonical headers open before scanning the elements tail.
|
||
* Out-of-order headers require the indexed interpreter to establish ownership first.
|
||
* Ordinals remain stable; ranges are retained only as records are requested. */
|
||
/** 浅捕获 options 后打开标准头：源最多 64MiB，单个元数据值上限 4MiB，字段名上限 1KiB。
|
||
* 先解析整份 canvases 并验证单根/父链，再返回 canvas-first 的 ordinal 接口；此时元素/尾部可尚未读取。
|
||
* 头顺序不满足拥有树前提时委托索引解释器，仍暴露 parserVersion=5，不能据版本推断一定渐进扫描。
|
||
* 指纹只是上层传入身份；源字节证明/完整 SHA 与最终元素/入口校验由读取端口和导入准备层负责。
|
||
* options.signal 用于打开阶段，返回后的每次 read 使用自己的 readSignal，不自动合并原 signal。 */
|
||
export async function openStreamingDocumentImportSource(input:Options):Promise<ImportSource>{
|
||
const options:Options=Object.freeze({...input});
|
||
const {bytes,signal}=options,max=options.maxRecordBytes??256*1024;
|
||
if(!(bytes instanceof Blob)||bytes.size>64*1024*1024||!options.fingerprint||!Number.isSafeInteger(max)||max<1)throw fault('invalid_import_source');
|
||
// Only range extension scans bytes; ready-record decoding uses independent slices.
|
||
// Keep one source-owned chunk across attempts instead of re-reading 16 KiB per row.
|
||
const cache:ScanCache={start:-1,bytes:new Uint8Array()},scanAt=(signal:AbortSignal,start=0)=>scanner(options,signal,start,cache);
|
||
const header=scanAt(signal),fields=new Map<string,any>();
|
||
await header.take(123);
|
||
while(true){
|
||
const key=await header.decode(await header.value(),1024);
|
||
if(typeof key!=='string'||!allowed.has(key)||fields.has(key))throw fault('invalid_document_import');
|
||
await header.take(58);
|
||
// 必需format/project/canvases还没出现则不能安全解释owner，回退全envelope索引而非猜测。
|
||
if(key==='elements'){
|
||
if(!fields.has('documentFormatVersion')||!fields.has('projectId')||!fields.has('canvases')){
|
||
const indexed=await openDocumentImportSource(options);
|
||
return Object.freeze({...indexed,parserVersion:5});
|
||
}
|
||
await header.take(91);fields.set(key,true);break;
|
||
}
|
||
fields.set(key,await header.decode(await header.value(),4*1024*1024));
|
||
await header.take(44);
|
||
}
|
||
if(fields.get('documentFormatVersion')!==1||typeof fields.get('projectId')!=='string'||!fields.get('projectId')||!Array.isArray(fields.get('canvases')))throw fault('invalid_document_import');
|
||
if(fields.has('tombstones'))tombstones(fields.get('tombstones'));
|
||
const canvases=fields.get('canvases') as Canvas[],byId=new Map<string,Canvas>();
|
||
const budget=createImportBudget(signal,options);
|
||
for(const canvas of canvases){
|
||
if(!canvas||typeof canvas.id!=='string'||!canvas.id||byId.has(canvas.id)||canvas.parent!==null&&(!canvas.parent||typeof canvas.parent.canvasId!=='string'||typeof canvas.parent.entryElementId!=='string'))throw fault('invalid_document_import');
|
||
byId.set(canvas.id,canvas);await budget.step();
|
||
}
|
||
// 验证单根和父链，rooted缓存已经验证的路径，减少深树中重复回溯工作。
|
||
const roots=canvases.filter(canvas=>canvas.parent===null);if(roots.length!==1)throw fault('invalid_document_import');
|
||
const rootId=roots[0].id,rooted=new Set([rootId]);
|
||
for(const canvas of canvases){
|
||
const path=new Set<string>();let current=canvas;
|
||
while(!rooted.has(current.id)){
|
||
if(path.has(current.id)||!current.parent)throw fault('invalid_document_import');
|
||
path.add(current.id);const parent=byId.get(current.parent.canvasId);if(!parent)throw fault('invalid_document_import');current=parent;await budget.step();
|
||
}
|
||
for(const id of path){rooted.add(id);await budget.step();}
|
||
}
|
||
const children=canvases.filter(canvas=>canvas.id!==rootId),ranges:Range[]=[];
|
||
// ranges 随请求积累元素字节范围，children 是已捕获子 canvas；ordinal 与原文件 byte offset 不同。
|
||
// 重开源需重建这些内存范围，持久 runner 保存的是稳定 ordinal/源身份，不是这里的 Promise 链。
|
||
// serial 只串行化范围延伸，failure 最多输出一条终结错误；请求很远的 ordinal 仍会扫描此前全部未定位项。
|
||
let cursor=header.position(),phase:'elements'|'tail'|'done'='elements',first=true,failure:string|undefined,serial=Promise.resolve();
|
||
/** 延伸一个element range，或验证剩余envelope与EOF；只在结构边界确认后更新cursor/phase。 */
|
||
async function scanNext(readSignal:AbortSignal){
|
||
const scan=scanAt(readSignal,cursor);
|
||
if(phase==='elements'){
|
||
await scan.spaces();
|
||
if(await scan.peek()===93){if(!first)throw fault('invalid_document_import');await scan.step();cursor=scan.position();phase='tail';}
|
||
else{
|
||
const range=await scan.value();await scan.spaces();const delimiter=await scan.peek();
|
||
if(delimiter!==44&&delimiter!==93)throw fault('invalid_document_import');
|
||
await scan.step();cursor=scan.position();phase=delimiter===93?'tail':'elements';first=false;ranges.push(range);return;
|
||
}
|
||
}
|
||
// elements结束后仍检查顶层重复/未知字段、可选tombstones与实际EOF，不能略过坏尾部。
|
||
const seen=new Set(fields.keys());await scan.spaces();
|
||
while(await scan.peek()!==125){
|
||
await scan.take(44);const key=await scan.decode(await scan.value(),1024);
|
||
if(typeof key!=='string'||!allowed.has(key)||seen.has(key))throw fault('invalid_document_import');
|
||
seen.add(key);await scan.take(58);const value=await scan.decode(await scan.value(),4*1024*1024);
|
||
if(key==='tombstones')tombstones(value);
|
||
await scan.spaces();
|
||
}
|
||
await scan.take(125);await scan.spaces();if(scan.position()!==bytes.size)throw fault('invalid_document_import');
|
||
cursor=scan.position();phase='done';
|
||
}
|
||
/** ordinal read：先 children，后 ranges 中元素；源根只作为 sourceRootId 别名，不作为导入子项输出。
|
||
* 子 canvas 可直接返回，元素定位后按独立 range 重读/解码；一个 payload 错误不抹掉其它已定位项。
|
||
* 结构错误记终结 failure；取消/已知字节证明错误只拒绝本次，允许同 ordinal 重试。
|
||
* EOF 仅在有效末序号返回 null，越过末尾仍拒绝，不能把任意大序号当作已完成。 */
|
||
async function read(position:number,readSignal:AbortSignal):Promise<ImportLine|null>{
|
||
if(readSignal.aborted)throw fault('import_aborted');
|
||
if(!Number.isSafeInteger(position)||position<0)throw fault('invalid_import_cursor');
|
||
const elementIndex=position-children.length;
|
||
// 未定位位置才推进源扫描；取消/byte证据错误抛出，其它结构失败转成尾部error记录。
|
||
while(elementIndex>=ranges.length&&phase!=='done'){
|
||
try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;failure=(error as {code?:string}).code??'invalid_document_import';phase='done';}
|
||
}
|
||
const count=children.length+ranges.length;
|
||
// 失败点恰好返回一条可报告记录，下一ordinal为EOF，不能吞掉错误而宣称正常结束。
|
||
if(position>=count){
|
||
if(failure&&position===count)return{cursor:position,nextCursor:position+1,key:`record:${position}`,error:failure};
|
||
if(position===count+(failure?1:0))return null;
|
||
throw fault('invalid_import_cursor');
|
||
}
|
||
const base={cursor:position,nextCursor:position+1,key:`record:${position}`};
|
||
try{
|
||
const kind=position<children.length?'canvas':'element',record=kind==='canvas'?children[position]:await scanner(options,readSignal).decode(ranges[elementIndex]);
|
||
if(!record||typeof record!=='object'||Array.isArray(record)||typeof record.id!=='string'||!record.id)throw fault('invalid_native_import_record');
|
||
return{...base,value:{kind,key:record.id,record,sourceRootId:rootId}};
|
||
}catch(error){if(readSignal.aborted)throw fault('import_aborted');if(isImportByteFailure(error))throw error;return{...base,error:(error as {code?:string}).code??'invalid_document_import'};}
|
||
}
|
||
/** 已定位record的重读不等未来scan的锁；仅扩张ranges通过serial排队，失败不会毒化后续Promise链。 */
|
||
return Object.freeze({fingerprint:options.fingerprint,parserVersion:5,read(position:number,readSignal:AbortSignal){
|
||
// A slow future scan must not block re-reading a pending unit for commit/recovery.
|
||
if(position<children.length+ranges.length)return read(position,readSignal);
|
||
const result=serial.then(()=>read(position,readSignal));serial=result.then(()=>{},()=>{});return result;
|
||
}});
|
||
}
|