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 read is answered from RAM — no read touches
// the network. There is deliberately no `sync()`, no `refresh()`, no
// `serverReachable` flag, and no "before each read" mode (design #36): the
// only ways state changes are the refreshes above.
//
// 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 { 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,
} from "./read.js";
import { makeMLDataAPI, type MLDataAPI } from "./mlsearch.js";

export {
    Album,
    Photo,
    type AlbumsAPI,
    type PhotosAPI,
    type TimelineAPI,
    type PhotoFilter,
    type TimelineGroup,
    type GroupBy,
} from "./read.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";

export const DEFAULT_REFRESH_INTERVAL_SECONDS = 3;

// 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>>;
}

// 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.
export interface RefreshEvent {
    operation: "refresh" | "fetchMLData";
    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 for later phases (backup, thumbnails); the
    // refresh loop does not use it.
    downloadDirectory?: string;
    refreshIntervalSeconds?: number;
    onProgress?: RefreshProgressCallback;
    // The bounded request pools (issue #45). ML data is fetched through the
    // metadata pool. Defaults to a fresh set at the design's caps.
    pools?: RequestPools;
}

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;
    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 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;
    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;

    private timer?: ReturnType<typeof setTimeout>;
    private refreshing = false;
    // Guards the ML fetch pass so a slow backfill never runs twice at once; a
    // refresh whose pass is still running kicks nothing new.
    private mlFetching = false;
    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;
    }) {
        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.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.photos = makePhotosAPI(derive);
        this.timeline = makeTimelineAPI(derive);
        // 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 ??
            join(envPaths("quak", { suffix: "" }).cache, String(userID));
        const store = await MetadataStore.load(
            join(cacheDirectory, "metadata.json"),
        );
        const intervalMs =
            (opts.refreshIntervalSeconds ?? DEFAULT_REFRESH_INTERVAL_SECONDS) *
            1000;

        // 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;

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

        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);
    }

    // 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();
        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,
            closed: this.closed,
        };
    }

    // Stop the background timer. Idempotent. An in-flight refresh is left to
    // finish; it will not schedule another cycle once closed.
    close(): void {
        this.closed = true;
        if (this.timer !== undefined) {
            clearTimeout(this.timer);
            this.timer = undefined;
        }
    }

    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?.();
    }

    // One refresh cycle, guarded so a failure never escapes and overlapping
    // cycles never run. Errors are reported, not thrown.
    private async runRefresh(): Promise<void> {
        if (this.closed || this.refreshing) return;
        this.refreshing = true;
        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.
            void this.runMLFetch();
        } catch (err) {
            const error = err instanceof Error ? err.message : String(err);
            this.lastError = error;
            this.emit({ operation: "refresh", status: "failed", error });
        } finally {
            this.refreshing = false;
        }
    }

    // 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;
        }

        // 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 || this.mlFetching) return;

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

        this.mlFetching = true;
        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 });
        } finally {
            this.mlFetching = false;
        }
    }

    // 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.
    private deriveNow(): DerivedRecords {
        const collections = this.store.listCollections();
        const files: EnteFile[] = [];
        for (const c of collections) files.push(...this.store.listFiles(c.id));
        return deriveRecords(collections, files);
    }

    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
        }
    }
}