quak/src/library/index.ts

// The library surface over the local cache.
//
// `Library.open()` loads the on-disk metadata store (issue #41), then starts
// the refresh loop. When the cache loaded empty it awaits the first refresh,
// so the library never opens onto an empty store it could have filled; when an
// existing copy loaded, that first refresh runs in the background and `open()`
// returns as soon as the cached data is ready to serve — a slow or unreachable
// server no longer stalls opening. A background timer then refreshes every
// `refreshIntervalSeconds`. Every default read is answered from RAM — no
// default read touches the network. There is deliberately no `sync()`, no
// `refresh()`, no `serverReachable` flag, and no "before each read" mode
// (design #36).
//
// `fresh()` is the one exception (issue #75, an owner amendment to #36): it
// forces a refresh, awaits it, and only then hands back the read namespaces, so
// a caller that needs server-current data can ask for it. Concurrent `fresh()`
// calls coalesce onto one in-flight refresh, and a refresh that fails rejects
// the caller (the default reads stay silent and serve the last good copy). The
// default methods and the background loop are unchanged.
//
// A refresh stages all of its network work first and only mutates the store
// once every fetch has succeeded. A refresh that fails partway therefore never
// becomes visible to reads: the last good snapshot stays in place, and the
// failure surfaces through `onProgress` and `status()` instead. A commit that
// mutates RAM but then fails to persist keeps `status().lastError` set and the
// store marked unsaved until a later save actually lands, so a stuck disk is
// never masked by a subsequent empty refresh.

import { rm } from "node:fs/promises";
import { join } from "node:path";
import envPaths from "env-paths";

import { MetadataStore } from "./store.js";
import { MLDataStore } from "./mldata.js";
import { RequestPools } from "./pools.js";
import {
    deriveRecords,
    snapshotFrom,
    diffRecords,
    type DerivedRecords,
    type LibrarySnapshot,
    type LibraryChange,
} from "./records.js";
import {
    makeAlbumsAPI,
    makePhotosAPI,
    makeTimelineAPI,
    type AlbumsAPI,
    type PhotosAPI,
    type TimelineAPI,
    type FreshReads,
} from "./read.js";
import {
    ContentCache,
    type ContentSource,
    type ThumbnailsAPI,
    type EnsureOptions,
    type EnsureResult,
} from "./content.js";
import { makeMLDataAPI, type MLDataAPI } from "./mlsearch.js";
import { Precache } from "./precache.js";

export {
    Album,
    Photo,
    type AlbumsAPI,
    type PhotosAPI,
    type TimelineAPI,
    type FreshReads,
    type PhotoFilter,
    type TimelineGroup,
    type GroupBy,
} from "./read.js";
export {
    type ContentSource,
    type ContentResult,
    type ContentEvent,
    type ContentOptions,
    type PhotoContent,
    type ThumbnailsAPI,
    type ThumbnailPriority,
    type EnsureOptions,
    type EnsureResult,
    type EnsureEvent,
} from "./content.js";
export { type MLDataAPI, type SimilarResult } from "./mlsearch.js";
import type { CollectionsPage, FilesPage } from "../client.js";
import { MLDATA_BATCH_SIZE, type MLData } from "../mldata-fetch.js";
import type { Collection, EnteFile } from "../model/types.js";
import { runBackup, type BackupOptions, type BackupResult } from "../backup.js";

export {
    runBackup,
    type BackupOptions,
    type BackupResult,
    type BackupError,
} from "../backup.js";

export const DEFAULT_REFRESH_INTERVAL_SECONDS = 3;

// The account's cache directory when `cacheDirectory` is not given: the
// env-paths cache directory plus the user id, so each account has its own.
export const defaultCacheDirectory = (userID: number): string =>
    join(envPaths("quak", { suffix: "" }).cache, String(userID));

// Project a metadata store into by-id records, filling each record's cache
// paths from the content cache when one is given. Shared by the live read
// projection and the precache's initial seeding at open().
const deriveRecordsFromStore = (
    store: MetadataStore,
    cache?: ContentCache,
): DerivedRecords => {
    const collections = store.listCollections();
    const files: EnteFile[] = [];
    for (const c of collections) files.push(...store.listFiles(c.id));
    return deriveRecords(
        collections,
        files,
        cache ? (fileID) => cache.pathsFor(fileID) : undefined,
    );
};

// The slice of `Client` the library depends on. Narrowing to an interface lets
// tests drive a mock with no crypto or network; the real `Client` satisfies it
// structurally.
export interface LibraryClient {
    whoami(): { email: string; userID: number };
    collectionsSince(args: { sinceTime: number }): Promise<CollectionsPage>;
    filesSince(args: {
        collectionID: number;
        collectionKey: Uint8Array;
        sinceTime: number;
    }): Promise<FilesPage>;
    // Fetch ML data (face detections + CLIP embeddings) for up to a batch of
    // files. Optional: a client without it simply disables ML fetching, leaving
    // the metadata refresh untouched.
    fetchMLData?(args: {
        fileIDs: number[];
        fileKeys: Map<number, Uint8Array>;
    }): Promise<Map<number, MLData>>;
    // The byte source for the on-disk content cache. Optional so a mock client
    // that only serves metadata still satisfies the interface; when absent (and
    // no explicit `contentSource` is passed to `open`) the content cache is
    // disabled and `Photo.original`/`thumbnail` and `thumbnails.ensure` throw.
    contentSource?(): ContentSource;
}

// A progress event for one unit of background work. A metadata "refresh" or an
// ML "fetchMLData" pass each fire "started" before their network work and then
// exactly one of "done" or "failed"; "failed" carries the error message and
// an ML "done" reports how many payloads it stored. The precache fills
// ("precacheThumbnails"/"precacheOriginals", #48) fire "started"/"done" around
// each sweep that has work, "done" reporting the count newly cached.
export interface RefreshEvent {
    operation:
        | "refresh"
        | "fetchMLData"
        | "precacheThumbnails"
        | "precacheOriginals";
    status: "started" | "done" | "failed";
    error?: string;
    fetched?: number;
}

export type RefreshProgressCallback = (event: RefreshEvent) => void;

export interface LibraryOptions {
    client: LibraryClient;
    // Where `metadata.json` lives. Defaults to the env-paths cache directory
    // plus the user id, so each account has its own cache.
    cacheDirectory?: string;
    // Persistent backup destination. The refresh loop does not use it; the
    // content cache treats an original already stored there as present.
    downloadDirectory?: string;
    refreshIntervalSeconds?: number;
    onProgress?: RefreshProgressCallback;
    // The bounded request pools (issue #45), shared by the ML-data fetch (the
    // metadata pool) and the content cache. Defaults to a fresh set at the
    // design's caps.
    pools?: RequestPools;
    // Overrides the client's own `contentSource()`; mainly for tests that drive
    // the cache with a stand-in source.
    contentSource?: ContentSource;
    // Bound on `cacheDirectory/originals` (default 100 GiB) and the free space
    // to protect on its volume (default 50 GiB). The effective limit adapts
    // down as the disk fills; `status().originalsLimitBytes` reports it.
    cacheOriginalsMaxBytes?: number;
    freeBelowBytes?: number;
    // An extra pinned predicate OR-ed with the precache's own pinned set
    // (favorites + latest week, #48). Pinned originals are never evicted.
    isOriginalPinned?: (fileID: number) => boolean;
    // The aggressive local precache (#48), all starting inside `open()` with no
    // caller input. Thumbnails: every file, newest first, until all are on
    // disk. Originals: the favorites album then the latest `precacheOriginalsDays`
    // window (the days ending at the newest file). Both default on; the days
    // default to 7.
    precacheThumbnails?: boolean;
    precacheOriginals?: boolean;
    precacheOriginalsDays?: number;
}

export interface LibraryStatus {
    userID: number;
    collections: number;
    files: number;
    // Wall-clock ms of the last refresh that succeeded, or undefined if none
    // has yet.
    lastRefreshAt?: number;
    // The message from the most recent refresh, set only while that refresh
    // failed; cleared by the next success.
    lastError?: string;
    // Wall-clock ms of the last ML fetch pass that succeeded, or undefined if
    // none has yet (or ML fetching is disabled).
    lastMLFetchAt?: number;
    // The most recent ML fetch pass's error, set only while it failed.
    lastMLError?: string;
    // ML payloads stored on disk and CLIP embeddings in the index; undefined
    // when ML fetching is disabled.
    mlStored?: number;
    mlIndexed?: number;
    // Bytes stored in the originals cache and the effective size limit as of the
    // last write or open; undefined when no content cache is open.
    originalsUsedBytes?: number;
    originalsLimitBytes?: number;
    // Precache progress (#48); undefined when no content cache is open. Totals
    // are the files targeted (0 when a fill is disabled); "cached" is how many
    // of them are on disk.
    thumbnailsCached?: number;
    thumbnailsTotal?: number;
    originalsCached?: number;
    originalsPinned?: number;
    closed: boolean;
}

export class Library {
    readonly cacheDirectory: string;
    readonly downloadDirectory?: string;

    // The in-process read surface (issue #44). Each namespace answers
    // synchronously from the live record projection; no read touches the
    // network.
    readonly albums: AlbumsAPI;
    readonly photos: PhotosAPI;
    readonly timeline: TimelineAPI;
    // The thumbnail-prefetch surface (issue #46): drives the thumbnail pool
    // with priority, dedup, and abort.
    readonly thumbnails: ThumbnailsAPI;
    // The content-similarity search surface over the CLIP index (issue #50).
    // Present whether or not ML fetching is enabled; with no ML store it
    // returns empty results.
    readonly mldata: MLDataAPI;

    private readonly client: LibraryClient;
    private readonly store: MetadataStore;
    // The on-disk content cache, or undefined when no content source is
    // available (a metadata-only client with no explicit source).
    private readonly cache?: ContentCache;
    private readonly userID: number;
    private readonly intervalMs: number;
    private readonly onProgress?: RefreshProgressCallback;
    private readonly pools: RequestPools;
    // The ML-data cache, present only when the client can fetch ML data.
    private readonly mlStore?: MLDataStore;
    // The local precache (#48), present only when the content cache is.
    private readonly precache?: Precache;

    private timer?: ReturnType<typeof setTimeout>;
    // The in-flight refresh cycle, or undefined when none runs. One slot serves
    // both paths: the background loop skips when it is set, and a fresh read
    // (issue #75) coalesces onto it or starts one. The promise carries the
    // cycle's real outcome (it rejects on failure); the background loop ignores
    // that, a fresh read propagates it.
    private cycle?: Promise<void>;
    // Guards the ML fetch pass so a slow backfill never runs twice at once; a
    // refresh whose pass is still running kicks nothing new. Holds the running
    // pass, so `close()` can wait for it.
    private mlFetch?: Promise<void>;
    private closed = false;
    private lastRefreshAt?: number;
    private lastError?: string;
    private lastMLFetchAt?: number;
    private lastMLError?: string;
    // The plain-record projection as of the last refresh, and the GUI change
    // subscribers. A refresh that alters the projection notifies each with the
    // delta; `lastRecords` is kept current every refresh so a subscriber that
    // joins later diffs against the state its own `snapshot()` already returned.
    private readonly subscribers = new Set<(change: LibraryChange) => void>();
    private lastRecords: DerivedRecords;
    // RAM holds changes disk has not yet accepted (an earlier save failed).
    // Cleared only when a save actually succeeds; keeps the store trying to
    // persist and the failure visible in `status()` until then.
    private unsaved = false;

    private constructor(args: {
        client: LibraryClient;
        store: MetadataStore;
        userID: number;
        cacheDirectory: string;
        downloadDirectory?: string;
        intervalMs: number;
        onProgress?: RefreshProgressCallback;
        pools: RequestPools;
        mldata?: MLDataStore;
        cache?: ContentCache;
        precache?: Precache;
    }) {
        this.client = args.client;
        this.store = args.store;
        this.userID = args.userID;
        this.cacheDirectory = args.cacheDirectory;
        this.downloadDirectory = args.downloadDirectory;
        this.intervalMs = args.intervalMs;
        this.onProgress = args.onProgress;
        this.pools = args.pools;
        this.mlStore = args.mldata;
        this.cache = args.cache;
        this.precache = args.precache;
        this.lastRecords = this.deriveNow();

        // The read namespaces derive fresh from the store on each call, so they
        // always reflect the latest refresh.
        const derive = (): DerivedRecords => this.deriveNow();
        this.albums = makeAlbumsAPI(derive, this.cache);
        this.photos = makePhotosAPI(derive, this.cache);
        this.timeline = makeTimelineAPI(derive);
        this.thumbnails = {
            ensure: (opts: EnsureOptions): Promise<EnsureResult[]> => {
                if (!this.cache) {
                    return Promise.reject(
                        new Error(
                            "thumbnails.ensure requires a library opened with a content cache",
                        ),
                    );
                }
                return this.cache.ensureThumbnails(opts);
            },
        };
        // Reads the ML store live so results grow as ML data is fetched.
        this.mldata = makeMLDataAPI(() => this.mlStore);
    }

    // Load the cache and start the refresh loop. With an empty cache the first
    // refresh is awaited, so `open()` resolves onto populated data whenever the
    // server is reachable; that awaited refresh may still fail, and the library
    // then opens empty with the failure recorded in `status()`. With an
    // existing cache the first refresh runs in the background and `open()`
    // returns as soon as the cached data is ready — an unreachable server does
    // not block opening.
    static async open(opts: LibraryOptions): Promise<Library> {
        const { userID } = opts.client.whoami();
        const cacheDirectory =
            opts.cacheDirectory ?? defaultCacheDirectory(userID);
        const metadataPath = join(cacheDirectory, "metadata.json");
        let store = await MetadataStore.load(metadataPath);
        // A cache directory given explicitly can hold another account's cache.
        // Its records and cursor are not this account's, so delete it and the
        // ML data beside it and start empty. A user ID of 0 means the cache
        // was never refreshed and so holds nothing to discard.
        if (store.userID !== 0 && store.userID !== userID) {
            await rm(metadataPath, { force: true });
            await rm(join(cacheDirectory, "mldata"), {
                recursive: true,
                force: true,
            });
            store = await MetadataStore.load(metadataPath);
        }
        const intervalMs =
            (opts.refreshIntervalSeconds ?? DEFAULT_REFRESH_INTERVAL_SECONDS) *
            1000;

        // One request-pool set serves both the ML-data fetch and the content
        // cache, so both honour the same concurrency caps.
        const pools = opts.pools ?? new RequestPools();

        // The ML cache only earns its keep when the client can fetch ML data;
        // a client without that capability opens no `mldata/` directory.
        const mldata = opts.client.fetchMLData
            ? await MLDataStore.open(join(cacheDirectory, "mldata"))
            : undefined;

        // Build the content cache from an explicit source or the client's own,
        // and take its record of what is already cached (and reap orphan temp
        // files) before the first projection, so cached paths are present from
        // the start and the first refresh raises no spurious path-change diff.
        const source = opts.contentSource ?? opts.client.contentSource?.();
        let cache: ContentCache | undefined;
        let precache: Precache | undefined;
        if (source) {
            // The precache owns the pinned set (favorites + latest week), which
            // is the cache's eviction predicate. It is built and seeded from
            // the loaded store first so the cache can wire `isPinned` to it, and
            // then bound to the cache it fills. A caller-supplied predicate is
            // OR-ed in so both survive.
            precache = new Precache({
                thumbnails: opts.precacheThumbnails,
                originals: opts.precacheOriginals,
                originalsDays: opts.precacheOriginalsDays,
                onEvent: opts.onProgress,
            });
            precache.update(deriveRecordsFromStore(store));
            const extraPinned = opts.isOriginalPinned;
            cache = new ContentCache({
                pools,
                source,
                cacheDirectory,
                downloadDirectory: opts.downloadDirectory,
                getFile: (fileID) => store.getFileByID(fileID),
                cacheOriginalsMaxBytes: opts.cacheOriginalsMaxBytes,
                freeBelowBytes: opts.freeBelowBytes,
                isPinned: (fileID) =>
                    precache!.isPinned(fileID) ||
                    (extraPinned?.(fileID) ?? false),
            });
            await cache.open();
            precache.bind(cache);
        }

        const lib = new Library({
            client: opts.client,
            store,
            userID,
            cacheDirectory,
            downloadDirectory: opts.downloadDirectory,
            intervalMs,
            onProgress: opts.onProgress,
            pools,
            mldata,
            cache,
            precache,
        });

        // Start filling from whatever the loaded store already holds; each
        // refresh below re-kicks with the new files (and retries any that
        // failed). An empty store starts empty here and fills after its first
        // refresh.
        precache?.start();

        if (store.loadedFromDisk) {
            // An existing copy already answers reads; refresh in the background
            // and start the interval once that first cycle settles.
            void lib.runRefresh().then(() => lib.scheduleNext());
        } else {
            // Nothing was cached: wait for the first refresh to fill the store
            // (or fail) rather than resolve onto an empty library.
            await lib.runRefresh();
            lib.scheduleNext();
        }
        return lib;
    }

    listCollections(): Collection[] {
        return this.store.listCollections();
    }

    getCollection(id: number): Collection | undefined {
        return this.store.getCollection(id);
    }

    listFiles(collectionID: number): EnteFile[] {
        return this.store.listFiles(collectionID);
    }

    getFile(collectionID: number, fileID: number): EnteFile | undefined {
        return this.store.getFile(collectionID, fileID);
    }

    // Any membership of a file, addressed by file id alone. A file's own
    // metadata (title, creationTime) is identical across the collections it
    // belongs to, so this serves the point commands that hold only a fileID.
    getFileByID(fileID: number): EnteFile | undefined {
        return this.store.getFileByID(fileID);
    }

    // A synchronous, RAM-only projection of the whole library into plain
    // records (no keys), the surface the GUI reads across IPC. Photos are
    // deduplicated to one record per file and ordered newest first.
    snapshot(): LibrarySnapshot {
        return snapshotFrom(this.deriveNow(), Date.now());
    }

    // Deliver a `LibraryChange` whenever a refresh alters the projection. A
    // refresh that changes nothing delivers nothing. The returned handle's
    // `unsubscribe` stops delivery.
    subscribe(args: { onChange: (change: LibraryChange) => void }): {
        unsubscribe: () => void;
    } {
        const { onChange } = args;
        this.subscribers.add(onChange);
        return {
            unsubscribe: () => {
                this.subscribers.delete(onChange);
            },
        };
    }

    status(): LibraryStatus {
        let files = 0;
        const collections = this.store.listCollections();
        for (const c of collections) {
            files += this.store.listFiles(c.id).length;
        }
        const ml = this.mlStore?.stats();
        const originals = this.cache?.originalsStatus();
        const pre = this.precache?.status();
        return {
            userID: this.store.userID,
            collections: collections.length,
            files,
            lastRefreshAt: this.lastRefreshAt,
            lastError: this.lastError,
            lastMLFetchAt: this.lastMLFetchAt,
            lastMLError: this.lastMLError,
            mlStored: ml?.stored,
            mlIndexed: ml?.indexed,
            originalsUsedBytes: originals?.usedBytes,
            originalsLimitBytes: originals?.limitBytes,
            thumbnailsCached: pre?.thumbnailsCached,
            thumbnailsTotal: pre?.thumbnailsTotal,
            originalsCached: pre?.originalsCached,
            originalsPinned: pre?.originalsPinned,
            closed: this.closed,
        };
    }

    // Fresh reads (issue #75, owner amendment to design #36). Force a refresh,
    // wait for it to complete and persist, then hand back the same
    // `albums`/`photos`/`timeline` namespaces — now guaranteed to reflect a
    // completed server round-trip. Concurrent calls coalesce onto one refresh;
    // a refresh that fails rejects here, where the default namespaces would
    // instead stay silent and serve the last good copy.
    async fresh(): Promise<FreshReads> {
        await this.refreshNow();
        return {
            albums: this.albums,
            photos: this.photos,
            timeline: this.timeline,
        };
    }

    // Back up every in-scope file to `downloadDirectory` in the historical
    // on-disk layout, with a durable failure ledger (issue #51). Refreshes
    // first, fetches pending originals (and optional thumbnails) through the
    // content cache and pools, then rebuilds the derived symlink/JSON views
    // from the model. Throws before any network work when no download directory
    // is available or no content cache backs the originals it must fetch.
    backup(opts?: BackupOptions): Promise<BackupResult> {
        const downloadDirectory =
            opts?.downloadDirectory ?? this.downloadDirectory;
        const includeOriginals = opts?.includeOriginals ?? true;
        const includeThumbnails = opts?.includeThumbnails ?? false;
        if (!downloadDirectory) {
            return Promise.reject(
                new Error(
                    "backup requires a downloadDirectory (pass one to " +
                        "backup() or open the library with one)",
                ),
            );
        }
        if ((includeOriginals || includeThumbnails) && !this.cache) {
            return Promise.reject(
                new Error(
                    "backup requires a library opened with a content cache",
                ),
            );
        }
        const cache = this.cache;
        return runBackup(
            {
                refresh: () => this.runRefresh(),
                listCollections: () => this.store.listCollections(),
                listFiles: (id) => this.store.listFiles(id),
                original: (fileID) => cache!.original(fileID),
                thumbnail: (fileID) => cache!.thumbnail(fileID),
            },
            { ...opts, downloadDirectory },
        );
    }

    // Stop the background timer. Idempotent. An in-flight refresh is left to
    // finish; it will not schedule another cycle once closed. The returned
    // promise resolves once that refresh (including its cache write), the ML
    // fetch pass and the precache fetches already running have all finished,
    // so a caller can then remove the cache directory. A refresh failure is
    // reported through `status()`, not thrown here.
    async close(): Promise<void> {
        this.closed = true;
        const precacheClosed = this.precache?.close();
        if (this.timer !== undefined) {
            clearTimeout(this.timer);
            this.timer = undefined;
        }
        await this.cycle?.catch(() => {});
        await this.mlFetch;
        await precacheClosed;
    }

    private scheduleNext(): void {
        if (this.closed) return;
        this.timer = setTimeout(() => {
            void this.runRefresh().then(() => this.scheduleNext());
        }, this.intervalMs);
        // Do not keep the process alive for the sake of the timer.
        this.timer.unref?.();
    }

    // The background loop's refresh: run a cycle unless one is already in flight
    // (or the library is closed), and never let a failure escape — the
    // background path reports errors through `status()`/`onProgress`, it does
    // not throw. Resolves once the cycle it started (or skipped past) settles.
    private runRefresh(): Promise<void> {
        if (this.closed || this.cycle) return Promise.resolve();
        return this.startCycle().catch(() => {});
    }

    // A fresh read's refresh (issue #75): force a cycle and await it, rejecting
    // if it fails. Concurrent fresh reads coalesce onto the one in-flight cycle
    // — the background loop's included — so they never fan out into redundant
    // server round-trips.
    private refreshNow(): Promise<void> {
        if (this.closed) {
            return Promise.reject(new Error("the library is closed"));
        }
        return this.cycle ?? this.startCycle();
    }

    // Start one refresh cycle and record it as the in-flight cycle so every
    // caller coalesces onto it. The returned promise carries the cycle's real
    // outcome; each caller attaches the handling its own path needs, and the
    // slot is cleared once the cycle settles.
    private startCycle(): Promise<void> {
        const cycle = this.refreshCycle();
        this.cycle = cycle;
        void cycle.then(
            () => {
                if (this.cycle === cycle) this.cycle = undefined;
            },
            () => {
                if (this.cycle === cycle) this.cycle = undefined;
            },
        );
        return cycle;
    }

    // One refresh cycle: the network fetch and commit, wrapped in the progress
    // events and status bookkeeping. Throws when the refresh fails so a fresh
    // read can reject; `runRefresh` swallows that throw for the background loop.
    private async refreshCycle(): Promise<void> {
        this.emit({ operation: "refresh", status: "started" });
        try {
            await this.refreshOnce();
            this.lastRefreshAt = Date.now();
            this.lastError = undefined;
            this.emit({ operation: "refresh", status: "done" });
            // Backfill ML data for the files this refresh knows about. It runs
            // outside the refresh's success/failure so a fetch or disk problem
            // there never marks the metadata refresh failed, and it is not
            // awaited so it never stalls the refresh interval.
            this.mlFetch ??= this.runMLFetch().finally(() => {
                this.mlFetch = undefined;
            });
        } catch (err) {
            const error = err instanceof Error ? err.message : String(err);
            this.lastError = error;
            this.emit({ operation: "refresh", status: "failed", error });
            throw err;
        }
    }

    // Fetch every change since the stored cursor, then commit. All network
    // reads happen before any store mutation, so a fetch that throws leaves the
    // store untouched and the previous snapshot intact.
    private async refreshOnce(): Promise<void> {
        const page = await this.client.collectionsSince({
            sinceTime: this.store.collectionsSinceTime,
        });

        // Stage per-collection file diffs. A collection's files are
        // re-enumerated only when its updationTime has advanced past the cached
        // copy; an unchanged album's file list cannot have changed. New
        // collections enumerate from the beginning of time.
        const filePages: { collectionID: number; page: FilesPage }[] = [];
        for (const collection of page.collections) {
            const known = this.store.getCollection(collection.id);
            if (known && collection.updationTime <= known.updationTime)
                continue;
            const filePage = await this.client.filesSince({
                collectionID: collection.id,
                collectionKey: collection.key,
                sinceTime: known ? known.updationTime : 0,
            });
            filePages.push({ collectionID: collection.id, page: filePage });
        }

        // Network work done; commit to the store and persist only if something
        // actually changed.
        let changed = false;

        if (this.store.userID !== this.userID) {
            this.store.userID = this.userID;
            changed = true;
        }

        for (const id of page.deleted) {
            if (this.store.getCollection(id)) {
                this.store.deleteCollection(id);
                changed = true;
            }
        }
        for (const collection of page.collections) {
            this.store.putCollection(collection);
            changed = true;
        }

        for (const { collectionID, page: filePage } of filePages) {
            for (const id of filePage.deleted) {
                if (this.store.getFile(collectionID, id)) {
                    this.store.deleteFile(collectionID, id);
                    changed = true;
                }
            }
            for (const f of filePage.files) {
                this.store.putFile(f);
                changed = true;
            }
        }

        if (page.cursor !== this.store.collectionsSinceTime) {
            this.store.collectionsSinceTime = page.cursor;
            changed = true;
        }

        if (changed) this.unsaved = true;

        // Reproject and notify subscribers of the delta. This tracks RAM (what
        // reads see), so it fires whether or not the save below succeeds; a
        // save failure surfaces separately through `status().lastError`.
        // `lastRecords` advances every changed refresh so the next diff is
        // against current state.
        if (changed) {
            const next = this.deriveNow();
            if (this.subscribers.size > 0) {
                const change = diffRecords(this.lastRecords, next, Date.now());
                if (change) this.notify(change);
            }
            this.lastRecords = next;
            // Recompute the fill orders and pinned set against the new library.
            this.precache?.update(next);
        }
        // Re-kick the fills every cycle: a finished sweep starts afresh to pick
        // up new files and retry any that failed, and a running one is left be.
        this.precache?.start();

        // Persist whenever RAM holds changes disk has not accepted — including
        // changes an earlier cycle staged whose save failed. `unsaved` clears
        // only once a save lands, so a save failure both stays visible through
        // `status().lastError` (the throw below records it) and keeps being
        // retried, instead of a later empty refresh silently clearing it while
        // the on-disk cache is still behind RAM.
        if (this.unsaved) {
            await this.store.save();
            this.unsaved = false;
        }
    }

    // One ML fetch pass: fetch, decrypt and store the ML data for every file
    // the store knows about that is not cached (or whose `updationTime` has
    // advanced), through the metadata pool, and update the CLIP index. Guarded
    // so passes never overlap; a failure is reported, not thrown.
    private async runMLFetch(): Promise<void> {
        const mldata = this.mlStore;
        // Bind so the call keeps the client as its receiver when invoked
        // through the pool below.
        const fetchMLData = this.client.fetchMLData?.bind(this.client);
        if (!mldata || !fetchMLData || this.closed) return;

        const files = this.uniqueFiles();
        const needed = mldata.neededFor(files);
        if (needed.length === 0) return;

        this.emit({ operation: "fetchMLData", status: "started" });
        try {
            const fileKeys = new Map<number, Uint8Array>();
            const updation = new Map<number, number>();
            for (const f of files) {
                fileKeys.set(f.id, f.key);
                updation.set(f.id, f.updationTime);
            }

            let stored = 0;
            for (let i = 0; i < needed.length; i += MLDATA_BATCH_SIZE) {
                if (this.closed) break;
                const batch = needed.slice(i, i + MLDATA_BATCH_SIZE);
                const payloads = await this.pools.metadata.run(
                    () => fetchMLData({ fileIDs: batch, fileKeys }),
                    { priority: "background" },
                );
                stored += (await mldata.storeFetched(payloads, updation))
                    .stored;
            }

            this.lastMLFetchAt = Date.now();
            this.lastMLError = undefined;
            this.emit({
                operation: "fetchMLData",
                status: "done",
                fetched: stored,
            });
        } catch (err) {
            const error = err instanceof Error ? err.message : String(err);
            this.lastMLError = error;
            this.emit({ operation: "fetchMLData", status: "failed", error });
        }
    }

    // The distinct files the store holds, one entry per fileID (a file in
    // several collections shares its ML data), each carrying the key and the
    // newest `updationTime` seen across its memberships.
    private uniqueFiles(): {
        id: number;
        key: Uint8Array;
        updationTime: number;
    }[] {
        const byID = new Map<
            number,
            { id: number; key: Uint8Array; updationTime: number }
        >();
        for (const collection of this.store.listCollections()) {
            for (const f of this.store.listFiles(collection.id)) {
                const seen = byID.get(f.id);
                if (seen === undefined || f.updationTime > seen.updationTime)
                    byID.set(f.id, {
                        id: f.id,
                        key: f.key,
                        updationTime: f.updationTime,
                    });
            }
        }
        return [...byID.values()];
    }

    // Gather every file membership and project the store into by-id records,
    // filling each record's cache paths from the content cache when present.
    private deriveNow(): DerivedRecords {
        return deriveRecordsFromStore(this.store, this.cache);
    }

    private notify(change: LibraryChange): void {
        for (const onChange of this.subscribers) {
            // A misbehaving subscriber must not break the loop or its peers.
            try {
                onChange(change);
            } catch {
                // ignore
            }
        }
    }

    private emit(event: RefreshEvent): void {
        if (!this.onProgress) return;
        // A misbehaving callback must not break the refresh loop.
        try {
            this.onProgress(event);
        } catch {
            // ignore
        }
    }
}