mirror of
https://github.com/kennethnym/aris.git
synced 2026-03-23 18:41:17 +00:00
feat(backend): add PUT /api/sources/:sourceId (#87)
Add a PUT endpoint that inserts or fully replaces a user's source config. Unlike PATCH (which deep-merges and requires an existing row), PUT requires both `enabled` and `config`, performs an upsert via INSERT ... ON CONFLICT DO UPDATE, and replaces config entirely. - Add `upsertConfig` to user-sources data layer - Add `upsertSourceConfig` to UserSessionManager - Add `addSource` to UserSession for new source registration - 12 new tests covering insert, replace, validation, and session refresh Co-authored-by: Ona <no-reply@ona.com>
This commit is contained in:
@@ -136,6 +136,52 @@ 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) {
|
||||||
|
const validated = provider.configSchema(data.config)
|
||||||
|
if (validated instanceof type.errors) {
|
||||||
|
throw new InvalidSourceConfigError(sourceId, validated.summary)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
await sources(this.db, userId).upsertConfig(sourceId, {
|
||||||
|
enabled: data.enabled,
|
||||||
|
config: data.config,
|
||||||
|
})
|
||||||
|
|
||||||
|
const session = this.sessions.get(userId)
|
||||||
|
if (session) {
|
||||||
|
if (!data.enabled) {
|
||||||
|
session.removeSource(sourceId)
|
||||||
|
} else {
|
||||||
|
const source = await provider.feedSourceForUser(userId, data.config)
|
||||||
|
if (session.hasSource(sourceId)) {
|
||||||
|
session.replaceSource(sourceId, source)
|
||||||
|
} else {
|
||||||
|
session.addSource(source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Replaces a provider and updates all active sessions.
|
* Replaces a provider and updates all active sessions.
|
||||||
* The new provider must have the same sourceId as an existing one.
|
* The new provider must have the same sourceId as an existing one.
|
||||||
|
|||||||
@@ -73,6 +73,32 @@ export class UserSession {
|
|||||||
return this.sources.has(sourceId)
|
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.
|
* Replaces a source in the engine and invalidates all caches.
|
||||||
* Stops and restarts the engine to re-establish reactive subscriptions.
|
* Stops and restarts the engine to re-establish reactive subscriptions.
|
||||||
|
|||||||
@@ -91,6 +91,20 @@ function createInMemoryStore() {
|
|||||||
existing.config = update.config as Record<string, unknown>
|
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>,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
},
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
@@ -124,6 +138,14 @@ 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
|
// Tests
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@@ -342,3 +364,195 @@ describe("PATCH /api/sources/:sourceId", () => {
|
|||||||
expect(row!.enabled).toBe(false)
|
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 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("accepts location source with arbitrary config (no 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(204)
|
||||||
|
const row = activeStore.rows.get(`${MOCK_USER_ID}:aelis.location`)
|
||||||
|
expect(row).toBeDefined()
|
||||||
|
expect(row!.config).toEqual({ something: "value" })
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|||||||
@@ -24,6 +24,11 @@ const UpdateSourceConfigRequestBody = type({
|
|||||||
"config?": "unknown",
|
"config?": "unknown",
|
||||||
})
|
})
|
||||||
|
|
||||||
|
const ReplaceSourceConfigRequestBody = type({
|
||||||
|
enabled: "boolean",
|
||||||
|
config: "unknown",
|
||||||
|
})
|
||||||
|
|
||||||
export function registerSourcesHttpHandlers(
|
export function registerSourcesHttpHandlers(
|
||||||
app: Hono,
|
app: Hono,
|
||||||
{ sessionManager, authSessionMiddleware }: SourcesHttpHandlersDeps,
|
{ sessionManager, authSessionMiddleware }: SourcesHttpHandlersDeps,
|
||||||
@@ -34,6 +39,7 @@ export function registerSourcesHttpHandlers(
|
|||||||
})
|
})
|
||||||
|
|
||||||
app.patch("/api/sources/:sourceId", inject, authSessionMiddleware, handleUpdateSource)
|
app.patch("/api/sources/:sourceId", inject, authSessionMiddleware, handleUpdateSource)
|
||||||
|
app.put("/api/sources/:sourceId", inject, authSessionMiddleware, handleReplaceSource)
|
||||||
}
|
}
|
||||||
|
|
||||||
async function handleUpdateSource(c: Context<Env>) {
|
async function handleUpdateSource(c: Context<Env>) {
|
||||||
@@ -83,3 +89,49 @@ async function handleUpdateSource(c: Context<Env>) {
|
|||||||
|
|
||||||
return c.body(null, 204)
|
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 parsed = ReplaceSourceConfigRequestBody(body)
|
||||||
|
if (parsed instanceof type.errors) {
|
||||||
|
return c.json({ error: parsed.summary }, 400)
|
||||||
|
}
|
||||||
|
|
||||||
|
const { enabled, config } = parsed
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
|||||||
@@ -72,6 +72,29 @@ 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. */
|
/** Updates the encrypted credentials for a source. Throws if the source row doesn't exist. */
|
||||||
async updateCredentials(sourceId: string, credentials: Buffer) {
|
async updateCredentials(sourceId: string, credentials: Buffer) {
|
||||||
const rows = await db
|
const rows = await db
|
||||||
|
|||||||
Reference in New Issue
Block a user