131 lines
13 KiB
TypeScript
131 lines
13 KiB
TypeScript
import { applyChanges } from '../canvas-core/document.ts';
|
|
import type { Change, Id, Json, OperationIntent } from '../canvas-core/types.ts';
|
|
import { clone, equal, fault, freeze, readMap, validJson } from '../canvas-core/values.ts';
|
|
|
|
export type Presence = 'absent' | 'exists' | 'unknown';
|
|
export type RequestPhase = 'frozen' | 'dispatch-marked' | 'waiting' | 'resolved';
|
|
export type Outcome = 'unresolved' | 'accepted' | 'applied' | 'not-applied';
|
|
export type FrozenRequest = {
|
|
readonly kind?: OperationIntent['operationType'];
|
|
readonly ownerCanvasId?: Id; readonly runId?: Id;
|
|
readonly id: Id; readonly protocolVersion: number; readonly generation: number; readonly body: string;
|
|
readonly operationIds: readonly Id[]; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>>;
|
|
readonly phase: RequestPhase; readonly outcome: Outcome; readonly error?: string; readonly attempt: number;
|
|
readonly retryAt?: number; readonly retryAction?: 'submit' | 'query';
|
|
};
|
|
export type OperationRecord = { readonly intent: OperationIntent; readonly phase: 'pending' | 'frozen' | 'resolved' | 'blocked'; readonly requestId?: Id; readonly blockers: readonly string[]; readonly resolution?: 'applied' | 'not-applied' | 'superseded' | 'discarded' | 'satisfied'; readonly recoveryId?: Id; readonly replacementId?: Id; readonly basePresence: Readonly<Record<Id, Presence>> };
|
|
export type LedgerSnapshot = { readonly generation: number; readonly durableGeneration: number; readonly operations: ReadonlyMap<Id, OperationRecord>; readonly requests: ReadonlyMap<Id, FrozenRequest> };
|
|
|
|
/** A field-safe compressor for a wholly unsent segment. Frozen records never enter here. */
|
|
export function compressChanges(changes: readonly Change[], options: { preserveLifecycle?: boolean; preserveChecks?: boolean } = {}): readonly Change[] {
|
|
const result: (Change | undefined)[] = [], positions = new Map<string, number[]>();
|
|
for (const change of changes) {
|
|
const id = change.kind === 'field' ? change.target.id : change.id;
|
|
const key = `${change.kind === 'canvas' ? 'canvas' : 'element'}:${id}`, stack = positions.get(key) ?? [];
|
|
const previousIndex = stack.at(-1), previous = previousIndex === undefined ? undefined : result[previousIndex];
|
|
const removePrevious = () => { result[previousIndex!] = undefined; stack.pop(); };
|
|
if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'field') {
|
|
// Apply the same conditional field semantics as the document; never discard a failed before check.
|
|
const document = { projectId: '', elements: new Map([[id, previous.after]]), canvases: new Map(), tombstones: new Map() };
|
|
const after = applyChanges(document, [change]).elements.get(id)!;
|
|
result[previousIndex!] = freeze({ ...previous, after });
|
|
} else if (previous?.kind === 'element' && previous.after && !previous.before && change.kind === 'element' && !change.after && equal(previous.after, change.before) && !options.preserveLifecycle) {
|
|
removePrevious();
|
|
} else if (previous?.kind === 'field' && change.kind === 'field' && previous.field === change.field && previous.target.incarnation === change.target.incarnation && equal(previous.after, change.before) && (previous.afterPresent ?? true) === (change.beforePresent ?? true)) {
|
|
const merged = freeze({ ...previous, after: clone(change.after), afterPresent: change.afterPresent ?? true });
|
|
if (!options.preserveChecks && equal(merged.before, merged.after) && (merged.beforePresent ?? true) === (merged.afterPresent ?? true)) removePrevious(); else result[previousIndex!] = merged;
|
|
} else { stack.push(result.length); positions.set(key, stack); result.push(clone(change)); }
|
|
}
|
|
return freeze(result.filter((change): change is Change => change !== undefined));
|
|
}
|
|
|
|
export function createLedger(initial?: LedgerSnapshot) {
|
|
let generation = initial?.generation ?? 0, durableGeneration = initial?.durableGeneration ?? 0;
|
|
const operations = new Map(initial?.operations), requests = new Map(initial?.requests);
|
|
let cached: LedgerSnapshot;
|
|
const refresh = () => { cached = Object.freeze({ generation, durableGeneration, operations: readMap(operations), requests: readMap(requests) }); };
|
|
refresh();
|
|
function changed() { generation++; refresh(); }
|
|
function requireRequest(id: Id) { const request = requests.get(id); if (!request) throw fault('request_missing', [id]); return request; }
|
|
function assertAcyclic(intent: OperationIntent) {
|
|
const visiting = new Set<Id>();
|
|
const visit = (id: Id): void => {
|
|
if (id === intent.id || visiting.has(id)) throw fault('dependency_cycle', [id]);
|
|
const operation = operations.get(id);
|
|
if (!operation) throw fault('dependency_missing', [id]);
|
|
visiting.add(id); for (const dependency of operation.intent.dependencies) visit(dependency); visiting.delete(id);
|
|
};
|
|
for (const id of intent.dependencies) visit(id);
|
|
}
|
|
return Object.freeze({
|
|
getSnapshot: () => cached,
|
|
add(intent: OperationIntent, basePresence: Readonly<Record<Id, Presence>>) {
|
|
if (!intent.id || !Number.isSafeInteger(intent.sequence) || intent.sequence < 1 || new Set(intent.dependencies).size !== intent.dependencies.length || Object.values(basePresence).some(value => !['absent', 'exists', 'unknown'].includes(value))) throw fault('invalid_operation');
|
|
const existing = operations.get(intent.id);
|
|
if (existing) { if (!equal(existing.intent, intent)) throw fault('operation_identity_conflict', [intent.id]); return; }
|
|
assertAcyclic(intent);
|
|
operations.set(intent.id, freeze({ intent: clone(intent), phase: 'pending', blockers: [], basePresence: { ...basePresence } })); changed();
|
|
},
|
|
eligible(id: Id) {
|
|
const operation = operations.get(id);
|
|
return !!operation && operation.phase === 'pending' && !operation.blockers.length && operation.intent.dependencies.every(dependency => operations.get(dependency)?.phase === 'resolved');
|
|
},
|
|
block(id: Id, reasons: readonly string[]) { const operation = operations.get(id); if (!operation || !['pending', 'blocked'].includes(operation.phase)) return; operations.set(id, freeze({ ...operation, phase: reasons.length ? 'blocked' : 'pending', blockers: [...reasons] })); changed(); },
|
|
freeze(input: { id: Id; kind?: FrozenRequest['kind']; ownerCanvasId?: Id; runId?: Id; protocolVersion?: number; operationIds: readonly Id[]; canvasIds: readonly Id[]; expectedVersions: Readonly<Record<Id, number>>; payload: Json }) {
|
|
if (!input.id || requests.has(input.id) || !Number.isSafeInteger(input.protocolVersion ?? 1) || (input.protocolVersion ?? 1) < 1 || !input.operationIds.length || (!input.canvasIds.length && input.kind!=='asset.upload' && input.kind!=='run.cancel') || !validJson(input.payload) || new Set(input.operationIds).size !== input.operationIds.length || new Set(input.canvasIds).size !== input.canvasIds.length || Object.keys(input.expectedVersions).length !== input.canvasIds.length || input.canvasIds.some(id => !id || !Number.isSafeInteger(input.expectedVersions[id]) || input.expectedVersions[id] < 0)) throw fault('invalid_frozen_request');
|
|
if (input.kind === 'run.start' && (!input.ownerCanvasId || !input.canvasIds.includes(input.ownerCanvasId))) throw fault('invalid_run_request');
|
|
if ((input.kind === 'run.cancel' || input.kind === 'run.apply') && (!input.ownerCanvasId || !input.runId)) throw fault('invalid_run_request');
|
|
for (const id of input.operationIds) {
|
|
const operation = operations.get(id);
|
|
if ((input.kind ?? 'document.change') !== operation?.intent.operationType) throw fault('request_kind_mismatch',[id]);
|
|
if (!operation || operation.phase !== 'pending' || operation.blockers.length || operation.intent.dependencies.some(dependency => operations.get(dependency)?.phase !== 'resolved' && !input.operationIds.includes(dependency))) throw fault('operation_not_ready', [id]);
|
|
}
|
|
const request: FrozenRequest = freeze({ id: input.id, kind: input.kind ?? 'document.change', ...(input.ownerCanvasId ? { ownerCanvasId: input.ownerCanvasId } : {}), ...(input.runId ? { runId: input.runId } : {}), protocolVersion: input.protocolVersion ?? 1, generation: generation + 1, body: JSON.stringify(input.payload), operationIds: [...input.operationIds], canvasIds: [...input.canvasIds].sort(), expectedVersions: { ...input.expectedVersions }, phase: 'frozen', outcome: 'unresolved', attempt: 0 });
|
|
requests.set(request.id, request);
|
|
for (const id of input.operationIds) operations.set(id, freeze({ ...operations.get(id)!, phase: 'frozen', requestId: request.id }));
|
|
changed(); return request;
|
|
},
|
|
markDispatch(id: Id) {
|
|
const request = requireRequest(id);
|
|
if (request.phase === 'resolved') throw fault('request_resolved', [id]);
|
|
const next: FrozenRequest = freeze({ ...request, phase: 'dispatch-marked', attempt: request.attempt + 1, retryAt: undefined, retryAction: undefined }); requests.set(id, next); changed(); return next;
|
|
},
|
|
result(id: Id, outcome: Outcome, error?: string, reconciled = true) {
|
|
const request = requireRequest(id);
|
|
if ((request.phase === 'resolved' || request.outcome === 'applied') && request.outcome !== outcome) throw fault('receipt_conflict', [id]);
|
|
const phase = (outcome === 'applied' && reconciled) || outcome === 'not-applied' ? 'resolved' : 'waiting';
|
|
requests.set(id, freeze({ ...request, phase, outcome, error, retryAt: undefined, retryAction: undefined }));
|
|
if (outcome === 'applied' && reconciled) for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'resolved', resolution: 'applied' }));
|
|
if (outcome === 'not-applied') for (const operationId of request.operationIds) operations.set(operationId, freeze({ ...operations.get(operationId)!, phase: 'blocked', blockers: [error || 'not_applied'] }));
|
|
changed();
|
|
},
|
|
resolveNoEffect(id: Id) {
|
|
const request=requireRequest(id);
|
|
if(request.outcome!=='not-applied'||!['run.start','run.cancel'].includes(request.kind??''))throw fault('invalid_no_effect_resolution');
|
|
for(const operationId of request.operationIds)operations.set(operationId,freeze({...operations.get(operationId)!,phase:'resolved',resolution:'not-applied'}));
|
|
changed();
|
|
},
|
|
/** Local retirement is not an applied receipt. Validate the complete closure before mutating. */
|
|
retire(recoveryId:Id,decisions:readonly {operationId:Id;resolution:'superseded'|'discarded'|'satisfied';replacementId?:Id}[]) {
|
|
const ids=new Set(decisions.map(decision=>decision.operationId));
|
|
if(!recoveryId||ids.size!==decisions.length||new Set(decisions.flatMap(decision=>decision.replacementId?[decision.replacementId]:[])).size!==decisions.filter(decision=>decision.replacementId).length)throw fault('invalid_recovery_decisions');
|
|
for(const decision of decisions){
|
|
const record=operations.get(decision.operationId),request=record?.requestId?requests.get(record.requestId):undefined;
|
|
if(!record||record.phase==='resolved'||record.intent.operationType==='asset.upload'||record.requestId&&(!request||request.phase!=='resolved'||request.outcome!=='not-applied'))throw fault('recovery_operation_not_safe',[decision.operationId]);
|
|
if(!['superseded','discarded','satisfied'].includes(decision.resolution)||(decision.resolution==='superseded')!==!!decision.replacementId||decision.replacementId&&(operations.has(decision.replacementId)||ids.has(decision.replacementId)))throw fault('invalid_recovery_decisions');
|
|
}
|
|
for(const record of operations.values())if(record.phase!=='resolved'&&!ids.has(record.intent.id)&&record.intent.dependencies.some(id=>ids.has(id)))throw fault('recovery_dependency_closure',[record.intent.id]);
|
|
for(const decision of decisions)operations.set(decision.operationId,freeze({...operations.get(decision.operationId)!,phase:'resolved',resolution:decision.resolution,recoveryId,...(decision.replacementId?{replacementId:decision.replacementId}:{})}));
|
|
if(decisions.length)changed();
|
|
},
|
|
scheduleRetry(id: Id, dueAt: number, action: 'submit' | 'query') {
|
|
const request = requireRequest(id);
|
|
if (request.phase === 'resolved' || !Number.isFinite(dueAt)) throw fault('invalid_retry', [id]);
|
|
requests.set(id, freeze({ ...request, retryAt: dueAt, retryAction: action })); changed();
|
|
},
|
|
blockRequest(id: Id, error: string) { const request = requireRequest(id); if (request.phase === 'resolved' || request.error === error) return; requests.set(id, freeze({ ...request, error, retryAt: undefined, retryAction: undefined })); changed(); },
|
|
markDurable(through: number) { if (!Number.isSafeInteger(through) || through < durableGeneration || through > generation) throw fault('invalid_durable_generation'); durableGeneration = through; refresh(); },
|
|
possiblyApplied(id: Id) { const request = requireRequest(id); return request.phase !== 'frozen' && request.outcome !== 'not-applied'; },
|
|
});
|
|
}
|