Compile IR to SQL

Author: kevin-dp

packages/electric-db-collection/src/electric.ts

@@ -13,12 +13,14 @@ import {
   ExpectedNumberInAwaitTxIdError,
   TimeoutWaitingForTxIdError,
 } from "./errors"
+import { compileSQL } from "./sql-compiler"
 import type {
   BaseCollectionConfig,
   CollectionConfig,
   DeleteMutationFnParams,
   Fn,
   InsertMutationFnParams,
+  OnLoadMoreOptions,
   SyncConfig,
   UpdateMutationFnParams,
   UtilsRecord,
@@ -494,15 +496,42 @@ function createElectricSync<T extends Row<unknown>>(
         }
       })
 
-      // Return the unsubscribe function
-      return () => {
-        // Unsubscribe from the stream
-        unsubscribeStream()
-        // Abort the abort controller to stop the stream
-        abortController.abort()
+      return {
+        onLoadMore: (opts) => onLoadMore(params, opts),
+        cleanup: () => {
+          // Unsubscribe from the stream
+          unsubscribeStream()
+          // Abort the abort controller to stop the stream
+          abortController.abort()
+        },
       }
     },
     // Expose the getSyncMetadata function
     getSyncMetadata,
   }
 }
+
+async function onLoadMore<T extends Row<unknown>>(
+  syncParams: Parameters<SyncConfig<T>[`sync`]>[0],
+  options: OnLoadMoreOptions
+) {
+  const { begin, write, commit } = syncParams
+
+  // TODO: optimize this by keeping track of which snapshot have been loaded already
+  //       and only load this one if it's not a subset of the ones that have been loaded already
+
+  const snapshotParams = compileSQL<T>(options)
+
+  const snapshot = await requestSnapshot(snapshotParams)
+
+  begin()
+
+  snapshot.data.forEach((row) => {
+    write({
+      type: `insert`,
+      value: row.value,
+    })
+  })
+
+  commit()
+}

packages/electric-db-collection/src/sql-compiler.ts

@@ -0,0 +1,146 @@
+import type { IR, OnLoadMoreOptions } from "@tanstack/db"
+
+export function compileSQL<T>(
+  options: OnLoadMoreOptions
+): ExternalSubsetParamsRecord {
+  const { where, orderBy, limit } = options
+
+  const params: Array<T> = []
+  const compiledSQL: ExternalSubsetParamsRecord = { params }
+
+  if (where) {
+    // TODO: this only works when the where expression's PropRefs directly reference a column of the collection
+    //       doesn't work if it goes through aliases because then we need to know the entire query to be able to follow the reference until the base collection (cf. followRef function)
+    compiledSQL.where = compileBasicExpression(where, params)
+  }
+
+  if (orderBy) {
+    compiledSQL.orderBy = compileOrderBy(orderBy, params)
+  }
+
+  if (limit) {
+    compiledSQL.limit = limit
+  }
+
+  return compiledSQL
+}
+
+/**
+ * Compiles the expression to a SQL string and mutates the params array with the values.
+ * @param exp - The expression to compile
+ * @param params - The params array
+ * @returns The compiled SQL string
+ */
+function compileBasicExpression(
+  exp: IR.BasicExpression<unknown>,
+  params: Array<unknown>
+): string {
+  switch (exp.type) {
+    case `val`:
+      params.push(exp.value)
+      return `$${params.length}`
+    case `ref`:
+      if (exp.path.length !== 1) {
+        throw new Error(
+          `Compiler can't handle nested properties: ${exp.path.join(`.`)}`
+        )
+      }
+      return exp.path[0]!
+    case `func`:
+      return compileFunction(exp, params)
+  }
+}
+
+function compileOrderBy(orderBy: IR.OrderBy, params: Array<unknown>): string {
+  const compiledOrderByClauses = orderBy.map((clause: IR.OrderByClause) =>
+    compileOrderByClause(clause, params)
+  )
+  return compiledOrderByClauses.join(`,`)
+}
+
+function compileOrderByClause(
+  clause: IR.OrderByClause,
+  params: Array<unknown>
+): string {
+  // TODO: what to do with stringSort and locale?
+  //       Correctly supporting them is tricky as it depends on Postgres' collation
+  const { expression, compareOptions } = clause
+  let sql = compileBasicExpression(expression, params)
+
+  if (compareOptions.direction === `desc`) {
+    sql = `${sql} DESC`
+  }
+
+  if (compareOptions.nulls === `first`) {
+    sql = `${sql} NULLS FIRST`
+  }
+
+  if (compareOptions.nulls === `last`) {
+    sql = `${sql} NULLS LAST`
+  }
+
+  return sql
+}
+
+function compileFunction(
+  exp: IR.Func<unknown>,
+  params: Array<unknown> = []
+): string {
+  const { name, args } = exp
+
+  const opName = getOpName(name)
+
+  const compiledArgs = args.map((arg: IR.BasicExpression) =>
+    compileBasicExpression(arg, params)
+  )
+
+  if (isBinaryOp(name)) {
+    if (compiledArgs.length !== 2) {
+      throw new Error(`Binary operator ${name} expects 2 arguments`)
+    }
+    const [lhs, rhs] = compiledArgs
+    return `${lhs} ${opName} ${rhs}`
+  }
+
+  return `${opName}(${compiledArgs.join(`,`)})`
+}
+
+function isBinaryOp(name: string): boolean {
+  const binaryOps = [`eq`, `gt`, `gte`, `lt`, `lte`, `and`, `or`]
+  return binaryOps.includes(name)
+}
+
+function getOpName(name: string): string {
+  const opNames = {
+    eq: `=`,
+    gt: `>`,
+    gte: `>=`,
+    lt: `<`,
+    lte: `<=`,
+    add: `+`,
+    and: `AND`,
+    or: `OR`,
+    not: `NOT`,
+    isUndefined: `IS NULL`,
+    isNull: `IS NULL`,
+    in: `IN`,
+    like: `LIKE`,
+    ilike: `ILIKE`,
+    upper: `UPPER`,
+    lower: `LOWER`,
+    length: `LENGTH`,
+    concat: `CONCAT`,
+    coalesce: `COALESCE`,
+  }
+  return opNames[name as keyof typeof opNames] || name
+}
+
+// TODO: remove this type once we rebase on top of Ilia's PR
+//       that type will be exported by Ilia's PR
+export type ExternalSubsetParamsRecord = {
+  where?: string
+  params?: Record<string, any>
+  limit?: number
+  offset?: number
+  orderBy?: string
+}