OOT HUD: Add ClickHouse queries, utility library, and unit tests (Part 1/3) (#8110)

## Summary

This is **Part 1 of 3** of the OOT HUD pipeline, split from #8069 per
reviewer request.

This PR adds the foundational data layer for the Out-of-Tree (OOT) HUD:

- **ClickHouse queries** for three views: OOT summary dashboard,
per-backend dashboard, and PR-level results
- **Shared utility library** (`torchci/lib/oot/ootUtils.ts`) with:
- Relay payload types (`RelayTrusted`, `RelayWorkflow`,
`RelayCallbackPayload`, etc.)
  - DynamoDB record extraction and validation (`extractDynamoRecord`)
  - UpdateItem write logic (`writeToDynamo`)
  - Payload size validation (`validatePayloadSize`)
  - UI helpers (`conclusionColor`, `conclusionLabel`)
  - `ApiError` class for structured HTTP error responses
- **Unit tests** (`torchci/test/ootUtils.test.ts`) covering:
  - Field mapping and DynamoDB key construction
  - Required field validation (job_name, check_run_id)
  - Test result computation (total from passed+failed+skipped)
  - Timing metric handling
  - Payload size validation
  - UI helper functions

### PR Stack
1. **This PR** — OOT HUD: Add ClickHouse queries, utility library, and
unit tests (Part 1/3)
2. #8111 — OOT HUD: Add frontend components — summary, dashboard, PR
section (Part 2/3)
3. #8112 — OOT HUD: Add API endpoint, PR page integration, and
replicator mapping (Part 3/3)

## Test Plan
- `ootUtils.test.ts` covers all utility functions
- ClickHouse queries can be tested once the `oot_workflow_job` table is
created (see #8105)

Author: subinz1

torchci/clickhouse_queries/oot_backend_dashboard/params.json

@@ -0,0 +1,12 @@
+{
+  "params": {
+    "repo": "String",
+    "days": "UInt64"
+  },
+  "tests": [
+    {
+      "repo": "<company>/<repo>",
+      "days": "7"
+    }
+  ]
+}

torchci/clickhouse_queries/oot_backend_dashboard/query.sql

@@ -0,0 +1,30 @@
+SELECT
+    upstream_repo,
+    pr_number,
+    pytorch_head_sha,
+    workflow_name,
+    job_name,
+    check_run_id,
+    run_id,
+    run_attempt,
+    status,
+    conclusion,
+    started_at,
+    completed_at,
+    duration_seconds,
+    total_tests,
+    passed_tests,
+    failed_tests,
+    skipped_tests,
+    workflow_run_url,
+    artifact_url,
+    queue_time,
+    execution_time
+FROM
+    default.oot_workflow_job FINAL
+WHERE
+    downstream_repo = {repo: String}
+    AND started_at > now() - INTERVAL {days: UInt64} DAY
+ORDER BY
+    started_at DESC
+LIMIT 500

torchci/clickhouse_queries/oot_pr_results/params.json

@@ -0,0 +1,10 @@
+{
+  "params": {
+    "pr": "UInt64"
+  },
+  "tests": [
+    {
+      "pr": "179565"
+    }
+  ]
+}

torchci/clickhouse_queries/oot_pr_results/query.sql

@@ -0,0 +1,22 @@
+SELECT
+    downstream_repo,
+    workflow_name,
+    job_name,
+    check_run_id,
+    run_id,
+    run_attempt,
+    status,
+    conclusion,
+    duration_seconds,
+    workflow_run_url,
+    artifact_url,
+    started_at,
+    queue_time,
+    execution_time
+FROM
+    default.oot_workflow_job FINAL
+WHERE
+    pr_number = {pr: UInt64}
+ORDER BY
+    downstream_repo, started_at DESC
+LIMIT 100

torchci/clickhouse_queries/oot_summary/params.json

@@ -0,0 +1,10 @@
+{
+  "params": {
+    "days": "UInt64"
+  },
+  "tests": [
+    {
+      "days": "7"
+    }
+  ]
+}

torchci/clickhouse_queries/oot_summary/query.sql

@@ -0,0 +1,18 @@
+SELECT
+    downstream_repo AS repo,
+    anyLast(downstream_repo_level) AS downstream_repo_level,
+    countIf(conclusion = 'success') AS successes,
+    countIf(conclusion = 'failure') AS failures,
+    count() AS total,
+    if(total > 0, successes / total, 0) AS pass_rate,
+    avg(duration_seconds) AS avg_duration_s,
+    max(started_at) AS last_run
+FROM
+    default.oot_workflow_job FINAL
+WHERE
+    started_at > now() - INTERVAL {days: UInt64} DAY
+    AND status = 'completed'
+GROUP BY
+    repo
+ORDER BY
+    pass_rate ASC

torchci/lib/oot/ootUtils.ts

@@ -0,0 +1,241 @@
+import { getDynamoClient } from "lib/dynamo";
+
+const OOT_TABLE = "torchci-oot-workflow-job";
+const MAX_PAYLOAD_BYTES = 2 * 1024 * 1024; // 2MB
+
+// ---- Types ----
+
+export interface RelayTrusted {
+  verified_repo: string;
+  downstream_repo_level?: string; // "L1" | "L2" | "L3" | "L4" — relay-determined from allowlist
+  ci_metrics?: {
+    queue_time?: number | null;
+    execution_time?: number | null;
+  };
+}
+
+export interface RelayWorkflow {
+  schema_version?: string;
+  status: string;
+  conclusion?: string | null;
+  name: string;
+  url: string;
+  job_name?: string;
+  check_run_id?: string;
+  run_id?: string;
+  run_attempt?: number | string;
+  started_at?: string;
+  completed_at?: string;
+  test_results?: {
+    passed?: number;
+    failed?: number;
+    skipped?: number;
+    total?: number;
+  };
+  artifact_url?: string;
+}
+
+export interface RelayCallbackPayload {
+  event_type: string;
+  delivery_id: string;
+  payload: {
+    pull_request?: { number: number; head?: { sha: string } };
+    repository?: { full_name: string };
+    [key: string]: any;
+  };
+  workflow: RelayWorkflow;
+}
+
+export interface RelayUntrusted {
+  callback_payload: RelayCallbackPayload;
+}
+
+export interface RelayPayload {
+  trusted: RelayTrusted;
+  untrusted: RelayUntrusted;
+}
+
+export interface OotWorkflowJobRecord {
+  dynamoKey: string;
+  status: string;
+  downstream_repo: string;
+  upstream_repo: string;
+  pr_number: number;
+  pytorch_head_sha: string;
+  delivery_id: string;
+  workflow_run_url: string;
+  workflow_name: string;
+  job_name: string;
+  check_run_id: string;
+  run_id: string;
+  run_attempt: number;
+  conclusion?: string;
+  queue_time?: number | null;
+  execution_time?: number | null;
+  started_at?: string;
+  completed_at?: string;
+  total_tests?: number;
+  passed_tests?: number;
+  failed_tests?: number;
+  skipped_tests?: number;
+  downstream_repo_level?: string;
+  artifact_url?: string;
+  environment?: string;
+}
+
+// ---- Validation ----
+
+export function validatePayloadSize(bodyString: string): void {
+  if (Buffer.byteLength(bodyString, "utf-8") > MAX_PAYLOAD_BYTES) {
+    throw new ApiError(413, "Payload exceeds 2MB limit");
+  }
+}
+
+// ---- Extraction ----
+
+export function extractDynamoRecord(
+  payload: RelayPayload
+): OotWorkflowJobRecord {
+  const { trusted, untrusted } = payload;
+  const cb = untrusted.callback_payload;
+  const wf = cb.workflow;
+  const pr = cb.payload?.pull_request;
+  const upstreamRepo = cb.payload?.repository?.full_name ?? "pytorch/pytorch";
+
+  if (!wf.job_name) {
+    throw new ApiError(400, "Missing required field: workflow.job_name");
+  }
+  if (wf.check_run_id == null) {
+    throw new ApiError(400, "Missing required field: workflow.check_run_id");
+  }
+  const jobName = wf.job_name;
+  const checkRunId = String(wf.check_run_id);
+  const runAttempt = Number(wf.run_attempt ?? 1) || 1;
+  const dynamoKey = `${trusted.verified_repo}/${cb.delivery_id}/${wf.name}/${jobName}/${checkRunId}`;
+
+  const record: OotWorkflowJobRecord = {
+    dynamoKey,
+    status: wf.status,
+    downstream_repo: trusted.verified_repo,
+    upstream_repo: upstreamRepo,
+    pr_number: pr?.number ?? 0,
+    pytorch_head_sha: pr?.head?.sha ?? "",
+    delivery_id: cb.delivery_id,
+    workflow_run_url: wf.url ?? "",
+    workflow_name: wf.name,
+    job_name: jobName,
+    check_run_id: checkRunId,
+    run_id: wf.run_id ?? "",
+    run_attempt: runAttempt,
+  };
+
+  if (trusted.downstream_repo_level) {
+    record.downstream_repo_level = trusted.downstream_repo_level;
+  }
+
+  // Only set timing metrics when the relay provides a non-null value.
+  // in_progress sets queue_time; completed sets execution_time.
+  // Using UpdateItem ensures the completed callback doesn't clobber
+  // queue_time with null.
+  if (trusted.ci_metrics?.queue_time != null) {
+    record.queue_time = trusted.ci_metrics.queue_time;
+  }
+  if (trusted.ci_metrics?.execution_time != null) {
+    record.execution_time = trusted.ci_metrics.execution_time;
+  }
+
+  // Use downstream-reported timestamps, not HUD wall-clock time
+  if (wf.started_at) {
+    record.started_at = wf.started_at;
+  }
+
+  if (wf.artifact_url) {
+    record.artifact_url = wf.artifact_url;
+  }
+
+  if (wf.status === "completed") {
+    record.conclusion = wf.conclusion ?? undefined;
+    if (wf.completed_at) {
+      record.completed_at = wf.completed_at;
+    }
+
+    if (wf.test_results) {
+      const tr = wf.test_results;
+      if (typeof tr.passed === "number") record.passed_tests = tr.passed;
+      if (typeof tr.failed === "number") record.failed_tests = tr.failed;
+      if (typeof tr.skipped === "number") record.skipped_tests = tr.skipped;
+      record.total_tests =
+        typeof tr.total === "number"
+          ? tr.total
+          : (tr.passed ?? 0) + (tr.failed ?? 0) + (tr.skipped ?? 0);
+    }
+  }
+
+  return record;
+}
+
+// ---- DynamoDB Write (UpdateItem) ----
+
+export async function writeToDynamo(
+  record: OotWorkflowJobRecord
+): Promise<void> {
+  const client = getDynamoClient();
+
+  // Build SET expression dynamically — only set non-undefined fields.
+  // This prevents completed callbacks from clobbering in_progress-only
+  // fields (queue_time, started_at) with null.
+  const expressionParts: string[] = [];
+  const expressionValues: Record<string, any> = {};
+  const expressionNames: Record<string, string> = {};
+
+  for (const [key, value] of Object.entries(record)) {
+    if (key === "dynamoKey" || value === undefined) continue;
+    const placeholder = `:v_${key}`;
+    const nameAlias = `#n_${key}`;
+    expressionParts.push(`${nameAlias} = ${placeholder}`);
+    expressionValues[placeholder] = value;
+    expressionNames[nameAlias] = key;
+  }
+
+  await client.update({
+    TableName: OOT_TABLE,
+    Key: { dynamoKey: record.dynamoKey },
+    UpdateExpression: `SET ${expressionParts.join(", ")}`,
+    ExpressionAttributeValues: expressionValues,
+    ExpressionAttributeNames: expressionNames,
+  });
+}
+
+// ---- UI Helpers ----
+
+export type ChipColor = "success" | "error" | "warning" | "info" | "default";
+
+export function conclusionColor(status: string, conclusion: string): ChipColor {
+  if (status === "in_progress") return "info";
+  switch (conclusion) {
+    case "success":
+      return "success";
+    case "failure":
+      return "error";
+    case "cancelled":
+    case "timed_out":
+      return "warning";
+    default:
+      return "default";
+  }
+}
+
+export function conclusionLabel(status: string, conclusion: string): string {
+  if (status === "in_progress") return "running";
+  return conclusion || status;
+}
+
+// ---- Error Helper ----
+
+export class ApiError extends Error {
+  statusCode: number;
+  constructor(statusCode: number, message: string) {
+    super(message);
+    this.statusCode = statusCode;
+  }
+}