Wire up
This commit is contained in:
parent
d5bdbbe304
commit
d70143988e
1 changed files with 347 additions and 392 deletions
|
|
@ -1,62 +1,78 @@
|
||||||
import type { PersistenceProvider } from "./persistence/persistence";
|
import type { NetworkConnectionStatus } from "./types/network-connection-status";
|
||||||
import type { HistoryEntry, HistoryStats } from "./tracing/sync-history";
|
import { ClientMetadataStore } from "./persistence/client-metadata-store";
|
||||||
import { SyncHistory } from "./tracing/sync-history";
|
|
||||||
import { Logger, LogLevel, LogLine } from "./tracing/logger";
|
|
||||||
import type { RelativePath, StoredDatabase } from "./persistence/database";
|
|
||||||
import { Database } from "./persistence/database";
|
|
||||||
import * as Sentry from "@sentry/browser";
|
import * as Sentry from "@sentry/browser";
|
||||||
import type { SyncSettings } from "./persistence/settings";
|
import { setUpTelemetry } from "./utils/set-up-telemetry";
|
||||||
import { DEFAULT_SETTINGS, Settings } from "./persistence/settings";
|
import type {
|
||||||
import { SyncService } from "./services/sync-service";
|
MetadataPersistenceProvider,
|
||||||
import { Syncer } from "./sync-operations/syncer";
|
StoredClient
|
||||||
|
} from "./persistence/metadata-persistence-provider";
|
||||||
|
import { Database, type RelativePath } from "./persistence/database";
|
||||||
|
import {
|
||||||
|
Settings,
|
||||||
|
DEFAULT_SETTINGS,
|
||||||
|
type SyncSettings
|
||||||
|
} from "./persistence/settings";
|
||||||
import type { FileSystemOperations } from "./file-operations/filesystem-operations";
|
import type { FileSystemOperations } from "./file-operations/filesystem-operations";
|
||||||
import { FileOperations } from "./file-operations/file-operations";
|
import { FileOperations } from "./file-operations/file-operations";
|
||||||
import { FetchController } from "./services/fetch-controller";
|
import { Logger, LogLevel } from "./tracing/logger";
|
||||||
import { UnrestrictedSyncer } from "./sync-operations/unrestricted-syncer";
|
import { SyncHistory } from "./tracing/sync-history";
|
||||||
import { rateLimit } from "./utils/rate-limit";
|
import { SyncService } from "./services/sync-service";
|
||||||
import type { NetworkConnectionStatus } from "./types/network-connection-status";
|
|
||||||
import { DocumentSyncStatus } from "./types/document-sync-status";
|
|
||||||
import { WebSocketManager } from "./services/websocket-manager";
|
|
||||||
import { createClientId } from "./utils/create-client-id";
|
|
||||||
import { CursorTracker } from "./sync-operations/cursor-tracker";
|
|
||||||
import type { CursorSpan } from "./services/types/CursorSpan";
|
|
||||||
import type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
|
|
||||||
import { FileChangeNotifier } from "./sync-operations/file-change-notifier";
|
|
||||||
import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache";
|
|
||||||
import { setUpTelemetry } from "./utils/set-up-telemetry";
|
|
||||||
import { DIFF_CACHE_SIZE_MB } from "./consts";
|
|
||||||
import { ServerConfig } from "./services/server-config";
|
import { ServerConfig } from "./services/server-config";
|
||||||
import type { EventListeners } from "./utils/data-structures/event-listeners";
|
import { WebSocketManager } from "./services/websocket-manager";
|
||||||
|
import { Syncer } from "./sync-operations/syncer";
|
||||||
|
import { CursorTracker } from "./sync-operations/cursor-tracker";
|
||||||
|
import { EventListeners } from "./utils/data-structures/event-listeners";
|
||||||
|
import { createClientId } from "./utils/create-client-id";
|
||||||
|
import { ClientPhase } from "./types/client-phase";
|
||||||
|
import { DocumentSyncStatus } from "./types/document-sync-status";
|
||||||
|
import type { CursorSpan } from "./services/types/CursorSpan";
|
||||||
|
import { isInternalPath } from "./utils/portable-path";
|
||||||
|
import { FixedSizeDocumentCache } from "./utils/data-structures/fix-sized-cache";
|
||||||
|
import { Lock } from "./utils/data-structures/locks";
|
||||||
|
import { getUnappliedChanges } from "./sync-operations/local-changes";
|
||||||
|
|
||||||
export class SyncClient {
|
export class SyncClient {
|
||||||
private hasStartedOfflineSync = false;
|
private readonly lifecycle = new Lock();
|
||||||
private hasFinishedOfflineSync = false;
|
private destroying?: Promise<void>;
|
||||||
private hasStarted = false;
|
private phase = ClientPhase.Created;
|
||||||
private hasBeenDestroyed = false;
|
|
||||||
private unloadTelemetry?: () => void;
|
private unloadTelemetry?: () => void;
|
||||||
private isDestroying = false;
|
private readonly unsubscribeLog: () => void;
|
||||||
private readonly eventUnsubscribers: (() => void)[] = [];
|
|
||||||
|
|
||||||
private constructor(
|
private constructor(
|
||||||
|
public readonly logger: Logger,
|
||||||
private readonly history: SyncHistory,
|
private readonly history: SyncHistory,
|
||||||
private readonly settings: Settings,
|
private readonly settingsStore: 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,
|
private readonly service: SyncService,
|
||||||
private readonly fetchController: FetchController,
|
|
||||||
private readonly cursorTracker: CursorTracker,
|
|
||||||
private readonly fileChangeNotifier: FileChangeNotifier,
|
|
||||||
private readonly contentCache: FixedSizeDocumentCache,
|
|
||||||
private readonly fileOperations: FileOperations,
|
|
||||||
private readonly serverConfig: ServerConfig,
|
private readonly serverConfig: ServerConfig,
|
||||||
private readonly persistence: PersistenceProvider<
|
private readonly cursorTracker: CursorTracker,
|
||||||
Partial<{
|
private readonly fileChanges: EventListeners<
|
||||||
settings: Partial<SyncSettings>;
|
(path: RelativePath) => unknown
|
||||||
database: Partial<StoredDatabase>;
|
>,
|
||||||
}>
|
private readonly contentCache: FixedSizeDocumentCache,
|
||||||
>
|
private readonly persistence: MetadataPersistenceProvider<StoredClient>
|
||||||
) {}
|
) {
|
||||||
|
syncer.onServerHistoryChanged.add(() => {
|
||||||
|
cursorTracker.reset();
|
||||||
|
});
|
||||||
|
|
||||||
|
syncer.onReadyForEvents.add(() => {
|
||||||
|
if (
|
||||||
|
this.phase === ClientPhase.Started &&
|
||||||
|
settingsStore.getSettings().isSyncEnabled
|
||||||
|
) {
|
||||||
|
webSocketManager.start();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
this.unsubscribeLog = logger.onLogEmitted.add((log) => {
|
||||||
|
if (log.level === LogLevel.ERROR && Sentry.isInitialized()) {
|
||||||
|
Sentry.captureMessage(log.message);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
public get documentCount(): number {
|
public get documentCount(): number {
|
||||||
return this.database.length;
|
return this.database.length;
|
||||||
|
|
@ -66,38 +82,23 @@ export class SyncClient {
|
||||||
return this.webSocketManager.isWebSocketConnected;
|
return this.webSocketManager.isWebSocketConnected;
|
||||||
}
|
}
|
||||||
|
|
||||||
public get onSyncHistoryUpdated(): EventListeners<
|
public get onSyncHistoryUpdated(): SyncHistory["onHistoryUpdated"] {
|
||||||
(stats: HistoryStats) => unknown
|
|
||||||
> {
|
|
||||||
this.checkIfDestroyed("onSyncHistoryUpdated getter");
|
|
||||||
return this.history.onHistoryUpdated;
|
return this.history.onHistoryUpdated;
|
||||||
}
|
}
|
||||||
|
|
||||||
public get onSettingsChanged(): EventListeners<
|
public get onSettingsChanged(): Settings["onSettingsChanged"] {
|
||||||
(newSettings: SyncSettings, oldSettings: SyncSettings) => unknown
|
return this.settingsStore.onSettingsChanged;
|
||||||
> {
|
|
||||||
this.checkIfDestroyed("onSettingsChanged getter");
|
|
||||||
return this.settings.onSettingsChanged;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public get onRemainingOperationsCountChanged(): EventListeners<
|
public get onRemainingOperationsCountChanged(): Syncer["onRemainingOperationsCountChanged"] {
|
||||||
(remainingOperationsCount: number) => unknown
|
|
||||||
> {
|
|
||||||
this.checkIfDestroyed("onRemainingOperationsCountChanged getter");
|
|
||||||
return this.syncer.onRemainingOperationsCountChanged;
|
return this.syncer.onRemainingOperationsCountChanged;
|
||||||
}
|
}
|
||||||
|
|
||||||
public get onWebSocketStatusChanged(): EventListeners<
|
public get onWebSocketStatusChanged(): WebSocketManager["onWebSocketStatusChanged"] {
|
||||||
(isConnected: boolean) => unknown
|
|
||||||
> {
|
|
||||||
this.checkIfDestroyed("onWebSocketStatusChanged getter");
|
|
||||||
return this.webSocketManager.onWebSocketStatusChanged;
|
return this.webSocketManager.onWebSocketStatusChanged;
|
||||||
}
|
}
|
||||||
|
|
||||||
public get onRemoteCursorsUpdated(): EventListeners<
|
public get onRemoteCursorsUpdated(): CursorTracker["onRemoteCursorsUpdated"] {
|
||||||
(cursors: MaybeOutdatedClientCursors[]) => unknown
|
|
||||||
> {
|
|
||||||
this.checkIfDestroyed("onRemoteCursorsUpdated getter");
|
|
||||||
return this.cursorTracker.onRemoteCursorsUpdated;
|
return this.cursorTracker.onRemoteCursorsUpdated;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -105,439 +106,393 @@ export class SyncClient {
|
||||||
fs,
|
fs,
|
||||||
persistence,
|
persistence,
|
||||||
fetch,
|
fetch,
|
||||||
webSocket,
|
webSocket
|
||||||
nativeLineEndings = "\n"
|
|
||||||
}: {
|
}: {
|
||||||
fs: FileSystemOperations;
|
fs: FileSystemOperations;
|
||||||
persistence: PersistenceProvider<
|
persistence: MetadataPersistenceProvider<StoredClient>;
|
||||||
Partial<{
|
|
||||||
settings: Partial<SyncSettings>;
|
|
||||||
database: Partial<StoredDatabase>;
|
|
||||||
}>
|
|
||||||
>;
|
|
||||||
fetch?: typeof globalThis.fetch;
|
fetch?: typeof globalThis.fetch;
|
||||||
webSocket?: typeof globalThis.WebSocket;
|
webSocket?: typeof globalThis.WebSocket;
|
||||||
nativeLineEndings?: string;
|
|
||||||
}): Promise<SyncClient> {
|
}): Promise<SyncClient> {
|
||||||
const logger = new Logger();
|
const logger = new Logger();
|
||||||
|
const metadata = await ClientMetadataStore.load(persistence);
|
||||||
|
const { stored } = metadata;
|
||||||
|
|
||||||
const deviceId = createClientId();
|
const save = async (update: StoredClient): Promise<void> =>
|
||||||
|
metadata.save(update);
|
||||||
|
|
||||||
logger.info(`Creating SyncClient with client id ${deviceId}`);
|
const settings = new Settings(logger, stored.settings, async (data) =>
|
||||||
|
save({ settings: data })
|
||||||
const history = new SyncHistory(logger);
|
|
||||||
|
|
||||||
let state = (await persistence.load()) ?? {
|
|
||||||
settings: undefined,
|
|
||||||
database: undefined
|
|
||||||
};
|
|
||||||
|
|
||||||
const settings = new Settings(
|
|
||||||
logger,
|
|
||||||
state.settings,
|
|
||||||
async (data): Promise<void> => {
|
|
||||||
state = { ...state, settings: data };
|
|
||||||
// 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
|
|
||||||
await persistence.save(state);
|
|
||||||
}
|
|
||||||
);
|
|
||||||
|
|
||||||
const rateLimitedSave = rateLimit(
|
|
||||||
persistence.save,
|
|
||||||
() => settings.getSettings().minimumSaveIntervalMs
|
|
||||||
);
|
);
|
||||||
|
|
||||||
const database = new Database(
|
const database = new Database(
|
||||||
logger,
|
stored.database,
|
||||||
state.database,
|
SyncClient.getVaultKey(settings.getSettings()),
|
||||||
async (data): Promise<void> => {
|
async (data) => save({ database: data }),
|
||||||
state = { ...state, database: data };
|
async () => metadata.reloadDatabase()
|
||||||
await rateLimitedSave(state);
|
);
|
||||||
|
|
||||||
|
const localChanges = getUnappliedChanges(
|
||||||
|
database.state,
|
||||||
|
stored.localChanges ?? []
|
||||||
|
);
|
||||||
|
|
||||||
|
if (
|
||||||
|
localChanges.length &&
|
||||||
|
stored.localChangesVaultKey !==
|
||||||
|
SyncClient.getVaultKey(settings.getSettings())
|
||||||
|
) {
|
||||||
|
throw new Error(
|
||||||
|
"Offline notifications belong to another vault; use a separate state store"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
);
|
|
||||||
|
|
||||||
const fetchController = new FetchController(
|
const deviceId = createClientId();
|
||||||
settings.getSettings().isSyncEnabled,
|
const service = new SyncService(deviceId, settings, fetch, {
|
||||||
logger
|
get: (): string | undefined =>
|
||||||
);
|
metadata.stored.historyCheckpoint?.vaultKey ===
|
||||||
|
SyncClient.getVaultKey(settings.getSettings())
|
||||||
|
? metadata.stored.historyCheckpoint.checkpoint
|
||||||
|
: undefined,
|
||||||
|
save: async (checkpoint): Promise<void> =>
|
||||||
|
save({
|
||||||
|
historyCheckpoint: {
|
||||||
|
vaultKey: SyncClient.getVaultKey(
|
||||||
|
settings.getSettings()
|
||||||
|
),
|
||||||
|
checkpoint
|
||||||
|
}
|
||||||
|
})
|
||||||
|
});
|
||||||
|
|
||||||
const syncService = new SyncService(
|
const serverConfig = new ServerConfig(service);
|
||||||
deviceId,
|
|
||||||
fetchController,
|
|
||||||
settings,
|
|
||||||
logger,
|
|
||||||
fetch
|
|
||||||
);
|
|
||||||
|
|
||||||
const serverConfig = new ServerConfig(syncService);
|
const notifier = new EventListeners<(path: RelativePath) => unknown>();
|
||||||
|
|
||||||
const fileOperations = new FileOperations(
|
const files = new FileOperations(
|
||||||
logger,
|
|
||||||
database,
|
|
||||||
fs,
|
fs,
|
||||||
serverConfig,
|
|
||||||
nativeLineEndings
|
|
||||||
);
|
|
||||||
|
|
||||||
const contentCache = new FixedSizeDocumentCache(
|
|
||||||
1024 * 1024 * DIFF_CACHE_SIZE_MB
|
|
||||||
);
|
|
||||||
const unrestrictedSyncer = new UnrestrictedSyncer(
|
|
||||||
logger,
|
|
||||||
database,
|
database,
|
||||||
settings,
|
serverConfig,
|
||||||
syncService,
|
(paths) => {
|
||||||
fileOperations,
|
for (const path of paths) {
|
||||||
history,
|
notifier.trigger(path);
|
||||||
contentCache,
|
}
|
||||||
serverConfig
|
},
|
||||||
|
{
|
||||||
|
entries: localChanges,
|
||||||
|
// Invoked after all components have been constructed.
|
||||||
|
// eslint-disable-next-line @typescript-eslint/no-use-before-define
|
||||||
|
flush: async (): Promise<void> => syncer.flushLocalChanges()
|
||||||
|
},
|
||||||
|
(path, size) =>
|
||||||
|
settings.isIgnored(path) || settings.isOversized(size)
|
||||||
);
|
);
|
||||||
|
|
||||||
const webSocketManager = new WebSocketManager(
|
const websocket = new WebSocketManager(
|
||||||
deviceId,
|
deviceId,
|
||||||
logger,
|
logger,
|
||||||
settings,
|
settings,
|
||||||
webSocket
|
webSocket
|
||||||
);
|
);
|
||||||
|
|
||||||
const syncer = new Syncer(
|
const history = new SyncHistory(logger);
|
||||||
deviceId,
|
|
||||||
|
const contentCache = new FixedSizeDocumentCache(
|
||||||
|
settings.getSettings().diffCacheSizeMB * 1024 * 1024
|
||||||
|
);
|
||||||
|
|
||||||
|
const syncer: Syncer = new Syncer(
|
||||||
logger,
|
logger,
|
||||||
database,
|
database,
|
||||||
settings,
|
settings,
|
||||||
syncService,
|
service,
|
||||||
webSocketManager,
|
websocket,
|
||||||
fileOperations,
|
files,
|
||||||
unrestrictedSyncer
|
serverConfig,
|
||||||
|
history,
|
||||||
|
contentCache,
|
||||||
|
{
|
||||||
|
entries: localChanges,
|
||||||
|
save: async (entries) =>
|
||||||
|
save({
|
||||||
|
localChanges: entries,
|
||||||
|
localChangesVaultKey: SyncClient.getVaultKey(
|
||||||
|
settings.getSettings()
|
||||||
|
)
|
||||||
|
})
|
||||||
|
}
|
||||||
);
|
);
|
||||||
|
|
||||||
const fileChangeNotifier = new FileChangeNotifier();
|
const cursors = new CursorTracker(database, websocket, fs, notifier);
|
||||||
const cursorTracker = new CursorTracker(
|
|
||||||
database,
|
return new SyncClient(
|
||||||
webSocketManager,
|
logger,
|
||||||
fileOperations,
|
|
||||||
fileChangeNotifier
|
|
||||||
);
|
|
||||||
const client = new SyncClient(
|
|
||||||
history,
|
history,
|
||||||
settings,
|
settings,
|
||||||
database,
|
database,
|
||||||
syncer,
|
syncer,
|
||||||
webSocketManager,
|
websocket,
|
||||||
logger,
|
service,
|
||||||
fetchController,
|
|
||||||
cursorTracker,
|
|
||||||
fileChangeNotifier,
|
|
||||||
contentCache,
|
|
||||||
fileOperations,
|
|
||||||
serverConfig,
|
serverConfig,
|
||||||
|
cursors,
|
||||||
|
notifier,
|
||||||
|
contentCache,
|
||||||
persistence
|
persistence
|
||||||
);
|
);
|
||||||
|
|
||||||
logger.info("SyncClient created successfully");
|
|
||||||
|
|
||||||
return client;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public async start(): Promise<void> {
|
private static getVaultKey(settings: SyncSettings): string {
|
||||||
this.checkIfDestroyed("start");
|
return JSON.stringify([
|
||||||
|
settings.remoteUri.replace(/\/$/u, ""),
|
||||||
if (this.hasStarted) {
|
settings.vaultName
|
||||||
throw new Error("SyncClient has already been started");
|
]);
|
||||||
}
|
|
||||||
this.hasStarted = true;
|
|
||||||
|
|
||||||
if (
|
|
||||||
!this.unloadTelemetry &&
|
|
||||||
this.settings.getSettings().enableTelemetry
|
|
||||||
) {
|
|
||||||
this.unloadTelemetry = setUpTelemetry();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
this.eventUnsubscribers.push(
|
public get historyEntries(): SyncHistory["entries"] {
|
||||||
this.settings.onSettingsChanged.add((newSettings, oldSettings) => {
|
|
||||||
if (oldSettings.isSyncEnabled != newSettings.isSyncEnabled) {
|
|
||||||
this.fetchController.canFetch = newSettings.isSyncEnabled;
|
|
||||||
}
|
|
||||||
})
|
|
||||||
);
|
|
||||||
|
|
||||||
this.eventUnsubscribers.push(
|
|
||||||
this.logger.onLogEmitted.add((log): void => {
|
|
||||||
if (log.level === LogLevel.ERROR && Sentry.isInitialized()) {
|
|
||||||
Sentry.captureMessage(log.message);
|
|
||||||
}
|
|
||||||
})
|
|
||||||
);
|
|
||||||
|
|
||||||
this.eventUnsubscribers.push(
|
|
||||||
this.settings.onSettingsChanged.add(
|
|
||||||
this.onSettingsChange.bind(this)
|
|
||||||
)
|
|
||||||
);
|
|
||||||
|
|
||||||
if (this.settings.getSettings().isSyncEnabled) {
|
|
||||||
this.logger.info("Starting SyncClient");
|
|
||||||
await this.startSyncing();
|
|
||||||
this.logger.info("SyncClient has successfully started");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Reload settings from disk overriding current in-memory settings.
|
|
||||||
* Missing values will be filled in from DEFAULT_SETTINGS rather than
|
|
||||||
* retaining current in-memory settings.
|
|
||||||
*/
|
|
||||||
public async reloadSettings(): Promise<void> {
|
|
||||||
this.checkIfDestroyed("reloadSettings");
|
|
||||||
|
|
||||||
const state = (await this.persistence.load()) ?? {
|
|
||||||
settings: undefined
|
|
||||||
};
|
|
||||||
|
|
||||||
const settings = {
|
|
||||||
...DEFAULT_SETTINGS,
|
|
||||||
...(state.settings ?? {})
|
|
||||||
};
|
|
||||||
|
|
||||||
await this.setSettings(settings);
|
|
||||||
}
|
|
||||||
|
|
||||||
public async checkConnection(): Promise<NetworkConnectionStatus> {
|
|
||||||
this.checkIfDestroyed("checkConnection");
|
|
||||||
|
|
||||||
const server = await this.serverConfig.checkConnection(true);
|
|
||||||
return {
|
|
||||||
isSuccessful: server.isSuccessful,
|
|
||||||
serverMessage: server.message,
|
|
||||||
isWebSocketConnected: this.webSocketManager.isWebSocketConnected
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
public getHistoryEntries(): readonly HistoryEntry[] {
|
|
||||||
return this.history.entries;
|
return this.history.entries;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
public get settings(): SyncSettings {
|
||||||
* Wait for the in-flight operations to finish, reset all tracking,
|
return this.settingsStore.getSettings();
|
||||||
* and the local database but retain the settings.
|
}
|
||||||
* The SyncClient can be used again after calling this method.
|
|
||||||
*/
|
|
||||||
public async reset(): Promise<void> {
|
|
||||||
this.checkIfDestroyed("reset");
|
|
||||||
|
|
||||||
this.logger.info(
|
public async start(): Promise<void> {
|
||||||
"Stopping SyncClient to apply changed connection settings"
|
await this.lifecycle.withLock(async () => {
|
||||||
|
this.assertNotDestroyed();
|
||||||
|
if (this.phase === ClientPhase.Started) {
|
||||||
|
throw new Error("SyncClient has already been started");
|
||||||
|
}
|
||||||
|
|
||||||
|
await this.syncer.reloadLocalState();
|
||||||
|
|
||||||
|
await this.database.bindVault(
|
||||||
|
SyncClient.getVaultKey(this.settingsStore.getSettings())
|
||||||
);
|
);
|
||||||
|
|
||||||
|
// destroy() can interrupt the metadata work above before acquiring
|
||||||
|
// the lifecycle lock. Its terminal state must not be overwritten.
|
||||||
|
if (this.phase === ClientPhase.Destroyed) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.phase = ClientPhase.Started;
|
||||||
|
this.configureTelemetry();
|
||||||
|
await this.resume();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
// Restart transports; never discard pending requests.
|
||||||
|
public async reset(): Promise<void> {
|
||||||
|
await this.lifecycle.withLock(async () => {
|
||||||
|
this.assertNotDestroyed();
|
||||||
await this.pause();
|
await this.pause();
|
||||||
|
|
||||||
// clear all local state
|
|
||||||
this.logger.info("Resetting SyncClient's local state");
|
|
||||||
this.database.reset();
|
|
||||||
await this.database.save(); // ensure the new database reads as empty
|
|
||||||
this.resetInMemoryState();
|
|
||||||
this.hasStartedOfflineSync = false;
|
|
||||||
this.hasFinishedOfflineSync = false;
|
|
||||||
this.serverConfig.reset();
|
this.serverConfig.reset();
|
||||||
|
this.cursorTracker.reset();
|
||||||
await this.startSyncing();
|
await this.syncer.retryRejectedRequests();
|
||||||
|
await this.resume();
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
public getSettings(): SyncSettings {
|
public async setSettings(change: Partial<SyncSettings>): Promise<void> {
|
||||||
return this.settings.getSettings();
|
const update = structuredClone(change);
|
||||||
|
|
||||||
|
const notification = await this.lifecycle.withLock(async () => {
|
||||||
|
this.assertNotDestroyed();
|
||||||
|
const old = this.settingsStore.getSettings();
|
||||||
|
const next = { ...old, ...update };
|
||||||
|
if (
|
||||||
|
SyncClient.getVaultKey(next) !== SyncClient.getVaultKey(old) &&
|
||||||
|
(this.phase === ClientPhase.Started ||
|
||||||
|
this.database.state.lastSeenUpdateId !== undefined ||
|
||||||
|
this.database.state.pending ||
|
||||||
|
this.syncer.hasLocalChanges ||
|
||||||
|
Object.keys(this.database.state.actualFileManifest).length)
|
||||||
|
) {
|
||||||
|
throw new Error(
|
||||||
|
"Changing vaults requires a separate state store and client; existing sync state must not be reset"
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
public async setSetting<T extends keyof SyncSettings>(
|
try {
|
||||||
key: T,
|
if (this.phase === ClientPhase.Started) {
|
||||||
value: SyncSettings[T]
|
await this.pause();
|
||||||
|
}
|
||||||
|
|
||||||
|
await this.database.reloadFromSave();
|
||||||
|
await this.settingsStore.setSettings(update, false);
|
||||||
|
await this.syncer.retryRejectedRequests();
|
||||||
|
this.contentCache.resize(next.diffCacheSizeMB * 1024 * 1024);
|
||||||
|
await this.database.bindVault(SyncClient.getVaultKey(next));
|
||||||
|
} catch (error) {
|
||||||
|
this.serverConfig.reset();
|
||||||
|
this.contentCache.resize(
|
||||||
|
this.settingsStore.getSettings().diffCacheSizeMB *
|
||||||
|
1024 *
|
||||||
|
1024
|
||||||
|
);
|
||||||
|
this.configureTelemetry();
|
||||||
|
await this.resume().catch((resumeError: unknown) => {
|
||||||
|
this.logger.error(
|
||||||
|
`Settings recovery could not resume sync: ${String(resumeError)}`
|
||||||
|
);
|
||||||
|
});
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.serverConfig.reset();
|
||||||
|
this.configureTelemetry();
|
||||||
|
await this.resume();
|
||||||
|
return [this.settingsStore.getSettings(), old] as const;
|
||||||
|
});
|
||||||
|
await this.settingsStore.onSettingsChanged.triggerAsync(
|
||||||
|
...notification
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async setSetting<K extends keyof SyncSettings>(
|
||||||
|
key: K,
|
||||||
|
value: SyncSettings[K]
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
this.checkIfDestroyed("setSetting");
|
await this.setSettings({ [key]: value });
|
||||||
|
|
||||||
await this.settings.setSetting(key, value);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public async setSettings(value: Partial<SyncSettings>): Promise<void> {
|
public async reloadSettings(): Promise<void> {
|
||||||
this.checkIfDestroyed("setSettings");
|
const stored = await this.persistence.load();
|
||||||
|
await this.setSettings({ ...DEFAULT_SETTINGS, ...stored?.settings });
|
||||||
await this.settings.setSettings(value);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public async syncLocallyCreatedFile(
|
public async checkConnection(): Promise<NetworkConnectionStatus> {
|
||||||
relativePath: RelativePath
|
const result = await this.serverConfig.checkConnection();
|
||||||
): Promise<void> {
|
return {
|
||||||
this.checkIfDestroyed("syncLocallyCreatedFile");
|
isSuccessful: result.isSuccessful,
|
||||||
|
serverMessage: result.message,
|
||||||
this.fileChangeNotifier.notifyOfFileChange(relativePath);
|
isWebSocketConnected: this.isWebSocketConnected
|
||||||
return this.syncer.syncLocallyCreatedFile(relativePath);
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
public async syncLocallyDeletedFile(
|
// Notifications enqueue work. Use waitUntilFinished() to await convergence.
|
||||||
relativePath: RelativePath
|
public async syncLocallyCreatedFile(path: RelativePath): Promise<void> {
|
||||||
): Promise<void> {
|
this.assertNotDestroyed();
|
||||||
this.checkIfDestroyed("syncLocallyDeletedFile");
|
if (isInternalPath(path)) {
|
||||||
|
return;
|
||||||
this.fileChangeNotifier.notifyOfFileChange(relativePath);
|
|
||||||
return this.syncer.syncLocallyDeletedFile(relativePath);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public async syncLocallyUpdatedFile({
|
this.fileChanges.trigger(path);
|
||||||
oldPath,
|
await this.syncer.syncLocallyCreatedFile(path);
|
||||||
relativePath
|
}
|
||||||
}: {
|
|
||||||
|
public async syncLocallyDeletedFile(path: RelativePath): Promise<void> {
|
||||||
|
this.assertNotDestroyed();
|
||||||
|
if (isInternalPath(path)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.fileChanges.trigger(path);
|
||||||
|
await this.syncer.syncLocallyDeletedFile(path);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async syncLocallyUpdatedFile(change: {
|
||||||
oldPath?: RelativePath;
|
oldPath?: RelativePath;
|
||||||
relativePath: RelativePath;
|
relativePath: RelativePath;
|
||||||
}): Promise<void> {
|
}): Promise<void> {
|
||||||
this.checkIfDestroyed("syncLocallyUpdatedFile");
|
this.assertNotDestroyed();
|
||||||
|
if (
|
||||||
this.fileChangeNotifier.notifyOfFileChange(relativePath);
|
isInternalPath(change.relativePath) ||
|
||||||
return this.syncer.syncLocallyUpdatedFile({
|
(change.oldPath !== undefined &&
|
||||||
oldPath,
|
change.oldPath !== "" &&
|
||||||
relativePath
|
isInternalPath(change.oldPath))
|
||||||
});
|
) {
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
public getDocumentSyncingStatus(
|
this.fileChanges.trigger(change.relativePath);
|
||||||
relativePath: RelativePath
|
await this.syncer.syncLocallyUpdatedFile(change);
|
||||||
): DocumentSyncStatus {
|
|
||||||
this.checkIfDestroyed("getDocumentSyncingStatus");
|
|
||||||
|
|
||||||
if (!this.settings.getSettings().isSyncEnabled) {
|
|
||||||
return DocumentSyncStatus.SYNCING_IS_DISABLED;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!this.syncer.isFirstSyncComplete || !this.hasFinishedOfflineSync) {
|
|
||||||
return DocumentSyncStatus.SYNCING;
|
|
||||||
}
|
|
||||||
|
|
||||||
const document =
|
|
||||||
this.database.getLatestDocumentByRelativePath(relativePath);
|
|
||||||
if (document === undefined) {
|
|
||||||
return DocumentSyncStatus.SYNCING;
|
|
||||||
}
|
|
||||||
return document.updates.length > 0
|
|
||||||
? DocumentSyncStatus.SYNCING
|
|
||||||
: DocumentSyncStatus.UP_TO_DATE;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public async updateLocalCursors(
|
public async updateLocalCursors(
|
||||||
documentToCursors: Record<RelativePath, CursorSpan[]>
|
cursors: Record<RelativePath, CursorSpan[]>
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
this.checkIfDestroyed("updateLocalCursors");
|
this.assertNotDestroyed();
|
||||||
|
await this.cursorTracker.sendLocalCursorsToServer(cursors);
|
||||||
|
}
|
||||||
|
|
||||||
await this.cursorTracker.sendLocalCursorsToServer(documentToCursors);
|
public getDocumentSyncingStatus(path: RelativePath): DocumentSyncStatus {
|
||||||
|
this.assertNotDestroyed();
|
||||||
|
if (!this.settingsStore.getSettings().isSyncEnabled) {
|
||||||
|
return DocumentSyncStatus.SYNCING_IS_DISABLED;
|
||||||
|
}
|
||||||
|
|
||||||
|
return this.syncer.isDocumentUpToDate(path)
|
||||||
|
? DocumentSyncStatus.UP_TO_DATE
|
||||||
|
: DocumentSyncStatus.SYNCING;
|
||||||
}
|
}
|
||||||
|
|
||||||
public async waitUntilFinished(): Promise<void> {
|
public async waitUntilFinished(): Promise<void> {
|
||||||
this.checkIfDestroyed("waitUntilIdle");
|
this.assertNotDestroyed();
|
||||||
await this.syncer.waitUntilFinished();
|
await this.syncer.waitUntilFinished();
|
||||||
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.
|
|
||||||
*/
|
|
||||||
public async destroy(): Promise<void> {
|
public async destroy(): Promise<void> {
|
||||||
this.checkIfDestroyed("destroy");
|
if (this.destroying) {
|
||||||
|
return this.destroying;
|
||||||
// Prevent concurrent destroy calls
|
|
||||||
if (this.isDestroying) {
|
|
||||||
this.logger.warn(
|
|
||||||
"destroy() called while already destroying, ignoring"
|
|
||||||
);
|
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
this.isDestroying = true;
|
|
||||||
|
|
||||||
// cancel everything that's in progress
|
|
||||||
await this.pause();
|
|
||||||
|
|
||||||
this.hasBeenDestroyed = true;
|
|
||||||
|
|
||||||
this.resetInMemoryState();
|
|
||||||
|
|
||||||
// Clean up event listeners to prevent memory leaks
|
|
||||||
this.eventUnsubscribers.forEach((unsubscribe) => {
|
|
||||||
unsubscribe();
|
|
||||||
});
|
|
||||||
this.eventUnsubscribers.length = 0;
|
|
||||||
|
|
||||||
this.logger.info("SyncClient has been successfully disposed");
|
|
||||||
|
|
||||||
|
this.phase = ClientPhase.Destroyed;
|
||||||
|
// Interrupt a start/reset awaiting network progress before waiting for
|
||||||
|
// its lifecycle operation to finish.
|
||||||
|
this.service.pause();
|
||||||
|
const stopped = this.syncer.stop();
|
||||||
|
this.destroying = this.lifecycle.withLock(async () => {
|
||||||
|
try {
|
||||||
|
await this.webSocketManager.stop();
|
||||||
|
await stopped;
|
||||||
|
await this.syncer.flushLocalChanges();
|
||||||
|
} finally {
|
||||||
|
this.cursorTracker.reset();
|
||||||
|
this.unsubscribeLog();
|
||||||
this.unloadTelemetry?.();
|
this.unloadTelemetry?.();
|
||||||
}
|
}
|
||||||
|
});
|
||||||
private async startSyncing(): Promise<void> {
|
return this.destroying;
|
||||||
this.checkIfDestroyed("startSyncing");
|
|
||||||
this.fetchController.finishReset();
|
|
||||||
|
|
||||||
await this.serverConfig.initialize();
|
|
||||||
this.webSocketManager.start();
|
|
||||||
|
|
||||||
if (!this.hasStartedOfflineSync) {
|
|
||||||
this.hasStartedOfflineSync = true;
|
|
||||||
await this.syncer.scheduleSyncForOfflineChanges();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
this.hasFinishedOfflineSync = true;
|
private assertNotDestroyed(): void {
|
||||||
|
if (this.phase === ClientPhase.Destroyed) {
|
||||||
|
throw new Error("SyncClient has been destroyed");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private async resume(): Promise<void> {
|
||||||
|
if (
|
||||||
|
this.phase !== ClientPhase.Started ||
|
||||||
|
!this.settingsStore.getSettings().isSyncEnabled
|
||||||
|
) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
await this.database.bindVault(
|
||||||
|
SyncClient.getVaultKey(this.settingsStore.getSettings())
|
||||||
|
);
|
||||||
|
this.service.resume();
|
||||||
|
this.syncer.start();
|
||||||
|
await this.syncer.waitUntilFinished();
|
||||||
}
|
}
|
||||||
|
|
||||||
private async pause(): Promise<void> {
|
private async pause(): Promise<void> {
|
||||||
this.fetchController.startReset();
|
const stopped = this.syncer.stop();
|
||||||
|
this.service.pause();
|
||||||
await this.webSocketManager.stop();
|
await this.webSocketManager.stop();
|
||||||
await this.waitUntilFinished();
|
await stopped;
|
||||||
}
|
}
|
||||||
|
|
||||||
private resetInMemoryState(): void {
|
private configureTelemetry(): void {
|
||||||
this.history.reset();
|
|
||||||
this.contentCache.reset();
|
|
||||||
// don't reset the logger
|
|
||||||
this.cursorTracker.reset();
|
|
||||||
this.syncer.reset();
|
|
||||||
this.fileOperations.reset();
|
|
||||||
}
|
|
||||||
|
|
||||||
private async onSettingsChange(
|
|
||||||
newSettings: SyncSettings,
|
|
||||||
oldSettings: SyncSettings
|
|
||||||
): Promise<void> {
|
|
||||||
this.checkIfDestroyed("onSettingsChange");
|
|
||||||
|
|
||||||
if (
|
if (
|
||||||
newSettings.vaultName !== oldSettings.vaultName ||
|
this.settingsStore.getSettings().enableTelemetry &&
|
||||||
newSettings.remoteUri !== oldSettings.remoteUri
|
typeof window !== "undefined"
|
||||||
) {
|
) {
|
||||||
await this.reset();
|
this.unloadTelemetry ??= setUpTelemetry();
|
||||||
}
|
|
||||||
|
|
||||||
if (newSettings.isSyncEnabled !== oldSettings.isSyncEnabled) {
|
|
||||||
if (newSettings.isSyncEnabled) {
|
|
||||||
await this.startSyncing();
|
|
||||||
} else {
|
|
||||||
await this.pause();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (newSettings.diffCacheSizeMB !== oldSettings.diffCacheSizeMB) {
|
|
||||||
this.contentCache.resize(newSettings.diffCacheSizeMB * 1024 * 1024);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (newSettings.enableTelemetry !== oldSettings.enableTelemetry) {
|
|
||||||
if (newSettings.enableTelemetry) {
|
|
||||||
this.unloadTelemetry = setUpTelemetry();
|
|
||||||
} else {
|
} else {
|
||||||
this.unloadTelemetry?.();
|
this.unloadTelemetry?.();
|
||||||
}
|
this.unloadTelemetry = undefined;
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private checkIfDestroyed(origin: string): void {
|
|
||||||
if (this.hasBeenDestroyed) {
|
|
||||||
throw new Error(
|
|
||||||
`SyncClient has been destroyed and can no longer be used; called from ${origin}`
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue