Made it possible to specify timezone on zongji listener configuration to stop unwanted timezone skew

Added serverId configuration
Added check for binlog existence before starting replication.

Author: Rentacookie

modules/module-mysql/package.json

@@ -32,7 +32,8 @@
     "@powersync/service-core": "workspace:*",
     "@powersync/service-sync-rules": "workspace:*",
     "@powersync/service-types": "workspace:*",
-    "@powersync/mysql-zongji": "0.0.0-dev-20241021144804",
+    "@powersync/mysql-zongji": "0.0.0-dev-20241022103518",
+    "date-fns": "^4.1.0",
     "semver": "^7.5.4",
     "async": "^3.2.4",
     "mysql2": "^3.11.0",

modules/module-mysql/src/common/mysql-to-sqlite.ts

@@ -1,10 +1,21 @@
+import { logger } from '@powersync/lib-services-framework';
 import * as sync_rules from '@powersync/service-sync-rules';
 import { ExpressionType } from '@powersync/service-sync-rules';
+import { ColumnDescriptor } from '@powersync/service-core';
+import { formatISO } from 'date-fns';
+import mysql from 'mysql2';
 
-export function toSQLiteRow(row: Record<string, any>): sync_rules.SqliteRow {
+export function toSQLiteRow(row: Record<string, any>, columns?: Map<string, ColumnDescriptor>): sync_rules.SqliteRow {
   for (let key in row) {
     if (row[key] instanceof Date) {
-      row[key] = row[key].toISOString();
+      logger.info(`Date before conversion: ${key}=${row[key].toISOString()}`);
+      const column = columns?.get(key);
+      if (column?.typeId == mysql.Types.DATE) {
+        row[key] = formatISO(row[key], { representation: 'date' });
+      } else {
+        row[key] = row[key].toISOString();
+      }
+      logger.info(`Converted date to string: ${key}=${row[key]}`);
     }
   }
   return sync_rules.toSyncRulesRow(row);

modules/module-mysql/src/common/read-executed-gtid.ts

@@ -41,3 +41,15 @@ export async function readExecutedGtid(connection: mysqlPromise.Connection): Pro
     raw_gtid: binlogStatus.Executed_Gtid_Set
   });
 }
+
+export async function isBinlogStillAvailable(
+  connection: mysqlPromise.Connection,
+  binlogFile: string
+): Promise<boolean> {
+  const [logFiles] = await mysql_utils.retriedQuery({
+    connection,
+    query: `SHOW BINARY LOGS;`
+  });
+
+  return logFiles.some((f) => f['Log_name'] == binlogFile);
+}

modules/module-mysql/src/replication/BinLogStream.ts

@@ -2,15 +2,16 @@ import { logger } from '@powersync/lib-services-framework';
 import * as sync_rules from '@powersync/service-sync-rules';
 import async from 'async';
 
-import { framework, getUuidReplicaIdentityBson, Metrics, storage } from '@powersync/service-core';
-import mysql from 'mysql2';
+import { ColumnDescriptor, framework, getUuidReplicaIdentityBson, Metrics, storage } from '@powersync/service-core';
+import mysql, { FieldPacket } from 'mysql2';
 
 import { BinLogEvent } from '@powersync/mysql-zongji';
 import * as common from '../common/common-index.js';
 import * as zongji_utils from './zongji/zongji-utils.js';
 import { MySQLConnectionManager } from './MySQLConnectionManager.js';
-import { ReplicatedGTID } from '../common/common-index.js';
+import { isBinlogStillAvailable, ReplicatedGTID } from '../common/common-index.js';
 import mysqlPromise from 'mysql2/promise';
+import { MySQLTypesMap } from '../utils/mysql_utils.js';
 
 export interface BinLogStreamOptions {
   connections: MySQLConnectionManager;
@@ -211,10 +212,23 @@ AND table_type = 'BASE TABLE';`,
    */
   protected async checkInitialReplicated(): Promise<boolean> {
     const status = await this.storage.getStatus();
+    const lastKnowGTID = status.checkpoint_lsn ? common.ReplicatedGTID.fromSerialized(status.checkpoint_lsn) : null;
     if (status.snapshot_done && status.checkpoint_lsn) {
-      logger.info(`Initial replication already done. MySQL appears healthy`);
+      logger.info(`Initial replication already done.`);
+
+      if (lastKnowGTID) {
+        // Check if the binlog is still available. If it isn't we need to snapshot again.
+        const connection = await this.connections.getConnection();
+        try {
+          return await isBinlogStillAvailable(connection, lastKnowGTID.position.filename);
+        } finally {
+          connection.release();
+        }
+      }
+
       return true;
     }
+
     return false;
   }
 
@@ -270,17 +284,24 @@ AND table_type = 'BASE TABLE';`,
     logger.info(`Replicating ${table.qualifiedName}`);
     // TODO count rows and log progress at certain batch sizes
 
+    const columns = new Map<string, ColumnDescriptor>();
     return new Promise<void>((resolve, reject) => {
       // MAX_EXECUTION_TIME(0) hint disables execution timeout for this query
       connection
         .query(`SELECT /*+ MAX_EXECUTION_TIME(0) */ * FROM ${table.schema}.${table.table}`)
-        .stream()
         .on('error', (err) => {
           reject(err);
         })
-        .on('data', async (row) => {
+        .on('fields', (fields: FieldPacket[]) => {
+          // Map the columns and their types
+          fields.forEach((field) => {
+            const columnType = MySQLTypesMap[field.type as number];
+            columns.set(field.name, { name: field.name, type: columnType, typeId: field.type });
+          });
+        })
+        .on('result', async (row) => {
           connection.pause();
-          const record = common.toSQLiteRow(row);
+          const record = common.toSQLiteRow(row, columns);
 
           await batch.save({
             tag: storage.SaveOperationTag.INSERT,

modules/module-mysql/src/replication/zongji/zongji.d.ts

@@ -4,6 +4,7 @@ declare module '@powersync/mysql-zongji' {
     user: string;
     password: string;
     dateStrings?: boolean;
+    timeZone?: string;
   };
 
   interface DatabaseFilter {
@@ -31,6 +32,11 @@ declare module '@powersync/mysql-zongji' {
      * BinLog position offset to start reading events from in file specified
      */
     position?: number;
+
+    /**
+     *  Unique server ID for this replication client.
+     */
+    serverId?: number;
   };
 
   export type ColumnSchema = {

modules/module-mysql/src/types/types.ts

@@ -14,6 +14,7 @@ export interface NormalizedMySQLConnectionConfig {
 
   username: string;
   password: string;
+  server_id: number;
 
   cacert?: string;
   client_certificate?: string;
@@ -29,6 +30,7 @@ export const MySQLConnectionConfig = service_types.configFile.DataSourceConfig.a
     username: t.string.optional(),
     password: t.string.optional(),
     database: t.string.optional(),
+    server_id: t.number.optional(),
 
     cacert: t.string.optional(),
     client_certificate: t.string.optional(),
@@ -97,6 +99,8 @@ export function normalizeConnectionConfig(options: MySQLConnectionConfig): Norma
     database,
 
     username,
-    password
+    password,
+
+    server_id: options.server_id ?? 1
   };
 }

modules/module-mysql/src/utils/mysql_utils.ts

@@ -1,8 +1,13 @@
 import { logger } from '@powersync/lib-services-framework';
-import mysql from 'mysql2';
+import mysql, { Types } from 'mysql2';
 import mysqlPromise from 'mysql2/promise';
 import * as types from '../types/types.js';
 
+export const MySQLTypesMap: { [key: number]: string } = {};
+for (const [name, code] of Object.entries(Types)) {
+  MySQLTypesMap[code as number] = name;
+}
+
 export type RetriedQueryOptions = {
   connection: mysqlPromise.Connection;
   query: string;
@@ -42,6 +47,9 @@ export function createPool(config: types.NormalizedMySQLConnectionConfig, option
     password: config.password,
     database: config.database,
     ssl: hasSSLOptions ? sslOptions : undefined,
+    supportBigNumbers: true,
+    // dateStrings: true,
+    timezone: 'Z', // Ensure no auto timezone manipulation of the dates occur
     ...(options || {})
   });
 }