/**
 * 作用域加载协调器测试：多个订阅者可共享同一渐进源，但各自保留导航元数据、取消状态和快照生命周期。
 * 块需绑定 query/权限/project/snapshot/version/sequence，完整标记后仍须正常 EOF；错误尾部只能保留只读前缀。
 * 活动并发、排队、单响应字节/块数及校验后缓存分别受限，旧任务迟到清理不能删除新代次。
 */
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createScopedLoader, loadQueryKey } from '../../src/lib/canvas-runtime/loader.ts';
// 完整查询作用域：账户、workspace、project/canvas、格式、过滤指纹、来源及权限/导航代次；导航代次不阻止共享源。
const scope = { accountId: 'account', workspaceId: 'workspace', projectId: 'project', canvasId: 'canvas', viewKey: 'canvas', format: 'v1', queryFingerprint: 'all', source: 'fixture', authEpoch: 1, navigationGeneration: 1 };
// 可手动 resolve/reject 的 Promise，用来暂停源迭代，检查取消与新代次的关系。
const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; };
// 让出至 setImmediate，使源执行及清理有机会推进，不模拟实际网络延迟。
const tick = () => new Promise(resolve => setImmediate(resolve));
// 构造绑定查询键的合法加载块，extra 专供逐字段篡改；覆盖信息按序累积，value 使用冻结样本。
const chunk = (query, sequence = 0, complete = true, extra = {}) => ({ queryKey: loadQueryKey(query), projectId: 'project', snapshotToken: 'snapshot', serverVersions: { canvas: 3 }, permissionEpoch: query.authEpoch, sequence, coverage: [`part-${sequence}`], complete, byteLength: 2, value: Object.freeze({ part: sequence }), ...extra });

// 两个导航订阅共用一次渐进读取；取消 A 不取消 B，A 保留取消时前缀且不再更新，B 收齐覆盖并完成。
test('subscribers share a progressive read while retaining navigation scopes; cancelling one preserves the other', async () => {
  const gate = deferred(), updates = []; let calls = 0, signal;
  const loader = createScopedLoader({ current: () => true, async *load(query, abort) { calls++; signal = abort; yield chunk(query, 0, false); await gate.promise; yield chunk(query, 1); } });
  const a = loader.acquire(scope, { onUpdate: value => updates.push(value.status) }); await tick();
  assert.equal(a.getSnapshot().status, 'initial-ready'); assert.equal(a.getSnapshot().chunks.length, 1);
  assert.strictEqual(a.getSnapshot(),a.getSnapshot());
  const b = loader.acquire({ ...scope, navigationGeneration: 2 }); assert.equal(b.getSnapshot().scope.navigationGeneration, 2);
  const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled;const stopped=a.getSnapshot(); assert.equal(signal.aborted, false);
  gate.resolve(); const result = await b.ready; assert.equal(calls, 1); assert.equal(result.status, 'complete'); assert.deepEqual(result.coverage, ['part-0', 'part-1']);
  assert.equal(a.getSnapshot().status,'cancelled');assert.equal(a.getSnapshot().chunks.length,1);assert.deepEqual(a.getSnapshot().coverage,['part-0']);
  assert.strictEqual(a.getSnapshot(),stopped);assert.strictEqual(b.getSnapshot(),result);
  assert.deepEqual(updates, ['queued', 'loading', 'initial-ready']); b.release(); loader.close();
});
// 尾部失败保留已接受首块为 failed-with-partial；重新 acquire 必须启动新源，不复用失败 inflight。
test('tail failure keeps the accepted prefix and a fresh retry replaces the failed inflight entry', async () => {
  let calls = 0;
  const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query, 0, false); if (calls === 1) throw Object.assign(Error('tail_failed'), { code: 'tail_failed' }); yield chunk(query, 1); } });
  const a = loader.acquire(scope); await assert.rejects(a.ready, { code: 'tail_failed' }); assert.equal(a.getSnapshot().status, 'failed-with-partial'); assert.equal(a.getSnapshot().chunks[0].value.part, 0);
  const b = loader.acquire(scope); assert.equal((await b.ready).chunks.length, 2); assert.equal(calls, 2); a.release(); b.release(); loader.close();
});
// 源先给 complete 再抛尾部校验错误，ready 必须拒绝且观察者从未收到 complete 状态。
test('a complete marker is not authoritative before iterator EOF; a checksum/trailing failure retains partial status', async () => {
  const statuses = [], loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query); throw Error('trailing_checksum'); } });
  const handle = loader.acquire(scope, { onUpdate: value => statuses.push(value.status) }); await assert.rejects(handle.ready, /trailing_checksum/);
  assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(statuses.includes('complete'), false); handle.release(); loader.close();
});
// 尾块换 snapshotToken、版本向量或 project 时拒绝，前缀 chunks/coverage 保留原值，不混入另一快照。
test('snapshot token, version vector and project changes reject the tail without mixing it into accepted coverage', async () => {
  for (const extra of [{ snapshotToken: 'other' }, { serverVersions: { canvas: 4 } }, { projectId: 'other' }]) {
    const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1, true, extra); } });
    const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); assert.equal(handle.getSnapshot().chunks.length, 1); assert.deepEqual(handle.getSnapshot().coverage, ['part-0']); handle.release(); loader.close();
  }
});
// 错误 queryKey、权限代次、序号、重复覆盖项、负字节数或非法版本在该块暴露前拒绝，结果为空 failed。
test('query, permission, sequence and malformed coverage headers reject before exposing a chunk', async () => {
  for (const extra of [{ queryKey: 'foreign' }, { permissionEpoch: 2 }, { sequence: 3 }, { coverage: ['x', 'x'] }, { byteLength: -1 }, { serverVersions: { canvas: 0 } }]) {
    const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, true, extra); } });
    const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'invalid_load_chunk' }); assert.equal(handle.getSnapshot().chunks.length, 0); assert.equal(handle.getSnapshot().status, 'failed'); handle.release(); loader.close();
  }
});
// EOF 前没有 complete，或 complete 之后又多一块，都不能成为完整结果，已有内容只作为失败前缀保留。
test('an incomplete EOF or extra chunk after a complete marker cannot become a complete result', async () => {
  for (const extra of [false, true]) {
    const loader = createScopedLoader({ current: () => true, async *load(query) { yield chunk(query, 0, extra); if (extra) yield chunk(query, 1); } });
    const handle = loader.acquire(scope); await assert.rejects(handle.ready); assert.equal(handle.getSnapshot().status, 'failed-with-partial'); handle.release(); loader.close();
  }
});
// 响应字节数或块数上限阻止接受第二块；保留首块计数，不能为满足预算静默丢块还宣称成功。
test('response byte and chunk budgets bound retained read copies while preserving earlier chunks', async () => {
  for (const limits of [{ maxResponseBytes: 3 }, { maxChunks: 1 }]) {
    const loader = createScopedLoader({ ...limits, current: () => true, async *load(query) { yield chunk(query, 0, false); yield chunk(query, 1); } });
    const handle = loader.acquire(scope); await assert.rejects(handle.ready, { code: 'load_response_budget' }); assert.equal(handle.getSnapshot().byteLength, 2); assert.equal(handle.getSnapshot().chunks.length, 1); handle.release(); loader.close();
  }
});
// 活动任务取消后仍占计算/读取槽位；旧迭代真正退出才启动新代次，其清理不能从共享表移除新任务。
test('cancelled active work retains its concurrency slot; its late cleanup cannot delete the newer generation', async () => {
  const first = deferred(), second = deferred(), entered = []; let calls = 0;
  const loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { const index = ++calls; entered.push(index); await (index === 1 ? first.promise : second.promise); yield chunk(query); } });
  const a = loader.acquire(scope); await tick(); const cancelled = assert.rejects(a.ready, { code: 'load_aborted' }); a.release(); await cancelled;
  // a 已经拒绝，但其生成器仍挂在 first；同键 b 必须以新代次排队，不能提前占第二个执行槽。
  const b = loader.acquire(scope); await tick(); assert.equal(calls, 1); assert.equal(b.getSnapshot().status, 'queued');
  assert.throws(() => loader.acquire({ ...scope, canvasId: 'third' }), { code: 'load_capacity' });
  first.resolve(); await tick(); assert.equal(calls, 2); const peer = loader.acquire(scope); assert.equal(peer.getSnapshot().generation, b.getSnapshot().generation);
  second.resolve(); await Promise.all([b.ready, peer.ready]); assert.deepEqual(entered, [1, 2]); b.release(); peer.release(); loader.close();
});
// 等待中的订阅取消可立即释放排队名额，放弃的 source 不执行；后续查询按空出的队列位置进入。
test('queued cancellation frees admission capacity and never executes the abandoned source', async () => {
  const gate = deferred(), called = [], loader = createScopedLoader({ concurrency: 1, maxQueued: 1, current: () => true, async *load(query) { called.push(query.canvasId); await gate.promise; yield chunk(query); } });
  const a = loader.acquire(scope), b = loader.acquire({ ...scope, canvasId: 'cancelled' }); const cancelled = assert.rejects(b.ready, { code: 'load_aborted' }); b.release(); await cancelled;
  const c = loader.acquire({ ...scope, canvasId: 'next' }); gate.resolve(); await Promise.all([a.ready, c.ready]); assert.deepEqual(called, ['canvas', 'next']); a.release(); c.release(); loader.close();
});
// 已完成读取缓存受字节 LRU、显式 source validation 和权限时间限制；不认可、过期及墙上时钟回退均重新读取。
test('read cache requires explicit source validation and is bounded by byte LRU and permission age', async () => {
  let time = 100, accepted = true; const calls = new Map();
  const loader = createScopedLoader({ maxCacheEntries: 10, maxCacheBytes: 4, cacheMaxAgeMs: 100, now: () => time, acceptCached: () => accepted, current: () => true,
    async *load(query) { calls.set(query.canvasId, (calls.get(query.canvasId) ?? 0) + 1); yield chunk(query); } });
  // 缓存用例统一领取、等待及释放；只有显式 allow-validated 才允许考虑已缓存结果。
  const read = async (canvasId, cached = false) => { const handle = loader.acquire({ ...scope, canvasId }, { cache: cached ? 'allow-validated' : 'fresh' }); await handle.ready; handle.release(); };
  // 每项声明 2 字节、总预算 4：访问 a 后再放 c 应淘汰 b；容量是协议 byteLength 计数，不是进程堆内存测量。
  await read('a'); await read('b'); await read('a', true); assert.equal(calls.get('a'), 1); await read('c'); await read('b', true); assert.equal(calls.get('b'), 2);
  accepted = false; await read('b', true); assert.equal(calls.get('b'), 3); accepted = true; time = 201; await read('b', true); assert.equal(calls.get('b'), 4);
  time = 0; await read('b', true); assert.equal(calls.get('b'), 5); loader.close();
});
// 相同 canvas 的不同过滤指纹或权限 epoch 不能共享请求，三个查询各自加载。
test('different query fingerprints and permission epochs do not share requests', async () => {
  let calls = 0; const loader = createScopedLoader({ current: () => true, async *load(query) { calls++; yield chunk(query); } });
  const handles = [scope, { ...scope, queryFingerprint: 'filtered' }, { ...scope, authEpoch: 2 }].map(query => loader.acquire(query));
  await Promise.all(handles.map(handle => handle.ready)); assert.equal(calls, 3); handles.forEach(handle => handle.release()); loader.close();
});
// 捕获 scope 只接管声明字段并冻结副本；外部额外对象仍可变，输入 canvas 后改不影响已领取查询。
test('scope capture owns only declared metadata without freezing unrelated caller state', async () => {
  const extra={filter:'mutable'},input={...scope,extra};const loader=createScopedLoader({current:()=>true,async *load(query){yield chunk(query);}});
  const handle=loader.acquire(input);assert.equal(Object.isFrozen(extra),false);input.canvasId='later';extra.filter='changed';
  const result=await handle.ready;assert.equal(result.scope.canvasId,'canvas');assert.equal(Object.hasOwn(result.scope,'extra'),false);assert.equal(Object.isFrozen(result.scope),true);handle.release();loader.close();
});
// 身份失效拒绝迟到块；close 即使源不理 abort 也立即拒绝等待者，迟到完成不能重开 loader。
test('identity changes discard late chunks, and close promptly rejects waiting readers even if the source ignores abort', async () => {
  const gate = deferred(); let current = true;
  const loader = createScopedLoader({ current: () => current, async *load(query) { await gate.promise; yield chunk(query); } });
  const a = loader.acquire(scope); const rejected = assert.rejects(a.ready, { code: 'identity_changed' }); current = false; gate.resolve(); await rejected; assert.equal(a.getSnapshot().chunks.length, 0); a.release(); loader.close();
  const pending = deferred(), closed = createScopedLoader({ current: () => true, async *load(query) { await pending.promise; yield chunk(query); } });
  const b = closed.acquire(scope); const stopped = assert.rejects(b.ready, { code: 'loader_closed' }); closed.close(); await stopped; pending.resolve(); await tick(); assert.equal(b.getSnapshot().status, 'cancelled'); assert.throws(() => closed.acquire(scope), { code: 'loader_closed' });
});
// 观察者回调重入 acquire 不应重通知旧订阅者；新观察者抛错通过专用错误口收集，不使源读取失败。
test('observer reentrancy does not re-notify old subscribers or allow observer exceptions to fail a read', async () => {
  const errors = []; let peer, updates = 0;
  const loader = createScopedLoader({ current: () => true, onObserverError: error => errors.push(error), async *load(query) { yield chunk(query); } });
  const a = loader.acquire(scope, { onUpdate: state => { updates++; if (state.status === 'initial-ready' && !peer) peer = loader.acquire(scope, { onUpdate: () => { throw Error('observer'); } }); } });
  await a.ready; await peer.ready; assert.equal(updates, 4); assert.ok(errors.length > 0); a.release(); peer.release(); loader.close();
});