Support uploads in node demo

Author: simolus3

demos/example-node/.env

@@ -0,0 +1,2 @@
+BACKEND=http://localhost:6060
+SYNC_SERVICE=http://localhost:8080

demos/example-node/README.md

@@ -2,18 +2,15 @@
 
 This demonstrates a small NodeJS client opening a database and connecting PowerSync.
 
-To get started with this example, you need a configured instance of the PowerSync Service. The easiest way to get started is to configure a PowerSync Cloud instance, docs [here](https://docs.powersync.com/installation/database-connection#create-a-powersync-cloud-instance).  
-This example expects the schema of our To-Do list example apps; ensure you deploy compatible Sync Rules to your instance which can be found [in our Supabase integration guide](https://docs.powersync.com/integration-guides/supabase-+-powersync#configure-sync-rules).
+This demo is configured to talk to an example backend [you can host yourself](https://github.com/powersync-ja/self-host-demo). To get started:
 
-Being a simple CLI program, this currently doesn't implement authentication. So, you'll need to
-create a [development token](https://docs.powersync.com/installation/authentication-setup/development-tokens#development-tokens) to run this example.
-This demo expects the URL of your PowerSync instance (this can be copied from the PowerSync Dashboard) in the `DEMO_ENDPOINT` environment variable,
-the development token is read from `DEMO_TOKEN`.
+1. Start one of the Node.js backend examples from [the self-host-demo repository](https://github.com/powersync-ja/self-host-demo).
+2. If necessary, change `.env` to point to the started backend and sync service.
+3. Run `pnpm install` and `pnpm build:packages` in the root of this repo.
+4. In this directory, run `pnpm run start`.
 
-Once you have these ready: 
-
-1. Make sure to run `pnpm install` and `pnpm build:packages` in the root of this repo.
-2. Then run the following, inserting your instance URL and developer token:
-```DEMO_ENDPOINT=https://yourinstance.powersync.journeyapps.com DEMO_TOKEN=YOURTOKEN pnpm run start
-```
 This opens the local database, connects to PowerSync, waits for a first sync and then runs a simple query.
+Results from the query are printed every time it changes. Try:
+
+1. Updating a row in the backend database and see changes reflected in the running client.
+2. Enter `add('my list')` and see the new list show up in the backend database.

demos/example-node/package.json

@@ -6,10 +6,12 @@
   "private": true,
   "scripts": {
     "build": "tsc -b",
-    "start": "node --loader ts-node/esm src/main.ts"
+    "watch": "tsc -b -w",
+    "start": "node --loader ts-node/esm -r dotenv/config src/main.ts"
   },
   "dependencies": {
-    "@powersync/node": "workspace:*"
+    "@powersync/node": "workspace:*",
+    "dotenv": "^16.4.7"
   },
   "devDependencies": {
     "ts-node": "^10.9.2",

demos/example-node/src/main.ts

@@ -1,17 +1,18 @@
-import {
-  AbstractPowerSyncDatabase,
-  column,
-  PowerSyncBackendConnector,
-  PowerSyncDatabase,
-  Schema,
-  SyncStreamConnectionMethod,
-  Table
-} from '@powersync/node';
+import repl_factory from 'node:repl';
+import { once } from 'node:events';
+
+import { PowerSyncDatabase, SyncStreamConnectionMethod } from '@powersync/node';
 import Logger from 'js-logger';
+import { AppSchema, DemoConnector } from './powersync.js';
+import { exit } from 'node:process';
 
 const main = async () => {
-  if (!('DEMO_ENDPOINT' in process.env) || !('DEMO_TOKEN' in process.env)) {
-    console.warn('Set the DEMO_ENDPOINT and DEMO_TOKEN environment variables to point at a sync service with a dev token to run this example.');
+  Logger.useDefaults({ defaultLevel: Logger.WARN });
+
+  if (!('BACKEND' in process.env) || !('SYNC_SERVICE' in process.env)) {
+    console.warn(
+      'Set the BACKEND and SYNC_SERVICE environment variables to point to a sync service and a running demo backend.'
+    );
     return;
   }
 
@@ -23,13 +24,8 @@ const main = async () => {
     logger: Logger
   });
   console.log(await db.get('SELECT powersync_rs_version();'));
-  db.registerListener({
-    statusChanged: (status) => {
-      console.log('Sync status', status);
-    }
-  });
 
-  await db.connect(new DemoConnector(), {connectionMethod: SyncStreamConnectionMethod.HTTP});
+  await db.connect(new DemoConnector(), { connectionMethod: SyncStreamConnectionMethod.HTTP });
   await db.waitForFirstSync();
   console.log('First sync complete!');
 
@@ -41,62 +37,29 @@ const main = async () => {
         hasFirstRow(null);
         hasFirstRow = null;
       }
-      console.log('Has todo items', rows.rows?._array);
+      console.log('Has todo lists', rows.rows?._array);
     }
   };
 
   watchLists();
   await firstRow;
 
-  //  await db.execute("INSERT INTO lists (id, created_at, name, owner_id) VALUEs (uuid(), 'test', 'test', 'test');");
-};
-
-class DemoConnector implements PowerSyncBackendConnector {
-  async fetchCredentials() {
-    console.log('fetchCredentials called');
-
-    return {
-      endpoint: process.env.DEMO_ENDPOINT!,
-      token: process.env.DEMO_TOKEN!,
-    };
-  }
+  console.log('Connected to PowerSync. Try updating the lists in the database and see it reflected here.');
+  console.log("To upload a list here, enter `await add('name of new list');`");
 
-  async uploadData(database: AbstractPowerSyncDatabase) {
-    await new Promise((resolve) => setTimeout(resolve, 10_000));
-    throw 'not implemented: DemoConnector.uploadData';
-  }
-}
-
-export const LIST_TABLE = 'lists';
-export const TODO_TABLE = 'todos';
-
-const todos = new Table(
-  {
-    list_id: column.text,
-    created_at: column.text,
-    completed_at: column.text,
-    description: column.text,
-    created_by: column.text,
-    completed_by: column.text,
-    completed: column.integer,
-    photo_id: column.text
-  },
-  { indexes: { list: ['list_id'] } }
-);
-
-const lists = new Table({
-  created_at: column.text,
-  name: column.text,
-  owner_id: column.text
-});
-
-export const AppSchema = new Schema({
-  lists,
-  todos
-});
+  const repl = repl_factory.start();
+  repl.context.add = async (name: string) => {
+    await db.execute(
+      "INSERT INTO lists (id, created_at, name, owner_id) VALUEs (uuid(), datetime('now'), ?, 'test');",
+      [name]
+    );
+  };
 
-export type Database = (typeof AppSchema)['types'];
-export type TodoRecord = Database['todos'];
-export type ListRecord = Database['lists'];
+  await once(repl, 'exit');
+  console.log('shutting down');
+  await db.disconnect();
+  await db.close();
+  exit(0);
+};
 
 await main();

demos/example-node/src/powersync.ts

@@ -0,0 +1,77 @@
+import { AbstractPowerSyncDatabase, column, PowerSyncBackendConnector, Schema, Table } from '@powersync/node';
+
+export class DemoConnector implements PowerSyncBackendConnector {
+  async fetchCredentials() {
+    const response = await fetch(`${process.env.BACKEND}/api/auth/token`);
+    if (response.status != 200) {
+      throw 'Could not fetch token';
+    }
+
+    const { token } = await response.json();
+
+    return {
+      endpoint: process.env.SYNC_SERVICE!,
+      token: token
+    };
+  }
+
+  async uploadData(database: AbstractPowerSyncDatabase) {
+    const batch = await database.getCrudBatch();
+    if (batch == null) {
+      return;
+    }
+
+    const entries: any[] = [];
+    for (const op of batch.crud) {
+      entries.push({
+        table: op.table,
+        op: op.op,
+        id: op.id,
+        data: op.opData
+      });
+    }
+
+    const response = await fetch(`${process.env.BACKEND}/api/data/`, {
+      method: 'POST',
+      headers: {'Content-Type': 'application/json'},
+      body: JSON.stringify({batch: entries}),
+    });
+    if (response.status !== 200) {
+      throw new Error(`Server returned HTTP ${response.status}: ${await response.text()}`);
+    }
+
+    await batch?.complete();
+  }
+}
+
+export const LIST_TABLE = 'lists';
+export const TODO_TABLE = 'todos';
+
+const todos = new Table(
+  {
+    list_id: column.text,
+    created_at: column.text,
+    completed_at: column.text,
+    description: column.text,
+    created_by: column.text,
+    completed_by: column.text,
+    completed: column.integer,
+    photo_id: column.text
+  },
+  { indexes: { list: ['list_id'] } }
+);
+
+const lists = new Table({
+  created_at: column.text,
+  name: column.text,
+  owner_id: column.text
+});
+
+export const AppSchema = new Schema({
+  lists,
+  todos
+});
+
+export type Database = (typeof AppSchema)['types'];
+export type TodoRecord = Database['todos'];
+export type ListRecord = Database['lists'];

packages/node/src/db/BetterSQLite3DBAdapter.ts

@@ -101,9 +101,9 @@ export class BetterSQLite3DBAdapter extends BaseObserver<DBAdapterListener> impl
   }
 
   async close() {
-    this.writeConnection.close();
+    await this.writeConnection.close();
     for (const connection of this.readConnections) {
-      connection.close();
+      await connection.close();
     }
   }