fix(webapp,llm-model-catalog): bound realtime and analytics inputs

Mono-RevId: 3064956970461fff4c7221c24116e20fbeab7685

Author: carderne

apps/webapp/app/presenters/v3/ErrorGroupPresenter.server.ts

@@ -1,5 +1,5 @@
 import { type ClickHouse, msToClickHouseInterval } from "@internal/clickhouse";
-import { TimeGranularity } from "~/utils/timeGranularity";
+import { isBoundedChartRange, TimeGranularity } from "~/utils/timeGranularity";
 import { ErrorId } from "@trigger.dev/core/v3/isomorphic";
 import { type ErrorGroupStatus, type PrismaClientOrTransaction } from "@trigger.dev/database";
 import { timeFilterFromTo } from "~/components/runs/v3/SharedFilters";
@@ -187,6 +187,9 @@ export class ErrorGroupPresenter extends BasePresenter {
     versions: string[];
   }> {
     const granularityMs = errorGroupGranularity.getTimeGranularityMs(from, to);
+    if (!isBoundedChartRange(from, to, granularityMs)) {
+      return { data: [], versions: [] };
+    }
     const intervalExpr = msToClickHouseInterval(granularityMs);
 
     const queryBuilder =

apps/webapp/app/presenters/v3/ErrorsListPresenter.server.ts

@@ -1,6 +1,6 @@
 import { z } from "zod";
 import { type ClickHouse, msToClickHouseInterval } from "@internal/clickhouse";
-import { TimeGranularity } from "~/utils/timeGranularity";
+import { isBoundedChartRange, TimeGranularity } from "~/utils/timeGranularity";
 
 const errorsListGranularity = new TimeGranularity([
   { max: "2h", granularity: "1m" },
@@ -358,6 +358,9 @@ export class ErrorsListPresenter extends BasePresenter {
     }
 
     const granularityMs = errorsListGranularity.getTimeGranularityMs(from, to);
+    if (!isBoundedChartRange(from, to, granularityMs)) {
+      return { data: {} };
+    }
     const intervalExpr = msToClickHouseInterval(granularityMs);
 
     const queryBuilder = this.clickhouse.errors.createOccurrencesQueryBuilder(intervalExpr);

apps/webapp/app/routes/tests.sse.stream.ts

@@ -6,6 +6,7 @@ import type { Request as ExpressRequest, Response as ExpressResponse, NextFuncti
 import { createHash } from "node:crypto";
 import { z } from "zod";
 import type { RedisWithClusterOptions } from "~/redis.server";
+import { getRouterPath } from "~/utils/sanitizeHttpUrl";
 import { logger } from "./logger.server";
 import type { Duration, Limiter } from "./rateLimiter.server";
 import { createRedisRateLimitClient, RateLimiter } from "./rateLimiter.server";
@@ -229,9 +230,10 @@ export function authorizationRateLimitMiddleware({
       return next();
     }
 
-    //first check if any of the pathMatchers match the request path
-    const path = req.path;
+    // Use the same normalized pathname that the application router receives.
+    const path = getRouterPath(req);
     if (
+      !path ||
       !pathMatchers.some((matcher) =>
         matcher instanceof RegExp ? matcher.test(path) : path === matcher
       )

apps/webapp/app/routes/tests.sse.tsx

@@ -23,9 +23,8 @@ export const engineRateLimiter = authorizationRateLimitMiddleware({
     stale: 60_000 * 20, // Date is stale after 20 minutes
     maxItems: 1000,
   },
-  pathMatchers: [/^\/engine/],
-  // Regex allow any path starting with /engine/v1/worker-actions/
-  pathWhiteList: [/^\/engine\/v1\/worker-actions\/.*/],
+  pathMatchers: [/^\/engine(?:\/|$)/],
+  pathWhiteList: [/^\/engine\/v1\/worker-actions(?:\/|$)/],
   log: {
     rejections: env.RUN_ENGINE_RATE_LIMIT_REJECTION_LOGS_ENABLED === "1",
     requests: env.RUN_ENGINE_RATE_LIMIT_REQUEST_LOGS_ENABLED === "1",

apps/webapp/app/services/authorizationRateLimitMiddleware.server.ts

@@ -22,6 +22,17 @@ type StreamChunk =
   | { type: "data"; redisId: string; data: string }
   | { type: "legacy-data"; redisId: string; data: string };
 
+function formatSseDataEvent(redisId: string, data: string): string {
+  const withoutDelimiterCarriageReturn = data.endsWith("\r") ? data.slice(0, -1) : data;
+  const normalized = withoutDelimiterCarriageReturn.replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+  const dataFields = normalized
+    .split("\n")
+    .map((part) => `data: ${part}`)
+    .join("\n");
+
+  return `id: ${redisId}\n${dataFields}\n\n`;
+}
+
 // Class implementing both interfaces
 export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
   private logger: Logger;
@@ -277,8 +288,7 @@ export class RedisRealtimeStreams implements StreamIngestor, StreamResponder {
             if (chunk.type === "ping") {
               controller.enqueue(`: ping\n\n`);
             } else if ((chunk.type === "data" || chunk.type === "legacy-data") && chunk.line) {
-              // Use Redis stream ID as SSE event ID
-              controller.enqueue(`id: ${chunk.redisId}\ndata: ${chunk.line}\n\n`);
+              controller.enqueue(formatSseDataEvent(chunk.redisId, chunk.line));
             }
           },
         })

apps/webapp/app/services/engineRateLimit.server.ts

@@ -1,5 +1,38 @@
 const AUTHORIZATION_CODE_PATH_PREFIX = "/account/authorization-code/";
 const REDACTED_AUTHORIZATION_CODE_PATH = "/account/authorization-code/[redacted]";
+const ROUTER_PATH = Symbol.for("trigger.routerPath");
+
+type RequestTarget = {
+  originalUrl?: string;
+  url?: string;
+  [ROUTER_PATH]?: string | null;
+};
+
+export function getRouterPath(request: RequestTarget): string | undefined {
+  const cached = request[ROUTER_PATH];
+  if (cached !== undefined) {
+    return cached ?? undefined;
+  }
+
+  const target = request.originalUrl ?? request.url;
+  if (!target) {
+    request[ROUTER_PATH] = null;
+    return undefined;
+  }
+
+  try {
+    const pathname = new URL(target, "http://localhost").pathname;
+    request[ROUTER_PATH] = pathname;
+    return pathname;
+  } catch {
+    request[ROUTER_PATH] = null;
+    return undefined;
+  }
+}
+
+export function pathHasPrefix(pathname: string, prefix: string): boolean {
+  return pathname === prefix || pathname.startsWith(`${prefix}/`);
+}
 
 export function sanitizeHttpUrl(value: string): string {
   const queryIndex = value.indexOf("?");

apps/webapp/app/services/realtime/redisRealtimeStreams.server.ts

@@ -59,3 +59,21 @@ export class TimeGranularity {
     return this.parsed[this.parsed.length - 1].granularityMs;
   }
 }
+
+const MAX_CHART_BUCKETS = 10_000;
+
+/**
+ * Rejects user-supplied chart ranges before they reach a query or a bucket
+ * loop: non-finite or inverted dates, or a span that would generate an
+ * unbounded number of buckets at the chosen granularity.
+ */
+export function isBoundedChartRange(from: Date, to: Date, granularityMs: number): boolean {
+  const fromMs = from.getTime();
+  const toMs = to.getTime();
+  return (
+    Number.isFinite(fromMs) &&
+    Number.isFinite(toMs) &&
+    fromMs <= toMs &&
+    (toMs - fromMs) / granularityMs <= MAX_CHART_BUCKETS
+  );
+}