fix(lds): Parse wal2json string values according to their type

[wal2json serialises all (non-numeric, non-boolean, non-null) database values as strings](https://github.com/eulerto/wal2json/blob/f8bab055fafc196539c9652b6b5584228498c6eb/wal2json.c#L1337-L1384) into the JSON,
and so far we failed to take their types into account when converting them to records.
This now
* makes wal2json output strings as far as possible (in particular for numeric types whose values did not always have a JSON number representation)
* parses the values [just like `node-postgres` does it](https://github.com/brianc/node-postgres/blob/32ec730b51a1fd73bf97d65105c949729fa9ec80/packages/pg/lib/result.js#L68-L73)
* instructs wal2json to output the type oids instead of the type names for this

Author: ab-pm

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

@@ -17,6 +17,7 @@ declare module "pg" {
 interface Keys {
   keynames: Array<string>;
   keytypes: Array<string>;
+  keytypeoids: Array<number>;
   keyvalues: Array<any>;
 }
 
@@ -34,7 +35,8 @@ export interface InsertChange extends Change {
 
   // https://github.com/eulerto/wal2json/blob/f81bf7af09324da656be87dfd53d20741c01e1e0/wal2json.c#L969
   columnnames: Array<string>;
-  columntypes: Array<string>;
+  columntypes: Array<string>; // with `include-types` option (default true)
+  columntypeoids: Array<number>; // with `include-type-oids` option (default false)
   columnvalues: Array<any>;
 }
 
@@ -43,7 +45,8 @@ export interface UpdateChange extends Change {
 
   // https://github.com/eulerto/wal2json/blob/f81bf7af09324da656be87dfd53d20741c01e1e0/wal2json.c#L973
   columnnames: Array<string>;
-  columntypes: Array<string>;
+  columntypes: Array<string>; // with `include-types` option (default true)
+  columntypeoids: Array<number>; // with `include-type-oids` option (default false)
   columnvalues: Array<any>;
 
   // https://github.com/eulerto/wal2json/blob/f81bf7af09324da656be87dfd53d20741c01e1e0/wal2json.c#L992-L1003
@@ -57,16 +60,26 @@ export interface DeleteChange extends Change {
   oldkeys: Keys;
 }
 
+const parse = (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
+  const parser = pg.types.getTypeParser(typeOid, "text");
+  return parser(value);
+};
+
 export const changeToRecord = (change: InsertChange | UpdateChange) => {
-  const { columnnames, columnvalues } = change;
-  return columnnames.reduce((memo, name, i) => {
-    memo[name] = columnvalues[i];
+  const { columnnames, columnvalues, columntypeoids } = change;
+  return columnnames.reduce<Record<string, any>>((memo, name, i) => {
+    memo[name] = parse(columnvalues[i], columntypeoids[i]);
     return memo;
   }, {});
 };
 
 export const changeToPk = (change: UpdateChange | DeleteChange) => {
-  return change.oldkeys.keyvalues;
+  const { keyvalues, keytypeoids } = change.oldkeys;
+  return keyvalues.map((value, i) => parse(value, keytypeoids[i]));
 };
 
 interface Payload {
@@ -173,7 +186,7 @@ export default class PgLogicalDecoding extends EventEmitter {
     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)`,
+        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],
         rowMode: "array",
       });