fix(test): add a test for the sync context, clean up some bugs with it

__tests__/index.spec.ts

@@ -48,7 +48,7 @@ describe('move tables from postgres to clickhouse', () => {
       },
       insert(stream) {
         return ch.insert({
-          table: 'identity__address_types',
+          table: 'address_types',
           values: stream,
           format: 'JSONEachRow',
         });
@@ -75,7 +75,7 @@ describe('move tables from postgres to clickhouse', () => {
         };
       },
       clickhouse: ch,
-      tableName: 'identity__address_types',
+      tableName: 'address_types',
     }, {});
     expect(detail.rows).toBe(2);
   });

__tests__/kysely.spec.ts

@@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, test } from 'vitest';
 import { ClickHouseClient } from '@clickhouse/client';
 import { Kysely } from 'kysely';
 
-import { copyTable, syncTable } from '../src/dbs/kysely';
+import { KyselySyncContext, copyTable, syncTable } from '../src/dbs/kysely';
 
 import { createChDb, createPgDb } from './db.fixtures';
 import { kyselyDb } from './kysely.fixtures';
@@ -30,14 +30,14 @@ describe('simple kysely interface', () => {
   test('should sync to clickhouse', async () => {
     const detail = await copyTable(db, ch, {}, {
       from: 'address_types',
-      to: 'identity__address_types',
+      to: 'address_types',
       pk: 'address_type_id',
     });
     expect(detail.rows).toBe(2);
 
     const indSpec = {
       from: 'individuals',
-      to: 'identity__individuals',
+      to: 'individuals',
       pk: 'individual_id',
       delaySeconds: 0,
     } as const;
@@ -62,4 +62,55 @@ describe('simple kysely interface', () => {
     const upd2 = await syncTable(db, ch, upd.bookmark, indSpec);
     expect(upd2.rows, 'Copy 2 rows after second update').toBe(2);
   });
+
+  test('should sync to clickhouse with simplified syntax', async () => {
+    await db.insertInto('individuals')
+      .columns(['updated_at'])
+      .values([{
+        updated_at: new Date('2024-01-01T00:00:00.000Z'),
+      }])
+      .execute();
+    let context = new KyselySyncContext(db, ch, {});
+    await Promise.all([
+      context.forwardOnly('address_types'),
+      context.withUpdatedAt('individuals'),
+    ]);
+    expect(context.results).toMatchInlineSnapshot(`
+      {
+        "address_types": {
+          "lastCount": 2,
+          "rowId": 2,
+        },
+        "individuals": {
+          "lastCount": 4,
+          "rowId": "4",
+          "rowTimestamp": "2024-01-01T00:00:00.000",
+        },
+      }
+    `);
+    context = new KyselySyncContext(db, ch, context.results);
+    const secondPass = await Promise.all([
+      context.forwardOnly('address_types'),
+      context.withUpdatedAt('individuals'),
+    ]);
+    expect(secondPass).toMatchInlineSnapshot(`
+      [
+        {
+          "bookmark": {
+            "lastCount": 0,
+            "rowId": 2,
+          },
+          "table": "address_types",
+        },
+        {
+          "bookmark": {
+            "lastCount": 1,
+            "rowId": "4",
+            "rowTimestamp": "2024-01-01T00:00:00.000",
+          },
+          "table": "individuals",
+        },
+      ]
+    `);
+  });
 });

__tests__/migrations/1_test.sql

@@ -1,11 +1,11 @@
-CREATE TABLE identity__address_types (
+CREATE TABLE address_types (
     address_type_id Int32,
     name String,
     created_at DateTime64
 ) ENGINE = ReplacingMergeTree(created_at)
 ORDER BY address_type_id;
 
-CREATE TABLE identity__individuals (
+CREATE TABLE individuals (
   individual_id Int32,
   individual_uuid String,
   favorite_color String,

src/dbs/kysely.ts

@@ -1,7 +1,7 @@
 import { type Kysely, type AnyColumn, sql } from 'kysely';
 import type { ClickHouseClient } from '@clickhouse/client';
 
-import type { Bookmark, ClickhouseRowRecord, RowMapper, SourceDatabaseRowRecord, SyncResult } from '../types';
+import type { Bookmark, ClickhouseRowRecord, RowMapper, SourceDatabaseRowRecord } from '../types';
 import { synchronizeTable } from '../stream-copy';
 
 type HasUpdatedAt<ColName extends string> = {
@@ -112,7 +112,7 @@ export async function copyTable<
     from: T;
     to: string;
     pk: PK;
-    optimize: boolean;
+    optimize?: boolean;
     rowMapper?: RowMapper;
   },
 ) {
@@ -159,8 +159,6 @@ export async function copyTable<
 export class KyselySyncContext<Schema, B extends Partial<Record<keyof Schema, Bookmark<string> | Bookmark<number>>>> {
   // Log function that can be overridden
   log: (level: 'info' | 'error', message: string, meta?: Record<string, unknown>) => void = () => { };
-  // The default name of the created_at column
-  createdAtColumn = 'created_at';
   // The default name of the updated_at column
   updatedAtColumn = 'updated_at';
   // The default name of the primary key column based on the table name (the default strips a plural 's' from the end and adds _id)
@@ -177,26 +175,27 @@ export class KyselySyncContext<Schema, B extends Partial<Record<keyof Schema, Bo
     private readonly bookmark?: B,
   ) { }
 
-  async table<T extends keyof Schema & string>(table: T, opts: {
+  /**
+   * Sync a table that only gets additions, no updates (or update tracking)
+   */
+  async forwardOnly<T extends keyof Schema & string>(table: T, opts?: {
     pk?: AnyColumn<Schema, T>,
-    timestampColumn?: AnyColumn<Schema, T>,
     rowMapper?: RowMapper,
-  }): Promise<SyncResult<T, string | number>> {
-    const { pk, timestampColumn, rowMapper } = opts;
-    return syncTable(
+  }) {
+    const { pk } = opts || {};
+    return copyTable(
       this.db,
       this.clickhouse,
       (this.bookmark?.[table as keyof B] || {}) as Bookmark<string | number>,
       {
+        ...opts,
         from: table,
         to: table,
         pk: (pk || this.getDefaultPrimaryKeyColumn(table)) as AnyColumn<Schema, T>,
         optimize: true,
-        timestampColumn,
-        rowMapper,
       })
       .then((result) => {
-        this.log('info', 'Sync complete', { table, rows: result.rows });
+        this.log('info', 'Copy complete', { table, rows: result.rows });
         const newBookmark = { ...result.bookmark, lastCount: result.rows };
         this.results[table as string] = newBookmark;
         return {
@@ -210,29 +209,39 @@ export class KyselySyncContext<Schema, B extends Partial<Record<keyof Schema, Bo
       });
   }
 
-  /**
-   * Sync a table that only gets additions, no updates (or update tracking)
-   */
-  async forwardOnly<T extends keyof Schema & string>(table: T, opts?: {
-    pk?: AnyColumn<Schema, T>,
-    rowMapper?: RowMapper,
-  }) {
-    return this.table(table, {
-      timestampColumn: this.createdAtColumn as AnyColumn<Schema, T>,
-      ...opts,
-    });
-  }
-
   /**
    * Sync a table that tracks its updates with an updated_at column
    */
   async withUpdatedAt<T extends keyof Schema & string>(table: T, opts?: {
     pk?: AnyColumn<Schema, T>,
+    timestampColumn?: AnyColumn<Schema, T>,
     rowMapper?: RowMapper,
   }) {
-    return this.table(table, {
-      timestampColumn: this.updatedAtColumn as AnyColumn<Schema, T>,
-      ...opts,
-    });
+    const { pk, timestampColumn } = opts || {};
+    return syncTable(
+      this.db,
+      this.clickhouse,
+      (this.bookmark?.[table as keyof B] || {}) as Bookmark<string | number>,
+      {
+        ...opts,
+        from: table,
+        to: table,
+        pk: (pk || this.getDefaultPrimaryKeyColumn(table)) as AnyColumn<Schema, T>,
+        optimize: true,
+        timestampColumn: timestampColumn || this.updatedAtColumn as AnyColumn<Schema, T>,
+      })
+      .then((result) => {
+        this.log('info', 'Sync complete', { table, rows: result.rows });
+        const newBookmark = { ...result.bookmark, lastCount: result.rows };
+        this.results[table as string] = newBookmark;
+        return {
+          table,
+          bookmark: newBookmark,
+        };
+      })
+      .catch((error) => {
+        this.log('error', 'Failed to sync table', { table, error });
+        throw error;
+      });
   }
 }

src/types.ts

@@ -12,11 +12,6 @@ export interface Bookmark<PK extends string | number> {
   lastCount?: number;
 }
 
-export interface SyncResult<T extends string, PK extends string | number> {
-  table: T;
-  bookmark: Bookmark<PK>;
-}
-
 export type RowFetchFunction<T, PK extends string | number> = (bookmark: Bookmark<PK>, limit: number) => AsyncIterableIterator<T>;
 
 export type RowMapper = (row: SourceDatabaseRowRecord) => ClickhouseRowRecord;