Use iso8601 timestamps as an opt-in

Author: simolus3

modules/module-mysql/src/api/MySQLRouteAPIAdapter.ts

@@ -102,7 +102,10 @@ export class MySQLRouteAPIAdapter implements api.RouteAPI {
              */
             return fields.map((c) => {
               const value = row[c.name];
-              const sqlValue = sync_rules.toSyncRulesValue(value);
+              const sqlValue = sync_rules.applyValueContext(
+                sync_rules.toSyncRulesValue(value),
+                sync_rules.CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY
+              );
               if (typeof sqlValue == 'bigint') {
                 return Number(value);
               } else if (value instanceof Date) {

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

@@ -103,7 +103,10 @@ export function toColumnDescriptorFromDefinition(column: ColumnDefinition): Colu
   };
 }
 
-export function toSQLiteRow(row: Record<string, any>, columns: Map<string, ColumnDescriptor>): sync_rules.SqliteRow {
+export function toSQLiteRow(
+  row: Record<string, any>,
+  columns: Map<string, ColumnDescriptor>
+): sync_rules.SqliteInputRow {
   let result: sync_rules.DatabaseInputRow = {};
   for (let key in row) {
     // We are very much expecting the column to be there

modules/module-postgres/src/api/PostgresRouteAPIAdapter.ts

@@ -105,7 +105,10 @@ export class PostgresRouteAPIAdapter implements api.RouteAPI {
           columns: result.columns.map((c) => c.name),
           rows: result.rows.map((row) => {
             return row.map((value) => {
-              const sqlValue = sync_rules.toSyncRulesValue(value);
+              const sqlValue = sync_rules.applyValueContext(
+                sync_rules.toSyncRulesValue(value),
+                sync_rules.CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY
+              );
               if (typeof sqlValue == 'bigint') {
                 return Number(value);
               } else if (sync_rules.isJsonValue(sqlValue)) {

modules/module-postgres/src/replication/WalStream.ts

@@ -20,7 +20,15 @@ import {
   storage
 } from '@powersync/service-core';
 import * as pgwire from '@powersync/service-jpgwire';
-import { DatabaseInputRow, SqliteRow, SqlSyncRules, TablePattern, toSyncRulesRow } from '@powersync/service-sync-rules';
+import {
+  applyValueContext,
+  CompatibilityContext,
+  DatabaseInputRow,
+  SqliteInputRow,
+  SqlSyncRules,
+  TablePattern,
+  toSyncRulesRow
+} from '@powersync/service-sync-rules';
 import * as pg_utils from '../utils/pgwire_utils.js';
 
 import { PgManager } from './PgManager.js';
@@ -500,7 +508,7 @@ WHERE  oid = $1::regclass`,
     await sendKeepAlive(db);
   }
 
-  static *getQueryData(results: Iterable<DatabaseInputRow>): Generator<SqliteRow> {
+  static *getQueryData(results: Iterable<DatabaseInputRow>): Generator<SqliteInputRow> {
     for (let row of results) {
       yield toSyncRulesRow(row);
     }
@@ -885,7 +893,8 @@ WHERE  oid = $1::regclass`,
           // The key should always be present in the "after" record.
           return;
         }
-        key[name] = value;
+        // We just need a consistent representation of the primary key, and don't care about fixed quirks.
+        key[name] = applyValueContext(value, CompatibilityContext.FULL_BACKWARDS_COMPATIBILITY);
       }
       resnapshot.push({
         table: record.sourceTable,

modules/module-postgres/src/utils/pgwire_utils.ts

@@ -1,13 +1,13 @@
 // Adapted from https://github.com/kagis/pgwire/blob/0dc927f9f8990a903f238737326e53ba1c8d094f/mod.js#L2218
 
 import * as pgwire from '@powersync/service-jpgwire';
-import { DatabaseInputRow, SqliteRow, toSyncRulesRow } from '@powersync/service-sync-rules';
+import { DatabaseInputRow, SqliteInputRow, toSyncRulesRow } from '@powersync/service-sync-rules';
 
 /**
  * pgwire message -> SQLite row.
  * @param message
  */
-export function constructAfterRecord(message: pgwire.PgoutputInsert | pgwire.PgoutputUpdate): SqliteRow {
+export function constructAfterRecord(message: pgwire.PgoutputInsert | pgwire.PgoutputUpdate): SqliteInputRow {
   const rawData = (message as any).afterRaw;
 
   const record = decodeTuple(message.relation, rawData);
@@ -18,7 +18,9 @@ export function constructAfterRecord(message: pgwire.PgoutputInsert | pgwire.Pgo
  * pgwire message -> SQLite row.
  * @param message
  */
-export function constructBeforeRecord(message: pgwire.PgoutputDelete | pgwire.PgoutputUpdate): SqliteRow | undefined {
+export function constructBeforeRecord(
+  message: pgwire.PgoutputDelete | pgwire.PgoutputUpdate
+): SqliteInputRow | undefined {
   const rawData = (message as any).beforeRaw;
   if (rawData == null) {
     return undefined;

modules/module-postgres/test/src/pg_test.test.ts

@@ -1,6 +1,6 @@
 import { constructAfterRecord } from '@module/utils/pgwire_utils.js';
 import * as pgwire from '@powersync/service-jpgwire';
-import { SqliteRow } from '@powersync/service-sync-rules';
+import { CustomSqliteType, SqliteInputRow, SqliteRow, TimeValue } from '@powersync/service-sync-rules';
 import { describe, expect, test } from 'vitest';
 import { clearTestDb, connectPgPool, connectPgWire, TEST_URI } from './util.js';
 import { WalStream } from '@module/replication/WalStream.js';
@@ -159,8 +159,8 @@ VALUES(10, ARRAY['null']::TEXT[]);
       id: 3n,
       date: '2023-03-06',
       time: '15:47:00',
-      timestamp: '2023-03-06 15:47:00',
-      timestamptz: '2023-03-06 13:47:00Z'
+      timestamp: new TimeValue('2023-03-06 15:47:00', '2023-03-06T15:47:00'),
+      timestamptz: new TimeValue('2023-03-06 13:47:00Z', '2023-03-06T13:47:00Z')
     });
 
     expect(transformed[3]).toMatchObject({
@@ -176,25 +176,25 @@ VALUES(10, ARRAY['null']::TEXT[]);
       id: 5n,
       date: '0000-01-01',
       time: '00:00:00',
-      timestamp: '0000-01-01 00:00:00',
-      timestamptz: '0000-01-01 00:00:00Z'
+      timestamp: new TimeValue('0000-01-01 00:00:00', '0000-01-01T00:00:00'),
+      timestamptz: new TimeValue('0000-01-01 00:00:00Z', '0000-01-01T00:00:00Z')
     });
 
     expect(transformed[5]).toMatchObject({
       id: 6n,
-      timestamp: '1970-01-01 00:00:00',
-      timestamptz: '1970-01-01 00:00:00Z'
+      timestamp: new TimeValue('1970-01-01 00:00:00', '1970-01-01T00:00:00'),
+      timestamptz: new TimeValue('1970-01-01 00:00:00Z', '1970-01-01T00:00:00Z')
     });
 
     expect(transformed[6]).toMatchObject({
       id: 7n,
-      timestamp: '9999-12-31 23:59:59',
-      timestamptz: '9999-12-31 23:59:59Z'
+      timestamp: new TimeValue('9999-12-31 23:59:59', '9999-12-31T23:59:59'),
+      timestamptz: new TimeValue('9999-12-31 23:59:59Z', '9999-12-31T23:59:59Z')
     });
 
     expect(transformed[7]).toMatchObject({
       id: 8n,
-      timestamptz: '0022-02-03 09:13:14Z'
+      timestamptz: new TimeValue('0022-02-03 09:13:14Z', '0022-02-03T09:13:14Z')
     });
 
     expect(transformed[8]).toMatchObject({
@@ -235,8 +235,11 @@ VALUES(10, ARRAY['null']::TEXT[]);
       id: 3n,
       date: `["2023-03-06"]`,
       time: `["15:47:00"]`,
-      timestamp: `["2023-03-06 15:47:00"]`,
-      timestamptz: `["2023-03-06 13:47:00Z","2023-03-06 13:47:00.12345Z"]`
+      timestamp: CustomSqliteType.wrapArray([new TimeValue('2023-03-06 15:47:00', '2023-03-06T15:47:00')]),
+      timestamptz: CustomSqliteType.wrapArray([
+        new TimeValue('2023-03-06 13:47:00Z', '2023-03-06T13:47:00Z'),
+        new TimeValue('2023-03-06 13:47:00.12345Z', '2023-03-06T13:47:00.12345Z')
+      ])
     });
 
     expect(transformed[3]).toMatchObject({
@@ -436,7 +439,7 @@ VALUES(10, ARRAY['null']::TEXT[]);
  * Return all the inserts from the first transaction in the replication stream.
  */
 async function getReplicationTx(replicationStream: pgwire.ReplicationStream) {
-  let transformed: SqliteRow[] = [];
+  let transformed: SqliteInputRow[] = [];
   for await (const batch of replicationStream.pgoutputDecode()) {
     for (const msg of batch.messages) {
       if (msg.tag == 'insert') {

packages/jpgwire/package.json

@@ -19,6 +19,7 @@
   },
   "dependencies": {
     "@powersync/service-jsonbig": "workspace:^",
+    "@powersync/service-sync-rules": "workspace:^",
     "date-fns": "^4.1.0",
     "pgwire": "github:kagis/pgwire#f1cb95f9a0f42a612bb5a6b67bb2eb793fc5fc87"
   }

packages/jpgwire/src/pgwire_types.ts

@@ -1,6 +1,7 @@
 // Adapted from https://github.com/kagis/pgwire/blob/0dc927f9f8990a903f238737326e53ba1c8d094f/mod.js#L2218
 
 import { JsonContainer } from '@powersync/service-jsonbig';
+import { type DatabaseInputValue } from '@powersync/service-sync-rules';
 import { dateToSqlite, lsnMakeComparable, timestampToSqlite, timestamptzToSqlite } from './util.js';
 
 export enum PgTypeOid {
@@ -105,7 +106,7 @@ export class PgType {
     return ELEM_OID_TO_ARRAY.get(typeOid);
   }
 
-  static decode(text: string, typeOid: number) {
+  static decode(text: string, typeOid: number): DatabaseInputValue {
     switch (typeOid) {
       // add line here when register new type
       case PgTypeOid.TEXT:
@@ -178,7 +179,14 @@ export class PgType {
           const decoded = this.decode(unescaped, elemTypeOid);
           stack[0].push(decoded);
         }
-        result = ch == 0x7d /*}*/ && stack.shift();
+
+        if (ch == 0x7d /*}*/) {
+          const entry = stack.shift();
+          result = entry;
+        } else {
+          result = false;
+        }
+
         elStart = i + 1; // TODO dry
       }
     }

packages/jpgwire/src/util.ts

@@ -5,6 +5,7 @@ import { DEFAULT_CERTS } from './certs.js';
 import * as pgwire from './pgwire.js';
 import { PgType } from './pgwire_types.js';
 import { ConnectOptions } from './socket_adapter.js';
+import { DatabaseInputValue, TimeValue } from '@powersync/service-sync-rules';
 
 // TODO this is duplicated, but maybe that is ok
 export interface NormalizedConnectionConfig {
@@ -132,7 +133,7 @@ export async function connectPgWire(
   return connection;
 }
 
-function _recvDataRow(this: any, _message: any, row: Uint8Array[], batch: any) {
+function _recvDataRow(this: any, _message: any, row: (Uint8Array | DatabaseInputValue)[], batch: any) {
   for (let i = 0; i < this._rowColumns.length; i++) {
     const valbuf = row[i];
     if (valbuf == null) {
@@ -232,17 +233,17 @@ export function lsnMakeComparable(text: string) {
  *
  * We have specific exceptions for -infinity and infinity.
  */
-export function timestamptzToSqlite(source?: string) {
+export function timestamptzToSqlite(source?: string): TimeValue | null {
   if (source == null) {
     return null;
   }
   // Make compatible with SQLite
   const match = /^([\d\-]+) ([\d:]+)(\.\d+)?([+-][\d:]+)$/.exec(source);
   if (match == null) {
     if (source == 'infinity') {
-      return '9999-12-31 23:59:59Z';
+      return new TimeValue('9999-12-31 23:59:59Z', '9999-12-31T23:59:59Z');
     } else if (source == '-infinity') {
-      return '0000-01-01 00:00:00Z';
+      return new TimeValue('0000-01-01 00:00:00Z', '0000-01-01T00:00:00Z');
     } else {
       return null;
     }
@@ -256,9 +257,11 @@ export function timestamptzToSqlite(source?: string) {
   if (isNaN(parsed.getTime())) {
     return null;
   }
-  const text = parsed.toISOString().replace('T', ' ').replace('.000', '').replace('Z', '');
 
-  return `${text}${precision ?? ''}Z`;
+  const baseValue = parsed.toISOString().replace('.000', '').replace('Z', '');
+  const baseText = `${baseValue}${precision ?? ''}Z`;
+
+  return new TimeValue(baseText.replace('T', ' '), baseText);
 }
 
 /**
@@ -268,16 +271,16 @@ export function timestamptzToSqlite(source?: string) {
  *
  * https://www.postgresql.org/docs/current/datatype-datetime.html#DATATYPE-DATETIME-SPECIAL-VALUES
  */
-export function timestampToSqlite(source?: string) {
+export function timestampToSqlite(source?: string): TimeValue | null {
   if (source == null) {
     return null;
   }
   if (source == 'infinity') {
-    return '9999-12-31 23:59:59';
+    return new TimeValue('9999-12-31 23:59:59', '9999-12-31T23:59:59');
   } else if (source == '-infinity') {
-    return '0000-01-01 00:00:00';
+    return new TimeValue('0000-01-01 00:00:00', '0000-01-01T00:00:00');
   } else {
-    return source;
+    return new TimeValue(source, source.replace(' ', 'T'));
   }
 }
 /**

packages/service-core/src/storage/BucketStorageBatch.ts

@@ -1,5 +1,11 @@
 import { ObserverClient } from '@powersync/lib-services-framework';
-import { EvaluatedParameters, EvaluatedRow, SqliteRow, ToastableSqliteRow } from '@powersync/service-sync-rules';
+import {
+  EvaluatedParameters,
+  EvaluatedRow,
+  SqliteInputRow,
+  SqliteRow,
+  ToastableSqliteRow
+} from '@powersync/service-sync-rules';
 import { BSON } from 'bson';
 import { ReplicationEventPayload } from './ReplicationEventPayload.js';
 import { SourceTable, TableSnapshotStatus } from './SourceTable.js';
@@ -132,7 +138,7 @@ export interface SaveInsert {
   sourceTable: SourceTable;
   before?: undefined;
   beforeReplicaId?: undefined;
-  after: SqliteRow;
+  after: SqliteInputRow;
   afterReplicaId: ReplicaId;
 }
 
@@ -143,7 +149,7 @@ export interface SaveUpdate {
   /**
    * This is only present when the id has changed, and will only contain replica identity columns.
    */
-  before?: SqliteRow;
+  before?: SqliteInputRow;
   beforeReplicaId?: ReplicaId;
 
   /**
@@ -158,7 +164,7 @@ export interface SaveUpdate {
 export interface SaveDelete {
   tag: SaveOperationTag.DELETE;
   sourceTable: SourceTable;
-  before?: SqliteRow;
+  before?: SqliteInputRow;
   beforeReplicaId: ReplicaId;
   after?: undefined;
   afterReplicaId?: undefined;