/**
 * 按元素 ID 精确路由的只读查询源。一个上游 editor 订阅服务所有元素桶；
 * 普通提交用 changes 找受影响 ID，整份文档替换或加载覆盖变化则做必要的保守复查。
 */
import type {Commit,Id} from '../canvas-core/types.ts';
import type {ElementLookup} from '../canvas-core/queries.ts';
import type {EngineSnapshot} from '../canvas-runtime/contracts.ts';
import {equal} from '../canvas-core/values.ts';
import type {ExternalSource} from './selectors.ts';

/** 注入根读取、发布订阅与元素查询，绑定层不另建可以编辑的元素表。 */
export type ElementSourcePorts={
 getSnapshot():EngineSnapshot;
 subscribe(listener:(commit:Commit,snapshot:EngineSnapshot)=>void):()=>void;
 lookup(snapshot:EngineSnapshot,id:Id):ElementLookup;
};
/** One upstream subscription, ID buckets and cached lookup values, never another document.
 * Explicit changes use ID routing; replacement/coverage cuts retain a checked fallback. */
/** buckets 按 elementId 共享上游查询结果/观察器，每个 forElement 另有即时读取缓存。
 * 首次 subscribe 才接上游，全部桶退订才释放 previous 根引用；创建源或直接读不持有上游租约。
 * 依赖不可变快照与完整 Commit.changes 合同，不从队列/网络结果猜测元素是否已改变。 */
export function createElementSources(ports:ElementSourcePorts){
 // 失败标记使下次成功查询即使值等价也通知恢复；不把异常缓存成一条虚假元素记录。
 type Bucket={value:ElementLookup;listeners:Set<()=>void>;failed?:boolean};
 const buckets=new Map<Id,Bucket>();let stop:(()=>void)|undefined,previous:EngineSnapshot|undefined;
 /** 先更新全部命中桶再通知，逐个隔离监听器错误，最终重抛首个错误给上游边界。 */
 function changed(commit:Commit,next:EngineSnapshot){
  // The editor snapshots its listeners; this callback can already be unsubscribed.
  if(!previous||!buckets.size)return;
  const before=previous!;previous=next;
  const ids=new Set<Id>();
  if(before.document!==next.document){
   // 显式变化只路由 field.target.id / element.id，不按 canvas 变更自动扩散到全体子元素。
   // 文档替换且无 changes 时复查全部已订阅桶；路由准确性依赖发布者提供完整变更及 lookup 的依赖合同。
   if(commit.changes.length){for(const change of commit.changes)if(change.kind==='field')ids.add(change.target.id);else if(change.kind==='element')ids.add(change.id);}
   else for(const id of buckets.keys())ids.add(id);
  }
  // loading 对象或其读取基线变动可改变 absent/unloaded 判定，因此复查非 present 桶。
  // 不将任意 runtime 更新都当元素变化；failed 桶也只有进入本次候选 IDs 才会在此重新查询。
  const coverageChanged=before.runtime.loading!==next.runtime.loading||!!next.runtime.loading&&(before.runtime.loadingRead?.document??before.runtime.acknowledged)!==(next.runtime.loadingRead?.document??next.runtime.acknowledged);
  if(coverageChanged)for(const[id,bucket]of buckets)if(bucket.value.status!=='present')ids.add(id);
  const notifications=new Set<()=>void>();
  for(const id of ids){
   const bucket=buckets.get(id);if(!bucket)continue;
   try{const value=ports.lookup(next,id);if(!bucket.failed&&equal(bucket.value,value))continue;bucket.value=value;bucket.failed=false;}catch{bucket.failed=true;/* Snapshot readers retry/raise through their own boundary. */}
   for(const notify of bucket.listeners)notifications.add(notify);
  }
  // All bucket values advance before observers run; a failing observer cannot starve others.
  let failed=false,error:unknown;
  for(const notify of notifications)try{notify();}catch(caught){if(!failed){failed=true;error=caught;}}
  if(failed)throw error;
 }
 /** 创建单元素外部源；单纯创建源不注册上游，只有 subscribe 才建立观察关系。 */
 return Object.freeze({forElement(id:Id):ExternalSource<ElementLookup>{
  let value:ElementLookup|undefined;
  return Object.freeze({
   /** 每次查询当前根，用 equal 保持等价 ElementLookup 的局部引用；读取不改共享桶通知基线。
    * 不深复制查询值，lookup 必须返回适合不可变使用的结果；异常直接抛给调用方读取边界。 */
   getSnapshot(){const next=ports.lookup(ports.getSnapshot(),id);if(!value||!equal(value,next))value=next;return value;},
   /** 每个订阅包装独立 listener，即使调用方传同一个函数也能分别释放。 */
   subscribe(notify:()=>void){
    let bucket=buckets.get(id);if(!bucket){bucket={value:ports.lookup(ports.getSnapshot(),id),listeners:new Set()};buckets.set(id,bucket);}
    // Distinct subscriptions may intentionally share the same callback identity.
    const listener=()=>notify();bucket.listeners.add(listener);
    // 首次连上游抛错时撤销这次 listener/空桶，避免半注册泄漏。
    if(!stop){previous=ports.getSnapshot();try{stop=ports.subscribe(changed);}catch(error){bucket.listeners.delete(listener);if(!bucket.listeners.size)buckets.delete(id);if(!buckets.size)previous=undefined;throw error;}}
    // 最后一个桶清空才 stop 上游并释放 previous；同回调的多次订阅因 listener 包装仍分别退订。
    // stop 异常不被捕获，后续状态重置不保证执行；宿主的取消订阅端口应不抛错。
    let active=true;return()=>{if(!active)return;active=false;bucket!.listeners.delete(listener);if(!bucket!.listeners.size)buckets.delete(id);if(!buckets.size){stop?.();stop=undefined;previous=undefined;}};
   },
  });
 }});
}