Merge branch 'main' of https://github.com/schmelczer/obsidian-shared-sync

.github/workflows/check.yml

@@ -30,32 +30,5 @@ jobs:
           sqlx database create --database-url sqlite://db.sqlite3
           sqlx migrate run --source src/app_state/database/migrations --database-url sqlite://db.sqlite3
 
-      - name: Lint sync-server
+      - name: Lint & test
-        run: |
+        run: scripts/check.sh
-          cd sync-server
-          cargo clippy --all-targets --all-features
-          cargo fmt --all -- --check
-          cargo machete
-
-      - name: Test sync-server
-        run: |
-          cd sync-server
-          cargo test --verbose
-
-      - name: Lint frontend
-        run: |
-          cd frontend
-          npm ci
-          npm run build
-          npm run lint
-          if [[ $(git status --porcelain) ]]; then
-              git status --porcelain
-              echo "Failing CI because the working directory is not clean after linting"
-              exit 1
-          fi
-
-      - name: Test frontend
-        run: |
-          cd frontend
-          npm ci
-          npm run test

frontend/obsidian-plugin/package.json

@@ -34,6 +34,7 @@
         "url": "^0.11.4",
         "virtual-scroller": "^1.13.1",
         "webpack": "^5.99.9",
-        "webpack-cli": "^6.0.1"
+        "webpack-cli": "^6.0.1",
+        "reconcile-text": "^0.5.0"
     }
 }

frontend/obsidian-plugin/src/debugging/flaky-websocket-factory.ts

@@ -24,15 +24,18 @@ export function flakyWebSocketFactory(
 
 		public set onmessage(callback: (event: MessageEvent) => void) {
 			super.onmessage = async (event: MessageEvent): Promise<void> => {
-				await this.locks.waitForLock(FlakyWebSocket.RECEIVE_KEY);
+				await this.locks.withLock(
-
+					FlakyWebSocket.RECEIVE_KEY,
+					async () => {
 						if (jitterScaleInSeconds > 0) {
-					await sleep(Math.random() * jitterScaleInSeconds * 1000);
+							await sleep(
+								Math.random() * jitterScaleInSeconds * 1000
+							);
 						}
 
 						callback(event);
-
+					}
-				this.locks.unlock(FlakyWebSocket.RECEIVE_KEY);
+				);
 			};
 		}
 
@@ -66,15 +69,12 @@ export function flakyWebSocketFactory(
 			data: string | ArrayBufferLike | Blob | ArrayBufferView
 		): Promise<void> {
 			// maintain message order
-			await this.locks.waitForLock(FlakyWebSocket.SEND_KEY);
+			await this.locks.withLock(FlakyWebSocket.SEND_KEY, async () => {
-
 				if (jitterScaleInSeconds > 0) {
 					await sleep(Math.random() * jitterScaleInSeconds * 1000);
 				}
-
 				super.send(data);
-
+			});
-			this.locks.unlock(FlakyWebSocket.SEND_KEY);
 		}
 	} as unknown as typeof WebSocket;
 }

frontend/obsidian-plugin/src/vault-link-plugin.ts

@@ -19,7 +19,7 @@ import { updateEditorStatusDisplay } from "./views/editor-sync-line/editor-sync-
 import { remoteCursorsTheme } from "./views/cursors/remote-cursor-theme";
 import {
 	remoteCursorsPlugin,
-	setCursors
+	RemoteCursorsPluginValue
 } from "./views/cursors/remote-cursors-plugin";
 import { LocalCursorUpdateListener } from "./views/cursors/local-cursor-update-listener";
 import { slowFetchFactory } from "./debugging/slow-fetch-factory";
@@ -93,7 +93,7 @@ export default class VaultLinkPlugin extends Plugin {
 		this.registerEditorExtension([remoteCursorsTheme, remoteCursorsPlugin]);
 
 		this.client.addRemoteCursorsUpdateListener((cursors) => {
-			setCursors(cursors, this.app);
+			RemoteCursorsPluginValue.setCursors(cursors, this.app);
 		});
 
 		const cursorListener = new LocalCursorUpdateListener(

frontend/obsidian-plugin/src/views/cursors/remote-cursors-plugin.ts

@@ -9,44 +9,136 @@ import type {
 	ViewUpdate
 } from "@codemirror/view";
 import { RemoteCursorWidget } from "./remote-cursor-widget";
-import type {
+import type { CursorSpan, MaybeOutdatedClientCursors } from "sync-client";
-	CursorSpan,
-	DocumentWithMaybeOutdatedClientCursors
-} from "sync-client";
 import type { App } from "obsidian";
 import { MarkdownView } from "obsidian";
 
-let cursors: {
-	name: string;
-	path: string;
-	span: CursorSpan;
-	deviceId: string;
-}[] = [];
-
 import { StateEffect } from "@codemirror/state";
 import { getRandomColor } from "src/utils/get-random-color";
-import { updateSelection } from "./update-selection";
+import type { SpanWithHistory } from "reconcile-text";
+import { reconcileWithHistory } from "reconcile-text";
+
+function findWhereToMoveCursor(
+	cursor: number,
+	spans: SpanWithHistory[]
+): number | null {
+	let position = 0;
+	for (const span of spans) {
+		// left and origin are the same
+		if (position === cursor && span.history === "AddedFromRight") {
+			return position + span.text.length;
+		}
+		position += span.text.length;
+		if (position === cursor && span.history === "RemovedFromRight") {
+			return position - span.text.length;
+		}
+	}
+
+	return null;
+}
 
 const forceUpdate = StateEffect.define();
 
 export class RemoteCursorsPluginValue implements PluginValue {
+	private static cursors: {
+		name: string;
+		path: string;
+		span: CursorSpan;
+		deviceId: string;
+		isOutdated: boolean;
+	}[] = [];
+
 	public decorations: DecorationSet = RangeSet.of([]);
 
-	public update(update: ViewUpdate): void {
+	public static setCursors(
-		update.changes.iterChanges((fromA, toA, fromB, toB, _inserted) => {
+		clients: MaybeOutdatedClientCursors[],
-			const spans = cursors.map((cursor) => cursor.span);
+		app: App
-			updateSelection({
+	): void {
-				fromA,
+		RemoteCursorsPluginValue.cursors = [
-				toA,
+			...RemoteCursorsPluginValue.cursors.filter(({ deviceId }) =>
-				fromB,
+				clients.some(
-				toB,
+					(client) =>
-				spans
+						client.deviceId === deviceId && client.isOutdated
+				)
+			),
+			...clients
+				.filter(
+					({ isOutdated, deviceId }) =>
+						!isOutdated ||
+						RemoteCursorsPluginValue.cursors.every(
+							(c) => deviceId !== c.deviceId
+						)
+				)
+				.flatMap((client) => {
+					const clientCursors = client.documentsWithCursors;
+					return clientCursors.flatMap((cursor) =>
+						cursor.cursors.map((span) => ({
+							name: client.userName,
+							path: cursor.relative_path,
+							deviceId: client.deviceId,
+							isOutdated: client.isOutdated,
+							span: { ...span }
+						}))
+					);
+				})
+		];
+
+		app.workspace
+			.getLeavesOfType("markdown")
+			.map((leaf) => leaf.view)
+			.filter((view) => view instanceof MarkdownView)
+			.forEach((view) => {
+				// @ts-expect-error, not typed
+				// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
+				const editor = view.editor.cm as EditorView;
+
+				editor.dispatch({
+					effects: [forceUpdate.of(null)]
 				});
 			});
+	}
+
+	public update(update: ViewUpdate): void {
+		const original = update.startState.doc.toString();
+		const edited = update.state.doc.toString();
+
+		const updatedPositions: number[] = [];
+		const reconciled = reconcileWithHistory(
+			original,
+			{
+				text: original,
+				cursors: RemoteCursorsPluginValue.cursors.flatMap(
+					({ span }, i) => [
+						{ id: i * 2, position: span.start },
+						{ id: i * 2 + 1, position: span.end }
+					]
+				)
+			},
+			edited,
+			"Character"
+		);
+
+		reconciled.cursors.forEach(({ id, position }) => {
+			const whereToJump = findWhereToMoveCursor(
+				position,
+				reconciled.history
+			);
+			if (whereToJump !== null) {
+				updatedPositions[id] = whereToJump;
+			} else {
+				updatedPositions[id] = position;
+			}
+		});
+
+		RemoteCursorsPluginValue.cursors.forEach(({ span }, i) => {
+			span.start = updatedPositions[i * 2];
+			span.end = updatedPositions[i * 2 + 1];
+		});
 
 		const decorations: Range<Decoration>[] = [];
 
-		cursors.forEach(({ name, span: { start, end } }) => {
+		RemoteCursorsPluginValue.cursors.forEach(
+			({ name, span: { start, end } }) => {
 				const color = getRandomColor(name);
 				const startLine = update.view.state.doc.lineAt(start);
 				const endLine = update.view.state.doc.lineAt(end);
@@ -76,7 +168,11 @@ export class RemoteCursorsPluginValue implements PluginValue {
 					});
 
 					// render text-selection in the lines between the first and last line
-				for (let i = startLine.number + 1; i < endLine.number; i++) {
+					for (
+						let i = startLine.number + 1;
+						i < endLine.number;
+						i++
+					) {
 						const currentLine = update.view.state.doc.line(i);
 						decorations.push({
 							from: currentLine.from,
@@ -106,7 +202,8 @@ export class RemoteCursorsPluginValue implements PluginValue {
 						widget: new RemoteCursorWidget(color, name)
 					})
 				});
-		});
+			}
+		);
 
 		this.decorations = Decoration.set(decorations, true);
 	}
@@ -118,43 +215,3 @@ export const remoteCursorsPlugin = ViewPlugin.fromClass(
 		decorations: (v) => v.decorations
 	}
 );
-
-export function setCursors(
-	clients: DocumentWithMaybeOutdatedClientCursors[],
-	app: App
-): void {
-	cursors = [
-		...cursors.filter(({ deviceId }) =>
-			clients.some(
-				(client) => client.deviceId === deviceId && client.isOutdated
-			)
-		),
-		...clients
-			.filter(({ isOutdated }) => !isOutdated)
-			.flatMap((client) => {
-				const clientCursors = client.documentsWithCursors;
-				return clientCursors.flatMap((cursor) =>
-					cursor.cursors.map((span) => ({
-						name: client.userName,
-						path: cursor.relative_path,
-						deviceId: client.deviceId,
-						span
-					}))
-				);
-			})
-	];
-
-	app.workspace
-		.getLeavesOfType("markdown")
-		.map((leaf) => leaf.view)
-		.filter((view) => view instanceof MarkdownView)
-		.forEach((view) => {
-			// @ts-expect-error, not typed
-			// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
-			const editor = view.editor.cm as EditorView;
-
-			editor.dispatch({
-				effects: [forceUpdate.of(null)]
-			});
-		});
-}

frontend/obsidian-plugin/src/views/cursors/update-selection.test.ts

@@ -1,111 +0,0 @@
-import { updateSelection } from "./update-selection";
-
-describe("Selection update", () => {
-	it("should handle span fully before - insert", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 0,
-			toA: 0,
-			fromB: 0,
-			toB: 2,
-			spans
-		});
-		expect(spans).toEqual([{ start: 5, end: 7 }]);
-	});
-
-	it("should handle span fully before - delete", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 0,
-			toA: 2,
-			fromB: 0,
-			toB: 0,
-			spans
-		});
-		expect(spans).toEqual([{ start: 1, end: 3 }]);
-	});
-
-	it("should handle span fully after - insert", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 6,
-			toA: 6,
-			fromB: 6,
-			toB: 10,
-			spans
-		});
-		expect(spans).toEqual([{ start: 3, end: 5 }]);
-	});
-
-	it("should handle span fully after - delete", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 6,
-			toA: 10,
-			fromB: 6,
-			toB: 6,
-			spans
-		});
-		expect(spans).toEqual([{ start: 3, end: 5 }]);
-	});
-
-	it("should handle span fully within - insert", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 4,
-			toA: 4,
-			fromB: 4,
-			toB: 6,
-			spans
-		});
-		expect(spans).toEqual([{ start: 3, end: 7 }]);
-	});
-
-	it("should handle span fully within - delete", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 4,
-			toA: 5,
-			fromB: 4,
-			toB: 4,
-			spans
-		});
-		expect(spans).toEqual([{ start: 3, end: 4 }]);
-	});
-
-	it("should handle span overlapping with start", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 2,
-			toA: 4,
-			fromB: 2,
-			toB: 2,
-			spans
-		});
-		expect(spans).toEqual([{ start: 2, end: 4 }]);
-	});
-
-	it("should handle span overlapping with end", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 4,
-			toA: 6,
-			fromB: 4,
-			toB: 4,
-			spans
-		});
-		expect(spans).toEqual([{ start: 3, end: 4 }]);
-	});
-
-	it("delete entire selection", () => {
-		const spans = [{ start: 3, end: 5 }];
-		updateSelection({
-			fromA: 0,
-			toA: 10,
-			fromB: 0,
-			toB: 0,
-			spans
-		});
-		expect(spans).toEqual([{ start: 0, end: 0 }]);
-	});
-});

frontend/obsidian-plugin/src/views/cursors/update-selection.ts

@@ -1,40 +0,0 @@
-import type { CursorSpan } from "sync-client";
-
-export const updateSelection = ({
-	fromA,
-	toA,
-	toB,
-	spans
-}: {
-	fromA: number;
-	toA: number;
-	fromB: number;
-	toB: number;
-	spans: CursorSpan[];
-}): void => {
-	spans.forEach((span) => {
-		if (fromA <= span.start) {
-			// the change covers the entirety of the selection
-			if (toA > span.end) {
-				span.start = toB;
-				span.end = toB;
-				return;
-			}
-
-			let change = toB - toA;
-			if (change < 0) {
-				// it's a deletion
-				// if overlaps with the start, we can't move it back more than the deleted range
-				change = Math.max(change, fromA - span.start);
-			}
-
-			span.start += change;
-			span.end += change;
-		} else if (toA <= span.end) {
-			span.end += toB - toA;
-		} else if (toB <= span.end) {
-			// a deletion overlaps with the end, so we move the end
-			span.end = toB;
-		}
-	});
-};

frontend/obsidian-plugin/src/views/editor-sync-line/editor-sync-line.ts

@@ -1,7 +1,7 @@
 import type { Workspace } from "obsidian";
 import { FileView, setIcon } from "obsidian";
 import type { SyncClient } from "sync-client";
-import { DocumentUpdateStatus } from "sync-client";
+import { DocumentSyncStatus } from "sync-client";
 import "./editor-sync-line.scss";
 
 export function updateEditorStatusDisplay(
@@ -35,7 +35,7 @@ export function updateEditorStatusDisplay(
 
 			const isLoading =
 				client.getDocumentSyncingStatus(filePath) ==
-				DocumentUpdateStatus.SYNCING;
+				DocumentSyncStatus.SYNCING;
 
 			if (isLoading) {
 				element.classList.add("loading");

frontend/package-lock.json

@@ -6373,6 +6373,7 @@
 				"jest": "^29.7.0",
 				"mini-css-extract-plugin": "^2.9.2",
 				"obsidian": "1.8.7",
+				"reconcile-text": "^0.5.0",
 				"resolve-url-loader": "^5.0.0",
 				"sass": "^1.89.1",
 				"sass-loader": "^16.0.5",

frontend/sync-client/src/file-operations/safe-filesystem-operations.ts

@@ -31,16 +31,17 @@ export class SafeFileSystemOperations implements FileSystemOperations {
 		this.logger.debug(`Reading file '${path}'`);
 		return this.safeOperation(
 			path,
-			this.decorateToHoldLock(path, async () => this.fs.read(path)),
+			async () =>
+				this.locks.withLock(path, async () => this.fs.read(path)),
 			"read"
 		);
 	}
 
 	public async write(path: RelativePath, content: Uint8Array): Promise<void> {
 		this.logger.debug(`Writing to file '${path}'`);
-		return this.decorateToHoldLock(path, async () =>
+		return this.locks.withLock(path, async () =>
 			this.fs.write(path, content)
-		)();
+		);
 	}
 
 	public async atomicUpdateText(
@@ -50,7 +51,8 @@ export class SafeFileSystemOperations implements FileSystemOperations {
 		this.logger.debug(`Atomically updating file '${path}'`);
 		return this.safeOperation(
 			path,
-			this.decorateToHoldLock(path, async () =>
+			async () =>
+				this.locks.withLock(path, async () =>
 					this.fs.atomicUpdateText(path, updater)
 				),
 			"atomicUpdateText"
@@ -61,7 +63,8 @@ export class SafeFileSystemOperations implements FileSystemOperations {
 		// Logging this would be too noisy
 		return this.safeOperation(
 			path,
-			this.decorateToHoldLock(path, async () =>
+			async () =>
+				this.locks.withLock(path, async () =>
 					this.fs.getFileSize(path)
 				),
 			"getFileSize"
@@ -70,23 +73,19 @@ export class SafeFileSystemOperations implements FileSystemOperations {
 
 	public async exists(path: RelativePath): Promise<boolean> {
 		this.logger.debug(`Checking if file '${path}' exists`);
-		return this.decorateToHoldLock(path, async () =>
+		return this.locks.withLock(path, async () => this.fs.exists(path));
-			this.fs.exists(path)
-		)();
 	}
 
 	public async createDirectory(path: RelativePath): Promise<void> {
 		this.logger.debug(`Creating directory '${path}'`);
-		return this.decorateToHoldLock(path, async () =>
+		return this.locks.withLock(path, async () =>
 			this.fs.createDirectory(path)
-		)();
+		);
 	}
 
 	public async delete(path: RelativePath): Promise<void> {
 		this.logger.debug(`Deleting file '${path}'`);
-		return this.decorateToHoldLock(path, async () =>
+		return this.locks.withLock(path, async () => this.fs.delete(path));
-			this.fs.delete(path)
-		)();
 	}
 
 	public async rename(
@@ -96,43 +95,14 @@ export class SafeFileSystemOperations implements FileSystemOperations {
 		this.logger.debug(`Renaming file '${oldPath}' to '${newPath}'`);
 		return this.safeOperation(
 			oldPath,
-			this.decorateToHoldLock([oldPath, newPath], async () =>
+			async () =>
+				this.locks.withLock([oldPath, newPath], async () =>
 					this.fs.rename(oldPath, newPath)
 				),
 			"rename"
 		);
 	}
 
-	/**
-	 * Decorate an operation to ensure that the file is locked before running it
-	 * and that the lock is released afterwards. This results in at-most one
-	 * concurrent operation running per file.
-	 */
-	private decorateToHoldLock<T>(
-		pathOrPaths: RelativePath | RelativePath[],
-		operation: () => Promise<T>
-	): () => Promise<T> {
-		return async () => {
-			const paths = Array.isArray(pathOrPaths)
-				? pathOrPaths
-				: [pathOrPaths];
-
-			await Promise.all(
-				paths.map(async (path) => this.locks.waitForLock(path))
-			);
-
-			try {
-				return await operation();
-			} finally {
-				await Promise.all(
-					paths.map((path) => {
-						this.locks.unlock(path);
-					})
-				);
-			}
-		};
-	}
-
 	/**
 	 * Decorate an operation to ensure that the file exists before running it.
 	 * If the operation fails, it will check if the file still exists and throw

frontend/sync-client/src/index.ts

@@ -18,8 +18,8 @@ export type { PersistenceProvider } from "./persistence/persistence";
 export type { CursorSpan } from "./services/types/CursorSpan";
 export type { ClientCursors } from "./services/types/ClientCursors";
 export type { NetworkConnectionStatus } from "./types/network-connection-status";
-export type { DocumentWithMaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
+export type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
-export { DocumentUpdateStatus } from "./types/document-update-status";
+export { DocumentSyncStatus } from "./types/document-sync-status";
 export { SyncClient } from "./sync-client";
 
 import { Locks } from "./utils/locks";

frontend/sync-client/src/persistence/database.ts

@@ -37,7 +37,7 @@ export interface DocumentRecord {
 	documentId: DocumentId;
 	metadata: DocumentMetadata | undefined;
 	isDeleted: boolean;
-	updates: Promise<void>[];
+	updates: Promise<unknown>[];
 	parallelVersion: number;
 }
 
@@ -135,7 +135,7 @@ export class Database {
 		this.save();
 	}
 
-	public removeDocumentPromise(promise: Promise<void>): void {
+	public removeDocumentPromise(promise: Promise<unknown>): void {
 		const entry = this.documents.find(({ updates }) =>
 			updates.includes(promise)
 		);
@@ -167,7 +167,7 @@ export class Database {
 
 	public async getResolvedDocumentByRelativePath(
 		relativePath: RelativePath,
-		promise: Promise<void>
+		promise: Promise<unknown>
 	): Promise<DocumentRecord> {
 		const entry = this.getLatestDocumentByRelativePath(relativePath);
 
@@ -191,7 +191,7 @@ export class Database {
 	public createNewPendingDocument(
 		documentId: DocumentId,
 		relativePath: RelativePath,
-		promise: Promise<void>
+		promise: Promise<unknown>
 	): DocumentRecord {
 		const previousEntry =
 			this.getLatestDocumentByRelativePath(relativePath);

frontend/sync-client/src/services/types/DocumentWithCursors.ts

@@ -2,7 +2,7 @@
 import type { CursorSpan } from "./CursorSpan";
 
 export interface DocumentWithCursors {
-	vault_update_id: number;
+	vault_update_id: number | null;
 	document_id: string;
 	relative_path: string;
 	cursors: CursorSpan[];

frontend/sync-client/src/services/websocket-manager.ts

@@ -152,7 +152,7 @@ export class WebSocketManager {
 				}
 				// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
 			} else if (message.type === "cursorPositions") {
-				this.logger.info(
+				this.logger.debug(
 					`Received cursor positions for ${JSON.stringify(message.clients)}`
 				);
 				this.remoteCursorsUpdateListeners.forEach((listener) => {

frontend/sync-client/src/sync-client.ts

@@ -14,26 +14,16 @@ import { ConnectionStatus } from "./services/connection-status";
 import { UnrestrictedSyncer } from "./sync-operations/unrestricted-syncer";
 import { rateLimit } from "./utils/rate-limit";
 import type { NetworkConnectionStatus } from "./types/network-connection-status";
-import { DocumentUpdateStatus } from "./types/document-update-status";
+import { DocumentSyncStatus } from "./types/document-sync-status";
 import { WebSocketManager } from "./services/websocket-manager";
 import { createClientId } from "./utils/create-client-id";
+import { CursorTracker } from "./sync-operations/cursor-tracker";
 import type { CursorSpan } from "./services/types/CursorSpan";
-import type { ClientCursors } from "./services/types/ClientCursors";
+import type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
-import type { DocumentWithCursors } from "./services/types/DocumentWithCursors";
+import { FileChangeNotifier } from "./sync-operations/file-change-notifier";
-import { hash } from "./utils/hash";
-import type { DocumentWithMaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
-
-enum DocumentUpToDateness {
-	UpToDate = "UpToDate",
-	Prior = "Prior",
-	Later = "Later"
-}
 
 export class SyncClient {
 	private static readonly MINIMUM_SAVE_INTERVAL_MS = 1000;
-	private lastCursorState: DocumentWithCursors[] = [];
-
-	private readonly knownClientCursors: ClientCursors[] = [];
 
 	// eslint-disable-next-line @typescript-eslint/max-params
 	private constructor(
@@ -45,7 +35,8 @@ export class SyncClient {
 		private readonly webSocketManager: WebSocketManager,
 		private readonly _logger: Logger,
 		private readonly connectionStatus: ConnectionStatus,
-		private readonly fileOperations: FileOperations
+		private readonly cursorTracker: CursorTracker,
+		private readonly fileChangeNotifier: FileChangeNotifier
 	) {
 		this.settings.addOnSettingsChangeListener(
 			async (newSettings, oldSettings) => {
@@ -54,10 +45,6 @@ export class SyncClient {
 				}
 			}
 		);
-
-		this.webSocketManager.addRemoteCursorsUpdateListener((cursors) => {
-			this.knownClientCursors.push(...cursors);
-		});
 	}
 
 	public get logger(): Logger {
@@ -148,7 +135,6 @@ export class SyncClient {
 		);
 
 		const syncer = new Syncer(
-			deviceId,
 			logger,
 			database,
 			settings,
@@ -166,6 +152,13 @@ export class SyncClient {
 			webSocket
 		);
 
+		const fileChangeNotifier = new FileChangeNotifier();
+		const cursorTracker = new CursorTracker(
+			database,
+			webSocketManager,
+			fileOperations,
+			fileChangeNotifier
+		);
 		const client = new SyncClient(
 			history,
 			settings,
@@ -175,7 +168,8 @@ export class SyncClient {
 			webSocketManager,
 			logger,
 			connectionStatus,
-			fileOperations
+			cursorTracker,
+			fileChangeNotifier
 		);
 
 		logger.info("SyncClient initialised");
@@ -264,12 +258,14 @@ export class SyncClient {
 	public async syncLocallyCreatedFile(
 		relativePath: RelativePath
 	): Promise<void> {
+		this.fileChangeNotifier.notifyOfFileChange(relativePath);
 		return this.syncer.syncLocallyCreatedFile(relativePath);
 	}
 
 	public async syncLocallyDeletedFile(
 		relativePath: RelativePath
 	): Promise<void> {
+		this.fileChangeNotifier.notifyOfFileChange(relativePath);
 		return this.syncer.syncLocallyDeletedFile(relativePath);
 	}
 
@@ -280,6 +276,7 @@ export class SyncClient {
 		oldPath?: RelativePath;
 		relativePath: RelativePath;
 	}): Promise<void> {
+		this.fileChangeNotifier.notifyOfFileChange(relativePath);
 		return this.syncer.syncLocallyUpdatedFile({
 			oldPath,
 			relativePath
@@ -288,154 +285,26 @@ export class SyncClient {
 
 	public getDocumentSyncingStatus(
 		relativePath: RelativePath
-	): DocumentUpdateStatus {
+	): DocumentSyncStatus {
 		const document =
 			this.database.getLatestDocumentByRelativePath(relativePath);
 		if (document === undefined) {
-			return DocumentUpdateStatus.SYNCING;
+			return DocumentSyncStatus.SYNCING;
 		}
 		return document.updates.length > 0
-			? DocumentUpdateStatus.SYNCING
+			? DocumentSyncStatus.SYNCING
-			: DocumentUpdateStatus.UP_TO_DATE;
+			: DocumentSyncStatus.UP_TO_DATE;
 	}
 
-	/// Update the local cursors for the given documents.
-	/// Can be called frequently as it only emits an event
-	// if the state has actually changed.
 	public async updateLocalCursors(
 		documentToCursors: Record<RelativePath, CursorSpan[]>
 	): Promise<void> {
-		const documentsWithCursors: DocumentWithCursors[] = [];
+		await this.cursorTracker.sendLocalCursorsToServer(documentToCursors);
-
-		for (const [relativePath, cursors] of Object.entries(
-			documentToCursors
-		)) {
-			const record =
-				this.database.getLatestDocumentByRelativePath(relativePath);
-
-			if (!record) {
-				continue; // Let's wait for the file to be created before sending cursors
-			}
-
-			const readContent = await this.fileOperations.read(relativePath);
-
-			if (record.metadata?.hash !== hash(readContent)) {
-				continue; // Wouldn't make sense to sync the positions in a dirty file
-			}
-
-			documentsWithCursors.push({
-				relative_path: relativePath,
-				document_id: record.documentId,
-				vault_update_id: record.metadata.parentVersionId,
-				cursors
-			});
-		}
-
-		if (
-			JSON.stringify(this.lastCursorState) ===
-			JSON.stringify(documentsWithCursors)
-		) {
-			return;
-		}
-
-		this.lastCursorState = documentsWithCursors;
-
-		this.webSocketManager.updateLocalCursors({ documentsWithCursors });
 	}
 
 	public addRemoteCursorsUpdateListener(
-		listener: (cursors: DocumentWithMaybeOutdatedClientCursors[]) => unknown
+		listener: (cursors: MaybeOutdatedClientCursors[]) => unknown
 	): void {
-		this.webSocketManager.addRemoteCursorsUpdateListener(async () => {
+		this.cursorTracker.addRemoteCursorsUpdateListener(listener);
-			listener(await this.getRelevantClientCursors());
-		});
-	}
-
-	private async getRelevantClientCursors(): Promise<
-		DocumentWithMaybeOutdatedClientCursors[]
-	> {
-		const result: DocumentWithMaybeOutdatedClientCursors[] = [];
-		const included = new Set<string>();
-		for (const clientCursors of [...this.knownClientCursors].reverse()) {
-			if (included.has(clientCursors.deviceId)) {
-				continue;
-			}
-
-			const upToDateness =
-				await this.getDocumentsUpToDateness(clientCursors);
-			if (upToDateness == DocumentUpToDateness.Later) {
-				continue;
-			}
-
-			result.push({
-				...clientCursors,
-				isOutdated: upToDateness == DocumentUpToDateness.Prior
-			});
-
-			included.add(clientCursors.deviceId);
-		}
-
-		return result;
-	}
-
-	private async getDocumentsUpToDateness(
-		clientCursor: ClientCursors
-	): Promise<DocumentUpToDateness> {
-		const results = [];
-		for (const document of clientCursor.documentsWithCursors) {
-			results.push(await this.getDocumentUpToDateness(document));
-		}
-
-		if (
-			results.every((result) => result === DocumentUpToDateness.UpToDate)
-		) {
-			return DocumentUpToDateness.UpToDate;
-		}
-
-		if (
-			results.every(
-				(result) =>
-					result === DocumentUpToDateness.UpToDate ||
-					result === DocumentUpToDateness.Prior
-			)
-		) {
-			return DocumentUpToDateness.Prior;
-		}
-
-		return DocumentUpToDateness.Later;
-	}
-
-	private async getDocumentUpToDateness(
-		document: DocumentWithCursors
-	): Promise<DocumentUpToDateness> {
-		const record = this.database.getLatestDocumentByRelativePath(
-			document.relative_path
-		);
-
-		if (!record) {
-			// the document of the cursor must be from the future
-			return DocumentUpToDateness.Later;
-		}
-
-		if (
-			(record.metadata?.parentVersionId ?? 0) < document.vault_update_id
-		) {
-			return DocumentUpToDateness.Later;
-		} else if (
-			document.vault_update_id < (record.metadata?.parentVersionId ?? 0)
-		) {
-			// the document of the cursor must be from the past
-			return DocumentUpToDateness.Prior;
-		}
-
-		const currentContent = await this.fileOperations.read(
-			document.relative_path
-		);
-
-		return this.database.getLatestDocumentByRelativePath(
-			document.relative_path
-		)?.metadata?.hash === hash(currentContent)
-			? DocumentUpToDateness.UpToDate
-			: DocumentUpToDateness.Prior;
 	}
 }

frontend/sync-client/src/sync-operations/cursor-tracker.ts

@@ -0,0 +1,253 @@
+import type { FileOperations } from "../file-operations/file-operations";
+import type { Database, RelativePath } from "../persistence/database";
+import type { ClientCursors } from "../services/types/ClientCursors";
+import type { CursorSpan } from "../services/types/CursorSpan";
+import type { DocumentWithCursors } from "../services/types/DocumentWithCursors";
+import type { WebSocketManager } from "../services/websocket-manager";
+import type { MaybeOutdatedClientCursors } from "../types/maybe-outdated-client-cursors";
+import { DocumentUpToDateness } from "../types/document-up-to-dateness";
+import { hash } from "../utils/hash";
+import type { FileChangeNotifier } from "./file-change-notifier";
+import { Lock } from "../utils/locks";
+
+// Cursor positions are updated separately from documents. However, a given cursor position is only
+// valid within a certain version of the document it belongs to. This class tracks previous and the latest
+// known remote cursor positions, and for each document, tries to return the latest cursor positions that are
+// not from the future.
+export class CursorTracker {
+	private readonly updateLock = new Lock();
+
+	private knownRemoteCursors: (ClientCursors & {
+		upToDateness: DocumentUpToDateness;
+	})[] = [];
+
+	private lastLocalCursorState: DocumentWithCursors[] = [];
+	private lastLocalCursorStateWithoutDirtyDocuments: DocumentWithCursors[] =
+		[];
+
+	public constructor(
+		private readonly database: Database,
+		private readonly webSocketManager: WebSocketManager,
+		private readonly fileOperations: FileOperations,
+		private readonly fileChangeNotifier: FileChangeNotifier
+	) {
+		this.webSocketManager.addRemoteCursorsUpdateListener(
+			async (clientCursors) => {
+				await this.updateLock.withLock(async () => {
+					// The latest message will contain all active clients, so we can delete the ones
+					// from the local list which are no longer active.
+					const allIds = new Set(
+						clientCursors.map((c) => c.deviceId)
+					);
+					const updatedKnownRemoteCursors =
+						this.knownRemoteCursors.filter((c) =>
+							allIds.has(c.deviceId)
+						);
+
+					for (const cursor of clientCursors.filter((client) =>
+						client.documentsWithCursors.every(
+							(doc) => doc.vault_update_id != null
+						)
+					)) {
+						updatedKnownRemoteCursors.push({
+							...cursor,
+							upToDateness:
+								await this.getDocumentsUpToDateness(cursor)
+						});
+					}
+
+					this.knownRemoteCursors = updatedKnownRemoteCursors;
+				});
+			}
+		);
+
+		this.fileChangeNotifier.addFileChangeListener(async (relativePath) =>
+			this.updateLock.withLock(async () => {
+				for (const clientCursor of this.knownRemoteCursors) {
+					if (
+						clientCursor.documentsWithCursors.some(
+							(document) =>
+								document.relative_path === relativePath
+						)
+					) {
+						clientCursor.upToDateness =
+							await this.getDocumentsUpToDateness(clientCursor);
+					}
+				}
+			})
+		);
+	}
+
+	/// Update the local cursors for the given documents.
+	/// Can be called frequently as it only emits an event
+	/// if the state has actually changed.
+	public async sendLocalCursorsToServer(
+		documentToCursors: Record<RelativePath, CursorSpan[]>
+	): Promise<void> {
+		const documentsWithCursors: DocumentWithCursors[] = [];
+
+		for (const [relativePath, cursors] of Object.entries(
+			documentToCursors
+		)) {
+			const record =
+				this.database.getLatestDocumentByRelativePath(relativePath);
+
+			if (!record) {
+				continue; // Let's wait for the file to be created before sending cursors
+			}
+
+			if (!record.metadata) {
+				continue; // this is a new document, no need to sync the cursors
+			}
+
+			documentsWithCursors.push({
+				relative_path: relativePath,
+				document_id: record.documentId,
+				vault_update_id: record.metadata.parentVersionId,
+				cursors: cursors.map(({ start, end }) => ({
+					start: Math.min(start, end),
+					end: Math.max(start, end)
+				})) // the client might send directional selections
+			});
+		}
+
+		if (
+			JSON.stringify(this.lastLocalCursorState) ===
+			JSON.stringify(documentsWithCursors)
+		) {
+			// Caching step to avoid reading the edited files all the time
+			return;
+		}
+		this.lastLocalCursorState = documentsWithCursors;
+
+		for (const doc of documentsWithCursors) {
+			const readContent = await this.fileOperations.read(
+				doc.relative_path
+			);
+			const record = this.database.getLatestDocumentByRelativePath(
+				doc.relative_path
+			);
+			if (record?.metadata?.hash !== hash(readContent)) {
+				doc.vault_update_id = null;
+			}
+		}
+
+		if (
+			JSON.stringify(this.lastLocalCursorStateWithoutDirtyDocuments) ===
+			JSON.stringify(documentsWithCursors)
+		) {
+			return;
+		}
+
+		this.lastLocalCursorStateWithoutDirtyDocuments = documentsWithCursors;
+
+		this.webSocketManager.updateLocalCursors({ documentsWithCursors });
+	}
+
+	// The returned position may be accurate, if it matches the document version, or outdated, in which case
+	// the client has to heuristically guess it's current position based on the local edits.
+	public addRemoteCursorsUpdateListener(
+		listener: (cursors: MaybeOutdatedClientCursors[]) => unknown
+	): void {
+		// CursorTracker registers its own event listener in the constructor so it must have been called before this
+		this.webSocketManager.addRemoteCursorsUpdateListener(async () => {
+			await this.updateLock.withLock(() =>
+				listener(this.getRelevantAndPruneKnownClientCursors())
+			);
+		});
+	}
+
+	private getRelevantAndPruneKnownClientCursors(): MaybeOutdatedClientCursors[] {
+		const result: MaybeOutdatedClientCursors[] = [];
+		const included = new Set<string>();
+
+		const relevantCursors = [];
+		for (const clientCursors of [...this.knownRemoteCursors].reverse()) {
+			if (included.has(clientCursors.deviceId)) {
+				continue;
+			}
+
+			if (clientCursors.upToDateness == DocumentUpToDateness.Later) {
+				continue;
+			}
+
+			result.push({
+				...clientCursors,
+				isOutdated:
+					clientCursors.upToDateness == DocumentUpToDateness.Prior
+			});
+
+			included.add(clientCursors.deviceId);
+			relevantCursors.unshift(clientCursors); // to reverse order back to normal
+		}
+
+		this.knownRemoteCursors = relevantCursors;
+
+		return result;
+	}
+
+	// We store up-to-dateness on a per-client basis to simplify the implementation.
+	// An individual client won't have too many documents open at once, so this is a reasonable trade-off.
+	private async getDocumentsUpToDateness(
+		clientCursor: ClientCursors
+	): Promise<DocumentUpToDateness> {
+		const results = [];
+		for (const document of clientCursor.documentsWithCursors) {
+			results.push(await this.getDocumentUpToDateness(document));
+		}
+
+		if (
+			results.every((result) => result === DocumentUpToDateness.UpToDate)
+		) {
+			return DocumentUpToDateness.UpToDate;
+		}
+
+		if (
+			results.every(
+				(result) =>
+					result === DocumentUpToDateness.UpToDate ||
+					result === DocumentUpToDateness.Prior
+			)
+		) {
+			return DocumentUpToDateness.Prior;
+		}
+
+		return DocumentUpToDateness.Later;
+	}
+
+	private async getDocumentUpToDateness(
+		document: DocumentWithCursors
+	): Promise<DocumentUpToDateness> {
+		const record = this.database.getLatestDocumentByRelativePath(
+			document.relative_path
+		);
+
+		if (!record) {
+			// the document of the cursor must be from the future
+			return DocumentUpToDateness.Later;
+		}
+
+		if (
+			(record.metadata?.parentVersionId ?? 0) <
+			(document.vault_update_id ?? 0)
+		) {
+			return DocumentUpToDateness.Later;
+		} else if (
+			(document.vault_update_id ?? 0) <
+			(record.metadata?.parentVersionId ?? 0)
+		) {
+			// the document of the cursor must be from the past
+			return DocumentUpToDateness.Prior;
+		}
+
+		const currentContent = await this.fileOperations.read(
+			document.relative_path
+		);
+
+		return this.database.getLatestDocumentByRelativePath(
+			document.relative_path
+		)?.metadata?.hash === hash(currentContent)
+			? DocumentUpToDateness.UpToDate
+			: DocumentUpToDateness.Prior;
+	}
+}

frontend/sync-client/src/sync-operations/file-change-notifier.ts

@@ -0,0 +1,15 @@
+import type { RelativePath } from "../persistence/database";
+
+export class FileChangeNotifier {
+	private readonly listeners: ((filePath: RelativePath) => unknown)[] = [];
+
+	public addFileChangeListener(
+		listener: (filePath: RelativePath) => unknown
+	): void {
+		this.listeners.push(listener);
+	}
+
+	public notifyOfFileChange(filePath: RelativePath): void {
+		this.listeners.forEach((listener) => listener(filePath));
+	}
+}

frontend/sync-client/src/sync-operations/syncer.ts

@@ -9,7 +9,7 @@ import type { Logger } from "../tracing/logger";
 import PQueue from "p-queue";
 import { hash } from "../utils/hash";
 import { v4 as uuidv4 } from "uuid";
-import type { Settings, SyncSettings } from "../persistence/settings";
+import type { Settings } from "../persistence/settings";
 import type { FileOperations } from "../file-operations/file-operations";
 import { findMatchingFile } from "../utils/find-matching-file";
 import type { UnrestrictedSyncer } from "./unrestricted-syncer";
@@ -27,12 +27,10 @@ export class Syncer {
 
 	private runningScheduleSyncForOfflineChanges: Promise<void> | undefined;
 
-	// eslint-disable-next-line @typescript-eslint/max-params
 	public constructor(
-		private readonly deviceId: string,
 		private readonly logger: Logger,
 		private readonly database: Database,
-		private readonly settings: Settings,
+		settings: Settings,
 		private readonly syncService: SyncService,
 		private readonly operations: FileOperations,
 		private readonly internalSyncer: UnrestrictedSyncer
@@ -261,20 +259,16 @@ export class Syncer {
 			remoteVersion.documentId
 		);
 
-		let hasLockToRelease = false;
 		if (document === undefined) {
 			// Let's avoid the same documents getting created in parallel multiple times.
 			// There might be multiple tasks waiting for the lock
-			await this.remoteDocumentsLock.waitForLock(
+			return this.remoteDocumentsLock.withLock(
-				remoteVersion.documentId
+				remoteVersion.documentId,
-			);
+				async () => {
-			hasLockToRelease = true;
 					document = this.database.getDocumentByDocumentId(
 						remoteVersion.documentId
 					);
-		}
 
-		try {
 					// We're either the first one to get the lock, so we have to create the document in `unrestrictedSyncRemotelyUpdatedFile`
 					if (document === undefined) {
 						await this.syncQueue.add(async () =>
@@ -308,11 +302,34 @@ export class Syncer {
 					}
 
 					this.database.addSeenUpdateId(remoteVersion.vaultUpdateId);
+				}
+			);
+		}
+
+		// We're either the first one to get the lock, so we have to create the document in `unrestrictedSyncRemotelyUpdatedFile`
+		const [promise, resolve, reject] = createPromise();
+
+		document = await this.database.getResolvedDocumentByRelativePath(
+			document.relativePath,
+			promise
+		);
+
+		try {
+			await this.syncQueue.add(async () =>
+				this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
+					remoteVersion,
+					document
+				)
+			);
+
+			resolve();
+		} catch (e) {
+			reject(e);
 		} finally {
-			if (hasLockToRelease) {
+			this.database.removeDocumentPromise(promise);
-				this.remoteDocumentsLock.unlock(remoteVersion.documentId);
-			}
 		}
+
+		this.database.addSeenUpdateId(remoteVersion.vaultUpdateId);
 	}
 
 	private async internalScheduleSyncForOfflineChanges(): Promise<void> {

frontend/sync-client/src/types/document-sync-status.ts

@@ -0,0 +1,4 @@
+export enum DocumentSyncStatus {
+	UP_TO_DATE = "UP_TO_DATE",
+	SYNCING = "SYNCING"
+}

frontend/sync-client/src/types/document-up-to-dateness.ts

@@ -0,0 +1,5 @@
+export enum DocumentUpToDateness {
+	UpToDate = "UpToDate", // easiest case, the client can just show the cursors as-is
+	Prior = "Prior", // The cursors are outdated, so the client has to guess the cursor positions based on local updates. This is only possible if this client's cursor has once been up-to-date in a given document.
+	Later = "Later" // The cursors are from a future version of a document, there's no way we can accuratly show them locally.
+}