91 lines
11 KiB
TypeScript
91 lines
11 KiB
TypeScript
/**
|
||
* 格式2的复制源：前段保存轻量topology和payload字节定位，后段保存原完整元素记录。
|
||
* 准备阶段只扫描索引/结构；真正复制某记录前才按range读取和校验payload，支持边转换边加入。
|
||
* 编码仍生成完整Blob，索引/定位仍占O(records)内存，不能据lazy解码宣称总内存常数。
|
||
* 打开索引不证明所有 payload 已读/哈希通过；字节完整性、权限和任务准入由外层负责。
|
||
*/
|
||
import type {CanvasRecord,Document,ElementRecord,Id,Json,Registry,Data} from '../canvas-core/types.ts';
|
||
import {createDocument,validateDocumentTopologySteps} from '../canvas-core/document.ts';
|
||
import {equal,fault,freeze,validJson} from '../canvas-core/values.ts';
|
||
import {createNdjsonImportSource} from './imports.ts';
|
||
import {createCopyBudget} from './copy-budget.ts';
|
||
import type {ImportScheduling} from './import-budget.ts';
|
||
/** Only ownership, endpoint and geometry fields needed to decide scope/coordinates. No component content. */
|
||
/** 保留ID/owner/端点/几何及contentScale/unitScale，清组件内容，边path置null用于范围/坐标规划。
|
||
* 返回结构记录不能直接作为用户元素发布，也不证明组件payload符合schema。
|
||
* 本函数只创建外壳：geometry/端点和保留的 presentation 值仍借用原引用，没有校验或深捕获。 */
|
||
export function copyTopology(record:ElementRecord):ElementRecord{
|
||
const raw=record.data.presentation,presentation=raw&&typeof raw==='object'&&!Array.isArray(raw)?Object.fromEntries(['contentScale','unitScale'].filter(key=>Object.hasOwn(raw,key)).map(key=>[key,raw[key]])):{};
|
||
const data:Data=Object.keys(presentation).length?{presentation}:{};
|
||
const common={id:record.id,incarnation:record.incarnation,ownerCanvasId:record.ownerCanvasId,componentId:record.componentId,dataVersion:record.dataVersion,data};
|
||
return record.structure==='node'?{...common,structure:'node',geometry:record.geometry}:{...common,structure:'connection',source:record.source,target:record.target,path:null};
|
||
}
|
||
/** Format 2 places a compact topology/byte-locator index before the uninterpreted payload region. */
|
||
/** 输出完整format2源：header/canvas/element-index/tombstone/payload标记，然后原元素NDJSON区域。
|
||
* 借用不可变 Document 合同，逐记录 await 后仍读取原表；没有在入口同步捕获整图或执行完整校验。
|
||
* index/payload 两个数组都留存到最终 Blob，预算只能在记录间让出，不能打断一次原生 stringify。 */
|
||
export async function encodeIndexedCopySnapshot(document:Document,signal:AbortSignal,scheduling:ImportScheduling={}):Promise<Blob>{
|
||
const index:BlobPart[]=[],payload:BlobPart[]=[];const budget=createCopyBudget(signal,scheduling);let indexBytes=0,payloadBytes=0;
|
||
/** 单记录序列化、UTF8计长与256KiB上限，保持原始记录/索引locator成本可核验。 */
|
||
function encode(value:unknown){if(signal.aborted)throw fault('copy_aborted');const text=JSON.stringify(value)+'\n',size=new TextEncoder().encode(text).byteLength;if(size>256*1024)throw fault('record_too_large');return{text,size};}
|
||
/** 添加索引记录并检查索引+payload总源上限，offset按payload区域相对字节计。 */
|
||
function append(value:unknown){const part=encode(value);index.push(part.text);indexBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');return part.size;}
|
||
append({kind:'header',version:2,projectId:document.projectId});
|
||
for(const record of document.canvases.values()){const size=append({kind:'canvas',record});await budget.step(Math.max(1,Math.ceil(size/1024)));}
|
||
// 每个元素同时写小topology索引和完整payload，locator保存精确offset/长度而非JS字符位置。
|
||
for(const record of document.elements.values()){
|
||
const part=encode({kind:'element',record}),indexSize=append({kind:'element-index',record:copyTopology(record),offset:payloadBytes,byteLength:part.size});payload.push(part.text);payloadBytes+=part.size;if(indexBytes+payloadBytes>64*1024*1024)throw fault('invalid_import_source');await budget.step(Math.max(1,Math.ceil((part.size+indexSize)/1024)));
|
||
}
|
||
for(const record of document.tombstones){const size=append({kind:'tombstone',record});await budget.step(Math.max(1,Math.ceil(size/1024)));}
|
||
append({kind:'payload',byteLength:payloadBytes});if(signal.aborted)throw fault('copy_aborted');return new Blob([...index,...payload],{type:'application/x-ndjson'});
|
||
}
|
||
/** document只含结构投影，readElement取得完整record；调用方必须明确使用topologyOnly规划模式。 */
|
||
export type IndexedCopySnapshot={readonly document:Document;readElement(id:Id,signal:AbortSignal):Promise<ElementRecord>};
|
||
/** Scan only the small index. Component JSON/schema checks occur just before each record enters the runner. */
|
||
/** 只解析前段索引，验证locator连续/总长/项目与结构，再返回有界窗口读取器。
|
||
* NDJSON reader 可预读越过 payload 标记的原始字节，但打开时不会解析/验证每条完整元素内容。
|
||
* document 是独立捕获的结构文档，locators/窗口表留在闭包；对象冻结不替代 readElement 的后续校验。 */
|
||
export async function openIndexedCopySnapshot(bytes:Blob,projectId:Id,registry:Registry,signal:AbortSignal,scheduling:ImportScheduling={}):Promise<IndexedCopySnapshot>{
|
||
if(!(bytes instanceof Blob)||bytes.size>64*1024*1024)throw fault('invalid_import_source');
|
||
const budget=createCopyBudget(signal,scheduling),source=createNdjsonImportSource({bytes,fingerprint:'indexed-copy-snapshot'}),canvases:CanvasRecord[]=[],elements:ElementRecord[]=[],tombstones=new Map<Id,number>(),locators=new Map<Id,{offset:number;byteLength:number}>();let cursor=0,header=false,payloadOffset=-1,expectedOffset=0;
|
||
while(true){
|
||
if(signal.aborted)throw fault('copy_aborted');const start=cursor,line=await source.read(cursor,signal);if(!line||line.error||line.skip||!line.value)throw fault('invalid_copy_snapshot');cursor=line.nextCursor;
|
||
const value=line.value as Record<string,Json>;
|
||
if(!header){if(value.kind!=='header'||value.version!==2||value.projectId!==projectId)throw fault('invalid_copy_snapshot');header=true;continue;}
|
||
// 标记声明长度必须等于所有locator连续总长，且恰好延伸到Blob EOF，不允许藏额外尾部。
|
||
if(value.kind==='payload'){if(value.byteLength!==expectedOffset||cursor+expectedOffset!==bytes.size)throw fault('invalid_copy_snapshot');payloadOffset=cursor;break;}
|
||
if(value.kind==='canvas')canvases.push(value.record as unknown as CanvasRecord);
|
||
else if(value.kind==='element-index'){
|
||
const record=value.record as unknown as ElementRecord,offset=value.offset as number,byteLength=value.byteLength as number;
|
||
// 每条offset严格等于上条结束，ID唯一且record确实是最小topology表示，不能把未知内容塞进结构索引。
|
||
if(!record||!Number.isSafeInteger(offset)||offset!==expectedOffset||!Number.isSafeInteger(byteLength)||byteLength<1||byteLength>256*1024||locators.has(record.id)||!equal(copyTopology(record),record))throw fault('invalid_copy_snapshot');
|
||
locators.set(record.id,{offset,byteLength});elements.push(record);expectedOffset+=byteLength;if(expectedOffset>bytes.size)throw fault('invalid_copy_snapshot');
|
||
}else if(value.kind==='tombstone'){const row=value.record as [Id,number];if(!Array.isArray(row)||row.length!==2||tombstones.has(row[0]))throw fault('invalid_copy_snapshot');tombstones.set(row[0],row[1]);}else throw fault('invalid_copy_snapshot');
|
||
await budget.step(Math.max(1,Math.ceil((cursor-start)/1024)));
|
||
}
|
||
// 这里只validateDocumentTopologySteps，不把缺payload的投影误送完整组件schema。
|
||
const document=createDocument(projectId,canvases,elements,tombstones);
|
||
for(const _ of validateDocumentTopologySteps(document,registry))await budget.step();
|
||
if(signal.aborted)throw fault('copy_aborted');
|
||
// Cache raw byte windows, never decoded component objects. Preserve parser batching across Blob task boundaries.
|
||
// 缓存原始 byte 窗口而非解码对象，表内最多4窗口/1MiB；命中不更新顺序，是插入/替换顺序淘汰。
|
||
// 上限只约束表内窗口，不包括 Blob、本次 raw、并发未完成的读取或解码结果；没有并发请求去重。
|
||
const windows=new Map<number,Uint8Array>(),windowBytes=64*1024;let cachedBytes=0;
|
||
/** readElement按locator截取原payload，fatal UTF8/JSON、topology一致及已知schema全部通过才返回冻结record。
|
||
* 每次命中仍重新解码/JSON.parse/校验，不缓存组件对象；此路径没有独立合作预算或异步 schema 执行。
|
||
* AbortSignal 只在入口及实际 Blob 读取后检查，不会打断同步 parse/validate；失败不清空已缓存字节。 */
|
||
return Object.freeze({document,async readElement(id:Id,signal:AbortSignal){
|
||
if(signal.aborted)throw fault('copy_aborted');const locator=locators.get(id);if(!locator)throw fault('element_missing',[id]);
|
||
// 从 payload 相对的64KiB边界起读，跨窗大记录扩大本次窗口；不靠 JS 字符下标定位多字节文本。
|
||
const start=Math.floor(locator.offset/windowBytes)*windowBytes,required=locator.offset+locator.byteLength-start;let window=windows.get(start);
|
||
if(!window||window.byteLength<required){const raw=await bytes.slice(payloadOffset+start,Math.min(bytes.size,payloadOffset+start+Math.max(windowBytes,required))).arrayBuffer();if(signal.aborted)throw fault('copy_aborted');const previous=windows.get(start);if(previous)cachedBytes-=previous.byteLength;window=new Uint8Array(raw);windows.delete(start);windows.set(start,window);cachedBytes+=window.byteLength;while(windows.size>4||cachedBytes>1024*1024){const[key,value]=windows.entries().next().value!;windows.delete(key);cachedBytes-=value.byteLength;}}
|
||
const raw=window.subarray(locator.offset-start,required);let value:{kind:string;record:ElementRecord};
|
||
try{value=JSON.parse(new TextDecoder('utf-8',{fatal:true}).decode(raw));}catch{throw fault('invalid_copy_snapshot');}
|
||
if(!value||value.kind!=='element'||!validJson(value.record)||!value.record.data||typeof value.record.data!=='object'||Array.isArray(value.record.data)||!equal(copyTopology(value.record),document.elements.get(id)))throw fault('invalid_copy_snapshot');
|
||
// 当前 registry 已知版本才执行其 schema；未知版本保留通用 JSON/拓扑，不等于已理解内容。
|
||
// readElement 不迁移版本、不执行复制解释器；这些要求由后续复制转换流程处理。
|
||
const record=value.record,definition=registry.components.get(record.componentId);if(registry.resolve(record)==='known'&&(!definition||definition.structure!==record.structure||!definition.validate(record.data)))throw fault('invalid_component_data',[id]);
|
||
return freeze(record);
|
||
}});
|
||
}
|