Wallet Connect Cache Clear

packages/engine/paima-sm/src/delegate-wallet.ts

@@ -7,7 +7,7 @@ import { CryptoManager } from '@paima/crypto';
 import type { IGetAddressFromAddressResult } from '@paima/db';
 import {
   addressCache,
-  deleteDelegationsFrom,
+  deleteDelegationTo,
   getAddressFromAddress,
   getAddressFromId,
   getDelegation,
@@ -241,7 +241,7 @@ export class DelegateWallet {
   private async cmdCancelDelegations(to: string): Promise<void> {
     const [toAddress] = await getAddressFromAddress.run({ address: to }, this.DBConn);
     if (!toAddress) throw new Error('Invalid Address');
-    await deleteDelegationsFrom.run({ from_id: toAddress.id }, this.DBConn);
+    await deleteDelegationTo.run({ to_id: toAddress.id }, this.DBConn);
 
     // TODO this is clears the entire cache. We can only clear necessary elements.
     addressCache.clear();

packages/engine/paima-sm/src/index.ts

@@ -1,17 +1,11 @@
 import { Pool } from 'pg';
-import type { PoolClient } from 'pg';
+import type { PoolClient, Client } from 'pg';
 
-import {
-  ChainDataExtensionDatumType,
-  doLog,
-  ENV,
-  Network,
-  SCHEDULED_DATA_ADDRESS,
-  SCHEDULED_DATA_ID,
-} from '@paima/utils';
+import { doLog, ENV, Network, SCHEDULED_DATA_ADDRESS, SCHEDULED_DATA_ID } from '@paima/utils';
 import {
   tx,
   getConnection,
+  getPersistentConnection,
   initializePaimaTables,
   storeGameInput,
   blockHeightDone,
@@ -53,6 +47,7 @@ const SM: GameStateMachineInitializer = {
     startBlockHeight
   ) => {
     const DBConn: Pool = getConnection(databaseInfo);
+    const persistentReadonlyDBConn: Client = getPersistentConnection(databaseInfo);
     const readonlyDBConn: Pool = getConnection(databaseInfo, true);
 
     return {
@@ -90,6 +85,9 @@ const SM: GameStateMachineInitializer = {
       getReadonlyDbConn: (): Pool => {
         return readonlyDBConn;
       },
+      getPersistentReadonlyDbConn: (): Client => {
+        return persistentReadonlyDBConn;
+      },
       getReadWriteDbConn: (): Pool => {
         return DBConn;
       },

packages/engine/paima-sm/src/types.ts

@@ -1,4 +1,4 @@
-import type { Pool, PoolClient, PoolConfig } from 'pg';
+import type { Client, Pool, PoolClient, PoolConfig } from 'pg';
 
 import type { SQLUpdate } from '@paima/db';
 import type {
@@ -325,6 +325,7 @@ export interface GameStateMachine {
   getPresyncBlockHeight: (dbTx?: PoolClient | Pool) => Promise<number>;
   getPresyncCardanoSlotHeight: (dbTx?: PoolClient | Pool) => Promise<number>;
   getReadonlyDbConn: () => Pool;
+  getPersistentReadonlyDbConn: () => Client;
   getReadWriteDbConn: () => Pool;
   process: (dbTx: PoolClient, chainData: ChainData) => Promise<void>;
   presyncProcess: (dbTx: PoolClient, latestCdeData: PresyncChainData) => Promise<void>;

packages/node-sdk/paima-db/migrations/up.sql

@@ -98,6 +98,59 @@ CREATE TABLE delegations (
  PRIMARY KEY (from_id, to_id)
 );
 
+-- Create a function to notify any change in address or delegate tables
+create or replace function public.notify_wallet_connect()
+  returns trigger
+  language plpgsql
+as $function$ 
+DECLARE
+  rec RECORD;
+  payload TEXT;
+  column_name TEXT;
+  column_value TEXT;
+  payload_items TEXT[];
+begin
+  CASE TG_OP
+    WHEN 'INSERT', 'UPDATE' THEN
+      rec := NEW;
+    WHEN 'DELETE' THEN
+      rec := OLD;
+    ELSE
+      RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP;
+  END CASE;
+
+   -- Get required fields
+  FOREACH column_name IN ARRAY TG_ARGV LOOP
+    EXECUTE format('SELECT $1.%I::TEXT', column_name)
+    INTO column_value
+    USING rec;
+    payload_items := array_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"');
+  END LOOP;
+	  -- Build the payload
+  payload := ''
+              || '{'
+              || '"timestamp":"' || CURRENT_TIMESTAMP                    || '",'
+              || '"operation":"' || TG_OP                                || '",'
+              || '"schema":"'    || TG_TABLE_SCHEMA                      || '",'
+              || '"table":"'     || TG_TABLE_NAME                        || '",'
+              || '"data":{'      || array_to_string(payload_items, ',')  || '}'
+              || '}';
+
+	perform pg_notify('wallet_connect_change', payload);
+ 	return null;
+end;
+$function$
+;
+
+CREATE OR REPLACE TRIGGER wallet_connect_insert_or_update
+  AFTER INSERT OR UPDATE or delete ON addresses
+  for each row execute procedure notify_wallet_connect('id', 'address');
+
+CREATE OR REPLACE TRIGGER wallet_connect_insert_or_update
+  AFTER INSERT OR UPDATE or delete ON delegations
+  for each row execute procedure notify_wallet_connect('from_id', 'to_id');
+
+
 CREATE TABLE cde_cardano_pool_delegation (
   cde_id INTEGER NOT NULL,
   address TEXT NOT NULL,

packages/node-sdk/paima-db/src/database-validation.ts

@@ -1,7 +1,7 @@
 import { doLog } from '@paima/utils';
 import type { PoolClient } from 'pg';
-import { TABLES } from './paima-tables.js';
-import { createTable, tableExists, tableIsValid } from './postgres-metadata.js';
+import { FUNCTIONS, TABLES } from './paima-tables.js';
+import { createFunctions, createTable, tableExists, tableIsValid } from './postgres-metadata.js';
 import type { TableData } from './table-types.js';
 
 const FAILURE_MESSAGE: string = `Please remove these tables from your database or update them to conform with
@@ -30,6 +30,15 @@ export async function initializePaimaTables(
     }
   }
 
+  for (const func of FUNCTIONS) {
+    try {
+      await createFunctions(pool, func);
+    } catch (err) {
+      doLog(`Error while initializing ${func}: ${err}`);
+      noIssues = false;
+    }
+  }
+
   if (!force && invalidTables.length > 0) {
     doLog('The following internal Paima tables were detected but have invalid structure:');
     for (const tableName of invalidTables) {

packages/node-sdk/paima-db/src/delegate-wallet.ts

@@ -1,3 +1,4 @@
+import { doLog } from '@paima/utils';
 import type {
   IGetDelegationsFromWithAddressResult,
   IGetDelegationsToWithAddressResult,
@@ -8,7 +9,7 @@ import {
   getDelegationsToWithAddress,
   getMainAddressFromAddress,
 } from './sql/wallet-delegation.queries.js';
-import type { PoolClient } from 'pg';
+import type { PoolClient, Notification, Client } from 'pg';
 
 export type WalletDelegate = { address: string; id: number };
 export const NO_USER_ID = -1;
@@ -18,6 +19,7 @@ export const NO_USER_ID = -1;
  *  This is a temporal fix as lru-cache module was
  *  not correctly packaged into a single file.
  */
+let useAddressCache = false;
 export const addressCache = new Map<string, WalletDelegate>();
 
 // Get Main Wallet and ID for address.
@@ -28,7 +30,7 @@ export async function getMainAddress(
 ): Promise<WalletDelegate> {
   const address = _address.toLocaleLowerCase();
   let addressMapping: WalletDelegate | undefined = addressCache.get(address);
-  if (addressMapping) return addressMapping;
+  if (useAddressCache && addressMapping) return addressMapping;
 
   // get main address.
   const [addressResult] = await getMainAddressFromAddress.run({ address }, DBConn);
@@ -78,3 +80,33 @@ export async function getRelatedWallets(
     id: to.length ? to[0].id : addressResult.id,
   };
 }
+
+// This improves performance by caching the results of the queries.
+// It is not enabled by default because `clearDelegateWalletCacheOnChanges` must also be called from client.
+export function enableDelegateWalletCache(): void {
+  useAddressCache = true;
+  addressCache.clear();
+}
+
+let isListening = false;
+export async function clearDelegateWalletCacheOnChanges(DBConn: Client): Promise<void> {
+  if (isListening) throw new Error('Already listening to wallet connect updates');
+  doLog('Listening to wallet connect updates');
+  await DBConn.query('LISTEN wallet_connect_change');
+  DBConn.on('notification', (_: Notification) => {
+    // { ...
+    //   payload: '{
+    //     "timestamp":"2023-12-18 14:26:57.186068-03",
+    //     "operation":"DELETE", /* INSERT | UPDATE | DELETE */
+    //     "schema":"public",
+    //     "table":"addresses", /* addresses | delegations */
+    //     "data":{"id":"16","address":"23"} /* addresses or delegations table fields */
+    //   }',
+    // }
+    addressCache.clear();
+  });
+  DBConn.on('end', () => {
+    doLog('Stopped listening to wallet connect updates?!');
+  });
+  isListening = true;
+}

packages/node-sdk/paima-db/src/index.ts

@@ -1,5 +1,5 @@
 import { tx } from './pg-tx.js';
-import { getConnection } from './pg-connection.js';
+import { getConnection, getPersistentConnection } from './pg-connection.js';
 import { createScheduledData, deleteScheduledData } from './scheduled-constructors.js';
 import { initializePaimaTables } from './database-validation.js';
 import { DataMigrations } from './data-migrations.js';
@@ -41,6 +41,7 @@ export type * from './sql/cde-tracking-cardano.queries.js';
 export {
   tx,
   getConnection,
+  getPersistentConnection,
   createScheduledData,
   deleteScheduledData,
   initializePaimaTables,

packages/node-sdk/paima-db/src/paima-tables.ts

@@ -334,6 +334,67 @@ const TABLE_DATA_DELEGATIONS: TableData = {
   creationQuery: QUERY_CREATE_TABLE_DELEGATIONS,
 };
 
+const FUNCTION_NOTIFY_WALLET_CONNECT: string = `
+create or replace function public.notify_wallet_connect()
+  returns trigger
+  language plpgsql
+as $function$ 
+DECLARE
+  rec RECORD;
+  payload TEXT;
+  column_name TEXT;
+  column_value TEXT;
+  payload_items TEXT[];
+begin
+  CASE TG_OP
+    WHEN 'INSERT', 'UPDATE' THEN
+      rec := NEW;
+    WHEN 'DELETE' THEN
+      rec := OLD;
+    ELSE
+      RAISE EXCEPTION 'Unknown TG_OP: "%". Should not occur!', TG_OP;
+  END CASE;
+ 
+   -- Get required fields
+  FOREACH column_name IN ARRAY TG_ARGV LOOP
+    EXECUTE format('SELECT $1.%I::TEXT', column_name)
+    INTO column_value
+    USING rec;
+    payload_items := array_append(payload_items, '"' || replace(column_name, '"', '\"') || '":"' || replace(column_value, '"', '\"') || '"');
+  END LOOP;
+	  -- Build the payload
+  payload := ''
+              || '{'
+              || '"timestamp":"' || CURRENT_TIMESTAMP                    || '",'
+              || '"operation":"' || TG_OP                                || '",'
+              || '"schema":"'    || TG_TABLE_SCHEMA                      || '",'
+              || '"table":"'     || TG_TABLE_NAME                        || '",'
+              || '"data":{'      || array_to_string(payload_items, ',')  || '}'
+              || '}';
+             
+	perform pg_notify('wallet_connect_change', payload);
+ 	return null;
+end;
+$function$
+;`;
+
+const FUNCTION_TRIGGER_ADDRESSES: string = `
+CREATE OR REPLACE TRIGGER wallet_connect_insert_or_update
+  AFTER INSERT OR UPDATE or delete ON addresses
+  for each row execute procedure notify_wallet_connect('id', 'address');
+`;
+
+const FUNCTION_TRIGGER_DELEGATIONS: string = `
+CREATE OR REPLACE TRIGGER wallet_connect_insert_or_update
+  AFTER INSERT OR UPDATE or delete ON delegations
+  for each row execute procedure notify_wallet_connect('from_id', 'to_id');
+`;
+
+export const FUNCTIONS: string[] = [
+  FUNCTION_NOTIFY_WALLET_CONNECT,
+  FUNCTION_TRIGGER_ADDRESSES,
+  FUNCTION_TRIGGER_DELEGATIONS,
+];
 export const TABLES: TableData[] = [
   TABLE_DATA_BLOCKHEIGHTS,
   TABLE_DATA_NONCES,

packages/node-sdk/paima-db/src/pg-connection.ts

@@ -1,4 +1,4 @@
-import type { PoolClient, PoolConfig } from 'pg';
+import type { Client, PoolClient, PoolConfig } from 'pg';
 import pg from 'pg';
 
 import { logError } from '@paima/utils';
@@ -25,3 +25,15 @@ export const getConnection = (creds: PoolConfig, readonly = false): pg.Pool => {
 
   return pool;
 };
+
+// For notifications use a direct (non-pooled) persistent connection.
+export const getPersistentConnection = (creds: PoolConfig): Client => {
+  const client = new pg.Client(creds);
+  client.connect(() => {});
+  client.on('error', err => logError(err));
+  // On each new client initiated, need to register for error(this is a serious bug on pg, the client throw errors although it should not)
+  client.on('error', (err: Error) => {
+    logError(err);
+  });
+  return client;
+};

packages/node-sdk/paima-db/src/postgres-metadata.ts

@@ -59,6 +59,17 @@ export async function createTable(pool: PoolClient, table: TableData): Promise<b
   return true;
 }
 
+export async function createFunctions(pool: PoolClient, query: string): Promise<boolean> {
+  try {
+    await pool.query(query);
+  } catch (err) {
+    doLog(`[database-validation] Error while creating functions: ${err}`);
+    return false;
+  }
+
+  return true;
+}
+
 async function checkTablePkey(
   pool: PoolClient,
   tableName: string,