Compare commits

..

6 Commits

Author SHA1 Message Date
54e4b0dcf7 feat(backend): add TflService
Manages per-user TflSource instances with individual line
configuration. Implements FeedSourceProvider so it can be
wired into FeedEngineService.

Adds TflSource.setLines() so line config can be mutated
in place, keeping engine references valid.

Also exports ITflApi from @aris/source-tfl for testability.

Co-authored-by: Ona <no-reply@ona.com>
2026-02-13 19:38:21 +00:00
d1102fe1ac Merge pull request #21 from kennethnym/feat/feed-engine-service
feat(backend): add FeedEngineService
2026-01-25 23:16:08 +00:00
db0c57f04b feat(backend): add FeedEngineService
Manages FeedEngine instances per user with auto-registration of
sources from FeedSourceProvider implementations.

- Add FeedSourceProvider interface
- Add FeedEngineService with providers array injection
- Update LocationService to implement FeedSourceProvider

Co-authored-by: Ona <no-reply@ona.com>
2026-01-25 23:12:48 +00:00
9e3fe2ea16 Merge pull request #20 from kennethnym/feat/location-router
feat(backend): add location router with tRPC factory pattern
2026-01-25 23:00:03 +00:00
949b7c8571 feat(backend): add location router with tRPC factory pattern
- Add createLocationRouter with location.update mutation
- Refactor tRPC to factory pattern (createTRPC, createTRPCRouter)
- Protected procedure by default (all routes require auth)
- Replace zod with arktype for input validation
- Wire location router in main() with dependency injection

Co-authored-by: Ona <no-reply@ona.com>
2026-01-25 22:58:32 +00:00
bd6cc3c963 Merge pull request #19 from kennethnym/feat/trpc
feat(backend): add tRPC with Hono adapter
2026-01-25 22:26:41 +00:00
14 changed files with 666 additions and 49 deletions

View File

@@ -11,13 +11,14 @@
"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",
"pg": "^8",
"zod": "^3"
"pg": "^8"
},
"devDependencies": {
"@types/pg": "^8"

View File

@@ -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)
})
})

View File

@@ -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
}
}

View File

@@ -0,0 +1,33 @@
import { TRPCError } from "@trpc/server"
import { type } from "arktype"
import { UserNotFoundError } from "../lib/error.ts"
import type { TRPC } from "../trpc/router.ts"
import type { LocationService } from "./service.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
}
}),
})
}

View File

@@ -1,11 +1,13 @@
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 {
export class LocationService implements FeedSourceProvider {
private sources = new Map<string, LocationSource>()
/**

View File

@@ -2,8 +2,14 @@ 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 { appRouter } from "./trpc/router.ts"
import { createTRPCRouter } from "./trpc/router.ts"
function main() {
const locationService = new LocationService()
const trpcRouter = createTRPCRouter({ locationService })
const app = new Hono()
@@ -14,11 +20,16 @@ registerAuthHandlers(app)
app.use(
"/trpc/*",
trpcServer({
router: appRouter,
router: trpcRouter,
createContext,
}),
)
return app
}
const app = main()
export default {
port: 3000,
fetch: app.fetch,

View File

@@ -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")
})
})
})

View File

@@ -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)
}
}

View File

@@ -1,5 +1,47 @@
import { router } from "./trpc.ts"
import { initTRPC, TRPCError } from "@trpc/server"
export const appRouter = router({})
import { createLocationRouter } from "../location/router.ts"
import type { LocationService } from "../location/service.ts"
import type { Context } from "./context.ts"
export type AppRouter = typeof appRouter
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
}
export function createTRPCRouter({ locationService }: TRPCRouterDeps) {
const t = createTRPC()
return t.router({
location: createLocationRouter(t, { locationService }),
})
}
export type TRPCRouter = ReturnType<typeof createTRPCRouter>

View File

@@ -1,22 +0,0 @@
import { initTRPC, TRPCError } from "@trpc/server"
import type { Context } from "./context.ts"
const t = initTRPC.context<Context>().create()
export const router = t.router
export const publicProcedure = t.procedure
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,
},
})
})
export const protectedProcedure = t.procedure.use(isAuthed)

View File

@@ -19,13 +19,14 @@
"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",
"pg": "^8",
"zod": "^3",
},
"devDependencies": {
"@types/pg": "^8",
@@ -204,12 +205,6 @@
"xtend": ["xtend@4.0.2", "", {}, "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ=="],
"zod": ["zod@3.25.76", "", {}, "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ=="],
"@better-auth/core/zod": ["zod@4.3.6", "", {}, "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg=="],
"better-auth/zod": ["zod@4.3.6", "", {}, "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg=="],
"better-call/zod": ["zod@4.3.6", "", {}, "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg=="],
"zod": ["zod@4.3.6", "", {}, "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg=="],
}
}

View File

@@ -2,6 +2,7 @@ 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,

View File

@@ -112,6 +112,49 @@ describe("TflSource", () => {
})
})
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 })

View File

@@ -42,18 +42,46 @@ const SEVERITY_PRIORITY: Record<TflAlertSeverity, number> = {
* ```
*/
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 readonly lines?: TflLineId[]
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
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[]> {