Demo: pglite & electric-sql

.electric/compose.yaml

@@ -0,0 +1,33 @@
+version: "3.3"
+name: "electric_rails"
+
+services:
+  postgres:
+    image: postgres:16-alpine
+    environment:
+      POSTGRES_DB: electric_rails
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: password
+    ports:
+      - 54321:5432
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+    tmpfs:
+      - /tmp
+    command:
+      - -c
+      - listen_addresses=*
+      - -c
+      - wal_level=logical
+
+  electric:
+    image: electricsql/electric
+    environment:
+      DATABASE_URL: postgresql://postgres:password@postgres:5432/electric_rails?sslmode=disable
+    ports:
+      - "3131:3000"
+    depends_on:
+      - postgres
+
+volumes:
+  postgres_data:

Gemfile

@@ -1,18 +1,18 @@
 source "https://rubygems.org"
 
 gem "actioncable-next", group: [:default, :wasm]
 # Bundle edge Rails instead: gem "rails", github: "rails/rails", branch: "main"
 gem "rails", "~> 7.2", group: [:default, :wasm]
 # The original asset pipeline for Rails [https://github.com/rails/sprockets-rails]
 gem "sprockets-rails", group: [:default, :wasm]
 # Use sqlite3 as the database for Active Record
-gem "sqlite3", ">= 1.4"
+gem "pg", "~> 1.5"
 # Use the Puma web server [https://github.com/puma/puma]
 gem "puma", ">= 5.0"
 # Use JavaScript with ESM import maps [https://github.com/rails/importmap-rails]
 gem "importmap-rails", group: [:default, :wasm]
 # Hotwire's SPA-like page accelerator [https://turbo.hotwired.dev]
 gem "turbo-rails", group: [:default, :wasm]
 # Hotwire's modest JavaScript framework [https://stimulus.hotwired.dev]
 gem "stimulus-rails", group: [:default, :wasm]
 # Use Tailwind CSS [https://github.com/rails/tailwindcss-rails]

Gemfile.lock

@@ -205,6 +205,7 @@ GEM
     parser (3.3.5.0)
       ast (~> 2.4.1)
       racc
+    pg (1.5.9)
     psych (5.1.2)
       stringio
     public_suffix (6.0.1)
@@ -308,16 +309,6 @@ GEM
       actionpack (>= 6.1)
       activesupport (>= 6.1)
       sprockets (>= 3.0.0)
-    sqlite3 (2.1.0-aarch64-linux-gnu)
-    sqlite3 (2.1.0-aarch64-linux-musl)
-    sqlite3 (2.1.0-arm-linux-gnu)
-    sqlite3 (2.1.0-arm-linux-musl)
-    sqlite3 (2.1.0-arm64-darwin)
-    sqlite3 (2.1.0-x86-linux-gnu)
-    sqlite3 (2.1.0-x86-linux-musl)
-    sqlite3 (2.1.0-x86_64-darwin)
-    sqlite3 (2.1.0-x86_64-linux-gnu)
-    sqlite3 (2.1.0-x86_64-linux-musl)
     stimulus-rails (1.3.4)
       railties (>= 6.0.0)
     stringio (3.1.1)
@@ -387,12 +378,12 @@ DEPENDENCIES
   debug
   importmap-rails
   jbuilder
+  pg (~> 1.5)
   puma (>= 5.0)
   rails (~> 7.2)
   rubocop-rails-omakase
   selenium-webdriver
   sprockets-rails
-  sqlite3 (>= 1.4)
   stimulus-rails
   tailwindcss-rails
   turbo-rails

config/database.yml

@@ -5,21 +5,25 @@
 #   gem "sqlite3"
 #
 default: &default
-  adapter: sqlite3
+  adapter: postgresql
   pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
   timeout: 5000
 
+# Go to the .electric folder and run `docker compose up` to start the database
 development:
   <<: *default
-  database: storage/development.sqlite3
+  host: localhost
+  port: 54321
+  username: postgres
+  password: password
+  database: electric_rails
 
 # Warning: The database defined as "test" will be erased and
 # re-generated from your development database when you run "rake".
 # Do not set this db to the same as development or production.
 test:
   <<: *default
-  database: storage/test.sqlite3
-
+  database: electric_rails_test
 
 # SQLite3 write its data on the local filesystem, as such it requires
 # persistent disks. If you are deploying to a managed service, you should
@@ -30,5 +34,6 @@ test:
 production:
   <<: *default
   # database: path/to/persistent/storage/production.sqlite3
+
 wasm:
   adapter: <%= ENV.fetch("ACTIVE_RECORD_ADAPTER") { "nulldb" } %>

pwa/package.json

@@ -14,7 +14,8 @@
     "vite-plugin-pwa": "^0.20.5"
   },
   "dependencies": {
-    "wasmify-rails": "~> 0.2.0",
-    "@sqlite.org/sqlite-wasm": "3.46.1-build3"
+    "@electric-sql/client": "^0.7.2",
+    "@electric-sql/pglite": "^0.2.13",
+    "wasmify-rails": "~> 0.2.0"
   }
 }

pwa/pglite-sync/index.cjs

@@ -0,0 +1,39 @@
+"use strict";var C=Object.defineProperty;var A=Object.getOwnPropertyDescriptor;var N=Object.getOwnPropertyNames;var K=Object.prototype.hasOwnProperty;var _=(a,t)=>{for(var s in t)C(a,s,{get:t[s],enumerable:!0})},D=(a,t,s,n)=>{if(t&&typeof t=="object"||typeof t=="function")for(let r of N(t))!K.call(a,r)&&r!==s&&C(a,r,{get:()=>t[r],enumerable:!(n=A(t,r))||n.enumerable});return a};var L=a=>D(C({},"__esModule",{value:!0}),a);var x={};_(x,{electricSync:()=>P});module.exports=L(x);var S=require("@electric-sql/client");async function R(a,t){let s=t?.debug??!1,n=t?.metadataSchema??"electric",r=[],c=new Map;return{namespaceObj:{syncShapeToTable:async e=>{if(c.has(e.table))throw new Error("Already syncing shape for table "+e.table);c.set(e.table);let i=null;e.shapeKey&&(i=await j({pg:a,metadataSchema:n,shapeKey:e.shapeKey}),s&&i&&console.log("resuming from shape state",i));let l=i===null&&e.useCopy,m=new AbortController;e.shape.signal&&e.shape.signal.addEventListener("abort",()=>m.abort(),{once:!0});let u=new S.ShapeStream({...e.shape,...i??{},signal:m.signal}),h=[],E=!1;return u.subscribe(async T=>{s&&console.log("sync messages received",T);for(let y of T){if((0,S.isChangeMessage)(y)){h.push(y);continue}if((0,S.isControlMessage)(y))switch(y.headers.control){case"must-refetch":s&&console.log("refetching shape"),E=!0,h=[];break;case"up-to-date":await a.transaction(async b=>{if(s&&console.log("up-to-date, committing all messages"),b.exec(`SET LOCAL ${n}.syncing = true;`),E&&(E=!1,await b.exec(`DELETE FROM ${e.table};`),e.shapeKey&&await F({pg:b,metadataSchema:n,shapeKey:e.shapeKey})),l){let f=[],O=[],$=!1;for(let M of h)!$&&M.headers.operation==="insert"?f.push(M):($=!0,O.push(M));f.length>0&&O.unshift(f.pop()),h=O,f.length>0&&(v({pg:b,table:e.table,schema:e.schema,messages:f,mapColumns:e.mapColumns,primaryKey:e.primaryKey,debug:s}),l=!1)}for(let f of h)await U({pg:b,table:e.table,schema:e.schema,message:f,mapColumns:e.mapColumns,primaryKey:e.primaryKey,debug:s});e.shapeKey&&h.length>0&&u.shapeId!==void 0&&await k({pg:b,metadataSchema:n,shapeKey:e.shapeKey,shapeId:u.shapeId,lastOffset:h[h.length-1].offset})}),h=[];break}}}),r.push({stream:u,aborter:m}),{unsubscribe:()=>{u.unsubscribeAll(),m.abort(),c.delete(e.table)},get isUpToDate(){return u.isUpToDate},get shapeId(){return u.shapeId},subscribeOnceToUpToDate:(T,y)=>u.subscribeOnceToUpToDate(T,y),unsubscribeAllUpToDateSubscribers:()=>{u.unsubscribeAllUpToDateSubscribers()}}}},close:async()=>{for(let{stream:e,aborter:i}of r)e.unsubscribeAll(),i.abort()},init:async()=>{await G({pg:a,metadataSchema:n})}}}function P(a){return{name:"ElectricSQL Sync",setup:async t=>{let{namespaceObj:s,close:n,init:r}=await R(t,a);return{namespaceObj:s,close:n,init:r}}}}function w(a,t){if(typeof a=="function")return a(t);{let s={};for(let[n,r]of Object.entries(a))s[n]=t.value[r];return s}}async function U({pg:a,table:t,schema:s="public",message:n,mapColumns:r,primaryKey:c,debug:g}){let p=r?w(r,n):n.value;switch(n.headers.operation){case"insert":{g&&console.log("inserting",p);let o=Object.keys(p);return await a.query(`
+            INSERT INTO "${s}"."${t}"
+            (${o.map(e=>'"'+e+'"').join(", ")})
+            VALUES
+            (${o.map((e,i)=>"$"+(i+1)).join(", ")})
+          `,o.map(e=>p[e]))}case"update":{g&&console.log("updating",p);let o=Object.keys(p).filter(e=>!c.includes(e));return o.length===0?void 0:await a.query(`
+            UPDATE "${s}"."${t}"
+            SET ${o.map((e,i)=>'"'+e+'" = $'+(i+1)).join(", ")}
+            WHERE ${c.map((e,i)=>'"'+e+'" = $'+(o.length+i+1)).join(" AND ")}
+          `,[...o.map(e=>p[e]),...c.map(e=>p[e])])}case"delete":return g&&console.log("deleting",p),await a.query(`
+            DELETE FROM "${s}"."${t}"
+            WHERE ${c.map((o,e)=>'"'+o+'" = $'+(e+1)).join(" AND ")}
+          `,[...c.map(o=>p[o])])}}async function v({pg:a,table:t,schema:s="public",messages:n,mapColumns:r,debug:c}){c&&console.log("applying messages with COPY");let g=n.map(i=>r?w(r,i):i.value),p=Object.keys(g[0]),o=g.map(i=>p.map(I=>{let l=i[I];return typeof l=="string"&&(l.includes(",")||l.includes('"')||l.includes(`
+`))?`"${l.replace(/"/g,'""')}"`:l===null?"\\N":l}).join(",")).join(`
+`),e=new Blob([o],{type:"text/csv"});await a.query(`
+      COPY "${s}"."${t}" (${p.map(i=>`"${i}"`).join(", ")})
+      FROM '/dev/blob'
+      WITH (FORMAT csv, NULL '\\N')
+    `,[],{blob:e}),c&&console.log(`Inserted ${n.length} rows using COPY`)}async function j({pg:a,metadataSchema:t,shapeKey:s}){let n=await a.query(`
+    SELECT shape_id, last_offset
+    FROM ${d(t)}
+    WHERE shape_key = $1
+  `,[s]);if(n.rows.length===0)return null;let{shape_id:r,last_offset:c}=n.rows[0];return{shapeId:r,offset:c}}async function k({pg:a,metadataSchema:t,shapeKey:s,shapeId:n,lastOffset:r}){await a.query(`
+    INSERT INTO ${d(t)} (shape_key, shape_id, last_offset)
+    VALUES ($1, $2, $3)
+    ON CONFLICT(shape_key)
+    DO UPDATE SET
+      shape_id = EXCLUDED.shape_id,
+      last_offset = EXCLUDED.last_offset;
+  `,[s,n,r])}async function F({pg:a,metadataSchema:t,shapeKey:s}){await a.query(`DELETE FROM ${d(t)} WHERE shape_key = $1`,[s])}async function G({pg:a,metadataSchema:t}){await a.exec(`
+    SET ${t}.syncing = false;
+    CREATE SCHEMA IF NOT EXISTS "${t}";
+    CREATE TABLE IF NOT EXISTS ${d(t)} (
+      shape_key TEXT PRIMARY KEY,
+      shape_id TEXT NOT NULL,
+      last_offset TEXT NOT NULL
+    );
+    `)}function d(a){return`"${a}"."${q}"`}var q="shape_subscriptions_metadata";0&&(module.exports={electricSync});
+//# sourceMappingURL=index.cjs.map

pwa/pglite.js

@@ -0,0 +1,24 @@
+import { PGlite } from '@electric-sql/pglite'
+import { electricSync } from './pglite-sync'
+import { live } from '@electric-sql/pglite/live'
+
+export const setupPGliteDatabase = async () => {
+  const db = await PGlite.create({
+    extensions: {
+      live,
+      electric: electricSync({debug: true}),
+    }
+  });
+  return db;
+};
+
+export const setupElectricSync = async (db, url) => {
+  await db.electric.syncShapeToTable({
+    shape: {
+      url: `http://localhost:3131/v1/shape`,
+      table: 'todos'
+    },
+    table: 'todos',
+    primaryKey: ['id'],
+  })
+}

pwa/rails.sw.js

@@ -1,42 +1,42 @@
 import {
   initRailsVM,
   Progress,
-  registerSQLiteWasmInterface,
   RackHandler,
+  registerPGliteWasmInterface,
 } from "wasmify-rails";
 
-import { setupSQLiteDatabase } from "./database.js";
+import { setupPGliteDatabase, setupElectricSync } from "./pglite.js";
 
 let db = null;
 
 const initDB = async (progress) => {
   if (db) return db;
 
-  progress?.updateStep("Initializing SQLite database...");
-  db = await setupSQLiteDatabase();
-  progress?.updateStep("SQLite database created.");
+  progress?.updateStep("Initializing PGlite database...");
+  db = await setupPGliteDatabase();
+  progress?.updateStep("PGlite database created.");
 
   return db;
 };
 
 let vm = null;
 
-const initVM = async (progress, opts = {}) => {
+export const initVM = async (progress, opts = {}) => {
   if (vm) return vm;
 
   if (!db) {
     await initDB(progress);
   }
 
-  registerSQLiteWasmInterface(self, db);
+  registerPGliteWasmInterface(self, db);
 
   let redirectConsole = true;
 
   const env = [];
 
   vm = await initRailsVM("/app.wasm", {
-    database: { adapter: "sqlite3_wasm" },
-    env,
+    database: { adapter: "pglite" },
+    async: true,
     progressCallback: (step) => {
       progress?.updateStep(step);
     },
@@ -49,7 +49,10 @@ const initVM = async (progress, opts = {}) => {
 
   // Ensure schema is loaded
   progress?.updateStep("Preparing database...");
-  vm.eval("ActiveRecord::Tasks::DatabaseTasks.prepare_all");
+  await vm.evalAsync("ActiveRecord::Tasks::DatabaseTasks.prepare_all");
+
+  progress?.updateStep("Enable electric-sql...");
+  await setupElectricSync(db, "http://localhost:3131");
 
   redirectConsole = false;
 
@@ -77,7 +80,7 @@ self.addEventListener("install", (event) => {
   event.waitUntil(installApp().then(() => self.skipWaiting()));
 });
 
-const rackHandler = new RackHandler(initVM, { assumeSSL: true, async: false });
+const rackHandler = new RackHandler(initVM, { assumeSSL: true, async: true });
 
 self.addEventListener("fetch", (event) => {
   const bootResources = ["/boot", "/boot.js", "/boot.html", "/rails.sw.js"];

pwa/vite.config.js

@@ -9,7 +9,7 @@ export default defineConfig({
     },
   },
   optimizeDeps: {
-    exclude: ["@sqlite.org/sqlite-wasm"],
+    exclude: ["@electric-sql/pglite"],
   },
   plugins: [
     VitePWA({