feat(backend): add DB persistence layer (#79)

* feat(backend): add DB persistence layer Replace raw pg Pool with Drizzle ORM backed by Bun.sql. Add per-user source configuration table (user_sources). Migrate Better Auth to drizzle-adapter. Add AES-256-GCM credential encryption. Co-authored-by: Ona <no-reply@ona.com> * fix(backend): set updatedAt explicitly in all mutations onConflictDoUpdate bypasses Drizzle's $onUpdate hook. Set updatedAt explicitly in all mutation methods. Co-authored-by: Ona <no-reply@ona.com> * fix(backend): add composite index on user_sources Add (user_id, enabled) index for the enabled() query path. Co-authored-by: Ona <no-reply@ona.com> --------- Co-authored-by: Ona <no-reply@ona.com>
2026-03-20 17:11:17 +00:00 · 2026-03-16 01:30:02 +00:00
parent 9ac88d921c
commit 61c1ade631
29 changed files with 2034 additions and 115 deletions
--- a/apps/aelis-backend/src/auth/http.ts
+++ b/apps/aelis-backend/src/auth/http.ts
@@ -1,7 +1,7 @@
 import type { Hono } from "hono"

-import { auth } from "./index.ts"
+import type { Auth } from "./index.ts"

-export function registerAuthHandlers(app: Hono): void {
+export function registerAuthHandlers(app: Hono, auth: Auth): void {
 	app.on(["POST", "GET"], "/api/auth/*", (c) => auth.handler(c.req.raw))
 }
--- a/apps/aelis-backend/src/auth/index.ts
+++ b/apps/aelis-backend/src/auth/index.ts
@@ -1,10 +1,19 @@
 import { betterAuth } from "better-auth"
+import { drizzleAdapter } from "better-auth/adapters/drizzle"

-import { pool } from "../db.ts"
+import type { Database } from "../db/index.ts"
+import * as schema from "../db/schema.ts"

-export const auth = betterAuth({
-	database: pool,
-	emailAndPassword: {
-		enabled: true,
-	},
-})
+export function createAuth(db: Database) {
+	return betterAuth({
+		database: drizzleAdapter(db, {
+			provider: "pg",
+			schema,
+		}),
+		emailAndPassword: {
+			enabled: true,
+		},
+	})
+}
+
+export type Auth = ReturnType<typeof createAuth>
--- a/apps/aelis-backend/src/auth/session-middleware.ts
+++ b/apps/aelis-backend/src/auth/session-middleware.ts
@@ -1,9 +1,8 @@
 import type { Context, MiddlewareHandler, Next } from "hono"

+import type { Auth } from "./index.ts"
 import type { AuthSession, AuthUser } from "./session.ts"

-import { auth } from "./index.ts"
-
 export interface SessionVariables {
 	user: AuthUser | null
 	session: AuthSession | null
@@ -18,46 +17,52 @@ declare module "hono" {
 }

 /**
- * Middleware that attaches session and user to the context.
- * Does not reject unauthenticated requests - use requireSession for that.
+ * Creates a middleware that attaches session and user to the context.
+ * Does not reject unauthenticated requests - use createRequireSession for that.
 */
-export async function sessionMiddleware(c: Context, next: Next): Promise<void> {
-	const session = await auth.api.getSession({ headers: c.req.raw.headers })
+export function createSessionMiddleware(auth: Auth): AuthSessionMiddleware {
+	return async (c: Context, next: Next): Promise<void> => {
+		const session = await auth.api.getSession({ headers: c.req.raw.headers })
+
+		if (session) {
+			c.set("user", session.user)
+			c.set("session", session.session)
+		} else {
+			c.set("user", null)
+			c.set("session", null)
+		}
+
+		await next()
+	}
+}
+
+/**
+ * Creates a middleware that requires a valid session. Returns 401 if not authenticated.
+ */
+export function createRequireSession(auth: Auth): AuthSessionMiddleware {
+	return async (c: Context, next: Next): Promise<Response | void> => {
+		const session = await auth.api.getSession({ headers: c.req.raw.headers })
+
+		if (!session) {
+			return c.json({ error: "Unauthorized" }, 401)
+		}

-	if (session) {
 		c.set("user", session.user)
 		c.set("session", session.session)
-	} else {
-		c.set("user", null)
-		c.set("session", null)
+		await next()
 	}
-
-	await next()
 }

 /**
- * Middleware that requires a valid session. Returns 401 if not authenticated.
+ * Creates a function to get session from headers. Useful for WebSocket upgrade validation.
 */
-export async function requireSession(c: Context, next: Next): Promise<Response | void> {
-	const session = await auth.api.getSession({ headers: c.req.raw.headers })
-
-	if (!session) {
-		return c.json({ error: "Unauthorized" }, 401)
+export function createGetSessionFromHeaders(auth: Auth) {
+	return async (
+		headers: Headers,
+	): Promise<{ user: AuthUser; session: AuthSession } | null> => {
+		const session = await auth.api.getSession({ headers })
+		return session
 	}
-
-	c.set("user", session.user)
-	c.set("session", session.session)
-	await next()
-}
-
-/**
- * Get session from headers. Useful for WebSocket upgrade validation.
- */
-export async function getSessionFromHeaders(
-	headers: Headers,
-): Promise<{ user: AuthUser; session: AuthSession } | null> {
-	const session = await auth.api.getSession({ headers })
-	return session
 }

 /**
--- a/apps/aelis-backend/src/auth/session.ts
+++ b/apps/aelis-backend/src/auth/session.ts
@@ -1,4 +1,4 @@
-import type { auth } from "./index.ts"
+import type { Auth } from "./index.ts"

-export type AuthUser = typeof auth.$Infer.Session.user
-export type AuthSession = typeof auth.$Infer.Session.session
+export type AuthUser = Auth["$Infer"]["Session"]["user"]
+export type AuthSession = Auth["$Infer"]["Session"]["session"]
--- a/apps/aelis-backend/src/db.ts
+++ b/apps/aelis-backend/src/db.ts
@@ -1,5 +0,0 @@
-import { Pool } from "pg"
-
-export const pool = new Pool({
-	connectionString: process.env.DATABASE_URL,
-})
--- a/apps/aelis-backend/src/db/auth-schema.ts
+++ b/apps/aelis-backend/src/db/auth-schema.ts
@@ -0,0 +1,91 @@
+import { relations } from "drizzle-orm";
+import { pgTable, text, timestamp, boolean, index } from "drizzle-orm/pg-core";
+
+export const user = pgTable("user", {
+  id: text("id").primaryKey(),
+  name: text("name").notNull(),
+  email: text("email").notNull().unique(),
+  emailVerified: boolean("email_verified").default(false).notNull(),
+  image: text("image"),
+  createdAt: timestamp("created_at").notNull(),
+  updatedAt: timestamp("updated_at")
+    .$onUpdate(() => new Date())
+    .notNull(),
+});
+
+export const session = pgTable(
+  "session",
+  {
+    id: text("id").primaryKey(),
+    expiresAt: timestamp("expires_at").notNull(),
+    token: text("token").notNull().unique(),
+    createdAt: timestamp("created_at").notNull(),
+    updatedAt: timestamp("updated_at")
+      .$onUpdate(() => new Date())
+      .notNull(),
+    ipAddress: text("ip_address"),
+    userAgent: text("user_agent"),
+    userId: text("user_id")
+      .notNull()
+      .references(() => user.id, { onDelete: "cascade" }),
+  },
+  (table) => [index("session_userId_idx").on(table.userId)],
+);
+
+export const account = pgTable(
+  "account",
+  {
+    id: text("id").primaryKey(),
+    accountId: text("account_id").notNull(),
+    providerId: text("provider_id").notNull(),
+    userId: text("user_id")
+      .notNull()
+      .references(() => user.id, { onDelete: "cascade" }),
+    accessToken: text("access_token"),
+    refreshToken: text("refresh_token"),
+    idToken: text("id_token"),
+    accessTokenExpiresAt: timestamp("access_token_expires_at"),
+    refreshTokenExpiresAt: timestamp("refresh_token_expires_at"),
+    scope: text("scope"),
+    password: text("password"),
+    createdAt: timestamp("created_at").notNull(),
+    updatedAt: timestamp("updated_at")
+      .$onUpdate(() => new Date())
+      .notNull(),
+  },
+  (table) => [index("account_userId_idx").on(table.userId)],
+);
+
+export const verification = pgTable(
+  "verification",
+  {
+    id: text("id").primaryKey(),
+    identifier: text("identifier").notNull(),
+    value: text("value").notNull(),
+    expiresAt: timestamp("expires_at").notNull(),
+    createdAt: timestamp("created_at").notNull(),
+    updatedAt: timestamp("updated_at")
+      .$onUpdate(() => new Date())
+      .notNull(),
+  },
+  (table) => [index("verification_identifier_idx").on(table.identifier)],
+);
+
+export const userRelations = relations(user, ({ many }) => ({
+  sessions: many(session),
+  accounts: many(account),
+}));
+
+export const sessionRelations = relations(session, ({ one }) => ({
+  user: one(user, {
+    fields: [session.userId],
+    references: [user.id],
+  }),
+}));
+
+export const accountRelations = relations(account, ({ one }) => ({
+  user: one(user, {
+    fields: [account.userId],
+    references: [user.id],
+  }),
+}));
--- a/apps/aelis-backend/src/db/index.ts
+++ b/apps/aelis-backend/src/db/index.ts
@@ -0,0 +1,23 @@
+import { SQL } from "bun"
+import { drizzle, type BunSQLDatabase } from "drizzle-orm/bun-sql"
+
+import * as schema from "./schema.ts"
+
+export type Database = BunSQLDatabase<typeof schema>
+
+export interface DatabaseConnection {
+	db: Database
+	close: () => Promise<void>
+}
+
+export function createDatabase(url: string): DatabaseConnection {
+	if (!url) {
+		throw new Error("DATABASE_URL is required")
+	}
+
+	const client = new SQL({ url })
+	return {
+		db: drizzle({ client, schema }),
+		close: () => client.close(),
+	}
+}
--- a/apps/aelis-backend/src/db/schema.ts
+++ b/apps/aelis-backend/src/db/schema.ts
@@ -0,0 +1,62 @@
+import {
+	boolean,
+	customType,
+	index,
+	jsonb,
+	pgTable,
+	text,
+	timestamp,
+	unique,
+	uuid,
+} from "drizzle-orm/pg-core"
+
+// ---------------------------------------------------------------------------
+// Better Auth core tables
+// Re-exported from CLI-generated schema.
+// Regenerate with: bunx --bun auth@latest generate --config auth.ts --output src/db/auth-schema.ts
+// ---------------------------------------------------------------------------
+
+export {
+	user,
+	session,
+	account,
+	verification,
+	userRelations,
+	sessionRelations,
+	accountRelations,
+} from "./auth-schema.ts"
+
+import { user } from "./auth-schema.ts"
+
+// ---------------------------------------------------------------------------
+// AELIS — per-user source configuration
+// ---------------------------------------------------------------------------
+
+const bytea = customType<{ data: Buffer }>({
+	dataType() {
+		return "bytea"
+	},
+})
+
+export const userSources = pgTable(
+	"user_sources",
+	{
+		id: uuid("id").primaryKey().defaultRandom(),
+		userId: text("user_id")
+			.notNull()
+			.references(() => user.id, { onDelete: "cascade" }),
+		sourceId: text("source_id").notNull(),
+		enabled: boolean("enabled").notNull().default(true),
+		config: jsonb("config").default({}),
+		credentials: bytea("credentials"),
+		createdAt: timestamp("created_at").notNull().defaultNow(),
+		updatedAt: timestamp("updated_at")
+			.notNull()
+			.defaultNow()
+			.$onUpdate(() => new Date()),
+	},
+	(t) => [
+		unique("user_sources_user_id_source_id_unique").on(t.userId, t.sourceId),
+		index("user_sources_user_id_enabled_idx").on(t.userId, t.enabled),
+	],
+)
--- a/apps/aelis-backend/src/lib/crypto.test.ts
+++ b/apps/aelis-backend/src/lib/crypto.test.ts
@@ -0,0 +1,62 @@
+import { randomBytes } from "node:crypto"
+import { describe, expect, test } from "bun:test"
+
+import { CredentialEncryptor } from "./crypto.ts"
+
+const TEST_KEY = randomBytes(32).toString("base64")
+
+describe("CredentialEncryptor", () => {
+	const encryptor = new CredentialEncryptor(TEST_KEY)
+
+	test("round-trip with simple string", () => {
+		const plaintext = "hello world"
+		const encrypted = encryptor.encrypt(plaintext)
+		expect(encryptor.decrypt(encrypted)).toBe(plaintext)
+	})
+
+	test("round-trip with JSON credentials", () => {
+		const credentials = JSON.stringify({
+			accessToken: "ya29.a0AfH6SMB...",
+			refreshToken: "1//0dx...",
+			expiresAt: "2025-12-01T00:00:00Z",
+		})
+		const encrypted = encryptor.encrypt(credentials)
+		expect(encryptor.decrypt(encrypted)).toBe(credentials)
+	})
+
+	test("round-trip with empty string", () => {
+		const encrypted = encryptor.encrypt("")
+		expect(encryptor.decrypt(encrypted)).toBe("")
+	})
+
+	test("round-trip with unicode", () => {
+		const plaintext = "日本語テスト 🔐"
+		const encrypted = encryptor.encrypt(plaintext)
+		expect(encryptor.decrypt(encrypted)).toBe(plaintext)
+	})
+
+	test("each encryption produces different ciphertext (unique IV)", () => {
+		const plaintext = "same input"
+		const a = encryptor.encrypt(plaintext)
+		const b = encryptor.encrypt(plaintext)
+		expect(a).not.toEqual(b)
+		expect(encryptor.decrypt(a)).toBe(plaintext)
+		expect(encryptor.decrypt(b)).toBe(plaintext)
+	})
+
+	test("tampered ciphertext throws", () => {
+		const encrypted = encryptor.encrypt("secret")
+		encrypted[13]! ^= 0xff
+		expect(() => encryptor.decrypt(encrypted)).toThrow()
+	})
+
+	test("truncated data throws", () => {
+		expect(() => encryptor.decrypt(Buffer.alloc(10))).toThrow("Encrypted data is too short")
+	})
+
+	test("throws when key is wrong length", () => {
+		expect(() => new CredentialEncryptor(Buffer.from("too-short").toString("base64"))).toThrow(
+			"must be 32 bytes",
+		)
+	})
+})
--- a/apps/aelis-backend/src/lib/crypto.ts
+++ b/apps/aelis-backend/src/lib/crypto.ts
@@ -0,0 +1,60 @@
+import { createCipheriv, createDecipheriv, randomBytes } from "node:crypto"
+
+const ALGORITHM = "aes-256-gcm"
+const IV_LENGTH = 12
+const AUTH_TAG_LENGTH = 16
+
+/**
+ * AES-256-GCM encryption for credential storage.
+ *
+ * Caches the parsed key on construction to avoid repeated
+ * env reads and Buffer allocations.
+ */
+export class CredentialEncryptor {
+	private readonly key: Buffer
+
+	constructor(base64Key: string) {
+		const key = Buffer.from(base64Key, "base64")
+		if (key.length !== 32) {
+			throw new Error(
+				`Encryption key must be 32 bytes (got ${key.length}). Generate with: openssl rand -base64 32`,
+			)
+		}
+		this.key = key
+	}
+
+	/**
+	 * Encrypts plaintext using AES-256-GCM.
+	 *
+	 * Output format: [12-byte IV][ciphertext][16-byte auth tag]
+	 */
+	encrypt(plaintext: string): Buffer {
+		const iv = randomBytes(IV_LENGTH)
+		const cipher = createCipheriv(ALGORITHM, this.key, iv)
+
+		const encrypted = Buffer.concat([cipher.update(plaintext, "utf8"), cipher.final()])
+		const authTag = cipher.getAuthTag()
+
+		return Buffer.concat([iv, encrypted, authTag])
+	}
+
+	/**
+	 * Decrypts a buffer produced by `encrypt`.
+	 *
+	 * Expects format: [12-byte IV][ciphertext][16-byte auth tag]
+	 */
+	decrypt(data: Buffer): string {
+		if (data.length < IV_LENGTH + AUTH_TAG_LENGTH) {
+			throw new Error("Encrypted data is too short")
+		}
+
+		const iv = data.subarray(0, IV_LENGTH)
+		const authTag = data.subarray(data.length - AUTH_TAG_LENGTH)
+		const ciphertext = data.subarray(IV_LENGTH, data.length - AUTH_TAG_LENGTH)
+
+		const decipher = createDecipheriv(ALGORITHM, this.key, iv)
+		decipher.setAuthTag(authTag)
+
+		return Buffer.concat([decipher.update(ciphertext), decipher.final()]).toString("utf8")
+	}
+}
--- a/apps/aelis-backend/src/location/http.ts
+++ b/apps/aelis-backend/src/location/http.ts
@@ -3,10 +3,9 @@ import type { Context, Hono } from "hono"
 import { type } from "arktype"
 import { createMiddleware } from "hono/factory"

+import type { AuthSessionMiddleware } from "../auth/session-middleware.ts"
 import type { UserSessionManager } from "../session/index.ts"

-import { requireSession } from "../auth/session-middleware.ts"
-
 type Env = { Variables: { sessionManager: UserSessionManager } }

 const locationInput = type({
@@ -16,16 +15,21 @@ const locationInput = type({
 	timestamp: "string.date.iso",
 })

+interface LocationHttpHandlersDeps {
+	sessionManager: UserSessionManager
+	authSessionMiddleware: AuthSessionMiddleware
+}
+
 export function registerLocationHttpHandlers(
 	app: Hono,
-	{ sessionManager }: { sessionManager: UserSessionManager },
+	{ sessionManager, authSessionMiddleware }: LocationHttpHandlersDeps,
 ) {
 	const inject = createMiddleware<Env>(async (c, next) => {
 		c.set("sessionManager", sessionManager)
 		await next()
 	})

-	app.post("/api/location", inject, requireSession, handleUpdateLocation)
+	app.post("/api/location", inject, authSessionMiddleware, handleUpdateLocation)
 }

 async function handleUpdateLocation(c: Context<Env>) {
--- a/apps/aelis-backend/src/location/provider.ts
+++ b/apps/aelis-backend/src/location/provider.ts
@@ -0,0 +1,25 @@
+import { LocationSource } from "@aelis/source-location"
+
+import type { Database } from "../db/index.ts"
+import type { FeedSourceProvider } from "../session/feed-source-provider.ts"
+
+import { SourceDisabledError } from "../sources/errors.ts"
+import { sources } from "../sources/user-sources.ts"
+
+export class LocationSourceProvider implements FeedSourceProvider {
+	private readonly db: Database
+
+	constructor(db: Database) {
+		this.db = db
+	}
+
+	async feedSourceForUser(userId: string): Promise<LocationSource> {
+		const row = await sources(this.db, userId).find("aelis.location")
+
+		if (!row || !row.enabled) {
+			throw new SourceDisabledError("aelis.location", userId)
+		}
+
+		return new LocationSource()
+	}
+}
--- a/apps/aelis-backend/src/server.ts
+++ b/apps/aelis-backend/src/server.ts
@@ -1,16 +1,21 @@
-import { LocationSource } from "@aelis/source-location"
 import { Hono } from "hono"

 import { registerAuthHandlers } from "./auth/http.ts"
-import { requireSession } from "./auth/session-middleware.ts"
+import { createAuth } from "./auth/index.ts"
+import { createRequireSession } from "./auth/session-middleware.ts"
+import { createDatabase } from "./db/index.ts"
 import { registerFeedHttpHandlers } from "./engine/http.ts"
 import { createFeedEnhancer } from "./enhancement/enhance-feed.ts"
 import { createLlmClient } from "./enhancement/llm-client.ts"
 import { registerLocationHttpHandlers } from "./location/http.ts"
+import { LocationSourceProvider } from "./location/provider.ts"
 import { UserSessionManager } from "./session/index.ts"
 import { WeatherSourceProvider } from "./weather/provider.ts"

 function main() {
+	const { db, close: closeDb } = createDatabase(process.env.DATABASE_URL!)
+	const auth = createAuth(db)
+
 	const openrouterApiKey = process.env.OPENROUTER_API_KEY
 	const feedEnhancer = openrouterApiKey
 		? createFeedEnhancer({
@@ -26,8 +31,9 @@ function main() {

 	const sessionManager = new UserSessionManager({
 		providers: [
-			async () => new LocationSource(),
+			new LocationSourceProvider(db),
 			new WeatherSourceProvider({
+				db,
 				credentials: {
 					privateKey: process.env.WEATHERKIT_PRIVATE_KEY!,
 					keyId: process.env.WEATHERKIT_KEY_ID!,
@@ -43,13 +49,20 @@ function main() {

 	app.get("/health", (c) => c.json({ status: "ok" }))

-	registerAuthHandlers(app)
+	const authSessionMiddleware = createRequireSession(auth)
+
+	registerAuthHandlers(app, auth)

 	registerFeedHttpHandlers(app, {
 		sessionManager,
-		authSessionMiddleware: requireSession,
+		authSessionMiddleware,
+	})
+	registerLocationHttpHandlers(app, { sessionManager, authSessionMiddleware })
+
+	process.on("SIGTERM", async () => {
+		await closeDb()
+		process.exit(0)
 	})
-	registerLocationHttpHandlers(app, { sessionManager })

 	return app
 }
--- a/apps/aelis-backend/src/session/user-session-manager.test.ts
+++ b/apps/aelis-backend/src/session/user-session-manager.test.ts
@@ -1,14 +1,11 @@
-import type { WeatherKitClient, WeatherKitResponse } from "@aelis/source-weatherkit"
-
 import { LocationSource } from "@aelis/source-location"
+import { WeatherSource } from "@aelis/source-weatherkit"
 import { describe, expect, mock, spyOn, test } from "bun:test"

-import { WeatherSourceProvider } from "../weather/provider.ts"
 import { UserSessionManager } from "./user-session-manager.ts"

-const mockWeatherClient: WeatherKitClient = {
-	fetch: async () => ({}) as WeatherKitResponse,
-}
+const mockWeatherProvider = async () =>
+	new WeatherSource({ client: { fetch: async () => ({}) as never } })

 describe("UserSessionManager", () => {
 	test("getOrCreate creates session on first call", async () => {
@@ -76,9 +73,8 @@ describe("UserSessionManager", () => {
 	})

 	test("accepts object providers", async () => {
-		const provider = new WeatherSourceProvider({ client: mockWeatherClient })
 		const manager = new UserSessionManager({
-			providers: [async () => new LocationSource(), provider],
+			providers: [async () => new LocationSource(), mockWeatherProvider],
 		})

 		const session = await manager.getOrCreate("user-1")
@@ -87,9 +83,8 @@ describe("UserSessionManager", () => {
 	})

 	test("accepts mixed providers", async () => {
-		const provider = new WeatherSourceProvider({ client: mockWeatherClient })
 		const manager = new UserSessionManager({
-			providers: [async () => new LocationSource(), provider],
+			providers: [async () => new LocationSource(), mockWeatherProvider],
 		})

 		const session = await manager.getOrCreate("user-1")
--- a/apps/aelis-backend/src/sources/errors.ts
+++ b/apps/aelis-backend/src/sources/errors.ts
@@ -0,0 +1,32 @@
+/**
+ * Thrown by a FeedSourceProvider when the source is not enabled for a user.
+ *
+ * UserSessionManager's Promise.allSettled handles this gracefully —
+ * the source is excluded from the session without crashing.
+ */
+export class SourceDisabledError extends Error {
+	readonly sourceId: string
+	readonly userId: string
+
+	constructor(sourceId: string, userId: string) {
+		super(`Source "${sourceId}" is not enabled for user "${userId}"`)
+		this.name = "SourceDisabledError"
+		this.sourceId = sourceId
+		this.userId = userId
+	}
+}
+
+/**
+ * Thrown when an operation targets a user source that doesn't exist.
+ */
+export class SourceNotFoundError extends Error {
+	readonly sourceId: string
+	readonly userId: string
+
+	constructor(sourceId: string, userId: string) {
+		super(`Source "${sourceId}" not found for user "${userId}"`)
+		this.name = "SourceNotFoundError"
+		this.sourceId = sourceId
+		this.userId = userId
+	}
+}
--- a/apps/aelis-backend/src/sources/user-sources.ts
+++ b/apps/aelis-backend/src/sources/user-sources.ts
@@ -0,0 +1,79 @@
+import { and, eq } from "drizzle-orm"
+
+import type { Database } from "../db/index.ts"
+
+import { userSources } from "../db/schema.ts"
+import { SourceNotFoundError } from "./errors.ts"
+
+export function sources(db: Database, userId: string) {
+	return {
+		/** Returns all enabled sources for the user. */
+		async enabled() {
+			return db
+				.select()
+				.from(userSources)
+				.where(and(eq(userSources.userId, userId), eq(userSources.enabled, true)))
+		},
+
+		/** Returns a specific source by ID, or undefined. */
+		async find(sourceId: string) {
+			const rows = await db
+				.select()
+				.from(userSources)
+				.where(and(eq(userSources.userId, userId), eq(userSources.sourceId, sourceId)))
+				.limit(1)
+
+			return rows[0]
+		},
+
+		/** Enables a source for the user. Throws if the source row doesn't exist. */
+		async enableSource(sourceId: string) {
+			const rows = await db
+				.update(userSources)
+				.set({ enabled: true, updatedAt: new Date() })
+				.where(and(eq(userSources.userId, userId), eq(userSources.sourceId, sourceId)))
+				.returning({ id: userSources.id })
+
+			if (rows.length === 0) {
+				throw new SourceNotFoundError(sourceId, userId)
+			}
+		},
+
+		/** Disables a source for the user. Throws if the source row doesn't exist. */
+		async disableSource(sourceId: string) {
+			const rows = await db
+				.update(userSources)
+				.set({ enabled: false, updatedAt: new Date() })
+				.where(and(eq(userSources.userId, userId), eq(userSources.sourceId, sourceId)))
+				.returning({ id: userSources.id })
+
+			if (rows.length === 0) {
+				throw new SourceNotFoundError(sourceId, userId)
+			}
+		},
+
+		/** Creates or updates the config for a source. */
+		async upsertConfig(sourceId: string, config: Record<string, unknown>) {
+			await db
+				.insert(userSources)
+				.values({ userId, sourceId, config })
+				.onConflictDoUpdate({
+					target: [userSources.userId, userSources.sourceId],
+					set: { config, updatedAt: new Date() },
+				})
+		},
+
+		/** Updates the encrypted credentials for a source. Throws if the source row doesn't exist. */
+		async updateCredentials(sourceId: string, credentials: Buffer) {
+			const rows = await db
+				.update(userSources)
+				.set({ credentials, updatedAt: new Date() })
+				.where(and(eq(userSources.userId, userId), eq(userSources.sourceId, sourceId)))
+				.returning({ id: userSources.id })
+
+			if (rows.length === 0) {
+				throw new SourceNotFoundError(sourceId, userId)
+			}
+		},
+	}
+}
--- a/apps/aelis-backend/src/tfl/provider.ts
+++ b/apps/aelis-backend/src/tfl/provider.ts
@@ -1,19 +1,47 @@
-import { TflSource, type ITflApi } from "@aelis/source-tfl"
+import { TflSource, type ITflApi, type TflLineId } from "@aelis/source-tfl"
+import { type } from "arktype"

+import type { Database } from "../db/index.ts"
 import type { FeedSourceProvider } from "../session/feed-source-provider.ts"

+import { SourceDisabledError } from "../sources/errors.ts"
+import { sources } from "../sources/user-sources.ts"
+
 export type TflSourceProviderOptions =
-	| { apiKey: string; client?: never }
-	| { apiKey?: never; client: ITflApi }
+	| { db: Database; apiKey: string; client?: never }
+	| { db: Database; apiKey?: never; client: ITflApi }
+
+const tflConfig = type({
+	"lines?": "string[]",
+})

 export class TflSourceProvider implements FeedSourceProvider {
-	private readonly options: TflSourceProviderOptions
+	private readonly db: Database
+	private readonly apiKey: string | undefined
+	private readonly client: ITflApi | undefined

 	constructor(options: TflSourceProviderOptions) {
-		this.options = options
+		this.db = options.db
+		this.apiKey = "apiKey" in options ? options.apiKey : undefined
+		this.client = "client" in options ? options.client : undefined
 	}

-	async feedSourceForUser(_userId: string): Promise<TflSource> {
-		return new TflSource(this.options)
+	async feedSourceForUser(userId: string): Promise<TflSource> {
+		const row = await sources(this.db, userId).find("aelis.tfl")
+
+		if (!row || !row.enabled) {
+			throw new SourceDisabledError("aelis.tfl", userId)
+		}
+
+		const parsed = tflConfig(row.config ?? {})
+		if (parsed instanceof type.errors) {
+			throw new Error(`Invalid TFL config for user ${userId}: ${parsed.summary}`)
+		}
+
+		return new TflSource({
+			apiKey: this.apiKey,
+			client: this.client,
+			lines: parsed.lines as TflLineId[] | undefined,
+		})
 	}
 }
--- a/apps/aelis-backend/src/weather/provider.ts
+++ b/apps/aelis-backend/src/weather/provider.ts
@@ -1,15 +1,53 @@
 import { WeatherSource, type WeatherSourceOptions } from "@aelis/source-weatherkit"
+import { type } from "arktype"

+import type { Database } from "../db/index.ts"
 import type { FeedSourceProvider } from "../session/feed-source-provider.ts"

-export class WeatherSourceProvider implements FeedSourceProvider {
-	private readonly options: WeatherSourceOptions
+import { SourceDisabledError } from "../sources/errors.ts"
+import { sources } from "../sources/user-sources.ts"

-	constructor(options: WeatherSourceOptions) {
-		this.options = options
+export interface WeatherSourceProviderOptions {
+	db: Database
+	credentials: WeatherSourceOptions["credentials"]
+	client?: WeatherSourceOptions["client"]
+}
+
+const weatherConfig = type({
+	"units?": "'metric' | 'imperial'",
+	"hourlyLimit?": "number",
+	"dailyLimit?": "number",
+})
+
+export class WeatherSourceProvider implements FeedSourceProvider {
+	private readonly db: Database
+	private readonly credentials: WeatherSourceOptions["credentials"]
+	private readonly client: WeatherSourceOptions["client"]
+
+	constructor(options: WeatherSourceProviderOptions) {
+		this.db = options.db
+		this.credentials = options.credentials
+		this.client = options.client
 	}

-	async feedSourceForUser(_userId: string): Promise<WeatherSource> {
-		return new WeatherSource(this.options)
+	async feedSourceForUser(userId: string): Promise<WeatherSource> {
+		const row = await sources(this.db, userId).find("aelis.weather")
+
+		if (!row || !row.enabled) {
+			throw new SourceDisabledError("aelis.weather", userId)
+		}
+
+		const parsed = weatherConfig(row.config ?? {})
+		if (parsed instanceof type.errors) {
+			throw new Error(`Invalid weather config for user ${userId}: ${parsed.summary}`)
+		}
+
+		return new WeatherSource({
+			credentials: this.credentials,
+			client: this.client,
+			units: parsed.units,
+			hourlyLimit: parsed.hourlyLimit,
+			dailyLimit: parsed.dailyLimit,
+		})
 	}
 }