feat: keep users' plugin credentials in the shared store without losing a concurrent save #1066

lib/__tests__/plugin-vault-shared-store.test.ts

@@ -0,0 +1,99 @@
+// @vitest-environment node
+import { mkdtempSync, readdirSync, rmSync, statSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import path from 'node:path';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { setStoreForTests } from '@/lib/shared-store';
+import { resetLocalStoreForTests } from '@/lib/shared-store/local';
+import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store';
+
+// Users' plugin credentials with a shared store (#1066). The vault used to
+// serialise writes with a queue in its own module, which no other process
+// (and no other module copy) sees: two writes on two replicas lost one.
+
+vi.mock('@/lib/auth/session-secret', () => ({
+  getSessionSecret: () => 's'.repeat(64),
+  hasSessionSecret: () => true,
+}));
+vi.mock('@/lib/logger', () => ({
+  logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} },
+}));
+
+const user = { username: 'alice@example.org', serverUrl: 'https://mail.example.org' };
+const input = (secret: string) => ({ origin: 'https://cloud.example.org', inject: { type: 'basic' } as const, label: 'Cloud', secret });
+
+type Vault = typeof import('@/lib/plugin-secrets/vault');
+
+/** A fresh copy of the vault module, as another process (or bundle) has. */
+async function replica(): Promise<Vault> {
+  vi.resetModules();
+  return import('@/lib/plugin-secrets/vault');
+}
+
+let disks: string;
+let store: SeparateDiskStore;
+
+beforeEach(() => {
+  disks = mkdtempSync(path.join(tmpdir(), 'bw-vault-replicas-'));
+  vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(disks, 'a'));
+  store = separateDiskStore();
+  setStoreForTests(store);
+});
+
+afterEach(() => {
+  setStoreForTests(null);
+  resetLocalStoreForTests();
+  vi.unstubAllEnvs();
+  store.cleanup();
+  rmSync(disks, { recursive: true, force: true });
+});
+
+describe('plugin credentials with a shared store', () => {
+  it('reads on replica B what replica A stored', async () => {
+    const a = await replica();
+    await a.putUserCredential(user, 'nextcloud', 'cloud', input('pw'));
+
+    vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(disks, 'b'));
+    const b = await replica();
+    expect(await b.getUserCredential(user, 'nextcloud', 'cloud')).toMatchObject({ secret: 'pw' });
+  });
+
+  it('keeps both of two credentials saved at once through two replicas', async () => {
+    const a = await replica();
+    const b = await replica();
+    await Promise.all([
+      a.putUserCredential(user, 'nextcloud', 'cloud', input('one')),
+      b.putUserCredential(user, 'paperless', 'docs', input('two')),
+      a.putUserCredential(user, 'radicale', 'dav', input('three')),
+      b.putUserCredential(user, 'immich', 'photos', input('four')),
+    ]);
+
+    const ids = (await a.listUserCredentials(user)).map((c) => c.pluginId).sort();
+    expect(ids).toEqual(['immich', 'nextcloud', 'paperless', 'radicale']);
+  });
+
+  it('answers with StoreUnavailableError while the store is down and writes nothing', async () => {
+    const vault = await replica();
+    await vault.putUserCredential(user, 'nextcloud', 'cloud', input('pw'));
+    store.down = true;
+    await expect(vault.putUserCredential(user, 'paperless', 'docs', input('x'))).rejects.toThrow('down (test)');
+    await expect(vault.listUserCredentials(user)).rejects.toThrow('down (test)');
+    store.down = false;
+    expect((await vault.listUserCredentials(user)).map((c) => c.pluginId)).toEqual(['nextcloud']);
+  });
+});
+
+describe('plugin credentials on a single instance', () => {
+  it('keep today\'s file: <sha256>.enc in the vault dir, mode 0600, gone when the last one goes', async () => {
+    setStoreForTests(null);
+    const vault = await replica();
+    await vault.putUserCredential(user, 'nextcloud', 'cloud', input('pw'));
+    const dir = path.join(disks, 'a');
+    const files = readdirSync(dir);
+    expect(files).toEqual([`${vault.userKey(user)}.enc`]);
+    if (process.platform !== 'win32') expect(statSync(path.join(dir, files[0])).mode & 0o777).toBe(0o600);
+
+    await vault.deleteUserCredentials(user, 'nextcloud');
+    expect(readdirSync(dir)).toEqual([]);
+  });
+});

lib/plugin-secrets/vault.ts

@@ -1,13 +1,13 @@
 // Per-user store of the credentials users connected for plugins. One
-// encrypted file per mail account; the secrets never leave the server again
-// except as an auth header on a request the proxy forwards to the
+// encrypted document per mail account; the secrets never leave the server
+// again except as an auth header on a request the proxy forwards to the
 // credential's own origin.
 
 import { createHash } from 'node:crypto';
-import { readFile, writeFile, mkdir, rename, unlink } from 'node:fs/promises';
 import path from 'node:path';
 import { decryptPayload, encryptPayload } from '@/lib/auth/crypto';
-import { StoredDataUnreadableError } from '@/lib/stored-data-errors';
+import { getStore, type CollectionOptions } from '@/lib/shared-store';
+import { isStoreUnavailable, StoredDataUnreadableError } from '@/lib/stored-data-errors';
 import type { CredentialInject } from './credentials';
 
 export interface VaultUser {
@@ -50,27 +50,33 @@ export function userKey(user: VaultUser): string {
   return createHash('sha256').update(`${user.username}:${user.serverUrl}`).digest('hex');
 }
 
-function vaultPath(user: VaultUser): string {
-  const dir = path.resolve(vaultDir());
-  const file = path.resolve(dir, `${userKey(user)}.enc`);
-  if (!file.startsWith(dir + path.sep)) throw new Error('Invalid vault path');
-  return file;
+const VAULT_FILE = /^[0-9a-f]{64}\.enc$/;
+
+/**
+ * One document per account, keyed by userKey (#1066). Local mode: today's
+ * `<vault dir>/<key>.enc`, base64 `encryptPayload` text, mode 0600.
+ */
+export const VAULT_COLLECTION: CollectionOptions = {
+  local: {
+    dir: () => vaultDir(),
+    file: (id) => `${id}.enc`,
+    idOf: (relative) => (VAULT_FILE.test(relative) ? relative.slice(0, -'.enc'.length) : null),
+    mode: 0o600,
+  },
+};
+
+function collection() {
+  return getStore().collection('plugin-secrets', VAULT_COLLECTION);
 }
 
 /**
- * The user's credentials; {} when none are stored. A file that exists but
- * can't be read or doesn't decrypt (another session secret) throws
- * StoredDataUnreadableError, so no write lands on top of it.
+ * The user's credentials from a stored document; {} for none. One that
+ * doesn't decrypt (another session secret) throws StoredDataUnreadableError,
+ * so no write lands on top of it.
  */
-async function load(user: VaultUser): Promise<VaultData> {
-  let raw: string;
-  try {
-    raw = await readFile(vaultPath(user), 'utf8');
-  } catch (error) {
-    if ((error as NodeJS.ErrnoException).code === 'ENOENT') return {};
-    throw new StoredDataUnreadableError('plugin credentials', { cause: error });
-  }
-  const payload = decryptPayload(raw, 'plugin-secrets');
+function decode(user: VaultUser, raw: Buffer | null): VaultData {
+  if (!raw) return {};
+  const payload = decryptPayload(raw.toString('utf8'), 'plugin-secrets');
   if (!payload) throw new StoredDataUnreadableError('plugin credentials');
   // Bound to this user: a file copied over from another account's name
   // decrypts, but carries the other account's key and is ignored.
@@ -79,28 +85,34 @@ async function load(user: VaultUser): Promise<VaultData> {
   return data && typeof data === 'object' ? (data as VaultData) : {};
 }
 
-async function save(user: VaultUser, data: VaultData): Promise<void> {
-  const file = vaultPath(user);
-  if (Object.keys(data).length === 0) {
-    try { await unlink(file); } catch { /* nothing stored */ }
-    return;
+async function load(user: VaultUser): Promise<VaultData> {
+  let stored;
+  try {
+    stored = await collection().get(userKey(user));
+  } catch (error) {
+    if (isStoreUnavailable(error)) throw error;
+    throw new StoredDataUnreadableError('plugin credentials', { cause: error });
   }
-  await mkdir(path.dirname(file), { recursive: true });
-  const tmp = `${file}.${process.pid}.tmp`;
-  await writeFile(tmp, encryptPayload({ u: userKey(user), d: data }, 'plugin-secrets'), { encoding: 'utf8', mode: 0o600 });
-  await rename(tmp, file);
+  return decode(user, stored?.data ?? null);
 }
 
-// Writes for one user run one after another, so two saves can't drop each
-// other's changes.
-const queues = new Map<string, Promise<unknown>>();
-function serialize<T>(user: VaultUser, task: () => Promise<T>): Promise<T> {
-  const key = userKey(user);
-  const run = (queues.get(key) ?? Promise.resolve()).then(task, task);
-  const tail = run.catch(() => {});
-  queues.set(key, tail);
-  void tail.then(() => { if (queues.get(key) === tail) queues.delete(key); });
-  return run;
+/**
+ * Change the user's credentials in one step against what is stored now, so
+ * two writes at once (two tabs, two replicas) both land. `change` edits the
+ * data it gets and says whether it changed anything; it may run more than
+ * once. An empty vault is deleted.
+ */
+async function modify<T>(user: VaultUser, change: (data: VaultData) => { result: T; changed: boolean }): Promise<T> {
+  let result!: T;
+  await collection().update(userKey(user), (current) => {
+    const data = decode(user, current);
+    const out = change(data);
+    result = out.result;
+    if (!out.changed) return undefined;
+    if (Object.keys(data).length === 0) return current ? null : undefined;
+    return encryptPayload({ u: userKey(user), d: data }, 'plugin-secrets');
+  });
+  return result;
 }
 
 export async function getUserCredential(user: VaultUser, pluginId: string, credentialId: string): Promise<StoredCredential | null> {
@@ -141,8 +153,7 @@ export function putUserCredential(
   credentialId: string,
   input: { origin: string; inject: CredentialInject; label: string; username?: string; secret?: string },
 ): Promise<void> {
-  return serialize(user, async () => {
-    const data = await load(user);
+  return modify(user, (data) => {
     const existing = data[pluginId]?.[credentialId];
     const count = Object.values(data).reduce((n, c) => n + Object.keys(c).length, 0);
     if (!existing && count >= MAX_USER_CREDENTIALS) throw new VaultFullError();
@@ -160,7 +171,7 @@ export function putUserCredential(
         updatedAt: now,
       },
     };
-    await save(user, data);
+    return { result: undefined, changed: true };
   });
 }
 
@@ -172,32 +183,29 @@ export function setUserCredentialSecret(
   secret: string,
   username?: string,
 ): Promise<boolean> {
-  return serialize(user, async () => {
-    const data = await load(user);
+  return modify(user, (data) => {
     const existing = data[pluginId]?.[credentialId];
-    if (!existing) return false;
+    if (!existing) return { result: false, changed: false };
     data[pluginId][credentialId] = {
       ...existing,
       secret,
       ...(username ? { username } : {}),
       updatedAt: new Date().toISOString(),
     };
-    await save(user, data);
-    return true;
+    return { result: true, changed: true };
   });
 }
 
 /** Remove one credential, or all of a plugin's when credentialId is omitted. */
 export function deleteUserCredentials(user: VaultUser, pluginId: string, credentialId?: string): Promise<void> {
-  return serialize(user, async () => {
-    const data = await load(user);
-    if (!data[pluginId]) return;
+  return modify(user, (data) => {
+    if (!data[pluginId]) return { result: undefined, changed: false };
     if (credentialId) {
       delete data[pluginId][credentialId];
       if (Object.keys(data[pluginId]).length === 0) delete data[pluginId];
     } else {
       delete data[pluginId];
     }
-    await save(user, data);
+    return { result: undefined, changed: true };
   });
 }

lib/settings-sync.ts

@@ -2,8 +2,8 @@ import { createHash, createCipheriv, createDecipheriv, randomBytes } from 'node:
 import path from 'node:path';
 import { logger } from '@/lib/logger';
 import { getSessionSecret } from '@/lib/auth/session-secret';
-import { getStore, StoreUnavailableError, type CollectionOptions } from '@/lib/shared-store';
-import { StoredDataUnreadableError } from '@/lib/stored-data-errors';
+import { getStore, type CollectionOptions } from '@/lib/shared-store';
+import { isStoreUnavailable, StoredDataUnreadableError } from '@/lib/stored-data-errors';
 
 const ALGORITHM = 'aes-256-gcm';
 const IV_LENGTH = 12;
@@ -87,7 +87,7 @@ export async function loadUserSettings(username: string, serverUrl: string): Pro
   try {
     stored = await collection().get(settingsId(username, serverUrl));
   } catch (error) {
-    if (error instanceof StoreUnavailableError) throw error;
+    if (isStoreUnavailable(error)) throw error;
     logger.warn('Failed to read user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
     throw new StoredDataUnreadableError('settings', { cause: error });
   }
@@ -113,7 +113,7 @@ export async function deleteUserSettings(username: string, serverUrl: string): P
   try {
     await collection().delete(settingsId(username, serverUrl));
   } catch (error) {
-    if (error instanceof StoreUnavailableError) throw error;
+    if (isStoreUnavailable(error)) throw error;
     logger.warn('Failed to delete user settings', { error: error instanceof Error ? error.message : 'Unknown error' });
   }
 }

lib/stored-data-errors.ts

@@ -1,5 +1,4 @@
 import { NextResponse } from 'next/server';
-import { StoreUnavailableError } from '@/lib/shared-store/types';
 
 /**
  * A user's stored document exists but can't be read: it doesn't decrypt
@@ -15,9 +14,24 @@ export class StoredDataUnreadableError extends Error {
   }
 }
 
+// By name, not instanceof: Next bundles each entry with its own copy of a
+// module, so the store's error class may not be the one this file imports.
+function named(error: unknown, name: string): boolean {
+  return error instanceof Error && error.name === name;
+}
+
+/** The shared store couldn't be reached (StoreUnavailableError from lib/shared-store). */
+export function isStoreUnavailable(error: unknown): boolean {
+  return named(error, 'StoreUnavailableError');
+}
+
+export function isStoredDataUnreadable(error: unknown): boolean {
+  return named(error, 'StoredDataUnreadableError');
+}
+
 /** True for the errors that mean "try again later, nothing was written". */
 export function isStoredDataUnavailable(error: unknown): boolean {
-  return error instanceof StoredDataUnreadableError || error instanceof StoreUnavailableError;
+  return isStoredDataUnreadable(error) || isStoreUnavailable(error);
 }
 
 /**
@@ -28,13 +42,13 @@ export function storedDataUnavailableResponse(
   error: unknown,
   headers?: Record<string, string>,
 ): NextResponse | null {
-  if (error instanceof StoredDataUnreadableError) {
+  if (isStoredDataUnreadable(error)) {
     return NextResponse.json(
       { error: 'Your stored data could not be read on the server. Nothing was changed.', code: 'stored_data_unreadable' },
       { status: 503, headers: { 'Retry-After': '10', ...headers } },
     );
   }
-  if (error instanceof StoreUnavailableError) {
+  if (isStoreUnavailable(error)) {
     return NextResponse.json(
       { error: 'Storage is temporarily unavailable. Nothing was changed.', code: 'store_unavailable' },
       { status: 503, headers: { 'Retry-After': '10', ...headers } },