From bcb333d40c90ed6b2f8802693b87d3fbc1d340fb Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sun, 1 Nov 2026 00:04:15 +0100 Subject: [PATCH] feat: share installed plugins and themes between instances #1066 --- .../extensions/[kind]/[id]/rollback/route.ts | 3 + app/api/admin/marketplace/route.ts | 3 + app/api/admin/plugins/[id]/bundle/route.ts | 9 +- app/api/admin/plugins/route.ts | 7 + app/api/admin/themes/route.ts | 7 + lib/admin/__tests__/bundle-cache.test.ts | 33 ++ .../__tests__/plugin-registry-shared.test.ts | 250 +++++++++ lib/admin/bundle-cache.ts | 70 +++ lib/admin/extension-history.ts | 110 +++- lib/admin/plugin-registry.ts | 511 ++++++++++++------ lib/admin/store-unavailable-response.ts | 15 + lib/admin/theme-asset.ts | 3 + 12 files changed, 821 insertions(+), 200 deletions(-) create mode 100644 lib/admin/__tests__/bundle-cache.test.ts create mode 100644 lib/admin/__tests__/plugin-registry-shared.test.ts create mode 100644 lib/admin/bundle-cache.ts create mode 100644 lib/admin/store-unavailable-response.ts diff --git a/app/api/admin/extensions/[kind]/[id]/rollback/route.ts b/app/api/admin/extensions/[kind]/[id]/rollback/route.ts index 651eefcb5..743225925 100644 --- a/app/api/admin/extensions/[kind]/[id]/rollback/route.ts +++ b/app/api/admin/extensions/[kind]/[id]/rollback/route.ts @@ -6,6 +6,7 @@ import { HistoryError, readHistoryEntry } from '@/lib/admin/extension-history'; import { getPlugin, getTheme, savePlugin, saveTheme } from '@/lib/admin/plugin-registry'; import { invalidateFrameOriginsCache } from '@/lib/admin/csp-frame-origins'; import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response'; +import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response'; import { EXTENSION_ID_RE } from '@/lib/extension-package'; /** @@ -87,6 +88,8 @@ export async function POST( } const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Extension rollback error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Rollback failed' }, { status: 500 }); } diff --git a/app/api/admin/marketplace/route.ts b/app/api/admin/marketplace/route.ts index bea46fc9d..95db4a5f3 100644 --- a/app/api/admin/marketplace/route.ts +++ b/app/api/admin/marketplace/route.ts @@ -23,6 +23,7 @@ import { import { configManager } from '@/lib/admin/config-manager'; import { checkBundleSignature } from '@/lib/admin/directory-keys'; import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response'; +import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response'; import { compareVersions } from '@/lib/version-compare'; import { RELEASE_VERSION } from '@/lib/release-version'; import { @@ -386,6 +387,8 @@ export async function POST(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; if (error instanceof DirectoryError) { return NextResponse.json({ error: error.message }, { status: 502 }); } diff --git a/app/api/admin/plugins/[id]/bundle/route.ts b/app/api/admin/plugins/[id]/bundle/route.ts index 06b398ae5..963d2eecf 100644 --- a/app/api/admin/plugins/[id]/bundle/route.ts +++ b/app/api/admin/plugins/[id]/bundle/route.ts @@ -3,6 +3,7 @@ import { getPluginBundle, getPlugin, getPluginSignature } from '@/lib/admin/plug import { getDevPlugin, readDevBundle } from '@/lib/admin/plugin-dev'; import { signBytes } from '@/lib/admin/plugin-signing'; import { canSeeInstanceDetails } from '@/lib/auth/instance-details'; +import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response'; async function safeSign(code: string): Promise { try { return await signBytes(code); } catch { return null; } @@ -56,7 +57,7 @@ export async function GET( return NextResponse.json({ error: 'Plugin is disabled' }, { status: 403 }); } - const code = await getPluginBundle(id); + const code = await getPluginBundle(id, plugin.bundleHash); if (!code) { return NextResponse.json({ error: 'Bundle not found' }, { status: 404 }); } @@ -71,7 +72,7 @@ export async function GET( }; if (etag) headers['ETag'] = etag; - const signature = await getPluginSignature(id, code); + const signature = await getPluginSignature(id, code, plugin.bundleHash); if (signature) headers['X-Bundle-Signature'] = signature; if (etag && request.headers.get('if-none-match') === etag) { @@ -80,7 +81,9 @@ export async function GET( headers['Content-Length'] = String(Buffer.byteLength(code, 'utf-8')); return new NextResponse(code, { headers }); - } catch { + } catch (error) { + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; return NextResponse.json({ error: 'Not found' }, { status: 404 }); } } diff --git a/app/api/admin/plugins/route.ts b/app/api/admin/plugins/route.ts index aeaf5c7bf..4e5958d90 100644 --- a/app/api/admin/plugins/route.ts +++ b/app/api/admin/plugins/route.ts @@ -16,6 +16,7 @@ import { keptVersions, listHistory } from '@/lib/admin/extension-history'; import { ExtensionPackageError, readPluginPackage, suspiciousCode } from '@/lib/extension-package'; import { MAX_PLUGIN_SIZE } from '@/lib/plugin-types'; import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response'; +import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response'; /** * GET /api/admin/plugins - List all admin-managed plugins @@ -133,6 +134,8 @@ export async function POST(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Plugin install error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } @@ -186,6 +189,8 @@ export async function PATCH(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Plugin update error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } @@ -219,6 +224,8 @@ export async function DELETE(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Plugin delete error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } diff --git a/app/api/admin/themes/route.ts b/app/api/admin/themes/route.ts index 8a2eeca7d..256f68497 100644 --- a/app/api/admin/themes/route.ts +++ b/app/api/admin/themes/route.ts @@ -15,6 +15,7 @@ import { keptVersions, listHistory } from '@/lib/admin/extension-history'; import { readThemePackage } from '@/lib/extension-package'; import { configReadOnlyResponse } from '@/lib/admin/config-read-only-response'; +import { storeUnavailableResponse } from '@/lib/admin/store-unavailable-response'; /** * GET /api/admin/themes - List all admin-managed themes @@ -85,6 +86,8 @@ export async function POST(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Theme install error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } @@ -124,6 +127,8 @@ export async function PATCH(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Theme update error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } @@ -156,6 +161,8 @@ export async function DELETE(request: NextRequest) { } catch (error) { const readOnly = configReadOnlyResponse(error); if (readOnly) return readOnly; + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Theme delete error', { error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } diff --git a/lib/admin/__tests__/bundle-cache.test.ts b/lib/admin/__tests__/bundle-cache.test.ts new file mode 100644 index 000000000..3da91de01 --- /dev/null +++ b/lib/admin/__tests__/bundle-cache.test.ts @@ -0,0 +1,33 @@ +// @vitest-environment node +import { beforeEach, describe, expect, it } from 'vitest'; +import { bundleCacheSize, cacheBundle, clearBundleCache, getCachedBundle, MAX_BYTES } from '../bundle-cache'; + +// The per-replica bundle cache (#1066) must stay under its 64 MB cap and +// drop the bundle used longest ago first. + +beforeEach(() => clearBundleCache()); + +const big = (c: string) => c.repeat(20 * 1024 * 1024); + +describe('bundle cache', () => { + it('drops the least recently used bundle once it holds more than the cap', () => { + cacheBundle('a', big('a')); + cacheBundle('b', big('b')); + cacheBundle('c', big('c')); + expect(getCachedBundle('a')).toBeTruthy(); // a is now the most recently used + cacheBundle('d', big('d')); + + expect(getCachedBundle('b')).toBeUndefined(); + expect(getCachedBundle('a')?.code[0]).toBe('a'); + expect(getCachedBundle('c')).toBeTruthy(); + expect(getCachedBundle('d')).toBeTruthy(); + expect(bundleCacheSize().bytes).toBeLessThanOrEqual(MAX_BYTES); + }); + + it('keeps a signature learned later with the bundle', () => { + cacheBundle('h', 'code()'); + getCachedBundle('h')!.signature = 'sig'; + cacheBundle('h', 'code()'); + expect(getCachedBundle('h')).toMatchObject({ code: 'code()', signature: 'sig' }); + }); +}); diff --git a/lib/admin/__tests__/plugin-registry-shared.test.ts b/lib/admin/__tests__/plugin-registry-shared.test.ts new file mode 100644 index 000000000..e81f213ca --- /dev/null +++ b/lib/admin/__tests__/plugin-registry-shared.test.ts @@ -0,0 +1,250 @@ +// @vitest-environment node +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createHash } from 'node:crypto'; +import { existsSync } from 'node:fs'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { NextRequest } from 'next/server'; +import JSZip from 'jszip'; + +// Several replicas share one store (#1066): a plugin or theme installed +// through one replica must be listed and served by every other one, two +// installs at once on different replicas must both stay in the registry, +// and an install while the store is down must answer 503 and leave nothing +// half-written behind. + +vi.mock('@/lib/admin/session', () => ({ + requireAdminAuth: vi.fn(async () => ({ session: { sub: 'admin' } })), + getClientIP: () => '127.0.0.1', +})); +vi.mock('@/lib/admin/audit', () => ({ auditLog: vi.fn(async () => {}) })); +vi.mock('@/lib/admin/config-manager', () => ({ + configManager: { + ensureLoaded: vi.fn(async () => {}), + get: vi.fn(() => undefined), + getPolicy: vi.fn(() => ({ forceEnabledPlugins: [] })), + updatePolicy: vi.fn(async () => {}), + }, +})); +vi.mock('@/lib/admin/plugin-dev', () => ({ + listDevPlugins: vi.fn(async () => []), + getDevPlugin: vi.fn(async () => null), + readDevBundle: vi.fn(), +})); +vi.mock('@/lib/auth/instance-details', () => ({ canSeeInstanceDetails: vi.fn(async () => true) })); + +import { setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import { INIT_EPOCH, resetChangesForTests } from '@/lib/shared-store/changes'; +import { separateFakeStore } from './shared-store-fake'; + +const MANIFEST = { + id: 'demo-plugin', name: 'Demo', version: '1.0.0', author: 'me', description: 'd', + type: 'hook', permissions: ['email:read'], entrypoint: 'index.js', +}; +const PLUGIN = { + id: 'pp', name: 'P', version: '1.0.0', author: 'a', description: '', type: 'hook', + permissions: [], entrypoint: 'index.js', enabled: true, installedAt: 'x', updatedAt: 'x', +}; + +let dir: string; +let store: ReturnType; + +/** + * Another replica: fresh module copies and nothing of this process's + * memory (registry copy, bundle cache, write queue, signing key). Only the + * shared store is common. + */ +async function otherReplica() { + const g = globalThis as Record; + for (const name of ['bulwark.admin.pluginRegistry', 'bulwark.admin.bundleCache', 'bulwark.admin.pluginSigning']) { + delete g[Symbol.for(name)]; + } + vi.resetModules(); + return { + registry: await import('../plugin-registry'), + plugins: await import('@/app/api/admin/plugins/route'), + themes: await import('@/app/api/admin/themes/route'), + bundle: await import('@/app/api/admin/plugins/[id]/bundle/route'), + themeCss: await import('@/app/api/admin/themes/[id]/css/route'), + }; +} + +async function zipOf(files: Record) { + const zip = new JSZip(); + for (const [n, d] of Object.entries(files)) zip.file(n, d); + return zip.generateAsync({ type: 'uint8array', compression: 'DEFLATE' }); +} + +async function install(replica: Awaited>, code: string) { + const form = new FormData(); + const bytes = await zipOf({ 'manifest.json': JSON.stringify(MANIFEST), 'index.js': code }); + form.append('file', new File([bytes as BlobPart], 'p.zip', { type: 'application/zip' })); + return replica.plugins.POST(new NextRequest('http://localhost/api/admin/plugins', { method: 'POST', body: form })); +} + +function getBundle(replica: Awaited>, id = MANIFEST.id) { + return replica.bundle.GET(new NextRequest(`http://localhost/api/admin/plugins/${id}/bundle`), { params: Promise.resolve({ id }) }); +} + +const localPluginsDir = () => path.join(dir, 'config', 'plugins'); + +beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), 'bw-registry-shared-')); + vi.stubEnv('ADMIN_CONFIG_DIR', path.join(dir, 'config')); + vi.stubEnv('ADMIN_STATE_DIR', path.join(dir, 'state')); + vi.stubEnv('PLUGIN_SIGNING_KEY_FILE', ''); + resetLocalStoreForTests(); + resetChangesForTests(); + store = separateFakeStore(path.join(dir, 'store')); + setStoreForTests(store); + await store.epochs.bump(INIT_EPOCH); +}); + +afterEach(async () => { + setStoreForTests(null); + resetLocalStoreForTests(); + resetChangesForTests(); + vi.unstubAllEnvs(); + await rm(dir, { recursive: true, force: true }); +}); + +describe('plugins in the shared store', () => { + it('lists and serves on another replica a plugin installed through one', async () => { + const a = await otherReplica(); + const code = 'module.exports = { hello: 1 }'; + expect((await install(a, code)).status).toBe(200); + expect((await store.epochs.all()).plugins).toBeGreaterThan(0); + + const b = await otherReplica(); + const list = await (await b.plugins.GET(new NextRequest('http://localhost/api/admin/plugins'))).json(); + expect(list.map((p: { id: string }) => p.id)).toEqual([MANIFEST.id]); + + const res = await getBundle(b); + expect(res.status).toBe(200); + expect(await res.text()).toBe(code); + expect(res.headers.get('etag')).toBe(`"${createHash('sha256').update(code).digest('hex')}"`); + expect(res.headers.get('x-bundle-signature')).toBeTruthy(); + + // Nothing went to this replica's own config dir. + expect(existsSync(localPluginsDir())).toBe(false); + }); + + it('keeps serving a bundle it already served while the store is down', async () => { + const a = await otherReplica(); + await install(a, 'module.exports = { cached: true }'); + const b = await otherReplica(); + expect((await getBundle(b)).status).toBe(200); + + store.down = true; + const res = await getBundle(b); + expect(res.status).toBe(200); + expect(await res.text()).toBe('module.exports = { cached: true }'); + + // A replica that never read it can't: 503, not "not found". + const c = await otherReplica(); + expect((await getBundle(c)).status).toBe(503); + }); + + it('serves the new bundle on another replica after an update', async () => { + const a = await otherReplica(); + const b = await otherReplica(); + await install(a, 'v1()'); + expect(await (await getBundle(b)).text()).toBe('v1()'); + await install(a, 'v2()'); + expect(await (await getBundle(b)).text()).toBe('v2()'); + expect(existsSync(localPluginsDir())).toBe(false); + }); + + it('keeps both plugins when two replicas install different ones at once', async () => { + const replicas = await Promise.all([otherReplica(), otherReplica()]); + // Each starts its install with nothing shared in memory. + const runs = replicas.map((r, i) => { + delete (globalThis as Record)[Symbol.for('bulwark.admin.pluginRegistry')]; + return r.registry.savePlugin({ ...PLUGIN, id: `plugin-${i}a` }, `code${i}()`); + }); + await Promise.all(runs); + + const c = await otherReplica(); + expect((await c.registry.getPluginRegistry()).plugins.map((p) => p.id).sort()).toEqual(['plugin-0a', 'plugin-1a']); + const docs = (await import('../extension-store')).pluginDocs(); + expect(JSON.parse((await docs.get('registry.json'))!.data.toString()).plugins).toHaveLength(2); + }); + + it('answers 503 for an install while the store is down and writes nothing', async () => { + const a = await otherReplica(); + store.down = true; + const res = await install(a, 'module.exports = {}'); + expect(res.status).toBe(503); + + store.down = false; + const docs = (await import('../extension-store')).pluginDocs(); + expect(await docs.list('')).toEqual([]); + expect(existsSync(localPluginsDir())).toBe(false); + }); + + it('answers 503 and writes nothing when the store goes down after the registry was read', async () => { + const a = await otherReplica(); + await a.registry.getPluginRegistry(); + store.down = true; + expect((await install(a, 'module.exports = {}')).status).toBe(503); + + store.down = false; + const docs = (await import('../extension-store')).pluginDocs(); + expect(await docs.list('')).toEqual([]); + }); + + it('fixes an old short bundle hash without losing an install that lands at the same time', async () => { + const docs = (await import('../extension-store')).pluginDocs(); + await docs.put('old-one.js', 'old()'); + await docs.put('registry.json', JSON.stringify({ plugins: [{ ...PLUGIN, id: 'old-one', bundleHash: 'abcdef0123456789' }] })); + + const [a, b] = await Promise.all([otherReplica(), otherReplica()]); + await Promise.all([a.registry.getPluginRegistry(), b.registry.savePlugin({ ...PLUGIN, id: 'new-one' }, 'new()')]); + + const stored = JSON.parse((await docs.get('registry.json'))!.data.toString()); + expect(stored.plugins.map((p: { id: string }) => p.id).sort()).toEqual(['new-one', 'old-one']); + expect(stored.plugins.find((p: { id: string }) => p.id === 'old-one').bundleHash) + .toBe(createHash('sha256').update('old()').digest('hex')); + }); + + it('keeps the history in the store and rolls back on another replica', async () => { + const a = await otherReplica(); + await a.registry.savePlugin(PLUGIN, 'v1()'); + await a.registry.savePlugin({ ...PLUGIN, version: '1.1.0' }, 'v2()'); + + const b = await otherReplica(); + const history = await import('../extension-history'); + expect((await history.listHistory('plugin', 'pp')).map((h) => h.version)).toEqual(['1.0.0']); + const kept = await history.readHistoryEntry('plugin', 'pp', '1.0.0'); + expect(kept.code).toBe('v1()'); + await b.registry.deletePlugin('pp'); + expect(await history.listHistory('plugin', 'pp')).toEqual([]); + expect(existsSync(localPluginsDir())).toBe(false); + }); +}); + +describe('themes in the shared store', () => { + it('serves on another replica a theme installed through one', async () => { + const a = await otherReplica(); + await a.registry.saveTheme( + { id: 'dark-blue', name: 'Dark blue', version: '1.0.0', author: 'a', description: '', variants: ['dark'], enabled: true, installedAt: 'x', updatedAt: 'x' }, + ':root { --x: 1 }', + '.skin {}', + ); + expect((await store.epochs.all()).themes).toBeGreaterThan(0); + + const b = await otherReplica(); + const res = await b.themeCss.GET(new NextRequest('http://localhost/api/admin/themes/dark-blue/css'), { params: Promise.resolve({ id: 'dark-blue' }) }); + expect(res.status).toBe(200); + expect(await res.text()).toBe(':root { --x: 1 }'); + expect(await b.registry.getThemeSkin('dark-blue')).toBe('.skin {}'); + expect(existsSync(path.join(dir, 'config', 'themes'))).toBe(false); + + store.down = true; + const c = await otherReplica(); + const down = await c.themeCss.GET(new NextRequest('http://localhost/api/admin/themes/dark-blue/css'), { params: Promise.resolve({ id: 'dark-blue' }) }); + expect(down.status).toBe(503); + }); +}); diff --git a/lib/admin/bundle-cache.ts b/lib/admin/bundle-cache.ts new file mode 100644 index 000000000..f0e2e243b --- /dev/null +++ b/lib/admin/bundle-cache.ts @@ -0,0 +1,70 @@ +/** + * Plugin bundles this replica served lately, keyed by their SHA-256 + * (`bundleHash`), so serving a bundle in shared mode (#1066) doesn't fetch + * up to 5 MB from the shared store on every request. Content-addressed: a + * new install has a new hash, so nothing here can go stale; entries are only + * added after the bytes were checked against the hash. Least recently used + * entries go first once the cache holds more than MAX_BYTES. + * + * Per process, on globalThis (one cache for every module copy). + */ + +export const MAX_BYTES = 64 * 1024 * 1024; + +export interface CachedBundle { + code: string; + /** The stored signature; undefined until it was read once. */ + signature?: string | null; +} + +interface BundleCacheState { + entries: Map; + bytes: number; +} + +const KEY = Symbol.for('bulwark.admin.bundleCache'); + +function state(): BundleCacheState { + const g = globalThis as typeof globalThis & { [KEY]?: BundleCacheState }; + return (g[KEY] ??= { entries: new Map(), bytes: 0 }); +} + +export function getCachedBundle(hash: string): CachedBundle | undefined { + const s = state(); + const entry = s.entries.get(hash); + if (!entry) return undefined; + // Most recently used last. + s.entries.delete(hash); + s.entries.set(hash, entry); + return entry; +} + +export function cacheBundle(hash: string, code: string, signature?: string | null): void { + const s = state(); + const previous = s.entries.get(hash); + if (previous) { + s.entries.delete(hash); + s.bytes -= previous.bytes; + } + const bytes = Buffer.byteLength(code, 'utf-8'); + if (bytes > MAX_BYTES) return; + s.entries.set(hash, { code, signature: signature ?? previous?.signature, bytes }); + s.bytes += bytes; + for (const [oldest, entry] of s.entries) { + if (s.bytes <= MAX_BYTES) break; + s.entries.delete(oldest); + s.bytes -= entry.bytes; + } +} + +/** How much the cache holds (tests). */ +export function bundleCacheSize(): { entries: number; bytes: number } { + const s = state(); + return { entries: s.entries.size, bytes: s.bytes }; +} + +export function clearBundleCache(): void { + const s = state(); + s.entries.clear(); + s.bytes = 0; +} diff --git a/lib/admin/extension-history.ts b/lib/admin/extension-history.ts index 080ed1534..cc9fcd24e 100644 --- a/lib/admin/extension-history.ts +++ b/lib/admin/extension-history.ts @@ -22,13 +22,19 @@ * current enabled/forced flags are kept. The snapshot's hash is checked * before it is restored: a history file edited on disk is refused rather * than installed for every user. + * + * The files are documents of the `plugins` and `themes` collections + * (extension-store.ts) under `history///`: the paths above in + * local mode, the shared store in shared mode (#1066). */ -import { mkdir, readFile, readdir, rm, writeFile } from 'node:fs/promises'; +import { readdir, rm } from 'node:fs/promises'; import { createHash } from 'node:crypto'; import path from 'node:path'; import { logger } from '@/lib/logger'; -import { getConfigDir } from './paths'; +import { isSharedMode, type DocCollection } from '@/lib/shared-store'; +import { isStoreUnavailable } from './admin-store'; +import { pluginDocs, pluginsOptions, themeDocs, themesOptions } from './extension-store'; import type { ServerPlugin, ServerTheme } from './plugin-registry'; export type ExtensionKind = 'plugin' | 'theme'; @@ -49,9 +55,39 @@ export interface HistoryEntry { const SAFE_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/; -function historyRoot(kind: ExtensionKind, id: string): string { +/** The extension's history folder: a document id prefix, without the trailing slash. */ +function historyRoot(id: string): string { if (!SAFE_ID_RE.test(id)) throw new Error('Invalid extension id'); - return path.join(getConfigDir(), kind === 'plugin' ? 'plugins' : 'themes', 'history', id); + return `history/${id}`; +} + +const docsOf = (kind: ExtensionKind): DocCollection => (kind === 'plugin' ? pluginDocs() : themeDocs()); +const localDir = (kind: ExtensionKind): string => (kind === 'plugin' ? pluginsOptions : themesOptions).local.dir(); + +async function readText(kind: ExtensionKind, id: string): Promise { + try { + const doc = await docsOf(kind).get(id); + return doc ? doc.data.toString('utf-8') : null; + } catch (error) { + if (isStoreUnavailable(error)) throw error; + return null; + } +} + +/** + * Remove a history folder. Local mode: `rm -r` of the folder, as before. + * Shared mode: entry.json first, so no reader sees a kept version without + * its files, then the rest. + */ +async function removeFolder(kind: ExtensionKind, folder: string): Promise { + if (!isSharedMode()) { + await rm(path.join(localDir(kind), ...folder.split('/')), { recursive: true, force: true }); + return; + } + const docs = docsOf(kind); + const ids = await docs.list(`${folder}/`); + const entry = `${folder}/entry.json`; + for (const id of [...ids.filter((i) => i === entry), ...ids.filter((i) => i !== entry)]) await docs.delete(id); } /** @@ -78,9 +114,11 @@ interface StoredEntry extends HistoryEntry { entry: ServerPlugin | ServerTheme; } -async function readEntry(dir: string): Promise { +async function readEntry(kind: ExtensionKind, dir: string): Promise { + const text = await readText(kind, `${dir}/entry.json`); + if (text === null) return null; try { - const raw = JSON.parse(await readFile(path.join(dir, 'entry.json'), 'utf-8')) as StoredEntry; + const raw = JSON.parse(text) as StoredEntry; if (!raw || typeof raw.version !== 'string' || typeof raw.savedAt !== 'string' || !raw.entry) return null; return raw; } catch { @@ -88,18 +126,36 @@ async function readEntry(dir: string): Promise { } } +/** + * The version folders under an extension's history. Local mode reads the + * one directory, as before (listing the collection would walk every file + * under C/plugins on each install). + */ +async function versionFolders(kind: ExtensionKind, root: string): Promise { + if (!isSharedMode()) { + return readdir(path.join(localDir(kind), ...root.split('/'))); + } + const names = new Set(); + for (const docId of await docsOf(kind).list(`${root}/`)) { + const name = docId.slice(root.length + 1).split('/')[0]; + if (name) names.add(name); + } + return [...names]; +} + async function readAll(kind: ExtensionKind, id: string): Promise> { - const root = historyRoot(kind, id); + const root = historyRoot(id); let names: string[]; try { - names = await readdir(root); - } catch { + names = await versionFolders(kind, root); + } catch (error) { + if (isStoreUnavailable(error)) throw error; return []; } const out: Array = []; for (const name of names) { - const dir = path.join(root, name); - const entry = await readEntry(dir); + const dir = `${root}/${name}`; + const entry = await readEntry(kind, dir); if (entry) out.push({ ...entry, dir }); } // Newest replacement first. @@ -110,7 +166,7 @@ async function readAll(kind: ExtensionKind, id: string): Promise { const all = await readAll(kind, id); for (const stale of all.slice(KEEP)) { - await rm(stale.dir, { recursive: true, force: true }); + await removeFolder(kind, stale.dir); } } @@ -142,8 +198,7 @@ export async function listHistory(kind: ExtensionKind, id: string): Promise { - const dir = path.join(historyRoot('plugin', current.id), dirNameFor(current.version)); - await mkdir(dir, { recursive: true }); + const dir = `${historyRoot(current.id)}/${dirNameFor(current.version)}`; const stored: StoredEntry = { kind: 'plugin', version: current.version, @@ -152,15 +207,15 @@ export async function snapshotPlugin(current: ServerPlugin, code: string): Promi ...(current.source ? { source: current.source } : {}), entry: current, }; - await writeFile(path.join(dir, 'bundle.js'), code, 'utf-8'); - await writeFile(path.join(dir, 'entry.json'), JSON.stringify(stored, null, 2), 'utf-8'); + const docs = docsOf('plugin'); + await docs.put(`${dir}/bundle.js`, code); + await docs.put(`${dir}/entry.json`, JSON.stringify(stored, null, 2)); await prune('plugin', current.id); } /** Same for a theme: its registry entry, CSS and skin. */ export async function snapshotTheme(current: ServerTheme, css: string, skin: string | null): Promise { - const dir = path.join(historyRoot('theme', current.id), dirNameFor(current.version)); - await mkdir(dir, { recursive: true }); + const dir = `${historyRoot(current.id)}/${dirNameFor(current.version)}`; const stored: StoredEntry = { kind: 'theme', version: current.version, @@ -169,10 +224,11 @@ export async function snapshotTheme(current: ServerTheme, css: string, skin: str ...(current.source ? { source: current.source } : {}), entry: current, }; - await writeFile(path.join(dir, 'theme.css'), css, 'utf-8'); - if (skin !== null) await writeFile(path.join(dir, 'skin.css'), skin, 'utf-8'); - else await rm(path.join(dir, 'skin.css'), { force: true }); - await writeFile(path.join(dir, 'entry.json'), JSON.stringify(stored, null, 2), 'utf-8'); + const docs = docsOf('theme'); + await docs.put(`${dir}/theme.css`, css); + if (skin !== null) await docs.put(`${dir}/skin.css`, skin); + else await docs.delete(`${dir}/skin.css`); + await docs.put(`${dir}/entry.json`, JSON.stringify(stored, null, 2)); await prune('theme', current.id); } @@ -186,7 +242,7 @@ export async function snapshotTheme(current: ServerTheme, css: string, skin: str export async function forgetVersion(kind: ExtensionKind, id: string, version: string): Promise { if (!SAFE_ID_RE.test(id)) return; for (const kept of await readAll(kind, id)) { - if (kept.version === version) await rm(kept.dir, { recursive: true, force: true }); + if (kept.version === version) await removeFolder(kind, kept.dir); } } @@ -194,7 +250,7 @@ export async function forgetVersion(kind: ExtensionKind, id: string, version: st export async function purgeHistory(kind: ExtensionKind, id: string): Promise { if (!SAFE_ID_RE.test(id)) return; try { - await rm(historyRoot(kind, id), { recursive: true, force: true }); + await removeFolder(kind, historyRoot(id)); } catch (err) { logger.warn('[extension-history] could not remove history', { kind, id, error: err instanceof Error ? err.message : String(err) }); } @@ -228,13 +284,13 @@ export async function readHistoryEntry(kind: K, id: str const stored = await find(kind, id, version); const damaged = () => new HistoryError(`The kept copy of ${id} ${version} does not match what was saved; it was not restored`, 409); if (kind === 'plugin') { - const code = await readFile(path.join(stored.dir, 'bundle.js'), 'utf-8').catch(() => null); + const code = await readText(kind, `${stored.dir}/bundle.js`); if (code === null || hashOf([code]) !== stored.contentHash) throw damaged(); const restored: RestoredPlugin = { kind: 'plugin', entry: stored.entry as ServerPlugin, code }; return restored as Restored; } - const css = await readFile(path.join(stored.dir, 'theme.css'), 'utf-8').catch(() => null); - const skin = await readFile(path.join(stored.dir, 'skin.css'), 'utf-8').catch(() => null); + const css = await readText(kind, `${stored.dir}/theme.css`); + const skin = await readText(kind, `${stored.dir}/skin.css`); if (css === null || hashOf([css, skin]) !== stored.contentHash) throw damaged(); const restored: RestoredTheme = { kind: 'theme', entry: stored.entry as ServerTheme, css, skin }; return restored as Restored; diff --git a/lib/admin/plugin-registry.ts b/lib/admin/plugin-registry.ts index ece86c968..ea2dd1332 100644 --- a/lib/admin/plugin-registry.ts +++ b/lib/admin/plugin-registry.ts @@ -1,20 +1,14 @@ import type { CredentialSpec } from '@/lib/plugin-secrets/credentials'; -import { readFile, writeFile, mkdir, rename, unlink } from 'node:fs/promises'; -import { existsSync } from 'node:fs'; -import { createHash } from 'node:crypto'; -import path from 'node:path'; +import { createHash, randomBytes } from 'node:crypto'; import { logger } from '@/lib/logger'; -import { getConfigDir, assertWritable } from './paths'; +import { getStore, isSharedMode, replicaId } from '@/lib/shared-store'; +import { announceChange } from '@/lib/shared-store/changes'; +import { assertWritable, isConfigReadOnly } from './paths'; +import { isStoreUnavailable } from './admin-store'; +import { cacheBundle, getCachedBundle } from './bundle-cache'; +import { pluginDocs, themeDocs } from './extension-store'; import { forgetVersion, purgeHistory, snapshotPlugin, snapshotTheme } from './extension-history'; -function getPluginsDir(): string { - return path.join(getConfigDir(), 'plugins'); -} - -function getThemesDir(): string { - return path.join(getConfigDir(), 'themes'); -} - // ─── Types ─────────────────────────────────────────────────── export interface PluginConfigField { @@ -145,48 +139,142 @@ interface ThemeRegistry { themes: ServerTheme[]; } -// ─── Plugin Registry ───────────────────────────────────────── +// ─── Storage ───────────────────────────────────────────────── +// +// Both registries, their bundles, signatures, CSS and history are +// documents of the `plugins` and `themes` collections (extension-store.ts): +// the files under C/plugins and C/themes in local mode, the shared store in +// shared mode (#1066). Every change writes the bundle (and its signature) +// before the registry entry that points at it, changes the registry with +// update() so a concurrent change of another entry stays, and then +// announces area `plugins` or `themes` so every replica drops what it +// derived from the old registry (the CSP frame origins, for one). -async function ensureDir(dir: string): Promise { - if (!existsSync(dir)) { - await mkdir(dir, { recursive: true }); - } +type RegistryKind = 'plugins' | 'themes'; + +interface RegistryGlobals { + /** One change at a time per registry in this process. */ + queues: Map>; + /** Shared mode: the last registry read, served while the store is down. */ + lastGood: Map; } -async function readJsonFile(filePath: string, fallback: T): Promise { +const GLOBAL_KEY = Symbol.for('bulwark.admin.pluginRegistry'); + +function registryGlobals(): RegistryGlobals { + const g = globalThis as typeof globalThis & { [GLOBAL_KEY]?: RegistryGlobals }; + return (g[GLOBAL_KEY] ??= { queues: new Map(), lastGood: new Map() }); +} + +const docsFor = (kind: RegistryKind) => (kind === 'plugins' ? pluginDocs() : themeDocs()); + +const REGISTRY_ID = 'registry.json'; + +function serialize(registry: unknown): string { + return JSON.stringify(registry, null, 2); +} + +/** A registry document, or `fallback` when it is missing or doesn't parse (as a broken file always did). */ +function parseRegistry(kind: RegistryKind, data: Buffer | null, fallback: () => T): T { + if (!data) return fallback(); try { - const raw = await readFile(filePath, 'utf-8'); - return JSON.parse(raw); + return JSON.parse(data.toString('utf-8')) as T; } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') return fallback; - logger.warn(`Failed to read ${filePath}`, { error: error instanceof Error ? error.message : 'Unknown error' }); - return fallback; + logger.warn(`Failed to read the ${kind} registry`, { error: error instanceof Error ? error.message : 'Unknown error' }); + return fallback(); } } -async function writeJsonFile(filePath: string, data: unknown): Promise { - const dir = path.dirname(filePath); - await ensureDir(dir); - // A temp name per write: two writers sharing one ".tmp" overwrote each - // other's file and the second rename failed (a 500 on concurrent installs). - const tmpPath = `${filePath}.${process.pid}.${Date.now().toString(36)}${Math.random().toString(36).slice(2, 8)}.tmp`; - await writeFile(tmpPath, JSON.stringify(data, null, 2), 'utf-8'); - await rename(tmpPath, filePath); +/** + * Read a registry. A store outage serves the last copy this replica read + * (config reads keep the last good copy, plan §6); with none it throws. + */ +async function readRegistry(kind: RegistryKind, fallback: () => T): Promise { + const g = registryGlobals(); + let data: Buffer | null; + try { + data = (await docsFor(kind).get(REGISTRY_ID))?.data ?? null; + } catch (error) { + if (isStoreUnavailable(error)) { + const last = g.lastGood.get(kind); + if (last !== undefined) return JSON.parse(last) as T; + throw error; + } + logger.warn(`Failed to read the ${kind} registry`, { error: error instanceof Error ? error.message : 'Unknown error' }); + return fallback(); + } + const registry = parseRegistry(kind, data, fallback); + if (isSharedMode()) g.lastGood.set(kind, serialize(registry)); + return registry; } -// Changes to one registry file run one after another, so two admin actions -// at once can't read the same old list and drop each other's change. -const registryQueues = new Map>(); -function withRegistryLock(name: 'plugins' | 'themes', task: () => Promise): Promise { - const run = (registryQueues.get(name) ?? Promise.resolve()).then(task, task); - const tail = run.catch(() => {}); - registryQueues.set(name, tail); +/** Change a registry with update(); `change` may run more than once. */ +async function updateRegistry(kind: RegistryKind, fallback: () => T, change: (registry: T) => boolean): Promise { + const result = await docsFor(kind).update(REGISTRY_ID, (current) => { + const registry = parseRegistry(kind, current, fallback); + return change(registry) ? serialize(registry) : undefined; + }); + if (result.data && isSharedMode()) registryGlobals().lastGood.set(kind, result.data.toString('utf-8')); +} + +async function readText(kind: RegistryKind, id: string): Promise { + const doc = await docsFor(kind).get(id); + return doc ? doc.data.toString('utf-8') : null; +} + +/** Delete a document; only a store outage counts as a failure (a file that is gone already is fine). */ +async function removeDoc(kind: RegistryKind, id: string): Promise { + try { + await docsFor(kind).delete(id); + } catch (error) { + if (isStoreUnavailable(error)) throw error; + } +} + +const LEASE_TTL_MS = 30_000; +const LEASE_WAIT_MS = 15_000; + +/** + * Changes to one registry run one after another: in this process through a + * promise chain on globalThis (every module copy shares it), and across + * replicas under a lease, so two installs of the same extension at once + * can't leave one's bundle under the other's registry entry. The registry + * itself is changed with update(), so even a change that couldn't get the + * lease in time keeps every other entry. A store outage fails here, before + * anything is written. + */ +function withRegistryLock(kind: RegistryKind, task: () => Promise): Promise { + const queues = registryGlobals().queues; + const locked = () => withLease(kind, task); + const run = (queues.get(kind) ?? Promise.resolve()).then(locked, locked); + queues.set(kind, run.catch(() => {})); return run; } +async function withLease(kind: RegistryKind, task: () => Promise): Promise { + if (!isSharedMode() || isConfigReadOnly()) return task(); + const store = getStore(); + const name = `extension-registry-${kind}`; + const holder = `${replicaId()}-${randomBytes(4).toString('hex')}`; + const deadline = Date.now() + LEASE_WAIT_MS; + let held = await store.leases.acquire(name, holder, LEASE_TTL_MS); + while (!held && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 100 + Math.random() * 100)); + held = await store.leases.acquire(name, holder, LEASE_TTL_MS); + } + if (!held) logger.warn(`Another instance is still changing the ${kind}; going ahead without waiting longer`); + try { + return await task(); + } finally { + if (held) await store.leases.release(name, holder).catch(() => {}); + } +} + +const sha256Hex = (data: string) => createHash('sha256').update(data).digest('hex'); + // ─── Plugin Operations ─────────────────────────────────────── -const pluginRegistryPath = () => path.join(getPluginsDir(), 'registry.json'); +const emptyPluginRegistry = (): PluginRegistry => ({ plugins: [] }); const FULL_HASH_RE = /^[0-9a-f]{64}$/; @@ -194,30 +282,48 @@ const FULL_HASH_RE = /^[0-9a-f]{64}$/; * Older builds wrote a 16-char SHA-256 prefix into `bundleHash`. The current * client-side verifyBundle requires equal-length hex (and the full digest for * real integrity), so any registry entry with a truncated or otherwise - * malformed hash needs to be re-hashed from the on-disk bundle. If the bundle - * file is missing the hash is cleared so verifyBundle skips the check rather + * malformed hash needs to be re-hashed from the stored bundle. If the bundle + * is missing the hash is cleared so verifyBundle skips the check rather * than refusing to load. + * + * Returns the fixed hash per plugin id and the hash it replaces (null = + * clear it), or null when nothing needs fixing. */ -async function migrateBundleHashes(registry: PluginRegistry): Promise { - let changed = false; +async function bundleHashFixes(registry: PluginRegistry): Promise | null> { + const fixes = new Map(); for (const plugin of registry.plugins) { if (!plugin.bundleHash || FULL_HASH_RE.test(plugin.bundleHash)) continue; - const bundlePath = path.join(getPluginsDir(), `${plugin.id}.js`); + let code: string | null = null; try { - const code = await readFile(bundlePath); - plugin.bundleHash = createHash('sha256').update(code).digest('hex'); + code = await readText('plugins', `${plugin.id}.js`); } catch { - delete plugin.bundleHash; + /* unreadable: cleared, as a missing bundle */ } + fixes.set(plugin.id, { from: plugin.bundleHash, to: code === null ? null : sha256Hex(code) }); + } + return fixes.size > 0 ? fixes : null; +} + +function applyHashFixes(registry: PluginRegistry, fixes: Map): boolean { + let changed = false; + for (const plugin of registry.plugins) { + const fix = fixes.get(plugin.id); + // Only an entry that still has the old hash: a save in between wins. + if (!fix || plugin.bundleHash !== fix.from) continue; + if (fix.to) plugin.bundleHash = fix.to; + else delete plugin.bundleHash; changed = true; } return changed; } export async function getPluginRegistry(): Promise { - const registry = await readJsonFile(pluginRegistryPath(), { plugins: [] }); - if (await migrateBundleHashes(registry)) { - try { await writeJsonFile(pluginRegistryPath(), registry); } catch { /* read-only fs ok */ } + const registry = await readRegistry('plugins', emptyPluginRegistry); + const fixes = await bundleHashFixes(registry); + if (fixes) { + applyHashFixes(registry, fixes); + // Stored with update(), so an install that lands meanwhile isn't undone. + try { await updateRegistry('plugins', emptyPluginRegistry, (stored) => applyHashFixes(stored, fixes)); } catch { /* read-only fs ok */ } } return registry; } @@ -229,8 +335,9 @@ export async function getPlugin(id: string): Promise { const SAFE_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/; -export function savePlugin(plugin: ServerPlugin, code: string): Promise { - return withRegistryLock('plugins', () => savePluginNow(plugin, code)); +export async function savePlugin(plugin: ServerPlugin, code: string): Promise { + await withRegistryLock('plugins', () => savePluginNow(plugin, code)); + await announceChange('plugins'); } async function savePluginNow( @@ -241,134 +348,179 @@ async function savePluginNow( if (!SAFE_ID_RE.test(plugin.id)) { throw new Error('Invalid plugin id'); } - const dir = getPluginsDir(); - await ensureDir(dir); - const bundleHash = createHash('sha256').update(code).digest('hex'); - const registry = await getPluginRegistry(); - const idx = registry.plugins.findIndex(p => p.id === plugin.id); + const bundleHash = sha256Hex(code); + const current = (await getPluginRegistry()).plugins.find(p => p.id === plugin.id); + // Sign now and keep the signature with the bundle. Signing at serve time + // would sign whatever is stored, so a bundle swapped after install would + // still verify. Signed before anything is written: a key that can't be + // loaded from the shared store fails the install, not half of it. + const signature = await signBundle(plugin.id, code); // Keep what is about to be replaced, so the update can be undone // (extension-history.ts). The same bytes under the same version are not // a replacement worth keeping. - const bundlePath = path.join(dir, `${plugin.id}.js`); - if (idx >= 0) { - const current = registry.plugins[idx]; + if (current) { await forgetVersion('plugin', plugin.id, plugin.version); if (current.version !== plugin.version || current.bundleHash !== bundleHash) { - const currentCode = await readFile(bundlePath, 'utf-8').catch(() => null); + const currentCode = await readText('plugins', `${plugin.id}.js`); if (currentCode !== null) await snapshotPlugin(current, currentCode); } } - // Save code bundle - await writeFile(bundlePath, code, 'utf-8'); - // Sign now and keep the signature with the bundle. Signing at serve time - // would sign whatever is on disk, so a bundle file swapped after install - // would still verify. - await storePluginSignature(plugin.id, code); + // The bundle and its signature first, then the registry entry pointing at them. + const docs = pluginDocs(); + await docs.put(`${plugin.id}.js`, code); + await storeSignature(plugin.id, signature); // Stamp content hash + updatedAt so clients can detect re-uploads even // when the manifest version hasn't changed. Preserve the original // installedAt across re-uploads. The full SHA-256 is required because the // client-side verifyBundle compares the entire digest length-checked. const now = new Date().toISOString(); - - const next: ServerPlugin = { - ...plugin, - bundleHash, - updatedAt: now, - installedAt: idx >= 0 ? registry.plugins[idx].installedAt : plugin.installedAt, - }; - if (idx >= 0) { - registry.plugins[idx] = next; - } else { - registry.plugins.push(next); - } - await writeJsonFile(pluginRegistryPath(), registry); + await updateRegistry('plugins', emptyPluginRegistry, (registry) => { + const idx = registry.plugins.findIndex(p => p.id === plugin.id); + const next: ServerPlugin = { + ...plugin, + bundleHash, + updatedAt: now, + installedAt: idx >= 0 ? registry.plugins[idx].installedAt : plugin.installedAt, + }; + if (idx >= 0) { + registry.plugins[idx] = next; + } else { + registry.plugins.push(next); + } + return true; + }); + if (isSharedMode()) cacheBundle(bundleHash, code, signature); } -export function updatePluginMeta(id: string, updates: Partial>): Promise { - return withRegistryLock('plugins', () => updatePluginMetaNow(id, updates)); +export async function updatePluginMeta(id: string, updates: Partial>): Promise { + const updated = await withRegistryLock('plugins', () => updatePluginMetaNow(id, updates)); + if (updated) await announceChange('plugins'); + return updated; } async function updatePluginMetaNow(id: string, updates: Partial>): Promise { assertWritable('update plugin metadata'); - const registry = await getPluginRegistry(); - const idx = registry.plugins.findIndex(p => p.id === id); - if (idx < 0) return null; - - registry.plugins[idx] = { ...registry.plugins[idx], ...updates, updatedAt: new Date().toISOString() }; - await writeJsonFile(pluginRegistryPath(), registry); - return registry.plugins[idx]; + const now = new Date().toISOString(); + let updated: ServerPlugin | null = null; + await updateRegistry('plugins', emptyPluginRegistry, (registry) => { + updated = null; + const idx = registry.plugins.findIndex(p => p.id === id); + if (idx < 0) return false; + registry.plugins[idx] = { ...registry.plugins[idx], ...updates, updatedAt: now }; + updated = registry.plugins[idx]; + return true; + }); + return updated; } -export function deletePlugin(id: string): Promise { - return withRegistryLock('plugins', () => deletePluginNow(id)); +export async function deletePlugin(id: string): Promise { + const deleted = await withRegistryLock('plugins', () => deletePluginNow(id)); + if (deleted) await announceChange('plugins'); + return deleted; } async function deletePluginNow(id: string): Promise { assertWritable('delete plugin'); - const registry = await getPluginRegistry(); - const idx = registry.plugins.findIndex(p => p.id === id); - if (idx < 0) return false; + let found = false; + await updateRegistry('plugins', emptyPluginRegistry, (registry) => { + const idx = registry.plugins.findIndex(p => p.id === id); + found = idx >= 0; + if (!found) return false; + registry.plugins.splice(idx, 1); + return true; + }); + if (!found) return false; - registry.plugins.splice(idx, 1); - await writeJsonFile(pluginRegistryPath(), registry); - - // Remove bundle and signature files - for (const file of [`${id}.js`, `${id}.sig`]) { - try { await unlink(path.join(getPluginsDir(), file)); } catch { /* ok if missing */ } + // Remove bundle and signature (the registry no longer points at them) + if (SAFE_ID_RE.test(id)) { + for (const file of [`${id}.js`, `${id}.sig`]) await removeDoc('plugins', file); } await purgeHistory('plugin', id); return true; } -async function storePluginSignature(id: string, code: string): Promise { - const sigPath = path.join(getPluginsDir(), `${id}.sig`); +/** Sign a bundle; null when no key can be loaded (the plugin then can't run privileged). */ +async function signBundle(id: string, code: string): Promise { try { const { signBytes } = await import('./plugin-signing'); - const signature = await signBytes(code); - await writeFile(sigPath, signature, 'utf-8'); - return signature; + return await signBytes(code); } catch (err) { + if (isStoreUnavailable(err)) throw err; // Without a key the bundle still installs; it just can't run privileged. - try { await unlink(sigPath); } catch { /* none */ } logger.warn('[plugin-registry] could not sign plugin bundle', { id, error: err instanceof Error ? err.message : String(err) }); return null; } } +async function storeSignature(id: string, signature: string | null): Promise { + if (signature) await pluginDocs().put(`${id}.sig`, signature); + else await removeDoc('plugins', `${id}.sig`); +} + /** * The signature made when the bundle was installed. Bundles installed before * signatures were stored get one on first request (trust on first use), so - * a later change to the file on disk stops verifying from then on. + * a later change to the stored bundle stops verifying from then on. + * + * `bundleHash` (the registry entry's) lets shared mode answer from this + * replica's bundle cache. */ -export async function getPluginSignature(id: string, code: string): Promise { +export async function getPluginSignature(id: string, code: string, bundleHash?: string): Promise { if (!SAFE_ID_RE.test(id)) return null; + const cached = bundleHash && isSharedMode() ? getCachedBundle(bundleHash) : undefined; + if (cached && cached.code === code && cached.signature !== undefined) return cached.signature; + let stored: string | null; try { - return (await readFile(path.join(getPluginsDir(), `${id}.sig`), 'utf-8')).trim() || null; - } catch { - return storePluginSignature(id, code); + stored = await readText('plugins', `${id}.sig`); + } catch (error) { + if (isStoreUnavailable(error)) throw error; + stored = null; } + let signature: string | null; + if (stored !== null) { + signature = stored.trim() || null; + } else { + signature = await signBundle(id, code); + await storeSignature(id, signature); + } + if (cached && cached.code === code) cached.signature = signature; + return signature; } -export async function getPluginBundle(id: string): Promise { - const bundlePath = path.join(getPluginsDir(), `${id}.js`); +/** + * The stored bundle of a plugin. With `bundleHash` (the registry entry's), + * shared mode serves it from this replica's cache once it was fetched and + * found to match. + */ +export async function getPluginBundle(id: string, bundleHash?: string): Promise { + if (!SAFE_ID_RE.test(id)) return null; + const useCache = !!bundleHash && isSharedMode(); + if (useCache) { + const cached = getCachedBundle(bundleHash!); + if (cached) return cached.code; + } + let code: string | null; try { - return await readFile(bundlePath, 'utf-8'); - } catch { + code = await readText('plugins', `${id}.js`); + } catch (error) { + if (isStoreUnavailable(error)) throw error; return null; } + if (code !== null && useCache && sha256Hex(code) === bundleHash) cacheBundle(bundleHash!, code); + return code; } // ─── Theme Operations ──────────────────────────────────────── -const themeRegistryPath = () => path.join(getThemesDir(), 'registry.json'); +const emptyThemeRegistry = (): ThemeRegistry => ({ themes: [] }); export async function getThemeRegistry(): Promise { - return readJsonFile(themeRegistryPath(), { themes: [] }); + return readRegistry('themes', emptyThemeRegistry); } export async function getTheme(id: string): Promise { @@ -376,8 +528,9 @@ export async function getTheme(id: string): Promise { return registry.themes.find(t => t.id === id) || null; } -export function saveTheme(theme: ServerTheme, css: string, skin: string | null = null): Promise { - return withRegistryLock('themes', () => saveThemeNow(theme, css, skin)); +export async function saveTheme(theme: ServerTheme, css: string, skin: string | null = null): Promise { + await withRegistryLock('themes', () => saveThemeNow(theme, css, skin)); + await announceChange('themes'); } async function saveThemeNow( @@ -389,75 +542,85 @@ async function saveThemeNow( if (!SAFE_ID_RE.test(theme.id)) { throw new Error('Invalid theme id'); } - const dir = getThemesDir(); - await ensureDir(dir); const contentHash = createHash('sha256').update(css).update('\0').update(skin ?? '').digest('hex'); - const registry = await getThemeRegistry(); - const idx = registry.themes.findIndex(t => t.id === theme.id); - const cssPath = path.join(dir, `${theme.id}.css`); - const skinPath = path.join(dir, `${theme.id}.skin.css`); + const current = (await getThemeRegistry()).themes.find(t => t.id === theme.id); + const cssId = `${theme.id}.css`; + const skinId = `${theme.id}.skin.css`; // Keep the theme being replaced for a rollback (extension-history.ts). - if (idx >= 0) { - const current = registry.themes[idx]; + if (current) { await forgetVersion('theme', theme.id, theme.version); if (current.version !== theme.version || current.contentHash !== contentHash) { - const currentCss = await readFile(cssPath, 'utf-8').catch(() => null); - const currentSkin = current.hasSkin ? await readFile(skinPath, 'utf-8').catch(() => null) : null; + const currentCss = await readText('themes', cssId); + const currentSkin = current.hasSkin ? await readText('themes', skinId) : null; if (currentCss !== null) await snapshotTheme(current, currentCss, currentSkin); } } - // Save CSS file - await writeFile(cssPath, css, 'utf-8'); + // The CSS and skin first, then the registry entry. + const docs = themeDocs(); + await docs.put(cssId, css); if (skin) { - await writeFile(skinPath, skin, 'utf-8'); + await docs.put(skinId, skin); } else { - try { await unlink(skinPath); } catch { /* none stored */ } + await removeDoc('themes', skinId); } - theme = { ...theme, hasSkin: !!skin, contentHash }; + const next: ServerTheme = { ...theme, hasSkin: !!skin, contentHash }; - // Update registry - if (idx >= 0) { - registry.themes[idx] = theme; - } else { - registry.themes.push(theme); - } - await writeJsonFile(themeRegistryPath(), registry); + await updateRegistry('themes', emptyThemeRegistry, (registry) => { + const idx = registry.themes.findIndex(t => t.id === theme.id); + if (idx >= 0) { + registry.themes[idx] = next; + } else { + registry.themes.push(next); + } + return true; + }); } -export function updateThemeMeta(id: string, updates: Partial>): Promise { - return withRegistryLock('themes', () => updateThemeMetaNow(id, updates)); +export async function updateThemeMeta(id: string, updates: Partial>): Promise { + const updated = await withRegistryLock('themes', () => updateThemeMetaNow(id, updates)); + if (updated) await announceChange('themes'); + return updated; } async function updateThemeMetaNow(id: string, updates: Partial>): Promise { assertWritable('update theme metadata'); - const registry = await getThemeRegistry(); - const idx = registry.themes.findIndex(t => t.id === id); - if (idx < 0) return null; - - registry.themes[idx] = { ...registry.themes[idx], ...updates, updatedAt: new Date().toISOString() }; - await writeJsonFile(themeRegistryPath(), registry); - return registry.themes[idx]; + const now = new Date().toISOString(); + let updated: ServerTheme | null = null; + await updateRegistry('themes', emptyThemeRegistry, (registry) => { + updated = null; + const idx = registry.themes.findIndex(t => t.id === id); + if (idx < 0) return false; + registry.themes[idx] = { ...registry.themes[idx], ...updates, updatedAt: now }; + updated = registry.themes[idx]; + return true; + }); + return updated; } -export function deleteTheme(id: string): Promise { - return withRegistryLock('themes', () => deleteThemeNow(id)); +export async function deleteTheme(id: string): Promise { + const deleted = await withRegistryLock('themes', () => deleteThemeNow(id)); + if (deleted) await announceChange('themes'); + return deleted; } async function deleteThemeNow(id: string): Promise { assertWritable('delete theme'); - const registry = await getThemeRegistry(); - const idx = registry.themes.findIndex(t => t.id === id); - if (idx < 0) return false; + let found = false; + await updateRegistry('themes', emptyThemeRegistry, (registry) => { + const idx = registry.themes.findIndex(t => t.id === id); + found = idx >= 0; + if (!found) return false; + registry.themes.splice(idx, 1); + return true; + }); + if (!found) return false; - registry.themes.splice(idx, 1); - await writeJsonFile(themeRegistryPath(), registry); - - // Remove CSS and skin files - for (const file of [`${id}.css`, `${id}.skin.css`]) { - try { await unlink(path.join(getThemesDir(), file)); } catch { /* ok if missing */ } + // Remove CSS and skin + if (SAFE_ID_RE.test(id)) { + for (const file of [`${id}.css`, `${id}.skin.css`]) await removeDoc('themes', file); } await purgeHistory('theme', id); @@ -466,9 +629,14 @@ async function deleteThemeNow(id: string): Promise { export async function getThemeSkin(id: string): Promise { if (!SAFE_ID_RE.test(id)) return null; + return readThemeAsset(`${id}.skin.css`); +} + +async function readThemeAsset(file: string): Promise { try { - return await readFile(path.join(getThemesDir(), `${id}.skin.css`), 'utf-8'); - } catch { + return await readText('themes', file); + } catch (error) { + if (isStoreUnavailable(error)) throw error; return null; } } @@ -489,10 +657,13 @@ export async function themeCSSResolver(): Promise<(id: string) => string | null> } export async function getThemeCSS(id: string): Promise { - const cssPath = path.join(getThemesDir(), `${id}.css`); - try { - return await readFile(cssPath, 'utf-8'); - } catch { - return null; - } + if (!SAFE_ID_RE.test(id)) return null; + return readThemeAsset(`${id}.css`); +} + +/** Forget the registries this replica keeps for an outage (tests). */ +export function resetRegistryCacheForTests(): void { + const g = registryGlobals(); + g.lastGood.clear(); + g.queues.clear(); } diff --git a/lib/admin/store-unavailable-response.ts b/lib/admin/store-unavailable-response.ts new file mode 100644 index 000000000..9a0a681a3 --- /dev/null +++ b/lib/admin/store-unavailable-response.ts @@ -0,0 +1,15 @@ +import { NextResponse } from 'next/server'; +import { isStoreUnavailable } from './admin-store'; + +/** + * The 503 a route answers when the shared store (#1066) couldn't be + * reached: nothing was written, and the client may try again shortly. + * Returns null for any other error. + */ +export function storeUnavailableResponse(error: unknown): NextResponse | null { + if (!isStoreUnavailable(error)) return null; + return NextResponse.json( + { error: 'Temporarily unavailable, try again in a moment' }, + { status: 503, headers: { 'Retry-After': '10' } }, + ); +} diff --git a/lib/admin/theme-asset.ts b/lib/admin/theme-asset.ts index 9dc700840..e9e892f7f 100644 --- a/lib/admin/theme-asset.ts +++ b/lib/admin/theme-asset.ts @@ -1,6 +1,7 @@ import { NextResponse } from 'next/server'; import { getThemeCSS, getThemeRegistry, getThemeSkin } from './plugin-registry'; import { logger } from '@/lib/logger'; +import { storeUnavailableResponse } from './store-unavailable-response'; /** * Serve an enabled theme's CSS or skin. Clients put the theme's content hash @@ -33,6 +34,8 @@ export async function serveThemeAsset(id: string, kind: 'css' | 'skin'): Promise }, }); } catch (error) { + const unavailable = storeUnavailableResponse(error); + if (unavailable) return unavailable; logger.error('Theme asset serve error', { kind, error: error instanceof Error ? error.message : 'Unknown error' }); return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); }