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'; type Range={start:number;end:number}; type ScanCache={start:number;bytes:Uint8Array}; type Canvas={id:string;parent:null|{canvasId:string;entryElementId:string}}; type Options=ImportScheduling&{bytes:Blob;fingerprint:string;signal:AbortSignal;maxRecordBytes?:number}; const allowed=new Set(['documentFormatVersion','projectId','canvases','elements','tombstones']); const whitespace=(byte:number)=>byte===9||byte===10||byte===13||byte===32; /** Each scan owns its cursor. An aborted scan cannot advance the resumable source. */ 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; const check=()=>{if(signal.aborted)throw fault('import_aborted');}; async function peek(){ check();if(cursor>=bytes.size)return -1; if(cursor=cache.start+cache.bytes.length){const loaded=new Uint8Array(await bytes.slice(cursor,cursor+16384).arrayBuffer());cache.start=cursor;cache.bytes=loaded;check();} return cache.bytes[cursor-cache.start]; } function step(){cursor++;return budget.step();} 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();} async function value():Promise{ 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}; } async function decode(range:Range,limit=options.maxRecordBytes??256*1024):Promise{ check();if(range.end-range.start>limit)throw fault('record_too_large'); let text:string; try{text=new TextDecoder('utf-8',{fatal:true}).decode(await bytes.slice(range.start,range.end).arrayBuffer());}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}; } 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. */ export async function openStreamingDocumentImportSource(input:Options):Promise{ 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(); 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); 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(); 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(); } 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();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[]=[]; let cursor=header.position(),phase:'elements'|'tail'|'done'='elements',first=true,failure:string|undefined,serial=Promise.resolve(); 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; } } 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'; } async function read(position:number,readSignal:AbortSignal):Promise{ if(readSignal.aborted)throw fault('import_aborted'); if(!Number.isSafeInteger(position)||position<0)throw fault('invalid_import_cursor'); const elementIndex=position-children.length; while(elementIndex>=ranges.length&&phase!=='done'){ try{await scanNext(readSignal);}catch(error){if(readSignal.aborted)throw fault('import_aborted');failure=(error as {code?:string}).code??'invalid_document_import';phase='done';} } const count=children.length+ranges.length; 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=positionread(position,readSignal));serial=result.then(()=>{},()=>{});return result; }}); }