Cleanup

Author: simolus3

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

@@ -9,11 +9,11 @@ import { getDebugTableInfo } from '../replication/replication-utils.js';
 import { KEEPALIVE_STATEMENT, PUBLICATION_NAME } from '../replication/WalStream.js';
 import * as types from '../types/types.js';
 import { getApplicationName } from '../utils/application-name.js';
-import { PostgresTypeCache } from '../types/cache.js';
-import { CustomTypeRegistry, isKnownType } from '../types/registry.js';
+import { CustomTypeRegistry } from '../types/registry.js';
+import { PostgresTypeResolver } from '../types/resolver.js';
 
 export class PostgresRouteAPIAdapter implements api.RouteAPI {
-  private typeCache: PostgresTypeCache;
+  private typeCache: PostgresTypeResolver;
   connectionTag: string;
   // TODO this should probably be configurable one day
   publicationName = PUBLICATION_NAME;
@@ -34,7 +34,7 @@ export class PostgresRouteAPIAdapter implements api.RouteAPI {
     connectionTag?: string,
     private config?: types.ResolvedConnectionConfig
   ) {
-    this.typeCache = new PostgresTypeCache(config?.typeRegistry ?? new CustomTypeRegistry(), pool);
+    this.typeCache = new PostgresTypeResolver(config?.typeRegistry ?? new CustomTypeRegistry(), pool);
     this.connectionTag = connectionTag ?? sync_rules.DEFAULT_TAG;
   }
 

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

@@ -2,7 +2,7 @@ import * as pgwire from '@powersync/service-jpgwire';
 import semver from 'semver';
 import { NormalizedPostgresConnectionConfig } from '../types/types.js';
 import { getApplicationName } from '../utils/application-name.js';
-import { PostgresTypeCache } from '../types/cache.js';
+import { PostgresTypeResolver } from '../types/resolver.js';
 import { getServerVersion } from '../utils/postgres_version.js';
 import { CustomTypeRegistry } from '../types/registry.js';
 
@@ -21,7 +21,7 @@ export class PgManager {
    */
   public readonly pool: pgwire.PgClient;
 
-  public readonly types: PostgresTypeCache;
+  public readonly types: PostgresTypeResolver;
 
   private connectionPromises: Promise<pgwire.PgConnection>[] = [];
 
@@ -31,7 +31,7 @@ export class PgManager {
   ) {
     // The pool is lazy - no connections are opened until a query is performed.
     this.pool = pgwire.connectPgWirePool(this.options, poolOptions);
-    this.types = new PostgresTypeCache(poolOptions.registry, this.pool);
+    this.types = new PostgresTypeResolver(poolOptions.registry, this.pool);
   }
 
   public get connectionTag() {

modules/module-postgres/src/types/registry.ts

@@ -276,7 +276,3 @@ export class CustomTypeRegistry {
     }
   }
 }
-
-export function isKnownType(type: MaybeKnownType): type is KnownType {
-  return type.type != 'unknown';
-}

modules/module-postgres/src/types/cache.ts renamed to modules/module-postgres/src/types/resolver.ts

@@ -5,9 +5,11 @@ import semver from 'semver';
 import { getServerVersion } from '../utils/postgres_version.js';
 
 /**
- * A cache of custom types for which information can be crawled from the source database.
+ * Resolves descriptions used to decode values for custom postgres types.
+ *
+ * Custom types are resolved from the source database, which also involves crawling inner types (e.g. for composites).
  */
-export class PostgresTypeCache {
+export class PostgresTypeResolver {
   private cachedVersion: semver.SemVer | null = null;
 
   constructor(
@@ -31,7 +33,7 @@ export class PostgresTypeCache {
    */
   async supportsMultiRanges() {
     const version = await this.fetchVersion();
-    return version.compare(PostgresTypeCache.minVersionForMultirange) >= 0;
+    return version.compare(PostgresTypeResolver.minVersionForMultirange) >= 0;
   }
 
   /**
@@ -138,7 +140,7 @@ WHERE t.oid = ANY($1)
   }
 
   /**
-   * Used for testing - fetches all custom types referenced by any column in the database.
+   * Crawls all custom types referenced by table columns in the current database.
    */
   public async fetchTypesForSchema() {
     const sql = `

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

@@ -10,7 +10,7 @@ import {
 import { describe, expect, test } from 'vitest';
 import { clearTestDb, connectPgPool, connectPgWire, TEST_URI } from './util.js';
 import { WalStream } from '@module/replication/WalStream.js';
-import { PostgresTypeCache } from '@module/types/cache.js';
+import { PostgresTypeResolver } from '@module/types/resolver.js';
 import { CustomTypeRegistry } from '@module/types/registry.js';
 
 describe('pg data types', () => {
@@ -551,7 +551,7 @@ INSERT INTO test_data(id, time, timestamp, timestamptz) VALUES (1, '17:42:01.12'
   test('test replication - multiranges', async () => {
     const db = await connectPgPool();
 
-    if (!(await new PostgresTypeCache(new CustomTypeRegistry(), db).supportsMultiRanges())) {
+    if (!(await new PostgresTypeResolver(new CustomTypeRegistry(), db).supportsMultiRanges())) {
       // This test requires Postgres 14 or later.
       return;
     }
@@ -620,7 +620,7 @@ INSERT INTO test_data(id, time, timestamp, timestamptz) VALUES (1, '17:42:01.12'
  * Return all the inserts from the first transaction in the replication stream.
  */
 async function getReplicationTx(db: pgwire.PgClient, replicationStream: pgwire.ReplicationStream) {
-  const typeCache = new PostgresTypeCache(new CustomTypeRegistry(), db);
+  const typeCache = new PostgresTypeResolver(new CustomTypeRegistry(), db);
   await typeCache.fetchTypesForSchema();
 
   let transformed: SqliteInputRow[] = [];

packages/jpgwire/src/pgwire_types.ts

@@ -156,7 +156,7 @@ export class PgType {
 
   static _decodeArray(text: string, elemTypeOid: number): DatabaseInputValue[] {
     text = text.replace(/^\[.+=/, ''); // skip dimensions
-    return new StructureParser(text).parseArray((raw) => (raw == null ? null : PgType.decode(raw, elemTypeOid)));
+    return new StructureParser(text).parseArray((raw) => PgType.decode(raw, elemTypeOid));
   }
 
   static _decodeBytea(text: string): Uint8Array {