feat(electric): Make it possible to select the SQL dialect in /api/migrations and Satellite

clients/typescript/src/_generated/protocol/satellite.ts

@@ -122,6 +122,11 @@ export interface SatInStartReplicationReq {
    * observed additional data before disconnect
    */
   observedTransactionData: Long[];
+  /**
+   * The SQL dialect used by the client
+   * Defaults to SQLite if not specified
+   */
+  sqlDialect?: SatInStartReplicationReq_Dialect | undefined;
 }
 
 export enum SatInStartReplicationReq_Option {
@@ -130,6 +135,12 @@ export enum SatInStartReplicationReq_Option {
   UNRECOGNIZED = -1,
 }
 
+export enum SatInStartReplicationReq_Dialect {
+  SQLITE = 0,
+  POSTGRES = 1,
+  UNRECOGNIZED = -1,
+}
+
 /** (Producer) The result of the start replication requests */
 export interface SatInStartReplicationResp {
   $type: "Electric.Satellite.SatInStartReplicationResp";
@@ -457,9 +468,15 @@ export interface SatOpMigrate_PgColumnType {
   size: number[];
 }
 
+/** reserved 2; */
 export interface SatOpMigrate_Column {
   $type: "Electric.Satellite.SatOpMigrate.Column";
   name: string;
+  /**
+   * deprecated
+   * leaving it here to avoid breaking TypeScript tests that have hard-coded,
+   * base64-encoded SatOpMigrate messages.
+   */
   sqliteType: string;
   pgType: SatOpMigrate_PgColumnType | undefined;
 }
@@ -1102,6 +1119,7 @@ function createBaseSatInStartReplicationReq(): SatInStartReplicationReq {
     subscriptionIds: [],
     schemaVersion: undefined,
     observedTransactionData: [],
+    sqlDialect: undefined,
   };
 }
 
@@ -1128,6 +1146,9 @@ export const SatInStartReplicationReq = {
       writer.uint64(v);
     }
     writer.ldelim();
+    if (message.sqlDialect !== undefined) {
+      writer.uint32(56).int32(message.sqlDialect);
+    }
     return writer;
   },
 
@@ -1193,6 +1214,13 @@ export const SatInStartReplicationReq = {
           }
 
           break;
+        case 7:
+          if (tag !== 56) {
+            break;
+          }
+
+          message.sqlDialect = reader.int32() as any;
+          continue;
       }
       if ((tag & 7) === 4 || tag === 0) {
         break;
@@ -1213,6 +1241,7 @@ export const SatInStartReplicationReq = {
     message.subscriptionIds = object.subscriptionIds?.map((e) => e) || [];
     message.schemaVersion = object.schemaVersion ?? undefined;
     message.observedTransactionData = object.observedTransactionData?.map((e) => Long.fromValue(e)) || [];
+    message.sqlDialect = object.sqlDialect ?? undefined;
     return message;
   },
 };

clients/typescript/src/config/index.ts

@@ -44,13 +44,18 @@ export interface ElectricConfig {
   connectionBackOffOptions?: ConnectionBackOffOptions
 }
 
+export type ElectricConfigWithDialect = ElectricConfig & {
+  dialect?: 'SQLite' | 'Postgres'
+}
+
 export type HydratedConfig = {
   auth: AuthConfig
   replication: {
     host: string
     port: number
     ssl: boolean
     timeout: number
+    dialect: 'SQLite' | 'Postgres'
   }
   debug: boolean
   connectionBackOffOptions: ConnectionBackOffOptions
@@ -68,7 +73,9 @@ export type InternalElectricConfig = {
   connectionBackOffOptions?: ConnectionBackOffOptions
 }
 
-export const hydrateConfig = (config: ElectricConfig): HydratedConfig => {
+export const hydrateConfig = (
+  config: ElectricConfigWithDialect
+): HydratedConfig => {
   const auth = config.auth ?? {}
 
   const debug = config.debug ?? false
@@ -86,6 +93,7 @@ export const hydrateConfig = (config: ElectricConfig): HydratedConfig => {
     port: port,
     ssl: sslEnabled,
     timeout: config.timeout ?? 3000,
+    dialect: config.dialect ?? 'SQLite',
   }
 
   const {

clients/typescript/src/electric/index.ts

@@ -1,4 +1,4 @@
-import { ElectricConfig, hydrateConfig } from '../config/index'
+import { ElectricConfigWithDialect, hydrateConfig } from '../config/index'
 import { DatabaseAdapter } from './adapter'
 import { BundleMigrator, Migrator } from '../migrators/index'
 import { EventNotifier, Notifier } from '../notifiers/index'
@@ -47,7 +47,7 @@ export const electrify = async <DB extends DbSchema<any>>(
   dbDescription: DB,
   adapter: DatabaseAdapter,
   socketFactory: SocketFactory,
-  config: ElectricConfig = {},
+  config: ElectricConfigWithDialect = {},
   opts?: Omit<ElectrifyOptions, 'adapter' | 'socketFactory'>
 ): Promise<ElectricClient<DB>> => {
   setLogLevel(config.debug ? 'TRACE' : 'WARN')

clients/typescript/src/migrators/triggers.ts

@@ -8,12 +8,7 @@ type ForeignKey = {
 }
 
 type ColumnName = string
-type SQLiteType = string
-type PgType = string
-type ColumnType = {
-  sqliteType: SQLiteType
-  pgType: PgType
-}
+type ColumnType = string
 type ColumnTypes = Record<ColumnName, ColumnType>
 
 export type Table = {
@@ -71,11 +66,11 @@ export function generateOplogTriggers(
       SELECT
         CASE
           ${primary
-            .map(
-              (col) =>
-                `WHEN old."${col}" != new."${col}" THEN\n\t\tRAISE (ABORT, 'cannot change the value of column ${col} as it belongs to the primary key')`
-            )
-            .join('\n')}
+        .map(
+          (col) =>
+            `WHEN old."${col}" != new."${col}" THEN\n\t\tRAISE (ABORT, 'cannot change the value of column ${col} as it belongs to the primary key')`
+        )
+        .join('\n')}
         END;
     END;
     `,
@@ -283,17 +278,15 @@ function joinColsForJSON(
   // Perform transformations on some columns to ensure consistent
   // serializability into JSON
   const transformIfNeeded = (col: string, targetedCol: string) => {
-    const tpes = colTypes[col]
-    const sqliteType = tpes.sqliteType
-    const pgType = tpes.pgType
+    const pgType = colTypes[col]
 
     // cast REALs, INT8s, BIGINTs to TEXT to work around SQLite's `json_object` bug
-    if (sqliteType === 'REAL' || pgType === 'INT8' || pgType === 'BIGINT') {
+    if (pgType === 'FLOAT4' || pgType === 'REAL' || pgType === 'INT8' || pgType === 'BIGINT') {
       return `cast(${targetedCol} as TEXT)`
     }
 
     // transform blobs/bytestrings into hexadecimal strings for JSON encoding
-    if (sqliteType === 'BLOB' || pgType === 'BYTEA') {
+    if (pgType === 'BYTEA') {
       return `CASE WHEN ${targetedCol} IS NOT NULL THEN hex(${targetedCol}) ELSE NULL END`
     }
     return targetedCol

clients/typescript/src/satellite/client.ts

@@ -26,6 +26,7 @@ import {
   Root,
   RootClientImpl,
   SatRpcRequest,
+  SatInStartReplicationReq_Dialect,
 } from '../_generated/protocol/satellite'
 import {
   getObjFromString,
@@ -131,6 +132,7 @@ type EventEmitter = AsyncEventEmitter<Events>
 
 export class SatelliteClient implements Client {
   private opts: Required<SatelliteClientOpts>
+  private dialect: SatInStartReplicationReq_Dialect
 
   private emitter: EventEmitter
 
@@ -194,6 +196,10 @@ export class SatelliteClient implements Client {
     this.emitter = new AsyncEventEmitter<Events>()
 
     this.opts = { ...satelliteClientDefaults, ...opts }
+    this.dialect =
+      opts.dialect === 'SQLite'
+        ? SatInStartReplicationReq_Dialect.SQLITE
+        : SatInStartReplicationReq_Dialect.POSTGRES
     this.socketFactory = socketFactory
 
     this.inbound = this.resetInboundReplication()
@@ -365,7 +371,10 @@ export class SatelliteClient implements Client {
           )
         )
       }
-      request = SatInStartReplicationReq.fromPartial({ schemaVersion })
+      request = SatInStartReplicationReq.fromPartial({
+        schemaVersion,
+        sqlDialect: this.dialect,
+      })
     } else {
       Log.info(
         `starting replication with lsn: ${base64.fromBytes(
@@ -376,6 +385,7 @@ export class SatelliteClient implements Client {
         lsn,
         subscriptionIds,
         observedTransactionData,
+        sqlDialect: this.dialect,
       })
     }
 
@@ -845,8 +855,7 @@ export class SatelliteClient implements Client {
         'error',
         new SatelliteError(
           SatelliteErrorCode.UNEXPECTED_STATE,
-          `unexpected state ${
-            ReplicationStatus[this.inbound.isReplicating]
+          `unexpected state ${ReplicationStatus[this.inbound.isReplicating]
           } handling 'relation' message`
         )
       )

clients/typescript/src/satellite/config.ts

@@ -64,6 +64,7 @@ export interface SatelliteClientOpts {
   ssl: boolean
   timeout: number
   pushPeriod?: number
+  dialect: 'SQLite' | 'Postgres'
 }
 
 export const validateConfig = (config: any) => {

clients/typescript/src/satellite/process.ts

@@ -1651,13 +1651,7 @@ export function generateTriggersForTable(tbl: MigrationTable): Statement[] {
       }
     }),
     columnTypes: Object.fromEntries(
-      tbl.columns.map((col) => [
-        col.name,
-        {
-          sqliteType: col.sqliteType.toUpperCase(),
-          pgType: col.pgType!.name.toUpperCase(),
-        },
-      ])
+      tbl.columns.map((col) => [col.name, col.pgType!.name.toUpperCase()])
     ),
   }
   const fullTableName = table.namespace + '.' + table.tableName

clients/typescript/src/satellite/registry.ts

@@ -208,6 +208,7 @@ export class GlobalRegistry extends BaseRegistry {
       port: config.replication.port,
       ssl: config.replication.ssl,
       timeout: config.replication.timeout,
+      dialect: config.replication.dialect,
     }
 
     const client = new SatelliteClient(

clients/typescript/test/satellite/client.test.ts

@@ -46,6 +46,7 @@ test.beforeEach((t) => {
     timeout: 10000,
     ssl: false,
     pushPeriod: 100,
+    dialect: 'SQLite',
   })
   const clientId = '91eba0c8-28ba-4a86-a6e8-42731c2c6694'
 

clients/typescript/test/satellite/common.ts

@@ -366,11 +366,11 @@ export const personTable: Table = {
   primary: ['id'],
   foreignKeys: [],
   columnTypes: {
-    id: { sqliteType: 'REAL', pgType: PgBasicType.PG_REAL },
-    name: { sqliteType: 'TEXT', pgType: PgBasicType.PG_TEXT },
-    age: { sqliteType: 'INTEGER', pgType: PgBasicType.PG_INTEGER },
-    bmi: { sqliteType: 'REAL', pgType: PgBasicType.PG_REAL },
-    int8: { sqliteType: 'INTEGER', pgType: PgBasicType.PG_INT8 },
-    blob: { sqliteType: 'BLOB', pgType: PgBasicType.PG_BYTEA },
+    id: PgBasicType.PG_REAL,
+    name: PgBasicType.PG_TEXT,
+    age: PgBasicType.PG_INTEGER,
+    bmi: PgBasicType.PG_REAL,
+    int8: PgBasicType.PG_INT8,
+    blob: PgBasicType.PG_BYTEA,
   },
 }

components/electric/lib/electric/plug/migrations.ex

@@ -46,6 +46,7 @@ defmodule Electric.Plug.Migrations do
   defp get_dialect(%{query_params: %{"dialect" => dialect_name}}) do
     case dialect_name do
       "sqlite" -> {:ok, Electric.Postgres.Dialect.SQLite}
+      "postgresql" -> {:ok, Electric.Postgres.Dialect.Postgresql}
       _ -> {:error, "unsupported dialect #{inspect(dialect_name)}"}
     end
   end

components/electric/lib/electric/postgres/replication.ex

@@ -69,7 +69,7 @@ defmodule Electric.Postgres.Replication do
         {:ok, [], []}
 
       propagate_ast ->
-        {msg, relations} = build_replication_msg(propagate_ast, schema_version, dialect)
+        {msg, relations} = build_replication_msg(propagate_ast, stmt, schema_version, dialect)
 
         {:ok, [msg], relations}
     end
@@ -90,6 +90,9 @@ defmodule Electric.Postgres.Replication do
     end
   end
 
+  defp to_sql(_ast, stmt, Dialect.Postgresql), do: stmt
+  defp to_sql(ast, _stmt, dialect), do: Dialect.to_sql(ast, dialect)
+
   def affected_tables(stmts, dialect \\ @default_dialect) when is_list(stmts) do
     stmts
     |> Enum.flat_map(&get_affected_table/1)
@@ -112,7 +115,7 @@ defmodule Electric.Postgres.Replication do
     []
   end
 
-  defp build_replication_msg(ast, schema_version, dialect) do
+  defp build_replication_msg(ast, stmt, schema_version, dialect) do
     affected_tables = affected_tables(ast, dialect)
 
     relations = Enum.map(affected_tables, &{&1.schema, &1.name})
@@ -133,7 +136,7 @@ defmodule Electric.Postgres.Replication do
         ast,
         &%SatOpMigrate.Stmt{
           type: stmt_type(&1),
-          sql: Dialect.to_sql(&1, dialect)
+          sql: to_sql(&1, stmt, dialect)
         }
       )
 
@@ -168,17 +171,17 @@ defmodule Electric.Postgres.Replication do
   defp replication_msg_table(%Proto.Table{} = table, dialect) do
     %SatOpMigrate.Table{
       name: Dialect.table_name(table.name, dialect),
-      columns: Enum.map(table.columns, &replication_msg_table_col(&1, dialect)),
+      columns: Enum.map(table.columns, &replication_msg_table_col(&1)),
       fks: Enum.flat_map(table.constraints, &replication_msg_table_fk(&1, dialect)),
-      pks: Enum.flat_map(table.constraints, &replication_msg_table_pk(&1, dialect))
+      pks: Enum.flat_map(table.constraints, &replication_msg_table_pk(&1))
     }
   end
 
-  defp replication_msg_table_col(%Proto.Column{} = column, dialect) do
+  defp replication_msg_table_col(%Proto.Column{} = column) do
     %SatOpMigrate.Column{
       name: column.name,
       pg_type: replication_msg_table_col_type(column.type),
-      sqlite_type: Dialect.type_name(column.type, dialect)
+      sqlite_type: Dialect.type_name(column.type, Dialect.SQLite)
     }
   end
 
@@ -190,13 +193,8 @@ defmodule Electric.Postgres.Replication do
     }
   end
 
-  defp replication_msg_table_pk(%Proto.Constraint{constraint: {:primary, pk}}, _dialect) do
-    pk.keys
-  end
-
-  defp replication_msg_table_pk(_constraint, _dialect) do
-    []
-  end
+  defp replication_msg_table_pk(%Proto.Constraint{constraint: {:primary, pk}}), do: pk.keys
+  defp replication_msg_table_pk(_constraint), do: []
 
   defp replication_msg_table_fk(%Proto.Constraint{constraint: {:foreign, fk}}, dialect) do
     [