Merge pull request #21 from kennethnym/feat/feed-engine-service

feat(backend): add FeedEngineService

apps/aris-backend/package.json

@@ -14,10 +14,10 @@
 		"@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"

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/location/router.ts

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

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

@@ -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>()
 
 	/**

apps/aris-backend/src/server.ts

@@ -2,22 +2,33 @@ 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"
 
-const app = new Hono()
+function main() {
+	const locationService = new LocationService()
 
-app.get("/health", (c) => c.json({ status: "ok" }))
+	const trpcRouter = createTRPCRouter({ locationService })
 
-registerAuthHandlers(app)
+	const app = new Hono()
 
-app.use(
+	app.get("/health", (c) => c.json({ status: "ok" }))
+
+	registerAuthHandlers(app)
+
+	app.use(
 		"/trpc/*",
 		trpcServer({
-		router: appRouter,
+			router: trpcRouter,
 			createContext,
 		}),
-)
+	)
+
+	return app
+}
+
+const app = main()
 
 export default {
 	port: 3000,

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

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

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

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

bun.lock

@@ -22,10 +22,10 @@
         "@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 +204,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=="],
   }
 }