template-app/convex/live.ts

// 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<Value> {
  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 }),
})