This commit is contained in:
Andras Schmelczer 2025-10-18 15:36:49 +01:00
commit 352c71af65
72 changed files with 20165 additions and 0 deletions

View file

@ -0,0 +1,33 @@
{
"name": "sync-client",
"version": "0.8.0",
"main": "dist/sync-client.node.js",
"browser": "dist/sync-client.web.js",
"types": "dist/types/index.d.ts",
"files": [
"dist/**/*"
],
"scripts": {
"dev": "webpack watch --mode development",
"build": "webpack --mode production",
"test": "tsx --test src/**/*.test.ts"
},
"dependencies": {
"byte-base64": "^1.1.0",
"minimatch": "^10.0.1",
"p-queue": "^8.1.0",
"reconcile-text": "^0.5.0",
"uuid": "^11.1.0"
},
"devDependencies": {
"@types/node": "^22.15.30",
"ts-loader": "^9.5.2",
"tslib": "2.8.1",
"tsx": "^4.20.5",
"typescript": "5.8.3",
"webpack": "^5.99.9",
"webpack-cli": "^6.0.1",
"webpack-merge": "^6.0.1",
"ws": "^8.18.3"
}
}

View file

@ -0,0 +1,24 @@
import type { SyncClient } from "../sync-client";
import type { LogLine } from "../tracing/logger";
import { LogLevel } from "../tracing/logger";
export function logToConsole(client: SyncClient): void {
client.logger.addOnMessageListener((logLine: LogLine) => {
const formatted = `${logLine.timestamp.toISOString()} ${logLine.level} ${logLine.message}`;
switch (logLine.level) {
case LogLevel.ERROR:
console.error(formatted);
break;
case LogLevel.WARNING:
console.warn(formatted);
break;
case LogLevel.INFO:
console.info(formatted);
break;
case LogLevel.DEBUG:
console.debug(formatted);
break;
}
});
}

View file

@ -0,0 +1,16 @@
import { sleep } from "../utils/sleep";
export const slowFetchFactory =
(jitterScaleInSeconds: number) =>
async (
input: string | URL | globalThis.Request,
init?: RequestInit
): Promise<Response> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
const response = await fetch(input, init);
return response;
};

View file

@ -0,0 +1,81 @@
import { sleep } from "../utils/sleep";
import { Locks } from "../utils/locks";
import type { Logger } from "../tracing/logger";
export function slowWebSocketFactory(
jitterScaleInSeconds: number,
logger: Logger
): typeof WebSocket {
// eslint-disable-next-line
return class FlakyWebSocket extends WebSocket {
private static readonly RECEIVE_KEY = "websocket-receive";
private static readonly SEND_KEY = "websocket-send";
private readonly locks = new Locks(logger);
public set onopen(callback: (event: Event) => void) {
super.onopen = async (event: Event): Promise<void> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
callback(event);
};
}
public set onmessage(callback: (event: MessageEvent) => void) {
super.onmessage = async (event: MessageEvent): Promise<void> => {
await this.locks.withLock(
FlakyWebSocket.RECEIVE_KEY,
async () => {
if (jitterScaleInSeconds > 0) {
await sleep(
Math.random() * jitterScaleInSeconds * 1000
);
}
callback(event);
}
);
};
}
public set onclose(callback: (event: CloseEvent) => void) {
super.onclose = async (event: CloseEvent): Promise<void> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
callback(event);
};
}
public set onerror(callback: (event: Event) => void) {
super.onerror = async (event: Event): Promise<void> => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
callback(event);
};
}
public send(
data: string | ArrayBufferLike | Blob | ArrayBufferView
): void {
this.waitingSend(data).catch((error: unknown) => {
logger.error(`Error sending WebSocket message: ${error}`);
});
}
private async waitingSend(
data: string | ArrayBufferLike | Blob | ArrayBufferView
): Promise<void> {
// maintain message order
await this.locks.withLock(FlakyWebSocket.SEND_KEY, async () => {
if (jitterScaleInSeconds > 0) {
await sleep(Math.random() * jitterScaleInSeconds * 1000);
}
super.send(data);
});
}
} as unknown as typeof WebSocket;
}

View file

@ -0,0 +1,6 @@
export class FileNotFoundError extends Error {
public constructor(message: string) {
super(message);
this.name = "FileNotFoundError";
}
}

View file

@ -0,0 +1,160 @@
import { describe, it } from "node:test";
import type {
Database,
DocumentRecord,
RelativePath
} from "../persistence/database";
import { FileOperations } from "./file-operations";
import { Logger } from "../tracing/logger";
import { assertSetContainsExactly } from "../utils/assert-set-contains-exactly";
import type { FileSystemOperations } from "./filesystem-operations";
import type { TextWithCursors } from "reconcile-text";
class MockDatabase implements Partial<Database> {
public getLatestDocumentByRelativePath(
_find: RelativePath
): DocumentRecord | undefined {
// no-op
return undefined;
}
public move(
_oldRelativePath: RelativePath,
_newRelativePath: RelativePath
): void {
// no-op
}
}
class FakeFileSystemOperations implements FileSystemOperations {
public readonly names = new Set<string>();
public async listAllFiles(): Promise<RelativePath[]> {
throw new Error("Method not implemented.");
}
public async read(_path: RelativePath): Promise<Uint8Array> {
throw new Error("Method not implemented.");
}
public async write(
path: RelativePath,
_content: Uint8Array
): Promise<void> {
this.names.add(path);
}
public async atomicUpdateText(
_path: RelativePath,
_updater: (current: TextWithCursors) => TextWithCursors
): Promise<string> {
throw new Error("Method not implemented.");
}
public async getFileSize(_path: RelativePath): Promise<number> {
throw new Error("Method not implemented.");
}
public async getModificationTime(_path: RelativePath): Promise<Date> {
throw new Error("Method not implemented.");
}
public async exists(path: RelativePath): Promise<boolean> {
return this.names.has(path);
}
public async createDirectory(_path: RelativePath): Promise<void> {
// this is called but irrelevant for this mock
}
public async delete(_path: RelativePath): Promise<void> {
throw new Error("Method not implemented.");
}
public async rename(
oldPath: RelativePath,
newPath: RelativePath
): Promise<void> {
this.names.delete(oldPath);
this.names.add(newPath);
}
}
describe("File operations", () => {
it("should deconflict renames", async () => {
const fileSystemOperations = new FakeFileSystemOperations();
const fileOperations = new FileOperations(
new Logger(),
new MockDatabase() as Database, // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
fileSystemOperations
);
await fileOperations.create("a", new Uint8Array());
assertSetContainsExactly(fileSystemOperations.names, "a");
await fileOperations.move("a", "b");
assertSetContainsExactly(fileSystemOperations.names, "b");
await fileOperations.create("c", new Uint8Array());
assertSetContainsExactly(fileSystemOperations.names, "b", "c");
await fileOperations.move("c", "b");
assertSetContainsExactly(fileSystemOperations.names, "b", "b (1)");
await fileOperations.create("c", new Uint8Array());
await fileOperations.move("c", "b");
assertSetContainsExactly(
fileSystemOperations.names,
"b",
"b (1)",
"b (2)"
);
});
it("should deconflict renames with file extension", async () => {
const fileSystemOperations = new FakeFileSystemOperations();
const fileOperations = new FileOperations(
new Logger(),
new MockDatabase() as Database, // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
fileSystemOperations
);
await fileOperations.create("b.md", new Uint8Array());
await fileOperations.create("c.md", new Uint8Array());
await fileOperations.move("c.md", "b.md");
assertSetContainsExactly(
fileSystemOperations.names,
"b.md",
"b (1).md"
);
await fileOperations.create("d.md", new Uint8Array());
await fileOperations.move("d.md", "b.md");
assertSetContainsExactly(
fileSystemOperations.names,
"b.md",
"b (1).md",
"b (2).md"
);
await fileOperations.create("file-23.md", new Uint8Array());
await fileOperations.create("file-23 (1).md", new Uint8Array());
await fileOperations.move("file-23.md", "file-23 (1).md");
assertSetContainsExactly(
fileSystemOperations.names,
"b.md",
"b (1).md",
"b (2).md",
"file-23 (1).md",
"file-23 (2).md"
);
});
it("should deconflict renames with paths", async () => {
const fileSystemOperations = new FakeFileSystemOperations();
const fileOperations = new FileOperations(
new Logger(),
new MockDatabase() as Database, // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
fileSystemOperations
);
await fileOperations.create("a/b.c/d", new Uint8Array());
await fileOperations.create("a/b.c/e", new Uint8Array());
await fileOperations.move("a/b.c/d", "a/b.c/e");
assertSetContainsExactly(
fileSystemOperations.names,
"a/b.c/e",
"a/b.c/e (1)"
);
});
});

View file

@ -0,0 +1,215 @@
import type { Logger } from "../tracing/logger";
import type { FileSystemOperations } from "./filesystem-operations";
import type { Database, RelativePath } from "../persistence/database";
import { SafeFileSystemOperations } from "./safe-filesystem-operations";
import type { TextWithCursors } from "reconcile-text";
import { isBinary, reconcile } from "reconcile-text";
import { isFileTypeMergable } from "../utils/is-file-type-mergable";
export class FileOperations {
private static readonly PARENTHESES_REGEX = / \((\d+)\)$/;
private readonly fs: SafeFileSystemOperations;
public constructor(
private readonly logger: Logger,
private readonly database: Database,
fs: FileSystemOperations,
private readonly nativeLineEndings = "\n"
) {
this.fs = new SafeFileSystemOperations(fs, logger);
}
public async listAllFiles(): Promise<RelativePath[]> {
return this.fs.listAllFiles();
}
public async read(path: RelativePath): Promise<Uint8Array> {
return this.fromNativeLineEndings(await this.fs.read(path));
}
/**
* Create a file at the specified path.
*
* If a file with the same name already exists, it is moved before creating the new one.
* Parent directories are created if necessary.
*/
public async create(
path: RelativePath,
newContent: Uint8Array
): Promise<void> {
await this.ensureClearPath(path);
return this.fs.write(path, this.toNativeLineEndings(newContent));
}
public async ensureClearPath(path: RelativePath): Promise<void> {
if (await this.fs.exists(path)) {
const deconflictedPath = await this.deconflictPath(path);
this.logger.debug(
`Didn't expect ${path} to exist, deconflicting by moving it to '${deconflictedPath}'`
);
this.database.move(path, deconflictedPath);
await this.fs.rename(path, deconflictedPath);
} else {
await this.createParentDirectories(path);
}
}
/**
* Update the file at the given path.
*
* Performs a 3-way merge before writing if the file's content differs from `expectedContent`.
* Does not recreate the file if it no longer exists, returning an empty array instead.
*/
public async write(
path: RelativePath,
expectedContent: Uint8Array,
newContent: Uint8Array
): Promise<void> {
if (!(await this.fs.exists(path))) {
this.logger.debug(
`The caller assumed ${path} exists, but it no longer, so we wont recreate it`
);
return;
}
if (
!isFileTypeMergable(path) ||
isBinary(expectedContent) ||
isBinary(newContent)
) {
this.logger.debug(
`The expected content is not mergable, so we won't perform a 3-way merge, just overwrite it`
);
await this.fs.write(
path,
// `newContent` might not be binary so we still have to ensure the line endings are correct
this.toNativeLineEndings(newContent)
);
return;
}
const expectedText = new TextDecoder().decode(expectedContent); // this comes from a previous read which must only have \n line endings
const newText = new TextDecoder().decode(newContent); // this comes from the server which stores text with \n line endings
await this.fs.atomicUpdateText(
path,
({ text, cursors }: TextWithCursors): TextWithCursors => {
this.logger.debug(
`Performing a 3-way merge for ${path} with the expected content`
);
text = text.replace(this.nativeLineEndings, "\n");
const merged = reconcile(
expectedText,
{ text, cursors },
newText
);
const resultText = merged.text.replace(
"\n",
this.nativeLineEndings
);
return {
text: resultText,
cursors: merged.cursors
};
}
);
}
public async delete(path: RelativePath): Promise<void> {
if (await this.exists(path)) {
return this.fs.delete(path);
} else {
this.logger.debug(`No need to delete '${path}', it doesn't exist`);
}
}
public async getFileSize(path: RelativePath): Promise<number> {
return this.fs.getFileSize(path);
}
public async exists(path: RelativePath): Promise<boolean> {
return this.fs.exists(path);
}
public async move(
oldPath: RelativePath,
newPath: RelativePath
): Promise<void> {
if (oldPath === newPath) {
return;
}
await this.ensureClearPath(newPath);
this.database.move(oldPath, newPath);
await this.fs.rename(oldPath, newPath);
}
private fromNativeLineEndings(content: Uint8Array): Uint8Array {
if (isBinary(content)) {
return content;
}
const decoder = new TextDecoder("utf-8");
let text = decoder.decode(content);
text = text.replace(this.nativeLineEndings, "\n");
return new TextEncoder().encode(text);
}
private toNativeLineEndings(content: Uint8Array): Uint8Array {
if (isBinary(content)) {
return content;
}
const decoder = new TextDecoder("utf-8");
let text = decoder.decode(content);
text = text.replace("\n", this.nativeLineEndings);
return new TextEncoder().encode(text);
}
private async createParentDirectories(path: string): Promise<void> {
const components = path.split("/");
if (components.length === 1) {
return;
}
for (let i = 1; i < components.length; i++) {
const parentDir = components.slice(0, i).join("/");
if (!(await this.fs.exists(parentDir))) {
await this.fs.createDirectory(parentDir);
}
}
}
private async deconflictPath(path: RelativePath): Promise<RelativePath> {
const pathParts = path.split("/");
const fileName = pathParts.pop();
if (fileName == "" || fileName == null) {
throw new Error(`Path '${path}' cannot be empty`);
}
let directory = pathParts.join("/");
if (directory) {
directory += "/";
}
const nameParts = fileName.split(".");
const extension =
nameParts.length > 1 ? "." + nameParts[nameParts.length - 1] : "";
let stem = extension ? nameParts.slice(0, -1).join(".") : fileName;
let currentCount = Number.parseInt(
FileOperations.PARENTHESES_REGEX.exec(stem)?.groups?.[0] ?? "0"
);
stem = stem.replace(FileOperations.PARENTHESES_REGEX, "");
let newName = path;
do {
currentCount++;
newName = `${directory}${stem} (${currentCount})${extension}`;
} while (await this.fs.exists(newName));
return newName;
}
}

View file

@ -0,0 +1,35 @@
import type { RelativePath } from "../persistence/database";
import type { TextWithCursors } from "reconcile-text";
export interface FileSystemOperations {
// List all files that should be synced.
listAllFiles: () => Promise<RelativePath[]>;
// Read the content of a file.
read: (path: RelativePath) => Promise<Uint8Array>;
// Create or overwrite a file with the given content.
write: (path: RelativePath, content: Uint8Array) => Promise<void>;
// Atomically update the content of a text file.
atomicUpdateText: (
path: RelativePath,
updater: (current: TextWithCursors) => TextWithCursors
) => Promise<string>;
// Get the size of a file in bytes.
getFileSize: (path: RelativePath) => Promise<number>;
// Check if a file exists.
exists: (path: RelativePath) => Promise<boolean>;
// Create a directory at the specified path. All parent directories must already exist.
createDirectory: (path: RelativePath) => Promise<void>;
// Delete a file. It is expected that the path points to an existing file.
delete: (path: RelativePath) => Promise<void>;
// Rename a file. It is expected that the oldPath points to an existing file and the newPath does not exist.
rename: (oldPath: RelativePath, newPath: RelativePath) => Promise<void>;
}

View file

@ -0,0 +1,137 @@
import type { RelativePath } from "../persistence/database";
import type { FileSystemOperations } from "./filesystem-operations";
import type { Logger } from "../tracing/logger";
import { Locks } from "../utils/locks";
import { FileNotFoundError } from "./file-not-found-error";
import type { TextWithCursors } from "reconcile-text";
/**
* Decorates `FileSystemOperations` to replace errors with `FileNotFoundError`
* if the accessed file doesn't exist. It also ensures that there's at most a
* single request in-flight for any one file through the use of locks.
*/
export class SafeFileSystemOperations implements FileSystemOperations {
private readonly locks: Locks<RelativePath>;
public constructor(
private readonly fs: FileSystemOperations,
private readonly logger: Logger
) {
this.locks = new Locks(logger);
}
public async listAllFiles(): Promise<RelativePath[]> {
this.logger.debug("Listing all files");
const result = await this.fs.listAllFiles();
this.logger.debug(`Listed ${result.length} files`);
return result;
}
public async read(path: RelativePath): Promise<Uint8Array> {
this.logger.debug(`Reading file '${path}'`);
return this.safeOperation(
path,
async () =>
this.locks.withLock(path, async () => this.fs.read(path)),
"read"
);
}
public async write(path: RelativePath, content: Uint8Array): Promise<void> {
this.logger.debug(`Writing to file '${path}'`);
return this.locks.withLock(path, async () =>
this.fs.write(path, content)
);
}
public async atomicUpdateText(
path: RelativePath,
updater: (current: TextWithCursors) => TextWithCursors
): Promise<string> {
this.logger.debug(`Atomically updating file '${path}'`);
return this.safeOperation(
path,
async () =>
this.locks.withLock(path, async () =>
this.fs.atomicUpdateText(path, updater)
),
"atomicUpdateText"
);
}
public async getFileSize(path: RelativePath): Promise<number> {
// Logging this would be too noisy
return this.safeOperation(
path,
async () =>
this.locks.withLock(path, async () =>
this.fs.getFileSize(path)
),
"getFileSize"
);
}
public async exists(path: RelativePath): Promise<boolean> {
this.logger.debug(`Checking if file '${path}' exists`);
return this.locks.withLock(path, async () => this.fs.exists(path));
}
public async createDirectory(path: RelativePath): Promise<void> {
this.logger.debug(`Creating directory '${path}'`);
return this.locks.withLock(path, async () =>
this.fs.createDirectory(path)
);
}
public async delete(path: RelativePath): Promise<void> {
this.logger.debug(`Deleting file '${path}'`);
return this.locks.withLock(path, async () => this.fs.delete(path));
}
public async rename(
oldPath: RelativePath,
newPath: RelativePath
): Promise<void> {
this.logger.debug(`Renaming file '${oldPath}' to '${newPath}'`);
return this.safeOperation(
oldPath,
async () =>
this.locks.withLock([oldPath, newPath], async () =>
this.fs.rename(oldPath, newPath)
),
"rename"
);
}
/**
* Decorate an operation to ensure that the file exists before running it.
* If the operation fails, it will check if the file still exists and throw
* a FileNotFoundError if it doesn't.
*/
private async safeOperation<T>(
path: RelativePath,
operation: () => Promise<T>,
operationName: string
): Promise<T> {
if (!(await this.fs.exists(path))) {
throw new FileNotFoundError(
`File '${path}' not found before trying to ${operationName}`
);
}
try {
return await operation();
} catch (error) {
// Without locking the file, this isn't atomic, however, it's good enough in practice.
// This will only break if the file exists, gets deleted and then immediately
// recreated while `operation` is running.
if (await this.fs.exists(path)) {
throw error;
} else {
throw new FileNotFoundError(
`File '${path}' not found when trying to ${operationName}`
);
}
}
}
}

View file

@ -0,0 +1,42 @@
import { logToConsole } from "./debugging/log-to-console";
import { slowFetchFactory } from "./debugging/slow-fetch-factory";
import { slowWebSocketFactory } from "./debugging/slow-web-socket-factory";
import { getRandomColor } from "./utils/get-random-color";
import { lineAndColumnToPosition } from "./utils/line-and-column-to-position";
import { positionToLineAndColumn } from "./utils/position-to-line-and-column";
export {
SyncType,
SyncStatus,
type HistoryStats,
type HistoryEntry,
type SyncDetails,
type SyncCreateDetails,
type SyncUpdateDetails,
type SyncMovedDetails,
type SyncDeleteDetails
} from "./tracing/sync-history";
export { Logger, LogLevel, LogLine } from "./tracing/logger";
export { type SyncSettings, DEFAULT_SETTINGS } from "./persistence/settings";
export { rateLimit } from "./utils/rate-limit";
export type { RelativePath, StoredDatabase } from "./persistence/database";
export type { FileSystemOperations } from "./file-operations/filesystem-operations";
export type { PersistenceProvider } from "./persistence/persistence";
export type { CursorSpan } from "./services/types/CursorSpan";
export type { ClientCursors } from "./services/types/ClientCursors";
export type { NetworkConnectionStatus } from "./types/network-connection-status";
export type { MaybeOutdatedClientCursors } from "./types/maybe-outdated-client-cursors";
export { DocumentSyncStatus } from "./types/document-sync-status";
export { SyncClient } from "./sync-client";
export const debugging = {
slowFetchFactory,
slowWebSocketFactory,
logToConsole
};
export const utils = {
getRandomColor,
positionToLineAndColumn,
lineAndColumnToPosition
};

View file

@ -0,0 +1,360 @@
import type { Logger } from "../tracing/logger";
import { EMPTY_HASH } from "../utils/hash";
import { CoveredValues } from "../utils/min-covered";
export type VaultUpdateId = number;
export type DocumentId = string;
export type RelativePath = string;
export interface DocumentMetadata {
parentVersionId: VaultUpdateId;
hash: string;
remoteRelativePath?: RelativePath;
}
export interface StoredDocumentMetadata {
relativePath: RelativePath;
documentId: DocumentId;
parentVersionId: VaultUpdateId;
remoteRelativePath?: RelativePath;
hash: string;
}
export interface StoredDatabase {
documents: StoredDocumentMetadata[];
lastSeenUpdateId: VaultUpdateId | undefined;
hasInitialSyncCompleted: boolean;
}
/**
* Represents a document in the database.
*
* It is mutable and its content should always represent the latest
* state of the document on disk based on the update events we have seen.
*/
export interface DocumentRecord {
relativePath: RelativePath;
documentId: DocumentId;
metadata: DocumentMetadata | undefined;
isDeleted: boolean;
updates: Promise<unknown>[];
parallelVersion: number;
}
export class Database {
private documents: DocumentRecord[];
private lastSeenUpdateIds: CoveredValues;
private hasInitialSyncCompleted: boolean;
public constructor(
private readonly logger: Logger,
initialState: Partial<StoredDatabase> | undefined,
private readonly saveData: (data: StoredDatabase) => Promise<void>
) {
initialState ??= {};
this.documents =
initialState.documents?.map(
({ relativePath, documentId, ...metadata }) => ({
relativePath,
documentId,
metadata,
isDeleted: false,
updates: [],
parallelVersion: 0
})
) ?? [];
this.ensureConsistency();
this.logger.debug(`Loaded ${this.documents.length} documents`);
const { lastSeenUpdateId } = initialState;
this.logger.debug(`Loaded last seen update id: ${lastSeenUpdateId}`);
this.lastSeenUpdateIds = new CoveredValues(
Math.max(0, lastSeenUpdateId ?? 0) // the first updateId will be 1 which is the first integer after -1
);
this.hasInitialSyncCompleted =
initialState.hasInitialSyncCompleted ?? false;
this.logger.debug(
`Loaded hasInitialSyncCompleted: ${this.hasInitialSyncCompleted}`
);
}
public get length(): number {
return this.documents.length;
}
public get resolvedDocuments(): DocumentRecord[] {
const paths = new Map<string, DocumentRecord[]>();
this.documents
.filter(({ metadata }) => metadata !== undefined)
.forEach((record) =>
paths.set(record.relativePath, [
record,
...(paths.get(record.relativePath) ?? [])
])
);
return Array.from(paths.values()).map((records) => {
records.sort(
(a, b) => b.parallelVersion - a.parallelVersion // descending
);
if (
records.length > 1 &&
records.some((current, i) =>
i === 0
? false
: records[i - 1].parallelVersion ===
current.parallelVersion
)
) {
throw new Error(
`Multiple documents with the same parallel version and path at ${records[0].relativePath}`
);
}
return records[0];
});
}
public updateDocumentMetadata(
metadata: {
parentVersionId: VaultUpdateId;
hash: string;
remoteRelativePath: RelativePath;
},
toUpdate: DocumentRecord
): void {
if (!this.documents.includes(toUpdate)) {
throw new Error("Document not found in database");
}
toUpdate.metadata = metadata;
this.save();
}
public removeDocumentPromise(promise: Promise<unknown>): void {
const entry = this.documents.find(({ updates }) =>
updates.includes(promise)
);
if (entry === undefined) {
// This method should be idempotent and tolerant of
// stragglers calling it after the databse has been reset.
return;
}
entry.updates = entry.updates.filter((update) => update !== promise);
// No need to save as Promises don't get serialized
}
public removeDocument(find: DocumentRecord): void {
this.documents = this.documents.filter((document) => document !== find);
this.save();
}
public getLatestDocumentByRelativePath(
find: RelativePath
): DocumentRecord | undefined {
const candidates = this.documents.filter(
({ relativePath }) => relativePath === find
);
candidates.sort((a, b) => b.parallelVersion - a.parallelVersion); // descending
return candidates[0];
}
public async getResolvedDocumentByRelativePath(
relativePath: RelativePath,
promise: Promise<unknown>
): Promise<DocumentRecord> {
const entry = this.getLatestDocumentByRelativePath(relativePath);
if (entry === undefined) {
throw new Error(
`Document not found by relative path: ${relativePath}, ${JSON.stringify(
this.documents,
null,
2
)}`
);
}
const currentPromises = entry.updates;
entry.updates = [...currentPromises, promise];
await Promise.all(currentPromises);
return entry;
}
public createNewPendingDocument(
documentId: DocumentId,
relativePath: RelativePath,
promise: Promise<unknown>
): DocumentRecord {
const previousEntry =
this.getLatestDocumentByRelativePath(relativePath);
const entry = {
relativePath,
documentId,
metadata: undefined,
isDeleted: false,
updates: [promise],
parallelVersion:
previousEntry?.parallelVersion === undefined
? 0
: previousEntry.parallelVersion + 1
};
this.documents.push(entry);
this.save();
return entry;
}
public createNewEmptyDocument(
documentId: DocumentId,
parentVersionId: VaultUpdateId,
relativePath: RelativePath
): DocumentRecord {
const entry = {
relativePath,
documentId,
metadata: {
parentVersionId,
hash: EMPTY_HASH,
remoteRelativePath: relativePath
},
isDeleted: false,
updates: [],
parallelVersion: 0
};
this.documents.push(entry);
this.save();
return entry;
}
public getDocumentByDocumentId(
find: DocumentId
): DocumentRecord | undefined {
return this.documents.find(({ documentId }) => documentId === find);
}
public move(
oldRelativePath: RelativePath,
newRelativePath: RelativePath
): void {
const oldDocument =
this.getLatestDocumentByRelativePath(oldRelativePath);
if (oldDocument === undefined) {
return;
}
const newDocument =
this.getLatestDocumentByRelativePath(newRelativePath);
if (newDocument?.isDeleted === false) {
throw new Error(
`Document already exists at new location: ${newRelativePath}`
);
}
oldDocument.relativePath = newRelativePath;
// We're in a strange state where the target of the move has just got deleted,
// however, its metadata might already have a bunch of updates queued up for
// the document at the new location. We need to keep these updates.
oldDocument.parallelVersion =
newDocument !== undefined ? newDocument.parallelVersion + 1 : 0;
this.save();
}
public delete(relativePath: RelativePath): void {
const candidate = this.getLatestDocumentByRelativePath(relativePath);
if (candidate === undefined) {
throw new Error(
`Document not found by relative path: ${relativePath}`
);
}
candidate.isDeleted = true;
}
public getHasInitialSyncCompleted(): boolean {
return this.hasInitialSyncCompleted;
}
public setHasInitialSyncCompleted(value: boolean): void {
this.hasInitialSyncCompleted = value;
this.save();
}
public getLastSeenUpdateId(): VaultUpdateId {
return this.lastSeenUpdateIds.min;
}
public addSeenUpdateId(value: number): void {
const previousMin = this.lastSeenUpdateIds.min;
this.lastSeenUpdateIds.add(value);
if (previousMin !== this.lastSeenUpdateIds.min) {
this.save();
}
}
public setLastSeenUpdateId(value: number): void {
this.lastSeenUpdateIds.min = value;
this.save();
}
public reset(): void {
this.documents = [];
this.lastSeenUpdateIds = new CoveredValues(
0 // the first updateId will be 1 which is the first integer after -1
);
this.hasInitialSyncCompleted = false;
this.save();
}
private save(): void {
this.ensureConsistency();
void this.saveData({
documents: this.resolvedDocuments.map(
({ relativePath, documentId, metadata }) => ({
documentId,
relativePath,
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
...metadata! // `resolvedDocuments` only returns docs with metadata set
})
),
lastSeenUpdateId: this.lastSeenUpdateIds.min,
hasInitialSyncCompleted: this.hasInitialSyncCompleted
}).catch((error: unknown) => {
this.logger.error(`Error saving data: ${error}`);
});
}
private ensureConsistency(): void {
const idToPath = new Map<string, string[]>();
this.resolvedDocuments.forEach(({ relativePath, documentId }) => {
idToPath.set(documentId, [
...(idToPath.get(documentId) ?? []),
relativePath
]);
});
const duplicates = Array.from(idToPath.entries())
.filter(([_, paths]) => paths.length > 1)
.map(([id, paths]) => `${id} (${paths.join(", ")})`);
if (duplicates.length > 0) {
throw new Error(
"Document IDs are not unique, found duplicates: " +
duplicates.join("; ")
);
}
}
}

View file

@ -0,0 +1,4 @@
export interface PersistenceProvider<T> {
load: () => Promise<T | undefined>;
save: (data: T) => Promise<void>;
}

View file

@ -0,0 +1,84 @@
import type { Logger } from "../tracing/logger";
export interface SyncSettings {
remoteUri: string;
token: string;
vaultName: string;
syncConcurrency: number;
isSyncEnabled: boolean;
maxFileSizeMB: number;
ignorePatterns: string[];
webSocketRetryIntervalMs: number;
}
export const DEFAULT_SETTINGS: SyncSettings = {
remoteUri: "",
token: "",
vaultName: "default",
syncConcurrency: 1,
isSyncEnabled: false,
maxFileSizeMB: 10,
ignorePatterns: [],
webSocketRetryIntervalMs: 3500
};
export class Settings {
private settings: SyncSettings;
private readonly onSettingsChangeHandlers: ((
newSettings: SyncSettings,
oldSettings: SyncSettings
) => unknown)[] = [];
public constructor(
private readonly logger: Logger,
initialState: Partial<SyncSettings> | undefined,
private readonly saveData: (data: SyncSettings) => Promise<void>
) {
this.settings = {
...DEFAULT_SETTINGS,
...(initialState ?? {})
};
this.logger.debug(
`Loaded settings: ${JSON.stringify(this.settings, null, 2)}`
);
}
public getSettings(): SyncSettings {
return this.settings;
}
public addOnSettingsChangeListener(
handler: (settings: SyncSettings, oldSettings: SyncSettings) => unknown
): void {
this.onSettingsChangeHandlers.push(handler);
}
public async setSetting<T extends keyof SyncSettings>(
key: T,
value: SyncSettings[T]
): Promise<void> {
this.logger.debug(`Setting '${key}' to '${value}'`);
await this.setSettings({
[key]: value
});
}
public async setSettings(value: Partial<SyncSettings>): Promise<void> {
const oldSettings = this.settings;
this.settings = {
...this.settings,
...value
};
this.onSettingsChangeHandlers.forEach((handler) => {
handler(this.settings, oldSettings);
});
await this.save();
}
private async save(): Promise<void> {
await this.saveData(this.settings);
}
}

View file

@ -0,0 +1,98 @@
import type { Settings } from "../persistence/settings";
import type { Logger } from "../tracing/logger";
import { createPromise } from "../utils/create-promise";
import { SyncResetError } from "./sync-reset-error";
export class ConnectionStatus {
private static readonly UNTIL_RESOLUTION = Symbol();
private canFetch: boolean;
private until: Promise<symbol>;
private resolveUntil: (result: symbol) => unknown;
private rejectUntil: (reason: unknown) => unknown;
public constructor(
settings: Settings,
private readonly logger: Logger
) {
this.canFetch = settings.getSettings().isSyncEnabled;
[this.until, this.resolveUntil, this.rejectUntil] =
createPromise<symbol>();
settings.addOnSettingsChangeListener((newSettings, oldSettings) => {
if (oldSettings.isSyncEnabled != newSettings.isSyncEnabled) {
this.canFetch = newSettings.isSyncEnabled;
this.resolveUntil(ConnectionStatus.UNTIL_RESOLUTION);
[this.until, this.resolveUntil, this.rejectUntil] =
createPromise<symbol>();
}
});
}
private static getUrlFromInput(input: RequestInfo | URL): string {
if (input instanceof URL) {
return input.href;
}
if (typeof input === "string") {
return input;
}
return input.url;
}
public startReset(): void {
this.rejectUntil(new SyncResetError());
}
public finishReset(): void {
[this.until, this.resolveUntil, this.rejectUntil] = createPromise();
}
public getFetchImplementation(
logger: Logger,
fetch: typeof globalThis.fetch = globalThis.fetch
): typeof globalThis.fetch {
return async (
input: RequestInfo | URL,
init?: RequestInit
): Promise<Response> => {
while (!this.canFetch) {
await this.until;
}
try {
// https://github.com/jonbern/fetch-retry/blob/8684ef4e688375f623bd76f13add76dbc1d67cfb/index.js#L67C1-L70C21
const _input =
typeof Request !== "undefined" && input instanceof Request
? input.clone()
: input;
const fetchPromise = fetch(_input, init);
// We only want to catch rejections from `this.until`
let result: symbol | Response | undefined = undefined;
do {
result = await Promise.race([this.until, fetchPromise]);
} while (result === ConnectionStatus.UNTIL_RESOLUTION);
const fetchResult: Response = result as Response; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
if (!fetchResult.ok) {
this.logger.warn(
`Fetch for ${ConnectionStatus.getUrlFromInput(
input
)}, got status ${fetchResult.status}`
);
}
return fetchResult;
} catch (error) {
logger.warn(
`Fetch for ${ConnectionStatus.getUrlFromInput(
input
)}, got error: ${error}`
);
throw error;
}
};
}
}

View file

@ -0,0 +1,6 @@
export class SyncResetError extends Error {
public constructor() {
super("Sync was reset");
this.name = "SyncResetError";
}
}

View file

@ -0,0 +1,325 @@
import type {
DocumentId,
RelativePath,
VaultUpdateId
} from "../persistence/database";
import type { Logger } from "../tracing/logger";
import type { Settings } from "../persistence/settings";
import type { ConnectionStatus } from "./connection-status";
import { sleep } from "../utils/sleep";
import { SyncResetError } from "./sync-reset-error";
import type { SerializedError } from "./types/SerializedError";
import type { DocumentVersionWithoutContent } from "./types/DocumentVersionWithoutContent";
import type { DocumentUpdateResponse } from "./types/DocumentUpdateResponse";
import type { DocumentVersion } from "./types/DocumentVersion";
import type { FetchLatestDocumentsResponse } from "./types/FetchLatestDocumentsResponse";
import type { PingResponse } from "./types/PingResponse";
import type { DeleteDocumentVersion } from "./types/DeleteDocumentVersion";
export interface CheckConnectionResult {
isSuccessful: boolean;
message: string;
}
export class SyncService {
private static readonly NETWORK_RETRY_INTERVAL_MS = 1000;
private readonly client: typeof globalThis.fetch;
private readonly pingClient: typeof globalThis.fetch;
public constructor(
private readonly deviceId: string,
private readonly connectionStatus: ConnectionStatus,
private readonly settings: Settings,
private readonly logger: Logger,
fetchImplementation: typeof globalThis.fetch = globalThis.fetch
) {
// ensure that if it's called a method, `this` won't be bound to the instance
const unboundFetch: typeof globalThis.fetch = async (...args) =>
fetchImplementation(...args);
this.client = this.connectionStatus.getFetchImplementation(
this.logger,
unboundFetch
);
this.pingClient = unboundFetch;
}
private static formatError(error: SerializedError): string {
let result = error.message;
if (error.causes.length > 0) {
const causes = error.causes.join(", ");
result += ` caused by: ${causes}`;
}
return result;
}
public async create({
documentId,
relativePath,
contentBytes
}: {
documentId?: DocumentId;
relativePath: RelativePath;
contentBytes: Uint8Array;
}): Promise<DocumentVersionWithoutContent> {
return this.withRetries(async () => {
const formData = new FormData();
if (documentId !== undefined) {
formData.append("document_id", documentId);
}
formData.append("relative_path", relativePath);
formData.append(
"content",
new Blob([new Uint8Array(contentBytes)])
);
const response = await this.client(this.getUrl("/documents"), {
method: "POST",
body: formData,
headers: this.getDefaultHeaders()
});
const result: SerializedError | DocumentVersionWithoutContent =
(await response.json()) as // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
| SerializedError
| DocumentVersionWithoutContent;
if ("errorType" in result) {
throw new Error(
`Failed to create document: ${SyncService.formatError(result)}`
);
}
this.logger.debug(
`Created document ${JSON.stringify(result)} with id ${
result.documentId
}`
);
return result;
});
}
public async put({
parentVersionId,
documentId,
relativePath,
contentBytes
}: {
parentVersionId: VaultUpdateId;
documentId: DocumentId;
relativePath: RelativePath;
contentBytes: Uint8Array;
}): Promise<DocumentUpdateResponse> {
return this.withRetries(async () => {
this.logger.debug(
`Updating document ${documentId} with parent version ${parentVersionId} and relative path ${relativePath}`
);
const formData = new FormData();
formData.append("parent_version_id", parentVersionId.toString());
formData.append("relative_path", relativePath);
formData.append(
"content",
new Blob([new Uint8Array(contentBytes)])
);
const response = await this.client(
this.getUrl(`/documents/${documentId}`),
{
method: "PUT",
body: formData,
headers: this.getDefaultHeaders()
}
);
const result: SerializedError | DocumentUpdateResponse =
(await response.json()) as // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
| SerializedError
| DocumentUpdateResponse;
if ("errorType" in result) {
throw new Error(
`Failed to update document: ${SyncService.formatError(result)}`
);
}
this.logger.debug(
`Updated document ${JSON.stringify(result)} with id ${
result.documentId
}}`
);
return result;
});
}
public async delete({
documentId,
relativePath
}: {
documentId: DocumentId;
relativePath: RelativePath;
}): Promise<DocumentVersionWithoutContent> {
return this.withRetries(async () => {
const request: DeleteDocumentVersion = {
relativePath
};
const response = await this.client(
this.getUrl(`/documents/${documentId}`),
{
method: "DELETE",
body: JSON.stringify(request),
headers: {
"Content-Type": "application/json",
...this.getDefaultHeaders()
}
}
);
const result: SerializedError | DocumentVersionWithoutContent =
(await response.json()) as // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
| SerializedError
| DocumentVersionWithoutContent;
if ("errorType" in result) {
throw new Error(
`Failed to delete document: ${SyncService.formatError(result)}`
);
}
this.logger.debug(
`Deleted document ${relativePath} with id ${documentId}`
);
return result;
});
}
public async get({
documentId
}: {
documentId: DocumentId;
}): Promise<DocumentVersion> {
return this.withRetries(async () => {
const response = await this.client(
this.getUrl(`/documents/${documentId}`),
{
headers: this.getDefaultHeaders()
}
);
const result: SerializedError | DocumentVersion =
(await response.json()) as SerializedError | DocumentVersion; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
if ("errorType" in result) {
throw new Error(
`Failed to get document: ${SyncService.formatError(result)}`
);
}
this.logger.debug(
`Get document ${result.relativePath} with id ${result.documentId}`
);
return result;
});
}
public async getAll(
since?: VaultUpdateId
): Promise<FetchLatestDocumentsResponse> {
return this.withRetries(async () => {
const url = new URL(this.getUrl("/documents"));
if (since !== undefined) {
url.searchParams.append("since", since.toString());
}
const response = await this.client(url.toString(), {
headers: this.getDefaultHeaders()
});
const result: SerializedError | FetchLatestDocumentsResponse =
(await response.json()) as // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
| SerializedError
| FetchLatestDocumentsResponse;
if ("errorType" in result) {
throw new Error(
`Failed to get documents: ${SyncService.formatError(result)}`
);
}
this.logger.debug(
`Got ${result.latestDocuments.length} document metadata`
);
return result;
});
}
public async checkConnection(): Promise<CheckConnectionResult> {
try {
const response = await this.pingClient(this.getUrl("/ping"), {
headers: this.getDefaultHeaders()
});
const result: PingResponse | SerializedError =
(await response.json()) as PingResponse | SerializedError; // eslint-disable-line @typescript-eslint/no-unsafe-type-assertion
if ("errorType" in result) {
throw new Error(
`Failed to ping server: ${SyncService.formatError(result)}`
);
}
if (result.isAuthenticated) {
return {
isSuccessful: true,
message: `Successfully connected to server (version: ${result.serverVersion}) and authenticated`
};
}
return {
isSuccessful: false,
message: `Successfully connected to server (version: ${result.serverVersion}) but failed to authenticate`
};
} catch (e) {
return {
isSuccessful: false,
message: `Failed to connect to server: ${e}`
};
}
}
private getUrl(path: string): string {
const { vaultName, remoteUri } = this.settings.getSettings();
const safeRemoteUri = remoteUri.replace(/\/+$/, "");
return `${safeRemoteUri}/vaults/${vaultName}${path}`;
}
private getDefaultHeaders(): Record<string, string> {
return {
"device-id": this.deviceId,
authorization: `Bearer ${this.settings.getSettings().token}`
};
}
private async withRetries<T>(fn: () => Promise<T>): Promise<T> {
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
while (true) {
try {
return await fn();
} catch (e) {
// We must not retry errors coming from reset
if (e instanceof SyncResetError) {
throw e;
}
this.logger.error(
`Failed network call (${e}), retrying in ${SyncService.NETWORK_RETRY_INTERVAL_MS}ms`
);
await sleep(SyncService.NETWORK_RETRY_INTERVAL_MS);
}
}
}
}

View file

@ -0,0 +1,8 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { DocumentWithCursors } from "./DocumentWithCursors";
export interface ClientCursors {
userName: string;
deviceId: string;
documentsWithCursors: DocumentWithCursors[];
}

View file

@ -0,0 +1,13 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export interface CreateDocumentVersion {
/**
* The client can decide the document id (if it wishes to) in order
* to help with syncing. If the client does not provide a document id,
* the server will generate one. If the client provides a document id
* it must not already exist in the database.
*/
document_id: string | null;
relative_path: string;
content: number[];
}

View file

@ -0,0 +1,6 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { DocumentWithCursors } from "./DocumentWithCursors";
export interface CursorPositionFromClient {
documentsWithCursors: DocumentWithCursors[];
}

View file

@ -0,0 +1,6 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { ClientCursors } from "./ClientCursors";
export interface CursorPositionFromServer {
clients: ClientCursors[];
}

Some files were not shown because too many files have changed in this diff Show more