Add createdAt filter to realtime subscribing with tags

Author: ericallam

apps/webapp/app/routes/realtime.v1.runs.ts

@@ -9,6 +9,7 @@ const SearchParamsSchema = z.object({
     .transform((value) => {
       return value ? value.split(",") : undefined;
     }),
+  createdAt: z.string().optional(),
 });
 
 export const loader = createLoaderApiRoute(

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

@@ -1,14 +1,18 @@
 import { json } from "@remix-run/server-runtime";
-import Redis, { Callback, Result, type RedisOptions } from "ioredis";
+import { tryCatch } from "@trigger.dev/core/utils";
+import { safeParseNaturalLanguageDurationAgo } from "@trigger.dev/core/v3/isomorphic";
+import { Callback, Result } from "ioredis";
 import { randomUUID } from "node:crypto";
+import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
 import { longPollingFetch } from "~/utils/longPollingFetch";
 import { logger } from "./logger.server";
-import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
 
 export interface CachedLimitProvider {
   getCachedLimit: (organizationId: string, defaultValue: number) => Promise<number | undefined>;
 }
 
+const MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS = 7 * 24 * 60 * 60 * 1000;
+
 export type RealtimeClientOptions = {
   electricOrigin: string;
   redis: RedisWithClusterOptions;
@@ -24,6 +28,7 @@ export type RealtimeEnvironment = {
 
 export type RealtimeRunsParams = {
   tags?: string[];
+  createdAt?: string;
 };
 
 export class RealtimeClient {
@@ -92,9 +97,91 @@ export class RealtimeClient {
       whereClauses.push(`"runTags" @> ARRAY[${params.tags.map((t) => `'${t}'`).join(",")}]`);
     }
 
+    const createdAtFilter = await this.#calculateCreatedAtFilter(url, params.createdAt);
+
+    if (createdAtFilter) {
+      whereClauses.push(`"createdAt" > '${createdAtFilter.toISOString()}'`);
+    }
+
     const whereClause = whereClauses.join(" AND ");
 
-    return this.#streamRunsWhere(url, environment, whereClause, clientVersion);
+    const response = await this.#streamRunsWhere(url, environment, whereClause, clientVersion);
+
+    if (createdAtFilter) {
+      const [setCreatedAtFilterError] = await tryCatch(
+        this.#setCreatedAtFilterFromResponse(response, createdAtFilter)
+      );
+
+      if (setCreatedAtFilterError) {
+        logger.error("[realtimeClient] Failed to set createdAt filter", {
+          error: setCreatedAtFilterError,
+          createdAtFilter,
+          responseHeaders: Object.fromEntries(response.headers.entries()),
+          responseStatus: response.status,
+        });
+      }
+    }
+
+    return response;
+  }
+
+  async #calculateCreatedAtFilter(url: URL | string, createdAt?: string) {
+    const duration = createdAt ?? "24h";
+    const $url = new URL(url.toString());
+    const shapeId = extractShapeId($url);
+
+    if (!shapeId) {
+      // This means we need to calculate the createdAt filter and store it in redis after we get back the response
+      const createdAtFilter = safeParseNaturalLanguageDurationAgo(duration);
+
+      // Validate that the createdAt filter is in the past, and not more than 1 week in the past.
+      // if it's more than 1 week in the past, just return 1 week ago Date
+      if (
+        createdAtFilter &&
+        createdAtFilter < new Date(Date.now() - MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS)
+      ) {
+        return new Date(Date.now() - MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS);
+      }
+
+      return createdAtFilter;
+    } else {
+      // We need to get the createdAt filter value from redis, if there is none we need to return undefined
+      const [createdAtFilterError, createdAtFilter] = await tryCatch(
+        this.#getCreatedAtFilter(shapeId)
+      );
+
+      if (createdAtFilterError) {
+        logger.error("[realtimeClient] Failed to get createdAt filter", {
+          shapeId,
+          error: createdAtFilterError,
+        });
+
+        return;
+      }
+
+      return createdAtFilter;
+    }
+  }
+
+  async #getCreatedAtFilter(shapeId: string) {
+    // TODO: replace this with unkey cache so we can use in memory
+    const createdAtFilterRawValue = await this.redis.get(`shapes:${shapeId}:filters:createdAt`);
+
+    if (!createdAtFilterRawValue) {
+      return;
+    }
+
+    return new Date(createdAtFilterRawValue);
+  }
+
+  async #setCreatedAtFilterFromResponse(response: Response, createdAtFilter: Date) {
+    const shapeId = extractShapeIdFromResponse(response);
+
+    if (!shapeId) {
+      return;
+    }
+
+    await this.redis.set(`shapes:${shapeId}:filters:createdAt`, createdAtFilter.toISOString());
   }
 
   async #streamRunsWhere(
@@ -172,6 +259,8 @@ export class RealtimeClient {
   ) {
     const shapeId = extractShapeId(url);
 
+    url = await this.#handleCreatedAtFilter(url, shapeId);
+
     logger.debug("[realtimeClient] request", {
       url: url.toString(),
     });
@@ -232,6 +321,10 @@ export class RealtimeClient {
       // ... (rest of your existing code for the long polling request)
       const response = await longPollingFetch(url.toString(), { signal }, rewriteResponseHeaders);
 
+      // If this is the initial request, the response.headers['electric-handle'] will be the shapeId
+      // And we may need to set the "createdAt" filter timestamp keyed by the shapeId
+      // Then in the next request, we will get the createdAt timestamp value via the shapeId and use it to filter the results
+
       // Decrement the counter after the long polling request is complete
       await this.#decrementConcurrency(environment.id, requestId);
 
@@ -244,6 +337,10 @@ export class RealtimeClient {
     }
   }
 
+  async #handleCreatedAtFilter(url: URL, shapeId?: string | null) {
+    return url;
+  }
+
   async #incrementAndCheck(environmentId: string, requestId: string, limit: number) {
     const key = this.#getKey(environmentId);
     const now = Date.now();
@@ -314,7 +411,11 @@ export class RealtimeClient {
 }
 
 function extractShapeId(url: URL) {
-  return url.searchParams.get("handle");
+  return url.searchParams.get("handle") ?? url.searchParams.get("shape_id");
+}
+
+function extractShapeIdFromResponse(response: Response) {
+  return response.headers.get("electric-handle");
 }
 
 function isLiveRequestUrl(url: URL) {

packages/core/src/v3/apiClient/index.ts

@@ -904,10 +904,12 @@ export class ApiClient {
 
   subscribeToRunsWithTag<TRunTypes extends AnyRunTypes>(
     tag: string | string[],
+    filters?: { createdAt?: string },
     options?: { signal?: AbortSignal; onFetchError?: (error: Error) => void }
   ) {
     const searchParams = createSearchQueryForSubscribeToRuns({
       tags: tag,
+      ...(filters ? { createdAt: filters.createdAt } : {}),
     });
 
     return runShapeStream<TRunTypes>(
@@ -1043,6 +1045,10 @@ function createSearchQueryForSubscribeToRuns(query?: SubscribeToRunsQueryParams)
     if (query.tags) {
       searchParams.append("tags", Array.isArray(query.tags) ? query.tags.join(",") : query.tags);
     }
+
+    if (query.createdAt) {
+      searchParams.append("createdAt", query.createdAt);
+    }
   }
 
   return searchParams;

packages/core/src/v3/apiClient/types.ts

@@ -41,6 +41,7 @@ export interface ListProjectRunsQueryParams extends CursorPageParams, ListRunsQu
 export interface SubscribeToRunsQueryParams {
   tasks?: Array<string> | string;
   tags?: Array<string> | string;
+  createdAt?: string;
 }
 
 export interface ListWaitpointTokensQueryParams extends CursorPageParams {

packages/core/src/v3/isomorphic/duration.ts

@@ -1,55 +1,149 @@
 export function parseNaturalLanguageDuration(duration: string): Date | undefined {
-  const regexPattern = /^(\d+w)?(\d+d)?(\d+h)?(\d+m)?(\d+s)?$/;
+  // More flexible regex that captures all units individually regardless of order
+  const weekMatch = duration.match(/(\d+)w/);
+  const dayMatch = duration.match(/(\d+)d/);
+  const hourMatch = duration.match(/(\d+)(?:hr|h)/);
+  const minuteMatch = duration.match(/(\d+)m/);
+  const secondMatch = duration.match(/(\d+)s/);
 
-  const result: Date = new Date();
+  // Check if the entire string consists only of valid duration units
+  const validPattern = /^(\d+(?:w|d|hr|h|m|s))+$/;
+  if (!validPattern.test(duration)) {
+    return undefined;
+  }
+
+  let totalMilliseconds = 0;
   let hasMatch = false;
 
-  const elements = duration.match(regexPattern);
-  if (elements) {
-    if (elements[1]) {
-      const weeks = Number(elements[1].slice(0, -1));
-      if (weeks >= 0) {
-        result.setDate(result.getDate() + 7 * weeks);
-        hasMatch = true;
-      }
+  if (weekMatch) {
+    const weeks = Number(weekMatch[1]);
+    if (weeks >= 0) {
+      totalMilliseconds += weeks * 7 * 24 * 60 * 60 * 1000;
+      hasMatch = true;
     }
-    if (elements[2]) {
-      const days = Number(elements[2].slice(0, -1));
-      if (days >= 0) {
-        result.setDate(result.getDate() + days);
-        hasMatch = true;
-      }
+  }
+
+  if (dayMatch) {
+    const days = Number(dayMatch[1]);
+    if (days >= 0) {
+      totalMilliseconds += days * 24 * 60 * 60 * 1000;
+      hasMatch = true;
     }
-    if (elements[3]) {
-      const hours = Number(elements[3].slice(0, -1));
-      if (hours >= 0) {
-        result.setHours(result.getHours() + hours);
-        hasMatch = true;
-      }
+  }
+
+  if (hourMatch) {
+    const hours = Number(hourMatch[1]);
+    if (hours >= 0) {
+      totalMilliseconds += hours * 60 * 60 * 1000;
+      hasMatch = true;
     }
-    if (elements[4]) {
-      const minutes = Number(elements[4].slice(0, -1));
-      if (minutes >= 0) {
-        result.setMinutes(result.getMinutes() + minutes);
-        hasMatch = true;
-      }
+  }
+
+  if (minuteMatch) {
+    const minutes = Number(minuteMatch[1]);
+    if (minutes >= 0) {
+      totalMilliseconds += minutes * 60 * 1000;
+      hasMatch = true;
     }
-    if (elements[5]) {
-      const seconds = Number(elements[5].slice(0, -1));
-      if (seconds >= 0) {
-        result.setSeconds(result.getSeconds() + seconds);
-        hasMatch = true;
-      }
+  }
+
+  if (secondMatch) {
+    const seconds = Number(secondMatch[1]);
+    if (seconds >= 0) {
+      totalMilliseconds += seconds * 1000;
+      hasMatch = true;
     }
   }
 
   if (hasMatch) {
-    return result;
+    return new Date(Date.now() + totalMilliseconds);
   }
 
   return undefined;
 }
 
+export function safeParseNaturalLanguageDuration(duration: string): Date | undefined {
+  try {
+    return parseNaturalLanguageDuration(duration);
+  } catch (error) {
+    return undefined;
+  }
+}
+
+// ... existing code ...
+
+export function parseNaturalLanguageDurationAgo(duration: string): Date | undefined {
+  // More flexible regex that captures all units individually regardless of order
+  const weekMatch = duration.match(/(\d+)w/);
+  const dayMatch = duration.match(/(\d+)d/);
+  const hourMatch = duration.match(/(\d+)(?:hr|h)/);
+  const minuteMatch = duration.match(/(\d+)m/);
+  const secondMatch = duration.match(/(\d+)s/);
+
+  // Check if the entire string consists only of valid duration units
+  const validPattern = /^(\d+(?:w|d|hr|h|m|s))+$/;
+  if (!validPattern.test(duration)) {
+    return undefined;
+  }
+
+  let totalMilliseconds = 0;
+  let hasMatch = false;
+
+  if (weekMatch) {
+    const weeks = Number(weekMatch[1]);
+    if (weeks >= 0) {
+      totalMilliseconds += weeks * 7 * 24 * 60 * 60 * 1000;
+      hasMatch = true;
+    }
+  }
+
+  if (dayMatch) {
+    const days = Number(dayMatch[1]);
+    if (days >= 0) {
+      totalMilliseconds += days * 24 * 60 * 60 * 1000;
+      hasMatch = true;
+    }
+  }
+
+  if (hourMatch) {
+    const hours = Number(hourMatch[1]);
+    if (hours >= 0) {
+      totalMilliseconds += hours * 60 * 60 * 1000;
+      hasMatch = true;
+    }
+  }
+
+  if (minuteMatch) {
+    const minutes = Number(minuteMatch[1]);
+    if (minutes >= 0) {
+      totalMilliseconds += minutes * 60 * 1000;
+      hasMatch = true;
+    }
+  }
+
+  if (secondMatch) {
+    const seconds = Number(secondMatch[1]);
+    if (seconds >= 0) {
+      totalMilliseconds += seconds * 1000;
+      hasMatch = true;
+    }
+  }
+
+  if (hasMatch) {
+    return new Date(Date.now() - totalMilliseconds);
+  }
+
+  return undefined;
+}
+
+export function safeParseNaturalLanguageDurationAgo(duration: string): Date | undefined {
+  try {
+    return parseNaturalLanguageDurationAgo(duration);
+  } catch (error) {
+    return undefined;
+  }
+}
+
 export function stringifyDuration(seconds: number): string | undefined {
   if (seconds <= 0) {
     return;