POC for onUpdate implementation for better-sqlite3.

Author: rkistner

src/driver-api.ts

@@ -16,6 +16,11 @@ export interface SqliteDriverConnection {
     options?: ExecuteOptions
   ): Promise<ResultSet>;
 
+  onUpdate(
+    listener: UpdateListener,
+    options?: { tables?: string[]; batchLimit?: number }
+  ): () => void;
+
   close(): Promise<void>;
 }
 
@@ -31,6 +36,23 @@ export interface SqliteDriverConnectionPool {
   ): Promise<ReservedConnection>;
 
   close(): Promise<void>;
+
+  onUpdate(
+    listener: UpdateListener,
+    options?: { tables?: string[]; batchLimit?: number }
+  ): () => void;
+}
+
+export type UpdateListener = (event: BatchedUpdateEvent) => void;
+
+export interface BatchedUpdateEvent {
+  events: UpdateEvent[];
+}
+
+export interface UpdateEvent {
+  table: string;
+  type: "insert" | "update" | "delete";
+  rowId: bigint;
 }
 
 export interface ReservedConnection {

src/driver-util.ts

@@ -3,6 +3,7 @@ import {
   ReservedConnection,
   SqliteDriverConnection,
   SqliteDriverConnectionPool,
+  UpdateListener,
 } from "./driver-api.js";
 
 interface QueuedItem {
@@ -78,6 +79,15 @@ export class SingleConnectionPool implements SqliteDriverConnectionPool {
       break;
     }
   }
+
+  onUpdate(
+    listener: UpdateListener,
+    options?:
+      | { tables?: string[] | undefined; batchLimit?: number | undefined }
+      | undefined
+  ): () => void {
+    return this.connection.onUpdate(listener, options);
+  }
 }
 
 export interface DriverFactory {
@@ -161,6 +171,16 @@ class MultiConnectionPool implements SqliteDriverConnectionPool {
       await con.close();
     }
   }
+
+  onUpdate(
+    listener: UpdateListener,
+    options?:
+      | { tables?: string[] | undefined; batchLimit?: number | undefined }
+      | undefined
+  ): () => void {
+    // No-op
+    return () => {};
+  }
 }
 
 export class ReadWriteConnectionPool implements SqliteDriverConnectionPool {
@@ -193,4 +213,13 @@ export class ReadWriteConnectionPool implements SqliteDriverConnectionPool {
     await this.readPool.close();
     await this.writePool?.close();
   }
+
+  onUpdate(
+    listener: UpdateListener,
+    options?:
+      | { tables?: string[] | undefined; batchLimit?: number | undefined }
+      | undefined
+  ): () => void {
+    return this.writePool!.onUpdate(listener, options);
+  }
 }

src/drivers/better-sqlite3-async-driver.ts

@@ -5,6 +5,7 @@ import {
   RunResults,
   SqliteDriverConnection,
   SqliteDriverConnectionPool,
+  UpdateListener,
 } from "../driver-api.js";
 
 import { EventIterator } from "event-iterator";
@@ -104,4 +105,13 @@ export class BetterSqliteAsyncConnection implements SqliteDriverConnection {
     }
     return results!;
   }
+
+  onUpdate(
+    listener: UpdateListener,
+    options?:
+      | { tables?: string[] | undefined; batchLimit?: number | undefined }
+      | undefined
+  ): () => void {
+    throw new Error("Not implemented");
+  }
 }

src/drivers/better-sqlite3-driver.ts

@@ -6,6 +6,7 @@ import {
   RunResults,
   SqliteDriverConnection,
   SqliteDriverConnectionPool,
+  UpdateListener,
 } from "../driver-api.js";
 const Database = require("better-sqlite3");
 
@@ -118,4 +119,46 @@ export class BetterSqliteConnection implements SqliteDriverConnection {
   dispose(): void {
     // No-op
   }
+
+  onUpdate(
+    listener: UpdateListener,
+    options?:
+      | { tables?: string[] | undefined; batchLimit?: number | undefined }
+      | undefined
+  ): () => void {
+    // Proof-of-concept implementation, based on the idea here:
+    // https://github.com/WiseLibs/better-sqlite3/issues/62
+    // TODO:
+    // 1. Handle multiple registrations.
+    // 2. Don't re-register triggers.
+    // 3. De-register listener.
+    // 4. Batching.
+    //
+    // More fundamental limitations:
+    // 1. The table needs to exist before registering the listener.
+    // 2. Deleting and re-creating the same will dereigster the listener for that table.
+
+    this.con.function("_logger", function (table: any, type: any, rowid: any) {
+      listener({ events: [{ table, rowId: rowid, type }] });
+    });
+    let tables = options?.tables;
+    if (tables == null) {
+      tables = this.con
+        .prepare(`select name from sqlite_master where type = 'table'`)
+        .all()
+        .map((row) => (row as any).name as string);
+    }
+    for (let table of tables) {
+      this.con.exec(
+        `CREATE TEMPORARY TRIGGER IF NOT EXISTS _logger_notification_${table}__update AFTER UPDATE ON ${table} BEGIN SELECT _logger('${table}', 'update', NEW.rowid); END`
+      );
+      this.con.exec(
+        `CREATE TEMPORARY TRIGGER IF NOT EXISTS _logger_notification_${table}__insert AFTER INSERT ON ${table} BEGIN SELECT _logger('${table}', 'insert', NEW.rowid); END`
+      );
+      this.con.exec(
+        `CREATE TEMPORARY TRIGGER IF NOT EXISTS _logger_notification_${table}__delete AFTER DELETE ON ${table} BEGIN SELECT _logger('${table}', 'delete', OLD.rowid); END`
+      );
+    }
+    return () => {};
+  }
 }

src/drivers/sqlite3-driver.ts

@@ -6,6 +6,7 @@ import {
   RunResults,
   SqliteDriverConnection,
   SqliteDriverConnectionPool,
+  UpdateListener,
 } from "../driver-api.js";
 
 import { ReadWriteConnectionPool } from "../driver-util.js";
@@ -93,6 +94,15 @@ export class Sqlite3Connection implements SqliteDriverConnection {
   async close() {
     this.con.close();
   }
+
+  onUpdate(
+    listener: UpdateListener,
+    options?:
+      | { tables?: string[] | undefined; batchLimit?: number | undefined }
+      | undefined
+  ): () => void {
+    throw new Error("Not implemented");
+  }
 }
 
 function transformResults(rows: any[], options?: ExecuteOptions): ResultSet {

test/tests/drivers.ts

@@ -165,5 +165,28 @@ export function describeDriverTests(
         release();
       }
     });
+
+    test.skip("onUpdate", async () => {
+      // Skipped: Not properly implemented yet.
+
+      const driver = await open();
+      const { connection, release } = await driver.reserveConnection();
+      try {
+        await connection.run(
+          "create table test_data(id integer primary key, data text)"
+        );
+        // TODO: test the results
+        connection.onUpdate(({ events }) => {
+          console.log("update", events);
+        });
+
+        await connection.run(
+          "insert into test_data(data) values(123) returning id"
+        );
+        await connection.run("update test_data set data = data || 'test'");
+      } finally {
+        release();
+      }
+    });
   });
 }