Upgrade to to the latest electric sql client and server

Author: ericallam

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

@@ -70,7 +70,7 @@ export class RealtimeClient {
   #constructElectricUrl(url: URL | string, whereClause: string): URL {
     const $url = new URL(url.toString());
 
-    const electricUrl = new URL(`${this.options.electricOrigin}/v1/shape/public."TaskRun"`);
+    const electricUrl = new URL(`${this.options.electricOrigin}/v1/shape`);
 
     // Copy over all the url search params to the electric url
     $url.searchParams.forEach((value, key) => {
@@ -86,6 +86,7 @@ export class RealtimeClient {
     // });
 
     electricUrl.searchParams.set("where", whereClause);
+    electricUrl.searchParams.set("table", 'public."TaskRun"');
 
     return electricUrl;
   }
@@ -231,7 +232,7 @@ export class RealtimeClient {
 }
 
 function extractShapeId(url: URL) {
-  return url.searchParams.get("shape_id");
+  return url.searchParams.get("handle");
 }
 
 function isLiveRequestUrl(url: URL) {

apps/webapp/test/realtimeClient.test.ts

@@ -86,8 +86,10 @@ describe("RealtimeClient", () => {
 
       const headers = Object.fromEntries(response.headers.entries());
 
-      const shapeId = headers["electric-shape-id"];
-      const chunkOffset = headers["electric-chunk-last-offset"];
+      console.log(headers);
+
+      const shapeId = headers["electric-handle"];
+      const chunkOffset = headers["electric-offset"];
 
       expect(response.status).toBe(200);
       expect(response2.status).toBe(200);
@@ -96,15 +98,15 @@ describe("RealtimeClient", () => {
 
       // Okay, now we will do two live requests, and the second one should fail because of the concurrency limit
       const liveResponsePromise = client.streamRun(
-        `http://localhost:3000?offset=0_0&live=true&shape_id=${shapeId}`,
+        `http://localhost:3000?offset=0_0&live=true&handle=${shapeId}`,
         environment,
         run.id
       );
 
       const liveResponsePromise2 = new Promise<Response>((resolve) => {
         setTimeout(async () => {
           const response = await client.streamRun(
-            `http://localhost:3000?offset=0_0&live=true&shape_id=${shapeId}`,
+            `http://localhost:3000?offset=0_0&live=true&handle=${shapeId}`,
             environment,
             run.id
           );
@@ -200,8 +202,8 @@ describe("RealtimeClient", () => {
 
       const headers = Object.fromEntries(response.headers.entries());
 
-      const shapeId = headers["electric-shape-id"];
-      const chunkOffset = headers["electric-chunk-last-offset"];
+      const shapeId = headers["electric-handle"];
+      const chunkOffset = headers["electric-offset"];
 
       expect(response.status).toBe(200);
       expect(shapeId).toBeDefined();

docker/docker-compose.yml

@@ -61,7 +61,7 @@ services:
       - 6379:6379
 
   electric:
-    image: electricsql/electric:0.7.5
+    image: electricsql/electric:0.8.1
     restart: always
     environment:
       DATABASE_URL: postgresql://postgres:postgres@database:5432/postgres?sslmode=disable

internal-packages/testcontainers/src/utils.ts

@@ -55,7 +55,7 @@ export async function createElectricContainer(
     network.getName()
   )}:5432/${postgresContainer.getDatabase()}?sslmode=disable`;
 
-  const container = await new GenericContainer("electricsql/electric:0.7.5")
+  const container = await new GenericContainer("electricsql/electric:0.8.1")
     .withExposedPorts(3000)
     .withNetwork(network)
     .withEnvironment({

packages/core/package.json

@@ -182,7 +182,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@electric-sql/client": "0.6.3",
+    "@electric-sql/client": "0.7.1",
     "@google-cloud/precise-date": "^4.0.0",
     "@jsonhero/path": "^1.0.21",
     "@opentelemetry/api": "1.9.0",

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

@@ -23,14 +23,14 @@ export async function zodShapeStream<TShapeSchema extends z.ZodTypeAny>(
 
   const shape = new Shape(stream);
 
-  const initialValue = await shape.value;
+  const initialRows = await shape.rows;
 
-  for (const shapeRow of initialValue.values()) {
+  for (const shapeRow of initialRows) {
     await callback(schema.parse(shapeRow));
   }
 
   return shape.subscribe(async (newShape) => {
-    for (const shapeRow of newShape.values()) {
+    for (const shapeRow of newShape.rows) {
       await callback(schema.parse(shapeRow));
     }
   });