feat(lds): make wal2json parameters configurable

* add `LdsOptions.params`
* fix `numeric-data-types-as-string` to be set only if `types` are passed
* adjust the `parse` function accordingly, to handle `number` values without a type parser

Author: ab-pm

packages/lds/src/pg-logical-decoding.ts

@@ -83,45 +83,58 @@ export interface LdsOptions {
   temporary?: boolean;
   /** (Custom) [type parsers](https://node-postgres.com/features/queries#types) to deserialise the wal2json column string values. Pass `pg.types` to get the default type parsing. Defaults to `undefined`, that is raw values will get emitted. */
   types?: pg.CustomTypesConfig;
+  /** Extra [parameters to be passed to wal2json](https://github.com/eulerto/wal2json?tab=readme-ov-file#parameters). Use e.g. `{'numeric-data-types-as-string', 't'}` to make the type parsers apply to numeric values. */
+  params?: Record<string, string>;
 }
 
 export default class PgLogicalDecoding extends EventEmitter {
   public readonly slotName: string;
   public readonly temporary: boolean;
-  private connectionString: string;
-  private tablePattern: string;
+  private readonly getChangesQueryText: string;
+  private readonly parse: (value: any, typeOid: number) => any;
   private pool: pg.Pool | null;
   private client: Promise<pg.PoolClient> | null;
-  private readonly parse: (value: any, typeOid: number) => any;
 
   constructor(connectionString: string, options?: LdsOptions) {
     super();
-    this.connectionString = connectionString;
     const {
       tablePattern = "*.*",
       slotName = "postgraphile",
       temporary = false,
       types,
+      params,
     } = options || {};
-    this.tablePattern = tablePattern;
     this.slotName = slotName;
     this.temporary = temporary;
+    const parametersSql = Object.entries({
+      "add-tables": tablePattern != "*.*" ? tablePattern : null,
+      "include-types": "f", // type names are unnecessary
+      "include-type-oids": types ? "t" : null,
+      "numeric-data-types-as-string": types ? "t" : null,
+      ...params,
+    })
+      .flatMap(entry => (typeof entry[1] == "string" ? entry : []))
+      .map(pg.Client.prototype.escapeLiteral)
+      .join(", ");
+    this.getChangesQueryText = `SELECT lsn, data FROM pg_catalog.pg_logical_slot_get_changes($1, $2, $3, ${parametersSql})`;
     this.parse = types
       ? (value: any, typeOid: number) => {
           if (value === null) return null;
-          // wal2json always outputs `bool`s as boolean, not as string, irregardless of `numeric-data-types-as-string`.
-          if (typeOid === pg.types.builtins.BOOL) return value;
-          // FIXME: this should use `client.getTypeParser` or have some other non-global option to configure type parsing
+          // wal2json always outputs `bool`s as boolean
+          if (typeof value === "boolean") return value; // assert: typeOid === pg.types.builtins.BOOL
+          // wal2json outputs numeric data as numbers, unless `numeric-data-types-as-string` is set
+          if (typeof value === "number") return value;
           const parser = pg.types.getTypeParser(typeOid, "text");
           return parser(value);
         }
       : id;
     // We just use the pool to get better error handling
     this.pool = new pg.Pool({
-      connectionString: this.connectionString,
+      connectionString,
       max: 1,
     });
     this.pool.on("error", this.onPoolError);
+    this.client = null;
   }
 
   public async dropStaleSlots() {
@@ -182,9 +195,9 @@ export default class PgLogicalDecoding extends EventEmitter {
     const client = await this.getClient();
     await this.trackSelf(client);
     try {
-      const { rows } = await client.query({
-        text: `SELECT lsn, data FROM pg_catalog.pg_logical_slot_get_changes($1, $2, $3, 'add-tables', $4::text, 'include-types', 'f', 'include-type-oids', 't', 'numeric-data-types-as-string', 't')`,
-        values: [this.slotName, uptoLsn, uptoNchanges, this.tablePattern],
+      const { rows } = await client.query<[lsn: string, data: string]>({
+        text: this.getChangesQueryText,
+        values: [this.slotName, uptoLsn, uptoNchanges],
         rowMode: "array",
       });
       return rows.map(toLsnData);