platform events, and thumbnails warmed the moment files arrive

a closed, named list of events — `files:uploaded`, `archive:extracted`,
`files:placed` — each carrying absolute paths and an owner. NOT an event bus in the
open sense: `emit(anyString, anything)` becomes an API nobody can change, because
the moment a plugin listens for a string that string is a contract with no list
saying so. a union means adding one is a deliberate edit and removing one fails to
compile.

TWO GUARANTEES, AND THEY ARE THE WHOLE DESIGN. emit never blocks: it schedules and
returns, so an upload does not wait for a subscriber thumbnailing a HEIC. and a
subscriber cannot fail the operation that triggered it — the upload is written and
answered before anybody is told, so a plugin that is down, slow or throwing gets a
line in the log. tested both ways, sync and async, plus that one failing listener
does not stop the next.

PLUGINS SUBSCRIBE THROUGH THE MANIFEST, same shape as `fileTypes`: `hooks: {
'files:uploaded': '/v1/on-upload' }`. resolved at EMIT time rather than at boot,
because plugins are installed, updated and disabled while the server runs and a
registration made at startup describes a machine that has since changed. nothing
here names a plugin.

CORE IS THE FIRST SUBSCRIBER, and that is deliberate rather than a shortcut. the
thumbnail cache belongs to the platform — its key, its route — so a plugin doing
this work would produce files the browser never looks at. the warmer calls the same
`buildThumbnail` the request path calls, which means HEIC still warms through
av-club via the fileTypes claim, without this file knowing that plugin exists.

AND A SINGLE-FLIGHT MAP, which the warmer made necessary. two callers now want the
same thumbnail as a matter of course: the warmer starts on upload and the browser
asks a moment later. measured before the fix — four concurrent requests plus the
warmer each decoded the same 4032x3024 HEIC and took a slot from a limit of four,
which is a warmer competing with the person waiting rather than helping them. after:
all five share one build, 5.5s once, then 130ms.

verified end to end: upload returns in 56ms, four concurrent requests all 200 in a
single 5.5s build, cached 130ms. 1679 tests, docs:check green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Officer-Session: 656f5c85-e49a-4f1f-b9f3-042a04b4d043
This commit is contained in:
2026-08-29 08:20:35 +01:00
co-authored by Claude Opus 5
parent 39c1122651
commit 000df7fe45
8 changed files with 456 additions and 27 deletions
+15
View File
@@ -5,6 +5,8 @@ import { join } from 'node:path';
import { honoServer, PROTECTED_API_PREFIXES, UNPROTECTED_API_PREFIXES } from './servers/hono';
import { assertPermissionTotality } from './servers/permissions/totality';
import { bootstrapCaldav } from './servers/caldav-bootstrap';
import { registerThumbnailWarmer } from './servers/events/warm-thumbnails';
import { registerPluginHooks } from './servers/events/plugin-hooks';
import { refreshPluginMounts, snapshotPlugins } from './servers/plugins/mount';
import { listPluginRootClaims } from './servers/plugins/public-paths';
import { generatePluginsModule, rebuildFrontend, BUILD_DIR, SHELL_FILE } from './servers/plugins/generate';
@@ -252,6 +254,19 @@ if (secured.repaired.length) {
}
for (const path of secured.failed) console.error(`[boot] could not close ${path} — check its owner and mode`);
/*
Subscribe to platform events before anything can emit one.
Both are pure registration — no I/O, nothing that can fail — so they sit above the plugin mount rather
than inside its try. If either threw, the server not starting would be the correct outcome, and
wrapping them in the plugin block's `catch` would hide that.
The thumbnail warmer is core subscribing to core: the cache belongs to the platform, so the work has to
happen here. Plugin hooks are the general mechanism, resolved per event from whatever is installed.
*/
registerThumbnailWarmer();
registerPluginHooks();
// Mount whatever is installed, BEFORE serve().
//
// Without this a restart silently unmounts every plugin: `honoServer` is built with none at import,
+32
View File
@@ -11,6 +11,7 @@ import { readOcrConfig } from '@@/api/server-settings/ocr';
import { getUserSettings } from 'officerdb';
import { buildThumbnail, clampSize, thumbnailKind } from './thumbnails';
import { pluginFileTypes, sidecarFetch } from './plugin-file-types';
import { emit } from '../../events/bus';
async function getUserTtsVoice(userId: number): Promise<string | null> {
try {
@@ -333,6 +334,7 @@ router.post('/upload', async (ctx) => {
going to be, next to where it belongs. Nothing sweeps them up — a job deleting files it did not
create is a worse trade than a fragment somebody can see.
*/
const written: string[] = [];
for (const file of files) {
if (!(file instanceof File)) continue;
const filePath = resolve(targetDir, file.name);
@@ -340,8 +342,19 @@ router.post('/upload', async (ctx) => {
await mkdir(dirname(filePath), { recursive: true });
await writeAtomic(filePath, file);
written.push(filePath);
}
/*
Announced AFTER the writes and BEFORE the response, which is the only ordering that is correct.
After, because a subscriber reads these from disk and they have to be there. Before returning is
fine because `emit` schedules and returns — the upload does not wait for anybody. What it must never
become is an await: a subscriber thumbnailing a HEIC would add five seconds to an upload that has
already finished.
*/
emit('files:uploaded', { paths: written, email: user.email });
return ctx.json({ ok: true });
});
@@ -1235,6 +1248,25 @@ router.post('/extract', async (ctx) => {
throw errors.BAD_REQUEST(stderr.trim() || 'Archive extraction failed');
}
/*
What actually came out, walked once rather than guessed from the archive listing.
Bounded, because an archive can hold a hundred thousand files and this runs on the request's way
out. The warmer has its own cap too; this one keeps the WALK from being the expensive part.
*/
const unpacked: string[] = [];
const collect = async (dir: string, depth = 0): Promise<void> => {
if (depth > 8 || unpacked.length >= 500) return;
for (const entry of await readdir(dir, { withFileTypes: true }).catch(() => [])) {
if (unpacked.length >= 500) return;
const full = join(dir, entry.name);
if (entry.isDirectory()) await collect(full, depth + 1);
else unpacked.push(full);
}
};
await collect(finalDest);
emit('archive:extracted', { paths: unpacked, email: user.email });
const extractedPath = '/' + finalDest.slice(rootDir.length).replace(/^\/+/, '');
return ctx.json({ extractedPath });
});
+53 -27
View File
@@ -138,6 +138,19 @@ const release = () => {
* file another chance after somebody has installed a codec or replaced the file. Bounded, because a
* long-lived process browsing a lot of broken media should not grow without limit.
*/
/**
* Builds in flight, by cache key — so the same thumbnail is never made twice at once.
*
* Two callers want the same file all the time now: the warmer starts on upload, and the browser asks
* for it a moment later. Without this they both decode the same 4032x3024 HEIC, take a slot each from a
* limit of four, and finish at the same time having done the work twice — which turns a warmer meant to
* help into one competing with the person waiting.
*
* Keyed on the same string the cache is, so it deduplicates exactly what would collide: same file, same
* size, same content. Removed in a `finally`, or one failure would wedge that key until a restart.
*/
const inFlight = new Map<string, Promise<ThumbnailResult>>();
const failed = new Set<string>();
const FAILED_LIMIT = 5_000;
@@ -245,34 +258,39 @@ export async function buildThumbnail(
const cached = await stat(cachePath).catch(() => null);
if (cached?.isFile() && cached.size > 0) return { ok: true, path: cachePath };
await mkdir(cacheDir, { recursive: true });
await acquire();
try {
/*
// Somebody else is already making this one. Wait for theirs rather than starting a second.
const running = inFlight.get(key);
if (running) return running;
const work = (async (): Promise<ThumbnailResult> => {
await mkdir(cacheDir, { recursive: true });
await acquire();
try {
/*
`force_original_aspect_ratio=decrease` fits inside the box without cropping, and never enlarges
past the source — a 40px icon blown up to 256 is worse than a small one. `-2` on the odd dimension
keeps it even, which the JPEG encoder requires for chroma subsampling.
*/
const scale = `scale=${size}:${size}:force_original_aspect_ratio=decrease`;
const common = ['-v', 'error', '-y', '-frames:v', '1', '-vf', scale, '-f', 'mjpeg', cachePath];
const scale = `scale=${size}:${size}:force_original_aspect_ratio=decrease`;
const common = ['-v', 'error', '-y', '-frames:v', '1', '-vf', scale, '-f', 'mjpeg', cachePath];
let ok: boolean;
if (produce) {
const bytes = await produce().catch(() => null);
if (bytes?.length) await Bun.write(cachePath, bytes);
ok = !!bytes?.length;
} else if (kind === 'image') {
ok = await resizeImage(absPath, cachePath, size);
} else {
// `-ss` BEFORE `-i` is the fast seek — it jumps rather than decoding a second of video first.
ok = await spawnFfmpeg(['-ss', SEEK_SECONDS, '-i', absPath, ...common], cachePath);
// A clip shorter than the seek yields no frame, and that is a fact about its length rather than a
// broken file. Take the first frame instead before giving up on it.
if (!ok) ok = await spawnFfmpeg(['-i', absPath, ...common], cachePath);
}
let ok: boolean;
if (produce) {
const bytes = await produce().catch(() => null);
if (bytes?.length) await Bun.write(cachePath, bytes);
ok = !!bytes?.length;
} else if (kind === 'image') {
ok = await resizeImage(absPath, cachePath, size);
} else {
// `-ss` BEFORE `-i` is the fast seek — it jumps rather than decoding a second of video first.
ok = await spawnFfmpeg(['-ss', SEEK_SECONDS, '-i', absPath, ...common], cachePath);
// A clip shorter than the seek yields no frame, and that is a fact about its length rather than a
// broken file. Take the first frame instead before giving up on it.
if (!ok) ok = await spawnFfmpeg(['-i', absPath, ...common], cachePath);
}
if (!ok) {
/*
if (!ok) {
/*
A PRODUCER'S FAILURE IS NOT REMEMBERED, and that distinction is the whole point of this branch.
When sharp or ffmpeg fails, the file is the problem — it will fail identically forever, and
@@ -284,12 +302,20 @@ export async function buildThumbnail(
the SERVER was restarted — no client retry could ever get past it, because the answer was being
served from memory without the plugin being asked again.
*/
if (!produce) rememberFailure(key);
return { ok: false, reason: 'could not read an image from this file' };
}
if (!produce) rememberFailure(key);
return { ok: false, reason: 'could not read an image from this file' };
}
return { ok: true, path: cachePath };
return { ok: true, path: cachePath };
} finally {
release();
}
})();
inFlight.set(key, work);
try {
return await work;
} finally {
release();
inFlight.delete(key);
}
}
+71
View File
@@ -0,0 +1,71 @@
import { expect, test } from 'bun:test';
import { emit, listenerCount, on } from './bus';
const settle = () => new Promise((r) => setTimeout(r, 10));
test('a listener is told, with the payload', async () => {
const seen: string[][] = [];
const off = on('files:uploaded', (p) => void seen.push(p.paths));
emit('files:uploaded', { paths: ['/a', '/b'], email: 'x@y' });
await settle();
expect(seen).toEqual([['/a', '/b']]);
off();
});
test('unsubscribing stops it', async () => {
let calls = 0;
const off = on('files:uploaded', () => void calls++);
off();
emit('files:uploaded', { paths: ['/a'], email: 'x@y' });
await settle();
expect(calls).toBe(0);
});
// The property the whole design rests on: a subscriber must never be able to fail the operation that
// triggered it. An upload is already written and answered by the time this runs.
test('emit does not throw when a listener throws synchronously', async () => {
const off = on('files:uploaded', () => {
throw new Error('boom');
});
expect(() => emit('files:uploaded', { paths: [], email: 'x@y' })).not.toThrow();
await settle();
off();
});
test('emit does not throw when a listener rejects', async () => {
const off = on('files:uploaded', async () => {
throw new Error('boom');
});
expect(() => emit('files:uploaded', { paths: [], email: 'x@y' })).not.toThrow();
await settle();
off();
});
// One bad subscriber must not stop the others — they are independent, not a chain.
test('a failing listener does not stop the next one', async () => {
let reached = false;
const offBad = on('archive:extracted', () => {
throw new Error('boom');
});
const offGood = on('archive:extracted', () => void (reached = true));
emit('archive:extracted', { paths: [], email: 'x@y' });
await settle();
expect(reached).toBe(true);
offBad();
offGood();
});
test('emit returns immediately rather than awaiting listeners', () => {
const off = on('files:placed', async () => {
await new Promise((r) => setTimeout(r, 200));
});
const started = performance.now();
emit('files:placed', { paths: [], email: 'x@y' });
expect(performance.now() - started).toBeLessThan(50);
off();
});
test('an event with no listeners is a no-op', () => {
expect(listenerCount('files:placed')).toBe(0);
expect(() => emit('files:placed', { paths: [], email: 'x@y' })).not.toThrow();
});
+88
View File
@@ -0,0 +1,88 @@
/**
* Platform events, and who is told about them.
*
* ── A NAMED LIST, not an event bus ──
*
* The type below is closed on purpose. An open bus — `emit(anyString, anyPayload)` — becomes an API
* nobody can change: the moment a plugin listens for `file:thing`, that string and the shape beside it
* are a contract, and there is no list anywhere saying so. A closed union means adding an event is a
* deliberate edit here, and removing one fails to compile rather than failing quietly at runtime.
*
* ── Every payload is absolute paths and an owner ──
*
* Not a file's contents, not a request, not a user object. A subscriber that needs more should read it
* from the path, because by the time this fires the work is done and the disk is the truth. Passing
* more would be inventing a second description of something already on the filesystem.
*
* ── Emitting NEVER blocks and NEVER throws ──
*
* This is fired after a response is decided, and the response must not wait for it. An upload that
* takes five extra seconds because a subscriber is thumbnailing a HEIC is an upload somebody reports as
* broken; a failing subscriber must not turn a completed upload into an error the caller sees.
*
* So `emit` returns immediately and swallows everything. That is deliberate rather than lazy: a
* subscriber's failure is ITS failure, logged with its name, and the operation that triggered it has
* already succeeded.
*/
/** Absolute paths that now exist, and whose account they belong to. */
export type FileEventPayload = {
/** Absolute. Files, not directories — a subscriber wanting the folder can take the dirname. */
paths: string[];
/** The account the operation was performed for. */
email: string;
};
export type PlatformEvents = {
/** One or more files arrived through `POST /file-browser/upload`. */
'files:uploaded': FileEventPayload;
/** An archive was unpacked. `paths` is what came out of it. */
'archive:extracted': FileEventPayload;
/** Files were copied or moved to a new location. `paths` is where they are NOW. */
'files:placed': FileEventPayload;
};
export type EventName = keyof PlatformEvents;
type Listener<K extends EventName> = (payload: PlatformEvents[K]) => void | Promise<void>;
const listeners = new Map<EventName, Set<Listener<EventName>>>();
/**
* Subscribe. Returns the unsubscribe, which nothing currently calls — core's own subscribers live for
* the process — but returning it means a future one can be torn down without changing this.
*/
export function on<K extends EventName>(event: K, listener: Listener<K>): () => void {
const set = listeners.get(event) ?? new Set();
set.add(listener as Listener<EventName>);
listeners.set(event, set);
return () => set.delete(listener as Listener<EventName>);
}
/**
* Tell everyone, and get out of the way.
*
* Synchronous return, work scheduled after. `queueMicrotask` rather than awaiting: the caller is a
* request handler about to write a response, and the point is that it does not wait.
*/
export function emit<K extends EventName>(event: K, payload: PlatformEvents[K]): void {
const set = listeners.get(event);
if (!set?.size) return;
for (const listener of set) {
queueMicrotask(() => {
try {
// A listener returning a rejected promise is the common case — most of them are async — and it
// needs catching separately from a synchronous throw.
void Promise.resolve(listener(payload)).catch((err: unknown) => {
console.error(`[events] ${event} listener failed —`, err instanceof Error ? err.message : err);
});
} catch (err) {
console.error(`[events] ${event} listener threw —`, err instanceof Error ? err.message : err);
}
});
}
}
/** For tests: how many listeners an event has. Nothing in the server should need this. */
export const listenerCount = (event: EventName): number => listeners.get(event)?.size ?? 0;
+76
View File
@@ -0,0 +1,76 @@
import { snapshotPlugins } from '../plugins/mount';
import { sidecarPort } from '../sidecar-registry';
import { on, type EventName, type FileEventPayload } from './bus';
/**
* Letting plugins subscribe to platform events.
*
* ── The same shape as `fileTypes`, deliberately ──
*
* A manifest declares `hooks: { 'files:uploaded': '/v1/on-upload' }`. Core resolves that at emit time
* from whatever is installed and enabled, POSTs the payload to the plugin's own sidecar, and knows
* nothing about which plugin it is. Disable the plugin and the hook stops firing, with nothing to
* unregister — the same property that makes a claimed file type disappear cleanly.
*
* ── Resolved at EMIT time, not at boot ──
*
* The obvious design registers every plugin's hooks once at startup. It is wrong here: plugins are
* installed, updated, enabled and disabled while the server runs, and a registration made at boot
* describes a machine that has since changed. Reading the current set costs one database call on an
* event that has just done filesystem work, which is not the expensive part of anything.
*
* ── A hook cannot fail the operation that triggered it ──
*
* `emit` already swallows, and this adds nothing that could throw past it. The upload succeeded before
* anybody was told. A plugin that is down, slow, or returns 500 gets a line in the log and no more —
* because the alternative is an upload that reports failure for work that has nothing to do with it.
*/
/** Long enough for real work, short enough that a wedged plugin does not hold a socket all day. */
const HOOK_TIMEOUT_MS = 120_000;
async function dispatch(event: EventName, payload: FileEventPayload): Promise<void> {
let states: Awaited<ReturnType<typeof snapshotPlugins>>['states'] = [];
try {
({ states } = await snapshotPlugins());
} catch {
return; // a database that cannot be read is a louder problem elsewhere
}
for (const state of states) {
if (!state.install?.enabled) continue;
const path = state.plugin.manifest?.hooks?.[event];
if (!path) continue;
const port = sidecarPort(state.plugin.appName);
// Declared but not running — restarting, or never started. Skipped silently: a plugin being down is
// not an error of the upload that happened to coincide with it.
if (!port) continue;
try {
const res = await fetch(`http://127.0.0.1:${port}${path}`, {
method: 'POST',
headers: { 'content-type': 'application/json', 'x-officer-user': payload.email },
body: JSON.stringify({ event, ...payload }),
signal: AbortSignal.timeout(HOOK_TIMEOUT_MS),
});
if (!res.ok) {
console.warn(`[hooks] ${state.plugin.appName} ${event} -> ${res.status}`);
}
} catch (err) {
console.warn(`[hooks] ${state.plugin.appName} ${event} failed —`, err instanceof Error ? err.message : err);
}
}
}
/**
* Wire plugin dispatch into every event. Called once, from the server's own start-up.
*
* One listener per event rather than one per plugin: which plugins care is resolved when the event
* fires, so there is nothing to add or remove as they come and go.
*/
export function registerPluginHooks(): void {
const events: EventName[] = ['files:uploaded', 'archive:extracted', 'files:placed'];
for (const event of events) on(event, (payload) => dispatch(event, payload));
}
+107
View File
@@ -0,0 +1,107 @@
import { join } from 'node:path';
import { DATA_PATH } from '../data-path';
import { buildThumbnail, thumbnailKind } from '../api/file-browser/thumbnails';
import { pluginFileTypes, sidecarFetch } from '../api/file-browser/plugin-file-types';
import { on, type FileEventPayload } from './bus';
/**
* Make the thumbnails before anybody asks for them.
*
* ── Why this is worth doing at all ──
*
* Thumbnails are generated on demand and cached. That is right — most files are never looked at — but
* it means the first person to open a folder waits for the work. For a jpeg that is 40ms nobody
* notices. For a HEIC it is four and a half seconds each, because ImageMagick decodes the whole
* 4032x3024 frame before sharp sees it. A folder of photographs from a phone is thirteen seconds of
* staring at file icons.
*
* An upload is the one moment the platform KNOWS which files are new, so it is the moment to do it.
*
* ── This is core subscribing to core, and that is the point ──
*
* The thumbnail cache belongs to the platform: keyed on its scheme, read by its route. A plugin doing
* this work would produce them somewhere the file browser never looks. So the warmer lives here and
* uses the same `buildThumbnail` the request path uses — which means a HEIC still warms through
* av-club, via the `fileTypes` claim, without this file knowing that plugin exists.
*
* ── ONE size, and it is the one both views ask for ──
*
* The cache key includes the size. Warming 128 when the browser asks for 256 is work that produces
* nothing anybody reads, so this must stay equal to what `FileItem` requests. Both views were made to
* ask for 256 for exactly this reason.
*/
const WARM_SIZE = 256;
/**
* How many to warm at once.
*
* Lower than the request path's own limit on purpose. This is background work competing with a person
* who is actually waiting: somebody opening a different folder while an upload warms must not queue
* behind it. `buildThumbnail` has its own cap too, so this is a second, tighter gate rather than the
* only one.
*/
const WARM_CONCURRENCY = 2;
/** Stop after this many. An unpacked archive can hold thousands, and warming all of them is not a favour. */
const WARM_LIMIT = 200;
async function warm(payload: FileEventPayload): Promise<void> {
const claimed = await pluginFileTypes();
// Only what will actually be asked for: core's own types plus whatever a plugin claims. Anything else
// has no thumbnail to make and would be a spawn per file for nothing.
const targets = payload.paths
.filter((path) => {
const name = path.slice(path.lastIndexOf('/') + 1);
return !!thumbnailKind(name) || claimed.has(name.slice(name.lastIndexOf('.') + 1).toLowerCase());
})
.slice(0, WARM_LIMIT);
if (targets.length === 0) return;
const userDataDir = join(DATA_PATH, payload.email);
let index = 0;
const worker = async () => {
while (index < targets.length) {
const path = targets[index++]!;
const name = path.slice(path.lastIndexOf('/') + 1);
const kind = thumbnailKind(name);
const claim = kind ? null : claimed.get(name.slice(name.lastIndexOf('.') + 1).toLowerCase());
/*
Exactly what the request path does, including its cache and its remembered failures — so a file
warmed here is a cache HIT when the browser asks, and a file that cannot be thumbnailed is not
attempted twice.
*/
await buildThumbnail(
path,
userDataDir,
kind ?? 'image',
WARM_SIZE,
claim?.thumbnail
? async () => {
const res = await sidecarFetch(
`${claim.thumbnail}?path=${encodeURIComponent(path)}&size=${WARM_SIZE}`,
payload.email,
);
return res.ok ? Buffer.from(await res.arrayBuffer()) : null;
}
: undefined,
).catch(() => {
// Already logged by the builder where it matters. A warm that fails costs nothing: the request
// path will try again and answer honestly.
});
}
};
await Promise.all(Array.from({ length: Math.min(WARM_CONCURRENCY, targets.length) }, worker));
}
/** Called once from start-up. */
export function registerThumbnailWarmer(): void {
on('files:uploaded', warm);
on('archive:extracted', warm);
on('files:placed', warm);
}
+14
View File
@@ -509,6 +509,20 @@ export type PluginManifest = {
render?: string;
};
/**
* Platform events this plugin wants to be told about, and where to tell it.
*
* `{ 'files:uploaded': '/v1/on-upload' }` — the key is an event from `servers/events/bus.ts`, the
* value a path on this plugin's own sidecar. Core POSTs `{ event, paths, email }` to it after the
* operation has already succeeded.
*
* **A hook cannot fail the thing that triggered it.** The upload is done and answered before anybody
* is told; a plugin that is down, slow or erroring gets a line in the log. Nothing here is a way to
* veto or modify an operation, and that is deliberate — a hook that could refuse an upload would make
* every upload's reliability the worst of every installed plugin's.
*/
hooks?: Partial<Record<string, string>>;
osDependencies?: OsDependency[];
/**