diff --git a/.codex/environments/environment.toml b/.codex/environments/environment.toml new file mode 100644 index 0000000..29fa228 --- /dev/null +++ b/.codex/environments/environment.toml @@ -0,0 +1,36 @@ +# This file is shared by Codex desktop local environments. +version = 1 + +name = "compendium" + +[setup] +script = ''' +set -euo pipefail + +# Codex creates each agent worktree at a new path. This repository is +# explicitly trusted, so register only its checked-in mise configuration +# before using mise to install the local quality gates. +mise trust --yes mise.toml +npm ci +mise x prek@0.4.8 -- prek install --hook-type pre-commit --hook-type pre-push +''' + +[[actions]] +name = "Pre-commit" +icon = "check" +command = "mise x prek@0.4.8 -- prek run --all-files" + +[[actions]] +name = "Type check" +icon = "check" +command = "npm run check" + +[[actions]] +name = "Coverage" +icon = "test" +command = "npm run test:coverage" + +[[actions]] +name = "End-to-end tests" +icon = "test" +command = "sh scripts/pre-push-check.sh" diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index fcc48bf..8e0efde 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -22,3 +22,9 @@ repos: files: \.(ts|svelte)$ exclude: ^(node_modules|\.svelte-kit|build) pass_filenames: false + - id: e2e + name: Build and run end-to-end tests + language: system + entry: sh scripts/pre-push-check.sh + pass_filenames: false + stages: [pre-push] diff --git a/docs/specifications/audit-coverage.md b/docs/specifications/audit-coverage.md index cd9f960..761d628 100644 --- a/docs/specifications/audit-coverage.md +++ b/docs/specifications/audit-coverage.md @@ -14,9 +14,9 @@ The fix is a generic observer on the server's own `Y.Doc` (`src/lib/server/audit ## 2. How the observer attributes without new tagging -y-protocols/sync applies an incoming client update via `Y.applyUpdate(doc, update, ws)` (`yjs-ws-server.ts`), so the resulting transaction's `origin` is that connection's own `ws` object. Every service-layer write, by contrast, calls `doc.transact(fn)` with no origin, which defaults to `null` (yjs's own `transact(doc, f, origin = null)`). That distinction already existed, unmodified, on every write path in this codebase — `attachDocAuditObserver` (called once per resolved `Y.Doc`, from `workspace-store.ts`'s `createContext()`, after the initial snapshot load) just reads it: +`src/lib/mutation-origin.ts` defines the closed mutation-origin contract. The WebSocket server uses a named `remote-ui` origin, the browser uses `local-ui`, services use `service`, and migration/replay/undo-redo/test paths have their own explicit origins. `attachDocAuditObserver` rejects an unrecognized origin rather than inferring meaning from nullability or object shape. -- `transaction.origin == null` → a service-layer write. Already audited by its own `logAudit` call — the observer does nothing, so nothing is double-counted. +- `service`, `migration`, `replay`, and `undo-redo` are intentionally not observer-audited. Service calls retain their own `logAudit` operation, preventing duplicates. - `transaction.origin` is anything else → a real y-websocket client wrote directly to the doc. The observer resolves what changed and logs it, attributed to `CURRENT_USER` (Phase 0/1 is single-tenant — every live UI connection is the one workspace owner; see `service-layer.md` §3's note that UI writes are currently unscoped). Resolving _what_ changed walks `transaction.changed` (the Yjs-native per-transaction diff): a key added/removed directly on `documents`/`collections`/`records` (the three top-level maps) is a whole entry created or deleted; anything else is walked up via `.parent` until it reaches one of those three maps, attributing the change to whichever entry owns it — e.g. a record's `content` Y.Text's parent is that record's own `Y.Map`, whose parent is the top-level `records` map. A document's `recordIds` reorder (dragging a block, or a record being added/removed from it) resolves to that _document_, not the moved record — reordering is a structural fact about the document distinct from the record's own content, and is logged as its own `update_document`/`update_collection` event alongside whatever `create_record`/`delete_record` event fired in the same transaction. @@ -50,12 +50,12 @@ Create and delete are discrete, comparatively rare events — logged immediately Two operational details this implies: - `flushPendingAuditEvents()` (exported from `audit-observer.ts`) writes any pending debounced events immediately instead of waiting out the window — called on process shutdown (`ydoc.ts`'s `SIGINT`/`SIGTERM` handler) so a debounced edit made just before shutdown isn't lost, and available to tests that don't want to wait out real time. -- The debounce is in-memory and per-process — a pending event is lost (not written) if the process crashes ungracefully before the window elapses or before `flushPendingAuditEvents` runs. Acceptable for Phase 0/1's local-trust, single-process scope; revisit if audit completeness under crash needs a stronger guarantee than "graceful shutdown flushes." +- The audit debounce is in-memory and per-process; graceful shutdown flushes it. Catalog projection does **not** share this limitation: direct UI metadata changes project synchronously (§3). Retention itself (pruning old rows) is out of scope for this feature — `queryAuditLog`'s existing `since`/`until`/`limit` filtering (`persistence.md` §1, `src/lib/server/audit.ts`) is the only volume control today beyond the debounce above; see the tracked follow-up issue for periodic pruning. ## 5. Testing -- `src/lib/server/audit-observer.test.ts` — unit tests against a bare `Y.Doc` (no server, no websocket): create/update/delete for all three entry kinds, the null-origin no-op, debounce coalescing (including a content-edit-plus-field-edit-in-one-transaction case, proving exactly one event per entry per burst, not one per changed Yjs type), `flushPendingAuditEvents`, and the `recordIds`-reorder-attributes-to-the-parent-not-the-record case. +- `src/lib/server/audit-observer.test.ts` — unit tests cover named remote/service/test origins, rejection of unknown origins, debounce coalescing, explicit flush, and parent record-order attribution. - `src/lib/services/services.test.ts` — the new denied-attempt audit tests (`create_record_denied`, `get_document_denied`, no denial logged for a human caller, a denial for a nonexistent record carries no extra metadata). - `tests/e2e/tier-a.test.ts` (#11) — a _real_ y-websocket client (the same harness every other Tier A test uses) mutates its own `Y.Doc` directly, with zero MCP/service-layer calls, mirroring exactly what `BlockEditor.svelte`/`Sidebar.svelte` do today; asserts the server's `audit_log` picks up `create_document`, `create_record`, a debounced-then-flushed `update_record`, and `delete_record`, each written exactly once, and that the live record projection carries the editor and newer timestamp. diff --git a/docs/specifications/persistence.md b/docs/specifications/persistence.md index 6957e15..b14caf0 100644 --- a/docs/specifications/persistence.md +++ b/docs/specifications/persistence.md @@ -12,7 +12,7 @@ **Idle unload (#122, implementing §6's "unloads only when it has no live connections, no active holds, and no unflushed state"):** `workspace-store.ts` runs a process-wide `sweepIdleContexts()` on a 60s (`IDLE_SWEEP_INTERVAL_MS`) timer, wired once per process (`wireIdleSweepOnce()`, mirroring the existing shutdown-hook wiring). Each tick re-evaluates every currently-resolved context and releases (`releaseContextIfIdle`) any with zero live WebSocket connections _and_ zero active holds — checked separately, because an MCP agent's hold is a synthetic Awareness client with no WebSocket connection at all (`holds.ts`), so "zero connections" alone is not a safe unload signal. A release always flushes first, so an idle-unloaded context never loses dirty state. No context is exempt, including the Phase 0 default key — once genuinely idle it unloads and reloads like any other. - **Recovery on restart or reload:** there is no separate "restart mode" — `resolveWorkspaceContext()`'s existing lazy-load path (load-latest-snapshot-if-present, else start empty) is the same code whether the calling process just started or a given key was idle-unloaded ten seconds ago and is now being re-resolved by a new connection. **Data-loss bound:** a graceful shutdown (`SIGINT`/`SIGTERM`, already wired) flushes every resolved context plus pending audit/catalog-mirror events before exit, so it loses nothing. An ungraceful crash can lose up to `SAVE_INTERVAL_MS` (30s) of changes to a context that was neither idle-released (which flushes first) nor yet due for its periodic save tick — the same class of bound `audit-coverage.md` already documents for its own debounce window, not a new risk this introduces. + **Recovery on restart or reload:** there is no separate "restart mode" — `resolveWorkspaceContext()`'s existing lazy-load path (load-latest-snapshot-if-present, else start empty) is the same code whether the calling process just started or a given key was idle-unloaded ten seconds ago and is now being re-resolved by a new connection. Snapshot replay is explicitly tagged `replay`; graceful shutdown flushes every resolved context and pending audit events. Catalog metadata projection is synchronous and has no pending debounce queue. An ungraceful crash can lose up to `SAVE_INTERVAL_MS` (30s) of Yjs changes not yet snapshotted. **Compaction: not applicable.** Each snapshot is a complete `Y.encodeStateAsUpdate` dump on every save, not an incremental update log, so there is nothing to compact — this is a deliberate consequence of the periodic-full-snapshot design above, not a gap. diff --git a/docs/specifications/undo-redo.md b/docs/specifications/undo-redo.md index b9632f3..4462941 100644 --- a/docs/specifications/undo-redo.md +++ b/docs/specifications/undo-redo.md @@ -12,7 +12,7 @@ Each browser tab tracks only its own locally-originated Yjs transactions — nev `src/lib/client/undo.ts` constructs one [`Y.UndoManager`](https://docs.yjs.dev/api/undo-manager) per `Y.Doc` (see §5: since #120, that's per-Document-shard, not a single workspace-wide doc), scoped to the three top-level shared types every local write lands in: the Documents index, the Collections index, and Records (`doc.getMap('documents' | 'collections' | 'records')`). Y.UndoManager tracks a transaction if any type it touched has one of the scope types as an ancestor, not just the scope types themselves — so this also covers every block's nested `Y.Text` content, every record's properties, and each parent's block/row-order array, without needing a separate UndoManager per block. -**Why this requires no new origin-tagging convention:** Y.UndoManager's own default `trackedOrigins` is `new Set([null])` — it only tracks transactions whose origin is `null`. Every local write already goes through an untagged `doc.transact(...)` call (or a single untransacted `.set()`, which Yjs auto-wraps the same way) in `src/lib/data/records.ts` and the block editor, and Yjs assigns those the `null` origin by default. Remote edits — from a collaborator's browser tab or an MCP agent — always arrive over this tab's y-websocket connection and are applied by y-protocols' sync handler with the `WebsocketProvider` instance itself as the transaction origin (see y-websocket's `readSyncMessage` call), which is never `null` and so is never tracked. The local/remote split that "local, per-actor" requires falls directly out of the existing transport architecture (`architecture.md` §1) — no call site needs to opt in. +**Explicit origin contract:** `Y.UndoManager` tracks only `LOCAL_UI_ORIGIN`. Remote UI, service/MCP, migration, replay, and test transactions use distinct named origins from `mutation-origin.ts`; undo/redo transactions are registered as `undo-redo`. No production behavior infers actor class from `null` or an object identity. ## 3. Grouping and correctness under concurrent edits diff --git a/scripts/pre-push-check.sh b/scripts/pre-push-check.sh new file mode 100644 index 0000000..d4f640c --- /dev/null +++ b/scripts/pre-push-check.sh @@ -0,0 +1,5 @@ +#!/bin/sh +set -eu + +npm run build +npm run test:e2e diff --git a/src/lib/client/undo.test.ts b/src/lib/client/undo.test.ts index f6b5900..2371f49 100644 --- a/src/lib/client/undo.test.ts +++ b/src/lib/client/undo.test.ts @@ -12,6 +12,7 @@ import { } from '$lib/data/records'; import type { ActorId } from '$lib/data/types'; import { createUndoManager, redo, subscribeUndoRedoState, undo } from './undo'; +import { LOCAL_UI_ORIGIN, remoteUiOrigin } from '../mutation-origin'; // createUndoManager(doc) is exercised directly against real Y.Doc instances // here — these are the tests that prove the actual CRDT-level guarantees the @@ -26,23 +27,30 @@ const REMOTE_ACTOR: ActorId = { kind: 'human', userId: 'collaborator' }; /** * Applies `sourceDoc`'s current state to `doc` as a single incoming update, - * with a non-null object identity as the transaction origin — standing in - * for the WebsocketProvider instance that origin would be on a real client - * (see undo.ts's module comment). This is the same mechanism a real second - * browser tab's edit — or an MCP agent's edit relayed through the shared - * Y.Doc — arrives by, so it's the right way to simulate "a collaborator's - * intervening work" in a unit test without booting a real server/websocket. + * with the named remote-UI origin used by the WebSocket server. This is the + * same transaction class a collaborator's browser edit arrives as. */ function deliverAsRemoteUpdate(doc: Y.Doc, sourceDoc: Y.Doc): void { const update = Y.encodeStateAsUpdate(sourceDoc, Y.encodeStateVector(doc)); - Y.applyUpdate(doc, update, {}); + Y.applyUpdate(doc, update, remoteUiOrigin('undo-test-peer')); +} + +/** Legacy data helpers intentionally omit an origin; make their test-side + * direct calls model the browser boundary, where every local action is named. */ +function makeTestDoc(): Y.Doc { + const doc = new Y.Doc(); + const transact = doc.transact.bind(doc); + vi.spyOn(doc, 'transact').mockImplementation((fn, origin) => + transact(fn, origin ?? LOCAL_UI_ORIGIN) + ); + return doc; } describe('createUndoManager: scope', () => { let doc: Y.Doc; beforeEach(() => { - doc = new Y.Doc(); + doc = makeTestDoc(); }); afterEach(() => { @@ -177,7 +185,7 @@ describe('createUndoManager: per-actor isolation', () => { let doc: Y.Doc; beforeEach(() => { - doc = new Y.Doc(); + doc = makeTestDoc(); }); afterEach(() => { @@ -328,8 +336,8 @@ describe('undo/redo per-doc manager cache', () => { let docB: Y.Doc; beforeEach(() => { - docA = new Y.Doc(); - docB = new Y.Doc(); + docA = makeTestDoc(); + docB = makeTestDoc(); }); afterEach(() => { diff --git a/src/lib/client/undo.ts b/src/lib/client/undo.ts index 1fde4c2..f6b8087 100644 --- a/src/lib/client/undo.ts +++ b/src/lib/client/undo.ts @@ -1,22 +1,14 @@ import * as Y from 'yjs'; +import { LOCAL_UI_ORIGIN, registerUndoRedoOrigin } from '../mutation-origin.js'; // Local, per-actor undo/redo (#8): each browser tab tracks only its own // locally-originated Yjs transactions, never a collaborator's — including a // stateless MCP agent's writes, which is exactly the "never a collaborator's // intervening work" guarantee the issue asks for. // -// This falls out of Y.UndoManager's own default `trackedOrigins: new -// Set([null])` combined with how every local write already runs through an -// untagged `doc.transact(...)` call (or a single untransacted `.set()`, -// which Yjs auto-wraps the same way) in src/lib/data/records.ts and the -// block editor — Yjs assigns those the `null` origin. Remote edits, whether -// from a collaborator's browser tab or an MCP agent, always arrive over this -// tab's y-websocket connection and are applied by y-protocols' sync handler -// with the WebsocketProvider instance itself as the transaction origin (see -// y-websocket's `readSyncMessage` call), which is never `null` and so is -// never tracked. No extra origin-tagging is needed on either side for that -// separation to hold — it's a byproduct of the existing transact() calls, -// not a new convention call sites need to opt into. +// Local UI writes use LOCAL_UI_ORIGIN. Remote, service, migration, and replay +// transactions have their own named origins and therefore cannot enter this +// tab's undo history. const SCOPE_MAP_NAMES = ['documents', 'collections', 'records'] as const; /** @@ -30,7 +22,14 @@ const SCOPE_MAP_NAMES = ['documents', 'collections', 'records'] as const; * scope types themselves. */ export function createUndoManager(doc: Y.Doc): Y.UndoManager { - return new Y.UndoManager(SCOPE_MAP_NAMES.map((name) => doc.getMap(name))); + const manager = new Y.UndoManager( + SCOPE_MAP_NAMES.map((name) => doc.getMap(name)), + { + trackedOrigins: new Set([LOCAL_UI_ORIGIN]) + } + ); + registerUndoRedoOrigin(manager); + return manager; } // Keyed by Y.Doc instance, not a single module-level singleton: each diff --git a/src/lib/mcp/server.test.ts b/src/lib/mcp/server.test.ts index 5025b3e..74a09a6 100644 --- a/src/lib/mcp/server.test.ts +++ b/src/lib/mcp/server.test.ts @@ -3,7 +3,29 @@ import type { CallToolResult } from '@modelcontextprotocol/sdk/types.js'; import { createMcpServer } from './server'; import { createToken } from './tokens'; import { resolveWorkspaceContext } from '$lib/server/workspace-store'; -import { createCollection, createDocument, createRecord, setPrimaryField } from '$lib/data/records'; +import { + createCollection as rawCreateCollection, + createDocument as rawCreateDocument, + createRecord as rawCreateRecord, + setPrimaryField as rawSetPrimaryField +} from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; + +function createDocument(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCreateDocument(...args)); +} + +function createCollection(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCreateCollection(...args)); +} + +function createRecord(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCreateRecord(...args)); +} + +function setPrimaryField(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawSetPrimaryField(...args)); +} interface ToolHolder { _registeredTools: Record< diff --git a/src/lib/mutation-origin.ts b/src/lib/mutation-origin.ts new file mode 100644 index 0000000..4f4bade --- /dev/null +++ b/src/lib/mutation-origin.ts @@ -0,0 +1,76 @@ +import * as Y from 'yjs'; + +/** Every Yjs write source accepted by Compendium's projection observers. */ +export type MutationSource = + 'local-ui' | 'remote-ui' | 'service' | 'migration' | 'replay' | 'undo-redo' | 'test'; + +export interface MutationOrigin { + readonly source: MutationSource; + readonly detail?: string; +} + +// The E2E browser harness loads the WebSocket server from source and the +// SvelteKit request handler from `build/handler.js`. Those module graphs each +// evaluate this module, but their Yjs transactions meet in the same Node +// process. Keep the identity registry on globalThis so an origin registered +// by one graph remains recognizable by observers in the other. A plain object +// carrying a matching `source` property is still rejected: only objects added +// to this shared WeakMap are accepted. +const ORIGIN_REGISTRY = Symbol.for('compendium.mutation-origin-registry'); + +type GlobalWithOriginRegistry = typeof globalThis & { + [ORIGIN_REGISTRY]?: WeakMap; +}; + +const originRegistry = globalThis as GlobalWithOriginRegistry; +const origins = (originRegistry[ORIGIN_REGISTRY] ??= new WeakMap()); + +function origin(source: MutationSource, detail?: string): MutationOrigin { + const value = Object.freeze({ source, detail }); + origins.set(value, source); + return value; +} + +export const LOCAL_UI_ORIGIN = origin('local-ui'); +export const SERVICE_ORIGIN = origin('service'); +export const MIGRATION_ORIGIN = origin('migration'); +export const REPLAY_ORIGIN = origin('replay'); +export const TEST_ORIGIN = origin('test'); + +/** A connection-specific server-side origin for an update received over y-websocket. */ +export function remoteUiOrigin(connectionId: string): MutationOrigin { + return origin('remote-ui', connectionId); +} + +/** Runs a mutation under one of the named origins; nested Yjs transactions retain it. */ +export function transactWithOrigin( + doc: Y.Doc, + transactionOrigin: MutationOrigin, + mutate: () => T +): T { + let result!: T; + doc.transact(() => { + result = mutate(); + }, transactionOrigin); + return result; +} + +/** Registers a Y.UndoManager's internal transaction origin with the shared classifier. */ +export function registerUndoRedoOrigin(originObject: object): void { + origins.set(originObject, 'undo-redo'); +} + +/** Returns the recognized source for a Yjs transaction origin, if any. */ +export function mutationSource(originValue: unknown): MutationSource | undefined { + return typeof originValue === 'object' && originValue !== null + ? origins.get(originValue) + : undefined; +} + +/** Raised when an observer sees a mutation that bypasses the origin contract. */ +export class UnknownMutationOriginError extends Error { + constructor(originValue: unknown) { + super(`Yjs mutation used an unrecognized origin: ${String(originValue)}`); + this.name = 'UnknownMutationOriginError'; + } +} diff --git a/src/lib/server/audit-observer.test.ts b/src/lib/server/audit-observer.test.ts index dc64621..64d31e1 100644 --- a/src/lib/server/audit-observer.test.ts +++ b/src/lib/server/audit-observer.test.ts @@ -1,7 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import * as Y from 'yjs'; import { - createRecord as crdtCreateRecord, + createRecord as rawCreateRecord, createDocument as crdtCreateDocument, reorderRecord as crdtReorderRecord } from '$lib/data/records'; @@ -13,8 +13,14 @@ import { resetAuditObserverForTests } from './audit-observer'; import type { ActorId } from '$lib/data/types'; +import { remoteUiOrigin, SERVICE_ORIGIN, transactWithOrigin } from '../mutation-origin'; const human: ActorId = { kind: 'human', userId: 'local' }; +const REMOTE_UI_ORIGIN = remoteUiOrigin('audit-observer-test'); + +function crdtCreateRecord(...args: Parameters) { + return transactWithOrigin(args[0], SERVICE_ORIGIN, () => rawCreateRecord(...args)); +} function recentActions(targetRecordId: string): string[] { return queryAuditLog() @@ -38,15 +44,15 @@ describe('audit-observer: generic UI-mutation audit trail', () => { it('logs create_record for a client-origin transaction that adds a top-level record entry', () => { doc.transact(() => { doc.getMap('records').set('r1', new Y.Map()); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions('r1')).toContain('create_record'); }); - it('does not log anything for a null-origin (service-layer-style) transaction', () => { + it('does not log anything for a named service transaction', () => { doc.transact(() => { doc.getMap('records').set('r-service', new Y.Map()); - }); + }, SERVICE_ORIGIN); expect(recentActions('r-service')).toHaveLength(0); }); @@ -54,11 +60,11 @@ describe('audit-observer: generic UI-mutation audit trail', () => { it('logs delete_record for a client-origin transaction that removes a top-level record entry', () => { doc.transact(() => { doc.getMap('records').set('r2', new Y.Map()); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); doc.transact(() => { doc.getMap('records').delete('r2'); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); // queryAuditLog orders newest-first. expect(recentActions('r2')).toEqual(['delete_record', 'create_record']); @@ -67,24 +73,24 @@ describe('audit-observer: generic UI-mutation audit trail', () => { it('logs create_document and delete_document for whole-entry changes on the documents map', () => { doc.transact(() => { doc.getMap('documents').set('d1', new Y.Map()); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions('d1')).toContain('create_document'); doc.transact(() => { doc.getMap('documents').delete('d1'); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions('d1')).toContain('delete_document'); }); it('logs create_collection and delete_collection for whole-entry changes on the collections map', () => { doc.transact(() => { doc.getMap('collections').set('c1', new Y.Map()); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions('c1')).toContain('create_collection'); doc.transact(() => { doc.getMap('collections').delete('c1'); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions('c1')).toContain('delete_collection'); }); @@ -106,11 +112,11 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const yrecord = doc.getMap('records').get(record.id) as Y.Map; const content = yrecord.get('content') as Y.Text; - doc.transact(() => content.insert(0, 'a'), 'fake-ws-connection'); + doc.transact(() => content.insert(0, 'a'), REMOTE_UI_ORIGIN); vi.advanceTimersByTime(1_000); - doc.transact(() => content.insert(1, 'b'), 'fake-ws-connection'); + doc.transact(() => content.insert(1, 'b'), REMOTE_UI_ORIGIN); vi.advanceTimersByTime(1_000); - doc.transact(() => content.insert(2, 'c'), 'fake-ws-connection'); + doc.transact(() => content.insert(2, 'c'), REMOTE_UI_ORIGIN); // Still inside the debounce window from the most recent edit. expect(recentActions(record.id).filter((a) => a === 'update_record')).toHaveLength(0); @@ -131,7 +137,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { doc.transact(() => { content.insert(0, 'hello'); yrecord.set('checked', true); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); vi.advanceTimersByTime(3_000); expect(recentActions(record.id).filter((a) => a === 'update_record')).toHaveLength(1); @@ -146,7 +152,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const yrecord = doc.getMap('records').get(record.id) as Y.Map; const content = yrecord.get('content') as Y.Text; - doc.transact(() => content.insert(0, 'x'), 'fake-ws-connection'); + doc.transact(() => content.insert(0, 'x'), REMOTE_UI_ORIGIN); expect(recentActions(record.id).filter((a) => a === 'update_record')).toHaveLength(0); flushPendingAuditEvents(); @@ -164,11 +170,11 @@ describe('audit-observer: generic UI-mutation audit trail', () => { // Edit, then delete moments later — still well inside the 3s debounce // window, so the update event is still only pending, not yet written. - doc.transact(() => content.insert(0, 'edited just before deletion'), 'fake-ws-connection'); + doc.transact(() => content.insert(0, 'edited just before deletion'), REMOTE_UI_ORIGIN); vi.advanceTimersByTime(500); doc.transact(() => { doc.getMap('records').delete(record.id); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); // Without the fix, the pending update wouldn't surface until the full // debounce window elapses — after the delete already logged — putting @@ -205,12 +211,18 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const yrecordA = doc.getMap('records').get(sharedId) as Y.Map; const yrecordB = docB.getMap('records').get(sharedId) as Y.Map; - doc.transact(() => (yrecordA.get('content') as Y.Text).insert(0, 'from A'), 'ws-a'); + doc.transact( + () => (yrecordA.get('content') as Y.Text).insert(0, 'from A'), + remoteUiOrigin('ws-a') + ); vi.advanceTimersByTime(1_000); // Doc B's edit for the same record id arrives inside doc A's // debounce window. Without per-doc scoping this would reset/steal // the timer keyed by "record:shared-record-id". - docB.transact(() => (yrecordB.get('content') as Y.Text).insert(0, 'from B'), 'ws-b'); + docB.transact( + () => (yrecordB.get('content') as Y.Text).insert(0, 'from B'), + remoteUiOrigin('ws-b') + ); vi.advanceTimersByTime(3_000); // Both docs' pending updates must have fired on their own schedule — @@ -235,7 +247,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const yrecord = doc.getMap('records').get(record.id) as Y.Map; const content = yrecord.get('content') as Y.Text; - doc.transact(() => content.insert(0, 'x'), 'fake-ws-connection'); + doc.transact(() => content.insert(0, 'x'), REMOTE_UI_ORIGIN); expect(recentActions(colonId).filter((a) => a === 'update_record')).toHaveLength(0); flushPendingAuditEvents(); @@ -251,7 +263,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const yrecord = doc.getMap('records').get(record.id) as Y.Map; const content = yrecord.get('content') as Y.Text; - doc.transact(() => content.insert(0, 'x'), 'fake-ws-connection'); + doc.transact(() => content.insert(0, 'x'), REMOTE_UI_ORIGIN); expect(pendingTimerDocCountForTests()).toBe(1); vi.advanceTimersByTime(3_000); @@ -267,7 +279,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const yrecord = doc.getMap('records').get(record.id) as Y.Map; const content = yrecord.get('content') as Y.Text; - doc.transact(() => content.insert(0, 'x'), 'fake-ws-connection'); + doc.transact(() => content.insert(0, 'x'), REMOTE_UI_ORIGIN); expect(pendingTimerDocCountForTests()).toBe(1); flushPendingAuditEvents(); @@ -286,7 +298,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const ids = recordIds.toArray(); recordIds.delete(0, ids.length); recordIds.insert(0, [...ids].reverse()); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); vi.advanceTimersByTime(3_000); expect(recentActions(documentId).filter((a) => a === 'update_document')).toHaveLength(1); @@ -299,7 +311,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { const a = crdtCreateRecord(doc, { parentId: documentId, blockType: 'paragraph' }, human); const b = crdtCreateRecord(doc, { parentId: documentId, blockType: 'paragraph' }, human); - doc.transact(() => crdtReorderRecord(doc, a.id, b.id), 'fake-ws-connection'); + doc.transact(() => crdtReorderRecord(doc, a.id, b.id), REMOTE_UI_ORIGIN); vi.advanceTimersByTime(3_000); expect(recentActions(documentId).filter((act) => act === 'update_document')).toHaveLength(1); @@ -325,9 +337,9 @@ describe('audit-observer: generic UI-mutation audit trail', () => { 'content' ) as Y.Text; - doc.transact(() => contentA.insert(0, 'a'), 'fake-ws-connection'); + doc.transact(() => contentA.insert(0, 'a'), REMOTE_UI_ORIGIN); vi.advanceTimersByTime(1_000); - doc.transact(() => contentB.insert(0, 'b'), 'fake-ws-connection'); + doc.transact(() => contentB.insert(0, 'b'), REMOTE_UI_ORIGIN); // A's timer fires first, alone — B's is still pending for this same // doc, so the doc's own entry in pendingUpdateTimers must survive. @@ -346,7 +358,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { doc.transact(() => { doc.getMap('records').delete(record.id); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions(record.id)).toEqual(['delete_record']); }); @@ -364,7 +376,7 @@ describe('audit-observer: generic UI-mutation audit trail', () => { doc.transact(() => { content.insert(0, 'edited and deleted together'); doc.getMap('records').delete(record.id); - }, 'fake-ws-connection'); + }, REMOTE_UI_ORIGIN); expect(recentActions(record.id)).toEqual(['delete_record']); }); @@ -372,6 +384,8 @@ describe('audit-observer: generic UI-mutation audit trail', () => { /** Creates a bare Document meta entry directly (no service layer, no audit) as a parent for record tests. */ function makeDoc(doc: Y.Doc): string { - const created = crdtCreateDocument(doc, { title: 'Parent' }); + const created = transactWithOrigin(doc, SERVICE_ORIGIN, () => + crdtCreateDocument(doc, { title: 'Parent' }) + ); return created.id; } diff --git a/src/lib/server/audit-observer.ts b/src/lib/server/audit-observer.ts index f4d4eb1..5e85df9 100644 --- a/src/lib/server/audit-observer.ts +++ b/src/lib/server/audit-observer.ts @@ -1,6 +1,7 @@ import * as Y from 'yjs'; import { logAudit } from './audit.js'; import { CURRENT_USER } from './current-user.js'; +import { mutationSource, UnknownMutationOriginError } from '../mutation-origin.js'; // The UI edits the workspace by mutating the client's own Y.Doc directly // (src/lib/data/records.ts, called straight from Svelte components — see @@ -253,7 +254,9 @@ export function attachDocAuditObserver(doc: Y.Doc): void { const maps = topLevelMaps(doc); doc.on('afterTransaction', (transaction: Y.Transaction) => { - if (transaction.origin == null) return; // service-layer write — already audited itself + const source = mutationSource(transaction.origin); + if (!source) throw new UnknownMutationOriginError(transaction.origin); + if (source !== 'local-ui' && source !== 'remote-ui' && source !== 'test') return; const finalized = new Map(); collectTopLevelEntryChanges(transaction, maps, finalized); diff --git a/src/lib/server/catalog-mirror-observer.test.ts b/src/lib/server/catalog-mirror-observer.test.ts index 3f0fa5c..33fda2e 100644 --- a/src/lib/server/catalog-mirror-observer.test.ts +++ b/src/lib/server/catalog-mirror-observer.test.ts @@ -1,4 +1,4 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import * as Y from 'yjs'; import { createDocument as crdtCreateDocument, @@ -19,9 +19,18 @@ import { flushPendingCatalogMirrorEvents, resetCatalogMirrorObserverForTests } from './catalog-mirror-observer'; +import { + LOCAL_UI_ORIGIN, + remoteUiOrigin, + SERVICE_ORIGIN, + transactWithOrigin, + UnknownMutationOriginError +} from '../mutation-origin'; +import { createUndoManager } from '../client/undo'; const WS = 'default'; const SHARD = 'default'; +const REMOTE_UI_ORIGIN = remoteUiOrigin('catalog-mirror-observer-test'); function bootstrap(doc: Y.Doc) { attachCatalogMirrorObserver(WS, doc); @@ -29,7 +38,7 @@ function bootstrap(doc: Y.Doc) { } function seedDocument(doc: Y.Doc, spaceId: string, title = 'Original Title') { - const meta = crdtCreateDocument(doc, { title }); + const meta = transactWithOrigin(doc, SERVICE_ORIGIN, () => crdtCreateDocument(doc, { title })); recordCatalogDocumentCreated({ workspaceId: WS, spaceId, @@ -42,7 +51,9 @@ function seedDocument(doc: Y.Doc, spaceId: string, title = 'Original Title') { } function seedCollection(doc: Y.Doc, spaceId: string, title = 'Original Title') { - const meta = crdtCreateCollection(doc, { title, schema: [] }); + const meta = transactWithOrigin(doc, SERVICE_ORIGIN, () => + crdtCreateCollection(doc, { title, schema: [] }) + ); recordCatalogCollectionCreated({ workspaceId: WS, spaceId, @@ -68,47 +79,37 @@ describe('catalog-mirror-observer: mirroring direct UI title/hierarchy edits int beforeEach(() => { doc = new Y.Doc(); ({ defaultSpaceId: spaceId } = bootstrap(doc)); - vi.useFakeTimers(); }); afterEach(() => { - vi.useRealTimers(); resetCatalogMirrorObserverForTests(); doc.destroy(); }); - it('mirrors a direct (client-origin) Document title edit into the catalog after the debounce window', () => { + it('mirrors a direct remote UI Document title edit into the catalog synchronously', () => { const document = seedDocument(doc, spaceId); doc.transact(() => { crdtUpdateDocumentTitle(doc, document.id, 'Renamed From The UI'); - }, 'fake-ws-connection'); - - expect(catalogDocTitle(document.id)).toBe('Original Title'); - vi.advanceTimersByTime(3_000); + }, REMOTE_UI_ORIGIN); expect(catalogDocTitle(document.id)).toBe('Renamed From The UI'); }); - it('mirrors a direct Collection title edit into the catalog after the debounce window', () => { + it('mirrors a direct remote UI Collection title edit into the catalog synchronously', () => { const collection = seedCollection(doc, spaceId); doc.transact(() => { crdtUpdateCollectionTitle(doc, collection.id, 'Renamed Collection'); - }, 'fake-ws-connection'); - - expect(catalogCollectionTitle(collection.id)).toBe('Original Title'); - vi.advanceTimersByTime(3_000); + }, REMOTE_UI_ORIGIN); expect(catalogCollectionTitle(collection.id)).toBe('Renamed Collection'); }); - it('does not mirror a null-origin (service-layer) write — the service function already dual-writes the catalog itself', () => { + it('does not mirror a named service write — services own their catalog projection', () => { const document = seedDocument(doc, spaceId); doc.transact(() => { crdtUpdateDocumentTitle(doc, document.id, 'Service Layer Rename'); - }); - - vi.advanceTimersByTime(3_000); + }, SERVICE_ORIGIN); // The catalog is never told about this rename by the observer — a real // service-layer caller would have called recordCatalogDocumentTitleChanged // itself, which this test deliberately doesn't do, to prove the observer @@ -122,107 +123,46 @@ describe('catalog-mirror-observer: mirroring direct UI title/hierarchy edits int doc.transact(() => { crdtUpdateDocumentParent(doc, child.id, parent.id); - }, 'fake-ws-connection'); - - vi.advanceTimersByTime(3_000); + }, REMOTE_UI_ORIGIN); const row = listCatalogDocuments(WS).find((d) => d.id === child.id); expect(row?.parentDocumentId).toBe(parent.id); }); - it('coalesces rapid successive edits to the same Document into one catalog write, using its latest value', () => { + it('projects each UI edit immediately, retaining its latest value', () => { const document = seedDocument(doc, spaceId); - doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'First'), 'fake-ws-connection'); - vi.advanceTimersByTime(1_000); - doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'Second'), 'fake-ws-connection'); - vi.advanceTimersByTime(1_000); - doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'Third'), 'fake-ws-connection'); - - // Still inside the debounce window from the most recent edit. - expect(catalogDocTitle(document.id)).toBe('Original Title'); - - vi.advanceTimersByTime(3_000); + doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'First'), REMOTE_UI_ORIGIN); + doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'Second'), REMOTE_UI_ORIGIN); + doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'Third'), REMOTE_UI_ORIGIN); expect(catalogDocTitle(document.id)).toBe('Third'); }); - it('flushPendingCatalogMirrorEvents writes a pending debounced mirror immediately', () => { + it('projects undo and redo of a local UI title change', () => { const document = seedDocument(doc, spaceId); + const undoManager = createUndoManager(doc); + undoManager.stopCapturing(); - doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'Flushed'), 'fake-ws-connection'); - expect(catalogDocTitle(document.id)).toBe('Original Title'); - - flushPendingCatalogMirrorEvents(); - expect(catalogDocTitle(document.id)).toBe('Flushed'); - }); - - it('is a no-op when the entry was deleted before its debounce window elapsed', () => { - const document = seedDocument(doc, spaceId); + transactWithOrigin(doc, LOCAL_UI_ORIGIN, () => + crdtUpdateDocumentTitle(doc, document.id, 'Renamed From The UI') + ); + expect(catalogDocTitle(document.id)).toBe('Renamed From The UI'); - doc.transact(() => crdtUpdateDocumentTitle(doc, document.id, 'Renamed'), 'fake-ws-connection'); - doc.transact(() => { - doc.getMap('documents').delete(document.id); - }, 'fake-ws-connection'); + undoManager.undo(); + expect(catalogDocTitle(document.id)).toBe('Original Title'); - // Must not throw when the pending mirror fires against a now-deleted entry. - expect(() => vi.advanceTimersByTime(3_000)).not.toThrow(); + undoManager.redo(); + expect(catalogDocTitle(document.id)).toBe('Renamed From The UI'); }); - it('is a no-op when a Collection was deleted before its debounce window elapsed', () => { - const collection = seedCollection(doc, spaceId); - - doc.transact( - () => crdtUpdateCollectionTitle(doc, collection.id, 'Renamed'), - 'fake-ws-connection' - ); - doc.transact(() => { - doc.getMap('collections').delete(collection.id); - }, 'fake-ws-connection'); - - expect(() => vi.advanceTimersByTime(3_000)).not.toThrow(); + it('has no pending projection for shutdown to flush', () => { + expect(() => flushPendingCatalogMirrorEvents()).not.toThrow(); }); - it("debounces a same-id entry independently per Y.Doc, so one workspace does not clobber another's pending mirror", () => { - // A document id is only unique within its own workspace (catalog_documents' - // primary key is (workspace_id, id) — see db/schema.ts), so two different - // docs sharing an id, as this test needs to prove per-Y.Doc scoping, must - // belong to two different workspaces, not the same one. - const WS_B = 'other-workspace'; - const docB = new Y.Doc(); - attachCatalogMirrorObserver(WS_B, docB); - const { defaultSpaceId: spaceIdB } = ensureCatalogBootstrapped(WS_B, SHARD, docB); - - try { - const sharedId = 'shared-doc-id'; - crdtCreateDocument(doc, { id: sharedId, title: 'A Original' }); - recordCatalogDocumentCreated({ - workspaceId: WS, - spaceId, - id: sharedId, - title: 'A Original', - order: 'a0', - shardId: SHARD - }); - crdtCreateDocument(docB, { id: sharedId, title: 'B Original' }); - recordCatalogDocumentCreated({ - workspaceId: WS_B, - spaceId: spaceIdB, - id: sharedId, - title: 'B Original', - order: 'a0', - shardId: SHARD - }); - - doc.transact(() => crdtUpdateDocumentTitle(doc, sharedId, 'A Renamed'), 'ws-a'); - vi.advanceTimersByTime(1_000); - docB.transact(() => crdtUpdateDocumentTitle(docB, sharedId, 'B Renamed'), 'ws-b'); - - vi.advanceTimersByTime(3_000); - // Both docs' pending mirrors must have fired on their own schedule — - // neither cleared the other's timer. - expect(catalogDocTitle(sharedId)).toBe('A Renamed'); - expect(listCatalogDocuments(WS_B).find((d) => d.id === sharedId)?.title).toBe('B Renamed'); - } finally { - docB.destroy(); - } + it('rejects an unregistered origin even when it has a recognized source name', () => { + expect(() => + doc.transact(() => doc.getMap('documents').set('bad', new Y.Map()), { + source: 'service' + }) + ).toThrow(UnknownMutationOriginError); }); }); diff --git a/src/lib/server/catalog-mirror-observer.ts b/src/lib/server/catalog-mirror-observer.ts index d7e0848..6bfb4a1 100644 --- a/src/lib/server/catalog-mirror-observer.ts +++ b/src/lib/server/catalog-mirror-observer.ts @@ -5,9 +5,9 @@ import { } from '../data/records.js'; import { recordCatalogCollectionTitleChanged, - recordCatalogDocumentMoved, - recordCatalogDocumentTitleChanged + recordCatalogDocumentMetadataChanged } from './catalog.js'; +import { mutationSource, UnknownMutationOriginError } from '../mutation-origin.js'; // A direct UI mutation (the title input on /doc/[id] or /table/[id], or a // future Sidebar drag-and-drop reorder) writes straight to its own shard's @@ -19,10 +19,8 @@ import { // becomes visible anywhere outside the page that made it — not even after a // refresh, since the catalog itself was never updated. // -// Same origin-based distinction audit-observer.ts already established: -// transaction.origin === null is a service-layer write (services/documents.ts -// and services/collections.ts already dual-write the catalog themselves), so -// this observer only needs to act on origin !== null (direct UI) writes. +// Services have their own authoritative catalog write in the same operation. +// This observer owns the corresponding projection for direct UI transactions. type EntryKind = 'document' | 'collection'; @@ -63,49 +61,11 @@ function resolveOwningEntry( return undefined; } -// Debounced the same way audit-observer.ts debounces content edits — a title -// input fires on every keystroke, and coalescing into one catalog write per -// quiet period avoids one SQLite transaction (plus an outbox/revision bump) -// per character typed. -const UPDATE_DEBOUNCE_MS = 3_000; - -// Keyed by Y.Doc instance for the same reason as audit-observer.ts: more than -// one shard's Y.Doc can be live in one process, and two shards can perfectly -// well contain an entry with the same id. workspaceId travels alongside each -// doc's timer map (not passed into flush separately) so a process-wide -// shutdown flush needs no external per-doc bookkeeping of its own. -interface PendingForDoc { - workspaceId: string; - timers: Map>; -} - -const pendingByDoc = new Map(); - -function pendingFor(workspaceId: string, doc: Y.Doc): PendingForDoc { - let pending = pendingByDoc.get(doc); - if (!pending) { - pending = { workspaceId, timers: new Map() }; - pendingByDoc.set(doc, pending); - } - return pending; -} - -function pruneIfEmpty(doc: Y.Doc, pending: PendingForDoc): void { - if (pending.timers.size === 0) pendingByDoc.delete(doc); -} - -// JSON-encoded, not a template string — same reason as audit-observer.ts's -// timerKey: entry ids are caller-supplied with no format restriction. -function timerKey(kind: EntryKind, id: string): string { - return JSON.stringify([kind, id]); -} - function mirrorNow(workspaceId: string, doc: Y.Doc, kind: EntryKind, id: string): void { if (kind === 'document') { const meta = crdtGetDocument(doc, id); if (!meta) return; // deleted since this was scheduled — nothing to mirror - recordCatalogDocumentTitleChanged(workspaceId, id, meta.title); - recordCatalogDocumentMoved(workspaceId, id, meta.parentDocumentId, meta.order); + recordCatalogDocumentMetadataChanged(workspaceId, id, meta); } else { const meta = crdtGetCollection(doc, id); if (!meta) return; @@ -113,41 +73,17 @@ function mirrorNow(workspaceId: string, doc: Y.Doc, kind: EntryKind, id: string) } } -function scheduleMirror(workspaceId: string, doc: Y.Doc, kind: EntryKind, id: string): void { - const pending = pendingFor(workspaceId, doc); - const key = timerKey(kind, id); - const existing = pending.timers.get(key); - if (existing) clearTimeout(existing); - const timer = setTimeout(() => { - pending.timers.delete(key); - pruneIfEmpty(doc, pending); - mirrorNow(workspaceId, doc, kind, id); - }, UPDATE_DEBOUNCE_MS); - timer.unref?.(); - pending.timers.set(key, timer); -} - -/** Test/shutdown hook: write any debounced mirror events immediately instead of waiting out the window, across every doc with a pending timer. */ +/** + * Kept as a shutdown compatibility hook. Catalog projections are synchronous, + * so there is deliberately no in-memory queue left to flush. + */ export function flushPendingCatalogMirrorEvents(): void { - for (const [doc, pending] of pendingByDoc) { - for (const key of [...pending.timers.keys()]) { - const timer = pending.timers.get(key); - if (!timer) continue; - clearTimeout(timer); - pending.timers.delete(key); - pruneIfEmpty(doc, pending); - const [kind, id] = JSON.parse(key) as [EntryKind, string]; - mirrorNow(pending.workspaceId, doc, kind, id); - } - } + // No-op by design. } -/** Test-only: drop any pending debounce timers without flushing them. */ +/** Test-only compatibility hook. */ export function resetCatalogMirrorObserverForTests(): void { - for (const pending of pendingByDoc.values()) { - for (const timer of pending.timers.values()) clearTimeout(timer); - } - pendingByDoc.clear(); + // No-op by design. } /** @@ -165,19 +101,33 @@ export function attachCatalogMirrorObserver(workspaceId: string, doc: Y.Doc): vo const maps = topLevelMaps(doc); doc.on('afterTransaction', (transaction: Y.Transaction) => { - if (transaction.origin == null) return; // service-layer write — already mirrors itself - - const touched = new Set(); + const source = mutationSource(transaction.origin); + if (!source) throw new UnknownMutationOriginError(transaction.origin); + if ( + source !== 'local-ui' && + source !== 'remote-ui' && + source !== 'undo-redo' && + source !== 'test' + ) + return; + + const touched = new Map(); transaction.changed.forEach((_keys, type) => { if (maps.some((t) => t.map === type)) return; // whole-entry create/delete — not mirrored here const owner = resolveOwningEntry(maps, type); if (!owner) return; - touched.add(timerKey(owner.kind, owner.id)); + touched.set(JSON.stringify([owner.kind, owner.id]), owner); }); - for (const key of touched) { - const [kind, id] = JSON.parse(key) as [EntryKind, string]; - scheduleMirror(workspaceId, doc, kind, id); + for (const { kind, id } of touched.values()) { + try { + mirrorNow(workspaceId, doc, kind, id); + } catch (error) { + // The Yjs mutation is authoritative and must still reach the + // context's update listener, which marks it dirty for the durable + // flush/load reconciliation path. + console.error('Catalog projection deferred until reconciliation', error); + } } }); } diff --git a/src/lib/server/catalog.test.ts b/src/lib/server/catalog.test.ts index f09c468..6cc1df0 100644 --- a/src/lib/server/catalog.test.ts +++ b/src/lib/server/catalog.test.ts @@ -16,6 +16,7 @@ import { import { createSpace, ensureCatalogBootstrapped, + reconcileCatalogMetadata, listCatalogCollections, listCatalogDocuments, recordCatalogDocumentCreated, @@ -53,6 +54,40 @@ describe('catalog: record locator uniqueness (#113 Phase A, §3.1)', () => { }); describe('catalog: bootstrap and backfill', () => { + it('reconciles stale catalog metadata from the authoritative Yjs document idempotently', () => { + const { doc, defaultSpaceId } = bootstrap(); + const document = crdtCreateDocument(doc, { id: 'reconcile-doc', title: 'Authoritative' }); + reserveDocumentLocator(WS, defaultSpaceId, document.id, SHARD); + recordCatalogDocumentCreated({ + workspaceId: WS, + spaceId: defaultSpaceId, + id: document.id, + title: 'Stale', + order: document.order, + shardId: SHARD + }); + + reconcileCatalogMetadata(WS, doc); + expect(listCatalogDocuments(WS).find((meta) => meta.id === document.id)?.title).toBe( + 'Authoritative' + ); + const revisionAfterRepair = + getDb() + .select({ revision: catalogRevisions.revision }) + .from(catalogRevisions) + .where(eq(catalogRevisions.workspaceId, WS)) + .get()?.revision ?? 0; + + reconcileCatalogMetadata(WS, doc); + expect( + getDb() + .select({ revision: catalogRevisions.revision }) + .from(catalogRevisions) + .where(eq(catalogRevisions.workspaceId, WS)) + .get()?.revision + ).toBe(revisionAfterRepair); + }); + it('creates exactly one default Space, idempotently, even across repeated calls', () => { const doc = new Y.Doc(); const first = ensureCatalogBootstrapped(WS, SHARD, doc); diff --git a/src/lib/server/catalog.ts b/src/lib/server/catalog.ts index ffce3f4..9922510 100644 --- a/src/lib/server/catalog.ts +++ b/src/lib/server/catalog.ts @@ -302,6 +302,61 @@ export function recordCatalogDocumentMoved( }); } +/** + * Projects a document's catalog-owned metadata in one SQLite transaction. + * Used for direct Yjs edits, where title and hierarchy can change together. + */ +export function recordCatalogDocumentMetadataChanged( + workspaceId: string, + id: string, + metadata: { title: string; parentDocumentId?: string; order: string } +): void { + getDb().transaction((tx) => { + tx.update(catalogDocuments) + .set({ + title: metadata.title, + parentDocumentId: metadata.parentDocumentId ?? null, + order: metadata.order, + updatedAt: Date.now() + }) + .where(and(eq(catalogDocuments.workspaceId, workspaceId), eq(catalogDocuments.id, id))) + .run(); + bumpRevisionAndAppendOutbox(tx, workspaceId, { documents: [id], op: 'update' }); + }); +} + +/** + * Repairs catalog metadata from the authoritative Yjs document. It performs + * only value-changing writes, making repeated calls safe after a failed + * immediate projection or process restart. + */ +export function reconcileCatalogMetadata(workspaceId: string, doc: Y.Doc): void { + const catalogDocumentsById = new Map( + listCatalogDocuments(workspaceId).map((meta) => [meta.id, meta]) + ); + for (const meta of crdtListDocuments(doc)) { + const catalog = catalogDocumentsById.get(meta.id); + if ( + catalog && + (catalog.title !== meta.title || + catalog.parentDocumentId !== meta.parentDocumentId || + catalog.order !== meta.order) + ) { + recordCatalogDocumentMetadataChanged(workspaceId, meta.id, meta); + } + } + + const catalogCollectionsById = new Map( + listCatalogCollections(workspaceId).map((meta) => [meta.id, meta]) + ); + for (const meta of crdtListCollections(doc)) { + const catalog = catalogCollectionsById.get(meta.id); + if (catalog && catalog.title !== meta.title) { + recordCatalogCollectionTitleChanged(workspaceId, meta.id, meta.title); + } + } +} + /** * Deletes a Document and its descendants from the catalog, mirroring * data/records.ts's recursive deleteDocument. Walks the *catalog's own* diff --git a/src/lib/server/migration.test.ts b/src/lib/server/migration.test.ts index 1a33306..08c9f73 100644 --- a/src/lib/server/migration.test.ts +++ b/src/lib/server/migration.test.ts @@ -1,13 +1,14 @@ import { describe, expect, it } from 'vitest'; import { eq } from 'drizzle-orm'; import { - createCollection as crdtCreateCollection, - createDocument as crdtCreateDocument, - createRecord as crdtCreateRecord, + createCollection as rawCrdtCreateCollection, + createDocument as rawCrdtCreateDocument, + createRecord as rawCrdtCreateRecord, getCollection as crdtGetCollection, getDocument as crdtGetDocument, getRecord as crdtGetRecord } from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { CURRENT_USER } from './current-user'; import { listDocuments } from '../services/documents'; import { listCatalogDocuments } from './catalog'; @@ -19,6 +20,18 @@ import { migrateWorkspace } from './migration'; const WS = 'migration-test-ws'; const actor = CURRENT_USER; +function crdtCreateDocument(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateDocument(...args)); +} + +function crdtCreateCollection(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateCollection(...args)); +} + +function crdtCreateRecord(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateRecord(...args)); +} + describe('migration: lossless content migration (#114/#132, workspace-sharding.md §7)', () => { it('migrates a Document hierarchy, a page-link reference, and a Collection with a relation, preserving every id/order/content field exactly', () => { const { doc } = resolveWorkspaceContext({ workspaceId: WS }); diff --git a/src/lib/server/migration.ts b/src/lib/server/migration.ts index 4fee3d8..ed99871 100644 --- a/src/lib/server/migration.ts +++ b/src/lib/server/migration.ts @@ -21,6 +21,7 @@ import { listDocuments as crdtListDocuments } from '../data/records.js'; import type { CollectionMeta, DocumentMeta } from '../data/types.js'; +import { MIGRATION_ORIGIN, transactWithOrigin } from '../mutation-origin.js'; // Moves every legacy Document/Collection — content still living in the // shared default shard because it predates the catalog/shard system (#113) @@ -140,9 +141,13 @@ function migrateTarget( }); if (target.kind === 'document') { - copyDocumentVerbatim(legacyDoc, shardDoc, target.legacyId); + transactWithOrigin(shardDoc, MIGRATION_ORIGIN, () => + copyDocumentVerbatim(legacyDoc, shardDoc, target.legacyId) + ); } else { - copyCollectionVerbatim(legacyDoc, shardDoc, target.legacyId); + transactWithOrigin(shardDoc, MIGRATION_ORIGIN, () => + copyCollectionVerbatim(legacyDoc, shardDoc, target.legacyId) + ); } const snapshotStore = getSnapshotStore(workspaceId, shardId); diff --git a/src/lib/server/space-isolation.test.ts b/src/lib/server/space-isolation.test.ts index 53041f4..c2849b0 100644 --- a/src/lib/server/space-isolation.test.ts +++ b/src/lib/server/space-isolation.test.ts @@ -1,10 +1,11 @@ import { describe, expect, it } from 'vitest'; import { - createCollection as crdtCreateCollection, - createDocument as crdtCreateDocument, - createRecord as crdtCreateRecord, - updateRecordContent + createCollection as rawCrdtCreateCollection, + createDocument as rawCrdtCreateDocument, + createRecord as rawCrdtCreateRecord, + updateRecordContent as rawUpdateRecordContent } from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { CURRENT_USER } from './current-user'; import { createDocument, @@ -29,6 +30,22 @@ import type { AccessToken } from '$lib/mcp/tokens'; const actor = CURRENT_USER; +function crdtCreateDocument(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateDocument(...args)); +} + +function crdtCreateCollection(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateCollection(...args)); +} + +function crdtCreateRecord(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateRecord(...args)); +} + +function updateRecordContent(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawUpdateRecordContent(...args)); +} + // Two-Space fixture (#114/#133, workspace-sharding.md §9): Space A content // goes through the real service layer (which always writes to // resolveWorkspaceContext()'s defaultSpaceId — Space A here), Space B diff --git a/src/lib/server/workspace-store.test.ts b/src/lib/server/workspace-store.test.ts index 47bfbcc..09f0008 100644 --- a/src/lib/server/workspace-store.test.ts +++ b/src/lib/server/workspace-store.test.ts @@ -1,9 +1,11 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { TEST_ORIGIN, transactWithOrigin } from '../mutation-origin'; import { closeDb } from './store'; import { clientIdForToken, requestAgentHold } from './holds'; +import * as catalog from './catalog'; import { flush, registerConnection, @@ -23,6 +25,8 @@ describe('workspace-store: snapshot persistence survives a process restart', () afterEach(() => { resetWorkspaceStoreForTests(); + vi.useRealTimers(); + vi.restoreAllMocks(); closeDb(); delete process.env.DATABASE_URL; rmSync(dir, { recursive: true, force: true }); @@ -30,7 +34,9 @@ describe('workspace-store: snapshot persistence survives a process restart', () it('reloads the last saved state after the in-memory doc is dropped ("restart")', () => { const { doc: doc1 } = resolveWorkspaceContext(); - doc1.getMap('workspace').set('greeting', 'hello from before restart'); + transactWithOrigin(doc1, TEST_ORIGIN, () => + doc1.getMap('workspace').set('greeting', 'hello from before restart') + ); flush(); // Simulate a process restart: drop the registry and DB handle, force a @@ -47,6 +53,28 @@ describe('workspace-store: snapshot persistence survives a process restart', () flush(); // nothing written yet -> dirty is false expect(() => flush()).not.toThrow(); }); + + it('logs a transient periodic reconciliation failure and retries it on the next interval', () => { + vi.useFakeTimers(); + const context = resolveWorkspaceContext(); + const reconcile = vi.spyOn(catalog, 'reconcileCatalogMetadata').mockImplementationOnce(() => { + throw new Error('temporary catalog outage'); + }); + const error = vi.spyOn(console, 'error').mockImplementation(() => {}); + + transactWithOrigin(context.doc, TEST_ORIGIN, () => + context.doc.getMap('workspace').set('needs-reconciliation', true) + ); + + vi.advanceTimersByTime(30_000); + expect(error).toHaveBeenCalledWith( + 'Failed to flush workspace context; will retry', + expect.any(Error) + ); + + vi.advanceTimersByTime(30_000); + expect(reconcile).toHaveBeenCalledTimes(2); + }); }); describe('workspace-store: isolation between independently-resolved contexts', () => { @@ -74,8 +102,8 @@ describe('workspace-store: isolation between independently-resolved contexts', ( const a = resolveWorkspaceContext({ workspaceId: 'space-a', shardId: 'main' }); const b = resolveWorkspaceContext({ workspaceId: 'space-b', shardId: 'main' }); - a.doc.getMap('workspace').set('title', 'Space A'); - b.doc.getMap('workspace').set('title', 'Space B'); + transactWithOrigin(a.doc, TEST_ORIGIN, () => a.doc.getMap('workspace').set('title', 'Space A')); + transactWithOrigin(b.doc, TEST_ORIGIN, () => b.doc.getMap('workspace').set('title', 'Space B')); expect(a.doc.getMap('workspace').get('title')).toBe('Space A'); expect(b.doc.getMap('workspace').get('title')).toBe('Space B'); @@ -90,8 +118,8 @@ describe('workspace-store: isolation between independently-resolved contexts', ( const a = resolveWorkspaceContext({ workspaceId: 'space::main', shardId: 'primary' }); const b = resolveWorkspaceContext({ workspaceId: 'space', shardId: 'main::primary' }); - a.doc.getMap('workspace').set('title', 'A'); - b.doc.getMap('workspace').set('title', 'B'); + transactWithOrigin(a.doc, TEST_ORIGIN, () => a.doc.getMap('workspace').set('title', 'A')); + transactWithOrigin(b.doc, TEST_ORIGIN, () => b.doc.getMap('workspace').set('title', 'B')); expect(a.doc).not.toBe(b.doc); expect(a.doc.getMap('workspace').get('title')).toBe('A'); @@ -116,8 +144,8 @@ describe('workspace-store: isolation between independently-resolved contexts', ( const a = resolveWorkspaceContext({ workspaceId: 'space-a', shardId: 'main' }); const b = resolveWorkspaceContext({ workspaceId: 'space-b', shardId: 'main' }); - a.doc.getMap('workspace').set('title', 'Space A'); - b.doc.getMap('workspace').set('title', 'Space B'); + transactWithOrigin(a.doc, TEST_ORIGIN, () => a.doc.getMap('workspace').set('title', 'Space A')); + transactWithOrigin(b.doc, TEST_ORIGIN, () => b.doc.getMap('workspace').set('title', 'Space B')); flush(); resetWorkspaceStoreForTests(); @@ -209,7 +237,9 @@ describe('workspace-store: isolation between independently-resolved contexts', ( it('survives an idle-unload-then-reload cycle with content intact', () => { const a = resolveWorkspaceContext({ workspaceId: 'space-a', shardId: 'main' }); - a.doc.getMap('workspace').set('title', 'Space A content'); + transactWithOrigin(a.doc, TEST_ORIGIN, () => + a.doc.getMap('workspace').set('title', 'Space A content') + ); const released = releaseContextIfIdle('space-a', 'main'); expect(released).toBe(true); diff --git a/src/lib/server/workspace-store.ts b/src/lib/server/workspace-store.ts index 95d336c..e838266 100644 --- a/src/lib/server/workspace-store.ts +++ b/src/lib/server/workspace-store.ts @@ -12,8 +12,9 @@ import { resetCatalogMirrorObserverForTests } from './catalog-mirror-observer.js'; import { aggregateHolds, initHoldEviction, resetHoldEvictionForTests } from './holds.js'; -import { ensureCatalogBootstrapped } from './catalog.js'; +import { ensureCatalogBootstrapped, reconcileCatalogMetadata } from './catalog.js'; import { getInstanceWorkspaceId } from './instance.js'; +import { REPLAY_ORIGIN } from '../mutation-origin.js'; // This is the one place a {workspaceId, shardId} selector resolves to a live // Y.Doc/Awareness/persistence/connection bundle. Every boundary that used to @@ -98,13 +99,14 @@ function createContext(workspaceId: string, shardId: string): InternalContext { const snapshotStore = getSnapshotStore(workspaceId, shardId); const snapshot = snapshotStore.loadLatest(); if (snapshot) { - Y.applyUpdate(doc, snapshot); + Y.applyUpdate(doc, snapshot, REPLAY_ORIGIN); } // Backfills/bootstraps the catalog from this doc's current content the // first time this {workspaceId, shardId} resolves — see catalog.ts. Runs // after the snapshot load (so it sees real content) and before the audit // observer attaches (so it never produces a spurious audit trail). const { defaultSpaceId } = ensureCatalogBootstrapped(workspaceId, shardId, doc); + reconcileCatalogMetadata(workspaceId, doc); attachDocAuditObserver(doc); attachCatalogMirrorObserver(workspaceId, doc); @@ -126,7 +128,15 @@ function createContext(workspaceId: string, shardId: string): InternalContext { context.dirty = true; }); - context.saveTimer = setInterval(() => flushContext(context, snapshotStore), SAVE_INTERVAL_MS); + context.saveTimer = setInterval(() => { + try { + flushContext(context, snapshotStore); + } catch (error) { + // Keep dirty state intact: the next interval retries reconciliation + // from the already-persisted authoritative Yjs snapshot. + console.error('Failed to flush workspace context; will retry', error); + } + }, SAVE_INTERVAL_MS); context.saveTimer.unref?.(); wireShutdownOnce(); @@ -159,6 +169,10 @@ function flushContext( ): void { if (!context.dirty) return; snapshotStore.save(Y.encodeStateAsUpdate(context.doc)); + // A direct Yjs update can commit while SQLite is temporarily unavailable. + // Keep this context dirty until reconciliation succeeds so the next flush + // retries from the authoritative snapshot without duplicating catalog rows. + reconcileCatalogMetadata(context.workspaceId, context.doc); context.dirty = false; } diff --git a/src/lib/server/yjs-ws-server.test.ts b/src/lib/server/yjs-ws-server.test.ts index 99d4231..ac15447 100644 --- a/src/lib/server/yjs-ws-server.test.ts +++ b/src/lib/server/yjs-ws-server.test.ts @@ -6,6 +6,7 @@ import type { WebSocket } from 'ws'; import { setupWSConnection } from './yjs-ws-server'; import { resolveWorkspaceContext, resetWorkspaceStoreForTests } from './workspace-store'; import { resetHoldsForTests } from './holds'; +import { TEST_ORIGIN, transactWithOrigin } from '../mutation-origin'; // Minimal stand-in for the `ws` library's WebSocket, just enough surface for // setupWSConnection: EventEmitter for on/emit, plus the properties/methods it @@ -84,7 +85,8 @@ describe('yjs-ws-server: disconnect cleanup', () => { ws.sendImpl = sendSpy; setupWSConnection(ws as unknown as WebSocket); - resolveWorkspaceContext().doc.getMap('workspace').set('key', 'value'); // triggers doc's 'update' event + const { doc } = resolveWorkspaceContext(); + transactWithOrigin(doc, TEST_ORIGIN, () => doc.getMap('workspace').set('key', 'value')); expect(sendSpy).not.toHaveBeenCalled(); }); @@ -96,8 +98,9 @@ describe('yjs-ws-server: disconnect cleanup', () => { }; setupWSConnection(ws as unknown as WebSocket); + const { doc } = resolveWorkspaceContext(); expect(() => - resolveWorkspaceContext().doc.getMap('workspace').set('key2', 'value2') + transactWithOrigin(doc, TEST_ORIGIN, () => doc.getMap('workspace').set('key2', 'value2')) ).not.toThrow(); }); diff --git a/src/lib/server/yjs-ws-server.ts b/src/lib/server/yjs-ws-server.ts index 96ca624..db6e58e 100644 --- a/src/lib/server/yjs-ws-server.ts +++ b/src/lib/server/yjs-ws-server.ts @@ -11,6 +11,7 @@ import { } from './workspace-store.js'; import { getInstanceWorkspaceId } from './instance.js'; import { isKnownShard } from './catalog.js'; +import { remoteUiOrigin } from '../mutation-origin.js'; // y-websocket's npm package ships the browser client only as of v3 — the // server side (formerly bin/utils.js) is reimplemented here against the same @@ -69,6 +70,7 @@ export function setupWSConnection(ws: WebSocket, selector?: WorkspaceSelector): ws.binaryType = 'arraybuffer'; const ownedClientIds = new Set(); + const transactionOrigin = remoteUiOrigin(`ws:${crypto.randomUUID()}`); let closed = false; let pongReceived = true; @@ -144,7 +146,7 @@ export function setupWSConnection(ws: WebSocket, selector?: WorkspaceSelector): case MESSAGE_SYNC: { const encoder = encoding.createEncoder(); encoding.writeVarUint(encoder, MESSAGE_SYNC); - syncProtocol.readSyncMessage(decoder, encoder, doc, ws); + syncProtocol.readSyncMessage(decoder, encoder, doc, transactionOrigin); if (encoding.length(encoder) > 1) send(encoder); break; } diff --git a/src/lib/services/collections.ts b/src/lib/services/collections.ts index 8cfeba2..3a3274e 100644 --- a/src/lib/services/collections.ts +++ b/src/lib/services/collections.ts @@ -23,6 +23,7 @@ import { import { grantCollectionAccess, tokenAllowsParent } from '$lib/server/token-store'; import type { CollectionMeta, PropertyDefinition, WorkspaceRecord } from '$lib/data/types'; import { nanoid } from 'nanoid'; +import { SERVICE_ORIGIN, transactWithOrigin } from '../mutation-origin.js'; import { actorForCaller, isAccessToken, @@ -84,11 +85,13 @@ export function createCollection( } reserveCollectionLocator(workspaceId, targetSpaceId, id, shardId); - const collection = crdtCreateCollection(doc, { - id, - title: input.title, - schema: input.schema ?? [] - }); + const collection = transactWithOrigin(doc, SERVICE_ORIGIN, () => + crdtCreateCollection(doc, { + id, + title: input.title, + schema: input.schema ?? [] + }) + ); recordCatalogCollectionCreated({ workspaceId, @@ -184,7 +187,7 @@ export function deleteCollection(caller: CallerIdentity, collectionId: string): const actor = actorForCaller(caller); requireAccessibleParent(caller, collectionId, 'delete_collection'); - crdtDeleteCollection(doc, collectionId); + transactWithOrigin(doc, SERVICE_ORIGIN, () => crdtDeleteCollection(doc, collectionId)); recordCatalogCollectionDeleted(workspaceId, collectionId); logAudit({ actor, action: 'delete_collection', targetRecordId: collectionId }); } @@ -199,7 +202,9 @@ export function updateCollectionTitle( const actor = actorForCaller(caller); requireAccessibleParent(caller, collectionId, 'update_collection_title'); - crdtUpdateCollectionTitle(doc, collectionId, title); + transactWithOrigin(doc, SERVICE_ORIGIN, () => + crdtUpdateCollectionTitle(doc, collectionId, title) + ); recordCatalogCollectionTitleChanged(workspaceId, collectionId, title); logAudit({ actor, diff --git a/src/lib/services/documents.ts b/src/lib/services/documents.ts index 3df51f5..8afdcf0 100644 --- a/src/lib/services/documents.ts +++ b/src/lib/services/documents.ts @@ -39,6 +39,7 @@ import type { } from '$lib/data/types'; import type * as Y from 'yjs'; import { nanoid } from 'nanoid'; +import { SERVICE_ORIGIN, transactWithOrigin } from '../mutation-origin.js'; import { actorForCaller, isAccessToken, @@ -159,11 +160,17 @@ export function createDocument(caller: CallerIdentity, input: CreateDocumentInpu ); const order = computeSiblingOrder(siblings, input.afterDocumentId); - const document = crdtCreateDocument(doc, { - id, - title: input.title, - parentDocumentId: input.parentDocumentId, - order + const document = transactWithOrigin(doc, SERVICE_ORIGIN, () => { + const created = crdtCreateDocument(doc, { + id, + title: input.title, + parentDocumentId: input.parentDocumentId, + order + }); + if (input.createInitialBlock) { + crdtCreateRecord(doc, { parentId: created.id, blockType: 'paragraph' }, actor); + } + return created; }); recordCatalogDocumentCreated({ @@ -176,10 +183,6 @@ export function createDocument(caller: CallerIdentity, input: CreateDocumentInpu shardId }); - if (input.createInitialBlock) { - crdtCreateRecord(doc, { parentId: document.id, blockType: 'paragraph' }, actor); - } - // Persist access grant in SQLite so subsequent tool calls from this token succeed if (isAccessToken(caller)) { grantDocumentAccess(caller.tokenHash, document.id); @@ -243,12 +246,14 @@ export function moveDocument( ); const order = computeSiblingOrder(siblings, options.afterDocumentId); - crdtUpdateDocumentParent( - doc, - documentId, - options.parentDocumentId, - options.afterDocumentId, - order + transactWithOrigin(doc, SERVICE_ORIGIN, () => + crdtUpdateDocumentParent( + doc, + documentId, + options.parentDocumentId, + options.afterDocumentId, + order + ) ); recordCatalogDocumentMoved(workspaceId, documentId, options.parentDocumentId, order); logAudit({ @@ -295,7 +300,7 @@ export function deleteDocument(caller: CallerIdentity, documentId: string): void const descendantIds = collectDescendantIds(listCatalogDocuments(workspaceId), documentId); for (const id of descendantIds) { const { doc } = resolveParentWorkspaceContext(id); - crdtDeleteDocument(doc, id); + transactWithOrigin(doc, SERVICE_ORIGIN, () => crdtDeleteDocument(doc, id)); } // One call cascades the whole subtree — recordCatalogDocumentDeleted @@ -314,7 +319,7 @@ export function updateDocumentTitle( const actor = actorForCaller(caller); requireAccessibleParent(caller, documentId, 'update_document_title'); - crdtUpdateDocumentTitle(doc, documentId, title); + transactWithOrigin(doc, SERVICE_ORIGIN, () => crdtUpdateDocumentTitle(doc, documentId, title)); recordCatalogDocumentTitleChanged(workspaceId, documentId, title); logAudit({ actor, diff --git a/src/lib/services/records.ts b/src/lib/services/records.ts index bc084d9..4525f7e 100644 --- a/src/lib/services/records.ts +++ b/src/lib/services/records.ts @@ -1,4 +1,5 @@ import { nanoid } from 'nanoid'; +import { SERVICE_ORIGIN, transactWithOrigin } from '../mutation-origin.js'; import type * as Y from 'yjs'; import type { Awareness } from 'y-protocols/awareness'; import { clientIdForToken, isHeldByClient, releaseAgentHold } from '$lib/server/holds'; @@ -222,19 +223,21 @@ export function createRecord( let record: WorkspaceRecord; try { - record = crdtCreateRecord( - doc, - { - id, - parentId: input.parentId, - afterRecordId: input.afterRecordId, - blockType: input.blockType, - properties: input.properties, - referencedRecordId: input.referencedRecordId, - viewConfig: input.viewConfig, - childPagesDepth: input.childPagesDepth - }, - actor + record = transactWithOrigin(doc, SERVICE_ORIGIN, () => + crdtCreateRecord( + doc, + { + id, + parentId: input.parentId, + afterRecordId: input.afterRecordId, + blockType: input.blockType, + properties: input.properties, + referencedRecordId: input.referencedRecordId, + viewConfig: input.viewConfig, + childPagesDepth: input.childPagesDepth + }, + actor + ) ); } catch (err) { releaseRecordLocator(workspaceId, id); @@ -402,27 +405,26 @@ export function writeRecord( validateViewConfig(record.blockType, input.viewConfig); } - if (input.markdown !== undefined) { - writeRecordMarkdown(caller, doc, awareness, recordId, actor, input.markdown); - } - - if (input.properties) { - updateRecordProperties(doc, recordId, input.properties, actor); - logAudit({ - actor, - action: 'write_record', - targetRecordId: recordId, - diff: { properties: input.properties } - }); - } - - if (input.referencedRecordId !== undefined) { - applyReferencedRecordIdWrite(doc, record, recordId, actor, input.referencedRecordId); - } - - if (input.viewConfig !== undefined) { - applyViewConfigWrite(doc, record, recordId, actor, input.viewConfig); - } + transactWithOrigin(doc, SERVICE_ORIGIN, () => { + if (input.markdown !== undefined) { + writeRecordMarkdown(caller, doc, awareness, recordId, actor, input.markdown); + } + if (input.properties) { + updateRecordProperties(doc, recordId, input.properties, actor); + logAudit({ + actor, + action: 'write_record', + targetRecordId: recordId, + diff: { properties: input.properties } + }); + } + if (input.referencedRecordId !== undefined) { + applyReferencedRecordIdWrite(doc, record, recordId, actor, input.referencedRecordId); + } + if (input.viewConfig !== undefined) { + applyViewConfigWrite(doc, record, recordId, actor, input.viewConfig); + } + }); } /** @@ -436,7 +438,7 @@ export function deleteRecord(caller: CallerIdentity, recordId: string): void { const actor = actorForCaller(caller); requireAccessibleRecord(caller, recordId, 'delete_record'); - crdtDeleteRecord(doc, recordId); + transactWithOrigin(doc, SERVICE_ORIGIN, () => crdtDeleteRecord(doc, recordId)); // The CRDT delete has already committed at this point — a release failure // here must not throw back to the caller as if deletion itself failed. Log // it instead: a stale locator row for a since-deleted record fails safe diff --git a/src/lib/services/search.test.ts b/src/lib/services/search.test.ts index 42482dd..059c985 100644 --- a/src/lib/services/search.test.ts +++ b/src/lib/services/search.test.ts @@ -7,16 +7,29 @@ import { writeRecord } from './index'; import { - createCollection as crdtCreateCollection, - createDocument as crdtCreateDocument, - createRecord as crdtCreateRecord + createCollection as rawCrdtCreateCollection, + createDocument as rawCrdtCreateDocument, + createRecord as rawCrdtCreateRecord } from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { createToken } from '$lib/mcp/tokens'; import { resolveWorkspaceContext } from '$lib/server/workspace-store'; import type { ActorId } from '$lib/data/types'; const human: ActorId = { kind: 'human', userId: 'brylie' }; +function crdtCreateDocument(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateDocument(...args)); +} + +function crdtCreateCollection(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateCollection(...args)); +} + +function crdtCreateRecord(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateRecord(...args)); +} + describe('searchWorkspace: snippet boundaries', () => { it('omits the leading ellipsis when the match is at the very start of the text', () => { const doc = createDocument(human, { title: 'Snippet Doc', createInitialBlock: true }); diff --git a/src/lib/services/services.test.ts b/src/lib/services/services.test.ts index 95b20ee..143bc27 100644 --- a/src/lib/services/services.test.ts +++ b/src/lib/services/services.test.ts @@ -26,14 +26,15 @@ import { createToken, verifyToken } from '$lib/mcp/tokens'; import { queryAuditLog } from '$lib/server/audit'; import { resolveWorkspaceContext } from '$lib/server/workspace-store'; import { - createRecord as crdtCreateRecord, - createDocument as crdtCreateDocument, - createCollection as crdtCreateCollection, + createRecord as rawCrdtCreateRecord, + createDocument as rawCrdtCreateDocument, + createCollection as rawCrdtCreateCollection, getDocument as crdtGetDocument, getCollection as crdtGetCollection, getRecord as crdtGetRecord, getRecordYText as crdtGetRecordYText } from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { listCatalogCollections, listCatalogDocuments, @@ -47,6 +48,22 @@ import type { ActorId, EmbeddedViewConfig } from '$lib/data/types'; const human: ActorId = { kind: 'human', userId: 'brylie' }; +function crdtCreateDocument(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateDocument(...args)); +} + +function crdtCreateCollection(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateCollection(...args)); +} + +function crdtCreateRecord(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCrdtCreateRecord(...args)); +} + +function writeTestYText(doc: Parameters[0], write: () => void): void { + transactWithOrigin(doc, TEST_ORIGIN, write); +} + describe('service layer: centralized business rules & side effects', () => { it('creates documents, logs audit, and persists token grants in SQLite', () => { // 1. Create document as human @@ -401,7 +418,9 @@ describe('service layer: centralized business rules & side effects', () => { }, human ); - crdtGetRecordYText(doc, callout.id)!.insert(0, 'Handle with care.'); + writeTestYText(doc, () => + crdtGetRecordYText(doc, callout.id)!.insert(0, 'Handle with care.') + ); const result = getDocument(human, docPublic.id); const record = result?.records.find((r) => r.id === callout.id); @@ -421,7 +440,9 @@ describe('service layer: centralized business rules & side effects', () => { }, human ); - crdtGetRecordYText(doc, callout.id)!.insert(0, 'First line\nSecond line'); + writeTestYText(doc, () => + crdtGetRecordYText(doc, callout.id)!.insert(0, 'First line\nSecond line') + ); const result = getDocument(human, docPublic.id); const record = result?.records.find((r) => r.id === callout.id); @@ -457,7 +478,9 @@ describe('service layer: centralized business rules & side effects', () => { }, human ); - crdtGetRecordYText(doc, callout.id)!.insert(0, 'Plain text, styled cell only.'); + writeTestYText(doc, () => + crdtGetRecordYText(doc, callout.id)!.insert(0, 'Plain text, styled cell only.') + ); const result = getDocument(human, docPublic.id); const record = result?.records.find((r) => r.id === callout.id); @@ -473,7 +496,7 @@ describe('service layer: centralized business rules & side effects', () => { { parentId: docPublic.id, blockType: 'callout' }, human ); - crdtGetRecordYText(doc, callout.id)!.insert(0, 'Careful!'); + writeTestYText(doc, () => crdtGetRecordYText(doc, callout.id)!.insert(0, 'Careful!')); const result = getDocument(human, docPublic.id); const record = result?.records.find((r) => r.id === callout.id); diff --git a/src/routes/mcp/server.test.ts b/src/routes/mcp/server.test.ts index 0d93d4f..276d63f 100644 --- a/src/routes/mcp/server.test.ts +++ b/src/routes/mcp/server.test.ts @@ -4,11 +4,19 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { Client } from '@modelcontextprotocol/sdk/client/index.js'; import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; import { GET, POST, DELETE } from './+server'; -import { createDocument } from '$lib/data/records'; +import { createDocument as createDocumentRaw } from '$lib/data/records'; import { createToken } from '$lib/mcp/tokens'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { resolveWorkspaceContext } from '$lib/server/workspace-store'; import { closeTestServer, listenOnLoopback } from '../../../tests/e2e/listener'; +function createDocument( + doc: Parameters[0], + input: Parameters[1] +) { + return transactWithOrigin(doc, TEST_ORIGIN, () => createDocumentRaw(doc, input)); +} + // Mirrors tests/e2e/harness.ts's node-request/web-request bridge, but points // at this route's own exported handlers rather than reimplementing MCP // server wiring, so this test exercises extractBearerToken/handle() for real. diff --git a/src/routes/space/[spaceId]/doc/[id]/+page.svelte b/src/routes/space/[spaceId]/doc/[id]/+page.svelte index 27263fe..a185ebe 100644 --- a/src/routes/space/[spaceId]/doc/[id]/+page.svelte +++ b/src/routes/space/[spaceId]/doc/[id]/+page.svelte @@ -5,6 +5,7 @@ import { resolve } from '$app/paths'; import { getShardAwareness, getShardDoc } from '$lib/client/yjs-client'; import { CURRENT_USER } from '$lib/client/actor'; + import { LOCAL_UI_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { createRecord, deleteRecord, @@ -448,10 +449,13 @@ blockType: BlockType = 'paragraph' ): Promise { if (!ydoc) return; - const record = createRecord( - ydoc, - { parentId: data.documentId, blockType, afterRecordId: afterId }, - CURRENT_USER + const currentDoc = ydoc; + const record = transactWithOrigin(ydoc, LOCAL_UI_ORIGIN, () => + createRecord( + currentDoc, + { parentId: data.documentId, blockType, afterRecordId: afterId }, + CURRENT_USER + ) ); await tick(); blockRefs[record.id]?.focusEditor(true); @@ -504,6 +508,7 @@ nextBlockType: BlockType ): Promise { if (!ydoc) return; + const currentDoc = ydoc; const ytext = getRecordYText(ydoc, block.id); const richText = ytext ? yTextToRichText(ytext) : { runs: [] }; const offset = ytext ? Math.min(Math.max(0, caretOffset), ytext.length) : 0; @@ -512,18 +517,24 @@ if (ytext && offset < ytext.length) { const doc = ytext.doc; const trim = () => ytext.delete(offset, ytext.length - offset); - if (doc) doc.transact(trim); + if (doc) doc.transact(trim, LOCAL_UI_ORIGIN); else trim(); } - const record = createRecord( - ydoc, - { parentId: data.documentId, blockType: nextBlockType, afterRecordId: block.id }, - CURRENT_USER + const record = transactWithOrigin(ydoc, LOCAL_UI_ORIGIN, () => + createRecord( + currentDoc, + { parentId: data.documentId, blockType: nextBlockType, afterRecordId: block.id }, + CURRENT_USER + ) ); if (after.runs.length > 0) { const newYtext = getRecordYText(ydoc, record.id); - if (newYtext) applyRichTextToYText(newYtext, after); + if (newYtext) { + transactWithOrigin(currentDoc, LOCAL_UI_ORIGIN, () => + applyRichTextToYText(newYtext, after) + ); + } } await tick(); // Caret at the very start (offset 0): `block` becomes the empty line @@ -809,7 +820,7 @@ if (ytext) { const doc = ytext.doc; const clear = () => ytext.delete(0, ytext.length); - if (doc) doc.transact(clear); + if (doc) doc.transact(clear, LOCAL_UI_ORIGIN); else clear(); } setBlockType(ydoc, blockId, blockType, CURRENT_USER); diff --git a/src/routes/space/[spaceId]/doc/[id]/BlockEditor.svelte b/src/routes/space/[spaceId]/doc/[id]/BlockEditor.svelte index 7aa061b..256e616 100644 --- a/src/routes/space/[spaceId]/doc/[id]/BlockEditor.svelte +++ b/src/routes/space/[spaceId]/doc/[id]/BlockEditor.svelte @@ -3,6 +3,7 @@ import { page } from '$app/state'; import { resolve } from '$app/paths'; import { diffPlainText } from '$lib/client/text-diff'; + import { LOCAL_UI_ORIGIN } from '$lib/mutation-origin'; import { getCaretClientX, getCaretOffset, @@ -181,7 +182,7 @@ if (diff.deleteCount > 0) ytext.delete(diff.start, diff.deleteCount); if (diff.insertText) ytext.insert(diff.start, diff.insertText); }; - if (doc) doc.transact(apply); + if (doc) doc.transact(apply, LOCAL_UI_ORIGIN); else apply(); lastPlainText = newText; onInputText(); @@ -354,7 +355,7 @@ const doc = ytext.doc; const apply = () => ytext.format(offsets.start, offsets.end - offsets.start, { [mark]: nextValue }); - if (doc) doc.transact(apply); + if (doc) doc.transact(apply, LOCAL_UI_ORIGIN); else apply(); } diff --git a/src/routes/space/[spaceId]/doc/[id]/editing-conventions.svelte.test.ts b/src/routes/space/[spaceId]/doc/[id]/editing-conventions.svelte.test.ts index d0a220b..8d3a458 100644 --- a/src/routes/space/[spaceId]/doc/[id]/editing-conventions.svelte.test.ts +++ b/src/routes/space/[spaceId]/doc/[id]/editing-conventions.svelte.test.ts @@ -16,6 +16,7 @@ import * as Y from 'yjs'; import { createDocument, createRecord, getDocument, getRecordYText } from '$lib/data/records'; import { plainText, yTextToRichText } from '$lib/data/richtext'; import type { ActorId } from '$lib/data/types'; +import { LOCAL_UI_ORIGIN } from '$lib/mutation-origin'; import Page from './+page.svelte'; vi.mock('$app/state', () => ({ @@ -311,6 +312,8 @@ describe('editing conventions: Enter, Backspace, and toolbar block controls', () const record = createRecord(ydoc, { parentId: 'doc-1', blockType: 'paragraph' }, HUMAN); getRecordYText(ydoc, record.id)!.insert(0, 'Hello world'); const { container } = await renderDoc(); + const splitOrigins: unknown[] = []; + ydoc.on('afterTransaction', (transaction) => splitOrigins.push(transaction.origin)); const editor = container.querySelector('[contenteditable]') as HTMLElement; selectRange(editor, 5, 5); // caret right after "Hello" @@ -323,6 +326,8 @@ describe('editing conventions: Enter, Backspace, and toolbar block controls', () expect(recordIds).toHaveLength(2); expect(textOf(record.id)).toBe('Hello'); expect(textOf(recordIds[1])).toBe(' world'); + expect(splitOrigins).toContain(LOCAL_UI_ORIGIN); + expect(splitOrigins).not.toContain(null); }); it.each(['quote', 'callout', 'code'] as const)( diff --git a/src/routes/space/[spaceId]/doc/[id]/page.server.test.ts b/src/routes/space/[spaceId]/doc/[id]/page.server.test.ts index f2cb5b5..899b616 100644 --- a/src/routes/space/[spaceId]/doc/[id]/page.server.test.ts +++ b/src/routes/space/[spaceId]/doc/[id]/page.server.test.ts @@ -1,11 +1,16 @@ import { describe, expect, it } from 'vitest'; import { load } from './+page.server'; -import { createDocument } from '$lib/data/records'; +import { createDocument as rawCreateDocument } from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { createDocument as createDocumentService } from '$lib/services'; import { CURRENT_USER } from '$lib/server/current-user'; import { resolveWorkspaceContext } from '$lib/server/workspace-store'; import { resolveRequestContext } from '$lib/server/request-context'; +function createDocument(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCreateDocument(...args)); +} + function loadEvent(id: string, spaceId?: string): Parameters[0] { return { params: { id, spaceId: spaceId ?? resolveWorkspaceContext().defaultSpaceId }, diff --git a/src/routes/space/[spaceId]/table/[id]/page.server.test.ts b/src/routes/space/[spaceId]/table/[id]/page.server.test.ts index 95c9a2d..db9ccca 100644 --- a/src/routes/space/[spaceId]/table/[id]/page.server.test.ts +++ b/src/routes/space/[spaceId]/table/[id]/page.server.test.ts @@ -1,10 +1,15 @@ import { describe, expect, it } from 'vitest'; import { load } from './+page.server'; -import { createCollection } from '$lib/data/records'; +import { createCollection as rawCreateCollection } from '$lib/data/records'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; import { createCollection as createCollectionService } from '$lib/services'; import { CURRENT_USER } from '$lib/server/current-user'; import { resolveWorkspaceContext } from '$lib/server/workspace-store'; +function createCollection(...args: Parameters) { + return transactWithOrigin(args[0], TEST_ORIGIN, () => rawCreateCollection(...args)); +} + describe('routes/table/[id]/+page.server', () => { it('returns the collection title for an existing collection written directly to the default doc', () => { const { doc, defaultSpaceId } = resolveWorkspaceContext(); diff --git a/tests/e2e/tier-a.test.ts b/tests/e2e/tier-a.test.ts index ebe3a29..b0e9609 100644 --- a/tests/e2e/tier-a.test.ts +++ b/tests/e2e/tier-a.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { createTestHarness, type TestHarness } from './harness'; import { createCollection, - createDocument, + createDocument as createDocumentRaw, createRecord, getRecord, getRecordYText, @@ -22,9 +22,20 @@ import { serviceModules, serviceSurfaces } from '$lib/services/manifest'; import { flushPendingAuditEvents } from '$lib/server/audit-observer'; import { deleteRecord as crdtDeleteRecord } from '$lib/data/records'; import type { ActorId } from '$lib/data/types'; +import { TEST_ORIGIN, transactWithOrigin } from '$lib/mutation-origin'; const human: ActorId = { kind: 'human', userId: 'brylie' }; +// The server-side workspace contexts used by the cross-space tests have +// projection observers attached. Keep direct fixture mutations explicit so +// they exercise the same origin contract as every other test write. +function createDocument( + doc: Parameters[0], + input: Parameters[1] +) { + return transactWithOrigin(doc, TEST_ORIGIN, () => createDocumentRaw(doc, input)); +} + // A generic assertion-sugar helper: T is used only once in the // signature (this rule's own "replace with the constraint" fix would // collapse every call site's return type to `unknown`), but that single @@ -300,16 +311,16 @@ describe('Tier A: Protocol-Level MCP & Yjs E2E Parity', () => { order: docB.order, shardId: docBShard.shardId }); - const recordB = createRecord( - docBShard.doc, - { parentId: docB.id, blockType: 'paragraph' }, - human + const recordB = transactWithOrigin(docBShard.doc, TEST_ORIGIN, () => + createRecord(docBShard.doc, { parentId: docB.id, blockType: 'paragraph' }, human) ); - updateRecordContent( - docBShard.doc, - recordB.id, - { runs: [{ text: 'unicornsparkle', marks: {} }] }, - human + transactWithOrigin(docBShard.doc, TEST_ORIGIN, () => + updateRecordContent( + docBShard.doc, + recordB.id, + { runs: [{ text: 'unicornsparkle', marks: {} }] }, + human + ) ); // Grant this same token access to docB too — otherwise the pre-existing