feat: surface per-user credentials to feed source providers (#110)

Add credentials parameter to FeedSourceProvider.feedSourceForUser so
providers can receive decrypted per-user credentials (OAuth tokens,
passwords) from the user_sources table.

Wire CredentialEncryptor into UserSessionManager to handle
encrypt/decrypt. Providers receive plaintext and handle validation
internally. Existing providers ignore the new parameter.

Co-authored-by: Ona <no-reply@ona.com>

apps/aelis-backend/src/session/feed-source-provider.ts

@@ -8,5 +8,5 @@ export interface FeedSourceProvider {
 	readonly sourceId: string
 	/** Arktype schema for validating user-provided config. Omit if the source has no config. */
 	readonly configSchema?: ConfigSchema
-	feedSourceForUser(userId: string, config: unknown): Promise<FeedSource>
+	feedSourceForUser(userId: string, config: unknown, credentials: unknown): Promise<FeedSource>
 }

apps/aelis-backend/src/session/user-session-manager.test.ts

@@ -7,6 +7,12 @@ import { beforeEach, describe, expect, mock, spyOn, test } from "bun:test"
 import type { Database } from "../db/index.ts"
 import type { FeedSourceProvider } from "./feed-source-provider.ts"
 
+import { CredentialEncryptor } from "../lib/crypto.ts"
+import {
+	CredentialStorageUnavailableError,
+	InvalidSourceCredentialsError,
+} from "../sources/errors.ts"
+import { SourceNotFoundError } from "../sources/errors.ts"
 import { UserSessionManager } from "./user-session-manager.ts"
 
 /**
@@ -38,6 +44,13 @@ function getEnabledSourceIds(userId: string): string[] {
  */
 let mockFindResult: unknown | undefined
 
+/**
+ * Spy for `updateCredentials` calls. Tests can inspect calls via
+ * `mockUpdateCredentialsCalls` or override behavior.
+ */
+const mockUpdateCredentialsCalls: Array<{ sourceId: string; credentials: Buffer }> = []
+let mockUpdateCredentialsError: Error | null = null
+
 // Mock the sources module so UserSessionManager's DB query returns controlled data.
 mock.module("../sources/user-sources.ts", () => ({
 	sources: (_db: Database, userId: string) => ({
@@ -68,6 +81,12 @@ mock.module("../sources/user-sources.ts", () => ({
 				updatedAt: now,
 			}
 		},
+		async updateCredentials(sourceId: string, credentials: Buffer) {
+			if (mockUpdateCredentialsError) {
+				throw mockUpdateCredentialsError
+			}
+			mockUpdateCredentialsCalls.push({ sourceId, credentials })
+		},
 	}),
 }))
 
@@ -93,8 +112,11 @@ function createStubSource(id: string, items: FeedItem[] = []): FeedSource {
 
 function createStubProvider(
 	sourceId: string,
-	factory: (userId: string, config: Record<string, unknown>) => Promise<FeedSource> = async () =>
-		createStubSource(sourceId),
+	factory: (
+		userId: string,
+		config: Record<string, unknown>,
+		credentials: unknown,
+	) => Promise<FeedSource> = async () => createStubSource(sourceId),
 ): FeedSourceProvider {
 	return { sourceId, feedSourceForUser: factory }
 }
@@ -116,6 +138,8 @@ const weatherProvider: FeedSourceProvider = {
 beforeEach(() => {
 	enabledByUser.clear()
 	mockFindResult = undefined
+	mockUpdateCredentialsCalls.length = 0
+	mockUpdateCredentialsError = null
 })
 
 describe("UserSessionManager", () => {
@@ -681,3 +705,122 @@ describe("UserSessionManager.replaceProvider", () => {
 		expect(feedAfter.items[0]!.data.version).toBe(1)
 	})
 })
+
+const TEST_ENCRYPTION_KEY = "/bv1nbzC4ozZkT/pcv5oQfl+JAMuMZDUSVDesG2dur8="
+const testEncryptor = new CredentialEncryptor(TEST_ENCRYPTION_KEY)
+
+describe("UserSessionManager.updateSourceCredentials", () => {
+	test("encrypts and persists credentials", async () => {
+		setEnabledSources(["test"])
+		const provider = createStubProvider("test")
+		const manager = new UserSessionManager({
+			db: fakeDb,
+			providers: [provider],
+			credentialEncryptor: testEncryptor,
+		})
+
+		await manager.updateSourceCredentials("user-1", "test", { token: "secret-123" })
+
+		expect(mockUpdateCredentialsCalls).toHaveLength(1)
+		expect(mockUpdateCredentialsCalls[0]!.sourceId).toBe("test")
+
+		// Verify the persisted buffer decrypts to the original credentials
+		const decrypted = JSON.parse(testEncryptor.decrypt(mockUpdateCredentialsCalls[0]!.credentials))
+		expect(decrypted).toEqual({ token: "secret-123" })
+	})
+
+	test("throws CredentialStorageUnavailableError when encryptor is not configured", async () => {
+		setEnabledSources(["test"])
+		const provider = createStubProvider("test")
+		const manager = new UserSessionManager({
+			db: fakeDb,
+			providers: [provider],
+			// no credentialEncryptor
+		})
+
+		await expect(
+			manager.updateSourceCredentials("user-1", "test", { token: "x" }),
+		).rejects.toBeInstanceOf(CredentialStorageUnavailableError)
+	})
+
+	test("throws SourceNotFoundError for unknown source", async () => {
+		setEnabledSources([])
+		const manager = new UserSessionManager({
+			db: fakeDb,
+			providers: [],
+			credentialEncryptor: testEncryptor,
+		})
+
+		await expect(
+			manager.updateSourceCredentials("user-1", "unknown", { token: "x" }),
+		).rejects.toBeInstanceOf(SourceNotFoundError)
+	})
+
+	test("propagates InvalidSourceCredentialsError from provider", async () => {
+		setEnabledSources(["test"])
+		let callCount = 0
+		const provider: FeedSourceProvider = {
+			sourceId: "test",
+			async feedSourceForUser(_userId: string, _config: unknown, _credentials: unknown) {
+				callCount++
+				// Succeed on first call (session creation), throw on refresh
+				if (callCount > 1) {
+					throw new InvalidSourceCredentialsError("test", "bad credentials")
+				}
+				return createStubSource("test")
+			},
+		}
+		const manager = new UserSessionManager({
+			db: fakeDb,
+			providers: [provider],
+			credentialEncryptor: testEncryptor,
+		})
+
+		// Create a session first so the refresh path is exercised
+		await manager.getOrCreate("user-1")
+
+		await expect(
+			manager.updateSourceCredentials("user-1", "test", { token: "bad" }),
+		).rejects.toBeInstanceOf(InvalidSourceCredentialsError)
+
+		// Credentials should still have been persisted before the provider threw
+		expect(mockUpdateCredentialsCalls).toHaveLength(1)
+	})
+
+	test("refreshes source in active session after credential update", async () => {
+		setEnabledSources(["test"])
+		let receivedCredentials: unknown = null
+		const provider = createStubProvider("test", async (_userId, _config, credentials) => {
+			receivedCredentials = credentials
+			return createStubSource("test")
+		})
+		const manager = new UserSessionManager({
+			db: fakeDb,
+			providers: [provider],
+			credentialEncryptor: testEncryptor,
+		})
+
+		await manager.getOrCreate("user-1")
+		await manager.updateSourceCredentials("user-1", "test", { token: "refreshed" })
+
+		expect(receivedCredentials).toEqual({ token: "refreshed" })
+	})
+
+	test("persists credentials without session refresh when no active session", async () => {
+		setEnabledSources(["test"])
+		const factory = mock(async () => createStubSource("test"))
+		const provider: FeedSourceProvider = { sourceId: "test", feedSourceForUser: factory }
+		const manager = new UserSessionManager({
+			db: fakeDb,
+			providers: [provider],
+			credentialEncryptor: testEncryptor,
+		})
+
+		// No session created — just update credentials
+		await manager.updateSourceCredentials("user-1", "test", { token: "stored" })
+
+		expect(mockUpdateCredentialsCalls).toHaveLength(1)
+		// feedSourceForUser should not have been called (no session to refresh)
+		expect(factory).not.toHaveBeenCalled()
+	})
+})

apps/aelis-backend/src/session/user-session-manager.ts

@@ -5,9 +5,14 @@ import merge from "lodash.merge"
 
 import type { Database } from "../db/index.ts"
 import type { FeedEnhancer } from "../enhancement/enhance-feed.ts"
+import type { CredentialEncryptor } from "../lib/crypto.ts"
 import type { FeedSourceProvider } from "./feed-source-provider.ts"
 
-import { InvalidSourceConfigError, SourceNotFoundError } from "../sources/errors.ts"
+import {
+	CredentialStorageUnavailableError,
+	InvalidSourceConfigError,
+	SourceNotFoundError,
+} from "../sources/errors.ts"
 import { sources } from "../sources/user-sources.ts"
 import { UserSession } from "./user-session.ts"
 
@@ -15,6 +20,7 @@ export interface UserSessionManagerConfig {
 	db: Database
 	providers: FeedSourceProvider[]
 	feedEnhancer?: FeedEnhancer | null
+	credentialEncryptor?: CredentialEncryptor | null
 }
 
 export class UserSessionManager {
@@ -23,7 +29,7 @@ export class UserSessionManager {
 	private readonly db: Database
 	private readonly providers = new Map<string, FeedSourceProvider>()
 	private readonly feedEnhancer: FeedEnhancer | null
-	private readonly db: Database
+	private readonly encryptor: CredentialEncryptor | null
 
 	constructor(config: UserSessionManagerConfig) {
 		this.db = config.db
@@ -31,7 +37,7 @@ export class UserSessionManager {
 			this.providers.set(provider.sourceId, provider)
 		}
 		this.feedEnhancer = config.feedEnhancer ?? null
-		this.db = config.db
+		this.encryptor = config.credentialEncryptor ?? null
 	}
 
 	getProvider(sourceId: string): FeedSourceProvider | undefined {
@@ -120,14 +126,15 @@ export class UserSessionManager {
 			return
 		}
 
-		// When config is provided, fetch existing to deep-merge before validating.
+		// Fetch the existing row for config merging and credential access.
 		// NOTE: find + updateConfig is not atomic. A concurrent update could
 		// read stale config. Use SELECT FOR UPDATE or atomic jsonb merge if
 		// this becomes a problem.
+		const existingRow = await sources(this.db, userId).find(sourceId)
+
 		let mergedConfig: Record<string, unknown> | undefined
 		if (update.config !== undefined && provider.configSchema) {
-			const existing = await sources(this.db, userId).find(sourceId)
-			const existingConfig = (existing?.config ?? {}) as Record<string, unknown>
+			const existingConfig = (existingRow?.config ?? {}) as Record<string, unknown>
 			mergedConfig = merge({}, existingConfig, update.config)
 
 			const validated = provider.configSchema(mergedConfig)
@@ -149,7 +156,10 @@ export class UserSessionManager {
 			if (update.enabled === false) {
 				session.removeSource(sourceId)
 			} else {
-				const source = await provider.feedSourceForUser(userId, mergedConfig ?? {})
+				const credentials = existingRow?.credentials
+					? this.decryptCredentials(existingRow.credentials)
+					: null
+				const source = await provider.feedSourceForUser(userId, mergedConfig ?? {}, credentials)
 				session.replaceSource(sourceId, source)
 			}
 		}
@@ -182,6 +192,11 @@ export class UserSessionManager {
 		}
 
 		const config = data.config ?? {}
+
+		// Fetch existing row before upsert to capture credentials for session refresh.
+		// For new rows this will be undefined — credentials will be null.
+		const existingRow = await sources(this.db, userId).find(sourceId)
+
 		await sources(this.db, userId).upsertConfig(sourceId, {
 			enabled: data.enabled,
 			config,
@@ -192,7 +207,10 @@ export class UserSessionManager {
 			if (!data.enabled) {
 				session.removeSource(sourceId)
 			} else {
-				const source = await provider.feedSourceForUser(userId, config)
+				const credentials = existingRow?.credentials
+					? this.decryptCredentials(existingRow.credentials)
+					: null
+				const source = await provider.feedSourceForUser(userId, config, credentials)
 				if (session.hasSource(sourceId)) {
 					session.replaceSource(sourceId, source)
 				} else {
@@ -202,6 +220,44 @@ export class UserSessionManager {
 		}
 	}
 
+	/**
+	 * Validates, encrypts, and persists per-user credentials for a source,
+	 * then refreshes the active session.
+	 *
+	 * @throws {SourceNotFoundError} if the source row doesn't exist or has no registered provider
+	 * @throws {CredentialStorageUnavailableError} if no CredentialEncryptor is configured
+	 */
+	async updateSourceCredentials(
+		userId: string,
+		sourceId: string,
+		credentials: unknown,
+	): Promise<void> {
+		const provider = this.providers.get(sourceId)
+		if (!provider) {
+			throw new SourceNotFoundError(sourceId, userId)
+		}
+
+		if (!this.encryptor) {
+			throw new CredentialStorageUnavailableError()
+		}
+
+		const encrypted = this.encryptor.encrypt(JSON.stringify(credentials))
+		await sources(this.db, userId).updateCredentials(sourceId, encrypted)
+
+		// Refresh the source in the active session.
+		// If feedSourceForUser throws (e.g. provider rejects the credentials),
+		// the DB already has the new credentials but the session keeps the old
+		// source. The next session creation will pick up the persisted credentials.
+		const session = this.sessions.get(userId)
+		if (session && session.hasSource(sourceId)) {
+			const row = await sources(this.db, userId).find(sourceId)
+			if (row?.enabled) {
+				const source = await provider.feedSourceForUser(userId, row.config ?? {}, credentials)
+				session.replaceSource(sourceId, source)
+			}
+		}
+	}
+
 	/**
 	 * Replaces a provider and updates all active sessions.
 	 * The new provider must have the same sourceId as an existing one.
@@ -254,7 +310,12 @@ export class UserSessionManager {
 			const row = await sources(this.db, session.userId).find(provider.sourceId)
 			if (!row?.enabled) return
 
-			const newSource = await provider.feedSourceForUser(session.userId, row.config ?? {})
+			const credentials = row.credentials ? this.decryptCredentials(row.credentials) : null
+			const newSource = await provider.feedSourceForUser(
+				session.userId,
+				row.config ?? {},
+				credentials,
+			)
 			session.replaceSource(provider.sourceId, newSource)
 		} catch (err) {
 			console.error(
@@ -271,7 +332,8 @@ export class UserSessionManager {
 		for (const row of enabledRows) {
 			const provider = this.providers.get(row.sourceId)
 			if (provider) {
-				promises.push(provider.feedSourceForUser(userId, row.config ?? {}))
+				const credentials = row.credentials ? this.decryptCredentials(row.credentials) : null
+				promises.push(provider.feedSourceForUser(userId, row.config ?? {}, credentials))
 			}
 		}
 
@@ -302,4 +364,19 @@ export class UserSessionManager {
 
 		return new UserSession(userId, feedSources, this.feedEnhancer)
 	}
+
+	/**
+	 * Decrypts a credentials buffer from the DB, returning parsed JSON or null.
+	 * Returns null (with a warning) if decryption or parsing fails — e.g. due to
+	 * key rotation, data corruption, or malformed JSON.
+	 */
+	private decryptCredentials(credentials: Buffer): unknown {
+		if (!this.encryptor) return null
+		try {
+			return JSON.parse(this.encryptor.decrypt(credentials))
+		} catch (err) {
+			console.warn("[UserSessionManager] Failed to decrypt credentials:", err)
+			return null
+		}
+	}
 }