Admin dashboard for jobs

Extremely simple jobs dashboard to sanity-check what we've got going on in the job queue.

app/@types/nextjs-routes.d.ts

@@ -12,6 +12,7 @@ declare module "nextjs-routes" {
 
   export type Route =
     | StaticRoute<"/account/signin">
+    | StaticRoute<"/admin/jobs">
     | DynamicRoute<"/api/auth/[...nextauth]", { "nextauth": string[] }>
     | StaticRoute<"/api/experiments/og-image">
     | DynamicRoute<"/api/trpc/[trpc]", { "trpc": string }>

app/package.json

@@ -18,6 +18,7 @@
     "lint": "next lint",
     "start": "TZ=UTC next start",
     "codegen:clients": "tsx src/server/scripts/client-codegen.ts",
+    "codegen:db": "prisma generate && kysely-codegen --dialect postgres --out-file src/server/db.types.ts",
     "seed": "tsx prisma/seed.ts",
     "check": "concurrently 'pnpm lint' 'pnpm tsc' 'pnpm prettier . --check'",
     "test": "pnpm vitest"
@@ -65,6 +66,7 @@
     "json-stringify-pretty-compact": "^4.0.0",
     "jsonschema": "^1.4.1",
     "kysely": "^0.26.1",
+    "kysely-codegen": "^0.10.1",
     "lodash-es": "^4.17.21",
     "lucide-react": "^0.265.0",
     "marked": "^7.0.3",

app/prisma/seed.ts

@@ -10,6 +10,14 @@ await prisma.project.deleteMany({
   where: { id: defaultId },
 });
 
+// Mark all users as admins
+await prisma.user.updateMany({
+  where: {},
+  data: {
+    role: "ADMIN",
+  },
+});
+
 // If there's an existing project, just seed into it
 const project =
   (await prisma.project.findFirst({})) ??
@@ -18,12 +26,16 @@ const project =
   }));
 
 if (env.OPENPIPE_API_KEY) {
-  await prisma.apiKey.create({
+  await prisma.apiKey.upsert({
-    data: {
+    where: {
+      apiKey: env.OPENPIPE_API_KEY,
+    },
+    create: {
       projectId: project.id,
       name: "Default API Key",
       apiKey: env.OPENPIPE_API_KEY,
     },
+    update: {},
   });
 }
 

app/src/modelProviders/openai-ChatCompletion/refinementActions.ts

@@ -12,7 +12,6 @@ export const refinementActions: Record<string, RefinementAction> = {
   
       definePrompt("openai/ChatCompletion", {
           model: "gpt-4",
-          stream: true,
           messages: [
               {
               role: "system",
@@ -29,7 +28,6 @@ export const refinementActions: Record<string, RefinementAction> = {
   
       definePrompt("openai/ChatCompletion", {
           model: "gpt-4",
-          stream: true,
           messages: [
               {
               role: "system",
@@ -126,7 +124,6 @@ export const refinementActions: Record<string, RefinementAction> = {
   
       definePrompt("openai/ChatCompletion", {
         model: "gpt-4",
-        stream: true,
         messages: [
           {
             role: "system",
@@ -143,7 +140,6 @@ export const refinementActions: Record<string, RefinementAction> = {
   
       definePrompt("openai/ChatCompletion", {
         model: "gpt-4",
-        stream: true,
         messages: [
           {
             role: "system",
@@ -237,7 +233,6 @@ export const refinementActions: Record<string, RefinementAction> = {
   
       definePrompt("openai/ChatCompletion", {
         model: "gpt-3.5-turbo",
-        stream: true,
         messages: [
           {
             role: "system",

app/src/pages/admin/jobs/index.tsx

@@ -0,0 +1,54 @@
+import { Card, Table, Tbody, Td, Th, Thead, Tr } from "@chakra-ui/react";
+import dayjs from "dayjs";
+import { isDate, isObject, isString } from "lodash-es";
+import AppShell from "~/components/nav/AppShell";
+import { type RouterOutputs, api } from "~/utils/api";
+
+const fieldsToShow: (keyof RouterOutputs["adminJobs"]["list"][0])[] = [
+  "id",
+  "queue_name",
+  "payload",
+  "priority",
+  "attempts",
+  "last_error",
+  "created_at",
+  "key",
+  "locked_at",
+  "run_at",
+];
+
+export default function Jobs() {
+  const jobs = api.adminJobs.list.useQuery({});
+
+  return (
+    <AppShell title="Admin Jobs">
+      <Card m={4} overflowX="auto">
+        <Table>
+          <Thead>
+            <Tr>
+              {fieldsToShow.map((field) => (
+                <Th key={field}>{field}</Th>
+              ))}
+            </Tr>
+          </Thead>
+          <Tbody>
+            {jobs.data?.map((job) => (
+              <Tr key={job.id}>
+                {fieldsToShow.map((field) => {
+                  // Check if object
+                  let value = job[field];
+                  if (isDate(value)) {
+                    value = dayjs(value).format("YYYY-MM-DD HH:mm:ss");
+                  } else if (isObject(value) && !isString(value)) {
+                    value = JSON.stringify(value);
+                  } // check if date
+                  return <Td key={field}>{value}</Td>;
+                })}
+              </Tr>
+            ))}
+          </Tbody>
+        </Table>
+      </Card>
+    </AppShell>
+  );
+}

app/src/server/api/root.router.ts

@@ -12,6 +12,7 @@ import { projectsRouter } from "./routers/projects.router";
 import { dashboardRouter } from "./routers/dashboard.router";
 import { loggedCallsRouter } from "./routers/loggedCalls.router";
 import { usersRouter } from "./routers/users.router";
+import { adminJobsRouter } from "./routers/adminJobs.router";
 
 /**
  * This is the primary router for your server.
@@ -32,6 +33,7 @@ export const appRouter = createTRPCRouter({
   dashboard: dashboardRouter,
   loggedCalls: loggedCallsRouter,
   users: usersRouter,
+  adminJobs: adminJobsRouter,
 });
 
 // export type definition of API

app/src/server/api/routers/adminJobs.router.ts

@@ -0,0 +1,18 @@
+import { z } from "zod";
+
+import { createTRPCRouter, protectedProcedure } from "~/server/api/trpc";
+import { kysely } from "~/server/db";
+import { requireIsAdmin } from "~/utils/accessControl";
+
+export const adminJobsRouter = createTRPCRouter({
+  list: protectedProcedure.input(z.object({})).query(async ({ ctx }) => {
+    await requireIsAdmin(ctx);
+
+    return await kysely
+      .selectFrom("graphile_worker.jobs")
+      .limit(100)
+      .selectAll()
+      .orderBy("created_at", "desc")
+      .execute();
+  }),
+});

app/src/server/api/routers/experiments.router.ts

@@ -335,7 +335,6 @@ export const experimentsRouter = createTRPCRouter({
           
           definePrompt("openai/ChatCompletion", {
             model: "gpt-3.5-turbo-0613",
-            stream: true,
             messages: [
               {
                 role: "system",

app/src/server/db.ts

@@ -1,27 +1,6 @@
-import {
+import { type DB } from "./db.types";
-  type Experiment,
+
-  type PromptVariant,
+import { PrismaClient } from "@prisma/client";
-  type TestScenario,
-  type TemplateVariable,
-  type ScenarioVariantCell,
-  type ModelResponse,
-  type Evaluation,
-  type OutputEvaluation,
-  type Dataset,
-  type DatasetEntry,
-  type Project,
-  type ProjectUser,
-  type WorldChampEntrant,
-  type LoggedCall,
-  type LoggedCallModelResponse,
-  type LoggedCallTag,
-  type ApiKey,
-  type Account,
-  type Session,
-  type User,
-  type VerificationToken,
-  PrismaClient,
-} from "@prisma/client";
 import { Kysely, PostgresDialect } from "kysely";
 // TODO: Revert to normal import when our tsconfig.json is fixed
 // import { Pool } from "pg";
@@ -32,30 +11,6 @@ const Pool = (UntypedPool.default ? UntypedPool.default : UntypedPool) as typeof
 
 import { env } from "~/env.mjs";
 
-interface DB {
-  Experiment: Experiment;
-  PromptVariant: PromptVariant;
-  TestScenario: TestScenario;
-  TemplateVariable: TemplateVariable;
-  ScenarioVariantCell: ScenarioVariantCell;
-  ModelResponse: ModelResponse;
-  Evaluation: Evaluation;
-  OutputEvaluation: OutputEvaluation;
-  Dataset: Dataset;
-  DatasetEntry: DatasetEntry;
-  Project: Project;
-  ProjectUser: ProjectUser;
-  WorldChampEntrant: WorldChampEntrant;
-  LoggedCall: LoggedCall;
-  LoggedCallModelResponse: LoggedCallModelResponse;
-  LoggedCallTag: LoggedCallTag;
-  ApiKey: ApiKey;
-  Account: Account;
-  Session: Session;
-  User: User;
-  VerificationToken: VerificationToken;
-}
-
 const globalForPrisma = globalThis as unknown as {
   prisma: PrismaClient | undefined;
 };

app/src/server/db.types.ts

@@ -0,0 +1,336 @@
+import type { ColumnType } from "kysely";
+
+export type Generated<T> = T extends ColumnType<infer S, infer I, infer U>
+  ? ColumnType<S, I | undefined, U>
+  : ColumnType<T, T | undefined, T>;
+
+export type Int8 = ColumnType<string, string | number | bigint, string | number | bigint>;
+
+export type Json = ColumnType<JsonValue, string, string>;
+
+export type JsonArray = JsonValue[];
+
+export type JsonObject = {
+  [K in string]?: JsonValue;
+};
+
+export type JsonPrimitive = boolean | null | number | string;
+
+export type JsonValue = JsonArray | JsonObject | JsonPrimitive;
+
+export type Numeric = ColumnType<string, string | number, string | number>;
+
+export type Timestamp = ColumnType<Date, Date | string, Date | string>;
+
+export interface _PrismaMigrations {
+  id: string;
+  checksum: string;
+  finished_at: Timestamp | null;
+  migration_name: string;
+  logs: string | null;
+  rolled_back_at: Timestamp | null;
+  started_at: Generated<Timestamp>;
+  applied_steps_count: Generated<number>;
+}
+
+export interface Account {
+  id: string;
+  userId: string;
+  type: string;
+  provider: string;
+  providerAccountId: string;
+  refresh_token: string | null;
+  refresh_token_expires_in: number | null;
+  access_token: string | null;
+  expires_at: number | null;
+  token_type: string | null;
+  scope: string | null;
+  id_token: string | null;
+  session_state: string | null;
+}
+
+export interface ApiKey {
+  id: string;
+  name: string;
+  apiKey: string;
+  projectId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface Dataset {
+  id: string;
+  name: string;
+  projectId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface DatasetEntry {
+  id: string;
+  input: string;
+  output: string | null;
+  datasetId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface Evaluation {
+  id: string;
+  label: string;
+  value: string;
+  evalType: "CONTAINS" | "DOES_NOT_CONTAIN" | "GPT4_EVAL";
+  experimentId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface Experiment {
+  id: string;
+  label: string;
+  sortIndex: Generated<number>;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+  projectId: string;
+}
+
+export interface GraphileWorkerJobQueues {
+  queue_name: string;
+  job_count: number;
+  locked_at: Timestamp | null;
+  locked_by: string | null;
+}
+
+export interface GraphileWorkerJobs {
+  id: Generated<Int8>;
+  queue_name: string | null;
+  task_identifier: string;
+  payload: Generated<Json>;
+  priority: Generated<number>;
+  run_at: Generated<Timestamp>;
+  attempts: Generated<number>;
+  max_attempts: Generated<number>;
+  last_error: string | null;
+  created_at: Generated<Timestamp>;
+  updated_at: Generated<Timestamp>;
+  key: string | null;
+  locked_at: Timestamp | null;
+  locked_by: string | null;
+  revision: Generated<number>;
+  flags: Json | null;
+}
+
+export interface GraphileWorkerKnownCrontabs {
+  identifier: string;
+  known_since: Timestamp;
+  last_execution: Timestamp | null;
+}
+
+export interface GraphileWorkerMigrations {
+  id: number;
+  ts: Generated<Timestamp>;
+}
+
+export interface LoggedCall {
+  id: string;
+  requestedAt: Timestamp;
+  cacheHit: boolean;
+  modelResponseId: string | null;
+  projectId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+  model: string | null;
+}
+
+export interface LoggedCallModelResponse {
+  id: string;
+  reqPayload: Json;
+  statusCode: number | null;
+  respPayload: Json | null;
+  errorMessage: string | null;
+  requestedAt: Timestamp;
+  receivedAt: Timestamp;
+  cacheKey: string | null;
+  durationMs: number | null;
+  inputTokens: number | null;
+  outputTokens: number | null;
+  finishReason: string | null;
+  completionId: string | null;
+  cost: Numeric | null;
+  originalLoggedCallId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface LoggedCallTag {
+  id: string;
+  name: string;
+  value: string | null;
+  loggedCallId: string;
+  projectId: string;
+}
+
+export interface ModelResponse {
+  id: string;
+  cacheKey: string;
+  respPayload: Json | null;
+  inputTokens: number | null;
+  outputTokens: number | null;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+  scenarioVariantCellId: string;
+  cost: number | null;
+  requestedAt: Timestamp | null;
+  receivedAt: Timestamp | null;
+  statusCode: number | null;
+  errorMessage: string | null;
+  retryTime: Timestamp | null;
+  outdated: Generated<boolean>;
+}
+
+export interface OutputEvaluation {
+  id: string;
+  result: number;
+  details: string | null;
+  modelResponseId: string;
+  evaluationId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface Project {
+  id: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+  personalProjectUserId: string | null;
+  name: Generated<string>;
+}
+
+export interface ProjectUser {
+  id: string;
+  role: "ADMIN" | "MEMBER" | "VIEWER";
+  projectId: string;
+  userId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface PromptVariant {
+  id: string;
+  label: string;
+  uiId: string;
+  visible: Generated<boolean>;
+  sortIndex: Generated<number>;
+  experimentId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+  promptConstructor: string;
+  model: string;
+  promptConstructorVersion: number;
+  modelProvider: string;
+}
+
+export interface ScenarioVariantCell {
+  id: string;
+  errorMessage: string | null;
+  promptVariantId: string;
+  testScenarioId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+  retrievalStatus: Generated<"COMPLETE" | "ERROR" | "IN_PROGRESS" | "PENDING">;
+  prompt: Json | null;
+  jobQueuedAt: Timestamp | null;
+  jobStartedAt: Timestamp | null;
+}
+
+export interface Session {
+  id: string;
+  sessionToken: string;
+  userId: string;
+  expires: Timestamp;
+}
+
+export interface TemplateVariable {
+  id: string;
+  label: string;
+  experimentId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface TestScenario {
+  id: string;
+  variableValues: Json;
+  uiId: string;
+  visible: Generated<boolean>;
+  sortIndex: Generated<number>;
+  experimentId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface User {
+  id: string;
+  name: string | null;
+  email: string | null;
+  emailVerified: Timestamp | null;
+  image: string | null;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Generated<Timestamp>;
+  role: Generated<"ADMIN" | "USER">;
+}
+
+export interface UserInvitation {
+  id: string;
+  projectId: string;
+  email: string;
+  role: "ADMIN" | "MEMBER" | "VIEWER";
+  invitationToken: string;
+  senderId: string;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface VerificationToken {
+  identifier: string;
+  token: string;
+  expires: Timestamp;
+}
+
+export interface WorldChampEntrant {
+  id: string;
+  userId: string;
+  approved: Generated<boolean>;
+  createdAt: Generated<Timestamp>;
+  updatedAt: Timestamp;
+}
+
+export interface DB {
+  _prisma_migrations: _PrismaMigrations;
+  Account: Account;
+  ApiKey: ApiKey;
+  Dataset: Dataset;
+  DatasetEntry: DatasetEntry;
+  Evaluation: Evaluation;
+  Experiment: Experiment;
+  "graphile_worker.job_queues": GraphileWorkerJobQueues;
+  "graphile_worker.jobs": GraphileWorkerJobs;
+  "graphile_worker.known_crontabs": GraphileWorkerKnownCrontabs;
+  "graphile_worker.migrations": GraphileWorkerMigrations;
+  LoggedCall: LoggedCall;
+  LoggedCallModelResponse: LoggedCallModelResponse;
+  LoggedCallTag: LoggedCallTag;
+  ModelResponse: ModelResponse;
+  OutputEvaluation: OutputEvaluation;
+  Project: Project;
+  ProjectUser: ProjectUser;
+  PromptVariant: PromptVariant;
+  ScenarioVariantCell: ScenarioVariantCell;
+  Session: Session;
+  TemplateVariable: TemplateVariable;
+  TestScenario: TestScenario;
+  User: User;
+  UserInvitation: UserInvitation;
+  VerificationToken: VerificationToken;
+  WorldChampEntrant: WorldChampEntrant;
+}

app/src/utils/accessControl.ts

@@ -17,6 +17,8 @@ export const requireNothing = (ctx: TRPCContext) => {
 };
 
 export const requireIsProjectAdmin = async (projectId: string, ctx: TRPCContext) => {
+  ctx.markAccessControlRun();
+
   const userId = ctx.session?.user.id;
   if (!userId) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
@@ -33,11 +35,11 @@ export const requireIsProjectAdmin = async (projectId: string, ctx: TRPCContext)
   if (!isAdmin) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
   }
-
-  ctx.markAccessControlRun();
 };
 
 export const requireCanViewProject = async (projectId: string, ctx: TRPCContext) => {
+  ctx.markAccessControlRun();
+
   const userId = ctx.session?.user.id;
   if (!userId) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
@@ -53,11 +55,11 @@ export const requireCanViewProject = async (projectId: string, ctx: TRPCContext)
   if (!canView) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
   }
-
-  ctx.markAccessControlRun();
 };
 
 export const requireCanModifyProject = async (projectId: string, ctx: TRPCContext) => {
+  ctx.markAccessControlRun();
+
   const userId = ctx.session?.user.id;
   if (!userId) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
@@ -74,11 +76,11 @@ export const requireCanModifyProject = async (projectId: string, ctx: TRPCContex
   if (!canModify) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
   }
-
-  ctx.markAccessControlRun();
 };
 
 export const requireCanViewDataset = async (datasetId: string, ctx: TRPCContext) => {
+  ctx.markAccessControlRun();
+
   const dataset = await prisma.dataset.findFirst({
     where: {
       id: datasetId,
@@ -96,8 +98,6 @@ export const requireCanViewDataset = async (datasetId: string, ctx: TRPCContext)
   if (!dataset) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
   }
-
-  ctx.markAccessControlRun();
 };
 
 export const requireCanModifyDataset = async (datasetId: string, ctx: TRPCContext) => {
@@ -105,13 +105,10 @@ export const requireCanModifyDataset = async (datasetId: string, ctx: TRPCContex
   await requireCanViewDataset(datasetId, ctx);
 };
 
-export const requireCanViewExperiment = async (experimentId: string, ctx: TRPCContext) => {
+export const requireCanViewExperiment = (experimentId: string, ctx: TRPCContext): Promise<void> => {
-  await prisma.experiment.findFirst({
-    where: { id: experimentId },
-  });
-
   // Right now all experiments are publicly viewable, so this is a no-op.
   ctx.markAccessControlRun();
+  return Promise.resolve();
 };
 
 export const canModifyExperiment = async (experimentId: string, userId: string) => {
@@ -136,6 +133,8 @@ export const canModifyExperiment = async (experimentId: string, userId: string)
 };
 
 export const requireCanModifyExperiment = async (experimentId: string, ctx: TRPCContext) => {
+  ctx.markAccessControlRun();
+
   const userId = ctx.session?.user.id;
   if (!userId) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
@@ -144,6 +143,17 @@ export const requireCanModifyExperiment = async (experimentId: string, ctx: TRPC
   if (!(await canModifyExperiment(experimentId, userId))) {
     throw new TRPCError({ code: "UNAUTHORIZED" });
   }
-
+};
-  ctx.markAccessControlRun();
+
+export const requireIsAdmin = async (ctx: TRPCContext) => {
+  ctx.markAccessControlRun();
+
+  const userId = ctx.session?.user.id;
+  if (!userId) {
+    throw new TRPCError({ code: "UNAUTHORIZED" });
+  }
+
+  if (!(await isAdmin(userId))) {
+    throw new TRPCError({ code: "UNAUTHORIZED" });
+  }
 };

pnpm-lock.yaml

@@ -134,6 +134,9 @@ importers:
       kysely:
         specifier: ^0.26.1
         version: 0.26.1
+      kysely-codegen:
+        specifier: ^0.10.1
+        version: 0.10.1(kysely@0.26.1)(pg@8.11.2)
       lodash-es:
         specifier: ^4.17.21
         version: 4.17.21
@@ -6391,6 +6394,30 @@ packages:
       object.values: 1.1.6
     dev: true
 
+  /kysely-codegen@0.10.1(kysely@0.26.1)(pg@8.11.2):
+    resolution: {integrity: sha512-8Bslh952gN5gtucRv4jTZDFD18RBioS6M50zHfe5kwb5iSyEAunU4ZYMdHzkHraa4zxjg5/183XlOryBCXLRIw==}
+    hasBin: true
+    peerDependencies:
+      better-sqlite3: '>=7.6.2'
+      kysely: '>=0.19.12'
+      mysql2: ^2.3.3 || ^3.0.0
+      pg: ^8.8.0
+    peerDependenciesMeta:
+      better-sqlite3:
+        optional: true
+      mysql2:
+        optional: true
+      pg:
+        optional: true
+    dependencies:
+      chalk: 4.1.2
+      dotenv: 16.3.1
+      kysely: 0.26.1
+      micromatch: 4.0.5
+      minimist: 1.2.8
+      pg: 8.11.2
+    dev: false
+
   /kysely@0.26.1:
     resolution: {integrity: sha512-FVRomkdZofBu3O8SiwAOXrwbhPZZr8mBN5ZeUWyprH29jzvy6Inzqbd0IMmGxpd4rcOCL9HyyBNWBa8FBqDAdg==}
     engines: {node: '>=14.0.0'}
@@ -6611,7 +6638,6 @@ packages:
     dependencies:
       braces: 3.0.2
       picomatch: 2.3.1
-    dev: true
 
   /mime-db@1.52.0:
     resolution: {integrity: sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg==}