Compare commits

..

1 Commits

Author SHA1 Message Date
efe5232cf6 feat(backend): add PATCH /api/sources/:sourceId
Add endpoint for users to update their source config
and enabled state. Config is deep-merged with existing
values via lodash.merge and validated against the
provider's schema before persisting.

Co-authored-by: Ona <no-reply@ona.com>
2026-03-22 17:55:07 +00:00
7 changed files with 10 additions and 469 deletions

View File

@@ -5,10 +5,10 @@ import merge from "lodash.merge"
import type { Database } from "../db/index.ts"
import type { FeedEnhancer } from "../enhancement/enhance-feed.ts"
import type { FeedSourceProvider } from "./feed-source-provider.ts"
import { InvalidSourceConfigError, SourceNotFoundError } from "../sources/errors.ts"
import { sources } from "../sources/user-sources.ts"
import type { FeedSourceProvider } from "./feed-source-provider.ts"
import { UserSession } from "./user-session.ts"
export interface UserSessionManagerConfig {
@@ -104,14 +104,16 @@ export class UserSessionManager {
// read stale config. Use SELECT FOR UPDATE or atomic jsonb merge if
// this becomes a problem.
let mergedConfig: Record<string, unknown> | undefined
if (update.config !== undefined && provider.configSchema) {
if (update.config !== undefined) {
const existing = await sources(this.db, userId).find(sourceId)
const existingConfig = (existing?.config ?? {}) as Record<string, unknown>
mergedConfig = merge({}, existingConfig, update.config)
const validated = provider.configSchema(mergedConfig)
if (validated instanceof type.errors) {
throw new InvalidSourceConfigError(sourceId, validated.summary)
if (provider.configSchema) {
const validated = provider.configSchema(mergedConfig)
if (validated instanceof type.errors) {
throw new InvalidSourceConfigError(sourceId, validated.summary)
}
}
}
@@ -134,53 +136,6 @@ export class UserSessionManager {
}
}
/**
* Validates, persists, and upserts a user's source config, then
* refreshes the cached session. Unlike updateSourceConfig, this
* inserts a new row if one doesn't exist and fully replaces config
* (no merge).
*
* @throws {SourceNotFoundError} if the sourceId has no registered provider
* @throws {InvalidSourceConfigError} if config fails schema validation
*/
async upsertSourceConfig(
userId: string,
sourceId: string,
data: { enabled: boolean; config?: unknown },
): Promise<void> {
const provider = this.providers.get(sourceId)
if (!provider) {
throw new SourceNotFoundError(sourceId, userId)
}
if (provider.configSchema && data.config !== undefined) {
const validated = provider.configSchema(data.config)
if (validated instanceof type.errors) {
throw new InvalidSourceConfigError(sourceId, validated.summary)
}
}
const config = data.config ?? {}
await sources(this.db, userId).upsertConfig(sourceId, {
enabled: data.enabled,
config,
})
const session = this.sessions.get(userId)
if (session) {
if (!data.enabled) {
session.removeSource(sourceId)
} else {
const source = await provider.feedSourceForUser(userId, config)
if (session.hasSource(sourceId)) {
session.replaceSource(sourceId, source)
} else {
session.addSource(source)
}
}
}
}
/**
* Replaces a provider and updates all active sessions.
* The new provider must have the same sourceId as an existing one.

View File

@@ -73,32 +73,6 @@ export class UserSession {
return this.sources.has(sourceId)
}
/**
* Registers a new source in the engine and invalidates all caches.
* Stops and restarts the engine to establish reactive subscriptions.
*/
addSource(source: FeedSource): void {
if (this.sources.has(source.id)) {
throw new Error(`Cannot add source "${source.id}": already registered`)
}
const wasStarted = this.engine.isStarted()
if (wasStarted) {
this.engine.stop()
}
this.engine.register(source)
this.sources.set(source.id, source)
this.invalidateEnhancement()
this.enhancingPromise = null
if (wasStarted) {
this.engine.start()
}
}
/**
* Replaces a source in the engine and invalidates all caches.
* Stops and restarts the engine to re-establish reactive subscriptions.

View File

@@ -91,20 +91,6 @@ function createInMemoryStore() {
existing.config = update.config as Record<string, unknown>
}
},
async upsertConfig(sourceId: string, data: { enabled: boolean; config: unknown }) {
const existing = rows.get(key(userId, sourceId))
if (existing) {
existing.enabled = data.enabled
existing.config = data.config as Record<string, unknown>
} else {
rows.set(key(userId, sourceId), {
userId,
sourceId,
enabled: data.enabled,
config: (data.config ?? {}) as Record<string, unknown>,
})
}
},
}
},
}
@@ -138,14 +124,6 @@ function patch(app: Hono, sourceId: string, body: unknown) {
})
}
function put(app: Hono, sourceId: string, body: unknown) {
return app.request(`/api/sources/${sourceId}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body),
})
}
// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------
@@ -217,31 +195,6 @@ describe("PATCH /api/sources/:sourceId", () => {
expect(body.error).toContain("Invalid JSON")
})
test("returns 400 when request body contains unknown fields", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.weather")
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await patch(app, "aelis.weather", {
enabled: true,
unknownField: "hello",
})
expect(res.status).toBe(400)
})
test("returns 400 when weather config contains unknown fields", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.weather")
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await patch(app, "aelis.weather", {
config: { units: "metric", unknownField: "hello" },
})
expect(res.status).toBe(400)
})
test("returns 400 when weather config fails validation", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.weather")
@@ -365,7 +318,7 @@ describe("PATCH /api/sources/:sourceId", () => {
removeSpy.mockRestore()
})
test("returns 400 when config is provided for source without schema", async () => {
test("accepts location source with arbitrary config (no schema)", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.location")
const { app } = createApp([createStubProvider("aelis.location")], MOCK_USER_ID)
@@ -374,19 +327,7 @@ describe("PATCH /api/sources/:sourceId", () => {
config: { something: "value" },
})
expect(res.status).toBe(400)
})
test("returns 400 when empty config is provided for source without schema", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.location")
const { app } = createApp([createStubProvider("aelis.location")], MOCK_USER_ID)
const res = await patch(app, "aelis.location", {
config: {},
})
expect(res.status).toBe(400)
expect(res.status).toBe(204)
})
test("updates enabled on location source", async () => {
@@ -401,240 +342,3 @@ describe("PATCH /api/sources/:sourceId", () => {
expect(row!.enabled).toBe(false)
})
})
// ---------------------------------------------------------------------------
// PUT /api/sources/:sourceId
// ---------------------------------------------------------------------------
describe("PUT /api/sources/:sourceId", () => {
test("returns 401 without auth", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)])
const res = await put(app, "aelis.weather", { enabled: true, config: {} })
expect(res.status).toBe(401)
})
test("returns 404 for unknown source", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "unknown.source", { enabled: true, config: {} })
expect(res.status).toBe(404)
const body = (await res.json()) as { error: string }
expect(body.error).toContain("not found")
})
test("returns 400 for invalid JSON", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await app.request("/api/sources/aelis.weather", {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: "not json",
})
expect(res.status).toBe(400)
const body = (await res.json()) as { error: string }
expect(body.error).toContain("Invalid JSON")
})
test("returns 400 when enabled is missing", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", { config: {} })
expect(res.status).toBe(400)
})
test("returns 400 when config is missing", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", { enabled: true })
expect(res.status).toBe(400)
})
test("returns 400 when request body contains unknown fields", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", {
enabled: true,
config: { units: "metric" },
unknownField: "hello",
})
expect(res.status).toBe(400)
})
test("returns 400 when weather config contains unknown fields", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", {
enabled: true,
config: { units: "metric", unknownField: "hello" },
})
expect(res.status).toBe(400)
})
test("returns 400 when config fails schema validation", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", {
enabled: true,
config: { units: "invalid" },
})
expect(res.status).toBe(400)
})
test("returns 204 and inserts when row does not exist", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", {
enabled: true,
config: { units: "metric" },
})
expect(res.status).toBe(204)
const row = activeStore.rows.get(`${MOCK_USER_ID}:aelis.weather`)
expect(row).toBeDefined()
expect(row!.enabled).toBe(true)
expect(row!.config).toEqual({ units: "metric" })
})
test("returns 204 and fully replaces existing row", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.weather", {
enabled: true,
config: { units: "metric", hourlyLimit: 12 },
})
const { app } = createApp([createStubProvider("aelis.weather", weatherConfig)], MOCK_USER_ID)
const res = await put(app, "aelis.weather", {
enabled: false,
config: { units: "imperial" },
})
expect(res.status).toBe(204)
const row = activeStore.rows.get(`${MOCK_USER_ID}:aelis.weather`)
expect(row!.enabled).toBe(false)
// hourlyLimit should be gone — full replace, not merge
expect(row!.config).toEqual({ units: "imperial" })
})
test("refreshes source in active session after upsert", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.weather", {
config: { units: "metric" },
})
const { app, sessionManager } = createApp(
[createStubProvider("aelis.weather", weatherConfig)],
MOCK_USER_ID,
)
const session = await sessionManager.getOrCreate(MOCK_USER_ID)
const replaceSpy = spyOn(session, "replaceSource")
const res = await put(app, "aelis.weather", {
enabled: true,
config: { units: "imperial" },
})
expect(res.status).toBe(204)
expect(replaceSpy).toHaveBeenCalled()
replaceSpy.mockRestore()
})
test("removes source from session when disabled via upsert", async () => {
activeStore = createInMemoryStore()
activeStore.seed(MOCK_USER_ID, "aelis.weather", {
enabled: true,
config: { units: "metric" },
})
const { app, sessionManager } = createApp(
[createStubProvider("aelis.weather", weatherConfig)],
MOCK_USER_ID,
)
const session = await sessionManager.getOrCreate(MOCK_USER_ID)
const removeSpy = spyOn(session, "removeSource")
const res = await put(app, "aelis.weather", {
enabled: false,
config: { units: "metric" },
})
expect(res.status).toBe(204)
expect(removeSpy).toHaveBeenCalledWith("aelis.weather")
removeSpy.mockRestore()
})
test("adds source to active session when inserting a new source", async () => {
activeStore = createInMemoryStore()
// Seed a different source so the session can be created
activeStore.seed(MOCK_USER_ID, "aelis.location", { enabled: true })
const { app, sessionManager } = createApp(
[createStubProvider("aelis.location"), createStubProvider("aelis.weather", weatherConfig)],
MOCK_USER_ID,
)
// Create session — only has aelis.location
const session = await sessionManager.getOrCreate(MOCK_USER_ID)
expect(session.hasSource("aelis.weather")).toBe(false)
// PUT a new source that didn't exist before
const res = await put(app, "aelis.weather", {
enabled: true,
config: { units: "metric" },
})
expect(res.status).toBe(204)
expect(session.hasSource("aelis.weather")).toBe(true)
})
test("returns 400 when config is provided for source without schema", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.location")], MOCK_USER_ID)
const res = await put(app, "aelis.location", {
enabled: true,
config: { something: "value" },
})
expect(res.status).toBe(400)
})
test("returns 400 when empty config is provided for source without schema", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.location")], MOCK_USER_ID)
const res = await put(app, "aelis.location", {
enabled: true,
config: {},
})
expect(res.status).toBe(400)
})
test("returns 204 without config field for source without schema", async () => {
activeStore = createInMemoryStore()
const { app } = createApp([createStubProvider("aelis.location")], MOCK_USER_ID)
const res = await put(app, "aelis.location", {
enabled: true,
})
expect(res.status).toBe(204)
})
})

View File

@@ -20,22 +20,10 @@ interface SourcesHttpHandlersDeps {
}
const UpdateSourceConfigRequestBody = type({
"+": "reject",
"enabled?": "boolean",
"config?": "unknown",
})
const ReplaceSourceConfigRequestBody = type({
"+": "reject",
enabled: "boolean",
config: "unknown",
})
const ReplaceSourceConfigNoConfigRequestBody = type({
"+": "reject",
enabled: "boolean",
})
export function registerSourcesHttpHandlers(
app: Hono,
{ sessionManager, authSessionMiddleware }: SourcesHttpHandlersDeps,
@@ -46,7 +34,6 @@ export function registerSourcesHttpHandlers(
})
app.patch("/api/sources/:sourceId", inject, authSessionMiddleware, handleUpdateSource)
app.put("/api/sources/:sourceId", inject, authSessionMiddleware, handleReplaceSource)
}
async function handleUpdateSource(c: Context<Env>) {
@@ -76,10 +63,6 @@ async function handleUpdateSource(c: Context<Env>) {
return c.json({ error: parsed.summary }, 400)
}
if (!provider.configSchema && "config" in parsed) {
return c.json({ error: `Source "${sourceId}" does not accept config` }, 400)
}
const { enabled, config: newConfig } = parsed
const user = c.get("user")!
@@ -100,53 +83,3 @@ async function handleUpdateSource(c: Context<Env>) {
return c.body(null, 204)
}
async function handleReplaceSource(c: Context<Env>) {
const sourceId = c.req.param("sourceId")
if (!sourceId) {
return c.body(null, 404)
}
const sessionManager = c.get("sessionManager")
const provider = sessionManager.getProvider(sourceId)
if (!provider) {
return c.json({ error: `Source "${sourceId}" not found` }, 404)
}
let body: unknown
try {
body = await c.req.json()
} catch {
return c.json({ error: "Invalid JSON" }, 400)
}
const schema = provider.configSchema
? ReplaceSourceConfigRequestBody
: ReplaceSourceConfigNoConfigRequestBody
const parsed = schema(body)
if (parsed instanceof type.errors) {
return c.json({ error: parsed.summary }, 400)
}
const { enabled } = parsed
const config = "config" in parsed ? parsed.config : undefined
const user = c.get("user")!
try {
await sessionManager.upsertSourceConfig(user.id, sourceId, {
enabled,
config,
})
} catch (err) {
if (err instanceof SourceNotFoundError) {
return c.json({ error: err.message }, 404)
}
if (err instanceof InvalidSourceConfigError) {
return c.json({ error: err.message }, 400)
}
throw err
}
return c.body(null, 204)
}

View File

@@ -72,29 +72,6 @@ export function sources(db: Database, userId: string) {
}
},
/** Inserts a new user source row or fully replaces enabled/config on an existing one. */
async upsertConfig(sourceId: string, data: { enabled: boolean; config: unknown }) {
const now = new Date()
await db
.insert(userSources)
.values({
userId,
sourceId,
enabled: data.enabled,
config: data.config,
createdAt: now,
updatedAt: now,
})
.onConflictDoUpdate({
target: [userSources.userId, userSources.sourceId],
set: {
enabled: data.enabled,
config: data.config,
updatedAt: now,
},
})
},
/** Updates the encrypted credentials for a source. Throws if the source row doesn't exist. */
async updateCredentials(sourceId: string, credentials: Buffer) {
const rows = await db

View File

@@ -8,7 +8,6 @@ export type TflSourceProviderOptions =
| { apiKey?: never; client: ITflApi }
export const tflConfig = type({
"+": "reject",
"lines?": "string[]",
})

View File

@@ -9,7 +9,6 @@ export interface WeatherSourceProviderOptions {
}
export const weatherConfig = type({
"+": "reject",
"units?": "'metric' | 'imperial'",
"hourlyLimit?": "number",
"dailyLimit?": "number",