refactor(plugins): outbound OAuth tokens and per-user plugin settings through repositories

server/src/db/repositories/PluginOauthState.repository.ts

@@ -6,9 +6,86 @@ import { TrekRepository } from './_shared/trek-repository';
  * cascade's own delete (PR43) — the rest of this table's surface belongs to
  * Task 4's `oauth/plugin-oauth.service.ts`, which adds its own additive methods.
  */
+
+/** `consumeByState`'s own projection — PO5's columns plus `plugin_id` (the
+ *  legacy SELECT's own WHERE-bound column, re-checked in JS once the atomic
+ *  delete below no longer filters on it — see that method's docstring). */
+export interface PluginOauthStateRow {
+  plugin_id: string;
+  user_id: number;
+  verifier: string;
+  created_at: number;
+}
+
+/** `consumeByState`'s own Kysely DB shape. */
+interface PluginOauthStateKyselyDB {
+  plugin_oauth_state: {
+    state: string;
+    plugin_id: string;
+    user_id: number;
+    verifier: string;
+    created_at: number;
+  };
+}
+
 export class PluginOauthStateRepository extends TrekRepository<PluginOauthState> {
   /** PR43 (uninstall cascade, `deleteData` branch) — `DELETE FROM plugin_oauth_state WHERE plugin_id = ?`. */
   async deleteAllForPlugin(pluginId: string): Promise<void> {
     await this.nativeDelete({ plugin_id: pluginId });
   }
+
+  /**
+   * PO3 (`startConnect`'s stale-state clear, before minting a fresh one) / PO10
+   * (`disconnect`) — dup text `DELETE FROM plugin_oauth_state WHERE plugin_id = ?
+   * AND user_id = ?`, one method for both call sites.
+   */
+  async deleteForUser(pluginId: string, userId: number): Promise<void> {
+    await this.nativeDelete({ plugin_id: pluginId, user_id: userId });
+  }
+
+  /** PO4 (`startConnect`) — `INSERT INTO plugin_oauth_state (state, plugin_id, user_id, verifier, created_at) VALUES (?, ?, ?, ?, ?)`. */
+  async insertState(state: string, pluginId: string, userId: number, verifier: string, createdAt: number): Promise<void> {
+    await this.insert({ state, plugin_id: pluginId, user_id: userId, verifier, created_at: createdAt });
+  }
+
+  /**
+   * PO5+PO6+PO7 combined (`completeCallback`'s one-shot consume). The legacy
+   * sequence was SELECT-then-DELETE — `SELECT verifier, user_id, created_at
+   * FROM plugin_oauth_state WHERE state = ? AND plugin_id = ?`, then an
+   * UNCONDITIONAL `DELETE FROM plugin_oauth_state WHERE state = ?` on EVERY
+   * path out of `completeCallback` (the refused branch and the success
+   * branch both delete by `state` alone) — but every statement was a
+   * synchronous `better-sqlite3` call with no `await` anywhere between the
+   * read and the delete, so two concurrent `completeCallback` calls for the
+   * SAME state could never interleave: JS ran one call's check+delete to
+   * completion before the other's synchronous block even started.
+   *
+   * Splitting that into a `findOne` (PO5) and a separate `nativeDelete`
+   * (PO6/PO7) — each now genuinely async through MikroORM/Kysely — reopens
+   * exactly that window: a second concurrent caller's read can land before
+   * the first caller's delete does, and BOTH would see a live row and both
+   * would proceed to exchange the code, spending the one-shot state twice.
+   * Collapsed into ONE atomic `DELETE ... RETURNING`, unconditional on
+   * `state` alone (matching the legacy delete's own condition exactly —
+   * `plugin_id`/`user_id`/TTL are re-checked by the caller against the
+   * RETURNED row, same as it always did against the SELECTed one), the same
+   * `WebauthnChallengesRepository.claimChallenge` precedent this program
+   * already uses for an identical single-use-consume race. Whichever
+   * concurrent caller's DELETE the database serialises first gets the real
+   * row back; the other gets `null` and fails closed — restoring the
+   * legacy's effectively-atomic outcome instead of reopening a TOCTOU
+   * window the conversion would otherwise introduce. Built on `this.kysely()`
+   * (validates the request context first), not `em.getKysely()` directly,
+   * and `.returning([...])` (not the QueryBuilder's own `.returning()` —
+   * `claimChallenge`'s docstring records the installed MikroORM version's
+   * gap on a bare DELETE).
+   */
+  async consumeByState(state: string): Promise<PluginOauthStateRow | null> {
+    const row = await this.kysely<PluginOauthStateKyselyDB>()
+      .deleteFrom('plugin_oauth_state')
+      .where('state', '=', state)
+      .returning(['plugin_id', 'user_id', 'verifier', 'created_at'])
+      .executeTakeFirst();
+    return row ?? null;
+  }
 }

server/src/db/repositories/PluginOauthTokens.repository.ts

@@ -1,4 +1,5 @@
 import type { PluginOauthTokens } from '../entities/PluginOauthTokens.entity';
+import { currentTimestampKysely } from '../dialect/sql-functions';
 import { TrekRepository } from './_shared/trek-repository';
 
 /**
@@ -9,6 +10,34 @@ import { TrekRepository } from './_shared/trek-repository';
  * Task 4's `oauth/plugin-oauth.service.ts`. Task 4 adds its own methods to this
  * same repository; this one is additive and does not collide with them.
  */
+
+/** PO8 (`getAccessToken`)'s own projection. Ciphertext columns pass through
+ *  untouched — `access_token`/`refresh_token` are encrypted/decrypted in
+ *  `plugin-oauth.service.ts` only (3e's R6 boundary), never in this repository. */
+export interface PluginOauthTokenRow {
+  access_token: string | null;
+  refresh_token: string | null;
+  expires_at: number | null;
+}
+
+/**
+ * PO11's insert-only shape (`storeToken`'s VALUES list — `updated_at` is always
+ * `datetime('now')`/`CURRENT_TIMESTAMP`, never caller-supplied, same reasoning
+ * `DocumentSyncItemsRepository`'s own write-only Kysely interface docstring gives
+ * for a SEPARATE interface rather than widening the read-shaped row above).
+ */
+interface PluginOauthTokensWriteKyselyDB {
+  plugin_oauth_tokens: {
+    plugin_id: string;
+    user_id: number;
+    access_token: string | null;
+    refresh_token: string | null;
+    expires_at: number | null;
+    scope: string | null;
+    updated_at: string;
+  };
+}
+
 export class PluginOauthTokensRepository extends TrekRepository<PluginOauthTokens> {
   /** PR42 (uninstall cascade, `deleteData` branch) — `DELETE FROM plugin_oauth_tokens WHERE plugin_id = ?`. */
   async deleteAllForPlugin(pluginId: string): Promise<void> {
@@ -20,4 +49,81 @@ export class PluginOauthTokensRepository extends TrekRepository<PluginOauthToken
     const rows = await this.find({ user_id: userId }, { fields: ['plugin_id'] });
     return [...new Set(rows.map((r) => r.plugin_id))];
   }
+
+  /** PO2 (`status`) — `SELECT 1 FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ? AND access_token IS NOT NULL`, as a boolean existence check. */
+  async hasAccessToken(pluginId: string, userId: number): Promise<boolean> {
+    const row = await this.findOne({ plugin_id: pluginId, user_id: userId, access_token: { $ne: null } }, { fields: ['plugin_id'] });
+    return row !== null;
+  }
+
+  /** PO8 (`getAccessToken`) — `SELECT access_token, refresh_token, expires_at FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ?`. */
+  async findTokenRow(pluginId: string, userId: number): Promise<PluginOauthTokenRow | null> {
+    const row = await this.findOne({ plugin_id: pluginId, user_id: userId }, { fields: ['access_token', 'refresh_token', 'expires_at'] });
+    return row ? { access_token: row.access_token ?? null, refresh_token: row.refresh_token ?? null, expires_at: row.expires_at ?? null } : null;
+  }
+
+  /** PO9 (`disconnect`) — `DELETE FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ?`. */
+  async deleteForUser(pluginId: string, userId: number): Promise<void> {
+    await this.nativeDelete({ plugin_id: pluginId, user_id: userId });
+  }
+
+  /**
+   * PO11 (`storeToken`, R-oauth-upsert) — the composite-key upsert on `(plugin_id,
+   * user_id)`. Legacy text:
+   * `` INSERT INTO plugin_oauth_tokens (plugin_id, user_id, access_token,
+   * refresh_token, expires_at, scope, updated_at) VALUES (?, ?, ?, ?, ?, ?,
+   * datetime('now')) ON CONFLICT(plugin_id, user_id) DO UPDATE SET
+   * access_token = excluded.access_token, refresh_token = COALESCE(excluded.refresh_token,
+   * plugin_oauth_tokens.refresh_token), expires_at = excluded.expires_at,
+   * scope = excluded.scope, updated_at = excluded.updated_at ``.
+   *
+   * The `COALESCE` operand order is load-bearing (R-oauth-upsert's own warning,
+   * read twice): NEW value (`excluded.refresh_token`) FIRST, EXISTING
+   * (`plugin_oauth_tokens.refresh_token`) SECOND — a refresh response that DOES
+   * supply a new refresh token uses it; one that omits it (many providers do on a
+   * plain refresh grant) keeps the row's existing token. Reversing the two operands
+   * would silently erase stored offline access on every refresh that omits the
+   * field, with nothing in the request/response cycle ever surfacing it.
+   *
+   * `em.upsert` cannot express a column-level COALESCE merge (`onConflictMergeFields`
+   * only ever re-assigns `excluded.<col>` verbatim or omits the column, never "merge
+   * with the row's own current value") — hand-typed Kysely, the same
+   * `insertInto(...).onConflict(oc => oc.columns([...]).doUpdateSet({...}))` shape
+   * `DocumentSyncItemsRepository.insertOrUpsertOnConflict` established as this
+   * program's precedent for a COALESCE-preserving composite upsert. Ciphertext
+   * passes through untouched: `access_token`/`refresh_token` are already encrypted
+   * by the caller (`plugin-oauth.service.ts#storeToken`, 3e's R6 boundary) — this
+   * repository never calls `encrypt_api_key`/`decrypt_api_key`.
+   */
+  async storeToken(data: {
+    plugin_id: string;
+    user_id: number;
+    access_token: string;
+    refresh_token: string | null;
+    expires_at: number | null;
+    scope: string | null;
+  }): Promise<void> {
+    const platform = this.getEntityManager().getPlatform();
+    await this.kysely<PluginOauthTokensWriteKyselyDB>()
+      .insertInto('plugin_oauth_tokens')
+      .values({
+        plugin_id: data.plugin_id,
+        user_id: data.user_id,
+        access_token: data.access_token,
+        refresh_token: data.refresh_token,
+        expires_at: data.expires_at,
+        scope: data.scope,
+        updated_at: currentTimestampKysely(platform),
+      })
+      .onConflict((oc) =>
+        oc.columns(['plugin_id', 'user_id']).doUpdateSet({
+          access_token: (eb) => eb.ref('excluded.access_token'),
+          refresh_token: (eb) => eb.fn.coalesce(eb.ref('excluded.refresh_token'), eb.ref('plugin_oauth_tokens.refresh_token')),
+          expires_at: (eb) => eb.ref('excluded.expires_at'),
+          scope: (eb) => eb.ref('excluded.scope'),
+          updated_at: () => currentTimestampKysely(platform),
+        }),
+      )
+      .execute();
+  }
 }

server/src/db/repositories/PluginSettingsFields.repository.ts

@@ -92,4 +92,88 @@ export class PluginSettingsFieldsRepository extends TrekRepository<PluginSetting
   async deleteAllForPlugin(pluginId: string): Promise<void> {
     await this.nativeDelete({ plugin_id: pluginId });
   }
+
+  /**
+   * PU1 (Plan 3j Task 4, `plugin-user-settings.service.ts#readOne`) — `SELECT
+   * secret FROM plugin_settings_fields WHERE plugin_id = ? AND field_key = ? AND
+   * scope = 'user'`, as a boolean secret flag deciding whether `readOne` decrypts
+   * the stored value.
+   */
+  async isUserFieldSecret(pluginId: string, fieldKey: string): Promise<boolean> {
+    const row = await this.findOne({ plugin_id: pluginId, field_key: fieldKey, scope: 'user' }, { fields: ['secret'] });
+    return row?.secret === 1;
+  }
+
+  /**
+   * `plugin-user-settings.service.ts#readAll` (Plan 3j Task 4) — `SELECT
+   * field_key, secret FROM plugin_settings_fields WHERE plugin_id = ? AND scope =
+   * ?` (`scope` is always `'user'` at this call site; kept as a parameter to match
+   * this repository's own established convention on every other `scope`-taking
+   * method here).
+   */
+  async listFieldKeysWithSecretFlag(pluginId: string, scope: 'instance' | 'user'): Promise<Array<{ field_key: string; secret: number }>> {
+    const rows = await this.find({ plugin_id: pluginId, scope }, { fields: ['field_key', 'secret'] });
+    return rows.map((r) => ({ field_key: r.field_key, secret: r.secret }));
+  }
+
+  /**
+   * DI8 (Plan 3j Task 3, `install/discovery.ts#upsert`) — `` INSERT INTO
+   * plugin_settings_fields (plugin_id, field_key, label, input_type, placeholder, hint,
+   * required, secret, scope, options, oauth_config, default_value, sort_order) VALUES
+   * (?,?,?,?,?,?,?,?,?,?,?,?,?) ``, looped once per manifest settings field, paired with
+   * DI7's `deleteAllForPlugin` above (the same delete-then-reinsert re-declare sequence).
+   * `insertMany`, one batched native insert — same rows the legacy per-row `.run()` loop
+   * produced. A no-op on an empty manifest `settings[]`.
+   */
+  async insertFields(pluginId: string, fields: NewPluginSettingsFieldRow[]): Promise<void> {
+    if (!fields.length) return;
+    await this.insertMany(
+      fields.map((f) => ({
+        plugin_id: pluginId,
+        field_key: f.field_key,
+        label: f.label,
+        input_type: f.input_type,
+        placeholder: f.placeholder,
+        hint: f.hint,
+        required: f.required,
+        secret: f.secret,
+        scope: f.scope,
+        options: f.options,
+        oauth_config: f.oauth_config,
+        default_value: f.default_value,
+        sort_order: f.sort_order,
+      })),
+    );
+  }
+
+  /**
+   * SD1 (Plan 3j Task 3, `settings-defaults.ts#settingDefaults`) — `SELECT field_key,
+   * default_value FROM plugin_settings_fields WHERE plugin_id = ? AND scope = ? AND
+   * secret = 0 AND default_value IS NOT NULL`. Secrets never carry a default (the
+   * manifest parse drops it before it ever reaches a row) — the `secret = 0` filter is
+   * belt-and-braces, kept exactly as the legacy statement had it.
+   */
+  async listDefaults(pluginId: string, scope: 'instance' | 'user'): Promise<Array<{ field_key: string; default_value: string }>> {
+    const rows = await this.find(
+      { plugin_id: pluginId, scope, secret: 0, default_value: { $ne: null } },
+      { fields: ['field_key', 'default_value'] },
+    );
+    return rows.map((r) => ({ field_key: r.field_key, default_value: r.default_value ?? '' }));
+  }
+}
+
+/** DI8's own row shape — `discoverPlugins#upsert`'s per-manifest-settings-field insert. */
+export interface NewPluginSettingsFieldRow {
+  field_key: string;
+  label: string | null;
+  input_type: string;
+  placeholder: string | null;
+  hint: string | null;
+  required: number;
+  secret: number;
+  scope: string;
+  options: string | null;
+  oauth_config: string | null;
+  default_value: string | null;
+  sort_order: number;
 }

server/src/nest/plugins/oauth/plugin-oauth.module.ts

@@ -1,4 +1,9 @@
 import { Module } from '@nestjs/common';
+import { MikroOrmModule } from '@mikro-orm/nestjs';
+import { Plugins } from '../../../db/entities/Plugins.entity';
+import { PluginOauthTokens } from '../../../db/entities/PluginOauthTokens.entity';
+import { PluginOauthState } from '../../../db/entities/PluginOauthState.entity';
+import { PluginSettingsFields } from '../../../db/entities/PluginSettingsFields.entity';
 import { PluginOAuthController } from './plugin-oauth.controller';
 import { PluginOAuthService } from './plugin-oauth.service';
 
@@ -9,8 +14,19 @@ import { PluginOAuthService } from './plugin-oauth.service';
  *
  * A leaf on purpose — it owns credentials and nothing else, so it must not depend on
  * the runtime or the registry. PluginsRuntimeModule imports it for the token read.
+ *
+ * Plan 3j Task 4 — `PluginOAuthService` now injects `PluginsRepository` (PO1, the
+ * provider-config read), `PluginOauthTokensRepository`, `PluginOauthStateRepository`
+ * and `PluginSettingsFieldsRepository` (the manifest-default fold `settingDefaults`
+ * now needs, per Task 3's own conversion of `settings-defaults.ts`); this module is
+ * where the class is actually constructed, so it (not PluginsRuntimeModule, which is
+ * where the OTHER copies of these same entities are registered for
+ * `PluginRuntimeService`/`PluginsService`/`PluginUserSettingsService`) owns this
+ * `forFeature` list — `MikroOrmModule.forFeature` is idempotent per entity within one
+ * DI graph, so the two modules registering the same entities is not a conflict.
  */
 @Module({
+  imports: [MikroOrmModule.forFeature([Plugins, PluginOauthTokens, PluginOauthState, PluginSettingsFields])],
   controllers: [PluginOAuthController],
   providers: [PluginOAuthService],
   exports: [PluginOAuthService],

server/src/nest/plugins/oauth/plugin-oauth.service.ts

@@ -1,11 +1,19 @@
 import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@mikro-orm/nestjs';
 import crypto from 'node:crypto';
-import { DatabaseService } from '../../database/database.service';
 import { encrypt_api_key, decrypt_api_key } from '../../common/crypto/apiKeyCrypto';
 import { applySettingDefaults, settingDefaults } from '../settings-defaults';
 import { getAppUrl } from '../../../app-config';
 import { isPrivateIp } from '../install/safe-fetch';
 import { safeFetchAdminConfigured } from '../../../utils/ssrfGuard';
+import { Plugins } from '../../../db/entities/Plugins.entity';
+import type { PluginsRepository } from '../../../db/repositories/Plugins.repository';
+import { PluginOauthTokens } from '../../../db/entities/PluginOauthTokens.entity';
+import type { PluginOauthTokensRepository } from '../../../db/repositories/PluginOauthTokens.repository';
+import { PluginOauthState } from '../../../db/entities/PluginOauthState.entity';
+import type { PluginOauthStateRepository } from '../../../db/repositories/PluginOauthState.repository';
+import { PluginSettingsFields } from '../../../db/entities/PluginSettingsFields.entity';
+import type { PluginSettingsFieldsRepository } from '../../../db/repositories/PluginSettingsFields.repository';
 
 /**
  * Host-brokered outbound OAuth (#plugins). A plugin becomes an OAuth *client* of a
@@ -66,26 +74,28 @@ function assertSafeHttps(urlStr: string, what: string): URL {
 
 @Injectable()
 export class PluginOAuthService {
-  constructor(private readonly dbs: DatabaseService) {}
-
-  private get db() {
-    return this.dbs.connection;
-  }
+  constructor(
+    @InjectRepository(Plugins) private readonly pluginsRepo: PluginsRepository,
+    @InjectRepository(PluginOauthTokens) private readonly tokensRepo: PluginOauthTokensRepository,
+    @InjectRepository(PluginOauthState) private readonly stateRepo: PluginOauthStateRepository,
+    @InjectRepository(PluginSettingsFields) private readonly settingsFieldsRepo: PluginSettingsFieldsRepository,
+  ) {}
 
   /** The plugin's decrypted OAuth provider config from its INSTANCE settings, or null
    *  when any required piece is missing/blank. */
   async providerConfig(pluginId: string): Promise<OAuthProviderConfig | null> {
-    const row = this.db.prepare('SELECT config FROM plugins WHERE id = ?').get(pluginId) as { config: string } | undefined;
-    if (!row) return null;
+    // PO1 — dup shape of PR18/PS5/PS13, reuses PluginsRepository.findConfig.
+    const config = await this.pluginsRepo.findConfig(pluginId);
+    if (config === null) return null;
     let cfg: Record<string, unknown>;
     try {
-      cfg = JSON.parse(row.config || '{}');
+      cfg = JSON.parse(config || '{}');
     } catch {
       return null;
     }
     // A provider plugin ships its endpoints/scopes as manifest defaults; the admin types
     // only the client id/secret (secrets — never defaulted).
-    cfg = applySettingDefaults(cfg, await settingDefaults(this.db, pluginId, 'instance'));
+    cfg = applySettingDefaults(cfg, await settingDefaults(this.settingsFieldsRepo, pluginId, 'instance'));
     const authorizeUrl = String(cfg.oauth_authorize_url ?? '').trim();
     const tokenUrl = String(cfg.oauth_token_url ?? '').trim();
     const clientId = cfg.oauth_client_id ? String(decrypt_api_key(cfg.oauth_client_id)) : '';
@@ -102,8 +112,8 @@ export class PluginOAuthService {
   /** Whether the acting user has a stored token for this plugin. */
   async status(pluginId: string, userId: number): Promise<{ configured: boolean; connected: boolean }> {
     const configured = (await this.providerConfig(pluginId)) !== null;
-    const tok = this.db.prepare('SELECT 1 FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ? AND access_token IS NOT NULL').get(pluginId, userId);
-    return { configured, connected: !!tok };
+    const connected = await this.tokensRepo.hasAccessToken(pluginId, userId); // PO2
+    return { configured, connected };
   }
 
   /** Begin the authorize flow: mint PKCE + state, persist them, return the provider URL. */
@@ -118,14 +128,8 @@ export class PluginOAuthService {
     const state = b64url(crypto.randomBytes(24));
 
     // Drop this user's stale states for the plugin, then store the fresh one.
-    this.db.prepare('DELETE FROM plugin_oauth_state WHERE plugin_id = ? AND user_id = ?').run(pluginId, userId);
-    this.db.prepare('INSERT INTO plugin_oauth_state (state, plugin_id, user_id, verifier, created_at) VALUES (?, ?, ?, ?, ?)').run(
-      state,
-      pluginId,
-      userId,
-      verifier,
-      nowMs,
-    );
+    await this.stateRepo.deleteForUser(pluginId, userId); // PO3
+    await this.stateRepo.insertState(state, pluginId, userId, verifier, nowMs); // PO4
 
     authorize.searchParams.set('response_type', 'code');
     authorize.searchParams.set('client_id', cfg.clientId);
@@ -139,16 +143,18 @@ export class PluginOAuthService {
 
   /** Complete the callback: verify state, exchange the code, store the tokens. */
   async completeCallback(pluginId: string, userId: number, code: string, state: string, nowMs: number): Promise<void> {
-    const row = this.db.prepare('SELECT verifier, user_id, created_at FROM plugin_oauth_state WHERE state = ? AND plugin_id = ?').get(state, pluginId) as
-      | { verifier: string; user_id: number; created_at: number }
-      | undefined;
-    // State must exist, belong to THIS user, and be fresh — this binds the callback to
-    // the connect request and blocks CSRF / a replayed/foreign state.
-    if (!row || row.user_id !== userId || nowMs - row.created_at > STATE_TTL_MS) {
-      this.db.prepare('DELETE FROM plugin_oauth_state WHERE state = ?').run(state);
+    // PO5+PO6+PO7, atomically — the state is consumed (deleted) by this ONE call
+    // regardless of outcome, same as the legacy SELECT-then-unconditional-DELETE
+    // sequence, but without the SELECT→await→DELETE race the split conversion
+    // would have reopened (see PluginOauthStateRepository.consumeByState's own
+    // docstring). Single-use is therefore enforced HERE, before the checks below,
+    // not on a later path-specific call.
+    const row = await this.stateRepo.consumeByState(state);
+    // State must exist, belong to THIS plugin + user, and be fresh — this binds the
+    // callback to the connect request and blocks CSRF / a replayed/foreign state.
+    if (!row || row.plugin_id !== pluginId || row.user_id !== userId || nowMs - row.created_at > STATE_TTL_MS) {
       throw new Error('invalid or expired OAuth state');
     }
-    this.db.prepare('DELETE FROM plugin_oauth_state WHERE state = ?').run(state); // single-use
 
     const cfg = await this.providerConfig(pluginId);
     if (!cfg) throw new Error('OAuth is not configured for this plugin');
@@ -165,9 +171,7 @@ export class PluginOAuthService {
   /** A valid access token for the acting user, refreshing it if it is expiring. Null when
    *  the user hasn't connected. The plugin never receives the refresh token. */
   async getAccessToken(pluginId: string, userId: number, nowMs: number): Promise<string | null> {
-    const row = this.db.prepare('SELECT access_token, refresh_token, expires_at FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ?').get(pluginId, userId) as
-      | { access_token: string | null; refresh_token: string | null; expires_at: number | null }
-      | undefined;
+    const row = await this.tokensRepo.findTokenRow(pluginId, userId); // PO8
     if (!row || !row.access_token) return null;
 
     const notExpiring = row.expires_at == null || row.expires_at - REFRESH_SKEW_S * 1000 > nowMs;
@@ -187,8 +191,8 @@ export class PluginOAuthService {
   }
 
   async disconnect(pluginId: string, userId: number): Promise<void> {
-    this.db.prepare('DELETE FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ?').run(pluginId, userId);
-    this.db.prepare('DELETE FROM plugin_oauth_state WHERE plugin_id = ? AND user_id = ?').run(pluginId, userId);
+    await this.tokensRepo.deleteForUser(pluginId, userId); // PO9
+    await this.stateRepo.deleteForUser(pluginId, userId); // PO10
   }
 
   // --- internals ---
@@ -224,22 +228,16 @@ export class PluginOAuthService {
   private async storeToken(pluginId: string, userId: number, token: { access_token?: string; refresh_token?: string; expires_in?: number; scope?: string }, nowMs: number): Promise<void> {
     if (!token.access_token) throw new Error('token endpoint returned no access_token');
     const expiresAt = token.expires_in ? nowMs + token.expires_in * 1000 : null;
-    this.db.prepare(
-      `INSERT INTO plugin_oauth_tokens (plugin_id, user_id, access_token, refresh_token, expires_at, scope, updated_at)
-       VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
-       ON CONFLICT(plugin_id, user_id) DO UPDATE SET
-         access_token = excluded.access_token,
-         refresh_token = COALESCE(excluded.refresh_token, plugin_oauth_tokens.refresh_token),
-         expires_at = excluded.expires_at,
-         scope = excluded.scope,
-         updated_at = excluded.updated_at`,
-    ).run(
-      pluginId,
-      userId,
-      encrypt_api_key(token.access_token),
-      token.refresh_token ? encrypt_api_key(token.refresh_token) : null,
-      expiresAt,
-      token.scope ?? null,
-    );
+    // PO11 (R-oauth-upsert) — encryption happens HERE, in the service, before the
+    // repository call; the repository's `storeToken` passes ciphertext through
+    // untouched and never calls encrypt_api_key/decrypt_api_key (3e's R6 boundary).
+    await this.tokensRepo.storeToken({
+      plugin_id: pluginId,
+      user_id: userId,
+      access_token: encrypt_api_key(token.access_token),
+      refresh_token: token.refresh_token ? encrypt_api_key(token.refresh_token) : null,
+      expires_at: expiresAt,
+      scope: token.scope ?? null,
+    });
   }
 }

server/src/nest/plugins/plugin-user-settings.service.ts

@@ -1,8 +1,12 @@
 import { Injectable } from '@nestjs/common';
-import { DatabaseService } from '../database/database.service';
+import { InjectRepository } from '@mikro-orm/nestjs';
 import { decrypt_api_key } from '../common/crypto/apiKeyCrypto';
 import { safeParseConfig } from './plugin-config-parse';
 import { isFilled, settingDefaults } from './settings-defaults';
+import { PluginSettingsFields } from '../../db/entities/PluginSettingsFields.entity';
+import type { PluginSettingsFieldsRepository } from '../../db/repositories/PluginSettingsFields.repository';
+import { PluginUserConfig } from '../../db/entities/PluginUserConfig.entity';
+import type { PluginUserConfigRepository } from '../../db/repositories/PluginUserConfig.repository';
 
 /**
  * A plugin's per-user settings, decrypted host-side.
@@ -17,22 +21,16 @@ import { isFilled, settingDefaults } from './settings-defaults';
  */
 @Injectable()
 export class PluginUserSettingsService {
-  constructor(private readonly dbs: DatabaseService) {}
-
-  private get db() {
-    return this.dbs.connection;
-  }
+  constructor(
+    @InjectRepository(PluginSettingsFields) private readonly settingsFields: PluginSettingsFieldsRepository,
+    @InjectRepository(PluginUserConfig) private readonly userConfig: PluginUserConfigRepository,
+  ) {}
 
   /** One decrypted value for the acting user, for the runtime's `ctx.settings.get()`. */
   async readOne(pluginId: string, userId: number, key: string): Promise<unknown> {
-    const isSecret =
-      (
-        this.db
-          .prepare("SELECT secret FROM plugin_settings_fields WHERE plugin_id = ? AND field_key = ? AND scope = 'user'")
-          .get(pluginId, key) as { secret: number } | undefined
-      )?.secret === 1;
+    const isSecret = await this.settingsFields.isUserFieldSecret(pluginId, key); // PU1
     const value = (await this.storedFor(pluginId, userId))[key];
-    if (value == null) return (await settingDefaults(this.db, pluginId, 'user'))[key]; // unset → the manifest default, if any
+    if (value == null) return (await settingDefaults(this.settingsFields, pluginId, 'user'))[key]; // unset → the manifest default, if any
     return isSecret ? decrypt_api_key(value as string) : value;
   }
 
@@ -44,14 +42,12 @@ export class PluginUserSettingsService {
    * resolves against the acting user) would return undefined there.
    */
   async readAll(pluginId: string, userId: number): Promise<Record<string, unknown>> {
-    const fields = this.db
-      .prepare("SELECT field_key, secret FROM plugin_settings_fields WHERE plugin_id = ? AND scope = 'user'")
-      .all(pluginId) as Array<{ field_key: string; secret: number }>;
+    const fields = await this.settingsFields.listFieldKeysWithSecretFlag(pluginId, 'user');
     const stored = await this.storedFor(pluginId, userId);
     // Null-prototype for the same reason as safeParseConfig: never let a field key write
     // through to Object.prototype on the way out to the plugin.
     const out: Record<string, unknown> = Object.create(null);
-    const defaults = await settingDefaults(this.db, pluginId, 'user');
+    const defaults = await settingDefaults(this.settingsFields, pluginId, 'user');
     for (const field of fields) {
       const value = stored[field.field_key];
       if (value == null) {
@@ -74,21 +70,15 @@ export class PluginUserSettingsService {
    * must not leave them "not configured".
    */
   async hasRequired(pluginId: string, userId: number): Promise<boolean> {
-    const required = this.db
-      .prepare(
-        "SELECT field_key FROM plugin_settings_fields WHERE plugin_id = ? AND scope = 'user' AND required = 1 AND input_type != 'checkbox'",
-      )
-      .all(pluginId) as Array<{ field_key: string }>;
+    const required = await this.settingsFields.listRequiredFieldKeys(pluginId, 'user');
     if (required.length === 0) return true;
     const stored = await this.storedFor(pluginId, userId);
-    const defaults = await settingDefaults(this.db, pluginId, 'user');
-    return required.every((field) => isFilled(stored[field.field_key] ?? defaults[field.field_key]));
+    const defaults = await settingDefaults(this.settingsFields, pluginId, 'user');
+    return required.every((key) => isFilled(stored[key] ?? defaults[key]));
   }
 
   private async storedFor(pluginId: string, userId: number): Promise<Record<string, unknown>> {
-    const row = this.db
-      .prepare('SELECT config FROM plugin_user_config WHERE plugin_id = ? AND user_id = ?')
-      .get(pluginId, userId) as { config: string } | undefined;
-    return safeParseConfig(row?.config ?? '{}');
+    const config = await this.userConfig.findConfig(pluginId, userId); // PU2 — reuses Task 2's PS7/PS9/PS11 method
+    return safeParseConfig(config ?? '{}');
   }
 }

server/tests/integration/plugins/plugin-runtime.task2-conversion.test.ts

@@ -76,7 +76,7 @@ async function buildRuntime(registrar?: CronRegistrarService): Promise<PluginRun
   const dbs = new DatabaseService(testDb);
   const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
   const addons = await createTestAddonsService(testDb, dbs);
-  const userSettings = new PluginUserSettingsService(dbs);
+  const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
   const uow = new UnitOfWork(t.em);
   return new PluginRuntimeService(
     dbs, audit, addons, userSettings,
@@ -158,7 +158,7 @@ describe('Plan 3j Task 2 — R-install-gates named accept+refuse pairs (PR17/PR2
       const dbs = new DatabaseService(testDb);
       const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
       const addons = await createTestAddonsService(testDb, dbs);
-      const userSettings = new PluginUserSettingsService(dbs);
+      const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
       const rtU = new PluginRuntimeService(
         dbs, audit, addons, userSettings,
         t.repo(Plugins), t.repo(PluginErrorLog), t.repo(PluginScheduledTasks), t.repo(PluginUserErasureQueue),
@@ -181,7 +181,7 @@ describe('Plan 3j Task 2 — R-install-gates named accept+refuse pairs (PR17/PR2
       const dbs = new DatabaseService(testDb);
       const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
       const addons = await createTestAddonsService(testDb, dbs);
-      const userSettings = new PluginUserSettingsService(dbs);
+      const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
       const rtU = new PluginRuntimeService(
         dbs, audit, addons, userSettings,
         t.repo(Plugins), t.repo(PluginErrorLog), t.repo(PluginScheduledTasks), t.repo(PluginUserErasureQueue),
@@ -216,7 +216,7 @@ describe('Plan 3j Task 2 — R-install-gates named accept+refuse pairs (PR17/PR2
       const dbs = new DatabaseService(testDb);
       const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
       const addons = await createTestAddonsService(testDb, dbs);
-      const userSettings = new PluginUserSettingsService(dbs);
+      const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
       const registry = registryFor('pr28-accept');
       const rtR = new PluginRuntimeService(
         dbs, audit, addons, userSettings,
@@ -241,7 +241,7 @@ describe('Plan 3j Task 2 — R-install-gates named accept+refuse pairs (PR17/PR2
       const dbs = new DatabaseService(testDb);
       const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
       const addons = await createTestAddonsService(testDb, dbs);
-      const userSettings = new PluginUserSettingsService(dbs);
+      const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
       const registry = registryFor('pr28-refuse');
       vi.mocked(registry.assertRetrustable).mockRejectedValue(new Error('nothing to re-trust'));
       const rtR = new PluginRuntimeService(

server/tests/integration/plugins/plugin-scheduler-sweep.test.ts

@@ -131,7 +131,7 @@ async function buildRuntime(registrar?: CronRegistrarService, orm?: TestOrm['orm
   const dbs = new DatabaseService(testDb);
   const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
   const addons = await createTestAddonsService(testDb, dbs);
-  const userSettings = new PluginUserSettingsService(dbs);
+  const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
   return new PluginRuntimeService(
     dbs,
     audit,

server/tests/unit/db/repositories/PluginOauthTokens.repository.test.ts

@@ -0,0 +1,115 @@
+/**
+ * `PluginOauthTokensRepository.storeToken` (Plan 3j Task 4, PO11 — R-oauth-upsert):
+ * the composite-key (`plugin_id`, `user_id`) upsert whose `refresh_token` column is
+ * COALESCE-preserving. This is the repository-level proof the task brief names as
+ * mandatory: `plugin-oauth.service.ts#getAccessToken`'s own refresh path pre-fills a
+ * missing `refresh_token` in JS before calling `storeToken` (so it never actually
+ * reaches this repository's COALESCE with a null), and a brand-new row has no
+ * existing value for COALESCE to fall back to either way — so neither of those two
+ * call sites, alone, proves the repository's own COALESCE branch. This file calls
+ * `storeToken` directly, against an EXISTING row, with a genuinely omitted
+ * `refresh_token`.
+ */
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
+import { createSnapshotTestDb } from '../../../helpers/db-mock';
+import { resetTestDb } from '../../../helpers/test-db';
+import { createTestOrm, type TestOrm } from '../../../helpers/test-orm';
+import { PluginOauthTokens } from '../../../../src/db/entities/PluginOauthTokens.entity';
+import type { PluginOauthTokensRepository } from '../../../../src/db/repositories/PluginOauthTokens.repository';
+import { currentTimestampKysely } from '../../../../src/db/dialect/sql-functions';
+
+const testDb = createSnapshotTestDb();
+let t: TestOrm;
+let repo: PluginOauthTokensRepository;
+
+beforeAll(async () => {
+  t = await createTestOrm(testDb);
+  repo = t.repo(PluginOauthTokens);
+});
+beforeEach(() => { resetTestDb(testDb); t.clear(); });
+afterAll(async () => { await t.close(); testDb.close(); });
+
+function storedRow(pluginId: string, userId: number) {
+  return testDb
+    .prepare('SELECT plugin_id, user_id, access_token, refresh_token, expires_at, scope, updated_at FROM plugin_oauth_tokens WHERE plugin_id = ? AND user_id = ?')
+    .get(pluginId, userId) as { plugin_id: string; user_id: number; access_token: string | null; refresh_token: string | null; expires_at: number | null; scope: string | null; updated_at: string } | undefined;
+}
+
+describe('PluginOauthTokensRepository', () => {
+  describe('storeToken (PO11, R-oauth-upsert — the COALESCE-preserving composite upsert)', () => {
+    it('PO11REPO-001: pinned rendered SQL for the ON CONFLICT DO UPDATE target and SET clause', () => {
+      const platform = t.em.getPlatform();
+      const compiled = t.em
+        .getKysely<{ plugin_oauth_tokens: Record<string, unknown> }>()
+        .insertInto('plugin_oauth_tokens')
+        .values({
+          plugin_id: 'p', user_id: 1, access_token: 'AT', refresh_token: 'RT', expires_at: 111, scope: 'read',
+          updated_at: currentTimestampKysely(platform),
+        })
+        .onConflict((oc) =>
+          oc.columns(['plugin_id', 'user_id']).doUpdateSet({
+            access_token: (eb) => eb.ref('excluded.access_token'),
+            refresh_token: (eb) => eb.fn.coalesce(eb.ref('excluded.refresh_token'), eb.ref('plugin_oauth_tokens.refresh_token')),
+            expires_at: (eb) => eb.ref('excluded.expires_at'),
+            scope: (eb) => eb.ref('excluded.scope'),
+            updated_at: () => currentTimestampKysely(platform),
+          }),
+        )
+        .compile();
+      // Pinned for the task report — the exact ON CONFLICT clause Kysely renders
+      // for this table. The COALESCE operand order is the load-bearing part
+      // (R-oauth-upsert): NEW value (excluded) first, EXISTING second.
+      expect(compiled.sql).toContain('on conflict ("plugin_id", "user_id") do update set');
+      expect(compiled.sql).toContain('"access_token" = "excluded"."access_token"');
+      expect(compiled.sql).toContain('"refresh_token" = coalesce("excluded"."refresh_token", "plugin_oauth_tokens"."refresh_token")');
+      expect(compiled.sql).toContain('"expires_at" = "excluded"."expires_at"');
+      expect(compiled.sql).toContain('"scope" = "excluded"."scope"');
+      expect(compiled.sql).toContain('"updated_at" = CURRENT_TIMESTAMP');
+      // Pinned in full, for the task report:
+      expect(compiled.sql).toBe(
+        'insert into "plugin_oauth_tokens" ("plugin_id", "user_id", "access_token", "refresh_token", "expires_at", "scope", "updated_at") values (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) on conflict ("plugin_id", "user_id") do update set "access_token" = "excluded"."access_token", "refresh_token" = coalesce("excluded"."refresh_token", "plugin_oauth_tokens"."refresh_token"), "expires_at" = "excluded"."expires_at", "scope" = "excluded"."scope", "updated_at" = CURRENT_TIMESTAMP',
+      );
+    });
+
+    it('PO11REPO-002 (R-oauth-upsert parity): a refresh response that OMITS refresh_token keeps the existing stored value unchanged', async () => {
+      // Seed an existing token row with a non-null refresh_token — the state a
+      // previously-connected user is in when a later refresh comes back without one.
+      await repo.storeToken({ plugin_id: 'p', user_id: 42, access_token: 'AT-1', refresh_token: 'RT-original', expires_at: 1000, scope: 'read' });
+      expect(storedRow('p', 42)).toMatchObject({ access_token: 'AT-1', refresh_token: 'RT-original' });
+
+      // A refresh response that omits refresh_token — the repository call the
+      // service makes when it has NOT pre-filled the field itself.
+      await repo.storeToken({ plugin_id: 'p', user_id: 42, access_token: 'AT-2', refresh_token: null, expires_at: 2000, scope: 'read write' });
+      const row = storedRow('p', 42)!;
+      // access_token/expires_at/scope always take the new value (plain excluded.* reassignment)...
+      expect(row.access_token).toBe('AT-2');
+      expect(row.expires_at).toBe(2000);
+      expect(row.scope).toBe('read write');
+      // ...but refresh_token is UNCHANGED — still the ORIGINAL value, not null, not overwritten.
+      expect(row.refresh_token).toBe('RT-original');
+      expect(row.refresh_token).not.toBeNull();
+    });
+
+    it('PO11REPO-003 (R-oauth-upsert, the other COALESCE direction): a refresh response that DOES supply a new refresh_token overwrites the stored one', async () => {
+      await repo.storeToken({ plugin_id: 'p', user_id: 42, access_token: 'AT-1', refresh_token: 'RT-original', expires_at: 1000, scope: 'read' });
+      await repo.storeToken({ plugin_id: 'p', user_id: 42, access_token: 'AT-2', refresh_token: 'RT-rotated', expires_at: 2000, scope: 'read' });
+      const row = storedRow('p', 42)!;
+      expect(row.refresh_token).toBe('RT-rotated');
+    });
+
+    it('PO11REPO-004: a brand-new row (no prior state) inserts plainly — no COALESCE fallback target to speak of yet', async () => {
+      expect(storedRow('fresh', 7)).toBeUndefined();
+      await repo.storeToken({ plugin_id: 'fresh', user_id: 7, access_token: 'AT', refresh_token: null, expires_at: null, scope: null });
+      const row = storedRow('fresh', 7)!;
+      expect(row.access_token).toBe('AT');
+      expect(row.refresh_token).toBeNull();
+    });
+
+    it('PO11REPO-005: the composite key isolates rows — a second user of the same plugin gets its own row, the first is untouched', async () => {
+      await repo.storeToken({ plugin_id: 'p', user_id: 1, access_token: 'AT-1', refresh_token: 'RT-1', expires_at: null, scope: null });
+      await repo.storeToken({ plugin_id: 'p', user_id: 2, access_token: 'AT-2', refresh_token: 'RT-2', expires_at: null, scope: null });
+      expect(storedRow('p', 1)).toMatchObject({ access_token: 'AT-1', refresh_token: 'RT-1' });
+      expect(storedRow('p', 2)).toMatchObject({ access_token: 'AT-2', refresh_token: 'RT-2' });
+    });
+  });
+});

server/tests/unit/plugins/plugin-oauth.service.test.ts

@@ -16,9 +16,11 @@ vi.mock('../../../src/app-config', async (importOriginal) => {
 });
 
 const { getDb } = vi.hoisted(() => ({ getDb: { current: null as unknown } }));
+// The mock stays even though nothing in this file imports `db` directly any more
+// (the service dropped its last raw-connection use) — a transitive importer of
+// `src/db/database` elsewhere in the module graph would otherwise try to open a
+// real connection at import time.
 vi.mock('../../../src/db/database', () => ({ get db() { return getDb.current; } }));
-import { db as dbConn } from '../../../src/db/database';
-import { DatabaseService } from '../../../src/nest/database/database.service';
 
 // The token POST now runs through the SSRF guard (ssrfGuard.safeFetchLlm), which
 // resolves the host before fetching. Stub DNS so the fake provider.example host
@@ -31,6 +33,11 @@ vi.mock('node:dns/promises', () => {
 
 import Database from 'better-sqlite3';
 import { PluginOAuthService } from '../../../src/nest/plugins/oauth/plugin-oauth.service';
+import { sharedTestOrm } from '../../helpers/test-uow';
+import { Plugins } from '../../../src/db/entities/Plugins.entity';
+import { PluginOauthTokens } from '../../../src/db/entities/PluginOauthTokens.entity';
+import { PluginOauthState } from '../../../src/db/entities/PluginOauthState.entity';
+import { PluginSettingsFields } from '../../../src/db/entities/PluginSettingsFields.entity';
 
 const CFG = {
   oauth_authorize_url: 'https://provider.example/authorize',
@@ -44,7 +51,7 @@ function freshDb(cfg: Record<string, unknown> = CFG) {
   const d = new Database(':memory:');
   d.exec(`
     CREATE TABLE plugins (id TEXT PRIMARY KEY, config TEXT, status TEXT);
-    CREATE TABLE plugin_settings_fields (plugin_id TEXT, field_key TEXT, scope TEXT, secret INTEGER, default_value TEXT);
+    CREATE TABLE plugin_settings_fields (id INTEGER PRIMARY KEY AUTOINCREMENT, plugin_id TEXT, field_key TEXT, scope TEXT, secret INTEGER, default_value TEXT);
     CREATE TABLE plugin_oauth_tokens (plugin_id TEXT, user_id INTEGER, access_token TEXT, refresh_token TEXT, expires_at INTEGER, scope TEXT, updated_at TEXT, PRIMARY KEY (plugin_id, user_id));
     CREATE TABLE plugin_oauth_state (state TEXT PRIMARY KEY, plugin_id TEXT, user_id INTEGER, verifier TEXT, created_at INTEGER);
   `);
@@ -54,14 +61,25 @@ function freshDb(cfg: Record<string, unknown> = CFG) {
 
 const NOW = 1_700_000_000_000;
 
+/**
+ * `PluginOAuthService`, built with real repositories over whichever fresh
+ * `getDb.current` the caller just set — `sharedTestOrm` is memoized per db HANDLE,
+ * and every test below assigns a brand-new `:memory:` db, so each call here gets
+ * its own ORM, matching the pattern `plugin-user-settings.test.ts` established.
+ */
+async function makeOauthService(): Promise<PluginOAuthService> {
+  const orm = await sharedTestOrm(getDb.current as Database.Database);
+  return new PluginOAuthService(orm.repo(Plugins), orm.repo(PluginOauthTokens), orm.repo(PluginOauthState), orm.repo(PluginSettingsFields));
+}
+
 describe('PluginOAuthService', () => {
   let svc: PluginOAuthService;
-  beforeEach(() => { getDb.current = freshDb(); svc = new PluginOAuthService(new DatabaseService(dbConn)); vi.restoreAllMocks(); dnsState.address = '93.184.216.34'; dnsState.family = 4; });
+  beforeEach(async () => { getDb.current = freshDb(); svc = await makeOauthService(); vi.restoreAllMocks(); dnsState.address = '93.184.216.34'; dnsState.family = 4; });
 
   it('providerConfig returns null unless every piece is present, decrypting the secrets', async () => {
     expect(await svc.providerConfig('p')).toMatchObject({ clientId: 'client-123', clientSecret: 'secret-abc', scopes: 'read write' });
     getDb.current = freshDb({ ...CFG, oauth_client_secret: '' });
-    expect(await new PluginOAuthService(new DatabaseService(dbConn)).providerConfig('p')).toBeNull();
+    expect(await (await makeOauthService()).providerConfig('p')).toBeNull();
   });
 
   it('providerConfig falls back to the manifest defaults for the endpoints the admin left unset', async () => {
@@ -74,7 +92,7 @@ describe('PluginOAuthService', () => {
     ins.run('oauth_authorize_url', JSON.stringify('https://provider.example/authorize'));
     ins.run('oauth_token_url', JSON.stringify('https://provider.example/token'));
     ins.run('oauth_scopes', JSON.stringify('read'));
-    svc = new PluginOAuthService(new DatabaseService(dbConn));
+    svc = (await makeOauthService());
     expect(await svc.providerConfig('p')).toEqual({
       authorizeUrl: 'https://provider.example/authorize',
       tokenUrl: 'https://provider.example/token',
@@ -102,18 +120,18 @@ describe('PluginOAuthService', () => {
 
   it('rejects a non-https / loopback / metadata / internal authorize endpoint', async () => {
     getDb.current = freshDb({ ...CFG, oauth_authorize_url: 'http://provider.example/authorize' });
-    await expect(new PluginOAuthService(new DatabaseService(dbConn)).startConnect('p', 42, NOW)).rejects.toThrow(/https/);
+    await expect((await makeOauthService()).startConnect('p', 42, NOW)).rejects.toThrow(/https/);
     getDb.current = freshDb({ ...CFG, oauth_token_url: 'https://127.0.0.1/token' });
-    await expect(new PluginOAuthService(new DatabaseService(dbConn)).startConnect('p', 42, NOW)).rejects.toThrow(/loopback|private/);
+    await expect((await makeOauthService()).startConnect('p', 42, NOW)).rejects.toThrow(/loopback|private/);
     // IPv6-literal loopback must not slip past the fast-fail
     getDb.current = freshDb({ ...CFG, oauth_token_url: 'https://[::1]/token' });
-    await expect(new PluginOAuthService(new DatabaseService(dbConn)).startConnect('p', 42, NOW)).rejects.toThrow(/loopback/);
+    await expect((await makeOauthService()).startConnect('p', 42, NOW)).rejects.toThrow(/loopback/);
     // cloud-metadata by literal is refused too
     getDb.current = freshDb({ ...CFG, oauth_token_url: 'https://169.254.169.254/token' });
-    await expect(new PluginOAuthService(new DatabaseService(dbConn)).startConnect('p', 42, NOW)).rejects.toThrow(/loopback|metadata/);
+    await expect((await makeOauthService()).startConnect('p', 42, NOW)).rejects.toThrow(/loopback|metadata/);
     // an internal name suffix is refused
     getDb.current = freshDb({ ...CFG, oauth_token_url: 'https://idp.internal/token' });
-    await expect(new PluginOAuthService(new DatabaseService(dbConn)).startConnect('p', 42, NOW)).rejects.toThrow(/local/);
+    await expect((await makeOauthService()).startConnect('p', 42, NOW)).rejects.toThrow(/local/);
   });
 
   it('completeCallback verifies state (single-use, user-bound, TTL), exchanges the code, encrypts tokens', async () => {
@@ -180,4 +198,55 @@ describe('PluginOAuthService', () => {
     await svc.disconnect('p', 42);
     expect((await svc.status('p', 42)).connected).toBe(false);
   });
+
+  /**
+   * PO5+PO6+PO7 — single-use is enforced by `consumeByState`'s ONE atomic
+   * delete, on EVERY path out of `completeCallback`: the expired/foreign-state
+   * refusal and the success path both leave the row gone, not just the
+   * success one.
+   */
+  it('single-use: the state row is gone after EITHER the refused path or the success path', async () => {
+    const rows = getDb.current as unknown as InstanceType<typeof Database>;
+
+    // Refused path: wrong user — the row is still consumed.
+    const stateA = new URL(await svc.startConnect('p', 42, NOW)).searchParams.get('state')!;
+    expect(rows.prepare('SELECT 1 FROM plugin_oauth_state WHERE state = ?').get(stateA)).toBeTruthy();
+    await expect(svc.completeCallback('p', 99, 'code', stateA, NOW + 1000)).rejects.toThrow(/state/);
+    expect(rows.prepare('SELECT 1 FROM plugin_oauth_state WHERE state = ?').get(stateA)).toBeUndefined();
+
+    // Success path.
+    const stateB = new URL(await svc.startConnect('p', 42, NOW)).searchParams.get('state')!;
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue({ ok: true, json: async () => ({ access_token: 'AT' }) } as Response);
+    await svc.completeCallback('p', 42, 'code', stateB, NOW + 1000);
+    expect(rows.prepare('SELECT 1 FROM plugin_oauth_state WHERE state = ?').get(stateB)).toBeUndefined();
+  });
+
+  /**
+   * R-oauth-upsert's TRAP: the state consume is a check-then-act primitive.
+   * `consumeByState`'s atomic `DELETE ... RETURNING` (PluginOauthState.repository.ts)
+   * exists specifically so two concurrent `completeCallback` calls for the SAME
+   * state cannot both see a live row — racing them here is the proof, matching
+   * the legacy synchronous-JS outcome (exactly one consumer wins, the token
+   * exchange fires exactly once).
+   */
+  it('the state consume race: two concurrent completeCallback calls for the SAME state — exactly one wins', async () => {
+    const state = new URL(await svc.startConnect('p', 42, NOW)).searchParams.get('state')!;
+    const fetchMock = vi.spyOn(globalThis, 'fetch').mockResolvedValue({
+      ok: true, json: async () => ({ access_token: 'AT', refresh_token: 'RT', expires_in: 3600 }),
+    } as Response);
+
+    const results = await Promise.allSettled([
+      svc.completeCallback('p', 42, 'code', state, NOW + 1000),
+      svc.completeCallback('p', 42, 'code', state, NOW + 1000),
+    ]);
+    const fulfilled = results.filter((r) => r.status === 'fulfilled');
+    const rejected = results.filter((r) => r.status === 'rejected');
+    expect(fulfilled).toHaveLength(1); // exactly one consumer wins the state
+    expect(rejected).toHaveLength(1);
+    expect((rejected[0] as PromiseRejectedResult).reason as Error).toMatchObject({ message: expect.stringMatching(/state/) });
+    // The token exchange fired exactly once — the loser never reached it.
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const rows = getDb.current as unknown as InstanceType<typeof Database>;
+    expect(rows.prepare('SELECT 1 FROM plugin_oauth_state WHERE state = ?').get(state)).toBeUndefined();
+  });
 });

server/tests/unit/plugins/plugin-runtime.boot-no-orm.test.ts

@@ -64,7 +64,7 @@ async function buildRuntime(withOrm: boolean): Promise<PluginRuntimeService> {
   const dbs = new DatabaseService(testDb);
   const audit = new AuditService(t.repo(AuditLog), t.repo(Users));
   const addons = await createTestAddonsService(testDb, dbs);
-  const userSettings = new PluginUserSettingsService(dbs);
+  const userSettings = new PluginUserSettingsService(t.repo(PluginSettingsFields), t.repo(PluginUserConfig));
   return new PluginRuntimeService(
     dbs,
     audit,

server/tests/unit/plugins/plugin-user-settings.test.ts

@@ -46,8 +46,15 @@ async function makePluginsService(dbs: DatabaseService): Promise<PluginsService>
     orm.repo(PluginCapabilityAudit),
   );
 }
-/** The host-side settings reads, over the same connection the test seeded. */
-const userSettings = () => new PluginUserSettingsService(new DatabaseService(dbConn));
+/**
+ * The host-side settings reads, over the same connection the test seeded — a fresh
+ * `orm` per test (`sharedTestOrm` is memoized per db HANDLE, and `beforeEach` swaps
+ * `getDb.current` to a brand-new `:memory:` db every time, same as `makePluginsService`).
+ */
+async function userSettings(): Promise<PluginUserSettingsService> {
+  const orm = await sharedTestOrm(getDb.current as Database.Database);
+  return new PluginUserSettingsService(orm.repo(PluginSettingsFields), orm.repo(PluginUserConfig));
+}
 
 function freshDb() {
   const d = new Database(':memory:');
@@ -84,14 +91,14 @@ describe('per-user plugin settings', () => {
     expect(masked.apiKey).toBe('••••••••');       // never echoed
     expect(masked.units).toBe('metric');
     // decrypted runtime read returns the real value; the stored form is ciphertext
-    expect(await userSettings().readOne('p', 42, 'apiKey')).toBe('sk-123');
+    expect(await (await userSettings()).readOne('p', 42, 'apiKey')).toBe('sk-123');
     expect((await svc.getUserConfig('p', 42)).apiKey).toBe('••••••••');
   });
 
   it('an unchanged secret (the mask) keeps the stored ciphertext', async () => {
     await svc.updateUserConfig('p', 42, { apiKey: 'sk-123' });
     await svc.updateUserConfig('p', 42, { apiKey: '••••••••', units: 'imperial' }); // mask = untouched
-    expect(await userSettings().readOne('p', 42, 'apiKey')).toBe('sk-123'); // still the original
+    expect(await (await userSettings()).readOne('p', 42, 'apiKey')).toBe('sk-123'); // still the original
     expect((await svc.getUserConfig('p', 42)).units).toBe('imperial');
   });
 
@@ -110,7 +117,7 @@ describe('per-user plugin settings', () => {
     // test is that user 99 sees none of user 42's values.
     await svc.updateUserConfig('p', 42, { apiKey: 'sk-123', units: 'metric' });
     expect((await svc.getUserConfig('p', 99)).units).toBeUndefined();
-    expect(await userSettings().readOne('p', 99, 'apiKey')).toBeUndefined();
+    expect(await (await userSettings()).readOne('p', 99, 'apiKey')).toBeUndefined();
   });
 });
 
@@ -132,28 +139,28 @@ describe('manifest defaults reach the runtime reads', () => {
   });
 
   it('readOne falls back to the declared default when nothing is stored', async () => {
-    expect(await userSettings().readOne('p', 42, 'region')).toBe('eu');
-    expect(await userSettings().readOne('p', 42, 'retries')).toBe(3); // JSON round-trip keeps the type
-    expect(await userSettings().readOne('p', 42, 'apiKey')).toBeUndefined(); // no default declared
+    expect(await (await userSettings()).readOne('p', 42, 'region')).toBe('eu');
+    expect(await (await userSettings()).readOne('p', 42, 'retries')).toBe(3); // JSON round-trip keeps the type
+    expect(await (await userSettings()).readOne('p', 42, 'apiKey')).toBeUndefined(); // no default declared
   });
 
   it('a stored value wins over the default', async () => {
     await svc.updateUserConfig('p', 42, { apiKey: 'sk-123', region: 'us' });
-    expect(await userSettings().readOne('p', 42, 'region')).toBe('us');
-    expect(await userSettings().readOne('p', 99, 'region')).toBe('eu'); // another user still gets the default
+    expect(await (await userSettings()).readOne('p', 42, 'region')).toBe('us');
+    expect(await (await userSettings()).readOne('p', 99, 'region')).toBe('eu'); // another user still gets the default
   });
 
   it('readAll folds defaults in for the fields the user left unset', async () => {
     await svc.updateUserConfig('p', 42, { apiKey: 'sk-123', region: 'us' });
-    expect(await userSettings().readAll('p', 42)).toMatchObject({ apiKey: 'sk-123', region: 'us', retries: 3, endpoint: 'https://api.example' });
+    expect(await (await userSettings()).readAll('p', 42)).toMatchObject({ apiKey: 'sk-123', region: 'us', retries: 3, endpoint: 'https://api.example' });
   });
 
   it('hasRequired treats a required field with a default as filled', async () => {
     // apiKey is required with no default → not configured until stored.
-    expect(await userSettings().hasRequired('p', 42)).toBe(false);
+    expect(await (await userSettings()).hasRequired('p', 42)).toBe(false);
     await svc.updateUserConfig('p', 42, { apiKey: 'sk-123' });
     // endpoint is required too, but its default satisfies it.
-    expect(await userSettings().hasRequired('p', 42)).toBe(true);
+    expect(await (await userSettings()).hasRequired('p', 42)).toBe(true);
   });
 });
 
@@ -173,13 +180,13 @@ describe('hasRequired applies the same "filled" rule as the save gate', () => {
 
   it('exempts a required checkbox (consent, not a settings field)', async () => {
     await svc.updateUserConfig('p', 42, { apiKey: 'sk-123' }); // consent never set
-    expect(await userSettings().hasRequired('p', 42)).toBe(true);
+    expect(await (await userSettings()).hasRequired('p', 42)).toBe(true);
   });
 
   it('treats a whitespace-only value as empty', async () => {
     (getDb.current as import('better-sqlite3').Database)
       .prepare("INSERT INTO plugin_user_config (plugin_id, user_id, config, updated_at) VALUES ('p', 42, ?, '')")
       .run(JSON.stringify({ apiKey: '   ', consent: true }));
-    expect(await userSettings().hasRequired('p', 42)).toBe(false);
+    expect(await (await userSettings()).hasRequired('p', 42)).toBe(false);
   });
 });

server/tests/unit/plugins/settings-isolation.test.ts

@@ -46,8 +46,15 @@ import { PluginMetaMigrations } from '../../../src/db/entities/PluginMetaMigrati
 import { PluginCapabilityAudit } from '../../../src/db/entities/PluginCapabilityAudit.entity';
 import { Settings } from '../../../src/db/entities/Settings.entity';
 import { NotificationChannelPreferences } from '../../../src/db/entities/NotificationChannelPreferences.entity';
-/** The host-side settings reads, over the same connection the test seeded. */
-const userSettings = () => new PluginUserSettingsService(new DatabaseService(dbConn));
+/**
+ * The host-side settings reads, over the same connection the test seeded. `t` is
+ * initialized in `beforeAll` (below) before any test runs — this reaches for
+ * `PluginSettingsFieldsRepository`/`PluginUserConfigRepository` through it, matching
+ * PSET-007's own `PluginRuntimeService` construction further down, which reuses the
+ * SAME `t` rather than a second ORM over the same connection.
+ */
+const userSettings = () =>
+  new PluginUserSettingsService((t as TestOrm).repo(PluginSettingsFields), (t as TestOrm).repo(PluginUserConfig));
 
 let uid: number;
 let t: TestOrm | undefined;
@@ -66,7 +73,14 @@ function setUserConfig(pluginId: string, config: Record<string, unknown>) {
   testDb.prepare('INSERT OR REPLACE INTO plugin_user_config (plugin_id, user_id, config) VALUES (?, ?, ?)').run(pluginId, uid, JSON.stringify(config));
 }
 
-beforeAll(() => { createTables(testDb); runMigrations(testDb); });
+beforeAll(async () => {
+  createTables(testDb);
+  runMigrations(testDb);
+  // Plan 3j Task 4 — `userSettings()` above now needs repositories, so the ORM this
+  // file eventually built only inside PSET-007 is built here instead, before any
+  // test runs (PSET-001..006 call `userSettings()` with no ORM of their own).
+  t = await createTestOrm(dbConn);
+});
 beforeEach(() => {
   testDb.prepare('DELETE FROM plugin_settings_fields').run();
   testDb.prepare('DELETE FROM plugin_user_config').run();
@@ -149,7 +163,6 @@ describe('a plugin channel label is bounded by the host', () => {
        VALUES ('loud', 'Loud', 'active', 1, '1.0.0', '[]', '[]', ?, '{}')`,
     ).run(JSON.stringify({ notificationChannel: { title: '🎉'.repeat(5) + 'A'.repeat(500) } }));
 
-    t = await createTestOrm(dbConn);
     const rt = new PluginRuntimeService(
       new DatabaseService(dbConn),
       new AuditService(t.repo(AuditLog), t.repo(Users)),