clawbot
fbb8ae44a7
check / check (push) Successful in 30s
Adds a synchronous, key-free snapshot() returning LibrarySnapshot (one PhotoRecord per fileID, deduped, newest first) with edited-name/time precedence (pubMagicMetadata over basic metadata) in milliseconds, plus AlbumRecord (favorites identified by type). subscribe({onChange}) delivers LibraryChange (changed/removed albums and files, refreshedAt) only when a refresh changes something; unsubscribe stops delivery. Built on the existing refresh loop; served from RAM, safe to send over IPC.
Model: opus-4-8
387 lines
15 KiB
TypeScript
387 lines
15 KiB
TypeScript
// 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 {
|
|
deriveRecords,
|
|
snapshotFrom,
|
|
diffRecords,
|
|
type DerivedRecords,
|
|
type LibrarySnapshot,
|
|
type LibraryChange,
|
|
} from "./records.js";
|
|
import type { CollectionsPage, FilesPage } from "../client.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>;
|
|
}
|
|
|
|
// A single refresh cycle's progress. "started" fires before the network work,
|
|
// then exactly one of "done" or "failed"; "failed" carries the error message.
|
|
export interface RefreshEvent {
|
|
operation: "refresh";
|
|
status: "started" | "done" | "failed";
|
|
error?: string;
|
|
}
|
|
|
|
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;
|
|
}
|
|
|
|
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;
|
|
closed: boolean;
|
|
}
|
|
|
|
export class Library {
|
|
readonly cacheDirectory: string;
|
|
readonly downloadDirectory?: string;
|
|
|
|
private readonly client: LibraryClient;
|
|
private readonly store: MetadataStore;
|
|
private readonly userID: number;
|
|
private readonly intervalMs: number;
|
|
private readonly onProgress?: RefreshProgressCallback;
|
|
|
|
private timer?: ReturnType<typeof setTimeout>;
|
|
private refreshing = false;
|
|
private closed = false;
|
|
private lastRefreshAt?: number;
|
|
private lastError?: 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;
|
|
}) {
|
|
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.lastRecords = this.deriveNow();
|
|
}
|
|
|
|
// 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;
|
|
|
|
const lib = new Library({
|
|
client: opts.client,
|
|
store,
|
|
userID,
|
|
cacheDirectory,
|
|
downloadDirectory: opts.downloadDirectory,
|
|
intervalMs,
|
|
onProgress: opts.onProgress,
|
|
});
|
|
|
|
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;
|
|
}
|
|
return {
|
|
userID: this.store.userID,
|
|
collections: collections.length,
|
|
files,
|
|
lastRefreshAt: this.lastRefreshAt,
|
|
lastError: this.lastError,
|
|
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" });
|
|
} 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;
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
}
|
|
}
|