Method fetchChanges

packages/rpc/src/server.ts:165–192

(input: { after?: ChangeCursor; ignore?: string[] })

  }

  async fetchChanges(input: { after?: ChangeCursor; ignore?: string[] }): Promise<{
    currentCursor: ChangeCursor;
    appliedPushCursor: ChangeCursor;
    stream: ReadableStream<ChangeEntry>;
  }> {
    if (this.options.beforeFetch !== undefined) {
      try {
        await this.options.beforeFetch();
      } catch (err) {
        // Settle hook failures must not surface as fetch failures —
        // we still want to stream whatever's already in the store.
        // Log so the operator notices a wedged shim, then carry on.
        console.warn("[SyncRPCServer] beforeFetch hook failed:", err);
      }
    }
    const after = input.after ?? { rev: 0, path: null };
    const ignore =
      input.ignore ?? (this.options.ignore.length > 0 ? this.options.ignore : DEFAULT_IGNORE);
    const snapshotRev = currentRev(this.db);
    const currentCursor = { rev: snapshotRev, path: null };
    return {
      currentCursor,
      appliedPushCursor: readFetchCursor(this.db),
      stream: iterableToReadableStream(
        coalesceChanges(this.db, after, { ignore, through: currentCursor }),
      ),
    };
  }

  async readEntry(path: string): Promise<ChangeEntry | null> {
    return materialiseChange(this.db, path);