// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0 // Live collections on the Convex profile (umbrella TS6, the Convex mapping of `@conusai-kit/ui/data`). // The browser keeps the facade it has on every profile -- `liveCollection`, `useLive`, `useWrite` over the // app's CLP/1 stream, `POST /api/v1/orgs/{org}/live` -- and the web process serves that stream from these // functions (apps/web/src/server/convex-live.ts): // // open who may stream what, decided HERE, over this deployment's rows, with the caller's verified // identity (the BFF attaches it with the admin key) and the kernel's tenancy: the same rule as // the Postgres profile's (`decideLiveGrants`). Another organization, a workspace the caller does // not resolve, or no identity is refused; a collection the caller may not read is denied. // The BFF answers a refusal as the Postgres profile's live problem (401, 404). // recheck each heartbeat's question, under the caller's identity: null while the caller still reaches // the organization and every topic and still reads every granted collection, else why the // stream ends (`grants_changed` when a read permission was lost) // head the organization's log head; the web subscribes to it as a Convex reactive query, so a // committed write wakes the organization's streams at once (internal: admin key only) // changes the organization's changes after a cursor, with the head of the same snapshot (internal) // prune the hourly cron: changes older than 24 hours go, as on the Postgres profile // // The log holds ids only; a browser reads each changed row back through the collection's batch-get // procedure, a generated tenant query under its own permission and the repository's tenant predicate. import { AppError, toProblem } from "@conusai-kit/core" import { LIVE_RETENTION_MS, liveLogHead, pruneLiveLog, readLiveLog } from "@conusai-kit/data-convex" import { decideLiveGrants, recheckLiveScopes } from "@conusai-kit/live-server/grants" import { ConvexError, type Value, v } from "convex/values" import { internalMutation, internalQuery, query } from "./_generated/server" import { callContextOf, registry, tenancyOver } from "./lib/runtime" const topic = v.object({ ws: v.string(), project: v.optional(v.string()) }) /** A refusal as the BFF reads it back: the live problem's own fields, or the kernel's problem. */ function refusal(error: unknown): ConvexError { if (error instanceof AppError) return new ConvexError(toProblem(error) as unknown as Value) const problem = error as { status?: unknown; type?: unknown; title?: unknown; name?: unknown } | null if (problem?.name === "LiveProblem" && typeof problem.status === "number") { return new ConvexError({ type: String(problem.type), title: String(problem.title), status: problem.status, code: problem.status === 401 ? "unauthenticated" : problem.status === 404 ? "not_found" : "live", }) } return new ConvexError(toProblem(error) as unknown as Value) } /** The grants of one stream for the calling identity, or its refusal (401 signed out, 404 not reached). */ export const open = query({ args: { org: v.string(), topics: v.array(topic), entities: v.array(v.string()) }, handler: async (ctx, args) => { try { const call = await callContextOf(ctx.auth) const grants = await decideLiveGrants({ tenancy: tenancyOver(ctx.db), collections: registry.collections, org: args.org, actor: call.actor, ...(call.pinnedOrgId ? { pinnedOrgId: call.pinnedOrgId } : {}), topics: args.topics, entities: args.entities, }) return { org: grants.org, topics: grants.topics.map((t) => ({ ...t })), entities: [...grants.entities], denied: grants.denied.map((d) => ({ ...d })), } } catch (error) { throw refusal(error) } }, }) /** * The heartbeat's re-check: null while the caller still reaches the organization and every topic and holds * the read permission of every entity the stream was granted (`granted`), else why the stream ends. */ export const recheck = query({ args: { org: v.string(), topics: v.array(topic), granted: v.array(v.string()) }, handler: async (ctx, args) => { try { const call = await callContextOf(ctx.auth) return await recheckLiveScopes({ tenancy: tenancyOver(ctx.db), collections: registry.collections, org: args.org, actor: call.actor, ...(call.pinnedOrgId ? { pinnedOrgId: call.pinnedOrgId } : {}), topics: args.topics, granted: args.granted, }) } catch (error) { throw refusal(error) } }, }) export const head = internalQuery({ args: { orgId: v.string() }, handler: async (ctx, { orgId }) => liveLogHead(ctx.db as never, orgId), }) export const changes = internalQuery({ args: { orgId: v.string(), after: v.object({ txid: v.number(), seq: v.number() }), limit: v.number() }, handler: async (ctx, args) => readLiveLog(ctx.db as never, args), }) export const prune = internalMutation({ args: {}, handler: async (ctx) => pruneLiveLog(ctx.db as never, { before: Date.now() - LIVE_RETENTION_MS, limit: 4000 }), })