Merge branch 'main' into enchance-profile-search

src/pipeline.ts

@@ -5,6 +5,7 @@ import { LRUCache } from 'lru-cache';
 import { z } from 'zod';
 
 import { Conf } from '@/config.ts';
+import { DittoTables } from '@/db/DittoTables.ts';
 import { DittoEvent } from '@/interfaces/DittoEvent.ts';
 import { pipelineEventsCounter, policyEventsCounter } from '@/metrics.ts';
 import { RelayError } from '@/RelayError.ts';
@@ -14,11 +15,11 @@ import { Storages } from '@/storages.ts';
 import { eventAge, parseNip05, Time } from '@/utils.ts';
 import { policyWorker } from '@/workers/policy.ts';
 import { verifyEventWorker } from '@/workers/verify.ts';
+import { getAmount } from '@/utils/bolt11.ts';
 import { nip05Cache } from '@/utils/nip05.ts';
+import { purifyEvent } from '@/utils/purify.ts';
 import { updateStats } from '@/utils/stats.ts';
 import { getTagSet } from '@/utils/tags.ts';
-import { getAmount } from '@/utils/bolt11.ts';
-import { DittoTables } from '@/db/DittoTables.ts';
 
 const debug = Debug('ditto:pipeline');
 
@@ -55,7 +56,7 @@ async function handleEvent(event: DittoEvent, signal: AbortSignal): Promise<void
   const kysely = await Storages.kysely();
 
   await Promise.all([
-    storeEvent(event, signal),
+    storeEvent(purifyEvent(event), signal),
     handleZaps(kysely, event),
     handleAuthorSearch(kysely, event),
     parseMetadata(event, signal),
@@ -118,10 +119,11 @@ async function hydrateEvent(event: DittoEvent, signal: AbortSignal): Promise<voi
 async function storeEvent(event: DittoEvent, signal?: AbortSignal): Promise<undefined> {
   if (NKinds.ephemeral(event.kind)) return;
   const store = await Storages.db();
-  const kysely = await Storages.kysely();
 
-  await updateStats({ event, store, kysely }).catch(debug);
+  await store.transaction(async (store, kysely) => {
+    await updateStats({ event, store, kysely });
     await store.event(event, { signal });
+  });
 }
 
 /** Parse kind 0 metadata and track indexes in the database. */

src/storages.ts

@@ -12,7 +12,7 @@ import { seedZapSplits } from '@/utils/zap-split.ts';
 
 export class Storages {
   private static _db: Promise<EventsDB> | undefined;
-  private static _database: DittoDatabase | undefined;
+  private static _database: Promise<DittoDatabase> | undefined;
   private static _admin: Promise<AdminStore> | undefined;
   private static _client: Promise<NPool> | undefined;
   private static _pubsub: Promise<InternalRelay> | undefined;
@@ -20,8 +20,11 @@ export class Storages {
 
   public static async database(): Promise<DittoDatabase> {
     if (!this._database) {
-      this._database = DittoDB.create(Conf.databaseUrl, { poolSize: Conf.pg.poolSize });
-      await DittoDB.migrate(this._database.kysely);
+      this._database = (async () => {
+        const db = DittoDB.create(Conf.databaseUrl, { poolSize: Conf.pg.poolSize });
+        await DittoDB.migrate(db.kysely);
+        return db;
+      })();
     }
     return this._database;
   }
@@ -35,7 +38,7 @@ export class Storages {
   public static async db(): Promise<EventsDB> {
     if (!this._db) {
       this._db = (async () => {
-        const { kysely } = await this.database();
+        const kysely = await this.kysely();
         const store = new EventsDB({ kysely, pubkey: Conf.pubkey, timeout: Conf.db.timeouts.default });
         await seedZapSplits(store);
         return store;

src/storages/EventsDB.ts

@@ -1,17 +1,7 @@
 // deno-lint-ignore-file require-await
 
-import { NPostgres } from '@nostrify/db';
-import {
-  NIP50,
-  NKinds,
-  NostrEvent,
-  NostrFilter,
-  NostrRelayCLOSED,
-  NostrRelayEOSE,
-  NostrRelayEVENT,
-  NSchema as n,
-  NStore,
-} from '@nostrify/nostrify';
+import { NPostgres, NPostgresSchema } from '@nostrify/db';
+import { NIP50, NKinds, NostrEvent, NostrFilter, NSchema as n } from '@nostrify/nostrify';
 import { Stickynotes } from '@soapbox/stickynotes';
 import { Kysely } from 'kysely';
 import { nip27 } from 'nostr-tools';
@@ -41,8 +31,7 @@ interface EventsDBOpts {
 }
 
 /** SQL database storage adapter for Nostr events. */
-class EventsDB implements NStore {
-  private store: NPostgres;
+class EventsDB extends NPostgres {
   private console = new Stickynotes('ditto:db:events');
 
   /** Conditions for when to index certain tags. */
@@ -63,7 +52,7 @@ class EventsDB implements NStore {
   };
 
   constructor(private opts: EventsDBOpts) {
-    this.store = new NPostgres(opts.kysely, {
+    super(opts.kysely, {
       indexTags: EventsDB.indexTags,
       indexSearch: EventsDB.searchText,
     });
@@ -82,7 +71,7 @@ class EventsDB implements NStore {
     await this.deleteEventsAdmin(event);
 
     try {
-      await this.store.event(event, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
+      await super.event(event, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
     } catch (e) {
       if (e.message === 'Cannot add a deleted event') {
         throw new RelayError('blocked', 'event deleted by user');
@@ -155,12 +144,9 @@ class EventsDB implements NStore {
     }
   }
 
-  /** Stream events from the database. */
-  req(
-    filters: NostrFilter[],
-    opts: { signal?: AbortSignal } = {},
-  ): AsyncIterable<NostrRelayEVENT | NostrRelayEOSE | NostrRelayCLOSED> {
-    return this.store.req(filters, opts);
+  protected getFilterQuery(trx: Kysely<NPostgresSchema>, filter: NostrFilter) {
+    const query = super.getFilterQuery(trx, filter);
+    return query;
   }
 
   /** Get events for filters from the database. */
@@ -185,32 +171,28 @@ class EventsDB implements NStore {
     }
 
     if (opts.signal?.aborted) return Promise.resolve([]);
-    if (!filters.length) return Promise.resolve([]);
 
     this.console.debug('REQ', JSON.stringify(filters));
 
-    return this.store.query(filters, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
+    return super.query(filters, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
   }
 
   /** Delete events based on filters from the database. */
   async remove(filters: NostrFilter[], opts: { signal?: AbortSignal; timeout?: number } = {}): Promise<void> {
-    if (!filters.length) return Promise.resolve();
     this.console.debug('DELETE', JSON.stringify(filters));
-
-    return this.store.remove(filters, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
+    return super.remove(filters, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
   }
 
   /** Get number of events that would be returned by filters. */
   async count(
     filters: NostrFilter[],
     opts: { signal?: AbortSignal; timeout?: number } = {},
-  ): Promise<{ count: number; approximate: boolean }> {
+  ): Promise<{ count: number; approximate: any }> {
     if (opts.signal?.aborted) return Promise.reject(abortError());
-    if (!filters.length) return Promise.resolve({ count: 0, approximate: false });
 
     this.console.debug('COUNT', JSON.stringify(filters));
 
-    return this.store.count(filters, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
+    return super.count(filters, { ...opts, timeout: opts.timeout ?? this.opts.timeout });
   }
 
   /** Return only the tags that should be indexed. */
@@ -316,6 +298,10 @@ class EventsDB implements NStore {
 
     return filters;
   }
+
+  async transaction(callback: (store: NPostgres, kysely: Kysely<any>) => Promise<void>): Promise<void> {
+    return super.transaction((store, kysely) => callback(store, kysely as unknown as Kysely<DittoTables>));
+  }
 }
 
 export { EventsDB };