From 194b151c0e31da45b8c11ed00aa33479655c7fb5 Mon Sep 17 00:00:00 2001 From: sammy Date: Thu, 28 Dec 2023 10:36:36 -0500 Subject: [PATCH] server db is sync --- src/lib/server/sync-db/db.ts | 4 +-- .../websockets/features/offline/sync.ts | 34 +++++++++---------- 2 files changed, 18 insertions(+), 20 deletions(-) diff --git a/src/lib/server/sync-db/db.ts b/src/lib/server/sync-db/db.ts index 8294db7..01a9832 100644 --- a/src/lib/server/sync-db/db.ts +++ b/src/lib/server/sync-db/db.ts @@ -1,14 +1,14 @@ import Database from 'better-sqlite3'; import { extensionPath } from '@vlcn.io/crsqlite'; -export async function dbFrom(filename) { +export function dbFrom(filename) { // TODO: should be an env-var so we can use Render's persistent disk as the path const db = new Database(`./sync-dbs/${filename}`); db.pragma('journal_mode = WAL'); db.loadExtension(extensionPath); // TODO: import schema from same place as frontend - await db.exec(`CREATE TABLE IF NOT EXISTS todos (id PRIMARY KEY NOT NULL, content, complete); + db.exec(`CREATE TABLE IF NOT EXISTS todos (id PRIMARY KEY NOT NULL, content, complete); SELECT crsql_as_crr('todos'); CREATE TABLE IF NOT EXISTS todonts (id PRIMARY KEY NOT NULL, content, complete); SELECT crsql_as_crr('todonts'); diff --git a/src/lib/server/websockets/features/offline/sync.ts b/src/lib/server/websockets/features/offline/sync.ts index 6bd2848..ff089c4 100644 --- a/src/lib/server/websockets/features/offline/sync.ts +++ b/src/lib/server/websockets/features/offline/sync.ts @@ -38,8 +38,8 @@ export class Sync { clientSiteId: string; clientVersion: number; }) { - const db = await dbFrom(`${ws.session.user.userId}.db`); - const { siteId, version } = await db + const db = dbFrom(`${ws.session.user.userId}.db`); + const { siteId, version } = db .prepare('SELECT hex(crsql_site_id()) as siteId, crsql_db_version() as version;') .get(); const sync = new Sync({ ws, stream, db, version, siteId, clientSiteId }); @@ -66,21 +66,19 @@ export class Sync { 1. client inserts a new entry and sends an update 2. client receives a message of `type: 'connected'`, then it sends up all changes */ - changes.forEach(async (change, i) => { - await db.prepare(INSERT_CHANGES).run(...change); + changes.forEach((change, i) => { + db.prepare(INSERT_CHANGES).run(...change); }); const changeSiteVersions = latestVersions(changes); - changeSiteVersions.forEach(async ([changeSiteId, changeDbVersion]) => { - await db - .prepare( - `INSERT INTO crsql_tracked_peers (site_id, version, tag, event) + changeSiteVersions.forEach(([changeSiteId, changeDbVersion]) => { + db.prepare( + `INSERT INTO crsql_tracked_peers (site_id, version, tag, event) VALUES (unhex(?), ?, 0, 0) ON CONFLICT([site_id], [tag], [event]) DO UPDATE SET version=excluded.version` - ) - .run(changeSiteId, changeDbVersion); + ).run(changeSiteId, changeDbVersion); }); await sync.receive(data); @@ -122,27 +120,27 @@ export class Sync { this.clientSiteId = clientSiteId; } - async catchUpServer(clientSiteId) { + catchUpServer(clientSiteId) { // Here we can send down the last seen id or something. // that way we don't need the entire contents of the db - const result = await this.db + const result = this.db .prepare(`SELECT version FROM crsql_tracked_peers WHERE site_id = unhex(?)`) .get(clientSiteId); const version = result?.version ?? 0; - await this.notify(JSON.stringify({ type: 'connected', siteId: clientSiteId, version })); + this.notify(JSON.stringify({ type: 'connected', siteId: clientSiteId, version })); } - async catchUpClient(clientSiteId: string) { + catchUpClient(clientSiteId: string) { // Maybe we can do something to only send down what's needed. // just updates after the last update by `${clientSiteId} - const result = await this.db + const result = this.db .prepare(`SELECT version FROM crsql_tracked_peers WHERE site_id = unhex(?)`) .get(clientSiteId); const lastVersion = result?.version ?? 0; - const changes = await this.db + const changes = this.db .prepare( `SELECT "table", hex("pk") as pk, "cid", "val", "col_version", "db_version", hex("site_id") as site_id, "cl", "seq" FROM crsql_changes WHERE site_id != unhex(:clientSiteId) @@ -155,8 +153,8 @@ export class Sync { const changeSiteVersions = latestVersions(changes.map((change) => Object.values(change))); - changeSiteVersions.forEach(async ([changeSiteId, changeDbVersion]) => { - await this.db + changeSiteVersions.forEach(([changeSiteId, changeDbVersion]) => { + this.db .prepare( `INSERT INTO crsql_tracked_peers (site_id, version, tag, event) VALUES (unhex(?), ?, 0, 0)