src/vault/changeQueue.ts - master - raiden/obsidian-git-encrypt

src/vault/changeQueue.ts
196 строк6 KB
Robert Kuzhin
feat: sync local vault changes to encrypted mirror
09 авг 2026, 10:28
09 авг 2026, 10:28994d9e6
import type { VaultChange } from "./vaultTypes.js";

export interface ChangeQueueOptions {
  readonly debounceMs?: number;
  readonly onError?: (error: unknown) => void;
}

export class ChangeQueue {
  private pending: VaultChange[] = [];
  private timer: ReturnType<typeof setTimeout> | undefined;
  private chain: Promise<void> = Promise.resolve();
  private stopped = false;
  private readonly debounceMs: number;
  private readonly onError: (error: unknown) => void;

  public constructor(
    private readonly handler: (changes: readonly VaultChange[]) => Promise<void>,
    options: ChangeQueueOptions = {},
  ) {
    this.debounceMs = options.debounceMs ?? 250;
    this.onError = options.onError ?? (() => undefined);
    if (!Number.isSafeInteger(this.debounceMs) || this.debounceMs < 0) {
      throw new RangeError("debounceMs must be a non-negative safe integer");
    }
  }

  public enqueue(change: VaultChange): void {
    if (this.stopped) throw new Error("Change queue is stopped");
    this.pending = coalesceChanges([...this.pending, change]);
    this.schedule();
  }

  public flush(): Promise<void> {
    this.clearTimer();
    const run = async (): Promise<void> => {
      while (this.pending.length > 0) {
        const batch = this.pending;
        this.pending = [];
        try {
          await this.handler(batch);
        } catch (error) {
          this.pending = coalesceChanges([...batch, ...this.pending]);
          throw error;
        }
      }
    };
    const result = this.chain.then(run, run);
    this.chain = result.catch(() => undefined);
    return result;
  }

  public async stop(): Promise<void> {
    this.stopped = true;
    await this.flush();
  }

  private schedule(): void {
    this.clearTimer();
    this.timer = setTimeout(() => {
      this.timer = undefined;
      void this.flush().catch((error: unknown) => {
        try {
          this.onError(error);
        } catch {
          // Error observers must not create an unhandled rejection or alter retry state.
        }
      });
    }, this.debounceMs);
  }

  private clearTimer(): void {
    if (this.timer !== undefined) {
      clearTimeout(this.timer);
      this.timer = undefined;
    }
  }
}

export function coalesceChanges(changes: readonly VaultChange[]): VaultChange[] {
  const result: VaultChange[] = [];
  for (const change of changes) {
    if (change.type === "modify") {
      if (hasPendingCreate(result, change.path) || hasPendingModify(result, change.path)) continue;
      result.push(change);
      continue;
    }

    if (change.type === "create") {
      result.push(change);
      continue;
    }

    if (change.type === "rename") {
      const createIndex = findLastIndex(
        result,
        (candidate) => candidate.type === "create" && candidate.path === change.oldPath,
      );
      if (createIndex >= 0) {
        result[createIndex] = { type: "create", path: change.newPath };
        removeModifies(result, change.oldPath);
        continue;
      }

      const renameIndex = findLastIndex(
        result,
        (candidate) => candidate.type === "rename" && candidate.newPath === change.oldPath,
      );
      if (renameIndex >= 0) {
        const previous = result[renameIndex]!;
        if (previous.type === "rename") {
          result[renameIndex] = { type: "rename", oldPath: previous.oldPath, newPath: change.newPath };
          retargetModifies(result, change.oldPath, change.newPath);
          continue;
        }
      }

      const hadModify = removeModifies(result, change.oldPath);
      result.push(change);
      if (hadModify) result.push({ type: "modify", path: change.newPath });
      continue;
    }

    const createIndex = findLastIndex(
      result,
      (candidate) => candidate.type === "create" && candidate.path === change.path,
    );
    if (createIndex >= 0) {
      result.splice(createIndex, 1);
      removeModifies(result, change.path);
      continue;
    }

    const renameIndex = findLastIndex(
      result,
      (candidate) => candidate.type === "rename" && candidate.newPath === change.path,
    );
    if (renameIndex >= 0) {
      const previous = result[renameIndex]!;
      if (previous.type === "rename") {
        result.splice(renameIndex, 1);
        removeModifies(result, change.path);
        appendDelete(result, previous.oldPath);
        continue;
      }
    }

    removeModifies(result, change.path);
    appendDelete(result, change.path);
  }
  return result;
}

function hasPendingCreate(changes: readonly VaultChange[], path: string): boolean {
  return changes.some((change) => change.type === "create" && change.path === path);
}

function hasPendingModify(changes: readonly VaultChange[], path: string): boolean {
  return changes.some((change) => change.type === "modify" && change.path === path);
}

function removeModifies(changes: VaultChange[], path: string): boolean {
  let removed = false;
  for (let index = changes.length - 1; index >= 0; index -= 1) {
    const change = changes[index]!;
    if (change.type === "modify" && change.path === path) {
      changes.splice(index, 1);
      removed = true;
    }
  }
  return removed;
}

function retargetModifies(changes: VaultChange[], oldPath: string, newPath: string): void {
  for (let index = 0; index < changes.length; index += 1) {
    const change = changes[index]!;
    if (change.type === "modify" && change.path === oldPath) {
      changes[index] = { type: "modify", path: newPath };
    }
  }
}

function appendDelete(changes: VaultChange[], path: string): void {
  if (!changes.some((change) => change.type === "delete" && change.path === path)) {
    changes.push({ type: "delete", path });
  }
}

function findLastIndex(
  changes: readonly VaultChange[],
  predicate: (change: VaultChange) => boolean,
): number {
  for (let index = changes.length - 1; index >= 0; index -= 1) {
    if (predicate(changes[index]!)) return index;
  }
  return -1;
}