Merge pull request #22 from kennethnym/feat/weather-service

feat(backend): add WeatherService

.devcontainer/devcontainer.json

@@ -10,6 +10,7 @@
 		"context": ".",
 		"dockerfile": "Dockerfile"
 	},
+	"postCreateCommand": "bun install",
 	"postStartCommand": "./scripts/setup-git.sh && ./scripts/setup-nvim.sh"
 	// Features add additional features to your environment. See https://containers.dev/features
 	// Beware: features are not supported on all platforms and may have unintended side-effects.

README.md

@@ -8,14 +8,14 @@ bun install
 
 ## Packages
 
-### @aris/data-source-tfl
+### @aris/source-tfl
 
-TfL (Transport for London) data source for tube, overground, and Elizabeth line alerts.
+TfL (Transport for London) feed source for tube, overground, and Elizabeth line alerts.
 
 #### Testing
 
 ```bash
-cd packages/aris-data-source-tfl
+cd packages/aris-source-tfl
 bun run test
 ```
 

apps/aris-backend/.env.example

@@ -0,0 +1,14 @@
+# PostgreSQL connection string
+DATABASE_URL=postgresql://user:password@localhost:5432/aris
+
+# BetterAuth secret (min 32 chars, generate with: openssl rand -base64 32)
+BETTER_AUTH_SECRET=
+
+# Base URL of the backend
+BETTER_AUTH_URL=http://localhost:3000
+
+# Apple WeatherKit credentials
+WEATHERKIT_PRIVATE_KEY=
+WEATHERKIT_KEY_ID=
+WEATHERKIT_TEAM_ID=
+WEATHERKIT_SERVICE_ID=

apps/aris-backend/package.json

@@ -2,18 +2,25 @@
 	"name": "@aris/backend",
 	"version": "0.0.0",
 	"type": "module",
-	"main": "src/index.ts",
+	"main": "src/server.ts",
 	"scripts": {
-		"dev": "bun run --watch src/index.ts",
-		"start": "bun run src/index.ts",
+		"dev": "bun run --watch src/server.ts",
+		"start": "bun run src/server.ts",
 		"test": "bun test src/"
 	},
 	"dependencies": {
 		"@aris/core": "workspace:*",
 		"@aris/source-location": "workspace:*",
+		"@aris/source-tfl": "workspace:*",
 		"@aris/source-weatherkit": "workspace:*",
+		"@hono/trpc-server": "^0.3",
+		"@trpc/server": "^11",
+		"arktype": "^2.1.29",
 		"better-auth": "^1",
 		"hono": "^4",
-		"postgres": "^3"
+		"pg": "^8"
+	},
+	"devDependencies": {
+		"@types/pg": "^8"
 	}
 }

apps/aris-backend/src/auth/http.ts

@@ -0,0 +1,7 @@
+import type { Hono } from "hono"
+
+import { auth } from "./index.ts"
+
+export function registerAuthHandlers(app: Hono): void {
+	app.on(["POST", "GET"], "/api/auth/*", (c) => auth.handler(c.req.raw))
+}

apps/aris-backend/src/auth/index.ts

@@ -0,0 +1,10 @@
+import { betterAuth } from "better-auth"
+
+import { pool } from "../db.ts"
+
+export const auth = betterAuth({
+	database: pool,
+	emailAndPassword: {
+		enabled: true,
+	},
+})

apps/aris-backend/src/auth/session-middleware.ts

@@ -0,0 +1,54 @@
+import type { Context, Next } from "hono"
+
+import { auth } from "./index.ts"
+
+type SessionUser = typeof auth.$Infer.Session.user
+type Session = typeof auth.$Infer.Session.session
+
+export interface SessionVariables {
+	user: SessionUser | null
+	session: Session | null
+}
+
+/**
+ * Middleware that attaches session and user to the context.
+ * Does not reject unauthenticated requests - use requireSession for that.
+ */
+export async function sessionMiddleware(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()
+}
+
+/**
+ * Middleware that requires a valid session. Returns 401 if not authenticated.
+ */
+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)
+	}
+
+	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: SessionUser; session: Session } | null> {
+	const session = await auth.api.getSession({ headers })
+	return session
+}

apps/aris-backend/src/db.ts

@@ -0,0 +1,5 @@
+import { Pool } from "pg"
+
+export const pool = new Pool({
+	connectionString: process.env.DATABASE_URL,
+})

apps/aris-backend/src/feed/service.test.ts

@@ -0,0 +1,162 @@
+import { describe, expect, mock, test } from "bun:test"
+
+import { LocationService } from "../location/service.ts"
+import { FeedEngineService } from "./service.ts"
+
+describe("FeedEngineService", () => {
+	test("engineForUser creates engine on first call", () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+
+		const engine = service.engineForUser("user-1")
+
+		expect(engine).toBeDefined()
+	})
+
+	test("engineForUser returns same engine for same user", () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+
+		const engine1 = service.engineForUser("user-1")
+		const engine2 = service.engineForUser("user-1")
+
+		expect(engine1).toBe(engine2)
+	})
+
+	test("engineForUser returns different engines for different users", () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+
+		const engine1 = service.engineForUser("user-1")
+		const engine2 = service.engineForUser("user-2")
+
+		expect(engine1).not.toBe(engine2)
+	})
+
+	test("engineForUser registers sources from all providers", async () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+
+		const engine = service.engineForUser("user-1")
+		const result = await engine.refresh()
+
+		expect(result.errors).toHaveLength(0)
+	})
+
+	test("engineForUser works with empty providers array", async () => {
+		const service = new FeedEngineService([])
+
+		const engine = service.engineForUser("user-1")
+		const result = await engine.refresh()
+
+		expect(result.errors).toHaveLength(0)
+		expect(result.items).toHaveLength(0)
+	})
+
+	test("refresh returns feed result", async () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+
+		const result = await service.refresh("user-1")
+
+		expect(result).toHaveProperty("context")
+		expect(result).toHaveProperty("items")
+		expect(result).toHaveProperty("errors")
+		expect(result.context.time).toBeInstanceOf(Date)
+	})
+
+	test("refresh uses location from LocationService", async () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+		const location = {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		}
+
+		// Create engine first, then update location
+		service.engineForUser("user-1")
+		locationService.updateUserLocation("user-1", location)
+
+		const result = await service.refresh("user-1")
+
+		expect(result.context.location).toEqual(location)
+	})
+
+	test("subscribe receives updates", async () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+		const callback = mock()
+
+		service.subscribe("user-1", callback)
+
+		// Push location to trigger update
+		locationService.updateUserLocation("user-1", {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		})
+
+		// Wait for async update propagation
+		await new Promise((resolve) => setTimeout(resolve, 10))
+
+		expect(callback).toHaveBeenCalled()
+	})
+
+	test("subscribe returns unsubscribe function", async () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+		const callback = mock()
+
+		const unsubscribe = service.subscribe("user-1", callback)
+
+		unsubscribe()
+
+		locationService.updateUserLocation("user-1", {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		})
+
+		await new Promise((resolve) => setTimeout(resolve, 10))
+
+		expect(callback).not.toHaveBeenCalled()
+	})
+
+	test("removeUser stops engine and removes it", async () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+		const callback = mock()
+
+		service.subscribe("user-1", callback)
+
+		service.removeUser("user-1")
+
+		// Push location - should not trigger update since engine is stopped
+		locationService.feedSourceForUser("user-1")
+		locationService.updateUserLocation("user-1", {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		})
+
+		await new Promise((resolve) => setTimeout(resolve, 10))
+
+		expect(callback).not.toHaveBeenCalled()
+	})
+
+	test("removeUser allows new engine to be created", () => {
+		const locationService = new LocationService()
+		const service = new FeedEngineService([locationService])
+
+		const engine1 = service.engineForUser("user-1")
+		service.removeUser("user-1")
+		const engine2 = service.engineForUser("user-1")
+
+		expect(engine1).not.toBe(engine2)
+	})
+})

apps/aris-backend/src/feed/service.ts

@@ -0,0 +1,75 @@
+import { FeedEngine, type FeedResult, type FeedSource, type FeedSubscriber } from "@aris/core"
+
+/**
+ * Provides a FeedSource instance for a user.
+ */
+export interface FeedSourceProvider {
+	feedSourceForUser(userId: string): FeedSource
+}
+
+/**
+ * Manages FeedEngine instances per user.
+ *
+ * Receives FeedSource instances from injected providers and wires them
+ * into per-user engines. Engines are auto-started on creation.
+ */
+export class FeedEngineService {
+	private engines = new Map<string, FeedEngine>()
+
+	constructor(private readonly providers: FeedSourceProvider[]) {}
+
+	/**
+	 * Get or create a FeedEngine for a user.
+	 * Automatically registers sources and starts the engine.
+	 */
+	engineForUser(userId: string): FeedEngine {
+		let engine = this.engines.get(userId)
+		if (!engine) {
+			engine = this.createEngine(userId)
+			this.engines.set(userId, engine)
+		}
+		return engine
+	}
+
+	/**
+	 * Refresh a user's feed.
+	 */
+	async refresh(userId: string): Promise<FeedResult> {
+		const engine = this.engineForUser(userId)
+		return engine.refresh()
+	}
+
+	/**
+	 * Subscribe to feed updates for a user.
+	 * Returns unsubscribe function.
+	 */
+	subscribe(userId: string, callback: FeedSubscriber): () => void {
+		const engine = this.engineForUser(userId)
+		return engine.subscribe(callback)
+	}
+
+	/**
+	 * Remove a user's FeedEngine.
+	 * Stops the engine and cleans up resources.
+	 */
+	removeUser(userId: string): void {
+		const engine = this.engines.get(userId)
+		if (engine) {
+			engine.stop()
+			this.engines.delete(userId)
+		}
+	}
+
+	private createEngine(userId: string): FeedEngine {
+		const engine = new FeedEngine()
+
+		for (const provider of this.providers) {
+			const source = provider.feedSourceForUser(userId)
+			engine.register(source)
+		}
+
+		engine.start()
+
+		return engine
+	}
+}

apps/aris-backend/src/index.ts

@@ -1,10 +0,0 @@
-import { Hono } from "hono"
-
-const app = new Hono()
-
-app.get("/health", (c) => c.json({ status: "ok" }))
-
-export default {
-	port: 3000,
-	fetch: app.fetch,
-}

apps/aris-backend/src/lib/error.ts

@@ -0,0 +1,8 @@
+export class UserNotFoundError extends Error {
+	constructor(
+		public readonly userId: string,
+		message?: string,
+	) {
+		super(message ? `${message}: user not found: ${userId}` : `User not found: ${userId}`)
+	}
+}

apps/aris-backend/src/location/router.ts

@@ -0,0 +1,37 @@
+import { TRPCError } from "@trpc/server"
+import { type } from "arktype"
+
+import type { TRPC } from "../trpc/router.ts"
+import type { LocationService } from "./service.ts"
+
+import { UserNotFoundError } from "../lib/error.ts"
+
+const locationInput = type({
+	lat: "number",
+	lng: "number",
+	accuracy: "number",
+	timestamp: "Date",
+})
+
+export function createLocationRouter(
+	t: TRPC,
+	{ locationService }: { locationService: LocationService },
+) {
+	return t.router({
+		update: t.procedure.input(locationInput).mutation(({ input, ctx }) => {
+			try {
+				locationService.updateUserLocation(ctx.user.id, {
+					lat: input.lat,
+					lng: input.lng,
+					accuracy: input.accuracy,
+					timestamp: input.timestamp,
+				})
+			} catch (error) {
+				if (error instanceof UserNotFoundError) {
+					throw new TRPCError({ code: "NOT_FOUND", message: error.message })
+				}
+				throw error
+			}
+		}),
+	})
+}

apps/aris-backend/src/location/service.test.ts

@@ -0,0 +1,111 @@
+import { describe, expect, test } from "bun:test"
+
+import { UserNotFoundError } from "../lib/error.ts"
+import { LocationService } from "./service.ts"
+
+describe("LocationService", () => {
+	test("feedSourceForUser creates source on first call", () => {
+		const service = new LocationService()
+		const source = service.feedSourceForUser("user-1")
+
+		expect(source).toBeDefined()
+		expect(source.id).toBe("location")
+	})
+
+	test("feedSourceForUser returns same source for same user", () => {
+		const service = new LocationService()
+		const source1 = service.feedSourceForUser("user-1")
+		const source2 = service.feedSourceForUser("user-1")
+
+		expect(source1).toBe(source2)
+	})
+
+	test("feedSourceForUser returns different sources for different users", () => {
+		const service = new LocationService()
+		const source1 = service.feedSourceForUser("user-1")
+		const source2 = service.feedSourceForUser("user-2")
+
+		expect(source1).not.toBe(source2)
+	})
+
+	test("updateUserLocation updates the source", () => {
+		const service = new LocationService()
+		const source = service.feedSourceForUser("user-1")
+		const location = {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		}
+
+		service.updateUserLocation("user-1", location)
+
+		expect(source.lastLocation).toEqual(location)
+	})
+
+	test("updateUserLocation throws if source does not exist", () => {
+		const service = new LocationService()
+		const location = {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		}
+
+		expect(() => service.updateUserLocation("user-1", location)).toThrow(UserNotFoundError)
+	})
+
+	test("lastUserLocation returns null for unknown user", () => {
+		const service = new LocationService()
+
+		expect(service.lastUserLocation("unknown")).toBeNull()
+	})
+
+	test("lastUserLocation returns last location", () => {
+		const service = new LocationService()
+		service.feedSourceForUser("user-1")
+		const location1 = {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		}
+		const location2 = {
+			lat: 52.0,
+			lng: -0.2,
+			accuracy: 5,
+			timestamp: new Date(),
+		}
+
+		service.updateUserLocation("user-1", location1)
+		service.updateUserLocation("user-1", location2)
+
+		expect(service.lastUserLocation("user-1")).toEqual(location2)
+	})
+
+	test("removeUser removes the source", () => {
+		const service = new LocationService()
+		service.feedSourceForUser("user-1")
+		const location = {
+			lat: 51.5074,
+			lng: -0.1278,
+			accuracy: 10,
+			timestamp: new Date(),
+		}
+
+		service.updateUserLocation("user-1", location)
+		service.removeUser("user-1")
+
+		expect(service.lastUserLocation("user-1")).toBeNull()
+	})
+
+	test("removeUser allows new source to be created", () => {
+		const service = new LocationService()
+		const source1 = service.feedSourceForUser("user-1")
+
+		service.removeUser("user-1")
+		const source2 = service.feedSourceForUser("user-1")
+
+		expect(source1).not.toBe(source2)
+	})
+})

apps/aris-backend/src/location/service.ts

@@ -0,0 +1,57 @@
+import { LocationSource, type Location } from "@aris/source-location"
+
+import type { FeedSourceProvider } from "../feed/service.ts"
+
+import { UserNotFoundError } from "../lib/error.ts"
+
+/**
+ * Manages LocationSource instances per user.
+ */
+export class LocationService implements FeedSourceProvider {
+	private sources = new Map<string, LocationSource>()
+
+	/**
+	 * Get or create a LocationSource for a user.
+	 * @param userId - The user's unique identifier
+	 * @returns The user's LocationSource instance
+	 */
+	feedSourceForUser(userId: string): LocationSource {
+		let source = this.sources.get(userId)
+		if (!source) {
+			source = new LocationSource()
+			this.sources.set(userId, source)
+		}
+		return source
+	}
+
+	/**
+	 * Update location for a user.
+	 * @param userId - The user's unique identifier
+	 * @param location - The new location data
+	 * @throws {UserNotFoundError} If no source exists for the user
+	 */
+	updateUserLocation(userId: string, location: Location): void {
+		const source = this.sources.get(userId)
+		if (!source) {
+			throw new UserNotFoundError(userId)
+		}
+		source.pushLocation(location)
+	}
+
+	/**
+	 * Get last known location for a user.
+	 * @param userId - The user's unique identifier
+	 * @returns The last location, or null if none exists
+	 */
+	lastUserLocation(userId: string): Location | null {
+		return this.sources.get(userId)?.lastLocation ?? null
+	}
+
+	/**
+	 * Remove a user's LocationSource.
+	 * @param userId - The user's unique identifier
+	 */
+	removeUser(userId: string): void {
+		this.sources.delete(userId)
+	}
+}

apps/aris-backend/src/server.ts

@@ -0,0 +1,46 @@
+import { trpcServer } from "@hono/trpc-server"
+import { Hono } from "hono"
+
+import { registerAuthHandlers } from "./auth/http.ts"
+import { LocationService } from "./location/service.ts"
+import { createContext } from "./trpc/context.ts"
+import { createTRPCRouter } from "./trpc/router.ts"
+import { WeatherService } from "./weather/service.ts"
+
+function main() {
+	const locationService = new LocationService()
+
+	const weatherService = new WeatherService({
+		credentials: {
+			privateKey: process.env.WEATHERKIT_PRIVATE_KEY!,
+			keyId: process.env.WEATHERKIT_KEY_ID!,
+			teamId: process.env.WEATHERKIT_TEAM_ID!,
+			serviceId: process.env.WEATHERKIT_SERVICE_ID!,
+		},
+	})
+
+	const trpcRouter = createTRPCRouter({ locationService, weatherService })
+
+	const app = new Hono()
+
+	app.get("/health", (c) => c.json({ status: "ok" }))
+
+	registerAuthHandlers(app)
+
+	app.use(
+		"/trpc/*",
+		trpcServer({
+			router: trpcRouter,
+			createContext,
+		}),
+	)
+
+	return app
+}
+
+const app = main()
+
+export default {
+	port: 3000,
+	fetch: app.fetch,
+}

apps/aris-backend/src/tfl/service.test.ts

@@ -0,0 +1,206 @@
+import type { Context } from "@aris/core"
+import type { ITflApi, StationLocation, TflLineId, TflLineStatus } from "@aris/source-tfl"
+
+import { describe, expect, test } from "bun:test"
+
+import { UserNotFoundError } from "../lib/error.ts"
+import { TflService } from "./service.ts"
+
+class StubTflApi implements ITflApi {
+	private statuses: TflLineStatus[]
+
+	constructor(statuses: TflLineStatus[] = []) {
+		this.statuses = statuses
+	}
+
+	async fetchLineStatuses(lines?: TflLineId[]): Promise<TflLineStatus[]> {
+		if (lines) {
+			return this.statuses.filter((s) => lines.includes(s.lineId))
+		}
+		return this.statuses
+	}
+
+	async fetchStations(): Promise<StationLocation[]> {
+		return [
+			{
+				id: "940GZZLUKSX",
+				name: "King's Cross",
+				lat: 51.5308,
+				lng: -0.1238,
+				lines: ["northern", "victoria"],
+			},
+		]
+	}
+}
+
+function createContext(): Context {
+	return { time: new Date("2026-01-15T12:00:00Z") }
+}
+
+const sampleStatuses: TflLineStatus[] = [
+	{
+		lineId: "northern",
+		lineName: "Northern",
+		severity: "minor-delays",
+		description: "Minor delays on the Northern line",
+	},
+	{
+		lineId: "victoria",
+		lineName: "Victoria",
+		severity: "major-delays",
+		description: "Severe delays on the Victoria line",
+	},
+	{
+		lineId: "central",
+		lineName: "Central",
+		severity: "closure",
+		description: "Central line suspended",
+	},
+]
+
+describe("TflService", () => {
+	test("feedSourceForUser creates source on first call", () => {
+		const service = new TflService(new StubTflApi())
+		const source = service.feedSourceForUser("user-1")
+
+		expect(source).toBeDefined()
+		expect(source.id).toBe("tfl")
+	})
+
+	test("feedSourceForUser returns same source for same user", () => {
+		const service = new TflService(new StubTflApi())
+		const source1 = service.feedSourceForUser("user-1")
+		const source2 = service.feedSourceForUser("user-1")
+
+		expect(source1).toBe(source2)
+	})
+
+	test("feedSourceForUser returns different sources for different users", () => {
+		const service = new TflService(new StubTflApi())
+		const source1 = service.feedSourceForUser("user-1")
+		const source2 = service.feedSourceForUser("user-2")
+
+		expect(source1).not.toBe(source2)
+	})
+
+	test("updateLinesOfInterest mutates the existing source in place", () => {
+		const service = new TflService(new StubTflApi())
+		const original = service.feedSourceForUser("user-1")
+
+		service.updateLinesOfInterest("user-1", ["northern", "victoria"])
+		const after = service.feedSourceForUser("user-1")
+
+		expect(after).toBe(original)
+	})
+
+	test("updateLinesOfInterest throws if source does not exist", () => {
+		const service = new TflService(new StubTflApi())
+
+		expect(() => service.updateLinesOfInterest("user-1", ["northern"])).toThrow(UserNotFoundError)
+	})
+
+	test("removeUser removes the source", () => {
+		const service = new TflService(new StubTflApi())
+		const source1 = service.feedSourceForUser("user-1")
+
+		service.removeUser("user-1")
+		const source2 = service.feedSourceForUser("user-1")
+
+		expect(source1).not.toBe(source2)
+	})
+
+	test("removeUser clears line configuration", async () => {
+		const api = new StubTflApi(sampleStatuses)
+		const service = new TflService(api)
+		service.feedSourceForUser("user-1")
+		service.updateLinesOfInterest("user-1", ["northern"])
+
+		service.removeUser("user-1")
+		const items = await service.feedSourceForUser("user-1").fetchItems(createContext())
+
+		expect(items.length).toBe(3)
+	})
+
+	test("shares single api instance across users", () => {
+		const api = new StubTflApi()
+		const service = new TflService(api)
+
+		service.feedSourceForUser("user-1")
+		service.feedSourceForUser("user-2")
+
+		expect(service.feedSourceForUser("user-1").id).toBe("tfl")
+		expect(service.feedSourceForUser("user-2").id).toBe("tfl")
+	})
+
+	describe("returned source fetches items", () => {
+		test("source returns feed items from api", async () => {
+			const api = new StubTflApi(sampleStatuses)
+			const service = new TflService(api)
+			const source = service.feedSourceForUser("user-1")
+
+			const items = await source.fetchItems(createContext())
+
+			expect(items.length).toBe(3)
+			for (const item of items) {
+				expect(item.type).toBe("tfl-alert")
+				expect(item.id).toMatch(/^tfl-alert-/)
+				expect(typeof item.priority).toBe("number")
+				expect(item.timestamp).toBeInstanceOf(Date)
+			}
+		})
+
+		test("source returns items sorted by priority descending", async () => {
+			const api = new StubTflApi(sampleStatuses)
+			const service = new TflService(api)
+			const source = service.feedSourceForUser("user-1")
+
+			const items = await source.fetchItems(createContext())
+
+			for (let i = 1; i < items.length; i++) {
+				expect(items[i - 1]!.priority).toBeGreaterThanOrEqual(items[i]!.priority)
+			}
+		})
+
+		test("source returns empty array when no disruptions", async () => {
+			const api = new StubTflApi([])
+			const service = new TflService(api)
+			const source = service.feedSourceForUser("user-1")
+
+			const items = await source.fetchItems(createContext())
+
+			expect(items).toEqual([])
+		})
+
+		test("updateLinesOfInterest filters items to configured lines", async () => {
+			const api = new StubTflApi(sampleStatuses)
+			const service = new TflService(api)
+
+			const before = await service.feedSourceForUser("user-1").fetchItems(createContext())
+			expect(before.length).toBe(3)
+
+			service.updateLinesOfInterest("user-1", ["northern"])
+			const after = await service.feedSourceForUser("user-1").fetchItems(createContext())
+
+			expect(after.length).toBe(1)
+			expect(after[0]!.data.line).toBe("northern")
+		})
+
+		test("different users get independent line configs", async () => {
+			const api = new StubTflApi(sampleStatuses)
+			const service = new TflService(api)
+			service.feedSourceForUser("user-1")
+			service.feedSourceForUser("user-2")
+
+			service.updateLinesOfInterest("user-1", ["northern"])
+			service.updateLinesOfInterest("user-2", ["central"])
+
+			const items1 = await service.feedSourceForUser("user-1").fetchItems(createContext())
+			const items2 = await service.feedSourceForUser("user-2").fetchItems(createContext())
+
+			expect(items1.length).toBe(1)
+			expect(items1[0]!.data.line).toBe("northern")
+			expect(items2.length).toBe(1)
+			expect(items2[0]!.data.line).toBe("central")
+		})
+	})
+})

apps/aris-backend/src/tfl/service.ts

@@ -0,0 +1,40 @@
+import { TflSource, type ITflApi, type TflLineId } from "@aris/source-tfl"
+
+import type { FeedSourceProvider } from "../feed/service.ts"
+
+import { UserNotFoundError } from "../lib/error.ts"
+
+/**
+ * Manages per-user TflSource instances with individual line configuration.
+ */
+export class TflService implements FeedSourceProvider {
+	private sources = new Map<string, TflSource>()
+
+	constructor(private readonly api: ITflApi) {}
+
+	feedSourceForUser(userId: string): TflSource {
+		let source = this.sources.get(userId)
+		if (!source) {
+			source = new TflSource({ client: this.api })
+			this.sources.set(userId, source)
+		}
+		return source
+	}
+
+	/**
+	 * Update monitored lines for a user. Mutates the existing TflSource
+	 * so that references held by FeedEngine remain valid.
+	 * @throws {UserNotFoundError} If no source exists for the user
+	 */
+	updateLinesOfInterest(userId: string, lines: TflLineId[]): void {
+		const source = this.sources.get(userId)
+		if (!source) {
+			throw new UserNotFoundError(userId)
+		}
+		source.setLinesOfInterest(lines)
+	}
+
+	removeUser(userId: string): void {
+		this.sources.delete(userId)
+	}
+}

apps/aris-backend/src/trpc/context.ts

@@ -0,0 +1,14 @@
+import type { FetchCreateContextFnOptions } from "@trpc/server/adapters/fetch"
+
+import { auth } from "../auth/index.ts"
+
+export async function createContext(opts: FetchCreateContextFnOptions) {
+	const session = await auth.api.getSession({ headers: opts.req.headers })
+
+	return {
+		user: session?.user ?? null,
+		session: session?.session ?? null,
+	}
+}
+
+export type Context = Awaited<ReturnType<typeof createContext>>

apps/aris-backend/src/trpc/router.ts

@@ -0,0 +1,50 @@
+import { initTRPC, TRPCError } from "@trpc/server"
+
+import type { LocationService } from "../location/service.ts"
+import type { WeatherService } from "../weather/service.ts"
+import type { Context } from "./context.ts"
+
+import { createLocationRouter } from "../location/router.ts"
+
+interface AuthedContext {
+	user: NonNullable<Context["user"]>
+	session: NonNullable<Context["session"]>
+}
+
+function createTRPC() {
+	const t = initTRPC.context<Context>().create()
+
+	const isAuthed = t.middleware(({ ctx, next }) => {
+		if (!ctx.user || !ctx.session) {
+			throw new TRPCError({ code: "UNAUTHORIZED" })
+		}
+		return next({
+			ctx: {
+				user: ctx.user,
+				session: ctx.session,
+			},
+		})
+	})
+
+	return {
+		router: t.router,
+		procedure: t.procedure.use(isAuthed),
+	}
+}
+
+export type TRPC = ReturnType<typeof createTRPC>
+
+export interface TRPCRouterDeps {
+	locationService: LocationService
+	weatherService: WeatherService
+}
+
+export function createTRPCRouter({ locationService }: TRPCRouterDeps) {
+	const t = createTRPC()
+
+	return t.router({
+		location: createLocationRouter(t, { locationService }),
+	})
+}
+
+export type TRPCRouter = ReturnType<typeof createTRPCRouter>

apps/aris-backend/src/weather/service.test.ts

@@ -0,0 +1,116 @@
+import type { Context } from "@aris/core"
+
+import { LocationKey } from "@aris/source-location"
+import {
+	Units,
+	WeatherFeedItemType,
+	type WeatherKitClient,
+	type WeatherKitResponse,
+} from "@aris/source-weatherkit"
+import { describe, expect, test } from "bun:test"
+
+import fixture from "../../../../packages/aris-source-weatherkit/fixtures/san-francisco.json"
+import { WeatherService } from "./service.ts"
+
+const mockClient = createMockClient(fixture.response as WeatherKitResponse)
+
+function createMockClient(response: WeatherKitResponse): WeatherKitClient {
+	return {
+		fetch: async () => response,
+	}
+}
+
+function createMockContext(location?: { lat: number; lng: number }): Context {
+	const ctx: Context = { time: new Date("2026-01-17T00:00:00Z") }
+	if (location) {
+		ctx[LocationKey] = { ...location, accuracy: 10, timestamp: new Date() }
+	}
+	return ctx
+}
+
+describe("WeatherService", () => {
+	test("feedSourceForUser creates source on first call", () => {
+		const service = new WeatherService({ client: mockClient })
+		const source = service.feedSourceForUser("user-1")
+
+		expect(source).toBeDefined()
+		expect(source.id).toBe("weather")
+	})
+
+	test("feedSourceForUser returns same source for same user", () => {
+		const service = new WeatherService({ client: mockClient })
+		const source1 = service.feedSourceForUser("user-1")
+		const source2 = service.feedSourceForUser("user-1")
+
+		expect(source1).toBe(source2)
+	})
+
+	test("feedSourceForUser returns different sources for different users", () => {
+		const service = new WeatherService({ client: mockClient })
+		const source1 = service.feedSourceForUser("user-1")
+		const source2 = service.feedSourceForUser("user-2")
+
+		expect(source1).not.toBe(source2)
+	})
+
+	test("feedSourceForUser applies hourly and daily limits", async () => {
+		const service = new WeatherService({
+			client: mockClient,
+			hourlyLimit: 3,
+			dailyLimit: 2,
+		})
+		const source = service.feedSourceForUser("user-1")
+		const context = createMockContext({ lat: 37.7749, lng: -122.4194 })
+
+		const items = await source.fetchItems(context)
+
+		const hourly = items.filter((i) => i.type === WeatherFeedItemType.hourly)
+		const daily = items.filter((i) => i.type === WeatherFeedItemType.daily)
+
+		expect(hourly).toHaveLength(3)
+		expect(daily).toHaveLength(2)
+	})
+
+	test("feedSourceForUser applies units", async () => {
+		const service = new WeatherService({
+			client: mockClient,
+			units: Units.imperial,
+		})
+		const source = service.feedSourceForUser("user-1")
+		const context = createMockContext({ lat: 37.7749, lng: -122.4194 })
+
+		const items = await source.fetchItems(context)
+		const current = items.find((i) => i.type === WeatherFeedItemType.current)
+
+		expect(current).toBeDefined()
+		// Fixture has ~15.87°C, imperial should be ~60.6°F
+		expect(current!.data.temperature).toBeGreaterThan(50)
+	})
+
+	test("removeUser removes the source", () => {
+		const service = new WeatherService({ client: mockClient })
+		service.feedSourceForUser("user-1")
+
+		service.removeUser("user-1")
+
+		// After removal, feedSourceForUser should create a new instance
+		const source2 = service.feedSourceForUser("user-1")
+		expect(source2).toBeDefined()
+	})
+
+	test("removeUser allows new source to be created", () => {
+		const service = new WeatherService({ client: mockClient })
+		const source1 = service.feedSourceForUser("user-1")
+
+		service.removeUser("user-1")
+		const source2 = service.feedSourceForUser("user-1")
+
+		expect(source1).not.toBe(source2)
+	})
+
+	test("removeUser is no-op for unknown user", () => {
+		const service = new WeatherService({ client: mockClient })
+
+		expect(() => service.removeUser("unknown")).not.toThrow()
+	})
+})

apps/aris-backend/src/weather/service.ts

@@ -0,0 +1,40 @@
+import { WeatherSource, type WeatherSourceOptions } from "@aris/source-weatherkit"
+
+import type { FeedSourceProvider } from "../feed/service.ts"
+
+/**
+ * Options forwarded to every per-user WeatherSource.
+ * Must include either `credentials` or `client` (same requirement as WeatherSourceOptions).
+ */
+export type WeatherServiceOptions = WeatherSourceOptions
+
+/**
+ * Manages WeatherSource instances per user.
+ */
+export class WeatherService implements FeedSourceProvider {
+	private sources = new Map<string, WeatherSource>()
+	private readonly options: WeatherServiceOptions
+
+	constructor(options: WeatherServiceOptions) {
+		this.options = options
+	}
+
+	/**
+	 * Get or create a WeatherSource for a user.
+	 */
+	feedSourceForUser(userId: string): WeatherSource {
+		let source = this.sources.get(userId)
+		if (!source) {
+			source = new WeatherSource(this.options)
+			this.sources.set(userId, source)
+		}
+		return source
+	}
+
+	/**
+	 * Remove a user's WeatherSource.
+	 */
+	removeUser(userId: string): void {
+		this.sources.delete(userId)
+	}
+}

apps/aris-backend/tsconfig.json

@@ -1,7 +1,4 @@
 {
 	"extends": "../../tsconfig.json",
-	"compilerOptions": {
-		"types": ["bun-types"]
-	},
 	"include": ["src"]
 }

bun.lock

@@ -19,24 +19,23 @@
       "dependencies": {
         "@aris/core": "workspace:*",
         "@aris/source-location": "workspace:*",
+        "@aris/source-tfl": "workspace:*",
         "@aris/source-weatherkit": "workspace:*",
+        "@hono/trpc-server": "^0.3",
+        "@trpc/server": "^11",
+        "arktype": "^2.1.29",
         "better-auth": "^1",
         "hono": "^4",
-        "postgres": "^3",
+        "pg": "^8",
+      },
+      "devDependencies": {
+        "@types/pg": "^8",
       },
     },
     "packages/aris-core": {
       "name": "@aris/core",
       "version": "0.0.0",
     },
-    "packages/aris-data-source-tfl": {
-      "name": "@aris/data-source-tfl",
-      "version": "0.0.0",
-      "dependencies": {
-        "@aris/core": "workspace:*",
-        "arktype": "^2.1.0",
-      },
-    },
     "packages/aris-data-source-weatherkit": {
       "name": "@aris/data-source-weatherkit",
       "version": "0.0.0",
@@ -52,6 +51,15 @@
         "@aris/core": "workspace:*",
       },
     },
+    "packages/aris-source-tfl": {
+      "name": "@aris/source-tfl",
+      "version": "0.0.0",
+      "dependencies": {
+        "@aris/core": "workspace:*",
+        "@aris/source-location": "workspace:*",
+        "arktype": "^2.1.0",
+      },
+    },
     "packages/aris-source-weatherkit": {
       "name": "@aris/source-weatherkit",
       "version": "0.0.0",
@@ -67,12 +75,12 @@
 
     "@aris/core": ["@aris/core@workspace:packages/aris-core"],
 
-    "@aris/data-source-tfl": ["@aris/data-source-tfl@workspace:packages/aris-data-source-tfl"],
-
     "@aris/data-source-weatherkit": ["@aris/data-source-weatherkit@workspace:packages/aris-data-source-weatherkit"],
 
     "@aris/source-location": ["@aris/source-location@workspace:packages/aris-source-location"],
 
+    "@aris/source-tfl": ["@aris/source-tfl@workspace:packages/aris-source-tfl"],
+
     "@aris/source-weatherkit": ["@aris/source-weatherkit@workspace:packages/aris-source-weatherkit"],
 
     "@ark/schema": ["@ark/schema@0.56.0", "", { "dependencies": { "@ark/util": "0.56.0" } }, "sha512-ECg3hox/6Z/nLajxXqNhgPtNdHWC9zNsDyskwO28WinoFEnWow4IsERNz9AnXRhTZJnYIlAJ4uGn3nlLk65vZA=="],
@@ -87,6 +95,8 @@
 
     "@better-fetch/fetch": ["@better-fetch/fetch@1.1.21", "", {}, "sha512-/ImESw0sskqlVR94jB+5+Pxjf+xBwDZF/N5+y2/q4EqD7IARUTSpPfIo8uf39SYpCxyOCtbyYpUrZ3F/k0zT4A=="],
 
+    "@hono/trpc-server": ["@hono/trpc-server@0.3.4", "", { "peerDependencies": { "@trpc/server": "^10.10.0 || >11.0.0-rc", "hono": ">=4.*" } }, "sha512-xFOPjUPnII70FgicDzOJy1ufIoBTu8eF578zGiDOrYOrYN8CJe140s9buzuPkX+SwJRYK8LjEBHywqZtxdm8aA=="],
+
     "@noble/ciphers": ["@noble/ciphers@2.1.1", "", {}, "sha512-bysYuiVfhxNJuldNXlFEitTVdNnYUc+XNJZd7Qm2a5j1vZHgY+fazadNFWFaMK/2vye0JVlxV3gHmC0WDfAOQw=="],
 
     "@noble/hashes": ["@noble/hashes@2.0.1", "", {}, "sha512-XlOlEbQcE9fmuXxrVTXCTlG2nlRXa9Rj3rr5Ue/+tX+nmkgbX720YHh0VR3hBF9xDvwnb8D2shVGOwNx+ulArw=="],
@@ -125,10 +135,14 @@
 
     "@standard-schema/spec": ["@standard-schema/spec@1.1.0", "", {}, "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w=="],
 
+    "@trpc/server": ["@trpc/server@11.8.1", "", { "peerDependencies": { "typescript": ">=5.7.2" } }, "sha512-P4rzZRpEL7zDFgjxK65IdyH0e41FMFfTkQkuq0BA5tKcr7E6v9/v38DEklCpoDN6sPiB1Sigy/PUEzHENhswDA=="],
+
     "@types/bun": ["@types/bun@1.3.6", "", { "dependencies": { "bun-types": "1.3.6" } }, "sha512-uWCv6FO/8LcpREhenN1d1b6fcspAB+cefwD7uti8C8VffIv0Um08TKMn98FynpTiU38+y2dUO55T11NgDt8VAA=="],
 
     "@types/node": ["@types/node@25.0.9", "", { "dependencies": { "undici-types": "~7.16.0" } }, "sha512-/rpCXHlCWeqClNBwUhDcusJxXYDjZTyE8v5oTO7WbL8eij2nKhUeU89/6xgjU7N4/Vh3He0BtyhJdQbDyhiXAw=="],
 
+    "@types/pg": ["@types/pg@8.16.0", "", { "dependencies": { "@types/node": "*", "pg-protocol": "*", "pg-types": "^2.2.0" } }, "sha512-RmhMd/wD+CF8Dfo+cVIy3RR5cl8CyfXQ0tGgW6XBL8L4LM/UTEbNXYRbLwU6w+CgrKBNbrQWt4FUtTfaU5jSYQ=="],
+
     "arkregex": ["arkregex@0.0.5", "", { "dependencies": { "@ark/util": "0.56.0" } }, "sha512-ncYjBdLlh5/QnVsAA8De16Tc9EqmYM7y/WU9j+236KcyYNUXogpz3sC4ATIZYzzLxwI+0sEOaQLEmLmRleaEXw=="],
 
     "arktype": ["arktype@2.1.29", "", { "dependencies": { "@ark/schema": "0.56.0", "@ark/util": "0.56.0", "arkregex": "0.0.5" } }, "sha512-jyfKk4xIOzvYNayqnD8ZJQqOwcrTOUbIU4293yrzAjA3O1dWh61j71ArMQ6tS/u4pD7vabSPe7nG3RCyoXW6RQ=="],
@@ -153,18 +167,44 @@
 
     "oxlint": ["oxlint@1.39.0", "", { "optionalDependencies": { "@oxlint/darwin-arm64": "1.39.0", "@oxlint/darwin-x64": "1.39.0", "@oxlint/linux-arm64-gnu": "1.39.0", "@oxlint/linux-arm64-musl": "1.39.0", "@oxlint/linux-x64-gnu": "1.39.0", "@oxlint/linux-x64-musl": "1.39.0", "@oxlint/win32-arm64": "1.39.0", "@oxlint/win32-x64": "1.39.0" }, "peerDependencies": { "oxlint-tsgolint": ">=0.10.0" }, "optionalPeers": ["oxlint-tsgolint"], "bin": { "oxlint": "bin/oxlint" } }, "sha512-wSiLr0wjG+KTU6c1LpVoQk7JZ7l8HCKlAkVDVTJKWmCGazsNxexxnOXl7dsar92mQcRnzko5g077ggP3RINSjA=="],
 
-    "postgres": ["postgres@3.4.8", "", {}, "sha512-d+JFcLM17njZaOLkv6SCev7uoLaBtfK86vMUXhW1Z4glPWh4jozno9APvW/XKFJ3CCxVoC7OL38BqRydtu5nGg=="],
+    "pg": ["pg@8.17.2", "", { "dependencies": { "pg-connection-string": "^2.10.1", "pg-pool": "^3.11.0", "pg-protocol": "^1.11.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, "optionalDependencies": { "pg-cloudflare": "^1.3.0" }, "peerDependencies": { "pg-native": ">=3.0.1" }, "optionalPeers": ["pg-native"] }, "sha512-vjbKdiBJRqzcYw1fNU5KuHyYvdJ1qpcQg1CeBrHFqV1pWgHeVR6j/+kX0E1AAXfyuLUGY1ICrN2ELKA/z2HWzw=="],
+
+    "pg-cloudflare": ["pg-cloudflare@1.3.0", "", {}, "sha512-6lswVVSztmHiRtD6I8hw4qP/nDm1EJbKMRhf3HCYaqud7frGysPv7FYJ5noZQdhQtN2xJnimfMtvQq21pdbzyQ=="],
+
+    "pg-connection-string": ["pg-connection-string@2.10.1", "", {}, "sha512-iNzslsoeSH2/gmDDKiyMqF64DATUCWj3YJ0wP14kqcsf2TUklwimd+66yYojKwZCA7h2yRNLGug71hCBA2a4sw=="],
+
+    "pg-int8": ["pg-int8@1.0.1", "", {}, "sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw=="],
+
+    "pg-pool": ["pg-pool@3.11.0", "", { "peerDependencies": { "pg": ">=8.0" } }, "sha512-MJYfvHwtGp870aeusDh+hg9apvOe2zmpZJpyt+BMtzUWlVqbhFmMK6bOBXLBUPd7iRtIF9fZplDc7KrPN3PN7w=="],
+
+    "pg-protocol": ["pg-protocol@1.11.0", "", {}, "sha512-pfsxk2M9M3BuGgDOfuy37VNRRX3jmKgMjcvAcWqNDpZSf4cUmv8HSOl5ViRQFsfARFn0KuUQTgLxVMbNq5NW3g=="],
+
+    "pg-types": ["pg-types@2.2.0", "", { "dependencies": { "pg-int8": "1.0.1", "postgres-array": "~2.0.0", "postgres-bytea": "~1.0.0", "postgres-date": "~1.0.4", "postgres-interval": "^1.1.0" } }, "sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA=="],
+
+    "pgpass": ["pgpass@1.0.5", "", { "dependencies": { "split2": "^4.1.0" } }, "sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug=="],
+
+    "postgres-array": ["postgres-array@2.0.0", "", {}, "sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA=="],
+
+    "postgres-bytea": ["postgres-bytea@1.0.1", "", {}, "sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ=="],
+
+    "postgres-date": ["postgres-date@1.0.7", "", {}, "sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q=="],
+
+    "postgres-interval": ["postgres-interval@1.2.0", "", { "dependencies": { "xtend": "^4.0.0" } }, "sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ=="],
 
     "rou3": ["rou3@0.7.12", "", {}, "sha512-iFE4hLDuloSWcD7mjdCDhx2bKcIsYbtOTpfH5MHHLSKMOUyjqQXTeZVa289uuwEGEKFoE/BAPbhaU4B774nceg=="],
 
     "set-cookie-parser": ["set-cookie-parser@2.7.2", "", {}, "sha512-oeM1lpU/UvhTxw+g3cIfxXHyJRc/uidd3yK1P242gzHds0udQBYzs3y8j4gCCW+ZJ7ad0yctld8RYO+bdurlvw=="],
 
+    "split2": ["split2@4.2.0", "", {}, "sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg=="],
+
     "tinypool": ["tinypool@2.0.0", "", {}, "sha512-/RX9RzeH2xU5ADE7n2Ykvmi9ED3FBGPAjw9u3zucrNNaEBIO0HPSYgL0NT7+3p147ojeSdaVu08F6hjpv31HJg=="],
 
     "typescript": ["typescript@5.9.3", "", { "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" } }, "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw=="],
 
     "undici-types": ["undici-types@7.16.0", "", {}, "sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw=="],
 
+    "xtend": ["xtend@4.0.2", "", {}, "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ=="],
+
     "zod": ["zod@4.3.6", "", {}, "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg=="],
   }
 }

docs/backend-spec.md

@@ -7,12 +7,14 @@ ARIS needs a backend service that manages per-user FeedEngine instances and deli
 ## Requirements
 
 ### Authentication
+
 - Email/password authentication using BetterAuth
 - PostgreSQL for session and user storage
 - Session tokens validated via `Authorization: Bearer <token>` header
 - Auth endpoints exposed via BetterAuth's built-in routes
 
 ### FeedEngine Management
+
 - Each authenticated user gets their own FeedEngine instance
 - Instances are cached in memory with a 30-minute TTL
 - TTL resets on any activity (WebSocket message, location update)
@@ -20,6 +22,7 @@ ARIS needs a backend service that manages per-user FeedEngine instances and deli
 - Source configuration is hardcoded initially (customization deferred)
 
 ### WebSocket Connection
+
 - Single endpoint: `GET /ws` (upgrades to WebSocket)
 - Authentication via `Authorization: Bearer <token>` header on upgrade request
 - Rejected before upgrade if token is invalid
@@ -28,39 +31,44 @@ ARIS needs a backend service that manages per-user FeedEngine instances and deli
 - On connect: immediately send current feed state
 
 ### JSON-RPC Protocol
+
 All WebSocket communication uses JSON-RPC 2.0.
 
 **Client → Server (Requests):**
+
 ```json
 { "jsonrpc": "2.0", "method": "location.update", "params": { "lat": 51.5, "lng": -0.1, "accuracy": 10, "timestamp": "2025-01-01T12:00:00Z" }, "id": 1 }
 { "jsonrpc": "2.0", "method": "feed.refresh", "params": {}, "id": 2 }
 ```
 
 **Server → Client (Responses):**
+
 ```json
 { "jsonrpc": "2.0", "result": { "ok": true }, "id": 1 }
 ```
 
 **Server → Client (Notifications - no id):**
+
 ```json
 { "jsonrpc": "2.0", "method": "feed.update", "params": { "items": [...], "errors": [...] } }
 ```
 
 ### JSON-RPC Methods
 
-| Method | Params | Description |
-|--------|--------|-------------|
+| Method            | Params                              | Description                                 |
+| ----------------- | ----------------------------------- | ------------------------------------------- |
 | `location.update` | `{ lat, lng, accuracy, timestamp }` | Push location update, triggers feed refresh |
-| `feed.refresh` | `{}` | Force manual feed refresh |
+| `feed.refresh`    | `{}`                                | Force manual feed refresh                   |
 
 ### Server Notifications
 
-| Method | Params | Description |
-|--------|--------|-------------|
-| `feed.update` | `{ context, items, errors }` | Feed state changed |
-| `error` | `{ code, message, data? }` | Source or system error |
+| Method        | Params                       | Description            |
+| ------------- | ---------------------------- | ---------------------- |
+| `feed.update` | `{ context, items, errors }` | Feed state changed     |
+| `error`       | `{ code, message, data? }`   | Source or system error |
 
 ### Error Handling
+
 - Source failures during refresh are reported via `error` notification
 - Format: `{ "jsonrpc": "2.0", "method": "error", "params": { "code": -32000, "message": "...", "data": { "sourceId": "weather" } } }`
 
@@ -96,16 +104,19 @@ All WebSocket communication uses JSON-RPC 2.0.
 ## Implementation Approach
 
 ### Phase 1: Project Setup
+
 1. Create `apps/aris-backend` with Hono
 2. Configure TypeScript, add dependencies (hono, better-auth, postgres driver)
 3. Set up database connection and BetterAuth
 
 ### Phase 2: Authentication
+
 4. Configure BetterAuth with email/password provider
 5. Mount BetterAuth routes at `/api/auth/*`
 6. Create session validation helper for extracting user from token
 
 ### Phase 3: FeedEngine Manager
+
 7. Create `FeedEngineManager` class:
    - `getOrCreate(userId): FeedEngine` - returns cached or creates new
    - `touch(userId)` - resets TTL
@@ -114,22 +125,26 @@ All WebSocket communication uses JSON-RPC 2.0.
 8. Factory function to create FeedEngine with default sources
 
 ### Phase 4: WebSocket Handler
+
 9. Create WebSocket upgrade endpoint at `/ws`
 10. Validate `Authorization` header before upgrade
 11. On connect: register connection, send initial feed state
 12. On disconnect: unregister connection
 
 ### Phase 5: JSON-RPC Handler
+
 13. Create JSON-RPC message parser and dispatcher
 14. Implement `location.update` method
 15. Implement `feed.refresh` method
 16. Wire FeedEngine subscription to broadcast `feed.update` to all user connections
 
 ### Phase 6: Connection Manager
+
 17. Create `ConnectionManager` to track WebSocket connections per user
 18. Broadcast helper to send to all connections for a user
 
 ### Phase 7: Integration & Testing
+
 19. Integration test: auth → connect → location update → receive feed
 20. Test multiple connections receive same updates
 21. Test TTL cleanup
@@ -158,15 +173,15 @@ apps/aris-backend/
 
 ```json
 {
-  "dependencies": {
-    "hono": "^4",
-    "better-auth": "^1",
-    "postgres": "^3",
-    "@aris/core": "workspace:*",
-    "@aris/source-location": "workspace:*",
-    "@aris/source-weatherkit": "workspace:*",
-    "@aris/data-source-tfl": "workspace:*"
-  }
+	"dependencies": {
+		"hono": "^4",
+		"better-auth": "^1",
+		"postgres": "^3",
+		"@aris/core": "workspace:*",
+		"@aris/source-location": "workspace:*",
+		"@aris/source-weatherkit": "workspace:*",
+		"@aris/data-source-tfl": "workspace:*"
+	}
 }
 ```
 

packages/aris-data-source-tfl/src/data-source.ts

@@ -1,115 +0,0 @@
-import type { Context, DataSource } from "@aris/core"
-
-import type {
-	StationLocation,
-	TflAlertData,
-	TflAlertFeedItem,
-	TflAlertSeverity,
-	TflDataSourceConfig,
-	TflDataSourceOptions,
-	TflLineId,
-} from "./types.ts"
-
-import { TflApi, type ITflApi } from "./tfl-api.ts"
-
-const SEVERITY_PRIORITY: Record<TflAlertSeverity, number> = {
-	closure: 100,
-	"major-delays": 80,
-	"minor-delays": 60,
-}
-
-function haversineDistance(lat1: number, lng1: number, lat2: number, lng2: number): number {
-	const R = 6371 // Earth's radius in km
-	const dLat = ((lat2 - lat1) * Math.PI) / 180
-	const dLng = ((lng2 - lng1) * Math.PI) / 180
-	const a =
-		Math.sin(dLat / 2) * Math.sin(dLat / 2) +
-		Math.cos((lat1 * Math.PI) / 180) *
-			Math.cos((lat2 * Math.PI) / 180) *
-			Math.sin(dLng / 2) *
-			Math.sin(dLng / 2)
-	const c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a))
-	return R * c
-}
-
-function findClosestStationDistance(
-	lineId: TflLineId,
-	stations: StationLocation[],
-	userLat: number,
-	userLng: number,
-): number | null {
-	const lineStations = stations.filter((s) => s.lines.includes(lineId))
-	if (lineStations.length === 0) return null
-
-	let minDistance = Infinity
-	for (const station of lineStations) {
-		const distance = haversineDistance(userLat, userLng, station.lat, station.lng)
-		if (distance < minDistance) {
-			minDistance = distance
-		}
-	}
-
-	return minDistance
-}
-
-export class TflDataSource implements DataSource<TflAlertFeedItem, TflDataSourceConfig> {
-	readonly type = "tfl-alert"
-	private api: ITflApi
-
-	constructor(options: TflDataSourceOptions)
-	constructor(api: ITflApi)
-	constructor(optionsOrApi: TflDataSourceOptions | ITflApi) {
-		if ("fetchLineStatuses" in optionsOrApi) {
-			this.api = optionsOrApi
-		} else {
-			this.api = new TflApi(optionsOrApi.apiKey)
-		}
-	}
-
-	async query(context: Context, config: TflDataSourceConfig): Promise<TflAlertFeedItem[]> {
-		const [statuses, stations] = await Promise.all([
-			this.api.fetchLineStatuses(config.lines),
-			this.api.fetchStations(),
-		])
-
-		const items: TflAlertFeedItem[] = statuses.map((status) => {
-			const closestStationDistance = context.location
-				? findClosestStationDistance(
-						status.lineId,
-						stations,
-						context.location.lat,
-						context.location.lng,
-					)
-				: null
-
-			const data: TflAlertData = {
-				line: status.lineId,
-				lineName: status.lineName,
-				severity: status.severity,
-				description: status.description,
-				closestStationDistance,
-			}
-
-			return {
-				id: `tfl-alert-${status.lineId}-${status.severity}`,
-				type: this.type,
-				priority: SEVERITY_PRIORITY[status.severity],
-				timestamp: context.time,
-				data,
-			}
-		})
-
-		// Sort by severity (desc), then by proximity (asc) if location available
-		items.sort((a, b) => {
-			if (b.priority !== a.priority) {
-				return b.priority - a.priority
-			}
-			if (a.data.closestStationDistance !== null && b.data.closestStationDistance !== null) {
-				return a.data.closestStationDistance - b.data.closestStationDistance
-			}
-			return 0
-		})
-
-		return items
-	}
-}

packages/aris-data-source-tfl/src/index.ts

@@ -1,11 +0,0 @@
-export { TflDataSource } from "./data-source.ts"
-export { TflApi, type ITflApi, type TflLineStatus } from "./tfl-api.ts"
-export type {
-	TflAlertData,
-	TflAlertFeedItem,
-	TflAlertSeverity,
-	TflDataSourceConfig,
-	TflDataSourceOptions,
-	TflLineId,
-	StationLocation,
-} from "./types.ts"

packages/aris-data-source-tfl/src/integration.test.ts

@@ -1,208 +0,0 @@
-import type { Context } from "@aris/core"
-
-import { describe, expect, test } from "bun:test"
-
-import type { ITflApi, TflLineStatus } from "./tfl-api.ts"
-import type { StationLocation, TflLineId } from "./types.ts"
-
-import fixtures from "../fixtures/tfl-responses.json"
-import { TflDataSource } from "./data-source.ts"
-
-// Mock API that returns fixture data
-class FixtureTflApi implements ITflApi {
-	async fetchLineStatuses(_lines?: TflLineId[]): Promise<TflLineStatus[]> {
-		const statuses: TflLineStatus[] = []
-
-		for (const line of fixtures.lineStatuses as Record<string, unknown>[]) {
-			for (const status of line.lineStatuses as Record<string, unknown>[]) {
-				const severityCode = status.statusSeverity as number
-				const severity = this.mapSeverity(severityCode)
-				if (severity) {
-					statuses.push({
-						lineId: line.id as TflLineId,
-						lineName: line.name as string,
-						severity,
-						description: (status.reason as string) ?? (status.statusSeverityDescription as string),
-					})
-				}
-			}
-		}
-
-		return statuses
-	}
-
-	async fetchStations(): Promise<StationLocation[]> {
-		const stationMap = new Map<string, StationLocation>()
-
-		for (const [lineId, stops] of Object.entries(fixtures.stopPoints)) {
-			for (const stop of stops as Record<string, unknown>[]) {
-				const id = stop.naptanId as string
-				const existing = stationMap.get(id)
-				if (existing) {
-					if (!existing.lines.includes(lineId as TflLineId)) {
-						existing.lines.push(lineId as TflLineId)
-					}
-				} else {
-					stationMap.set(id, {
-						id,
-						name: stop.commonName as string,
-						lat: stop.lat as number,
-						lng: stop.lon as number,
-						lines: [lineId as TflLineId],
-					})
-				}
-			}
-		}
-
-		return Array.from(stationMap.values())
-	}
-
-	private mapSeverity(code: number): "minor-delays" | "major-delays" | "closure" | null {
-		const map: Record<number, "minor-delays" | "major-delays" | "closure" | null> = {
-			1: "closure",
-			2: "closure",
-			3: "closure",
-			4: "closure",
-			5: "closure",
-			6: "major-delays",
-			7: "major-delays",
-			8: "major-delays",
-			9: "minor-delays",
-			10: null,
-		}
-		return map[code] ?? null
-	}
-}
-
-const createContext = (location?: { lat: number; lng: number }): Context => ({
-	time: new Date("2026-01-15T12:00:00Z"),
-	location: location ? { ...location, accuracy: 10 } : undefined,
-})
-
-describe("TfL Feed Items (using fixture data)", () => {
-	const api = new FixtureTflApi()
-
-	test("query returns feed items array", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext(), {})
-		expect(Array.isArray(items)).toBe(true)
-	})
-
-	test("feed items have correct base structure", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext({ lat: 51.5074, lng: -0.1278 }), {})
-
-		for (const item of items) {
-			expect(typeof item.id).toBe("string")
-			expect(item.id).toMatch(/^tfl-alert-/)
-			expect(item.type).toBe("tfl-alert")
-			expect(typeof item.priority).toBe("number")
-			expect(item.timestamp).toBeInstanceOf(Date)
-		}
-	})
-
-	test("feed items have correct data structure", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext({ lat: 51.5074, lng: -0.1278 }), {})
-
-		for (const item of items) {
-			expect(typeof item.data.line).toBe("string")
-			expect(typeof item.data.lineName).toBe("string")
-			expect(["minor-delays", "major-delays", "closure"]).toContain(item.data.severity)
-			expect(typeof item.data.description).toBe("string")
-			expect(
-				item.data.closestStationDistance === null ||
-					typeof item.data.closestStationDistance === "number",
-			).toBe(true)
-		}
-	})
-
-	test("feed item ids are unique", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext(), {})
-
-		const ids = items.map((item) => item.id)
-		const uniqueIds = new Set(ids)
-		expect(uniqueIds.size).toBe(ids.length)
-	})
-
-	test("feed items are sorted by priority descending", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext(), {})
-
-		for (let i = 1; i < items.length; i++) {
-			const prev = items[i - 1]!
-			const curr = items[i]!
-			expect(prev.priority).toBeGreaterThanOrEqual(curr.priority)
-		}
-	})
-
-	test("priority values match severity levels", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext(), {})
-
-		const severityPriority: Record<string, number> = {
-			closure: 100,
-			"major-delays": 80,
-			"minor-delays": 60,
-		}
-
-		for (const item of items) {
-			expect(item.priority).toBe(severityPriority[item.data.severity]!)
-		}
-	})
-
-	test("closestStationDistance is number when location provided", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext({ lat: 51.5074, lng: -0.1278 }), {})
-
-		for (const item of items) {
-			expect(typeof item.data.closestStationDistance).toBe("number")
-			expect(item.data.closestStationDistance!).toBeGreaterThan(0)
-		}
-	})
-
-	test("closestStationDistance is null when no location provided", async () => {
-		const dataSource = new TflDataSource(api)
-		const items = await dataSource.query(createContext(), {})
-
-		for (const item of items) {
-			expect(item.data.closestStationDistance).toBeNull()
-		}
-	})
-})
-
-describe("TfL Fixture Data Shape", () => {
-	test("fixtures have expected structure", () => {
-		expect(typeof fixtures.fetchedAt).toBe("string")
-		expect(Array.isArray(fixtures.lineStatuses)).toBe(true)
-		expect(typeof fixtures.stopPoints).toBe("object")
-	})
-
-	test("line statuses have required fields", () => {
-		for (const line of fixtures.lineStatuses as Record<string, unknown>[]) {
-			expect(typeof line.id).toBe("string")
-			expect(typeof line.name).toBe("string")
-			expect(Array.isArray(line.lineStatuses)).toBe(true)
-
-			for (const status of line.lineStatuses as Record<string, unknown>[]) {
-				expect(typeof status.statusSeverity).toBe("number")
-				expect(typeof status.statusSeverityDescription).toBe("string")
-			}
-		}
-	})
-
-	test("stop points have required fields", () => {
-		for (const [lineId, stops] of Object.entries(fixtures.stopPoints)) {
-			expect(typeof lineId).toBe("string")
-			expect(Array.isArray(stops)).toBe(true)
-
-			for (const stop of stops as Record<string, unknown>[]) {
-				expect(typeof stop.naptanId).toBe("string")
-				expect(typeof stop.commonName).toBe("string")
-				expect(typeof stop.lat).toBe("number")
-				expect(typeof stop.lon).toBe("number")
-			}
-		}
-	})
-})

packages/aris-data-source-tfl/src/types.ts

@@ -1,33 +0,0 @@
-import type { FeedItem } from "@aris/core"
-
-import type { TflLineId } from "./tfl-api.ts"
-
-export type { TflLineId } from "./tfl-api.ts"
-
-export type TflAlertSeverity = "minor-delays" | "major-delays" | "closure"
-
-export interface TflAlertData extends Record<string, unknown> {
-	line: TflLineId
-	lineName: string
-	severity: TflAlertSeverity
-	description: string
-	closestStationDistance: number | null
-}
-
-export type TflAlertFeedItem = FeedItem<"tfl-alert", TflAlertData>
-
-export interface TflDataSourceConfig {
-	lines?: TflLineId[]
-}
-
-export interface TflDataSourceOptions {
-	apiKey: string
-}
-
-export interface StationLocation {
-	id: string
-	name: string
-	lat: number
-	lng: number
-	lines: TflLineId[]
-}

packages/aris-source-tfl/package.json

@@ -1,5 +1,5 @@
 {
-	"name": "@aris/data-source-tfl",
+	"name": "@aris/source-tfl",
 	"version": "0.0.0",
 	"type": "module",
 	"main": "src/index.ts",
@@ -10,6 +10,7 @@
 	},
 	"dependencies": {
 		"@aris/core": "workspace:*",
+		"@aris/source-location": "workspace:*",
 		"arktype": "^2.1.0"
 	}
 }

packages/aris-source-tfl/src/index.ts

@@ -0,0 +1,12 @@
+export { TflSource } from "./tfl-source.ts"
+export { TflApi } from "./tfl-api.ts"
+export type { TflLineId } from "./tfl-api.ts"
+export type {
+	ITflApi,
+	StationLocation,
+	TflAlertData,
+	TflAlertFeedItem,
+	TflAlertSeverity,
+	TflLineStatus,
+	TflSourceOptions,
+} from "./types.ts"

packages/aris-source-tfl/src/tfl-api.ts

@@ -1,6 +1,6 @@
 import { type } from "arktype"
 
-import type { StationLocation, TflAlertSeverity } from "./types.ts"
+import type { StationLocation, TflAlertSeverity, TflLineStatus } from "./types.ts"
 
 const TFL_API_BASE = "https://api.tfl.gov.uk"
 
@@ -50,19 +50,7 @@ const SEVERITY_MAP: Record<number, TflAlertSeverity | null> = {
 	20: null, // Service Closed
 }
 
-export interface TflLineStatus {
-	lineId: TflLineId
-	lineName: string
-	severity: TflAlertSeverity
-	description: string
-}
-
-export interface ITflApi {
-	fetchLineStatuses(lines?: TflLineId[]): Promise<TflLineStatus[]>
-	fetchStations(): Promise<StationLocation[]>
-}
-
-export class TflApi implements ITflApi {
+export class TflApi {
 	private apiKey: string
 	private stationsCache: StationLocation[] | null = null
 

packages/aris-source-tfl/src/tfl-source.test.ts

@@ -0,0 +1,286 @@
+import type { Context } from "@aris/core"
+
+import { LocationKey, type Location } from "@aris/source-location"
+import { describe, expect, test } from "bun:test"
+
+import type {
+	ITflApi,
+	StationLocation,
+	TflAlertSeverity,
+	TflLineId,
+	TflLineStatus,
+} from "./types.ts"
+
+import fixtures from "../fixtures/tfl-responses.json"
+import { TflSource } from "./tfl-source.ts"
+
+// Mock API that returns fixture data
+class FixtureTflApi implements ITflApi {
+	async fetchLineStatuses(_lines?: TflLineId[]): Promise<TflLineStatus[]> {
+		const statuses: TflLineStatus[] = []
+
+		for (const line of fixtures.lineStatuses as Record<string, unknown>[]) {
+			for (const status of line.lineStatuses as Record<string, unknown>[]) {
+				const severityCode = status.statusSeverity as number
+				const severity = this.mapSeverity(severityCode)
+				if (severity) {
+					statuses.push({
+						lineId: line.id as TflLineId,
+						lineName: line.name as string,
+						severity,
+						description: (status.reason as string) ?? (status.statusSeverityDescription as string),
+					})
+				}
+			}
+		}
+
+		return statuses
+	}
+
+	async fetchStations(): Promise<StationLocation[]> {
+		const stationMap = new Map<string, StationLocation>()
+
+		for (const [lineId, stops] of Object.entries(fixtures.stopPoints)) {
+			for (const stop of stops as Record<string, unknown>[]) {
+				const id = stop.naptanId as string
+				const existing = stationMap.get(id)
+				if (existing) {
+					if (!existing.lines.includes(lineId as TflLineId)) {
+						existing.lines.push(lineId as TflLineId)
+					}
+				} else {
+					stationMap.set(id, {
+						id,
+						name: stop.commonName as string,
+						lat: stop.lat as number,
+						lng: stop.lon as number,
+						lines: [lineId as TflLineId],
+					})
+				}
+			}
+		}
+
+		return Array.from(stationMap.values())
+	}
+
+	private mapSeverity(code: number): TflAlertSeverity | null {
+		const map: Record<number, TflAlertSeverity | null> = {
+			1: "closure",
+			2: "closure",
+			3: "closure",
+			4: "closure",
+			5: "closure",
+			6: "major-delays",
+			7: "major-delays",
+			8: "major-delays",
+			9: "minor-delays",
+			10: null,
+		}
+		return map[code] ?? null
+	}
+}
+
+function createContext(location?: Location): Context {
+	const ctx: Context = { time: new Date("2026-01-15T12:00:00Z") }
+	if (location) {
+		ctx[LocationKey] = location
+	}
+	return ctx
+}
+
+describe("TflSource", () => {
+	const api = new FixtureTflApi()
+
+	describe("interface", () => {
+		test("has correct id", () => {
+			const source = new TflSource({ client: api })
+			expect(source.id).toBe("tfl")
+		})
+
+		test("depends on location", () => {
+			const source = new TflSource({ client: api })
+			expect(source.dependencies).toEqual(["location"])
+		})
+
+		test("implements fetchItems", () => {
+			const source = new TflSource({ client: api })
+			expect(source.fetchItems).toBeDefined()
+		})
+
+		test("throws if neither client nor apiKey provided", () => {
+			expect(() => new TflSource({})).toThrow("Either client or apiKey must be provided")
+		})
+	})
+
+	describe("setLinesOfInterest", () => {
+		const lineFilteringApi: ITflApi = {
+			async fetchLineStatuses(lines?: TflLineId[]): Promise<TflLineStatus[]> {
+				const all: TflLineStatus[] = [
+					{
+						lineId: "northern",
+						lineName: "Northern",
+						severity: "minor-delays",
+						description: "Delays",
+					},
+					{ lineId: "central", lineName: "Central", severity: "closure", description: "Closed" },
+				]
+				return lines ? all.filter((s) => lines.includes(s.lineId)) : all
+			},
+			async fetchStations(): Promise<StationLocation[]> {
+				return []
+			},
+		}
+
+		test("changes which lines are fetched", async () => {
+			const source = new TflSource({ client: lineFilteringApi })
+			const before = await source.fetchItems(createContext())
+			expect(before.length).toBe(2)
+
+			source.setLinesOfInterest(["northern"])
+			const after = await source.fetchItems(createContext())
+
+			expect(after.length).toBe(1)
+			expect(after[0]!.data.line).toBe("northern")
+		})
+
+		test("DEFAULT_LINES_OF_INTEREST restores all lines", async () => {
+			const source = new TflSource({ client: lineFilteringApi, lines: ["northern"] })
+			const filtered = await source.fetchItems(createContext())
+			expect(filtered.length).toBe(1)
+
+			source.setLinesOfInterest([...TflSource.DEFAULT_LINES_OF_INTEREST])
+			const all = await source.fetchItems(createContext())
+
+			expect(all.length).toBe(2)
+		})
+	})
+
+	describe("fetchItems", () => {
+		test("returns feed items array", async () => {
+			const source = new TflSource({ client: api })
+			const items = await source.fetchItems(createContext())
+			expect(Array.isArray(items)).toBe(true)
+		})
+
+		test("feed items have correct base structure", async () => {
+			const source = new TflSource({ client: api })
+			const location: Location = { lat: 51.5074, lng: -0.1278, accuracy: 10, timestamp: new Date() }
+			const items = await source.fetchItems(createContext(location))
+
+			for (const item of items) {
+				expect(typeof item.id).toBe("string")
+				expect(item.id).toMatch(/^tfl-alert-/)
+				expect(item.type).toBe("tfl-alert")
+				expect(typeof item.priority).toBe("number")
+				expect(item.timestamp).toBeInstanceOf(Date)
+			}
+		})
+
+		test("feed items have correct data structure", async () => {
+			const source = new TflSource({ client: api })
+			const location: Location = { lat: 51.5074, lng: -0.1278, accuracy: 10, timestamp: new Date() }
+			const items = await source.fetchItems(createContext(location))
+
+			for (const item of items) {
+				expect(typeof item.data.line).toBe("string")
+				expect(typeof item.data.lineName).toBe("string")
+				expect(["minor-delays", "major-delays", "closure"]).toContain(item.data.severity)
+				expect(typeof item.data.description).toBe("string")
+				expect(
+					item.data.closestStationDistance === null ||
+						typeof item.data.closestStationDistance === "number",
+				).toBe(true)
+			}
+		})
+
+		test("feed item ids are unique", async () => {
+			const source = new TflSource({ client: api })
+			const items = await source.fetchItems(createContext())
+
+			const ids = items.map((item) => item.id)
+			const uniqueIds = new Set(ids)
+			expect(uniqueIds.size).toBe(ids.length)
+		})
+
+		test("feed items are sorted by priority descending", async () => {
+			const source = new TflSource({ client: api })
+			const items = await source.fetchItems(createContext())
+
+			for (let i = 1; i < items.length; i++) {
+				const prev = items[i - 1]!
+				const curr = items[i]!
+				expect(prev.priority).toBeGreaterThanOrEqual(curr.priority)
+			}
+		})
+
+		test("priority values match severity levels", async () => {
+			const source = new TflSource({ client: api })
+			const items = await source.fetchItems(createContext())
+
+			const severityPriority: Record<string, number> = {
+				closure: 1.0,
+				"major-delays": 0.8,
+				"minor-delays": 0.6,
+			}
+
+			for (const item of items) {
+				expect(item.priority).toBe(severityPriority[item.data.severity]!)
+			}
+		})
+
+		test("closestStationDistance is number when location provided", async () => {
+			const source = new TflSource({ client: api })
+			const location: Location = { lat: 51.5074, lng: -0.1278, accuracy: 10, timestamp: new Date() }
+			const items = await source.fetchItems(createContext(location))
+
+			for (const item of items) {
+				expect(typeof item.data.closestStationDistance).toBe("number")
+				expect(item.data.closestStationDistance!).toBeGreaterThan(0)
+			}
+		})
+
+		test("closestStationDistance is null when no location provided", async () => {
+			const source = new TflSource({ client: api })
+			const items = await source.fetchItems(createContext())
+
+			for (const item of items) {
+				expect(item.data.closestStationDistance).toBeNull()
+			}
+		})
+	})
+})
+
+describe("TfL Fixture Data Shape", () => {
+	test("fixtures have expected structure", () => {
+		expect(typeof fixtures.fetchedAt).toBe("string")
+		expect(Array.isArray(fixtures.lineStatuses)).toBe(true)
+		expect(typeof fixtures.stopPoints).toBe("object")
+	})
+
+	test("line statuses have required fields", () => {
+		for (const line of fixtures.lineStatuses as Record<string, unknown>[]) {
+			expect(typeof line.id).toBe("string")
+			expect(typeof line.name).toBe("string")
+			expect(Array.isArray(line.lineStatuses)).toBe(true)
+
+			for (const status of line.lineStatuses as Record<string, unknown>[]) {
+				expect(typeof status.statusSeverity).toBe("number")
+				expect(typeof status.statusSeverityDescription).toBe("string")
+			}
+		}
+	})
+
+	test("stop points have required fields", () => {
+		for (const [lineId, stops] of Object.entries(fixtures.stopPoints)) {
+			expect(typeof lineId).toBe("string")
+			expect(Array.isArray(stops)).toBe(true)
+
+			for (const stop of stops as Record<string, unknown>[]) {
+				expect(typeof stop.naptanId).toBe("string")
+				expect(typeof stop.commonName).toBe("string")
+				expect(typeof stop.lat).toBe("number")
+				expect(typeof stop.lon).toBe("number")
+			}
+		}
+	})
+})

packages/aris-source-tfl/src/tfl-source.ts

@@ -0,0 +1,164 @@
+import type { Context, FeedSource } from "@aris/core"
+
+import { contextValue } from "@aris/core"
+import { LocationKey } from "@aris/source-location"
+
+import type {
+	ITflApi,
+	StationLocation,
+	TflAlertData,
+	TflAlertFeedItem,
+	TflAlertSeverity,
+	TflLineId,
+	TflSourceOptions,
+} from "./types.ts"
+
+import { TflApi } from "./tfl-api.ts"
+
+const SEVERITY_PRIORITY: Record<TflAlertSeverity, number> = {
+	closure: 1.0,
+	"major-delays": 0.8,
+	"minor-delays": 0.6,
+}
+
+/**
+ * A FeedSource that provides TfL (Transport for London) service alerts.
+ *
+ * Depends on location source for proximity-based sorting. Produces feed items
+ * for tube, overground, and Elizabeth line disruptions.
+ *
+ * @example
+ * ```ts
+ * const tflSource = new TflSource({
+ *   apiKey: process.env.TFL_API_KEY!,
+ *   lines: ["northern", "victoria", "jubilee"],
+ * })
+ *
+ * const engine = new FeedEngine()
+ *   .register(locationSource)
+ *   .register(tflSource)
+ *
+ * const { items } = await engine.refresh()
+ * ```
+ */
+export class TflSource implements FeedSource<TflAlertFeedItem> {
+	static readonly DEFAULT_LINES_OF_INTEREST: readonly TflLineId[] = [
+		"bakerloo",
+		"central",
+		"circle",
+		"district",
+		"hammersmith-city",
+		"jubilee",
+		"metropolitan",
+		"northern",
+		"piccadilly",
+		"victoria",
+		"waterloo-city",
+		"lioness",
+		"mildmay",
+		"windrush",
+		"weaver",
+		"suffragette",
+		"liberty",
+		"elizabeth",
+	]
+
+	readonly id = "tfl"
+	readonly dependencies = ["location"]
+
+	private readonly client: ITflApi
+	private lines: TflLineId[]
+
+	constructor(options: TflSourceOptions) {
+		if (!options.client && !options.apiKey) {
+			throw new Error("Either client or apiKey must be provided")
+		}
+		this.client = options.client ?? new TflApi(options.apiKey!)
+		this.lines = options.lines ?? [...TflSource.DEFAULT_LINES_OF_INTEREST]
+	}
+
+	/**
+	 * Update the set of monitored lines. Takes effect on the next fetchItems call.
+	 */
+	setLinesOfInterest(lines: TflLineId[]): void {
+		this.lines = lines
+	}
+
+	async fetchItems(context: Context): Promise<TflAlertFeedItem[]> {
+		const [statuses, stations] = await Promise.all([
+			this.client.fetchLineStatuses(this.lines),
+			this.client.fetchStations(),
+		])
+
+		const location = contextValue(context, LocationKey)
+
+		const items: TflAlertFeedItem[] = statuses.map((status) => {
+			const closestStationDistance = location
+				? findClosestStationDistance(status.lineId, stations, location.lat, location.lng)
+				: null
+
+			const data: TflAlertData = {
+				line: status.lineId,
+				lineName: status.lineName,
+				severity: status.severity,
+				description: status.description,
+				closestStationDistance,
+			}
+
+			return {
+				id: `tfl-alert-${status.lineId}-${status.severity}`,
+				type: "tfl-alert",
+				priority: SEVERITY_PRIORITY[status.severity],
+				timestamp: context.time,
+				data,
+			}
+		})
+
+		// Sort by severity (desc), then by proximity (asc) if location available
+		items.sort((a, b) => {
+			if (b.priority !== a.priority) {
+				return b.priority - a.priority
+			}
+			if (a.data.closestStationDistance !== null && b.data.closestStationDistance !== null) {
+				return a.data.closestStationDistance - b.data.closestStationDistance
+			}
+			return 0
+		})
+
+		return items
+	}
+}
+
+function haversineDistance(lat1: number, lng1: number, lat2: number, lng2: number): number {
+	const R = 6371 // Earth's radius in km
+	const dLat = ((lat2 - lat1) * Math.PI) / 180
+	const dLng = ((lng2 - lng1) * Math.PI) / 180
+	const a =
+		Math.sin(dLat / 2) * Math.sin(dLat / 2) +
+		Math.cos((lat1 * Math.PI) / 180) *
+			Math.cos((lat2 * Math.PI) / 180) *
+			Math.sin(dLng / 2) *
+			Math.sin(dLng / 2)
+	const c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a))
+	return R * c
+}
+
+function findClosestStationDistance(
+	lineId: TflLineId,
+	stations: StationLocation[],
+	userLat: number,
+	userLng: number,
+): number | null {
+	const lineStations = stations.filter((s) => s.lines.includes(lineId))
+	if (lineStations.length === 0) return null
+
+	let minDistance = Infinity
+	for (const station of lineStations) {
+		const distance = haversineDistance(userLat, userLng, station.lat, station.lng)
+		if (distance < minDistance) {
+			minDistance = distance
+		}
+	}
+
+	return minDistance
+}

packages/aris-source-tfl/src/types.ts

@@ -0,0 +1,50 @@
+import type { FeedItem } from "@aris/core"
+
+import type { TflLineId } from "./tfl-api.ts"
+
+export type { TflLineId } from "./tfl-api.ts"
+
+export const TflAlertSeverity = {
+	MinorDelays: "minor-delays",
+	MajorDelays: "major-delays",
+	Closure: "closure",
+} as const
+
+export type TflAlertSeverity = (typeof TflAlertSeverity)[keyof typeof TflAlertSeverity]
+
+export interface TflAlertData extends Record<string, unknown> {
+	line: TflLineId
+	lineName: string
+	severity: TflAlertSeverity
+	description: string
+	closestStationDistance: number | null
+}
+
+export type TflAlertFeedItem = FeedItem<"tfl-alert", TflAlertData>
+
+export interface TflSourceOptions {
+	apiKey?: string
+	client?: ITflApi
+	/** Lines to monitor. Defaults to all lines. */
+	lines?: TflLineId[]
+}
+
+export interface StationLocation {
+	id: string
+	name: string
+	lat: number
+	lng: number
+	lines: TflLineId[]
+}
+
+export interface ITflApi {
+	fetchLineStatuses(lines?: TflLineId[]): Promise<TflLineStatus[]>
+	fetchStations(): Promise<StationLocation[]>
+}
+
+export interface TflLineStatus {
+	lineId: TflLineId
+	lineName: string
+	severity: TflAlertSeverity
+	description: string
+}