100 lines
9.1 KiB
TypeScript
100 lines
9.1 KiB
TypeScript
/**
|
|
* 工作区级生命周期：共享 digest/display/asset depot，按 projectId 复用一个 ProjectEngine。
|
|
* view 只借用 engine，离开页面不关闭持久任务；workspace.close 统一收尾所有项目及资源池。
|
|
* 关闭前先停止新准入，任一项目关闭失败就保留共享池/存储供重试，不用强制清空掩盖未保存状态。
|
|
*/
|
|
import {createAssetDepot} from './asset-depot.ts';
|
|
import type {LoadingState} from './loading.ts';
|
|
import type { Document, Id, Registry } from '../canvas-core/types.ts';
|
|
import { fault, readMap } from '../canvas-core/values.ts';
|
|
import { openProjectEngine, type EngineOptions } from './engine.ts';
|
|
import {createDigestPool,type DigestPort} from './digest-pool.ts';
|
|
import type { StoragePort } from './storage-port.ts';
|
|
import { createDisplayCache } from './display-cache.ts';
|
|
|
|
/** 异步打开的项目引擎类型，文档、历史和运行事实的真正拥有者。 */
|
|
export type ProjectEngine = Awaited<ReturnType<typeof openProjectEngine>>;
|
|
/** 装配身份、存储、共享计算端口与每项目 service adapters；不把凭证写入持久文档。 */
|
|
export type WorkspaceOptions = {
|
|
accountId: Id; workspaceId: Id; authEpoch: number; registry: Registry; storage: StoragePort; digests?:DigestPort; images?:import('./image-decoder.ts').ImageDecoderPort; queryBudget?:EngineOptions['queryBudget'];
|
|
isCurrentIdentity(): boolean;
|
|
adapters(projectId: Id, current: () => boolean): Pick<EngineOptions, 'persistence' | 'assets' | 'runs'>;
|
|
onError?: (error: unknown) => void;
|
|
};
|
|
|
|
/** Workspace owns persistent projects; a view only borrows one and never closes it. */
|
|
/** 建立 workspace 共享池和 projectId→打开 Promise 表，不在构造时 claim 项目存储或恢复文档。
|
|
* 注入的 digests/images/storage 也由本容器在最终 close 时关闭，跨 workspace 共享它们需宿主另行管理所有权。
|
|
* options 仍由闭包读取，身份/端口应保持稳定；公开标量与 assetDepot scope 并非每次随外部 options 重写。
|
|
* getProjects 捕获只读表成员，但 Promise 兑现的 engine 仍是业务 API，不是不可操作的纯数据快照。 */
|
|
export function createWorkspaceRuntime(options: WorkspaceOptions) {
|
|
const digests=options.digests??createDigestPool();
|
|
const displays=createDisplayCache({onError:options.onError});
|
|
// 缓存打开中的 Promise 而非仅缓存完成实例，防止并发两个 view 为同项目申请两个 leader。
|
|
const projects = new Map<Id, Promise<ProjectEngine>>();
|
|
let accepting = true, closed = false, closing: Promise<void> | undefined;
|
|
/** 准入同时要求工作区仍接受任务和宿主身份仍有效，关闭一开始就让旧调用失效。 */
|
|
const current = () => accepting && !closed && options.isCurrentIdentity();
|
|
const assetDepot=options.storage.assetDepot?createAssetDepot({scope:options,port:options.storage.assetDepot,current,digest:(bytes,signal)=>digests.digest(bytes,signal)}):undefined;
|
|
/** current 通过后按 document.projectId 复用在途/已打开 Promise，避免同 workspace 同项目重复 claim leader。
|
|
* 命中时不以新 document/versions/loading 替换初始参数；open/openPartial 随后再协调新输入。
|
|
* 打开拒绝仅移除仍对应自己的 Promise，准入有效时下次可重试；adapters 的同步抛错直接传出。 */
|
|
function acquire(document:Document,versions:ReadonlyMap<Id,number>,loading?:LoadingState){
|
|
if(!current())return Promise.reject(fault('identity_changed'));
|
|
const existing=projects.get(document.projectId);if(existing)return existing;
|
|
const opening=openProjectEngine({scope:{accountId:options.accountId,workspaceId:options.workspaceId,projectId:document.projectId},registry:options.registry,document,versions,loading,storage:options.storage,assetDepot,displays,digests,images:options.images,queryBudget:options.queryBudget,...options.adapters(document.projectId,current),onError:options.onError,isCurrentIdentity:current,policy:()=>current()?{enabled:true}:{enabled:false,code:'identity_changed',targets:[]},traversalPolicy:()=>current()?{enabled:true}:{enabled:false,code:'identity_changed',targets:[]}});
|
|
projects.set(document.projectId,opening);void opening.catch(()=>{if(projects.get(document.projectId)===opening)projects.delete(document.projectId);});return opening;
|
|
}
|
|
return Object.freeze({
|
|
accountId: options.accountId, workspaceId: options.workspaceId, authEpoch: options.authEpoch, registry: options.registry,
|
|
isCurrentIdentity: current,
|
|
assetDepot,
|
|
/** 查看项目打开句柄的只读表，不授予外部关闭/替换内部 ownership 的能力。 */
|
|
getProjects: () => readMap(projects),
|
|
/** 取得 engine 后复查身份：同 sourceToken 的 loading 用 complete，否则尝试普通 refresh。
|
|
* 任一传入版本低于当前已知值就跳过整个 refresh；其余一致性由 engine 校验，不在这里合并版本向量。
|
|
* complete 拒绝总会抛错；refresh 拒绝仅在原 engine 仍 loading 时抛错，非 loading 可保留原 engine 返回。
|
|
* 因此返回 engine 表示复用成功，不等于传入 document 已被采纳或服务器写入已完成。 */
|
|
open(document: Document, versions: ReadonlyMap<Id, number>,settings:{sourceToken?:string}={}) {
|
|
return acquire(document,versions).then(async engine=>{if(!current())throw fault('identity_changed');const loading=engine.loading.getState();if(loading&&settings.sourceToken===loading.sourceToken){const result=await engine.loading.complete(document,versions,{sourceToken:settings.sourceToken});if(result.status==='rejected')throw fault(result.code,result.targets);}else{const previous=engine.getSnapshot().runtime.versions;if(![...versions].some(([id,value])=>(previous.get(id)??0)>value)){const result=engine.refresh(document,versions);if(result.status==='rejected'&&loading)throw fault(result.code,result.targets);}}return engine;});
|
|
},
|
|
/** 新项目以 partial 文档/loading 打开；已有项目只在当前 loading.sourceToken 匹配时 accept。
|
|
* 不同 token 或已无 loading 时直接返回原 engine，不替换其文档/加载任务；匹配但 accept 拒绝则抛错。 */
|
|
openPartial(document:Document,versions:ReadonlyMap<Id,number>,loading:LoadingState){
|
|
return acquire(document,versions,loading).then(engine=>{if(!current())throw fault('identity_changed');const previous=engine.loading.getState();if(previous?.sourceToken===loading.sourceToken){const result=engine.loading.accept(document,versions,loading);if(result.status==='rejected')throw fault(result.code,result.targets);}return engine;});
|
|
},
|
|
/** view 离开后仍可 resume 已打开项目的背景加载；等待 opening 前后检查 current。
|
|
* loading.resume 的在途身份/结果围栏由 engine 负责，此方法在其返回后不额外再查 current。
|
|
* 错误仅在当前身份仍有效时报告 onError，随后原样拒绝；不吞错或清除项目句柄。 */
|
|
async resumeLoading(projectId:Id){
|
|
try{if(!current())throw fault('identity_changed');const opening=projects.get(projectId);if(!opening)throw fault('project_not_open');const engine=await opening;if(!current())throw fault('identity_changed');const result=await engine.loading.resume();if(result.status==='rejected')throw fault(result.code,result.targets);return result;}
|
|
catch(error){if(current()){try{options.onError?.(error);}catch{}}throw error;}
|
|
},
|
|
/** 先 accepting=false 永久停止此实例新准入，再单飞等待全部在途打开/engine 关闭。
|
|
* 打开失败视为无需关闭；engine 关闭失败聚合抛错并保留项目/共享池，后续 close 可重试但不重新 accepting。
|
|
* 项目全成功后才依次同步关闭共享端口；这段不是原子事务，注入端口 close 应可安全重复，抛错也允许重试。
|
|
* 全部收尾成功才清 projects/置 closed；此接口不证明外部服务或服务器生成任务也已停止。 */
|
|
close() {
|
|
accepting = false; // Fence commands/transports before awaiting pending opens or storage.
|
|
if (closed) return Promise.resolve();
|
|
if (closing) return closing;
|
|
const task = (async () => {
|
|
// 未成功打开的项目无需关闭；已打开项目各自等待 durable 收尾，allSettled 不漏掉其它项目的错误。
|
|
const results = await Promise.allSettled([...projects.values()].map(async pending => {
|
|
let engine: ProjectEngine;
|
|
try { engine = await pending; } catch { return; }
|
|
await engine.close();
|
|
}));
|
|
const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason] : []);
|
|
if (errors.length) throw new AggregateError(errors, 'workspace_close_failed');
|
|
// 只有所有项目可靠关闭才到这里；失败走 finally 清 closing，下一次 close 可重试，资源不先释放。
|
|
assetDepot?.close();displays.close();digests.close();options.images?.close();options.storage.close(); projects.clear(); closed = true;
|
|
})();
|
|
closing = task.finally(() => { closing = undefined; });
|
|
return closing;
|
|
},
|
|
});
|
|
}
|
|
/** 公共工作区 API 从实现推导，避免另一份可变 store 合同。 */
|
|
export type WorkspaceRuntime = ReturnType<typeof createWorkspaceRuntime>;
|