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

Add event handler class

Browse source
This commit is contained in:
Andras Schmelczer 2025-12-07 13:30:45 +00:00
commit ad3191957a
14 changed files with 2583 additions and 2464 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,7 +1,6 @@
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;
@ -37,10 +36,9 @@ 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,
@ -61,18 +59,6 @@ export class Settings {
return this.settings; return this.settings;
} }
public addOnSettingsChangeListener(
listener: (settings: SyncSettings, oldSettings: SyncSettings) => unknown
): void {
this.onSettingsChangeHandlers.push(listener);
}
public removeOnSettingsChangeListener(
listener: (settings: SyncSettings, oldSettings: SyncSettings) => unknown
): void {
removeFromArray(this.onSettingsChangeHandlers, listener);
}
public async setSetting<T extends keyof SyncSettings>( public async setSetting<T extends keyof SyncSettings>(
key: T, key: T,
value: SyncSettings[T] value: SyncSettings[T]
@ -93,14 +79,9 @@ export class Settings {
...value ...value
}; };
await awaitAll( await this.onSettingsChanged.triggerAsync(
this.onSettingsChangeHandlers this.settings,
.map((handler) => { oldSettings
return handler(this.settings, oldSettings);
})
.filter((result): result is Promise<unknown> => {
return result instanceof Promise;
})
); );
await this.save(); await this.save();

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;
}); });

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

@ -6,22 +6,23 @@ 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;
@ -60,24 +61,6 @@ export class WebSocketManager {
); );
} }
public addWebSocketStatusChangeListener(
listener: (isConnected: boolean) => unknown
): void {
this.webSocketStatusChangeListeners.push(listener);
}
public addRemoteCursorsUpdateListener(
listener: (cursors: ClientCursors[]) => Promise<void>
): void {
this.remoteCursorsUpdateListeners.push(listener);
}
public addRemoteVaultUpdateListener(
listener: (update: WebSocketVaultUpdate) => Promise<void>
): void {
this.remoteVaultUpdateListeners.push(listener);
}
public start(): void { public start(): void {
this.isStopped = false; this.isStopped = false;
this.initializeWebSocket(); this.initializeWebSocket();
@ -206,9 +189,7 @@ export class WebSocketManager {
this.webSocket.onopen = (): void => { this.webSocket.onopen = (): void => {
this.logger.info("WebSocket connection opened"); this.logger.info("WebSocket connection opened");
this.webSocketStatusChangeListeners.forEach((listener) => this.onWebSocketStatusChanged.trigger(true);
listener(true)
);
}; };
this.webSocket.onmessage = (event): void => { this.webSocket.onmessage = (event): void => {
@ -246,9 +227,7 @@ export class WebSocketManager {
this.logger.warn( this.logger.warn(
`WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})` `WebSocket closed with code ${event.code} (${event.reason == "" ? "unknown reason" : event.reason})`
); );
this.webSocketStatusChangeListeners.forEach((listener) => this.onWebSocketStatusChanged.trigger(false);
listener(false)
);
if (this.isStopped) { if (this.isStopped) {
this.resolveDisconnectingPromise?.(); this.resolveDisconnectingPromise?.();
@ -266,15 +245,7 @@ export class WebSocketManager {
message: WebSocketServerMessage message: WebSocketServerMessage
): Promise<void> { ): Promise<void> {
if (message.type === "vaultUpdate") { if (message.type === "vaultUpdate") {
await awaitAll( await this.onRemoteVaultUpdateReceived.triggerAsync(message);
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 // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
} else if (message.type === "cursorPositions") { } else if (message.type === "cursorPositions") {
@ -282,14 +253,8 @@ export class WebSocketManager {
`Received cursor positions for ${JSON.stringify(message.clients)}` `Received cursor positions for ${JSON.stringify(message.clients)}`
); );
await awaitAll( await this.onRemoteCursorsUpdateReceived.triggerAsync(
this.remoteCursorsUpdateListeners.map(async (listener) => { message.clients
await listener(message.clients).catch((error: unknown) => {
this.logger.error(
`Error in cursor positions listener: ${String(error)}`
);
});
})
); );
} else { } else {
this.logger.warn( this.logger.warn(

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

@ -26,6 +26,7 @@ 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;
@ -122,7 +123,7 @@ export class SyncClient {
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;
} }
@ -221,13 +222,13 @@ export class SyncClient {
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)
); );
@ -273,14 +274,6 @@ export class SyncClient {
return this.history.entries; return this.history.entries;
} }
public addSyncHistoryUpdateListener(
listener: (stats: HistoryStats) => unknown
): void {
this.checkIfDestroyed("addSyncHistoryUpdateListener");
this.history.addSyncHistoryUpdateListener(listener);
}
/** /**
* Wait for the in-flight operations to finish, reset all tracking, * Wait for the in-flight operations to finish, reset all tracking,
* and the local database but retain the settings. * and the local database but retain the settings.
@ -325,26 +318,35 @@ export class SyncClient {
await this.settings.setSettings(value); await this.settings.setSettings(value);
} }
public addOnSettingsChangeListener( public get onSyncHistoryUpdated(): EventListeners<
listener: (settings: SyncSettings, oldSettings: SyncSettings) => unknown (stats: HistoryStats) => unknown
): void { > {
this.checkIfDestroyed("addOnSettingsChangeListener"); this.checkIfDestroyed("onSyncHistoryUpdated getter");
return this.history.onHistoryUpdated;
this.settings.addOnSettingsChangeListener(listener);
} }
public addRemainingSyncOperationsListener(
listener: (remainingOperations: number) => unknown
): void {
this.checkIfDestroyed("addRemainingSyncOperationsListener");
this.syncer.addRemainingOperationsListener(listener);
public get onSettingsChanged(): EventListeners<
(newSettings: SyncSettings, oldSettings: SyncSettings) => unknown
> {
this.checkIfDestroyed("onSettingsChanged getter");
return this.settings.onSettingsChanged;
} }
public addWebSocketStatusChangeListener(listener: () => unknown): void { public get onRemainingOperationsCountChanged(): EventListeners<
this.checkIfDestroyed("addWebSocketStatusChangeListener"); (remainingOperationsCount: number) => unknown
> {
this.checkIfDestroyed("onRemainingOperationsCountChanged getter");
return this.syncer.onRemainingOperationsCountChanged;
}
this.webSocketManager.addWebSocketStatusChangeListener(listener); public get onWebSocketStatusChanged(): EventListeners<
(isConnected: boolean) => unknown
> {
this.checkIfDestroyed("onWebSocketStatusChanged getter");
return this.webSocketManager.onWebSocketStatusChanged;
} }
public async syncLocallyCreatedFile( public async syncLocallyCreatedFile(
@ -412,12 +414,12 @@ export class SyncClient {
await this.cursorTracker.sendLocalCursorsToServer(documentToCursors); await this.cursorTracker.sendLocalCursorsToServer(documentToCursors);
} }
public addRemoteCursorsUpdateListener(
listener: (cursors: MaybeOutdatedClientCursors[]) => unknown
): void {
this.checkIfDestroyed("addRemoteCursorsUpdateListener");
this.cursorTracker.addRemoteCursorsUpdateListener(listener); public get onRemoteCursorsUpdated(): EventListeners<
(cursors: MaybeOutdatedClientCursors[]) => unknown
> {
this.checkIfDestroyed("onRemoteCursorsUpdated getter");
return this.cursorTracker.onRemoteCursorsUpdated;
} }
public async waitUntilFinished(): Promise<void> { public async waitUntilFinished(): Promise<void> {

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

@ -9,6 +9,7 @@ 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
@ -17,6 +18,12 @@ import { Lock } from "../utils/data-structures/locks";
export class CursorTracker { export class CursorTracker {
private readonly updateLock = new Lock(); private readonly updateLock = new Lock();
// 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 readonly onRemoteCursorsUpdated = new EventListeners<
(cursors: MaybeOutdatedClientCursors[]) => unknown
>();
private knownRemoteCursors: (ClientCursors & { private knownRemoteCursors: (ClientCursors & {
upToDateness: DocumentUpToDateness; upToDateness: DocumentUpToDateness;
})[] = []; })[] = [];
@ -31,7 +38,7 @@ export class CursorTracker {
private readonly fileOperations: FileOperations, private readonly fileOperations: FileOperations,
private readonly fileChangeNotifier: FileChangeNotifier private readonly fileChangeNotifier: FileChangeNotifier
) { ) {
this.webSocketManager.addRemoteCursorsUpdateListener( this.webSocketManager.onRemoteCursorsUpdateReceived.add(
async (clientCursors) => { async (clientCursors) => {
await this.updateLock.withLock(async () => { await this.updateLock.withLock(async () => {
// The latest message will contain all active clients, so we can delete the ones // The latest message will contain all active clients, so we can delete the ones
@ -58,10 +65,15 @@ export class CursorTracker {
this.knownRemoteCursors = updatedKnownRemoteCursors; this.knownRemoteCursors = updatedKnownRemoteCursors;
}); });
this.onRemoteCursorsUpdated.trigger(
this.getRelevantAndPruneKnownClientCursors()
);
} }
); );
this.fileChangeNotifier.addFileChangeListener(async (relativePath) =>
this.fileChangeNotifier.onFileChanged.add(async (relativePath) =>
this.updateLock.withLock(async () => { this.updateLock.withLock(async () => {
for (const clientCursor of this.knownRemoteCursors) { for (const clientCursor of this.knownRemoteCursors) {
if ( if (
@ -144,18 +156,6 @@ export class CursorTracker {
this.webSocketManager.updateLocalCursors({ 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())
);
});
}
public reset(): void { public reset(): void {
this.knownRemoteCursors = []; this.knownRemoteCursors = [];

20
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( >();
listener: (filePath: RelativePath) => unknown
): void {
this.listeners.push(listener);
}
public removeFileChangeListener(
listener: (filePath: RelativePath) => unknown
): void {
removeFromArray(this.listeners, listener);
}
public notifyOfFileChange(filePath: RelativePath): void { public notifyOfFileChange(filePath: RelativePath): void {
this.listeners.forEach((listener) => listener(filePath)); this.onFileChanged.trigger(filePath);
} }
} }

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

@ -21,12 +21,13 @@ import type { WebSocketVaultUpdate } from "../services/types/WebSocketVaultUpdat
import type { WebSocketManager } from "../services/websocket-manager"; import type { WebSocketManager } from "../services/websocket-manager";
import type { WebSocketClientMessage } from "../services/types/WebSocketClientMessage"; import type { WebSocketClientMessage } from "../services/types/WebSocketClientMessage";
import { awaitAll } from "../utils/await-all"; import { awaitAll } from "../utils/await-all";
import { EventListeners } from "../utils/data-structures/event-listeners";
export class Syncer { export class Syncer {
private readonly remoteDocumentsLock: Locks<DocumentId>; private readonly remoteDocumentsLock: Locks<DocumentId>;
private readonly remainingOperationsListeners: (( public readonly onRemainingOperationsCountChanged = new EventListeners<
remainingOperations: number (remainingOperations: number) => unknown
) => unknown)[] = []; >();
// FIFO to limit the number of concurrent sync operations // FIFO to limit the number of concurrent sync operations
private readonly syncQueue: PQueue; private readonly syncQueue: PQueue;
@ -50,19 +51,17 @@ export class Syncer {
this.remoteDocumentsLock = new Locks<DocumentId>(this.logger); this.remoteDocumentsLock = new Locks<DocumentId>(this.logger);
settings.addOnSettingsChangeListener((newSettings, oldSettings) => { settings.onSettingsChanged.add((newSettings, oldSettings) => {
if (newSettings.syncConcurrency !== oldSettings.syncConcurrency) { if (newSettings.syncConcurrency !== oldSettings.syncConcurrency) {
this.syncQueue.concurrency = newSettings.syncConcurrency; this.syncQueue.concurrency = newSettings.syncConcurrency;
} }
}); });
this.syncQueue.on("active", () => { this.syncQueue.on("active", () => {
this.remainingOperationsListeners.forEach((listener) => { this.onRemainingOperationsCountChanged.trigger(this.syncQueue.size);
listener(this.syncQueue.size);
});
}); });
this.webSocketManager.addWebSocketStatusChangeListener( this.webSocketManager.onWebSocketStatusChanged.add(
(isConnected) => { (isConnected) => {
if (isConnected) { if (isConnected) {
// The JS WebSocket API doesn't support setting headers, so we have to send the token as a message // The JS WebSocket API doesn't support setting headers, so we have to send the token as a message
@ -70,7 +69,7 @@ export class Syncer {
} }
} }
); );
this.webSocketManager.addRemoteVaultUpdateListener( this.webSocketManager.onRemoteVaultUpdateReceived.add(
this.syncRemotelyUpdatedFile.bind(this) this.syncRemotelyUpdatedFile.bind(this)
); );
} }
@ -79,12 +78,6 @@ export class Syncer {
return this._isFirstSyncComplete; return this._isFirstSyncComplete;
} }
public addRemainingOperationsListener(
listener: (remainingOperations: number) => unknown
): void {
this.remainingOperationsListeners.push(listener);
}
public async syncLocallyCreatedFile( public async syncLocallyCreatedFile(
relativePath: RelativePath relativePath: RelativePath
): Promise<void> { ): Promise<void> {

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

@ -52,7 +52,7 @@ export class UnrestrictedSyncer {
this.logger this.logger
); );
this.settings.addOnSettingsChangeListener((newSettings) => { this.settings.onSettingsChanged.add((newSettings) => {
this.ignorePatterns = globsToRegexes( this.ignorePatterns = globsToRegexes(
newSettings.ignorePatterns, newSettings.ignorePatterns,
this.logger this.logger
@ -540,8 +540,7 @@ export class UnrestrictedSyncer {
type: SyncType.SKIPPED, type: SyncType.SKIPPED,
relativePath relativePath
}, },
message: `File size of ${sizeInMB} MB exceeds the maximum file size limit of ${ message: `File size of ${sizeInMB} MB exceeds the maximum file size limit of ${maxFileSizeMB
maxFileSizeMB
} MB` } MB`
}; };
} }

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

@ -1,5 +1,5 @@
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",
@ -25,13 +25,10 @@ export class LogLine {
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);
@ -57,16 +54,6 @@ export class Logger {
); );
} }
public addOnMessageListener(listener: (message: LogLine) => unknown): void {
this.onMessageListeners.push(listener);
}
public removeOnMessageListener(
listener: (message: LogLine) => unknown
): void {
removeFromArray(this.onMessageListeners, listener);
}
public reset(): void { public reset(): void {
this.messages.length = 0; this.messages.length = 0;
this.debug("Logger has been reset"); this.debug("Logger has been reset");
@ -80,8 +67,6 @@ export class Logger {
this.messages.shift(); this.messages.shift();
} }
this.onMessageListeners.forEach((listener) => { this.onLogEmitted.trigger(logLine);
listener(logLine);
});
} }
} }

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

@ -5,6 +5,7 @@ import {
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;
@ -71,9 +72,9 @@ export interface HistoryStats {
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,
@ -113,18 +114,7 @@ export class SyncHistory {
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;
@ -132,9 +122,7 @@ export class SyncHistory {
success: 0, success: 0,
error: 0 error: 0
}; };
this.syncHistoryUpdateListeners.forEach((listener) => { this.onHistoryUpdated.trigger(this.status);
listener(this.status);
});
} }
private findSimilarRecentUpdateEntry( private findSimilarRecentUpdateEntry(
@ -176,8 +164,6 @@ export class SyncHistory {
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) {