..
This commit is contained in:
parent
17a1f4d060
commit
c9cf3239db
10 changed files with 200 additions and 509 deletions
|
|
@ -152,13 +152,32 @@
|
|||
|
||||
async function executeRestore() {
|
||||
const api = auth.api;
|
||||
if (!api || !restoreTarget) return;
|
||||
if (!api || !restoreTarget || !latest) return;
|
||||
restoring = true;
|
||||
try {
|
||||
await api.restoreVersion(
|
||||
// Restore = re-submit the target version's bytes at its path
|
||||
// as if it were a fresh edit. `update_document` short-circuits
|
||||
// on `is_deleted`, so resurrecting a deleted doc has to go
|
||||
// through `create_document`; a live doc takes the normal
|
||||
// update path with the current latest as its parent.
|
||||
const bytes = await api.fetchDocumentVersionContent(
|
||||
documentId,
|
||||
restoreTarget.vaultUpdateId
|
||||
);
|
||||
if (latest.isDeleted) {
|
||||
await api.createDocument(
|
||||
latest.vaultUpdateId,
|
||||
restoreTarget.relativePath,
|
||||
bytes
|
||||
);
|
||||
} else {
|
||||
await api.updateBinaryDocument(
|
||||
documentId,
|
||||
latest.vaultUpdateId,
|
||||
restoreTarget.relativePath,
|
||||
bytes
|
||||
);
|
||||
}
|
||||
toasts.add(
|
||||
`Restored to version #${restoreTarget.vaultUpdateId}`,
|
||||
"success"
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import type {
|
||||
DocumentUpdateResponse,
|
||||
DocumentVersion,
|
||||
DocumentVersionWithoutContent,
|
||||
FetchLatestDocumentsResponse,
|
||||
|
|
@ -108,19 +109,47 @@ export class ApiClient {
|
|||
);
|
||||
}
|
||||
|
||||
async restoreVersion(
|
||||
/**
|
||||
* Upload a new version of an existing (non-deleted) document. The
|
||||
* server treats this like any other edit — server-side merging,
|
||||
* path dedupe, and broadcast still apply. Used by the UI to restore
|
||||
* an old version by re-submitting its bytes on top of the latest.
|
||||
*/
|
||||
async updateBinaryDocument(
|
||||
documentId: string,
|
||||
vaultUpdateId: number
|
||||
): Promise<DocumentVersionWithoutContent> {
|
||||
parentVersionId: number,
|
||||
relativePath: string,
|
||||
content: ArrayBuffer
|
||||
): Promise<DocumentUpdateResponse> {
|
||||
const form = new FormData();
|
||||
form.append("parent_version_id", String(parentVersionId));
|
||||
form.append("relative_path", relativePath);
|
||||
form.append("content", new Blob([content]));
|
||||
return this.fetchJson(
|
||||
`${this.baseUrl}/documents/${documentId}/restore`,
|
||||
{
|
||||
method: "POST",
|
||||
headers: {
|
||||
"Content-Type": "application/json"
|
||||
},
|
||||
body: JSON.stringify({ vaultUpdateId })
|
||||
}
|
||||
`${this.baseUrl}/documents/${documentId}/binary`,
|
||||
{ method: "PUT", body: form }
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new document. Used by the UI to restore a deleted
|
||||
* document: `update_document` short-circuits on `is_deleted`, so
|
||||
* resurrection has to go through `create_document` — which detects
|
||||
* an existing doc at the same path, merges or dedupes as needed,
|
||||
* and returns the resulting version.
|
||||
*/
|
||||
async createDocument(
|
||||
lastSeenVaultUpdateId: number,
|
||||
relativePath: string,
|
||||
content: ArrayBuffer
|
||||
): Promise<DocumentUpdateResponse> {
|
||||
const form = new FormData();
|
||||
form.append("last_seen_vault_update_id", String(lastSeenVaultUpdateId));
|
||||
form.append("relative_path", relativePath);
|
||||
form.append("content", new Blob([content]));
|
||||
return this.fetchJson(`${this.baseUrl}/documents`, {
|
||||
method: "POST",
|
||||
body: form
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import type { DocumentVersionWithoutContent } from "./DocumentVersionWithoutContent";
|
||||
|
||||
export type { DocumentUpdateResponse } from "./DocumentUpdateResponse";
|
||||
export type { DocumentVersion } from "./DocumentVersion";
|
||||
export type { DocumentVersionWithoutContent } from "./DocumentVersionWithoutContent";
|
||||
export type { FetchLatestDocumentsResponse } from "./FetchLatestDocumentsResponse";
|
||||
|
|
|
|||
|
|
@ -3,9 +3,12 @@ import type { RelativePath } from "./types";
|
|||
// Local-only files displaced by `FileOperations.ensureClearPath` are named
|
||||
// `conflict-<uuid>-<originalName>`. The UUID is a full RFC-4122 v4 value so
|
||||
// a user-authored filename that happens to start with `conflict-` doesn't
|
||||
// get misclassified.
|
||||
const CONFLICT_UUID_REGEX =
|
||||
/^conflict-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}-/u;
|
||||
// get misclassified. The leading `(?:^|\/)` and trailing `[^/]*$` anchor the
|
||||
// match to the final path segment so intermediate directories named after
|
||||
// old conflict files (if a user renames one into a directory) don't ignore
|
||||
// everything beneath them.
|
||||
export const CONFLICT_PATH_REGEX =
|
||||
/(?:^|\/)conflict-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}-[^/]*$/u;
|
||||
|
||||
// Safe segment length for common filesystems (ext4 / NTFS / APFS all cap
|
||||
// at 255 bytes). `conflict-<36-char-uuid>-` adds 46 bytes; reserve a few
|
||||
|
|
@ -61,6 +64,5 @@ function truncateFileNameToByteLimit(
|
|||
* strictly local and must stay invisible to the server.
|
||||
*/
|
||||
export function isConflictPath(path: RelativePath): boolean {
|
||||
const fileName = path.substring(path.lastIndexOf("/") + 1);
|
||||
return CONFLICT_UUID_REGEX.test(fileName);
|
||||
return CONFLICT_PATH_REGEX.test(path);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -278,7 +278,7 @@ describe("SyncEventQueue", () => {
|
|||
const queue = createQueue();
|
||||
queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
|
||||
const promise = queue.getCreatePromise("a.md");
|
||||
const promise = queue.getLatestCreatePromise("a.md");
|
||||
assert.ok(promise !== undefined);
|
||||
|
||||
// The syncer resolves via event.resolvers after dequeuing
|
||||
|
|
@ -294,7 +294,7 @@ describe("SyncEventQueue", () => {
|
|||
const queue = createQueue();
|
||||
queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
|
||||
const promise = queue.getCreatePromise("a.md");
|
||||
const promise = queue.getLatestCreatePromise("a.md");
|
||||
assert.ok(promise !== undefined);
|
||||
|
||||
const event = await queue.next();
|
||||
|
|
@ -311,8 +311,8 @@ describe("SyncEventQueue", () => {
|
|||
queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
queue.enqueue({ type: SyncEventType.LocalCreate, path: "b.md" });
|
||||
|
||||
const promiseA = queue.getCreatePromise("a.md");
|
||||
const promiseB = queue.getCreatePromise("b.md");
|
||||
const promiseA = queue.getLatestCreatePromise("a.md");
|
||||
const promiseB = queue.getLatestCreatePromise("b.md");
|
||||
assert.ok(promiseA !== undefined);
|
||||
assert.ok(promiseB !== undefined);
|
||||
|
||||
|
|
@ -481,7 +481,7 @@ describe("SyncEventQueue", () => {
|
|||
const queue = createQueue();
|
||||
|
||||
queue.enqueue({ type: SyncEventType.LocalCreate, path: "a.md" });
|
||||
const createPromise = queue.getCreatePromise("a.md")!;
|
||||
const createPromise = queue.getLatestCreatePromise("a.md")!;
|
||||
|
||||
// Dependent events enqueued while create is still pending
|
||||
queue.enqueue({ type: SyncEventType.LocalUpdate, path: "a.md" });
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import type { Settings } from "../persistence/settings";
|
||||
import type { Logger } from "../tracing/logger";
|
||||
import { globsToRegexes } from "../utils/globs-to-regexes";
|
||||
import { isConflictPath } from "./conflict-path";
|
||||
import { CONFLICT_PATH_REGEX } from "./conflict-path";
|
||||
import { removeFromArray } from "../utils/remove-from-array";
|
||||
import {
|
||||
SyncEventType,
|
||||
|
|
@ -44,7 +44,7 @@ export class SyncEventQueue {
|
|||
private savePending = false;
|
||||
|
||||
|
||||
private readonly lastSeenUpdateId: VaultUpdateId;
|
||||
public readonly lastSeenUpdateId: VaultUpdateId;
|
||||
|
||||
public constructor(
|
||||
private readonly settings: Settings,
|
||||
|
|
@ -52,16 +52,19 @@ export class SyncEventQueue {
|
|||
initialState: Partial<StoredSyncState> | undefined,
|
||||
private readonly saveData: (data: StoredSyncState) => Promise<void>
|
||||
) {
|
||||
this.ignorePatterns = globsToRegexes(
|
||||
this.ignorePatterns = [
|
||||
CONFLICT_PATH_REGEX,
|
||||
...globsToRegexes(
|
||||
this.settings.getSettings().ignorePatterns,
|
||||
this.logger
|
||||
);
|
||||
)
|
||||
];
|
||||
|
||||
this.settings.onSettingsChanged.add((newSettings) => {
|
||||
this.ignorePatterns = globsToRegexes(
|
||||
newSettings.ignorePatterns,
|
||||
this.logger
|
||||
);
|
||||
this.ignorePatterns = [
|
||||
CONFLICT_PATH_REGEX,
|
||||
...globsToRegexes(newSettings.ignorePatterns, this.logger)
|
||||
];
|
||||
});
|
||||
|
||||
initialState ??= {};
|
||||
|
|
@ -84,6 +87,100 @@ export class SyncEventQueue {
|
|||
return this.documents.size;
|
||||
}
|
||||
|
||||
public enqueue(input: FileSyncEvent): void {
|
||||
if (input.type === SyncEventType.RemoteUpdate) {
|
||||
this.events.push(input);
|
||||
return;
|
||||
}
|
||||
|
||||
const { path } = input;
|
||||
|
||||
if (this.isIgnored(path)) {
|
||||
this.logger.info(
|
||||
`Ignoring ${input.type} for ${path} as it matches ignore patterns`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (input.type === SyncEventType.LocalCreate) {
|
||||
this.events.push({ type: SyncEventType.LocalCreate, path, originalPath: path });
|
||||
return;
|
||||
}
|
||||
|
||||
const lookupPath = (input.type === SyncEventType.LocalUpdate && input.oldPath) ? input.oldPath : path;
|
||||
const record = this.documents.get(lookupPath);
|
||||
const documentId: DocumentId | Promise<DocumentId> | undefined =
|
||||
this.getLatestCreatePromise(lookupPath) ?? record?.documentId;
|
||||
if (documentId === undefined) return;
|
||||
|
||||
if (input.type === SyncEventType.LocalDelete) {
|
||||
this.events.push({ type: SyncEventType.LocalDelete, documentId });
|
||||
return;
|
||||
}
|
||||
|
||||
if (input.oldPath !== undefined) {
|
||||
if (typeof documentId === "string") {
|
||||
this.documents.delete(input.oldPath);
|
||||
this.documents.set(path, record!);
|
||||
for (const e of this.events) {
|
||||
// It already has a docId, so there can't be a pending create event for it
|
||||
if (e.type === SyncEventType.LocalUpdate && e.documentId === documentId) {
|
||||
e.path = path;
|
||||
}
|
||||
}
|
||||
this.saveInTheBackground();
|
||||
} else {
|
||||
this.updatePendingCreatePath(input.oldPath, path);
|
||||
}
|
||||
}
|
||||
this.events.push({ type: SyncEventType.LocalUpdate, documentId, path, originalPath: path });
|
||||
}
|
||||
|
||||
|
||||
|
||||
public async next(): Promise<SyncEvent | undefined> {
|
||||
return this.events.shift();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Call once a create has been acknowledged by the server.
|
||||
*/
|
||||
public resolveCreate(
|
||||
event: Extract<SyncEvent, { type: SyncEventType.LocalCreate }>,
|
||||
record: DocumentRecord
|
||||
): void {
|
||||
const promise = event.resolvers?.promise;
|
||||
|
||||
this.documents.set(event.path, record);
|
||||
event.resolvers?.resolve(record.documentId);
|
||||
|
||||
if (promise !== undefined) {
|
||||
for (const e of this.events) {
|
||||
if (
|
||||
(e.type === SyncEventType.LocalUpdate || e.type === SyncEventType.LocalDelete) &&
|
||||
e.documentId === promise
|
||||
) {
|
||||
(e as { documentId: DocumentId | Promise<DocumentId> }).documentId = record.documentId;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.saveInTheBackground();
|
||||
}
|
||||
|
||||
public async save(): Promise<void> {
|
||||
return this.saveData({
|
||||
documents: Array.from(this.documents.entries()).map(
|
||||
([relativePath, record]) => ({
|
||||
relativePath,
|
||||
...record
|
||||
})
|
||||
),
|
||||
lastSeenUpdateId: this.lastSeenUpdateId
|
||||
});
|
||||
}
|
||||
|
||||
// todo: let's remove
|
||||
public getSettledDocumentByPath(path: RelativePath): DocumentRecord | undefined {
|
||||
return this.documents.get(path);
|
||||
|
|
@ -110,87 +207,10 @@ export class SyncEventQueue {
|
|||
this.saveInTheBackground();
|
||||
}
|
||||
|
||||
/**
|
||||
* Reflect a local rename in the queue's disk-path index.
|
||||
*
|
||||
* Mirrors the `input.oldPath !== undefined` branch of `enqueue`, but
|
||||
* without emitting a new `SyncLocal` — used by `FileOperations.move`
|
||||
* when the rename is a byproduct of another sync operation (e.g. the
|
||||
* user dragging a file) and the caller will push the resulting event
|
||||
* separately, or not at all.
|
||||
*
|
||||
* If the rename targets a path that already holds a settled record
|
||||
* (e.g. concurrent clobber), the destination's record is dropped: the
|
||||
* caller is expected to have moved the displaced file out of the way
|
||||
* via `ensureClearPath` already, so the dropped record reflects the
|
||||
* now-orphaned disk state.
|
||||
*/
|
||||
public moveDocument(
|
||||
oldPath: RelativePath,
|
||||
newPath: RelativePath
|
||||
): void {
|
||||
if (oldPath === newPath) return;
|
||||
|
||||
const record = this.documents.get(oldPath);
|
||||
if (record !== undefined) {
|
||||
// If `newPath` already holds a settled record, overwriting it
|
||||
// silently would orphan that document's identity. Warn so the
|
||||
// bug is visible; the caller is expected to have freed the
|
||||
// destination via `ensureClearPath` first.
|
||||
const clobbered = this.documents.get(newPath);
|
||||
if (clobbered !== undefined) {
|
||||
this.logger.warn(
|
||||
`moveDocument(${oldPath} → ${newPath}) is overwriting a settled record for document ${clobbered.documentId}; caller should have displaced it first`
|
||||
);
|
||||
}
|
||||
|
||||
this.documents.delete(oldPath);
|
||||
this.documents.set(newPath, record);
|
||||
for (const e of this.events) {
|
||||
if (
|
||||
e.type === SyncEventType.LocalUpdate &&
|
||||
e.documentId === record.documentId
|
||||
) {
|
||||
e.path = newPath;
|
||||
}
|
||||
}
|
||||
this.saveInTheBackground();
|
||||
return;
|
||||
}
|
||||
|
||||
// No settled record — the rename may be over a pending Create
|
||||
// whose document hasn't been persisted on the server yet.
|
||||
this.updatePendingCreatePath(oldPath, newPath);
|
||||
}
|
||||
|
||||
/**
|
||||
* Call once a create has been acknowledged by the server.
|
||||
*/
|
||||
public resolveCreate(
|
||||
event: Extract<SyncEvent, { type: SyncEventType.LocalCreate }>,
|
||||
record: DocumentRecord
|
||||
): void {
|
||||
const promise = event.resolvers?.promise;
|
||||
|
||||
this.documents.set(event.path, record);
|
||||
event.resolvers?.resolve(record.documentId);
|
||||
|
||||
if (promise !== undefined) {
|
||||
for (const e of this.events) {
|
||||
if (
|
||||
(e.type === SyncEventType.LocalUpdate || e.type === SyncEventType.LocalDelete) &&
|
||||
e.documentId === promise
|
||||
) {
|
||||
(e as { documentId: DocumentId | Promise<DocumentId> }).documentId = record.documentId;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.saveInTheBackground();
|
||||
}
|
||||
|
||||
public getCreatePromise(path: RelativePath): Promise<DocumentId> | undefined {
|
||||
const event = this.findLastCreate(path);
|
||||
public getLatestCreatePromise(path: RelativePath): Promise<DocumentId> | undefined {
|
||||
const event = this.findLatestCreate(path);
|
||||
if (event === undefined) return undefined;
|
||||
event.resolvers ??= Promise.withResolvers<DocumentId>();
|
||||
return event.resolvers.promise;
|
||||
|
|
@ -254,17 +274,6 @@ export class SyncEventQueue {
|
|||
);
|
||||
}
|
||||
|
||||
public async save(): Promise<void> {
|
||||
return this.saveData({
|
||||
documents: Array.from(this.documents.entries()).map(
|
||||
([relativePath, record]) => ({
|
||||
relativePath,
|
||||
...record
|
||||
})
|
||||
),
|
||||
lastSeenUpdateId: this.lastSeenUpdateId
|
||||
});
|
||||
}
|
||||
|
||||
public resetState(): void {
|
||||
this.rejectAllPendingCreates();
|
||||
|
|
@ -277,161 +286,11 @@ export class SyncEventQueue {
|
|||
this.events.length = 0;
|
||||
}
|
||||
|
||||
public enqueue(input: FileSyncEvent): void {
|
||||
if (input.type === SyncEventType.RemoteUpdate) {
|
||||
this.events.push(input);
|
||||
return;
|
||||
}
|
||||
|
||||
const { path } = input;
|
||||
|
||||
// Conflict-displaced files are local-only bookkeeping so a conflict
|
||||
// hit is a debug-level event. A hit against a user-configured glob
|
||||
// is a higher-signal "we're deliberately not syncing this" and
|
||||
// stays at info.
|
||||
if (isConflictPath(path)) {
|
||||
this.logger.debug(
|
||||
`Ignoring ${input.type} for ${path}: conflict-displaced file`
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (this.matchesUserIgnorePattern(path)) {
|
||||
this.logger.info(
|
||||
`Ignoring ${input.type} for ${path} as it matches ignore patterns`
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
if (input.type === SyncEventType.LocalCreate) {
|
||||
this.events.push({ type: SyncEventType.LocalCreate, path, originalPath: path });
|
||||
return;
|
||||
}
|
||||
|
||||
const lookupPath = (input.type === SyncEventType.LocalUpdate && input.oldPath) ? input.oldPath : path;
|
||||
const record = this.documents.get(lookupPath);
|
||||
const documentId: DocumentId | Promise<DocumentId> | undefined =
|
||||
record?.documentId ?? this.getCreatePromise(lookupPath);
|
||||
if (documentId === undefined) return;
|
||||
|
||||
if (input.type === SyncEventType.LocalDelete) {
|
||||
this.events.push({ type: SyncEventType.LocalDelete, documentId });
|
||||
return;
|
||||
}
|
||||
|
||||
if (input.oldPath !== undefined) {
|
||||
if (typeof documentId === "string") {
|
||||
this.documents.delete(input.oldPath);
|
||||
this.documents.set(path, record!);
|
||||
for (const e of this.events) {
|
||||
if (e.type === SyncEventType.LocalUpdate && e.documentId === documentId) {
|
||||
e.path = path;
|
||||
}
|
||||
}
|
||||
this.saveInTheBackground();
|
||||
} else {
|
||||
this.updatePendingCreatePath(input.oldPath, path);
|
||||
}
|
||||
}
|
||||
this.events.push({ type: SyncEventType.LocalUpdate, documentId, path, originalPath: path });
|
||||
}
|
||||
|
||||
|
||||
|
||||
public async next(): Promise<SyncEvent | undefined> {
|
||||
if (this.events.length === 0) return undefined;
|
||||
|
||||
const [first] = this.events;
|
||||
|
||||
// Creates are always returned immediately (FIFO)
|
||||
if (first.type === SyncEventType.LocalCreate) {
|
||||
this.events.shift();
|
||||
return first;
|
||||
}
|
||||
|
||||
// Deletes are returned immediately; also discard any subsequent
|
||||
// events for the same documentId so stale broadcasts don't
|
||||
// resurrect the document. If the documentId is still a pending
|
||||
// `Promise<DocumentId>` (the originating Create hasn't landed
|
||||
// yet), awaiting it may reject — handle that: the Create was
|
||||
// cancelled, so the Delete has nothing to delete, just drop it.
|
||||
if (first.type === SyncEventType.LocalDelete) {
|
||||
this.events.shift();
|
||||
const { documentId } = first;
|
||||
let resolvedId: DocumentId;
|
||||
try {
|
||||
resolvedId = await documentId;
|
||||
} catch {
|
||||
this.logger.debug(
|
||||
"Dropping Delete whose Create was cancelled before it could be synced"
|
||||
);
|
||||
return this.next();
|
||||
}
|
||||
this.removeAllEventsForDocumentId(resolvedId);
|
||||
return first;
|
||||
}
|
||||
|
||||
if (first.type === SyncEventType.LocalUpdate) {
|
||||
const { documentId } = first;
|
||||
|
||||
// If there's a later delete for the same documentId, discard
|
||||
// all sync-locals for that document and return the delete
|
||||
const deleteEvent = this.events.find(
|
||||
(e) =>
|
||||
e.type === SyncEventType.LocalDelete &&
|
||||
e.documentId === documentId
|
||||
);
|
||||
if (deleteEvent !== undefined) {
|
||||
let resolvedId: DocumentId;
|
||||
try {
|
||||
resolvedId = await documentId;
|
||||
} catch {
|
||||
this.logger.debug(
|
||||
"Dropping SyncLocal+Delete whose Create was cancelled before it could be synced"
|
||||
);
|
||||
return this.next();
|
||||
}
|
||||
this.removeAllEventsForDocumentId(resolvedId);
|
||||
return deleteEvent;
|
||||
}
|
||||
|
||||
// Coalesce multiple sync-locals for the same documentId and
|
||||
// original path to the last one
|
||||
const matching = this.events.filter(
|
||||
(e) =>
|
||||
e.type === SyncEventType.LocalUpdate &&
|
||||
e.documentId === documentId &&
|
||||
e.originalPath === first.originalPath // can't coalesce moves as they can depend on each other so we have to sync them in the same order, could do topological sort but let's keep it simple for now
|
||||
);
|
||||
const result = matching[matching.length - 1];
|
||||
for (const item of matching) {
|
||||
removeFromArray(this.events, item);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
// Coalesce multiple RemoteUpdate events for the same documentId
|
||||
// down to the last one — the `.next` walk already short-circuits
|
||||
// on obsolete versions via `parentVersionId` checks, but compacting
|
||||
// here keeps the queue bounded under burst remote activity.
|
||||
const { documentId } = first.remoteVersion;
|
||||
const matching = this.events.filter(
|
||||
(e) =>
|
||||
e.type === SyncEventType.RemoteUpdate &&
|
||||
e.remoteVersion.documentId === documentId
|
||||
);
|
||||
const result = matching[matching.length - 1];
|
||||
for (const item of matching) {
|
||||
removeFromArray(this.events, item);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private matchesUserIgnorePattern(path: RelativePath): boolean {
|
||||
return this.ignorePatterns.some((pattern) => pattern.test(path));
|
||||
}
|
||||
|
||||
private isIgnored(path: RelativePath): boolean {
|
||||
return isConflictPath(path) || this.matchesUserIgnorePattern(path);
|
||||
return this.ignorePatterns.some((pattern) => pattern.test(path));
|
||||
}
|
||||
|
||||
public removeAllEventsForDocumentId(documentId: DocumentId): void {
|
||||
|
|
@ -455,7 +314,7 @@ export class SyncEventQueue {
|
|||
oldPath: RelativePath,
|
||||
newPath: RelativePath
|
||||
): void {
|
||||
const createEvent = this.findLastCreate(oldPath);
|
||||
const createEvent = this.findLatestCreate(oldPath);
|
||||
if (createEvent === undefined) return;
|
||||
|
||||
const promise = createEvent.resolvers?.promise;
|
||||
|
|
@ -473,22 +332,7 @@ export class SyncEventQueue {
|
|||
}
|
||||
}
|
||||
|
||||
private findCreatePathByPromise(
|
||||
promise: Promise<DocumentId>
|
||||
): RelativePath | undefined {
|
||||
for (let i = this.events.length - 1; i >= 0; i--) {
|
||||
const e = this.events[i];
|
||||
if (
|
||||
e.type === SyncEventType.LocalCreate &&
|
||||
e.resolvers?.promise === promise
|
||||
) {
|
||||
return e.path;
|
||||
}
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
private findLastCreate(
|
||||
private findLatestCreate(
|
||||
path: RelativePath
|
||||
): Extract<SyncEvent, { type: SyncEventType.LocalCreate }> | undefined {
|
||||
for (let i = this.events.length - 1; i >= 0; i--) {
|
||||
|
|
@ -506,7 +350,7 @@ export class SyncEventQueue {
|
|||
* merging it with a concurrent remote create.
|
||||
*/
|
||||
public hasPendingCreateAt(path: RelativePath): boolean {
|
||||
return this.findLastCreate(path) !== undefined;
|
||||
return this.findLatestCreate(path) !== undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -517,7 +361,7 @@ export class SyncEventQueue {
|
|||
* and cancelled.
|
||||
*/
|
||||
public cancelPendingCreate(path: RelativePath): boolean {
|
||||
const event = this.findLastCreate(path);
|
||||
const event = this.findLatestCreate(path);
|
||||
if (event === undefined) return false;
|
||||
|
||||
if (event.resolvers !== undefined) {
|
||||
|
|
|
|||
|
|
@ -44,7 +44,7 @@ export class Syncer {
|
|||
|
||||
private readonly queue: SyncEventQueue;
|
||||
|
||||
private _isFirstSyncComplete = false;
|
||||
private _isFirstSyncStarted = false;
|
||||
private runningScheduleSyncForOfflineChanges: Promise<void> | undefined;
|
||||
private draining: Promise<void> | undefined;
|
||||
private previousRemainingOperationsCount = 0;
|
||||
|
|
@ -66,14 +66,6 @@ export class Syncer {
|
|||
this.webSocketManager.onWebSocketStatusChanged.add((isConnected) => {
|
||||
if (isConnected) {
|
||||
this.sendHandshakeMessage();
|
||||
// The server no longer carries an `is_initial_sync`
|
||||
// terminator: it streams missed versions as individual
|
||||
// VaultUpdates and then behaves like a live subscription.
|
||||
// Mark first-sync as complete once we've observed the
|
||||
// transition to "connected" — per-path sync status still
|
||||
// relies on `hasPendingEventsForPath`, which correctly
|
||||
// shows SYNCING while catch-up events are in flight.
|
||||
this._isFirstSyncComplete = true;
|
||||
}
|
||||
});
|
||||
this.webSocketManager.onRemoteVaultUpdateReceived.add(
|
||||
|
|
@ -82,7 +74,7 @@ export class Syncer {
|
|||
}
|
||||
|
||||
public get isFirstSyncComplete(): boolean {
|
||||
return this._isFirstSyncComplete;
|
||||
return this._isFirstSyncStarted;
|
||||
}
|
||||
|
||||
public syncLocallyCreatedFile(relativePath: RelativePath): void {
|
||||
|
|
@ -110,11 +102,7 @@ export class Syncer {
|
|||
}
|
||||
|
||||
|
||||
// Handler for every `WebSocketVaultUpdate` the server emits. The
|
||||
// server filters out messages authored by this device, so every
|
||||
// update here comes from a peer (or is part of the catch-up stream
|
||||
// the server replays on connect for versions we missed while
|
||||
// offline).
|
||||
|
||||
public async syncRemotelyUpdatedFile(
|
||||
message: WebSocketVaultUpdate
|
||||
): Promise<void> {
|
||||
|
|
@ -126,6 +114,8 @@ export class Syncer {
|
|||
});
|
||||
|
||||
this.ensureDraining();
|
||||
|
||||
this._isFirstSyncStarted = true;
|
||||
}
|
||||
|
||||
public async scheduleSyncForOfflineChanges(): Promise<void> {
|
||||
|
|
@ -167,7 +157,7 @@ export class Syncer {
|
|||
|
||||
|
||||
public reset(): void {
|
||||
this._isFirstSyncComplete = false;
|
||||
this._isFirstSyncStarted = false;
|
||||
this.queue.clear();
|
||||
// Don't null the reference synchronously — if the scan is
|
||||
// still in flight, the next reconnect would spawn a second
|
||||
|
|
@ -220,14 +210,12 @@ export class Syncer {
|
|||
);
|
||||
});
|
||||
|
||||
await this.scheduleDrain();
|
||||
this.ensureDraining();
|
||||
await this.draining;
|
||||
}
|
||||
|
||||
|
||||
|
||||
private ensureDraining(): void {
|
||||
void this.chainOntoDrain(async () => this.drain());
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialize a unit of work onto the same promise chain the drain
|
||||
|
|
@ -248,12 +236,11 @@ export class Syncer {
|
|||
);
|
||||
return chained;
|
||||
}
|
||||
|
||||
private async scheduleDrain(): Promise<void> {
|
||||
this.ensureDraining();
|
||||
await this.draining;
|
||||
private ensureDraining(): void {
|
||||
void this.chainOntoDrain(async () => this.drain());
|
||||
}
|
||||
|
||||
|
||||
private async drain(): Promise<void> {
|
||||
let event = await this.queue.next();
|
||||
while (event !== undefined) {
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ export interface StoredDocument extends DocumentRecord {
|
|||
}
|
||||
|
||||
export interface StoredSyncState {
|
||||
documents: StoredDocument[];
|
||||
documents: StoredDocument[] | undefined;
|
||||
lastSeenUpdateId: VaultUpdateId | undefined;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ mod ping;
|
|||
mod rate_limit;
|
||||
mod requests;
|
||||
mod responses;
|
||||
mod restore_document_version;
|
||||
mod update_document;
|
||||
mod websocket;
|
||||
|
||||
|
|
@ -174,10 +173,6 @@ fn get_authed_routes(app_state: AppState) -> Router<AppState> {
|
|||
"/vaults/:vault_id/documents/:document_id",
|
||||
delete(delete_document::delete_document),
|
||||
)
|
||||
.route(
|
||||
"/vaults/:vault_id/documents/:document_id/restore",
|
||||
post(restore_document_version::restore_document_version),
|
||||
)
|
||||
.route(
|
||||
"/vaults/:vault_id/history",
|
||||
get(fetch_vault_history::fetch_vault_history),
|
||||
|
|
|
|||
|
|
@ -1,186 +0,0 @@
|
|||
use anyhow::anyhow;
|
||||
use axum::{
|
||||
Extension, Json,
|
||||
extract::{Path, State},
|
||||
};
|
||||
use axum_extra::TypedHeader;
|
||||
use log::{debug, info};
|
||||
use serde::Deserialize;
|
||||
|
||||
use super::device_id_header::DeviceIdHeader;
|
||||
use crate::{
|
||||
app_state::{
|
||||
AppState,
|
||||
database::{
|
||||
InsertBroadcast,
|
||||
models::{
|
||||
DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId,
|
||||
VaultUpdateId,
|
||||
},
|
||||
},
|
||||
},
|
||||
config::user_config::User,
|
||||
errors::{
|
||||
SyncServerError, client_error, not_found_error, server_error, write_transaction_error,
|
||||
},
|
||||
utils::{find_first_available_path::find_first_available_path, normalize::normalize},
|
||||
};
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct RestorePathParams {
|
||||
#[serde(deserialize_with = "normalize")]
|
||||
vault_id: VaultId,
|
||||
|
||||
document_id: DocumentId,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RestoreDocumentVersionRequest {
|
||||
pub vault_update_id: VaultUpdateId,
|
||||
}
|
||||
|
||||
#[axum::debug_handler]
|
||||
pub async fn restore_document_version(
|
||||
Path(RestorePathParams {
|
||||
vault_id,
|
||||
document_id,
|
||||
}): Path<RestorePathParams>,
|
||||
Extension(user): Extension<User>,
|
||||
TypedHeader(device_id): TypedHeader<DeviceIdHeader>,
|
||||
State(state): State<AppState>,
|
||||
Json(request): Json<RestoreDocumentVersionRequest>,
|
||||
) -> Result<Json<DocumentVersionWithoutContent>, SyncServerError> {
|
||||
debug!(
|
||||
"Restoring document `{document_id}` in vault `{vault_id}` to version `{}`",
|
||||
request.vault_update_id
|
||||
);
|
||||
|
||||
if request.vault_update_id <= 0 {
|
||||
return Err(client_error(anyhow!(
|
||||
"Invalid vault_update_id: `{}`",
|
||||
request.vault_update_id
|
||||
)));
|
||||
}
|
||||
|
||||
let mut transaction = state
|
||||
.database
|
||||
.create_write_transaction(&vault_id)
|
||||
.await
|
||||
.map_err(write_transaction_error)?;
|
||||
|
||||
let target_version = state
|
||||
.database
|
||||
.get_document_version(&vault_id, request.vault_update_id, Some(&mut *transaction))
|
||||
.await
|
||||
.map_err(server_error)?
|
||||
.ok_or_else(|| {
|
||||
not_found_error(anyhow!("Version `{}` not found", request.vault_update_id))
|
||||
})?;
|
||||
|
||||
if target_version.document_id != document_id {
|
||||
transaction.rollback().await.map_err(server_error)?;
|
||||
return Err(not_found_error(anyhow!(
|
||||
"Version `{}` does not belong to document `{document_id}`",
|
||||
request.vault_update_id,
|
||||
)));
|
||||
}
|
||||
|
||||
if target_version.is_deleted {
|
||||
transaction.rollback().await.map_err(server_error)?;
|
||||
return Err(client_error(anyhow!(
|
||||
"Cannot restore to a deleted version `{}`",
|
||||
request.vault_update_id,
|
||||
)));
|
||||
}
|
||||
|
||||
let existing = state
|
||||
.database
|
||||
.get_latest_non_deleted_document_by_path(
|
||||
&vault_id,
|
||||
&target_version.relative_path,
|
||||
Some(&mut *transaction),
|
||||
)
|
||||
.await
|
||||
.map_err(server_error)?;
|
||||
|
||||
let restore_path = if let Some(existing_doc) = &existing
|
||||
&& existing_doc.document_id != document_id
|
||||
{
|
||||
find_first_available_path(
|
||||
&vault_id,
|
||||
&target_version.relative_path,
|
||||
&state.database,
|
||||
&mut transaction,
|
||||
)
|
||||
.await
|
||||
.map_err(server_error)?
|
||||
} else {
|
||||
target_version.relative_path.clone()
|
||||
};
|
||||
|
||||
let last_update_id = state
|
||||
.database
|
||||
.get_max_update_id_in_vault(&vault_id, Some(&mut *transaction))
|
||||
.await
|
||||
.map_err(server_error)?;
|
||||
|
||||
// The current latest (pre-restore) is our baseline for deciding
|
||||
// whether content and/or path actually change.
|
||||
let current_latest = state
|
||||
.database
|
||||
.get_latest_document(&vault_id, &document_id, Some(&mut *transaction))
|
||||
.await
|
||||
.map_err(server_error)?;
|
||||
|
||||
let new_version = StoredDocumentVersion {
|
||||
vault_update_id: last_update_id + 1,
|
||||
creation_vault_update_id: target_version.creation_vault_update_id,
|
||||
document_id,
|
||||
relative_path: restore_path,
|
||||
content: target_version.content,
|
||||
updated_date: chrono::Utc::now(),
|
||||
is_deleted: false,
|
||||
user_id: user.name.clone(),
|
||||
device_id: device_id.0.clone(),
|
||||
has_been_merged: false,
|
||||
};
|
||||
|
||||
let (content_changed, path_changed) = match ¤t_latest {
|
||||
Some(prev) => (
|
||||
prev.content != new_version.content || prev.is_deleted,
|
||||
// Mirror `update_document`: `path_changed` is true when the
|
||||
// stored path differs from either the prior stored path (peers
|
||||
// need to learn about the move) *or* from the path the caller
|
||||
// implicitly requested (`target_version.relative_path`, so the
|
||||
// origin learns if the server deduped its requested restore
|
||||
// path).
|
||||
prev.relative_path != new_version.relative_path
|
||||
|| target_version.relative_path != new_version.relative_path,
|
||||
),
|
||||
// No prior version (shouldn't happen in practice — target_version
|
||||
// already proved the document exists — but treat defensively).
|
||||
None => (true, true),
|
||||
};
|
||||
|
||||
state
|
||||
.database
|
||||
.insert_document_version(
|
||||
&vault_id,
|
||||
&new_version,
|
||||
transaction,
|
||||
InsertBroadcast {
|
||||
content_changed,
|
||||
path_changed,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.map_err(server_error)?;
|
||||
|
||||
info!(
|
||||
"Restored document `{document_id}` to version `{}` as new version `{}`",
|
||||
request.vault_update_id, new_version.vault_update_id
|
||||
);
|
||||
|
||||
Ok(Json(new_version.into()))
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue