Move cache up

Author: simolus3

modules/module-postgres/src/module/PostgresModule.ts

@@ -19,8 +19,11 @@ import { WalStreamReplicator } from '../replication/WalStreamReplicator.js';
 import * as types from '../types/types.js';
 import { PostgresConnectionConfig } from '../types/types.js';
 import { getApplicationName } from '../utils/application-name.js';
+import { CustomTypeRegistry } from '../types/registry.js';
 
 export class PostgresModule extends replication.ReplicationModule<types.PostgresConnectionConfig> {
+  private customTypes: CustomTypeRegistry = new CustomTypeRegistry();
+
   constructor() {
     super({
       name: 'Postgres',
@@ -48,7 +51,7 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
   protected createReplicator(context: system.ServiceContext): replication.AbstractReplicator {
     const normalisedConfig = this.resolveConfig(this.decodedConfig!);
     const syncRuleProvider = new ConfigurationFileSyncRulesProvider(context.configuration.sync_rules);
-    const connectionFactory = new ConnectionManagerFactory(normalisedConfig);
+    const connectionFactory = new ConnectionManagerFactory(normalisedConfig, this.customTypes);
 
     return new WalStreamReplicator({
       id: this.getDefaultId(normalisedConfig.database),
@@ -75,7 +78,8 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
     const connectionManager = new PgManager(normalisedConfig, {
       idleTimeout: 30_000,
       maxSize: 1,
-      applicationName: getApplicationName()
+      applicationName: getApplicationName(),
+      registry: this.customTypes
     });
 
     try {
@@ -106,7 +110,8 @@ export class PostgresModule extends replication.ReplicationModule<types.Postgres
     const connectionManager = new PgManager(normalizedConfig, {
       idleTimeout: 30_000,
       maxSize: 1,
-      applicationName: getApplicationName()
+      applicationName: getApplicationName(),
+      registry: new CustomTypeRegistry()
     });
     const connection = await connectionManager.snapshotConnection();
     try {

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

@@ -2,18 +2,22 @@ import { PgManager } from './PgManager.js';
 import { NormalizedPostgresConnectionConfig } from '../types/types.js';
 import { PgPoolOptions } from '@powersync/service-jpgwire';
 import { logger } from '@powersync/lib-services-framework';
+import { CustomTypeRegistry } from '../types/registry.js';
 
 export class ConnectionManagerFactory {
   private readonly connectionManagers: PgManager[];
   public readonly dbConnectionConfig: NormalizedPostgresConnectionConfig;
 
-  constructor(dbConnectionConfig: NormalizedPostgresConnectionConfig) {
+  constructor(
+    dbConnectionConfig: NormalizedPostgresConnectionConfig,
+    private readonly registry: CustomTypeRegistry
+  ) {
     this.dbConnectionConfig = dbConnectionConfig;
     this.connectionManagers = [];
   }
 
   create(poolOptions: PgPoolOptions) {
-    const manager = new PgManager(this.dbConnectionConfig, poolOptions);
+    const manager = new PgManager(this.dbConnectionConfig, { ...poolOptions, registry: this.registry });
     this.connectionManagers.push(manager);
     return manager;
   }

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

@@ -4,6 +4,11 @@ import { NormalizedPostgresConnectionConfig } from '../types/types.js';
 import { getApplicationName } from '../utils/application-name.js';
 import { PostgresTypeCache } from '../types/cache.js';
 import { getServerVersion } from '../utils/postgres_version.js';
+import { CustomTypeRegistry } from '../types/registry.js';
+
+export interface PgManagerOptions extends pgwire.PgPoolOptions {
+  registry: CustomTypeRegistry;
+}
 
 /**
  * Shorter timeout for snapshot connections than for replication connections.
@@ -22,11 +27,11 @@ export class PgManager {
 
   constructor(
     public options: NormalizedPostgresConnectionConfig,
-    public poolOptions: pgwire.PgPoolOptions
+    public poolOptions: PgManagerOptions
   ) {
     // The pool is lazy - no connections are opened until a query is performed.
     this.pool = pgwire.connectPgWirePool(this.options, poolOptions);
-    this.types = new PostgresTypeCache(this.pool, () => this.getServerVersion());
+    this.types = new PostgresTypeCache(poolOptions.registry, this.pool, () => this.getServerVersion());
   }
 
   public get connectionTag() {

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

@@ -7,10 +7,10 @@ import semver from 'semver';
  * A cache of custom types for which information can be crawled from the source database.
  */
 export class PostgresTypeCache {
-  readonly registry: CustomTypeRegistry;
   private cachedVersion: semver.SemVer | null = null;
 
   constructor(
+    readonly registry: CustomTypeRegistry,
     private readonly pool: pgwire.PgClient,
     private readonly getVersion: () => Promise<semver.SemVer | null>
   ) {