Nodejs pglite api

.env.sample

@@ -1,3 +1,5 @@
 ELECTRIC_URL=https://api.electric-sql.cloud
+ELECTRIC_SOURCE_ID=
+ELECTRIC_SOURCE_SECRET=
 NEON_API_KEY=
 

durable-object/sst-env.d.ts

@@ -6,6 +6,15 @@
 import "sst"
 declare module "sst" {
   export interface Resource {
+    "ElectricUrl": {
+      "type": "sst.sst.Linkable"
+      "url": string
+    }
+    "FastifyApi": {
+      "name": string
+      "type": "sst.aws.Function"
+      "url": string
+    }
     "examples-infra-shared-examplesInfraVpcShared": {
       "type": "sst.aws.Vpc"
     }

package.json

@@ -4,15 +4,18 @@
   "description": "",
   "main": "index.js",
   "scripts": {
-    "db:load-data": "dotenv -e .env -- node ./db/load-data.mjs",
+    "db:load-data": "dotenv -e .env -- node ./db/load-data.js",
     "test": "echo \"Error: no test specified\" && exit 1"
   },
   "keywords": [],
   "author": "",
   "license": "ISC",
   "dependencies": {
     "@electric-sql/client": "1.0.0-beta.3",
+    "@electric-sql/pglite": "^0.2.16",
+    "@electric-sql/pglite-sync": "^0.2.18",
     "dotenv": "^16.4.7",
+    "fastify": "^5.2.1",
     "postgres": "^3.4.5",
     "redis": "^4.7.0",
     "sst": "^3.7.2",
@@ -25,8 +28,10 @@
     "@faker-js/faker": "^9.4.0",
     "@types/pg": "^8.11.11",
     "@types/uuid": "^10.0.0",
+    "@types/uuid-validate": "^0.0.3",
     "camelcase": "^8.0.0",
     "dotenv-cli": "^8.0.0",
-    "tsx": "^4.19.2"
+    "tsx": "^4.19.2",
+    "uuid-validate": "^0.0.3"
   }
 }

server/app.mjs

@@ -0,0 +1,139 @@
+import { electricSync } from "@electric-sql/pglite-sync";
+import { PGlite } from "@electric-sql/pglite";
+import { live } from "@electric-sql/pglite/live";
+import dotenv from "dotenv";
+
+import postgres from "postgres";
+
+dotenv.config();
+
+import { Bench } from "tinybench";
+
+import { Resource } from "sst";
+
+import Fastify from "fastify";
+import validate from "uuid-validate";
+
+import { generateAndSyncToElectric } from "./populate-pglite.js";
+
+const PORT = 8000;
+
+const fastify = Fastify({
+  logger: true,
+});
+
+//instantiate pglite electric sync
+try {
+  let runBenchmarks = false;
+
+  const db = await PGlite.create({
+    extensions: {
+      live,
+      electric: electricSync({
+        debug: true,
+      }),
+    },
+  });
+
+  console.log("Waiting for db to be ready");
+  await db.waitReady;
+
+  console.log("Syncing shapes");
+
+  let initialSyncStart = Date.now;
+  await generateAndSyncToElectric(
+    db,
+    "https://api.electric-sql.cloud", //Resource.ElectricUrl.url,
+    process.env.SOURCE_ID, //Resource.ElectricUrl.sourceId,
+    process.env.SOURCE_SECRET, //Resource.ElectricUrl.sourceSecret,
+    Bench,
+  );
+
+  //post sync
+  const duration = Date.now() - initialSyncStart;
+  if (!runBenchmarks) {
+    runBenchmarks = true;
+    console.log(`Did the initial sync in ${(duration / 1000).toFixed(2)}`);
+    runIncrementalBenchmark(db);
+  }
+
+  fastify.get("/users", async (_req, reply) => {
+    const res = await db.exec(
+      `
+        SELECT * FROM users;
+      `,
+    );
+    console.log(res);
+    reply.send(JSON.stringify(res.rows, null, 2));
+  });
+
+  fastify.get("/users/:userId", async (req, reply) => {
+    const { userId } = req.params;
+
+    //check if valid uuid
+    if (!validate(userId)) {
+      reply.send(`The id provided ${userId} is not a valid UUID`);
+    }
+
+    const res = await db.exec(
+      `
+      SELECT * from users WHERE id = '${userId}'
+    `,
+    );
+    reply.send(JSON.stringify(res.rows, null, 2));
+  });
+} catch (err) {
+  console.error(`Failed to create pglite electric sync instance ${err}`);
+}
+
+fastify.listen({ port: PORT }, () => {
+  console.log(`Server is running on port ${PORT}`);
+});
+
+async function runIncrementalBenchmark(db) {
+  console.log("Starting benchmark setup");
+  const bench = new Bench({ time: 2000 });
+  const sql = postgres(process.env.DATABASE_URL, {
+    max: 10, // Max number of connections
+    idle_timeout: 20, // Idle connection timeout in seconds
+  });
+
+  // Get a random user to update
+  const result = await sql`SELECT id FROM users LIMIT 1`;
+  console.log({ result });
+
+  const userId = result[0].id;
+  console.log(`Selected user ID for testing:`, userId);
+
+  let newName = "";
+
+  bench.add(
+    "sync latency",
+    async () => {
+      await new Promise((resolve) => {
+        const checkInterval = setInterval(async () => {
+          const res = await db.exec(
+            `SELECT * from users WHERE id = '5abd3773-678f-45f4-a28f-405f28b00911'`,
+          );
+
+          if (res && res[0].first_name === newName) {
+            clearInterval(checkInterval);
+            resolve(true);
+          }
+        }, 0);
+      });
+    },
+    {
+      beforeEach: async () => {
+        newName = `User ${Date.now()}`;
+        await sql`UPDATE users SET first_name = ${newName} WHERE id = ${userId}`;
+      },
+    },
+  );
+
+  console.log("\nStarting benchmark runs...");
+  await bench.run();
+
+  console.log("\nBenchmark Results:");
+  console.table(bench.table());
+}

server/populate-pglite.js

@@ -0,0 +1,41 @@
+//generate table and sync to electric//
+export async function generateAndSyncToElectric(
+  db,
+  electricUrl,
+  sourceId,
+  sourceSecret,
+) {
+  await db.exec(`
+  CREATE TABLE IF NOT EXISTS users (
+    id uuid PRIMARY KEY,
+    email TEXT UNIQUE NOT NULL,
+    password_hash TEXT NOT NULL,
+    first_name TEXT,
+    last_name TEXT,
+    phone TEXT,
+    email_verified INTEGER DEFAULT 0,
+    two_factor_enabled INTEGER DEFAULT 0,
+    last_login_at TEXT,
+    failed_login_attempts INTEGER DEFAULT 0,
+    status TEXT DEFAULT 'active' CHECK (status IN ('active', 'inactive', 'suspended')),
+    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
+    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
+  )
+`);
+
+  const shape = await db.electric.syncShapeToTable({
+    shape: {
+      url: `${electricUrl}/v1/shape`,
+      params: {
+        table: "users",
+        source_id: sourceId,
+        source_secret: sourceSecret,
+      },
+    },
+    table: "users",
+    primaryKey: ["id"],
+    shapeKey: "",
+  });
+
+  return shape;
+}

sst-env.d.ts

@@ -5,6 +5,15 @@
 
 declare module "sst" {
   export interface Resource {
+    "ElectricUrl": {
+      "type": "sst.sst.Linkable"
+      "url": string
+    }
+    "FastifyApi": {
+      "name": string
+      "type": "sst.aws.Function"
+      "url": string
+    }
     "examples-infra-shared-examplesInfraVpcShared": {
       "type": "sst.aws.Vpc"
     }

sst.config.ts

@@ -1,6 +1,6 @@
 /// <reference path="./.sst/platform/config.d.ts" />
 
-import { execSync } from "node:child_process";
+import {execSync} from "node:child_process";
 
 export default $config({
   app(input) {
@@ -18,9 +18,10 @@ export default $config({
     };
   },
   async run() {
-    const { getNeonConnectionString, createNeonDb } = await import("./neon");
+    // env var checks
+    const {getNeonConnectionString, createNeonDb} = await import("./neon");
     // Create a db in Neon
-    const project = neon.getProjectOutput({ id: `square-flower-52864146` });
+    const project = neon.getProjectOutput({id: `square-flower-52864146`});
     const dbName = `user_benchmark_${$app.stage.replace(/-/g, `_`)}`;
     const branchId = `br-blue-morning-a4ywuv4l`;
 
@@ -33,7 +34,7 @@ export default $config({
       pooled: false,
     };
 
-    const { dbName: resultingDbName, ownerName } = createNeonDb({
+    const {dbName: resultingDbName, ownerName} = createNeonDb({
       projectId: project.id,
       branchId,
       dbName,
@@ -52,6 +53,14 @@ export default $config({
       applyMigrations(url);
     });
 
+    const electricUrlLink = new sst.Linkable("ElectricUrl", {
+      properties: {
+        url: process.env.ELECTRIC_URL,
+        sourceId: process.env.ELECTRIC_SOURCE_ID,
+        sourceSecret: process.env.ELECTRIC_SOURCE_SECRET,
+      },
+    });
+
     const vpc = sst.aws.Vpc.get(
       "examples-infra-shared-examplesInfraVpcShared",
       "vpc-044836d73fc26a218",
@@ -99,7 +108,18 @@ export default $config({
       ],
     });
 
+    //add lambda
+    const node = new sst.aws.Function("FastifyApi", {
+      url: true,
+      link: [electricUrlLink],
+      handler: "server/lambda.handler",
+      timeout: "3 minutes",
+      memory: "1024 MB",
+    });
+
+
     return {
+      node: node.url,
       dbUrl: postgres.properties.url,
     };
   },