feat(core): Add trusted source store with CacheService (no-changelog) (#39976)

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

packages/@n8n/backend-services/src/services/cache/__tests__/cache.service.test.ts

@@ -162,6 +162,15 @@ for (const backend of ['memory', 'redis'] as const) {
 				await expect(promise).resolves.toBe('refreshValue');
 			});
 
+			test('should honor ttl on a refreshed value', async () => {
+				const promise = cacheService.get('key', { refreshFn: createRefreshFn(), ttl: 100 });
+				await expect(promise).resolves.toBe('refreshValue');
+
+				await sleep(200);
+
+				await expect(cacheService.get('key')).resolves.toBeUndefined();
+			});
+
 			test('should handle non-ASCII key', async () => {
 				const nonAsciiKey = 'ԱԲԳ';
 				await cacheService.set(nonAsciiKey, 'value');

packages/@n8n/backend-services/src/services/cache/cache.service.ts

@@ -181,7 +181,12 @@ export class CacheService extends TypedEmitter<CacheEvents> {
 		{
 			fallbackValue,
 			refreshFn,
-		}: { fallbackValue?: T; refreshFn?: (key: string) => Promise<T | undefined> } = {},
+			ttl,
+		}: {
+			fallbackValue?: T;
+			refreshFn?: (key: string) => Promise<T | undefined>;
+			ttl?: number;
+		} = {},
 	) {
 		if (!this.cache) await this.init();
 
@@ -201,7 +206,7 @@ export class CacheService extends TypedEmitter<CacheEvents> {
 			this.emit('metrics.cache.update');
 
 			const refreshValue = await refreshFn(key);
-			await this.set(key, refreshValue);
+			await this.set(key, refreshValue, ttl);
 
 			return refreshValue;
 		}

packages/@n8n/backend-test-utils/src/test-db.ts

@@ -233,6 +233,8 @@ type EntityName =
 	| 'TokenExchangeJti'
 	| 'TrustedKeySourceEntity'
 	| 'TrustedKeyEntity'
+	| 'TrustedSourceEntity'
+	| 'TrustedSourceIdentityEntity'
 	| 'WorkflowStatisticsDelta';
 
 /**

packages/@n8n/inbound-auth/src/__tests__/trusted-source-config.test.ts

@@ -5,7 +5,6 @@ import {
 	migrateToLatest,
 	TrustedSourceConfigSchema,
 	trustedSourceConfigSchemaFor,
-	TrustedSourceSecretsSchema,
 	type TrustedSourceConfig,
 } from '../trusted-source-config';
 import { trustedSourceConfigV1Fixture } from './fixtures/trusted-source-config.v1';
@@ -127,14 +126,18 @@ describe('TrustedSourceConfigSchema shape', () => {
 		expect(issuesOf(result)).toContainEqual(expect.objectContaining({ path }));
 	});
 
-	test('a registered client carries no secret in the document', () => {
+	test('a registered client keeps its client secret in the document', () => {
 		const document = withAuthentication({
-			client: { kind: 'registered', clientId: 'n8n', clientSecret: 'leaked' },
+			client: { kind: 'registered', clientId: 'n8n', clientSecret: 's3cret' },
 		});
 
 		const parsed = TrustedSourceConfigSchema.parse(document);
 
-		expect(parsed.authentication.client).toEqual({ kind: 'registered', clientId: 'n8n' });
+		expect(parsed.authentication.client).toEqual({
+			kind: 'registered',
+			clientId: 'n8n',
+			clientSecret: 's3cret',
+		});
 	});
 });
 
@@ -322,17 +325,6 @@ describe('manual discovery endpoints must be https', () => {
 	);
 });
 
-describe('TrustedSourceSecretsSchema', () => {
-	test('accepts a client secret and refuses unknown keys', () => {
-		expect(
-			TrustedSourceSecretsSchema.safeParse({ version: 1, clientSecret: 's3cret' }).success,
-		).toBe(true);
-		expect(TrustedSourceSecretsSchema.safeParse({ version: 1, clientId: 'n8n' }).success).toBe(
-			false,
-		);
-	});
-});
-
 describe('migrateToLatest', () => {
 	const asConfig = (document: Record<string, unknown>) =>
 		document as unknown as TrustedSourceConfig;

packages/@n8n/inbound-auth/src/trusted-source-config.ts

@@ -75,12 +75,12 @@ export const Oauth2AuthenticationSchema = z.object({
 	// Upper bounds keep both windows from disabling `exp`/`nbf` checks. Relaxing later is additive.
 	maxTokenLifetimeSeconds: z.number().int().positive().max(86400).default(86400),
 	clockSkewSeconds: z.number().int().nonnegative().max(300).default(60),
-	// The client secret is not part of the document: see `TrustedSourceSecretsSchema`.
 	client: z
 		.discriminatedUnion('kind', [
 			z.object({
 				kind: z.literal('registered'),
 				clientId: z.string().min(1),
+				clientSecret: z.string().min(1).optional(),
 				scopes: z.array(z.string()).optional(),
 			}),
 			z.object({ kind: z.literal('virtual') }),
@@ -150,6 +150,7 @@ export const TrustedSourceConfigV1Schema = z.object({
 });
 export type TrustedSourceConfigV1 = z.infer<typeof TrustedSourceConfigV1Schema>;
 export type TrustedSourceConfigV1Input = z.input<typeof TrustedSourceConfigV1Schema>;
+export type TrustedSourceConfigInput = z.input<typeof TrustedSourceConfigSchema>;
 
 const ownerRoleIssue = (path: Array<string | number>): z.IssueData => ({
 	code: z.ZodIssueCode.custom,
@@ -256,19 +257,6 @@ export const trustedSourceConfigSchemaFor = (managedBy: ManagedBy) =>
 		}
 	});
 
-/**
- * Secret material of a trusted source. Stored apart from the config document in its own encrypted
- * column, so the document stays safe to return to an admin UI unchanged. Strict: a secret payload
- * with an unknown key is a bug, not a forward-compatible read.
- */
-export const TrustedSourceSecretsSchema = z
-	.object({
-		version: z.literal(1),
-		clientSecret: z.string().min(1).optional(),
-	})
-	.strict();
-export type TrustedSourceSecrets = z.infer<typeof TrustedSourceSecretsSchema>;
-
 type ConfigMigration = (config: TrustedSourceConfig) => TrustedSourceConfig;
 
 /** One step per version, keyed by the version it migrates from. Empty until a v2 exists. */

packages/cli/src/modules/inbound-auth-core/database/repositories/trusted-source-identity.repository.ts

@@ -1,4 +1,4 @@
-import { BaseRepository, TransactionRunner } from '@n8n/db';
+import { BaseRepository, OperationContext, TransactionRunner } from '@n8n/db';
 import { Service } from '@n8n/di';
 import { DataSource } from '@n8n/typeorm';
 
@@ -8,14 +8,23 @@ class TrustedSourceIdentityStore extends BaseRepository<TrustedSourceIdentityEnt
 	constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
 		super(TrustedSourceIdentityEntity, dataSource.manager, transactionRunner);
 	}
+
+	override managerFor(ctx: OperationContext) {
+		return super.managerFor(ctx);
+	}
 }
 
 @Service()
 export class TrustedSourceIdentityRepository {
-	// ponytail: protected only so noUnusedLocals accepts a field nothing reads yet; private with the first method
-	protected readonly store: TrustedSourceIdentityStore;
+	private readonly store: TrustedSourceIdentityStore;
 
 	constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
 		this.store = new TrustedSourceIdentityStore(dataSource, transactionRunner);
 	}
+
+	async clearByTrustedSourceId(trustedSourceId: string, ctx: OperationContext = {}): Promise<void> {
+		await this.store
+			.managerFor(ctx)
+			.delete(TrustedSourceIdentityEntity, { sourceId: trustedSourceId });
+	}
 }

packages/cli/src/modules/inbound-auth-core/database/repositories/trusted-source.repository.ts

@@ -1,21 +1,66 @@
-import { BaseRepository, TransactionRunner } from '@n8n/db';
+import { BaseRepository, OperationContext, TransactionRunner } from '@n8n/db';
 import { Service } from '@n8n/di';
 import { DataSource } from '@n8n/typeorm';
 
 import { TrustedSourceEntity } from '../entities/trusted-source.entity';
 
-class TrustedSourceStore extends BaseRepository<TrustedSourceEntity> {
+class TrustedSourceTable extends BaseRepository<TrustedSourceEntity> {
 	constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
 		super(TrustedSourceEntity, dataSource.manager, transactionRunner);
 	}
+
+	override managerFor(ctx: OperationContext) {
+		return super.managerFor(ctx);
+	}
 }
 
+export type NewTrustedSourceRow = Pick<
+	TrustedSourceEntity,
+	| 'name'
+	| 'type'
+	| 'issuer'
+	| 'managedBy'
+	| 'status'
+	| 'lastError'
+	| 'lastCheckedAt'
+	| 'configVersion'
+	| 'config'
+>;
+
+export type TrustedSourceRowChanges = Partial<
+	Pick<TrustedSourceEntity, 'name' | 'issuer' | 'configVersion' | 'config'>
+>;
+
 @Service()
 export class TrustedSourceRepository {
-	// ponytail: protected only so noUnusedLocals accepts a field nothing reads yet; private with the first method
-	protected readonly store: TrustedSourceStore;
+	private readonly table: TrustedSourceTable;
 
 	constructor(dataSource: DataSource, transactionRunner: TransactionRunner) {
-		this.store = new TrustedSourceStore(dataSource, transactionRunner);
+		this.table = new TrustedSourceTable(dataSource, transactionRunner);
+	}
+
+	async findById(id: string, ctx: OperationContext = {}) {
+		return await this.table.managerFor(ctx).findOneBy(TrustedSourceEntity, { id });
+	}
+
+	async findByIssuer(issuer: string, ctx: OperationContext = {}) {
+		return await this.table.managerFor(ctx).findOneBy(TrustedSourceEntity, { issuer });
+	}
+
+	async findAll(ctx: OperationContext = {}) {
+		return await this.table.managerFor(ctx).find(TrustedSourceEntity, { order: { name: 'ASC' } });
+	}
+
+	/** `save`, not `insert`: the `@BeforeInsert` id generator only runs on an entity instance. */
+	async insertRow(row: NewTrustedSourceRow, ctx: OperationContext = {}) {
+		return await this.table.managerFor(ctx).save(TrustedSourceEntity, this.table.create(row));
+	}
+
+	async updateById(id: string, changes: TrustedSourceRowChanges, ctx: OperationContext = {}) {
+		await this.table.managerFor(ctx).update(TrustedSourceEntity, { id }, changes);
+	}
+
+	async deleteById(id: string, ctx: OperationContext = {}) {
+		await this.table.managerFor(ctx).delete(TrustedSourceEntity, { id });
 	}
 }

packages/cli/src/modules/inbound-auth-core/trusted-source.store.ts

@@ -0,0 +1,211 @@
+import { Logger } from '@n8n/backend-common';
+import { Service } from '@n8n/di';
+import { NotFoundError } from '@n8n/errors';
+import {
+	migrateToLatest,
+	trustedSourceConfigSchemaFor,
+	type SurfaceId,
+	type TrustedSourceConfigInput,
+	type TrustedSourceConfigLatest,
+} from '@n8n/inbound-auth';
+import { ensureError } from '@n8n/utils/errors/ensure-error';
+import { Cipher } from 'n8n-core';
+import { jsonParse, UserError } from 'n8n-workflow';
+
+import type {
+	TrustedSourceEntity,
+	TrustedSourceManagedBy,
+	TrustedSourceStatus,
+	TrustedSourceType,
+} from './database/entities/trusted-source.entity';
+import { TrustedSourceIdentityRepository } from './database/repositories/trusted-source-identity.repository';
+import {
+	TrustedSourceRepository,
+	type TrustedSourceRowChanges,
+} from './database/repositories/trusted-source.repository';
+import { TransactionRunner } from '@n8n/db';
+import { Time } from '@n8n/constants';
+import { CacheService } from '@n8n/backend-services';
+
+/**
+ * A trusted source as consumers see it: decrypted, validated and migrated to the latest config
+ * version. Dates are ISO strings so the shape survives a Redis round-trip unchanged.
+ */
+export type TrustedSource = {
+	id: string;
+	name: string;
+	type: TrustedSourceType;
+	issuer: string;
+	managedBy: TrustedSourceManagedBy;
+	status: TrustedSourceStatus;
+	lastError: string | null;
+	lastCheckedAt: string | null;
+	createdAt: string;
+	updatedAt: string;
+	config: TrustedSourceConfigLatest;
+};
+
+/** The store still validates `config`: the admin-vs-system rules depend on the row, not the type. */
+export type CreateTrustedSourceInput = {
+	name: string;
+	issuer: string;
+	config: TrustedSourceConfigInput;
+};
+export type UpdateTrustedSourceInput = Partial<CreateTrustedSourceInput>;
+
+export class SystemTrustedSourceModificationError extends UserError {
+	constructor(action: 'update' | 'delete') {
+		super(`Cannot ${action} a system-managed trusted source.`);
+	}
+}
+
+const idKey = (id: string) => `trusted-source:id:${id}`;
+const issuerKey = (issuer: string) => `trusted-source:issuer:${issuer}`;
+const ALL_KEY = 'trusted-source:all';
+// Reads are look-aside, so a read that started before a concurrent write can re-cache the old
+// row after the write invalidated it. We accept that; this TTL caps the stale window at 5 minutes.
+const CACHE_TTL = 5 * Time.minutes.toMilliseconds;
+
+@Service()
+export class TrustedSourceStore {
+	constructor(
+		private readonly logger: Logger,
+		private readonly trustedSourceRepository: TrustedSourceRepository,
+		private readonly cipher: Cipher,
+		private readonly cacheService: CacheService,
+		private readonly trustedSourceIdentityRepository: TrustedSourceIdentityRepository,
+		private readonly transactionRunner: TransactionRunner,
+	) {}
+
+	async getById(id: string): Promise<TrustedSource | undefined> {
+		return await this.cacheService.get(idKey(id), {
+			ttl: CACHE_TTL,
+			refreshFn: async () => await this.load(await this.trustedSourceRepository.findById(id)),
+		});
+	}
+
+	async getByIssuer(issuer: string): Promise<TrustedSource | undefined> {
+		return await this.cacheService.get(issuerKey(issuer), {
+			ttl: CACHE_TTL,
+			refreshFn: async () =>
+				await this.load(await this.trustedSourceRepository.findByIssuer(issuer)),
+		});
+	}
+
+	async listBySurface(surface: SurfaceId): Promise<TrustedSource[]> {
+		const allSources =
+			(await this.cacheService.get(ALL_KEY, {
+				ttl: CACHE_TTL,
+				refreshFn: async () => await this.loadAll(),
+			})) ?? [];
+		return allSources.filter((source) => source.config.surfaces[surface] !== undefined);
+	}
+
+	async create(input: CreateTrustedSourceInput): Promise<TrustedSource> {
+		// Persist the parsed, migrated output, not the raw input: stored rows hold the latest version
+		// with defaults filled in, so a later default change must not alter them.
+		const config = migrateToLatest(trustedSourceConfigSchemaFor('admin').parse(input.config));
+		const row = await this.trustedSourceRepository.insertRow({
+			name: input.name,
+			issuer: input.issuer,
+			type: config.authentication.type,
+			managedBy: 'admin',
+			status: 'unchecked',
+			lastError: null,
+			lastCheckedAt: null,
+			configVersion: config.version,
+			config: await this.cipher.encryptV2(config),
+		});
+		await this.invalidateCache(row);
+		return this.toRuntime(row, config);
+	}
+
+	async update(
+		id: string,
+		input: UpdateTrustedSourceInput,
+		options?: { clearBindings: boolean },
+	): Promise<void> {
+		const row = await this.loadAdminRow(id, 'update');
+		const changes: TrustedSourceRowChanges = {};
+		if (input.name !== undefined) changes.name = input.name;
+		if (input.issuer !== undefined) changes.issuer = input.issuer;
+		if (input.config !== undefined) {
+			const config = migrateToLatest(
+				trustedSourceConfigSchemaFor(row.managedBy).parse(input.config),
+			);
+			changes.configVersion = config.version;
+			changes.config = await this.cipher.encryptV2(config);
+		}
+		await this.transactionRunner.run({}, async (ctx) => {
+			if (Object.keys(changes).length > 0) {
+				await this.trustedSourceRepository.updateById(id, changes, ctx);
+			}
+			// After the row write, so a failed update does not drop bindings.
+			if (options?.clearBindings) {
+				await this.trustedSourceIdentityRepository.clearByTrustedSourceId(id, ctx);
+			}
+		});
+		await this.invalidateCache(row, changes.issuer);
+	}
+
+	async delete(id: string): Promise<void> {
+		const row = await this.loadAdminRow(id, 'delete');
+		await this.trustedSourceRepository.deleteById(id); // identities go with it via FK cascade
+		await this.invalidateCache(row);
+	}
+
+	private async loadAdminRow(id: string, action: 'update' | 'delete') {
+		const row = await this.trustedSourceRepository.findById(id);
+		if (!row) throw new NotFoundError(`Trusted source "${id}" not found`);
+		if (row.managedBy === 'system') throw new SystemTrustedSourceModificationError(action);
+		return row;
+	}
+
+	/** `row` is the state before the write, so `row.issuer` is the old issuer. */
+	private async invalidateCache(row: TrustedSourceEntity, newIssuer = row.issuer) {
+		await this.cacheService.deleteMany([
+			idKey(row.id),
+			issuerKey(row.issuer),
+			issuerKey(newIssuer),
+			ALL_KEY,
+		]);
+	}
+
+	/** Any failure makes the row unusable: warn and skip it. Reads never write. */
+	private async load(row: TrustedSourceEntity | null): Promise<TrustedSource | undefined> {
+		if (!row) return undefined;
+		try {
+			const document = jsonParse<unknown>(await this.cipher.decryptV2(row.config));
+			const config = trustedSourceConfigSchemaFor(row.managedBy).parse(document);
+			return this.toRuntime(row, migrateToLatest(config));
+		} catch (error) {
+			this.logger.warn('Skipping unusable trusted source', {
+				id: row.id,
+				reason: ensureError(error).message,
+			});
+			return undefined;
+		}
+	}
+
+	private async loadAll(): Promise<TrustedSource[]> {
+		const rows = await this.trustedSourceRepository.findAll();
+		const sources = await Promise.all(rows.map(async (row) => await this.load(row)));
+		return sources.filter((source): source is TrustedSource => source !== undefined);
+	}
+
+	private toRuntime(row: TrustedSourceEntity, config: TrustedSourceConfigLatest): TrustedSource {
+		return {
+			id: row.id,
+			name: row.name,
+			type: row.type,
+			issuer: row.issuer,
+			managedBy: row.managedBy,
+			status: row.status,
+			lastError: row.lastError,
+			lastCheckedAt: row.lastCheckedAt?.toISOString() ?? null,
+			createdAt: row.createdAt.toISOString(),
+			updatedAt: row.updatedAt.toISOString(),
+			config,
+		};
+	}
+}

packages/cli/test/integration/inbound-auth-core/trusted-source.store.integration.test.ts

@@ -0,0 +1,352 @@
+import { Logger } from '@n8n/backend-common';
+import { testDb, testModules } from '@n8n/backend-test-utils';
+import { Container } from '@n8n/di';
+import { NotFoundError } from '@n8n/errors';
+import {
+	trustedSourceConfigSchemaFor,
+	type SurfaceId,
+	type TrustedSourceConfigInput,
+} from '@n8n/inbound-auth';
+import { DataSource, type Repository } from '@n8n/typeorm';
+import { Cipher } from 'n8n-core';
+
+import { TrustedSourceIdentityEntity } from '@/modules/inbound-auth-core/database/entities/trusted-source-identity.entity';
+import { TrustedSourceEntity } from '@/modules/inbound-auth-core/database/entities/trusted-source.entity';
+import {
+	SystemTrustedSourceModificationError,
+	TrustedSourceStore,
+	type TrustedSource,
+} from '@/modules/inbound-auth-core/trusted-source.store';
+import { CacheService } from '@n8n/backend-services';
+
+import { createOwner } from '../shared/db/users';
+
+let store: TrustedSourceStore;
+let cacheService: CacheService;
+let cipher: Cipher;
+/** Direct table access, so the tests do not depend on the repository under construction. */
+let rows: Repository<TrustedSourceEntity>;
+let bindings: Repository<TrustedSourceIdentityEntity>;
+
+const minimalConfig = (
+	surfaces: Partial<Record<SurfaceId, { audiences?: string[] }>> = { 'public-api': {} },
+): TrustedSourceConfigInput => ({
+	version: 1,
+	authentication: { type: 'oauth2' },
+	surfaces,
+});
+
+const ids = (sources: TrustedSource[]) => sources.map((source) => source.id).sort();
+
+let seeded = 0;
+
+async function seedRow(overrides: Partial<TrustedSourceEntity> = {}) {
+	seeded += 1;
+	const config = trustedSourceConfigSchemaFor(overrides.managedBy ?? 'admin').parse(
+		minimalConfig(),
+	);
+	return await rows.save(
+		rows.create({
+			name: `source-${seeded}`,
+			type: 'oauth2',
+			issuer: `https://issuer-${seeded}.example.com`,
+			managedBy: 'admin',
+			status: 'unchecked',
+			lastError: null,
+			lastCheckedAt: null,
+			configVersion: 1,
+			config: await cipher.encryptV2(config),
+			...overrides,
+		}),
+	);
+}
+
+async function encryptedAdminConfig(
+	surfaces: Partial<Record<SurfaceId, { audiences?: string[] }>>,
+) {
+	return await cipher.encryptV2(
+		trustedSourceConfigSchemaFor('admin').parse(minimalConfig(surfaces)),
+	);
+}
+
+beforeAll(async () => {
+	await testModules.loadModules(['inbound-auth-core']);
+	await testDb.init();
+
+	store = Container.get(TrustedSourceStore);
+	cacheService = Container.get(CacheService);
+	cipher = Container.get(Cipher);
+	rows = Container.get(DataSource).getRepository(TrustedSourceEntity);
+	bindings = Container.get(DataSource).getRepository(TrustedSourceIdentityEntity);
+	// `reset()` assumes an initialised backend.
+	await cacheService.init();
+});
+
+beforeEach(async () => {
+	await testDb.truncate(['TrustedSourceIdentityEntity', 'TrustedSourceEntity', 'User']);
+	await cacheService.reset();
+});
+
+afterAll(async () => {
+	await testDb.terminate();
+});
+
+describe('TrustedSourceStore (integration)', () => {
+	describe('create', () => {
+		it('persists an encrypted, defaulted admin source and returns the runtime shape', async () => {
+			// Prime the list, so the test fails if `create` stops invalidating it.
+			expect(await store.listBySurface('public-api')).toEqual([]);
+
+			const source = await store.create({
+				name: 'Acme',
+				issuer: 'https://acme.example.com',
+				config: minimalConfig(),
+			});
+
+			const row = await rows.findOneByOrFail({ id: source.id });
+			expect(row).toMatchObject({
+				name: 'Acme',
+				issuer: 'https://acme.example.com',
+				type: 'oauth2',
+				managedBy: 'admin',
+				status: 'unchecked',
+				configVersion: 1,
+			});
+			expect(row.config).not.toContain('oauth2');
+			expect(JSON.parse(await cipher.decryptV2(row.config))).toMatchObject({
+				version: 1,
+				authentication: { type: 'oauth2', clockSkewSeconds: 60 },
+				identity: { subject: 'binding' },
+			});
+
+			expect(source).toMatchObject({
+				id: row.id,
+				name: 'Acme',
+				managedBy: 'admin',
+				status: 'unchecked',
+				lastError: null,
+				lastCheckedAt: null,
+				createdAt: row.createdAt.toISOString(),
+				updatedAt: row.updatedAt.toISOString(),
+			});
+			expect(source.config.authentication.clockSkewSeconds).toBe(60);
+			expect(ids(await store.listBySurface('public-api'))).toEqual([source.id]);
+		});
+
+		it('refuses a system-only field on an admin source and inserts nothing', async () => {
+			await expect(
+				store.create({
+					name: 'Acme',
+					issuer: 'https://acme.example.com',
+					config: { ...minimalConfig(), identity: { subject: 'n8n-user-id' } },
+				}),
+			).rejects.toThrow(/only allowed on system-managed sources/);
+
+			expect(await rows.count()).toBe(0);
+		});
+
+		it('keeps the client secret inside the encrypted document', async () => {
+			const source = await store.create({
+				name: 'Acme',
+				issuer: 'https://acme.example.com',
+				config: {
+					...minimalConfig(),
+					authentication: {
+						type: 'oauth2',
+						client: { kind: 'registered', clientId: 'n8n', clientSecret: 's3cret' },
+					},
+				},
+			});
+
+			const row = await rows.findOneByOrFail({ id: source.id });
+			expect(row.config).not.toContain('s3cret');
+
+			const fetched = await store.getById(source.id);
+			expect(fetched?.config.authentication.client).toEqual({
+				kind: 'registered',
+				clientId: 'n8n',
+				clientSecret: 's3cret',
+			});
+		});
+	});
+
+	describe('reads', () => {
+		it('getById and getByIssuer read through the cache and return undefined on a miss', async () => {
+			const row = await seedRow({ name: 'Acme', issuer: 'https://acme.example.com' });
+
+			expect(await store.getById(row.id)).toMatchObject({
+				id: row.id,
+				name: 'Acme',
+				config: { version: 1 },
+			});
+			expect(await store.getByIssuer('https://acme.example.com')).toMatchObject({ id: row.id });
+			expect(await store.getById('missing')).toBeUndefined();
+			expect(await store.getByIssuer('https://nobody.example.com')).toBeUndefined();
+
+			// A change behind the store's back is invisible until the entry expires or is invalidated.
+			await rows.update({ id: row.id }, { name: 'Renamed' });
+			expect((await store.getById(row.id))?.name).toBe('Acme');
+			await cacheService.reset();
+			expect((await store.getById(row.id))?.name).toBe('Renamed');
+		});
+
+		it('does not cache an issuer miss', async () => {
+			expect(await store.getByIssuer('https://late.example.com')).toBeUndefined();
+
+			const row = await seedRow({ issuer: 'https://late.example.com' });
+
+			expect(await store.getByIssuer('https://late.example.com')).toMatchObject({ id: row.id });
+		});
+
+		it('listBySurface returns the sources that register the surface', async () => {
+			const publicApi = await seedRow({ config: await encryptedAdminConfig({ 'public-api': {} }) });
+			const mcp = await seedRow({ config: await encryptedAdminConfig({ 'instance-mcp': {} }) });
+			const both = await seedRow({
+				config: await encryptedAdminConfig({ 'public-api': {}, 'instance-mcp': {} }),
+			});
+
+			expect(ids(await store.listBySurface('public-api'))).toEqual([publicApi.id, both.id].sort());
+			expect(ids(await store.listBySurface('instance-mcp'))).toEqual([mcp.id, both.id].sort());
+			expect(await store.listBySurface('trigger')).toEqual([]);
+		});
+
+		it.each<[string, () => Promise<string>]>([
+			['ciphertext that does not decrypt', async () => 'garbage'],
+			['a document that is not JSON', async () => await cipher.encryptV2('not-json')],
+			['a document that fails the schema', async () => await cipher.encryptV2({ version: 1 })],
+			[
+				'an admin row that holds a system-only field',
+				async () =>
+					await cipher.encryptV2(
+						trustedSourceConfigSchemaFor('system').parse({
+							...minimalConfig(),
+							authentication: { type: 'oauth2', keys: { kind: 'local-keystore' } },
+						}),
+					),
+			],
+		])('skips and logs a row with %s without writing to it', async (_, ciphertext) => {
+			const warn = vi.spyOn(Container.get(Logger), 'warn');
+			const bad = await seedRow({ config: await ciphertext() });
+			const good = await seedRow();
+
+			expect(await store.getById(bad.id)).toBeUndefined();
+			expect(await store.getByIssuer(bad.issuer)).toBeUndefined();
+			expect(ids(await store.listBySurface('public-api'))).toEqual([good.id]);
+
+			// One warning per read path; the good row never warns.
+			expect(warn).toHaveBeenCalledTimes(3);
+			expect(warn).toHaveBeenCalledWith(
+				expect.any(String),
+				expect.objectContaining({ id: bad.id }),
+			);
+			expect(await rows.findOneByOrFail({ id: bad.id })).toMatchObject({
+				config: bad.config,
+				status: 'unchecked',
+				lastError: null,
+			});
+		});
+	});
+
+	describe('update', () => {
+		it('rewrites the row and invalidates id, old issuer, new issuer and the list', async () => {
+			const row = await seedRow({ name: 'Old', issuer: 'https://old.example.com' });
+			// Prime every entry the write must invalidate.
+			await store.getById(row.id);
+			await store.getByIssuer('https://old.example.com');
+			expect(await store.listBySurface('instance-mcp')).toEqual([]);
+
+			await store.update(row.id, {
+				name: 'New',
+				issuer: 'https://new.example.com',
+				config: minimalConfig({ 'instance-mcp': {} }),
+			});
+
+			const updated = await rows.findOneByOrFail({ id: row.id });
+			expect(updated).toMatchObject({
+				name: 'New',
+				issuer: 'https://new.example.com',
+				configVersion: 1,
+			});
+			expect(updated.config).not.toBe(row.config);
+
+			expect(await store.getByIssuer('https://old.example.com')).toBeUndefined();
+			expect(await store.getByIssuer('https://new.example.com')).toMatchObject({ name: 'New' });
+			expect((await store.getById(row.id))?.name).toBe('New');
+			expect(ids(await store.listBySurface('instance-mcp'))).toEqual([row.id]);
+		});
+
+		it('clears the bindings of the updated source only when asked', async () => {
+			const user = await createOwner();
+			const a = await seedRow();
+			const b = await seedRow();
+			const binding = (sourceId: string, subject: string) =>
+				bindings.create({
+					sourceId,
+					subject,
+					userId: user.id,
+					provenance: 'admin',
+					status: 'active',
+					lastSeenAt: null,
+				});
+			await bindings.save([binding(a.id, 'alice'), binding(a.id, 'bob'), binding(b.id, 'carol')]);
+
+			await store.update(a.id, { name: 'Still bound' });
+			expect(await bindings.countBy({ sourceId: a.id })).toBe(2);
+
+			await store.update(a.id, { name: 'Re-keyed' }, { clearBindings: true });
+			expect(await bindings.countBy({ sourceId: a.id })).toBe(0);
+			expect(await bindings.countBy({ sourceId: b.id })).toBe(1);
+			expect((await rows.findOneByOrFail({ id: a.id })).name).toBe('Re-keyed');
+		});
+
+		it('revalidates the config against the row schema', async () => {
+			const row = await seedRow();
+
+			await expect(
+				store.update(row.id, {
+					config: { ...minimalConfig(), identity: { subject: 'n8n-user-id' } },
+				}),
+			).rejects.toThrow(/only allowed on system-managed sources/);
+
+			expect((await rows.findOneByOrFail({ id: row.id })).config).toBe(row.config);
+		});
+	});
+
+	describe('delete', () => {
+		it('removes the row and every cached view of it', async () => {
+			const row = await seedRow();
+			await store.getById(row.id);
+			await store.listBySurface('public-api');
+
+			await store.delete(row.id);
+
+			expect(await rows.findOneBy({ id: row.id })).toBeNull();
+			expect(await store.getById(row.id)).toBeUndefined();
+			expect(await store.listBySurface('public-api')).toEqual([]);
+		});
+	});
+
+	describe('guards', () => {
+		it.each(['update', 'delete'] as const)('%s refuses a system-managed row', async (action) => {
+			const row = await seedRow({ managedBy: 'system' });
+
+			const call =
+				action === 'update' ? store.update(row.id, { name: 'Renamed' }) : store.delete(row.id);
+			await expect(call).rejects.toThrow(SystemTrustedSourceModificationError);
+
+			expect(await rows.findOneByOrFail({ id: row.id })).toMatchObject({ name: row.name });
+		});
+
+		it.each(['update', 'delete'] as const)(
+			'%s throws NotFoundError for an unknown id',
+			async (action) => {
+				const call =
+					action === 'update'
+						? store.update('missing', { name: 'Renamed' })
+						: store.delete('missing');
+
+				await expect(call).rejects.toThrow(NotFoundError);
+			},
+		);
+	});
+});