Make realtime server backwards compat with 3.1.2 release

Author: ericallam

apps/webapp/app/routes/realtime.v1.batches.$batchId.ts

@@ -31,6 +31,11 @@ export const loader = createLoaderApiRoute(
       return json({ error: "Batch not found" }, { status: 404 });
     }
 
-    return realtimeClient.streamBatch(request.url, authentication.environment, batchRun.id);
+    return realtimeClient.streamBatch(
+      request.url,
+      authentication.environment,
+      batchRun.id,
+      request.headers.get("x-trigger-electric-version") ?? undefined
+    );
   }
 );

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

@@ -31,6 +31,11 @@ export const loader = createLoaderApiRoute(
       return json({ error: "Run not found" }, { status: 404 });
     }
 
-    return realtimeClient.streamRun(request.url, authentication.environment, run.id);
+    return realtimeClient.streamRun(
+      request.url,
+      authentication.environment,
+      run.id,
+      request.headers.get("x-trigger-electric-version") ?? undefined
+    );
   }
 );

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

@@ -23,6 +23,11 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ searchParams, authentication, request }) => {
-    return realtimeClient.streamRuns(request.url, authentication.environment, searchParams);
+    return realtimeClient.streamRuns(
+      request.url,
+      authentication.environment,
+      searchParams,
+      request.headers.get("x-trigger-electric-version") ?? undefined
+    );
   }
 );

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

@@ -37,18 +37,29 @@ export class RealtimeClient {
     this.#registerCommands();
   }
 
-  async streamRun(url: URL | string, environment: RealtimeEnvironment, runId: string) {
-    return this.#streamRunsWhere(url, environment, `id='${runId}'`);
+  async streamRun(
+    url: URL | string,
+    environment: RealtimeEnvironment,
+    runId: string,
+    clientVersion?: string
+  ) {
+    return this.#streamRunsWhere(url, environment, `id='${runId}'`, clientVersion);
   }
 
-  async streamBatch(url: URL | string, environment: RealtimeEnvironment, batchId: string) {
-    return this.#streamRunsWhere(url, environment, `"batchId"='${batchId}'`);
+  async streamBatch(
+    url: URL | string,
+    environment: RealtimeEnvironment,
+    batchId: string,
+    clientVersion?: string
+  ) {
+    return this.#streamRunsWhere(url, environment, `"batchId"='${batchId}'`, clientVersion);
   }
 
   async streamRuns(
     url: URL | string,
     environment: RealtimeEnvironment,
-    params: RealtimeRunsParams
+    params: RealtimeRunsParams,
+    clientVersion?: string
   ) {
     const whereClauses: string[] = [`"runtimeEnvironmentId"='${environment.id}'`];
 
@@ -58,16 +69,21 @@ export class RealtimeClient {
 
     const whereClause = whereClauses.join(" AND ");
 
-    return this.#streamRunsWhere(url, environment, whereClause);
+    return this.#streamRunsWhere(url, environment, whereClause, clientVersion);
   }
 
-  async #streamRunsWhere(url: URL | string, environment: RealtimeEnvironment, whereClause: string) {
-    const electricUrl = this.#constructElectricUrl(url, whereClause);
+  async #streamRunsWhere(
+    url: URL | string,
+    environment: RealtimeEnvironment,
+    whereClause: string,
+    clientVersion?: string
+  ) {
+    const electricUrl = this.#constructElectricUrl(url, whereClause, clientVersion);
 
-    return this.#performElectricRequest(electricUrl, environment);
+    return this.#performElectricRequest(electricUrl, environment, clientVersion);
   }
 
-  #constructElectricUrl(url: URL | string, whereClause: string): URL {
+  #constructElectricUrl(url: URL | string, whereClause: string, clientVersion?: string): URL {
     const $url = new URL(url.toString());
 
     const electricUrl = new URL(`${this.options.electricOrigin}/v1/shape`);
@@ -77,36 +93,42 @@ export class RealtimeClient {
       electricUrl.searchParams.set(key, value);
     });
 
-    // const electricParams = ["shape_id", "live", "offset", "columns", "cursor"];
-
-    // electricParams.forEach((param) => {
-    //   if ($url.searchParams.has(param) && $url.searchParams.get(param)) {
-    //     electricUrl.searchParams.set(param, $url.searchParams.get(param)!);
-    //   }
-    // });
-
     electricUrl.searchParams.set("where", whereClause);
     electricUrl.searchParams.set("table", 'public."TaskRun"');
 
+    if (!clientVersion) {
+      // If the client version is not provided, that means we're using an older client
+      // This means the client will be sending shape_id instead of handle
+      electricUrl.searchParams.set("handle", electricUrl.searchParams.get("shape_id") ?? "");
+    }
+
     return electricUrl;
   }
 
-  async #performElectricRequest(url: URL, environment: RealtimeEnvironment) {
+  async #performElectricRequest(
+    url: URL,
+    environment: RealtimeEnvironment,
+    clientVersion?: string
+  ) {
     const shapeId = extractShapeId(url);
 
     logger.debug("[realtimeClient] request", {
       url: url.toString(),
     });
 
+    const rewriteResponseHeaders: Record<string, string> = clientVersion
+      ? {}
+      : { "electric-handle": "electric-shape-id", "electric-offset": "electric-chunk-last-offset" };
+
     if (!shapeId) {
       // If the shapeId is not present, we're just getting the initial value
-      return longPollingFetch(url.toString());
+      return longPollingFetch(url.toString(), {}, rewriteResponseHeaders);
     }
 
     const isLive = isLiveRequestUrl(url);
 
     if (!isLive) {
-      return longPollingFetch(url.toString());
+      return longPollingFetch(url.toString(), {}, rewriteResponseHeaders);
     }
 
     const requestId = randomUUID();
@@ -148,7 +170,7 @@ export class RealtimeClient {
 
     try {
       // ... (rest of your existing code for the long polling request)
-      const response = await longPollingFetch(url.toString());
+      const response = await longPollingFetch(url.toString(), {}, rewriteResponseHeaders);
 
       // Decrement the counter after the long polling request is complete
       await this.#decrementConcurrency(environment.id, requestId);

apps/webapp/app/utils/longPollingFetch.ts

@@ -6,20 +6,44 @@
 import { logger } from "~/services/logger.server";
 
 // Similar-ish problem to https://github.com/wintercg/fetch/issues/23
-export async function longPollingFetch(url: string, options?: RequestInit) {
+export async function longPollingFetch(
+  url: string,
+  options?: RequestInit,
+  rewriteResponseHeaders?: Record<string, string>
+) {
   try {
     let response = await fetch(url, options);
 
     if (response.headers.get("content-encoding")) {
       const headers = new Headers(response.headers);
       headers.delete("content-encoding");
       headers.delete("content-length");
+
+      response = new Response(response.body, {
+        headers,
+        status: response.status,
+        statusText: response.statusText,
+      });
+    }
+
+    if (rewriteResponseHeaders) {
+      const headers = new Headers(response.headers);
+
+      for (const [fromKey, toKey] of Object.entries(rewriteResponseHeaders)) {
+        const value = headers.get(fromKey);
+        if (value) {
+          headers.set(toKey, value);
+          headers.delete(fromKey);
+        }
+      }
+
       response = new Response(response.body, {
         headers,
         status: response.status,
         statusText: response.statusText,
       });
     }
+
     return response;
   } catch (error) {
     if (error instanceof TypeError) {