Skip to content

Commit 19d4e93

Browse files
simolus3simolus3
committed
Complete simple nodejs tests
1 parent 3e1d9fd commit 19d4e93

File tree

7 files changed

+244
-190
lines changed

7 files changed

+244
-190
lines changed

‎demos/example-node/src/main.ts‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,10 @@ import {
99
import Logger from 'js-logger';
1010

1111
const main = async () => {
12+
if (!('DEMO_ENDPOINT' in process.env) || !('DEMO_TOKEN' in process.env)) {
13+
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.');
14+
}
15+
1216
const db = new PowerSyncDatabase({
1317
schema: AppSchema,
1418
database: {
@@ -27,7 +31,7 @@ const main = async () => {
2731
await db.waitForFirstSync();
2832

2933
let hasFirstRow: ((value: any) => void) | null = null;
30-
const firstRow = new Promise((resolve) => hasFirstRow = resolve);
34+
const firstRow = new Promise((resolve) => (hasFirstRow = resolve));
3135
const watchLists = async () => {
3236
for await (const rows of db.watch('SELECT * FROM lists')) {
3337
if (hasFirstRow) {
@@ -41,14 +45,14 @@ const main = async () => {
4145
watchLists();
4246
await firstRow;
4347

44-
// await db.execute("INSERT INTO lists (id, created_at, name, owner_id) VALUEs (uuid(), 'test', 'test', 'test');");
48+
// await db.execute("INSERT INTO lists (id, created_at, name, owner_id) VALUEs (uuid(), 'test', 'test', 'test');");
4549
};
4650

4751
class DemoConnector implements PowerSyncBackendConnector {
4852
async fetchCredentials() {
4953
return {
50-
endpoint: 'https://678775966cf706da85f4e447.powersync.journeyapps.com', // todo
51-
token: 'eyJhbGciOiJSUzI1NiIsImtpZCI6InBvd2Vyc3luYy1kZXYtMzIyM2Q0ZTMifQ.eyJzdWIiOiI5MGQ5MzBhZi1iYmVkLTQ0NDgtOWM0NS1kYWQyZGYzMDAwNWYiLCJpYXQiOjE3NDEwMTU4NjUsImlzcyI6Imh0dHBzOi8vcG93ZXJzeW5jLWFwaS5qb3VybmV5YXBwcy5jb20iLCJhdWQiOiJodHRwczovLzY3ODc3NTk2NmNmNzA2ZGE4NWY0ZTQ0Ny5wb3dlcnN5bmMuam91cm5leWFwcHMuY29tIiwiZXhwIjoxNzQxMDU5MDY1fQ.aQqLOulvaOMKFtOUfbYuywLRjxmhs1J9hpdlkoSjw2m1_OTmnoVJkMRFKvO45k2I_ZD1GLg2sb6HoV2KNQHJPp2yMeL5eBENXt1HEy-WKU5ObrbwoQT0knkHnwZRmbGNwPYz3R21GibKDdD8chILVtNkSXoy3LjAUmywdzzhMBmmJwiIxP5Ew_K4XAxBU2pDyjX3FNvISHdB60IPgGBQUz3Ke_t-4ZD-k-EUHVLDufhxsDRSwhIKG26PPUqJdZ4YPlqwtjjZrVHy_B4XQBtRsAjqhf61ZLFEx6xeHcze-xZRNVsTw3qTmAFC4Vf9Ezka5jhMppM_HAtPn_wyzWUauA' // todo
54+
endpoint: process.env.DEMO_ENDPOINT!,
55+
token: process.env.DEMO_TOKEN!,
5256
};
5357
}
5458

‎packages/node/src/db/AsyncBetterSqlite.ts‎

Lines changed: 97 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -1,104 +1,141 @@
1-
import { QueryResult } from "@powersync/common";
1+
import { QueryResult } from '@powersync/common';
22
import BetterSQLite3Database, { Database } from 'better-sqlite3';
33
import * as Comlink from 'comlink';
4-
import { MessagePort, Worker, parentPort } from 'node:worker_threads';
4+
import { MessagePort, parentPort, threadId } from 'node:worker_threads';
55
import OS from 'node:os';
66
import url from 'node:url';
77

88
export type ProxiedQueryResult = Omit<QueryResult, 'rows'> & {
9-
rows?: {
10-
_array: any[];
11-
length: number;
12-
};
9+
rows?: {
10+
_array: any[];
11+
length: number;
1312
};
13+
};
1414

1515
export interface AsyncDatabase {
1616
execute: (query: string, params: any[]) => Promise<ProxiedQueryResult>;
1717
executeBatch: (query: string, params: any[][]) => Promise<ProxiedQueryResult>;
1818
close: () => Promise<void>;
19+
// Collect table updates made since the last call to collectCommittedUpdates.
20+
// This happens on the worker because we otherwise get race conditions when wrapping
21+
// callbacks to invoke on the main thread (we need a guarantee that collectCommittedUpdates
22+
// contains entries immediately after calling COMMIT).
23+
collectCommittedUpdates: () => Promise<string[]>;
1924
}
2025

2126
class BlockingAsyncDatabase implements AsyncDatabase {
22-
private readonly db: Database
27+
private readonly db: Database;
2328

24-
constructor(db: Database) {
25-
this.db = db;
26-
}
29+
private readonly uncommittedUpdatedTables = new Set<string>();
30+
private readonly committedUpdatedTables = new Set<string>();
2731

28-
async close() {
29-
this.db.close();
30-
}
32+
constructor(db: Database) {
33+
this.db = db;
34+
35+
db.function('node_thread_id', () => threadId);
36+
}
37+
38+
collectCommittedUpdates() {
39+
const resolved = Promise.resolve([...this.committedUpdatedTables]);
40+
this.committedUpdatedTables.clear();
41+
return resolved;
42+
}
43+
44+
installUpdateHooks() {
45+
this.db.updateHook((_op: string, _dbName: string, tableName: string, _rowid: bigint) => {
46+
this.uncommittedUpdatedTables.add(tableName);
47+
});
3148

32-
async execute(query: string, params: any[]) {
33-
const stmt = this.db.prepare(query);
34-
if (stmt.reader) {
35-
const rows = stmt.all(params);
36-
return {
37-
rowsAffected: 0,
38-
rows: {
39-
_array: rows,
40-
length: rows.length,
41-
},
42-
};
43-
} else {
44-
const info = stmt.run(params);
45-
return {
46-
rowsAffected: info.changes,
47-
insertId: Number(info.lastInsertRowid),
48-
};
49+
this.db.commitHook(() => {
50+
for (const tableName of this.uncommittedUpdatedTables) {
51+
this.committedUpdatedTables.add(tableName);
52+
}
53+
this.uncommittedUpdatedTables.clear();
54+
return true;
55+
});
56+
57+
this.db.rollbackHook(() => {
58+
this.uncommittedUpdatedTables.clear();
59+
});
60+
}
61+
62+
async close() {
63+
this.db.close();
64+
}
65+
66+
async execute(query: string, params: any[]) {
67+
const stmt = this.db.prepare(query);
68+
if (stmt.reader) {
69+
const rows = stmt.all(params);
70+
return {
71+
rowsAffected: 0,
72+
rows: {
73+
_array: rows,
74+
length: rows.length
4975
}
76+
};
77+
} else {
78+
const info = stmt.run(params);
79+
return {
80+
rowsAffected: info.changes,
81+
insertId: Number(info.lastInsertRowid)
82+
};
5083
}
84+
}
5185

52-
async executeBatch(query: string, params: any[][]) {
53-
params = params ?? [];
86+
async executeBatch(query: string, params: any[][]) {
87+
params = params ?? [];
5488

55-
let rowsAffected = 0;
56-
57-
const stmt = this.db.prepare(query);
58-
for (const paramSet of params) {
59-
const info = stmt.run(paramSet);
60-
rowsAffected += info.changes;
61-
}
62-
63-
return { rowsAffected };
89+
let rowsAffected = 0;
90+
91+
const stmt = this.db.prepare(query);
92+
for (const paramSet of params) {
93+
const info = stmt.run(paramSet);
94+
rowsAffected += info.changes;
6495
}
96+
97+
return { rowsAffected };
98+
}
6599
}
66100

67101
export class BetterSqliteWorker {
68-
open(path: string, isWriter: boolean): AsyncDatabase {
69-
const baseDB = new BetterSQLite3Database(path);
70-
baseDB.pragma('journal_mode = WAL');
71-
loadExtension(baseDB);
72-
if (!isWriter) {
73-
baseDB.pragma('query_only = true');
74-
}
75-
76-
return Comlink.proxy(new BlockingAsyncDatabase(baseDB));
102+
open(path: string, isWriter: boolean): AsyncDatabase {
103+
const baseDB = new BetterSQLite3Database(path);
104+
baseDB.pragma('journal_mode = WAL');
105+
loadExtension(baseDB);
106+
if (!isWriter) {
107+
baseDB.pragma('query_only = true');
77108
}
109+
110+
const asyncDb = new BlockingAsyncDatabase(baseDB);
111+
asyncDb.installUpdateHooks();
112+
113+
return Comlink.proxy(asyncDb);
114+
}
78115
}
79116

80117
const platform = OS.platform();
81118
let extensionPath: string;
82-
if (platform === "win32") {
119+
if (platform === 'win32') {
83120
extensionPath = 'powersync.dll';
84-
} else if (platform === "linux") {
121+
} else if (platform === 'linux') {
85122
extensionPath = 'libpowersync.so';
86-
} else if (platform === "darwin") {
123+
} else if (platform === 'darwin') {
87124
extensionPath = 'libpowersync.dylib';
88125
}
89126

90127
const loadExtension = (db: Database) => {
91128
const resolved = url.fileURLToPath(new URL(`../${extensionPath}`, import.meta.url));
92129
db.loadExtension(resolved, 'sqlite3_powersync_init');
93-
}
130+
};
94131

95132
function toComlink(port: MessagePort): Comlink.Endpoint {
96-
return {
97-
postMessage: port.postMessage.bind(port),
98-
start: port.start && port.start.bind(port),
99-
addEventListener: port.addEventListener.bind(port),
100-
removeEventListener: port.removeEventListener.bind(port),
101-
};
133+
return {
134+
postMessage: port.postMessage.bind(port),
135+
start: port.start && port.start.bind(port),
136+
addEventListener: port.addEventListener.bind(port),
137+
removeEventListener: port.removeEventListener.bind(port)
138+
};
102139
}
103140

104141
Comlink.expose(new BetterSqliteWorker(), toComlink(parentPort!));

0 commit comments

Comments
 (0)