133 lines
10 KiB
TypeScript
133 lines
10 KiB
TypeScript
import type { Document, Id, Registry } from '../canvas-core/types.ts';
|
|
import { createEditor, type EditorOptions } from '../canvas-core/editor.ts';
|
|
import { createDocument } from '../canvas-core/document.ts';
|
|
import { createSessions } from '../canvas-core/gestures.ts';
|
|
import { equal, fault, freeze, readMap } from '../canvas-core/values.ts';
|
|
import type { EngineSnapshot, PersistencePort, RuntimeState } from './contracts.ts';
|
|
import { createLedger, type FrozenRequest } from './ledger.ts';
|
|
import { operationCanvases, replayPending } from './reconcile.ts';
|
|
import { decodeRecovery, encodeRecovery } from './recovery.ts';
|
|
import { createScheduler, type Clock, type OperationStatus } from './scheduler.ts';
|
|
import type { StoragePort, StorageScope } from './storage.ts';
|
|
|
|
export type EngineOptions = {
|
|
scope: StorageScope; registry: Registry; document: Document; versions: ReadonlyMap<Id, number>;
|
|
storage: StoragePort; persistence: PersistencePort; createId?: () => Id;
|
|
historyLimit?: number; policy?: EditorOptions['policy']; clock?: Clock;
|
|
onError?: (error: unknown) => void;
|
|
};
|
|
|
|
/** One project owner. Detaching a view does not close this engine or abort persistent work. */
|
|
export async function openProjectEngine(options: EngineOptions) {
|
|
if (options.scope.projectId !== options.document.projectId) throw fault('project_mismatch');
|
|
const leader = await options.storage.claim(options.scope);
|
|
try {
|
|
const image = await options.storage.recover(leader), recovered = image ? decodeRecovery(image, options.registry) : undefined;
|
|
if (recovered && recovered.document.projectId !== options.scope.projectId) throw fault('project_mismatch');
|
|
for (const canvas of options.document.canvases.keys()) if (!Number.isSafeInteger(options.versions.get(canvas)) || options.versions.get(canvas)! < 1) throw fault('missing_server_version', [canvas]);
|
|
let closing = false, closed = false, durableRootRevision = image?.rootRevision ?? -1;
|
|
const report = (error: unknown) => { try { options.onError?.(error); } catch { /* Error reporting cannot alter state. */ } };
|
|
const initial: RuntimeState = recovered?.runtime ?? Object.freeze({ acknowledged: createDocument(options.document.projectId, [...options.document.canvases.values()], [...options.document.elements.values()], options.document.tombstones), versions: readMap(options.versions), ledger: createLedger().getSnapshot(), receipts: readMap(new Map()), conflicts: readMap(new Map()) });
|
|
const editor = createEditor({ registry: options.registry, document: options.document, recovery: recovered, runtime: initial, createId: options.createId, historyLimit: options.historyLimit,
|
|
policy(command, document) { if (closing || closed) return { enabled: false, code: 'engine_closing', targets: [] }; return options.policy?.(command, document) ?? { enabled: true }; },
|
|
traversalPolicy() { return closing || closed ? { enabled: false, code: 'engine_closing', targets: [] } : { enabled: true }; },
|
|
onListenerError: report,
|
|
reduceRuntime(previous, next, commit) {
|
|
if (!commit.operationIds.length) return previous;
|
|
const ledger = createLedger(previous.ledger);
|
|
for (const id of commit.operationIds) {
|
|
const intent = next.operations.get(id)!;
|
|
// Last unresolved writer on every participating canvas preserves causal order.
|
|
const last = new Map<Id, Id>();
|
|
for (const record of [...previous.ledger.operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) if (record.phase !== 'resolved') for (const canvas of record.intent.canvasIds) last.set(canvas, record.intent.id);
|
|
const dependencies = [...new Set([...intent.dependencies, ...intent.canvasIds.flatMap(canvas => last.has(canvas) ? [last.get(canvas)!] : [])])];
|
|
const basePresence: Record<Id, 'absent' | 'exists'> = Object.create(null);
|
|
for (const change of intent.changes) if (change.kind === 'element') basePresence[change.id] = change.before ? 'exists' : 'absent'; else if (change.kind === 'field') basePresence[change.target.id] ??= 'exists';
|
|
ledger.add(freeze({ ...intent, dependencies }), basePresence);
|
|
}
|
|
return Object.freeze({ ...previous, ledger: ledger.getSnapshot() });
|
|
},
|
|
});
|
|
const pending = new Map<number, EngineSnapshot>();
|
|
if (!recovered) pending.set(editor.getSnapshot().rootRevision, editor.getSnapshot());
|
|
let writing: Promise<void> | undefined, scheduled = false;
|
|
function persist(): Promise<void> {
|
|
if (writing) return writing;
|
|
const task = (async () => {
|
|
while (pending.size) {
|
|
const [revision, snapshot] = pending.entries().next().value!;
|
|
const receipt = await options.storage.commit(leader, { image: encodeRecovery(snapshot) });
|
|
if (receipt.durableRootRevision !== revision) throw fault('invalid_durable_receipt');
|
|
durableRootRevision = revision; pending.delete(revision);
|
|
if (editor.getSnapshot().runtime.ledger.durableGeneration < snapshot.runtime.ledger.generation) mutate(ledger => ledger.markDurable(snapshot.runtime.ledger.generation));
|
|
}
|
|
})();
|
|
writing = task.finally(() => { writing = undefined; });
|
|
return writing;
|
|
}
|
|
function mutate(reduce: (ledger: ReturnType<typeof createLedger>) => void) {
|
|
const result = editor.updateRuntime(previous => { const ledger = createLedger(previous.ledger); reduce(ledger); return Object.freeze({ ...previous, ledger: ledger.getSnapshot() }); });
|
|
if (result.status === 'rejected') throw fault(result.code, result.targets);
|
|
}
|
|
function complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void {
|
|
const root = editor.getSnapshot(), previous = root.runtime, ledger = createLedger(previous.ledger);
|
|
if (request.phase === 'resolved') return;
|
|
if (status.outcome !== 'applied') { mutate(draft => draft.result(request.id, status.outcome, status.error)); return; }
|
|
const receipts = new Map(previous.receipts), conflicts = new Map(previous.conflicts);
|
|
if (status.canonical !== undefined) receipts.set(request.id, freeze(status.canonical));
|
|
try {
|
|
const proposal = options.persistence.canonical(request, status, root);
|
|
if (proposal.document.projectId !== root.document.projectId) throw fault('project_mismatch');
|
|
for (const canvas of request.canvasIds) {
|
|
const version = proposal.versions.get(canvas);
|
|
if (!version || version <= request.expectedVersions[canvas] || version < (previous.versions.get(canvas) ?? 0)) throw fault('stale_canonical_version', [canvas]);
|
|
}
|
|
// A codec cannot roll back an unrelated canvas while reconciling this request.
|
|
for (const [canvas, version] of previous.versions) if (!request.canvasIds.includes(canvas)) {
|
|
if (proposal.versions.get(canvas) !== version || !equal(proposal.document.canvases.get(canvas), previous.acknowledged.canvases.get(canvas)) || !equal([...proposal.document.elements.values()].filter(element => element.ownerCanvasId === canvas), [...previous.acknowledged.elements.values()].filter(element => element.ownerCanvasId === canvas))) throw fault('canonical_scope_mismatch', [canvas]);
|
|
}
|
|
ledger.result(request.id, 'applied');
|
|
const acknowledged = createDocument(proposal.document.projectId, [...proposal.document.canvases.values()], [...proposal.document.elements.values()], proposal.document.tombstones);
|
|
const document = replayPending(acknowledged, ledger.getSnapshot(), options.registry);
|
|
conflicts.delete(request.id);
|
|
const runtime: RuntimeState = Object.freeze({ ledger: ledger.getSnapshot(), acknowledged, versions: readMap(proposal.versions), receipts: readMap(receipts), conflicts: readMap(conflicts) });
|
|
const result = editor.reconcile(document, runtime, root.rootRevision);
|
|
if (result.status === 'rejected') throw fault(result.code, result.targets);
|
|
} catch (error) {
|
|
// Preserve both the known receipt and the user's working document. No rollback or lane release.
|
|
const waiting = createLedger(previous.ledger); waiting.result(request.id, 'applied', (error as Error).message, false);
|
|
conflicts.set(request.id, (error as Error).message || 'reconcile_conflict');
|
|
const result = editor.updateRuntime(() => Object.freeze({ ...previous, ledger: waiting.getSnapshot(), receipts: readMap(receipts), conflicts: readMap(conflicts) }), root.rootRevision);
|
|
if (result.status === 'rejected') throw fault(result.code, result.targets);
|
|
report(error); return { blocked: true };
|
|
}
|
|
}
|
|
const scheduler = createScheduler({ getLedger: () => editor.getSnapshot().runtime.ledger, mutate, persist,
|
|
createId: options.createId ?? (() => crypto.randomUUID()),
|
|
participants: intent => operationCanvases(intent, editor.getSnapshot().document),
|
|
compile: (intent, id) => options.persistence.compile(intent, id, editor.getSnapshot()),
|
|
submit: (request, signal) => options.persistence.submit(request, signal), query: (request, signal) => options.persistence.query(request, signal), supportsRequest: options.persistence.supportsRequest, complete, onError: report, clock: options.clock,
|
|
});
|
|
const sessions = createSessions(editor);
|
|
const unsubscribe = editor.subscribe((commit, snapshot) => {
|
|
pending.set(snapshot.rootRevision, snapshot);
|
|
if (!scheduled) { scheduled = true; queueMicrotask(() => { scheduled = false; void persist().catch(report); }); }
|
|
// Intents already exist in this root; notification only schedules their IDs.
|
|
queueMicrotask(() => { if (!closing && !closed) scheduler.notify(commit.operationIds); });
|
|
});
|
|
scheduler.start();
|
|
return Object.freeze({
|
|
editor, sessions, getSnapshot: editor.getSnapshot, persist,
|
|
getDurability: () => Object.freeze({ rootRevision: editor.getSnapshot().rootRevision, durableRootRevision }),
|
|
retry: () => scheduler.resume(),
|
|
async settle() { await Promise.resolve(); await scheduler.settle(); await persist(); },
|
|
async close() {
|
|
if (closed) return;
|
|
closing = true; sessions.dispose(); scheduler.stop();
|
|
// Failure keeps the leader/recovery material available for an explicit retry of close.
|
|
await persist(); unsubscribe(); editor.dispose(); await leader.release(); closed = true;
|
|
},
|
|
});
|
|
} catch (error) { await leader.release(); throw error; }
|
|
}
|