Expose a generic listen method on the database adapters

src/db/DittoDatabase.ts

@@ -1,4 +1,3 @@
-import { NostrEvent } from '@nostrify/nostrify';
 import { Kysely } from 'kysely';
 
 import { DittoTables } from '@/db/DittoTables.ts';
@@ -7,7 +6,7 @@ export interface DittoDatabase {
   readonly kysely: Kysely<DittoTables>;
   readonly poolSize: number;
   readonly availableConnections: number;
-  readonly listenNostr: (onEvent: (event: NostrEvent) => void) => void;
+  listen(channel: string, callback: (payload: string) => void): void;
 }
 
 export interface DittoDatabaseOpts {

src/db/adapters/DittoPglite.ts

@@ -1,6 +1,5 @@
 import { PGlite } from '@electric-sql/pglite';
 import { pg_trgm } from '@electric-sql/pglite/contrib/pg_trgm';
-import { NostrEvent } from '@nostrify/nostrify';
 import { PgliteDialect } from '@soapbox/kysely-pglite';
 import { Kysely } from 'kysely';
 
@@ -27,17 +26,15 @@ export class DittoPglite {
       log: KyselyLogger,
     });
 
-    const listenNostr = (onEvent: (event: NostrEvent) => void): void => {
+    const listen = (channel: string, callback: (payload: string) => void): void => {
-      pglite.listen('nostr_event', (payload) => {
+      pglite.listen(channel, callback);
-        onEvent(JSON.parse(payload));
-      });
     };
 
     return {
       kysely,
       poolSize: 1,
       availableConnections: 1,
-      listenNostr,
+      listen,
     };
   }
 }

src/db/adapters/DittoPostgres.ts

@@ -1,4 +1,3 @@
-import { NostrEvent } from '@nostrify/nostrify';
 import {
   BinaryOperationNode,
   FunctionNode,
@@ -41,10 +40,8 @@ export class DittoPostgres {
       log: KyselyLogger,
     });
 
-    const listenNostr = (onEvent: (event: NostrEvent) => void): void => {
+    const listen = (channel: string, callback: (payload: string) => void): void => {
-      pg.listen('nostr_event', (payload) => {
+      pg.listen(channel, callback);
-        onEvent(JSON.parse(payload));
-      });
     };
 
     return {
@@ -55,7 +52,7 @@ export class DittoPostgres {
       get availableConnections() {
         return pg.connections.idle;
       },
-      listenNostr,
+      listen,
     };
   }
 }

src/notify.ts

@@ -6,11 +6,12 @@ import { Storages } from '@/storages.ts';
 const sem = new Semaphore(1);
 
 export async function startNotify(): Promise<void> {
-  const { listenNostr } = await Storages.database();
+  const { listen } = await Storages.database();
 
-  listenNostr((event) => {
+  listen('nostr_event', (payload) => {
     sem.lock(async () => {
       try {
+        const event = JSON.parse(payload);
         await pipeline.handleEvent(event, AbortSignal.timeout(5000));
       } catch (e) {
         console.warn(e);