feat (client): extract sync API and make DAL optional

.changeset/six-knives-notice.md

@@ -0,0 +1,6 @@
+---
+"electric-sql": patch
+"@electric-sql/prisma-generator": patch
+---
+
+Extract the sync API out of the DAL and make the DAL optional.

.github/workflows/e2e.yml

@@ -113,17 +113,19 @@ jobs:
             https://analytics-api.buildkite.com/v1/uploads
 
   e2e_satellite_tests:
-    name: E2E Satellite tests
     runs-on: electric-e2e-8-32
     strategy:
       matrix:
         dialect: [SQLite, Postgres]
+        dal: [true, false]
+    name: E2E Satellite tests (Dialect ${{ matrix.dialect }} - uses DAL? ${{ matrix.dal }})
     defaults:
       run:
         working-directory: e2e
     env:
       BUILDKITE_ANALYTICS_TOKEN: ${{ secrets.BUILDKITE_TEST_ANALYTICS_E2E }}
       DIALECT: ${{ matrix.dialect }}
+      DAL: ${{ matrix.dal }}
     steps:
       - uses: actions/checkout@v3
         with:

clients/typescript/package.json

@@ -54,6 +54,7 @@
     "./node": "./dist/drivers/better-sqlite3/index.js",
     "./node-postgres": "./dist/drivers/node-postgres/index.js",
     "./pglite": "./dist/drivers/pglite/index.js",
+    "./protocol": "./dist/_generated/protocol/satellite.js",
     "./react": "./dist/frameworks/react/index.js",
     "./tauri-postgres": "./dist/drivers/tauri-postgres/index.js",
     "./vuejs": "./dist/frameworks/vuejs/index.js",
@@ -78,6 +79,9 @@
       "capacitor": [
         "./dist/drivers/capacitor-sqlite/index.d.ts"
       ],
+      "client": [
+        "./dist/client/index.d.ts"
+      ],
       "expo": [
         "./dist/drivers/expo-sqlite/index.d.ts"
       ],
@@ -96,6 +100,9 @@
       "pglite": [
         "./dist/drivers/pglite/index.d.ts"
       ],
+      "protocol": [
+        "./dist/_generated/protocol/satellite.d.ts"
+      ],
       "react": [
         "./dist/frameworks/react/index.d.ts"
       ],
@@ -181,6 +188,7 @@
     "lodash.flow": "^3.5.0",
     "lodash.groupby": "^4.6.0",
     "lodash.isequal": "^4.5.0",
+    "lodash.keyby": "^4.6.0",
     "lodash.mapvalues": "^4.6.0",
     "lodash.omitby": "^4.6.0",
     "lodash.partition": "^4.6.0",
@@ -209,6 +217,7 @@
     "@types/lodash.flow": "^3.5.7",
     "@types/lodash.groupby": "^4.6.7",
     "@types/lodash.isequal": "^4.5.6",
+    "@types/lodash.keyby": "^4.6.9",
     "@types/lodash.mapvalues": "^4.6.7",
     "@types/lodash.omitby": "^4.6.7",
     "@types/lodash.partition": "^4.6.7",

clients/typescript/src/client/conversions/index.ts

@@ -0,0 +1,3 @@
+export { postgresConverter } from './postgres'
+export { sqliteConverter } from './sqlite'
+export { PgBasicType } from './types'

clients/typescript/src/client/conversions/input.ts

@@ -247,7 +247,7 @@ export class InputTransformer {
     value: any,
     fields: Fields
   ): any {
-    const pgType = fields.get(field)
+    const pgType = fields[field]
 
     if (!pgType) throw new InvalidArgumentError(`Unknown field ${field}`)
 
@@ -335,7 +335,7 @@ export function transformFields(
   // as those will be transformed later when the query on the related field is processed.
   const copied: Record<string, any> = { ...o }
   Object.entries(o).forEach(([field, value]) => {
-    const pgType = fields.get(field)
+    const pgType = fields[field]
     // Skip anything that's not an actual column on the table
     if (pgType === undefined) return
 
@@ -363,7 +363,10 @@ export function isFilterObject(value: any): boolean {
  * @returns A filtered object.
  */
 function keepTableFieldsOnly(o: object, fields: Fields) {
-  return filterKeys(o, fields)
+  return filterKeys(o, {
+    ...fields,
+    has: (x) => Object.hasOwn(fields, x),
+  })
 }
 
 /**

clients/typescript/src/client/index.ts

@@ -0,0 +1,3 @@
+export type { TableName, AnyTable, AnyTableSchema } from './model'
+export { type DbSchema, createDbDescription } from './util/relations'
+export * from './conversions'

clients/typescript/src/client/model/builder.ts

@@ -251,7 +251,7 @@ export class Builder {
    * The DAL will convert the string into a BigInt in the `fromSqlite` function from `../conversions/sqlite.ts`.
    */
   private castBigIntToText(field: string) {
-    const pgType = this._tableDescription.fields.get(field)
+    const pgType = this._tableDescription.fields[field]
     if (pgType === PgBasicType.PG_INT8 && this.dialect === 'SQLite') {
       const quotedField = quoteIdentifier(field)
       return `cast(${quotedField} as TEXT) AS ${quotedField}`
@@ -308,7 +308,7 @@ export class Builder {
       // if field is of type BigInt cast the result to TEXT
       // because not all adapters deal well with BigInts
       // the DAL will convert the string into a BigInt in the `fromSqlite` function from `../conversions/sqlite.ts`.
-      const pgType = this._tableDescription.fields.get(field)
+      const pgType = this._tableDescription.fields[field]
       if (pgType === PgBasicType.PG_INT8 && this.dialect === 'SQLite') {
         // make a raw string and quote the field name ourselves
         // because otherwise Squel would add quotes around the entire cast

clients/typescript/src/client/model/client.ts

@@ -1,17 +1,32 @@
 import { ElectricNamespace } from '../../electric/namespace'
-import { DbSchema, TableSchema } from './schema'
+import { DbSchema, TableSchema, TableSchemas } from './schema'
 import { rawQuery, liveRawQuery, unsafeExec, Table } from './table'
-import { Row, Statement } from '../../util'
+import {
+  QualifiedTablename,
+  ReplicatedRowTransformer,
+  Row,
+  Statement,
+} from '../../util'
 import { LiveResultContext } from './model'
 import { Notifier } from '../../notifiers'
 import { DatabaseAdapter } from '../../electric/adapter'
-import { GlobalRegistry, Registry, Satellite } from '../../satellite'
-import { ReplicationTransformManager } from './transforms'
+import {
+  GlobalRegistry,
+  Registry,
+  Satellite,
+  ShapeSubscription,
+} from '../../satellite'
+import {
+  IReplicationTransformManager,
+  ReplicationTransformManager,
+  setReplicationTransform,
+} from './transforms'
 import { Dialect } from '../../migrators/query-builder/builder'
 import { InputTransformer } from '../conversions/input'
 import { sqliteConverter } from '../conversions/sqlite'
 import { postgresConverter } from '../conversions/postgres'
 import { IShapeManager } from './shapes'
+import { ShapeInputWithTable, sync } from './sync'
 
 export type ClientTables<DB extends DbSchema<any>> = {
   [Tbl in keyof DB['tables']]: DB['tables'][Tbl] extends TableSchema<
@@ -96,25 +111,62 @@ interface RawQueries {
 export class ElectricClient<
   DB extends DbSchema<any>
 > extends ElectricNamespace {
-  public sync: Omit<IShapeManager, 'subscribe'>
+  public sync: Omit<IShapeManager, 'subscribe'> & {
+    /**
+     * Subscribes to the given shape, returnig a {@link ShapeSubscription} object which
+     * can be used to wait for the shape to sync initial data.
+     *
+     * NOTE: If you establish a shape subscription that has already synced its initial data,
+     * awaiting `shape.synced` will always resolve immediately as shape subscriptions are persisted.
+     * i.e.: imagine that you re-sync the same shape during subsequent application loads.
+     * Awaiting `shape.synced` a second time will only ensure that the initial
+     * shape load is complete. It does not ensure that the replication stream
+     * has caught up to the central DB's more recent state.
+     *
+     * @param i - The shape to subscribe to
+     * @param key - An optional unique key that identifies the subscription
+     * @returns A shape subscription
+     */
+    subscribe: (
+      i: ShapeInputWithTable,
+      key?: string
+    ) => Promise<ShapeSubscription>
+  }
 
   private constructor(
     public db: ClientTables<DB> & RawQueries,
     dbName: string,
+    private _dbDescription: DB,
     adapter: DatabaseAdapter,
     notifier: Notifier,
     public readonly satellite: Satellite,
-    registry: Registry | GlobalRegistry
+    registry: Registry | GlobalRegistry,
+    private _replicationTransformManager: IReplicationTransformManager
   ) {
     super(dbName, adapter, notifier, registry)
     this.satellite = satellite
     // Expose the Shape Sync API without additional properties
     this.sync = {
       syncStatus: this.satellite.syncStatus.bind(this.satellite),
+      subscribe: sync.bind(null, this.satellite, this._dbDescription),
       unsubscribe: this.satellite.unsubscribe.bind(this.satellite),
     }
   }
 
+  setReplicationTransform<
+    T extends Record<string, unknown> = Record<string, unknown>
+  >(
+    qualifiedTableName: QualifiedTablename,
+    i: ReplicatedRowTransformer<T>
+  ): void {
+    setReplicationTransform<T>(
+      this._dbDescription,
+      this._replicationTransformManager,
+      qualifiedTableName,
+      i
+    )
+  }
+
   /**
    * Connects to the Electric sync service.
    * This method is idempotent, it is safe to call it multiple times.
@@ -136,7 +188,10 @@ export class ElectricClient<
     this.satellite.clientDisconnect()
   }
 
-  // Builds the DAL namespace from a `dbDescription` object
+  /**
+   * Builds the DAL namespace from a `dbDescription` object
+   * @param minimalDbDescription - A minimal description of the database schema can be provided in order to use Electric without the DAL.
+   */
   static create<DB extends DbSchema<any>>(
     dbName: string,
     dbDescription: DB,
@@ -154,30 +209,44 @@ export class ElectricClient<
     )
     const inputTransformer = new InputTransformer(converter)
 
-    const createTable = (tableName: string) => {
-      return new Table(
-        tableName,
-        adapter,
-        notifier,
-        satellite,
-        replicationTransformManager,
-        dbDescription,
-        inputTransformer,
-        dialect
-      )
-    }
+    // Check if we need to create the DAL
+    // If the schemas are missing from the `dbDescription``
+    // it means that the user did not generate the Electric client
+    // and thus we don't create the DAL.
+    // This is needed because we piggyback the minimal DB description (that is used without the DAL)
+    // on the same DB description argument as the one that is used with the DAL.
+    const ts: Array<[string, TableSchemas]> = Object.entries(
+      dbDescription.tables
+    )
+    const withDal = ts.length > 0 && ts[0][1].modelSchema !== undefined
+    let dal = {} as ClientTables<DB>
 
-    // Create all tables
-    const dal = Object.fromEntries(
-      Object.keys(tables).map((tableName) => {
-        return [tableName, createTable(tableName)]
-      })
-    ) as ClientTables<DB>
+    if (withDal) {
+      const createTable = (tableName: string) => {
+        return new Table(
+          tableName,
+          adapter,
+          notifier,
+          satellite,
+          replicationTransformManager,
+          dbDescription,
+          inputTransformer,
+          dialect
+        )
+      }
 
-    // Now inform each table about all tables
-    Object.keys(dal).forEach((tableName) => {
-      dal[tableName].setTables(new Map(Object.entries(dal)))
-    })
+      // Create all tables
+      dal = Object.fromEntries(
+        Object.keys(tables).map((tableName) => {
+          return [tableName, createTable(tableName)]
+        })
+      ) as ClientTables<DB>
+
+      // Now inform each table about all tables
+      Object.keys(dal).forEach((tableName) => {
+        dal[tableName].setTables(new Map(Object.entries(dal)))
+      })
+    }
 
     const db: ClientTables<DB> & RawQueries = {
       ...dal,
@@ -191,10 +260,12 @@ export class ElectricClient<
     return new ElectricClient(
       db,
       dbName,
+      dbDescription,
       adapter,
       notifier,
       satellite,
-      registry
+      registry,
+      replicationTransformManager
     )
   }
 }

clients/typescript/src/client/model/index.ts

@@ -1,7 +1,13 @@
 export { ElectricClient } from './client'
 export type { ClientTables } from './client'
-export type { TableSchema } from './schema'
+export type {
+  TableSchema,
+  TableSchemas,
+  TableName,
+  AnyTableSchema,
+} from './schema'
 export { DbSchema, Relation } from './schema'
 export { Table } from './table'
+export type { AnyTable } from './table'
 export type { HKT } from '../util/hkt'
 export type { SyncStatus } from './shapes'

clients/typescript/src/client/model/schema.ts

@@ -17,7 +17,7 @@ export type TableName = string
 export type FieldName = string
 export type RelationName = string
 
-export type Fields = Map<FieldName, PgType>
+export type Fields = Record<FieldName, PgType>
 
 export type TableSchema<
   T extends Record<string, any>,
@@ -76,11 +76,21 @@ export type ExtendedTableSchema<
   incomingRelations: Relation[]
 }
 
-export type TableSchemas = Record<
-  TableName,
-  TableSchema<any, any, any, any, any, any, any, any, any, HKT>
+export type AnyTableSchema = TableSchema<
+  any,
+  any,
+  any,
+  any,
+  any,
+  any,
+  any,
+  any,
+  any,
+  HKT
 >
 
+export type TableSchemas = Record<TableName, AnyTableSchema>
+
 export type ExtendedTableSchemas = Record<
   TableName,
   ExtendedTableSchema<any, any, any, any, any, any, any, any, any, HKT>
@@ -190,7 +200,7 @@ export class DbSchema<T extends TableSchemas> {
   }
 
   getFieldNames(table: TableName): FieldName[] {
-    return Array.from(this.getFields(table).keys())
+    return Array.from(Object.keys(this.getFields(table)))
   }
 
   hasRelationForField(table: TableName, field: FieldName): boolean {