322 lines
13 KiB
TypeScript
322 lines
13 KiB
TypeScript
// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
|
|
// The Convex profile's live stream in the web process (apps/web/src/server/convex-live.ts) and the write
|
|
// txid of its procedure transport (convex-caller.ts, umbrella TS6.3), against fake Convex clients: the
|
|
// log source reads `live:head`/`live:changes` and subscribes to `live:head` once per organization while a
|
|
// stream of it is open; `live:open`'s refusals are the live problems the Postgres profile answers (401
|
|
// signed out, 404 not reached) and its denials pass through; the heartbeat asks `live:recheck`; a write
|
|
// is called with `conusaiLiveTxid` and its envelope's txid becomes the request's `ConusAI-Txid`.
|
|
// The Convex functions themselves (convex/live.ts) run in a deployment; this file does not start one.
|
|
|
|
import { captureTxids } from "@conusai-kit/data-kysely"
|
|
import { convexLiveLog, createLiveServer, LiveProblem, type OpenRequest } from "@conusai-kit/live-server"
|
|
import { getFunctionName, type UserIdentityAttributes } from "convex/server"
|
|
import { ConvexError } from "convex/values"
|
|
import { describe, expect, it, vi } from "vitest"
|
|
|
|
/** Every call a fake ConvexHttpClient of convex-caller.ts received: [kind, function, args]. */
|
|
const httpCalls: Array<[string, string, Record<string, unknown>]> = []
|
|
/** What that client answers; a test sets `answer` for the calls it makes. */
|
|
const http: { answer: (kind: string, name: string, args: Record<string, unknown>) => unknown } = { answer: () => null }
|
|
|
|
vi.mock("convex/browser", () => {
|
|
class FakeHttpClient {
|
|
setAdminAuth(): void {}
|
|
async query(reference: never, args: Record<string, unknown>) {
|
|
httpCalls.push(["query", getFunctionName(reference), args])
|
|
return http.answer("query", getFunctionName(reference), args)
|
|
}
|
|
async mutation(reference: never, args: Record<string, unknown>) {
|
|
httpCalls.push(["mutation", getFunctionName(reference), args])
|
|
return http.answer("mutation", getFunctionName(reference), args)
|
|
}
|
|
async action(reference: never, args: Record<string, unknown>) {
|
|
httpCalls.push(["action", getFunctionName(reference), args])
|
|
return http.answer("action", getFunctionName(reference), args)
|
|
}
|
|
}
|
|
return { ConvexHttpClient: FakeHttpClient, ConvexClient: class {} }
|
|
})
|
|
|
|
const { convexLiveAuthorize, convexLiveSource } = await import("../apps/web/src/server/convex-live")
|
|
const { unwrapLiveTxid } = await import("../apps/web/src/server/convex-caller")
|
|
const { createConvexCaller } = await import("../apps/web/src/server/convex-caller") // kit:notes
|
|
|
|
const ORG = "org-1"
|
|
const HEAD = { txid: 4, prunedThrough: 0 }
|
|
|
|
/** A fake subscribing client: what is subscribed now, and a push of a new head. */
|
|
function fakeWatcher() {
|
|
const subscriptions = new Map<number, { orgId: string; callback: (value: unknown) => unknown }>()
|
|
let next = 0
|
|
const client = {
|
|
opened: 0,
|
|
closed: 0,
|
|
subscriptions,
|
|
onUpdate(reference: never, args: never, callback: (value: unknown) => unknown) {
|
|
expect(getFunctionName(reference)).toBe("live:head")
|
|
const id = next++
|
|
subscriptions.set(id, { orgId: (args as { orgId: string }).orgId, callback })
|
|
callback(HEAD)
|
|
return () => subscriptions.delete(id)
|
|
},
|
|
async close() {
|
|
client.closed += 1
|
|
},
|
|
push(orgId: string, head: unknown) {
|
|
for (const entry of subscriptions.values()) if (entry.orgId === orgId) entry.callback(head)
|
|
},
|
|
}
|
|
return client
|
|
}
|
|
|
|
function fakeSource() {
|
|
const watcher = fakeWatcher()
|
|
const reads: Array<[string, Record<string, unknown>]> = []
|
|
const source = convexLiveSource({
|
|
admin: {
|
|
async query(reference: never, args: never) {
|
|
const name = getFunctionName(reference)
|
|
reads.push([name, args as Record<string, unknown>])
|
|
if (name === "live:head") return HEAD
|
|
if (name === "live:changes") return { rows: [], head: HEAD, more: false }
|
|
throw new Error(`unexpected ${name}`)
|
|
},
|
|
},
|
|
watcher: () => {
|
|
watcher.opened += 1
|
|
return watcher
|
|
},
|
|
})
|
|
return { source, watcher, reads }
|
|
}
|
|
|
|
const identityOf = (userId: string): UserIdentityAttributes => ({
|
|
issuer: "https://conusai.internal/bff",
|
|
subject: userId,
|
|
tokenIdentifier: `https://conusai.internal/bff|${userId}`,
|
|
})
|
|
|
|
const OPEN: OpenRequest = {
|
|
protocol: "clp/1",
|
|
after: null,
|
|
entities: ["notes.note", "notes.secret"],
|
|
topics: [{ ws: "main", project: "notes" }],
|
|
}
|
|
|
|
const request = () => new Request(`http://app.test/api/v1/orgs/acme/live`, { method: "POST" })
|
|
|
|
const problem = (status: number, code: string) =>
|
|
new ConvexError({ type: `/problems/${code}`, title: code, status, code })
|
|
|
|
/** The authorize of convex-live.ts over a fake caller and a fake `live:open`/`live:recheck`. */
|
|
function authorizeWith(answer: (name: string, args: Record<string, unknown>) => unknown, userId = () => "u-ada") {
|
|
const calls: Array<[string, UserIdentityAttributes, Record<string, unknown>]> = []
|
|
const warn = vi.fn()
|
|
const authorize = convexLiveAuthorize({
|
|
caller: async () => ({ userId: userId(), session: "s-1", identity: identityOf(userId()) }),
|
|
clientFor: (identity) => ({
|
|
async query(reference: never, args: never) {
|
|
const name = getFunctionName(reference)
|
|
calls.push([name, identity, args as Record<string, unknown>])
|
|
return answer(name, args as Record<string, unknown>)
|
|
},
|
|
}),
|
|
orgSlugOf: () => "acme",
|
|
warn,
|
|
})
|
|
return { authorize, calls, warn }
|
|
}
|
|
|
|
const granted = {
|
|
org: ORG,
|
|
topics: [{ ws: "main", project: "notes", ids: { org: ORG, ws: "w-1", project: "p-1" } }],
|
|
entities: ["notes.note"],
|
|
denied: [{ entity: "notes.secret", reason: "unknown" }],
|
|
}
|
|
|
|
describe("the Convex live log source", () => {
|
|
it("reads the head and the changes with the admin client, and positions them as convexLiveLog does", async () => {
|
|
const { source, reads } = fakeSource()
|
|
const log = convexLiveLog({ source })
|
|
expect(await log.head(ORG)).toEqual({ position: { txid: 5n, seq: 0n }, w: 5n })
|
|
const read = await log.read(ORG, { txid: 3n, seq: 2n }, 50)
|
|
expect(read).toMatchObject({ final: [], more: false, w: 5n, position: { txid: 5n, seq: 0n } })
|
|
expect(reads).toEqual([
|
|
["live:head", { orgId: ORG }],
|
|
["live:changes", { orgId: ORG, after: { txid: 3, seq: 2 }, limit: 50 }],
|
|
])
|
|
})
|
|
|
|
it("subscribes to an organization's head once while its streams are open, and ends it with the last one", async () => {
|
|
const { source, watcher } = fakeSource()
|
|
const log = convexLiveLog({ source })
|
|
const woken: Array<string | null> = []
|
|
log.onWake((org) => woken.push(org))
|
|
const first = log.watch?.(ORG)
|
|
const second = log.watch?.(ORG)
|
|
const other = log.watch?.("org-2")
|
|
// One websocket client, one subscription per organization.
|
|
expect(watcher.opened).toBe(1)
|
|
expect([...watcher.subscriptions.values()].map(({ orgId }) => orgId)).toEqual([ORG, "org-2"])
|
|
woken.length = 0
|
|
watcher.push(ORG, { txid: 5, prunedThrough: 0 })
|
|
expect(woken).toEqual([ORG])
|
|
first?.()
|
|
first?.()
|
|
expect(watcher.subscriptions.size).toBe(2)
|
|
second?.()
|
|
expect([...watcher.subscriptions.values()].map(({ orgId }) => orgId)).toEqual(["org-2"])
|
|
other?.()
|
|
expect(watcher.subscriptions.size).toBe(0)
|
|
await log.close()
|
|
expect(watcher.closed).toBe(1)
|
|
})
|
|
|
|
it("is what a live server's streams subscribe through: the subscription follows the open streams", async () => {
|
|
const { source, watcher } = fakeSource()
|
|
const { authorize } = authorizeWith(() => granted)
|
|
const server = createLiveServer({
|
|
engine: "convex",
|
|
log: convexLiveLog({ source }),
|
|
authorize,
|
|
heartbeatMs: 60_000,
|
|
})
|
|
const open = async () => {
|
|
const abort = new AbortController()
|
|
const response = await server.handle(
|
|
new Request("http://app.test/api/v1/orgs/acme/live", {
|
|
method: "POST",
|
|
headers: { "content-type": "application/json" },
|
|
body: JSON.stringify({ protocol: "clp/1", after: null, entities: ["notes.note"], topics: [] }),
|
|
signal: abort.signal,
|
|
}),
|
|
)
|
|
expect(response.status).toBe(200)
|
|
const reader = response.body?.getReader()
|
|
const first = new TextDecoder().decode((await reader?.read())?.value)
|
|
expect(first).toContain("event: hello")
|
|
return { abort, reader }
|
|
}
|
|
const a = await open()
|
|
const b = await open()
|
|
expect(server.open).toBe(2)
|
|
expect(watcher.subscriptions.size).toBe(1)
|
|
a.abort.abort()
|
|
expect(watcher.subscriptions.size).toBe(1)
|
|
b.abort.abort()
|
|
expect(watcher.subscriptions.size).toBe(0)
|
|
expect(server.open).toBe(0)
|
|
await server.close()
|
|
expect(watcher.closed).toBe(1)
|
|
})
|
|
})
|
|
|
|
describe("the Convex profile's stream authorization", () => {
|
|
it("asks live:open as the caller and passes its grants and denials through", async () => {
|
|
const { authorize, calls } = authorizeWith(() => granted)
|
|
const grant = await authorize(request(), {
|
|
...OPEN,
|
|
topics: [{ ws: "main", project: "notes", extra: "dropped" } as never],
|
|
})
|
|
expect(grant.org).toBe(ORG)
|
|
expect(grant.session).toBe("s-1")
|
|
expect(grant.granted).toEqual({ entities: ["notes.note"], topics: granted.topics })
|
|
expect(grant.denied).toEqual([{ entity: "notes.secret", reason: "unknown" }])
|
|
expect(calls).toEqual([
|
|
[
|
|
"live:open",
|
|
identityOf("u-ada"),
|
|
{ org: "acme", topics: [{ ws: "main", project: "notes" }], entities: ["notes.note", "notes.secret"] },
|
|
],
|
|
])
|
|
})
|
|
|
|
it("answers live:open's refusals as the Postgres profile does: 401 signed out, 404 not reached", async () => {
|
|
for (const [status, type] of [
|
|
[401, "urn:conusai:live:unauthenticated"],
|
|
[404, "urn:conusai:live:not-found"],
|
|
] as const) {
|
|
const { authorize } = authorizeWith(() => {
|
|
throw problem(status, status === 401 ? "unauthenticated" : "not_found")
|
|
})
|
|
const refused = await authorize(request(), OPEN).catch((error: unknown) => error)
|
|
expect(refused).toBeInstanceOf(LiveProblem)
|
|
expect(refused).toMatchObject({ status, type })
|
|
}
|
|
// Another problem of the kernel's stays the kernel's (serveLive answers it as a procedure's).
|
|
const { authorize } = authorizeWith(() => {
|
|
throw problem(429, "rate_limited")
|
|
})
|
|
await expect(authorize(request(), OPEN)).rejects.toMatchObject({ status: 429, code: "rate_limited" })
|
|
})
|
|
|
|
it("re-checks every heartbeat with live:recheck, and revokes a changed caller, a lost membership or a lost read", async () => {
|
|
let recheck: () => unknown = () => null
|
|
let user = "u-ada"
|
|
const { authorize, calls, warn } = authorizeWith(
|
|
(name) => (name === "live:open" ? granted : recheck()),
|
|
() => user,
|
|
)
|
|
const grant = await authorize(request(), OPEN)
|
|
expect(await grant.recheck()).toBeNull()
|
|
expect(calls.at(-1)).toEqual([
|
|
"live:recheck",
|
|
identityOf("u-ada"),
|
|
// The entities the stream was granted, so the deployment re-checks each one's read permission.
|
|
{ org: "acme", topics: [{ ws: "main", project: "notes" }], granted: ["notes.note"] },
|
|
])
|
|
recheck = () => "grants_changed"
|
|
expect(await grant.recheck()).toBe("grants_changed")
|
|
recheck = () => "membership"
|
|
expect(await grant.recheck()).toBe("membership")
|
|
recheck = () => {
|
|
throw problem(404, "not_found")
|
|
}
|
|
expect(await grant.recheck()).toBe("membership")
|
|
recheck = () => {
|
|
throw problem(401, "unauthenticated")
|
|
}
|
|
expect(await grant.recheck()).toBe("session")
|
|
recheck = () => {
|
|
throw new Error("the deployment is unreachable")
|
|
}
|
|
expect(await grant.recheck()).toBeNull()
|
|
expect(warn).toHaveBeenCalledWith("live: the re-check failed: Error")
|
|
user = "u-grace"
|
|
expect(await grant.recheck()).toBe("session")
|
|
})
|
|
})
|
|
|
|
describe("the Convex profile's write txid", () => {
|
|
it("unwraps a tenant mutation's envelope and records its txid with the router's capture", async () => {
|
|
const written = await captureTxids(async () => unwrapLiveTxid({ value: { id: "n-1" }, txid: "12" }))
|
|
expect(written).toEqual({ value: { id: "n-1" }, txid: "12" })
|
|
const nothing = await captureTxids(async () => unwrapLiveTxid({ value: null, txid: null }))
|
|
expect(nothing).toEqual({ value: null, txid: null })
|
|
expect(() => unwrapLiveTxid({ id: "n-1" })).toThrow()
|
|
})
|
|
|
|
// kit:notes:begin
|
|
it("asks a write for its txid and answers it as ConusAI-Txid's capture; a read is called unchanged", async () => {
|
|
httpCalls.length = 0
|
|
http.answer = (kind) => (kind === "mutation" ? { value: { noteId: "n-1" }, txid: "7" } : { items: [] })
|
|
const caller = createConvexCaller({
|
|
url: "http://convex.test",
|
|
adminKey: "admin-key",
|
|
principal: { actor: { userId: "u-ada" }, method: "bearer" } as never,
|
|
objectStore: {} as never,
|
|
identity: identityOf("u-ada"),
|
|
})
|
|
const input = { org: "acme", workspace: "main", project: "notes", title: "A" }
|
|
const write = await captureTxids(() => caller.callByName("notes.note.create", input))
|
|
expect(write).toEqual({ value: { noteId: "n-1" }, txid: "7" })
|
|
const read = await captureTxids(() => caller.callByName("notes.note.list", { org: "acme" }))
|
|
expect(read).toEqual({ value: { items: [] }, txid: null })
|
|
expect(httpCalls).toEqual([
|
|
["mutation", "generated/notes:notes_note_create", { input, conusaiLiveTxid: true }],
|
|
["query", "generated/notes:notes_note_list", { input: { org: "acme" } }],
|
|
])
|
|
})
|
|
// kit:notes:end
|
|
})
|