Matthewz Datapuller Button

.github/workflows/cd-deploy.yaml

@@ -69,6 +69,9 @@ jobs:
               # Check if staff helm chart exists
               helm status bt-staff &>/dev/null && staff_status=true || staff_status=false
 
+              # Delete deployment if selector labels changed (immutable field)
+              kubectl delete deployment bt-staff-frontend -n bt --ignore-not-found=true
+
               # Upgrade staff helm chart, or install if not exists
               helm upgrade bt-staff oci://registry-1.docker.io/octoberkeleytime/bt-staff \
                 --install \

.github/workflows/cd-dev.yaml

@@ -83,6 +83,7 @@ jobs:
       environment: development
       name: bt-dev-app-${{ needs.compute-sha.outputs.sha_short }}
       version: 0.1.0-dev-${{ needs.compute-sha.outputs.sha_short }}
+      deploy_staff: false
       values: |
         env: dev
         ttl: ${{ inputs.ttl }}

apps/backend/package.json

@@ -44,6 +44,16 @@
     "@graphql-tools/schema": "^10.0.25",
     "@graphql-tools/utils": "^10.9.1",
     "@keyv/redis": "^5.1.2",
+    "@kubernetes/client-node": "^1.4.0",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/auto-instrumentations-node": "^0.53.0",
+    "@opentelemetry/exporter-logs-otlp-http": "^0.56.0",
+    "@opentelemetry/exporter-metrics-otlp-http": "^0.56.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.56.0",
+    "@opentelemetry/resources": "^1.29.0",
+    "@opentelemetry/sdk-logs": "^0.56.0",
+    "@opentelemetry/sdk-metrics": "1.29.0",
+    "@opentelemetry/sdk-node": "^0.56.0",
     "@repo/common": "*",
     "@repo/gql-typedefs": "*",
     "@repo/shared": "*",
@@ -65,8 +75,9 @@
     "papaparse": "^5.5.3",
     "passport": "^0.7.0",
     "passport-google-oauth20": "^2.0.0",
+    "pino": "^9.6.0",
     "redis": "^5.8.3",
-    "uuid": "^13.0.0",
-    "undici": "^7.14.0"
+    "undici": "^7.14.0",
+    "uuid": "^13.0.0"
   }
 }

apps/backend/src/modules/datapuller/controller.ts

@@ -0,0 +1,154 @@
+import { BatchV1Api, KubeConfig, V1Job } from "@kubernetes/client-node";
+import { GraphQLError } from "graphql";
+
+export type DatapullerJob =
+  | "TERMS_ALL"
+  | "TERMS_NEARBY"
+  | "COURSES"
+  | "SECTIONS_ACTIVE"
+  | "SECTIONS_LAST_FIVE_YEARS"
+  | "CLASSES_ACTIVE"
+  | "CLASSES_LAST_FIVE_YEARS"
+  | "GRADES_RECENT"
+  | "GRADES_LAST_FIVE_YEARS"
+  | "ENROLLMENTS"
+  | "ENROLLMENT_TIMEFRAME";
+
+const JOB_SUFFIX: Record<DatapullerJob, string> = {
+  TERMS_ALL: "terms-all",
+  TERMS_NEARBY: "terms-nearby",
+  COURSES: "courses",
+  SECTIONS_ACTIVE: "sections-active",
+  SECTIONS_LAST_FIVE_YEARS: "sections-l5y",
+  CLASSES_ACTIVE: "classes-active",
+  CLASSES_LAST_FIVE_YEARS: "classes-l5y",
+  GRADES_RECENT: "grades-recent",
+  GRADES_LAST_FIVE_YEARS: "grades-l5y",
+  ENROLLMENTS: "enrollments",
+  ENROLLMENT_TIMEFRAME: "enrollment-timeframe",
+};
+
+const TTL_SECONDS = 300; // 5 minutes
+
+function getBatchApi() {
+  const kc = new KubeConfig();
+  // loadFromDefault uses in-cluster service account when deployed to k8s
+  kc.loadFromDefault();
+  return kc.makeApiClient(BatchV1Api);
+}
+
+function getCronJobName(job: DatapullerJob): string {
+  const prefix = process.env.DATAPULLER_CRONJOB_PREFIX ?? "bt-prod-datapuller";
+  return `${prefix}-${JOB_SUFFIX[job]}`;
+}
+
+function getNamespace(): string {
+  return process.env.K8S_NAMESPACE ?? "bt";
+}
+
+export async function triggerDatapuller(
+  job: DatapullerJob
+): Promise<{ jobName: string; success: boolean; message: string }> {
+  // set SKIP_K8S=true in .env to skip the real k8s call
+  if (process.env.SKIP_K8S === "true") {
+    const mockJobName = `${getCronJobName(job)}-manual-${Math.floor(Date.now() / 1000)}`;
+    return {
+      jobName: mockJobName,
+      success: true,
+      message: `[MOCK] Job ${mockJobName} would have been created.`,
+    };
+  }
+
+  const batchApi = getBatchApi();
+  const namespace = getNamespace();
+  const cronJobName = getCronJobName(job);
+  const jobName = `${cronJobName}-manual-${Math.floor(Date.now() / 1000)}`;
+
+  // check for existing job
+  const existingJobs = await batchApi.listNamespacedJob({ namespace });
+  const alreadyRunning = existingJobs.items.some(
+    (j) =>
+      j.metadata?.annotations?.["cronjob.kubernetes.io/instantiate"] ===
+        "manual" &&
+      j.metadata?.name?.startsWith(cronJobName) &&
+      (j.status?.active ?? 0) > 0
+  );
+
+  if (alreadyRunning) {
+    throw new GraphQLError(
+      `A job for ${JOB_SUFFIX[job]} is already running. Wait for it to finish before triggering again.`,
+      { extensions: { code: "CONFLICT" } }
+    );
+  }
+
+  // get job spec + job name from original cronjob
+  const cronJob = await batchApi.readNamespacedCronJob({
+    name: cronJobName,
+    namespace,
+  });
+
+  const jobSpec = cronJob.spec?.jobTemplate?.spec;
+  if (!jobSpec) {
+    throw new GraphQLError(`Could not read spec for CronJob ${cronJobName}`, {
+      extensions: { code: "INTERNAL_SERVER_ERROR" },
+    });
+  }
+
+  const newJob: V1Job = {
+    apiVersion: "batch/v1",
+    kind: "Job",
+    metadata: {
+      name: jobName,
+      namespace,
+      annotations: {
+        "cronjob.kubernetes.io/instantiate": "manual",
+      },
+    },
+    spec: {
+      ...jobSpec,
+      ttlSecondsAfterFinished: TTL_SECONDS,
+    },
+  };
+
+  await batchApi.createNamespacedJob({ namespace, body: newJob });
+
+  return {
+    jobName,
+    success: true,
+    message: `Job ${jobName} created successfully.`,
+  };
+}
+
+export async function getDatapullerJobStatus(
+  jobName: string
+): Promise<{ jobName: string; phase: string; message: string | null }> {
+  if (process.env.SKIP_K8S === "true") {
+    return { jobName, phase: "Succeeded", message: "[MOCK] Job completed." };
+  }
+
+  const batchApi = getBatchApi();
+  const namespace = getNamespace();
+
+  try {
+    const job = await batchApi.readNamespacedJob({ name: jobName, namespace });
+    const status = job.status ?? {};
+
+    let phase = "Pending";
+    if ((status.active ?? 0) > 0) phase = "Running";
+    else if ((status.succeeded ?? 0) > 0) phase = "Succeeded";
+    else if ((status.failed ?? 0) > 0) phase = "Failed";
+
+    return { jobName, phase, message: null };
+  } catch (e: unknown) {
+    const status = (e as { response?: { statusCode?: number } })?.response
+      ?.statusCode;
+    if (status === 404) {
+      return {
+        jobName,
+        phase: "NotFound",
+        message: "Job not found or already cleaned up.",
+      };
+    }
+    throw e;
+  }
+}

apps/backend/src/modules/datapuller/index.ts

@@ -0,0 +1,8 @@
+import { datapullerTypeDef } from "@repo/gql-typedefs";
+
+import resolver from "./resolver";
+
+export default {
+  resolver,
+  typeDef: datapullerTypeDef,
+};

apps/backend/src/modules/datapuller/resolver.ts

@@ -0,0 +1,32 @@
+import { RequestContext } from "../../types/request-context";
+import { requireStaffAuth } from "../analytics/helpers/staff-auth";
+import {
+  DatapullerJob,
+  getDatapullerJobStatus,
+  triggerDatapuller,
+} from "./controller";
+
+const resolver = {
+  Query: {
+    datapullerJobStatus: async (
+      _: unknown,
+      { jobName }: { jobName: string },
+      context: RequestContext
+    ) => {
+      await requireStaffAuth(context);
+      return getDatapullerJobStatus(jobName);
+    },
+  },
+  Mutation: {
+    triggerDatapuller: async (
+      _: unknown,
+      { job }: { job: DatapullerJob },
+      context: RequestContext
+    ) => {
+      await requireStaffAuth(context);
+      return triggerDatapuller(job);
+    },
+  },
+};
+
+export default resolver;

apps/backend/src/modules/index.ts

@@ -2,6 +2,7 @@ import { merge } from "lodash";
 
 import Analytics from "./analytics";
 import Banner from "./banner";
+import Datapuller from "./datapuller";
 import Catalog from "./catalog";
 import Class from "./class";
 import ClickTracking from "./click-tracking";
@@ -23,6 +24,7 @@ import User from "./user";
 const modules = [
   Analytics,
   Banner,
+  Datapuller,
   ClickTracking,
   User,
   GradeDistribution,

apps/staff-frontend/src/app/Dashboard/Dashboard.module.scss

@@ -217,6 +217,119 @@
   }
 }
 
+// Data tab
+.dataTab {
+  display: flex;
+  flex-direction: column;
+  gap: 16px;
+  max-width: 480px;
+}
+
+.dataTabTitle {
+  font-size: 18px;
+  font-weight: 600;
+  margin: 0;
+}
+
+.dataTabDescription {
+  font-size: 14px;
+  color: var(--label-color);
+  margin: 0;
+}
+
+.splitButton {
+  position: relative;
+  display: inline-flex;
+  border-radius: 8px;
+  overflow: visible;
+}
+
+.splitButtonMain {
+  flex: 1;
+  padding: 8px 16px;
+  font-size: 14px;
+  font-weight: 500;
+  background: var(--blue-500);
+  color: #fff;
+  border: none;
+  border-radius: 8px 0 0 8px;
+  cursor: pointer;
+
+  &:disabled {
+    opacity: 0.6;
+    cursor: not-allowed;
+  }
+
+  &:not(:disabled):hover {
+    background: var(--blue-600, #1a56db);
+  }
+}
+
+.splitButtonCaret {
+  padding: 8px 10px;
+  font-size: 12px;
+  background: var(--blue-500);
+  color: #fff;
+  border: none;
+  border-left: 1px solid rgba(255, 255, 255, 0.3);
+  border-radius: 0 8px 8px 0;
+  cursor: pointer;
+
+  &:disabled {
+    opacity: 0.6;
+    cursor: not-allowed;
+  }
+
+  &:not(:disabled):hover {
+    background: var(--blue-600, #1a56db);
+  }
+}
+
+.splitButtonDropdown {
+  position: absolute;
+  top: calc(100% + 4px);
+  left: 0;
+  z-index: 100;
+  min-width: 220px;
+  background: var(--surface-color, #fff);
+  border: 1px solid var(--border-color, #e5e7eb);
+  border-radius: 8px;
+  box-shadow: 0 4px 12px rgba(0, 0, 0, 0.1);
+  list-style: none;
+  margin: 0;
+  padding: 4px 0;
+}
+
+.splitButtonDropdownItem {
+  padding: 8px 16px;
+  font-size: 14px;
+  cursor: pointer;
+
+  &:hover {
+    background: var(--hover-color, #f3f4f6);
+  }
+}
+
+.splitButtonDropdownItemActive {
+  font-weight: 600;
+  color: var(--blue-500);
+}
+
+.datapullerSuccess {
+  font-size: 13px;
+  color: var(--green-500);
+}
+
+.datapullerError {
+  font-size: 13px;
+  color: var(--red-500);
+}
+
+.datapullerStatus {
+  font-size: 13px;
+  color: var(--label-color);
+}
+
 .noLink {
   font-size: 13px;
   color: var(--label-color);