feature(client): PGlite driver and example (#1058)

PGlite driver for Electric

Builds on parameterized query support for PGlite:
electric-sql/pglite#39

---------

Co-authored-by: msfstef <msfstef@gmail.com>

clients/typescript/package.json

@@ -56,6 +56,7 @@
     "./generic": "./dist/drivers/generic/index.js",
     "./node": "./dist/drivers/better-sqlite3/index.js",
     "./node-postgres": "./dist/drivers/node-postgres/index.js",
+    "./pglite": "./dist/drivers/pglite/index.js",
     "./react": "./dist/frameworks/react/index.js",
     "./tauri-postgres": "./dist/drivers/tauri-postgres/index.js",
     "./vuejs": "./dist/frameworks/vuejs/index.js",
@@ -95,6 +96,9 @@
       "node-postgres": [
         "./dist/drivers/node-postgres/index.d.ts"
       ],
+      "pglite": [
+        "./dist/drivers/pglite/index.d.ts"
+      ],
       "react": [
         "./dist/frameworks/react/index.d.ts"
       ],
@@ -212,6 +216,7 @@
     "zod": "3.21.1"
   },
   "devDependencies": {
+    "@electric-sql/pglite": "^0.1.4",
     "@electric-sql/prisma-generator": "workspace:*",
     "@op-engineering/op-sqlite": ">= 2.0.16",
     "@tauri-apps/plugin-sql": "2.0.0-alpha.5",
@@ -268,6 +273,7 @@
   },
   "peerDependencies": {
     "@capacitor-community/sqlite": ">= 5.6.2",
+    "@electric-sql/pglite": ">= 0.1.4",
     "@op-engineering/op-sqlite": ">= 2.0.16",
     "@tauri-apps/plugin-sql": "2.0.0-alpha.5",
     "embedded-postgres": "16.1.1-beta.9",

clients/typescript/src/drivers/pglite/adapter.ts

@@ -0,0 +1,26 @@
+import { Database } from './database'
+import { Row } from '../../util/types'
+import { Statement } from '../../util'
+import { SerialDatabaseAdapter as GenericDatabaseAdapter } from '../generic'
+import { RunResult } from '../../electric/adapter'
+
+export class DatabaseAdapter extends GenericDatabaseAdapter {
+  readonly db: Database
+  readonly defaultNamespace = 'public'
+
+  constructor(db: Database) {
+    super()
+    this.db = db
+  }
+
+  async _run(statement: Statement): Promise<RunResult> {
+    const res = await this.db.query(statement.sql, statement.args)
+    return {
+      rowsAffected: res.affectedRows ?? 0,
+    }
+  }
+
+  async _query(statement: Statement): Promise<Row[]> {
+    return (await this.db.query<Row>(statement.sql, statement.args)).rows
+  }
+}

clients/typescript/src/drivers/pglite/database.ts

@@ -0,0 +1,7 @@
+import type { PGlite } from '@electric-sql/pglite'
+
+// The relevant subset of the SQLitePlugin database client API
+// that we need to ensure the client we're electrifying provides.
+export interface Database
+  extends Pick<PGlite, 'query' | 'dataDir'> {
+}

clients/typescript/src/drivers/pglite/index.ts

@@ -0,0 +1,45 @@
+import { DatabaseAdapter as DatabaseAdapterI } from '../../electric/adapter'
+import { DatabaseAdapter } from './adapter'
+import { Database } from './database'
+import { ElectricConfig } from '../../config'
+import { electrify as baseElectrify, ElectrifyOptions } from '../../electric'
+import { WebSocketWeb } from '../../sockets/web'
+import { ElectricClient, DbSchema } from '../../client/model'
+import { PgBundleMigrator } from '../../migrators/bundle'
+
+export { DatabaseAdapter }
+export type { Database }
+
+export const electrify = async <T extends Database, DB extends DbSchema<any>>(
+  db: T,
+  dbDescription: DB,
+  config: ElectricConfig,
+  opts?: ElectrifyOptions
+): Promise<ElectricClient<DB>> => {
+  const dbName = db.dataDir?.split('/').pop() ?? 'memory'
+  const adapter = opts?.adapter || new DatabaseAdapter(db)
+  const migrator =
+    opts?.migrator || new PgBundleMigrator(adapter, dbDescription.pgMigrations)
+  const socketFactory = opts?.socketFactory || WebSocketWeb
+  const prepare = async (_connection: DatabaseAdapterI) => undefined
+
+  const configWithDialect = {
+    ...config,
+    dialect: 'Postgres',
+  } as const
+
+  const client = await baseElectrify(
+    dbName,
+    dbDescription,
+    adapter,
+    socketFactory,
+    configWithDialect,
+    {
+      migrator,
+      prepare,
+      ...opts,
+    }
+  )
+
+  return client
+}

clients/typescript/src/drivers/pglite/mock.ts

@@ -0,0 +1,30 @@
+import { Database } from './database'
+import type { PGliteOptions, QueryOptions, Results } from '@electric-sql/pglite'
+
+export class MockDatabase implements Database {
+  dataDir?: string
+  fail: Error | undefined
+
+  constructor(dataDir?: string, options?: PGliteOptions) {
+    this.dataDir = dataDir
+  }
+
+  async query<T>(
+    query: string,
+    params?: any[],
+    options?: QueryOptions
+  ): Promise<Results<T>> {
+    if (typeof this.fail !== 'undefined') throw this.fail
+
+    return {
+      rows: [{ val: 1 } as T, { val: 2 } as T],
+      affectedRows: 0,
+      fields: [
+        {
+          name: 'val',
+          dataTypeID: 0,
+        },
+      ],
+    }
+  }
+}

clients/typescript/test/drivers/pglite.test.ts

@@ -0,0 +1,173 @@
+import test from 'ava'
+
+import { MockDatabase } from '../../src/drivers/pglite/mock'
+import { DatabaseAdapter } from '../../src/drivers/pglite'
+import { PGlite } from '@electric-sql/pglite'
+
+test('database adapter run works', async (t) => {
+  const db = new MockDatabase('test.db')
+  const adapter = new DatabaseAdapter(db)
+
+  const sql = 'drop table badgers'
+  const result = await adapter.run({ sql })
+
+  t.is(result.rowsAffected, 0)
+})
+
+test('database adapter query works', async (t) => {
+  const db = new MockDatabase('test.db')
+  const adapter = new DatabaseAdapter(db)
+
+  const sql = 'select * from bars'
+  const result = await adapter.query({ sql })
+
+  t.deepEqual(result, [
+    {
+      val: 1,
+    },
+    {
+      val: 2,
+    },
+  ])
+})
+
+// Test with an actual PGlite
+async function makeAdapter() {
+  const db = new PGlite()
+  const adapter = new DatabaseAdapter(db)
+  const createTableSql =
+    'CREATE TABLE IF NOT EXISTS Post(id TEXT PRIMARY KEY, title TEXT, contents TEXT, nbr integer);'
+  await adapter.run({ sql: createTableSql })
+  return adapter
+}
+
+test('adapter run works on real DB', async (t) => {
+  const adapter = await makeAdapter()
+  const insertRecordSql =
+    "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't1', 'c1', 18)"
+  const res = await adapter.run({ sql: insertRecordSql })
+  t.is(res.rowsAffected, 1)
+})
+
+test('adapter query works on real DB', async (t) => {
+  const adapter = await makeAdapter()
+  const insertRecordSql =
+    "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't1', 'c1', 18)"
+  await adapter.run({ sql: insertRecordSql })
+
+  const selectSql =
+    "SELECT * FROM Post WHERE (id = ('i1')) AND (nbr = (18)) LIMIT 1"
+  const res = await adapter.query({ sql: selectSql })
+  t.deepEqual(res, [{ id: 'i1', title: 't1', contents: 'c1', nbr: 18 }])
+})
+
+test('adapter runInTransaction works on real DB', async (t) => {
+  const adapter = await makeAdapter()
+  const insertRecord1Sql =
+    "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't1', 'c1', 18)"
+  const insertRecord2Sql =
+    "INSERT INTO Post (id, title, contents, nbr) VALUES ('i2', 't2', 'c2', 25)"
+
+  const txRes = await adapter.runInTransaction(
+    { sql: insertRecord1Sql },
+    { sql: insertRecord2Sql }
+  )
+
+  t.is(txRes.rowsAffected, 2)
+
+  const selectAll = 'SELECT id FROM Post'
+  const res = await adapter.query({ sql: selectAll })
+
+  t.deepEqual(res, [{ id: 'i1' }, { id: 'i2' }])
+})
+
+test('adapter runInTransaction rolls back on conflict', async (t) => {
+  const adapter = await makeAdapter()
+  const insertRecord1Sql =
+    "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't1', 'c1', 18)"
+  const insertRecord2Sql =
+    "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't2', 'c2', 25)"
+
+  try {
+    await adapter.runInTransaction(
+      { sql: insertRecord1Sql },
+      { sql: insertRecord2Sql }
+    )
+    t.fail() // the transaction should be rejected because the primary key of the second record already exists
+  } catch (err) {
+    const castError = err as { code: string; detail: string }
+    t.is(castError.code, '23505')
+    t.is(castError.detail, 'Key (id)=(i1) already exists.')
+
+    // Check that no posts were added to the DB
+    const selectAll = 'SELECT id FROM Post'
+    const res = await adapter.query({ sql: selectAll })
+    t.deepEqual(res, [])
+  }
+})
+
+test('adapter supports dependent queries in transaction on real DB', async (t) => {
+  const adapter = await makeAdapter()
+  const [txRes, rowsAffected] = (await adapter.transaction<Array<number>>(
+    (tx, setResult) => {
+      let rowsAffected = 0
+      tx.run(
+        {
+          sql: "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't1', 'c1', 18)",
+        },
+        (tx2, res) => {
+          rowsAffected += res.rowsAffected
+          const select = { sql: "SELECT nbr FROM Post WHERE id = 'i1'" }
+          tx2.query(select, (tx3, rows) => {
+            const [res] = rows as unknown as Array<{ nbr: number }>
+            const newNbr = res.nbr + 2
+            tx3.run(
+              {
+                sql: `INSERT INTO Post (id, title, contents, nbr) VALUES ('i2', 't2', 'c2', ${newNbr})`,
+              },
+              (_, res) => {
+                rowsAffected += res.rowsAffected
+                setResult([newNbr, rowsAffected])
+              }
+            )
+          })
+        }
+      )
+    }
+  )) as unknown as Array<number>
+
+  t.is(txRes, 20)
+  t.is(rowsAffected, 2)
+
+  const selectAll = 'SELECT * FROM Post'
+  const res = await adapter.query({ sql: selectAll })
+
+  t.deepEqual(res, [
+    { id: 'i1', title: 't1', contents: 'c1', nbr: 18 },
+    { id: 'i2', title: 't2', contents: 'c2', nbr: 20 },
+  ])
+})
+
+test('adapter rolls back dependent queries on conflict', async (t) => {
+  const adapter = await makeAdapter()
+  try {
+    await adapter.transaction((tx) => {
+      tx.run({
+        sql: "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't1', 'c1', 18)",
+      })
+      tx.run({
+        sql: "INSERT INTO Post (id, title, contents, nbr) VALUES ('i1', 't2', 'c2', 20)",
+      })
+    })
+    t.fail() // the transaction should be rejected because the primary key of the second record already exists
+  } catch (err) {
+    const castError = err as { code: string; detail: string }
+    t.is(castError.code, '23505')
+    t.is(castError.detail, 'Key (id)=(i1) already exists.')
+
+    // Check that no posts were added to the DB
+    const selectAll = 'SELECT id FROM Post'
+    const res = await adapter.query({ sql: selectAll })
+    t.deepEqual(res, [])
+  }
+})

examples/web-pglite/.env

@@ -0,0 +1,3 @@
+ELECTRIC_SERVICE=http://localhost:5133
+ELECTRIC_PG_PROXY_PORT=65432
+ELECTRIC_IMAGE=electric:local-build

examples/web-pglite/.eslintrc.cjs

@@ -0,0 +1,18 @@
+module.exports = {
+  root: true,
+  env: { browser: true, es2020: true },
+  extends: [
+    'eslint:recommended',
+    'plugin:@typescript-eslint/recommended',
+    'plugin:react-hooks/recommended',
+  ],
+  ignorePatterns: ['dist', '.eslintrc.cjs'],
+  parser: '@typescript-eslint/parser',
+  plugins: ['react-refresh'],
+  rules: {
+    'react-refresh/only-export-components': [
+      'warn',
+      { allowConstantExport: true },
+    ],
+  },
+}

examples/web-pglite/.gitignore

@@ -0,0 +1,31 @@
+# Logs
+logs
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+pnpm-debug.log*
+lerna-debug.log*
+
+node_modules
+dist
+dist-ssr
+*.local
+
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?
+
+# Wasm
+public/wa-sqlite-async.wasm
+
+# Env files
+.env.local
+.env.*.local