works
This commit is contained in:
parent
408afa3626
commit
e3196c2dc0
2 changed files with 337 additions and 300 deletions
|
|
@ -3,6 +3,7 @@ import type { SyncService } from "src/services/sync-service";
|
|||
import type { Logger } from "src/tracing/logger";
|
||||
import type { SyncHistory } from "src/tracing/sync-history";
|
||||
import PQueue from "p-queue";
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import { hash } from "src/utils/hash";
|
||||
import type { components } from "src/services/types";
|
||||
import type { Settings } from "src/persistence/settings";
|
||||
|
|
@ -27,7 +28,7 @@ export class Syncer {
|
|||
public constructor(
|
||||
private readonly logger: Logger,
|
||||
private readonly database: Database,
|
||||
private readonly settings: Settings,
|
||||
settings: Settings,
|
||||
private readonly syncService: SyncService,
|
||||
private readonly operations: FileOperations,
|
||||
history: SyncHistory
|
||||
|
|
@ -43,9 +44,11 @@ export class Syncer {
|
|||
this.syncQueue.concurrency = newSettings.syncConcurrency;
|
||||
});
|
||||
|
||||
this.syncQueue.on("active", () => {
|
||||
this.emitRemainingOperationsChange(this.syncQueue.size);
|
||||
});
|
||||
this.syncQueue.on("active", () =>
|
||||
this.remainingOperationsListeners.forEach((listener) =>
|
||||
listener(this.syncQueue.size)
|
||||
)
|
||||
);
|
||||
|
||||
this.internalSyncer = new UnrestrictedSyncer(
|
||||
logger,
|
||||
|
|
@ -82,29 +85,32 @@ export class Syncer {
|
|||
}
|
||||
|
||||
public async syncLocallyCreatedFile(
|
||||
relativePath: RelativePath,
|
||||
updateTime?: Date
|
||||
relativePath: RelativePath
|
||||
): Promise<void> {
|
||||
if (!this.settings.getSettings().isSyncEnabled) {
|
||||
this.logger.info(
|
||||
`Syncing is disabled, not syncing '${relativePath}'`
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(relativePath)
|
||||
?.isDeleted === false
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} already exists in the database, skipping`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
const proposedDocumentId = uuidv4();
|
||||
|
||||
// Most likely, we're waiting for the previous delete to finish on the file at this path
|
||||
await this.database.getResolvedDocumentByRelativePath(
|
||||
this.database.getNewResolvedDocumentByRelativePath(
|
||||
proposedDocumentId,
|
||||
relativePath,
|
||||
promise
|
||||
);
|
||||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
await this.syncQueue.add(() =>
|
||||
this.internalSyncer.unrestrictedSyncLocallyCreatedFile(
|
||||
() => this.database.getDocumentByUpdatePromise(promise),
|
||||
updateTime
|
||||
proposedDocumentId,
|
||||
() => this.database.getDocumentByUpdatePromise(promise)
|
||||
)
|
||||
);
|
||||
|
||||
|
|
@ -119,13 +125,8 @@ export class Syncer {
|
|||
public async syncLocallyDeletedFile(
|
||||
relativePath: RelativePath
|
||||
): Promise<void> {
|
||||
if (!this.settings.getSettings().isSyncEnabled) {
|
||||
this.logger.info(
|
||||
`Syncing is disabled, not syncing '${relativePath}'`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// We have to have a record of the delete in case there's an in-flight update for the same
|
||||
// document which finishes after the delete has succeeded and would introduce a phantom metadata record.
|
||||
this.database.delete(relativePath);
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
|
@ -137,13 +138,9 @@ export class Syncer {
|
|||
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncLocallyDeletedFile(() => {
|
||||
this.logger.debug(
|
||||
`aaaahg ${relativePath} has been deleted locally, syncing to delete it`
|
||||
);
|
||||
|
||||
return this.database.getDocumentByUpdatePromise(promise);
|
||||
})
|
||||
this.internalSyncer.unrestrictedSyncLocallyDeletedFile(() =>
|
||||
this.database.getDocumentByUpdatePromise(promise)
|
||||
)
|
||||
);
|
||||
|
||||
resolve();
|
||||
|
|
@ -156,23 +153,22 @@ export class Syncer {
|
|||
|
||||
public async syncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
relativePath,
|
||||
updateTime
|
||||
relativePath
|
||||
}: {
|
||||
oldPath?: RelativePath;
|
||||
relativePath: RelativePath;
|
||||
updateTime?: Date;
|
||||
}): Promise<void> {
|
||||
if (!this.settings.getSettings().isSyncEnabled) {
|
||||
this.logger.info(
|
||||
`Syncing is disabled, not syncing '${relativePath}'`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
||||
if (oldPath !== undefined) {
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(oldPath)
|
||||
?.isDeleted === true
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${oldPath} has been deleted locally, skipping`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (oldPath === relativePath) {
|
||||
throw new Error(
|
||||
`Old path and new path are the same: ${oldPath}`
|
||||
|
|
@ -182,6 +178,18 @@ export class Syncer {
|
|||
this.database.move(oldPath, relativePath);
|
||||
}
|
||||
|
||||
if (
|
||||
this.database.getLatestDocumentByRelativePath(relativePath)
|
||||
?.isDeleted === true
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${relativePath} has been deleted locally, skipping`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
||||
await this.database.getResolvedDocumentByRelativePath(
|
||||
relativePath,
|
||||
promise
|
||||
|
|
@ -191,7 +199,6 @@ export class Syncer {
|
|||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
updateTime,
|
||||
getLatestDocument: () =>
|
||||
this.database.getDocumentByUpdatePromise(promise)
|
||||
})
|
||||
|
|
@ -206,13 +213,6 @@ export class Syncer {
|
|||
}
|
||||
|
||||
public async scheduleSyncForOfflineChanges(): Promise<void> {
|
||||
if (!this.settings.getSettings().isSyncEnabled) {
|
||||
this.logger.debug(
|
||||
`Syncing is disabled, not uploading local changes`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.runningScheduleSyncForOfflineChanges !== undefined) {
|
||||
this.logger.debug("Uploading local changes is already in progress");
|
||||
return this.runningScheduleSyncForOfflineChanges;
|
||||
|
|
@ -234,13 +234,6 @@ export class Syncer {
|
|||
}
|
||||
|
||||
public async applyRemoteChangesLocally(): Promise<void> {
|
||||
if (!this.settings.getSettings().isSyncEnabled) {
|
||||
this.logger.debug(
|
||||
`Syncing is disabled, not fetching remote changes`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.runningApplyRemoteChangesLocally != null) {
|
||||
this.logger.debug(
|
||||
"Applying remote changes locally is already in progress"
|
||||
|
|
@ -272,6 +265,35 @@ export class Syncer {
|
|||
this.internalSyncer.reset();
|
||||
}
|
||||
|
||||
private async internalApplyRemoteChangesLocally(): Promise<void> {
|
||||
const remote = await this.syncQueue.add(async () =>
|
||||
this.syncService.getAll(this.database.getLastSeenUpdateId())
|
||||
);
|
||||
|
||||
if (!remote) {
|
||||
throw new Error("Failed to fetch remote changes");
|
||||
}
|
||||
|
||||
if (remote.latestDocuments.length === 0) {
|
||||
this.logger.debug("No remote changes to apply");
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.info("Applying remote changes locally");
|
||||
|
||||
await Promise.all(
|
||||
remote.latestDocuments.map(this.syncRemotelyUpdatedFile.bind(this))
|
||||
);
|
||||
|
||||
const lastSeenUpdateId = this.database.getLastSeenUpdateId();
|
||||
if (
|
||||
lastSeenUpdateId === undefined ||
|
||||
remote.lastUpdateId > lastSeenUpdateId
|
||||
) {
|
||||
this.database.setLastSeenUpdateId(remote.lastUpdateId);
|
||||
}
|
||||
}
|
||||
|
||||
private async syncRemotelyUpdatedFile(
|
||||
remoteVersion: components["schemas"]["DocumentVersionWithoutContent"]
|
||||
): Promise<void> {
|
||||
|
|
@ -279,45 +301,38 @@ export class Syncer {
|
|||
remoteVersion.documentId
|
||||
);
|
||||
|
||||
if (document === undefined) {
|
||||
const candidate = this.database.getLatestDocumentByRelativePath(
|
||||
remoteVersion.relativePath
|
||||
);
|
||||
if (candidate !== undefined && candidate.metadata === undefined) {
|
||||
document = candidate;
|
||||
}
|
||||
}
|
||||
|
||||
if (document === undefined) {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion
|
||||
)
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
const [promise, resolve, reject] = createPromise();
|
||||
|
||||
await this.database.getResolvedDocumentByRelativePath(
|
||||
document.relativePath,
|
||||
promise
|
||||
);
|
||||
|
||||
try {
|
||||
if (document === undefined) {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion,
|
||||
() => this.database.getDocumentByUpdatePromise(promise)
|
||||
() =>
|
||||
this.database.getDocumentByDocumentId(
|
||||
remoteVersion.documentId
|
||||
)
|
||||
)
|
||||
);
|
||||
} else {
|
||||
await this.database.getResolvedDocumentByRelativePath(
|
||||
document.relativePath,
|
||||
promise
|
||||
);
|
||||
|
||||
resolve();
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
try {
|
||||
await this.syncQueue.add(async () =>
|
||||
this.internalSyncer.unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion,
|
||||
() => this.database.getDocumentByUpdatePromise(promise)
|
||||
)
|
||||
);
|
||||
|
||||
resolve();
|
||||
} catch (e) {
|
||||
reject(e);
|
||||
} finally {
|
||||
this.database.removeDocumentPromise(promise);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -405,35 +420,4 @@ export class Syncer {
|
|||
|
||||
await Promise.all([updates, deletes]);
|
||||
}
|
||||
|
||||
private async internalApplyRemoteChangesLocally(): Promise<void> {
|
||||
const remote = await this.syncService.getAll(
|
||||
this.database.getLastSeenUpdateId()
|
||||
);
|
||||
|
||||
if (remote.latestDocuments.length === 0) {
|
||||
this.logger.debug("No remote changes to apply");
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.info("Applying remote changes locally");
|
||||
|
||||
await Promise.all(
|
||||
remote.latestDocuments.map(this.syncRemotelyUpdatedFile.bind(this))
|
||||
);
|
||||
|
||||
const lastSeenUpdateId = this.database.getLastSeenUpdateId();
|
||||
if (
|
||||
lastSeenUpdateId === undefined ||
|
||||
remote.lastUpdateId > lastSeenUpdateId
|
||||
) {
|
||||
this.database.setLastSeenUpdateId(remote.lastUpdateId);
|
||||
}
|
||||
}
|
||||
|
||||
private emitRemainingOperationsChange(remainingOperations: number): void {
|
||||
this.remainingOperationsListeners.forEach((listener) => {
|
||||
listener(remainingOperations);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import type {
|
||||
Database,
|
||||
DocumentId,
|
||||
DocumentRecord,
|
||||
RelativePath
|
||||
} from "../persistence/database";
|
||||
|
|
@ -15,6 +16,7 @@ import type { Settings } from "src/persistence/settings";
|
|||
import type { FileOperations } from "src/file-operations/file-operations";
|
||||
import { FileNotFoundError } from "src/file-operations/safe-filesystem-operations";
|
||||
import { DocumentLocks } from "../file-operations/document-locks";
|
||||
import { createPromise } from "src/utils/create-promise";
|
||||
|
||||
export class UnrestrictedSyncer {
|
||||
private readonly locks: DocumentLocks;
|
||||
|
|
@ -31,8 +33,8 @@ export class UnrestrictedSyncer {
|
|||
}
|
||||
|
||||
public async unrestrictedSyncLocallyCreatedFile(
|
||||
getLatestDocument: () => DocumentRecord,
|
||||
updateTime?: Date
|
||||
proposedDocumentId: DocumentId,
|
||||
getLatestDocument: () => DocumentRecord
|
||||
): Promise<void> {
|
||||
let latestDocument = getLatestDocument();
|
||||
|
||||
|
|
@ -41,29 +43,15 @@ export class UnrestrictedSyncer {
|
|||
SyncType.CREATE,
|
||||
SyncSource.PUSH,
|
||||
async () => {
|
||||
if (
|
||||
latestDocument.metadata !== undefined &&
|
||||
!latestDocument.isDeleted
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${latestDocument.relativePath} already exists in the database, no need to create it again`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBytes = await this.operations.read(
|
||||
latestDocument.relativePath
|
||||
); // this can throw FileNotFoundError
|
||||
const contentHash = hash(contentBytes);
|
||||
|
||||
updateTime ??= await this.operations.getModificationTime(
|
||||
latestDocument.relativePath
|
||||
); // this can throw FileNotFoundError
|
||||
|
||||
const response = await this.syncService.create({
|
||||
documentId: proposedDocumentId,
|
||||
relativePath: latestDocument.relativePath,
|
||||
contentBytes,
|
||||
createdDate: updateTime
|
||||
contentBytes
|
||||
});
|
||||
|
||||
latestDocument = getLatestDocument();
|
||||
|
|
@ -76,15 +64,15 @@ export class UnrestrictedSyncer {
|
|||
type: SyncType.CREATE
|
||||
});
|
||||
|
||||
const newMetadata = {
|
||||
relativePath: latestDocument.relativePath,
|
||||
documentId: response.documentId,
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: contentHash,
|
||||
isDeleted: false
|
||||
};
|
||||
|
||||
this.database.setDocument(newMetadata, latestDocument.identity);
|
||||
this.database.setDocument(
|
||||
{
|
||||
relativePath: latestDocument.relativePath,
|
||||
documentId: response.documentId,
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: contentHash
|
||||
},
|
||||
latestDocument.identity
|
||||
);
|
||||
|
||||
this.tryIncrementVaultUpdateId(response.vaultUpdateId);
|
||||
}
|
||||
|
|
@ -100,17 +88,9 @@ export class UnrestrictedSyncer {
|
|||
SyncType.DELETE,
|
||||
SyncSource.PUSH,
|
||||
async () => {
|
||||
if (document.metadata === undefined) {
|
||||
this.logger.debug(
|
||||
`Document '${document.relativePath}' has been created yet so deleting it remotely can be skipped`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const response = await this.syncService.delete({
|
||||
documentId: document.metadata.documentId,
|
||||
relativePath: document.relativePath,
|
||||
createdDate: new Date() // We've got the event now, so it must have been deleted just now
|
||||
documentId: document.documentId,
|
||||
relativePath: document.relativePath
|
||||
});
|
||||
|
||||
this.history.addHistoryEntry({
|
||||
|
|
@ -123,8 +103,6 @@ export class UnrestrictedSyncer {
|
|||
|
||||
document = getLatestDocument();
|
||||
|
||||
// We have to have a record of the delete in case there's an in-flight update for the same
|
||||
// document which finishes after the delete has succeeded and would introduce a phantom metadata record.
|
||||
this.database.setDocument(
|
||||
{
|
||||
relativePath: document.relativePath,
|
||||
|
|
@ -140,12 +118,10 @@ export class UnrestrictedSyncer {
|
|||
|
||||
public async unrestrictedSyncLocallyUpdatedFile({
|
||||
oldPath,
|
||||
getLatestDocument,
|
||||
updateTime
|
||||
getLatestDocument
|
||||
}: {
|
||||
oldPath?: RelativePath;
|
||||
getLatestDocument: () => DocumentRecord;
|
||||
updateTime?: Date;
|
||||
}): Promise<void> {
|
||||
let document = getLatestDocument();
|
||||
|
||||
|
|
@ -178,19 +154,13 @@ export class UnrestrictedSyncer {
|
|||
return;
|
||||
}
|
||||
|
||||
updateTime ??= await this.operations.getModificationTime(
|
||||
document.relativePath
|
||||
); // this can throw FileNotFoundError;
|
||||
|
||||
const response = await this.syncService.put({
|
||||
documentId: document.metadata.documentId,
|
||||
documentId: document.documentId,
|
||||
parentVersionId: document.metadata.parentVersionId,
|
||||
relativePath: document.relativePath,
|
||||
contentBytes,
|
||||
createdDate: updateTime
|
||||
contentBytes
|
||||
});
|
||||
|
||||
// Update relativePath which is the only property that can change while this is running (due to a move)
|
||||
document = getLatestDocument();
|
||||
|
||||
if (document.isDeleted) {
|
||||
|
|
@ -252,6 +222,11 @@ export class UnrestrictedSyncer {
|
|||
}
|
||||
|
||||
if (response.relativePath != document.relativePath) {
|
||||
// this.database.getNewResolvedDocumentByRelativePath(
|
||||
// response.relativePath,
|
||||
// promise
|
||||
// );
|
||||
|
||||
await this.operations.move(
|
||||
document.relativePath,
|
||||
response.relativePath,
|
||||
|
|
@ -283,10 +258,7 @@ export class UnrestrictedSyncer {
|
|||
this.database.setDocument(
|
||||
{
|
||||
documentId: response.documentId,
|
||||
relativePath:
|
||||
response.relativePath != document.relativePath
|
||||
? response.relativePath
|
||||
: document.relativePath,
|
||||
relativePath: document.relativePath,
|
||||
parentVersionId: response.vaultUpdateId,
|
||||
hash: contentHash
|
||||
},
|
||||
|
|
@ -300,20 +272,19 @@ export class UnrestrictedSyncer {
|
|||
|
||||
public async unrestrictedSyncRemotelyUpdatedFile(
|
||||
remoteVersion: components["schemas"]["DocumentVersionWithoutContent"],
|
||||
getLatestDocument?: () => DocumentRecord
|
||||
getLatestDocument: () => DocumentRecord | undefined
|
||||
): Promise<void> {
|
||||
await this.executeSync(
|
||||
[remoteVersion.relativePath],
|
||||
SyncType.UPDATE,
|
||||
SyncSource.PULL,
|
||||
async () => {
|
||||
const localMetadata =
|
||||
getLatestDocument?.() ??
|
||||
this.database.getDocumentByDocumentId(
|
||||
remoteVersion.documentId
|
||||
);
|
||||
let localMetadata = getLatestDocument();
|
||||
|
||||
if (localMetadata?.metadata !== undefined) {
|
||||
if (
|
||||
localMetadata !== undefined &&
|
||||
localMetadata?.metadata !== undefined
|
||||
) {
|
||||
// If the file exists locally, let's pretend the user has updated it
|
||||
// and deal with remote update/deletion within `unrestrictedSyncLocallyUpdatedFile`
|
||||
if (
|
||||
|
|
@ -329,7 +300,7 @@ export class UnrestrictedSyncer {
|
|||
return this.unrestrictedSyncLocallyUpdatedFile({
|
||||
getLatestDocument: () =>
|
||||
this.database.getDocumentByIdentity(
|
||||
localMetadata.identity
|
||||
localMetadata!.identity
|
||||
)
|
||||
});
|
||||
} else if (remoteVersion.isDeleted) {
|
||||
|
|
@ -347,27 +318,26 @@ export class UnrestrictedSyncer {
|
|||
})
|
||||
).contentBase64;
|
||||
|
||||
const latestDocument =
|
||||
getLatestDocument?.() ??
|
||||
this.database.getDocumentByDocumentId(
|
||||
remoteVersion.documentId
|
||||
);
|
||||
localMetadata = getLatestDocument();
|
||||
|
||||
if (latestDocument?.isDeleted) {
|
||||
if (localMetadata?.isDeleted === true) {
|
||||
this.logger.info(
|
||||
`Document ${remoteVersion.relativePath} has been deleted locally before we could finish updating it`
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (
|
||||
localMetadata?.metadata?.parentVersionId ??
|
||||
-1 >= remoteVersion.vaultUpdateId
|
||||
) {
|
||||
this.logger.debug(
|
||||
`Document ${remoteVersion.relativePath} is already more up to date than the fetched version`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const contentBytes = deserialize(content);
|
||||
|
||||
await this.operations.create(
|
||||
remoteVersion.relativePath,
|
||||
contentBytes,
|
||||
remoteVersion.documentId
|
||||
);
|
||||
|
||||
this.database.setDocument(
|
||||
{
|
||||
documentId: remoteVersion.documentId,
|
||||
|
|
@ -375,7 +345,13 @@ export class UnrestrictedSyncer {
|
|||
parentVersionId: remoteVersion.vaultUpdateId,
|
||||
hash: hash(contentBytes)
|
||||
},
|
||||
latestDocument?.identity
|
||||
localMetadata?.identity
|
||||
);
|
||||
|
||||
await this.operations.create(
|
||||
remoteVersion.relativePath,
|
||||
contentBytes,
|
||||
remoteVersion.documentId
|
||||
);
|
||||
|
||||
this.history.addHistoryEntry({
|
||||
|
|
@ -390,19 +366,12 @@ export class UnrestrictedSyncer {
|
|||
}
|
||||
|
||||
public async executeSync<T>(
|
||||
lockedPaths: RelativePath[],
|
||||
paths: RelativePath[],
|
||||
syncType: SyncType,
|
||||
syncSource: SyncSource,
|
||||
fn: () => Promise<T>
|
||||
): Promise<T | undefined> {
|
||||
const relativePath = lockedPaths[lockedPaths.length - 1];
|
||||
|
||||
if (!this.settings.getSettings().isSyncEnabled) {
|
||||
this.logger.info(
|
||||
`Syncing is disabled, not syncing '${relativePath}'`
|
||||
);
|
||||
return;
|
||||
}
|
||||
const relativePath = paths[paths.length - 1];
|
||||
|
||||
if (!this.operations.isFileEligibleForSync(relativePath)) {
|
||||
this.history.addHistoryEntry({
|
||||
|
|
|
|||
Loading…
Reference in a new issue