Reduce lodash usage.

Author: rkistner

packages/powersync-sdk-common/src/client/AbstractPowerSyncDatabase.ts

@@ -1,24 +1,25 @@
-import _ from 'lodash';
 import { Mutex } from 'async-mutex';
+import { EventIterator } from 'event-iterator';
 import Logger, { ILogger } from 'js-logger';
+import intersection from 'lodash/intersection';
+import throttle from 'lodash/throttle';
 import { DBAdapter, QueryResult, Transaction, isBatchedUpdateNotification } from '../db/DBAdapter';
-import { Schema } from '../db/schema/Schema';
 import { SyncStatus } from '../db/crud/SyncStatus';
 import { UploadQueueStats } from '../db/crud/UploadQueueStatus';
+import { Schema } from '../db/schema/Schema';
+import { BaseObserver } from '../utils/BaseObserver';
+import { mutexRunExclusive } from '../utils/mutex';
+import { quoteIdentifier } from '../utils/strings';
 import { PowerSyncBackendConnector } from './connection/PowerSyncBackendConnector';
+import { BucketStorageAdapter, PSInternalTable } from './sync/bucket/BucketStorageAdapter';
+import { CrudBatch } from './sync/bucket/CrudBatch';
+import { CrudEntry, CrudEntryJSON } from './sync/bucket/CrudEntry';
+import { CrudTransaction } from './sync/bucket/CrudTransaction';
 import {
   AbstractStreamingSyncImplementation,
   DEFAULT_CRUD_UPLOAD_THROTTLE_MS,
   StreamingSyncImplementationListener
 } from './sync/stream/AbstractStreamingSyncImplementation';
-import { CrudBatch } from './sync/bucket/CrudBatch';
-import { CrudTransaction } from './sync/bucket/CrudTransaction';
-import { BucketStorageAdapter, PSInternalTable } from './sync/bucket/BucketStorageAdapter';
-import { CrudEntry, CrudEntryJSON } from './sync/bucket/CrudEntry';
-import { mutexRunExclusive } from '../utils/mutex';
-import { BaseObserver } from '../utils/BaseObserver';
-import { EventIterator } from 'event-iterator';
-import { quoteIdentifier } from '../utils/strings';
 
 export interface DisconnectAndClearOptions {
   /** When set to false, data in local-only tables is preserved. */
@@ -538,25 +539,26 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Source tables are automatically detected using `EXPLAIN QUERY PLAN`.
    */
   async *watch(sql: string, parameters?: any[], options?: SQLWatchOptions): AsyncIterable<QueryResult> {
-    //Fetch initial data
+    // Fetch initial data
     yield await this.executeReadOnly(sql, parameters);
 
-    const resolvedTables = options?.tables ?? [];
+    const resolvedTables = options?.tables ? [...options.tables] : [];
     if (!options?.tables) {
       const explained = await this.getAll<{ opcode: string; p3: number; p2: number }>(`EXPLAIN ${sql}`, parameters);
-      const rootPages = _.chain(explained)
-        .filter((row) => row['opcode'] == 'OpenRead' && row['p3'] == 0 && _.isNumber(row['p2']))
-        .map((row) => row['p2'])
-        .value();
+      const rootPages = explained
+        .filter((row) => row.opcode == 'OpenRead' && row.p3 == 0 && typeof row.p2 == 'number')
+        .map((row) => row.p2);
       const tables = await this.getAll<{ tbl_name: string }>(
-        `SELECT tbl_name FROM sqlite_master WHERE rootpage IN (SELECT json_each.value FROM json_each(?))`,
+        `SELECT DISTINCT tbl_name FROM sqlite_master WHERE rootpage IN (SELECT json_each.value FROM json_each(?))`,
         [JSON.stringify(rootPages)]
       );
-      tables.forEach((t) => resolvedTables.push(t.tbl_name.replace(POWERSYNC_TABLE_MATCH, '')));
+      for (let table of tables) {
+        resolvedTables.push(table.tbl_name.replace(POWERSYNC_TABLE_MATCH, ''));
+      }
     }
     for await (const event of this.onChange({
       ...(options ?? {}),
-      tables: _.uniq(resolvedTables)
+      tables: resolvedTables
     })) {
       yield await this.executeReadOnly(sql, parameters);
     }
@@ -578,12 +580,12 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     const throttleMs = resolvedOptions.throttleMs ?? DEFAULT_WATCH_THROTTLE_MS;
 
     return new EventIterator<WatchOnChangeEvent>((eventOptions) => {
-      const flushTableUpdates = _.throttle(
+      const flushTableUpdates = throttle(
         () => {
-          const intersection = _.intersection(watchedTables, throttledTableUpdates);
-          if (intersection.length) {
+          const changedTables = intersection(watchedTables, throttledTableUpdates);
+          if (changedTables.length > 0) {
             eventOptions.push({
-              changedTables: intersection
+              changedTables
             });
           }
           throttledTableUpdates = [];

packages/powersync-sdk-common/src/client/sync/bucket/CrudEntry.ts

@@ -1,4 +1,4 @@
-import _ from 'lodash';
+import isEqual from 'lodash/isEqual';
 
 /**
  * 64-bit unsigned integer stored as a string in base-10.
@@ -109,7 +109,7 @@ export class CrudEntry {
   }
 
   equals(entry: CrudEntry) {
-    return _.isEqual(this.toComparisonArray(), entry.toComparisonArray());
+    return isEqual(this.toComparisonArray(), entry.toComparisonArray());
   }
 
   /**

packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -1,4 +1,6 @@
-import _ from 'lodash';
+import throttle from 'lodash/throttle';
+import defer from 'lodash/defer';
+
 import Logger, { ILogger } from 'js-logger';
 
 import {
@@ -76,7 +78,7 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
       }
     });
 
-    this.triggerCrudUpload = _.throttle(
+    this.triggerCrudUpload = throttle(
       () => {
         if (!this.syncStatus.connected || this.syncStatus.dataFlowStatus.uploading) {
           return;
@@ -206,6 +208,7 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
           after: after
         }));
 
+        // These are compared by reference
         let targetCheckpoint: Checkpoint | null = null;
         let validatedCheckpoint: Checkpoint | null = null;
         let appliedCheckpoint: Checkpoint | null = null;
@@ -223,7 +226,7 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
           // A connection is active and messages are being received
           if (!this.syncStatus.connected) {
             // There is a connection now
-            _.defer(() => this.triggerCrudUpload());
+            defer(() => this.triggerCrudUpload());
             this.updateSyncStatus({
               connected: true
             });
@@ -256,7 +259,7 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
               // Continue waiting.
               // landing here the whole time
             } else {
-              appliedCheckpoint = _.clone(targetCheckpoint);
+              appliedCheckpoint = targetCheckpoint;
               this.logger.debug('validated checkpoint', appliedCheckpoint);
               this.updateSyncStatus({
                 connected: true,
@@ -267,7 +270,7 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
               });
             }
 
-            validatedCheckpoint = _.clone(targetCheckpoint);
+            validatedCheckpoint = targetCheckpoint;
           } else if (isStreamingSyncCheckpointDiff(line)) {
             // TODO: It may be faster to just keep track of the diff, instead of the entire checkpoint
             if (targetCheckpoint == null) {
@@ -319,12 +322,12 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
           } else {
             this.logger.debug('Sync complete');
 
-            if (_.isEqual(targetCheckpoint, appliedCheckpoint)) {
+            if (targetCheckpoint === appliedCheckpoint) {
               this.updateSyncStatus({
                 connected: true,
                 lastSyncedAt: new Date()
               });
-            } else if (_.isEqual(validatedCheckpoint, targetCheckpoint)) {
+            } else if (validatedCheckpoint === targetCheckpoint) {
               const result = await this.options.adapter.syncLocalDatabase(targetCheckpoint!);
               if (!result.checkpointValid) {
                 // This means checksums failed. Start again with a new checkpoint.
@@ -335,7 +338,7 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
                 // Checksums valid, but need more data for a consistent checkpoint.
                 // Continue waiting.
               } else {
-                appliedCheckpoint = _.clone(targetCheckpoint);
+                appliedCheckpoint = targetCheckpoint;
                 this.updateSyncStatus({
                   connected: true,
                   lastSyncedAt: new Date(),
@@ -378,7 +381,10 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
     const updatedStatus = new SyncStatus({
       connected: options.connected ?? this.syncStatus.connected,
       lastSyncedAt: options.lastSyncedAt ?? this.syncStatus.lastSyncedAt,
-      dataFlow: _.merge(_.clone(this.syncStatus.dataFlowStatus), options.dataFlow ?? {})
+      dataFlow: {
+        ...this.syncStatus.dataFlowStatus,
+        ...options.dataFlow
+      }
     });
 
     if (!this.syncStatus.isEqual(updatedStatus)) {

packages/powersync-sdk-common/src/db/crud/SyncStatus.ts

@@ -1,4 +1,4 @@
-import _ from 'lodash';
+import isEqual from 'lodash/isEqual';
 
 export type SyncDataFlowStatus = Partial<{
   downloading: boolean;
@@ -49,7 +49,7 @@ export class SyncStatus {
   }
 
   isEqual(status: SyncStatus) {
-    return _.isEqual(this.options, status.options);
+    return isEqual(this.options, status.options);
   }
 
   getMessage() {

packages/powersync-sdk-common/src/db/schema/Table.ts

@@ -1,4 +1,3 @@
-import _, { indexOf } from 'lodash';
 import { Column } from '../Column';
 import type { Index } from './Index';
 import { TableV2 } from './TableV2';
@@ -86,10 +85,13 @@ export class Table {
   }
 
   get validName() {
-    return _.chain([this.name, this.viewNameOverride])
-      .compact()
-      .every((name) => !InvalidSQLCharacters.test(name))
-      .value();
+    if (InvalidSQLCharacters.test(this.name)) {
+      return false;
+    }
+    if (this.viewNameOverride != null && InvalidSQLCharacters.test(this.viewNameOverride)) {
+      return false;
+    }
+    return true;
   }
 
   validate() {

packages/powersync-sdk-web/src/db/adapters/AbstractWebPowerSyncDatabaseOpenFactory.ts

@@ -1,4 +1,3 @@
-import _ from 'lodash';
 import {
   AbstractPowerSyncDatabase,
   AbstractPowerSyncDatabaseOpenFactory,
@@ -73,10 +72,14 @@ export abstract class AbstractWebPowerSyncDatabaseOpenFactory extends AbstractPo
   }
 
   protected resolveDBFlags(): WebPowerSyncFlags {
-    return _.merge(_.clone(DEFAULT_POWERSYNC_FLAGS), {
-      ssrMode: this.isServerSide(),
-      enableMultiTabs: this.options.flags?.enableMultiTabs
-    });
+    let flags = {
+      ...DEFAULT_POWERSYNC_FLAGS,
+      ssrMode: this.isServerSide()
+    };
+    if (typeof this.options.flags?.enableMultiTabs != 'undefined') {
+      flags.enableMultiTabs = this.options.flags.enableMultiTabs;
+    }
+    return flags;
   }
 
   generateInstance(options: PowerSyncDatabaseOptions): AbstractPowerSyncDatabase {

packages/powersync-sdk-web/src/db/sync/SSRWebStreamingSyncImplementation.ts

@@ -1,4 +1,3 @@
-import _ from 'lodash';
 import {
   AbstractStreamingSyncImplementation,
   AbstractStreamingSyncImplementationOptions,

packages/powersync-sdk-web/src/db/sync/SharedWebStreamingSyncImplementation.ts

@@ -1,4 +1,3 @@
-import _ from 'lodash';
 import { v4 as uuid } from 'uuid';
 import * as Comlink from 'comlink';
 import {

packages/powersync-sdk-web/src/db/sync/WebStreamingSyncImplementation.ts

@@ -1,4 +1,3 @@
-import _ from 'lodash';
 import {
   AbstractStreamingSyncImplementation,
   AbstractStreamingSyncImplementationOptions,

packages/powersync-sdk-web/src/worker/db/open-db.ts

@@ -1,6 +1,5 @@
 import * as SQLite from '@journeyapps/wa-sqlite';
 import '@journeyapps/wa-sqlite';
-import _ from 'lodash';
 import * as Comlink from 'comlink';
 import { v4 as uuid } from 'uuid';
 import { QueryResult } from '@journeyapps/powersync-sdk-common';
@@ -93,21 +92,16 @@ export async function _openDB(dbFileName: string): Promise<DBWorkerInterface> {
         }
       }
 
-      const rows = _.chain(results)
-        .filter(({ rows }) => !!rows.length)
-        .flatMap(({ columns, rows }) =>
-          _.map(rows, (row) =>
-            _.reduce(
-              columns,
-              (out: Record<string, any>, key: string, index) => {
-                out[key] = row[index];
-                return out;
-              },
-              {}
-            )
-          )
-        )
-        .value();
+      let rows: Record<string, any>[] = [];
+      for (let resultset of results) {
+        for (let row of resultset.rows) {
+          let outRow: Record<string, any> = {};
+          resultset.columns.forEach((key, index) => {
+            outRow[key] = row[index];
+          });
+          rows.push(outRow);
+        }
+      }
 
       const result = {
         insertId: sqlite3.last_insert_id(db),