Endoleap/tests/headless/import-document-stream.test.mjs

import {test} from 'node:test';
import assert from 'node:assert/strict';
import {openStreamingDocumentImportSource} from '../../src/lib/canvas-runtime/import-document-stream.ts';
const signal=()=>new AbortController().signal;
const header={documentFormatVersion:1,projectId:'source',canvases:[{id:'root',parent:null}]};
const open=(bytes,options={})=>openStreamingDocumentImportSource({bytes:bytes instanceof Blob?bytes:new Blob([typeof bytes==='string'?bytes:JSON.stringify(bytes)]),fingerprint:'immutable',signal:signal(),...options});
const deferred=()=>{let resolve;const promise=new Promise(done=>resolve=done);return{promise,resolve};};

test('canonical document opens and emits its first record without reading the gated payload tail; aborted scanning can retry the same ordinal',async()=>{
 const gate=deferred(),entered=deferred(),reads=[];let block=true;
 class Source extends Blob{slice(start=0,end=this.size,type){reads.push([start,end]);const part=super.slice(start,end,type);if(start<32768||!block)return part;return{async arrayBuffer(){entered.resolve();await gate.promise;return part.arrayBuffer();}};}}
 const bytes=new Source([JSON.stringify({...header,elements:[{id:'first'},{id:'second',text:'x'.repeat(100000)}],tombstones:{}})]),reader=await open(bytes);
 assert.equal(reader.parserVersion,5);assert.equal(reads.some(([start])=>start>=32768),false);
 assert.equal((await reader.read(0,signal())).value.record.id,'first');assert.equal(reads.some(([start])=>start>=32768),false);
 const controller=new AbortController(),pending=reader.read(1,controller.signal);await entered.promise;let timer;try{const repeated=await Promise.race([reader.read(0,signal()),new Promise((_,reject)=>timer=setTimeout(()=>reject(Error('a blocked future scan prevented re-reading a ready record')),500))]);assert.equal(repeated.value.record.id,'first');}finally{clearTimeout(timer);controller.abort();block=false;gate.resolve();}await assert.rejects(pending,/import_aborted/);
 assert.equal((await reader.read(1,signal())).value.record.id,'second');assert.equal(await reader.read(2,signal()),null);
});

test('reopening and concurrent reads keep stable canvas-first ordinals and original root aliases',async()=>{
 const input={...header,canvases:[...header.canvases,{id:'nested',parent:{canvasId:'root',entryElementId:'frame'}}],elements:[{id:'frame'},{id:'member',ownerCanvasId:'nested',data:{text:'图,] \\"'}}],tombstones:{old:2}},reader=await open(input);
 const rows=await Promise.all([0,1,2,1].map(position=>reader.read(position,signal())));assert.deepEqual(rows.map(row=>row.value.record.id),['nested','frame','member','frame']);assert.equal(rows[2].value.sourceRootId,'root');const reopened=await open(input);assert.deepEqual(await reopened.read(2,signal()),rows[2]);assert.equal(await reader.read(3,signal()),null);await assert.rejects(reader.read(5,signal()),/invalid_import_cursor/);
});

test('bad envelope tails preserve legal records and become one explicit terminal failure',async()=>{
 for(const suffix of [',"tombstones":{"bad":0}}','} trailing',',"elements":[]}',',"unknown":true}',',}']){
  const reader=await open(JSON.stringify({...header,elements:[{id:'first'},{id:'last'}]}).slice(0,-1)+suffix);
  assert.equal((await reader.read(0,signal())).value.record.id,'first');assert.equal((await reader.read(1,signal())).value.record.id,'last');const failure=await reader.read(2,signal());assert.ok(failure.error,suffix);assert.equal(failure.nextCursor,3);assert.equal(await reader.read(3,signal()),null);
 }
 const trailing=await open(JSON.stringify(header).slice(0,-1)+',"elements":[{"id":"first"},]}');assert.equal((await trailing.read(0,signal())).value.record.id,'first');assert.equal((await trailing.read(1,signal())).error,'invalid_document_import');
});

test('out-of-order headers retain indexed compatibility; invalid headers and cycles are rejected before publishing',async()=>{
 const reader=await open({elements:[{id:'first'}],...header});assert.equal(reader.parserVersion,5);assert.equal((await reader.read(0,signal())).value.record.id,'first');
 for(const input of [{...header,documentFormatVersion:2,elements:[]},{...header,canvases:[...header.canvases,{id:'cycle',parent:{canvasId:'cycle',entryElementId:'self'}}],elements:[]},{...header,canvases:[...header.canvases,{id:'lost',parent:{canvasId:'missing',entryElementId:'x'}}],elements:[]}])await assert.rejects(open(input),/invalid_document_import/);
});

test('oversized payload is scanned without decoding and later records still import',async()=>{
 const reader=await open({...header,elements:[{id:'huge',text:'x'.repeat(300000)},{id:'next'}]});assert.equal((await reader.read(0,signal())).error,'record_too_large');assert.equal((await reader.read(1,signal())).value.record.id,'next');assert.equal(await reader.read(2,signal()),null);
});

test('sequential records reuse the source scan chunk rather than reading a new 16 KiB window per row',async()=>{
 let chunks=0;class Counted extends Blob{slice(start=0,end=this.size,...rest){if(end-start===16384)chunks++;return super.slice(start,end,...rest);}}
 const input={...header,elements:Array.from({length:1000},(_,index)=>({id:`element-${index}`,text:'small'}))},bytes=new Counted([JSON.stringify(input)]),reader=await open(bytes);
 for(let index=0;index<1000;index++)assert.equal((await reader.read(index,signal())).value.record.id,`element-${index}`);
 assert.equal(await reader.read(1000,signal()),null);assert.ok(chunks<=Math.ceil(bytes.size/16384)+1,`${chunks} scan reads for ${bytes.size} bytes`);
});

test('source identity and bytes are captured before the first await, independent of caller option mutation',async()=>{
 const options={bytes:new Blob([JSON.stringify({...header,elements:[{id:'original'}]})]),fingerprint:'original-file',signal:signal()},opening=openStreamingDocumentImportSource(options);options.bytes=new Blob([JSON.stringify({...header,elements:[{id:'replaced'}]})]);options.fingerprint='replacement-file';const reader=await opening;assert.equal(reader.fingerprint,'original-file');assert.equal((await reader.read(0,signal())).value.record.id,'original');assert.equal(await reader.read(1,signal()),null);
});