andras/vault-link
1
0
Fork 0
Code Issues Pull requests 18 Projects Releases 55 Packages Wiki Activity Actions 1

Add event handler class

Browse source
This commit is contained in:
Andras Schmelczer 2025-12-07 13:30:45 +00:00
parent 1ea465fcf8
commit 496db06213
14 changed files with 2428 additions and 2309 deletions
Download patch file Download diff file Expand all files Collapse all files

78
frontend/obsidian-plugin/src/views/cursors/file-explorer.ts
View file

@ -2,54 +2,54 @@ import "./file-explorer.scss";
import type { App, View } from "obsidian"; import type { App, View } from "obsidian";
import { import {
utils, utils,
type MaybeOutdatedClientCursors, type MaybeOutdatedClientCursors,
type RelativePath type RelativePath
} from "sync-client"; } from "sync-client";
const REMOTE_USER_CONTAINER_CLASS = "remote-users"; const REMOTE_USER_CONTAINER_CLASS = "remote-users";
export function renderCursorsInFileExplorer( export function renderCursorsInFileExplorer(
cursors: MaybeOutdatedClientCursors[], cursors: MaybeOutdatedClientCursors[],
app: App app: App
): void { ): void {
const fileExplorers = app.workspace.getLeavesOfType("file-explorer"); const fileExplorers = app.workspace.getLeavesOfType("file-explorer");
if (fileExplorers.length == 0) return; if (fileExplorers.length == 0) return;
const [fileExplorer] = fileExplorers; const [fileExplorer] = fileExplorers;
// eslint-disable-next-line @typescript-eslint/no-unsafe-assignment // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
const fileExplorerView: View & { const fileExplorerView: View & {
fileItems: Record<RelativePath, { el: Element }>; // it's an internal API fileItems: Record<RelativePath, { el: Element }>; // it's an internal API
} = fileExplorer.view as any; // eslint-disable-line } = fileExplorer.view as any; // eslint-disable-line
for (const key in fileExplorerView.fileItems) { for (const key in fileExplorerView.fileItems) {
const element = const element =
fileExplorerView.fileItems[key].el.querySelector(".tree-item-self"); fileExplorerView.fileItems[key].el.querySelector(".tree-item-self");
const customElement = createDiv( const customElement = createDiv(
{ {
cls: REMOTE_USER_CONTAINER_CLASS cls: REMOTE_USER_CONTAINER_CLASS
}, },
(parent) => { (parent) => {
cursors.forEach((cursor) => { cursors.forEach((cursor) => {
cursor.documentsWithCursors.forEach((document) => { cursor.documentsWithCursors.forEach((document) => {
if (document.relative_path.startsWith(key)) { if (document.relative_path.startsWith(key)) {
parent.appendChild( parent.appendChild(
createSpan({ createSpan({
text: cursor.userName, text: cursor.userName,
attr: { attr: {
style: `border-color: ${utils.getRandomColor(cursor.userName)}` style: `border-color: ${utils.getRandomColor(cursor.userName)}`
} }
}) })
); );
} }
}); });
}); });
} }
); );
element?.querySelector("." + REMOTE_USER_CONTAINER_CLASS)?.remove(); element?.querySelector("." + REMOTE_USER_CONTAINER_CLASS)?.remove();
element?.appendChild(customElement); element?.appendChild(customElement);
} }
} }

167
frontend/sync-client/src/persistence/settings.ts
View file

@ -1,113 +1,94 @@
import type { Logger } from "../tracing/logger"; import type { Logger } from "../tracing/logger";
import { awaitAll } from "../utils/await-all";
import { Lock } from "../utils/data-structures/locks"; import { Lock } from "../utils/data-structures/locks";
import { removeFromArray } from "../utils/remove-from-array"; import { EventListeners } from "../utils/data-structures/event-listeners";
export interface SyncSettings { export interface SyncSettings {
remoteUri: string; remoteUri: string;
token: string; token: string;
vaultName: string; vaultName: string;
syncConcurrency: number; syncConcurrency: number;
isSyncEnabled: boolean; isSyncEnabled: boolean;
maxFileSizeMB: number; maxFileSizeMB: number;
ignorePatterns: string[]; ignorePatterns: string[];
webSocketRetryIntervalMs: number; webSocketRetryIntervalMs: number;
diffCacheSizeMB: number; diffCacheSizeMB: number;
enableTelemetry: boolean; enableTelemetry: boolean;
networkRetryIntervalMs: number; networkRetryIntervalMs: number;
minimumSaveIntervalMs: number; minimumSaveIntervalMs: number;
} }
export const DEFAULT_SETTINGS: SyncSettings = { export const DEFAULT_SETTINGS: SyncSettings = {
remoteUri: "", remoteUri: "",
token: "", token: "",
vaultName: "default", vaultName: "default",
syncConcurrency: 1, syncConcurrency: 1,
isSyncEnabled: false, isSyncEnabled: false,
maxFileSizeMB: 10, maxFileSizeMB: 10,
ignorePatterns: [], ignorePatterns: [],
webSocketRetryIntervalMs: 3500, webSocketRetryIntervalMs: 3500,
diffCacheSizeMB: 4, diffCacheSizeMB: 4,
enableTelemetry: false, enableTelemetry: false,
networkRetryIntervalMs: 1000, networkRetryIntervalMs: 1000,
minimumSaveIntervalMs: 1000 minimumSaveIntervalMs: 1000
}; };
export class Settings { export class Settings {
private settings: SyncSettings; private settings: SyncSettings;
private readonly lock: Lock = new Lock(); private readonly lock: Lock = new Lock();
private readonly onSettingsChangeHandlers: (( public readonly onSettingsChanged = new EventListeners<
newSettings: SyncSettings, (newSettings: SyncSettings, oldSettings: SyncSettings) => unknown
oldSettings: SyncSettings >();
) => unknown)[] = [];
public constructor( public constructor(
private readonly logger: Logger, private readonly logger: Logger,
initialState: Partial<SyncSettings> | undefined, initialState: Partial<SyncSettings> | undefined,
private readonly saveData: (data: SyncSettings) => Promise<void> private readonly saveData: (data: SyncSettings) => Promise<void>
) { ) {
this.settings = { this.settings = {
...DEFAULT_SETTINGS, ...DEFAULT_SETTINGS,
...(initialState ?? {}) ...(initialState ?? {})
}; };
this.logger.debug( this.logger.debug(
`Loaded settings: ${JSON.stringify(this.settings, null, 2)}` `Loaded settings: ${JSON.stringify(this.settings, null, 2)}`
); );
} }
public getSettings(): SyncSettings { public getSettings(): SyncSettings {
return this.settings; return this.settings;
} }
public addOnSettingsChangeListener( public async setSetting<T extends keyof SyncSettings>(
listener: (settings: SyncSettings, oldSettings: SyncSettings) => unknown key: T,
): void { value: SyncSettings[T]
this.onSettingsChangeHandlers.push(listener); ): Promise<void> {
} await this.setSettings({
[key]: value
});
}
public removeOnSettingsChangeListener( public async setSettings(value: Partial<SyncSettings>): Promise<void> {
listener: (settings: SyncSettings, oldSettings: SyncSettings) => unknown await this.lock.withLock(async () => {
): void { this.logger.debug(
removeFromArray(this.onSettingsChangeHandlers, listener); `Updating settings with: ${JSON.stringify(value)}`
} );
const oldSettings = this.settings;
this.settings = {
...this.settings,
...value
};
public async setSetting<T extends keyof SyncSettings>( await this.onSettingsChanged.triggerAsync(
key: T, this.settings,
value: SyncSettings[T] oldSettings
): Promise<void> { );
await this.setSettings({
[key]: value
});
}
public async setSettings(value: Partial<SyncSettings>): Promise<void> { await this.save();
await this.lock.withLock(async () => { });
this.logger.debug( }
`Updating settings with: ${JSON.stringify(value)}`
);
const oldSettings = this.settings;
this.settings = {
...this.settings,
...value
};
await awaitAll( private async save(): Promise<void> {
this.onSettingsChangeHandlers await this.saveData(this.settings);
.map((handler) => { }
return handler(this.settings, oldSettings);
})
.filter((result): result is Promise<unknown> => {
return result instanceof Promise;
})
);
await this.save();
});
}
private async save(): Promise<void> {
await this.saveData(this.settings);
}
} }

8
frontend/sync-client/src/services/websocket-manager.test.ts
View file

@ -122,7 +122,7 @@ describe("WebSocketManager", () => {
MockWebSocket as unknown as typeof WebSocket MockWebSocket as unknown as typeof WebSocket
); );
manager.addRemoteVaultUpdateListener(async () => { manager.onRemoteVaultUpdateReceived.add(async () => {
await new Promise((resolve) => setTimeout(resolve, 10)); await new Promise((resolve) => setTimeout(resolve, 10));
}); });
manager.start(); manager.start();
@ -152,7 +152,7 @@ describe("WebSocketManager", () => {
MockWebSocket as unknown as typeof WebSocket MockWebSocket as unknown as typeof WebSocket
); );
manager.addRemoteCursorsUpdateListener(async () => { manager.onRemoteCursorsUpdateReceived.add(async () => {
await new Promise((resolve) => setTimeout(resolve, 10)); await new Promise((resolve) => setTimeout(resolve, 10));
}); });
manager.start(); manager.start();
@ -227,7 +227,7 @@ describe("WebSocketManager", () => {
); );
let statusChangeCount = 0; let statusChangeCount = 0;
manager.addWebSocketStatusChangeListener(() => { manager.onWebSocketStatusChanged.add(() => {
statusChangeCount++; statusChangeCount++;
}); });
@ -269,7 +269,7 @@ describe("WebSocketManager", () => {
resolveListener = resolve; resolveListener = resolve;
}); });
manager.addRemoteVaultUpdateListener(async () => { manager.onRemoteVaultUpdateReceived.add(async () => {
await listenerPromise; await listenerPromise;
}); });

471
frontend/sync-client/src/services/websocket-manager.ts
View file

@ -6,295 +6,260 @@ import type { CursorPositionFromClient } from "./types/CursorPositionFromClient"
import type { ClientCursors } from "./types/ClientCursors"; import type { ClientCursors } from "./types/ClientCursors";
import { createPromise } from "../utils/create-promise"; import { createPromise } from "../utils/create-promise";
import type { WebSocketVaultUpdate } from "./types/WebSocketVaultUpdate"; import type { WebSocketVaultUpdate } from "./types/WebSocketVaultUpdate";
import { awaitAll } from "../utils/await-all";
import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S } from "../consts"; import { WEBSOCKET_DISCONNECT_TIMEOUT_IN_S } from "../consts";
import { removeFromArray } from "../utils/remove-from-array"; import { removeFromArray } from "../utils/remove-from-array";
import { EventListeners } from "../utils/data-structures/event-listeners";
import { awaitAll } from "../utils/await-all";
export class WebSocketManager { export class WebSocketManager {
private readonly webSocketStatusChangeListeners: (( public readonly onWebSocketStatusChanged = new EventListeners<
isConnected: boolean (isConnected: boolean) => unknown
) => unknown)[] = []; >();
private readonly remoteVaultUpdateListeners: (( public readonly onRemoteVaultUpdateReceived = new EventListeners<
update: WebSocketVaultUpdate (update: WebSocketVaultUpdate) => Promise<void>
) => Promise<void>)[] = []; >();
private readonly remoteCursorsUpdateListeners: (( public readonly onRemoteCursorsUpdateReceived = new EventListeners<
cursors: ClientCursors[] (cursors: ClientCursors[]) => Promise<void>
) => Promise<void>)[] = []; >();
private isStopped = true; private isStopped = true;
private resolveDisconnectingPromise: null | (() => unknown) = null; private resolveDisconnectingPromise: null | (() => unknown) = null;
private reconnectTimeoutId: ReturnType<typeof setTimeout> | undefined; private reconnectTimeoutId: ReturnType<typeof setTimeout> | undefined;
private readonly outstandingPromises: Promise<unknown>[] = []; private readonly outstandingPromises: Promise<unknown>[] = [];
private webSocket: WebSocket | undefined; private webSocket: WebSocket | undefined;
private readonly webSocketFactoryImplementation: typeof globalThis.WebSocket; private readonly webSocketFactoryImplementation: typeof globalThis.WebSocket;
public constructor( public constructor(
private readonly deviceId: string, private readonly deviceId: string,
private readonly logger: Logger, private readonly logger: Logger,
private readonly settings: Settings, private readonly settings: Settings,
webSocketImplementation?: typeof globalThis.WebSocket webSocketImplementation?: typeof globalThis.WebSocket
) { ) {
if (webSocketImplementation) { if (webSocketImplementation) {
this.webSocketFactoryImplementation = webSocketImplementation; this.webSocketFactoryImplementation = webSocketImplementation;
} else { } else {
if ( if (
typeof globalThis !== "undefined" && typeof globalThis !== "undefined" &&
typeof globalThis.WebSocket === "undefined" typeof globalThis.WebSocket === "undefined"
) { ) {
// eslint-disable-next-line // eslint-disable-next-line
this.webSocketFactoryImplementation = require("ws"); // polyfill for WebSocket in Node.js this.webSocketFactoryImplementation = require("ws"); // polyfill for WebSocket in Node.js
} else { } else {
this.webSocketFactoryImplementation = WebSocket; this.webSocketFactoryImplementation = WebSocket;
} }
} }
} }
public get isWebSocketConnected(): boolean { public get isWebSocketConnected(): boolean {
return ( return (
this.webSocket?.readyState === this.webSocket?.readyState ===
this.webSocketFactoryImplementation.OPEN this.webSocketFactoryImplementation.OPEN
); );
} }
public addWebSocketStatusChangeListener( public start(): void {
listener: (isConnected: boolean) => unknown this.isStopped = false;
): void { this.initializeWebSocket();
this.webSocketStatusChangeListeners.push(listener); }
}
public addRemoteCursorsUpdateListener( public async stop(): Promise<void> {
listener: (cursors: ClientCursors[]) => Promise<void> const [promise, resolve] = createPromise();
): void { this.resolveDisconnectingPromise = resolve;
this.remoteCursorsUpdateListeners.push(listener);
}
public addRemoteVaultUpdateListener( this.isStopped = true;
listener: (update: WebSocketVaultUpdate) => Promise<void>
): void {
this.remoteVaultUpdateListeners.push(listener);
}
public start(): void { if (this.reconnectTimeoutId !== undefined) {
this.isStopped = false; clearTimeout(this.reconnectTimeoutId);
this.initializeWebSocket(); this.reconnectTimeoutId = undefined;
} }
public async stop(): Promise<void> { this.webSocket?.close(1000, "WebSocketManager has been stopped");
const [promise, resolve] = createPromise();
this.resolveDisconnectingPromise = resolve;
this.isStopped = true; // eslint-disable-next-line @typescript-eslint/init-declarations
let timeoutId: ReturnType<typeof setTimeout> | undefined;
const timeoutPromise = new Promise<void>((_, reject) => {
timeoutId = setTimeout(() => {
reject(
new Error(
`Timeout waiting for WebSocket to close after ${WEBSOCKET_DISCONNECT_TIMEOUT_IN_S} seconds`
)
);
}, WEBSOCKET_DISCONNECT_TIMEOUT_IN_S * 1000);
});
if (this.reconnectTimeoutId !== undefined) { try {
clearTimeout(this.reconnectTimeoutId); while (this.isWebSocketConnected) {
this.reconnectTimeoutId = undefined; await Promise.race([promise, timeoutPromise]);
} }
} catch (error) {
this.logger.error(
`Error while waiting for WebSocket to close: ${String(error)}`
);
// Force cleanup even if close didn't work
this.resolveDisconnectingPromise();
this.resolveDisconnectingPromise = null;
} finally {
// Clear timeout to prevent unhandled rejection
if (timeoutId !== undefined) {
clearTimeout(timeoutId);
}
}
this.webSocket?.close(1000, "WebSocketManager has been stopped"); await this.waitUntilFinished();
}
// eslint-disable-next-line @typescript-eslint/init-declarations public async waitUntilFinished(): Promise<void> {
let timeoutId: ReturnType<typeof setTimeout> | undefined; await awaitAll(this.outstandingPromises);
const timeoutPromise = new Promise<void>((_, reject) => { }
timeoutId = setTimeout(() => {
reject(
new Error(
`Timeout waiting for WebSocket to close after ${WEBSOCKET_DISCONNECT_TIMEOUT_IN_S} seconds`
)
);
}, WEBSOCKET_DISCONNECT_TIMEOUT_IN_S * 1000);
});
try { public sendHandshakeMessage(
while (this.isWebSocketConnected) { message: WebSocketClientMessage & { type: "handshake" }
await Promise.race([promise, timeoutPromise]); ): void {
} const { webSocket } = this;
} catch (error) { if (!webSocket) {
this.logger.error( throw new Error(
`Error while waiting for WebSocket to close: ${String(error)}` "WebSocket is not connected, cannot send handshake message"
); );
// Force cleanup even if close didn't work }
this.resolveDisconnectingPromise();
this.resolveDisconnectingPromise = null;
} finally {
// Clear timeout to prevent unhandled rejection
if (timeoutId !== undefined) {
clearTimeout(timeoutId);
}
}
await this.waitUntilFinished(); try {
} webSocket.send(JSON.stringify(message));
} catch (error) {
this.logger.error(
`Failed to send handshake message: ${String(error)}`
);
throw error;
}
}
public async waitUntilFinished(): Promise<void> { public updateLocalCursors(cursorPositions: CursorPositionFromClient): void {
await awaitAll(this.outstandingPromises); if (!this.isWebSocketConnected || !this.webSocket) {
} // A missing cursor update is fine, we can just skip it if needed
this.logger.warn(
"WebSocket is not connected, cannot send cursor positions"
);
return;
}
public sendHandshakeMessage( const message: WebSocketClientMessage = {
message: WebSocketClientMessage & { type: "handshake" } type: "cursorPositions",
): void { ...cursorPositions
const { webSocket } = this; };
if (!webSocket) {
throw new Error(
"WebSocket is not connected, cannot send handshake message"
);
}
try { try {
webSocket.send(JSON.stringify(message)); this.webSocket.send(JSON.stringify(message));
} catch (error) { this.logger.debug(
this.logger.error( `Sent cursor positions: ${JSON.stringify(cursorPositions)}`
`Failed to send handshake message: ${String(error)}` );
); } catch (error) {
throw error; this.logger.warn(
} `Failed to send cursor positions: ${String(error)}`
} );
}
}
public updateLocalCursors(cursorPositions: CursorPositionFromClient): void { private initializeWebSocket(): void {
if (!this.isWebSocketConnected || !this.webSocket) { // Clean up old WebSocket handlers to prevent race conditions
// A missing cursor update is fine, we can just skip it if needed if (this.webSocket) {
this.logger.warn( try {
"WebSocket is not connected, cannot send cursor positions" // Remove handlers to prevent them from firing after new connection
); this.webSocket.onopen = null;
return; this.webSocket.onclose = null;
} this.webSocket.onmessage = null;
this.webSocket.onerror = null;
this.webSocket.close();
} catch (e) {
this.logger.error(
`Failed to close previous WebSocket connection: ${e}`
);
}
}
const message: WebSocketClientMessage = { const wsUri = new URL(this.settings.getSettings().remoteUri);
type: "cursorPositions", wsUri.protocol = wsUri.protocol === "https" ? "wss" : "ws";
...cursorPositions wsUri.pathname = `/vaults/${this.settings.getSettings().vaultName}/ws`;
};
try { this.logger.info(`Connecting to WebSocket at ${wsUri.toString()}`);
this.webSocket.send(JSON.stringify(message));
this.logger.debug(
`Sent cursor positions: ${JSON.stringify(cursorPositions)}`
);
} catch (error) {
this.logger.warn(
`Failed to send cursor positions: ${String(error)}`
);
}
}
private initializeWebSocket(): void { this.webSocket = new this.webSocketFactoryImplementation(wsUri);
// Clean up old WebSocket handlers to prevent race conditions
if (this.webSocket) {
try {
// Remove handlers to prevent them from firing after new connection
this.webSocket.onopen = null;
this.webSocket.onclose = null;
this.webSocket.onmessage = null;
this.webSocket.onerror = null;
this.webSocket.close();
} catch (e) {
this.logger.error(
`Failed to close previous WebSocket connection: ${e}`
);
}
}
const wsUri = new URL(this.settings.getSettings().remoteUri); this.webSocket.onopen = (): void => {
wsUri.protocol = wsUri.protocol === "https" ? "wss" : "ws"; this.logger.info("WebSocket connection opened");
wsUri.pathname = `/vaults/${this.settings.getSettings().vaultName}/ws`; this.onWebSocketStatusChanged.trigger(true);
};
this.logger.info(`Connecting to WebSocket at ${wsUri.toString()}`); this.webSocket.onmessage = (event): void => {
try {
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
const message = JSON.parse(
event.data
) as WebSocketServerMessage;
this.webSocket = new this.webSocketFactoryImplementation(wsUri); // Track the message handling promise
const messageHandlingPromise = this.handleWebSocketMessage(
message
)
.catch((error: unknown) => {
this.logger.error(
`Error handling WebSocket message: ${String(error)}`
);
})
.finally(() => {
removeFromArray(
this.outstandingPromises,
messageHandlingPromise
);
});
this.webSocket.onopen = (): void => { void this.outstandingPromises.push(messageHandlingPromise); // ignore the returned promise
this.logger.info("WebSocket connection opened"); } catch (error) {
this.webSocketStatusChangeListeners.forEach((listener) => this.logger.error(
listener(true) `Error parsing WebSocket message: ${String(error)}`
); );
}; }
};
this.webSocket.onmessage = (event): void => { this.webSocket.onclose = (event): void => {
try { this.logger.warn(
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion `WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})`
const message = JSON.parse( );
event.data this.onWebSocketStatusChanged.trigger(false);
) as WebSocketServerMessage;
// Track the message handling promise if (this.isStopped) {
const messageHandlingPromise = this.handleWebSocketMessage( this.resolveDisconnectingPromise?.();
message this.resolveDisconnectingPromise = null;
) } else {
.catch((error: unknown) => { this.reconnectTimeoutId = setTimeout(() => {
this.logger.error( this.reconnectTimeoutId = undefined;
`Error handling WebSocket message: ${String(error)}` this.initializeWebSocket();
); }, this.settings.getSettings().webSocketRetryIntervalMs);
}) }
.finally(() => { };
removeFromArray( }
this.outstandingPromises,
messageHandlingPromise
);
});
void this.outstandingPromises.push(messageHandlingPromise); // ignore the returned promise private async handleWebSocketMessage(
} catch (error) { message: WebSocketServerMessage
this.logger.error( ): Promise<void> {
`Error parsing WebSocket message: ${String(error)}` if (message.type === "vaultUpdate") {
); await this.onRemoteVaultUpdateReceived.triggerAsync(message);
}
};
this.webSocket.onclose = (event): void => { // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
this.logger.warn( } else if (message.type === "cursorPositions") {
`WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})` this.logger.debug(
); `Received cursor positions for ${JSON.stringify(message.clients)}`
this.webSocketStatusChangeListeners.forEach((listener) => );
listener(false)
);
if (this.isStopped) { await this.onRemoteCursorsUpdateReceived.triggerAsync(
this.resolveDisconnectingPromise?.(); message.clients
this.resolveDisconnectingPromise = null; );
} else { } else {
this.reconnectTimeoutId = setTimeout(() => { this.logger.warn(
this.reconnectTimeoutId = undefined; `Received unknown message type: ${JSON.stringify(message)}`
this.initializeWebSocket(); );
}, this.settings.getSettings().webSocketRetryIntervalMs); }
} }
};
}
private async handleWebSocketMessage(
message: WebSocketServerMessage
): Promise<void> {
if (message.type === "vaultUpdate") {
await awaitAll(
this.remoteVaultUpdateListeners.map(async (listener) => {
await listener(message).catch((error: unknown) => {
this.logger.error(
`Error in vault update listener: ${String(error)}`
);
});
})
);
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
} else if (message.type === "cursorPositions") {
this.logger.debug(
`Received cursor positions for ${JSON.stringify(message.clients)}`
);
await awaitAll(
this.remoteCursorsUpdateListeners.map(async (listener) => {
await listener(message.clients).catch((error: unknown) => {
this.logger.error(
`Error in cursor positions listener: ${String(error)}`
);
});
})
);
} else {
this.logger.warn(
`Received unknown message type: ${JSON.stringify(message)}`
);
}
}
} }

980
frontend/sync-client/src/sync-client.ts
View file

@ -26,495 +26,497 @@ import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache"
import { setUpTelemetry } from "./utils/set-up-telemetry"; import { setUpTelemetry } from "./utils/set-up-telemetry";
import { DIFF_CACHE_SIZE_MB } from "./consts"; import { DIFF_CACHE_SIZE_MB } from "./consts";
import { ServerConfig } from "./services/server-config"; import { ServerConfig } from "./services/server-config";
import { EventListeners } from "./utils/data-structures/event-listeners";
export class SyncClient { export class SyncClient {
private hasStartedOfflineSync = false; private hasStartedOfflineSync = false;
private hasFinishedOfflineSync = false; private hasFinishedOfflineSync = false;
private hasStarted = false; private hasStarted = false;
private hasBeenDestroyed = false; private hasBeenDestroyed = false;
private unloadTelemetry?: () => void; private unloadTelemetry?: () => void;
private constructor( private constructor(
private readonly history: SyncHistory, private readonly history: SyncHistory,
private readonly settings: Settings, private readonly settings: Settings,
private readonly database: Database, private readonly database: Database,
private readonly syncer: Syncer, private readonly syncer: Syncer,
private readonly webSocketManager: WebSocketManager, private readonly webSocketManager: WebSocketManager,
public readonly logger: Logger, public readonly logger: Logger,
private readonly fetchController: FetchController, private readonly fetchController: FetchController,
private readonly cursorTracker: CursorTracker, private readonly cursorTracker: CursorTracker,
private readonly fileChangeNotifier: FileChangeNotifier, private readonly fileChangeNotifier: FileChangeNotifier,
private readonly contentCache: FixedSizeDocumentCache, private readonly contentCache: FixedSizeDocumentCache,
private readonly fileOperations: FileOperations, private readonly fileOperations: FileOperations,
private readonly serverConfig: ServerConfig, private readonly serverConfig: ServerConfig,
private readonly persistence: PersistenceProvider< private readonly persistence: PersistenceProvider<
Partial<{ Partial<{
settings: Partial<SyncSettings>; settings: Partial<SyncSettings>;
database: Partial<StoredDatabase>; database: Partial<StoredDatabase>;
}> }>
> >
) {} ) { }
public get documentCount(): number { public get documentCount(): number {
return this.database.length; return this.database.length;
} }
public get isWebSocketConnected(): boolean { public get isWebSocketConnected(): boolean {
return this.webSocketManager.isWebSocketConnected; return this.webSocketManager.isWebSocketConnected;
} }
public static async create({ public static async create({
fs, fs,
persistence, persistence,
fetch, fetch,
webSocket, webSocket,
nativeLineEndings = "\n" nativeLineEndings = "\n"
}: { }: {
fs: FileSystemOperations; fs: FileSystemOperations;
persistence: PersistenceProvider< persistence: PersistenceProvider<
Partial<{ Partial<{
settings: Partial<SyncSettings>; settings: Partial<SyncSettings>;
database: Partial<StoredDatabase>; database: Partial<StoredDatabase>;
}> }>
>; >;
fetch?: typeof globalThis.fetch; fetch?: typeof globalThis.fetch;
webSocket?: typeof globalThis.WebSocket; webSocket?: typeof globalThis.WebSocket;
nativeLineEndings?: string; nativeLineEndings?: string;
}): Promise<SyncClient> { }): Promise<SyncClient> {
const logger = new Logger(); const logger = new Logger();
const deviceId = createClientId(); const deviceId = createClientId();
logger.info(`Creating SyncClient with client id ${deviceId}`); logger.info(`Creating SyncClient with client id ${deviceId}`);
const history = new SyncHistory(logger); const history = new SyncHistory(logger);
let state = (await persistence.load()) ?? { let state = (await persistence.load()) ?? {
settings: undefined, settings: undefined,
database: undefined database: undefined
}; };
const settings = new Settings( const settings = new Settings(
logger, logger,
state.settings, state.settings,
async (data): Promise<void> => { async (data): Promise<void> => {
state = { ...state, settings: data }; state = { ...state, settings: data };
// we're not rate-limiting settings saves as (1) we need to initialise the settings to know the rate limit // we're not rate-limiting settings saves as (1) we need to initialise the settings to know the rate limit
// and (2) settings changes are infrequent enough that rate-limiting is not necessary // and (2) settings changes are infrequent enough that rate-limiting is not necessary
await persistence.save(state); await persistence.save(state);
} }
); );
const rateLimitedSave = rateLimit( const rateLimitedSave = rateLimit(
persistence.save, persistence.save,
() => settings.getSettings().minimumSaveIntervalMs () => settings.getSettings().minimumSaveIntervalMs
); );
const database = new Database( const database = new Database(
logger, logger,
state.database, state.database,
async (data): Promise<void> => { async (data): Promise<void> => {
state = { ...state, database: data }; state = { ...state, database: data };
await rateLimitedSave(state); await rateLimitedSave(state);
} }
); );
const fetchController = new FetchController( const fetchController = new FetchController(
settings.getSettings().isSyncEnabled, settings.getSettings().isSyncEnabled,
logger logger
); );
settings.addOnSettingsChangeListener((newSettings, oldSettings) => { settings.onSettingsChanged.add((newSettings, oldSettings) => {
if (oldSettings.isSyncEnabled != newSettings.isSyncEnabled) { if (oldSettings.isSyncEnabled != newSettings.isSyncEnabled) {
fetchController.canFetch = newSettings.isSyncEnabled; fetchController.canFetch = newSettings.isSyncEnabled;
} }
}); });
const syncService = new SyncService( const syncService = new SyncService(
deviceId, deviceId,
fetchController, fetchController,
settings, settings,
logger, logger,
fetch fetch
); );
const serverConfig = new ServerConfig(syncService); const serverConfig = new ServerConfig(syncService);
const fileOperations = new FileOperations( const fileOperations = new FileOperations(
logger, logger,
database, database,
fs, fs,
serverConfig, serverConfig,
nativeLineEndings nativeLineEndings
); );
const contentCache = new FixedSizeDocumentCache( const contentCache = new FixedSizeDocumentCache(
1024 * 1024 * DIFF_CACHE_SIZE_MB 1024 * 1024 * DIFF_CACHE_SIZE_MB
); );
const unrestrictedSyncer = new UnrestrictedSyncer( const unrestrictedSyncer = new UnrestrictedSyncer(
logger, logger,
database, database,
settings, settings,
syncService, syncService,
fileOperations, fileOperations,
history, history,
contentCache, contentCache,
serverConfig serverConfig
); );
const webSocketManager = new WebSocketManager( const webSocketManager = new WebSocketManager(
deviceId, deviceId,
logger, logger,
settings, settings,
webSocket webSocket
); );
const syncer = new Syncer( const syncer = new Syncer(
deviceId, deviceId,
logger, logger,
database, database,
settings, settings,
syncService, syncService,
webSocketManager, webSocketManager,
fileOperations, fileOperations,
unrestrictedSyncer unrestrictedSyncer
); );
const fileChangeNotifier = new FileChangeNotifier(); const fileChangeNotifier = new FileChangeNotifier();
const cursorTracker = new CursorTracker( const cursorTracker = new CursorTracker(
database, database,
webSocketManager, webSocketManager,
fileOperations, fileOperations,
fileChangeNotifier fileChangeNotifier
); );
const client = new SyncClient( const client = new SyncClient(
history, history,
settings, settings,
database, database,
syncer, syncer,
webSocketManager, webSocketManager,
logger, logger,
fetchController, fetchController,
cursorTracker, cursorTracker,
fileChangeNotifier, fileChangeNotifier,
contentCache, contentCache,
fileOperations, fileOperations,
serverConfig, serverConfig,
persistence persistence
); );
logger.info("SyncClient created successfully"); logger.info("SyncClient created successfully");
return client; return client;
} }
public async start(): Promise<void> { public async start(): Promise<void> {
this.checkIfDestroyed("start"); this.checkIfDestroyed("start");
if (this.hasStarted) { if (this.hasStarted) {
throw new Error("SyncClient has already been started"); throw new Error("SyncClient has already been started");
} }
this.hasStarted = true; this.hasStarted = true;
if ( if (
!this.unloadTelemetry && !this.unloadTelemetry &&
this.settings.getSettings().enableTelemetry this.settings.getSettings().enableTelemetry
) { ) {
this.unloadTelemetry = setUpTelemetry(); this.unloadTelemetry = setUpTelemetry();
} }
this.logger.addOnMessageListener((log): void => { this.logger.onLogEmitted.add((log): void => {
if (log.level === LogLevel.ERROR && Sentry.isInitialized()) { if (log.level === LogLevel.ERROR && Sentry.isInitialized()) {
Sentry.captureMessage(log.message); Sentry.captureMessage(log.message);
} }
}); });
this.settings.addOnSettingsChangeListener( this.settings.onSettingsChanged.add(
this.onSettingsChange.bind(this) this.onSettingsChange.bind(this)
); );
if (this.settings.getSettings().isSyncEnabled) { if (this.settings.getSettings().isSyncEnabled) {
this.logger.info("Starting SyncClient"); this.logger.info("Starting SyncClient");
await this.startSyncing(); await this.startSyncing();
this.logger.info("SyncClient has successfully started"); this.logger.info("SyncClient has successfully started");
} }
} }
/** /**
* Reload settings from disk overriding current in-memory settings. * Reload settings from disk overriding current in-memory settings.
* Missing values will be filled in from DEFAULT_SETTINGS rather than * Missing values will be filled in from DEFAULT_SETTINGS rather than
* retaining current in-memory settings. * retaining current in-memory settings.
*/ */
public async reloadSettings(): Promise<void> { public async reloadSettings(): Promise<void> {
this.checkIfDestroyed("reloadSettings"); this.checkIfDestroyed("reloadSettings");
const state = (await this.persistence.load()) ?? { const state = (await this.persistence.load()) ?? {
settings: undefined settings: undefined
}; };
const settings = { const settings = {
...DEFAULT_SETTINGS, ...DEFAULT_SETTINGS,
...(state.settings ?? {}) ...(state.settings ?? {})
}; };
await this.setSettings(settings); await this.setSettings(settings);
} }
public async checkConnection(): Promise<NetworkConnectionStatus> { public async checkConnection(): Promise<NetworkConnectionStatus> {
this.checkIfDestroyed("checkConnection"); this.checkIfDestroyed("checkConnection");
const server = await this.serverConfig.checkConnection(true); const server = await this.serverConfig.checkConnection(true);
return { return {
isSuccessful: server.isSuccessful, isSuccessful: server.isSuccessful,
serverMessage: server.message, serverMessage: server.message,
isWebSocketConnected: this.webSocketManager.isWebSocketConnected isWebSocketConnected: this.webSocketManager.isWebSocketConnected
}; };
} }
public getHistoryEntries(): readonly HistoryEntry[] { public getHistoryEntries(): readonly HistoryEntry[] {
return this.history.entries; return this.history.entries;
} }
public addSyncHistoryUpdateListener( /**
listener: (stats: HistoryStats) => unknown * Wait for the in-flight operations to finish, reset all tracking,
): void { * and the local database but retain the settings.
this.checkIfDestroyed("addSyncHistoryUpdateListener"); * The SyncClient can be used again after calling this method.
*/
this.history.addSyncHistoryUpdateListener(listener); public async reset(): Promise<void> {
} this.checkIfDestroyed("reset");
/** this.logger.info(
* Wait for the in-flight operations to finish, reset all tracking, "Stopping SyncClient to apply changed connection settings"
* and the local database but retain the settings. );
* The SyncClient can be used again after calling this method. await this.pause();
*/
public async reset(): Promise<void> { // clear all local state
this.checkIfDestroyed("reset"); this.logger.info("Resetting SyncClient's local state");
this.database.reset();
this.logger.info( await this.database.save(); // ensure the new database reads as empty
"Stopping SyncClient to apply changed connection settings" this.resetInMemoryState();
); this.hasStartedOfflineSync = false;
await this.pause(); this.hasFinishedOfflineSync = false;
this.serverConfig.reset();
// clear all local state
this.logger.info("Resetting SyncClient's local state"); await this.startSyncing();
this.database.reset(); }
await this.database.save(); // ensure the new database reads as empty
this.resetInMemoryState(); public getSettings(): SyncSettings {
this.hasStartedOfflineSync = false; return this.settings.getSettings();
this.hasFinishedOfflineSync = false; }
this.serverConfig.reset();
public async setSetting<T extends keyof SyncSettings>(
await this.startSyncing(); key: T,
} value: SyncSettings[T]
): Promise<void> {
public getSettings(): SyncSettings { this.checkIfDestroyed("setSetting");
return this.settings.getSettings();
} await this.settings.setSetting(key, value);
}
public async setSetting<T extends keyof SyncSettings>(
key: T, public async setSettings(value: Partial<SyncSettings>): Promise<void> {
value: SyncSettings[T] this.checkIfDestroyed("setSettings");
): Promise<void> {
this.checkIfDestroyed("setSetting"); await this.settings.setSettings(value);
}
await this.settings.setSetting(key, value);
} public get onSyncHistoryUpdated(): EventListeners<
(stats: HistoryStats) => unknown
public async setSettings(value: Partial<SyncSettings>): Promise<void> { > {
this.checkIfDestroyed("setSettings"); this.checkIfDestroyed("onSyncHistoryUpdated getter");
return this.history.onHistoryUpdated;
await this.settings.setSettings(value); }
}
public addOnSettingsChangeListener(
listener: (settings: SyncSettings, oldSettings: SyncSettings) => unknown
): void { public get onSettingsChanged(): EventListeners<
this.checkIfDestroyed("addOnSettingsChangeListener"); (newSettings: SyncSettings, oldSettings: SyncSettings) => unknown
> {
this.settings.addOnSettingsChangeListener(listener); this.checkIfDestroyed("onSettingsChanged getter");
} return this.settings.onSettingsChanged;
}
public addRemainingSyncOperationsListener(
listener: (remainingOperations: number) => unknown public get onRemainingOperationsCountChanged(): EventListeners<
): void { (remainingOperationsCount: number) => unknown
this.checkIfDestroyed("addRemainingSyncOperationsListener"); > {
this.checkIfDestroyed("onRemainingOperationsCountChanged getter");
this.syncer.addRemainingOperationsListener(listener); return this.syncer.onRemainingOperationsCountChanged;
} }
public addWebSocketStatusChangeListener(listener: () => unknown): void { public get onWebSocketStatusChanged(): EventListeners<
this.checkIfDestroyed("addWebSocketStatusChangeListener"); (isConnected: boolean) => unknown
> {
this.webSocketManager.addWebSocketStatusChangeListener(listener); this.checkIfDestroyed("onWebSocketStatusChanged getter");
} return this.webSocketManager.onWebSocketStatusChanged;
}
public async syncLocallyCreatedFile(
relativePath: RelativePath public async syncLocallyCreatedFile(
): Promise<void> { relativePath: RelativePath
this.checkIfDestroyed("syncLocallyCreatedFile"); ): Promise<void> {
this.checkIfDestroyed("syncLocallyCreatedFile");
this.fileChangeNotifier.notifyOfFileChange(relativePath);
return this.syncer.syncLocallyCreatedFile(relativePath); this.fileChangeNotifier.notifyOfFileChange(relativePath);
} return this.syncer.syncLocallyCreatedFile(relativePath);
}
public async syncLocallyDeletedFile(
relativePath: RelativePath public async syncLocallyDeletedFile(
): Promise<void> { relativePath: RelativePath
this.checkIfDestroyed("syncLocallyDeletedFile"); ): Promise<void> {
this.checkIfDestroyed("syncLocallyDeletedFile");
this.fileChangeNotifier.notifyOfFileChange(relativePath);
return this.syncer.syncLocallyDeletedFile(relativePath); this.fileChangeNotifier.notifyOfFileChange(relativePath);
} return this.syncer.syncLocallyDeletedFile(relativePath);
}
public async syncLocallyUpdatedFile({
oldPath, public async syncLocallyUpdatedFile({
relativePath oldPath,
}: { relativePath
oldPath?: RelativePath; }: {
relativePath: RelativePath; oldPath?: RelativePath;
}): Promise<void> { relativePath: RelativePath;
this.checkIfDestroyed("syncLocallyUpdatedFile"); }): Promise<void> {
this.checkIfDestroyed("syncLocallyUpdatedFile");
this.fileChangeNotifier.notifyOfFileChange(relativePath);
return this.syncer.syncLocallyUpdatedFile({ this.fileChangeNotifier.notifyOfFileChange(relativePath);
oldPath, return this.syncer.syncLocallyUpdatedFile({
relativePath oldPath,
}); relativePath
} });
}
public getDocumentSyncingStatus(
relativePath: RelativePath public getDocumentSyncingStatus(
): DocumentSyncStatus { relativePath: RelativePath
this.checkIfDestroyed("getDocumentSyncingStatus"); ): DocumentSyncStatus {
this.checkIfDestroyed("getDocumentSyncingStatus");
if (!this.settings.getSettings().isSyncEnabled) {
return DocumentSyncStatus.SYNCING_IS_DISABLED; if (!this.settings.getSettings().isSyncEnabled) {
} return DocumentSyncStatus.SYNCING_IS_DISABLED;
}
if (!this.syncer.isFirstSyncComplete || !this.hasFinishedOfflineSync) {
return DocumentSyncStatus.SYNCING; if (!this.syncer.isFirstSyncComplete || !this.hasFinishedOfflineSync) {
} return DocumentSyncStatus.SYNCING;
}
const document =
this.database.getLatestDocumentByRelativePath(relativePath); const document =
if (document === undefined) { this.database.getLatestDocumentByRelativePath(relativePath);
return DocumentSyncStatus.SYNCING; if (document === undefined) {
} return DocumentSyncStatus.SYNCING;
return document.updates.length > 0 }
? DocumentSyncStatus.SYNCING return document.updates.length > 0
: DocumentSyncStatus.UP_TO_DATE; ? DocumentSyncStatus.SYNCING
} : DocumentSyncStatus.UP_TO_DATE;
}
public async updateLocalCursors(
documentToCursors: Record<RelativePath, CursorSpan[]> public async updateLocalCursors(
): Promise<void> { documentToCursors: Record<RelativePath, CursorSpan[]>
this.checkIfDestroyed("updateLocalCursors"); ): Promise<void> {
this.checkIfDestroyed("updateLocalCursors");
await this.cursorTracker.sendLocalCursorsToServer(documentToCursors);
} await this.cursorTracker.sendLocalCursorsToServer(documentToCursors);
}
public addRemoteCursorsUpdateListener(
listener: (cursors: MaybeOutdatedClientCursors[]) => unknown
): void { public get onRemoteCursorsUpdated(): EventListeners<
this.checkIfDestroyed("addRemoteCursorsUpdateListener"); (cursors: MaybeOutdatedClientCursors[]) => unknown
> {
this.cursorTracker.addRemoteCursorsUpdateListener(listener); this.checkIfDestroyed("onRemoteCursorsUpdated getter");
} return this.cursorTracker.onRemoteCursorsUpdated;
}
public async waitUntilFinished(): Promise<void> {
this.checkIfDestroyed("waitUntilIdle"); public async waitUntilFinished(): Promise<void> {
await this.syncer.waitUntilFinished(); this.checkIfDestroyed("waitUntilIdle");
await this.webSocketManager.waitUntilFinished(); await this.syncer.waitUntilFinished();
await this.database.save(); // flush all changes to disk await this.webSocketManager.waitUntilFinished();
} await this.database.save(); // flush all changes to disk
}
/**
* Completely destroy the SyncClient, cancelling all in-progress operations. /**
* After calling this method, the SyncClient cannot be used again. * Completely destroy the SyncClient, cancelling all in-progress operations.
*/ * After calling this method, the SyncClient cannot be used again.
public async destroy(): Promise<void> { */
this.checkIfDestroyed("destroy"); public async destroy(): Promise<void> {
this.checkIfDestroyed("destroy");
// cancel everything that's in progress
await this.pause(); // cancel everything that's in progress
await this.pause();
this.hasBeenDestroyed = true;
this.hasBeenDestroyed = true;
this.resetInMemoryState();
this.resetInMemoryState();
this.logger.info("SyncClient has been successfully disposed");
this.logger.info("SyncClient has been successfully disposed");
this.unloadTelemetry?.();
} this.unloadTelemetry?.();
}
private async startSyncing(): Promise<void> {
this.checkIfDestroyed("startSyncing"); private async startSyncing(): Promise<void> {
this.fetchController.finishReset(); this.checkIfDestroyed("startSyncing");
this.fetchController.finishReset();
await this.serverConfig.initialize();
this.webSocketManager.start(); await this.serverConfig.initialize();
this.webSocketManager.start();
if (!this.hasStartedOfflineSync) {
this.hasStartedOfflineSync = true; if (!this.hasStartedOfflineSync) {
await this.syncer.scheduleSyncForOfflineChanges(); this.hasStartedOfflineSync = true;
} await this.syncer.scheduleSyncForOfflineChanges();
}
this.hasFinishedOfflineSync = true;
} this.hasFinishedOfflineSync = true;
}
private async pause(): Promise<void> {
this.fetchController.startReset(); private async pause(): Promise<void> {
await this.webSocketManager.stop(); this.fetchController.startReset();
await this.waitUntilFinished(); await this.webSocketManager.stop();
} await this.waitUntilFinished();
}
private resetInMemoryState(): void {
this.history.reset(); private resetInMemoryState(): void {
this.contentCache.reset(); this.history.reset();
// don't reset the logger this.contentCache.reset();
this.cursorTracker.reset(); // don't reset the logger
this.syncer.reset(); this.cursorTracker.reset();
this.fileOperations.reset(); this.syncer.reset();
} this.fileOperations.reset();
}
private async onSettingsChange(
newSettings: SyncSettings, private async onSettingsChange(
oldSettings: SyncSettings newSettings: SyncSettings,
): Promise<void> { oldSettings: SyncSettings
this.checkIfDestroyed("onSettingsChange"); ): Promise<void> {
this.checkIfDestroyed("onSettingsChange");
if (
newSettings.vaultName !== oldSettings.vaultName || if (
newSettings.remoteUri !== oldSettings.remoteUri newSettings.vaultName !== oldSettings.vaultName ||
) { newSettings.remoteUri !== oldSettings.remoteUri
await this.reset(); ) {
} await this.reset();
}
if (newSettings.isSyncEnabled !== oldSettings.isSyncEnabled) {
if (newSettings.isSyncEnabled) { if (newSettings.isSyncEnabled !== oldSettings.isSyncEnabled) {
await this.startSyncing(); if (newSettings.isSyncEnabled) {
} else { await this.startSyncing();
await this.pause(); } else {
} await this.pause();
} }
}
if (newSettings.diffCacheSizeMB !== oldSettings.diffCacheSizeMB) {
this.contentCache.resize(newSettings.diffCacheSizeMB * 1024 * 1024); if (newSettings.diffCacheSizeMB !== oldSettings.diffCacheSizeMB) {
} this.contentCache.resize(newSettings.diffCacheSizeMB * 1024 * 1024);
}
if (newSettings.enableTelemetry !== oldSettings.enableTelemetry) {
if (newSettings.enableTelemetry) { if (newSettings.enableTelemetry !== oldSettings.enableTelemetry) {
this.unloadTelemetry = setUpTelemetry(); if (newSettings.enableTelemetry) {
} else { this.unloadTelemetry = setUpTelemetry();
this.unloadTelemetry?.(); } else {
} this.unloadTelemetry?.();
} }
} }
}
private checkIfDestroyed(origin: string): void {
if (this.hasBeenDestroyed) { private checkIfDestroyed(origin: string): void {
throw new Error( if (this.hasBeenDestroyed) {
`SyncClient has been destroyed and can no longer be used; called from ${origin}` throw new Error(
); `SyncClient has been destroyed and can no longer be used; called from ${origin}`
} );
} }
}
} }

416
frontend/sync-client/src/sync-operations/cursor-tracker.ts
View file

@ -9,252 +9,252 @@ import { DocumentUpToDateness } from "../types/document-up-to-dateness";
import { hash } from "../utils/hash"; import { hash } from "../utils/hash";
import type { FileChangeNotifier } from "./file-change-notifier"; import type { FileChangeNotifier } from "./file-change-notifier";
import { Lock } from "../utils/data-structures/locks"; import { Lock } from "../utils/data-structures/locks";
import { EventListeners } from "../utils/data-structures/event-listeners";
// Cursor positions are updated separately from documents. However, a given cursor position is only // 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 // 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 // known remote cursor positions, and for each document, tries to return the latest cursor positions that are
// not from the future. // not from the future.
export class CursorTracker { export class CursorTracker {
private readonly updateLock = new Lock(); private readonly updateLock = new Lock();
private knownRemoteCursors: (ClientCursors & { // The returned position may be accurate, if it matches the document version, or outdated, in which case
upToDateness: DocumentUpToDateness; // the client has to heuristically guess it's current position based on the local edits.
})[] = []; public readonly onRemoteCursorsUpdated = new EventListeners<
(cursors: MaybeOutdatedClientCursors[]) => unknown
>();
private lastLocalCursorState: DocumentWithCursors[] = []; private knownRemoteCursors: (ClientCursors & {
private lastLocalCursorStateWithoutDirtyDocuments: DocumentWithCursors[] = upToDateness: DocumentUpToDateness;
[]; })[] = [];
public constructor( private lastLocalCursorState: DocumentWithCursors[] = [];
private readonly database: Database, private lastLocalCursorStateWithoutDirtyDocuments: DocumentWithCursors[] =
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) => public constructor(
client.documentsWithCursors.every( private readonly database: Database,
(doc) => doc.vault_update_id != null private readonly webSocketManager: WebSocketManager,
) private readonly fileOperations: FileOperations,
)) { private readonly fileChangeNotifier: FileChangeNotifier
updatedKnownRemoteCursors.push({ ) {
...cursor, this.webSocketManager.onRemoteCursorsUpdateReceived.add(
upToDateness: async (clientCursors) => {
await this.getDocumentsUpToDateness(cursor) 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)
);
this.knownRemoteCursors = updatedKnownRemoteCursors; 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.fileChangeNotifier.addFileChangeListener(async (relativePath) => this.knownRemoteCursors = updatedKnownRemoteCursors;
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. this.onRemoteCursorsUpdated.trigger(
/// Can be called frequently as it only emits an event this.getRelevantAndPruneKnownClientCursors()
/// 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) { this.fileChangeNotifier.onFileChanged.add(async (relativePath) =>
continue; // Let's wait for the file to be created before sending cursors 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);
}
}
})
);
}
if (!record.metadata) { /// Update the local cursors for the given documents.
continue; // this is a new document, no need to sync the cursors /// 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[] = [];
documentsWithCursors.push({ for (const [relativePath, cursors] of Object.entries(
relative_path: relativePath, documentToCursors
document_id: record.documentId, )) {
vault_update_id: record.metadata.parentVersionId, const record =
cursors: cursors.map(({ start, end }) => ({ this.database.getLatestDocumentByRelativePath(relativePath);
start: Math.min(start, end),
end: Math.max(start, end)
})) // the client might send directional selections
});
}
if ( if (!record) {
JSON.stringify(this.lastLocalCursorState) === continue; // Let's wait for the file to be created before sending cursors
JSON.stringify(documentsWithCursors) }
) {
// Caching step to avoid reading the edited files all the time
return;
}
this.lastLocalCursorState = documentsWithCursors;
for (const doc of documentsWithCursors) { if (!record.metadata) {
const readContent = await this.fileOperations.read( continue; // this is a new document, no need to sync the cursors
doc.relative_path }
);
const record = this.database.getLatestDocumentByRelativePath(
doc.relative_path
);
if (record?.metadata?.hash !== hash(readContent)) {
doc.vault_update_id = null;
}
}
if ( documentsWithCursors.push({
JSON.stringify(this.lastLocalCursorStateWithoutDirtyDocuments) === relative_path: relativePath,
JSON.stringify(documentsWithCursors) document_id: record.documentId,
) { vault_update_id: record.metadata.parentVersionId,
return; cursors: cursors.map(({ start, end }) => ({
} start: Math.min(start, end),
end: Math.max(start, end)
})) // the client might send directional selections
});
}
this.lastLocalCursorStateWithoutDirtyDocuments = documentsWithCursors; if (
JSON.stringify(this.lastLocalCursorState) ===
JSON.stringify(documentsWithCursors)
) {
// Caching step to avoid reading the edited files all the time
return;
}
this.lastLocalCursorState = documentsWithCursors;
this.webSocketManager.updateLocalCursors({ 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;
}
}
// The returned position may be accurate, if it matches the document version, or outdated, in which case if (
// the client has to heuristically guess it's current position based on the local edits. JSON.stringify(this.lastLocalCursorStateWithoutDirtyDocuments) ===
public addRemoteCursorsUpdateListener( JSON.stringify(documentsWithCursors)
listener: (cursors: MaybeOutdatedClientCursors[]) => unknown ) {
): void { return;
// 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())
);
});
}
public reset(): void { this.lastLocalCursorStateWithoutDirtyDocuments = documentsWithCursors;
this.knownRemoteCursors = [];
this.lastLocalCursorState = [];
this.lastLocalCursorStateWithoutDirtyDocuments = [];
this.updateLock.reset();
}
private getRelevantAndPruneKnownClientCursors(): MaybeOutdatedClientCursors[] { this.webSocketManager.updateLocalCursors({ documentsWithCursors });
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) { public reset(): void {
continue; this.knownRemoteCursors = [];
} this.lastLocalCursorState = [];
this.lastLocalCursorStateWithoutDirtyDocuments = [];
this.updateLock.reset();
}
result.push({ private getRelevantAndPruneKnownClientCursors(): MaybeOutdatedClientCursors[] {
...clientCursors, const result: MaybeOutdatedClientCursors[] = [];
isOutdated: const included = new Set<string>();
clientCursors.upToDateness === DocumentUpToDateness.Prior
});
included.add(clientCursors.deviceId); const relevantCursors = [];
relevantCursors.unshift(clientCursors); // to reverse order back to normal for (const clientCursors of [...this.knownRemoteCursors].reverse()) {
} if (included.has(clientCursors.deviceId)) {
continue;
}
this.knownRemoteCursors = relevantCursors; if (clientCursors.upToDateness === DocumentUpToDateness.Later) {
continue;
}
return result; result.push({
} ...clientCursors,
isOutdated:
clientCursors.upToDateness === DocumentUpToDateness.Prior
});
// We store up-to-dateness on a per-client basis to simplify the implementation. included.add(clientCursors.deviceId);
// An individual client won't have too many documents open at once, so this is a reasonable trade-off. relevantCursors.unshift(clientCursors); // to reverse order back to normal
private async getDocumentsUpToDateness( }
clientCursor: ClientCursors
): Promise<DocumentUpToDateness> {
const results = [];
for (const document of clientCursor.documentsWithCursors) {
results.push(await this.getDocumentUpToDateness(document));
}
if ( this.knownRemoteCursors = relevantCursors;
results.every((result) => result === DocumentUpToDateness.UpToDate)
) {
return DocumentUpToDateness.UpToDate;
}
if ( return result;
results.every( }
(result) =>
result === DocumentUpToDateness.UpToDate ||
result === DocumentUpToDateness.Prior
)
) {
return DocumentUpToDateness.Prior;
}
return DocumentUpToDateness.Later; // 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));
}
private async getDocumentUpToDateness( if (
document: DocumentWithCursors results.every((result) => result === DocumentUpToDateness.UpToDate)
): Promise<DocumentUpToDateness> { ) {
const record = this.database.getLatestDocumentByRelativePath( return DocumentUpToDateness.UpToDate;
document.relative_path }
);
if (!record) { if (
// the document of the cursor must be from the future results.every(
return DocumentUpToDateness.Later; (result) =>
} result === DocumentUpToDateness.UpToDate ||
result === DocumentUpToDateness.Prior
)
) {
return DocumentUpToDateness.Prior;
}
if ( return DocumentUpToDateness.Later;
(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( private async getDocumentUpToDateness(
document.relative_path document: DocumentWithCursors
); ): Promise<DocumentUpToDateness> {
const record = this.database.getLatestDocumentByRelativePath(
document.relative_path
);
return this.database.getLatestDocumentByRelativePath( if (!record) {
document.relative_path // the document of the cursor must be from the future
)?.metadata?.hash === hash(currentContent) return DocumentUpToDateness.Later;
? DocumentUpToDateness.UpToDate }
: DocumentUpToDateness.Prior;
} 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;
}
} }

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

@ -1,22 +1,12 @@
import type { RelativePath } from "../persistence/database"; import type { RelativePath } from "../persistence/database";
import { removeFromArray } from "../utils/remove-from-array"; import { EventListeners } from "../utils/data-structures/event-listeners";
export class FileChangeNotifier { export class FileChangeNotifier {
private readonly listeners: ((filePath: RelativePath) => unknown)[] = []; public readonly onFileChanged = new EventListeners<
(filePath: RelativePath) => unknown
>();
public addFileChangeListener( public notifyOfFileChange(filePath: RelativePath): void {
listener: (filePath: RelativePath) => unknown this.onFileChanged.trigger(filePath);
): void { }
this.listeners.push(listener);
}
public removeFileChangeListener(
listener: (filePath: RelativePath) => unknown
): void {
removeFromArray(this.listeners, listener);
}
public notifyOfFileChange(filePath: RelativePath): void {
this.listeners.forEach((listener) => listener(filePath));
}
} }

997
frontend/sync-client/src/sync-operations/syncer.ts
View file

File diff suppressed because it is too large Load diff

1021
frontend/sync-client/src/sync-operations/unrestricted-syncer.ts
View file

File diff suppressed because it is too large Load diff

113
frontend/sync-client/src/tracing/logger.ts
View file

@ -1,87 +1,72 @@
import { MAX_LOG_MESSAGE_COUNT } from "../consts"; import { MAX_LOG_MESSAGE_COUNT } from "../consts";
import { removeFromArray } from "../utils/remove-from-array"; import { EventListeners } from "../utils/data-structures/event-listeners";
export enum LogLevel { export enum LogLevel {
DEBUG = "DEBUG", DEBUG = "DEBUG",
INFO = "INFO", INFO = "INFO",
WARNING = "WARNING", WARNING = "WARNING",
ERROR = "ERROR" ERROR = "ERROR"
} }
const LOG_LEVEL_ORDER = { const LOG_LEVEL_ORDER = {
[LogLevel.DEBUG]: 0, [LogLevel.DEBUG]: 0,
[LogLevel.INFO]: 1, [LogLevel.INFO]: 1,
[LogLevel.WARNING]: 2, [LogLevel.WARNING]: 2,
[LogLevel.ERROR]: 3 [LogLevel.ERROR]: 3
}; };
export class LogLine { export class LogLine {
public timestamp = new Date(); public timestamp = new Date();
public constructor( public constructor(
public level: LogLevel, public level: LogLevel,
public message: string public message: string
) {} ) { }
} }
export class Logger { export class Logger {
private readonly messages: LogLine[] = []; private readonly messages: LogLine[] = [];
private readonly onMessageListeners: ((message: LogLine) => unknown)[] = []; public readonly onLogEmitted = new EventListeners<
(message: LogLine) => unknown
>();
public constructor(
...onMessageListeners: ((message: LogLine) => unknown)[]
) {
this.onMessageListeners = onMessageListeners;
}
public debug(message: string): void { public debug(message: string): void {
this.pushMessage(message, LogLevel.DEBUG); this.pushMessage(message, LogLevel.DEBUG);
} }
public info(message: string): void { public info(message: string): void {
this.pushMessage(message, LogLevel.INFO); this.pushMessage(message, LogLevel.INFO);
} }
public warn(message: string): void { public warn(message: string): void {
this.pushMessage(message, LogLevel.WARNING); this.pushMessage(message, LogLevel.WARNING);
} }
public error(message: string): void { public error(message: string): void {
this.pushMessage(message, LogLevel.ERROR); this.pushMessage(message, LogLevel.ERROR);
} }
public getMessages(mininumSeverity: LogLevel): LogLine[] { public getMessages(mininumSeverity: LogLevel): LogLine[] {
return this.messages.filter( return this.messages.filter(
(message) => (message) =>
LOG_LEVEL_ORDER[message.level] >= LOG_LEVEL_ORDER[message.level] >=
LOG_LEVEL_ORDER[mininumSeverity] LOG_LEVEL_ORDER[mininumSeverity]
); );
} }
public addOnMessageListener(listener: (message: LogLine) => unknown): void { public reset(): void {
this.onMessageListeners.push(listener); this.messages.length = 0;
} this.debug("Logger has been reset");
}
public removeOnMessageListener( private pushMessage(message: string, level: LogLevel): void {
listener: (message: LogLine) => unknown const logLine = new LogLine(level, message);
): void { this.messages.push(logLine);
removeFromArray(this.onMessageListeners, listener);
}
public reset(): void { while (this.messages.length > MAX_LOG_MESSAGE_COUNT) {
this.messages.length = 0; this.messages.shift();
this.debug("Logger has been reset"); }
}
private pushMessage(message: string, level: LogLevel): void { this.onLogEmitted.trigger(logLine);
const logLine = new LogLine(level, message); }
this.messages.push(logLine);
while (this.messages.length > MAX_LOG_MESSAGE_COUNT) {
this.messages.shift();
}
this.onMessageListeners.forEach((listener) => {
listener(logLine);
});
}
} }

242
frontend/sync-client/src/tracing/sync-history.ts
View file

@ -1,183 +1,169 @@
import { import {
MAX_HISTORY_ENTRY_COUNT, MAX_HISTORY_ENTRY_COUNT,
TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS
} from "../consts"; } from "../consts";
import type { RelativePath } from "../persistence/database"; import type { RelativePath } from "../persistence/database";
import type { Logger } from "./logger"; import type { Logger } from "./logger";
import { removeFromArray } from "../utils/remove-from-array"; import { removeFromArray } from "../utils/remove-from-array";
import { EventListeners } from "../utils/data-structures/event-listeners";
export interface SyncCreateDetails { export interface SyncCreateDetails {
type: SyncType.CREATE; type: SyncType.CREATE;
relativePath: RelativePath; relativePath: RelativePath;
} }
export interface SyncUpdateDetails { export interface SyncUpdateDetails {
type: SyncType.UPDATE; type: SyncType.UPDATE;
relativePath: RelativePath; relativePath: RelativePath;
} }
export interface SyncMovedDetails { export interface SyncMovedDetails {
type: SyncType.MOVE; type: SyncType.MOVE;
relativePath: RelativePath; relativePath: RelativePath;
movedFrom: RelativePath; movedFrom: RelativePath;
} }
export interface SyncDeleteDetails { export interface SyncDeleteDetails {
type: SyncType.DELETE; type: SyncType.DELETE;
relativePath: RelativePath; relativePath: RelativePath;
} }
export interface SyncSkippedDetails { export interface SyncSkippedDetails {
type: SyncType.SKIPPED; type: SyncType.SKIPPED;
relativePath: RelativePath; relativePath: RelativePath;
} }
export type SyncDetails = export type SyncDetails =
| SyncCreateDetails | SyncCreateDetails
| SyncUpdateDetails | SyncUpdateDetails
| SyncDeleteDetails | SyncDeleteDetails
| SyncMovedDetails | SyncMovedDetails
| SyncSkippedDetails; | SyncSkippedDetails;
export interface CommonHistoryEntry { export interface CommonHistoryEntry {
status: SyncStatus; status: SyncStatus;
message: string; message: string;
details: SyncDetails; details: SyncDetails;
author?: string; author?: string;
timestamp?: Date; timestamp?: Date;
} }
export enum SyncType { export enum SyncType {
CREATE = "CREATE", CREATE = "CREATE",
UPDATE = "UPDATE", UPDATE = "UPDATE",
DELETE = "DELETE", DELETE = "DELETE",
MOVE = "MOVE", MOVE = "MOVE",
SKIPPED = "SKIPPED" SKIPPED = "SKIPPED"
} }
export enum SyncStatus { export enum SyncStatus {
SUCCESS = "SUCCESS", SUCCESS = "SUCCESS",
ERROR = "ERROR", ERROR = "ERROR",
SKIPPED = "SKIPPED" SKIPPED = "SKIPPED"
} }
export type HistoryEntry = CommonHistoryEntry & { timestamp: Date }; export type HistoryEntry = CommonHistoryEntry & { timestamp: Date };
export interface HistoryStats { export interface HistoryStats {
success: number; success: number;
error: number; error: number;
} }
export class SyncHistory { export class SyncHistory {
private readonly _entries: HistoryEntry[] = []; private readonly _entries: HistoryEntry[] = [];
private readonly syncHistoryUpdateListeners: (( public readonly onHistoryUpdated = new EventListeners<
status: HistoryStats (status: HistoryStats) => unknown
) => unknown)[] = []; >();
private status: HistoryStats = { private status: HistoryStats = {
success: 0, success: 0,
error: 0 error: 0
}; };
public constructor(private readonly logger: Logger) {} public constructor(private readonly logger: Logger) { }
public get entries(): readonly HistoryEntry[] { public get entries(): readonly HistoryEntry[] {
return this._entries; return this._entries;
} }
/** /**
* Insert the entry at the beginning of the history list. If the entry * Insert the entry at the beginning of the history list. If the entry
* already in the list, it will get moved to the beginning and updated. * already in the list, it will get moved to the beginning and updated.
* *
* If the entry list is too long, the oldest entry will be removed. * If the entry list is too long, the oldest entry will be removed.
*/ */
public addHistoryEntry(entry: CommonHistoryEntry): void { public addHistoryEntry(entry: CommonHistoryEntry): void {
const historyEntry = { const historyEntry = {
...entry, ...entry,
timestamp: entry.timestamp ?? new Date() timestamp: entry.timestamp ?? new Date()
}; };
const candidate = this.findSimilarRecentUpdateEntry(historyEntry); const candidate = this.findSimilarRecentUpdateEntry(historyEntry);
if (candidate !== undefined) { if (candidate !== undefined) {
removeFromArray(this._entries, candidate); removeFromArray(this._entries, candidate);
} }
// Insert the entry at the beginning // Insert the entry at the beginning
this._entries.unshift(historyEntry); this._entries.unshift(historyEntry);
if (this._entries.length > MAX_HISTORY_ENTRY_COUNT) { if (this._entries.length > MAX_HISTORY_ENTRY_COUNT) {
this._entries.pop(); this._entries.pop();
} }
this.updateSuccessCount(historyEntry); this.updateSuccessCount(historyEntry);
} }
public addSyncHistoryUpdateListener(
listener: (stats: HistoryStats) => unknown
): void {
this.syncHistoryUpdateListeners.push(listener);
listener({ ...this.status });
}
public removeSyncHistoryUpdateListener(
listener: (stats: HistoryStats) => unknown
): void {
removeFromArray(this.syncHistoryUpdateListeners, listener);
}
public reset(): void { public reset(): void {
this._entries.length = 0; this._entries.length = 0;
this.status = { this.status = {
success: 0, success: 0,
error: 0 error: 0
}; };
this.syncHistoryUpdateListeners.forEach((listener) => { this.onHistoryUpdated.trigger(this.status);
listener(this.status); }
});
}
private findSimilarRecentUpdateEntry( private findSimilarRecentUpdateEntry(
entry: HistoryEntry entry: HistoryEntry
): HistoryEntry | undefined { ): HistoryEntry | undefined {
if (entry.details.type !== SyncType.UPDATE) { if (entry.details.type !== SyncType.UPDATE) {
return; return;
} }
const candidate = this._entries.find( const candidate = this._entries.find(
(e) => (e) =>
e.details.type === SyncType.UPDATE && e.details.type === SyncType.UPDATE &&
e.details.relativePath === entry.details.relativePath e.details.relativePath === entry.details.relativePath
); );
if ( if (
candidate !== undefined && candidate !== undefined &&
(this._entries[0] === candidate || (this._entries[0] === candidate ||
candidate.timestamp.getTime() + candidate.timestamp.getTime() +
TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS * 1000 > TIMEOUT_FOR_MERGING_HISTORY_ENTRIES_IN_SECONDS * 1000 >
entry.timestamp.getTime()) entry.timestamp.getTime())
) { ) {
return candidate; return candidate;
} }
} }
private updateSuccessCount(entry: HistoryEntry): void { private updateSuccessCount(entry: HistoryEntry): void {
const message = `${entry.details.relativePath} - ${entry.message} (${entry.details.type.toLocaleLowerCase()})`; const message = `${entry.details.relativePath} - ${entry.message} (${entry.details.type.toLocaleLowerCase()})`;
switch (entry.status) { switch (entry.status) {
case SyncStatus.SUCCESS: case SyncStatus.SUCCESS:
this.status.success++; this.status.success++;
this.logger.info(`History entry: ${message}`); this.logger.info(`History entry: ${message}`);
break; break;
case SyncStatus.ERROR: case SyncStatus.ERROR:
this.status.error++; this.status.error++;
this.logger.error(`Cannot sync file: ${message}`); this.logger.error(`Cannot sync file: ${message}`);
break; break;
case SyncStatus.SKIPPED: case SyncStatus.SKIPPED:
this.logger.warn(`Skipping file: ${message}`); this.logger.warn(`Skipping file: ${message}`);
break; break;
} }
this.syncHistoryUpdateListeners.forEach((listener) => { this.onHistoryUpdated.trigger(this.status);
listener(this.status); }
});
}
} }

147
frontend/sync-client/src/utils/data-structures/event-listeners.test.ts Normal file
View file

@ -0,0 +1,147 @@
import { describe, it } from "node:test";
import assert from "node:assert";
import { EventListeners } from "./event-listeners";
describe("EventListeners", () => {
it("should add & remove listeners", () => {
const listeners = new EventListeners<() => void>();
const listener = () => { };
listeners.add(listener);
assert.strictEqual(listeners.count, 1);
const removed = listeners.remove(listener);
assert.strictEqual(removed, true);
assert.strictEqual(listeners.count, 0);
});
it("should remove listeners using unsubscribe function", () => {
const listeners = new EventListeners<() => void>();
const listener = () => { };
const unsubscribe = listeners.add(listener);
unsubscribe();
assert.strictEqual(listeners.count, 0);
});
it("should return false when removing non-existent listener", () => {
const listeners = new EventListeners<() => void>();
const listener = () => { };
const removed = listeners.remove(listener);
assert.strictEqual(removed, false);
});
it("should handle multiple listeners", () => {
const listeners = new EventListeners<() => void>();
const listener1 = () => { };
const listener2 = () => { };
const listener3 = () => { };
listeners.add(listener1);
listeners.add(listener2);
listeners.add(listener3);
assert.strictEqual(listeners.count, 3);
listeners.remove(listener2);
assert.strictEqual(listeners.count, 2);
});
it("should trigger all listeners synchronously", () => {
const listeners = new EventListeners<(value: string) => void>();
const calls: string[] = [];
listeners.add((value) => calls.push(`listener1-${value}`));
listeners.add((value) => calls.push(`listener2-${value}`));
listeners.trigger("test");
assert.deepStrictEqual(calls, ["listener1-test", "listener2-test"]);
});
it("should trigger listeners with multiple arguments", () => {
const listeners = new EventListeners<
(a: number, b: string, c: boolean) => void
>();
const calls: [number, string, boolean][] = [];
listeners.add((a, b, c) => calls.push([a, b, c]));
listeners.trigger(42, "hello", true);
assert.deepStrictEqual(calls, [[42, "hello", true]]);
});
it("should not trigger removed listeners", () => {
const listeners = new EventListeners<() => void>();
let count1 = 0;
let count2 = 0;
const listener1 = () => {
count1++;
};
const listener2 = () => {
count2++;
};
listeners.add(listener1);
const unsubscribe = listeners.add(listener2);
unsubscribe();
listeners.trigger();
assert.strictEqual(count1, 1);
assert.strictEqual(count2, 0);
});
it("should trigger all listeners and await promises", async () => {
const listeners = new EventListeners<
(value: string) => Promise<void> | void
>();
const results: string[] = [];
listeners.add(async (value) => {
await new Promise((resolve) => setTimeout(resolve, 10));
results.push(`async1-${value}`);
});
listeners.add((value) => {
results.push(`sync-${value}`);
});
listeners.add(async (value) => {
await new Promise((resolve) => setTimeout(resolve, 5));
results.push(`async2-${value}`);
});
await listeners.triggerAsync("test");
assert.ok(results.includes("async1-test"));
assert.ok(results.includes("sync-test"));
assert.ok(results.includes("async2-test"));
assert.strictEqual(results.length, 3);
});
it("should not trigger cleared listeners", () => {
const listeners = new EventListeners<() => void>();
let called = false;
const listener = () => {
called = true;
};
listeners.add(listener);
listeners.clear();
assert.strictEqual(listeners.count, 0);
listeners.trigger();
assert.strictEqual(called, false);
});
});

71
frontend/sync-client/src/utils/data-structures/event-listeners.ts Normal file
View file

@ -0,0 +1,71 @@
import { removeFromArray } from "../remove-from-array";
import { awaitAll } from "../await-all";
/**
* A utility class for managing event listeners with type-safe add/remove operations.
*/
export class EventListeners<TListener extends (...args: any[]) => any> {
private readonly listeners: TListener[] = [];
/**
* Adds a new listener to the collection.
*
* @param listener The listener callback to add
* @returns An unsubscribe function that removes this listener when called
*/
public add(listener: TListener): () => void {
this.listeners.push(listener);
return () => this.remove(listener);
}
/**
* Removes a listener from the collection.
*
* @param listener The listener callback to remove
* @returns true if the listener was found and removed, false otherwise
*/
public remove(listener: TListener): boolean {
return removeFromArray(this.listeners, listener);
}
/**
* Triggers all listeners synchronously with the provided arguments.
* Any returned promises are ignored. Use triggerAsync() to await them.
*
* @param args The arguments to pass to each listener
*/
public trigger(...args: Parameters<TListener>): void {
this.listeners.forEach((listener) => {
listener(...args);
});
}
/**
* Triggers all listeners and awaits any promises they return.
* Synchronous listeners are called immediately, and any async listeners
* are awaited in parallel.
*
* @param args The arguments to pass to each listener
*/
public async triggerAsync(...args: Parameters<TListener>): Promise<void> {
await awaitAll(
this.listeners
.map((listener) => {
return listener(...args);
})
.filter((result): result is Promise<unknown> => {
return result instanceof Promise;
})
);
}
public clear(): void {
this.listeners.length = 0;
}
public get count(): number {
return this.listeners.length;
}
}

2
frontend/sync-client/src/utils/debugging/log-to-console.ts
View file

@ -3,7 +3,7 @@ import type { LogLine } from "../../tracing/logger";
import { LogLevel } from "../../tracing/logger"; import { LogLevel } from "../../tracing/logger";
export function logToConsole(client: SyncClient): void { export function logToConsole(client: SyncClient): void {
client.logger.addOnMessageListener((logLine: LogLine) => { client.logger.onLogEmitted.add((logLine: LogLine) => {
const formatted = `${logLine.timestamp.toISOString()} ${logLine.level} ${logLine.message}`; const formatted = `${logLine.timestamp.toISOString()} ${logLine.level} ${logLine.message}`;
switch (logLine.level) { switch (logLine.level) {